style: better formatting and docstrings

Author: fstagni

src/DIRAC/WorkloadManagementSystem/Agent/PushJobAgent.py

@@ -25,7 +25,6 @@
 from DIRAC.WorkloadManagementSystem.Service.WMSUtilities import getGridEnv
 from DIRAC.WorkloadManagementSystem.Agent.JobAgent import JobAgent
 from DIRAC.WorkloadManagementSystem.private.ConfigHelper import findGenericPilotCredentials
-from DIRAC.WorkloadManagementSystem.Client import JobStatus
 
 MAX_JOBS_MANAGED = 100
 

src/DIRAC/WorkloadManagementSystem/Agent/StalledJobAgent.py

@@ -1,14 +1,12 @@
-""" The StalledJobAgent hunts for stalled jobs in the Job database. Jobs in "running"
-    state not receiving a heart beat signal for more than stalledTime
-    seconds will be assigned the "Stalled" state.
-
+"""The StalledJobAgent hunts for stalled jobs in the Job database. Jobs in
+"running" state not receiving a heart beat signal for more than stalledTime
+seconds will be assigned the "Stalled" state.
 
 .. literalinclude:: ../ConfigTemplate.cfg
   :start-after: ##BEGIN StalledJobAgent
   :end-before: ##END
   :dedent: 2
   :caption: StalledJobAgent options
-
 """
 import concurrent.futures
 import datetime
@@ -32,23 +30,28 @@
 
 
 class StalledJobAgent(AgentModule):
-    """Agent for setting Running jobs Stalled, and Stalled jobs Failed. And a few more."""
+    """Agent for setting Running jobs Stalled, and Stalled jobs Failed.
+
+    And a few more.
+    """
 
     def __init__(self, *args, **kwargs):
-        """c'tor"""
+        """c'tor."""
         super().__init__(*args, **kwargs)
 
         self.jobDB = None
         self.logDB = None
         self.matchedTime = 7200
         self.rescheduledTime = 600
         self.submittingTime = 300
+        self.stalledJobsToleranceTime = 0
         self.stalledJobsTolerantSites = []
         self.stalledJobsToRescheduleSites = []
+        self.threadPoolExecutor = None
 
     #############################################################################
     def initialize(self):
-        """Sets default parameters"""
+        """Sets default parameters."""
         self.jobDB = JobDB()
         self.logDB = JobLoggingDB()
 
@@ -57,19 +60,21 @@ def initialize(self):
         if not self.am_getOption("Enable", True):
             self.log.info("Stalled Job Agent running in disabled mode")
 
-        wms_instance = getSystemInstance("WorkloadManagement")
-        if not wms_instance:
+        wmsInstance = getSystemInstance("WorkloadManagement")
+        if not wmsInstance:
             return S_ERROR("Can not get the WorkloadManagement system instance")
-        self.stalledJobsTolerantSites = self.am_getOption("StalledJobsTolerantSites", [])
-        self.stalledJobsToleranceTime = self.am_getOption("StalledJobsToleranceTime", 0)
+        self.stalledJobsTolerantSites = self.am_getOption("StalledJobsTolerantSites", self.stalledJobsTolerantSites)
+        self.stalledJobsToleranceTime = self.am_getOption("StalledJobsToleranceTime", self.stalledJobsToleranceTime)
 
-        self.stalledJobsToRescheduleSites = self.am_getOption("StalledJobsToRescheduleSites", [])
+        self.stalledJobsToRescheduleSites = self.am_getOption(
+            "StalledJobsToRescheduleSites", self.stalledJobsToRescheduleSites
+        )
 
         self.submittingTime = self.am_getOption("SubmittingTime", self.submittingTime)
         self.matchedTime = self.am_getOption("MatchedTime", self.matchedTime)
         self.rescheduledTime = self.am_getOption("RescheduledTime", self.rescheduledTime)
 
-        wrapperSection = cfgPath("Systems", "WorkloadManagement", wms_instance, "JobWrapper")
+        wrapperSection = cfgPath("Systems", "WorkloadManagement", wmsInstance, "JobWrapper")
 
         failedTime = self.am_getOption("FailedTimeHours", 6)
         watchdogCycle = gConfig.getValue(cfgPath(wrapperSection, "CheckingTime"), 30 * 60)
@@ -89,14 +94,14 @@ def initialize(self):
 
         # setting up the threading
         maxNumberOfThreads = self.am_getOption("MaxNumberOfThreads", 15)
-        self.log.verbose("Multithreaded with %d threads" % maxNumberOfThreads)
+        self.log.verbose(f"Multithreaded with {maxNumberOfThreads} threads")
         self.threadPoolExecutor = concurrent.futures.ThreadPoolExecutor(max_workers=maxNumberOfThreads)
 
         return S_OK()
 
     #############################################################################
     def execute(self):
-        """The main agent execution method"""
+        """The main agent execution method."""
         # Now we are getting what's going to be checked
         futures = []
 
@@ -162,22 +167,23 @@ def execute(self):
         return S_OK()
 
     def finalize(self):
-        """graceful finalization"""
+        """Graceful finalization."""
 
         self.log.info("Wait for threads to get empty before terminating the agent")
         self.threadPoolExecutor.shutdown()
         self.log.info("Threads are empty, terminating the agent...")
         return S_OK()
 
     def _execute(self, job_Op):
-        """
-        Doing the actual job. This is run inside the threads
+        """Doing the actual job.
+
+        This is run inside the threads
         """
         jobID, jobOp = job_Op.split(":")
         jobID = int(jobID)
         res = getattr(self, f"{jobOp}")(jobID)
         if not res["OK"]:
-            self.log.error(f"Failure executing {jobOp}", "on %d: %s" % (jobID, res["Message"]))
+            self.log.error(f"Failure executing {jobOp}", f"on {jobID}: {res['Message']}")
 
     #############################################################################
     def _markStalledJobs(self, jobID):
@@ -205,8 +211,8 @@ def _markStalledJobs(self, jobID):
 
     #############################################################################
     def _failStalledJobs(self, jobID):
-        """
-        Changes the Stalled status to Failed for jobs long in the Stalled status.
+        """Changes the Stalled status to Failed for jobs long in the Stalled
+        status.
 
         Run inside thread.
         """
@@ -215,7 +221,7 @@ def _failStalledJobs(self, jobID):
         # Check if the job pilot is lost
         result = self._getJobPilotStatus(jobID)
         if not result["OK"]:
-            self.log.error("Failed to get pilot status", "for job %d: %s" % (jobID, result["Message"]))
+            self.log.error("Failed to get pilot status", f"for job {jobID}: {result['Message']}")
             return result
         pilotStatus = result["Value"]
         if pilotStatus != "Running":
@@ -224,7 +230,7 @@ def _failStalledJobs(self, jobID):
             # Verify that there was no sign of life for long enough
             result = self._getLatestUpdateTime(jobID)
             if not result["OK"]:
-                self.log.error("Failed to get job update time", "for job %d: %s" % (jobID, result["Message"]))
+                self.log.error("Failed to get job update time", f"for job {jobID}: {result['Message']}")
                 return result
             elapsedTime = toEpoch() - result["Value"]
             if elapsedTime > self.failedTime:
@@ -233,7 +239,9 @@ def _failStalledJobs(self, jobID):
         # Set the jobs Failed, send them a kill signal in case they are not really dead
         # and send accounting info
         if setFailed:
-            self._sendKillCommand(jobID)  # always returns None
+            res = self._sendKillCommand(jobID)
+            if not res["OK"]:
+                self.log.error("Failed to kill job", jobID)
 
             # For some sites we might want to reschedule rather than fail the jobs
             if self.stalledJobsToRescheduleSites:
@@ -249,7 +257,7 @@ def _failStalledJobs(self, jobID):
         return S_OK()
 
     def _getJobPilotStatus(self, jobID):
-        """Get the job pilot status"""
+        """Get the job pilot status."""
         result = JobMonitoringClient().getJobParameter(jobID, "Pilot_Reference")
         if not result["OK"]:
             return result
@@ -261,9 +269,9 @@ def _getJobPilotStatus(self, jobID):
         result = PilotManagerClient().getPilotInfo(pilotReference)
         if not result["OK"]:
             if DErrno.cmpError(result, DErrno.EWMSNOPILOT):
-                self.log.warn("No pilot found", "for job %d: %s" % (jobID, result["Message"]))
+                self.log.warn("No pilot found", f"for job {jobID}: {result['Message']}")
                 return S_OK("NoPilot")
-            self.log.error("Failed to get pilot information", "for job %d: %s" % (jobID, result["Message"]))
+            self.log.error("Failed to get pilot information", f"for job {jobID}: {result['Message']}")
             return result
         pilotStatus = result["Value"][pilotReference]["Status"]
 
@@ -272,8 +280,7 @@ def _getJobPilotStatus(self, jobID):
     #############################################################################
     def _checkJobStalled(self, job, stalledTime):
         """Compares the most recent of LastUpdateTime and HeartBeatTime against
-        the stalledTime limit.
-        """
+        the stalledTime limit."""
         result = self._getLatestUpdateTime(job)
         if not result["OK"]:
             return result
@@ -290,7 +297,7 @@ def _checkJobStalled(self, job, stalledTime):
 
     #############################################################################
     def _getLatestUpdateTime(self, job):
-        """Returns the most recent of HeartBeatTime and LastUpdateTime"""
+        """Returns the most recent of HeartBeatTime and LastUpdateTime."""
         result = self.jobDB.getJobAttributes(job, ["HeartBeatTime", "LastUpdateTime"])
         if not result["OK"] or not result["Value"]:
             self.log.error(
@@ -318,7 +325,7 @@ def _getLatestUpdateTime(self, job):
 
     #############################################################################
     def _updateJobStatus(self, job, status, minorStatus=None, force=False):
-        """This method updates the job status in the JobDB"""
+        """This method updates the job status in the JobDB."""
 
         if not self.am_getOption("Enable", True):
             return S_OK("Disabled")
@@ -328,23 +335,21 @@ def _updateJobStatus(self, job, status, minorStatus=None, force=False):
         self.log.debug(f"self.jobDB.setJobAttribute({job},'Status','{status}',update=True)")
         result = self.jobDB.setJobAttribute(job, "Status", status, update=True, force=force)
         if not result["OK"]:
-            self.log.error("Failed setting Status", "%s for job %d: %s" % (status, job, result["Message"]))
+            self.log.error("Failed setting Status", f"{status} for job {job}: {result['Message']}")
             toRet = result
         if minorStatus:
             self.log.debug(f"self.jobDB.setJobAttribute({job},'MinorStatus','{minorStatus}',update=True)")
             result = self.jobDB.setJobAttribute(job, "MinorStatus", minorStatus, update=True)
             if not result["OK"]:
-                self.log.error(
-                    "Failed setting MinorStatus", "%s for job %d: %s" % (minorStatus, job, result["Message"])
-                )
+                self.log.error("Failed setting MinorStatus", f"{minorStatus} for job {job}: {result['Message']}")
                 toRet = result
 
         if not minorStatus:  # Retain last minor status for stalled jobs
             result = self.jobDB.getJobAttributes(job, ["MinorStatus"])
             if result["OK"]:
                 minorStatus = result["Value"]["MinorStatus"]
             else:
-                self.log.error("Failed getting MinorStatus", "for job %d: %s" % (job, result["Message"]))
+                self.log.error("Failed getting MinorStatus", f"for job {job}: {result['Message']}")
                 minorStatus = "idem"
                 toRet = result
 
@@ -356,7 +361,8 @@ def _updateJobStatus(self, job, status, minorStatus=None, force=False):
         return toRet
 
     def _getProcessingType(self, jobID):
-        """Get the Processing Type from the JDL, until it is promoted to a real Attribute"""
+        """Get the Processing Type from the JDL, until it is promoted to a real
+        Attribute."""
         processingType = "unknown"
         result = self.jobDB.getJobJDL(jobID, original=True)
         if not result["OK"]:
@@ -367,8 +373,7 @@ def _getProcessingType(self, jobID):
         return processingType
 
     def _sendAccounting(self, jobID):
-        """
-        Send WMS accounting data for the given job.
+        """Send WMS accounting data for the given job.
 
         Run inside thread.
         """
@@ -448,11 +453,11 @@ def _sendAccounting(self, jobID):
         if result["OK"]:
             self.jobDB.setJobAttribute(jobID, "AccountedFlag", "True")
         else:
-            self.log.error("Failed to send accounting report", "Job: %d, Error: %s" % (int(jobID), result["Message"]))
+            self.log.error("Failed to send accounting report", f"for job {jobID}: {result['Message']}")
         return result
 
     def _checkHeartBeat(self, jobID, jobDict):
-        """Get info from HeartBeat"""
+        """Get info from HeartBeat."""
         result = self.jobDB.getHeartBeatData(jobID)
         lastCPUTime = 0
         lastWallTime = 0
@@ -482,7 +487,7 @@ def _checkHeartBeat(self, jobID, jobDict):
         return lastCPUTime, lastWallTime, lastHeartBeatTime
 
     def _checkLoggingInfo(self, jobID, jobDict):
-        """Get info from JobLogging"""
+        """Get info from JobLogging."""
         logList = []
         result = self.logDB.getJobLoggingInfo(jobID)
         if result["OK"]:
@@ -516,7 +521,8 @@ def _checkLoggingInfo(self, jobID, jobDict):
         return startTime, endTime
 
     def _kickStuckJobs(self):
-        """Reschedule jobs stuck in initialization status Rescheduled, Matched"""
+        """Reschedule jobs stuck in initialization status Rescheduled,
+        Matched."""
 
         message = ""
 
@@ -563,6 +569,7 @@ def _kickStuckJobs(self):
 
     def _failSubmittingJobs(self):
         """Failed Jobs stuck in Submitting Status for a long time.
+
         They are due to a failed bulk submission transaction.
         """
 
@@ -587,11 +594,17 @@ def _sendKillCommand(self, job):
         :param int job: ID of job to send kill command
         """
 
-        owner = self.jobDB.getJobAttribute(job, "Owner")["Value"]
-        ownerGroup = self.jobDB.getJobAttribute(job, "OwnerGroup")["Value"]
+        res = self.jobDB.getJobAttribute(job, "Owner")
+        if not res["OK"]:
+            return res
+        owner = res["Value"]
+
+        res = self.jobDB.getJobAttribute(job, "OwnerGroup")
+        if not res["OK"]:
+            return res
+        ownerGroup = res["Value"]
+
         wmsClient = WMSClient(
             useCertificates=True, delegatedDN=getDNForUsername(owner)["Value"][0], delegatedGroup=ownerGroup
         )
-        resKill = wmsClient.killJob(job)
-        if not resKill["OK"]:
-            self.log.error("Failed to send kill command to job", f"{job}: {resKill['Message']}")
+        return wmsClient.killJob(job)

src/DIRAC/WorkloadManagementSystem/Service/SandboxStoreHandler.py

@@ -7,24 +7,23 @@
   :dedent: 2
   :caption: SandboxStore options
 """
+import hashlib
 import os
-import time
-import threading
 import tempfile
-import hashlib
+import threading
+import time
 
-from DIRAC import gLogger, S_OK, S_ERROR
-from DIRAC.ConfigurationSystem.Client.Helpers.Registry import getDNForUsername
+from DIRAC import S_ERROR, S_OK, gLogger
 from DIRAC.Core.DISET.RequestHandler import RequestHandler
 from DIRAC.Core.Security import Locations, Properties, X509Certificate
 from DIRAC.Core.Utilities.File import mkDir
 from DIRAC.Core.Utilities.ObjectLoader import ObjectLoader
 from DIRAC.DataManagementSystem.Client.DataManager import DataManager
 from DIRAC.DataManagementSystem.Service.StorageElementHandler import getDiskSpace
+from DIRAC.RequestManagementSystem.Client.File import File
+from DIRAC.RequestManagementSystem.Client.Operation import Operation
 from DIRAC.RequestManagementSystem.Client.ReqClient import ReqClient
 from DIRAC.RequestManagementSystem.Client.Request import Request
-from DIRAC.RequestManagementSystem.Client.Operation import Operation
-from DIRAC.RequestManagementSystem.Client.File import File
 from DIRAC.Resources.Storage.StorageElement import StorageElement
 from DIRAC.Core.Utilities.File import getGlobbedTotalSize
 

tests/Integration/WorkloadManagementSystem/Test_Client_WMS.py

@@ -182,7 +182,6 @@ def test_submitJob(jobType, inputData, expectedSite):
     #     # Check that the JDL contains some fields
     #     assert jobDescription.lookupAttribute("Owner") is True
     #     assert jobDescription.lookupAttribute("OwnerGroup") is True
-    #     assert jobDescription.lookupAttribute("OwnerDN") is True
     #     assert jobDescription.lookupAttribute("CPUTime") is True
     #     assert jobDescription.lookupAttribute("Priority") is True
     #     assert jobDescription.lookupAttribute("JobID") is True
@@ -193,7 +192,6 @@ def test_submitJob(jobType, inputData, expectedSite):
 
     #     resourceDescription = {
     #         "OwnerGroup": jobDescription.getAttributeString("OwnerGroup"),
-    #         "OwnerDN": jobDescription.getAttributeString("OwnerDN"),
     #         "VirtualOrganization": jobDescription.getAttributeString("VirtualOrganization"),
     #         "CPUTime": jobDescription.getAttributeInt("CPUTime"),
     #         "DIRACVersion": "pippo",
@@ -324,7 +322,6 @@ def test_WMSClient_rescheduleJob():
 
         # resourceDescription = {
         #     "OwnerGroup": jobDescription.getAttributeString("OwnerGroup"),
-        #     "OwnerDN": jobDescription.getAttributeString("OwnerDN"),
         #     "VirtualOrganization": jobDescription.getAttributeString("VirtualOrganization"),
         #     "CPUTime": jobDescription.getAttributeInt("CPUTime"),
         #     "DIRACVersion": "pippo",