feat: add scitag support

Author: fstagni

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

@@ -25,25 +25,22 @@
 import re
 from collections import defaultdict
 
+import requests
+from cachetools import TTLCache, cached
+
 # # from DIRAC
-from DIRAC import S_OK, S_ERROR, gLogger
-from DIRAC.Core.Utilities.Adler import compareAdler, hexAdlerToInt, intAdlerToHex
+from DIRAC import S_ERROR, S_OK, gLogger
 from DIRAC.Core.Security.ProxyInfo import getVOfromProxyGroup
-
-from DIRAC.DataManagementSystem.Client.DataManager import DataManager
+from DIRAC.Core.Utilities.Adler import compareAdler, hexAdlerToInt, intAdlerToHex
 from DIRAC.DataManagementSystem.Agent.RequestOperations.DMSRequestOperationsBase import DMSRequestOperationsBase
-
-from DIRAC.Resources.Storage.StorageElement import StorageElement
-from DIRAC.Resources.Catalog.FileCatalog import FileCatalog
-
-from DIRAC.DataManagementSystem.Client.FTS3Operation import FTS3TransferOperation
-from DIRAC.DataManagementSystem.Client.FTS3File import FTS3File
+from DIRAC.DataManagementSystem.Client.DataManager import DataManager
 from DIRAC.DataManagementSystem.Client.FTS3Client import FTS3Client
+from DIRAC.DataManagementSystem.Client.FTS3File import FTS3File
+from DIRAC.DataManagementSystem.Client.FTS3Operation import FTS3TransferOperation
 from DIRAC.DataManagementSystem.private.FTS3Utilities import getFTS3Plugin
-
-from DIRAC.ConfigurationSystem.Client.Helpers import Registry
-
 from DIRAC.MonitoringSystem.Client.MonitoringReporter import MonitoringReporter
+from DIRAC.Resources.Catalog.FileCatalog import FileCatalog
+from DIRAC.Resources.Storage.StorageElement import StorageElement
 
 
 def filterReplicas(opFile, logger=None, dataManager=None, opSources=None, activeReplicas=None):
@@ -154,6 +151,44 @@ def filterReplicas(opFile, logger=None, dataManager=None, opSources=None, active
     return S_OK(result)
 
 
+def get_scitag(vo: str, activity: str = None):
+    """
+    Get the scitag based on the VO and activity.
+    If the VO is not found in the scitag.json, it defaults to 1.
+    If no specific activity is provided, it defaults to the "default" activityName.
+
+    :param vo: The VO for which to get the scitag
+    :param activity: The activity for which to get the scitag
+    :return: The scitag value
+    """
+
+    # Create a TTL cache: max 1 item, expires after 84400 seconds (1 day)
+    json_cache = TTLCache(maxsize=1, ttl=86400)
+
+    @cached(json_cache)
+    def get_remote_json():
+        gLogger.verbose("Fetching https://scitags.org/api.json from the network")
+        response = requests.get("https://scitags.org/api.json")
+        response.raise_for_status()
+        return response.json()
+
+    # Load the JSON data from the cache or network
+    sj = get_remote_json()
+
+    vo_id = 1  # Default VO ID
+    activity_id = 1  # Default activity ID
+
+    for experiment in sj.get("experiments", []):
+        if experiment.get("expName") == vo.lower():
+            vo_id = experiment.get("expId")
+            for act in experiment.get("activities", []):
+                if act.get("activityName") == activity:
+                    activity_id = act.get("activityId")
+
+    # Logic to determine the scitag based on vo and activity (this is what FTS wants)
+    return vo_id << 6 | activity_id  # Example logic, replace with actual implementation
+
+
 ########################################################################
 class ReplicateAndRegister(DMSRequestOperationsBase):
     """
@@ -472,14 +507,23 @@ def fts3Transfer(self):
             fts3Operation = FTS3TransferOperation.fromRMSObjects(self.request, self.operation)
             fts3Operation.ftsFiles = fts3Files
 
+            vo = getVOfromProxyGroup().get("Value")
             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
 
+            try:
+                if not fts3Operation.scitag:
+                    fts3Operation.scitag = get_scitag(
+                        vo=vo,
+                        activity=fts3Operation.activity if fts3Operation.activity else None,
+                    )
+            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/FTS3Job.py

@@ -2,13 +2,13 @@
 
 import datetime
 import errno
-from packaging.version import Version
 
+import fts3.rest.client.easy as fts3
 
 # Requires at least version 3.3.3
 from fts3 import __version__ as fts3_version
-import fts3.rest.client.easy as fts3
 from fts3.rest.client.exceptions import FTS3ClientException, NotFound
+from packaging.version import Version
 
 # There is a breaking change in the API in 3.13
 # https://gitlab.cern.ch/fts/fts-rest-flask/-/commit/5faa283e0cd4b80a0139a547c4a6356522c8449d
@@ -22,16 +22,13 @@
 # default pycurl. See https://its.cern.ch/jira/browse/FTS-261
 from fts3.rest.client.request import Request as ftsSSLRequest
 
-from DIRAC.Resources.Storage.StorageElement import StorageElement
-
-from DIRAC.FrameworkSystem.Client.Logger import gLogger
-from DIRAC.FrameworkSystem.Client.TokenManagerClient import gTokenManager
-
-from DIRAC.Core.Utilities.ReturnValues import S_OK, S_ERROR
 from DIRAC.Core.Utilities.DErrno import cmpError
-
 from DIRAC.Core.Utilities.JEncode import JSerializable
+from DIRAC.Core.Utilities.ReturnValues import S_ERROR, S_OK
 from DIRAC.DataManagementSystem.Client.FTS3File import FTS3File
+from DIRAC.FrameworkSystem.Client.Logger import gLogger
+from DIRAC.FrameworkSystem.Client.TokenManagerClient import gTokenManager
+from DIRAC.Resources.Storage.StorageElement import StorageElement
 
 # 3 days in seconds
 BRING_ONLINE_TIMEOUT = 259200
@@ -104,6 +101,7 @@ def __init__(self):
         self.targetSE = None
         self.filesToSubmit = []
         self.activity = None
+        self.scitag = None
         self.priority = None
         self.vo = None
         self.rmsReqID = None
@@ -485,6 +483,7 @@ def _constructTransferJob(self, pinTime, allLFNs, target_spacetoken, protocols=N
                         filesize=ftsFile.size,
                         metadata=stageTrans_metadata,
                         activity=self.activity,
+                        scitag=self.scitag,
                     )
                     transfers.append(stageTrans)
 
@@ -553,6 +552,7 @@ def _constructTransferJob(self, pinTime, allLFNs, target_spacetoken, protocols=N
                     activity=self.activity,
                     source_token=srcToken,
                     destination_token=dstToken,
+                    scitag=self.scitag,
                 )
 
                 transfers.append(trans)

src/DIRAC/DataManagementSystem/Client/FTS3Operation.py

@@ -1,33 +1,27 @@
 import datetime
 import errno
 import json
+
 from sqlalchemy import orm
 
+from DIRAC import S_ERROR, S_OK
+from DIRAC.ConfigurationSystem.Client.Helpers.Registry import getVOForGroup
+from DIRAC.Core.Utilities.DErrno import cmpError
+from DIRAC.Core.Utilities.JEncode import JSerializable
+from DIRAC.Core.Utilities.List import breakListIntoChunks
+from DIRAC.Core.Utilities.ReturnValues import returnSingleResult
+from DIRAC.DataManagementSystem.Client.DataManager import DataManager
+from DIRAC.DataManagementSystem.Client.FTS3File import FTS3File
 from DIRAC.DataManagementSystem.Client.FTS3Job import FTS3Job
 from DIRAC.DataManagementSystem.private import FTS3Utilities
 from DIRAC.DataManagementSystem.Utilities.DMSHelpers import DMSHelpers
-
-from DIRAC.DataManagementSystem.Client.DataManager import DataManager
 from DIRAC.FrameworkSystem.Client.Logger import gLogger
-
-from DIRAC.Resources.Storage.StorageElement import StorageElement
-from DIRAC.Core.Utilities.ReturnValues import returnSingleResult
-from DIRAC.Core.Utilities.DErrno import cmpError
-
-
-from DIRAC import S_OK, S_ERROR
-
-from DIRAC.Core.Utilities.List import breakListIntoChunks
-from DIRAC.ResourceStatusSystem.Client.ResourceStatus import ResourceStatus
-from DIRAC.DataManagementSystem.Client.FTS3File import FTS3File
-from DIRAC.Core.Utilities.JEncode import JSerializable
-
-from DIRAC.RequestManagementSystem.Client.ReqClient import ReqClient
-from DIRAC.RequestManagementSystem.Client.Operation import Operation as rmsOperation
 from DIRAC.RequestManagementSystem.Client.File import File as rmsFile
+from DIRAC.RequestManagementSystem.Client.Operation import Operation as rmsOperation
+from DIRAC.RequestManagementSystem.Client.ReqClient import ReqClient
 from DIRAC.RequestManagementSystem.Client.Request import Request as rmsRequest
-
-from DIRAC.ConfigurationSystem.Client.Helpers.Registry import getVOForGroup
+from DIRAC.Resources.Storage.StorageElement import StorageElement
+from DIRAC.ResourceStatusSystem.Client.ResourceStatus import ResourceStatus
 
 
 class FTS3Operation(JSerializable):
@@ -83,6 +77,7 @@ def __init__(
         sourceSEs=None,
         activity=None,
         priority=None,
+        scitag=None,
     ):
         """
         :param ftsFiles: list of FTS3Files object that belongs to the operation
@@ -93,6 +88,7 @@ def __init__(
         :param sourceSEs: list of SE to be used as source (if applicable)
         :param activity: FTS activity to use
         :param priority: FTS priority to use
+        :param scitag: optional scitag to be used for the transfers
 
         """
         ############################
@@ -113,6 +109,7 @@ def __init__(
 
         self.activity = activity
         self.priority = priority
+        self.scitag = scitag
 
         self.ftsJobs = []
 
@@ -254,6 +251,7 @@ def _createNewJob(self, jobType, ftsFiles, targetSE, sourceSE=None, multiHopSE=N
         newJob.targetSE = targetSE
         newJob.activity = self.activity
         newJob.priority = self.priority
+        newJob.scitag = self.scitag
         newJob.username = self.username
         newJob.userGroup = self.userGroup
         newJob.vo = self.vo
@@ -387,6 +385,7 @@ def fromRMSObjects(cls, rmsReq, rmsOp):
 
             ftsOp.activity = argumentDic["activity"]
             ftsOp.priority = argumentDic["priority"]
+            ftsOp.scitag = argumentDic.get("scitag", 65)  # 65 is 1 << 6 | 1 (default experiment, default activity)
         except Exception:
             pass
 

src/DIRAC/DataManagementSystem/DB/FTS3DB.py

@@ -104,6 +104,7 @@
     Column("sourceSEs", String(255)),
     Column("activity", String(255)),
     Column("priority", SmallInteger),
+    Column("scitag", SmallInteger, server_default="65"),  # 65 is 1 << 6 | 1 (default experiment, default activity)
     Column("creationTime", DateTime),
     Column("lastUpdate", DateTime, onupdate=utc_timestamp()),
     Column("status", Enum(*FTS3Operation.ALL_STATES), server_default=FTS3Operation.INIT_STATE, index=True),