feat(JobWrapper): Report the status of the job to DiracX by Loxeris · Pull Request #111 · DIRACGrid/dirac-cwl

Loxeris · 2026-02-09T17:11:18Z

Reports the status of the job to DiracX using diracx-api.

closes #83

aldbr

Thanks @Loxeris, it's a good conversion from the existing DIRAC JobWrapper. Based on that, and my experience with it, I have a few comments to improve it.

aldbr · 2026-02-10T07:48:42Z

test/test_job_wrapper.py

 This module tests the functionalities of the job wrapper.
 """

+import asyncio


Could you try to use pytest-asyncio as it's done in diracx please?
https://github.com/DIRACGrid/diracx/blob/b7358959d3bdc4aee25e74cf2bfbf0da0e59a521/diracx-testing/pyproject.toml#L17

aldbr · 2026-02-10T07:49:04Z

test/test_integration.py

 usage scenarios.
 """

+import asyncio


Same comment here

aldbr · 2026-02-10T07:49:27Z

test/test_execution_hooks_core.py

 abstract interfaces.
 """

+import asyncio


Same comment here too

aldbr · 2026-02-10T07:58:30Z

src/dirac_cwl/job/job_wrapper_template.py


 if __name__ == "__main__":
-    main()
+    asyncio.run(main())


Instead of calling sys.exit from the main (now async) function, I think it would be cleaner to return an int, and sys.exit from here like:

return 1 if __name__ == "__main__: sys.exit(asyncio.run(main())

aldbr · 2026-02-10T07:59:18Z

src/dirac_cwl/data_management_mocks/status.py

I think you can just rename the data_management_mocks into mocks

aldbr · 2026-02-10T08:21:06Z

src/dirac_cwl/job/job_wrapper_template.py

+        ret = await set_job_status(
+            jobID,
+            "Done",
+            "Execution Complete",
+            source=src,
+        )


Can we move the calls to set_job_status within the JobWrapper?
The reason is that we also use the JobWrapper but without the JobWrapperTemplate with HPC with no external connectivity.

aldbr · 2026-02-10T08:21:49Z

src/dirac_cwl/job/job_wrapper_template.py

+        ret = await set_job_status(
+            jobID,
+            "Failed",
+            source=src,
+        )


aldbr · 2026-02-10T08:29:09Z

src/dirac_cwl/job/job_wrapper.py

        :param job_path: Path to the job working directory.
        """
        assert arguments.sandbox is not None
+        await set_job_status(self.jobID, minor_status="Downloading InputSandbox", source=src)


FYI, in DIRAC, there is a list of the existing minor status: https://github.com/DIRACGrid/DIRAC/blob/88fb54782ecae01d2b58205d0720efef0ae8d8ac/src/DIRAC/WorkloadManagementSystem/Client/JobMinorStatus.py

For now, you can make an StrEnum in dirac-cwl out of it (that will be later moved to diracx).
I think you could potentially even add more minor status like: InputSandbox successfully downloaded for instance.

aldbr · 2026-02-10T08:30:44Z

src/dirac_cwl/job/job_wrapper.py

+
+    async def initialize(self) -> None:
+        """Initialize the JobWrapper."""
+        await set_job_status(self.jobID, "Running", "Job Initialization", source=src)


And you could name them uniformly like: Initializing job..., Downloading input sandbox..., Resolving input data..., etc.

aldbr · 2026-02-10T08:31:47Z

src/dirac_cwl/job/job_wrapper.py


            # Execute the task
            logger.info("Executing Task: %s", command)
+            await set_job_status(self.jobID, minor_status="Application", source=src)


Executing payload...

aldbr · 2026-02-23T16:56:20Z

src/dirac_cwl/job/job_report.py

+class JobStatus(StrEnum):
+    """List of all available job statuses."""
+
+    SUBMITTING = "Submitting"
+    RECEIVED = "Received"
+    CHECKING = "Checking"
+    STAGING = "Staging"
+    WAITING = "Waiting"
+    MATCHED = "Matched"
+    RUNNING = "Running"
+    STALLED = "Stalled"
+    COMPLETING = "Completing"
+    DONE = "Done"
+    COMPLETED = "Completed"
+    FAILED = "Failed"
+    DELETED = "Deleted"
+    KILLED = "Killed"
+    RESCHEDULED = "Rescheduled"


Can't you reuse the job status from diracx-core?

Suggested change

class JobStatus(StrEnum):

"""List of all available job statuses."""

SUBMITTING = "Submitting"

RECEIVED = "Received"

CHECKING = "Checking"

STAGING = "Staging"

WAITING = "Waiting"

MATCHED = "Matched"

RUNNING = "Running"

STALLED = "Stalled"

COMPLETING = "Completing"

DONE = "Done"

COMPLETED = "Completed"

FAILED = "Failed"

DELETED = "Deleted"

KILLED = "Killed"

RESCHEDULED = "Rescheduled"

from diracx.core.models import JobStatus

aldbr · 2026-02-23T16:58:49Z

src/dirac_cwl/job/job_report.py

+        # add job status record
+        self.job_status_info.append((status, minor_status, application_status.strip(' "' + "'"), timestamp))
+
+    def setApplicationStatus(self, appStatus):


To be (almost) consistent with setJobStatus?

Suggested change

def setApplicationStatus(self, appStatus):

def setApplicationStatus(self, application_status: str | None = None):

Not entirely sure if the best is str = "" or str | None = None, can you check?

aldbr · 2026-02-23T16:59:18Z

src/dirac_cwl/job/job_report.py

+        self._client = client
+
+    def setJobStatus(
+        self, status: JobStatus | None = None, minor_status: JobMinorStatus | None = None, application_status: str = ""


Isn't it better like this? I am just surprised you can have status = None with application status = ""
Shouldn't JobStatus be mandatory?

Suggested change

self, status: JobStatus | None = None, minor_status: JobMinorStatus | None = None, application_status: str = ""

self, status: JobStatus | None = None, minor_status: JobMinorStatus | None = None, application_status: str | None = None

aldbr · 2026-02-23T17:03:09Z

src/dirac_cwl/job/job_report.py

+        if appStatus:
+            self.job_status_info.append(("", "", (appStatus.strip(' "' + "'"), timeStamp)))
+
+    def setJobParameter(self, par_name, par_value):


What about?

Suggested change

def setJobParameter(self, par_name, par_value):

def setJobParameter(self, key: str, value: str):

aldbr · 2026-02-23T17:04:27Z

src/dirac_cwl/job/job_report.py

+        :param par_name: name of the parameter
+        :param par_value: value of the parameter
+        """
+        self.job_parameters.append((par_name, par_value))


Any reason of having a list of tuples instead of a dictionary here? Because then I see you are building a dictionary from that when you send them

I kept how it was in DIRAC. A dictionary sounds better indeed.

aldbr · 2026-02-23T17:07:49Z

src/dirac_cwl/job/job_report.py

+    def dump(self):
+        """Print out the contents of the internal cached information."""
+        print("Job status info:")
+        for status, minor, app, timeStamp in self.job_status_info:
+            if not status:
+                status = ""
+            if not minor:
+                minor = ""
+            print(status.ljust(20), minor.ljust(30), app.ljust(30), timeStamp)
+
+        print("Job parameters:")
+        for pname, pvalue in self.job_parameters:
+            print(pname.ljust(20), pvalue.ljust(30))


I don't think we want that function here? At least not for now and not with print statements.

Suggested change

def dump(self):

"""Print out the contents of the internal cached information."""

print("Job status info:")

for status, minor, app, timeStamp in self.job_status_info:

if not status:

status = ""

if not minor:

minor = ""

print(status.ljust(20), minor.ljust(30), app.ljust(30), timeStamp)

print("Job parameters:")

for pname, pvalue in self.job_parameters:

print(pname.ljust(20), pvalue.ljust(30))

I copied it from DIRAC implementation but I agree it should probably go.

aldbr · 2026-02-23T17:12:49Z

src/dirac_cwl/job/job_wrapper.py

+    async def initialize(self) -> None:
+        """Initialize the JobWrapper."""
+        self.job_report.setJobStatus(JobStatus.RUNNING, JobMinorStatus.JOB_INITIALIZATION)
+        await self.job_report.commit()


Let's say there is no initialize for now (I want to see if we will need it in the future). The job_report call is then made in __init__.

Suggested change

async def initialize(self) -> None:

"""Initialize the JobWrapper."""

self.job_report.setJobStatus(JobStatus.RUNNING, JobMinorStatus.JOB_INITIALIZATION)

await self.job_report.commit()

self.job_report.setJobStatus(JobStatus.RUNNING, JobMinorStatus.JOB_INITIALIZATION)

await self.job_report.commit()

We need an await to commit the job report, await needs to be in an async method and __init__ can't be one afaik. I guess we could just remove the problem entirely by not committing here. I need to check when it's needed.

Ah ok I see. I think it's better to do the commit at the end of every public methods.
So I don't say anything wrong, that would be at the end of preprocess/execute/postprocess.
What do you think?

I agree this should probably work out. Maybe we could also add a commit just after the JobMinorStatus.APPLICATION so the user can see the payload is running in case it takes a while.

aldbr · 2026-02-23T17:14:12Z

src/dirac_cwl/job/job_wrapper.py

        assert arguments.sandbox is not None
+        self.job_report.setJobStatus(minor_status=JobMinorStatus.DOWNLOADING_INPUT_SANDBOX)
        if not self.execution_hooks_plugin:
+            self.job_report.setJobStatus(minor_status=JobMinorStatus.FAILED_DOWNLOADING_INPUT_SANDBOX)


Don't you need to commit here?
Unless you want to catch exceptions in the JobWrapperTemplate and then commit?

I actually think it's fine not to commit here because Exceptions are caught in a try catch in run_job() and there is a commit there.

except Exception: logger.exception("JobWrapper: Failed to execute workflow") self.job_report.setJobStatus(JobStatus.FAILED) await self.job_report.commit() return False

I might also just move this commit to the finally block, to not have one for each return of the method too.

aldbr · 2026-02-23T17:14:59Z

src/dirac_cwl/job/job_wrapper.py

+        else:
+            self.job_report = JobReport(self.job_id, src, AsyncDiracClient())
+
+    async def initialize(self) -> None:


Good thing to make everything async! (I think the whole code should be async to fit with the diracx code base)

aldbr · 2026-02-23T17:16:30Z

src/dirac_cwl/job/job_wrapper.py

+            await self.execution_hooks_plugin.store_output(outputs)
+            self.job_report.setJobStatus(status=JobStatus.COMPLETING, minor_status=JobMinorStatus.OUTPUT_DATA_UPLOADED)
+        except RuntimeError as err:
+            self.job_report.setJobStatus(status=JobStatus.FAILED, minor_status=JobMinorStatus.UPLOADING_OUTPUT_DATA)


Same here, no commit?

Loxeris · 2026-03-04T10:28:42Z

src/dirac_cwl/job/job_report.py

+class JobMinorStatus(StrEnum):
+    """List of all available job minor statuses."""
+
+    APPLICATION = "Executing Payload"
+    APP_ERRORS = "Application Finished With Errors"
+    # APP_NOT_FOUND = "Application not found"
+    APP_SUCCESS = "Application Finished Successfully"
+    # APP_THREAD_FAILED = "Application thread failed"
+    # APP_THREAD_NOT_COMPLETE = "Application thread did not complete"
+    DOWNLOADING_INPUT_SANDBOX = "Downloading InputSandbox"
+    # DOWNLOADING_INPUT_SANDBOX_LFN = "Downloading InputSandbox LFN(s)"
+    # EXCEPTION_DURING_EXEC = "Exception During Execution"
+    EXEC_COMPLETE = "Execution Complete"
+    FAILED_DOWNLOADING_INPUT_SANDBOX = "Failed Downloading InputSandbox"
+    # FAILED_DOWNLOADING_INPUT_SANDBOX_LFN = "Failed Downloading InputSandbox LFN(s)"
+    # FAILED_SENDING_REQUESTS = "Failed sending requests"
+    # GOING_RESCHEDULE = "Going to reschedule job"
+    # ILLEGAL_JOB_JDL = "Illegal Job JDL"
+    INPUT_DATA_RESOLUTION = "Resolving Input Data"
+    # INPUT_NOT_AVAILABLE = "Input Data Not Available"
+    # JOB_EXCEEDED_CPU = "Job has reached the CPU limit of the queue"
+    # JOB_EXCEEDED_WALL_CLOCK = "Job has exceeded maximum wall clock time"
+    JOB_INITIALIZATION = "Initializing Job"
+    # JOB_INSUFFICIENT_DISK = "Job has insufficient disk space to continue"
+    # JOB_WRAPPER_EXECUTION = "JobWrapper execution"
+    # JOB_WRAPPER_INITIALIZATION = "Job Wrapper Initialization"
+    # MARKED_FOR_TERMINATION = "Marked for termination"
+    # NO_CANDIDATE_SITE_FOUND = "No candidate sites available"
+    OUTPUT_DATA_UPLOADED = "Output Data Uploaded"
+    OUTPUT_SANDBOX_UPLOADED = "Output Sandbox Uploaded"
+    # PENDING_REQUESTS = "Pending Requests"
+    # PILOT_AGENT_SUBMISSION = "Pilot Agent Submission"
+    # RECEIVED_KILL_SIGNAL = "Received Kill signal"
+    # REQUESTS_DONE = "Requests done"
+    # RESCHEDULED = "Job Rescheduled"
+    RESOLVING_OUTPUT_SANDBOX = "Resolving Output Sandbox"
+    # STALLED_PILOT_NOT_RUNNING = "Job stalled: pilot not running"
+    # UPLOADING_JOB_OUTPUTS = "Uploading Outputs"
+    UPLOADING_OUTPUT_DATA = "Uploading Output Data"
+    UPLOADING_OUTPUT_SANDBOX = "Uploading Output Sandbox"
+    # WATCHDOG_STALLED = "Watchdog identified this job as stalled"


I commented all minor statuses unused in the job wrapper when working on it. I'm wondering if I should restore all these minor statuses, keep them commented, or delete them entirely from the enum.

Loxeris force-pushed the diracx-job-statuses branch from 71983e2 to 2ff453b Compare February 9, 2026 17:19

aldbr requested changes Feb 10, 2026

View reviewed changes

Loxeris added 4 commits February 20, 2026 14:46

feat: report job status

2fca114

fix: fix tests

06f9fde

feat: add JobReport class and apply pr suggestions

c881093

feat: add job parameters to job report

140b8bc

Loxeris force-pushed the diracx-job-statuses branch from 2ff453b to 140b8bc Compare February 23, 2026 16:53

aldbr requested changes Feb 23, 2026

View reviewed changes

Loxeris added 6 commits March 2, 2026 16:09

refactor: use dicts to store job reports

e506f62

refactor: remove initialize method

f278eb3

style: prettier local job report

00d283e

fix: fix store_output not being awaited in test

d12076a

refactor: add job report commits along job wrapper

7e19467

fix: fix deserialization error

ae4abc4

Loxeris force-pushed the diracx-job-statuses branch from 4e793fe to ae4abc4 Compare March 3, 2026 15:41

test: test reports of job status and parameters

c7b7188

Loxeris force-pushed the diracx-job-statuses branch from a6ac83f to c7b7188 Compare March 4, 2026 10:19

Loxeris commented Mar 4, 2026

View reviewed changes

Loxeris marked this pull request as ready for review March 4, 2026 10:54

Loxeris requested a review from aldbr March 4, 2026 10:54

	def setApplicationStatus(self, appStatus):
	def setApplicationStatus(self, application_status: str \| None = None):

	self, status: JobStatus \| None = None, minor_status: JobMinorStatus \| None = None, application_status: str = ""
	self, status: JobStatus \| None = None, minor_status: JobMinorStatus \| None = None, application_status: str \| None = None

	def setJobParameter(self, par_name, par_value):
	def setJobParameter(self, key: str, value: str):

Conversation

Loxeris commented Feb 9, 2026

Uh oh!

aldbr left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants