From 7ede2abd1c168e55e2df6368885e0183cae57484 Mon Sep 17 00:00:00 2001 From: David Davis <86290+daviddavis@users.noreply.github.com> Date: Fri, 17 Jul 2026 13:46:45 +0000 Subject: [PATCH] Add periodic workflow and workflow run support Mirror pulp_workflow PR #41 in the CLI/glue: - Add --dispatch-interval to 'workflow create' for recurring workflows. - 'workflow cancel' now stops a workflow (removes schedule, cancels in-flight runs) and is idempotent. - Add PulpWorkflowRunContext and a 'workflow run' command group (list/show/cancel) for the new workflow-runs resource. - Drop the workflow-level 'state' filter (state now lives on runs). --- CHANGES/+periodic-workflows.feature | 4 ++ .../+periodic-workflows.feature | 2 + README.md | 12 +++++ .../src/pulp_glue/workflow/context.py | 20 +++++++++ src/pulpcore/cli/workflow/__init__.py | 44 ++++++++++++++---- src/pulpcore/cli/workflow/workflow.py | 45 ++++++++++++++++--- tests/scripts/pulp_workflow/test_workflow.sh | 31 +++++++++---- 7 files changed, 135 insertions(+), 23 deletions(-) create mode 100644 CHANGES/+periodic-workflows.feature create mode 100644 CHANGES/pulp-glue-workflow/+periodic-workflows.feature diff --git a/CHANGES/+periodic-workflows.feature b/CHANGES/+periodic-workflows.feature new file mode 100644 index 0000000..d14f588 --- /dev/null +++ b/CHANGES/+periodic-workflows.feature @@ -0,0 +1,4 @@ +Added `pulp workflow create --dispatch-interval` to schedule a workflow to re-run on a recurring +interval, and a new `pulp workflow run` command group (`list`, `show`, `cancel`) to inspect and +cancel the individual runs of a workflow. `pulp workflow cancel` now stops a workflow by removing +its schedule and canceling any in-flight runs. diff --git a/CHANGES/pulp-glue-workflow/+periodic-workflows.feature b/CHANGES/pulp-glue-workflow/+periodic-workflows.feature new file mode 100644 index 0000000..1307979 --- /dev/null +++ b/CHANGES/pulp-glue-workflow/+periodic-workflows.feature @@ -0,0 +1,2 @@ +Added `PulpWorkflowRunContext` for the new `workflow-runs` resource and a `dispatch_interval` field +on workflow creation to support periodic (recurring) workflows. diff --git a/README.md b/README.md index 7836606..324fbd3 100644 --- a/README.md +++ b/README.md @@ -8,6 +8,18 @@ A pulp-cli plugin for managing [Pulp workflows](https://github.com/daviddavis/pu pulp workflow list pulp workflow show --name pulp workflow create --name --task '' +pulp workflow create --name --task '' --dispatch-interval '1 00:00:00' pulp workflow cancel --name pulp workflow label set --name --key --value + +# Inspect the individual runs (executions) of workflows +pulp workflow run list --workflow +pulp workflow run show --href +pulp workflow run cancel --href ``` + +A workflow is a definition plus a schedule. Each time its schedule fires, a +`WorkflowRun` records that execution. Set `--dispatch-interval` to re-run a +workflow on a recurring schedule; otherwise it runs once at `--start-time`. +`pulp workflow cancel` stops a workflow (removes its schedule and cancels any +in-flight runs), while `pulp workflow run cancel` cancels a single run. diff --git a/pulp-glue-workflow/src/pulp_glue/workflow/context.py b/pulp-glue-workflow/src/pulp_glue/workflow/context.py index 1894f16..951b036 100644 --- a/pulp-glue-workflow/src/pulp_glue/workflow/context.py +++ b/pulp-glue-workflow/src/pulp_glue/workflow/context.py @@ -15,6 +15,7 @@ class PulpWorkflowContext(PulpEntityContext): ENTITY = _("workflow") ENTITIES = _("workflows") HREF = "workflow_workflow_href" + HREF_PATTERN = r"workflow/workflows/[^/]+/" ID_PREFIX = "workflow_workflows" NEEDS_PLUGINS = [PluginRequirement("workflow")] NULLABLES: t.ClassVar[set[str]] = set() @@ -31,3 +32,22 @@ def cancel(self) -> t.Any: def preprocess_entity(self, body: EntityDefinition, partial: bool = False) -> EntityDefinition: body = super().preprocess_entity(body, partial=partial) return body + + +class PulpWorkflowRunContext(PulpEntityContext): + ENTITY = _("workflow run") + ENTITIES = _("workflow runs") + HREF = "workflow_workflow_run_href" + HREF_PATTERN = r"workflow/workflow-runs/[^/]+/" + ID_PREFIX = "workflow_workflow_runs" + NEEDS_PLUGINS = [PluginRequirement("workflow")] + NULLABLES: t.ClassVar[set[str]] = set() + + CANCEL_ID = "workflow_runs_cancel" + + def cancel(self) -> t.Any: + return self.call( + "cancel", + parameters={self.HREF: self.pulp_href}, + body={"state": "canceled"}, + ) diff --git a/src/pulpcore/cli/workflow/__init__.py b/src/pulpcore/cli/workflow/__init__.py index f781f58..4f3076b 100644 --- a/src/pulpcore/cli/workflow/__init__.py +++ b/src/pulpcore/cli/workflow/__init__.py @@ -10,13 +10,14 @@ name_option, pass_pulp_context, pulp_group, + resource_option, show_command, ) from pulp_glue.common.i18n import get_translation -from pulp_glue.workflow.context import PulpWorkflowContext +from pulp_glue.workflow.context import PulpWorkflowContext, PulpWorkflowRunContext -from pulpcore.cli.workflow.workflow import cancel, create +from pulpcore.cli.workflow.workflow import cancel, create, run_cancel translation = get_translation(__package__) _ = translation.gettext @@ -26,16 +27,28 @@ lookup_options = [href_option, name_option] filter_options = [ click.option("--name"), - click.option( - "--state", - type=click.Choice( - ["waiting", "skipped", "running", "completed", "failed", "canceled"], - case_sensitive=False, - ), - ), label_select_option, ] +state_choice = click.Choice( + ["waiting", "skipped", "running", "completed", "failed", "canceled"], + case_sensitive=False, +) + +workflow_option = resource_option( + "--workflow", + default_plugin="workflow", + default_type="workflow", + context_table={"workflow:workflow": PulpWorkflowContext}, + href_pattern=PulpWorkflowContext.HREF_PATTERN, + help=_("Workflow to filter runs by, in the form or by href."), +) +run_filter_options = [ + workflow_option, + click.option("--state", type=state_choice), +] +run_lookup_options = [href_option] + @pulp_group(name="workflow") @pass_pulp_context @@ -44,10 +57,23 @@ def workflow_group(ctx: click.Context, pulp_ctx: PulpCLIContext, /) -> None: ctx.obj = PulpWorkflowContext(pulp_ctx) +@pulp_group(name="run") +@pass_pulp_context +@click.pass_context +def run_group(ctx: click.Context, pulp_ctx: PulpCLIContext, /) -> None: + ctx.obj = PulpWorkflowRunContext(pulp_ctx) + + +run_group.add_command(list_command(decorators=run_filter_options)) +run_group.add_command(show_command(decorators=run_lookup_options)) +run_group.add_command(run_cancel) + + def mount(main: click.Group, **kwargs: t.Any) -> None: workflow_group.add_command(list_command(decorators=filter_options)) workflow_group.add_command(show_command(decorators=lookup_options)) workflow_group.add_command(label_command(decorators=lookup_options)) workflow_group.add_command(create) workflow_group.add_command(cancel) + workflow_group.add_command(run_group) main.add_command(workflow_group) diff --git a/src/pulpcore/cli/workflow/workflow.py b/src/pulpcore/cli/workflow/workflow.py index 243e833..fe48432 100644 --- a/src/pulpcore/cli/workflow/workflow.py +++ b/src/pulpcore/cli/workflow/workflow.py @@ -14,7 +14,7 @@ from pulp_glue.common.context import DATETIME_FORMATS, PulpEntityContext from pulp_glue.common.i18n import get_translation -from pulp_glue.workflow.context import PulpWorkflowContext +from pulp_glue.workflow.context import PulpWorkflowContext, PulpWorkflowRunContext translation = get_translation(__name__) _ = translation.gettext @@ -27,7 +27,18 @@ "start_time", default=None, type=click.DateTime(formats=DATETIME_FORMATS), - help=_("ISO 8601 datetime for when the workflow should start. Defaults to now."), + help=_("ISO 8601 datetime for when the workflow should first run. Defaults to now."), +) +@click.option( + "--dispatch-interval", + "dispatch_interval", + default=None, + type=click.STRING, + help=_( + "If set, the interval on which the workflow re-runs, creating a new run each time " + "(e.g. '1 00:00:00' for daily or '01:00:00' for hourly). If omitted, the workflow " + "runs exactly once at start-time." + ), ) @click.option( "--task", @@ -55,6 +66,7 @@ def create( /, name: str, start_time: t.Optional[datetime], + dispatch_interval: t.Optional[str], tasks: tuple[str, ...], pulp_labels: tuple[str, ...], ) -> None: @@ -66,6 +78,9 @@ def create( if start_time is not None: body["start_time"] = start_time + if dispatch_interval is not None: + body["dispatch_interval"] = dispatch_interval + if tasks: parsed_tasks = [] for task_json in tasks: @@ -100,14 +115,34 @@ def cancel( entity_ctx: PulpEntityContext, /, ) -> None: - """Cancel a waiting or running workflow.""" + """Stop a workflow. + + Removes the workflow's schedule so no further runs are created and cancels any of its + runs that are still in progress. This is idempotent. + """ assert isinstance(entity_ctx, PulpWorkflowContext) + result = entity_ctx.cancel() + pulp_ctx.output_result(result) + + +@pulp_command(name="cancel") +@href_option +@pass_entity_context +@pass_pulp_context +def run_cancel( + pulp_ctx: PulpCLIContext, + entity_ctx: PulpEntityContext, + /, +) -> None: + """Cancel a waiting or running workflow run.""" + assert isinstance(entity_ctx, PulpWorkflowRunContext) + entity = entity_ctx.entity if entity["state"] not in ("waiting", "running"): raise click.ClickException( - _("Workflow '{name}' is in state '{state}' and cannot be canceled.").format( - name=entity["name"], state=entity["state"] + _("Workflow run '{href}' is in state '{state}' and cannot be canceled.").format( + href=entity["pulp_href"], state=entity["state"] ) ) result = entity_ctx.cancel() diff --git a/tests/scripts/pulp_workflow/test_workflow.sh b/tests/scripts/pulp_workflow/test_workflow.sh index 84626b3..9da26e2 100755 --- a/tests/scripts/pulp_workflow/test_workflow.sh +++ b/tests/scripts/pulp_workflow/test_workflow.sh @@ -8,11 +8,13 @@ set -eu WORKFLOW_NAME="test_cli_workflow_$$" CANCEL_NAME="test_cli_workflow_cancel_$$" FUTURE_NAME="test_cli_workflow_future_$$" +PERIODIC_NAME="test_cli_workflow_periodic_$$" cleanup() { pulp workflow cancel --name "${WORKFLOW_NAME}" 2>/dev/null || true pulp workflow cancel --name "${CANCEL_NAME}" 2>/dev/null || true pulp workflow cancel --name "${FUTURE_NAME}" 2>/dev/null || true + pulp workflow cancel --name "${PERIODIC_NAME}" 2>/dev/null || true } trap cleanup EXIT @@ -36,9 +38,6 @@ assert "$(echo "$OUTPUT" | jq -r '.pulp_labels.test_key')" = "test_value" expect_succ pulp workflow list --name "${WORKFLOW_NAME}" assert "$(echo "$OUTPUT" | jq -r '.[0].name')" = "${WORKFLOW_NAME}" -# Test: list with state filter -expect_succ pulp workflow list --state waiting - # Test: label set expect_succ pulp workflow label set --name "${WORKFLOW_NAME}" --key "env" --value "ci" expect_succ pulp workflow show --name "${WORKFLOW_NAME}" @@ -49,17 +48,31 @@ expect_succ pulp workflow label unset --name "${WORKFLOW_NAME}" --key "env" expect_succ pulp workflow show --name "${WORKFLOW_NAME}" assert "$(echo "$OUTPUT" | jq -r '.pulp_labels | has("env")')" = "false" -# Test: cancel a waiting workflow +# Test: list the runs of a workflow (may be empty, but must be valid JSON) +expect_succ pulp workflow run list --workflow "${WORKFLOW_NAME}" +assert "$OUTPUT" != "null" + +# Test: create a periodic workflow with --dispatch-interval +expect_succ pulp workflow create \ + --name "${PERIODIC_NAME}" \ + --dispatch-interval "01:00:00" \ + --task '{"task_name": "pulpcore.app.tasks.base.general_create", "task_args": [], "task_kwargs": []}' +expect_succ pulp workflow show --name "${PERIODIC_NAME}" +assert "$(echo "$OUTPUT" | jq -r '.name')" = "${PERIODIC_NAME}" +# Stopping a periodic workflow halts its schedule. +expect_succ pulp workflow cancel --name "${PERIODIC_NAME}" + +# Test: stop a workflow scheduled in the future expect_succ pulp workflow create \ --name "${CANCEL_NAME}" \ --task '{"task_name": "pulpcore.app.tasks.base.general_create", "task_args": [], "task_kwargs": []}' \ --start-time "2099-01-01T00:00:00" expect_succ pulp workflow cancel --name "${CANCEL_NAME}" +# Stopping is idempotent: a second stop still succeeds. +expect_succ pulp workflow cancel --name "${CANCEL_NAME}" +# The workflow definition is still readable after being stopped. expect_succ pulp workflow show --name "${CANCEL_NAME}" -assert "$(echo "$OUTPUT" | jq -r '.state')" = "canceled" - -# Test: canceling an already-canceled workflow should fail -expect_fail pulp workflow cancel --name "${CANCEL_NAME}" +assert "$(echo "$OUTPUT" | jq -r '.name')" = "${CANCEL_NAME}" # Test: create with --start-time in the future FUTURE_NAME="test_cli_workflow_future_$$" @@ -68,7 +81,7 @@ expect_succ pulp workflow create \ --task '{"task_name": "pulpcore.app.tasks.base.general_create", "task_args": [], "task_kwargs": []}' \ --start-time "2099-01-01T00:00:00" expect_succ pulp workflow show --name "${FUTURE_NAME}" -assert "$(echo "$OUTPUT" | jq -r '.state')" = "waiting" +assert "$(echo "$OUTPUT" | jq -r '.name')" = "${FUTURE_NAME}" # Clean up the future workflow expect_succ pulp workflow cancel --name "${FUTURE_NAME}"