diff --git a/docs/changelog.rst b/docs/changelog.rst index 542a8b7..0e792e3 100644 --- a/docs/changelog.rst +++ b/docs/changelog.rst @@ -4,6 +4,11 @@ Changelog Unreleased ---------- +:octicon:`codescan-checkmark` Rules ++++++++++++++++++++++++++++++++++++ + +* New rule set :doc:`rules/dag` to validate DAG templates and their tasks. + :octicon:`plug` Enhancements ++++++++++++++++++++++++++++ diff --git a/docs/index.rst b/docs/index.rst index 0da3a44..05de4e2 100644 --- a/docs/index.rst +++ b/docs/index.rst @@ -155,6 +155,7 @@ For more information on how to use Tugboat, runs its help command: :caption: Rules rules/cron-workflow + rules/dag rules/fatal-errors rules/manifest-errors rules/step diff --git a/docs/rules/dag.rst b/docs/rules/dag.rst new file mode 100644 index 0000000..ea45597 --- /dev/null +++ b/docs/rules/dag.rst @@ -0,0 +1,398 @@ +DAG Rules (``DAG``) +=================== + +Code ``DAG`` is used for errors related to the `DAG`_ in a `template`_. + +.. _dag: https://argo-workflows.readthedocs.io/en/latest/walk-through/dag/ +.. _template: https://argo-workflows.readthedocs.io/en/latest/fields/#template + +.. DAG1xx duplicated items + +.. rule:: DAG101 Duplicate task names + + The template contains multiple tasks with the same name. + + .. code-block:: yaml + :emphasize-lines: 11,17 + + apiVersion: argoproj.io/v1alpha1 + kind: Workflow + metadata: + generateName: steps- + spec: + entrypoint: hello-hello + templates: + - name: hello-hello + dag: + tasks: + - name: hello + template: print-message + arguments: + parameters: + - name: message + value: "hello-1" + - name: hello + template: print-message + arguments: + parameters: + - name: message + value: "hello-2" + +.. rule:: DAG102 Duplicate input parameters + + The task contains several input parameters (``.arguments.parameters``) that share the same name, + which means the parameter was set multiple times. + + .. code-block:: yaml + :emphasize-lines: 15,17 + + apiVersion: argoproj.io/v1alpha1 + kind: Workflow + metadata: + name: test- + spec: + entrypoint: main + templates: + - name: main + dag: + tasks: + - name: hello + template: print-message + arguments: + parameters: + - name: message + value: "hello" + - name: message + value: "world" + +.. rule:: DAG103 Duplicate input artifacts + + The task includes several input artifacts (``.arguments.artifacts``) that share the same name. + The artifact was set multiple times. + + .. code-block:: yaml + :emphasize-lines: 15,18 + + apiVersion: argoproj.io/v1alpha1 + kind: Workflow + metadata: + name: test- + spec: + entrypoint: main + templates: + - name: main + dag: + tasks: + - name: process-data + template: data-processor + arguments: + artifacts: + - name: input-data + raw: + data: "data-1" + - name: input-data + raw: + data: "data-2" + +.. DAG2xx template reference issues + +.. rule:: DAG201 Self-referencing task + + The task references the current template in the ``template`` field. This may cause an infinite loop. + + Since this can still be intentional, this rule defaults to warning level. + + .. code-block:: yaml + :emphasize-lines: 12 + + apiVersion: argoproj.io/v1alpha1 + kind: Workflow + metadata: + generateName: dag- + spec: + entrypoint: main + templates: + - name: main + dag: + tasks: + - name: process + template: main + +.. rule:: DAG202 Reference to a non-existent template + + The task references a template that does not exist in the workflow. + + .. code-block:: yaml + :emphasize-lines: 12 + + apiVersion: argoproj.io/v1alpha1 + kind: Workflow + metadata: + generateName: dag- + spec: + entrypoint: main + templates: + - name: main + dag: + tasks: + - name: process + template: not-exist-template + +.. DAG3xx argument reference issues + +.. rule:: DAG301 Invalid parameter reference + + A task input parameter contains a reference that cannot be resolved. + This rule builds on :rule:`VAR002` but narrows the report to the task argument where the typo or mismatch occurs. + +.. rule:: DAG302 Invalid artifact reference + + A task input artifact points to an artifact that is not available in the current context. + This can happen with bare references (``inputs.artifacts.foo``) or templated expressions (``{{ inputs.artifacts.foo }}``) when the name is misspelled or simply not defined. + + .. code-block:: yaml + :emphasize-lines: 16 + + apiVersion: argoproj.io/v1alpha1 + kind: Workflow + metadata: + generateName: dag- + spec: + entrypoint: main + templates: + - name: main + dag: + tasks: + - name: process + template: compute + arguments: + artifacts: + - name: data + from: "{{ inputs.artifacts.not-exist }}" + +.. rule:: DAG303 Invalid raw artifact reference + + A task input artifact uses the :py:attr:`~tugboat.schemas.Artifact.raw` block to embed a template expression, but the expression does not resolve to a known parameter. + + Raw artifacts only support parameter references (e.g. ``{{ inputs.parameters.foo }}``); pointing at artifacts leaves the expression unresolved. + + .. code-block:: yaml + :emphasize-lines: 17 + + apiVersion: argoproj.io/v1alpha1 + kind: Workflow + metadata: + generateName: dag- + spec: + entrypoint: main + templates: + - name: main + dag: + tasks: + - name: process + template: compute + arguments: + artifacts: + - name: data + raw: + data: "{{ inputs.artifacts.any }}" + +.. rule:: DAG304 Unexpected argument parameters + + The task provides parameters that the referenced template does not declare. + Remove the extra entries or update the target template so both sides align. + + .. code-block:: yaml + :emphasize-lines: 17 + + apiVersion: argoproj.io/v1alpha1 + kind: Workflow + metadata: + generateName: dag- + spec: + entrypoint: main + templates: + - name: main + dag: + tasks: + - name: process + template: print-message + arguments: + parameters: + - name: message + value: hello + - name: extra-param + value: ignored + + - name: print-message + inputs: + parameters: + - name: message + - name: role + default: user + +.. rule:: DAG305 Missing argument parameters + + The task skips parameters that are required by the referenced template. + Ensure every parameter without a default or static value on the template side is provided by the task. + + .. code-block:: yaml + :emphasize-lines: 11-12 + + apiVersion: argoproj.io/v1alpha1 + kind: Workflow + metadata: + generateName: dag- + spec: + entrypoint: main + templates: + - name: main + dag: + tasks: + - name: process + template: print-message + + - name: print-message + inputs: + parameters: + - name: message + +.. rule:: DAG306 Unexpected argument artifacts + + The task provides artifacts that the referenced template does not declare. + Remove the extra entries or update the target template so both sides align. + + .. code-block:: yaml + :emphasize-lines: 17 + + apiVersion: argoproj.io/v1alpha1 + kind: Workflow + metadata: + generateName: dag- + spec: + entrypoint: main + templates: + - name: main + dag: + tasks: + - name: process + template: data-processor + arguments: + artifacts: + - name: input-data + raw: + data: "data" + - name: extra-artifact + raw: + data: "ignored" + + - name: data-processor + inputs: + artifacts: + - name: input-data + +.. rule:: DAG307 Missing required argument artifacts + + The task skips artifacts that are required by the referenced template. + Ensure every artifact without a default or static value on the template side is provided by the task. + + .. code-block:: yaml + :emphasize-lines: 11-12 + + apiVersion: argoproj.io/v1alpha1 + kind: Workflow + metadata: + generateName: dag- + spec: + entrypoint: main + templates: + - name: main + dag: + tasks: + - name: process + template: data-processor + + - name: data-processor + inputs: + artifacts: + - name: input-data + + +.. STP4xx definition issues + +.. rule:: DAG401 Invalid task definition + + The task definition is invalid according to the schema. + This can happen when fields have the wrong type or unexpected values. + + For example, nested a dag inside a task is not allowed: + + .. code-block:: yaml + :emphasize-lines: 12-14 + + apiVersion: argoproj.io/v1alpha1 + kind: Workflow + metadata: + generateName: dag- + spec: + entrypoint: main + templates: + - name: main + dag: + tasks: + - name: process + inline: + dag: + tasks: [] + + +.. DAG9xx deprecated items + +.. rule:: DAG901 Deprecated Field: ``onExit`` + + The ``onExit`` field in a task definition is deprecated. + + As of Argo Workflows 3.1, ``onExit`` is deprecated by the :py:class:`~tugboat.schemas.DagTask.hooks` field. + This rule is a variant of :rule:`STP901` specific to DAG templates. + + .. code-block:: yaml + :caption: ❌ Example manifest that violates this rule + :emphasize-lines: 18 + + apiVersion: argoproj.io/v1alpha1 + kind: Workflow + metadata: + generateName: dag- + spec: + entrypoint: main + templates: + - name: main + dag: + tasks: + - name: build + onExit: exit + template: run-build + + - name: run-build + container: + image: alpine + + .. code-block:: yaml + :caption: ✅ Example manifest that complies with this rule + :emphasize-lines: 13-15 + + apiVersion: argoproj.io/v1alpha1 + kind: Workflow + metadata: + generateName: dag- + spec: + entrypoint: main + templates: + - name: main + dag: + tasks: + - name: build + template: run-build + hooks: + exit: + template: cleanup diff --git a/docs/rules/template.rst b/docs/rules/template.rst index bb058b3..9cc6d0f 100644 --- a/docs/rules/template.rst +++ b/docs/rules/template.rst @@ -25,7 +25,7 @@ Argo Workflows offers various types of templates. However, Tugboat currently sup * - `DAG template `_ - :octicon:`alert` Partial (:py:class:`~tugboat.schemas.DagTask`) - - :octicon:`x` + - :octicon:`check` :doc:`dag` * - `Data template `_ - :octicon:`x` diff --git a/tests/analyzers/test_dag.py b/tests/analyzers/test_dag.py new file mode 100644 index 0000000..adcf228 --- /dev/null +++ b/tests/analyzers/test_dag.py @@ -0,0 +1,330 @@ +import pytest +from dirty_equals import AnyThing + +from tests.dirty_equals import HasSubstring, IsPartialModel +from tugboat.engine import analyze_yaml_stream + + +def test_analyze_task(diagnoses_logger): + diagnoses = analyze_yaml_stream( + """ + apiVersion: argoproj.io/v1alpha1 + kind: Workflow + metadata: + generateName: test- + spec: + entrypoint: main + templates: + - name: main + dag: + tasks: + - name: task-2 + depends: task-1 + dependencies: + - task-1 + template: test + - name: deprecated + onExit: exit + template: print-message + """ + ) + diagnoses_logger(diagnoses) + + loc = ("spec", "templates", 0, "dag", "tasks", 0) + assert IsPartialModel(code="M201", loc=(*loc, "depends")) in diagnoses + assert IsPartialModel(code="M201", loc=(*loc, "dependencies")) in diagnoses + + +def test_check_argument_parameters(diagnoses_logger): + diagnoses = analyze_yaml_stream( + """ + apiVersion: argoproj.io/v1alpha1 + kind: Workflow + metadata: + generateName: test- + spec: + entrypoint: main + templates: + - name: main + dag: + tasks: + - name: step1 + arguments: + parameters: + - name: param1 + value: "value1" + - name: param1 + value: "value2" + """ + ) + diagnoses_logger(diagnoses) + + loc_prefix = ("spec", "templates", 0, "dag", "tasks", 0, "arguments", "parameters") + assert IsPartialModel(code="DAG102", loc=(*loc_prefix, 0, "name")) in diagnoses + assert IsPartialModel(code="DAG102", loc=(*loc_prefix, 1, "name")) in diagnoses + + +def test_check_argument_parameter_fields(diagnoses_logger): + diagnoses = analyze_yaml_stream( + """ + apiVersion: argoproj.io/v1alpha1 + kind: Workflow + metadata: + generateName: test- + spec: + entrypoint: main + templates: + - name: main + dag: + tasks: + - name: task1 + template: compute + arguments: + parameters: + - name: message + valueFrom: + path: /tmp/message + - name: invalid-ref + value: "{{ workflow.invalid }}" + - name: "" + value: foo + - name: both + value: baz + valueFrom: + parameter: message + - name: compute + container: + image: alpine:3.19 + """ + ) + diagnoses_logger(diagnoses) + + loc_prefix = ("spec", "templates", 0, "dag", "tasks", 0, "arguments", "parameters") + assert ( + IsPartialModel(code="M102", loc=(*loc_prefix, 0, "valueFrom", "path")) + in diagnoses + ) + assert IsPartialModel(code="M101", loc=(*loc_prefix, 0, "valueFrom")) in diagnoses + assert IsPartialModel(code="DAG301", loc=(*loc_prefix, 1, "value")) in diagnoses + assert IsPartialModel(code="M202", loc=(*loc_prefix, 2, "name")) in diagnoses + assert IsPartialModel(code="M201", loc=(*loc_prefix, 3, "value")) in diagnoses + assert IsPartialModel(code="M201", loc=(*loc_prefix, 3, "valueFrom")) in diagnoses + + +def test_check_argument_parameters_usage(diagnoses_logger): + diagnoses = analyze_yaml_stream( + """ + apiVersion: argoproj.io/v1alpha1 + kind: Workflow + metadata: + generateName: test- + spec: + entrypoint: main + templates: + - name: main + dag: + tasks: + - name: hello + template: print-message + arguments: + parameters: + - name: role + value: admin + - name: extra-param + value: blah + - name: print-message + inputs: + parameters: + - name: message + - name: role + default: user + """ + ) + diagnoses_logger(diagnoses) + + loc_prefix = ("spec", "templates", 0, "dag", "tasks", 0, "arguments", "parameters") + assert IsPartialModel(code="DAG304", loc=(*loc_prefix, 1, "name")) in diagnoses + assert IsPartialModel(code="DAG305", loc=loc_prefix) in diagnoses + + +def test_check_argument_artifacts(diagnoses_logger): + diagnoses = analyze_yaml_stream( + """ + apiVersion: argoproj.io/v1alpha1 + kind: Workflow + metadata: + generateName: test- + spec: + entrypoint: main + templates: + - name: main + dag: + tasks: + - name: step1 + template: compute + arguments: + artifacts: + - name: artifact1 + from: "{{ workflow.invalid }}" + - name: artifact1 + raw: + data: "{{ workflow.invalid }}" + """ + ) + diagnoses_logger(diagnoses) + + loc_prefix = ("spec", "templates", 0, "dag", "tasks", 0, "arguments", "artifacts") + assert IsPartialModel(code="DAG103", loc=(*loc_prefix, 0, "name")) in diagnoses + assert IsPartialModel(code="DAG103", loc=(*loc_prefix, 1, "name")) in diagnoses + assert IsPartialModel(code="DAG302", loc=(*loc_prefix, 0, "from")) in diagnoses + assert ( + IsPartialModel(code="DAG303", loc=(*loc_prefix, 1, "raw", "data")) in diagnoses + ) + + +def test_check_argument_artifact_usage(diagnoses_logger): + diagnoses = analyze_yaml_stream( + """ + apiVersion: argoproj.io/v1alpha1 + kind: Workflow + metadata: + generateName: test- + spec: + entrypoint: main + templates: + - name: main + dag: + tasks: + - name: hello + template: print-message + arguments: + artifacts: + - name: role + from: "{{ workflow.artifact-role }}" + - name: extra-artifact + from: "{{ workflow.artifact-extra }}" + + - name: print-message + inputs: + artifacts: + - name: message + - name: role + optional: true + """ + ) + diagnoses_logger(diagnoses) + + loc_prefix = ("spec", "templates", 0, "dag", "tasks", 0, "arguments", "artifacts") + + assert IsPartialModel(code="DAG306", loc=(*loc_prefix, 1, "name")) in diagnoses + assert IsPartialModel(code="DAG307", loc=loc_prefix) in diagnoses + + +def test_check_referenced_template(caplog: pytest.LogCaptureFixture, diagnoses_logger): + with caplog.at_level("DEBUG", "tugboat.analyzers.dag"): + diagnoses = analyze_yaml_stream( + """ + apiVersion: argoproj.io/v1alpha1 + kind: WorkflowTemplate + metadata: + name: demo + spec: + templates: + - name: main + dag: + tasks: + - name: loop + template: main + + - name: invalid-reference + dag: + tasks: + - name: missing + templateRef: + name: demo + template: not-exist-template + - name: external + templateRef: + name: another-workflow + template: whatever + + - name: another + suspend: {} + """ + ) + + diagnoses_logger(diagnoses) + + assert ( + IsPartialModel( + code="DAG201", + loc=("spec", "templates", 0, "dag", "tasks", 0, "template"), + ) + in diagnoses + ) + assert ( + IsPartialModel( + code="DAG202", + loc=( + "spec", + "templates", + 1, + "dag", + "tasks", + 0, + "templateRef", + "template", + ), + msg=( + AnyThing + & HasSubstring( + "Template 'not-exist-template' does not exist in the workflow." + ) + & HasSubstring("Available templates: 'another' or 'main'") + ), + ) + in diagnoses + ) + assert ( + "Task 'external': Referenced template 'another-workflow' is not the same as current workflow 'demo'. Skipping." + in caplog.text + ) + + +def test_check_inline_template(diagnoses_logger): + diagnoses = analyze_yaml_stream( + """ + apiVersion: argoproj.io/v1alpha1 + kind: Workflow + metadata: + generateName: test- + spec: + entrypoint: main + templates: + - name: main + dag: + tasks: + - name: task1 + inline: + dag: + tasks: [] + """ + ) + diagnoses_logger(diagnoses) + + assert ( + IsPartialModel( + code="DAG401", + loc=( + "spec", + "templates", + 0, + "dag", + "tasks", + 0, + "inline", + "dag", + ), + ) + in diagnoses + ) diff --git a/tests/analyzers/test_template.py b/tests/analyzers/test_template.py index 81cea41..d6a6732 100644 --- a/tests/analyzers/test_template.py +++ b/tests/analyzers/test_template.py @@ -74,6 +74,49 @@ def test_check_duplicate_step_names(self, diagnoses_logger): in diagnoses ) + def test_check_duplicate_task_names(self, diagnoses_logger): + diagnoses = analyze_yaml_stream( + """ + apiVersion: argoproj.io/v1alpha1 + kind: Workflow + metadata: + generateName: steps- + spec: + entrypoint: hello-hello + templates: + - name: hello-hello + dag: + tasks: + - name: hello + template: print-message + arguments: + parameters: + - name: message + value: "hello-1" + - name: hello + template: print-message + arguments: + parameters: + - name: message + value: "hello-2" + """ + ) + diagnoses_logger(diagnoses) + assert ( + IsPartialModel( + code="DAG101", + loc=("spec", "templates", 0, "dag", "tasks", 0, "name"), + ) + in diagnoses + ) + assert ( + IsPartialModel( + code="DAG101", + loc=("spec", "templates", 0, "dag", "tasks", 1, "name"), + ) + in diagnoses + ) + MANIFEST_AMBIGUOUS_TYPE = """ apiVersion: argoproj.io/v1alpha1 diff --git a/tugboat/analyzers/dag.py b/tugboat/analyzers/dag.py new file mode 100644 index 0000000..2ad33cb --- /dev/null +++ b/tugboat/analyzers/dag.py @@ -0,0 +1,218 @@ +from __future__ import annotations + +import logging +import typing + +from rapidfuzz.process import extractOne + +import tugboat.analyzers.step as step_analyzer +from tugboat.constraints import mutually_exclusive +from tugboat.core import hookimpl +from tugboat.references import get_task_context +from tugboat.types import Field +from tugboat.utils import find_duplicate_names, join_with_or, prepend_loc + +if typing.TYPE_CHECKING: + from collections.abc import Iterable + + from tugboat.schemas import DagTask, Template, Workflow, WorkflowTemplate + from tugboat.types import Diagnosis + + type WorkflowCompatible = Workflow | WorkflowTemplate + +logger = logging.getLogger(__name__) + + +@hookimpl +def analyze_task( + task: DagTask, template: Template, workflow: Workflow | WorkflowTemplate +) -> Iterable[Diagnosis]: + yield from mutually_exclusive( + task, + fields=["template", "templateRef", "inline"], + require_one=True, + ) + yield from mutually_exclusive( + task, + fields=["withItems", "withParam", "withSequence"], + ) + yield from mutually_exclusive( + task, + fields=["depends", "dependencies"], + ) + + if task.onExit: + yield { + "type": "warning", + "code": "DAG901", + "loc": ("onExit",), + "summary": "Deprecated field", + "msg": "Field 'onExit' is deprecated. Please use 'hooks[exit].template' instead.", + "input": Field("onExit"), + } + + +@hookimpl(specname="analyze_task") +def check_argument_parameters( + task: DagTask, template: Template, workflow: WorkflowCompatible +) -> Iterable[Diagnosis]: + if not task.arguments: + return + if not task.arguments.parameters: + return + + # report duplicate names + for idx, name in find_duplicate_names(task.arguments.parameters): + yield { + "code": "DAG102", + "loc": ("arguments", "parameters", idx, "name"), + "summary": "Duplicate parameter name", + "msg": f"Parameter name '{name}' is duplicated.", + "input": name, + } + + # check fields for each parameter + ctx = get_task_context(workflow, template, task) + + for idx, param in enumerate(task.arguments.parameters): + for diag in step_analyzer.check_argument_parameter_fields(param, ctx): + diag["loc"] = ["arguments", "parameters", idx, *diag["loc"]] + + match diag["code"]: + case "STP301": + diag["code"] = "DAG301" + yield diag + + +@hookimpl(specname="analyze_task") +def check_argument_artifacts( + task: DagTask, template: Template, workflow: WorkflowCompatible +) -> Iterable[Diagnosis]: + if not task.arguments: + return + if not task.arguments.artifacts: + return + + # report duplicate names + for idx, name in find_duplicate_names(task.arguments.artifacts): + yield { + "code": "DAG103", + "loc": ("arguments", "artifacts", idx, "name"), + "summary": "Duplicate artifact name", + "msg": f"Artifact name '{name}' is duplicated.", + "input": name, + } + + # check fields for each parameter + ctx = get_task_context(workflow, template, task) + + for idx, artifact in enumerate(task.arguments.artifacts): + for diag in step_analyzer.check_argument_artifact_fields(artifact, ctx): + diag["loc"] = ["arguments", "artifacts", idx, *diag["loc"]] + + match diag["code"]: + case "STP302": + diag["code"] = "DAG302" + case "STP303": + diag["code"] = "DAG303" + yield diag + + +@hookimpl(specname="analyze_task") +def check_argument_parameters_usage( + task: DagTask, workflow: WorkflowCompatible +) -> Iterable[Diagnosis]: + for diag in step_analyzer.check_argument_parameters_usage(task, workflow): + match diag["code"]: + case "STP304": + diag["code"] = "DAG304" + case "STP305": + diag["code"] = "DAG305" + yield diag + + +@hookimpl(specname="analyze_task") +def check_argument_artifact_usage( + task: DagTask, workflow: WorkflowCompatible +) -> Iterable[Diagnosis]: + for diag in step_analyzer.check_argument_artifact_usage(task, workflow): + match diag["code"]: + case "STP306": + diag["code"] = "DAG306" + case "STP307": + diag["code"] = "DAG307" + yield diag + + +@hookimpl(specname="analyze_task") +def check_referenced_template( + task: DagTask, template: Template, workflow: WorkflowCompatible +) -> Iterable[Diagnosis]: + if task.template: + yield from prepend_loc( + ("template",), + _check_referenced_template(task.template, template, workflow), + ) + + elif task.templateRef: + if task.templateRef.name == workflow.metadata.name: + yield from prepend_loc( + ("templateRef", "template"), + _check_referenced_template( + task.templateRef.template, template, workflow + ), + ) + else: + logger.debug( + "Task '%s': Referenced template '%s' is not the same as current workflow '%s'. Skipping.", + task.name, + task.templateRef.name, + workflow.name, + ) + + +def _check_referenced_template( + target_template_name: str, template: Template, workflow: WorkflowCompatible +) -> Iterable[Diagnosis]: + if target_template_name == template.name: + yield { + "type": "warning", + "code": "DAG201", + "loc": (), + "summary": "Self-referencing", + "msg": "Self-referencing may cause infinite recursion.", + "input": target_template_name, + } + + if target_template_name not in workflow.template_dict: + templates = set(workflow.template_dict) + templates -= {template.name} + + suggestion = None + if result := extractOne(target_template_name, templates): + suggestion, _, _ = result + + yield { + "code": "DAG202", + "loc": (), + "summary": "Template not found", + "msg": ( + f""" + Template '{target_template_name}' does not exist in the workflow. + Available templates: {join_with_or(templates)} + """ + ), + "input": target_template_name, + "fix": suggestion, + } + + +@hookimpl(specname="analyze_task") +def check_inline_template( + task: DagTask, workflow: WorkflowCompatible +) -> Iterable[Diagnosis]: + for diagnosis in step_analyzer.check_inline_template(task, workflow): + match diagnosis["code"]: + case "STP401": + diagnosis["code"] = "DAG401" + yield diagnosis diff --git a/tugboat/analyzers/step.py b/tugboat/analyzers/step.py index 25b1870..eeeab5e 100644 --- a/tugboat/analyzers/step.py +++ b/tugboat/analyzers/step.py @@ -26,6 +26,7 @@ from tugboat.references import Context from tugboat.schemas import ( Artifact, + DagTask, Parameter, Step, Template, @@ -34,6 +35,7 @@ ) from tugboat.types import Diagnosis + type TaskCompatible = DagTask | Step type WorkflowCompatible = Workflow | WorkflowTemplate logger = logging.getLogger(__name__) @@ -86,11 +88,11 @@ def check_argument_parameters( for idx, param in enumerate(step.arguments.parameters or ()): yield from prepend_loc( ("arguments", "parameters", idx), - _check_argument_parameter_fields(param, ctx), + check_argument_parameter_fields(param, ctx), ) -def _check_argument_parameter_fields( +def check_argument_parameter_fields( param: Parameter, context: Context ) -> Iterable[Diagnosis]: yield from require_all(param, fields=["name"]) @@ -134,7 +136,7 @@ def _check_argument_parameter_fields( @hookimpl(specname="analyze_step") def check_argument_parameters_usage( - step: Step, workflow: WorkflowCompatible + step: TaskCompatible, workflow: WorkflowCompatible ) -> Iterable[Diagnosis]: # early exit: referenced template not found ref_template = _get_template_by_ref(step, workflow) @@ -189,7 +191,9 @@ def check_argument_parameters_usage( } -def _get_template_by_ref(step: Step, workflow: WorkflowCompatible) -> Template | None: +def _get_template_by_ref( + step: TaskCompatible, workflow: WorkflowCompatible +) -> Template | None: if step.template: return workflow.template_dict.get(step.template) if step.templateRef and step.templateRef.name == workflow.metadata.name: @@ -220,11 +224,11 @@ def check_argument_artifacts( for idx, artifact in enumerate(step.arguments.artifacts or []): yield from prepend_loc( ("arguments", "artifacts", idx), - _check_argument_artifact_fields(artifact, ctx), + check_argument_artifact_fields(artifact, ctx), ) -def _check_argument_artifact_fields( +def check_argument_artifact_fields( artifact: Artifact, context: Context ) -> Iterable[Diagnosis]: yield from require_all( @@ -318,7 +322,7 @@ def _check_argument_artifact_fields( @hookimpl(specname="analyze_step") def check_argument_artifact_usage( - step: Step, workflow: WorkflowCompatible + step: TaskCompatible, workflow: WorkflowCompatible ) -> Iterable[Diagnosis]: # early exit: referenced template not found ref_template = _get_template_by_ref(step, workflow) @@ -463,7 +467,7 @@ def check_fields_references( @hookimpl(specname="analyze_step") def check_inline_template( - step: Step, workflow: WorkflowCompatible + step: TaskCompatible, workflow: WorkflowCompatible ) -> Iterable[Diagnosis]: if not step.inline: return diff --git a/tugboat/analyzers/template/__init__.py b/tugboat/analyzers/template/__init__.py index 5368f9d..ef6e2de 100644 --- a/tugboat/analyzers/template/__init__.py +++ b/tugboat/analyzers/template/__init__.py @@ -74,6 +74,38 @@ def check_steps( ) +@hookimpl(specname="analyze_template") +def check_dag(template: Template, workflow: WorkflowCompatible) -> Iterable[Diagnosis]: + if not template.dag: + return + + # check for duplicate task names + task_names = collections.defaultdict(list) + for idx, task in enumerate(template.dag.tasks or ()): + if task.name: + task_names[task.name].append(("dag", "tasks", idx, "name")) + + for name, locs in task_names.items(): + if len(locs) > 1: + for loc in locs: + yield { + "code": "DAG101", + "loc": loc, + "summary": "Duplicate task name", + "msg": f"Task name '{name}' is duplicated.", + "input": name, + } + + # investigate tasks + pm = get_plugin_manager() + + for idx, task in enumerate(template.dag.tasks or ()): + yield from prepend_loc.from_iterables( + ["dag", "tasks", idx], + pm.hook.analyze_task(task=task, template=template, workflow=workflow), + ) + + @hookimpl(specname="analyze_template") def check_metrics( template: Template, workflow: WorkflowCompatible diff --git a/tugboat/core.py b/tugboat/core.py index 64e22ed..d88b536 100644 --- a/tugboat/core.py +++ b/tugboat/core.py @@ -23,6 +23,7 @@ def get_plugin_manager() -> pluggy.PluginManager: # built-in hook implementations import tugboat.analyzers.container import tugboat.analyzers.cron_workflow + import tugboat.analyzers.dag import tugboat.analyzers.step import tugboat.analyzers.template import tugboat.analyzers.template.inputs @@ -33,6 +34,7 @@ def get_plugin_manager() -> pluggy.PluginManager: pm.register(tugboat.analyzers.container) pm.register(tugboat.analyzers.cron_workflow) + pm.register(tugboat.analyzers.dag) pm.register(tugboat.analyzers.step) pm.register(tugboat.analyzers.template) pm.register(tugboat.analyzers.template.inputs) diff --git a/tugboat/hookspecs/workflow.py b/tugboat/hookspecs/workflow.py index 473cf85..4abaf0b 100644 --- a/tugboat/hookspecs/workflow.py +++ b/tugboat/hookspecs/workflow.py @@ -15,7 +15,7 @@ if typing.TYPE_CHECKING: from collections.abc import Iterable - from tugboat.schemas import Step, Template, Workflow, WorkflowTemplate + from tugboat.schemas import DagTask, Step, Template, Workflow, WorkflowTemplate from tugboat.types import Diagnosis @@ -84,3 +84,25 @@ def analyze_step( workflow : Workflow | WorkflowTemplate The workflow that the template is part of. """ + + +@hookspec +def analyze_task( + task: DagTask, template: Template, workflow: Workflow | WorkflowTemplate +) -> Iterable[Diagnosis]: # type: ignore[reportReturnType] + """ + Analyze a DAG template task. + + For issues reported by this hook, the ``loc`` attribute of the diagnosis + should start from the ``task`` object itself. The framework will manage + the outer structure and its location path. + + Parameters + ---------- + task : DagTask + The task to analyze. + template : Template + The template that the task is part of. + workflow : Workflow | WorkflowTemplate + The workflow that the template is part of. + """