refactor: moved threadScheduler out of ProxyManagerHandler. Agent ProxyRenewalAgent now necessary

Author: fstagni

docs/source/AdministratorGuide/Configuration/ConfReference/Systems/Framework/Agents/MyProxyRenewalAgent/index.rst

@@ -39,4 +39,3 @@ Agents associated with Framework System:
    :maxdepth: 2
 
    CAUpdateAgent/index
-   MyProxyRenewalAgent/index

docs/source/AdministratorGuide/Configuration/ConfReference/Systems/Framework/Agents/index.rst

@@ -17,10 +17,7 @@
     ("DIRAC.DataManagementSystem.Agent.FTS3Agent", {}),
     ("DIRAC.FrameworkSystem.Agent.CAUpdateAgent", {}),
     ("DIRAC.FrameworkSystem.Agent.ComponentSupervisionAgent", {}),
-    (
-        "DIRAC.FrameworkSystem.Agent.MyProxyRenewalAgent",
-        {"IgnoreOptions": ["MinValidity", "ValidityPeriod", "MinimumLifeTime", "RenewedLifeTime"]},
-    ),
+    ("DIRAC.FrameworkSystem.Agent.ProxyRenewalAgent", {}),
     ("DIRAC.RequestManagementSystem.Agent.CleanReqDBAgent", {}),
     (
         "DIRAC.RequestManagementSystem.Agent.RequestExecutingAgent",

src/DIRAC/ConfigurationSystem/test/Test_agentOptions.py

@@ -0,0 +1,85 @@
+"""  ProxyRenewalAgent keeps the proxy repository clean.
+
+    .. literalinclude:: ../ConfigTemplate.cfg
+      :start-after: ##BEGIN MyProxyRenewalAgent
+      :end-before: ##END
+      :dedent: 2
+      :caption: MyProxyRenewalAgent options
+"""
+import concurrent.futures
+
+from DIRAC import S_OK
+
+from DIRAC.Core.Base.AgentModule import AgentModule
+from DIRAC.FrameworkSystem.DB.ProxyDB import ProxyDB
+
+DEFAULT_MAIL_FROM = "[email protected]"
+
+
+class ProxyRenewalAgent(AgentModule):
+    def initialize(self):
+
+        requiredLifeTime = self.am_getOption("MinimumLifeTime", 3600)
+        renewedLifeTime = self.am_getOption("RenewedLifeTime", 54000)
+        mailFrom = self.am_getOption("MailFrom", DEFAULT_MAIL_FROM)
+        self.useMyProxy = self.am_getOption("UseMyProxy", False)
+        self.proxyDB = ProxyDB(useMyProxy=self.useMyProxy, mailFrom=mailFrom)
+
+        self.log.info(f"Minimum Life time      : {requiredLifeTime}")
+        self.log.info(f"Life time on renew     : {renewedLifeTime}")
+        if self.useMyProxy:
+            self.log.info(f"MyProxy server         : {self.proxyDB.getMyProxyServer()}")
+            self.log.info(f"MyProxy max proxy time : {self.proxyDB.getMyProxyMaxLifeTime()}")
+
+        return S_OK()
+
+    def __renewProxyForCredentials(self, userDN, userGroup):
+        lifeTime = self.am_getOption("RenewedLifeTime", 54000)
+        self.log.info(f"Renewing for {userDN}@{userGroup} {lifeTime} secs")
+        res = self.proxyDB.renewFromMyProxy(userDN, userGroup, lifeTime=lifeTime)
+        if not res["OK"]:
+            self.log.error("Failed to renew proxy", f"for {userDN}@{userGroup} : {res['Message']}")
+        else:
+            self.log.info(f"Renewed proxy for {userDN}@{userGroup}")
+
+    def execute(self):
+        """The main agent execution method"""
+        self.log.verbose("Purging expired requests")
+        res = self.proxyDB.purgeExpiredRequests()
+        if not res["OK"]:
+            self.log.error(res["Message"])
+        else:
+            self.log.info(f"Purged {res['Value']} requests")
+
+        self.log.verbose("Purging expired tokens")
+        res = self.proxyDB.purgeExpiredTokens()
+        if not res["OK"]:
+            self.log.error(res["Message"])
+        else:
+            self.log.info(f"Purged {res['Value']} tokens")
+
+        self.log.verbose("Purging expired proxies")
+        res = self.proxyDB.purgeExpiredProxies()
+        if not res["OK"]:
+            self.log.error(res["Message"])
+        else:
+            self.log.info(f"Purged {res['Value']} proxies")
+
+        self.log.verbose("Purging logs")
+        res = self.proxyDB.purgeLogs()
+        if not res["OK"]:
+            self.log.error(res["Message"])
+
+        if self.useMyProxy:
+            res = self.proxyDB.getCredentialsAboutToExpire(self.am_getOption("MinimumLifeTime", 3600))
+            if not res["OK"]:
+                return res
+            data = res["Value"]
+            self.log.info(f"Renewing {len(data)} proxies...")
+            with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
+                futures = []
+                for record in data:
+                    userDN = record[0]
+                    userGroup = record[1]
+                    futures.append(executor.submit(self.__renewProxyForCredentials, userDN, userGroup))
+            return S_OK()

src/DIRAC/FrameworkSystem/Agent/MyProxyRenewalAgent.py

@@ -77,7 +77,6 @@ Services
   TornadoProxyManager
   {
     Protocol = https
-    MaxThreads = 100
     # Flag to use myproxy server
     UseMyProxy = False
     # Email to use as a sender for the expiration reminder
@@ -154,7 +153,7 @@ Services
       Default = authenticated
       FileTransfer
       {
-	Default = authenticated
+        Default = authenticated
       }
     }
   }
@@ -218,15 +217,16 @@ Services
 }
 Agents
 {
-  ##BEGIN MyProxyRenewalAgent
-  MyProxyRenewalAgent
+  ##BEGIN ProxyRenewalAgent
+  ProxyRenewalAgent
   {
-    PollingTime = 1800
-    MinValidity = 10000
-    #The period for which the proxy will be extended. The value is in hours
-    ValidityPeriod = 15
+    PollingTime = 900
     # Email to use as a sender for the expiration reminder
     MailFrom = [email protected]
+    # Renew, or not, the proxies stored on the MyProxy service
+    UseMyProxy = False
+    MinimumLifeTime = 3600
+    RenewedLifeTime = 54000
   }
   ##END
   CAUpdateAgent

src/DIRAC/FrameworkSystem/Agent/ProxyRenewalAgent.py

@@ -27,7 +27,6 @@
 from DIRAC.Core.Security.X509Request import X509Request  # pylint: disable=import-error
 from DIRAC.Core.Security.X509Chain import X509Chain, isPUSPdn  # pylint: disable=import-error
 from DIRAC.ConfigurationSystem.Client.Helpers import Registry
-from DIRAC.ConfigurationSystem.Client.PathFinder import getDatabaseSection
 from DIRAC.FrameworkSystem.Client.NotificationClient import NotificationClient
 from DIRAC.Resources.ProxyProvider.ProxyProviderFactory import ProxyProviderFactory
 
@@ -1192,8 +1191,9 @@ def purgeLogs(self):
 
         :return: S_OK()/S_ERROR()
         """
-        cmd = "DELETE FROM `ProxyDB_Log` WHERE TIMESTAMPDIFF( SECOND, Timestamp, UTC_TIMESTAMP() ) > 15552000"
-        return self._update(cmd)
+        return self._update(
+            "DELETE FROM `ProxyDB_Log` WHERE TIMESTAMPDIFF( SECOND, Timestamp, UTC_TIMESTAMP() ) > 15552000"
+        )
 
     def getLogsContent(self, selDict, sortList, start=0, limit=0):
         """

src/DIRAC/FrameworkSystem/ConfigTemplate.cfg

@@ -9,7 +9,6 @@
 from DIRAC import gLogger, S_OK, S_ERROR
 from DIRAC.Core.DISET.RequestHandler import RequestHandler, getServiceOption
 from DIRAC.Core.Security import Properties
-from DIRAC.Core.Utilities.ThreadScheduler import gThreadScheduler
 from DIRAC.Core.Utilities.ObjectLoader import ObjectLoader
 from DIRAC.ConfigurationSystem.Client.Helpers import Registry
 
@@ -37,11 +36,7 @@ def initializeHandler(cls, serviceInfoDict):
             cls.__proxyDB = dbClass(useMyProxy=useMyProxy, mailFrom=mailFrom)
 
         except RuntimeError as excp:
-            return S_ERROR("Can't connect to ProxyDB: %s" % excp)
-        gThreadScheduler.addPeriodicTask(900, cls.__proxyDB.purgeExpiredTokens, elapsedTime=900)
-        gThreadScheduler.addPeriodicTask(900, cls.__proxyDB.purgeExpiredRequests, elapsedTime=900)
-        gThreadScheduler.addPeriodicTask(21600, cls.__proxyDB.purgeLogs)
-        gThreadScheduler.addPeriodicTask(3600, cls.__proxyDB.purgeExpiredProxies)
+            return S_ERROR("Can't connect to ProxyDB", repr(excp))
         if useMyProxy:
             gLogger.info("MyProxy: %s\n MyProxy Server: %s" % (useMyProxy, cls.__proxyDB.getMyProxyServer()))
         return S_OK()

src/DIRAC/FrameworkSystem/DB/ProxyDB.py

@@ -6,7 +6,7 @@
       CREATE TABLE `JobParameters` (
         `JobID` INT(11) UNSIGNED NOT NULL,
         `Name` VARCHAR(100) NOT NULL,
-	`Value` TEXT NOT NULL,
+        `Value` TEXT NOT NULL,
         PRIMARY KEY (`JobID`,`Name`),
         FOREIGN KEY (`JobID`) REFERENCES `Jobs`(`JobID`)
       ) ENGINE=InnoDB DEFAULT CHARSET=latin1;

src/DIRAC/FrameworkSystem/Service/ProxyManagerHandler.py

@@ -10,7 +10,7 @@ LocalInstallation
   InstanceName = Production
   SkipCADownload = yes
   UseServerCertificate = yes
-  #  ConfigurationServer = dips://myprimaryserver.name:9135/Configuration/Server
+  #  ConfigurationServer = https://myprimaryserver.name:8443/Configuration/Server
   ConfigurationName = Production
   #LogLevel of the installed components
   LogLevel = DEBUG