From ed0b4aef798d3c0de51246c86290342d0254d4fd Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 3 Aug 2026 21:10:12 +0000 Subject: [PATCH 1/2] Add trigger_workflow action: fire a platform Triggered Workflow from an alert MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit New alert-triggered playbook action that POSTs the entire alert payload (labels, annotations, status, timestamps, generatorURL, fingerprint) plus the cluster name to the Robusta platform /webhooks endpoint with one or more workflow_id params, starting a Triggered Workflow run — typically a Holmes investigation. Optional route_to_alert_cluster targets the workflow at the cluster the alert fired in via the cluster URL param. Tested with a real HTTP capture server against a KubeNodeUnschedulable (node cordoned) alert; the request contract is mirrored end-to-end in relay's tests/test_workflow_trigger_alert_e2e.py. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01SSJqTBghtypCiZUwDDvP7a --- .../robusta_playbooks/workflow_trigger.py | 116 ++++++++++++ tests/test_workflow_trigger.py | 170 ++++++++++++++++++ 2 files changed, 286 insertions(+) create mode 100644 playbooks/robusta_playbooks/workflow_trigger.py create mode 100644 tests/test_workflow_trigger.py diff --git a/playbooks/robusta_playbooks/workflow_trigger.py b/playbooks/robusta_playbooks/workflow_trigger.py new file mode 100644 index 000000000..c15439246 --- /dev/null +++ b/playbooks/robusta_playbooks/workflow_trigger.py @@ -0,0 +1,116 @@ +"""Trigger a Robusta platform Triggered Workflow from an alert. + +Fires the platform's ``POST /webhooks`` endpoint with the entire alert +payload (labels, annotations, status, timestamps, generatorURL, fingerprint) +plus the cluster name, so a Triggered Workflow — typically a Holmes +investigation — runs in response to the alert. + +Example playbook configuration:: + + customPlaybooks: + - triggers: + - on_prometheus_alert: + alert_name: NodeCordonedManually + actions: + - trigger_workflow: + workflow_id: "b7f9d2e4-1234-4c56-9abc-0123456789ab" + api_key: "{{ env.ROBUSTA_PLATFORM_API_KEY }}" +""" + +import json +import logging +from typing import List, Optional, Union + +import requests +from pydantic import SecretStr +from robusta.api import ActionException, ActionParams, ErrorCodes, PrometheusKubernetesAlert, action + + +class TriggerWorkflowParams(ActionParams): + """ + :var workflow_id: One or more Triggered Workflow ids to run. A single id, + or a list to trigger several workflows from the same alert. + :var api_key: Robusta platform account API key with ``alerts:WRITE`` + permission. Sent as ``Authorization: Bearer ``. + :var url: The platform webhooks endpoint. + :var account_id: (optional) Robusta account id. Defaults to the account + this runner is connected to. + :var origin: (optional) Origin label stored with the event, shown in the + platform Delivery Log. + :var route_to_alert_cluster: (optional) (Default: False) When True, the + workflow runs against the cluster this alert fired in (via the + ``cluster`` URL parameter), overriding the cluster configured on the + workflow definition. + :var timeout: (optional) (Default: 30) Request timeout in seconds. + """ + + workflow_id: Union[str, List[str]] + api_key: SecretStr + url: str = "https://api.robusta.dev/webhooks" + account_id: Optional[str] = None + origin: str = "robusta-runner" + route_to_alert_cluster: bool = False + timeout: int = 30 + + +def build_workflow_trigger_payload(alert: PrometheusKubernetesAlert) -> dict: + """The webhook body: the entire alert payload plus the cluster name. + + The alert is nested under ``alert`` untouched (labels, annotations, + status, startsAt/endsAt, generatorURL, fingerprint), so workflow filters + can match on any alert field; ``cluster_name`` rides alongside it. + """ + context = alert.get_context() + return { + "cluster_name": context.cluster_name, + "alert": json.loads(alert.alert.json()), + } + + +@action +def trigger_workflow(alert: PrometheusKubernetesAlert, params: TriggerWorkflowParams): + """ + Trigger one or more Robusta platform Triggered Workflows (e.g. a Holmes + investigation), sending the entire alert payload and the cluster name as + the workflow's trigger payload. + """ + workflow_ids = params.workflow_id if isinstance(params.workflow_id, list) else [params.workflow_id] + workflow_ids = [w.strip() for w in workflow_ids if w and w.strip()] + if not workflow_ids: + raise ActionException(ErrorCodes.ACTION_UNEXPECTED_ERROR, "trigger_workflow: no workflow_id provided") + + context = alert.get_context() + account_id = params.account_id or context.account_id + + query_params: List[tuple] = [("account_id", account_id), ("origin", params.origin)] + query_params.extend(("workflow_id", workflow_id) for workflow_id in workflow_ids) + if params.route_to_alert_cluster: + query_params.append(("cluster", context.cluster_name)) + + payload = build_workflow_trigger_payload(alert) + + try: + response = requests.post( + params.url, + params=query_params, + json=payload, + headers={"Authorization": f"Bearer {params.api_key.get_secret_value()}"}, + timeout=params.timeout, + ) + except Exception as e: + raise ActionException( + ErrorCodes.ACTION_UNEXPECTED_ERROR, + f"trigger_workflow: failed to reach {params.url} for alert {alert.alert_name}: {e}", + ) + + if not (200 <= response.status_code < 300): + raise ActionException( + ErrorCodes.ACTION_UNEXPECTED_ERROR, + f"trigger_workflow: {params.url} returned {response.status_code} " + f"for alert {alert.alert_name}: {response.text[:500]}", + ) + + logging.info( + f"trigger_workflow: triggered workflow(s) {workflow_ids} for alert " + f"{alert.alert_name} on cluster {context.cluster_name}" + ) diff --git a/tests/test_workflow_trigger.py b/tests/test_workflow_trigger.py new file mode 100644 index 000000000..115497f80 --- /dev/null +++ b/tests/test_workflow_trigger.py @@ -0,0 +1,170 @@ +import json +import threading +from datetime import datetime, timezone +from http.server import BaseHTTPRequestHandler, HTTPServer +from urllib.parse import parse_qs, urlparse + +import pytest +from pydantic import SecretStr +from robusta.api import ActionException +from robusta.core.model.events import ExecutionContext +from robusta.integrations.prometheus.models import PrometheusAlert, PrometheusKubernetesAlert + +from playbooks.robusta_playbooks.workflow_trigger import ( + TriggerWorkflowParams, + build_workflow_trigger_payload, + trigger_workflow, +) + +CLUSTER_NAME = "prod-us-east-1" +ACCOUNT_ID = "11111111-2222-3333-4444-555555555555" +WORKFLOW_ID = "b7f9d2e4-0000-4c56-9abc-0123456789ab" +API_KEY = "test-api-key" + +# A KubeNodeUnschedulable alert — what fires when a node is cordoned. +NODE_CORDONED_ALERT = { + "status": "firing", + "labels": { + "alertname": "KubeNodeUnschedulable", + "node": "ip-10-0-1-17.ec2.internal", + "severity": "warning", + }, + "annotations": { + "summary": "Node is unschedulable.", + "description": "ip-10-0-1-17.ec2.internal is unschedulable for more than 15 minutes.", + }, + "startsAt": "2026-08-03T10:00:00Z", + "endsAt": "0001-01-01T00:00:00Z", + "generatorURL": "http://prometheus/graph?g0.expr=kube_node_spec_unschedulable+%3D%3D+1", + "fingerprint": "abcdef0123456789", +} + + +def make_alert() -> PrometheusKubernetesAlert: + alert = PrometheusKubernetesAlert( + alert=PrometheusAlert(**NODE_CORDONED_ALERT), + alert_name=NODE_CORDONED_ALERT["labels"]["alertname"], + alert_severity=NODE_CORDONED_ALERT["labels"]["severity"], + named_sinks=[], + ) + alert.set_context(ExecutionContext(account_id=ACCOUNT_ID, cluster_name=CLUSTER_NAME)) + return alert + + +class _CaptureServer: + """Minimal HTTP server capturing webhook requests, responding 200.""" + + def __init__(self, status_code: int = 200): + self.requests = [] + capture = self + + class Handler(BaseHTTPRequestHandler): + def do_POST(self): + body = self.rfile.read(int(self.headers.get("Content-Length", 0))) + capture.requests.append( + { + "path": self.path, + "headers": dict(self.headers), + "body": body, + } + ) + self.send_response(status_code) + self.send_header("Content-Type", "application/json") + self.end_headers() + self.wfile.write(b'{"status": "stored"}') + + def log_message(self, *args): + pass + + self.server = HTTPServer(("127.0.0.1", 0), Handler) + self.thread = threading.Thread(target=self.server.serve_forever, daemon=True) + + def __enter__(self): + self.thread.start() + return self + + def __exit__(self, *exc): + self.server.shutdown() + + @property + def url(self) -> str: + return f"http://127.0.0.1:{self.server.server_port}/webhooks" + + +def test_payload_contains_entire_alert_and_cluster_name(): + payload = build_workflow_trigger_payload(make_alert()) + assert payload["cluster_name"] == CLUSTER_NAME + # the entire alert payload survives, byte-for-byte on every field + assert payload["alert"]["labels"] == NODE_CORDONED_ALERT["labels"] + assert payload["alert"]["annotations"] == NODE_CORDONED_ALERT["annotations"] + assert payload["alert"]["status"] == "firing" + assert payload["alert"]["generatorURL"] == NODE_CORDONED_ALERT["generatorURL"] + assert payload["alert"]["fingerprint"] == NODE_CORDONED_ALERT["fingerprint"] + assert datetime.fromisoformat(payload["alert"]["startsAt"]) == datetime(2026, 8, 3, 10, 0, tzinfo=timezone.utc) + # payload is JSON-serializable as-is (datetimes already rendered) + json.dumps(payload) + + +def test_trigger_workflow_posts_alert_to_webhooks_endpoint(): + with _CaptureServer() as server: + trigger_workflow( + make_alert(), + TriggerWorkflowParams(workflow_id=WORKFLOW_ID, api_key=SecretStr(API_KEY), url=server.url), + ) + + assert len(server.requests) == 1 + request = server.requests[0] + parsed = urlparse(request["path"]) + query = parse_qs(parsed.query) + + assert parsed.path == "/webhooks" + assert query["account_id"] == [ACCOUNT_ID] + assert query["workflow_id"] == [WORKFLOW_ID] + assert query["origin"] == ["robusta-runner"] + assert "cluster" not in query # off unless route_to_alert_cluster is set + assert request["headers"]["Authorization"] == f"Bearer {API_KEY}" + + body = json.loads(request["body"]) + assert body["cluster_name"] == CLUSTER_NAME + assert body["alert"]["labels"] == NODE_CORDONED_ALERT["labels"] + assert body["alert"]["annotations"] == NODE_CORDONED_ALERT["annotations"] + + +def test_trigger_workflow_multiple_ids_and_cluster_routing(): + other_workflow_id = "c8f9d2e4-0000-4c56-9abc-0123456789ab" + with _CaptureServer() as server: + trigger_workflow( + make_alert(), + TriggerWorkflowParams( + workflow_id=[WORKFLOW_ID, other_workflow_id], + api_key=SecretStr(API_KEY), + url=server.url, + route_to_alert_cluster=True, + ), + ) + + query = parse_qs(urlparse(server.requests[0]["path"]).query) + assert query["workflow_id"] == [WORKFLOW_ID, other_workflow_id] + assert query["cluster"] == [CLUSTER_NAME] + + +def test_trigger_workflow_raises_on_http_error(): + with _CaptureServer(status_code=401) as server: + with pytest.raises(ActionException): + trigger_workflow( + make_alert(), + TriggerWorkflowParams(workflow_id=WORKFLOW_ID, api_key=SecretStr(API_KEY), url=server.url), + ) + + +def test_trigger_workflow_raises_when_unreachable(): + with pytest.raises(ActionException): + trigger_workflow( + make_alert(), + TriggerWorkflowParams( + workflow_id=WORKFLOW_ID, + api_key=SecretStr(API_KEY), + url="http://127.0.0.1:1/webhooks", + timeout=2, + ), + ) From 7921ea9973ca5c01c2536f06671405c4e6388451 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 3 Aug 2026 21:18:46 +0000 Subject: [PATCH 2/2] trigger_workflow: route_to_alert_cluster defaults to True The triggered workflow now runs against the cluster the alert fired in by default; set route_to_alert_cluster: false to keep the workflow definition's configured cluster. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01SSJqTBghtypCiZUwDDvP7a --- .../robusta_playbooks/workflow_trigger.py | 7 +++--- tests/test_workflow_trigger.py | 22 ++++++++++++++++--- 2 files changed, 23 insertions(+), 6 deletions(-) diff --git a/playbooks/robusta_playbooks/workflow_trigger.py b/playbooks/robusta_playbooks/workflow_trigger.py index c15439246..d65c1351c 100644 --- a/playbooks/robusta_playbooks/workflow_trigger.py +++ b/playbooks/robusta_playbooks/workflow_trigger.py @@ -37,10 +37,11 @@ class TriggerWorkflowParams(ActionParams): this runner is connected to. :var origin: (optional) Origin label stored with the event, shown in the platform Delivery Log. - :var route_to_alert_cluster: (optional) (Default: False) When True, the + :var route_to_alert_cluster: (optional) (Default: True) When True, the workflow runs against the cluster this alert fired in (via the ``cluster`` URL parameter), overriding the cluster configured on the - workflow definition. + workflow definition. Set False to always use the workflow's + configured cluster. :var timeout: (optional) (Default: 30) Request timeout in seconds. """ @@ -49,7 +50,7 @@ class TriggerWorkflowParams(ActionParams): url: str = "https://api.robusta.dev/webhooks" account_id: Optional[str] = None origin: str = "robusta-runner" - route_to_alert_cluster: bool = False + route_to_alert_cluster: bool = True timeout: int = 30 diff --git a/tests/test_workflow_trigger.py b/tests/test_workflow_trigger.py index 115497f80..65d5c492b 100644 --- a/tests/test_workflow_trigger.py +++ b/tests/test_workflow_trigger.py @@ -121,7 +121,8 @@ def test_trigger_workflow_posts_alert_to_webhooks_endpoint(): assert query["account_id"] == [ACCOUNT_ID] assert query["workflow_id"] == [WORKFLOW_ID] assert query["origin"] == ["robusta-runner"] - assert "cluster" not in query # off unless route_to_alert_cluster is set + # route_to_alert_cluster defaults to True: the run targets the alert's cluster + assert query["cluster"] == [CLUSTER_NAME] assert request["headers"]["Authorization"] == f"Bearer {API_KEY}" body = json.loads(request["body"]) @@ -130,7 +131,7 @@ def test_trigger_workflow_posts_alert_to_webhooks_endpoint(): assert body["alert"]["annotations"] == NODE_CORDONED_ALERT["annotations"] -def test_trigger_workflow_multiple_ids_and_cluster_routing(): +def test_trigger_workflow_multiple_ids(): other_workflow_id = "c8f9d2e4-0000-4c56-9abc-0123456789ab" with _CaptureServer() as server: trigger_workflow( @@ -139,7 +140,6 @@ def test_trigger_workflow_multiple_ids_and_cluster_routing(): workflow_id=[WORKFLOW_ID, other_workflow_id], api_key=SecretStr(API_KEY), url=server.url, - route_to_alert_cluster=True, ), ) @@ -148,6 +148,22 @@ def test_trigger_workflow_multiple_ids_and_cluster_routing(): assert query["cluster"] == [CLUSTER_NAME] +def test_trigger_workflow_cluster_routing_opt_out(): + with _CaptureServer() as server: + trigger_workflow( + make_alert(), + TriggerWorkflowParams( + workflow_id=WORKFLOW_ID, + api_key=SecretStr(API_KEY), + url=server.url, + route_to_alert_cluster=False, + ), + ) + + query = parse_qs(urlparse(server.requests[0]["path"]).query) + assert "cluster" not in query # the workflow's configured cluster applies + + def test_trigger_workflow_raises_on_http_error(): with _CaptureServer(status_code=401) as server: with pytest.raises(ActionException):