From defe4a6340c97d084390e6d167c7fb481b540f91 Mon Sep 17 00:00:00 2001 From: Federico Stagni Date: Wed, 21 Oct 2026 17:03:20 +0200 Subject: [PATCH] feat: removed JobDB JobParameters table --- .../Systems/WorkloadManagement/index.rst | 4 +- .../Agent/JobCleaningAgent.py | 6 +- .../Agent/PushJobAgent.py | 4 +- .../Agent/StalledJobAgent.py | 16 ++-- .../Agent/test/Test_Agent_JobCleaningAgent.py | 84 +++++++++++++------ .../Agent/test/Test_Agent_StalledJobAgent.py | 8 +- .../WorkloadManagementSystem/DB/JobDB.py | 69 +-------------- .../WorkloadManagementSystem/DB/JobDB.sql | 10 --- .../DB/JobParametersDB.py | 35 +++----- .../Service/JobMonitoringHandler.py | 6 +- 10 files changed, 96 insertions(+), 146 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 4380806ffbf..bf7f3142793 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 getJobParametersUtility 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 = getJobParametersUtility(jobIDs=jobIDList, parName="OutputSandboxLFN") if not result["OK"]: return result osLFNDict = result["Value"] diff --git a/src/DIRAC/WorkloadManagementSystem/Agent/PushJobAgent.py b/src/DIRAC/WorkloadManagementSystem/Agent/PushJobAgent.py index 75614eeee8c..7c8321bab57 100644 --- a/src/DIRAC/WorkloadManagementSystem/Agent/PushJobAgent.py +++ b/src/DIRAC/WorkloadManagementSystem/Agent/PushJobAgent.py @@ -34,7 +34,7 @@ from DIRAC.WorkloadManagementSystem.Client import JobMinorStatus, JobStatus, PilotStatus from DIRAC.WorkloadManagementSystem.Client.JobMonitoringClient import JobMonitoringClient from DIRAC.WorkloadManagementSystem.Client.JobReport import JobReport -from DIRAC.WorkloadManagementSystem.DB.JobParametersDB import getJobParameters +from DIRAC.WorkloadManagementSystem.DB.JobParametersDB import getJobParametersUtility from DIRAC.WorkloadManagementSystem.JobWrapper.JobWrapper import JobWrapper from DIRAC.WorkloadManagementSystem.JobWrapper.JobWrapperUtilities import ( getJobWrapper, @@ -801,7 +801,7 @@ def _checkSubmittedJobWrappers(self, ce: ComputingElement, site: str): return S_OK() # Get their parameters - if not (result := getJobParameters(jobs, ["GridCE", "TaskID", "Stamp"]))["OK"]: + if not (result := getJobParametersUtility(jobs, ["GridCE", "TaskID", "Stamp"]))["OK"]: self.log.error("Failed to get the list of taskIDs", result["Message"]) return result diff --git a/src/DIRAC/WorkloadManagementSystem/Agent/StalledJobAgent.py b/src/DIRAC/WorkloadManagementSystem/Agent/StalledJobAgent.py index e2ebcd61d95..5b33f31feaa 100755 --- a/src/DIRAC/WorkloadManagementSystem/Agent/StalledJobAgent.py +++ b/src/DIRAC/WorkloadManagementSystem/Agent/StalledJobAgent.py @@ -21,7 +21,7 @@ from DIRAC.Core.Utilities.ObjectLoader import ObjectLoader from DIRAC.Core.Utilities.TimeUtilities import DiracTime, fromString, second, toEpoch from DIRAC.WorkloadManagementSystem.Client import JobMinorStatus, JobStatus -from DIRAC.WorkloadManagementSystem.DB.JobParametersDB import getJobParameters +from DIRAC.WorkloadManagementSystem.DB.JobParametersDB import getJobParametersUtility from DIRAC.WorkloadManagementSystem.DB.StatusUtils import kill_delete_jobs from DIRAC.WorkloadManagementSystem.Service.JobPolicy import RIGHT_KILL from DIRAC.WorkloadManagementSystem.Utilities.Utils import rescheduleJobs @@ -290,12 +290,11 @@ def _failStalledJobs(self, jobID): def _getJobPilotStatus(self, jobID): """Get the job pilot status.""" - result = getJobParameters([jobID], "Pilot_Reference") + result = getJobParametersUtility([jobID], "Pilot_Reference") if not result["OK"]: return result - # getJobParameters returns {jobID: {parName: value, ...}, ...}, with jobID being either an int (JobDB) - # or a str (JobParametersDB), so look it up both ways - jobParameters = result["Value"].get(jobID) or result["Value"].get(str(jobID)) or {} + # getJobParametersUtility returns {jobID: {parName: value, ...}, ...}, keyed by the int jobID + jobParameters = result["Value"].get(jobID) or {} pilotReference = jobParameters.get("Pilot_Reference") if not pilotReference: # There is no pilot reference, hence its status is unknown @@ -428,10 +427,9 @@ def _sendAccounting(self, jobID): if lastHeartBeatTime is not None and lastHeartBeatTime > endTime: endTime = lastHeartBeatTime - result = getJobParameters([jobID], "CPUNormalizationFactor") - # getJobParameters returns {jobID: {parName: value, ...}, ...}, with jobID being either an int (JobDB) - # or a str (JobParametersDB), so look it up both ways - jobParameters = result.get("Value", {}).get(jobID) or result.get("Value", {}).get(str(jobID)) or {} + result = getJobParametersUtility([jobID], "CPUNormalizationFactor") + # getJobParametersUtility returns {jobID: {parName: value, ...}, ...}, keyed by the int jobID + jobParameters = result.get("Value", {}).get(jobID) or {} if not result["OK"] or not jobParameters.get("CPUNormalizationFactor"): errorMsg = f"- {result['Message']}" if "Message" in result else "empty" self.log.warn( diff --git a/src/DIRAC/WorkloadManagementSystem/Agent/test/Test_Agent_JobCleaningAgent.py b/src/DIRAC/WorkloadManagementSystem/Agent/test/Test_Agent_JobCleaningAgent.py index ef53fd61e93..67c72baa3df 100644 --- a/src/DIRAC/WorkloadManagementSystem/Agent/test/Test_Agent_JobCleaningAgent.py +++ b/src/DIRAC/WorkloadManagementSystem/Agent/test/Test_Agent_JobCleaningAgent.py @@ -11,14 +11,9 @@ gLogger.setLevel("DEBUG") # Mock Objects -mockReply = MagicMock() mockAM = MagicMock() mockNone = MagicMock() mockNone.return_value = None -mockJMC = MagicMock() -mockJobDB = MagicMock() -mockJobDB.getDistinctJobAttributes = mockReply -mockJobDB.selectJobs = mockReply @pytest.fixture @@ -29,11 +24,32 @@ def jca(mocker): side_effect=lambda x, y=None: y, create=True, ) - mocker.patch("DIRAC.WorkloadManagementSystem.Agent.JobCleaningAgent.AgentModule.am_getOption", return_value=mockAM) + + def mock_am_getOption(option, default=None): + defaults = { + "ProductionTypes": [], + "MaxJobsAtOnce": 500, + "RemoveStatusDelay/Done": 7, + "RemoveStatusDelay/Killed": 7, + "RemoveStatusDelay/Failed": 7, + "RemoveStatusDelay/Any": -1, + "RemoveStatusDelayHB/Done": -1, + "RemoveStatusDelayHB/Killed": -1, + "RemoveStatusDelayHB/Failed": -1, + "MaxHBJobsAtOnce": 0, + } + return defaults.get(option, default) + + mocker.patch( + "DIRAC.WorkloadManagementSystem.Agent.JobCleaningAgent.AgentModule.am_getOption", + side_effect=mock_am_getOption, + ) + + mocker.patch("DIRAC.WorkloadManagementSystem.Agent.JobCleaningAgent.Operations") def mock_load_object(module_path, class_name): mocks = { - "JobDB": mockJobDB, + "JobDB": MagicMock(), "TaskQueueDB": MagicMock(), "PilotAgentsDB": MagicMock(), "SandboxMetadataDB": MagicMock(), @@ -55,54 +71,65 @@ def mock_load_object(module_path, class_name): @pytest.mark.parametrize( - "mockReplyInput, expected", + "dbReply, expected", [ - ({"OK": True, "Value": ""}, {"OK": True, "Value": []}), - ({"OK": False, "Message": ""}, {"OK": False, "Message": ""}), + ({"OK": True, "Value": []}, {"OK": True, "Value": []}), + ({"OK": False, "Message": "error"}, {"OK": False, "Message": "error"}), ], ) -def test__getAllowedJobTypes(jca, mockReplyInput, expected): +def test__getAllowedJobTypes(jca, mocker, dbReply, expected): """Testing JobCleaningAgent()._getAllowedJobTypes()""" - mockReply.return_value = mockReplyInput + jca.jobDB.getDistinctJobAttributes.return_value = dbReply result = jca._getAllowedJobTypes() - assert result == expected + assert result["OK"] == expected["OK"] + if result["OK"]: + assert result["Value"] == expected["Value"] + else: + assert result["Message"] == expected["Message"] @pytest.mark.parametrize( "mockReplyInput, expected", [ - ({"OK": True, "Value": ""}, {"OK": True, "Value": None}), - ({"OK": False, "Message": ""}, {"OK": False, "Message": ""}), + ({"OK": True, "Value": []}, {"OK": True, "Value": None}), + ({"OK": False, "Message": "error"}, {"OK": False, "Message": "error"}), ], ) -def test_removeJobsByStatus(jca, mockReplyInput, expected): +def test_removeJobsByStatus(jca, mocker, mockReplyInput, expected): """Testing JobCleaningAgent().removeDeletedJobs()""" - mockReply.return_value = mockReplyInput + jca.jobDB.selectJobs.return_value = mockReplyInput result = jca.removeDeletedJobs() - assert result == expected + assert result["OK"] == expected["OK"] + if result["OK"]: + assert result["Value"] is None + else: + assert result["Message"] == expected["Message"] @pytest.mark.parametrize( "conditions, mockReplyInput, expected", [ - ({"JobType": "", "Status": "Deleted"}, {"OK": True, "Value": ""}, {"OK": True, "Value": None}), - ({"JobType": "", "Status": "Deleted"}, {"OK": False, "Message": ""}, {"OK": False, "Message": ""}), - ({"JobType": [], "Status": "Deleted"}, {"OK": True, "Value": ""}, {"OK": True, "Value": None}), + ({"JobType": [], "Status": "Deleted"}, {"OK": True, "Value": []}, {"OK": True, "Value": None}), + ({"JobType": [], "Status": "Deleted"}, {"OK": False, "Message": "error"}, {"OK": False, "Message": "error"}), ( - {"JobType": ["some", "status"], "Status": ["Deleted", "Cancelled"]}, - {"OK": True, "Value": ""}, + {"JobType": ["User"], "Status": ["Deleted", "Killed"]}, + {"OK": True, "Value": []}, {"OK": True, "Value": None}, ), ], ) -def test_deleteJobsByStatus(jca, conditions, mockReplyInput, expected): +def test_deleteJobsByStatus(jca, mocker, conditions, mockReplyInput, expected): """Testing JobCleaningAgent().deleteJobsByStatus()""" - mockReply.return_value = mockReplyInput + jca.jobDB.selectJobs.return_value = mockReplyInput result = jca.deleteJobsByStatus(conditions) - assert result == expected + assert result["OK"] == expected["OK"] + if result["OK"]: + assert result["Value"] is None + else: + assert result["Message"] == expected["Message"] @pytest.mark.parametrize( @@ -140,7 +167,10 @@ def test_deleteJobOversizedSandbox(mocker, inputs, params, expected): mocker.patch( "DIRAC.WorkloadManagementSystem.Agent.JobCleaningAgent.getDNForUsername", return_value=S_OK(["/bih/boh/DN"]) ) - mocker.patch("DIRAC.WorkloadManagementSystem.Agent.JobCleaningAgent.getJobParameters", return_value=params) + mockGetJobParamsUtility = mocker.patch( + "DIRAC.WorkloadManagementSystem.Agent.JobCleaningAgent.getJobParametersUtility" + ) + mockGetJobParamsUtility.return_value = params def mock_load_object(module_path, class_name): mocks = { diff --git a/src/DIRAC/WorkloadManagementSystem/Agent/test/Test_Agent_StalledJobAgent.py b/src/DIRAC/WorkloadManagementSystem/Agent/test/Test_Agent_StalledJobAgent.py index 0ef0485a976..0f49936cdb2 100644 --- a/src/DIRAC/WorkloadManagementSystem/Agent/test/Test_Agent_StalledJobAgent.py +++ b/src/DIRAC/WorkloadManagementSystem/Agent/test/Test_Agent_StalledJobAgent.py @@ -1,5 +1,5 @@ -""" Test class for Stalled Job Agent -""" +"""Test class for Stalled Job Agent""" + from unittest.mock import MagicMock import pytest @@ -50,7 +50,9 @@ def mock_load_object(module_path, class_name): ) mocker.patch("DIRAC.WorkloadManagementSystem.Agent.StalledJobAgent.rescheduleJobs", return_value=MagicMock()) - mocker.patch("DIRAC.WorkloadManagementSystem.Agent.StalledJobAgent.getJobParameters", return_value=MagicMock()) + mocker.patch( + "DIRAC.WorkloadManagementSystem.Agent.StalledJobAgent.getJobParametersUtility", return_value=MagicMock() + ) mocker.patch("DIRAC.WorkloadManagementSystem.Agent.StalledJobAgent.kill_delete_jobs", return_value=MagicMock()) stalledJobAgent = StalledJobAgent() diff --git a/src/DIRAC/WorkloadManagementSystem/DB/JobDB.py b/src/DIRAC/WorkloadManagementSystem/DB/JobDB.py index d1b1f08c096..b5e8cb005ad 100755 --- a/src/DIRAC/WorkloadManagementSystem/DB/JobDB.py +++ b/src/DIRAC/WorkloadManagementSystem/DB/JobDB.py @@ -1,4 +1,4 @@ -""" The JobDB class is a front-end to the main WMS database containing +"""The JobDB class is a front-end to the main WMS database containing job definitions and status information. It is used in most of the WMS components @@ -11,6 +11,7 @@ * *CompressJDLs*: Enable compression of JDLs when they are stored in the database, default *False*. """ + from __future__ import annotations import datetime @@ -21,7 +22,6 @@ from DIRAC.ConfigurationSystem.Client.Helpers.Resources import getSiteTier from DIRAC.Core.Base.DB import DB from DIRAC.Core.Utilities.ClassAd.ClassAdLight import ClassAd -from DIRAC.Core.Utilities.Decorators import deprecated from DIRAC.Core.Utilities.DErrno import EWMSJMAN, EWMSSUBM, cmpError from DIRAC.Core.Utilities.ReturnValues import ( S_ERROR, @@ -35,7 +35,7 @@ from DIRAC.FrameworkSystem.Client.Logger import contextLogger from DIRAC.ResourceStatusSystem.Client.SiteStatus import SiteStatus from DIRAC.WorkloadManagementSystem.Client import JobMinorStatus, JobStatus -from DIRAC.WorkloadManagementSystem.Client.JobMonitoringClient import JobMonitoringClient +from DIRAC.WorkloadManagementSystem.DB.JobParametersDB import JobParametersDB from DIRAC.WorkloadManagementSystem.DB.JobDBUtils import ( checkAndAddOwner, checkAndPrepareJob, @@ -112,53 +112,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. @@ -263,16 +217,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.""" @@ -911,7 +855,6 @@ def removeJobFromDB(self, jobIDs): for table in [ "InputData", - "JobParameters", "AtticJobParameters", "HeartBeatLoggingInfo", "OptimizerParameters", @@ -976,7 +919,7 @@ def rescheduleJob(self, jobID): jobAttrs = {"RescheduleCounter": rescheduleCounter} # Save the job parameters for later debugging - result = JobMonitoringClient().getJobParameters(jobID) + result = JobParametersDB().getJobParameters(jobID, getVOForGroup(resultDict["OwnerGroup"])) if result["OK"]: parDict = result["Value"] for key, value in parDict.get(int(jobID), {}).items(): @@ -984,10 +927,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/DB/JobParametersDB.py b/src/DIRAC/WorkloadManagementSystem/DB/JobParametersDB.py index 9030a6ff611..b5d6749eb4e 100644 --- a/src/DIRAC/WorkloadManagementSystem/DB/JobParametersDB.py +++ b/src/DIRAC/WorkloadManagementSystem/DB/JobParametersDB.py @@ -1,5 +1,5 @@ """Module containing a front-end to the OpenSearch-based JobParametersDB. -This is a drop-in replacement for MySQL-based table JobDB.JobParameters. +This OpenSearch-based DB replaces the former MySQL-based table JobDB.JobParameters. The following class methods are provided for public usage - getJobParameters() @@ -7,9 +7,12 @@ - deleteJobParameters() """ +from typing import Any + from DIRAC import S_ERROR, S_OK from DIRAC.Core.Base.ElasticDB import ElasticDB from DIRAC.Core.Utilities import TimeUtilities +from DIRAC.Core.Utilities.ReturnValues import DReturnType mapping = { "JobID": {"type": "long"}, @@ -33,17 +36,19 @@ } -def getJobParameters(jobIDs: list[int], parName: str | None, vo: str = "") -> dict: +def getJobParametersUtility( + jobIDs: list[int], parName: str | None, vo: str = "" +) -> DReturnType[dict[int, dict[str, Any]]]: """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. + Instantiates the JobParametersDB, and, when no VO is passed, the JobDB, to find + the VO each job belongs to. :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 + :return: S_OK/S_ERROR, with a dictionary with jobID (int) as key and the + parameters as value """ from DIRAC.WorkloadManagementSystem.DB.JobDB import JobDB @@ -80,22 +85,7 @@ def getJobParameters(jobIDs: list[int], parName: str | None, vo: str = "") -> di return res parameters.update(res["Value"]) - # Need anyway to get also from JobDB, for those jobs with parameters registered in MySQL or in both backends - res = jobDB.getJobParameters(jobIDs, parName) - if not res["OK"]: - return res - parametersM = res["Value"] - - # and now combine - final = dict(parametersM) - # if job in JobDB, update with parameters from ES if any - for jobID in final: - final[jobID].update(parameters.get(jobID, {})) - # if job in ES and not in JobDB, take ES - for jobID in parameters: - if jobID not in final: - final[jobID] = parameters[jobID] - return S_OK(final) + return S_OK(parameters) class JobParametersDB(ElasticDB): @@ -141,6 +131,7 @@ def getJobParameters(self, jobIDs: int | list[int], vo: str, paramList: list[str :param self: self reference :param jobID: Job ID + :param vo: virtual organization :param paramList: list of parameters to be returned (also a string is treated) :return: dict with all Job Parameter values """ diff --git a/src/DIRAC/WorkloadManagementSystem/Service/JobMonitoringHandler.py b/src/DIRAC/WorkloadManagementSystem/Service/JobMonitoringHandler.py index d8e9bc4167c..e387c7254dc 100755 --- a/src/DIRAC/WorkloadManagementSystem/Service/JobMonitoringHandler.py +++ b/src/DIRAC/WorkloadManagementSystem/Service/JobMonitoringHandler.py @@ -10,7 +10,7 @@ from DIRAC.Core.Utilities.DEncode import ignoreEncodeWarning from DIRAC.Core.Utilities.JEncode import strToIntDict from DIRAC.Core.Utilities.ObjectLoader import ObjectLoader -from DIRAC.WorkloadManagementSystem.DB.JobParametersDB import getJobParameters +from DIRAC.WorkloadManagementSystem.DB.JobParametersDB import getJobParametersUtility class JobMonitoringHandlerMixin: @@ -147,7 +147,7 @@ def export_getJobParameter(self, jobID, parName): :param str/int jobID: one single Job ID :param str parName: one single parameter name """ - res = getJobParameters([int(jobID)], parName, self.vo or "") + res = getJobParametersUtility([int(jobID)], parName, self.vo or "") if not res["OK"]: return res return S_OK(res["Value"].get(int(jobID), {})) @@ -165,7 +165,7 @@ def export_getJobParameters(self, jobIDs, parName=None): jobIDs = [jobIDs] jobIDs = [int(jobID) for jobID in jobIDs] - return getJobParameters(jobIDs, parName, self.vo or "") + return getJobParametersUtility(jobIDs, parName, self.vo or "") ############################################################################## types_getJobAttributes = [int]