[#626][#627] client redirection is no longer active by default.

When first implemented, the client redirect feature was on by default, meaning
that an open() call on a data object always favored a connection to the server hosting
the replica indicated by the resource hierarchy resolution outcome, and then subsequent
data movements would involve the new server host.  This proved problematic in some
scenarios, notably whenever the client was unable to connect using the server hostname
attached to the storage resource in question - DNS often being a factor.

As of now, therefore, the client redirects will no longer happen without explicit
"approval" by way of assigning the value of True to either the open() method's
allow_redirect parameter or (more globally) to a new configuration setting,
named irods.client_configuration.data_objects.allow_redirect

Author: d-w-moore

README.md

@@ -414,6 +414,18 @@ The current value of the setting is global in scope (i.e. applies to all
 sessions, whenever created) and is always consulted for the creation of
 any data object handle to govern that handle's cleanup behavior.
 
+Also, alternatively, the client may opt into a special "redirect" behavior
+in which data objects' open() method forge a connection directly to whichever
+iRODS server is found to host the selected replica.  Data reads and
+writes will therefore happen on that alternate network route, instead of
+through the originally-connected server.  Though not the client's default
+behavior, this approach can turn out to be more efficient, particularly
+if several concurrent data uploads ("puts") and downloads ("gets") are 
+happening which might increase traffic on the client's main communication
+route with the server.  (See, in [Python iRODS Client Settings File](#python-irods-client-settings-file),
+the client configuration setting `data_objects.allow_redirect`, which may be
+set to True for the aforementioned opt-in.)
+
 Python iRODS Client Settings File
 ---------------------------------
 
@@ -470,6 +482,12 @@ variables serving as overrides:
     -   Default Value: `False`
     -   Environment Variable Override: `PYTHON_IRODSCLIENT_CONFIG__DATA_OBJECTS__AUTO_CLOSE`
 
+-   Setting: Let a call to data object open() redirect to the server whose storage resource hosts the given object's preferred replica.
+    -   Dotted Name: `data_objects.allow_redirect`
+    -   Type: `bool`
+    -   Default Value: `False`
+    -   Environment Variable Override: `PYTHON_IRODSCLIENT_CONFIG__DATA_OBJECTS__ALLOW_REDIRECT`
+
 -   Setting: Number of hours to request for the new password entry's TTL (Time To Live) when auto-renewing PAM-authenticated sessions.
     - Dotted Name: `legacy_auth.pam.time_to_live_in_hours`
     - Type: `int`

irods/client_configuration/__init__.py

@@ -57,7 +57,8 @@ def xml_parser_default(self,str_value):
 # (irods.client_configuration.data_objects is one such category):
 
 class DataObjects(iRODSConfiguration):
-    __slots__ = ('auto_close',)
+    __slots__ = ('auto_close',
+                 'allow_redirect',)
 
     def __init__(self):
 
@@ -71,6 +72,7 @@ def __init__(self):
         # >>> irods.client_configuration.data_objects.auto_close = True
 
         self.auto_close = False
+        self.allow_redirect = False
 
 # #############################################################################
 #

irods/manager/data_object_manager.py

@@ -422,7 +422,7 @@ def open(self, path, mode,
                    auto_close = client_config.getter('data_objects','auto_close'), # The default value will be a lambda returning the
                                                                                    # global setting. Use True or False as an override.
                    returned_values = None,    # Used to update session reference, for forging more conns to same host, in irods.parallel.io_main
-                   allow_redirect = True,     # This may be set to False to disallow the client redirect-to-resource.
+                   allow_redirect = client_config.getter('data_objects','allow_redirect'), 
                    **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,
@@ -475,6 +475,9 @@ def make_FileOpenRequest(**extra_opts):
 
         use_get_rescinfo_apis = False
 
+        if callable(allow_redirect):
+            allow_redirect = allow_redirect()
+
         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',

irods/test/data_obj_test.py

@@ -834,59 +834,62 @@ def do_test_redirect_in_data_object_put_and_get__issue_452(self, content, data_c
             self.skipTest('Expects iRODS server version 4.3.1')
         LOCAL_FILE = mktemp()
         filename = ''
-        try:
-            with self.create_simple_resc(hostname = 'localhost') as rescName:
-                with NamedTemporaryFile(delete = False) as f:
-                    filename = f.name
-                    f.write(content)
-                data_ctx['initialize']()
-                sess = data_ctx['session']
-                remote_name = data_ctx['path']
-                PUT_LOG = self.In_Memory_Stream(always_unicode = True)
-                with helpers.enableLogging(logging.getLogger('irods.manager.data_object_manager'),
-                                           logging.StreamHandler, (PUT_LOG,), level_ = logging.DEBUG),\
-                     helpers.enableLogging(logging.getLogger('irods.parallel'),
-                                           logging.StreamHandler, (PUT_LOG,), level_ = logging.DEBUG):
-                    sess.data_objects.put(filename, remote_name, **{kw.DEST_RESC_NAME_KW: rescName})
-                def srch(BUF):
-                    nthr = 0
-                    search_text = BUF.getvalue()
-                    find_iterator = itertools.chain( re.finditer(u'redirect_to_host = (\S+)', search_text),
-                                                     re.finditer(u'target_host = (\S+)', search_text) )
-                    for match in find_iterator:
-                        nthr += 1
-                        self.assertEqual(match.group(1), u'localhost')
-                    occur_threshold = (1 if len(content) <= 32*MEBI else 2)
-                    self.assertGreaterEqual(nthr, occur_threshold)
-                srch(PUT_LOG)
-                generator = None
-                # Activate a read ticket on a new session if necessary, and attempt a GET
-                if data_ctx['ticket_access']:
-                    for access in iRODSAccess('own',remote_name,self.sess.username), \
-                                  iRODSAccess('null',remote_name,sess.username):
-                        self.sess.acls.set(access, admin = True)
-                    generator = self._data_object_and_associated_ticket(data_name=remote_name, auto_delete_data = False, ticket_access='read')
-                    # Emulate the 'with'-block construction for the read ticket:
-                    data_ctx_get = next(generator)
-                    data_ctx_get['initialize']()
-                    sess = data_ctx_get['session']
-                GET_LOG = self.In_Memory_Stream(always_unicode = True)
-                with helpers.enableLogging(logging.getLogger('irods.manager.data_object_manager'),
-                                           logging.StreamHandler, (GET_LOG,), level_ = logging.DEBUG),\
-                     helpers.enableLogging(logging.getLogger('irods.parallel'),
-                                           logging.StreamHandler, (GET_LOG,), level_ = logging.DEBUG):
-                    sess.data_objects.get(remote_name,LOCAL_FILE)
-                srch(GET_LOG)
-                with open(LOCAL_FILE,'rb') as get_result:
-                    self.assertTrue(content, get_result.read())
-                # Finalize the emulated 'with'-block construction for the read ticket, if active:
-                del generator
-                data_ctx['finalize']()
-        finally:
-            if os.path.isfile(LOCAL_FILE):
-                os.unlink(LOCAL_FILE)
-            if filename:
-                os.unlink(filename)
+        with config.loadlines(entries=[dict(setting='data_objects.allow_redirect',value=True)]):
+            try:
+                with self.create_simple_resc(hostname = 'localhost') as rescName:
+                    with NamedTemporaryFile(delete = False) as f:
+                        filename = f.name
+                        f.write(content)
+                    data_ctx['initialize']()
+                    sess = data_ctx['session']
+                    remote_name = data_ctx['path']
+                    PUT_LOG = self.In_Memory_Stream(always_unicode = True)
+                    with helpers.enableLogging(logging.getLogger('irods.manager.data_object_manager'),
+                                               logging.StreamHandler, (PUT_LOG,), level_ = logging.DEBUG),\
+                         helpers.enableLogging(logging.getLogger('irods.parallel'),
+                                               logging.StreamHandler, (PUT_LOG,), level_ = logging.DEBUG):
+                        sess.data_objects.put(filename, remote_name, **{kw.DEST_RESC_NAME_KW: rescName})
+                    # Within a buffer 'BUF' (is expected to be an io.StringIO object) assert the presence of certain
+                    # log text that will indicate a redirection was performed.
+                    def assert_expected_redirection_logging(BUF):
+                        nthr = 0
+                        search_text = BUF.getvalue()
+                        find_iterator = itertools.chain( re.finditer(u'redirect_to_host = (\S+)', search_text),
+                                                         re.finditer(u'target_host = (\S+)', search_text) )
+                        for match in find_iterator:
+                            nthr += 1
+                            self.assertEqual(match.group(1), u'localhost')
+                        occur_threshold = (1 if len(content) <= 32*MEBI else 2)
+                        self.assertGreaterEqual(nthr, occur_threshold)
+                    assert_expected_redirection_logging(PUT_LOG)
+                    generator = None
+                    # Activate a read ticket on a new session if necessary, and attempt a GET
+                    if data_ctx['ticket_access']:
+                        for access in iRODSAccess('own',remote_name,self.sess.username), \
+                                      iRODSAccess('null',remote_name,sess.username):
+                            self.sess.acls.set(access, admin = True)
+                        generator = self._data_object_and_associated_ticket(data_name=remote_name, auto_delete_data = False, ticket_access='read')
+                        # Emulate the 'with'-block construction for the read ticket:
+                        data_ctx_get = next(generator)
+                        data_ctx_get['initialize']()
+                        sess = data_ctx_get['session']
+                    GET_LOG = self.In_Memory_Stream(always_unicode = True)
+                    with helpers.enableLogging(logging.getLogger('irods.manager.data_object_manager'),
+                                               logging.StreamHandler, (GET_LOG,), level_ = logging.DEBUG),\
+                         helpers.enableLogging(logging.getLogger('irods.parallel'),
+                                               logging.StreamHandler, (GET_LOG,), level_ = logging.DEBUG):
+                        sess.data_objects.get(remote_name,LOCAL_FILE)
+                    assert_expected_redirection_logging(GET_LOG)
+                    with open(LOCAL_FILE,'rb') as get_result:
+                        self.assertTrue(content, get_result.read())
+                    # Finalize the emulated 'with'-block construction for the read ticket, if active:
+                    del generator
+                    data_ctx['finalize']()
+            finally:
+                if os.path.isfile(LOCAL_FILE):
+                    os.unlink(LOCAL_FILE)
+                if filename:
+                    os.unlink(filename)
 
     def test_redirect_in_data_object_open__issue_452(self):
         self._skip_unless_connected_to_local_computer_by_other_than_localhost_synonym()
@@ -895,18 +898,19 @@ def test_redirect_in_data_object_open__issue_452(self):
         sess = self.sess
         home = helpers.home_collection(sess)
 
-        with self.create_simple_resc(hostname = 'localhost') as rescName:
-            try:
-                test_path = home + '/data_open_452'
-                desc = sess.data_objects.open(test_path, 'w', **{kw.RESC_NAME_KW: rescName})
-                self.assertEqual('localhost', desc.raw.session.host)
-                desc.close()
-                desc = sess.data_objects.open(test_path, 'r')
-                self.assertEqual('localhost', desc.raw.session.host)
-                desc.close()
-            finally:
-                if sess.data_objects.exists(test_path):
-                    sess.data_objects.unlink(test_path, force=True)
+        with config.loadlines(entries=[dict(setting='data_objects.allow_redirect',value=True)]):
+            with self.create_simple_resc(hostname = 'localhost') as rescName:
+                try:
+                    test_path = home + '/data_open_452'
+                    desc = sess.data_objects.open(test_path, 'w', **{kw.RESC_NAME_KW: rescName})
+                    self.assertEqual('localhost', desc.raw.session.host)
+                    desc.close()
+                    desc = sess.data_objects.open(test_path, 'r')
+                    self.assertEqual('localhost', desc.raw.session.host)
+                    desc.close()
+                finally:
+                    if sess.data_objects.exists(test_path):
+                        sess.data_objects.unlink(test_path, force=True)
 
 
     def test_create_with_checksum(self):
@@ -2174,35 +2178,53 @@ def test_touch_operation_does_not_work_when_given_a_collection__525(self):
         with self.assertRaises(ex.InvalidInputArgument):
             user_session.data_objects.touch(home_collection_path)
 
+    def assert_redirect_happens_on_open(self, open_opts):
+        name = 'redirect_happens_' + unique_name (my_function_name(), datetime.now())
+        data_path = '{self.coll_path}/{name}'.format(**locals())
+        try:
+            PUT_LOG = self.In_Memory_Stream(always_unicode = True)
+            with helpers.enableLogging(logging.getLogger('irods.manager.data_object_manager'),
+                                       logging.StreamHandler, (PUT_LOG,), level_ = logging.DEBUG):
+                with self.sess.data_objects.open(data_path,'w',**open_opts):
+                    pass
+                log_text = PUT_LOG.getvalue()
+                self.assertIn('redirect_to_host',log_text)
+        finally:
+            if self.sess.data_objects.exists(data_path):
+                self.sess.data_objects.unlink(data_path, force = True)
+
     @unittest.skipIf(six.PY2, "Python2 won't destruct an out-of-scope iRODSSession due to lazy GC ref-cycle detection.")
     def test_client_redirect_lets_go_of_connections__issue_562(self):
         self._skip_unless_connected_to_local_computer_by_other_than_localhost_synonym()
         # Force data object connections to redirect by enforcing a non-equivalent hostname for their resource
         total_conns = lambda session: len(session.pool.idle | session.pool.active)
-        with self.create_simple_resc(hostname = 'localhost') as resc_name:
-            # A reasonable number of data objects to create without eliciting problems.
-            # (But before resolution of #562, a NetworkException was eventually thrown from
-            # this test loop if a session cleanup() did not intervene between open() calls.)
-            REPS_TO_REPRODUCE_CONNECT_ERROR = 100
-            paths=[]
-            prev_conns = None
-            try:
-                # Try to exhaust connections
-                for n in range(REPS_TO_REPRODUCE_CONNECT_ERROR):
-                    data_path = '{self.coll_path}/issue_562_test_obj_{n:03d}.dat'.format(**locals())
-                    paths.append(data_path)
-                    with self.sess.data_objects.open(data_path, 'w', **{kw.DEST_RESC_NAME_KW: resc_name}) as f:
-                        pass
-                    # Assert number of connections does not increase
-                    current_conns = total_conns(self.sess)
-                    if isinstance(prev_conns,int):
-                        self.assertLessEqual(current_conns, prev_conns)
-                    prev_conns = current_conns
-            finally:
-                # Clean up data objects before resource is deleted.
-                for data_path in paths:
-                    if self.sess.data_objects.exists(data_path):
-                        self.sess.data_objects.unlink(data_path, force = True)
+        with config.loadlines(entries=[dict(setting='data_objects.allow_redirect',value=True)]):
+            with self.create_simple_resc(hostname = 'localhost') as resc_name:
+                self.assert_redirect_happens_on_open( {kw.DEST_RESC_NAME_KW: resc_name} )
+                # A reasonable number of data objects to create without eliciting problems.
+                # (But before resolution of #562, a NetworkException was eventually thrown from
+                # this test loop if a session cleanup() did not intervene between open() calls.)
+                REPS_TO_REPRODUCE_CONNECT_ERROR = 100
+                paths=[]
+                prev_conns = None
+                try:
+                    # Try to exhaust connections
+                    for n in range(REPS_TO_REPRODUCE_CONNECT_ERROR):
+                        data_path = '{self.coll_path}/issue_562_test_obj_{n:03d}.dat'.format(**locals())
+                        paths.append(data_path)
+                        with self.sess.data_objects.open(data_path, 'w', **{kw.DEST_RESC_NAME_KW: resc_name}) as f:
+                            pass
+                        # Assert number of connections does not increase
+                        current_conns = total_conns(self.sess)
+                        if isinstance(prev_conns,int):
+                            self.assertLessEqual(current_conns, prev_conns)
+                        prev_conns = current_conns
+                finally:
+                    # Clean up data objects before resource is deleted.
+                    for data_path in paths:
+                        if self.sess.data_objects.exists(data_path):
+                            self.sess.data_objects.unlink(data_path, force = True)
+
 
     @unittest.skipIf(progressbar is None, "progressbar is not installed")
     def test_progressbar_style_of_pbar_without_registering__issue_574(self):
@@ -2397,6 +2419,34 @@ def test_replica_truncate_related_errors__issue_534(self):
             if data_objs.exists(data_path):
                 data_objs.unlink(data_path, force = True)
 
+    def test_allow_redirect_configuration_setting__issue_627(self):
+
+        self._skip_unless_connected_to_local_computer_by_other_than_localhost_synonym()
+
+        logical_paths = ['{}/issue_627_{}_{}'.format(self.coll_path, n, unique_name(my_function_name(), datetime.now())) for n in range(2)]
+
+        with self.create_simple_resc(hostname = 'localhost') as newResc1,\
+             self.create_simple_resc() as newResc2:
+
+            if self.sess.resources.get(newResc1).location == self.sess.resources.get(newResc2).location:
+                self.skipTest('test runs only if host locations differ between experimental and control resource')
+
+            for use_redirect in (True,False):
+
+                with config.loadlines(entries = [dict(setting = 'data_objects.allow_redirect',value = use_redirect)]):
+
+                    try:
+                        with self.sess.data_objects.open(logical_paths[0], 'w', **{kw.RESC_NAME_KW:newResc1}) as d1,\
+                             self.sess.data_objects.open(logical_paths[1], 'w', **{kw.RESC_NAME_KW:newResc2}) as d2:
+
+                            hostname_inequality_relation = (d1.raw.session.host != d2.raw.session.host)
+                            self.assertEqual(use_redirect, hostname_inequality_relation)
+                    finally:
+                        for path in logical_paths:
+                            if self.sess.data_objects.exists(path):
+                                self.sess.data_objects.unlink(path, force = True)
+
+
     def test_replica_truncate__issue_534(self):
         sess = self.sess
         data_objs = self.sess.data_objects