[for minor release] feat: removed JobDB JobParameters table

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.

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"]

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 = {

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.")

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` (

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)