Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions CHANGES/+periodic-workflows.feature
Original file line number Diff line number Diff line change
@@ -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.
2 changes: 2 additions & 0 deletions CHANGES/pulp-glue-workflow/+periodic-workflows.feature
Original file line number Diff line number Diff line change
@@ -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.
12 changes: 12 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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.
20 changes: 20 additions & 0 deletions pulp-glue-workflow/src/pulp_glue/workflow/context.py
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Expand All @@ -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"},
)
44 changes: 35 additions & 9 deletions src/pulpcore/cli/workflow/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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
Expand All @@ -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)
45 changes: 40 additions & 5 deletions src/pulpcore/cli/workflow/workflow.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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",
Expand Down Expand Up @@ -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:
Expand All @@ -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:
Expand Down Expand Up @@ -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()
Expand Down
31 changes: 22 additions & 9 deletions tests/scripts/pulp_workflow/test_workflow.sh
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand All @@ -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}"
Expand All @@ -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_$$"
Expand All @@ -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}"
Expand Down
Loading