Merge pull request #6854 from chaen/v8.0_FIX_repAndRegisterCheckSEs

[8.0] ReplicateAndRegister  with FTS checks SE status

Author: fstagni

src/DIRAC/DataManagementSystem/Agent/FTS3Agent.py

@@ -12,7 +12,10 @@
   :caption: FTS3Agent options
 
 """
+import datetime
 import errno
+import os
+from urllib import parse
 import time
 
 # from threading import current_thread
@@ -52,6 +55,10 @@
 # when running multiple agents, we rather do it in steps
 JOB_MONITORING_BATCH_SIZE = 20
 
+# We do not monitor a job more often
+# than MONITORING_DELAY in minutes
+MONITORING_DELAY = 10
+
 
 class FTS3Agent(AgentModule):
     """
@@ -108,6 +115,8 @@ def initialize(self):
         # name that will be used in DB for assignment tag
         self.assignmentTag = gethostname().split(".")[0]
 
+        self.workDirectory = self.am_getWorkDirectory()
+
         res = self.__readConf()
 
         # We multiply by two because of the two threadPools
@@ -165,13 +174,19 @@ def getFTS3Context(self, username, group, ftsServer, threadID):
 
             log.debug(f"UserDN {userDN}")
 
+            # Chose a meaningful proxy name for easier debugging
+            srvName = parse.urlparse(ftsServer).netloc.split(":")[0]
+            proxyFile = os.path.join(
+                self.workDirectory, f"{int(time.time())}_{username}_{group}_{srvName}_{threadID}.pem"
+            )
+
             # We dump the proxy to a file.
             # It has to have a lifetime of self.proxyLifetime
             # Because the FTS3 servers cache it for 2/3rd of the lifetime
             # we should make our cache a bit less than 2/3rd of the lifetime
             cacheTime = int(2 * self.proxyLifetime / 3) - 600
             res = gProxyManager.downloadVOMSProxyToFile(
-                userDN, group, requiredTimeLeft=self.proxyLifetime, cacheTime=cacheTime
+                userDN, group, requiredTimeLeft=self.proxyLifetime, cacheTime=cacheTime, filePath=proxyFile
             )
             if not res["OK"]:
                 return res
@@ -289,18 +304,29 @@ def monitorJobsLoop(self):
         if mod:
             nbOfLoops += 1
 
+        # Not only is it pointless to monitor right after submission
+        # but also we would end up fetching multiple time the same job otherwise
+        # as we call getActiveJobs by batch
+        lastMonitor = datetime.datetime.utcnow() - datetime.timedelta(minutes=MONITORING_DELAY)
+
         log.debug("Getting active jobs")
 
         for loopId in range(nbOfLoops):
             log.info("Getting next batch of jobs to monitor", f"{loopId}/{nbOfLoops}")
             # get jobs from DB
-            res = self.fts3db.getActiveJobs(limit=JOB_MONITORING_BATCH_SIZE, jobAssignmentTag=self.assignmentTag)
+            res = self.fts3db.getActiveJobs(
+                limit=JOB_MONITORING_BATCH_SIZE, lastMonitor=lastMonitor, jobAssignmentTag=self.assignmentTag
+            )
 
             if not res["OK"]:
                 log.error("Could not retrieve ftsJobs from the DB", res)
                 return res
 
             activeJobs = res["Value"]
+            if not activeJobs:
+                log.info("No more jobs to monitor")
+                break
+
             log.info("Jobs queued for monitoring", len(activeJobs))
 
             # We store here the AsyncResult object on which we are going to wait

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

@@ -184,6 +184,35 @@ def __call__(self):
         if self.rmsMonitoring:
             self.rmsMonitoringReporter = MonitoringReporter(monitoringType="RMSMonitoring")
 
+        sourceSE = self.operation.SourceSE if self.operation.SourceSE else None
+        if sourceSE:
+            # check sourceSE for read
+            bannedSource = self.checkSEsRSS(sourceSE, "ReadAccess")
+            if not bannedSource["OK"]:
+                if self.rmsMonitoring:
+                    for status in ["Attempted", "Failed"]:
+                        self.rmsMonitoringReporter.addRecord(self.createRMSRecord(status, len(self.operation)))
+                    self.rmsMonitoringReporter.commit()
+                return bannedSource
+
+            if bannedSource["Value"]:
+                self.operation.Error = f"SourceSE {sourceSE} is banned for reading"
+                self.log.info(self.operation.Error)
+                return S_OK(self.operation.Error)
+
+        # check targetSEs for write
+        bannedTargets = self.checkSEsRSS()
+        if not bannedTargets["OK"]:
+            if self.rmsMonitoring:
+                for status in ["Attempted", "Failed"]:
+                    self.rmsMonitoringReporter.addRecord(self.createRMSRecord(status, len(self.operation)))
+                self.rmsMonitoringReporter.commit()
+            return bannedTargets
+
+        if bannedTargets["Value"]:
+            self.operation.Error = f"{','.join(bannedTargets['Value'])} targets are banned for writing"
+            return S_OK(self.operation.Error)
+
         # # check replicas first
         checkReplicas = self.__checkReplicas()
         if not checkReplicas["OK"]:
@@ -480,36 +509,6 @@ def fts3Transfer(self):
 
     def dmTransfer(self, fromFTS=False):
         """replicate and register using dataManager"""
-        # # get waiting files. If none just return
-        # # source SE
-        sourceSE = self.operation.SourceSE if self.operation.SourceSE else None
-        if sourceSE:
-            # # check source se for read
-            bannedSource = self.checkSEsRSS(sourceSE, "ReadAccess")
-            if not bannedSource["OK"]:
-                if self.rmsMonitoring:
-                    for status in ["Attempted", "Failed"]:
-                        self.rmsMonitoringReporter.addRecord(self.createRMSRecord(status, len(self.operation)))
-                    self.rmsMonitoringReporter.commit()
-                return bannedSource
-
-            if bannedSource["Value"]:
-                self.operation.Error = f"SourceSE {sourceSE} is banned for reading"
-                self.log.info(self.operation.Error)
-                return S_OK(self.operation.Error)
-
-        # # check targetSEs for write
-        bannedTargets = self.checkSEsRSS()
-        if not bannedTargets["OK"]:
-            if self.rmsMonitoring:
-                for status in ["Attempted", "Failed"]:
-                    self.rmsMonitoringReporter.addRecord(self.createRMSRecord(status, len(self.operation)))
-                self.rmsMonitoringReporter.commit()
-            return bannedTargets
-
-        if bannedTargets["Value"]:
-            self.operation.Error = f"{','.join(bannedTargets['Value'])} targets are banned for writing"
-            return S_OK(self.operation.Error)
 
         # Can continue now
         self.log.verbose("No targets banned for writing")