Repository: rucio/rucio
Issue: Crash when trying to find matching scheme

From the CMS instance, I noticed this exception thrown in the `conveyor-transfer-submitter` after creating some minimal RSEs and trying to trigger a transfer
```
2018-01-11 20:59:36,983	4217	CRITICAL	Exception happened when trying to get transfer for request 8a818df9ca4047e3a6b36f95b98a5c7e: Traceback (most recent call last):
  File "/usr/lib/python2.7/site-packages/rucio/core/transfer.py", line 567, in get_transfer_requests_and_source_replicas
    scheme=current_schemes)
  File "/usr/lib/python2.7/site-packages/rucio/rse/rsemanager.py", line 640, in find_matching_scheme
    dest_candidates = sorted(dest_candidates, key=lambda k: k['domains'][domain][operation_dest])
  File "/usr/lib/python2.7/site-packages/rucio/rse/rsemanager.py", line 640, in <lambda>
    dest_candidates = sorted(dest_candidates, key=lambda k: k['domains'][domain][operation_dest])
KeyError: None
```


Changed files:
lib/rucio/api/replica.py, lib/rucio/core/did.py, lib/rucio/core/replica.py, lib/rucio/tests/test_replica.py, lib/rucio/core/transfer.py

diff --git a/lib/rucio/api/replica.py b/lib/rucio/api/replica.py
index 29ff9c692..cac687083 100644
--- a/lib/rucio/api/replica.py
+++ b/lib/rucio/api/replica.py
@@ -6,7 +6,7 @@
 #
 # Authors:
 # - Vincent Garonne, <vincent.garonne@cern.ch>, 2012-2016
-# - Mario Lassnig, <mario.lassnig@cern.ch>, 2012, 2017
+# - Mario Lassnig, <mario.lassnig@cern.ch>, 2012, 2017-2018
 # - Cedric Serfon, <cedric.serfon@cern.ch>, 2014-2016
 # - Thomas Beermann, <thomas.beermann@cern.ch>, 2014
 
@@ -81,7 +81,7 @@ def get_did_from_pfns(pfns, rse):
 
 def list_replicas(dids, schemes=None, unavailable=False, request_id=None,
                   ignore_availability=True, all_states=False, rse_expression=None,
-                  client_location=None):
+                  client_location=None, domain=None):
     """
     List file replicas for a list of data identifiers.
 
@@ -92,13 +92,14 @@ def list_replicas(dids, schemes=None, unavailable=False, request_id=None,
     :param all_states: Return all replicas whatever state they are in. Adds an extra 'states' entry in the result dictionary.
     :param rse_expression: The RSE expression to restrict replicas on a set of RSEs.
     :param client_location: Client location dictionary for PFN modification {'ip', 'fqdn', 'site'}
+    :param domain: The network domain for the call, either None, 'wan' or 'lan'. Compatibility fallback: None falls back to 'wan'.
     """
     validate_schema(name='r_dids', obj=dids)
     return replica.list_replicas(dids=dids, schemes=schemes, unavailable=unavailable,
                                  request_id=request_id,
                                  ignore_availability=ignore_availability,
                                  all_states=all_states, rse_expression=rse_expression,
-                                 client_location=client_location)
+                                 client_location=client_location, domain=domain)
 
 
 def add_replicas(rse, files, issuer, ignore_availability=False):
diff --git a/lib/rucio/core/did.py b/lib/rucio/core/did.py
index 53bee55a4..863ac7d52 100644
--- a/lib/rucio/core/did.py
+++ b/lib/rucio/core/did.py
@@ -7,7 +7,7 @@
   http://www.apache.org/licenses/LICENSE-2.0
 
   Authors:
-  - Vincent Garonne, <vincent.garonne@cern.ch>, 2012-2017
+  - Vincent Garonne, <vincent.garonne@cern.ch>, 2012-2018
   - Mario Lassnig, <mario.lassnig@cern.ch>, 2012-2015, 2017
   - Yun-Pin Sun, <yun-pin.sun@cern.ch>, 2013
   - Cedric Serfon, <cedric.serfon@cern.ch>, 2013-2015, 2017
@@ -788,11 +788,14 @@ def set_new_dids(dids, new_flag, session=None):
     :param new_flag: A boolean to flag new DIDs.
     :param session: The database session in use.
     """
+    if session.bind.dialect.name == 'postgresql':
+        new_flag = bool(new_flag)
     for did in dids:
         try:
+
             rowcount = session.query(models.DataIdentifier).\\
                 filter_by(scope=did['scope'], name=did['name']).\\
-                update({'is_new': bool(new_flag)}, synchronize_session=False)
+                update({'is_new': new_flag}, synchronize_session=False)
             if not rowcount:
                 raise exception.DataIdentifierNotFound("Data identifier '%s:%s' not found" % (did['scope'], did['name']))
         except DatabaseError as error:
diff --git a/lib/rucio/core/replica.py b/lib/rucio/core/replica.py
index da941ae22..a480fd67d 100644
--- a/lib/rucio/core/replica.py
+++ b/lib/rucio/core/replica.py
@@ -11,7 +11,7 @@
   - Cedric Serfon <cedric.serfon@cern.ch>, 2013-2017
   - Ralph Vigne <ralph.vigne@cern.ch>, 2013-2014
   - Martin Barisits <martin.barisits@cern.ch>, 2013-2016
-  - Mario Lassnig <mario.lassnig@cern.ch>, 2014-2017
+  - Mario Lassnig <mario.lassnig@cern.ch>, 2014-2018
   - David Cameron <d.g.cameron@gmail.com>, 2014
   - Thomas Beermann <thomas.beermann@cern.ch>, 2014-2016
   - Wen Guan, <wen.guan@cern.ch>, 2014-2015
@@ -719,7 +719,7 @@ def _list_replicas_for_files(file_clause, state_clause, files, rse_clause, sessi
         #    raise exception.DataIdentifierNotFound("Files not found %s", str(files))
 
 
-def _list_replicas(dataset_clause, file_clause, state_clause, show_pfns, schemes, files, rse_clause, client_location, session):
+def _list_replicas(dataset_clause, file_clause, state_clause, show_pfns, schemes, files, rse_clause, client_location, domain, session):
 
     files = [dataset_clause and _list_replicas_for_datasets(dataset_clause, state_clause, rse_clause, session),
              file_clause and _list_replicas_for_files(file_clause, state_clause, files, rse_clause, session)]
@@ -740,7 +740,8 @@ def _list_replicas(dataset_clause, file_clause, state_clause, show_pfns, schemes
                     if not rse_schemes:
                         try:
                             rse_schemes = [rsemgr.select_protocol(rse_settings=rse_info[rse],
-                                                                  operation='read')['scheme']]
+                                                                  operation='read',
+                                                                  domain=domain)['scheme']]
                         except:
                             print format_exc()
 
@@ -749,7 +750,8 @@ def _list_replicas(dataset_clause, file_clause, state_clause, show_pfns, schemes
                         try:
                             protocols.append(rsemgr.create_protocol(rse_settings=rse_info[rse],
                                                                     operation='read',
-                                                                    scheme=s))
+                                                                    scheme=s,
+                                                                    domain=domain))
                         except exception.RSEProtocolNotSupported:
                             pass  # no need to be verbose
                         except:
@@ -828,7 +830,8 @@ def _list_replicas(dataset_clause, file_clause, state_clause, show_pfns, schemes
 @stream_session
 def list_replicas(dids, schemes=None, unavailable=False, request_id=None,
                   ignore_availability=True, all_states=False, pfns=True,
-                  rse_expression=None, client_location=None, session=None):
+                  rse_expression=None, client_location=None, domain=None,
+                  session=None):
     """
     List file replicas for a list of data identifiers (DIDs).
 
@@ -840,8 +843,16 @@ def list_replicas(dids, schemes=None, unavailable=False, request_id=None,
     :param all_states: Return all replicas whatever state they are in. Adds an extra 'states' entry in the result dictionary.
     :param rse_expression: The RSE expression to restrict list_replicas on a set of RSEs.
     :param client_location: Client location dictionary for PFN modification {'ip', 'fqdn', 'site'}
+    :param domain: The network domain for the call, either None, 'wan' or 'lan'. Compatibility fallback: None falls back to 'wan'.
     :param session: The database session in use.
     """
+
+    # Compatibility fallback:
+    # Old clients expect WAN replicas always, domain=None could potentially retrieve LAN replicas for WAN sources.
+    # TODO: Automated WAN/LAN selection in _list_replicas
+    if not domain:
+        domain = 'wan'
+
     file_clause, dataset_clause, state_clause, files = _resolve_dids(dids=dids, unavailable=unavailable,
                                                                      ignore_availability=ignore_availability,
                                                                      all_states=all_states, session=session)
@@ -851,7 +862,7 @@ def list_replicas(dids, schemes=None, unavailable=False, request_id=None,
         for rse in parse_expression(expression=rse_expression, session=session):
             rse_clause.append(models.RSEFileAssociation.rse_id == rse['id'])
 
-    for file in _list_replicas(dataset_clause, file_clause, state_clause, pfns, schemes, files, rse_clause, client_location, session):
+    for file in _list_replicas(dataset_clause, file_clause, state_clause, pfns, schemes, files, rse_clause, client_location, domain, session):
         yield file
 
 
diff --git a/lib/rucio/core/transfer.py b/lib/rucio/core/transfer.py
index c3ae3e0f9..35b6cd1e6 100644
--- a/lib/rucio/core/transfer.py
+++ b/lib/rucio/core/transfer.py
@@ -7,7 +7,7 @@
 #
 # Authors:
 # - Martin Barisits, <martin.barisits@cern.ch>, 2017
-# - Mario Lassnig, <mario.lassnig@cern.ch>, 2017
+# - Mario Lassnig, <mario.lassnig@cern.ch>, 2017-2018
 
 import datetime
 import json
@@ -558,12 +558,13 @@ def get_transfer_requests_and_source_replicas(process=None, total_processes=None
                 allow_tape_source = True
 
                 # Find matching scheme between destination and source
+                # TODO: In the future, move to domain "third_party_copy"
                 try:
                     matching_scheme = rsemgr.find_matching_scheme(rse_settings_dest=rses_info[dest_rse_id],
                                                                   rse_settings_src=rses_info[source_rse_id],
                                                                   operation_src='read',
                                                                   operation_dest='write',
-                                                                  domain=None,
+                                                                  domain='wan',
                                                                   scheme=current_schemes)
                 except RSEProtocolNotSupported:
                     logging.error('Operation "write" not supported by %s with schemes %s' % (rses_info[dest_rse_id]['rse'], current_schemes))
diff --git a/lib/rucio/tests/test_replica.py b/lib/rucio/tests/test_replica.py
index 5f0a1b3c7..3226d76d6 100644
--- a/lib/rucio/tests/test_replica.py
+++ b/lib/rucio/tests/test_replica.py
@@ -7,7 +7,7 @@
 #
 # Authors:
 # - Vincent Garonne, <vincent.garonne@cern.ch>, 2013-2015
-# - Mario Lassnig, <mario.lassnig@cern.ch>, 2013-2014, 2016-2017
+# - Mario Lassnig, <mario.lassnig@cern.ch>, 2013-2014, 2016-2018
 # - Cedric Serfon, <cedric.serfon@cern.ch>, 2014-2016
 # - Thomas Beermann, <thomas.beermann@cern.ch>, 2014
 
@@ -32,8 +32,10 @@ from rucio.core.replica import (add_replica, add_replicas, delete_replicas,
                                 declare_bad_file_replicas, list_bad_replicas,
                                 update_replicas_paths, update_replica_state,
                                 get_replica_atime, touch_replica)
+from rucio.core.rse import add_rse, add_protocol
 from rucio.daemons.necromancer import run
 from rucio.rse import rsemanager as rsemgr
+from rucio.tests.common import rse_name_generator
 from rucio.web.rest.authentication import APP as auth_app
 from rucio.web.rest.replica import APP as rep_app
 
@@ -246,6 +248,58 @@ class TestReplicaCore:
 
         assert_equal(nbfiles, replica_cpt)
 
+    def test_list_replica_with_domain(self):
+        """ REPLICA (CORE): Add and list file replicas forcing domain"""
+
+        tmp_rse = rse_name_generator()
+        add_rse(tmp_rse)
+
+        protocols = [{'scheme': 'MOCK',
+                      'hostname': 'localhost',
+                      'port': 17,
+                      'prefix': '/i/prefer/the/lan',
+                      'impl': 'rucio.rse.protocols.mock.Default',
+                      'domains': {
+                          'lan': {'read': 1,
+                                  'write': 1,
+                                  'delete': 1},
+                          'wan': {'read': 2,
+                                  'write': 2,
+                                  'delete': 2}}},
+                     {'scheme': 'MOCK',
+                      'hostname': 'localhost',
+                      'port': 18,
+                      'prefix': '/i/prefer/the/wan',
+                      'impl': 'rucio.rse.protocols.mock.Default',
+                      'domains': {
+                          'lan': {'read': 2,
+                                  'write': 2,
+                                  'delete': 2},
+                          'wan': {'read': 1,
+                                  'write': 1,
+                                  'delete': 1}}}, ]
+        for p in protocols:
+            add_protocol(tmp_rse, p)
+
+        nbfiles = 3
+        files = [{'scope': 'mock',
+                  'name': 'file_%s' % generate_uuid(),
+                  'bytes': 1234L,
+                  'adler32': '01234567',
+                  'meta': {'events': 1234}} for i in xrange(nbfiles)]
+
+        add_replicas(rse=tmp_rse, files=files, account='root', ignore_availability=True)
+
+        for replica in list_replicas(dids=[{'scope': f['scope'], 'name': f['name'], 'type': DIDType.FILE} for f in files],
+                                     schemes=['MOCK'],
+                                     domain='wan'):
+            assert_in('/i/prefer/the/wan', replica['pfns'].keys()[0])
+
+        for replica in list_replicas(dids=[{'scope': f['scope'], 'name': f['name'], 'type': DIDType.FILE} for f in files],
+                                     schemes=['MOCK'],
+                                     domain='lan'):
+            assert_in('/i/prefer/the/lan', replica['pfns'].keys()[0])
+
 
 class TestReplicaClients: