feat: ship API and worker logs to ClickHouse over OTLP [#241]

.env.example

@@ -264,3 +264,13 @@ KC_SMTP_STARTTLS=true
 KC_SMTP_SSL=false
 KC_SMTP_USERNAME=
 KC_SMTP_PASSWORD=
+
+
+# --- Logs to ClickHouse over OTLP (ADR 0007). Empty endpoint = off.
+# Local: http://otel-collector:4318 with `docker compose --profile observability up -d`
+# Shared env: http://clickstack-otel-collector.observability.svc.cluster.local:4318
+OTEL_EXPORTER_OTLP_ENDPOINT=
+# Shared env only: authorization=<hyperdx-ingestion-api-key>
+OTEL_EXPORTER_OTLP_HEADERS=
+# Seconds before an export is dropped
+OTEL_EXPORTER_OTLP_TIMEOUT=5

AGENTS.md

@@ -19,6 +19,7 @@ Full narrative docs (read these before deep architectural changes, don't re-deri
 # Run the full stack (api, worker, temporal, minio, qdrant, ui)
 docker compose up -d --build
 docker compose down
+docker compose --profile observability up -d   # + otel-collector & ClickHouse (needs OTEL_EXPORTER_OTLP_ENDPOINT=http://otel-collector:4318)
 
 # Tests (pytest.ini: testpaths=tests, markers unit/integration/slow/api/db/workflow)
 uv run pytest
@@ -77,6 +78,7 @@ Core modules (all under `pipeline/`):
 - `discovery_publish_service.py` — `DiscoveryPublishService`: owns the Publish to Network env vars and the HTTP call. Deliberately Temporal-free
 - `models.py` — Pydantic models, including `DocumentStage` enum and `PIPELINE_STAGES` (the stepper-UI stage list)
 - `config.py` — `Config` dataclass reading env vars, with defaults
+- `observability.py` — bridges root-logger `INFO`+ lines to OTLP when `OTEL_EXPORTER_OTLP_ENDPOINT` is set (ClickHouse/HyperDX); no-op otherwise. See ADR 0007
 - `instances.py` — instance/tenant (state) code ↔ human name mapping; `bv` = Bharat Vistaar platform-wide, others are state codes matching Keycloak group paths (`/states/MH/...` → `mh`)
 - `master_catalog_pg.py`, `scheme_catalog.py` — Postgres-backed master scheme catalog, synced to Redis for the AI layer (bharat-oan-api) to consume
 - `ocr/`, `translation/`, `chunking/`, `vector_store/` — pluggable provider packages, each with a `base.py` protocol/ABC and a `service.py` (or equivalent) that dispatches to the configured provider

ENV.md

@@ -66,6 +66,17 @@ Root `.env` (see `.env.example`). Required by FastAPI (`pipeline/api/`), Tempora
 | `MINIO_BUCKET` | `documents` | Object storage bucket |
 | `DOCUMENT_DB_PATH` | `/data/documents.db` | SQLite path (use `./data/documents.db` for local non-Docker) |
 
+### Logs to ClickHouse (OTLP)
+
+Off unless `OTEL_EXPORTER_OTLP_ENDPOINT` is set. When it is set, API and worker log lines at `INFO` and above (root logger, including `/search` query text) also go over OTLP/HTTP to a collector, which writes them to ClickHouse for HyperDX. See `docs/ADR/0007-ship-logs-over-otlp.md`.
+
+| Variable | Default | Purpose |
+|----------|---------|---------|
+| `OTEL_EXPORTER_OTLP_ENDPOINT` | *(unset — off)* | Collector base URL. Local: `http://otel-collector:4318` with `docker compose --profile observability`. Shared env: the ClickStack collector, e.g. `http://clickstack-otel-collector.observability.svc.cluster.local:4318` |
+| `OTEL_EXPORTER_OTLP_HEADERS` | *(unset)* | Headers sent with each export, e.g. `authorization=<HyperDX ingestion API key>`; ClickStack rejects unauthenticated telemetry |
+| `OTEL_SERVICE_NAME` | `knowledge-provider` | Labels our lines in the shared store. Compose sets `knowledge-provider-api` / `knowledge-provider-worker`; set the same per process in other deployments |
+| `OTEL_EXPORTER_OTLP_TIMEOUT` | `5` (seconds) | How long one export may take before it is dropped; kept under Docker's 10s stop grace so shutdown with the collector down stays clean |
+
 ### API HTTP surface
 
 | Variable | Default | Purpose |

README.md

@@ -333,6 +333,18 @@ docker compose up -d --build
 docker compose down
 ```
 
+### Logs in ClickHouse (optional)
+
+Set `OTEL_EXPORTER_OTLP_ENDPOINT=http://otel-collector:4318` in `.env`, then:
+
+```bash
+docker compose --profile observability up -d --build
+curl 'http://127.0.0.1:18123/?user=knowledge&password=knowledge' \
+  --data-binary 'SELECT ServiceName, Body FROM otel.otel_logs ORDER BY Timestamp DESC LIMIT 20'
+```
+
+API and worker lines land in `otel.otel_logs`. Shared environments send to the ClickStack collector instead; see `ENV.md` and `docs/ADR/0007-ship-logs-over-otlp.md`.
+
 ### Health Checks
 
 Useful endpoints after startup:

docker-compose.yml

@@ -14,6 +14,8 @@
 # Profiles:
 #   (default) — all core services
 #   ops       — also starts temporal-ui and keycloak-ui (already included in keycloak)
+#   observability — also starts otel-collector and clickhouse; set
+#                   OTEL_EXPORTER_OTLP_ENDPOINT=http://otel-collector:4318 in .env
 #
 # Host ports (all bound to 127.0.0.1 for nginx reverse-proxy; override in .env):
 #   API          127.0.0.1:${API_HOST_PORT:-8011}:8001
@@ -23,6 +25,7 @@
 #   MinIO API    127.0.0.1:9000:9000
 #   MinIO UI     127.0.0.1:9001:9001
 #   Temporal UI  127.0.0.1:${TEMPORAL_UI_PORT:-8090}:8080  (profile: ops)
+#   ClickHouse   127.0.0.1:${CLICKHOUSE_HTTP_PORT:-18123}:8123  (profile: observability)
 
 services:
 
@@ -254,6 +257,7 @@ services:
       # HF cache path propagated into the container
       HF_HOME: ${HF_HOME:-/opt/hf-cache}
       HF_HUB_CACHE: ${HF_HOME:-/opt/hf-cache}
+      OTEL_SERVICE_NAME: knowledge-provider-api
     ports:
       - "127.0.0.1:${API_HOST_PORT:-8011}:8001"
     depends_on:
@@ -295,6 +299,7 @@ services:
       ALLOWED_FILE_PATHS: /app/books,/data/documents
       HF_HOME: ${HF_HOME:-/opt/hf-cache}
       HF_HUB_CACHE: ${HF_HOME:-/opt/hf-cache}
+      OTEL_SERVICE_NAME: knowledge-provider-worker
     depends_on:
       temporal:
         condition: service_healthy
@@ -343,6 +348,39 @@ services:
       - api
     restart: unless-stopped
 
+  otel-collector:
+    image: otel/opentelemetry-collector-contrib:${OTEL_COLLECTOR_TAG:-0.161.0}
+    profiles: [observability]
+    command:
+      - --config=/etc/otel/collector.yaml
+    volumes:
+      - ./otel/collector.local.yaml:/etc/otel/collector.yaml:ro
+    environment:
+      CLICKHOUSE_USER: knowledge
+      CLICKHOUSE_PASSWORD: knowledge
+    depends_on:
+      clickhouse:
+        condition: service_healthy
+    restart: unless-stopped
+
+  clickhouse:
+    image: clickhouse/clickhouse-server:${CLICKHOUSE_TAG:-25.12}
+    profiles: [observability]
+    environment:
+      CLICKHOUSE_DB: otel
+      CLICKHOUSE_USER: knowledge
+      CLICKHOUSE_PASSWORD: knowledge
+    volumes:
+      - clickhouse-data:/var/lib/clickhouse
+    ports:
+      - "127.0.0.1:${CLICKHOUSE_HTTP_PORT:-18123}:8123"
+    healthcheck:
+      test: wget --no-verbose --tries=1 --spider http://localhost:8123/ping || exit 1
+      interval: 5s
+      timeout: 5s
+      retries: 10
+    restart: unless-stopped
+
 volumes:
   temporal-db-data:
   qdrant-data:
@@ -351,3 +389,4 @@ volumes:
   minio-data:
   sqlite-data:
   documents-data:
+  clickhouse-data:

docs/ADR/0007-ship-logs-over-otlp.md

@@ -0,0 +1,26 @@
+---
+status: accepted
+---
+
+# Ship API and worker logs to ClickHouse over OTLP
+
+Context: The API and worker only log to stdout, so a support engineer has to reach into a server to see why a farmer's search went wrong. The decision support system already sends its telemetry to ClickHouse, where HyperDX reads it (engineering-tracker#241).
+
+Decision:
+- The API and worker attach an OpenTelemetry log handler to the root logger when `OTEL_EXPORTER_OTLP_ENDPOINT` is set. With it unset, nothing changes. The handler goes on root because `/search` and `api.py` log through `logging.info(...)` directly, not through module loggers.
+- Only `INFO` and above is exported. The level sits on the handler, so `LOG_LEVEL=DEBUG` still prints locally without shipping debug output.
+- `opentelemetry.*` records are not exported, so the exporter's own failures can't feed back into it.
+- Lines are batched and dropped when the collector is unreachable. Search and uploads never wait on the log store.
+- Endpoint and auth headers come from the standard `OTEL_*` variables. Service name falls back to `knowledge-provider` when `OTEL_SERVICE_NAME` is unset, so lines are never labelled `unknown_service`.
+- An export gives up after 5 seconds unless `OTEL_EXPORTER_OTLP_TIMEOUT` says otherwise, so shutdown with the collector down fits inside Docker's 10-second stop grace.
+- Query text, the raw `/search` payload and the caller's username or email stay in the exported lines, as do email-OTP lines naming an operator's email. Operators are staff, so shipping their identity is accepted. Farmer questions carry no personal data, and a support engineer needs the question to judge a poor answer. Unlike the DSS collector, there is no allowlist.
+- This repo's compose ships a collector and ClickHouse under the `observability` profile, for laptops only. Shared environments point at the existing ClickStack collector instead of running their own.
+
+Consequences:
+- Logs only. Traces and metrics are not emitted.
+- Every API request logs one line on `pipeline.api.requests`: method, path with query string, status and duration, never a body. A pure ASGI middleware times the whole response, so streamed downloads are timed to their last byte and a failure mid-stream is logged.
+- Status below 400 logs at INFO (successful `/health` at DEBUG, since the UI polls it). 4xx logs at WARNING and 5xx at ERROR, with the error detail (validation errors keep only type, field and message, never the submitted value), or the HTTP reason when there is none. A 5xx raised from a caught exception carries that exception's traceback. Unhandled exceptions log at ERROR with their traceback. uvicorn's own access log doesn't reach the root logger and is not shipped.
+- A Temporal worker interceptor (`pipeline/activity_logging.py`) logs every workflow's and activity's start and its completion, failure or cancellation, so new workflows and activities are covered without extra code. Activity failures log at WARNING without a traceback, because each attempt may be retried and Temporal's own `Completing activity as failed` line already carries it. Workflow failures log at ERROR with the traceback.
+- Startup messages from the API and worker are log lines, not `print`s. A failed API startup or a worker crash logs at ERROR with its traceback, and the log buffer is flushed before the process exits.
+- `/search` lines don't carry transaction or message ids yet. That comes with the next story, once callers send them.
+- The shared-environment Helm values (endpoint, ingestion key) are set in the helmcharts repo, not here.

otel/collector.local.yaml

@@ -0,0 +1,24 @@
+receivers:
+  otlp:
+    protocols:
+      http:
+        endpoint: 0.0.0.0:4318
+
+processors:
+  batch:
+
+exporters:
+  clickhouse:
+    endpoint: tcp://clickhouse:9000
+    username: ${env:CLICKHOUSE_USER}
+    password: ${env:CLICKHOUSE_PASSWORD}
+    database: otel
+    create_schema: true
+    ttl: 720h
+
+service:
+  pipelines:
+    logs:
+      receivers: [otlp]
+      processors: [batch]
+      exporters: [clickhouse]

pipeline/activity_logging.py

@@ -0,0 +1,87 @@
+from __future__ import annotations
+
+import asyncio
+import logging
+import time
+from typing import Any
+
+from temporalio import activity, workflow
+from temporalio.worker import (
+    ActivityInboundInterceptor,
+    ExecuteActivityInput,
+    ExecuteWorkflowInput,
+    Interceptor,
+    WorkflowInboundInterceptor,
+    WorkflowInterceptorClassInput,
+)
+
+logger = logging.getLogger(__name__)
+
+
+class ActivityLogging(Interceptor):
+    def intercept_activity(self, next: ActivityInboundInterceptor) -> ActivityInboundInterceptor:
+        return _LogActivity(next)
+
+    def workflow_interceptor_class(
+        self, input: WorkflowInterceptorClassInput
+    ) -> type[WorkflowInboundInterceptor]:
+        return _LogWorkflow
+
+
+class _LogActivity(ActivityInboundInterceptor):
+    async def execute_activity(self, input: ExecuteActivityInput) -> Any:
+        info = activity.info()
+        fields = (info.activity_type, info.workflow_id, info.attempt)
+        logger.info("activity started name=%s workflow_id=%s attempt=%s", *fields)
+        started = time.perf_counter()
+        try:
+            result = await self.next.execute_activity(input)
+        except asyncio.CancelledError:
+            logger.warning(
+                "activity cancelled name=%s workflow_id=%s attempt=%s duration_ms=%d",
+                *fields,
+                _elapsed_ms(started),
+            )
+            raise
+        except Exception as exc:
+            logger.warning(
+                "activity failed name=%s workflow_id=%s attempt=%s duration_ms=%d error=%r",
+                *fields,
+                _elapsed_ms(started),
+                exc,
+            )
+            raise
+        logger.info(
+            "activity completed name=%s workflow_id=%s attempt=%s duration_ms=%d",
+            *fields,
+            _elapsed_ms(started),
+        )
+        return result
+
+
+class _LogWorkflow(WorkflowInboundInterceptor):
+    async def execute_workflow(self, input: ExecuteWorkflowInput) -> Any:
+        info = workflow.info()
+        fields = (info.workflow_type, info.workflow_id, info.run_id)
+        workflow.logger.info("workflow started name=%s workflow_id=%s run_id=%s", *fields)
+        try:
+            result = await self.next.execute_workflow(input)
+        except asyncio.CancelledError:
+            workflow.logger.warning(
+                "workflow cancelled name=%s workflow_id=%s run_id=%s", *fields
+            )
+            raise
+        except Exception as exc:
+            workflow.logger.error(
+                "workflow failed name=%s workflow_id=%s run_id=%s error=%r",
+                *fields,
+                exc,
+                exc_info=True,
+            )
+            raise
+        workflow.logger.info("workflow completed name=%s workflow_id=%s run_id=%s", *fields)
+        return result
+
+
+def _elapsed_ms(started: float) -> int:
+    return int((time.perf_counter() - started) * 1000)