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).

Author: daviddavis

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.

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.

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 <name>
 pulp workflow create --name <name> --task '<json>'
+pulp workflow create --name <name> --task '<json>' --dispatch-interval '1 00:00:00'
 pulp workflow cancel --name <name>
 pulp workflow label set --name <name> --key <key> --value <value>
+
+# Inspect the individual runs (executions) of workflows
+pulp workflow run list --workflow <name>
+pulp workflow run show --href <href>
+pulp workflow run cancel --href <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.

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,28 @@ 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")
+    # Runs are canonically served from the nested route
+    # (/workflow/workflows/<workflow_pk>/runs/<pk>/), which is what a run's pulp_href points to
+    # and where retrieve/cancel live. ID_PREFIX and HREF therefore follow the nested viewset.
+    HREF = "workflow_workflows_workflow_run_href"
+    HREF_PATTERN = r"workflow/workflows/[^/]+/runs/[^/]+/"
+    ID_PREFIX = "workflow_workflows_runs"
+    # Listing uses the flat, list-only collection (/workflow/workflow-runs/) so runs can be
+    # scoped across workflows via the `workflow` query filter rather than a path parameter.
+    LIST_ID = "workflow_workflow_runs_list"
+    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"},
+        )

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 <name> 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)

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()

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,47 @@ 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}"
+
+# Test: inspect an individual run via its (nested) href. A periodic workflow starts at
+# start-time (now), so the scheduler creates a run shortly after creation; poll briefly.
+RUN_HREF=""
+for _ in $(seq 1 10); do
+  expect_succ pulp workflow run list --workflow "${PERIODIC_NAME}"
+  RUN_HREF="$(echo "$OUTPUT" | jq -r '.[0].pulp_href // empty')"
+  [ -n "${RUN_HREF}" ] && break
+  sleep 1
+done
+if [ -n "${RUN_HREF}" ]; then
+  # 'run show' must resolve the nested run href against the correct read operation.
+  expect_succ pulp workflow run show --href "${RUN_HREF}"
+  assert "$(echo "$OUTPUT" | jq -r '.pulp_href')" = "${RUN_HREF}"
+fi
+
+# 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 +97,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}"