From 16d924edb11fb7abf0acf95deda15c7f25c5b002 Mon Sep 17 00:00:00 2001 From: Federico Stagni Date: Mon, 6 Jul 2026 17:55:07 +0200 Subject: [PATCH] feat: removed JobDB JobParameters table --- .../Systems/WorkloadManagement/index.rst | 4 +- .../Agent/JobCleaningAgent.py | 6 +- .../Agent/test/Test_Agent_JobCleaningAgent.py | 6 +- .../WorkloadManagementSystem/DB/JobDB.py | 61 ------------------- .../WorkloadManagementSystem/DB/JobDB.sql | 10 --- .../Utilities/JobParameters.py | 46 ++++++++++++++ 6 files changed, 56 insertions(+), 77 deletions(-) diff --git a/docs/source/DeveloperGuide/Systems/WorkloadManagement/index.rst b/docs/source/DeveloperGuide/Systems/WorkloadManagement/index.rst index 7ec3701e700..58778dc911a 100644 --- a/docs/source/DeveloperGuide/Systems/WorkloadManagement/index.rst +++ b/docs/source/DeveloperGuide/Systems/WorkloadManagement/index.rst @@ -100,5 +100,5 @@ It is based on layered architecture and is based on DIRAC architecture: SandboxMetadataDB class is a front-end to the metadata for sandboxes. * JobParametersDB - JobParametersDB class is a front-end to the Elastic/OpenSearch based index providing Job Parameters. - It is used in most of the WMS components and is based on Elastic/OpenSearch. + JobParametersDB class is a front-end to the OpenSearch based index providing Job Parameters. + It is used in most of the WMS components and is based on OpenSearch. diff --git a/src/DIRAC/WorkloadManagementSystem/Agent/JobCleaningAgent.py b/src/DIRAC/WorkloadManagementSystem/Agent/JobCleaningAgent.py index 346c4e5458e..cdd56fe3db6 100644 --- a/src/DIRAC/WorkloadManagementSystem/Agent/JobCleaningAgent.py +++ b/src/DIRAC/WorkloadManagementSystem/Agent/JobCleaningAgent.py @@ -38,7 +38,7 @@ from DIRAC.RequestManagementSystem.Client.Request import Request from DIRAC.WorkloadManagementSystem.Client import JobStatus from DIRAC.WorkloadManagementSystem.Client.WMSClient import WMSClient -from DIRAC.WorkloadManagementSystem.DB.JobParametersDB import getJobParameters +from DIRAC.WorkloadManagementSystem.DB.JobParametersDB import JobParametersDB from DIRAC.WorkloadManagementSystem.DB.StatusUtils import kill_delete_jobs from DIRAC.WorkloadManagementSystem.Service.JobPolicy import RIGHT_DELETE @@ -333,8 +333,8 @@ def deleteJobOversizedSandbox(self, jobIDList): failed = {} successful = {} - jobIDs = [int(jobID) for jobID in jobIDList] - result = getJobParameters(jobIDs, "OutputSandboxLFN") + jobIDList = [int(jobID) for jobID in jobIDList] + result = JobParametersDB().getJobParameters(jobIDList, ["OutputSandboxLFN"]) if not result["OK"]: return result osLFNDict = result["Value"] diff --git a/src/DIRAC/WorkloadManagementSystem/Agent/test/Test_Agent_JobCleaningAgent.py b/src/DIRAC/WorkloadManagementSystem/Agent/test/Test_Agent_JobCleaningAgent.py index ef53fd61e93..616c68c6443 100644 --- a/src/DIRAC/WorkloadManagementSystem/Agent/test/Test_Agent_JobCleaningAgent.py +++ b/src/DIRAC/WorkloadManagementSystem/Agent/test/Test_Agent_JobCleaningAgent.py @@ -136,11 +136,15 @@ def test_deleteJobOversizedSandbox(mocker, inputs, params, expected): mocker.patch("DIRAC.WorkloadManagementSystem.Agent.JobCleaningAgent.AgentModule.__init__") mocker.patch("DIRAC.WorkloadManagementSystem.Agent.JobCleaningAgent.AgentModule.am_getOption", return_value=mockAM) + mocker.patch("DIRAC.WorkloadManagementSystem.Agent.JobCleaningAgent.JobDB.getJobAttributes", return_value=S_OK("")) + mocker.patch("DIRAC.WorkloadManagementSystem.Agent.JobCleaningAgent.JobDB", return_value=mockNone) + mocker.patch("DIRAC.WorkloadManagementSystem.Agent.JobCleaningAgent.SandboxMetadataDB", return_value=mockNone) mocker.patch("DIRAC.WorkloadManagementSystem.Agent.JobCleaningAgent.ReqClient", return_value=mockNone) mocker.patch( "DIRAC.WorkloadManagementSystem.Agent.JobCleaningAgent.getDNForUsername", return_value=S_OK(["/bih/boh/DN"]) ) - mocker.patch("DIRAC.WorkloadManagementSystem.Agent.JobCleaningAgent.getJobParameters", return_value=params) + mockJobParamsDB = mocker.patch("DIRAC.WorkloadManagementSystem.Agent.JobCleaningAgent.JobParametersDB") + mockJobParamsDB.return_value.getJobParameters.return_value = params def mock_load_object(module_path, class_name): mocks = { diff --git a/src/DIRAC/WorkloadManagementSystem/DB/JobDB.py b/src/DIRAC/WorkloadManagementSystem/DB/JobDB.py index ee5cd96fd09..3119e6a0766 100755 --- a/src/DIRAC/WorkloadManagementSystem/DB/JobDB.py +++ b/src/DIRAC/WorkloadManagementSystem/DB/JobDB.py @@ -111,53 +111,7 @@ def getDistinctJobAttributes(self, attribute, condDict=None, older=None, newer=N ) ############################################################################# - def getJobParameters(self, jobID, paramList=None): - """Get Job Parameters defined for jobID. - Returns a dictionary with the Job Parameters. - If parameterList is empty - all the parameters are returned. - """ - jobIDList = [jobID] if isinstance(jobID, (str, int)) else jobID - - resultDict = {} - if paramList: - if isinstance(paramList, str): - paramList = paramList.split(",") - cmd = "SELECT JobID, Name, Value FROM JobParameters WHERE JobID IN (" - cmd += ",".join(["%s"] * len(jobIDList)) - args = jobIDList - cmd += ") AND Name IN (" - cmd += ",".join(["%s"] * len(paramList)) - cmd += ")" - args.extend(paramList) - result = self._query(cmd, args=args) - if result["OK"]: - if result["Value"]: - for res_jobID, res_name, res_value in result["Value"]: - try: - res_value = res_value.decode(errors="replace") # account for use of BLOBs - except AttributeError: - pass - resultDict.setdefault(int(res_jobID), {})[res_name] = res_value - - return S_OK(resultDict) # there's a slim chance that this is an empty dictionary - else: - return S_ERROR("JobDB.getJobParameters: failed to retrieve parameters") - - else: - result = self.getFields("JobParameters", ["JobID", "Name", "Value"], {"JobID": jobID}) - if not result["OK"]: - return result - - for res_jobID, res_name, res_value in result["Value"]: - try: - res_value = res_value.decode(errors="replace") # account for use of BLOBs - except AttributeError: - pass - resultDict.setdefault(int(res_jobID), {})[res_name] = res_value - - return S_OK(resultDict) # there's a slim chance that this is an empty dictionary - ############################################################################# def getAtticJobParameters(self, jobID, paramList=None, rescheduleCounter=-1): """Get Attic Job Parameters defined for a job with jobID. Returns a dictionary with the Attic Job Parameters per each rescheduling cycle. @@ -262,16 +216,6 @@ def getJobAttribute(self, jobID, attribute): return result return S_OK(result["Value"].get(attribute)) - ############################################################################# - @deprecated("Use JobParametersDB instead") - def getJobParameter(self, jobID, parameter): - """Get the given parameter of a job specified by its jobID""" - - result = self.getJobParameters(jobID, [parameter]) - if not result["OK"]: - return result - return S_OK(result.get("Value", {}).get(int(jobID), {}).get(parameter)) - ############################################################################# def getJobOptParameter(self, jobID, parameter): """Get optimizer parameters for the given job.""" @@ -907,7 +851,6 @@ def removeJobFromDB(self, jobIDs): for table in [ "InputData", - "JobParameters", "AtticJobParameters", "HeartBeatLoggingInfo", "OptimizerParameters", @@ -980,10 +923,6 @@ def rescheduleJob(self, jobID): if not result["OK"]: break - res = self._update("DELETE FROM JobParameters WHERE JobID=%s", args=(str(jobID),)) - if not res["OK"]: - return res - # Delete optimizer parameters if not self._update("DELETE FROM OptimizerParameters WHERE JobID=%s", args=(str(jobID),))["OK"]: return S_ERROR("JobDB.removeJobOptParameter: operation failed.") diff --git a/src/DIRAC/WorkloadManagementSystem/DB/JobDB.sql b/src/DIRAC/WorkloadManagementSystem/DB/JobDB.sql index 1bc7ec83b58..01c638c7415 100755 --- a/src/DIRAC/WorkloadManagementSystem/DB/JobDB.sql +++ b/src/DIRAC/WorkloadManagementSystem/DB/JobDB.sql @@ -79,16 +79,6 @@ CREATE TABLE `InputData` ( FOREIGN KEY (`JobID`) REFERENCES `Jobs`(`JobID`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; --- ------------------------------------------------------------------------------ -DROP TABLE IF EXISTS `JobParameters`; -CREATE TABLE `JobParameters` ( - `JobID` INT(11) UNSIGNED NOT NULL, - `Name` VARCHAR(100) NOT NULL, - `Value` TEXT NOT NULL, - PRIMARY KEY (`JobID`,`Name`), - FOREIGN KEY (`JobID`) REFERENCES `Jobs`(`JobID`) -) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; - -- ------------------------------------------------------------------------------ DROP TABLE IF EXISTS `OptimizerParameters`; CREATE TABLE `OptimizerParameters` ( diff --git a/src/DIRAC/WorkloadManagementSystem/Utilities/JobParameters.py b/src/DIRAC/WorkloadManagementSystem/Utilities/JobParameters.py index e1d048741e6..3d472733f14 100644 --- a/src/DIRAC/WorkloadManagementSystem/Utilities/JobParameters.py +++ b/src/DIRAC/WorkloadManagementSystem/Utilities/JobParameters.py @@ -141,6 +141,52 @@ def getNumberOfGPUs(siteName=None, gridCE=None, queue=None): return 0 +def getJobParameters(jobIDs: list[int], parName: str | None, vo: str = "") -> dict: + """Utility to get a job parameter for a list of jobIDs pertaining to a VO. + If the jobID is not in the JobParametersDB, it will be looked up in the JobDB. + + Requires direct access to the JobParametersDB and JobDB. + + :param jobIDs: list of jobIDs + :param parName: name of the parameter to be retrieved + :param vo: VO of the jobIDs + :return: dictionary with jobID as key and the parameter as value + :rtype: dict + """ + from DIRAC.WorkloadManagementSystem.DB.JobParametersDB import JobParametersDB + + elasticJobParametersDB = JobParametersDB() + + if vo: # a user is connecting, with a proxy + res = elasticJobParametersDB.getJobParameters(jobIDs, vo, parName) + if not res["OK"]: + return res + parameters = res["Value"] + else: # a service is connecting, no proxy, e.g. StalledJobAgent + from DIRAC.WorkloadManagementSystem.DB.JobDB import JobDB + + q = f"SELECT JobID, VO FROM Jobs WHERE JobID IN ({','.join([str(jobID) for jobID in jobIDs])})" + res = JobDB()._query(q) + if not res["OK"]: + return res + if not res["Value"]: + return S_OK({}) + # get the VO for each jobID + voDict = {} + for jobID, vo in res["Value"]: + if vo not in voDict: + voDict[vo] = [] + voDict[vo].append(jobID) + # get the parameters for each VO + parameters = {} + for vo, jobIDs in voDict.items(): + res = elasticJobParametersDB.getJobParameters(jobIDs, vo, parName) + if not res["OK"]: + return res + parameters.update(res["Value"]) + return S_OK(parameters) + + def getAvailableRAM(siteName=None, gridCE=None, queue=None): """Gets the available RAM on a certain CE/queue/node (what the pilot administers)