Pilot manager handler simplifications

src/DIRAC/ResourceStatusSystem/Command/PilotCommand.py

@@ -1,14 +1,15 @@
-""" PilotCommand
+"""PilotCommand
 
-  The PilotCommand class is a command class to know about present pilots
-  efficiency.
+The PilotCommand class is a command class to know about present pilots
+efficiency.
 
 """
+
 from DIRAC import S_ERROR, S_OK
 from DIRAC.ConfigurationSystem.Client.Helpers.Resources import getCESiteMapping, getSites
 from DIRAC.ResourceStatusSystem.Client.ResourceManagementClient import ResourceManagementClient
 from DIRAC.ResourceStatusSystem.Command.Command import Command
-from DIRAC.WorkloadManagementSystem.Client.PilotManagerClient import PilotManagerClient
+from DIRAC.WorkloadManagementSystem.DB.PilotAgentsDB import PilotAgentsDB
 
 
 class PilotCommand(Command):
@@ -19,11 +20,7 @@ class PilotCommand(Command):
     def __init__(self, args=None, clients=None):
         super().__init__(args, clients)
 
-        if "Pilots" in self.apis:
-            self.pilots = self.apis["Pilots"]
-        else:
-            self.pilots = PilotManagerClient()
-
+        self.pilots = PilotAgentsDB()
         if "ResourceManagementClient" in self.apis:
             self.rmClient = self.apis["ResourceManagementClient"]
         else:

src/DIRAC/WorkloadManagementSystem/Agent/PilotStatusAgent.py

@@ -1,4 +1,4 @@
-"""  The Pilot Status Agent updates the status of the pilot jobs in the
+"""The Pilot Status Agent updates the status of the pilot jobs in the
      PilotAgents database.
 
 .. literalinclude:: ../ConfigTemplate.cfg
@@ -7,6 +7,7 @@
   :dedent: 2
   :caption: PilotStatusAgent options
 """
+
 import datetime
 
 from DIRAC import S_OK
@@ -16,7 +17,6 @@
 from DIRAC.Core.Base.AgentModule import AgentModule
 from DIRAC.Core.Utilities import TimeUtilities
 from DIRAC.WorkloadManagementSystem.Client import PilotStatus
-from DIRAC.WorkloadManagementSystem.Client.PilotManagerClient import PilotManagerClient
 from DIRAC.WorkloadManagementSystem.DB.JobDB import JobDB
 from DIRAC.WorkloadManagementSystem.DB.PilotAgentsDB import PilotAgentsDB
 from DIRAC.WorkloadManagementSystem.Service.WMSUtilities import killPilotsInQueues
@@ -52,7 +52,6 @@ def initialize(self):
         self.jobDB = JobDB()
         self.clearPilotsDelay = self.am_getOption("ClearPilotsDelay", 30)
         self.clearAbortedDelay = self.am_getOption("ClearAbortedPilotsDelay", 7)
-        self.pilots = PilotManagerClient()
 
         return S_OK()
 
@@ -70,7 +69,7 @@ def execute(self):
         # Now handle pilots not updated in the last N days and declare them Deleted.
         result = self.handleOldPilots(connection)
 
-        result = self.pilots.clearPilots(self.clearPilotsDelay, self.clearAbortedDelay)
+        result = self.pilotDB.clearPilots(self.clearPilotsDelay, self.clearAbortedDelay)
         if not result["OK"]:
             self.log.warn("Failed to clear old pilots in the PilotAgentsDB")
 

src/DIRAC/WorkloadManagementSystem/DB/PilotAgentsDB.py

@@ -1,22 +1,8 @@
-""" PilotAgentsDB class is a front-end to the Pilot Agent Database.
-    This database keeps track of all the submitted grid pilot jobs.
-    It also registers the mapping of the DIRAC jobs to the pilot
-    agents.
-
-    Available methods are:
-
-    addPilotReferences()
-    setPilotStatus()
-    deletePilot()
-    clearPilots()
-    setPilotDestinationSite()
-    storePilotOutput()
-    getPilotOutput()
-    setJobForPilot()
-    getPilotsSummary()
-    getGroupedPilotSummary()
-
+"""PilotAgentsDB class is a front-end to the Pilot Agent Database.
+This database keeps track of all the submitted grid pilot jobs.
+It also registers the mapping of the DIRAC jobs to the pilot agents.
 """
+
 import datetime
 import decimal
 import threading
@@ -152,25 +138,6 @@ def setPilotStatus(
         args.append(pilotRef)
         return self._update(req, args=args, conn=conn)
 
-    # ###########################################################################################
-    # FIXME: this can't work ATM because of how the DB table is made. Maybe it would be useful later.
-    #   def setPilotStatusBulk(self, pilotRefsStatusDict=None, statusReason=None,
-    #                          conn=False):
-    #     """ Set pilot job status in a bulk
-    #     """
-    #     if not pilotRefsStatusDict:
-    #       return S_OK()
-
-    #     # Building the request with "ON DUPLICATE KEY UPDATE"
-    #     reqBase = "INSERT INTO PilotAgents (PilotJobReference, Status, StatusReason) VALUES "
-
-    #     for pilotJobReference, status in pilotRefsStatusDict.items():
-    #       req = reqBase + ','.join("('%s', '%s', '%s')" % (pilotJobReference, status, statusReason))
-    #       req += " ON DUPLICATE KEY UPDATE Status=VALUES(Status),StatusReason=VALUES(StatusReason)"
-
-    #     return self._update(req, conn=conn)
-
-    ##########################################################################################
     def selectPilots(
         self, condDict, older=None, newer=None, timeStamp="SubmissionTime", orderAttribute=None, limit=None
     ):
@@ -364,34 +331,6 @@ def getPilotInfo(self, pilotRef=False, conn=False, paramNames=[], pilotID=False)
 
         return S_OK(resDict)
 
-    ##########################################################################################
-    def setPilotDestinationSite(self, pilotRef, destination, conn=False):
-        """Set the pilot agent destination site"""
-
-        gridSite = "Unknown"
-        res = getCESiteMapping(destination)
-        if res["OK"] and res["Value"]:
-            gridSite = res["Value"][destination]
-
-        req = "UPDATE PilotAgents SET DestinationSite=%s, GridSite=%s WHERE PilotJobReference=%s"
-        args = (
-            destination,
-            gridSite,
-            pilotRef,
-        )
-        return self._update(req, args=args, conn=conn)
-
-    ##########################################################################################
-    def setPilotBenchmark(self, pilotRef, mark):
-        """Set the pilot agent benchmark"""
-
-        req = "UPDATE PilotAgents SET BenchMark=%s WHERE PilotJobReference=%s"
-        args = (
-            f"{mark:f}",
-            pilotRef,
-        )
-        return self._update(req, args=args)
-
     ##########################################################################################
     def setAccountingFlag(self, pilotRef, mark="True"):
         """Set the pilot AccountingSent flag"""
@@ -540,20 +479,6 @@ def getPilotsForJobID(self, jobID):
         return S_OK([])
 
     ##########################################################################################
-    def getPilotCurrentJob(self, pilotRef):
-        """The job ID currently executed by the pilot"""
-        req = "SELECT CurrentJobID FROM PilotAgents WHERE PilotJobReference=%s"
-        result = self._query(req, args=(pilotRef,))
-        if not result["OK"]:
-            return result
-        if result["Value"]:
-            jobID = int(result["Value"][0][0])
-            return S_OK(jobID)
-        self.log.warn(f"Current job ID for pilot {pilotRef} is not known: pilot did not match jobs yet?")
-        return S_OK()
-
-    ##########################################################################################
-    # FIXME: investigate it getPilotSummaryShort can replace this method
     def getPilotSummary(self, startdate="", enddate=""):
         """Get summary of the pilot jobs status by site"""
         summary_dict = {}

src/DIRAC/WorkloadManagementSystem/Service/PilotManagerHandler.py

@@ -27,38 +27,6 @@ def initializeHandler(cls, serviceInfoDict):
 
         return S_OK()
 
-    ##############################################################################
-    types_getCurrentPilotCounters = [dict]
-
-    @classmethod
-    def export_getCurrentPilotCounters(cls, attrDict={}):
-        """Get pilot counters per Status with attrDict selection. Final statuses are given for
-        the last day.
-        """
-
-        result = cls.pilotAgentsDB.getCounters("PilotAgents", ["Status"], attrDict, timeStamp="LastUpdateTime")
-        if not result["OK"]:
-            return result
-        last_update = datetime.datetime.utcnow() - TimeUtilities.day
-        resultDay = cls.pilotAgentsDB.getCounters(
-            "PilotAgents", ["Status"], attrDict, newer=last_update, timeStamp="LastUpdateTime"
-        )
-        if not resultDay["OK"]:
-            return resultDay
-
-        resultDict = {}
-        for statusDict, count in result["Value"]:
-            status = statusDict["Status"]
-            resultDict[status] = count
-            if status in PilotStatus.PILOT_FINAL_STATES:
-                resultDict[status] = 0
-                for statusDayDict, ccount in resultDay["Value"]:
-                    if status == statusDayDict["Status"]:
-                        resultDict[status] = ccount
-                    break
-
-        return S_OK(resultDict)
-
     ##########################################################################################
     types_addPilotReferences = [list, str]
 
@@ -101,20 +69,6 @@ def export_getPilotSummary(cls, startdate="", enddate=""):
 
         return cls.pilotAgentsDB.getPilotSummary(startdate, enddate)
 
-    ##############################################################################
-    types_getGroupedPilotSummary = [list]
-
-    @classmethod
-    def export_getGroupedPilotSummary(cls, columnList):
-        """
-        Get pilot summary showing grouped by columns in columnList, all pilot states
-        and pilot efficiencies in a single row.
-
-        :param columnList: a list of columns to GROUP BY (less status column)
-        :return: a dictionary containing column names and data records
-        """
-        return cls.pilotAgentsDB.getGroupedPilotSummary(columnList)
-
     ##############################################################################
     types_getPilots = [[str, int]]
 
@@ -128,40 +82,6 @@ def export_getPilots(cls, jobID):
         return cls.pilotAgentsDB.getPilotInfo(pilotID=result["Value"])
 
     ##############################################################################
-    types_setJobForPilot = [[str, int], str]
-
-    @classmethod
-    def export_setJobForPilot(cls, jobID, pilotRef, destination=None):
-        """Report the DIRAC job ID which is executed by the given pilot job"""
-
-        result = cls.pilotAgentsDB.setJobForPilot(int(jobID), pilotRef)
-        if not result["OK"]:
-            return result
-        result = cls.pilotAgentsDB.setCurrentJobID(pilotRef, int(jobID))
-        if not result["OK"]:
-            return result
-        if destination:
-            result = cls.pilotAgentsDB.setPilotDestinationSite(pilotRef, destination)
-
-        return result
-
-    ##########################################################################################
-    types_setPilotBenchmark = [str, float]
-
-    @classmethod
-    def export_setPilotBenchmark(cls, pilotRef, mark):
-        """Set the pilot agent benchmark"""
-        return cls.pilotAgentsDB.setPilotBenchmark(pilotRef, mark)
-
-    ##########################################################################################
-    types_setAccountingFlag = [str]
-
-    @classmethod
-    def export_setAccountingFlag(cls, pilotRef, mark="True"):
-        """Set the pilot AccountingSent flag"""
-        return cls.pilotAgentsDB.setAccountingFlag(pilotRef, mark)
-
-    ##########################################################################################
     types_setPilotStatus = [str, str]
 
     @classmethod
@@ -171,38 +91,3 @@ def export_setPilotStatus(cls, pilotRef, status, destination=None, reason=None,
         return cls.pilotAgentsDB.setPilotStatus(
             pilotRef, status, destination=destination, statusReason=reason, gridSite=gridSite, queue=queue
         )
-
-    ##########################################################################################
-    types_countPilots = [dict]
-
-    @classmethod
-    def export_countPilots(cls, condDict, older=None, newer=None, timeStamp="SubmissionTime"):
-        """Set the pilot agent status"""
-
-        return cls.pilotAgentsDB.countPilots(condDict, older, newer, timeStamp)
-
-    ##########################################################################################
-    types_deletePilots = [[list, str, int]]
-
-    @classmethod
-    def export_deletePilots(cls, pilotIDs):
-        if isinstance(pilotIDs, str):
-            return cls.pilotAgentsDB.deletePilot(pilotIDs)
-
-        if isinstance(pilotIDs, int):
-            pilotIDs = [
-                pilotIDs,
-            ]
-
-        result = cls.pilotAgentsDB.deletePilots(pilotIDs)
-        if not result["OK"]:
-            return result
-
-        return S_OK()
-
-    ##############################################################################
-    types_clearPilots = [int, int]
-
-    @classmethod
-    def export_clearPilots(cls, interval=30, aborted_interval=7):
-        return cls.pilotAgentsDB.clearPilots(interval, aborted_interval)

tests/CI/install_client.sh

@@ -58,35 +58,33 @@ echo -e "*** $(date -u) **** Client INSTALLATION START ****\n"
 
 installDIRAC
 
-if [[ -z "${INSTALLATION_BRANCH}" ]]; then
-    echo 'Generate a pilot proxy, to be used by the pilot'
-    dirac-proxy-init -g pilot -C /ca/certs/pilot.pem -K /ca/certs/pilot.key "${DEBUG}"
-    mv /tmp/x509up_u$UID /ca/certs/pilot_proxy
+echo 'Generate a pilot proxy, to be used by the pilot'
+dirac-proxy-init -g pilot -C /ca/certs/pilot.pem -K /ca/certs/pilot.key "${DEBUG}"
+mv /tmp/x509up_u$UID /ca/certs/pilot_proxy
 
-    echo -e "*** $(date -u)  Getting a non privileged user\n" |& tee -a clientTestOutputs.txt
-    dirac-proxy-init "${DEBUG}" |& tee -a clientTestOutputs.txt
+echo -e "*** $(date -u)  Getting a non privileged user\n" |& tee -a clientTestOutputs.txt
+dirac-proxy-init "${DEBUG}" |& tee -a clientTestOutputs.txt
 
-    #-------------------------------------------------------------------------------#
-    echo -e "*** $(date -u) **** Submit a job ****\n"
+#-------------------------------------------------------------------------------#
+echo -e "*** $(date -u) **** Submit a job ****\n"
 
-    echo -e '[\n    Arguments = "Hello World";\n    Executable = "echo";\n    Site = "DIRAC.Jenkins.ch";' > test.jdl
-    echo "    JobName = \"${GITHUB_JOB}_$(date +"%Y-%m-%d_%T" | sed 's/://g')\"" >> test.jdl
-    echo "]" >> test.jdl
-    dirac-wms-job-submit test.jdl "${DEBUG}" |& tee -a clientTestOutputs.txt
+echo -e '[\n    Arguments = "Hello World";\n    Executable = "echo";\n    Site = "DIRAC.Jenkins.ch";' > test.jdl
+echo "    JobName = \"${GITHUB_JOB}_$(date +"%Y-%m-%d_%T" | sed 's/://g')\"" >> test.jdl
+echo "]" >> test.jdl
+dirac-wms-job-submit test.jdl "${DEBUG}" |& tee -a clientTestOutputs.txt
 
-    #-------------------------------------------------------------------------------#
-    echo -e "*** $(date -u) **** add a file ****\n"
+#-------------------------------------------------------------------------------#
+echo -e "*** $(date -u) **** add a file ****\n"
 
-    echo "${CLIENT_UPLOAD_BASE64}" > b64_lfn
-    base64 b64_lfn --decode > "${CLIENT_UPLOAD_FILE}"
-    dirac-dms-add-file "${CLIENT_UPLOAD_LFN}" "${CLIENT_UPLOAD_FILE}" S3-DIRECT
-    echo $?
+echo "${CLIENT_UPLOAD_BASE64}" > b64_lfn
+base64 b64_lfn --decode > "${CLIENT_UPLOAD_FILE}"
+dirac-dms-add-file "${CLIENT_UPLOAD_LFN}" "${CLIENT_UPLOAD_FILE}" S3-DIRECT
+echo $?
 
-    #-------------------------------------------------------------------------------#
-    echo -e "*** $(date -u) **** Submit a job with an input ****\n"
+#-------------------------------------------------------------------------------#
+echo -e "*** $(date -u) **** Submit a job with an input ****\n"
 
-    echo -e '[\n    Arguments = "Hello World";\n    Executable = "echo";\n    Site = "DIRAC.Jenkins.ch";\n    InputData = "/vo/test_lfn.txt";' > test_dl.jdl
-    echo "    JobName = \"${GITHUB_JOB}_$(date +"%Y-%m-%d_%T" | sed 's/://g')\"" >> test_dl.jdl
-    echo "]" >> test_dl.jdl
-    dirac-wms-job-submit test_dl.jdl "${DEBUG}" |& tee -a clientTestOutputs.txt
-fi
+echo -e '[\n    Arguments = "Hello World";\n    Executable = "echo";\n    Site = "DIRAC.Jenkins.ch";\n    InputData = "/vo/test_lfn.txt";' > test_dl.jdl
+echo "    JobName = \"${GITHUB_JOB}_$(date +"%Y-%m-%d_%T" | sed 's/://g')\"" >> test_dl.jdl
+echo "]" >> test_dl.jdl
+dirac-wms-job-submit test_dl.jdl "${DEBUG}" |& tee -a clientTestOutputs.txt

tests/CI/run_pilot.sh

@@ -11,12 +11,6 @@ set -x
 echo "Starting run_pilot.sh"
 source CONFIG
 
-if [[ -n "${INSTALLATION_BRANCH}" ]]; then
-  # Do not run this
-  echo "Not running the DIRAC Pilot"
-  exit
-fi
-
 # Creating "the worker node"
 mkdir -p /home/dirac/etc/grid-security/certificates
 mkdir -p /home/dirac/etc/grid-security/vomsdir