[v7r2] Add FTS3 archive monitoring feature and plugin based FTS activity

docs/source/AdministratorGuide/Resources/storage.rst

@@ -57,6 +57,7 @@ Configuration options are:
 * ``OccupancyLFN``: default (``/<vo>/occupancy.json``). LFN where the json file containing the space reporting is to be found
 * ``OccupancyPlugin``: default (``empty``). Plugin to find the occupancy of a given storage.
 * ``SpaceReservation``: just a name of a zone of the physical storage which can have some space reserved. Extends the SRM ``SpaceToken`` concept.
+* ``ArchiveTimeout``: for tape SE only. If set to a value in seconds, enables the `FTS Archive Monitoring feature <https://fts3-docs.web.cern.ch/fts3-docs/docs/archive_monitoring.html>`_
 
 VO specific paths
 -----------------

docs/source/AdministratorGuide/Systems/DataManagement/fts3.rst

@@ -122,55 +122,29 @@ The FTS server to which the job is sent is chose based on the policy. There are
 FTS3 state machines
 -------------------
 
-These are the states for FTS3File::
-
-  ALL_STATES = [ 'New',  # Nothing was attempted yet on this file
-                 'Submitted', # From FTS: Initial state of a file as soon it's dropped into the database
-                 'Ready', # From FTS: File is ready to become active
-                 'Active', # From FTS: File went active
-                 'Finished', # From FTS: File finished gracefully
-                 'Canceled', # From FTS: Canceled by the user
-                 'Staging', # From FTS: When staging of a file is requested
-                 'Failed', # From FTS: File failure
-                 'Defunct', # Totally fail, no more attempt will be made
-                 'Started', # From FTS: File transfer has started
-                 ]
-
-  FINAL_STATES = ['Canceled', 'Finished', 'Defunct']
-  FTS_FINAL_STATES = ['Canceled', 'Finished', 'Done']
-  INIT_STATE = 'New'
-
-These are the states for FTS3Operation::
-
-  ALL_STATES = ['Active',  # Default state until FTS has done everything
-                'Processed',  # Interactions with FTS done, but callback not done
-                'Finished',  # Everything was done
-                'Canceled',  # Canceled by the user
-                'Failed',  # I don't know yet
-               ]
-  FINAL_STATES = ['Finished', 'Canceled', 'Failed' ]
-  INIT_STATE = 'Active'
-
-States from the FTS3Job::
-
-  # States from FTS doc
-  ALL_STATES = ['Submitted',  # Initial state of a job as soon it's dropped into the database
-                'Ready', # One of the files within a job went to Ready state
-                'Active', # One of the files within a job went to Active state
-                'Finished', # All files Finished gracefully
-                'Canceled', # Job canceled
-                'Failed', # All files Failed
-                'Finisheddirty',  # Some files Failed
-                'Staging', # One of the files within a job went to Staging state
-               ]
-
-  FINAL_STATES = ['Canceled', 'Failed', 'Finished', 'Finisheddirty']
-  INIT_STATE = 'Submitted'
-
+The states for the ``FTS`` objects are as follow:
+
+.. literalinclude:: /../../src/DIRAC/DataManagementSystem/Client/FTS3Operation.py
+   :language: python
+   :start-after: # START states
+   :end-before: # END states
+   :caption: FTS3Operation states
+
+.. literalinclude:: /../../src/DIRAC/DataManagementSystem/Client/FTS3Job.py
+   :language: python
+   :start-after: # START states
+   :end-before: # END states
+   :caption: FTS3Job states
+
+.. literalinclude:: /../../src/DIRAC/DataManagementSystem/Client/FTS3File.py
+   :language: python
+   :start-after: # START states
+   :end-before: # END states
+   :caption: FTS3File states
 
 The status of the FTS3Jobs and FTSFiles are updated every time we monitor the matching job.
 
-The FTS3Operation goes to Processed when all the files are in a final state, and to Finished when the callback has been called successfully
+The FTS3Operation goes to ``Processed`` when all the files are in a final state, and to ``Finished`` when the callback has been called successfully
 
 
 FTS3 Plugins
@@ -180,4 +154,4 @@ FTS3 Plugins
     The ``FTS3Plugin`` option
 
 
-The ``FTS3Plugin`` option allows one to specify a plugin to alter some default choices made by the FTS3 system. These choices concern the list of third party protocols used, as well as the selection of a source storage element. This can be useful if you want to implement a matrix-like selection of protocols, or if some links require specific protocols, etc. The plugins must be placed in :py:mod:`DIRAC.DataManagementSystem.private.FTS3Plugins`. The default behaviors, as well as the documentation on how to implement your own plugin can be found in :py:mod:`DIRAC.DataManagementSystem.private.FTS3Plugins.DefaultFTS3Plugin`
+The ``FTS3Plugin`` option allows one to specify a plugin to alter some default choices made by the FTS3 system. These choices concern the list of third party protocols used, the selection of a source storage element as well as the FTS activity used. This can be useful if you want to implement a matrix-like selection of protocols, or if some links require specific protocols, etc. The plugins must be placed in :py:mod:`DIRAC.DataManagementSystem.private.FTS3Plugins`. The default behaviors, as well as the documentation on how to implement your own plugin can be found in :py:mod:`DIRAC.DataManagementSystem.private.FTS3Plugins.DefaultFTS3Plugin`

environment-py3.yml

@@ -91,6 +91,6 @@ dependencies:
     - git+https://github.com/DIRACGrid/tornado_m2crypto
     # fts-rest doesn't support Python 3, this branch changed has just enough to
     # make it importing possible
-    - git+https://gitlab.cern.ch/cburr/fts-rest.git@python3
+    - git+https://gitlab.cern.ch/chaen/fts-rest-flask.git@packaging_port_3.10
     - rucio-clients >=1.26.0rc1
     - -e .

src/DIRAC/DataManagementSystem/Agent/RequestOperations/ReplicateAndRegister.py

@@ -31,6 +31,7 @@
 # # from DIRAC
 from DIRAC import S_OK, S_ERROR, gLogger
 from DIRAC.Core.Utilities.Adler import compareAdler, hexAdlerToInt, intAdlerToHex
+from DIRAC.Core.Security.ProxyInfo import getVOfromProxyGroup
 from DIRAC.FrameworkSystem.Client.MonitoringClient import gMonitor
 
 from DIRAC.DataManagementSystem.Client.DataManager import DataManager
@@ -42,6 +43,8 @@
 from DIRAC.DataManagementSystem.Client.FTS3Operation import FTS3TransferOperation
 from DIRAC.DataManagementSystem.Client.FTS3File import FTS3File
 from DIRAC.DataManagementSystem.Client.FTS3Client import FTS3Client
+from DIRAC.DataManagementSystem.private.FTS3Utilities import getFTS3Plugin
+
 from DIRAC.ConfigurationSystem.Client.Helpers import Registry
 
 from DIRAC.MonitoringSystem.Client.MonitoringReporter import MonitoringReporter
@@ -58,7 +61,7 @@ def filterReplicas(opFile, logger=None, dataManager=None):
   log = logger.getSubLogger("filterReplicas")
   result = defaultdict(list)
 
-  replicas = dataManager.getActiveReplicas(opFile.LFN, getUrl=False)
+  replicas = dataManager.getActiveReplicas(opFile.LFN, getUrl=False, preferDisk=True)
   if not replicas["OK"]:
     log.error('Failed to get active replicas', replicas["Message"])
     return replicas
@@ -449,6 +452,14 @@ def fts3Transfer(self):
       fts3Operation = FTS3TransferOperation.fromRMSObjects(self.request, self.operation, username)
       fts3Operation.ftsFiles = fts3Files
 
+      try:
+        if not fts3Operation.activity:
+          vo = getVOfromProxyGroup().get('Value')
+          fts3Plugin = getFTS3Plugin(vo=vo)
+          fts3Operation.activity = fts3Plugin.inferFTSActivity(fts3Operation, self.request, self.operation)
+      except Exception:
+        pass
+
       ftsSchedule = FTS3Client().persistOperation(fts3Operation)
       if not ftsSchedule["OK"]:
         self.log.error("Completely failed to schedule to FTS3:", ftsSchedule["Message"])

src/DIRAC/DataManagementSystem/Client/FTS3File.py

@@ -10,6 +10,7 @@ class FTS3File(JSerializable):
       (Transfer, Staging) should be executed
    """
 
+  # START states
   ALL_STATES = ['New',  # Nothing was attempted yet on this file
                 'Submitted',  # From FTS: Initial state of a file as soon it's dropped into the database
                 'Ready',  # From FTS: File is ready to become active
@@ -21,6 +22,7 @@ class FTS3File(JSerializable):
                 'Defunct',  # Totally fail, no more attempt will be made
                 'Started',  # From FTS: File transfer has started
                 'Not_used',  # From FTS: Transfer not being considered yet, waiting for another one (multihop)
+                'Archiving',  # From FTS: file not yet migrated to tape
                 ]
 
   # These are the states that we consider final.
@@ -40,6 +42,8 @@ class FTS3File(JSerializable):
   FTS_FINAL_STATES = FTS_SUCCESS_STATES + FTS_FAILED_STATES
   INIT_STATE = 'New'
 
+  # END states
+
   _attrToSerialize = ['fileID', 'operationID', 'status', 'attempt', 'creationTime',
                       'lastUpdate', 'rmsFileID', 'checksum', 'size', 'lfn', 'error', 'targetSE', 'ftsGUID']
 

src/DIRAC/DataManagementSystem/Client/FTS3Job.py

@@ -34,6 +34,8 @@ class FTS3Job(JSerializable):
       to an FTS3Operation
   """
 
+  # START states
+
   # States from FTS doc https://fts3-docs.web.cern.ch/fts3-docs/docs/state_machine.html
   ALL_STATES = ['Submitted',  # Initial state of a job as soon it's dropped into the database
                 'Ready',  # One of the files within a job went to Ready state
@@ -43,11 +45,13 @@ class FTS3Job(JSerializable):
                 'Failed',  # All files Failed
                 'Finisheddirty',  # Some files Failed
                 'Staging',  # One of the files within a job went to Staging state
+                'Archiving',  # From FTS: one of the files within a job went to Archiving state
                 ]
 
   FINAL_STATES = ['Canceled', 'Failed', 'Finished', 'Finisheddirty']
   INIT_STATE = 'Submitted'
 
+  # END states
   _attrToSerialize = ['jobID', 'operationID', 'status', 'error', 'submitTime',
                       'lastUpdate', 'ftsServer', 'ftsGUID', 'completeness',
                       'username', 'userGroup']
@@ -291,6 +295,7 @@ def _constructTransferJob(self, pinTime, allLFNs, target_spacetoken, protocols=N
     sourceIsTape = self.__isTapeSE(self.sourceSE, self.vo)
     copy_pin_lifetime = pinTime if sourceIsTape else None
     bring_online = BRING_ONLINE_TIMEOUT if sourceIsTape else None
+    archive_timeout = None
 
     # getting all the (source, dest) surls
     res = dstSE.generateTransferURLsBetweenSEs(allLFNs, srcSE, protocols=protocols)
@@ -303,6 +308,13 @@ def _constructTransferJob(self, pinTime, allLFNs, target_spacetoken, protocols=N
       log.error("Could not get source SURL", "%s %s" % (lfn, reason))
 
     allSrcDstSURLs = res['Value']['Successful']
+    srcProto, destProto = res['Value']['Protocols']
+
+    # If the destination is a tape, and the protocol supports it,
+    # check if we want to have an archive timeout
+    dstIsTape = self.__isTapeSE(self.targetSE, self.vo)
+    if dstIsTape and destProto in dstSE.localStageProtocolList:
+      archive_timeout = dstSE.options.get('ArchiveTimeout')
 
     # This contains the staging URLs if they are different from the transfer URLs
     # (CTA...)
@@ -311,31 +323,29 @@ def _constructTransferJob(self, pinTime, allLFNs, target_spacetoken, protocols=N
     # In case we are transfering from a tape system, and the stage protocol
     # is not the same as the transfer protocol, we generate the staging URLs
     # to do a multihop transfer. See below.
-    if sourceIsTape:
-      srcProto, _destProto = res['Value']['Protocols']
-      if srcProto not in srcSE.localStageProtocolList:
+    if sourceIsTape and srcProto not in srcSE.localStageProtocolList:
 
-        # As of version 3.10, FTS can only handle one file per multi hop
-        # job. If we are here, that means that we need one, so make sure that
-        # we only have a single file to transfer (this should have been checked
-        # at the job construction step in FTS3Operation).
-        # This test is important, because multiple files would result in the source
-        # being deleted !
-        if len(allLFNs) != 1:
-          log.debug("Multihop job has %s files while only 1 allowed" % len(allLFNs))
-          return S_ERROR(errno.E2BIG, "Trying multihop job with more than one file !")
+      # As of version 3.10, FTS can only handle one file per multi hop
+      # job. If we are here, that means that we need one, so make sure that
+      # we only have a single file to transfer (this should have been checked
+      # at the job construction step in FTS3Operation).
+      # This test is important, because multiple files would result in the source
+      # being deleted !
+      if len(allLFNs) != 1:
+        log.debug("Multihop job has %s files while only 1 allowed" % len(allLFNs))
+        return S_ERROR(errno.E2BIG, "Trying multihop job with more than one file !")
 
-        res = srcSE.getURL(allSrcDstSURLs, protocol=srcSE.localStageProtocolList)
+      res = srcSE.getURL(allSrcDstSURLs, protocol=srcSE.localStageProtocolList)
 
-        if not res['OK']:
-          return res
+      if not res['OK']:
+        return res
 
-        for lfn, reason in res['Value']['Failed'].items():
-          failedLFNs.add(lfn)
-          log.error("Could not get stage SURL", "%s %s" % (lfn, reason))
-          allSrcDstSURLs.pop(lfn)
+      for lfn, reason in res['Value']['Failed'].items():
+        failedLFNs.add(lfn)
+        log.error("Could not get stage SURL", "%s %s" % (lfn, reason))
+        allSrcDstSURLs.pop(lfn)
 
-        allStageURLs = res['Value']['Successful']
+      allStageURLs = res['Value']['Successful']
 
     transfers = []
 
@@ -374,6 +384,13 @@ def _constructTransferJob(self, pinTime, allLFNs, target_spacetoken, protocols=N
         # We do not set a fileID in the metadata
         # such that we do not update the DB when monitoring
         stageTrans_metadata = {'desc': 'PreStage %s' % ftsFileID}
+
+        # If we use an activity, also set it as file metadata
+        # for WLCG monitoring purposes
+        # https://its.cern.ch/jira/projects/DOMATPC/issues/DOMATPC-14?
+        if self.activity:
+          stageTrans_metadata['activity'] = self.activity
+
         stageTrans = fts3.new_transfer(stageURL,
                                        stageURL,
                                        checksum='ADLER32:%s' % ftsFile.checksum,
@@ -384,6 +401,12 @@ def _constructTransferJob(self, pinTime, allLFNs, target_spacetoken, protocols=N
 
       trans_metadata = {'desc': 'Transfer %s' % ftsFileID, 'fileID': ftsFileID}
 
+      # If we use an activity, also set it as file metadata
+      # for WLCG monitoring purposes
+      # https://its.cern.ch/jira/projects/DOMATPC/issues/DOMATPC-14?
+      if self.activity:
+        trans_metadata['activity'] = self.activity
+
       # because of an xroot bug (https://github.com/xrootd/xrootd/issues/1433)
       # the checksum needs to be lowercase. It does not impact the other
       # protocol, so it's fine to put it here.
@@ -412,6 +435,9 @@ def _constructTransferJob(self, pinTime, allLFNs, target_spacetoken, protocols=N
         'sourceSE': self.sourceSE,
         'targetSE': self.targetSE}
 
+    if self.activity:
+      job_metadata['activity'] = self.activity
+
     job = fts3.new_job(transfers=transfers,
                        overwrite=True,
                        source_spacetoken=source_spacetoken,
@@ -422,7 +448,8 @@ def _constructTransferJob(self, pinTime, allLFNs, target_spacetoken, protocols=N
                        verify_checksum='target',  # Only check target vs specified, since we verify the source earlier
                        multihop=bool(allStageURLs),  # if we have stage urls, then we need multihop
                        metadata=job_metadata,
-                       priority=self.priority)
+                       priority=self.priority,
+                       archive_timeout=archive_timeout)
 
     return S_OK((job, fileIDsInTheJob))
 
@@ -550,6 +577,9 @@ def _constructStagingJob(self, pinTime, allLFNs, target_spacetoken):
         'sourceSE': self.sourceSE,
         'targetSE': self.targetSE}
 
+    if self.activity:
+      job_metadata['activity'] = self.activity
+
     job = fts3.new_job(transfers=transfers,
                        overwrite=True,
                        source_spacetoken=target_spacetoken,

src/DIRAC/DataManagementSystem/Client/FTS3Operation.py

@@ -47,6 +47,7 @@ class FTS3Operation(JSerializable):
       subclass
   """
 
+  # START states
   ALL_STATES = ['Active',  # Default state until FTS has done everything
                 'Processed',  # Interactions with FTS done, but callback not done
                 'Finished',  # Everything was done
@@ -55,6 +56,7 @@ class FTS3Operation(JSerializable):
                 ]
   FINAL_STATES = ['Finished', 'Canceled', 'Failed']
   INIT_STATE = 'Active'
+  # END states
 
   _attrToSerialize = ['operationID', 'username', 'userGroup', 'rmsReqID', 'rmsOpID',
                       'sourceSEs', 'ftsFiles', 'activity', 'priority',

src/DIRAC/DataManagementSystem/private/FTS3Plugins/DefaultFTS3Plugin.py

@@ -104,3 +104,15 @@ def selectSourceSE(self, ftsFile, replicaDict, allowedSources):
 
     randSource = random.choice(list(allowedReplicaSource))  # one has to convert to list
     return randSource
+
+  def inferFTSActivity(self, ftsOperation, rmsRequest, rmsOperation):
+    """
+      This will try to find which FTS activity should be applied to
+      the FTS3Operation.
+
+      If nothing can be found, it will return None.
+
+      Note that this is only called if there is no hardcoded Activity in the
+      RMS Operation Arguments.
+    """
+    return None