Merge pull request #6973 from simon-mazenoux/feat-validating-jdl-using-pydantic

[8.1] Validating the JDL format using pydantic

Author: fstagni

environment.yml

@@ -30,6 +30,7 @@ dependencies:
   - psutil >=4.2.0
   - pyasn1 >0.4.1
   - pyasn1-modules
+  - pydantic
   - python-json-logger >=0.1.8
   - pytz >=2015.7
   - pyyaml

pyproject.toml

@@ -69,6 +69,11 @@ ignored-modules = ["MySQLdb", "numpy"]
 # avoid hangs.
 jobs = 0
 
+# This allows pylint to not display the following false error :
+# No name 'BaseModel' in module 'pydantic' (no-name-in-module)
+# Related issue: https://github.com/pydantic/pydantic/issues/1961
+extension-pkg-whitelist = ["pydantic"]
+
 [tool.pylint.typecheck]
 # List of decorators that change the signature of a decorated function.
 signature-mutators = []

src/DIRAC/Core/Utilities/ClassAd/ClassAdLight.py

@@ -287,7 +287,7 @@ def getAttributeFloat(self, name):
                 value = None
         return value
 
-    def getAttributes(self):
+    def getAttributes(self) -> list[str]:
         """Get the list of all the attribute names
 
         :return: list of names as strings

src/DIRAC/Core/Utilities/Graphs/Palette.py

@@ -31,8 +31,6 @@
     JobMinorStatus.INPUT_NOT_AVAILABLE: "#2822A6",
     JobMinorStatus.INPUT_DATA_RESOLUTION: "#FFBE94",
     JobMinorStatus.DOWNLOADING_INPUT_SANDBOX: "#586CFF",
-    JobMinorStatus.INPUT_CONTAINS_SLASHES: "#AB7800",
-    JobMinorStatus.INPUT_INCORRECT: "#6812D6",
     JobMinorStatus.JOB_WRAPPER_INITIALIZATION: "#FFFFCC",
     JobMinorStatus.JOB_EXCEEDED_WALL_CLOCK: "#FF33CC",
     JobMinorStatus.JOB_INSUFFICIENT_DISK: "#33FFCC",

src/DIRAC/Core/Utilities/JDL.py

@@ -1,6 +1,46 @@
+"""Transformation classes around the JDL format."""
+
 from diraccfg import CFG
+from pydantic import ValidationError
+
 from DIRAC import S_OK, S_ERROR
 from DIRAC.Core.Utilities import List
+from DIRAC.Core.Utilities.ClassAd.ClassAdLight import ClassAd
+from DIRAC.WorkloadManagementSystem.Utilities.JobModel import BaseJobDescriptionModel
+
+ARGUMENTS = "Arguments"
+BANNED_SITES = "BannedSites"
+CPU_TIME = "CPUTime"
+EXECUTABLE = "Executable"
+EXECUTION_ENVIRONMENT = "ExecutionEnvironment"
+GRID_CE = "GridCE"
+INPUT_DATA = "InputData"
+INPUT_DATA_POLICY = "InputDataPolicy"
+INPUT_SANDBOX = "InputSandbox"
+JOB_CONFIG_ARGS = "JobConfigArgs"
+JOB_TYPE = "JobType"
+JOB_GROUP = "JobGroup"
+LOG_LEVEL = "LogLevel"
+NUMBER_OF_PROCESSORS = "NumberOfProcessors"
+MAX_NUMBER_OF_PROCESSORS = "MaxNumberOfProcessors"
+MIN_NUMBER_OF_PROCESSORS = "MinNumberOfProcessors"
+OUTPUT_DATA = "OutputData"
+OUTPUT_PATH = "OutputPath"
+OUTPUT_SE = "OutputSE"
+PLATFORM = "Platform"
+PRIORITY = "Priority"
+STD_ERROR = "StdError"
+STD_OUTPUT = "StdOutput"
+OUTPUT_SANDBOX = "OutputSandbox"
+JOB_NAME = "JobName"
+SITE = "Site"
+TAGS = "Tags"
+
+OWNER = "Owner"
+OWNER_GROUP = "OwnerGroup"
+VO = "VirtualOrganization"
+
+CREDENTIALS_FIELDS = {OWNER, OWNER_GROUP, VO}
 
 
 def loadJDLAsCFG(jdl):
@@ -138,7 +178,7 @@ def dumpCFGAsJDL(cfg, level=1, tab="  "):
         else:
             val = List.fromChar(cfg[key])
             # Some attributes are never lists
-            if len(val) < 2 or key in ["Arguments", "Executable", "StdOutput", "StdError"]:
+            if len(val) < 2 or key in [ARGUMENTS, EXECUTABLE, STD_OUTPUT, STD_ERROR]:
                 value = cfg[key]
                 try:
                     try_value = float(value)
@@ -157,3 +197,163 @@ def dumpCFGAsJDL(cfg, level=1, tab="  "):
                 contents.append("%s};" % indent)
     contents.append(f"{tab * (level - 1)}]")
     return "\n".join(contents)
+
+
+def jdlToBaseJobDescriptionModel(classAd: ClassAd):
+    """
+    Converts a JDL string into a JSON string for data validation from the BaseJob model
+    This method allows compatibility with older Client versions that used the _toJDL method
+    """
+    try:
+        jobDescription = BaseJobDescriptionModel(
+            executable=classAd.getAttributeString(EXECUTABLE),
+        )
+        if classAd.lookupAttribute(ARGUMENTS):
+            jobDescription.arguments = classAd.getAttributeString(ARGUMENTS)
+            classAd.deleteAttribute(ARGUMENTS)
+
+        if classAd.lookupAttribute(BANNED_SITES):
+            jobDescription.bannedSites = classAd.getListFromExpression(BANNED_SITES)
+            classAd.deleteAttribute(BANNED_SITES)
+
+        if classAd.lookupAttribute(CPU_TIME):
+            jobDescription.cpuTime = classAd.getAttributeInt(CPU_TIME)
+            classAd.deleteAttribute(CPU_TIME)
+
+        if classAd.lookupAttribute(EXECUTABLE):
+            jobDescription.executable = classAd.getAttributeString(EXECUTABLE)
+            classAd.deleteAttribute(EXECUTABLE)
+
+        if classAd.lookupAttribute(EXECUTION_ENVIRONMENT):
+            executionEnvironment = classAd.getListFromExpression(EXECUTION_ENVIRONMENT)
+            if executionEnvironment:
+                jobDescription.executionEnvironment = {}
+                for element in executionEnvironment:
+                    key, value = element.split("=")
+                    if value.isdigit():
+                        value = int(value)
+                    else:
+                        try:
+                            value = float(value)
+                        except ValueError:
+                            pass
+                    jobDescription.executionEnvironment[key] = value
+            classAd.deleteAttribute(EXECUTION_ENVIRONMENT)
+
+        if classAd.lookupAttribute(GRID_CE):
+            jobDescription.gridCE = classAd.getAttributeString(GRID_CE)
+            classAd.deleteAttribute(GRID_CE)
+
+        if classAd.lookupAttribute(INPUT_DATA):
+            jobDescription.inputData = classAd.getListFromExpression(INPUT_DATA)
+            classAd.deleteAttribute(INPUT_DATA)
+
+        if classAd.lookupAttribute(INPUT_DATA_POLICY):
+            jobDescription.inputDataPolicy = classAd.getAttributeString(INPUT_DATA_POLICY)
+            classAd.deleteAttribute(INPUT_DATA_POLICY)
+
+        if classAd.lookupAttribute(INPUT_SANDBOX):
+            jobDescription.inputSandbox = classAd.getListFromExpression(INPUT_SANDBOX)
+            classAd.deleteAttribute(INPUT_SANDBOX)
+
+        if classAd.lookupAttribute(JOB_CONFIG_ARGS):
+            jobDescription.jobConfigArgs = classAd.getAttributeString(JOB_CONFIG_ARGS)
+            classAd.deleteAttribute(JOB_CONFIG_ARGS)
+
+        if classAd.lookupAttribute(JOB_GROUP):
+            jobDescription.jobGroup = classAd.getAttributeString(JOB_GROUP)
+            classAd.deleteAttribute(JOB_GROUP)
+
+        if classAd.lookupAttribute(JOB_NAME):
+            jobDescription.jobName = classAd.getAttributeString(JOB_NAME)
+            classAd.deleteAttribute(JOB_NAME)
+
+        if classAd.lookupAttribute(JOB_TYPE):
+            jobDescription.jobType = classAd.getAttributeString(JOB_TYPE)
+            classAd.deleteAttribute(JOB_TYPE)
+
+        if classAd.lookupAttribute(LOG_LEVEL):
+            jobDescription.logLevel = classAd.getAttributeString(LOG_LEVEL)
+            classAd.deleteAttribute(LOG_LEVEL)
+
+        if classAd.lookupAttribute(NUMBER_OF_PROCESSORS):
+            jobDescription.maxNumberOfProcessors = classAd.getAttributeInt(NUMBER_OF_PROCESSORS)
+            jobDescription.minNumberOfProcessors = classAd.getAttributeInt(NUMBER_OF_PROCESSORS)
+            classAd.deleteAttribute(NUMBER_OF_PROCESSORS)
+            classAd.deleteAttribute(MAX_NUMBER_OF_PROCESSORS)
+            classAd.deleteAttribute(MIN_NUMBER_OF_PROCESSORS)
+        else:
+            if classAd.lookupAttribute(MAX_NUMBER_OF_PROCESSORS):
+                jobDescription.maxNumberOfProcessors = classAd.getAttributeInt(MAX_NUMBER_OF_PROCESSORS)
+                classAd.deleteAttribute(MAX_NUMBER_OF_PROCESSORS)
+            if classAd.lookupAttribute(MIN_NUMBER_OF_PROCESSORS):
+                jobDescription.minNumberOfProcessors = classAd.getAttributeInt(MIN_NUMBER_OF_PROCESSORS)
+                classAd.deleteAttribute(MIN_NUMBER_OF_PROCESSORS)
+
+        if classAd.lookupAttribute(OUTPUT_DATA):
+            jobDescription.outputData = set(classAd.getListFromExpression(OUTPUT_DATA))
+            classAd.deleteAttribute(OUTPUT_DATA)
+
+        if classAd.lookupAttribute(OUTPUT_SANDBOX):
+            jobDescription.outputSandbox = set(classAd.getListFromExpression(OUTPUT_SANDBOX))
+            classAd.deleteAttribute(OUTPUT_SANDBOX)
+
+        if classAd.lookupAttribute(OUTPUT_PATH):
+            jobDescription.outputPath = classAd.getAttributeString(OUTPUT_PATH)
+            classAd.deleteAttribute(OUTPUT_PATH)
+
+        if classAd.lookupAttribute(OUTPUT_SE):
+            jobDescription.outputSE = classAd.getAttributeString(OUTPUT_SE)
+            classAd.deleteAttribute(OUTPUT_SE)
+
+        if classAd.lookupAttribute(SITE):
+            jobDescription.sites = classAd.getListFromExpression(SITE)
+            classAd.deleteAttribute(SITE)
+
+        if classAd.lookupAttribute(PLATFORM):
+            jobDescription.platform = classAd.getAttributeString(PLATFORM)
+            classAd.deleteAttribute(PLATFORM)
+
+        if classAd.lookupAttribute(PRIORITY):
+            jobDescription.priority = classAd.getAttributeInt(PRIORITY)
+            classAd.deleteAttribute(PRIORITY)
+
+        if classAd.lookupAttribute(STD_OUTPUT):
+            jobDescription.stdout = classAd.getAttributeString(STD_OUTPUT)
+            classAd.deleteAttribute(STD_OUTPUT)
+
+        if classAd.lookupAttribute(STD_ERROR):
+            jobDescription.stderr = classAd.getAttributeString(STD_ERROR)
+            classAd.deleteAttribute(STD_ERROR)
+
+        if classAd.lookupAttribute(TAGS):
+            jobDescription.tags = classAd.getListFromExpression(TAGS)
+            classAd.deleteAttribute(TAGS)
+
+        # Remove credentials
+        for attribute in CREDENTIALS_FIELDS:
+            classAd.deleteAttribute(attribute)
+
+        # Remove legacy attributes
+        for attribute in {"DIRACSetup", "OwnerDN"}:
+            classAd.deleteAttribute(attribute)
+
+        for attribute in classAd.getAttributes():
+            if not jobDescription.extraFields:
+                jobDescription.extraFields = {}
+
+            value = classAd.getAttributeString(attribute)
+            if value.isdigit():
+                value = int(value)
+            else:
+                try:
+                    value = float(value)
+                except ValueError:
+                    pass
+
+            jobDescription.extraFields[attribute] = value
+
+    except ValidationError as e:
+        return S_ERROR(f"Invalid JDL: {e}")
+
+    return S_OK(jobDescription)

src/DIRAC/Core/Utilities/test/Test_JDL.py

@@ -0,0 +1,101 @@
+""" Unit tests for JDL module"""
+
+# pylint: disable=protected-access, invalid-name
+
+from io import StringIO
+from unittest.mock import patch
+
+import pytest
+
+from DIRAC import S_OK
+from DIRAC.Core.Utilities.ClassAd.ClassAdLight import ClassAd
+from DIRAC.Core.Utilities.JDL import jdlToBaseJobDescriptionModel
+from DIRAC.Interfaces.API.Job import Job
+from DIRAC.WorkloadManagementSystem.Utilities.JobModel import JobDescriptionModel
+
+
+def test_jdlToBaseJobDescriptionModel_valid():
+    """This test makes sure that a job object can be parsed by the jdlToBaseJobDescriptionModel method"""
+    # Arrange
+    with patch("DIRAC.Core.Base.API.getSites", return_value=S_OK(["LCG.IN2P3.fr"])):
+        job = Job()
+    job.setConfigArgs("configArgs")
+    job.setCPUTime(3600)
+    job.setExecutable("/bin/echo", arguments="arguments", logFile="logFile")
+    job.setName("JobName")
+    with patch(
+        "DIRAC.ConfigurationSystem.Client.Helpers.Operations.Operations.getValue",
+        return_value="DIRAC.WorkloadManagementSystem.Client.DownloadInputData",
+    ):
+        job.setInputDataPolicy("download")
+    job.setInputSandbox(["inputfile.opts"])
+    job.setOutputSandbox(["inputfile.opts"])
+    job.setInputData(["/lhcb/production/DC04/v2/DST/00000742_00003493_10.dst"])
+    job.setParameterSequence("IntSequence", [1, 2, 3])
+    job.setParameterSequence("StrSequence", ["a", "b", "c"])
+    job.setParameterSequence("FloatSequence", [1.0, 2.0, 3.0])
+
+    job.setOutputData(["outputfile.root"], outputSE="IN2P3-disk", outputPath="/myjobs/1234")
+    with patch("DIRAC.Interfaces.API.Job.getDIRACPlatforms", return_value=S_OK("x86_64-slc6-gcc49-opt")):
+        job.setPlatform("x86_64-slc6-gcc49-opt")
+    job.setPriority(10)
+
+    job.setDestination("LCG.IN2P3.fr")
+    job.setNumberOfProcessors(3)
+    with patch("DIRAC.Interfaces.API.Job.getCESiteMapping", return_value=S_OK({"some.ce.IN2P3.fr": "LCG.IN2P3.fr"})):
+        job.setDestinationCE("some.ce.IN2P3.fr")
+    job.setType("Test")
+    job.setTag(["WholeNode", "8GBMemory"])
+    job.setJobGroup("1234abcd")
+    job.setLogLevel("DEBUG")
+    job.setConfigArgs("configArgs")
+    job.setExecutionEnv({"INTVAR": 1, "STRVAR": "a"})
+    job._addJDLParameter("ExtraInt", 1)
+    job._addJDLParameter("ExtraFloat", 1.0)
+    job._addJDLParameter("ExtraString", "test")
+    # The 3 lines below are not a use case given that workflow parameters
+    # must be strings, ints, floats or booleans
+    # job._addJDLParameter("ExtraIntList", ";".join(["1", "2", "3"]))
+    # job._addJDLParameter("ExtraFloatList", ";".join(["1.0", "2.0", "3.0"]))
+    # job._addJDLParameter("ExtraStringList",";".join(["a", "b", "c"]))
+
+    # We make sure that the job above is valid
+    assert not job.errorDict
+
+    # Act
+    xml = job._toXML()
+    jdl = f"[{job._toJDL(jobDescriptionObject=StringIO(xml))}]"
+
+    # Assert
+    res = jdlToBaseJobDescriptionModel(ClassAd(jdl))
+    assert res["OK"], res["Message"]
+
+    data = res["Value"].dict()
+    with patch(
+        "DIRAC.WorkloadManagementSystem.Utilities.JobModel.getDIRACPlatforms",
+        return_value=S_OK(["x86_64-slc6-gcc49-opt"]),
+    ):
+        with patch(
+            "DIRAC.WorkloadManagementSystem.Utilities.JobModel.getSites",
+            return_value=S_OK(["LCG.IN2P3.fr"]),
+        ):
+            assert JobDescriptionModel(owner="owner", ownerGroup="ownerGroup", vo="lhcb", **data)
+
+
+@pytest.mark.parametrize(
+    "jdl",
+    [
+        """[]""",  # No executable
+        """[Executable="";]""",  # Empty executable
+        """Executable="executable";""",  # Missing brackets
+    ],
+)
+def test_jdlToBaseJobDescriptionModel_invalid(jdl):
+    """This test makes sure that a job object without an executable raises an error"""
+    # Arrange
+
+    # Act
+    res = jdlToBaseJobDescriptionModel(ClassAd(jdl))
+
+    # Assert
+    assert not res["OK"], res["Value"]

src/DIRAC/Interfaces/API/Job.py

@@ -40,7 +40,7 @@
 from DIRAC.Core.Utilities.ObjectLoader import ObjectLoader
 from DIRAC.ConfigurationSystem.Client.Helpers.Operations import Operations
 from DIRAC.ConfigurationSystem.Client.Helpers.Registry import getVOForGroup
-from DIRAC.ConfigurationSystem.Client.Helpers.Resources import getCESiteMapping
+from DIRAC.ConfigurationSystem.Client.Helpers.Resources import getCESiteMapping, getDIRACPlatforms
 from DIRAC.Interfaces.API.Dirac import Dirac
 from DIRAC.Workflow.Utilities.Utils import getStepDefinition, addStepToWorkflow
 
@@ -82,12 +82,6 @@ def __init__(self, script=None, stdout="std.out", stderr="std.err"):
         self.wfArguments = {}
         self.parametricWFArguments = {}
 
-        # loading the function that will be used to determine the platform (it can be VO specific)
-        res = ObjectLoader().loadObject("ConfigurationSystem.Client.Helpers.Resources", "getDIRACPlatforms")
-        if not res["OK"]:
-            self.log.fatal(res["Message"])
-        self.getDIRACPlatforms = res["Value"]
-
         self.script = script
         if not script:
             self.workflow = Workflow()
@@ -451,7 +445,7 @@ def setPlatform(self, platform):
             return self._reportError("Expected string for platform", **kwargs)
 
         if not platform.lower() == "any":
-            availablePlatforms = self.getDIRACPlatforms()
+            availablePlatforms = getDIRACPlatforms()
             if not availablePlatforms["OK"]:
                 return self._reportError("Can't check for platform", **kwargs)
             if platform in availablePlatforms["Value"]:

src/DIRAC/WorkloadManagementSystem/Client/JobMinorStatus.py

@@ -76,8 +76,4 @@
 #
 INPUT_NOT_AVAILABLE = "Input Data Not Available"
 #
-INPUT_CONTAINS_SLASHES = "Input data contains //"
-#
-INPUT_INCORRECT = "Input data not correctly specified"
-#
 NO_CANDIDATE_SITE_FOUND = "No candidate sites available"