feat: automatic job grouping#95
Conversation
Stellatsuu
temporarily deployed
to
github-pages
GitHub Actions
Inactive
Stellatsuu
temporarily deployed
to
github-pages
GitHub Actions
Inactive
Stellatsuu
temporarily deployed
to
github-pages
GitHub Actions
Inactive
|
@aldbr Regarding this part of the code: dirac-cwl/src/dirac_cwl_proto/transformation/__init__.py Lines 130 to 163 in 72956d5 Are we planning on keeping it? Just so I un-comment it and make the changes related to the |
|
Waiting on #66 (comment) and #95 (comment) approval about what we're doing, and then, PR should be ready to be fully reviewed (and potentially merged 🙏). |
Yes we want to keep it. A transformation should either get inputs from the CLI, or from a |
|
I’m also not sure whether the Also, the If you have any ideas. |
Since |
As far as I see, I'm not sure that any input_name is needed anymore. In the current QueryBasedPlugin, input_name is just used to build the LFN path, see: Probably we could just change get_input_query to not take any argument and just build LFN path as:
instead of:
Then, I guess that the group_size in yaml file should be specified as: instead of: @aldbr do you agree? (Maybe some other changes are needed that I haven't thought). |
Yes I agree. In any case, this is going to be revised at some point with the hints proposed in #69 |
Stellatsuu
temporarily deployed
to
github-pages
GitHub Actions
Inactive
|
Current PR status:
|
Stellatsuu
temporarily deployed
to
github-pages
GitHub Actions
Inactive
Stellatsuu
temporarily deployed
to
github-pages
GitHub Actions
Inactive
|
I don't know how to fix current lint error. I had to rebase and reword all my "wrong" commits and it's still not working because of old commits already pushed. |
|
Also, PyPi is failling on something I didn't touch directly (mostly sure about that), so I don't really know what to do about that too. If you have any ideas models.update(
{
"JobInputModel": JobInputModel, # <--- error happens here
"JobSubmissionModel": JobSubmissionModel,
"TransformationSubmissionModel": TransformationSubmissionModel,
"ProductionSubmissionModel": ProductionSubmissionModel,
}
)class JobInputModel(BaseModel): # <-- it's a BaseModel ?
"""Input data and sandbox files for a job execution."""
# Allow arbitrary types to be passed to the model
model_config = ConfigDict(arbitrary_types_allowed=True)
sandbox: list[str] | None
cwl: dict[str, Any]
@field_serializer("cwl")
def serialize_cwl(self, value):
"""Serialize CWL object to dictionary.
:param value: CWL object to serialize.
:return: Serialized CWL dictionary.
"""
return save(value) |
Stellatsuu
force-pushed
the
automatic-job-grouping
branch
from
f9337a8 to
52e6144
Compare
# Conflicts: # src/dirac_cwl/job/job_wrapper.py # test/test_integration.py # test/test_job_wrapper.py
aldbr
left a comment
aldbr
left a comment
There was a problem hiding this comment.
Since this PR was opened, there has been some changes in the design of the hints so we should be pragmatic and focus on the user interfaces mostly.
The input_data dictionary maps CWL input parameter names to lists of files. This avoids hardcoding any key name (like input-data) and makes the mapping between hint data and workflow inputs explicit.
CLI convenience: --inputs-file
Users can provide a standard CWL inputs YAML file via the CLI:
dirac-cwl transformation submit workflow.cwl --inputs-file data.yamlWhere data.yaml is:
simulation-files:
- /lfn/path/file1.root
- /lfn/path/file2.root
- /lfn/path/file3.rootThe CLI reads this file and populates dirac:ExecutionHooks.input_data in the hint before submission. This means:
- The router always reads input data from one place (the hint)
- The
--inputs-fileflag is syntactic sugar for populating the hint:
hints:
- class: dirac:ExecutionHooks
group_size: 5
input_data:
simulation-files:
- /lfn/path/file1.root
- /lfn/path/file2.root
- /lfn/path/file3.root- If the hint already contains
input_dataAND--inputs-fileis provided, emit a warning (--input_fileshould overrideinput_data). - The file format is a standard CWL inputs YAML, nothing new to learn. For simplicity, it should contain only 1 parameter I think (unless CTAO has a use case where they would need multiple static lists of inputs?).
Dynamic queries (what we already have with configuration)
For transformations that discover inputs at runtime (e.g., from upstream transformation outputs), the hint uses a plugin-based configuration instead:
hints:
- class: dirac:ExecutionHooks
group_size: 5
input_query:
plugin: QueryBasedPlugin
config:
query_root: "."
campaign: "pi"
data_type: "100"A transformation receives dynamic inputs from one upstream transformation only.
Mutual exclusivity
input_data (static) and input_query (dynamic) cannot coexist on the same transformation. The system should raise a clear error:
Cannot specify both static input data and dynamic input query.
Use input_data (or --inputs-file) for standalone transformations with known files.
Use input_query for transformations that discover inputs from upstream outputs.
Which input gets split?
A transformation receives input data through one file-array input only, whether static (input_data with a single key) or dynamic (input_query from one upstream transformation). group_size applies to that input.
If input_data contains multiple keys, raise an error. Multiple dynamic input sources per transformation are not supported, each transformation queries one upstream transformation only.
Non-file inputs (scalars, non-array types) defined in the workflow's regular CWL inputs are passed unchanged to every job.
Productions
Static inputs and the first transformation convention
When a production is submitted with --inputs-file, the static input data is passed to the first transformation by convention. This follows the natural pipeline structure:
t1 (receives static inputs) -> t2 (queries t1 outputs) -> t3 (queries t2 outputs)
Downstream transformations typically get their inputs via dynamic queries against the DataCatalog/Bookkeeping service.
In that way we start simple. If it's enough to cover CTAO and LHCb use cases, then it's fine. If we need something more complex (multiple input queries/data), then we can reevaluate based on the use cases.
There was a problem hiding this comment.
Looks like that shouldn't have been changed like that (we more and more need to update the lock file from github 😅 )
There was a problem hiding this comment.
Maybe we can add #22 to the current sprint? I don't think it'll take a lot of time (might be a weight 1)
| if os.getenv("DIRAC_PROTO_LOCAL") == "1": | ||
| from dirac_cwl.mocks.sandbox import create_sandbox, download_sandbox # type: ignore[no-redef] | ||
| from dirac_cwl.mocks.status import JobReportMock | ||
| else: | ||
| from diracx.api.jobs import create_sandbox, download_sandbox # type: ignore[no-redef] |
There was a problem hiding this comment.
Why did you move these lines?
There was a problem hiding this comment.
I had issues during tests: #95 (comment)
The problem was that the import weren't working correctly. IIRC, the value of os.getenv wasn't set yet when the code was called in the tests.
So, the JobWrapper would import diracx.api.jobs instead of dirac_cwl.mocks... because os.getenv("DIRAC_PROTO_LOCAL") was != "1" (None or 0 I don't remember)
| @property | ||
| def job_path(self): | ||
| """Return the job path.""" | ||
| return self._job_path |
There was a problem hiding this comment.
Where and why do you need this getter?
There was a problem hiding this comment.
This getter is used in this fixture: conftest.py
@pytest.fixture
def job_wrapper():
"""Create a JobWrapper instance and cleanup test files."""
job_wrapper = JobWrapper(job_id=0)
yield job_wrapper
task_file = job_wrapper.job_path / "task.cwl"
task_file.unlink(missing_ok=True)It’s used to clean up any files created during tests. As mentioned in one of our discussions, running JobWrapper-related tests generates a task.cwl file in the project directory, which isn’t automatically removed. This fixture handles that cleanup and can also provide a job_wrapper instance for use in tests when needed.
I had to create it since now, job_path is a private attribute.
| raise ValueError(f"Invalid DIRAC hints:\n{exc}") from exc | ||
|
|
||
| # Inputs from Transformation inputs_file | ||
| if transformation.input_data: |
There was a problem hiding this comment.
I guess input data should come either from input_data or from the query parameters but not both (I would make if/else instead of if/if).
What do you think?
There was a problem hiding this comment.
So, more like a if, elseif?
So we can keep this condition just in case: transformation_execution_hooks.configuration and transformation_execution_hooks.group_size or it's not needed anymore?
If I understand correctly, input_data is only defined in the hint now (at CLI-level) and shouldn't be part of SubmissionModels? |
|
Outdated question: (I didn’t see the
Edit: input_query: Dict[str, Any] = Field(
default_factory=dict, description="..."
)which would contain the plugin and the configuration, instead of having both: hook_plugin: str = Field(
default="QueryBasedPlugin",
description="Registry key for the metadata implementation class",
) # not sure if we include it in input_query?
# Enhanced fields for submission functionality
configuration: Dict[str, Any] = Field(
default_factory=dict, description="Additional parameters for metadata plugins"
)What do you think? Is it was we're looking for now? This would (potentially) work if hints:
- class: dirac:ExecutionHooks
group_size: 5
hook_plugin: ... # available for any ExecutionHookHints
configuration: ... # available for any ExecutionHookHints
input_query: # available only for TransformationExecutionHookHints
plugin: QueryBasedPlugin # this overides hook_plugin?
config: # this overides configuration?
query_root: "."
campaign: "pi"
data_type: "100"Or maybe |
I'm not sure that I remember correctly, so please @aldbr check my answer: First I would say that input_query is exclusive to TransformationHints as only the Transformation System (TS) makes use of input data query. The Production System does it as well but through the TS. Then about your proposal: I think it can work. Then to answer your specific questions:
I think they can coexist as if I'm not wrong they are used in different contexts:
I hope this answers to your questions, but I would prefer that @aldbr confirm/comment. |
aldbr
force-pushed
the
automatic-job-grouping
branch
from
c94c158 to
3ffa8de
Compare
|
@arrabito Yes I confirm. As the design of the hint has been moving a lot, I prefer not to spend too much time on that PR and therefore I am going to finish it by adding a |
| def submit_production_client( | ||
| task_path: str = typer.Argument(..., help="Path to the CWL file"), | ||
| inputs_file: str | None = typer.Option(None, help="Path to the CWL inputs file"), | ||
| chunk: str | None = typer.Option(None, help="Split an array input into jobs: PARAM=SIZE (e.g., input-data=3)"), |
There was a problem hiding this comment.
Me and @chrisburr have decided this is more trouble than it's worth due to various problems with assuming the site location of input files and creating file groups before optimising for this (essentially you risk creating jobs with impossible input data requirements) . It's better left as a detail for the transformation system to handle
There was a problem hiding this comment.
I will share a finalised diracx CLI design doc when it stabilises
aldbr
force-pushed
the
automatic-job-grouping
branch
from
3ffa8de to
d69260e
Compare
aldbr
force-pushed
the
automatic-job-grouping
branch
from
d69260e to
daf2576
Compare
5 participants
cc @aldbr @arrabito @natthan-pigoux
Closes: #66
Related to: #61
Changes:
input_data: list[str | File]toTransformationSubmissionModelandProductionSubmissionModelinputs-fileparameter to Transformation and Production CLIs:dirac-cwl transformation/production submit file.cwl --inputs-file file.yamlparameter-pathtoinput_files: list[str]in Job CLI:dirac-cwl job submit file.cwl --input-files file1.yaml file2.yaml ...group_sizeexecutionHooksHintto Transformation Workflows, such as:group_sizedetermines the number of jobs to be created and how many inputs files they will contain insubmit_transformation_router, by default, it equals 1, which mean a job will be created for each input in the inputs file. Once the list of jobs is created, it is sent to thejob_routerand processed.JobWrapperrelated tests:task.cwlwas created duringpost_processbut never cleared after running tests.TODO after this PR: