[v9.1] 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 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"]

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
 

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(

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

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

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,18 +919,14 @@ 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():
                 result = self.setAtticJobParameter(jobID, key, value, rescheduleCounter - 1)
                 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` (