feat(cwl): CWL job wrapper

setup.cfg

@@ -31,9 +31,11 @@ install_requires =
     cwltool
     diraccfg
     DIRACCommon
-    diracx-client >=v0.0.1
-    diracx-core >=v0.0.1
-    diracx-cli >=v0.0.1
+    diracx-api @ git+https://github.com/ryuwd/diracx.git@feat/cwl-job-submission#subdirectory=diracx-api
+    diracx-cli @ git+https://github.com/ryuwd/diracx.git@feat/cwl-job-submission#subdirectory=diracx-cli
+    diracx-client @ git+https://github.com/ryuwd/diracx.git@feat/cwl-job-submission#subdirectory=diracx-client
+    diracx-core @ git+https://github.com/ryuwd/diracx.git@feat/cwl-job-submission#subdirectory=diracx-core
+    diracx-logic @ git+https://github.com/ryuwd/diracx.git@feat/cwl-job-submission#subdirectory=diracx-logic
     db12
     fabric
     fts3

src/DIRAC/Core/Security/DiracX.py

@@ -6,6 +6,7 @@
     "diracxTokenFromPEM",
     "executeRPCStub",
     "FutureClient",
+    "writeDiracxTokenCache",
 )
 
 import base64
@@ -93,9 +94,18 @@ class FutureClient:
     ...
 
 
-@contextmanager
-def DiracXClient() -> Iterator[SyncDiracClient]:
-    """Get a DiracX client instance with the current user's credentials"""
+def writeDiracxTokenCache() -> tuple[str, Path]:
+    """Resolve the DiracX URL and write the proxy's token to a hash-named
+    cache file under ``gettempdir()``. Idempotent — re-uses an existing
+    cache file when its content already matches the current token.
+
+    Suitable for both ``DiracxPreferences(credentials_path=...)`` and the
+    ``DIRACX_CREDENTIALS_PATH`` env var.
+
+    :return: Tuple of (diracx_url, token_file_path).
+    :raises ValueError: If ``/DiracX/URL`` is unset or the proxy carries no
+        DiracX token.
+    """
     diracxUrl = gConfig.getValue("/DiracX/URL")
     if not diracxUrl:
         raise ValueError("Missing mandatory /DiracX/URL configuration")
@@ -112,6 +122,13 @@ def DiracXClient() -> Iterator[SyncDiracClient]:
         with secureOpenForWrite(token_file) as (fd, _):
             fd.write(json.dumps(diracxToken))
 
+    return diracxUrl, token_file
+
+
+@contextmanager
+def DiracXClient() -> Iterator[SyncDiracClient]:
+    """Get a DiracX client instance with the current user's credentials"""
+    diracxUrl, token_file = writeDiracxTokenCache()
     pref = DiracxPreferences(url=diracxUrl, credentials_path=token_file)
     with SyncDiracClient(diracx_preferences=pref) as api:
         yield api

src/DIRAC/WorkloadManagementSystem/Utilities/Utils.py

@@ -2,8 +2,6 @@
 
 import os
 from pathlib import Path
-from glob import glob
-import subprocess
 import sys
 import json
 
@@ -129,39 +127,35 @@ def createJobWrapper(
 
 
 def __createCWLJobWrapper(jobID, wrapperPath, log, rootLocation):
-    # Get the new JobWrapper
+    """Create a CWL job wrapper that fetches the workflow from the diracX API.
+
+    The CWL definition and input parameters are fetched at runtime from diracX
+    using the job ID. The wrapper script is extracted from diracx-logic via
+    importlib.resources and written to disk for execution.
+    """
+    import importlib.resources
+
     if not rootLocation:
         rootLocation = wrapperPath
-    protoPath = Path(wrapperPath) / f"proto{jobID}"
-    protoPath.unlink(missing_ok=True)
-    log.info("Cloning JobWrapper from repository https://github.com/DIRACGrid/dirac-cwl.git into", protoPath)
+
     try:
-        subprocess.run(["git", "clone", "https://github.com/DIRACGrid/dirac-cwl.git", str(protoPath)], check=True)
-    except subprocess.CalledProcessError:
-        return S_ERROR("Failed to clone the JobWrapper repository")
-    wrapperFound = glob(os.path.join(str(protoPath), "**", "job_wrapper_template.py"), recursive=True)
-    if len(wrapperFound) < 1 or not Path(wrapperFound[0]).is_file():
-        return S_ERROR("Could not find the JobWrapper in the cloned repository")
-    jobWrapperFile = wrapperFound[0]
-    directJobWrapperFile = str(Path(rootLocation) / Path(wrapperFound[0]).relative_to(wrapperPath))
-
-    jobWrapperJsonFile = Path(wrapperPath) / f"InputSandbox{jobID}" / "job.json"
-    directJobWrapperJsonFile = Path(rootLocation) / f"InputSandbox{jobID}" / "job.json"
-    # Create the executable file
+        wrapper_ref = importlib.resources.files("diracx.cli.internal") / "job_wrapper.py"
+        jobWrapperFile = os.path.join(wrapperPath, f"CWLWrapper_{jobID}.py")
+        with importlib.resources.as_file(wrapper_ref) as template_path:
+            with open(template_path) as src, open(jobWrapperFile, "w") as dst:
+                dst.write(src.read())
+    except (ImportError, FileNotFoundError) as e:
+        return S_ERROR(f"Could not load CWL job wrapper template: {e}")
+
+    jobWrapperJsonFile = os.path.join(wrapperPath, f"cwl_job_{jobID}.json")
+    with open(jobWrapperJsonFile, "w") as f:
+        json.dump({"JobID": jobID}, f)
+
+    directJobWrapperFile = str(Path(rootLocation) / Path(jobWrapperFile).relative_to(wrapperPath))
+
     jobExeFile = os.path.join(wrapperPath, f"Job{jobID}")
-    protoPath = str(Path(rootLocation) / Path(protoPath).relative_to(wrapperPath))
-    pixiPath = str(Path(rootLocation) / ".pixi")
     jobFileContents = f"""#!/bin/bash
-# Install pixi
-export PIXI_NO_PATH_UPDATE=1
-export PIXI_HOME={pixiPath}
-curl -fsSL https://pixi.sh/install.sh | bash
-export PATH="{pixiPath}/bin:$PATH"
-pixi install --manifest-path {protoPath}
-# Get json
-dirac-wms-job-get-input {jobID} -D {rootLocation}
-# Run JobWrapper
-pixi run --manifest-path {protoPath} python {directJobWrapperFile} {directJobWrapperJsonFile} {jobID}
+python {directJobWrapperFile} {jobID}
 """
     return S_OK((jobWrapperFile, jobWrapperJsonFile, jobExeFile, jobFileContents))