Render heartbeat with Slack Block Kit and group healthy clients

- post heartbeats via Block Kit (header / section / divider / context)
  instead of one mrkdwn blob; action alerts stay as plain text posts
- new send_blocks() on SlackNotifier with text fallback for notifications
- collapse online + canonical + distance=0 clients into one bucket;
  surface outliers (offline, synchronizing, non-canonical, lagging)
  above the healthy bucket with status emoji per row
- status emojis: green/yellow/orange/red circles for online/sync/opt/off
- dry-run patches both send and send_blocks; --debug dumps blocks JSON
  so it can be previewed in Slack's Block Kit Builder

Author: edg-l

dora_monitor/README.md

@@ -46,6 +46,8 @@ The process holds dedup state in `state_file` so restarts don't re-alert on alre
 
 Recoveries (fork resolved, caught up, back online) are posted as well.
 
+The periodic heartbeat digest uses Slack Block Kit (`{"blocks": [...]}`) with a plain-text fallback for notifications. Action alerts (offline / fork / lag / version change / missed-block) use plain mrkdwn `text` posts. Clients with status `online`, on the canonical fork, and at `distance == 0` from canonical head collapse into a single "online @ canonical" bucket so the digest highlights outliers instead of repeating identical rows; use `heartbeat_other_clients: detailed` (default) to list the healthy names, `summary` for just a count, or `off` to drop the section entirely.
+
 ## A note on what "client" means here
 
 `/api/v1/network/client_head_forks` lists Dora's **beacon (CL)** clients; their names embed the paired EL (e.g. `lighthouse-ethrex-1` is the Lighthouse beacon paired with an ethrex EL). So the offline / fork / lag signals are observed on the beacon side. An ethrex-EL crash shows up indirectly: the paired beacon's head stops advancing (sync_lag) or its status flips to non-online (offline).

dora_monitor/dora_monitor/checks.py

@@ -188,15 +188,59 @@ def check_version_drift(
             state.client_versions[name] = version
 
 
-def _format_heartbeat(
+_STATUS_EMOJI = {
+    "online": ":large_green_circle:",
+    "synchronizing": ":large_yellow_circle:",
+    "optimistic": ":large_orange_circle:",
+    "offline": ":red_circle:",
+}
+
+
+def _status_emoji(status: str) -> str:
+    return _STATUS_EMOJI.get(status, ":white_circle:")
+
+
+def _health_rank(entry: dict) -> int:
+    """Lower rank = surface higher. Sort key for ordering clients."""
+    status = entry["status"]
+    if status == "offline":
+        return 0
+    if status in ("synchronizing", "optimistic"):
+        return 1
+    if not entry["is_canonical_fork"]:
+        return 2
+    if entry["distance"] > 0:
+        return 3
+    return 4
+
+
+def _client_line(entry: dict) -> str:
+    """One-line mrkdwn rendering of a single client outlier row."""
+    parts = [
+        _status_emoji(entry["status"]),
+        f"`{entry['name']}`",
+        f"head `{entry['head_slot']}`",
+    ]
+    if entry["distance"] > 0:
+        parts.append(f"·  *{entry['distance']} behind*")
+    if not entry["is_canonical_fork"]:
+        parts.append("·  :fork_and_knife: non-canonical")
+    return "  ".join(parts)
+
+
+def _section(text: str) -> dict:
+    return {"type": "section", "text": {"type": "mrkdwn", "text": text}}
+
+
+def _build_heartbeat(
     dora: DoraClient,
     cfg: Config,
-) -> str:
-    """Compose the heartbeat digest text.
+) -> tuple[list[dict], str]:
+    """Build a Block Kit heartbeat plus a plain-text fallback.
 
-    Makes two separate HTTP requests (client_head_forks + slots) so the head
-    slot shown and the missed/orphaned counts are sampled at slightly
-    different instants. They may disagree by a slot or two; this is by
+    Makes two separate HTTP requests (client_head_forks + slots), so the
+    head slot and the missed/orphaned counts are sampled at slightly
+    different instants. They may disagree by a slot or two; that's by
     design, not a bug.
     """
     payload = dora.client_head_forks()
@@ -247,48 +291,112 @@ def _format_heartbeat(
         elif st == "orphaned":
             orphaned += 1
 
-    lines: list[str] = []
-    lines.append(
-        f":bar_chart: *Heartbeat* — canonical head slot `{canonical_slot}` "
-        f"(`{canonical_root[:14]}…`), {len(forks)} active fork(s)"
+    label = f" — {cfg.network_label}" if cfg.network_label else ""
+    blocks: list[dict] = []
+    blocks.append({
+        "type": "header",
+        "text": {"type": "plain_text", "text": f"\U0001F4CA Heartbeat{label}"},
+    })
+
+    # Network summary section.
+    status_mix = "  ".join(
+        f"{_status_emoji(k)} {v}" for k, v in sorted(status_counts.items())
+    ) or "no clients"
+    root_short = f"`{canonical_root[:14]}…`" if canonical_root else "`?`"
+    summary_text = (
+        f"Canonical head: slot `{canonical_slot}`  ·  root {root_short}\n"
+        f"Active forks: *{len(forks)}*  ·  Status mix: {status_mix}"
     )
-    status_summary = ", ".join(f"{k}:{v}" for k, v in sorted(status_counts.items())) or "no clients"
-    lines.append(f"Network clients: {status_summary}")
+    blocks.append(_section(summary_text))
+    blocks.append({"type": "divider"})
 
+    # Matched (client_match) section.
     if matched:
-        lines.append(f"*{cfg.client_match}* ({len(matched)} matched):")
-        for e in sorted(matched, key=lambda x: x["name"]):
-            mark = "" if e["is_canonical_fork"] else " :fork_and_knife:"
-            lines.append(
-                f"  • `{e['name']}` status=`{e['status']}` head=`{e['head_slot']}` "
-                f"distance=`{e['distance']}`{mark}"
-            )
-        lines.append(
-            f"  proposals in last {window} slots: {total_matched_proposals} "
-            f"(missed={missed}, orphaned={orphaned})"
+        matched_sorted = sorted(matched, key=lambda x: (_health_rank(x), x["name"]))
+        matched_lines = [
+            f":rocket: *{cfg.client_match}* ({len(matched_sorted)} matched)"
+        ]
+        # Collapse the healthy bucket if every matched client is healthy.
+        healthy = [e for e in matched_sorted if _health_rank(e) == 4]
+        outliers = [e for e in matched_sorted if _health_rank(e) != 4]
+        for e in outliers:
+            matched_lines.append(_client_line(e))
+        if healthy:
+            if len(healthy) == len(matched_sorted):
+                names = ", ".join(f"`{e['name']}`" for e in healthy)
+                matched_lines.append(
+                    f"{_status_emoji('online')} *all online @ canonical* "
+                    f"({len(healthy)}): {names}"
+                )
+            else:
+                for e in healthy:
+                    matched_lines.append(_client_line(e))
+        matched_lines.append("")
+        matched_lines.append(
+            f"Proposals in last {window} slots: *{total_matched_proposals}*  "
+            f"(missed *{missed}*, orphaned *{orphaned}*)"
         )
+        blocks.append(_section("\n".join(matched_lines)))
     else:
-        lines.append(f"No clients matching `{cfg.client_match}` found.")
+        blocks.append(_section(f":mag: No clients matching `{cfg.client_match}` found."))
 
+    # Other clients section (collapsed healthy bucket + per-client outliers).
     mode = (cfg.heartbeat_other_clients or "summary").lower()
     if others and mode != "off":
-        if mode == "detailed":
-            lines.append(f"Other clients ({len(others)}):")
-            for e in sorted(others, key=lambda x: x["name"]):
-                mark = "" if e["is_canonical_fork"] else " :fork_and_knife:"
+        blocks.append({"type": "divider"})
+        others_sorted = sorted(others, key=lambda x: (_health_rank(x), x["name"]))
+        healthy = [e for e in others_sorted if _health_rank(e) == 4]
+        outliers = [e for e in others_sorted if _health_rank(e) != 4]
+
+        lines = [f":desktop_computer: *Other clients* ({len(others_sorted)})"]
+        for e in outliers:
+            lines.append(_client_line(e))
+        if healthy:
+            if mode == "detailed":
+                names = ", ".join(f"`{e['name']}`" for e in healthy)
                 lines.append(
-                    f"  • `{e['name']}` status=`{e['status']}` head=`{e['head_slot']}` "
-                    f"distance=`{e['distance']}`{mark}"
+                    f"{_status_emoji('online')} *online @ canonical* "
+                    f"({len(healthy)}): {names}"
                 )
+            else:  # summary
+                lines.append(
+                    f"{_status_emoji('online')} *online @ canonical*: "
+                    f"{len(healthy)} client(s)"
+                )
+        blocks.append(_section("\n".join(lines)))
+
+    # Footer context block (small grey).
+    footer = (
+        f"_Polling `{cfg.dora_url}` every {cfg.poll_interval}s · "
+        f"matching `{cfg.client_match}`_"
+    )
+    blocks.append({"type": "context", "elements": [{"type": "mrkdwn", "text": footer}]})
+
+    # Plain-text fallback for notifications / non-Block-Kit clients.
+    fb_status_mix = ", ".join(f"{k}:{v}" for k, v in sorted(status_counts.items())) or "no clients"
+    fb_lines = [
+        f"Heartbeat — canonical head {canonical_slot} ({len(forks)} fork(s), {fb_status_mix})",
+    ]
+    if matched:
+        unhealthy_matched = sum(1 for e in matched if _health_rank(e) != 4)
+        if unhealthy_matched == 0:
+            fb_lines.append(
+                f"{cfg.client_match}: {len(matched)} client(s) all healthy @ {canonical_slot}; "
+                f"{total_matched_proposals} proposals (missed {missed}, orphan {orphaned})"
+            )
         else:
-            non_canonical = [e for e in others if not e["is_canonical_fork"]]
-            non_online = [e for e in others if e["status"] != "online"]
-            lines.append(
-                f"Other clients: {len(others)} total, "
-                f"{len(non_online)} non-online, {len(non_canonical)} off-canonical"
+            fb_lines.append(
+                f"{cfg.client_match}: {unhealthy_matched}/{len(matched)} unhealthy; "
+                f"{total_matched_proposals} proposals (missed {missed}, orphan {orphaned})"
             )
+    if others:
+        unhealthy_others = sum(1 for e in others if _health_rank(e) != 4)
+        fb_lines.append(
+            f"others: {len(others) - unhealthy_others}/{len(others)} healthy"
+        )
+    fallback = "\n".join(fb_lines)
 
-    return "\n".join(lines)
+    return blocks, fallback
 
 
 def maybe_heartbeat(
@@ -304,11 +412,11 @@ def maybe_heartbeat(
     if state.last_heartbeat_ts > 0 and (now - state.last_heartbeat_ts) < interval_s:
         return
     try:
-        text = _format_heartbeat(dora, cfg)
+        blocks, fallback = _build_heartbeat(dora, cfg)
     except Exception as e:
         log.exception("heartbeat compose failed: %s", e)
         return
-    slack.send(text)
+    slack.send_blocks(blocks, fallback)
     state.last_heartbeat_ts = now
 
 

dora_monitor/dora_monitor/main.py

@@ -41,10 +41,17 @@ def cli() -> None:
     dora = DoraClient(cfg.dora_url, timeout=cfg.http_timeout)
     slack = SlackNotifier(cfg.slack_webhook_url, cfg.network_label, timeout=cfg.http_timeout)
     if args.dry_run:
+        import json as _json
         prefix = slack._prefix()
         def _dry_send(text: str) -> None:
             print(f"[DRY-RUN] {prefix}{text}")
+        def _dry_send_blocks(blocks: list, fallback: str) -> None:
+            print(f"[DRY-RUN] {prefix}{fallback}")
+            if args.debug:
+                print("[DRY-RUN blocks JSON]")
+                print(_json.dumps(blocks, indent=2))
         slack.send = _dry_send  # type: ignore[assignment]
+        slack.send_blocks = _dry_send_blocks  # type: ignore[assignment]
 
     state = load_state(None if args.reset_state else cfg.state_file)
     if args.dry_run:

dora_monitor/dora_monitor/slack.py

@@ -40,6 +40,21 @@ def send(self, text: str) -> None:
         for i, chunk in enumerate(chunks, 1):
             self._post(f"{chunk}\n_({i}/{total})_")
 
+    def send_blocks(self, blocks: list[dict], fallback: str) -> None:
+        """Post a Slack Block Kit message. `fallback` is the plain-text
+        version shown in notifications and clients that can't render blocks.
+        """
+        body = {"blocks": blocks, "text": f"{self._prefix()}{fallback}"}
+        try:
+            r = requests.post(self.webhook_url, json=body, timeout=self.timeout)
+            if r.status_code == 429:
+                retry = r.headers.get("Retry-After", "?")
+                log.error("slack rate-limited (429, retry-after=%s); blocks dropped", retry)
+            elif r.status_code >= 300:
+                log.error("slack webhook (blocks) failed: %s %s", r.status_code, r.text[:200])
+        except requests.RequestException as e:
+            log.error("slack webhook (blocks) error: %s", e)
+
 
 def _split_on_lines(text: str, limit: int) -> list[str]:
     """Split text on newline boundaries into chunks of at most `limit` chars.