diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 72f93622..82927c04 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -10,7 +10,6 @@ repos: - id: trailing-whitespace - id: end-of-file-fixer - id: check-yaml - exclude: ^test/workflows/test_meta/override_dirac_hints_twice\.yaml$ - id: check-added-large-files - repo: https://github.com/astral-sh/ruff-pre-commit diff --git a/pixi.lock b/pixi.lock index 05759811..18d2eb30 100644 --- a/pixi.lock +++ b/pixi.lock @@ -1382,7 +1382,7 @@ packages: - pypi: ./ name: dirac-cwl-proto version: 0.1.0 - sha256: 9370f5ea4376b0165eae2c0c19c2528b08e088eb39914f8b6a919c7ccdf3eca5 + sha256: cccdf5a03254b3e093f392f662e9754eafc06b239a4c4baf01e14b91c1af0cb2 requires_dist: - cwl-utils - cwlformat diff --git a/pyproject.toml b/pyproject.toml index 71d09b88..12c76722 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -110,8 +110,8 @@ python scripts/generate_schemas.py \ --individual \ --unified && \ echo "Copying schemas to test locations..." && \ -cp generated_schemas/dirac-metadata.json test/workflows/test_meta/schemas/dirac-metadata.json && \ -cp generated_schemas/plugins-summary.json test/workflows/test_meta/schemas/plugins-summary.json +cp generated_schemas/dirac-metadata.json test/schemas/dirac-metadata.json && \ +cp generated_schemas/plugins-summary.json test/schemas/plugins-summary.json """ schemas-yaml = """ @@ -128,8 +128,8 @@ schemas = { depends-on = ["schemas-json", "schemas-yaml"] } clean-schemas = """ echo "Cleaning generated schemas..." && \ rm -rf generated_schemas/ && \ -rm -f test/workflows/test_meta/schemas/dirac-metadata.json \ - test/workflows/test_meta/schemas/plugins-summary.json +rm -f test/schemas/dirac-metadata.json \ + test/schemas/plugins-summary.json """ # Schema validation diff --git a/scripts/generate_schemas.py b/scripts/generate_schemas.py index cc616a51..ca143a17 100644 --- a/scripts/generate_schemas.py +++ b/scripts/generate_schemas.py @@ -47,9 +47,9 @@ def collect_pydantic_models() -> Dict[str, Any]: models.update( { "ExecutionHooksBasePlugin": ExecutionHooksBasePlugin, - "ExecutionHooksHint": ExecutionHooksHint, - "SchedulingHint": SchedulingHint, - "TransformationExecutionHooksHint": TransformationExecutionHooksHint, + "ExecutionHooks": ExecutionHooksHint, + "Scheduling": SchedulingHint, + "TransformationExecutionHooks": TransformationExecutionHooksHint, } ) logger.info("Collected core metadata models") diff --git a/src/dirac_cwl_proto/execution_hooks/core.py b/src/dirac_cwl_proto/execution_hooks/core.py index 04c9d7c3..16835bd0 100644 --- a/src/dirac_cwl_proto/execution_hooks/core.py +++ b/src/dirac_cwl_proto/execution_hooks/core.py @@ -9,7 +9,7 @@ import logging from abc import ABC, abstractmethod from pathlib import Path -from typing import Any, ClassVar, Dict, List, Mapping, Optional, TypeVar, Union +from typing import Any, ClassVar, Dict, List, Mapping, Optional, Self, TypeVar, Union from pydantic import BaseModel, ConfigDict, Field, PrivateAttr @@ -366,7 +366,7 @@ def from_cwl(cls: type[T], cwl_object: Any) -> T: hints = getattr(cwl_object, "hints", []) or [] for hint in hints: - if hint.get("class") == "dirac:scheduling": + if hint.get("class") == "dirac:Scheduling": hint_data = {k: v for k, v in hint.items() if k != "class"} descriptor = descriptor.model_copy(update=hint_data) @@ -407,7 +407,7 @@ def model_copy( update: Optional[Mapping[str, Any]] = None, *, deep: bool = False, - ) -> "ExecutionHooksHint": + ) -> Self: """Enhanced model copy with intelligent merging of dict fields (including configuration).""" if update is None: update = {} @@ -489,12 +489,12 @@ def _dash_to_snake(s: str) -> str: return get_registry().instantiate_plugin(descriptor) @classmethod - def from_cwl(cls, cwl_object: Any) -> "ExecutionHooksHint": + def from_cwl(cls, cwl_object: Any) -> Self: """Extract metadata descriptor from CWL object using Hint interface.""" descriptor = cls() hints = getattr(cwl_object, "hints", []) or [] for hint in hints: - if hint.get("class") == "dirac:execution-hooks": + if hint.get("class") == "dirac:ExecutionHooks": hint_data = {k: v for k, v in hint.items() if k != "class"} descriptor = descriptor.model_copy(update=hint_data) return descriptor diff --git a/src/dirac_cwl_proto/job/__init__.py b/src/dirac_cwl_proto/job/__init__.py index b3d006df..b98e8296 100644 --- a/src/dirac_cwl_proto/job/__init__.py +++ b/src/dirac_cwl_proto/job/__init__.py @@ -28,7 +28,6 @@ from dirac_cwl_proto.submission_models import ( JobInputModel, JobSubmissionModel, - extract_dirac_hints, ) app = AsyncTyper() @@ -80,16 +79,6 @@ async def submit_job_client( return typer.Exit(code=1) console.print(f"\t[green]:heavy_check_mark:[/green] Task {task_path}") - - # Extract and validate dirac hints; unknown hints are logged as warnings. - try: - job_metadata, job_scheduling = extract_dirac_hints(task) - except Exception as exc: - console.print( - f"[red]:heavy_multiplication_x:[/red] [bold]CLI:[/bold] Invalid DIRAC hints:\n{exc}" - ) - return typer.Exit(code=1) - console.print("\t[green]:heavy_check_mark:[/green] Hints") # Extract parameters if any @@ -104,17 +93,6 @@ async def submit_job_client( ) return typer.Exit(code=1) - overrides = parameter.pop("cwltool:overrides", {}) - if overrides: - override_hints = overrides[next(iter(overrides))].get("hints", {}) - if override_hints: - job_scheduling = job_scheduling.model_copy( - update=override_hints.pop("dirac:scheduling", {}) - ) - job_metadata = job_metadata.model_copy( - update=override_hints.pop("dirac:execution-hooks", {}) - ) - # Prepare files for the ISB isb_file_paths = prepare_input_sandbox(parameter) @@ -131,12 +109,7 @@ async def submit_job_client( f"\t[green]:heavy_check_mark:[/green] Parameter {parameter_p}" ) - job = JobSubmissionModel( - task=task, - parameters=parameters, - scheduling=job_scheduling, - execution_hooks=job_metadata, - ) + job = JobSubmissionModel(task=task, parameters=parameters) console.print( "[green]:heavy_check_mark:[/green] [bold]CLI:[/bold] Job(s) validated." ) @@ -175,8 +148,6 @@ def validate_jobs(job: JobSubmissionModel) -> list[JobSubmissionModel]: JobSubmissionModel( task=job.task, parameters=[parameter], - scheduling=job.scheduling, - execution_hooks=job.execution_hooks, ) ) console.print( diff --git a/src/dirac_cwl_proto/job/job_wrapper.py b/src/dirac_cwl_proto/job/job_wrapper.py index 52362b36..938826cc 100644 --- a/src/dirac_cwl_proto/job/job_wrapper.py +++ b/src/dirac_cwl_proto/job/job_wrapper.py @@ -21,6 +21,7 @@ from rich.text import Text from ruamel.yaml import YAML +from dirac_cwl_proto.execution_hooks import ExecutionHooksHint from dirac_cwl_proto.execution_hooks.core import ExecutionHooksBasePlugin from dirac_cwl_proto.submission_models import ( JobInputModel, @@ -161,8 +162,9 @@ def run_job(self, job: JobSubmissionModel) -> bool: logger = logging.getLogger("JobWrapper") # Instantiate runtime metadata from the serializable descriptor and # the job context so implementations can access task inputs/overrides. + job_execution_hooks = ExecutionHooksHint.from_cwl(job.task) self.runtime_metadata = ( - job.execution_hooks.to_runtime(job) if job.execution_hooks else None + job_execution_hooks.to_runtime(job) if job_execution_hooks else None ) # Isolate the job in a specific directory diff --git a/src/dirac_cwl_proto/job/submission_clients.py b/src/dirac_cwl_proto/job/submission_clients.py index efe206f3..eb79f82a 100644 --- a/src/dirac_cwl_proto/job/submission_clients.py +++ b/src/dirac_cwl_proto/job/submission_clients.py @@ -14,6 +14,7 @@ from diracx.client.aio import AsyncDiracClient from rich.console import Console +from dirac_cwl_proto.execution_hooks import SchedulingHint from dirac_cwl_proto.submission_models import JobSubmissionModel console = Console() @@ -169,11 +170,12 @@ def convert_to_jdl(self, job: JobSubmissionModel, sandbox_ids: list[str]) -> str jdl_lines.append("JobName = test;") jdl_lines.append("OutputSandbox = {std.out, std.err};") - if job.scheduling.priority: - jdl_lines.append(f"Priority = {job.scheduling.priority};") + job_scheduling = SchedulingHint.from_cwl(job.task) + if job_scheduling.priority: + jdl_lines.append(f"Priority = {job_scheduling.priority};") - if job.scheduling.sites: - jdl_lines.append(f"Site = {job.scheduling.sites};") + if job_scheduling.sites: + jdl_lines.append(f"Site = {job_scheduling.sites};") jdl_lines.append(f"InputSandbox = {sandbox_ids};") diff --git a/src/dirac_cwl_proto/production/__init__.py b/src/dirac_cwl_proto/production/__init__.py index a4008fae..3e5a8f94 100644 --- a/src/dirac_cwl_proto/production/__init__.py +++ b/src/dirac_cwl_proto/production/__init__.py @@ -4,7 +4,7 @@ import logging from concurrent.futures import ThreadPoolExecutor -from typing import Any, List, Optional +from typing import List, Optional import typer from cwl_utils.pack import pack @@ -18,13 +18,8 @@ ) from rich import print_json from rich.console import Console -from ruamel.yaml import YAML from schema_salad.exceptions import ValidationException -from dirac_cwl_proto.execution_hooks import ( - SchedulingHint, - TransformationExecutionHooksHint, -) from dirac_cwl_proto.submission_models import ( ProductionSubmissionModel, TransformationSubmissionModel, @@ -45,10 +40,6 @@ @app.command("submit") def submit_production_client( task_path: str = typer.Argument(..., help="Path to the CWL file"), - steps_metadata_path: str = typer.Option( - None, - help="Path to metadata file used to generate the transformations (one entry per step)", - ), # Specific parameter for the purpose of the prototype local: Optional[bool] = typer.Option( True, help="Run the job locally instead of submitting it to the router" @@ -78,33 +69,10 @@ def submit_production_client( ) return typer.Exit(code=1) console.print(f"\t[green]:heavy_check_mark:[/green] Task {task_path}") - - # Load the metadata: at this stage, only the structure is validated, not the content - steps_metadata = {} - if steps_metadata_path: - with open(steps_metadata_path, "r") as file: - steps_metadata = YAML(typ="safe").load(file) - - production_step_execution_hooks = {} - production_step_scheduling = {} - for step_name, step_data in steps_metadata.items(): - # Extract metadata and scheduling from step_data - metadata_config = step_data.get("execution-hooks", {}) - scheduling_config = step_data.get("scheduling", {}) - - # Create TransformationExecutionHooksHint with the metadata - production_step_execution_hooks[step_name] = TransformationExecutionHooksHint( - **metadata_config - ) - production_step_scheduling[step_name] = SchedulingHint(**scheduling_config) console.print("\t[green]:heavy_check_mark:[/green] Metadata") # Create the production - transformation = ProductionSubmissionModel( - task=task, - steps_execution_hooks=production_step_execution_hooks, - steps_scheduling=production_step_scheduling, - ) + production = ProductionSubmissionModel(task=task) console.print( "[green]:heavy_check_mark:[/green] [bold]CLI:[/bold] Production validated." ) @@ -113,8 +81,8 @@ def submit_production_client( console.print( "[blue]:information_source:[/blue] [bold]CLI:[/bold] Submitting the production..." ) - print_json(transformation.model_dump_json(indent=4)) - if not submit_production_router(transformation): + print_json(production.model_dump_json(indent=4)) + if not submit_production_router(production): console.print( "[red]:heavy_multiplication_x:[/red] [bold]CLI:[/bold] Failed to run production." ) @@ -170,31 +138,15 @@ def _get_transformations( """ # Create a subworkflow and a transformation for each step transformations = [] + for step in production.task.steps: step_task = _create_subworkflow( step, str(production.task.cwlVersion), production.task.inputs ) - configuration = _get_configuration(production.task) - - # Get the execution_hooks & description for the step - step_id = step.id.split("#")[-1] - step_data: TransformationExecutionHooksHint = ( - production.steps_execution_hooks.get( - step_id, - TransformationExecutionHooksHint(), - ) - ) - step_scheduling: SchedulingHint = production.steps_scheduling.get( - step_id, - SchedulingHint(), - ) - step_data.configuration.update(configuration) transformations.append( TransformationSubmissionModel( task=step_task, - execution_hooks=step_data, - scheduling=step_scheduling, ) ) return transformations @@ -263,16 +215,3 @@ def _create_subworkflow( break return new_workflow - - -def _get_configuration(task: Workflow) -> dict[str, Any]: - """Get the external inputs of a step. - - :param task: The task to get the query params for - - :return: A dictionary of query params - """ - configuration = {} - for input in task.inputs: - configuration[input.id.split("#")[-1]] = input.default - return configuration diff --git a/src/dirac_cwl_proto/submission_models.py b/src/dirac_cwl_proto/submission_models.py index 7933cd1e..2ee59bee 100644 --- a/src/dirac_cwl_proto/submission_models.py +++ b/src/dirac_cwl_proto/submission_models.py @@ -50,8 +50,6 @@ class JobSubmissionModel(BaseModel): task: CommandLineTool | Workflow | ExpressionTool parameters: list[JobInputModel] | None = None - scheduling: SchedulingHint - execution_hooks: ExecutionHooksHint @field_serializer("task") def serialize_task(self, value): @@ -60,6 +58,12 @@ def serialize_task(self, value): else: raise TypeError(f"Cannot serialize type {type(value)}") + @model_validator(mode="before") + def validate_hints(cls, values): + task = values.get("task") + ExecutionHooksHint.from_cwl(task), SchedulingHint.from_cwl(task) + return values + # ----------------------------------------------------------------------------- # Transformation models @@ -73,8 +77,6 @@ class TransformationSubmissionModel(BaseModel): model_config = ConfigDict(arbitrary_types_allowed=True) task: CommandLineTool | Workflow | ExpressionTool - execution_hooks: TransformationExecutionHooksHint - scheduling: SchedulingHint @field_serializer("task") def serialize_task(self, value): @@ -83,6 +85,12 @@ def serialize_task(self, value): else: raise TypeError(f"Cannot serialize type {type(value)}") + @model_validator(mode="before") + def validate_hints(cls, values): + task = values.get("task") + TransformationExecutionHooksHint.from_cwl(task), SchedulingHint.from_cwl(task) + return values + # ----------------------------------------------------------------------------- # Production models @@ -96,29 +104,6 @@ class ProductionSubmissionModel(BaseModel): model_config = ConfigDict(arbitrary_types_allowed=True) task: Workflow - # Key: step name, Value: description & execution_hooks of a transformation - steps_execution_hooks: dict[str, TransformationExecutionHooksHint] - # Key: step name, Value: scheduling configuration for a transformation - steps_scheduling: dict[str, SchedulingHint] = {} - - @model_validator(mode="before") - def validate_steps_metadata(cls, values): - task = values.get("task") - steps_execution_hooks = values.get("steps_execution_hooks") - - if task and steps_execution_hooks: - # Extract the available steps in the task - task_steps = set([step.id.split("#")[-1] for step in task.steps]) - metadata_keys = set(steps_execution_hooks.keys()) - - # Check if all metadata keys exist in the task's workflow steps - missing_steps = metadata_keys - task_steps - if missing_steps: - raise ValueError( - f"The following steps are missing from the task workflow: {missing_steps}" - ) - - return values @field_serializer("task") def serialize_task(self, value): @@ -126,18 +111,3 @@ def serialize_task(self, value): return save(value) else: raise TypeError(f"Cannot serialize type {type(value)}") - - -# ----------------------------------------------------------------------------- -# Module helpers -# ----------------------------------------------------------------------------- - - -def extract_dirac_hints(cwl: Any) -> tuple[ExecutionHooksHint, SchedulingHint]: - """Thin wrapper that returns (ExecutionHooksHint, SchedulingHint). - - Prefer the class-factory APIs `ExecutionHooksHint.from_cwl` and - `SchedulingHint.from_cwl` for new code. This helper remains for - convenience. - """ - return ExecutionHooksHint.from_cwl(cwl), SchedulingHint.from_cwl(cwl) diff --git a/src/dirac_cwl_proto/transformation/__init__.py b/src/dirac_cwl_proto/transformation/__init__.py index 58dda316..4b29f2e5 100644 --- a/src/dirac_cwl_proto/transformation/__init__.py +++ b/src/dirac_cwl_proto/transformation/__init__.py @@ -14,11 +14,9 @@ from cwl_utils.parser.cwl_v1_2 import File from rich import print_json from rich.console import Console -from ruamel.yaml import YAML from schema_salad.exceptions import ValidationException from dirac_cwl_proto.execution_hooks import ( - SchedulingHint, TransformationExecutionHooksHint, ) from dirac_cwl_proto.job import submit_job_router @@ -40,9 +38,6 @@ @app.command("submit") def submit_transformation_client( task_path: str = typer.Argument(..., help="Path to the CWL file"), - metadata_path: Optional[str] = typer.Option( - None, help="Path to metadata file used to generate the input query" - ), # Specific parameter for the purpose of the prototype local: Optional[bool] = typer.Option( True, help="Run the jobs locally instead of submitting them to the router" @@ -73,22 +68,7 @@ def submit_transformation_client( return typer.Exit(code=1) console.print(f"\t[green]:heavy_check_mark:[/green] Task {task_path}") - # Load the metadata: at this stage, only the structure is validated, not the content - metadata_model = TransformationExecutionHooksHint() - if metadata_path: - with open(metadata_path, "r") as file: - metadata = YAML(typ="safe").load(file) - metadata_model = TransformationExecutionHooksHint(**metadata) - console.print("\t[green]:heavy_check_mark:[/green] Metadata") - - transformation_scheduling = SchedulingHint.from_cwl(task) - console.print("\t[green]:heavy_check_mark:[/green] Description") - - transformation = TransformationSubmissionModel( - task=task, - execution_hooks=metadata_model, - scheduling=transformation_scheduling, - ) + transformation = TransformationSubmissionModel(task=task) console.print( "[green]:heavy_check_mark:[/green] [bold]CLI:[/bold] Transformation validated." ) @@ -121,7 +101,7 @@ def submit_transformation_router(transformation: TransformationSubmissionModel) :param transformation: The transformation to start - :return: True if the transformation executed successfully, False otherwise + :return: True if the transformation is executed successfully, False otherwise """ logger = logging.getLogger("TransformationRouter") @@ -134,12 +114,20 @@ def submit_transformation_router(transformation: TransformationSubmissionModel) # - if there is no execution_hooks, the transformation is not waiting for an input and can go on # - if there is execution_hooks, the transformation is waiting for an input job_model_params = [] + + try: + transformation_execution_hooks = TransformationExecutionHooksHint.from_cwl( + transformation.task + ) + except Exception as exc: + raise ValueError(f"Invalid DIRAC hints:\n{exc}") from exc + if ( - transformation.execution_hooks.configuration - and transformation.execution_hooks.group_size + transformation_execution_hooks.configuration + and transformation_execution_hooks.group_size ): # Get the metadata class - transformation_metadata = transformation.execution_hooks.to_runtime( + transformation_metadata = transformation_execution_hooks.to_runtime( transformation ) @@ -147,7 +135,7 @@ def submit_transformation_router(transformation: TransformationSubmissionModel) logger.info("Getting the input data for the transformation...") input_data_dict = {} min_length = None - for input_name, group_size in transformation.execution_hooks.group_size.items(): + for input_name, group_size in transformation_execution_hooks.group_size.items(): # Get input query logger.info(f"\t- Getting input query for {input_name}...") input_query = transformation_metadata.get_input_query(input_name) @@ -177,8 +165,6 @@ def submit_transformation_router(transformation: TransformationSubmissionModel) jobs = JobSubmissionModel( task=transformation.task, parameters=job_model_params, - scheduling=transformation.scheduling, - execution_hooks=transformation.execution_hooks, ) logger.info("Jobs built!") diff --git a/test/workflows/test_meta/schemas/dirac-metadata.json b/test/schemas/dirac-metadata.json similarity index 58% rename from test/workflows/test_meta/schemas/dirac-metadata.json rename to test/schemas/dirac-metadata.json index f945f101..fb99bd58 100644 --- a/test/workflows/test_meta/schemas/dirac-metadata.json +++ b/test/schemas/dirac-metadata.json @@ -1,18 +1,11 @@ { "$defs": { - "ExecutionHooksBasePlugin": { - "description": "Base metadata model for DIRAC jobs", - "dirac_description": "Base metadata model", - "dirac_vo": null, - "properties": {}, - "title": "DIRAC Metadata Model", - "type": "object" - }, - "ExecutionHooksHint": { + "ExecutionHooks": { "additionalProperties": true, "description": "Data management configuration for DIRAC jobs", "properties": { "configuration": { + "additionalProperties": true, "description": "Additional parameters for metadata plugins", "title": "Configuration", "type": "object" @@ -27,10 +20,19 @@ "title": "DIRAC Data Manager", "type": "object" }, + "ExecutionHooksBasePlugin": { + "description": "Base metadata model for DIRAC jobs", + "dirac_description": "Base metadata model", + "dirac_vo": null, + "properties": {}, + "title": "DIRAC Metadata Model", + "type": "object" + }, "JobInputModel": { "description": "Input data and sandbox files for a job execution.", "properties": { "cwl": { + "additionalProperties": true, "title": "Cwl", "type": "object" }, @@ -58,29 +60,11 @@ }, "JobSubmissionModel": { "$defs": { - "ExecutionHooksHint": { - "additionalProperties": true, - "description": "Data management configuration for DIRAC jobs", - "properties": { - "configuration": { - "description": "Additional parameters for metadata plugins", - "title": "Configuration", - "type": "object" - }, - "hook_plugin": { - "default": "QueryBasedPlugin", - "description": "Registry key for the metadata implementation class", - "title": "Hook Plugin", - "type": "string" - } - }, - "title": "DIRAC Data Manager", - "type": "object" - }, "JobInputModel": { "description": "Input data and sandbox files for a job execution.", "properties": { "cwl": { + "additionalProperties": true, "title": "Cwl", "type": "object" }, @@ -105,56 +89,10 @@ ], "title": "JobInputModel", "type": "object" - }, - "SchedulingHint": { - "additionalProperties": false, - "description": "Descriptor for job execution configuration.", - "properties": { - "platform": { - "anyOf": [ - { - "type": "string" - }, - { - "type": "null" - } - ], - "default": null, - "description": "Target platform (e.g., 'DIRAC', 'DIRACX')", - "title": "Platform" - }, - "priority": { - "default": 10, - "description": "Job priority (higher values = higher priority)", - "title": "Priority", - "type": "integer" - }, - "sites": { - "anyOf": [ - { - "items": { - "type": "string" - }, - "type": "array" - }, - { - "type": "null" - } - ], - "default": null, - "description": "Candidate execution sites", - "title": "Sites" - } - }, - "title": "SchedulingHint", - "type": "object" } }, "description": "Job definition sent to the router.", "properties": { - "execution_hooks": { - "$ref": "#/$defs/ExecutionHooksHint" - }, "parameters": { "anyOf": [ { @@ -170,18 +108,13 @@ "default": null, "title": "Parameters" }, - "scheduling": { - "$ref": "#/$defs/SchedulingHint" - }, "task": { "anyOf": [], "title": "Task" } }, "required": [ - "task", - "scheduling", - "execution_hooks" + "task" ], "title": "JobSubmissionModel", "type": "object" @@ -190,28 +123,13 @@ "additionalProperties": false, "description": "Schema for ProductionSubmissionModel with CWL type references", "properties": { - "steps_execution_hooks": { - "additionalProperties": { - "$ref": "#/$defs/TransformationExecutionHooksHint" - }, - "description": "Dictionary mapping strings to TransformationExecutionHooksHint", - "type": "object" - }, - "steps_scheduling": { - "additionalProperties": { - "$ref": "#/$defs/SchedulingHint" - }, - "description": "Dictionary mapping strings to SchedulingHint", - "type": "object" - }, "task": { "$ref": "https://json.schemastore.org/cwl-workflow.json", "description": "CWL Workflow definition" } }, "required": [ - "task", - "steps_execution_hooks" + "task" ], "title": "ProductionSubmissionModel", "type": "object" @@ -270,7 +188,7 @@ "title": "DIRAC Metadata Model", "type": "object" }, - "SchedulingHint": { + "Scheduling": { "additionalProperties": false, "description": "Descriptor for job execution configuration.", "properties": { @@ -313,11 +231,12 @@ "title": "SchedulingHint", "type": "object" }, - "TransformationExecutionHooksHint": { + "TransformationExecutionHooks": { "additionalProperties": true, "description": "Data management configuration for DIRAC jobs", "properties": { "configuration": { + "additionalProperties": true, "description": "Additional parameters for metadata plugins", "title": "Configuration", "type": "object" @@ -349,103 +268,15 @@ "type": "object" }, "TransformationSubmissionModel": { - "$defs": { - "SchedulingHint": { - "additionalProperties": false, - "description": "Descriptor for job execution configuration.", - "properties": { - "platform": { - "anyOf": [ - { - "type": "string" - }, - { - "type": "null" - } - ], - "default": null, - "description": "Target platform (e.g., 'DIRAC', 'DIRACX')", - "title": "Platform" - }, - "priority": { - "default": 10, - "description": "Job priority (higher values = higher priority)", - "title": "Priority", - "type": "integer" - }, - "sites": { - "anyOf": [ - { - "items": { - "type": "string" - }, - "type": "array" - }, - { - "type": "null" - } - ], - "default": null, - "description": "Candidate execution sites", - "title": "Sites" - } - }, - "title": "SchedulingHint", - "type": "object" - }, - "TransformationExecutionHooksHint": { - "additionalProperties": true, - "description": "Data management configuration for DIRAC jobs", - "properties": { - "configuration": { - "description": "Additional parameters for metadata plugins", - "title": "Configuration", - "type": "object" - }, - "group_size": { - "anyOf": [ - { - "additionalProperties": { - "type": "integer" - }, - "type": "object" - }, - { - "type": "null" - } - ], - "default": null, - "description": "Input grouping configuration for transformation jobs", - "title": "Group Size" - }, - "hook_plugin": { - "default": "QueryBasedPlugin", - "description": "Registry key for the metadata implementation class", - "title": "Hook Plugin", - "type": "string" - } - }, - "title": "DIRAC Data Manager", - "type": "object" - } - }, "description": "Transformation definition sent to the router.", "properties": { - "execution_hooks": { - "$ref": "#/$defs/TransformationExecutionHooksHint" - }, - "scheduling": { - "$ref": "#/$defs/SchedulingHint" - }, "task": { "anyOf": [], "title": "Task" } }, "required": [ - "task", - "execution_hooks", - "scheduling" + "task" ], "title": "TransformationSubmissionModel", "type": "object" diff --git a/test/workflows/test_meta/schemas/plugins-summary.json b/test/schemas/plugins-summary.json similarity index 100% rename from test/workflows/test_meta/schemas/plugins-summary.json rename to test/schemas/plugins-summary.json diff --git a/test/test_execution_hooks_core.py b/test/test_execution_hooks_core.py index c86d0019..70c6afc4 100644 --- a/test/test_execution_hooks_core.py +++ b/test/test_execution_hooks_core.py @@ -177,7 +177,7 @@ def test_from_cwl(self, mocker): mock_cwl = mocker.Mock() mock_cwl.hints = [ { - "class": "dirac:execution-hooks", + "class": "dirac:ExecutionHooks", "hook_plugin": "QueryBased", "campaign": "Run3", }, @@ -254,7 +254,7 @@ def test_from_cwl(self, mocker): mock_cwl = mocker.Mock() mock_cwl.hints = [ { - "class": "dirac:scheduling", + "class": "dirac:Scheduling", "platform": "DIRAC-v8", "priority": 8, "sites": ["LCG.CERN.ch"], diff --git a/test/test_workflows.py b/test/test_workflows.py index 7d1b2c22..c986af38 100644 --- a/test/test_workflows.py +++ b/test/test_workflows.py @@ -46,7 +46,7 @@ def pi_test_files(): result_file = job_dir / f"result_{i}.sim" with open(result_file, "w") as f: # Create different sample data for each file - f.write(f"0.{i} 0.{i+1}\n-0.{i+2} 0.{i+3}\n0.{i+4} -0.{i+5}\n") + f.write(f"0.{i} 0.{i + 1}\n-0.{i + 2} 0.{i + 3}\n0.{i + 4} -0.{i + 5}\n") result_files.append(result_file) yield @@ -87,13 +87,7 @@ def pi_test_files(): ], ), # --- Test metadata example --- - # A string input is passed - ( - "test/workflows/test_meta/test_meta.cwl", - [ - "test/workflows/test_meta/override_dirac_hints.yaml", - ], - ), + ("test/workflows/test_meta/test_meta.cwl", []), # --- Crypto example --- # Complete ( @@ -182,14 +176,6 @@ def test_run_job_success(cli_runner, cleanup, pi_test_files, cwl_file, inputs): [], "Recursingintostep", ), - # The configuration file is malformed: the hints are overridden more than once - ( - "test/workflows/test_meta/test_meta.cwl", - [ - "test/workflows/test_meta/override_dirac_hints_twice.yaml", - ], - "Failedtovalidatetheparameter", - ), ], ) def test_run_job_validation_failure( @@ -248,7 +234,7 @@ def test_run_job_parallely(): # This command forces the process 'dirac-cwl' to execute ONLY in # one core of the machine, independently of how many there are - # phisically available. + # physically available. # This simulates a sequential execution of the worklflow. command = [ "taskset", @@ -281,7 +267,7 @@ def test_run_job_parallely(): assert abs(1 - sequential_time / (2 * parallel_time)) < error_margin_percentage, ( "Difference between parallel and sequential time is too large", f"Sequential: {sequential_time} # Parallel: {parallel_time}", - f"Sequential time should be twice the parallel time with an error of {int(error_margin_percentage*100)}%", + f"Sequential time should be twice the parallel time with an error of {int(error_margin_percentage * 100)}%", ) @@ -291,38 +277,30 @@ def test_run_job_parallely(): @pytest.mark.parametrize( - "cwl_file, metadata", + "cwl_file", [ # --- Hello World example --- # There is no input expected - ("test/workflows/helloworld/description_basic.cwl", None), + "test/workflows/helloworld/description_basic.cwl", # --- Crypto example --- # Complete - ("test/workflows/crypto/description.cwl", None), + "test/workflows/crypto/description.cwl", # Caesar only - ("test/workflows/crypto/caesar.cwl", None), + "test/workflows/crypto/caesar.cwl", # ROT13 only - ("test/workflows/crypto/rot13.cwl", None), + "test/workflows/crypto/rot13.cwl", # Base64 only - ("test/workflows/crypto/base64.cwl", None), + "test/workflows/crypto/base64.cwl", # MD5 only - ("test/workflows/crypto/md5.cwl", None), + "test/workflows/crypto/md5.cwl", # --- Pi example --- # Pi simulate transformation - ( - "test/workflows/pi/pisimulate.cwl", - "test/workflows/pi/type_dependencies/transformation/metadata-pi_simulate.yaml", - ), + "test/workflows/pi/pisimulate.cwl", ], ) -def test_run_nonblocking_transformation_success( - cli_runner, cleanup, cwl_file, metadata -): +def test_run_nonblocking_transformation_success(cli_runner, cleanup, cwl_file): # CWL file is the first argument command = ["transformation", "submit", cwl_file] - # Add the metadata file - if metadata: - command.extend(["--metadata-path", metadata]) result = cli_runner.invoke(app, command) clean_output = strip_ansi_codes(result.stdout) @@ -332,13 +310,12 @@ def test_run_nonblocking_transformation_success( @pytest.mark.parametrize( - "cwl_file, metadata, destination_source_input_data", + "cwl_file, destination_source_input_data", [ # --- Pi example --- # Pi gather transformation (waits for simulation result files) ( "test/workflows/pi/pigather.cwl", - "test/workflows/pi/type_dependencies/transformation/metadata-pi_gather.yaml", { "filecatalog/pi/100/input-data": [ ("result_1.sim", "0.1 0.2\n-0.3 0.4\n0.5 -0.6\n"), @@ -352,13 +329,11 @@ def test_run_nonblocking_transformation_success( ], ) def test_run_blocking_transformation_success( - cli_runner, cleanup, cwl_file, metadata, destination_source_input_data + cli_runner, cleanup, cwl_file, destination_source_input_data ): # Define a function to run the transformation command and return the result def run_transformation(): command = ["transformation", "submit", cwl_file] - if metadata: - command.extend(["--metadata-path", metadata]) return cli_runner.invoke(app, command) # Start running the transformation in a separate thread and capture the result @@ -404,46 +379,39 @@ def run_and_capture(): @pytest.mark.parametrize( - "cwl_file, metadata, expected_error", + "cwl_file, expected_error", [ # The description file is malformed: class attribute is unknown ( "test/workflows/malformed_description/description_malformed_class.cwl", - None, "`class`containsundefinedreferenceto", ), # The description file is malformed: baseCommand is unknown ( "test/workflows/malformed_description/description_malformed_command.cwl", - None, "invalidfield`baseComand`", ), # The description file points to a non-existent file (subworkflow) ( "test/workflows/bad_references/reference_doesnotexists.cwl", - [], "Nosuchfileordirectory", ), # The description file points to another file point to it (circular dependency) ( "test/workflows/bad_references/reference_circular1.cwl", - [], "Recursingintostep", ), # The description file points to itself (another circular dependency) ( "test/workflows/bad_references/reference_circular1.cwl", - [], "Recursingintostep", ), ], ) def test_run_transformation_validation_failure( - cli_runner, cwl_file, cleanup, metadata, expected_error + cli_runner, cwl_file, cleanup, expected_error ): command = ["transformation", "submit", cwl_file] - if metadata: - command.extend(["--metadata-path", metadata]) result = cli_runner.invoke(app, command) clean_stdout = strip_ansi_codes(result.stdout) assert ( @@ -497,21 +465,16 @@ def test_run_transformation_validation_failure( @pytest.mark.parametrize( - "cwl_file, metadata", + "cwl_file", [ # --- Crypto example --- # Complete workflow with independent steps (ideal for production mode) - ("test/workflows/crypto/description.cwl", None), + "test/workflows/crypto/description.cwl" ], ) -def test_run_simple_production_success( - cli_runner, cleanup, pi_test_files, cwl_file, metadata -): +def test_run_simple_production_success(cli_runner, cleanup, pi_test_files, cwl_file): # CWL file is the first argument command = ["production", "submit", cwl_file] - # Add the metadata file - if metadata: - command.extend(["--steps-metadata-path", metadata]) result = cli_runner.invoke(app, command) clean_output = strip_ansi_codes(result.stdout) @@ -521,52 +484,44 @@ def test_run_simple_production_success( @pytest.mark.parametrize( - "cwl_file, metadata, expected_error", + "cwl_file, expected_error", [ # The description file is malformed: class attribute is unknown ( "test/workflows/malformed_description/description_malformed_class.cwl", - None, "`class`containsundefinedreferenceto", ), # The description file is malformed: baseCommand is unknown ( "test/workflows/malformed_description/description_malformed_command.cwl", - None, "invalidfield`baseComand`", ), # The description file points to a non-existent file (subworkflow) ( "test/workflows/bad_references/reference_doesnotexists.cwl", - [], "Nosuchfileordirectory", ), # The description file points to another file point to it (circular dependency) ( "test/workflows/bad_references/reference_circular1.cwl", - [], "Recursingintostep", ), # The description file points to itself (another circular dependency) ( "test/workflows/bad_references/reference_circular1.cwl", - [], "Recursingintostep", ), # The workflow is a CommandLineTool instead of a Workflow ( "test/workflows/helloworld/description_basic.cwl", - None, "InputshouldbeaninstanceofWorkflow", ), ], ) def test_run_production_validation_failure( - cli_runner, cleanup, cwl_file, metadata, expected_error + cli_runner, cleanup, cwl_file, expected_error ): command = ["production", "submit", cwl_file] - if metadata: - command.extend(["--steps-metadata-path", metadata]) result = cli_runner.invoke(app, command) clean_stdout = strip_ansi_codes(result.stdout) diff --git a/test/workflows/helloworld/type_dependencies/transformation/metadata-helloworld_with_inputs.yaml b/test/workflows/helloworld/type_dependencies/transformation/metadata-helloworld_with_inputs.yaml deleted file mode 100644 index 4ccf344c..00000000 --- a/test/workflows/helloworld/type_dependencies/transformation/metadata-helloworld_with_inputs.yaml +++ /dev/null @@ -1 +0,0 @@ -hook_plugin: QueryBased diff --git a/test/workflows/pi/description.cwl b/test/workflows/pi/description.cwl index bf989790..930d4b33 100644 --- a/test/workflows/pi/description.cwl +++ b/test/workflows/pi/description.cwl @@ -6,6 +6,12 @@ doc: > It generates random points in a square and calculates how many fall within a unit circle inscribed in the square. +$namespaces: + dirac: "../../schemas/dirac-metadata.json#/$defs/" + +$schemas: + - "../../schemas/dirac-metadata.json" + # Define the inputs of the workflow inputs: num-points: diff --git a/test/workflows/pi/pigather.cwl b/test/workflows/pi/pigather.cwl index 0c189cba..0ecc7f3c 100644 --- a/test/workflows/pi/pigather.cwl +++ b/test/workflows/pi/pigather.cwl @@ -6,6 +6,9 @@ requirements: coresMin: 1 ramMin: 1024 +hints: + $import: "type_dependencies/transformation/metadata-pi_gather.yaml" + inputs: input-data: type: File[] @@ -18,4 +21,4 @@ outputs: outputBinding: glob: "result*.sim" -baseCommand: [pi-gather] +baseCommand: [ pi-gather ] diff --git a/test/workflows/pi/pisimulate.cwl b/test/workflows/pi/pisimulate.cwl index 8d1a6349..444c464c 100644 --- a/test/workflows/pi/pisimulate.cwl +++ b/test/workflows/pi/pisimulate.cwl @@ -6,6 +6,9 @@ requirements: coresMin: 2 ramMin: 1024 +hints: + $import: "type_dependencies/transformation/metadata-pi_simulate.yaml" + inputs: num-points: type: int @@ -19,4 +22,4 @@ outputs: outputBinding: glob: "result*.sim" -baseCommand: [pi-simulate] +baseCommand: [ pi-simulate ] diff --git a/test/workflows/pi/type_dependencies/transformation/metadata-pi_gather.yaml b/test/workflows/pi/type_dependencies/transformation/metadata-pi_gather.yaml index 6c97a61b..1d3eacb8 100644 --- a/test/workflows/pi/type_dependencies/transformation/metadata-pi_gather.yaml +++ b/test/workflows/pi/type_dependencies/transformation/metadata-pi_gather.yaml @@ -1,8 +1,9 @@ -hook_plugin: QueryBasedPlugin -group_size: - input-data: 5 -configuration: - num_points: 100 - query_root: "./filecatalog" - campaign: "pi" - data_type: "100" +- class: dirac:ExecutionHooks + hook_plugin: QueryBasedPlugin + group_size: + input-data: 5 + configuration: + num_points: 100 + query_root: "./filecatalog" + campaign: "pi" + data_type: "100" diff --git a/test/workflows/pi/type_dependencies/transformation/metadata-pi_simulate.yaml b/test/workflows/pi/type_dependencies/transformation/metadata-pi_simulate.yaml index 0631d960..ecc2c045 100644 --- a/test/workflows/pi/type_dependencies/transformation/metadata-pi_simulate.yaml +++ b/test/workflows/pi/type_dependencies/transformation/metadata-pi_simulate.yaml @@ -1,3 +1,4 @@ # Override hints from CWL workflow -configuration: - num_points: 2000 # Override default 1000 from CWL +- class: dirac:ExecutionHooks + configuration: + num_points: 2000 # Override default 1000 from CWL diff --git a/test/workflows/test_meta/override_dirac_hints.yaml b/test/workflows/test_meta/override_dirac_hints.yaml deleted file mode 100644 index 4d787ddc..00000000 --- a/test/workflows/test_meta/override_dirac_hints.yaml +++ /dev/null @@ -1,14 +0,0 @@ -$namespaces: - dirac: "./schemas/dirac-metadata.json#" -cwltool:overrides: - test_meta.cwl#: - hints: - dirac:execution-hooks: - hook_plugin: "QueryBasedPlugin" - configuration: - site: "Paranal" - - dirac:scheduling: - platform: x86_64 - priority: 5 - sites: ["CSCS", "PIC"] diff --git a/test/workflows/test_meta/override_dirac_hints_twice.yaml b/test/workflows/test_meta/override_dirac_hints_twice.yaml deleted file mode 100644 index 9fb3a632..00000000 --- a/test/workflows/test_meta/override_dirac_hints_twice.yaml +++ /dev/null @@ -1,27 +0,0 @@ -$namespaces: - dirac: "./schemas/dirac-metadata.json#" -cwltool:overrides: - test_meta.cwl#: - hints: - dirac:execution-hooks: - hook_plugin: 'QueryBasedPlugin' - configuration: - site: 'Paranal' - - dirac:scheduling: - platform: x86_64 - priority: 5 - sites: ['CSCS', 'PIC'] - -cwltool:overrides: - test_meta.cwl#: - hints: - dirac:execution-hooks: - hook_plugin: 'QueryBasedPlugin' - configuration: - site: 'LaPalma' - - dirac:scheduling: - platform: arm64 - priority: 3 - sites: ['CSCS', 'PIC'] diff --git a/test/workflows/test_meta/test_meta.cwl b/test/workflows/test_meta/test_meta.cwl index 6d39f430..6ed1a147 100644 --- a/test/workflows/test_meta/test_meta.cwl +++ b/test/workflows/test_meta/test_meta.cwl @@ -3,21 +3,26 @@ cwlVersion: v1.2 class: CommandLineTool # The inputs for this process: none. -inputs: [] +inputs: [ ] # The outputs for this process: none. -outputs: [] +outputs: [ ] -baseCommand: ["echo", "Hello World"] +baseCommand: [ "echo", "Hello World" ] $namespaces: - dirac: "./schemas/dirac-metadata.json#" # Generated schema from Pydantic models + dirac: "../../schemas/dirac-metadata.json#/$defs/" # Generated schema from Pydantic models + +$schemas: + - "../../schemas/dirac-metadata.json" + hints: - dirac:execution-hooks: - hook_plugin: "QueryBased" + - class: dirac:ExecutionHooks + hook_plugin: "QueryBasedPlugin" configuration: campaign: PROD5 site: LaPalma - dirac:scheduling: + + - class: dirac:Scheduling platform: x86_64 priority: 10 sites: null