feat: add clickhouse telemetry backend and tooling

Author: haasonsaas

Makefile

@@ -1,4 +1,4 @@
-.PHONY: setup test compile run docs dashboard
+.PHONY: setup test compile run docs dashboard clickhouse-up
 
 setup:
 	uv sync --all-extras
@@ -17,3 +17,8 @@ docs:
 
 dashboard:
 	uv run --group dashboard -- streamlit run dashboards/agent_dashboard.py
+
+clickhouse-up:
+	docker run --rm -p 8123:8123 -p 9000:9000 \
+		-v $(PWD)/infrastructure/clickhouse/schema.sql:/docker-entrypoint-initdb.d/schema.sql \
+		clickhouse/clickhouse-server:latest

README.md

@@ -59,7 +59,7 @@ Copy `.env.example` to `.env` and adjust values locally if you prefer dotenv-sty
 | `PROVENANCE_RISK_HIGH_SEVERITY_THRESHOLD` | Number of high-severity findings before issuing a warn | `1` |
 | `PROVENANCE_ANALYTICS_DEFAULT_WINDOW` | Default lookback window for analytics | `7d` |
 | `PROVENANCE_SEMGREP_CONFIG_PATH` | Override path/URL for Semgrep configuration | *(bundled rules)* |
-| `PROVENANCE_TIMESERIES_BACKEND` | Backend for analytics event export (`file`, `bigquery`, `snowflake`, `off`) | `file` |
+| `PROVENANCE_TIMESERIES_BACKEND` | Backend for analytics event export (`file`, `bigquery`, `snowflake`, `clickhouse`, `off`) | `file` |
 | `PROVENANCE_TIMESERIES_PATH` | File path for JSONL events when using `file` backend | `data/timeseries_events.jsonl` |
 | `PROVENANCE_TIMESERIES_PROJECT` | Cloud project/account for warehouse exports | *(unset)* |
 | `PROVENANCE_TIMESERIES_DATABASE` | Warehouse database name (Snowflake only) | *(unset)* |
@@ -86,6 +86,10 @@ Copy `.env.example` to `.env` and adjust values locally if you prefer dotenv-sty
 | `PROVENANCE_GITHUB_REVIEWER_TEAM_MAP` | JSON map of reviewer logins to team names for cohort reporting | `{}` |
 | `PROVENANCE_TEAM_REVIEW_BUDGETS` | JSON map of team names to max expected human review counts per window | `{}` |
 | `PROVENANCE_AGENT_PUBLIC_KEYS` | JSON map of agent IDs to base64 Ed25519 public keys for attestation verification | `{}` |
+| `PROVENANCE_CLICKHOUSE_URL` | ClickHouse HTTP endpoint for analytics export (when backend=`clickhouse`) | *(unset)* |
+| `PROVENANCE_CLICKHOUSE_DATABASE` | ClickHouse database name (optional if table already qualified) | `provenance` |
+| `PROVENANCE_CLICKHOUSE_USER` | ClickHouse user for authenticated writes | *(unset)* |
+| `PROVENANCE_CLICKHOUSE_PASSWORD` | ClickHouse password for authenticated writes | *(unset)* |
 
 ## Detection Pipeline
 
@@ -183,6 +187,12 @@ The same process works against forks or sandboxes—helpful when validating new
 - Set `PROVENANCE_OTEL_ENABLED=true` to emit OpenTelemetry metrics (currently using the console exporter by default).
 - Event payloads include per-agent code volume, churn rates, complexity heuristics, and counts by finding category/severity.
 
+### ClickHouse quickstart
+
+- Run `make clickhouse-up` to launch a local ClickHouse instance with the starter schema from `infrastructure/clickhouse/schema.sql`.
+- Configure `PROVENANCE_TIMESERIES_BACKEND=clickhouse`, `PROVENANCE_CLICKHOUSE_URL=http://localhost:8123`, and `PROVENANCE_TIMESERIES_TABLE=analysis_events` (or point at your own table) to mirror analytics events into ClickHouse.
+- Downstream jobs can query the `provenance` database (tables: `analysis_events`, `findings`, `review_events`) for long-horizon reporting while Redis continues to serve hot state.
+
 ## Dashboard
 
 - Install dashboard dependencies: `uv sync --group dashboard`

app/core/config.py

@@ -44,6 +44,10 @@ class Settings(BaseSettings):
     github_reviewer_team_map: dict[str, str] = Field(default_factory=dict)
     github_team_review_budgets: dict[str, int] = Field(default_factory=dict)
     agent_public_keys: dict[str, str] = Field(default_factory=dict)
+    clickhouse_url: str | None = None
+    clickhouse_database: str | None = None
+    clickhouse_user: str | None = None
+    clickhouse_password: str | None = None
 
     model_config = SettingsConfigDict(env_prefix="provenance_", env_file=".env", extra="ignore")
 

app/telemetry/event_sink.py

@@ -8,6 +8,8 @@
 from pathlib import Path
 from typing import Protocol
 
+import requests
+
 from app.core.config import settings
 
 
@@ -180,6 +182,64 @@ def __del__(self):  # pragma: no cover - best effort cleanup
             return
 
 
+class ClickHouseEventSink:
+    """Writes events into ClickHouse via the HTTP interface."""
+
+    def __init__(
+        self,
+        url: str,
+        table: str,
+        *,
+        database: str | None = None,
+        user: str | None = None,
+        password: str | None = None,
+        batch_size: int = 25,
+    ) -> None:
+        self._url = url.rstrip("/")
+        self._table = table
+        self._database = database
+        self._batch_size = max(batch_size, 1)
+        self._auth = (user, password) if user and password else None
+        self._buffer: list[str] = []
+        self._lock = threading.Lock()
+        self._session = requests.Session()
+
+    def _table_reference(self) -> str:
+        if self._database and "." not in self._table:
+            return f"{self._database}.{self._table}"
+        return self._table
+
+    def publish(self, event: dict) -> None:
+        payload = json.dumps(event, separators=(",", ":"), sort_keys=True)
+        with self._lock:
+            self._buffer.append(payload)
+            if len(self._buffer) >= self._batch_size:
+                self._flush_locked()
+
+    def close(self) -> None:
+        with self._lock:
+            self._flush_locked()
+        self._session.close()
+
+    def _flush_locked(self) -> None:
+        if not self._buffer:
+            return
+        data = "\n".join(self._buffer)
+        query = f"INSERT INTO {self._table_reference()} FORMAT JSONEachRow\n{data}\n"
+        params = {"database": self._database} if self._database and "." not in self._table else None
+        response = self._session.post(
+            self._url,
+            params=params,
+            data=query.encode("utf-8"),
+            auth=self._auth,
+            headers={"Content-Type": "application/json"},
+            timeout=30,
+        )
+        if response.status_code >= 400:
+            raise RuntimeError(f"ClickHouse insert failed ({response.status_code}): {response.text}")
+        self._buffer.clear()
+
+
 def sink_from_settings() -> EventSink:
     """Factory to construct an event sink based on app settings."""
 
@@ -219,6 +279,17 @@ def sink_from_settings() -> EventSink:
             role=settings.timeseries_role,
             batch_size=settings.timeseries_batch_size,
         )
+    if backend == "clickhouse":
+        if not (settings.clickhouse_url and settings.timeseries_table):
+            raise ValueError("ClickHouse backend requires PROVENANCE_CLICKHOUSE_URL and PROVENANCE_TIMESERIES_TABLE")
+        return ClickHouseEventSink(
+            url=settings.clickhouse_url,
+            table=settings.timeseries_table,
+            database=settings.clickhouse_database,
+            user=settings.clickhouse_user,
+            password=settings.clickhouse_password,
+            batch_size=settings.timeseries_batch_size,
+        )
     if backend in {"off", "none", "disabled"}:
         return NullEventSink()
     raise ValueError(f"Unsupported timeseries backend: {settings.timeseries_backend}")

infrastructure/clickhouse/schema.sql

@@ -0,0 +1,33 @@
+CREATE DATABASE IF NOT EXISTS provenance;
+
+CREATE TABLE IF NOT EXISTS provenance.analysis_events
+(
+    event_time DateTime DEFAULT now(),
+    payload JSON
+)
+ENGINE = MergeTree
+ORDER BY (event_time);
+
+CREATE TABLE IF NOT EXISTS provenance.findings
+(
+    analysis_id String,
+    repo_id String,
+    pr_number String,
+    rule_key String,
+    severity String,
+    detected_at DateTime,
+    payload JSON
+)
+ENGINE = MergeTree
+ORDER BY (analysis_id, rule_key, detected_at);
+
+CREATE TABLE IF NOT EXISTS provenance.review_events
+(
+    repo_id String,
+    pr_number String,
+    agent_id String,
+    recorded_at DateTime DEFAULT now(),
+    payload JSON
+)
+ENGINE = MergeTree
+ORDER BY (repo_id, pr_number, recorded_at);

pyproject.toml

@@ -14,7 +14,8 @@ dependencies = [
     "semgrep>=1.64,<2.0",
     "opentelemetry-api>=1.25,<2.0",
     "opentelemetry-sdk>=1.25,<2.0",
-    "PyGithub>=2.4,<3.0"
+    "PyGithub>=2.4,<3.0",
+    "requests>=2.31,<3.0"
 ]
 
 [project.optional-dependencies]