Merge pull request #6530 from DIRACGridBot/cherry-pick-2-79fa433b7-integration

[sweep:integration] fix (resources): a few fixes related to ARC

Author: fstagni

src/DIRAC/Resources/Computing/ARC6ComputingElement.py

@@ -23,6 +23,19 @@ def __init__(self, ceUniqueID):
         # This should be reconstructed in getARCJob() to retrieve the outputs.
         self.restUrlPart = "rest/1.0/jobs/"
 
+        # Default ComputingInfo Endpoint, used to get details about the queues
+        self.computingInfoEndpoint = "org.nordugrid.ldapglue2"
+
+    def _reset(self):
+        super()._reset()
+        # ComputingInfoEndpoint to get information about queues
+        # https://www.nordugrid.org/arc/arc6/users/client_tools.html?#arcinfo
+        # Expected values are: [
+        # org.nordugrid.ldapng, org.nordugrid.ldapglue2, org.nordugrid.arcrest, org.ogf.glue.emies.resourceinfo
+        # ]
+        self.computingInfoEndpoint = self.ceParameters.get("ComputingInfoEndpoint", self.computingInfoEndpoint)
+        return S_OK()
+
     def _getARCJob(self, jobID):
         """Create an ARC Job with all the needed / possible parameters defined.
         By the time we come here, the environment variable X509_USER_PROXY should already be set
@@ -82,7 +95,7 @@ def submitJob(self, executableFile, proxy, numberOfJobs=1):
         stampDict = {}
 
         # Creating an endpoint
-        endpoint = arc.Endpoint(self.ceHost, arc.Endpoint.COMPUTINGINFO, "org.nordugrid.ldapglue2")
+        endpoint = arc.Endpoint(self.ceHost, arc.Endpoint.COMPUTINGINFO, self.computingInfoEndpoint)
 
         # Get the ExecutionTargets of the ComputingElement (Can be REST, EMI-ES or GRIDFTP)
         retriever = arc.ComputingServiceRetriever(self.usercfg, [endpoint])
@@ -163,7 +176,7 @@ def getCEStatus(self):
         self.usercfg.ProxyPath(os.environ["X509_USER_PROXY"])
 
         # Creating an endpoint
-        endpoint = arc.Endpoint(self.ceHost, arc.Endpoint.COMPUTINGINFO, "org.nordugrid.ldapglue2")
+        endpoint = arc.Endpoint(self.ceHost, arc.Endpoint.COMPUTINGINFO, self.computingInfoEndpoint)
 
         # Get the ExecutionTargets of the ComputingElement (Can be REST, EMI-ES or GRIDFTP)
         retriever = arc.ComputingServiceRetriever(self.usercfg, [endpoint])

src/DIRAC/Resources/Computing/ARCComputingElement.py

@@ -37,10 +37,10 @@
 """
 import os
 import stat
+import sys
 
 import arc  # Has to work if this module is called #pylint: disable=import-error
-from DIRAC import S_OK, S_ERROR, gConfig
-from DIRAC.ConfigurationSystem.Client.Helpers.Resources import getCESiteMapping
+from DIRAC import S_OK, S_ERROR
 from DIRAC.Core.Utilities.Subprocess import shellCall
 from DIRAC.Core.Utilities.File import makeGuid
 from DIRAC.Core.Utilities.List import breakListIntoChunks
@@ -50,13 +50,6 @@
 from DIRAC.WorkloadManagementSystem.Client import PilotStatus
 
 
-# Uncomment the following 5 lines for getting verbose ARC api output (debugging)
-# import sys
-# logstdout = arc.LogStream(sys.stdout)
-# logstdout.setFormat(arc.ShortFormat)
-# arc.Logger_getRootLogger().addDestination(logstdout)
-# arc.Logger_getRootLogger().setThreshold(arc.VERBOSE)
-
 MANDATORY_PARAMETERS = ["Queue"]  # Mandatory for ARC CEs in GLUE2?
 STATES_MAP = {
     "Accepted": PilotStatus.WAITING,
@@ -77,6 +70,8 @@
 
 class ARCComputingElement(ComputingElement):
 
+    _arcLevels = ["DEBUG", "VERBOSE", "INFO", "WARNING", "ERROR", "FATAL"]
+
     #############################################################################
     def __init__(self, ceUniqueID):
         """Standard constructor."""
@@ -117,14 +112,14 @@ def _getARCJob(self, jobID):
             j.JobStatusURL = arc.URL(str(statURL))
             j.JobStatusInterfaceName = "org.nordugrid.ldapng"
 
-            mangURL = "gsiftp://%s:2811/jobs/" % (self.ceHost)
+            mangURL = f"gsiftp://{self.ceHost}:2811/jobs/"
             j.JobManagementURL = arc.URL(str(mangURL))
             j.JobManagementInterfaceName = "org.nordugrid.gridftpjob"
 
             j.ServiceInformationURL = j.JobManagementURL
             j.ServiceInformationInterfaceName = "org.nordugrid.ldapng"
         else:
-            commonURL = "https://%s:8443/arex" % self.ceHost
+            commonURL = f"https://{self.ceHost}:8443/arex"
             j.JobStatusURL = arc.URL(str(commonURL))
             j.JobStatusInterfaceName = "org.ogf.glue.emies.activitymanagement"
 
@@ -195,7 +190,7 @@ def _writeXRSL(self, executableFile, inputs=None, outputs=None, executables=None
             if not isinstance(outputs, list):
                 outputs = [outputs]
             for outputFile in outputs:
-                xrslOutputs += '(%s "")' % (outputFile)
+                xrslOutputs += f'({outputFile} "")'
 
         xrsl = """
 &(executable="{executable}")
@@ -244,6 +239,23 @@ def _reset(self):
             self.log.warn("Unknown ARC endpoint, change to default", self.endpointType)
         else:
             self.endpointType = endpointType
+
+        # ARCLogLevel to enable/disable logs coming from the ARC client
+        # Because the ARC logger works independently from the standard logging library,
+        # it needs a specific initialization flag
+        # Expected values are: ["", "DEBUG", "VERBOSE", "INFO", "WARNING", "ERROR" and "FATAL"]
+        # Modifying the ARCLogLevel of an ARCCE instance would impact all existing instances within a same process.
+        logLevel = self.ceParameters.get("ARCLogLevel", "")
+        if logLevel:
+            arc.Logger_getRootLogger().removeDestinations()
+            if logLevel not in self._arcLevels:
+                self.log.warn("ARCLogLevel input is not known:", f"{logLevel} not in {self._arcLevels}")
+            else:
+                logstdout = arc.LogStream(sys.stdout)
+                logstdout.setFormat(arc.ShortFormat)
+                arc.Logger_getRootLogger().addDestination(logstdout)
+                arc.Logger_getRootLogger().setThreshold(getattr(arc, logLevel))
+
         return S_OK()
 
     #############################################################################
@@ -259,7 +271,7 @@ def submitJob(self, executableFile, proxy, numberOfJobs=1, inputs=None, outputs=
             return result
         self.usercfg.ProxyPath(os.environ["X509_USER_PROXY"])
 
-        self.log.verbose("Executable file path: %s" % executableFile)
+        self.log.verbose(f"Executable file path: {executableFile}")
         if not os.access(executableFile, 5):
             os.chmod(executableFile, stat.S_IRWXU | stat.S_IRGRP | stat.S_IXGRP | stat.S_IROTH + stat.S_IXOTH)
 
@@ -286,8 +298,8 @@ def submitJob(self, executableFile, proxy, numberOfJobs=1, inputs=None, outputs=
             jobdescs = arc.JobDescriptionList()
             # Get the job into the ARC way
             xrslString, diracStamp = self._writeXRSL(executableFile, inputs, outputs, executables)
-            self.log.debug("XRSL string submitted : %s" % xrslString)
-            self.log.debug("DIRAC stamp for job : %s" % diracStamp)
+            self.log.debug(f"XRSL string submitted : {xrslString}")
+            self.log.debug(f"DIRAC stamp for job : {diracStamp}")
             # The arc bindings don't accept unicode objects in Python 2 so xrslString must be explicitly cast
             result = arc.JobDescription.Parse(str(xrslString), jobdescs)
             if not result:

src/DIRAC/Resources/Computing/AREXComputingElement.py

@@ -4,9 +4,6 @@
     language is used in both cases. So, we retain the xrslExtraString and xrslMPExtraString strings.
 """
 
-
-__RCSID__ = "$Id$"
-
 import os
 import json
 import requests
@@ -59,13 +56,12 @@ def _reset(self):
         Note : This is not run from __init__ as the design of DIRAC means that ceParameters is
         filled with CEDefaults only at the time this class is initialised for the given CE
         """
-        # super()._reset()
+        super()._reset()
         self.log.debug("Testing if the REST interface is available", "for %s" % self.ceName)
 
         # Get options from the ceParameters dictionary
         self.port = self.ceParameters.get("Port", self.port)
         self.restVersion = self.ceParameters.get("RESTVersion", self.restVersion)
-        self.queue = self.ceParameters.get("Queue", self.queue)
 
         self.proxyTimeLeftBeforeRenewal = self.ceParameters.get(
             "proxyTimeLeftBeforeRenewal", self.proxyTimeLeftBeforeRenewal
@@ -432,10 +428,10 @@ def getCEStatus(self):
         result["SubmittedJobs"] = 0
 
         magic = self.queue + "_" + vo.lower()
-        for i in range(len(queueInfo)):
-            if queueInfo[i]["ID"].endswith(magic):
-                result["RunningJobs"] = queueInfo[i]["RunningJobs"]
-                result["WaitingJobs"] = queueInfo[i]["WaitingJobs"]
+        for _, qi in enumerate(queueInfo):
+            if qi["ID"].endswith(magic):
+                result["RunningJobs"] = qi["RunningJobs"]
+                result["WaitingJobs"] = qi["WaitingJobs"]
                 break  # Pick the first (should be only ...) matching queue + VO
 
         return result