Transitioning from (Dirac Worklfow, JDL) to (CWL, pydantic JobModel)

[aldbr]

Here is a possible transition plan (mostly prepared by @simon-mazenoux) that can be discussed:

Step1: Now

• Add cwl-tool in DIRACOS2
• Add a /submit endpoint, which would accept a CWL and DIRAC specific arguments (pydantic JobModel). The method would:
  • add the CWL into an input sandbox (might be problematic if there is already an input sandbox defined in the pydantic JobModel)
  • convert the CWL and the JobModel into a JDL such as:

    "Executable": "cwl-tool"
    "Arguments": "job.cwl --message 'Hola'"
    ...

  • the method calls /submit-jdl. Theoretically, DIRAC is flexible enough to handle it without any further modification.

Note: there would be 2 CLI: dirac jobs submit-jdl and dirac jobs submit.

Step2: Once the WMS tasks and routers are all implemented

• Add a Workflow table that will contain the CWL workflows (WorkflowID, Workflow). Instead of being stored in the input sandbox, the CWL will be sent in the Workflow DB.
• Add conditions within the code to know whether a given job is written as a JDL. It allows the components to know where to find information about the jobs.
• The /submit endpoint does not convert the CWL and the DIRAC specific arguments into a JDL anymore:
  • add the CWL into the Workflow table
  • call the executors by itself passing the JobModel

Step3: Once the /submit-jdl path is not used anymore

• Drop the /submit-jdl route
• Remove any reference to JDL and jobDescription.xml within the WMS

Alternatively: the Workflow table from Step2 could be introduced in DIRAC in Step1, to avoid the issue about having multiple input sandboxes for a same job. We have a preference for this option.

Any opinion about this plan?

[aldbr]

Draft of a new model based on CWL (v2)

The presented job submission model is designed to facilitate the execution of CWL (Common Workflow Language) tasks in diverse computing environments, particularly focusing on remote execution scenarios. This model introduces a flexible and efficient way to specify job requirements, including the source of executables and inputs, allowing for seamless integration with various data storage solutions such as local filesystems, sandbox stores, and storage elements (SE).

Model specifications:

• Endpoint: POST /jobs enables:
  • Single job submissions
  • Batch submissions of multiple jobs under the same CWL workflow with varying inputs.
  • Handling inputs/executables from diverse sources:
    • Local resources on the worker node (e.g., /cvmfs)
    • Sandbox environments for on-the-fly resource provisioning
    • Storage Elements (SE) for distributed data access

Key components of the model include:

• CWL Task: A standard, unmodified CWL definition that can be directly used or tested locally.
• Job Description: Metadata and execution preferences such as computing sites, priority, and output specifications.
• Job Parameters: List of detailed specifications for each job, including inputs and, if necessary, sandbox details for fetching executables or additional files.

Here is an example:

{
# CWL workflow
"cwl_task": {
	"cwlVersion": "v1.2",
	"class": "CommandLineTool",
	"label": "My Job",
	"doc": "A simple task.",
	# Requirements
	"requirements": {
		"ResourceRequirement": {
			"ramMin": 500, # Mebibytes
			"coresMax": 1,
		},
		"EnvVarRequirement": {
			"envDef": [
				{"envName": "HELLO", "envValue": "hello"}
			]
		}
	},
	# Command
	"baseCommand": "process.py",
	# Inputs
	"inputs": {
		"configuration": {
			"type": "File",
			"inputBinding": {
				"position": 1
			}
		},
		"run-number": {
			"type": "int",
			"inputBinding": {
				"position": 2
			}
		},
		"max-events": {
			"type": "int",
			"default": 1000,
			"inputBinding": {
				"position": 3
			}
		}
	},
	"outputs": 
	[
		{
			"id": "output",
			"type": "File",
			"outputBinding": {
				"glob": "output.sim"
			}
		}
	]
},
# List of parameters to pass to the CWL
# Each item represents a job
"parameters": [
	# 1st job
	{
		"sandbox": [
			"s3://mybucket/path/to/sandbox1.tar.bz",
		],
		"cwl": {
			"configuration": {
				"class": "File",
				"path": "config1.json",
			},
			"run-number": 123,
			"max-events": 1000,
		}
	},
	# 2nd job
	{
		# We could potentially have another sandbox here
		"sandbox": {
			"url": "s3://mybucket/path/to/sandbox2.tar.bz",
		},
		"cwl": {
			"configuration": {
				"class": "File",
				"path": "lfn:/path/to/config2.json",
			},
			"run-number": 124
		}
	}
],
# This is a common part to all the jobs that will be generated
"job_description": {
	"sites": ["Site1"],
	"cpu_work": 1500, # Should this be included?
	"platform": "x86_64-alma9",
	"priority": 1,
	"outputs": {
		"remote_data": ["output.sim"],
		"remote_se": "SE-USER"
		}
	}
}

Advantages:

• Flexibility: Accommodates a wide range of job configurations, from simple local executions to complex scenarios involving data from multiple sources.
• Simplicity: Maintains the CWL task integrity, allowing users to reuse existing CWL definitions without modifications.
• Efficiency: Reduces redundancy by allowing multiple jobs to reference the same sandbox without repeating details.
• Clarity: Clearly separates the execution environment specifications (job parameters and description) from the CWL task definition.

Considerations:

• Parameters Handling: Specific handling for lfn paths.
• Parametric Job Support: The current model does not support the Parameters/ParametersStart/ParametersStep pattern (see parametric-job-jdl), is this really needed?

This description and proposed model are open for discussion and further enhancements.

Changelog:

• v2:
  • feat: parameters[].sandbox section becomes a list of sandbox ids
  • feat: parameters[].cwl section contains the cwl parameters (this was directly accessible from parameters in v1
  • remove: job_description.jobGroup and type

[aldbr]

Example of a user job submission workflow (v1):

• Create a cwl: job.cwl:

{
	"cwlVersion": "v1.2",
	"class": "CommandLineTool",
	"label": "My Job",
	"doc": "A simple task.",

	"requirements": {
		"ResourceRequirement": {
			"ramMin": 500,
			"coresMax": 1,
		},
		"EnvVarRequirement": {
			"envDef": [
				{"envName": "HELLO", "envValue": "hello"}
			]
		}
	},

	"baseCommand": "process.py",

	"inputs": {
		"configuration": {
			"type": "File",
			"inputBinding": {
				"position": 1
			}
		},
		"run-number": {
			"type": "int",
			"inputBinding": {
				"position": 2
			}
		},
		"max-events": {
			"type": "int",
			"default": 1000,
			"inputBinding": {
				"position": 3
			}
		}
	},
	"outputs": 
	[
		{
			"id": "output",
			"type": "File",
			"outputBinding": {
				"glob": "output.sim"
			}
		}
	]
},

• Create some config files: local1.cwl, local2.cwl (can use cwltool --make-template to generate a first example):

{
	"configuration": {
			"class": "File",
			"path": "config1.json",
	},
	"run-number": 123,
	"max-events": 1000
}

{
	"configuration": {
		"class": "File",
		"path": "config2.json",
	},
	"run-number": 124
}

• Execute it locally with cwltool until it works as expected:

$ cwltool job.cwl local1.cwl

• Once the workflow is ready for large-scale execution, create additional config files (remote1.cwl and remote2.cwl):

{
	"configuration": {
		"class": "File",
		"path": "/cvmfs/repo/path/to/config3.json",
	},
	"run-number": 124
}

{
	"configuration": {
		"class": "File",
		"path": "lfn:/path/to/config4.json",
	},
	"run-number": 124
}

• Submit the jobs:

$ dirac jobs submit job.cwl
--params local1.cwl
--params local2.cwl
--params remote1.cwl
--params remote2.cwl
--site site1

# returns 4 job ids: [8543, 8544, 8545, 8546]

• Here is what happened under the hoods:

  • Check that the job and config files are correctly described with cwltool --validate
  • Extract the command from the baseCommand fields and check whether they are locally available: if yes, they need to be added as inputs
  • Go through the config files and extract the paths from the File fields, and check whether they are locally available: if yes, they need to be added as inputs
  • Create input sandboxes, with the local input files and command that will be needed on the remote computing resources. 3 inputs sandboxes should be created here: one with process.py and config1.json (for local1.cwl), another one with process.py and config2.json (for local2.cwl), and the last one with process.py(remote1.cwl and remote2.cwl)
  • Create a json object gathering everything such as:

   {
   	"cwl_task": {
   		# content of job.cwl
   	},
   	"parameters": [
   		# 1st job
   		{
   			"sandbox": [
   				# sandbox id if generated
   			],
   			"cwl": {
   				# content of local1.cwl
   			}
   		},
   		# 2nd job
   		{
   			# We could potentially have another sandbox here
   			"sandbox": [
   				# sandbox id if generated
   			],
   			"cwl": {
   				# content of local2.cwl					
   			}
   		}
   	  ...
   	],
   	# This is a common part to all the jobs that will be generated
   	"job_description": {
   		"sites": ["Site1"],
   		...
   	}
   }

  • Complete example:

   {
   	"cwl_task": {
   		"cwlVersion": "v1.2",
   		"class": "CommandLineTool",
   		"label": "My Job",
   		"doc": "A simple task.",
   		"requirements": {
   			"ResourceRequirement": {
   				"ramMin": 500, # Mebibytes
   				"coresMax": 1,
   			},
   			"EnvVarRequirement": {
   				"envDef": [
   					{"envName": "HELLO", "envValue": "hello"}
   				]
   			}
   		},
   		"baseCommand": "process.py",
   		"inputs": {
   			"configuration": {
   				"type": "File",
   				"inputBinding": {
   					"position": 1
   				}
   			},
   			"run-number": {
   				"type": "int",
   				"inputBinding": {
   					"position": 2
   				}
   			},
   			"max-events": {
   				"type": "int",
   				"default": 1000,
   				"inputBinding": {
   					"position": 3
   				}
   			}
   		},
   		"outputs": 
   		[
   			{
   				"id": "output",
   				"type": "File",
   				"outputBinding": {
   					"glob": "output.sim"
   				}
   			}
   		]
   	},
   	"parameters": [
   		{
   			"sandbox": [
   				"s3://mybucket/path/to/sandbox1.tar.bz",
   			],
   			"cwl": {
   				"configuration": {
   					"class": "File",
   					"path": "config1.json",
   				},
   				"run-number": 123,
   				"max-events": 1000
   			}
   		},
   		{
   			"sandbox": [
   				"s3://mybucket/path/to/sandbox2.tar.bz",
   			],
   			"cwl": {
   				"configuration": {
   					"class": "File",
   					"path": "config2.json",
   				},
   				"run-number": 124
   			}
   		},
   		{
   			"sandbox": [
   				"s3://mybucket/path/to/sandbox3.tar.bz",
   			],
   			"cwl": {
   				"configuration": {
   					"class": "File",
   					"path": "/cvmfs/repo/path/to/config3.json",
   				},
   				"run-number": 123,
   				"max-events": 1000
   			}
   		},
   		{
   			"sandbox": [
   				"s3://mybucket/path/to/sandbox3.tar.bz",
   			],
   			"cwl": {
   				"configuration": {
   					"class": "File",
   					"path": "lfn:/path/to/config4.json",
   				},
   				"run-number": 124
   			}
   		}
   	],
   	"job_description": {
   		"sites": ["Site1"],
   		"platform": "x86_64-alma9",
   		"priority": 1,
   		"outputs": {
   			"remote_data": ["output.sim"],
   			"remote_se": "SE-USER"
   		}
   	}
   }

  • The json object is submitted to the job manager service, which then creates one job per parameter (so 4), and for each:
    • Extract values from the CWL itself: cores, ram needed to gather similar jobs in the same queues
    • Extract values from the parameters, especially the lfn fields (files might be only available on tape and would need to be staged before submitting the job to remote computing resources).
  • Example of a job once handled by the job manager service:

   {
   	"cwl_task": {
   		"cwlVersion": "v1.2",
   		"class": "CommandLineTool",
   		"label": "My Job",
   		"doc": "A simple task.",
   		"requirements": {
   			"ResourceRequirement": {
   				"ramMin": 500,
   				"coresMax": 1,
   			},
   			"EnvVarRequirement": {
   				"envDef": [
   					{"envName": "HELLO", "envValue": "hello"}
   				]
   			}
   		},
   		"baseCommand": "process.py",
   		"inputs": {
   			"configuration": {
   				"type": "File",
   				"inputBinding": {
   					"position": 1
   				}
   			},
   			"run-number": {
   				"type": "int",
   				"inputBinding": {
   					"position": 2
   				}
   			},
   			"max-events": {
   				"type": "int",
   				"default": 1000,
   				"inputBinding": {
   					"position": 3
   				}
   			}
   		},
   		"outputs": 
   		[
   			{
   				"id": "output",
   				"type": "File",
   				"outputBinding": {
   					"glob": "output.sim"
   				}
   			}
   		]
   	},
   	"parameters": {
   		"sandbox": [
   			"s3://mybucket/path/to/sandbox1.tar.bz",
   		],
   		"cwl": {
   			"configuration": {
   				"class": "File",
   				"path": "config1.json",
   			},
   			"run-number": 123,
   			"max-events": 1000
   		}
   	},
   	"job_description": {
   		"sites": ["Site1"],
   		"platform": "x86_64-alma9",
   		"priority": 1,
   		"outputs": {
   			"remote_data": ["output.sim"],
   			"remote_se": "SE-USER"
   		}
   	}
   }

[arrabito]

Thanks a lot for this transitioning plan. I think that it's a very good plan which allows for a smooth transition.

Here below my comments/questions:

1. About :

Alternatively: the Workflow table from Step2 could be introduced in DIRAC in Step1, to avoid the issue about having multiple input sandboxes for a same job. We have a preference for this option.

I think that it would be better to introduce the Workflow table directly in Step1 since as you said it will simplify the management of sandoboxes.

2. About CWL Task I was wondering how it will be handled the resource requirements part, i.e. ram, core specs. As far as I understand these requirements would be still handled by the same tag mechanism existing in current DIRAC, right? I'm asking this just because, a user might have the impression that he can specify whatever resource requirement and the job will be executed remotely, while no job will be run by DIRAC if there are no queues configured matching the requirements. This is the case also in the current DIRAC when using job.setTag but now I think that for the user is clear this field is specific to DIRAC configuration and remote execution, similarly to job.setDestination, etc.
   Now the model you propose, similar specifications, i.e. resource requirements and computing site are described in 2 different components (CWL Task and Job description) which could be a little bit confusing for the user. On the other hand, since "resource requirement" is part of CWL features it's a good idea to leave the users the possibility to use it. Maybe we could leave it but add to the submit command a matching validation tool which checks if there are eligible remote sites/queues.

3. About cpu_work field in job description, as far as I understand from the current DIRAC, if nothing is specified some default value is used (which can be configured). The advantage of specifying this field is that if it is estimated correctly it allows to use more efficiently the cpu slot reserved by the pilot. If my understanding is correct, I think that this field can stay as optional.

4. Just for my understanding the specification of "remote_data" in job description is meant to work as in the current DIRAC?
   I.e. an automatically built lfn for the output data with the JobID in the directory structure? With also the possibility to specify an outputPath for the lfn? This is fine for me, just to be sure that I've understood correctly.

5. About parametric jobs, I'm not sure that it's really important to support this feature. For this use case we could simply envisage that users will submit productions (thus using the Transformation System) instead of jobs.

[aldbr]

2. About CWL Task I was wondering how it will be handled the resource requirements part, i.e. ram, core specs. As far as I understand these requirements would be still handled by the same tag mechanism existing in current DIRAC, right?

I haven't focused on that aspect yet, but yes I think that CWL resource requirements could be extracted by the JobManager service and then translated into tags.

Now the model you propose, similar specifications, i.e. resource requirements and computing site are described in 2 different components (CWL Task and Job description) which could be a little bit confusing for the user. On the other hand, since "resource requirement" is part of CWL features it's a good idea to leave the users the possibility to use it. Maybe we could leave it but add to the submit command a matching validation tool which checks if there are eligible remote sites/queues.

It seems confusing because we are not used to that distinction but I think it actually makes sense:

• One creates and runs a task (executable with dependencies) on his/her machine. One does not really need to care about describing the resource requirements.
• If one needs to share the task with a community, that's a different story. One now needs to describe the resource requirement so that other users trying it know what to expect (e.g. can this task run on my machine?).
• Now, if one wants to run a task on remote resources through a WMS (like Dirac), one may provide additional WMS-specific details such as the type of task, the priority, the site (though most of these WMS-specific parameters would likely be used by experts users and production managers)...

By having this distinction, nothing prevents the user to switch from a given WMS to Dirac, or vice versa. The workflow with resource requirements would still be valid, and the user would just need to modify the WMS-specific parameters.

3. About cpu_work field in job description, as far as I understand from the current DIRAC, if nothing is specified some default value is used (which can be configured). The advantage of specifying this field is that if it is estimated correctly it allows to use more efficiently the cpu slot reserved by the pilot. If my understanding is correct, I think that this field can stay as optional.

Yes exactly, and yes the field could stay as optional as it is now. This is something that could be worth discussing with the CWL developers by the way.

4. Just for my understanding the specification of "remote_data" in job description is meant to work as in the current DIRAC?
   I.e. an automatically built lfn for the output data with the JobID in the directory structure? With also the possibility to specify an outputPath for the lfn? This is fine for me, just to be sure that I've understood correctly.

Well yes I think so (I just took some Dirac-specific parameters from the JDL that are not related to CWL, and put them in this "job description" section.

5. About parametric jobs, I'm not sure that it's really important to support this feature. For this use case we could simply envisage that users will submit productions (thus using the Transformation System) instead of jobs.

I think it is important because it is actually useful for the Transformation System. Let's say you have a transformation and that the input query returns a very large number of inputs to process. To process them all you have to submit 100,000 jobs at the same time.

• if you don't support parametric jobs, then you have to submit the job one by one. You get network overhead for each submission + what happens if a subset of them fails to be submitted?
• if you support parametric jobs, they are submitted in one go and they are all submitted or not (no partial submission).

[arrabito]

6. For my understanding, the CWL Task should necessary be a 'CommandLineTool' or can be for instance a Workflow?

[aldbr]

For my understanding, the CWL Task should necessary be a 'CommandLineTool' or can be for instance a Workflow?

No it can either be a CommandLineTool or a Workflow, the CWL task will be executed with cwltool under the hoods.

[arrabito]

ok perfect. This makes me think to another question. Is the idea to support all the CWL features so that a user should be able to submit a generic CWL workflow or only a subset of them?

[aldbr]

Well, yes the goal is to support CWL entirely