ROB-834 Add workflow trigger action to fire Robusta platform workflows

playbooks/robusta_playbooks/workflow_trigger.py

@@ -0,0 +1,117 @@
+"""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 <key>``.
+    :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: 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. Set False to always use the workflow's
+        configured cluster.
+    :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 = True
+    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}"
+    )

tests/test_workflow_trigger.py

@@ -0,0 +1,186 @@
+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"]
+    # 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"])
+    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():
+    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,
+            ),
+        )
+
+    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_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):
+            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,
+            ),
+        )