From 281bff26deef231a59e3692a9f918589af660f2d Mon Sep 17 00:00:00 2001 From: mayankpande88 Date: Sun, 28 Jun 2026 15:11:52 +0530 Subject: [PATCH 1/2] feat: add cleanup_evicted_and_failed_pods template Mirror the net-new "Clean Up Evicted and Failed Pods" system template seeded by api-server migration V760. It deletes Evicted/Failed/Succeeded pods across a namespace (list -> approve -> delete -> verify) to reclaim resources after node-pressure evictions or completed batch runs. Brings the repo back to parity with the migration-seeded template set; V708 and V712 templates were already present. --- manifest.yaml | 3 + .../cleanup_evicted_and_failed_pods.yaml | 79 +++++++++++++++++++ 2 files changed, 82 insertions(+) create mode 100644 templates/cleanup_evicted_and_failed_pods.yaml diff --git a/manifest.yaml b/manifest.yaml index d0bc0a9..8ce21d2 100644 --- a/manifest.yaml +++ b/manifest.yaml @@ -1,4 +1,7 @@ # Generated by scripts/gen_manifest.py — do not edit by hand. +- slug: cleanup_evicted_and_failed_pods + path: templates/cleanup_evicted_and_failed_pods.yaml + sha256: d9952e67573780a7db443bd9db5ba593fe6dc8e7c9968cf674a35d5c7fe743d5 - slug: cordon_and_drain_kubernetes_node path: templates/cordon_and_drain_kubernetes_node.yaml sha256: 6af890543636e15289a31a135b8b1b7f029a9bd73596eb4016d4b63b04729901 diff --git a/templates/cleanup_evicted_and_failed_pods.yaml b/templates/cleanup_evicted_and_failed_pods.yaml new file mode 100644 index 0000000..6613d56 --- /dev/null +++ b/templates/cleanup_evicted_and_failed_pods.yaml @@ -0,0 +1,79 @@ +slug: cleanup_evicted_and_failed_pods +name: "Clean Up Evicted and Failed Pods" +description: "Delete Evicted, Failed, and Succeeded pods across a namespace to reclaim resources and reduce clutter. Useful after node-pressure evictions or completed batch runs leave terminated pods behind." +category: "kubernetes" +icon: "delete_sweep" +status: ACTIVE +tags: + labels: + - "delete" + event_sources: + - "prometheus" + - "kubernetes_api_server" + - "alertmanager" + alert_names: + - "KubePodEvicted" + - "KubeletTooManyPods" + - "KubePodCrashLooping" + subject_types: + - "pod" + - "namespace" +definition: + version: "v1" + inputs: + - id: "namespace" + type: "string" + description: "Kubernetes namespace to clean up" + required: true + - id: "account_id" + type: "string" + description: "Cloud account ID" + required: true + triggers: + - type: "manual" + tasks: + - id: "list_candidates" + type: "cloud.k8s.cli" + params: + command: "kubectl get pods -n {{ Inputs.namespace }} --field-selector status.phase=Failed -o wide && kubectl get pods -n {{ Inputs.namespace }} --field-selector status.phase=Succeeded -o wide" + account_id: "{{ Inputs.account_id }}" + - id: "approve" + type: "core.approval" + params: + message: "Delete all Evicted/Failed and Succeeded pods in namespace **{{ Inputs.namespace }}**? Review the list above before approving." + depends_on: + - "list_candidates" + - id: "delete_failed" + type: "cloud.k8s.cli" + params: + command: "kubectl delete pods -n {{ Inputs.namespace }} --field-selector status.phase=Failed" + account_id: "{{ Inputs.account_id }}" + depends_on: + - "approve" + - id: "delete_succeeded" + type: "cloud.k8s.cli" + params: + command: "kubectl delete pods -n {{ Inputs.namespace }} --field-selector status.phase=Succeeded" + account_id: "{{ Inputs.account_id }}" + depends_on: + - "approve" + - id: "verify" + type: "cloud.k8s.cli" + params: + command: "if kubectl get pods -n {{ Inputs.namespace }} --field-selector status.phase=Failed -o name | grep -q . ; then echo 'Failed pods still present' && exit 1; fi; if kubectl get pods -n {{ Inputs.namespace }} --field-selector status.phase=Succeeded -o name | grep -q . ; then echo 'Succeeded pods still present' && exit 1; fi; echo 'All Failed/Succeeded pods cleaned up'" + account_id: "{{ Inputs.account_id }}" + depends_on: + - "delete_failed" + - "delete_succeeded" + timeout: "10m" +template_variables: + - id: "namespace" + input_ref: "namespace" + display_name: "Namespace" + required: true + type: "string" + - id: "account_id" + input_ref: "account_id" + display_name: "Account" + required: true + type: "account_selector" From 71197022606dcb8b83543fe351afeec70468717d Mon Sep 17 00:00:00 2001 From: mayankpande88 Date: Sun, 28 Jun 2026 15:11:52 +0530 Subject: [PATCH 2/2] ci: validate workflow semantics offline in validate.py Add engine-level checks mirroring runbook-server's internal/model/validation.go and the task registry: every task type is a known engine task type (an unknown type is rejected at sync time as "task not found"), depends_on resolves to an existing task, no duplicate task ids, no dependency cycles, supported trigger types, and well-formed durations. This is the offline equivalent of `nbctl workflow validate`: no backend or secrets, so it runs on fork PRs too, unlike the server-side check in nudgebee-infra. It runs through the existing validate.yaml CI via scripts/validate.py, so no new workflow file is needed. --- README.md | 20 ++++- scripts/validate.py | 180 +++++++++++++++++++++++++++++++++++++++++++- 2 files changed, 196 insertions(+), 4 deletions(-) diff --git a/README.md b/README.md index e9b11ec..2702fde 100644 --- a/README.md +++ b/README.md @@ -16,9 +16,27 @@ templates/.yaml # one template per file; file name MUST equal `sl manifest.yaml # generated index [{slug, path, sha256}] — runbook-server verifies each file against this schema/template.schema.json # JSON Schema; CI validates every template scripts/gen_manifest.py # regenerate manifest.yaml (run before committing template changes) -scripts/validate.py # local equivalent of the CI validation +scripts/validate.py # local equivalent of the CI validation (schema + workflow semantics) ``` +## Validation layers + +`scripts/validate.py` (run in CI on every PR, including fork PRs — no secrets or +backend required) checks two things: + +1. **Structure** — the JSON Schema in `schema/template.schema.json`: required keys, + enums, `slug` == file name, no duplicate slugs. +2. **Workflow semantics** — the same rules the engine enforces in + `runbook-server/internal/model/validation.go` and the task registry: every task + `type` is a real engine task type (an unknown/mistyped type is rejected at sync + time as `task not found`), `depends_on` resolves to an existing task, no + duplicate task ids, no dependency cycles, supported trigger types, and + well-formed durations. This is the offline equivalent of `nbctl workflow + validate` — it needs no backend, so it gates external contributions too. + +When the engine registers a new task type, add it to the `TASK_TYPES` set in +`scripts/validate.py` (its source of truth is `runbook-server`'s task registry). + ## Authoring a template 1. Add or edit `templates/.yaml`. The `slug` field must match the file name diff --git a/scripts/validate.py b/scripts/validate.py index efc0e7d..202d55b 100755 --- a/scripts/validate.py +++ b/scripts/validate.py @@ -1,8 +1,25 @@ #!/usr/bin/env python3 -"""Validate every template file against the JSON schema and repo conventions. +"""Validate every template file against the JSON schema, repo conventions, and the +workflow-engine semantics that runbook-server enforces at sync/apply time. + +Two layers of checks: + +1. Structural (JSON Schema in schema/template.schema.json) — shape, enums, required + keys, slug == file name, no duplicate slugs. +2. Workflow semantics (this file) — the same rules the engine applies in + runbook-server/internal/model/validation.go and internal/tasks/registry.go. + These catch the failures the JSON schema can't: an unknown / mistyped task + `type` (rejected at sync time as "task not found"), a `depends_on` pointing at a + task that does not exist, duplicate task ids, dependency cycles, an unsupported + trigger type, or a malformed duration. This is the offline equivalent of + `nbctl workflow validate` — no backend or secrets required, so it also runs on + fork PRs. + Run by CI on every PR (see .github/workflows/validate.yaml).""" import glob +import json import os +import re import sys import jsonschema @@ -12,10 +29,162 @@ SCHEMA_PATH = os.path.join(ROOT, "schema", "template.schema.json") TEMPLATES_GLOB = os.path.join(ROOT, "templates", "*.yaml") +# Supported workflow task `type` strings. Source of truth: +# runbook-server/internal/tasks/registry.go (NewInitializedTaskRegistry). A +# template that references a type not in this set is rejected by the engine at +# sync time ("task not found: "). Keep this list in sync when the engine +# registers a new task type; an unknown type here is a hard error, not a warning. +TASK_TYPES = { + # core control flow + "core.approval", "core.call-workflow", "core.foreach", "core.group", + "core.print", "core.switch", "core.wait", + # data + "data.filter", "data.transform", + # database + "dbms.query", "dbms.redis.cli", + # events / audit + "events.store", + # crypto + "crypto.encode", "crypto.decode", "crypto.hash", "crypto.encrypt", "crypto.decrypt", + # chat + "google_chat.join_space", "slack.join_channel", + # integrations / scripting + "integrations.http", "integrations.ssh", "scripting.run_script", + # notifications + "notifications.add_reaction", "notifications.create_channel", "notifications.dm", + "notifications.email", "notifications.im", "notifications.read_thread", + # tickets / incidents + "tickets.acknowledge", "tickets.add_comment", "tickets.assign", "tickets.create", + "tickets.escalate", "tickets.get", "tickets.get_comments", "tickets.resolve", + "tickets.transition", "tickets.update", + # kubernetes + "k8s.cli", "k8s.continuous_rightsize", "k8s.horizontal_rightsize", + "k8s.node_graceful_shutdown", "k8s.pod_delete", "k8s.pv_rightsize", + "k8s.vertical_rightsize", "k8s.workload_restart", + # cloud CLIs (canonical + aliases) + "cloud.aws.cli", "cloud.azure.cli", "cloud.gcp.cli", "cloud.k8s.cli", + "aws.cli", "azure.cli", "gcp.cli", + # cicd / mq / scm + "cicd.argocd.cli", "mq.rabbitmqadmin.cli", "scm.github.cli", "scm.gitlab.cli", + # observability + "observability.logs", "observability.log_groups", "observability.metrics", + "observability.traces", + # network diagnostics + "network.dns", "network.ntp", "network.ping", "network.ssl", "network.tcp", + "network.traceroute", "network.whois", + # ai / llm + "llm.a2a_call", "llm.classify", "llm.event_investigate", "llm.investigate", + "llm.mcp_call", "llm.nubi", "llm.router", "llm.summary", + # internal optimization + "vertical_rightsize_generate", +} -def main(): - import json +# Triggers the engine accepts (model.ValidateWorkflowTrigger). +TRIGGER_TYPES = {"schedule", "manual", "webhook", "event", "optimization"} + +# Task id: alphanumeric / hyphen / underscore, 3-64 chars (model.ValidateTaskID). +TASK_ID_RE = re.compile(r"^[a-zA-Z0-9_-]+$") +# Workflow name: starts/ends alphanumeric, spaces/hyphens/underscores inside, +# 3-50 chars (model.ValidateWorkflowName). +NAME_RE = re.compile(r"^[a-zA-Z0-9](?:[a-zA-Z0-9 _-]*[a-zA-Z0-9])?$") +# Go time.ParseDuration units; durations may be concatenated (e.g. "1h30m"). +DURATION_RE = re.compile(r"^(\d+(\.\d+)?(ns|us|µs|ms|s|m|h))+$") + + +def _is_duration(s): + return isinstance(s, str) and bool(DURATION_RE.match(s)) + + +def validate_workflow_semantics(doc, rel): + """Return a list of error strings for the engine-level rules. Mirrors + runbook-server/internal/model/validation.go.""" + errs = [] + + name = doc.get("name", "") + if not (3 <= len(name) <= 50) or not NAME_RE.match(name): + errs.append(f"name {name!r} must be 3-50 chars, alphanumeric with spaces/_/- inside") + + definition = doc.get("definition") or {} + + version = definition.get("version") + if version is not None and version != "v1": + errs.append(f"definition.version must be \"v1\", got {version!r}") + + for trig in definition.get("triggers") or []: + ttype = trig.get("type") if isinstance(trig, dict) else None + if ttype not in TRIGGER_TYPES: + errs.append(f"unsupported trigger type {ttype!r} (allowed: {sorted(TRIGGER_TYPES)})") + + tasks = definition.get("tasks") or [] + ids = [t.get("id") for t in tasks if isinstance(t, dict)] + id_set = set() + for t in tasks: + if not isinstance(t, dict): + errs.append("each task must be a mapping") + continue + tid = t.get("id", "") + if not (3 <= len(str(tid)) <= 64) or not TASK_ID_RE.match(str(tid)): + errs.append(f"task id {tid!r} must be 3-64 chars, [a-zA-Z0-9_-]") + if tid in id_set: + errs.append(f"duplicate task id {tid!r}") + id_set.add(tid) + + ttype = t.get("type") + if ttype not in TASK_TYPES: + errs.append(f"task {tid!r}: unknown task type {ttype!r} (engine rejects at sync time)") + if t.get("timeout") and not _is_duration(t["timeout"]): + errs.append(f"task {tid!r}: invalid timeout {t['timeout']!r}") + + for dep in t.get("depends_on") or []: + if dep == tid: + errs.append(f"task {tid!r}: self-dependency") + elif dep not in ids: + errs.append(f"task {tid!r}: depends_on missing task {dep!r}") + + # Cycle detection over depends_on (DFS), matching the engine. + adj = {t.get("id"): list(t.get("depends_on") or []) for t in tasks if isinstance(t, dict)} + visited, rec = set(), set() + + def has_cycle(node): + if node in rec: + return True + if node in visited: + return False + visited.add(node) + rec.add(node) + for d in adj.get(node, []): + if has_cycle(d): + return True + rec.discard(node) + return False + + for node in list(adj): + if has_cycle(node): + errs.append(f"circular dependency involving task {node!r}") + break + + # Sum of task timeouts must not exceed the workflow timeout (engine rule). + wf_timeout = definition.get("timeout") + if _is_duration(wf_timeout): + wf_secs = _duration_seconds(wf_timeout) + total = sum(_duration_seconds(t["timeout"]) for t in tasks + if isinstance(t, dict) and _is_duration(t.get("timeout"))) + if total and total > wf_secs: + errs.append(f"sum of task timeouts ({total}s) exceeds workflow timeout ({wf_secs}s)") + + return errs + + +_UNIT_SECS = {"ns": 1e-9, "us": 1e-6, "µs": 1e-6, "ms": 1e-3, "s": 1, "m": 60, "h": 3600} + + +def _duration_seconds(s): + return sum(float(num) * _UNIT_SECS[unit] + for num, _frac, unit in re.findall(r"(\d+(\.\d+)?)(ns|us|µs|ms|s|m|h)", s)) + + +def main(): schema = json.load(open(SCHEMA_PATH)) files = sorted(glob.glob(TEMPLATES_GLOB)) if not files: @@ -26,6 +195,7 @@ def main(): slugs = set() for path in files: name = os.path.basename(path) + rel = os.path.relpath(path, ROOT) try: doc = yaml.safe_load(open(path, "rb").read()) except yaml.YAMLError as e: @@ -47,6 +217,10 @@ def main(): errors += 1 slugs.add(doc.get("slug")) + for msg in validate_workflow_semantics(doc, rel): + print(f"::error file={path}::{msg}") + errors += 1 + if errors: print(f"\n{errors} validation error(s).", file=sys.stderr) sys.exit(1)