fix: Refactor the PilotLoggingAgent - download proxies at initialisation (adress #7249)

Author: martynia

src/DIRAC/WorkloadManagementSystem/Agent/PilotLoggingAgent.py

@@ -12,12 +12,15 @@
 # # imports
 import os
 import time
+import tempfile
 from DIRAC import S_OK, S_ERROR, gConfig
 from DIRAC.ConfigurationSystem.Client.Helpers.Operations import Operations
 from DIRAC.ConfigurationSystem.Client.Helpers.Registry import getVOs
 from DIRAC.Core.Base.AgentModule import AgentModule
-from DIRAC.Core.Utilities.Proxy import executeWithUserProxy
+from DIRAC.Core.Utilities.Proxy import executeWithoutServerCertificate
+from DIRAC.Core.Utilities.Proxy import getProxy
 from DIRAC.DataManagementSystem.Client.DataManager import DataManager
+from DIRAC.ConfigurationSystem.Client.Helpers.Registry import getVOMSAttributeForGroup, getDNForUsername
 from DIRAC.WorkloadManagementSystem.Client.TornadoPilotLoggingClient import TornadoPilotLoggingClient
 
 
@@ -36,9 +39,8 @@ def __init__(self, *args, **kwargs):
     def initialize(self):
         """
         agent's initialisation. Use this agent's CS information to:
-        Determine what Defaults/Shifter shifter proxy to use.,
-        get the target SE name from the CS.
-        Obtain log file location from Tornado.
+        Determine VOs with remote logging enabled,
+        Determine what Defaults/Shifter shifter proxy to use., download the proxies.
 
         :param self: self reference
         """
@@ -52,25 +54,17 @@ def initialize(self):
 
         if isinstance(self.voList, str):
             self.voList = [self.voList]
+        # download shifter proxies for enabled VOs:
+        self.proxyDict = {}
 
-        return S_OK()
-
-    def execute(self):
-        """
-        Execute one agent cycle. Upload log files to the SE and register them in the DFC.
-        Use a shifter proxy dynamically loaded for every VO
-
-        :param self: self reference
-        """
-        voRes = {}
         for vo in self.voList:
-            self.opsHelper = Operations(vo=vo)
+            opsHelper = Operations(vo=vo)
             # is remote pilot logging enabled for the VO ?
-            pilotLogging = self.opsHelper.getValue("/Pilot/RemoteLogging", False)
+            pilotLogging = opsHelper.getValue("/Pilot/RemoteLogging", False)
             if pilotLogging:
-                res = self.opsHelper.getOptionsDict("Shifter/DataManager")
+                res = opsHelper.getOptionsDict("Shifter/DataManager")
                 if not res["OK"]:
-                    voRes[vo] = "No shifter defined - skipped"
+                    # voRes[vo] = "No shifter defined - skipped"
                     self.log.error(f"No shifter defined for VO: {vo} - skipping ...")
                     continue
 
@@ -80,36 +74,73 @@ def execute(self):
                     self.log.error(
                         f"No proxy user or group defined for pilot: VO: {vo}, User: {proxyUser}, Group: {proxyGroup}"
                     )
-                    voRes[vo] = "No proxy user or group defined - skipped"
                     continue
 
                 self.log.info(f"Proxy used for pilot logging: VO: {vo}, User: {proxyUser}, Group: {proxyGroup}")
-                res = self.executeForVO(  # pylint: disable=unexpected-keyword-arg
-                    vo, proxyUserName=proxyUser, proxyUserGroup=proxyGroup
-                )
-                if not res["OK"]:
-                    voRes[vo] = res["Message"]
+                # download a proxy and save a filename for future use:
+                result = getDNForUsername(proxyUser)
+                if not result["OK"]:
+                    self.log.error(f"Could not obtain a DN of user {proxyUser} for VO {vo}, skipped")
+                    continue
+                userDNs = result["Value"]  # a same user may have more than one DN
+                fd, filename = tempfile.mkstemp(prefix=vo + "__")
+                os.close(fd)
+                vomsAttr = getVOMSAttributeForGroup(proxyGroup)
+                result = getProxy(userDNs, proxyGroup, vomsAttr=vomsAttr, proxyFilePath=filename)
+
+                if not result["OK"]:
+                    self.log.error(
+                        f"Could not download a proxy for DN {userDNs}, group {proxyGroup} for VO {vo}, skipped"
+                    )
+                    return result
+                self.proxyDict[vo] = result["Value"]
+
+        return S_OK()
+
+    def execute(self):
+        """
+        Execute one agent cycle. Upload log files to the SE and register them in the DFC.
+        Consider only VOs we have proxies for.
+
+        :param self: self reference
+        """
+        voRes = {}
+        self.log.verbose(f"VOs configured for remote logging: {list(self.proxyDict.keys())}")
+        originalUserProxy = os.environ.get("X509_USER_PROXY")
+        for vo, proxy in self.proxyDict.items():
+            os.environ["X509_USER_PROXY"] = proxy
+            res = self.executeForVO(vo)
+            if not res["OK"]:
+                voRes[vo] = res["Message"]
+        # restore the original proxy:
+        if originalUserProxy:
+            os.environ["X509_USER_PROXY"] = originalUserProxy
+        else:
+            os.environ.pop("X509_USER_PROXY", None)
+
         if voRes:
             for key, value in voRes.items():
                 self.log.error(f"Error for {key} vo; message: {value}")
             voRes.update(S_ERROR("Agent cycle for some VO finished with errors"))
             return voRes
+
         return S_OK()
 
-    @executeWithUserProxy
+    @executeWithoutServerCertificate
     def executeForVO(self, vo):
         """
         Execute one agent cycle for a VO. It obtains VO-specific configuration pilot options from the CS:
         UploadPath - the path where the VO wants to upload pilot logs. It has to start with a VO name (/vo/path).
         UploadSE - Storage element where the logs will be kept.
 
-        :param str vo: vo enabled for remote pilot logging
+        :param str vo: vo enabled for remote pilot logging (and a successfully downloaded proxy for the VO)
         :return: S_OK or S_ERROR
         :rtype: dict
         """
 
         self.log.info(f"Pilot files upload cycle started for VO: {vo}")
-        res = self.opsHelper.getOptionsDict("Pilot")
+        opsHelper = Operations(vo=vo)
+        res = opsHelper.getOptionsDict("Pilot")
         if not res["OK"]:
             return S_ERROR(f"No pilot section for {vo} vo")
         pilotOptions = res["Value"]