Support of ISB in jobs submit CLI

[arrabito]

As far as I see the CLI for job submission does not support yet submission of jobs with ISB but I see that there is a 'create_sandbox' jobs api.

Can you please tell me if in order to support job submission with ISB is it enough to update the 'jobs submit' CLI to make use of the 'create_sandbox' jobs api ?

Thank you.

[aldbr]

You want to make a test CLI with CWL and the /jobs/jdl endpoint, right?
Then yes, I would say you just need to create the sandbox(es) in the CLI and then specify the path in the JDL as we are doing in DIRAC.

[arrabito]

I would say you just need to create the sandbox(es)

You mean calling:
https://github.com/DIRACGrid/diracx/blob/main/diracx-api/src/diracx/api/jobs.py#L51C11-L51C25

right?

Because currently if I just use:

dirac jobs submit test_job.jdl

with InputSandbox specified in the JDL, I get:

Error: No Input sandbox found for job 98495. Possible causes are: the job does not exist, no sandbox was registered or you do not have permission to access it.

While submitting the same jdl with the legacy dirac client:

dirac-wms-job-submit test_job.jdl

it correctly uploads the ISB.

[aldbr]

The CLI is still experimental, may be it does not work properly.
I can check that in 2 weeks.

[arrabito]

OK thank you.

[arrabito]

As a follow up, I've used:

diracx.api.jobs.create_sandbox

to create a sandbox.

Then I've submitted a job:

    async with AsyncDiracClient() as api:
       jobs = await api.jobs.submit_jdl_jobs([jdl])

and the I've assigned the previously created SB to the job with:

    async with AsyncDiracClient() as api:
        res = await api.jobs.assign_sandbox_to_job(job_id, json.dumps(pfn))

but I see that it assigns it as an OSB, while I would like to assign it as ISB.

Which is the correct way to do for the ISB?

Thank you.

[aldbr]

I think you don't need to manually assign the ISB to the job, you just need to specify the result returned by diracx.api.jobs.create_sandbox in the JDL, and the optimizers should assign the ISB to the jobs within DIRAC.

[arrabito]

I've just tried but it doesn't work

[arrabito]

I've talked too early.
It worked indeed. Thanks a lot.

[Stellatsuu]

Hello,
I’m currently working on adding ISB support for job submissions, and I tried to implement the approach mentioned here: #619 (reply in thread)

Here's what I did : 0f98a2d
However, when I run a job with an input sandbox, I get an error related to DiracXPreferences when calling create_sandbox, and I don’t understand what’s going wrong.

Am I allowed to call create_sandbox in this function or is there another way to do so?
I also tried calling it like this, explained in api.md, and I'm having the same issue:

async with AsyncDiracClient() as client:
    result = await create_sandbox(paths, client)

[Stellatsuu]

Hello Federico,

No worries about the late reply, and thanks also for the explaination!
I understand a bit more now, I'll check that in more details and change it when I have time :)

[Stellatsuu]

Hello @fstagni,

I'm back on this issue with the ISB support and I had some questions:

I'm not sure of what you're trying to say here:

the client, through the APIs, creates the sandbox and uploads it

I am able to create the sandboxes from the CLI, using the API when submitting a job. Such as:

@app.async_command()
async def submit(jdls: list[FileText]):
    # Create and upload InputSandboxes from JDLs
    for i, jdl in enumerate(jdls):
        original_jdl = jdl.read()

        # Fix possible lack of brackets
        if original_jdl.strip()[0] != "[":
            original_jdl = f"[{original_jdl}]"

        class_ad_job = ClassAd(original_jdl)
        if class_ad_job.lookupAttribute("InputSandbox"):
            isb = class_ad_job.getListFromExpression("InputSandbox")
            sandboxes_pfn = await create_sandbox(
                paths=[Path(file_path) for file_path in isb]
            )
            class_ad_job.set_expression("InputSandbox", {sandboxes_pfn})

        jdls[i] = class_ad_job.asJDL()

    # Submit JDL Jobs
    async with AsyncDiracClient() as api:
        jobs = await api.jobs.submit_jdl_jobs(list(jdls))
    print(
        f"Inserted {len(jobs)} jobs with ids: {','.join(map(str, (job.job_id for job in jobs)))}"
    )

This code takes the local ISB from the JDL, creates the sandboxes and uploads them (create_sandbox), then, replaces the local ISB by the sandbox PFN in the JDL, before submitting the updated jobs jdls. Job-ISB assignment isn't there yet, it's still in DIRAC JobSanity for now.
Although I'm not sure it is the right place to do that.

Maybe something like this would be a better approach? (I didn't test it tho):
In cli/jobs.py:

@app.async_command()
async def submit(jdls: list[FileText]):
    # Submit JDL Jobs
    jobs = await submit_jobs(jdls) # from api/jobs.py
    print(
        f"Inserted {len(jobs)} jobs with ids: {','.join(map(str, (job.job_id for job in jobs)))}"
    )

In api.jobs.py:

@with_client
async def submit_jobs(jdls: list[Job], *, client: AsyncDiracClient):
    # Create and upload InputSandboxes from JDLs
    for i, jdl in enumerate(jdls):
        original_jdl = jdl.read()

        # Fix possible lack of brackets
        if original_jdl.strip()[0] != "[":
            original_jdl = f"[{original_jdl}]"

        class_ad_job = ClassAd(original_jdl)
        if class_ad_job.lookupAttribute("InputSandbox"):
            isb = class_ad_job.getListFromExpression("InputSandbox")
            sandboxes_pfn = await create_sandbox(
                paths=[Path(file_path) for file_path in isb]
            )
            class_ad_job.set_expression("InputSandbox", {sandboxes_pfn})

        jdls[i] = class_ad_job.asJDL()

    jobs = await client.jobs.submit_jdl_jobs(list(jdls))
    return jobs

But then, this implementation would only work when calling from the CLI (?).

When using the web app, we'd use the /jobs/jdl endpoint which doesn't go through this part of the code, right? So, the ISB won't be created as sandboxes, uploaded and updated in the job JDL.
That's where it's unclear on how I should create the sandboxes for this particular case.

It would be appreciated if you could explain a bit more what you have in mind.

[fstagni]

Hi,
with

the client, through the APIs, creates the sandbox and uploads it

I meant that the sandbox should be created client side, and then uploaded, but I see you already understood correctly.

The phase of assigning a ISB to the job is right now done by DIRAC through the JobSanity Optimizer (server-side), which is of course something that will need to be re-implemented within DiracX through the tasks (a component that we have yet to properly think about). The sandbox_metadata DB in diracx already has a assign_sandbox_to_jobs call but it is not used anywhere.

I think it is OK to add the submit_jobs implementation in the api. A web app would need to effectively re-do the same steps.

[Stellatsuu]

So, if I understand correctly:

1. What I did in the CLI is good?
   We want to be able to create the sandbox, upload it and update the JDL only from the CLI command? Nowhere else?

2. As I can see, assign_sandbox_to_jobs is already used in assign_sandbox_to_job to assign an OutputSandbox to a job via /{job_id}/sandbox/output.
   Do we also want to handle the InputSandbox in the same way, or should we wait until the tasks component is ready? Handling it the same way would mean transforming the endpoint to /{job_id}/sandbox/{sandbox_type} so that it can assign an InputSandbox to a job when called

Apologies for all the questions, I'm feeling a bit lost here and I'm trying to be sure of what I'm doing and what is expected 🙂

[fstagni]

We want to be able to upload (and download) the ISB and OSB from several places, not only from the CLI. And indeed, if the TransformationSystem is used, effectively the vast majority of those operations are done through automations.
One of the clients that makes ISB uploads is DIRAC: https://github.com/DIRACGrid/DIRAC/blob/integration/src/DIRAC/WorkloadManagementSystem/Service/SandboxStoreHandler.py#L123

So, what you are doing is correct, just I would add the logic in the api rather than the cli.

We should implement the assignment of ISB to jobs too. I think you are right in replacing

@router.patch("/{job_id}/sandbox/output")

With

@router.patch("/{job_id}/sandbox/{sandbox_type}")

since anyway there's already a

@router.get("/{job_id}/sandbox/{sandbox_type}")