[#452] implement client redirection to resource server

Also refined session clone() method in order to implement the redirect.

Note: Instead of GET_HOST_FOR_{GET|PUT}, we're now using the newer
GET_RESOURCE_INFO_FOR_OPERATION api.  This is primarily for discovery
of the proper resource hierarchy string to apply to the OPEN operation
once we've redirected to the proper server.

Resulting from the use of the new api, the client redirect feature
will be available only in iRODS >= 4.3.1.

Slight corrections were made to eliminate errors introduced into older
tests by these changes:

- In test_that_connection_timeout_works__issue_377, we update a variable
  'sess' reference to make sure network timeout change is applied to the
  proper session.

- enableLogging() has moved to the irods.test.helpers module for use in
  tests generally.

Author: d-w-moore

irods/account.py

@@ -5,6 +5,11 @@ def __init__(self, irods_host, irods_port, irods_user_name, irods_zone_name,
                  password=None, client_user=None,
                  server_dn=None, client_zone=None, **kwargs):
 
+        # Allowed overrides when cloning sessions. (Currently hostname only.)
+        for k,v in kwargs.pop('_overrides',{}).items():
+            if k =='irods_host':
+                irods_host = v
+
         self.authentication_scheme = irods_authentication_scheme.lower()
         self.host = irods_host
         self.port = int(irods_port)

irods/api_number.py

@@ -176,6 +176,7 @@
     # 1100 - 1200 - SSL API calls
     "SSL_START_AN": 1100,
     "SSL_END_AN": 1101,
+    "GET_RESOURCE_INFO_FOR_OPERATION_AN":10220,
     "ATOMIC_APPLY_METADATA_OPERATIONS_APN": 20002,
     "GET_FILE_DESCRIPTOR_INFO_APN": 20000,
     "REPLICA_CLOSE_APN": 20004

irods/data_object.py

@@ -124,6 +124,8 @@ class iRODSDataObjectFileRaw(io.RawIOBase):
     """The raw object supporting file-like operations (read/write/seek) for the
        iRODSDataObject."""
 
+    session = None      # codacy
+
     def __init__(self, conn, descriptor, finalize_on_close = True, **options):
         """
         Constructor needs a connection and an iRODS data object descriptor. If the

irods/keywords.py

@@ -100,6 +100,7 @@
 TICKET_KW = "ticket"       # for ticket-based-access #
 PURGE_CACHE_KW = "purgeCache"   # purge the cache copy right after the operation JMC - backport 4537
 EMPTY_BUNDLE_ONLY_KW = "emptyBundleOnly"  # delete emptyBundleOnly # # JMC - backport 4552
+GET_RESOURCE_INFO_OP_TYPE_KW = "getResourceInfoOpType"
 
 # =-=-=-=-=-=-=-
 # JMC - backport 4599
@@ -247,6 +248,7 @@
 REPL_REQUESTED_KW = "repl_requested"
 # borrowed IN_PDMO_KW
 
+
 # =-=-=-=-=-=-=-
 # irods structured_object keyword definitions
 HOST_ADDR_KW = "host_addr"

irods/manager/__init__.py

@@ -12,4 +12,7 @@ def server_version(self):
         return tuple( self.__server_version )
 
     def __init__(self, sess):
+        self._set_manager_session(sess)
+
+    def _set_manager_session(self, sess):
         self.sess = sess

irods/manager/data_object_manager.py

@@ -5,7 +5,8 @@
 from irods.manager import Manager
 from irods.message import (
     iRODSMessage, FileOpenRequest, ObjCopyRequest, StringStringMap, DataObjInfo, ModDataObjMeta,
-    DataObjChksumRequest, DataObjChksumResponse, RErrorStack)
+    DataObjChksumRequest, DataObjChksumResponse, RErrorStack, STR_PI
+    )
 import irods.exception as ex
 from irods.api_number import api_number
 from irods.collection import iRODSCollection
@@ -19,14 +20,14 @@
 import json
 import logging
 
-
 MAXIMUM_SINGLE_THREADED_TRANSFER_SIZE = 32 * ( 1024 ** 2)
 
 DEFAULT_NUMBER_OF_THREADS = 0   # Defaults for reasonable number of threads -- optimized to be
                                 # performant but allow no more worker threads than available CPUs.
 
 DEFAULT_QUEUE_DEPTH = 32
 
+logger = logging.getLogger(__name__)
 
 class Server_Checksum_Warning(Exception):
     """Error from iRODS server indicating some replica checksums are missing or incorrect."""
@@ -96,16 +97,18 @@ def _download(self, obj, local_path, num_threads, **options):
         if os.path.exists(local_file) and kw.FORCE_FLAG_KW not in options:
             raise ex.OVERWRITE_WITHOUT_FORCE_FLAG
 
-        with open(local_file, 'wb') as f, self.open(obj, 'r', **options) as o:
-
-            if self.should_parallelize_transfer (num_threads, o):
-                f.close()
-                if not self.parallel_get( (obj,o), local_path, num_threads = num_threads,
-                                          target_resource_name = options.get(kw.RESC_NAME_KW,'')):
-                    raise RuntimeError("parallel get failed")
-            else:
-                for chunk in chunks(o, self.READ_BUFFER_SIZE):
-                    f.write(chunk)
+        data_open_returned_values_ = {}
+        with open(local_file, 'wb') as f:
+            with self.open(obj, 'r', returned_values = data_open_returned_values_, **options) as o:
+                if self.should_parallelize_transfer (num_threads, o):
+                    f.close()
+                    if not self.parallel_get( (obj,o), local_path, num_threads = num_threads,
+                                              target_resource_name = options.get(kw.RESC_NAME_KW,''),
+                                              data_open_returned_values = data_open_returned_values_):
+                        raise RuntimeError("parallel get failed")
+                else:
+                    for chunk in chunks(o, self.READ_BUFFER_SIZE):
+                        f.write(chunk)
 
 
     def get(self, path, local_path = None, num_threads = DEFAULT_NUMBER_OF_THREADS, **options):
@@ -214,6 +217,7 @@ def parallel_get(self,
                      async_ = False,
                      num_threads = 0,
                      target_resource_name = '',
+                     data_open_returned_values = None,
                      progressQueue = False):
         """Call into the irods.parallel library for multi-1247 GET.
 
@@ -224,6 +228,7 @@ def parallel_get(self,
         """
         return parallel.io_main( self.sess, data_or_path_, parallel.Oper.GET | (parallel.Oper.NONBLOCKING if async_ else 0), file_,
                                  num_threads = num_threads, target_resource_name = target_resource_name,
+                                 data_open_returned_values = data_open_returned_values,
                                  queueLength = (DEFAULT_QUEUE_DEPTH if progressQueue else 0))
 
     def parallel_put(self,
@@ -296,7 +301,9 @@ def open_with_FileRaw(self, *arg, **kw_options):
                                kw.RESC_HIER_STR_KW
                            ))
 
-    def open(self, path, mode, create = True, finalize_on_close = True, **options):
+
+    def open(self, path, mode, create = True, finalize_on_close = True, returned_values = None, allow_redirect = True, **options):
+
         _raw_fd_holder =  options.get('_raw_fd_holder',[])
         # If no keywords are used that would influence the server as to the choice of a storage resource,
         # then use the default resource in the client configuration.
@@ -317,29 +324,79 @@ def open(self, path, mode, create = True, finalize_on_close = True, **options):
         }[mode]
         # TODO: Use seek_to_end
 
+        if not isinstance(returned_values, dict):
+            returned_values = {}
+
         try:
             oprType = options[kw.OPR_TYPE_KW]
         except KeyError:
             oprType = 0
 
-        message_body = FileOpenRequest(
-            objPath=path,
-            createMode=0,
-            openFlags=flags,
-            offset=0,
-            dataSize=-1,
-            numThreads=self.sess.numThreads,
-            oprType=oprType,
-            KeyValPair_PI=StringStringMap(options),
-        )
-        message = iRODSMessage('RODS_API_REQ', msg=message_body,
-                               int_info=api_number['DATA_OBJ_OPEN_AN'])
+        def make_FileOpenRequest(**extra_opts):
+            options_ = dict(options) if extra_opts else options
+            options_.update(extra_opts)
+            return  FileOpenRequest(
+                objPath=path,
+                createMode=0,
+                openFlags=flags,
+                offset=0,
+                dataSize=-1,
+                numThreads=self.sess.numThreads,
+                oprType=oprType,
+                KeyValPair_PI=StringStringMap(options_),
+            )
+
+        requested_hierarchy = options.get(kw.RESC_HIER_STR_KW, None)
 
         conn = self.sess.pool.get_connection()
+        redirected_host = ''
+
+        use_get_rescinfo_apis = False
+
+        if allow_redirect and conn.server_version >= (4,3,1):
+            key = 'CREATE' if mode[0] in ('w','a') else 'OPEN'
+            message = iRODSMessage('RODS_API_REQ',
+                                   msg=make_FileOpenRequest(**{kw.GET_RESOURCE_INFO_OP_TYPE_KW:key}),
+                                   int_info=api_number['GET_RESOURCE_INFO_FOR_OPERATION_AN'])
+            conn.send(message)
+            response = conn.recv()
+            msg = response.get_main_message( STR_PI )
+            use_get_rescinfo_apis = True
+
+            # Get the information needed for the redirect
+            _ = json.loads(msg.myStr)
+            redirected_host = _["host"]
+            requested_hierarchy = _["resource_hierarchy"]
+
+        target_zone = list(filter(None, path.split('/')))
+        if target_zone:
+            target_zone = target_zone[0]
+
+        directed_sess = self.sess
+
+        if redirected_host and use_get_rescinfo_apis:
+            # Redirect only if the local zone is being targeted, and if the hostname is changed from the original.
+            if target_zone == self.sess.zone and (self.sess.host != redirected_host):
+                # This is the actual redirect.
+                directed_sess = self.sess.clone(host = redirected_host)
+                returned_values['session'] = directed_sess
+                conn = directed_sess.pool.get_connection()
+                logger.debug('redirect_to_host = %s', redirected_host)
+
+        # Restore RESC HIER for DATA_OBJ_OPEN call
+        if requested_hierarchy is not None:
+            options[kw.RESC_HIER_STR_KW] = requested_hierarchy
+        message_body = make_FileOpenRequest()
+
+        # Perform DATA_OBJ_OPEN call
+        message = iRODSMessage('RODS_API_REQ', msg=message_body,
+                               int_info=api_number['DATA_OBJ_OPEN_AN'])
         conn.send(message)
         desc = conn.recv().int_info
 
         raw = iRODSDataObjectFileRaw(conn, desc, finalize_on_close = finalize_on_close, **options)
+        raw.session = directed_sess
+
         (_raw_fd_holder).append(raw)
         return io.BufferedRandom(raw)
 

irods/parallel.py

@@ -74,28 +74,6 @@ def wait(self):
             self.barrier.release()
             return count - 1
 
-@contextlib.contextmanager
-def enableLogging(handlerType,args,level_ = logging.INFO):
-    """Context manager for temporarily enabling a logger. For debug or test.
-
-    Usage Example -
-    with irods.parallel.enableLogging(logging.FileHandler,('/tmp/logfile.txt',)):
-        # parallel put/get code here
-    """
-    h = None
-    saveLevel = logger.level
-    try:
-        logger.setLevel(level_)
-        h = handlerType(*args)
-        h.setLevel( level_ )
-        logger.addHandler(h)
-        yield
-    finally:
-        logger.setLevel(saveLevel)
-        if h in logger.handlers:
-            logger.removeHandler(h)
-
-
 RECOMMENDED_NUM_THREADS_PER_TRANSFER = 3
 
 verboseConnection = False
@@ -380,12 +358,13 @@ def bytes_range_for_thread( i, num_threads, total_bytes,  chunk ):
     for byte_range in ranges:
         if Io is None:
             Io = session.data_objects.open( Data_object.path, Operation.data_object_mode(initial_open = False),
-                                            create = False, finalize_on_close = False,
+                                            create = False, finalize_on_close = False, allow_redirect = False,
                                             **{ kw.NUM_THREADS_KW: str(num_threads),
                                                 kw.DATA_SIZE_KW: str(total_size),
                                                 kw.RESC_HIER_STR_KW: hier_str,
                                                 kw.REPLICA_TOKEN_KW: replica_token })
         mgr.add_io( Io )
+        logger.debug('target_host = %s', Io.raw.session.pool.account.host)
         if File is None: File = gen_file_handle()
         futures.append(executor.submit( _io_part, Io, byte_range, File, Operation, mgr, str(counter), queueObject))
         counter += 1
@@ -446,21 +425,29 @@ def io_main( session, Data, opr_, fname, R='', **kwopt):
         open_options[kw.NUM_THREADS_KW] = str(num_threads)
         open_options[kw.DATA_SIZE_KW] = str(total_bytes)
 
+    output_values = {}
     if (not Io):
         (Io, rawfile) = session.data_objects.open_with_FileRaw( (d_path or Data.path),
                                                                 Operation.data_object_mode(initial_open = True),
-                                                                finalize_on_close = True, **open_options )
+                                                                finalize_on_close = True, returned_values = output_values, **open_options )
     else:
         if type(Io) is deferred_call:
             Io[kw.NUM_THREADS_KW] = str(num_threads)
-            Io[kw.DATA_SIZE_KW] =  str(total_bytes)
+            Io[kw.DATA_SIZE_KW] = str(total_bytes)
+            Io['returned_values'] = output_values
             Io = Io()
         rawfile = Io.raw
 
+    if not output_values:
+        output_values = kwopt.get('data_open_returned_values',{})
+
+    if 'session' in output_values:
+        session = output_values['session']
+
     # At this point, the data object's existence in the catalog is guaranteed,
     # whether the Operation is a GET or PUT.
 
-    if not isinstance(Data,iRODSDataObject):
+    if not isinstance(Data,iRODSDataObject) or 'session' in output_values:
         Data = session.data_objects.get(d_path)
 
     # Determine total number of bytes for transfer.

irods/session.py

@@ -168,10 +168,17 @@ def __del__(self):
     def clone(self, **kwargs):
         other = copy.copy(self)
         other.pool = None
-        for k,v in vars(other).items():
-            setter = getattr(v,'_set_manager_session',None)
-            if setter:
-                setter(other)
+        for k,v in vars(self).items():
+            if getattr(v,'_set_manager_session',None) is not None:
+                vcopy = copy.copy(v)
+                # Deep-copy into the manager object for the cloned session and set its parent session
+                # reference to correspond to the clone.
+                setattr(other,k,vcopy)
+                vcopy._set_manager_session(other)
+            elif isinstance(v,iRODSAccount):
+                # Deep-copy the iRODSAccount subobject, since we might be setting the hostname on that object.
+                setattr(other,k,copy.copy(v))
+
         other.cleanup(new_host = kwargs.pop('host',''))
         other.ticket__ = kwargs.pop('ticket',self.ticket__)
         self.ticket_applied = weakref.WeakKeyDictionary() # conn -> ticket applied
@@ -255,7 +262,7 @@ def configure(self, **kwargs):
             account = self._configure_account(**kwargs)
         connection_refresh_time = self.get_connection_refresh_time(**kwargs)
         logger.debug("In iRODSSession's configure(). connection_refresh_time set to {}".format(connection_refresh_time))
-        self.pool = Pool(account, application_name=kwargs.pop('application_name',''), connection_refresh_time=connection_refresh_time)
+        self.pool = Pool(account, application_name=kwargs.pop('application_name',''), connection_refresh_time=connection_refresh_time, session = self)
         conn_timeout = getattr(self,'_cached_connection_timeout',None)
         self.pool.connection_timeout = conn_timeout
         return account

irods/test/connection_test.py

@@ -83,6 +83,10 @@ def test_that_connection_timeout_works__issue_377(self):
             # Set a very short socket timeout and remove all pre-existing socket connections.
             # This forces a new connection to be made for any ensuing connections to the iRODS server.
 
+            sess = obj.manager.sess # Because of client-redirect it is possible that self.sess and
+                                    # obj.manager.sess do not refer to the same object. In any case,
+                                    # it is the latter of the two iRODSSession objects that is
+                                    # involved in the data PUT connection.
             sess.connection_timeout = timeout = 0.01
             sess.cleanup()