Merge pull request DIRACGrid#5443 from chrisburr/rel-v7r2-merge-2021-10-01

[v7r2] Merge party for rel-v7r2

Author: chrisburr

src/DIRAC/Core/Utilities/List.py

@@ -111,6 +111,7 @@ def getChunk(aList, chunkSize):
         process( chunk )
 
   """
+  chunkSize = int(chunkSize)
   for i in range(0, len(aList), chunkSize):
     yield aList[i:i + chunkSize]
 

src/DIRAC/DataManagementSystem/Client/DataManager.py

@@ -13,12 +13,12 @@
 from __future__ import print_function
 
 # # imports
-import six
 from datetime import datetime, timedelta
 import fnmatch
 import os
 import time
 import errno
+import six
 
 # # from DIRAC
 import DIRAC
@@ -187,7 +187,6 @@ def __cleanDirectory(self, folder):
       return res
 
     # create a list of folders so that empty folders are also deleted
-    listOfFolders = []
     areDirs = self.fileCatalog.isDirectory(res['Value'])
     if not areDirs['OK']:
       return areDirs

src/DIRAC/RequestManagementSystem/private/RequestTask.py

@@ -29,9 +29,11 @@
 # # imports
 import os
 import time
+import datetime
 
 # # from DIRAC
 from DIRAC import gLogger, S_OK, S_ERROR, gConfig
+from DIRAC.Core.Utilities import DErrno
 from DIRAC.FrameworkSystem.Client.MonitoringClient import gMonitor
 from DIRAC.RequestManagementSystem.Client.ReqClient import ReqClient
 from DIRAC.RequestManagementSystem.Client.Request import Request
@@ -55,15 +57,8 @@ class RequestTask(object):
   request's processing task
   """
 
-  def __init__(
-          self,
-          requestJSON,
-          handlersDict,
-          csPath,
-          agentName,
-          standalone=False,
-          requestClient=None,
-          rmsMonitoring=False):
+  def __init__(self, requestJSON, handlersDict, csPath, agentName,
+               standalone=False, requestClient=None, rmsMonitoring=False):
     """c'tor
 
     :param self: self reference
@@ -190,6 +185,7 @@ def setupProxy(self):
 
   @staticmethod
   def getPluginName(pluginPath):
+    """ return plugin name """
     if not pluginPath:
       return ''
     if "/" in pluginPath:
@@ -282,19 +278,25 @@ def __call__(self):
     # # setup proxy for request owner
     setupProxy = self.setupProxy()
     if not setupProxy["OK"]:
+      userSuspended = "User is currently suspended"
       self.request.Error = setupProxy["Message"]
-      if 'has no proxy registered' in setupProxy["Message"]:
+      # In case the user does not have proxy
+      if DErrno.cmpError(setupProxy, DErrno.EPROXYFIND):
         self.log.error('Error setting proxy. Request set to Failed:', setupProxy["Message"])
         # If user is no longer registered, fail the request
         for operation in self.request:
           for opFile in operation:
             opFile.Status = 'Failed'
           operation.Status = 'Failed'
+      elif userSuspended in setupProxy['Message']:
+        # If user is suspended, wait for a long time
+        self.request.delayNextExecution(6 * 60)
+        self.request.Error = userSuspended
+        self.log.error("Error setting proxy: " + userSuspended, self.request.OwnerDN)
       else:
         self.log.error("Error setting proxy", setupProxy["Message"])
       return S_OK(self.request)
     shifter = setupProxy["Value"]["Shifter"]
-    proxyFile = setupProxy["Value"]["ProxyFile"]
 
     error = None
 

src/DIRAC/Resources/MessageQueue/StompMQConnector.py

@@ -158,7 +158,11 @@ def put(self, message, parameters=None):
     destination = parameters.get('destination', '')
 
     try:
-      self.connection.send(body=json.dumps(message), destination=destination)
+      try:
+        self.connection.send(body=json.dumps(message), destination=destination)
+      except stomp.exception.StompException:
+        self.connect()
+        self.connection.send(body=json.dumps(message), destination=destination)
     except Exception as e:
       log.debug("Failed to send message", repr(e))
       return S_ERROR(EMQUKN, 'Failed to send message: %s' % repr(e))

src/DIRAC/WorkloadManagementSystem/Executor/JobSanity.py

@@ -68,7 +68,7 @@ def optimizeJob(self, jid, jobState):
         return result
       self.jobLog.info("Found LFNs", result['Value'])
 
-    # Input Sandbox uploaded check
+    # Input Sandbox check
     if self.ex_getOption('InputSandboxCheck', True):
       result = self.checkInputSandbox(jobState, manifest)
       if not result['OK']:
@@ -141,9 +141,13 @@ def checkInputSandbox(self, jobState, manifest):
     isbList = manifest.getOption('InputSandbox', [])
     sbsToAssign = []
     for isb in isbList:
-      if isb.find("SB:") == 0:
-        self.jobLog.info("Found a sandbox", isb)
+      if isb.startswith("SB:"):
+        self.jobLog.debug("Found a sandbox", isb)
         sbsToAssign.append((isb, "Input"))
+      if isb.startswith("LFN:"):
+        self.jobLog.debug("Found a LFN sandbox", isb)
+        if isb[4] != '/':  # the LFN does not start with /
+          return S_ERROR("LFNs should always start with '/'")
     numSBsToAssign = len(sbsToAssign)
     if not numSBsToAssign:
       return S_OK(0)