Merge branch '3.8' into develop

packaging/rpm/fts-rest.spec

@@ -4,19 +4,19 @@
 %{!?nosetest_path: %global nosetest_path "/tmp"}
 
 Name:           fts-rest
-Version:        3.7.1
+Version:        3.8.0
 Release:        1%{?dist}
 BuildArch:      noarch
 Summary:        FTS3 Rest Interface
 Group:          Applications/Internet
 License:        ASL 2.0
 URL:            http://fts3-service.web.cern.ch/
-# git clone https://gitlab.cern.ch/fts/fts-rest.git --depth 1 -b master fts-rest-3.7.1
-# cd fts-rest-3.7.1
-# git checkout v3.7.1
+# git clone https://gitlab.cern.ch/fts/fts-rest.git --depth 1 -b master fts-rest-3.8.0
+# cd fts-rest-3.8.0
+# git checkout v3.8.0
 # git submodule init && git submodule update
 # cd ..
-# tar vczf fts-rest-3.7.1.tar.gz --exclude-vcs fts-rest-3.7.1
+# tar vczf fts-rest-3.8.0.tar.gz --exclude-vcs fts-rest-3.8.0
 Source0:        %{name}-%{version}.tar.gz
 
 BuildRequires:  gfal2-python

setup.py

@@ -66,7 +66,7 @@ base_dir = os.path.dirname(__file__)
 
 setup(
     name='fts3-rest',
-    version='3.7.1',
+    version='3.8.0',
     description='FTS3 Python Libraries',
     long_description='FTS3 Python Libraries',
     author='FTS3 Developers',

src/fts3rest/fts3rest/controllers/api.py

@@ -28,7 +28,7 @@ from fts3rest.lib.base import BaseController, Session
 from fts3rest.lib.helpers import jsonify
 from fts3rest.lib import api
 
-API_VERSION = dict(major=3, minor=7, patch=1)
+API_VERSION = dict(major=3, minor=8, patch=0)
 
 
 def _get_fts_core_version():

src/fts3rest/fts3rest/controllers/jobs.py

@@ -567,9 +567,9 @@ class JobsController(BaseController):
         # Send messages
         # Need to re-query so we get the file ids
         job = Session.query(Job).get(populated.job_id)
-        for transfer in job.files:
+        for i in range(len(job.files)):
             try:
-                submit_state_change(job, transfer)
+                submit_state_change(job, job.files[i], populated.files[0]['file_state'])
             except Exception, e:
                 log.warning("Failed to write state message to disk: %s" % e.message)
 

src/fts3rest/fts3rest/lib/JobBuilder.py

@@ -355,8 +355,8 @@ class JobBuilder(object):
         else:
             initial_file_state = 'SUBMITTED'
 
-        # Multiple replica job?
-        if len(file_dict['sources']) > 1:
+        # Multiple replica job or multihop? Then, the initial state is NOT_USED
+        if len(file_dict['sources']) > 1 or self.params['multihop']:
             if self.is_bringonline:
                 raise HTTPBadRequest('Staging with multiple replicas is not allowed')
             # On multiple replica job, we mark all files initially with NOT_USED
@@ -493,6 +493,9 @@ class JobBuilder(object):
             self.job['job_type'] = 'R'
             # Apply selection strategy
             self._apply_selection_strategy()
+        # For multihop, mark the first as SUBMITTED
+        elif self.params['multihop']:
+            self.files[0]['file_state'] = 'SUBMITTED'
 
         self._set_job_source_and_destination(self.files)
         

src/fts3rest/fts3rest/lib/helpers/msgbus.py

@@ -27,7 +27,7 @@ except:
 
 log = logging.getLogger(__name__)
 
-def submit_state_change(job, transfer):
+def submit_state_change(job, transfer, transfer_state):
     """
     Writes a state change message to the dirq
     """
@@ -49,7 +49,7 @@ def submit_state_change(job, transfer):
         job_id=job['job_id'],
         file_id=transfer['file_id'],
         job_state=job['job_state'],
-        file_state=transfer['file_state'],
+        file_state=transfer_state,
         retry_counter=0,
         retry_max=0,
         timestamp=time.time()*1000,
@@ -64,4 +64,4 @@ def submit_state_change(job, transfer):
 
     q = QueueSimple(path=mon_dir)
     q.add_path(tmpfile.name)
-    log.debug("Sent SUBMITTED state for %s %d" % (job['job_id'], transfer['file_id']))
\ No newline at end of file
+    log.debug("Sent SUBMITTED state for %s %d" % (job['job_id'], transfer['file_id']))