feat: implement context snapshot fetch (Task 8)

docs/hermes-implementation-plan.md

@@ -86,18 +86,13 @@
 
 ### Task 8: Implement context snapshot fetch
 
-**Objective:** Pre-fetch cheap context (peer count, sync status, container status) at receive time with Prometheus fallback.
-
-**Files:**
-- Modify: `webhook-receiver/src/server.py`
-- Create: `webhook-receiver/src/context_fetcher.py`
-
-**Steps:**
-1. Implement Prometheus queries for peer count, sync status, etc.
-2. Implement Docker socket queries for container status
-3. Implement fallback chain: primary source → Prometheus last value → "unavailable"
-4. Write tests
-5. Verify context snapshot is attached to HermesAlert
+[x] Complete — Created `webhook-receiver/src/webhook_receiver/context_fetcher.py`:
+  - `_query_prometheus`: Queries Prometheus for instant vector values (2s timeout)
+  - `_get_docker_container_status`: Queries Docker socket for container state
+  - `fetch_context_snapshot`: Orchestrates fallback chain (Docker → Prometheus → "unavailable")
+  - Queries peer count and validator count based on client type (lighthouse, prysm, teku, etc.)
+  - Integrated into `WebhookHandler` — context snapshot attached to each `HermesAlert` before processing
+  - 10 unit tests covering success, failure, and fallback scenarios
 
 ---
 

webhook-receiver/src/webhook_receiver/context_fetcher.py

@@ -0,0 +1,152 @@
+"""Context snapshot fetcher for webhook receiver.
+
+Pre-fetches cheap context (peer count, sync status, container status)
+at receive time with Prometheus fallback.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import os
+import socket
+import urllib.parse
+import urllib.request
+import urllib.error
+from typing import Optional, Tuple
+
+from .types import ContextSnapshot
+
+log = logging.getLogger(__name__)
+
+PROMETHEUS_URL = os.environ.get("PROMETHEUS_URL", "http://prometheus:9090")
+DOCKER_SOCKET_PATH = os.environ.get("DOCKER_SOCKET_PATH", "/var/run/docker.sock")
+
+
+def _query_prometheus(query: str) -> Optional[float]:
+    """Query Prometheus for a single instant vector value."""
+    url = f"{PROMETHEUS_URL}/api/v1/query?query={urllib.parse.quote(query)}"
+    try:
+        req = urllib.request.Request(url, method="GET")
+        with urllib.request.urlopen(req, timeout=2) as resp:
+            data = json.loads(resp.read())
+            result = data.get("data", {}).get("result", [])
+            if result and len(result) > 0:
+                return float(result[0]["value"][1])
+    except (
+        urllib.error.URLError,
+        json.JSONDecodeError,
+        KeyError,
+        ValueError,
+        TypeError,
+    ) as e:
+        log.debug("Prometheus query failed for %s: %s", query, e)
+    return None
+
+
+def _get_docker_container_status(container_name: str) -> Tuple[Optional[str], Optional[str]]:
+    """
+    Query Docker socket for container status.
+    Returns (status, note).
+    """
+    api_path = f"/containers/{container_name}/json"
+    request = (
+        f"GET {api_path} HTTP/1.1\r\n"
+        f"Host: docker\r\n"
+        f"Connection: close\r\n"
+        f"\r\n"
+    )
+
+    try:
+        with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as sock:
+            sock.settimeout(2.0)
+            sock.connect(DOCKER_SOCKET_PATH)
+            sock.sendall(request.encode('utf-8'))
+
+            response = b""
+            while True:
+                chunk = sock.recv(4096)
+                if not chunk:
+                    break
+                response += chunk
+
+            header_end = response.find(b"\r\n\r\n")
+            if header_end == -1:
+                return None, "malformed response"
+
+            body = response[header_end + 4:]
+            data = json.loads(body.decode('utf-8'))
+            state = data.get("State", {})
+            status = state.get("Status", "unknown")
+
+            if status == "running":
+                return "running", None
+            else:
+                return status, f"container is {status}"
+    except (socket.error, json.JSONDecodeError, UnicodeDecodeError, OSError) as e:
+        log.debug("Docker socket query failed for %s: %s", container_name, e)
+        return None, f"docker socket error: {type(e).__name__}"
+
+
+def fetch_context_snapshot(
+    host: str,
+    container: Optional[str] = None,
+    client: Optional[str] = None,
+) -> ContextSnapshot:
+    """
+    Fetch context snapshot for an alert.
+
+    Fallback chain:
+    1. Primary source (Docker socket for container status, direct metrics)
+    2. Prometheus last value
+    3. "unavailable"
+    """
+    snapshot = ContextSnapshot()
+    fallback_used = False
+
+    # 1. Container status
+    if container:
+        status, note = _get_docker_container_status(container)
+        if status:
+            snapshot.container_status = status
+            if note:
+                snapshot.container_status_note = note
+        else:
+            snapshot.container_status = "unavailable"
+            snapshot.container_status_note = "docker socket unreachable"
+            fallback_used = True
+
+    # 2. Peer count
+    peer_queries = []
+    if client == "lighthouse":
+        peer_queries = ["lighthouse_peers", "beacon_peers"]
+    elif client in ("prysm", "teku", "nimbus", "lodestar"):
+        peer_queries = ["beacon_peers", "p2p_peers"]
+
+    for q in peer_queries:
+        peers = _query_prometheus(q)
+        if peers is not None:
+            snapshot.peer_count = int(peers)
+            break
+    else:
+        if peer_queries:
+            fallback_used = True
+
+    # 3. Validator count
+    val_queries = []
+    if client == "lighthouse":
+        val_queries = ["lighthouse_validator_count", "validator_count"]
+    elif client in ("prysm", "teku", "nimbus", "lodestar"):
+        val_queries = ["beacon_validators_total", "validator_count"]
+
+    for q in val_queries:
+        val_count = _query_prometheus(q)
+        if val_count is not None:
+            snapshot.validator_count = int(val_count)
+            break
+    else:
+        if val_queries:
+            fallback_used = True
+
+    snapshot.prometheus_fallback_used = fallback_used
+    return snapshot

webhook-receiver/src/webhook_receiver/server.py

@@ -14,6 +14,7 @@
 
 from aiohttp import web
 
+from .context_fetcher import fetch_context_snapshot
 from .dedup import DedupLookup, should_process
 from .schema import ValidationError, validate_alertmanager_payload
 from .storm_protection import StormTracker
@@ -78,6 +79,14 @@ async def handle_webhook(self, request: web.Request) -> web.Response:
         if not alerts:
             return web.json_response({"status": "ok", "alerts_processed": 0})
 
+        # Pre-fetch context snapshot for each alert
+        for alert in alerts:
+            alert.context_snapshot = fetch_context_snapshot(
+                host=alert.host,
+                container=alert.container,
+                client=alert.client,
+            )
+
         offsets = []
         deduped_ids = []
         bundled_ids = []

webhook-receiver/tests/test_context_fetcher.py

@@ -0,0 +1,150 @@
+"""Tests for context_fetcher module."""
+
+import json
+from unittest.mock import patch, MagicMock
+
+import pytest
+
+from webhook_receiver.context_fetcher import (
+    _query_prometheus,
+    _get_docker_container_status,
+    fetch_context_snapshot,
+)
+
+
+class TestQueryPrometheus:
+    @patch("webhook_receiver.context_fetcher.urllib.request.urlopen")
+    def test_query_prometheus_success(self, mock_urlopen):
+        mock_resp = MagicMock()
+        mock_resp.read.return_value = json.dumps({
+            "data": {
+                "result": [{"value": [1704067200, "42.5"]}]
+            }
+        }).encode()
+        mock_urlopen.return_value.__enter__.return_value = mock_resp
+
+        result = _query_prometheus("lighthouse_peers")
+        assert result == 42.5
+
+    @patch("webhook_receiver.context_fetcher.urllib.request.urlopen")
+    def test_query_prometheus_failure(self, mock_urlopen):
+        import urllib.error
+        mock_urlopen.side_effect = urllib.error.URLError("Connection refused")
+        result = _query_prometheus("lighthouse_peers")
+        assert result is None
+
+    @patch("webhook_receiver.context_fetcher.urllib.request.urlopen")
+    def test_query_prometheus_empty_result(self, mock_urlopen):
+        mock_resp = MagicMock()
+        mock_resp.read.return_value = json.dumps({
+            "data": {
+                "result": []
+            }
+        }).encode()
+        mock_urlopen.return_value.__enter__.return_value = mock_resp
+
+        result = _query_prometheus("lighthouse_peers")
+        assert result is None
+
+
+class TestGetDockerContainerStatus:
+    @patch("webhook_receiver.context_fetcher.socket.socket")
+    def test_docker_status_running(self, mock_socket_cls):
+        mock_sock = MagicMock()
+        mock_socket_cls.return_value.__enter__.return_value = mock_sock
+
+        response = (
+            b"HTTP/1.1 200 OK\r\n"
+            b"Content-Type: application/json\r\n"
+            b"\r\n"
+            b'{"State": {"Status": "running"}}'
+        )
+        mock_sock.recv.side_effect = [response, b""]
+
+        status, note = _get_docker_container_status("consensus")
+        assert status == "running"
+        assert note is None
+
+        # Verify Connection: close header is sent to prevent hanging
+        sent_request = mock_sock.sendall.call_args[0][0].decode('utf-8')
+        assert "Connection: close\r\n" in sent_request
+
+    @patch("webhook_receiver.context_fetcher.socket.socket")
+    def test_docker_status_exited(self, mock_socket_cls):
+        mock_sock = MagicMock()
+        mock_socket_cls.return_value.__enter__.return_value = mock_sock
+
+        response = (
+            b"HTTP/1.1 200 OK\r\n"
+            b"Content-Type: application/json\r\n"
+            b"\r\n"
+            b'{"State": {"Status": "exited"}}'
+        )
+        mock_sock.recv.side_effect = [response, b""]
+
+        status, note = _get_docker_container_status("consensus")
+        assert status == "exited"
+        assert "exited" in note
+
+    @patch("webhook_receiver.context_fetcher.socket.socket")
+    def test_docker_status_connection_refused(self, mock_socket_cls):
+        mock_sock = MagicMock()
+        mock_sock.connect.side_effect = ConnectionRefusedError()
+        mock_socket_cls.return_value.__enter__.return_value = mock_sock
+
+        status, note = _get_docker_container_status("consensus")
+        assert status is None
+        assert "ConnectionRefusedError" in note
+
+
+class TestFetchContextSnapshot:
+    @patch("webhook_receiver.context_fetcher._get_docker_container_status")
+    @patch("webhook_receiver.context_fetcher._query_prometheus")
+    def test_fetch_success_no_fallback(self, mock_prom, mock_docker):
+        mock_docker.return_value = ("running", None)
+        mock_prom.return_value = 50.0
+
+        snapshot = fetch_context_snapshot("host1", "consensus", "lighthouse")
+
+        assert snapshot.container_status == "running"
+        assert snapshot.container_status_note is None
+        assert snapshot.peer_count == 50
+        assert snapshot.prometheus_fallback_used is False
+
+    @patch("webhook_receiver.context_fetcher._get_docker_container_status")
+    @patch("webhook_receiver.context_fetcher._query_prometheus")
+    def test_fetch_docker_fallback_to_prometheus(self, mock_prom, mock_docker):
+        mock_docker.return_value = (None, "docker socket error")
+        # Calls: 1. lighthouse_peers, 2. lighthouse_validator_count
+        mock_prom.side_effect = [50.0, 3.0]
+
+        snapshot = fetch_context_snapshot("host1", "consensus", "lighthouse")
+
+        assert snapshot.container_status == "unavailable"
+        assert snapshot.container_status_note == "docker socket unreachable"
+        assert snapshot.peer_count == 50
+        assert snapshot.validator_count == 3
+        assert snapshot.prometheus_fallback_used is True
+
+    @patch("webhook_receiver.context_fetcher._get_docker_container_status")
+    @patch("webhook_receiver.context_fetcher._query_prometheus")
+    def test_fetch_all_unavailable(self, mock_prom, mock_docker):
+        mock_docker.return_value = (None, "error")
+        mock_prom.return_value = None
+
+        snapshot = fetch_context_snapshot("host1", "consensus", "lighthouse")
+
+        assert snapshot.container_status == "unavailable"
+        assert snapshot.peer_count is None
+        assert snapshot.prometheus_fallback_used is True
+
+    @patch("webhook_receiver.context_fetcher._get_docker_container_status")
+    @patch("webhook_receiver.context_fetcher._query_prometheus")
+    def test_fetch_no_container(self, mock_prom, mock_docker):
+        mock_prom.return_value = 50.0
+
+        snapshot = fetch_context_snapshot("host1", container=None, client="lighthouse")
+
+        assert snapshot.container_status is None
+        assert snapshot.peer_count == 50
+        assert snapshot.prometheus_fallback_used is False