#!/usr/bin/env python3 """ deadman — a dead-man's switch for n8n workflows. Why this exists --------------- An n8n error workflow only fires when something *ran and failed*. The failure that costs people money is the one where nothing fires at all: the schedule stops producing executions and nobody notices for a day. That class is structurally invisible to an error workflow, because there is no error to catch — there is an absence. deadman looks at the absence. It checks three classes, in this order of precedence: 1. PROBE-FAULT the n8n REST API does not answer, or answers 401/403. This is checked FIRST and it stops every other verdict, because "my key expired" reported as "47 workflows are dead" is a lie. A lapsed Cloud trial and a rotated API key both look like this. 2. SILENCE a workflow that is expected to run on a schedule produced no execution inside its window (expected cadence + grace). 3. ERROR a new failed execution appeared that deadman has not already alerted on. Usage ----- python3 deadman.py --config config.toml python3 deadman.py --config config.example.toml --simulate demo.json # no instance needed python3 deadman.py --config config.toml --now 2026-11-13T09:00:00Z # deterministic Exit codes (documented; also printed by --help) ----------------------------------------------- 0 healthy: the probe reached n8n and no alert is open 1 at least one alert is open (silence or error) 2 the probe could not do its job: bad config, instance unreachable, auth refused, or malformed API reply. Never means "your workflows are fine". State and idempotency --------------------- State is one small JSON file: last-seen execution id and time per workflow, the last time the whole check succeeded, and which alerts are open. Re-running is idempotent: an alert fires ONCE when it opens, re-fires only after `cooldown_s`, and when the condition clears it sends exactly ONE recovery notice. That dedupe-and-recover behaviour is the product. Time is UTC everywhere. All cadence arithmetic is done on absolute instants, so daylight-saving changes and host timezone cannot shift a window. A `--now` string without an offset or `Z` is refused rather than guessed at. n8n API contract used here -------------------------- Every field name and query parameter below was read from n8n's own generated endpoint reference for "Retrieve all executions" (public API version 1.1.1): https://docs.n8n.io/connect/n8n-api/build-and-run/executions#get-executions (same page as Markdown: https://docs.n8n.io/connect/n8n-api/build-and-run/executions.md) and from the OpenAPI spec that page is generated from: https://n8n-openapi-dfb.pages.dev/v1/openapi.yml Verified facts taken from that page, and relied on by this program: * auth scheme `ApiKeyAuth`: {"type":"apiKey","in":"header","name":"X-N8N-API-KEY"} * query params: limit (default 100, maximum 250), cursor, status, workflowId, startedAfter, startedBefore (both `format: date-time`), includeData * response: an object with REQUIRED members `data` (array) and `nextCursor` (string or null) * each item in `data` has REQUIRED members: id, finished, mode, retryOf, retrySuccessId, status, startedAt, stoppedAt, workflowId, waitTill * `status` enum: canceled | crashed | error | new | running | success | unknown | waiting * `startedAt` is "Null until the execution leaves the queue and starts running" * `finished` is marked deprecated with the instruction "Use `status` instead", so this program does not read it * error responses: 400, 401, 403, 404, 429 (429 carries a Retry-After header) NOT VERIFIED from the docs, and therefore worked around rather than assumed: * the ORDER of `data` (the spec says nothing about sort order) — so nothing in this program depends on it; rows are filtered and compared client-side * whether `startedAfter` accepts (or requires) sub-second precision — the query is sent at whole-second UTC, and every window test is also applied client-side, so a lenient or strict server cannot change a verdict * whether `startedAfter` compares against `startedAt` or another column — the client-side filter is authoritative, the server-side one is only a transfer-size optimisation """ from __future__ import annotations import argparse import json import os import smtplib import sys import urllib.error import urllib.parse import urllib.request from datetime import datetime, timedelta, timezone try: import tomllib # Python 3.11+ except ModuleNotFoundError: # pragma: no cover - exercised only on <3.11 tomllib = None EXIT_OK = 0 EXIT_ALERT = 1 EXIT_PROBE = 2 EXIT_DOC = ( "exit codes:\n" " 0 healthy: n8n was reached and no alert is open\n" " 1 one or more alerts open (silence or error)\n" " 2 the probe could not do its job: bad config, unreachable instance,\n" " auth refused, or malformed API reply\n" ) # From the `status` enum on # https://docs.n8n.io/connect/n8n-api/build-and-run/executions#get-executions # "crashed" is in that enum and is a run that died without reporting an error, # so it counts as a failure alongside "error". "canceled" is a deliberate stop # by a human or by the API, so it is NOT a failure. FAILED_STATUSES = ("error", "crashed") ALL_STATUSES = ( "canceled", "crashed", "error", "new", "running", "success", "unknown", "waiting", ) # Verified as required members of each `data` item on the same page. EXECUTION_ID_FIELD = "id" EXECUTION_STATUS_FIELD = "status" # verified required EXECUTION_STARTED_FIELD = "startedAt" # verified required, nullable by design EXECUTION_WORKFLOW_FIELD = "workflowId" # verified required STATE_VERSION = 1 # n8n's own limit for this endpoint is maximum 250 (verified above). API_LIMIT_MAX = 250 UNIT_SECONDS = {"s": 1, "m": 60, "h": 3600, "d": 86400} class DeadmanError(Exception): """A problem that means deadman could not do its job -> exit 2.""" def __init__(self, message): super().__init__(message) self.message = message # -------------------------------------------------------------------------- # time and duration arithmetic (all of it on absolute instants, never wall clock) # -------------------------------------------------------------------------- def parse_duration(text, where): """'45s' / '90m' / '6h' / '2d' / '1h30m' / '3600' (bare number = seconds) -> float seconds. Every rejection names the field and shows the shape.""" if isinstance(text, bool): raise DeadmanError("%s: expected a duration like '90m', got a boolean" % where) if isinstance(text, (int, float)): seconds = float(text) if seconds <= 0: raise DeadmanError("%s: must be a positive duration, got %r" % (where, text)) return seconds if not isinstance(text, str): raise DeadmanError( "%s: expected a duration like '90m' or '1h30m', got %r" % (where, text)) s = text.strip().lower() if not s: raise DeadmanError("%s: duration is empty (use a form like '45s', '90m', '6h', '2d')" % where) if not s[-1].isdigit(): total = 0.0 consumed = "" for ch in s: if ch.isdigit(): consumed += ch continue unit = UNIT_SECONDS.get(ch) if unit is None or not consumed: raise DeadmanError( "%s: unreadable duration '%s' (use 45s / 90m / 6h / 2d, or a number of seconds)" % (where, text)) total += int(consumed) * unit consumed = "" if consumed or total <= 0: raise DeadmanError( "%s: unreadable duration '%s' (use 45s / 90m / 6h / 2d, or a number of seconds)" % (where, text)) return total try: seconds = float(s) except ValueError: raise DeadmanError( "%s: unreadable duration '%s' (use 45s / 90m / 6h / 2d, or a number of seconds)" % (where, text)) if seconds <= 0: raise DeadmanError("%s: must be a positive duration, got %r" % (where, text)) return seconds def parse_instant(text, where): """Parse an ISO-8601 timestamp that carries an explicit zone, return it in UTC. A naive timestamp is refused: guessing the host's zone is exactly how a monitoring window ends up an hour off twice a year. """ if not isinstance(text, str) or not text.strip(): raise DeadmanError("%s: expected an ISO-8601 timestamp, got %r" % (where, text)) s = text.strip() iso = s[:-1] + "+00:00" if s.endswith(("Z", "z")) else s try: dt = datetime.fromisoformat(iso) except ValueError: raise DeadmanError("%s: not a readable ISO-8601 timestamp: %r" % (where, text)) if dt.tzinfo is None: raise DeadmanError( "%s: timestamp %r has no UTC offset; deadman refuses to assume the host's " "zone (write it as 2026-11-13T09:00:00Z or with an offset)" % (where, text)) return dt.astimezone(timezone.utc) def iso(dt): """UTC ISO-8601 to the second with a 'Z' — the format the query is sent in.""" return dt.astimezone(timezone.utc).isoformat(timespec="seconds").replace("+00:00", "Z") def utc_now(): return datetime.now(timezone.utc) # -------------------------------------------------------------------------- # config # -------------------------------------------------------------------------- def _require_string(table, key, where): if key not in table: raise DeadmanError("%s: '%s' is required" % (where, key)) value = table[key] if not isinstance(value, str) or not value.strip(): raise DeadmanError("%s: '%s' must be a non-empty string" % (where, key)) return value.strip() def _reject_unknown(table, allowed, where): for key in table: if key not in allowed: raise DeadmanError("%s: unknown field '%s' (known: %s)" % (where, key, ", ".join(sorted(allowed)))) def _https_url(text, where): parsed = urllib.parse.urlsplit(text) if parsed.scheme not in ("http", "https") or not parsed.netloc: raise DeadmanError("%s: must be a full URL starting with http:// or https://, got %r" % (where, text)) return text.rstrip("/") PROBE_KEYS = {"state_file", "cooldown_s", "timeout_s", "error_lookback_s", "max_pages", "fail_statuses", "source"} INSTANCE_KEYS = {"base_url", "api_key", "api_key_env"} WORKFLOW_KEYS = {"id", "name", "every", "grace"} WEBHOOK_KEYS = {"url", "name", "timeout_s"} EMAIL_KEYS = {"name", "host", "port", "username", "password", "password_env", "from", "to", "security", "timeout_s"} ALERT_KEYS = {"webhook", "email"} ROOT_KEYS = {"probe", "instance", "alert", "workflow"} def load_config(path, simulate=False): """Read and validate the TOML config. Every failure is its own message. simulate=True relaxes [instance] only: a demo has no instance to point at. """ if tomllib is None: raise DeadmanError( "deadman needs Python 3.11 or newer for tomllib (this is %d.%d). " "Install python3.11+ or convert config.toml to the documented JSON twin." % sys.version_info[:2]) try: with open(path, "rb") as handle: raw = handle.read() except FileNotFoundError: raise DeadmanError("config file not found: %s" % path) except IsADirectoryError: raise DeadmanError("config path is a directory, not a file: %s" % path) except OSError as exc: raise DeadmanError("config file could not be read (%s): %s" % (path, exc)) try: cfg = tomllib.loads(raw.decode("utf-8")) except UnicodeDecodeError: raise DeadmanError("config file is not UTF-8 text: %s" % path) except tomllib.TOMLDecodeError as exc: raise DeadmanError("config is not valid TOML (%s): %s" % (path, exc)) if not isinstance(cfg, dict): raise DeadmanError("config must be a TOML table of sections, got %s" % type(cfg).__name__) _reject_unknown(cfg, ROOT_KEYS, "config") probe_in = cfg.get("probe", {}) if not isinstance(probe_in, dict): raise DeadmanError("[probe] must be a table") _reject_unknown(probe_in, PROBE_KEYS, "[probe]") if "state_file" not in probe_in: raise DeadmanError("[probe] 'state_file' is required (path of the little JSON state file)") state_file = probe_in["state_file"] if not isinstance(state_file, str) or not state_file.strip(): raise DeadmanError("[probe] 'state_file' must be a non-empty string") cooldown_s = float(probe_in.get("cooldown_s", 21600)) if isinstance(cooldown_s, bool) or not isinstance(cooldown_s, (int, float)): raise DeadmanError("[probe] 'cooldown_s' must be a number of seconds (0 disables re-alerting)") if cooldown_s < 0: raise DeadmanError("[probe] 'cooldown_s' must be >= 0 (0 = alert once per episode, never re-alert)") timeout_s = float(probe_in.get("timeout_s", 20)) if isinstance(timeout_s, bool) or not isinstance(timeout_s, (int, float)) or timeout_s <= 0: raise DeadmanError("[probe] 'timeout_s' must be a positive number of seconds") lookback_s = float(probe_in.get("error_lookback_s", 86400)) if isinstance(lookback_s, bool) or not isinstance(lookback_s, (int, float)) or lookback_s <= 0: raise DeadmanError("[probe] 'error_lookback_s' must be a positive number of seconds") max_pages = probe_in.get("max_pages", 3) if isinstance(max_pages, bool) or not isinstance(max_pages, int) or max_pages < 1: raise DeadmanError("[probe] 'max_pages' must be an integer >= 1 (pages of 250 executions per workflow)") fail_statuses = list(probe_in.get("fail_statuses", FAILED_STATUSES)) if not fail_statuses: raise DeadmanError("[probe] 'fail_statuses' must list at least one n8n execution status") for status in fail_statuses: if status not in ALL_STATUSES: raise DeadmanError("[probe] 'fail_statuses': %r is not an n8n execution status " "(choose from %s)" % (status, ", ".join(ALL_STATUSES))) instance_in = cfg.get("instance", {}) if not isinstance(instance_in, dict): raise DeadmanError("[instance] must be a table") _reject_unknown(instance_in, INSTANCE_KEYS, "[instance]") instance = None if simulate: # --simulate answers every query from a fixture, so the instance is not # used at all. Shape is still checked when a value is present, but # nothing is REQUIRED and no environment variable is read: that is what # lets a stranger with no n8n run the demo on the example config as it # ships, with N8N_API_KEY nowhere in their environment. if "base_url" in instance_in: _https_url(_require_string(instance_in, "base_url", "[instance]"), "[instance] 'base_url'") if "api_key" in instance_in and not isinstance(instance_in["api_key"], str): raise DeadmanError("[instance] 'api_key' must be a string") if "api_key_env" in instance_in and (not isinstance(instance_in["api_key_env"], str) or not instance_in["api_key_env"].strip()): raise DeadmanError("[instance] 'api_key_env' must be the NAME of an environment variable") else: base_url = _https_url(_require_string(instance_in, "base_url", "[instance]"), "[instance] 'base_url'") api_key, api_key_env = _instance_key(instance_in) instance = {"base_url": base_url, "api_key": api_key, "api_key_env": api_key_env} workflows_in = cfg.get("workflow", []) if not isinstance(workflows_in, list): raise DeadmanError("[[workflow]] must be a list of tables") if not workflows_in: raise DeadmanError("at least one [[workflow]] block is required (a workflow with an " "expected cadence is the whole point of a deadman switch)") workflows = [] seen_ids = set() for index, item in enumerate(workflows_in, start=1): where = "[[workflow]] #%d" % index if not isinstance(item, dict): raise DeadmanError("%s: must be a table" % where) _reject_unknown(item, WORKFLOW_KEYS, where) workflow_id = _require_string(item, "id", where) if workflow_id in seen_ids: raise DeadmanError("[[workflow]]: duplicate workflow id %r (each workflow is " "declared once)" % workflow_id) seen_ids.add(workflow_id) name = item.get("name", workflow_id) if not isinstance(name, str) or not name.strip(): raise DeadmanError("%s: 'name' must be a non-empty string" % where) if "every" not in item: raise DeadmanError("%s (id %s): 'every' is required — the expected cadence, " "like '15m' or '1d'" % (where, workflow_id)) cadence_s = parse_duration(item["every"], "%s (id %s) 'every'" % (where, workflow_id)) grace_s = parse_duration(item.get("grace", "5m"), "%s (id %s) 'grace'" % (where, workflow_id)) workflows.append({ "id": workflow_id, "name": name.strip(), "cadence_s": cadence_s, "grace_s": grace_s, "window_s": cadence_s + grace_s, }) alert_in = cfg.get("alert", {}) if not isinstance(alert_in, dict): raise DeadmanError("[alert] must be a table") _reject_unknown(alert_in, ALERT_KEYS, "[alert]") webhooks = [] hooks_in = alert_in.get("webhook", []) if not isinstance(hooks_in, list): raise DeadmanError("[[alert.webhook]] must be a list of tables") for index, item in enumerate(hooks_in, start=1): where = "[[alert.webhook]] #%d" % index if not isinstance(item, dict): raise DeadmanError("%s: must be a table" % where) _reject_unknown(item, WEBHOOK_KEYS, where) url = _https_url(_require_string(item, "url", where), "%s 'url'" % where) hook_timeout = float(item.get("timeout_s", timeout_s)) if isinstance(hook_timeout, bool) or not isinstance(hook_timeout, (int, float)) or hook_timeout <= 0: raise DeadmanError("%s: 'timeout_s' must be a positive number of seconds" % where) webhooks.append({"url": url, "name": (item.get("name") or url), "timeout_s": hook_timeout}) emails = [] mail_in = alert_in.get("email", []) if not isinstance(mail_in, list): raise DeadmanError("[[alert.email]] must be a list of tables") for index, item in enumerate(mail_in, start=1): where = "[[alert.email]] #%d" % index if not isinstance(item, dict): raise DeadmanError("%s: must be a table" % where) _reject_unknown(item, EMAIL_KEYS, where) host = _require_string(item, "host", where) port = item.get("port", 587) if isinstance(port, bool) or not isinstance(port, int) or not (0 < port < 65536): raise DeadmanError("%s: 'port' must be an integer between 1 and 65535" % where) sender = _require_string(item, "from", where) recipients = item.get("to") if isinstance(recipients, str): recipients = [recipients] if not isinstance(recipients, list) or not recipients: raise DeadmanError("%s: 'to' must be a non-empty list of addresses" % where) for address in recipients: if not isinstance(address, str) or "@" not in address.strip(): raise DeadmanError("%s: 'to' contains %r, which is not an address" % (where, address)) security = item.get("security", "starttls") if security not in ("starttls", "ssl", "none"): raise DeadmanError("%s: 'security' must be 'starttls', 'ssl' or 'none'" % where) password = item.get("password") if password is not None and not isinstance(password, str): raise DeadmanError("%s: 'password' must be a string" % where) password_env = item.get("password_env") if password_env is not None and not isinstance(password_env, str): raise DeadmanError("%s: 'password_env' must be the NAME of an environment variable" % where) emails.append({ "name": item.get("name") or ("%s -> %s" % (host, ", ".join(recipients))), "host": host, "port": port, "from": sender, "to": [a.strip() for a in recipients], "username": item.get("username"), "password": password, "password_env": password_env, "security": security, "timeout_s": float(item.get("timeout_s", timeout_s)), }) return { "path": path, "probe": { "state_file": state_file.strip(), "cooldown_s": cooldown_s, "timeout_s": timeout_s, "error_lookback_s": lookback_s, "max_pages": max_pages, "fail_statuses": fail_statuses, }, "instance": instance, "workflows": workflows, "webhooks": webhooks, "emails": emails, } def _instance_key(instance_in): """The API key, or the name of the environment variable holding it. A key in a config file is a real risk for a buyer; api_key_env is the documented way to keep it out of the file. """ api_key = instance_in.get("api_key") api_key_env = instance_in.get("api_key_env") if api_key is not None and not isinstance(api_key, str): raise DeadmanError("[instance] 'api_key' must be a string") if api_key_env is not None and (not isinstance(api_key_env, str) or not api_key_env.strip()): raise DeadmanError("[instance] 'api_key_env' must be the NAME of an environment variable") if api_key and api_key_env: raise DeadmanError("[instance]: set either 'api_key' or 'api_key_env', not both") if not api_key and not api_key_env: raise DeadmanError("[instance]: 'api_key' is required (or set 'api_key_env' to the name " "of the environment variable holding it)") if api_key_env and not api_key: value = os.environ.get(api_key_env.strip()) if not value: raise DeadmanError("[instance] 'api_key_env' names %r, which is not set in this " "environment" % api_key_env.strip()) api_key = value return api_key.strip(), (api_key_env.strip() if api_key_env else None) # -------------------------------------------------------------------------- # the two sources of execution rows: the live API, and a fixture # -------------------------------------------------------------------------- class SourceFault(Exception): """The probe could not get trustworthy data. Never treated as workflow silence.""" def __init__(self, kind, detail): super().__init__("%s: %s" % (kind, detail)) self.kind = kind self.detail = detail class N8nSource: """One paged request per workflow per poll. Verified against https://docs.n8n.io/connect/n8n-api/build-and-run/executions#get-executions : the endpoint is GET /api/v1/executions, authenticated by the header named in the spec's ApiKeyAuth scheme (X-N8N-API-KEY), and the response is {"data": [...], "nextCursor": }. """ name = "n8n" def __init__(self, base_url, api_key, timeout_s, opener=None): self.url = base_url.rstrip("/") + "/api/v1/executions" # servers[].url is "/api/v1" self.api_key = api_key self.timeout_s = timeout_s # opener is injectable so the test suite can drive a real HTTP shape # (HTTPError / URLError / body bytes) without touching the network. self.opener = opener or urllib.request.urlopen def page(self, params, cursor): query = dict(params) if cursor: query["cursor"] = cursor # verified param name full = self.url + "?" + urllib.parse.urlencode(query) request = urllib.request.Request( full, headers={ # header name verified in the spec's securitySchemes.ApiKeyAuth "X-N8N-API-KEY": self.api_key, "Accept": "application/json", "User-Agent": "deadman/1.0 (n8n deadman switch)", }, ) try: with self.opener(request, timeout=self.timeout_s) as response: body = response.read() except urllib.error.HTTPError as exc: # 401/403 is how a rotated key or a lapsed Cloud trial shows up. code = getattr(exc, "code", 0) if code in (401, 403): raise SourceFault("auth", "n8n refused the API key with HTTP %s at %s" % (code, self.url)) if code == 404: raise SourceFault("unreachable", "n8n answered 404 for %s — is the base_url right, and is the " "public API enabled (N8N_PUBLIC_API_DISABLED)?" % self.url) if code == 429: headers = getattr(exc, "headers", None) retry_after = headers.get("Retry-After") if headers is not None else None raise SourceFault("rate_limited", "n8n rate-limited the probe (429)%s" % ("; Retry-After %ss" % retry_after if retry_after else "")) raise SourceFault("http", "n8n answered HTTP %s at %s" % (code, self.url)) except urllib.error.URLError as exc: raise SourceFault("unreachable", "could not reach %s: %s" % (self.url, exc.reason)) except OSError as exc: # includes socket.timeout and connection resets raise SourceFault("unreachable", "could not reach %s: %s" % (self.url, exc)) try: payload = json.loads(body.decode("utf-8")) except (UnicodeDecodeError, ValueError) as exc: raise SourceFault("malformed", "n8n replied with something that is not JSON: %s" % exc) if not isinstance(payload, dict) or not isinstance(payload.get("data"), list): # `data` is a required member of the response per the spec. raise SourceFault("malformed", "n8n replied without the required 'data' array (keys: %s)" % ", ".join(sorted(payload) if isinstance(payload, dict) else str(payload)[:80])) return payload["data"], payload.get("nextCursor") def fetch(self, workflow_id, started_after, max_pages): params = { "workflowId": workflow_id, # verified param name "limit": API_LIMIT_MAX, # spec maximum for this endpoint "startedAfter": started_after, # verified param name (date-time) # includeData is deliberately NOT sent: its default is "false" per the # spec, and execution payloads are large and useless here. } rows, cursor, pages = [], None, 0 while True: page_rows, cursor = self.page(params, cursor) pages += 1 rows.extend(page_rows) if not cursor or pages >= max_pages: break return rows, bool(cursor) and pages >= max_pages class FixtureSource: """The same queries, answered from a JSON file on disk. Fixture shape (a plain, hand-writable file): { "fault": null, // or {"kind": "auth", "detail": "HTTP 401"} // or {"kind": "unreachable", "detail": "connection refused"} "executions": [ {"id": "12", "workflowId": "Ab12", "status": "success", "startedAt": "2026-11-13T09:00:00Z"}, ... ] } `executions` items use the same field names as n8n's `data` items (verified above), so a buyer can paste a real API response straight in. Filtering is applied here exactly as the decision logic applies it, which is what makes --simulate honest: the same code path runs, only the rows come from a file. """ name = "fixture" def __init__(self, path): self.path = path try: with open(path, "r", encoding="utf-8") as handle: payload = json.load(handle) except FileNotFoundError: raise DeadmanError("simulate fixture not found: %s" % path) except OSError as exc: raise DeadmanError("simulate fixture could not be read (%s): %s" % (path, exc)) except ValueError as exc: raise DeadmanError("simulate fixture is not valid JSON (%s): %s" % (path, exc)) if not isinstance(payload, dict): raise DeadmanError("simulate fixture must be a JSON object with 'executions' " "and optionally 'fault'") rows = payload.get("executions", []) if not isinstance(rows, list): raise DeadmanError("simulate fixture: 'executions' must be a list") for index, row in enumerate(rows): if not isinstance(row, dict): raise DeadmanError("simulate fixture: executions[%d] is not an object" % index) for field in (EXECUTION_ID_FIELD, EXECUTION_WORKFLOW_FIELD, EXECUTION_STATUS_FIELD): if field not in row: raise DeadmanError( "simulate fixture: executions[%d] has no '%s' (n8n's own API marks it required)" % (index, field)) self.rows = rows self.fault = payload.get("fault") if self.fault is not None: if not isinstance(self.fault, dict) or "kind" not in self.fault: raise DeadmanError("simulate fixture: 'fault' must be null or an object with 'kind' " "(auth | unreachable | http | malformed)") def fetch(self, workflow_id, started_after, max_pages): if self.fault is not None: raise SourceFault(str(self.fault.get("kind", "fault")), str(self.fault.get("detail", "injected by the fixture"))) # Return everything for this workflow; the decision logic applies the # window filter itself, so simulate and live cannot diverge. rows = [row for row in self.rows if str(row.get(EXECUTION_WORKFLOW_FIELD)) == str(workflow_id)] return rows, False # -------------------------------------------------------------------------- # state # -------------------------------------------------------------------------- def empty_state(): return {"version": STATE_VERSION, "last_check_ok": None, "last_fault": None, "alerts": {}, "workflows": {}} def load_state(path, warn): """Read state; a corrupt or unreadable file is set aside and we start fresh. Losing dedupe history is better than crashing: a monitor that dies on its own state file stops monitoring everything. """ try: with open(path, "r", encoding="utf-8") as handle: state = json.load(handle) except FileNotFoundError: return empty_state() except ValueError as exc: warn("state file %s is not valid JSON (%s): starting fresh, broken copy kept as %s" % (path, exc, path + ".corrupt")) _set_aside(path, "corrupt") return empty_state() except OSError as exc: warn("state file %s could not be read (%s): starting fresh" % (path, exc)) return empty_state() if not isinstance(state, dict) or not isinstance(state.get("alerts"), dict) \ or not isinstance(state.get("workflows"), dict): warn("state file %s is not the shape deadman writes (it needs 'alerts' and 'workflows' " "objects): starting fresh, broken copy kept as %s" % (path, path + ".corrupt")) _set_aside(path, "corrupt") return empty_state() version = state.get("version") if version != STATE_VERSION: warn("state file %s reports version %r, this deadman writes version %d: starting fresh" % (path, version, STATE_VERSION)) return empty_state() base = empty_state() base.update(state) return base def _set_aside(path, tag): try: os.replace(path, "%s.%s" % (path, tag)) except OSError: pass def save_state(path, state): """Atomic write: temp file in the same directory, then os.replace.""" state["version"] = STATE_VERSION directory = os.path.dirname(os.path.abspath(path)) or "." try: os.makedirs(directory, exist_ok=True) temporary = os.path.join(directory, ".%s.tmp%d" % (os.path.basename(path), os.getpid())) with open(temporary, "w", encoding="utf-8") as handle: json.dump(state, handle, indent=2, sort_keys=True) handle.write("\n") os.replace(temporary, path) except OSError as exc: raise DeadmanError("state file %s could not be written: %s" % (path, exc)) # -------------------------------------------------------------------------- # alert episodes: once, then cooldown, then recovery exactly once # -------------------------------------------------------------------------- def _new_episode(): return {"open": False, "first_seen": None, "last_alert_at": None, "alert_count": 0, "resolved_count": 0, "detail": None} def evaluate_episode(state, key, active, now_dt, cooldown_s, detail, deliver): """One alerting rule, applied to one key. Returns the notice sent, if any. * condition newly true -> ALERT now, once * still true -> RE-ALERT only after cooldown_s has passed since the last delivery (0 disables re-alerting) * condition now false -> RECOVERY, exactly once (the episode closes) `last_alert_at` advances only when `deliver` returns True, so a broken webhook cannot silently swallow an alert: it stays hot and is retried. """ now_text = iso(now_dt) episode = state["alerts"].get(key) if not isinstance(episode, dict): episode = _new_episode() action = None if active: if not episode.get("open"): episode["open"] = True episode["first_seen"] = now_text episode["last_alert_at"] = None action = "alert" else: last = episode.get("last_alert_at") if last is None: # The episode is open but nothing was ever delivered (every # channel refused it). Retry now rather than wait out a cooldown # clock that never started. action = "alert" elif cooldown_s: last_dt = parse_instant(last, "state %s last_alert_at" % key) if (now_dt - last_dt).total_seconds() >= cooldown_s: action = "re-alert" elif episode.get("open"): episode["open"] = False episode["resolved_count"] = int(episode.get("resolved_count", 0)) + 1 episode["last_alert_at"] = None action = "recovery" episode["detail"] = detail state["alerts"][key] = episode if action is None: return None notice = {"key": key, "action": action, "at": now_text, "open_since": episode.get("first_seen") if action != "recovery" else None, "detail": detail or {}} delivered = deliver(notice) if action in ("alert", "re-alert"): # The clock advances whether or not a remote channel took it, on purpose. # deadman's first channel is stdout plus the exit code, and that one # cannot fail; making delivery success gate the cooldown would turn one # dead webhook into an alert re-firing on every single poll, forever. # Instead a failing channel is said out loud every poll (stderr, where # cron mails it) and recorded on the episode as delivery_failed. episode["last_alert_at"] = now_text episode["alert_count"] = int(episode.get("alert_count", 0)) + 1 episode["delivery_failed"] = not delivered return notice def open_alerts(state): return sorted(key for key, episode in state["alerts"].items() if isinstance(episode, dict) and episode.get("open")) # -------------------------------------------------------------------------- # notice rendering and delivery # -------------------------------------------------------------------------- def notice_line(notice): detail = notice.get("detail") or {} who = detail.get("workflow") or "the probe itself" if notice["action"] == "recovery": return "RECOVERED %s (%s)" % (notice["key"], detail.get("summary", "")) return "%s %s %s" % (notice["action"].upper(), notice["key"], detail.get("summary", who)) def notice_body(notice): detail = notice.get("detail") or {} lines = ["%s %s" % (notice["action"].upper(), notice["key"]), "time: %s" % notice["at"]] if notice.get("open_since"): lines.append("open since: %s" % notice["open_since"]) for key in ("workflow", "cadence", "grace", "last_execution", "missing_for", "summary", "kind", "detail", "failures", "threshold", "url", "action_taken"): if detail.get(key) is not None: lines.append("%s: %s" % (key, detail[key])) lines.append("state: %s" % json.dumps({"action": notice["action"], "key": notice["key"]}, sort_keys=True)) return "\n".join(lines) class WebhookChannel: def __init__(self, spec, opener=None): self.name = spec["name"] self.url = spec["url"] self.timeout_s = spec["timeout_s"] self.opener = opener or urllib.request.urlopen def send(self, notice): payload = json.dumps({ "text": notice_line(notice), "subject": notice_line(notice), "action": notice["action"], "key": notice["key"], "at": notice["at"], "detail": notice.get("detail") or {}, }).encode("utf-8") request = urllib.request.Request( self.url, data=payload, method="POST", headers={"Content-Type": "application/json", "User-Agent": "deadman/1.0"}) with self.opener(request, timeout=self.timeout_s) as response: code = getattr(response, "status", 200) if not 200 <= int(code) < 300: raise DeadmanError("webhook %s answered HTTP %s" % (self.url, code)) class EmailChannel: def __init__(self, spec, mailer=None): self.spec = spec self.name = spec["name"] # mailer is injectable so tests never open an SMTP socket. self.mailer = mailer or self._smtp_send def _smtp_send(self, host, port, message, sender, recipients, username, password, security, timeout_s): if security == "ssl": connection = smtplib.SMTP_SSL(host, port, timeout=timeout_s) else: connection = smtplib.SMTP(host, port, timeout=timeout_s) try: if security == "starttls": connection.starttls() if username: connection.login(username or "", password or "") connection.sendmail(sender, recipients, message) finally: try: connection.quit() except Exception: # a half-open socket must not mask the send result pass def send(self, notice): spec = self.spec password = spec.get("password") if not password and spec.get("password_env"): password = os.environ.get(spec["password_env"]) body = ("From: %s\r\nTo: %s\r\nSubject: %s\r\nContent-Type: text/plain; charset=utf-8\r\n\r\n%s\r\n" % (spec["from"], ", ".join(spec["to"]), notice_line(notice), notice_body(notice))) self.mailer(spec["host"], spec["port"], body, spec["from"], spec["to"], spec.get("username"), password, spec["security"], spec["timeout_s"]) def build_channels(config, opener=None, mailer=None): channels = [WebhookChannel(spec, opener=opener) for spec in config["webhooks"]] channels += [EmailChannel(spec, mailer=mailer) for spec in config["emails"]] return channels # -------------------------------------------------------------------------- # the check itself # -------------------------------------------------------------------------- def _row_field(row, field): return row.get(field) if isinstance(row, dict) else None def _row_started(row): raw = _row_field(row, EXECUTION_STARTED_FIELD) if raw is None: return None try: # n8n sends ISO date-time strings ("format: date-time", verified). # A missing zone in a server reply is assumed UTC and said so nowhere: # this is the API's own field, not user config. return parse_instant(raw, "n8n %s" % EXECUTION_STARTED_FIELD) except DeadmanError: try: return datetime.fromtimestamp(float(raw), tz=timezone.utc) except (TypeError, ValueError, OSError): return None def check_workflows(config, source, state, now_dt, warn, deliver): """Silence and error verdicts for every declared workflow. A SourceFault raised by source.fetch propagates: if one workflow's read failed, the whole poll is untrustworthy, so run_check reports the PROBE and judges no workflow. That is what stops an expired API key from being reported as N dead workflows. """ probe = config["probe"] lookback_start = now_dt - timedelta(seconds=probe["error_lookback_s"]) notices = [] for workflow in config["workflows"]: workflow_id = workflow["id"] window_start = now_dt - timedelta(seconds=workflow["window_s"]) # Ask n8n for the EARLIER of the two instants, so one read covers both # questions. Taking the later one would silently cap the error lookback # at the silence window: with the default 24h lookback and an hourly # cadence, every failure older than an hour would have been invisible. rows, truncated = source.fetch( workflow_id, iso(lookback_start if lookback_start < window_start else window_start), probe["max_pages"]) if truncated: warn("workflow %s: more executions than %d pages of %d inside the lookback; the " "silence verdict is still exact, the error verdict only saw the pages it got " "(lower [probe] error_lookback_s or raise max_pages)" % (workflow_id, probe["max_pages"], API_LIMIT_MAX)) own = [row for row in rows if str(_row_field(row, EXECUTION_WORKFLOW_FIELD)) == str(workflow_id)] started = [] for row in own: moment = _row_started(row) if moment is not None: started.append((moment, row)) # n8n documents `startedAt` as null until the run leaves the queue, so a # queued-but-never-started execution is NOT evidence that it ran. started.sort(key=lambda pair: pair[0]) memory = state["workflows"].get(workflow_id) if not isinstance(memory, dict): memory = {"last_seen_id": None, "last_seen_at": None, "last_status": None, "newest_failed_at": None, "seen_error_ids": []} ran_in_window = [pair for pair in started if pair[0] >= window_start] newest = started[-1] if started else (None, None) failures = [pair for pair in started if _row_field(pair[1], EXECUTION_STATUS_FIELD) in probe["fail_statuses"] and pair[0] >= lookback_start] seen_ids = [str(x) for x in memory.get("seen_error_ids") or []] new_failures = [pair for pair in failures if str(_row_field(pair[1], EXECUTION_ID_FIELD)) not in seen_ids] if newest[1] is not None: memory["last_seen_id"] = str(_row_field(newest[1], EXECUTION_ID_FIELD)) memory["last_seen_at"] = iso(newest[0]) memory["last_status"] = _row_field(newest[1], EXECUTION_STATUS_FIELD) if failures: memory["newest_failed_at"] = iso(failures[-1][0]) for pair in failures: identifier = str(_row_field(pair[1], EXECUTION_ID_FIELD)) if identifier not in seen_ids: seen_ids.append(identifier) memory["seen_error_ids"] = seen_ids[-200:] # bounded, newest last state["workflows"][workflow_id] = memory last_execution = ("id %s at %s (status %s)" % (memory["last_seen_id"], memory["last_seen_at"], memory["last_status"]) if memory.get("last_seen_id") else "none the probe can see") # --- class 1: SILENCE ------------------------------------------------- silent = not ran_in_window detail = { "workflow": "%s (id %s)" % (workflow["name"], workflow_id), "summary": ("no execution in %ss (expected every %ss + %ss grace); last seen: %s" % (int(workflow["window_s"]), int(workflow["cadence_s"]), int(workflow["grace_s"]), last_execution)) if silent else "running again", "cadence": "%ss" % int(workflow["cadence_s"]), "grace": "%ss" % int(workflow["grace_s"]), "last_execution": last_execution, } notice = evaluate_episode( state, "silence:%s" % workflow_id, silent, now_dt, config["probe"]["cooldown_s"], detail, deliver) if notice: notices.append(notice) # --- class 2: ERROR --------------------------------------------------- # The episode OPENS on a failed execution id the probe has not reported # before, and stays open until a NEWER execution finished successfully. # "Nothing new failed since the last poll" is NOT recovery: a workflow # whose last run failed is still failed, and telling someone they # recovered when they did not is worse than staying quiet. error_key = "error:%s" % workflow_id previous = state["alerts"].get(error_key) already_open = bool(previous.get("open")) if isinstance(previous, dict) else False last_is_failure = memory.get("last_status") in probe["fail_statuses"] error_open = bool(new_failures) or (already_open and last_is_failure) if new_failures: identifiers = ", ".join(str(_row_field(p[1], EXECUTION_ID_FIELD)) for p in new_failures[:5]) summary = "%d new failed execution(s): %s (statuses %s)" % ( len(new_failures), identifiers, ", ".join(sorted({str(_row_field(p[1], EXECUTION_STATUS_FIELD)) for p in new_failures}))) elif error_open: summary = "still failing: last execution is %s" % memory.get("last_status") else: summary = "no new failed execution; last execution status %s" % memory.get("last_status") detail = { "workflow": "%s (id %s)" % (workflow["name"], workflow_id), "summary": summary, "failures": len(new_failures), "lookback_s": int(config["probe"]["error_lookback_s"]), } notice = evaluate_episode( state, error_key, error_open, now_dt, config["probe"]["cooldown_s"], detail, deliver) if notice: notices.append(notice) return notices def run_check(config, source, now_dt, out, deliver, warn=None): """One poll. Returns (exit code, notices sent this poll).""" warn = warn or (lambda message: print("warn: %s" % message, file=sys.stderr)) state = load_state(config["probe"]["state_file"], warn) notices = [] fault = None try: notices.extend(check_workflows(config, source, state, now_dt, warn, deliver)) except SourceFault as exc: fault = exc # Nothing else was measured; leave every workflow episode untouched. warn("probe fault (%s): %s — workflow verdicts are NOT computed this run" % (exc.kind, exc.detail)) # --- class 3: PROBE-FAULT ---------------------------------------------- probe_detail = None if fault is not None: probe_detail = { # One stable token for a person grepping a cron mail or a log. "summary": "PROBE-FAULT: deadman could not read n8n (%s): %s" % (fault.kind, fault.detail), "kind": fault.kind, "detail": fault.detail, "action_taken": "no workflow was judged; open workflow alerts are unchanged", } state["last_fault"] = ({"kind": fault.kind, "detail": fault.detail, "at": iso(now_dt)} if fault is not None else None) notice = evaluate_episode(state, "probe", fault is not None, now_dt, config["probe"]["cooldown_s"], probe_detail, deliver) if notice: notices.append(notice) if fault is None: state["last_check_ok"] = iso(now_dt) else: # Keep every declared workflow present in state so a reader can see # it was declared but not judged. for workflow in config["workflows"]: state["workflows"].setdefault(workflow["id"], {"last_seen_id": None, "last_seen_at": None, "last_status": None, "newest_failed_at": None, "seen_error_ids": []}) state["open_alerts"] = open_alerts(state) save_state(config["probe"]["state_file"], state) # A failing alert channel is itself an emergency, and it must not be a # one-line memory: as long as an alert is open and could not be delivered, # every poll says so on stderr, where cron mails it to the operator. for key in state["open_alerts"]: episode = state["alerts"].get(key) or {} if episode.get("delivery_failed"): warn("alert %s is still open and its alert channel(s) are still refusing to " "deliver it: the notice was counted as sent once and will not be repeated " "until the cooldown elapses" % key) for notice in notices: print("%s %s" % (notice["at"], notice_line(notice)), file=out) if not notices: # Said plainly: a poll with nothing NEW to send is not the same as a # healthy poll. Saying "healthy" here while an alert is still open would # be the exact lie that gets a buyer to close the terminal. if fault is not None: verdict = "PROBE-FAULT %s: %s" % (fault.kind, fault.detail) elif state["open_alerts"]: verdict = "unchanged: %d alert(s) still open (%s), already reported; last ok %s" % ( len(state["open_alerts"]), ", ".join(state["open_alerts"]), state.get("last_check_ok")) else: verdict = "healthy: %d workflow(s) checked, nothing open (last ok %s)" % ( len(config["workflows"]), state.get("last_check_ok")) print("%s %s" % (iso(now_dt), verdict), file=out) if fault is not None: return EXIT_PROBE, notices if state["open_alerts"]: return EXIT_ALERT, notices return EXIT_OK, notices def build_source(config, simulate_path, opener=None): if simulate_path: return FixtureSource(simulate_path) return N8nSource(config["instance"]["base_url"], config["instance"]["api_key"], config["probe"]["timeout_s"], opener=opener) def main(argv=None, stdout=None, opener=None, mailer=None): parser = argparse.ArgumentParser( prog="deadman.py", description=__doc__.split("\n\n")[0], epilog=EXIT_DOC, formatter_class=argparse.RawDescriptionHelpFormatter) parser.add_argument("--config", required=True, help="path to the TOML config") parser.add_argument("--simulate", default=None, help="answer the n8n queries from this JSON fixture instead of a live " "instance; runs the identical decision path") parser.add_argument("--now", default=None, help="evaluate as if it were this instant (ISO-8601, must carry Z or an offset)") parser.add_argument("--state", default=None, help="override [probe] state_file (handy for a dry run)") args = parser.parse_args(argv) out = stdout if stdout is not None else sys.stdout try: config = load_config(args.config, simulate=args.simulate is not None) if args.state: config["probe"]["state_file"] = args.state now_dt = parse_instant(args.now, "--now") if args.now else utc_now() source = build_source(config, args.simulate, opener=opener) channels = build_channels(config, opener=opener, mailer=mailer) def deliver(notice): """Whether every configured channel accepted it. stdout and the exit code are not a channel: they always carry the verdict, which is what makes the cooldown clock safe to advance even when a webhook is down. A failing channel is named on stderr on every poll it fails, because cron mails stderr to the operator. """ if not channels: return True delivered = 0 for channel in channels: try: channel.send(notice) delivered += 1 except Exception as exc: # one broken channel must not stop the others print("warn: alert channel %s failed: %s" % (channel.name, exc), file=sys.stderr) if delivered < len(channels): print("warn: %d of %d alert channel(s) refused this notice; the alert is still " "open (exit 1) and the channel keeps being reported here every poll" % (len(channels) - delivered, len(channels)), file=sys.stderr) return False return True exit_code, _notices = run_check(config, source, now_dt, out, deliver) return exit_code except DeadmanError as exc: print("deadman: %s" % exc.message, file=sys.stderr) return EXIT_PROBE if __name__ == "__main__": sys.exit(main())