From e259abe0469f95903128de56bab919d884226f20 Mon Sep 17 00:00:00 2001 From: Hariharan Thiagarajan Date: Thu, 5 Nov 2026 15:18:04 +0530 Subject: [PATCH 1/8] feat: ship API and worker logs to ClickHouse over OTLP [#241] Root-logger INFO+ lines go to an OTLP collector when OTEL_EXPORTER_OTLP_ENDPOINT is set; a local collector and ClickHouse run under the compose `observability` profile. See ADR 0007. Co-Authored-By: Claude Opus 5.5 --- AGENTS.md | 2 + ENV.md | 10 ++ README.md | 12 ++ docker-compose.yml | 41 +++++++ docs/ADR/0007-ship-logs-over-otlp.md | 22 ++++ otel/collector.local.yaml | 24 ++++ pipeline/api/api.py | 4 +- pipeline/observability.py | 67 +++++++++++ pipeline/worker.py | 9 ++ pyproject.toml | 4 + tests/test_observability.py | 86 +++++++++++++++ uv.lock | 159 +++++++++++++++++++++++++++ 12 files changed, 439 insertions(+), 1 deletion(-) create mode 100644 docs/ADR/0007-ship-logs-over-otlp.md create mode 100644 otel/collector.local.yaml create mode 100644 pipeline/observability.py create mode 100644 tests/test_observability.py diff --git a/AGENTS.md b/AGENTS.md index 66099bb..406cb6e 100644 --- a/AGENTS.md +++ b/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 diff --git a/ENV.md b/ENV.md index 00a4e61..cd09a38 100644 --- a/ENV.md +++ b/ENV.md @@ -66,6 +66,16 @@ 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=`; ClickStack rejects unauthenticated telemetry | +| `OTEL_SERVICE_NAME` | set by compose | `knowledge-provider-api` / `knowledge-provider-worker`; separates our lines from other services in the shared store | + ### API HTTP surface | Variable | Default | Purpose | diff --git a/README.md b/README.md index abe03e6..3534caa 100644 --- a/README.md +++ b/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: diff --git a/docker-compose.yml b/docker-compose.yml index 1f6f11c..a4cd717 100644 --- a/docker-compose.yml +++ b/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,8 @@ 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 + OTEL_EXPORTER_OTLP_ENDPOINT: ${OTEL_EXPORTER_OTLP_ENDPOINT:-} ports: - "127.0.0.1:${API_HOST_PORT:-8011}:8001" depends_on: @@ -295,6 +300,8 @@ 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 + OTEL_EXPORTER_OTLP_ENDPOINT: ${OTEL_EXPORTER_OTLP_ENDPOINT:-} depends_on: temporal: condition: service_healthy @@ -343,6 +350,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 +391,4 @@ volumes: minio-data: sqlite-data: documents-data: + clickhouse-data: diff --git a/docs/ADR/0007-ship-logs-over-otlp.md b/docs/ADR/0007-ship-logs-over-otlp.md new file mode 100644 index 0000000..2d49e6f --- /dev/null +++ b/docs/ADR/0007-ship-logs-over-otlp.md @@ -0,0 +1,22 @@ +--- +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, auth headers and service name come from the standard `OTEL_*` variables. The code parses none of them. +- Query text stays in the exported lines. 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. +- `print(...)` output, used for some worker and API startup messages, stays on stdout and is not shipped. +- `/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. diff --git a/otel/collector.local.yaml b/otel/collector.local.yaml new file mode 100644 index 0000000..3692218 --- /dev/null +++ b/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] diff --git a/pipeline/api/api.py b/pipeline/api/api.py index 593f2eb..5f00795 100644 --- a/pipeline/api/api.py +++ b/pipeline/api/api.py @@ -101,6 +101,7 @@ ) from ..network_constants import RESULT_REJECTED from ..network_group import validate_network_provider_config +from ..observability import configure_logs, reset_logs from ..workflows import ( ChunkingOnlyWorkflow, DocumentPipelineWorkflow, @@ -134,6 +135,7 @@ async def lifespan(app: FastAPI): """Initialize Temporal and MinIO clients on startup.""" global temporal_client, minio_client, MINIO_BUCKET + configure_logs() validate_network_provider_config() auth_cfg = load_auth_config() @@ -185,7 +187,7 @@ async def lifespan(app: FastAPI): print(f"Created MinIO bucket: {MINIO_BUCKET}") yield - # Cleanup if needed + reset_logs() app = FastAPI( diff --git a/pipeline/observability.py b/pipeline/observability.py new file mode 100644 index 0000000..ff29b53 --- /dev/null +++ b/pipeline/observability.py @@ -0,0 +1,67 @@ +from __future__ import annotations + +import logging +import os +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from opentelemetry.sdk._logs import LoggerProvider + +_MARKER = "_knowledge_provider_otel_handler" + +_provider: LoggerProvider | None = None +_provider_is_ours = False + + +class _SkipOpenTelemetry(logging.Filter): + def filter(self, record: logging.LogRecord) -> bool: + return not record.name.startswith("opentelemetry") + + +def is_otel_handler(handler: logging.Handler) -> bool: + return getattr(handler, _MARKER, False) + + +def configure_logs(*, logger_provider: LoggerProvider | None = None) -> None: + global _provider, _provider_is_ours + + _detach() + + ours = logger_provider is None + if ours: + if not os.environ.get("OTEL_EXPORTER_OTLP_ENDPOINT", "").strip(): + return + + from opentelemetry.exporter.otlp.proto.http._log_exporter import OTLPLogExporter + from opentelemetry.sdk._logs import LoggerProvider as SdkLoggerProvider + from opentelemetry.sdk._logs.export import BatchLogRecordProcessor + + logger_provider = SdkLoggerProvider() + logger_provider.add_log_record_processor(BatchLogRecordProcessor(OTLPLogExporter())) + + from opentelemetry.instrumentation.logging.handler import LoggingHandler + + handler = LoggingHandler(level=logging.INFO, logger_provider=logger_provider) + handler.addFilter(_SkipOpenTelemetry()) + setattr(handler, _MARKER, True) + logging.getLogger().addHandler(handler) + + _provider = logger_provider + _provider_is_ours = ours + + +def reset_logs() -> None: + global _provider, _provider_is_ours + + _detach() + _provider = None + _provider_is_ours = False + + +def _detach() -> None: + root = logging.getLogger() + for handler in [h for h in root.handlers if is_otel_handler(h)]: + root.removeHandler(handler) + + if _provider is not None and _provider_is_ours: + _provider.shutdown() diff --git a/pipeline/worker.py b/pipeline/worker.py index 6cd996b..ca861c8 100644 --- a/pipeline/worker.py +++ b/pipeline/worker.py @@ -27,6 +27,7 @@ update_document_state, ) from .network_group import validate_network_provider_config +from .observability import configure_logs, reset_logs from .workflows import ( ChunkingOnlyWorkflow, DocumentPipelineWorkflow, @@ -54,6 +55,14 @@ async def main(): """Start the worker.""" + configure_logs() + try: + await _run() + finally: + reset_logs() + + +async def _run(): validate_network_provider_config() translation_provider = os.environ.get("TRANSLATION_PROVIDER", "gemma_vllm").strip().lower() diff --git a/pyproject.toml b/pyproject.toml index ca47fb0..833e675 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -59,6 +59,10 @@ dependencies = [ # Auth (Keycloak JWT validation) "PyJWT[crypto]>=2.8.0", + + "opentelemetry-sdk>=1.44.0", + "opentelemetry-exporter-otlp-proto-http>=1.44.0", + "opentelemetry-instrumentation-logging>=0.65b0", ] [dependency-groups] diff --git a/tests/test_observability.py b/tests/test_observability.py new file mode 100644 index 0000000..04f31dd --- /dev/null +++ b/tests/test_observability.py @@ -0,0 +1,86 @@ +import logging + +import pytest +from opentelemetry.sdk._logs import LoggerProvider +from opentelemetry.sdk._logs.export import InMemoryLogRecordExporter, SimpleLogRecordProcessor + +from pipeline import observability + + +@pytest.fixture +def exported(monkeypatch): + monkeypatch.setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://collector:4318") + exporter = InMemoryLogRecordExporter() + provider = LoggerProvider() + provider.add_log_record_processor(SimpleLogRecordProcessor(exporter)) + root = logging.getLogger() + previous_level = root.level + root.setLevel(logging.DEBUG) + observability.configure_logs(logger_provider=provider) + yield exporter + observability.reset_logs() + root.setLevel(previous_level) + + +def _bodies(exporter): + return [item.log_record.body for item in exporter.get_finished_logs()] + + +def _our_handlers(): + return [h for h in logging.getLogger().handlers if observability.is_otel_handler(h)] + + +def test_nothing_is_attached_without_an_endpoint(monkeypatch): + monkeypatch.delenv("OTEL_EXPORTER_OTLP_ENDPOINT", raising=False) + + observability.configure_logs() + + assert _our_handlers() == [] + + +def test_an_endpoint_attaches_one_handler(monkeypatch): + monkeypatch.setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://collector:4318") + + observability.configure_logs() + try: + assert len(_our_handlers()) == 1 + finally: + observability.reset_logs() + + +def test_info_lines_from_root_and_module_loggers_are_exported(exported): + logging.info("search request query=%r", "cotton pest") + logging.getLogger("pipeline.activities").warning("ocr retry") + + assert _bodies(exported) == ["search request query='cotton pest'", "ocr retry"] + + +def test_exported_line_keeps_its_level(exported): + logging.getLogger("pipeline.activities").error("ingest failed") + + [item] = exported.get_finished_logs() + assert item.log_record.severity_text == "ERROR" + + +def test_debug_lines_are_not_exported(exported): + logging.getLogger("pipeline.activities").debug("page text") + + assert _bodies(exported) == [] + + +def test_opentelemetry_own_lines_are_not_exported(exported): + logging.getLogger("opentelemetry.exporter.otlp").warning("export failed") + + assert _bodies(exported) == [] + + +def test_configuring_twice_keeps_one_handler(exported): + observability.configure_logs(logger_provider=LoggerProvider()) + + assert len(_our_handlers()) == 1 + + +def test_reset_removes_the_handler(exported): + observability.reset_logs() + + assert _our_handlers() == [] diff --git a/uv.lock b/uv.lock index 9292f9a..5e9aedd 100644 --- a/uv.lock +++ b/uv.lock @@ -740,6 +740,18 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/fd/3c/6a2bf344106328fd04963664a60b9bb6496fc25df8e962fcdc1367285fb9/fsspec-2026.7.0-py3-none-any.whl", hash = "sha256:b57ddbafedfaef7018c1ecab32aa200a9d7ca26b77965f64e48b70061249d279", size = 206583, upload-time = "2026-07-28T16:34:49.538Z" }, ] +[[package]] +name = "googleapis-common-protos" +version = "1.76.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "protobuf" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/62/b6/bc095b23a3e0d978e75c8c18cabe4c11f791837410670e4486f0d539bfba/googleapis_common_protos-1.76.0.tar.gz", hash = "sha256:71db0e44a1d4c2dd024ebad44ca67129cc1e2892e4861df466764749d5ef8642", size = 156250, upload-time = "2026-10-22T23:43:36.263Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/05/6d/78dc67a9e779e594b49bf62604034d3a6aa428a55afc729cafda4b7333e1/googleapis_common_protos-1.76.0-py3-none-any.whl", hash = "sha256:59e13344a9f3651aa201a03e2bee815a3bf43a4baf1f9ff8a9b62a840ce25fac", size = 307707, upload-time = "2026-10-22T23:43:10.445Z" }, +] + [[package]] name = "grpcio" version = "1.83.0" @@ -1127,6 +1139,9 @@ dependencies = [ { name = "langchain-text-splitters" }, { name = "minio" }, { name = "openpyxl" }, + { name = "opentelemetry-exporter-otlp-proto-http" }, + { name = "opentelemetry-instrumentation-logging" }, + { name = "opentelemetry-sdk" }, { name = "pdf-inspector" }, { name = "pillow" }, { name = "psycopg2-binary" }, @@ -1165,6 +1180,9 @@ requires-dist = [ { name = "langchain-text-splitters", specifier = ">=0.0.1" }, { name = "minio", specifier = ">=7.0.0" }, { name = "openpyxl", specifier = ">=3.1.0" }, + { name = "opentelemetry-exporter-otlp-proto-http", specifier = ">=1.44.0" }, + { name = "opentelemetry-instrumentation-logging", specifier = ">=0.65b0" }, + { name = "opentelemetry-sdk", specifier = ">=1.44.0" }, { name = "pdf-inspector", specifier = ">=1.14.0" }, { name = "pillow", specifier = ">=10.0.0" }, { name = "psycopg2-binary", specifier = ">=2.9.0" }, @@ -1713,6 +1731,147 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/c0/da/977ded879c29cbd04de313843e76868e6e13408a94ed6b987245dc7c8506/openpyxl-3.1.5-py2.py3-none-any.whl", hash = "sha256:5282c12b107bffeef825f4617dc029afaf41d0ea60823bbb665ef3079dc79de2", size = 250910, upload-time = "2024-06-28T14:03:41.161Z" }, ] +[[package]] +name = "opentelemetry-api" +version = "1.45.2" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "typing-extensions" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/19/14/cf3a6b6448849656e1e5f4e2101f3988dc6789724050096aba1bc5ccf592/opentelemetry_api-1.45.2.tar.gz", hash = "sha256:d7b0008e21e5fd61cbe7a7f34ca3b6905a78ab677d9bc395e299359c75cda6b4", size = 73277, upload-time = "2026-10-29T17:29:37.888Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/a9/2d/610db864d7513158cebb8f36cc88e67e73bd6c7aa1905ef18d89e1281c81/opentelemetry_api-1.45.2-py3-none-any.whl", hash = "sha256:2cbd737377e710ee740d261c94e93fa0f08e6d037c18f17513cc0a2d94479a34", size = 60335, upload-time = "2026-10-29T17:29:06.955Z" }, +] + +[[package]] +name = "opentelemetry-exporter-http-transport" +version = "0.66b2" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "opentelemetry-api" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/a6/e6/dbeb3e217418071656232a4028c791060d219f218cef1249b664049d12c0/opentelemetry_exporter_http_transport-0.66b2.tar.gz", hash = "sha256:66d6ed810a515432c8891da9a834464332fd70badeca45bdeeae81b4280c903b", size = 11693, upload-time = "2026-10-29T17:29:39.444Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/a1/b2/c622a101f31c0e6670b82f4264c4e77c161bce415bc6fc158fda0b82e218/opentelemetry_exporter_http_transport-0.66b2-py3-none-any.whl", hash = "sha256:fd5f5b490c099acda1cd7b909d5e8981436bf18cc8c6098d302d114c271720e9", size = 12154, upload-time = "2026-10-29T17:29:10.244Z" }, +] + +[package.optional-dependencies] +requests = [ + { name = "requests" }, +] + +[[package]] +name = "opentelemetry-exporter-otlp-common" +version = "0.66b2" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "opentelemetry-sdk" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/5a/3c/3f2cc8c42bde2620e65024d71e2eb025f7bd0821bb363b52bcc5a3260dbd/opentelemetry_exporter_otlp_common-0.66b2.tar.gz", hash = "sha256:06f5a9b88638e310efa1f1d206cdf6b75393e112aba4e174a963416192bd1028", size = 14325, upload-time = "2026-10-29T17:29:41.701Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/a8/47/b3049e30b339c82e3c7ec1fc692185a5d5a3ba65a94c570bb39b2c0b8424/opentelemetry_exporter_otlp_common-0.66b2-py3-none-any.whl", hash = "sha256:ace86efb75489f925f1ef809856ae5f0c7db534e4afdc624530222228b1236a5", size = 12385, upload-time = "2026-10-29T17:29:13.667Z" }, +] + +[[package]] +name = "opentelemetry-exporter-otlp-proto-common" +version = "1.45.2" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "opentelemetry-proto" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/69/58/239c98b36d9ea317098769957043c251703636a26189c884fba46c1dad84/opentelemetry_exporter_otlp_proto_common-1.45.2.tar.gz", hash = "sha256:f1f2806f3f28ce7938033823e50934fcd91595a39de38e69a28a51e19f40fc6d", size = 18873, upload-time = "2026-10-29T17:29:44.496Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/e6/d6/0f30d38120a949aef70528f845bf539f80df62abb4b8e62fdf69c2b6df5a/opentelemetry_exporter_otlp_proto_common-1.45.2-py3-none-any.whl", hash = "sha256:af483b24217028892f017096731b794b196b785025d2e6808874720fd0b05a80", size = 15392, upload-time = "2026-10-29T17:29:17.913Z" }, +] + +[[package]] +name = "opentelemetry-exporter-otlp-proto-http" +version = "1.45.2" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "googleapis-common-protos" }, + { name = "opentelemetry-api" }, + { name = "opentelemetry-exporter-http-transport", extra = ["requests"] }, + { name = "opentelemetry-exporter-otlp-common" }, + { name = "opentelemetry-exporter-otlp-proto-common" }, + { name = "opentelemetry-proto" }, + { name = "opentelemetry-sdk" }, + { name = "requests" }, + { name = "typing-extensions" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/db/07/b7b01ecb69d02f540e679a9e36f7413d081f225bbbed4469d212ada10bb7/opentelemetry_exporter_otlp_proto_http-1.45.2.tar.gz", hash = "sha256:5984a3fb46bc4628e4e2b85d4a6e72637e03962aae6cf372791137768a4ad939", size = 28843, upload-time = "2026-10-29T17:29:46.15Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/32/65/f64d7d2450217e2989714e1dee15e19bf79a77a313387abac0cac77bc10e/opentelemetry_exporter_otlp_proto_http-1.45.2-py3-none-any.whl", hash = "sha256:6e4d025138829d0ee57946930c2ccad6491ad4d92a958e9c7eea69278fe60cfc", size = 22180, upload-time = "2026-10-29T17:29:20.154Z" }, +] + +[[package]] +name = "opentelemetry-instrumentation" +version = "0.66b2" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "opentelemetry-api" }, + { name = "opentelemetry-semantic-conventions" }, + { name = "packaging" }, + { name = "wrapt" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/76/b9/46a136199fa6928eecefec8418867a460d94f01c608a85d22bd2ed4e8d48/opentelemetry_instrumentation-0.66b2.tar.gz", hash = "sha256:02b094150e53b76f5906d75e628dbb3833bbc6e62fa109b5c7ba0c081865e923", size = 43305, upload-time = "2026-10-29T17:38:11.618Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/3b/41/014d40e2b08a5d9d9058ed4967ad992657c203ce0c292ded97364ee7b82a/opentelemetry_instrumentation-0.66b2-py3-none-any.whl", hash = "sha256:fa9ccfdab147a14af8107d19ef0a7b5015675273c5adced7a1fc38f28b42ec4c", size = 36902, upload-time = "2026-10-29T17:37:06.994Z" }, +] + +[[package]] +name = "opentelemetry-instrumentation-logging" +version = "0.66b2" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "opentelemetry-api" }, + { name = "opentelemetry-instrumentation" }, + { name = "opentelemetry-semantic-conventions" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/23/2f/900c10a10c780f06c0e90e885e5fdcb95f93a91efbd73d3b87d11c6ac72a/opentelemetry_instrumentation_logging-0.66b2.tar.gz", hash = "sha256:a94210291ae10395ad617c2b69bd7c2e10f77818b8beac11368460f47c370eb1", size = 20350, upload-time = "2026-10-29T17:38:34.161Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/6d/52/1f18b14487fa599f96537b2550c7096f824216fedeeb03fdb7f0f67db847/opentelemetry_instrumentation_logging-0.66b2-py3-none-any.whl", hash = "sha256:6141bc09a7868235172748e8a2cd9806549d742b4f5019ce2a571e8c157e5e7d", size = 15926, upload-time = "2026-10-29T17:37:36.938Z" }, +] + +[[package]] +name = "opentelemetry-proto" +version = "1.45.2" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "protobuf" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/24/9c/1aa4ce60e9bc72af20dc49c92cc0566825092d92b3d6a277439fb5fb7c01/opentelemetry_proto-1.45.2.tar.gz", hash = "sha256:bd8d23cf770605b0af9b7be2071d26a464fea289d35429abfd2a66226114ad64", size = 46486, upload-time = "2026-10-29T17:29:53.007Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/7e/0d/a0b8f2b9efe2ec1abdf639d3696db9b1563ebeb11d91aac08ba7341e9122/opentelemetry_proto-1.45.2-py3-none-any.whl", hash = "sha256:03a7ca1b43f89fd5b965d5975902bb2515c0c86e2428b456836e9a6149c6235a", size = 72484, upload-time = "2026-10-29T17:29:31.403Z" }, +] + +[[package]] +name = "opentelemetry-sdk" +version = "1.45.2" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "opentelemetry-api" }, + { name = "opentelemetry-semantic-conventions" }, + { name = "typing-extensions" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/7b/1a/74a21d4d6969fb8b432ac397e66112a8167a6deb267ea4559dbaf2fe91ef/opentelemetry_sdk-1.45.2.tar.gz", hash = "sha256:7eec4de13dd21ff3a1670dcb0e02079e692736f94d733dfdce5af461d884248d", size = 221705, upload-time = "2026-10-29T17:29:54.685Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/9f/a5/9a3b793ef48245f2336a16beb2324c2397dbb3f987cd68942f77cca06156/opentelemetry_sdk-1.45.2-py3-none-any.whl", hash = "sha256:ff974ea70ee4f1d51e2ec152db37c2b626c7bf82bb3f7d86114fd8704e7199c6", size = 141756, upload-time = "2026-10-29T17:29:33.722Z" }, +] + +[[package]] +name = "opentelemetry-semantic-conventions" +version = "0.66b2" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "opentelemetry-api" }, + { name = "typing-extensions" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/37/3f/50f95972b50e8e353f5b9342034c9cd17730f1956f009ce6a3fbbd89f9ce/opentelemetry_semantic_conventions-0.66b2.tar.gz", hash = "sha256:7da5f1a5dc4383eb66711d4d3d3e71824d9403ac1559fea021cc103c1a4c4976", size = 150246, upload-time = "2026-10-29T17:29:55.682Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/67/bd/6bf33ed31521c565931c037e9e81bac47c1acb000e850c6e24220e2a0c00/opentelemetry_semantic_conventions-0.66b2-py3-none-any.whl", hash = "sha256:4cbe61e5f9275d57ab31708446e7938ef4291ebbe6a7fca6f7a629c5acae87d8", size = 206280, upload-time = "2026-10-29T17:29:35.062Z" }, +] + [[package]] name = "orjson" version = "3.12.0" From 776983712b2c2f513a272e92976b0a8fecf769e7 Mon Sep 17 00:00:00 2001 From: Hariharan Thiagarajan Date: Thu, 5 Nov 2026 15:22:42 +0530 Subject: [PATCH 2/8] fix: address review on log shipping [#241] - Default service.name to knowledge-provider when OTEL_SERVICE_NAME is unset - Cap export at 5s so shutdown with the collector down fits Docker's stop grace - Keep operator emails out of shipped logs: /search logs user id, email-OTP lines move to a logger that is not exported - Let env_file carry OTEL_EXPORTER_OTLP_ENDPOINT instead of a compose override - Test provider ownership and the API/worker wiring Co-Authored-By: Claude Opus 5.5 --- ENV.md | 3 +- docker-compose.yml | 2 - docs/ADR/0007-ship-logs-over-otlp.md | 7 ++- pipeline/api/api.py | 7 ++- pipeline/api/search.py | 6 +-- pipeline/auth/email_otp.py | 12 +++-- pipeline/observability.py | 32 +++++++++-- tests/test_observability.py | 66 +++++++++++++++++++++++ tests/test_observability_wiring.py | 53 ++++++++++++++++++ tests/test_search_provider_integration.py | 22 ++++++++ 10 files changed, 190 insertions(+), 20 deletions(-) create mode 100644 tests/test_observability_wiring.py diff --git a/ENV.md b/ENV.md index cd09a38..beb1e05 100644 --- a/ENV.md +++ b/ENV.md @@ -74,7 +74,8 @@ Off unless `OTEL_EXPORTER_OTLP_ENDPOINT` is set. When it is set, API and worker |----------|---------|---------| | `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=`; ClickStack rejects unauthenticated telemetry | -| `OTEL_SERVICE_NAME` | set by compose | `knowledge-provider-api` / `knowledge-provider-worker`; separates our lines from other services in the shared store | +| `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 diff --git a/docker-compose.yml b/docker-compose.yml index a4cd717..b649b0e 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -258,7 +258,6 @@ services: HF_HOME: ${HF_HOME:-/opt/hf-cache} HF_HUB_CACHE: ${HF_HOME:-/opt/hf-cache} OTEL_SERVICE_NAME: knowledge-provider-api - OTEL_EXPORTER_OTLP_ENDPOINT: ${OTEL_EXPORTER_OTLP_ENDPOINT:-} ports: - "127.0.0.1:${API_HOST_PORT:-8011}:8001" depends_on: @@ -301,7 +300,6 @@ services: HF_HOME: ${HF_HOME:-/opt/hf-cache} HF_HUB_CACHE: ${HF_HOME:-/opt/hf-cache} OTEL_SERVICE_NAME: knowledge-provider-worker - OTEL_EXPORTER_OTLP_ENDPOINT: ${OTEL_EXPORTER_OTLP_ENDPOINT:-} depends_on: temporal: condition: service_healthy diff --git a/docs/ADR/0007-ship-logs-over-otlp.md b/docs/ADR/0007-ship-logs-over-otlp.md index 2d49e6f..a1b288c 100644 --- a/docs/ADR/0007-ship-logs-over-otlp.md +++ b/docs/ADR/0007-ship-logs-over-otlp.md @@ -11,12 +11,15 @@ Decision: - 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, auth headers and service name come from the standard `OTEL_*` variables. The code parses none of them. -- Query text stays in the exported lines. 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. +- 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. +- Operator emails are not shipped. `/search` lines name the caller by Keycloak user id, and email-OTP lines go to the `pipeline.auth.email_otp` logger, which is not exported. They still print locally. +- Query text and the raw `/search` payload stay in the exported lines. 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. +- uvicorn access lines don't propagate to root, so they are not shipped. - `print(...)` output, used for some worker and API startup messages, stays on stdout and is not shipped. - `/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. diff --git a/pipeline/api/api.py b/pipeline/api/api.py index 5f00795..1111807 100644 --- a/pipeline/api/api.py +++ b/pipeline/api/api.py @@ -58,6 +58,9 @@ send_otp, verify_otp, ) +from ..auth.email_otp import ( + logger as email_otp_logger, +) from ..auth.keycloak_admin import ( list_access_options, list_realm_users, @@ -611,7 +614,7 @@ async def auth_otp_request(request: Request, body: OtpRequestBody): """ email = normalize_email(body.email) result = await asyncio.to_thread(send_otp, email) - logging.info("email-otp: login code requested for %s", email) + email_otp_logger.info("email-otp: login code requested for %s", email) return result @@ -625,7 +628,7 @@ async def auth_otp_verify(request: Request, body: OtpVerifyBody): """ email = normalize_email(body.email) tokens = await asyncio.to_thread(verify_otp, email, body.code) - logging.info("email-otp: login succeeded for %s", email) + email_otp_logger.info("email-otp: login succeeded for %s", email) return tokens diff --git a/pipeline/api/search.py b/pipeline/api/search.py index 70beb7c..72c8e77 100644 --- a/pipeline/api/search.py +++ b/pipeline/api/search.py @@ -196,7 +196,7 @@ async def run_search(payload: dict, user: RequireSearch): logging.info( "search payload user=%s payload=%s", - user.username or user.email or user.user_id, + user.user_id, json.dumps(payload, ensure_ascii=False, default=str), ) @@ -208,7 +208,7 @@ async def run_search(payload: dict, user: RequireSearch): logging.info( "search request user=%s index=%s query=%r top_k=%s", - user.username or user.email or user.user_id, + user.user_id, index_name, query, payload.get("top_k"), @@ -291,7 +291,7 @@ async def run_search(payload: dict, user: RequireSearch): logging.info( "search response user=%s index=%s query=%r candidate_count=%s final_count=%s", - user.username or user.email or user.user_id, + user.user_id, index_name, query, len(hits), diff --git a/pipeline/auth/email_otp.py b/pipeline/auth/email_otp.py index 944abee..40349f5 100644 --- a/pipeline/auth/email_otp.py +++ b/pipeline/auth/email_otp.py @@ -48,6 +48,8 @@ from .. import db from . import keycloak_admin +logger = logging.getLogger(__name__) + OTP_LENGTH = 6 OTP_TTL_SECONDS = 300 OTP_RESEND_COOLDOWN_SECONDS = 30 @@ -209,7 +211,7 @@ def _send_email(to_addr: str, subject: str, body: str) -> None: smtp.login(cfg.username, cfg.password) smtp.send_message(message) except (smtplib.SMTPException, OSError) as exc: - logging.error("email-otp: failed to send code to %s: %s", to_addr, exc) + logger.error("email-otp: failed to send code to %s: %s", to_addr, exc) raise HTTPException( 503, "Could not send the email right now. Please try again shortly." ) from exc @@ -318,13 +320,13 @@ def send_otp(email: str) -> dict[str, Any]: f"This code expires in {OTP_TTL_SECONDS // 60} minutes. " "If you did not request this, you can safely ignore this email.", ) - logging.info("email-otp: login code sent to %s", email) + logger.info("email-otp: login code sent to %s", email) except HTTPException: # Never let an SMTP failure surface as a different response/status # than the "no such account" path — that would itself be an # enumeration oracle. The stored code is now simply unreachable; # the user's next resend attempt (after the cooldown) tries again. - logging.error("email-otp: send failed for %s, response withheld", email) + logger.error("email-otp: send failed for %s, response withheld", email) else: # Matches the typical SMTP round-trip _send_email would otherwise # take, so response latency doesn't leak account existence either. @@ -425,12 +427,12 @@ def _rotate_password() -> str: try: _rotate_password() except HTTPException: - logging.error("email-otp: post-grant password rotation failed for %s", email) + logger.error("email-otp: post-grant password rotation failed for %s", email) if status == 200 and payload.get("access_token"): return _token_response(payload) - logging.error( + logger.error( "email-otp: password-grant bridge failed for %s: status=%s body=%s", email, status, payload, ) diff --git a/pipeline/observability.py b/pipeline/observability.py index ff29b53..fecd833 100644 --- a/pipeline/observability.py +++ b/pipeline/observability.py @@ -8,20 +8,40 @@ from opentelemetry.sdk._logs import LoggerProvider _MARKER = "_knowledge_provider_otel_handler" +_TIMEOUT_ENV = ("OTEL_EXPORTER_OTLP_LOGS_TIMEOUT", "OTEL_EXPORTER_OTLP_TIMEOUT") + +DEFAULT_SERVICE_NAME = "knowledge-provider" +DEFAULT_EXPORT_TIMEOUT_SECONDS = 5.0 _provider: LoggerProvider | None = None _provider_is_ours = False -class _SkipOpenTelemetry(logging.Filter): +_NOT_EXPORTED = ("opentelemetry", "pipeline.auth.email_otp") + + +class _SkipUnexported(logging.Filter): def filter(self, record: logging.LogRecord) -> bool: - return not record.name.startswith("opentelemetry") + return not record.name.startswith(_NOT_EXPORTED) def is_otel_handler(handler: logging.Handler) -> bool: return getattr(handler, _MARKER, False) +def _resource(): + from opentelemetry.sdk.resources import SERVICE_NAME, Resource + + name = os.environ.get("OTEL_SERVICE_NAME", "").strip() or DEFAULT_SERVICE_NAME + return Resource.create({SERVICE_NAME: name}) + + +def _export_timeout() -> float | None: + if any(os.environ.get(name, "").strip() for name in _TIMEOUT_ENV): + return None + return DEFAULT_EXPORT_TIMEOUT_SECONDS + + def configure_logs(*, logger_provider: LoggerProvider | None = None) -> None: global _provider, _provider_is_ours @@ -36,13 +56,15 @@ def configure_logs(*, logger_provider: LoggerProvider | None = None) -> None: from opentelemetry.sdk._logs import LoggerProvider as SdkLoggerProvider from opentelemetry.sdk._logs.export import BatchLogRecordProcessor - logger_provider = SdkLoggerProvider() - logger_provider.add_log_record_processor(BatchLogRecordProcessor(OTLPLogExporter())) + logger_provider = SdkLoggerProvider(resource=_resource()) + logger_provider.add_log_record_processor( + BatchLogRecordProcessor(OTLPLogExporter(timeout=_export_timeout())) + ) from opentelemetry.instrumentation.logging.handler import LoggingHandler handler = LoggingHandler(level=logging.INFO, logger_provider=logger_provider) - handler.addFilter(_SkipOpenTelemetry()) + handler.addFilter(_SkipUnexported()) setattr(handler, _MARKER, True) logging.getLogger().addHandler(handler) diff --git a/tests/test_observability.py b/tests/test_observability.py index 04f31dd..eea43fa 100644 --- a/tests/test_observability.py +++ b/tests/test_observability.py @@ -74,6 +74,12 @@ def test_opentelemetry_own_lines_are_not_exported(exported): assert _bodies(exported) == [] +def test_login_code_lines_are_not_exported(exported): + logging.getLogger("pipeline.auth.email_otp").info("email-otp: login code sent to a@b.in") + + assert _bodies(exported) == [] + + def test_configuring_twice_keeps_one_handler(exported): observability.configure_logs(logger_provider=LoggerProvider()) @@ -84,3 +90,63 @@ def test_reset_removes_the_handler(exported): observability.reset_logs() assert _our_handlers() == [] + + +@pytest.fixture +def built_provider(monkeypatch): + monkeypatch.setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://collector:4318") + yield lambda: observability._provider + observability.reset_logs() + + +def test_lines_are_labelled_knowledge_provider_without_a_service_name(monkeypatch, built_provider): + monkeypatch.delenv("OTEL_SERVICE_NAME", raising=False) + + observability.configure_logs() + + assert built_provider().resource.attributes["service.name"] == "knowledge-provider" + + +def test_service_name_from_env_wins(monkeypatch, built_provider): + monkeypatch.setenv("OTEL_SERVICE_NAME", "knowledge-provider-worker") + + observability.configure_logs() + + assert built_provider().resource.attributes["service.name"] == "knowledge-provider-worker" + + +def test_export_gives_up_within_five_seconds_by_default(monkeypatch): + monkeypatch.delenv("OTEL_EXPORTER_OTLP_TIMEOUT", raising=False) + monkeypatch.delenv("OTEL_EXPORTER_OTLP_LOGS_TIMEOUT", raising=False) + + assert observability._export_timeout() == 5.0 + + +def test_export_timeout_from_env_is_left_to_the_sdk(monkeypatch): + monkeypatch.setenv("OTEL_EXPORTER_OTLP_TIMEOUT", "2") + + assert observability._export_timeout() is None + + +def test_reset_leaves_an_injected_provider_running(monkeypatch): + monkeypatch.setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://collector:4318") + exporter = InMemoryLogRecordExporter() + provider = LoggerProvider() + provider.add_log_record_processor(SimpleLogRecordProcessor(exporter)) + observability.configure_logs(logger_provider=provider) + + observability.reset_logs() + provider.get_logger("caller").emit(body="still mine") + + assert [item.log_record.body for item in exporter.get_finished_logs()] == ["still mine"] + + +def test_reset_shuts_down_a_provider_it_built(built_provider): + observability.configure_logs() + provider = built_provider() + shutdowns = [] + provider.shutdown = lambda: shutdowns.append(True) + + observability.reset_logs() + + assert shutdowns == [True] diff --git a/tests/test_observability_wiring.py b/tests/test_observability_wiring.py new file mode 100644 index 0000000..9ae1510 --- /dev/null +++ b/tests/test_observability_wiring.py @@ -0,0 +1,53 @@ +import asyncio +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from pipeline import worker +from pipeline.api import api + + +@pytest.fixture +def calls(): + return [] + + +def test_worker_configures_logs_and_resets_them_on_exit(monkeypatch, calls): + monkeypatch.setattr(worker, "configure_logs", lambda: calls.append("configure")) + monkeypatch.setattr(worker, "reset_logs", lambda: calls.append("reset")) + monkeypatch.setattr(worker, "_run", AsyncMock(side_effect=lambda: calls.append("run"))) + + asyncio.run(worker.main()) + + assert calls == ["configure", "run", "reset"] + + +def test_worker_resets_logs_when_it_fails(monkeypatch, calls): + monkeypatch.setattr(worker, "configure_logs", lambda: calls.append("configure")) + monkeypatch.setattr(worker, "reset_logs", lambda: calls.append("reset")) + monkeypatch.setattr(worker, "_run", AsyncMock(side_effect=RuntimeError("temporal down"))) + + with pytest.raises(RuntimeError): + asyncio.run(worker.main()) + + assert calls == ["configure", "reset"] + + +def test_api_lifespan_configures_logs_first_and_resets_them_on_shutdown(monkeypatch, calls): + monkeypatch.setattr(api, "configure_logs", lambda: calls.append("configure")) + monkeypatch.setattr(api, "reset_logs", lambda: calls.append("reset")) + monkeypatch.setattr(api, "validate_network_provider_config", lambda: calls.append("validate")) + monkeypatch.setattr(api, "load_auth_config", lambda: SimpleNamespace(disabled=True)) + monkeypatch.setattr(api, "validate_auth_config", lambda _cfg: None) + monkeypatch.setattr(api.db, "init_db", lambda: None) + monkeypatch.setattr(api.Client, "connect", AsyncMock(return_value=MagicMock())) + monkeypatch.setattr(api, "Minio", lambda *a, **k: MagicMock(bucket_exists=lambda _b: True)) + + async def run(): + async with api.lifespan(api.app): + calls.append("serving") + + asyncio.run(run()) + + assert calls == ["configure", "validate", "serving", "reset"] diff --git a/tests/test_search_provider_integration.py b/tests/test_search_provider_integration.py index bd28f3f..1ccd05d 100644 --- a/tests/test_search_provider_integration.py +++ b/tests/test_search_provider_integration.py @@ -64,3 +64,25 @@ def test_provider_composes_with_group(local_store, test_client): body = search(test_client, provider_id="bv.provider.oan.dev", group="mh") assert body["hits"] == [] + + +@pytest.mark.integration +@pytest.mark.api +def test_search_logs_name_the_caller_by_id_not_email(local_store, test_client, caplog): + from dataclasses import replace + + from pipeline.api import api + from pipeline.auth.deps import get_current_user + from pipeline.auth.models import local_bypass_user + + caller = replace(local_bypass_user(), user_id="kc-123", username="farmer.desk@state.gov.in") + api.app.dependency_overrides[get_current_user] = lambda: caller + try: + with caplog.at_level("INFO"): + search(test_client) + finally: + api.app.dependency_overrides.pop(get_current_user, None) + + search_lines = [r.getMessage() for r in caplog.records if r.getMessage().startswith("search ")] + assert search_lines and all("user=kc-123" in line for line in search_lines) + assert not any("@" in line.split("payload=")[0] for line in search_lines) From 1db8701a77dd7b8a3012015ad7ac4c6e8beac19f Mon Sep 17 00:00:00 2001 From: Hariharan Thiagarajan Date: Thu, 5 Nov 2026 15:49:12 +0530 Subject: [PATCH 3/8] docs: add OTEL log-shipping vars to .env.example [#241] --- .env.example | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/.env.example b/.env.example index 6578363..9745de9 100644 --- a/.env.example +++ b/.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= +OTEL_EXPORTER_OTLP_HEADERS= +# Seconds before an export is dropped +OTEL_EXPORTER_OTLP_TIMEOUT=5 \ No newline at end of file From c295e34a37f756a328f536c316f2a8ba70215291 Mon Sep 17 00:00:00 2001 From: Hariharan Thiagarajan Date: Thu, 5 Nov 2026 15:52:20 +0530 Subject: [PATCH 4/8] revert: keep username/email in shipped logs [#241] Operator identity is fine to ship; restore user.username/email on /search lines and export email-OTP lines again. Co-Authored-By: Claude Opus 5.5 --- .env.example | 2 +- docs/ADR/0007-ship-logs-over-otlp.md | 3 +-- pipeline/api/api.py | 7 ++----- pipeline/api/search.py | 6 +++--- pipeline/auth/email_otp.py | 12 +++++------- pipeline/observability.py | 2 +- tests/test_observability.py | 6 ------ tests/test_search_provider_integration.py | 21 --------------------- 8 files changed, 13 insertions(+), 46 deletions(-) diff --git a/.env.example b/.env.example index 9745de9..5f715e5 100644 --- a/.env.example +++ b/.env.example @@ -273,4 +273,4 @@ OTEL_EXPORTER_OTLP_ENDPOINT= # Shared env only: authorization= OTEL_EXPORTER_OTLP_HEADERS= # Seconds before an export is dropped -OTEL_EXPORTER_OTLP_TIMEOUT=5 \ No newline at end of file +OTEL_EXPORTER_OTLP_TIMEOUT=5 diff --git a/docs/ADR/0007-ship-logs-over-otlp.md b/docs/ADR/0007-ship-logs-over-otlp.md index a1b288c..a7211b9 100644 --- a/docs/ADR/0007-ship-logs-over-otlp.md +++ b/docs/ADR/0007-ship-logs-over-otlp.md @@ -13,8 +13,7 @@ Decision: - 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. -- Operator emails are not shipped. `/search` lines name the caller by Keycloak user id, and email-OTP lines go to the `pipeline.auth.email_otp` logger, which is not exported. They still print locally. -- Query text and the raw `/search` payload stay in the exported lines. 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. +- 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: diff --git a/pipeline/api/api.py b/pipeline/api/api.py index 1111807..5f00795 100644 --- a/pipeline/api/api.py +++ b/pipeline/api/api.py @@ -58,9 +58,6 @@ send_otp, verify_otp, ) -from ..auth.email_otp import ( - logger as email_otp_logger, -) from ..auth.keycloak_admin import ( list_access_options, list_realm_users, @@ -614,7 +611,7 @@ async def auth_otp_request(request: Request, body: OtpRequestBody): """ email = normalize_email(body.email) result = await asyncio.to_thread(send_otp, email) - email_otp_logger.info("email-otp: login code requested for %s", email) + logging.info("email-otp: login code requested for %s", email) return result @@ -628,7 +625,7 @@ async def auth_otp_verify(request: Request, body: OtpVerifyBody): """ email = normalize_email(body.email) tokens = await asyncio.to_thread(verify_otp, email, body.code) - email_otp_logger.info("email-otp: login succeeded for %s", email) + logging.info("email-otp: login succeeded for %s", email) return tokens diff --git a/pipeline/api/search.py b/pipeline/api/search.py index 72c8e77..70beb7c 100644 --- a/pipeline/api/search.py +++ b/pipeline/api/search.py @@ -196,7 +196,7 @@ async def run_search(payload: dict, user: RequireSearch): logging.info( "search payload user=%s payload=%s", - user.user_id, + user.username or user.email or user.user_id, json.dumps(payload, ensure_ascii=False, default=str), ) @@ -208,7 +208,7 @@ async def run_search(payload: dict, user: RequireSearch): logging.info( "search request user=%s index=%s query=%r top_k=%s", - user.user_id, + user.username or user.email or user.user_id, index_name, query, payload.get("top_k"), @@ -291,7 +291,7 @@ async def run_search(payload: dict, user: RequireSearch): logging.info( "search response user=%s index=%s query=%r candidate_count=%s final_count=%s", - user.user_id, + user.username or user.email or user.user_id, index_name, query, len(hits), diff --git a/pipeline/auth/email_otp.py b/pipeline/auth/email_otp.py index 40349f5..944abee 100644 --- a/pipeline/auth/email_otp.py +++ b/pipeline/auth/email_otp.py @@ -48,8 +48,6 @@ from .. import db from . import keycloak_admin -logger = logging.getLogger(__name__) - OTP_LENGTH = 6 OTP_TTL_SECONDS = 300 OTP_RESEND_COOLDOWN_SECONDS = 30 @@ -211,7 +209,7 @@ def _send_email(to_addr: str, subject: str, body: str) -> None: smtp.login(cfg.username, cfg.password) smtp.send_message(message) except (smtplib.SMTPException, OSError) as exc: - logger.error("email-otp: failed to send code to %s: %s", to_addr, exc) + logging.error("email-otp: failed to send code to %s: %s", to_addr, exc) raise HTTPException( 503, "Could not send the email right now. Please try again shortly." ) from exc @@ -320,13 +318,13 @@ def send_otp(email: str) -> dict[str, Any]: f"This code expires in {OTP_TTL_SECONDS // 60} minutes. " "If you did not request this, you can safely ignore this email.", ) - logger.info("email-otp: login code sent to %s", email) + logging.info("email-otp: login code sent to %s", email) except HTTPException: # Never let an SMTP failure surface as a different response/status # than the "no such account" path — that would itself be an # enumeration oracle. The stored code is now simply unreachable; # the user's next resend attempt (after the cooldown) tries again. - logger.error("email-otp: send failed for %s, response withheld", email) + logging.error("email-otp: send failed for %s, response withheld", email) else: # Matches the typical SMTP round-trip _send_email would otherwise # take, so response latency doesn't leak account existence either. @@ -427,12 +425,12 @@ def _rotate_password() -> str: try: _rotate_password() except HTTPException: - logger.error("email-otp: post-grant password rotation failed for %s", email) + logging.error("email-otp: post-grant password rotation failed for %s", email) if status == 200 and payload.get("access_token"): return _token_response(payload) - logger.error( + logging.error( "email-otp: password-grant bridge failed for %s: status=%s body=%s", email, status, payload, ) diff --git a/pipeline/observability.py b/pipeline/observability.py index fecd833..6a58f67 100644 --- a/pipeline/observability.py +++ b/pipeline/observability.py @@ -17,7 +17,7 @@ _provider_is_ours = False -_NOT_EXPORTED = ("opentelemetry", "pipeline.auth.email_otp") +_NOT_EXPORTED = ("opentelemetry",) class _SkipUnexported(logging.Filter): diff --git a/tests/test_observability.py b/tests/test_observability.py index eea43fa..d63d951 100644 --- a/tests/test_observability.py +++ b/tests/test_observability.py @@ -74,12 +74,6 @@ def test_opentelemetry_own_lines_are_not_exported(exported): assert _bodies(exported) == [] -def test_login_code_lines_are_not_exported(exported): - logging.getLogger("pipeline.auth.email_otp").info("email-otp: login code sent to a@b.in") - - assert _bodies(exported) == [] - - def test_configuring_twice_keeps_one_handler(exported): observability.configure_logs(logger_provider=LoggerProvider()) diff --git a/tests/test_search_provider_integration.py b/tests/test_search_provider_integration.py index 1ccd05d..244aa9f 100644 --- a/tests/test_search_provider_integration.py +++ b/tests/test_search_provider_integration.py @@ -65,24 +65,3 @@ def test_provider_composes_with_group(local_store, test_client): assert body["hits"] == [] - -@pytest.mark.integration -@pytest.mark.api -def test_search_logs_name_the_caller_by_id_not_email(local_store, test_client, caplog): - from dataclasses import replace - - from pipeline.api import api - from pipeline.auth.deps import get_current_user - from pipeline.auth.models import local_bypass_user - - caller = replace(local_bypass_user(), user_id="kc-123", username="farmer.desk@state.gov.in") - api.app.dependency_overrides[get_current_user] = lambda: caller - try: - with caplog.at_level("INFO"): - search(test_client) - finally: - api.app.dependency_overrides.pop(get_current_user, None) - - search_lines = [r.getMessage() for r in caplog.records if r.getMessage().startswith("search ")] - assert search_lines and all("user=kc-123" in line for line in search_lines) - assert not any("@" in line.split("payload=")[0] for line in search_lines) From 87860a8c84f94b85cba55fd196f5f4df1cb7aabc Mon Sep 17 00:00:00 2001 From: Hariharan Thiagarajan Date: Thu, 5 Nov 2026 17:09:33 +0530 Subject: [PATCH 5/8] feat: log startup, requests and activity lifecycle [#241] - Request middleware: method, URI with query, status, duration; error detail on non-2xx, traceback on unhandled exceptions; no bodies - Worker interceptor logs every activity's start, completion or failure - API and worker startup messages are log lines instead of prints Co-Authored-By: Claude Opus 5.5 --- docs/ADR/0007-ship-logs-over-otlp.md | 5 +- pipeline/activity_logging.py | 47 +++++++++++++ pipeline/api/api.py | 18 +++-- pipeline/api/request_logging.py | 76 +++++++++++++++++++++ pipeline/worker.py | 18 +++-- tests/test_activity_logging.py | 52 +++++++++++++++ tests/test_observability_wiring.py | 9 ++- tests/test_request_logging.py | 98 ++++++++++++++++++++++++++++ 8 files changed, 309 insertions(+), 14 deletions(-) create mode 100644 pipeline/activity_logging.py create mode 100644 pipeline/api/request_logging.py create mode 100644 tests/test_activity_logging.py create mode 100644 tests/test_request_logging.py diff --git a/docs/ADR/0007-ship-logs-over-otlp.md b/docs/ADR/0007-ship-logs-over-otlp.md index a7211b9..5ff32a3 100644 --- a/docs/ADR/0007-ship-logs-over-otlp.md +++ b/docs/ADR/0007-ship-logs-over-otlp.md @@ -18,7 +18,8 @@ Decision: Consequences: - Logs only. Traces and metrics are not emitted. -- uvicorn access lines don't propagate to root, so they are not shipped. -- `print(...)` output, used for some worker and API startup messages, stays on stdout and is not shipped. +- Every API request logs one line on `pipeline.api.requests`: method, path with query string, status and duration, never a body. Non-2xx responses log at WARNING with the error detail, and unhandled exceptions log at ERROR with the traceback. uvicorn's own access log doesn't reach the root logger, so it is not shipped and is not needed. +- Every activity logs start, completion or failure with name, workflow id, attempt and duration, through a worker interceptor (`pipeline/activity_logging.py`), so new activities are covered without per-activity code. +- API and worker startup messages are log lines, not `print`s, so they are shipped. - `/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. diff --git a/pipeline/activity_logging.py b/pipeline/activity_logging.py new file mode 100644 index 0000000..46ab6cb --- /dev/null +++ b/pipeline/activity_logging.py @@ -0,0 +1,47 @@ +from __future__ import annotations + +import logging +import time +from typing import Any + +from temporalio import activity +from temporalio.worker import ( + ActivityInboundInterceptor, + ExecuteActivityInput, + Interceptor, +) + +logger = logging.getLogger(__name__) + + +class ActivityLogging(Interceptor): + def intercept_activity(self, next: ActivityInboundInterceptor) -> ActivityInboundInterceptor: + return _LogActivity(next) + + +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 Exception as exc: + logger.error( + "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 + + +def _elapsed_ms(started: float) -> int: + return int((time.perf_counter() - started) * 1000) diff --git a/pipeline/api/api.py b/pipeline/api/api.py index 5f00795..25d328f 100644 --- a/pipeline/api/api.py +++ b/pipeline/api/api.py @@ -110,6 +110,7 @@ ReingestionWorkflow, TranslationOnlyWorkflow, ) +from . import request_logging LOG_LEVEL = getattr(logging, os.environ.get("LOG_LEVEL", "INFO").strip().upper(), logging.INFO) # uvicorn only configures its own "uvicorn"/"uvicorn.access" loggers, so plain @@ -156,12 +157,12 @@ async def lifespan(app: FastAPI): ) # Initialize SQLite database - print("Initializing SQLite database...") + logging.info("api startup: initializing SQLite database") db.init_db() # Temporal temporal_host = os.environ.get("TEMPORAL_HOST", "localhost:7233") - print(f"Connecting to Temporal at {temporal_host}") + logging.info("api startup: connecting to Temporal at %s", temporal_host) temporal_client = await Client.connect(temporal_host) # MinIO - credentials required via environment variables @@ -173,7 +174,7 @@ async def lifespan(app: FastAPI): if not minio_access_key or not minio_secret_key: raise RuntimeError("MINIO_ACCESS_KEY and MINIO_SECRET_KEY environment variables are required") - print(f"Connecting to MinIO at {minio_endpoint}") + logging.info("api startup: connecting to MinIO at %s", minio_endpoint) minio_client = Minio( minio_endpoint, access_key=minio_access_key, @@ -184,9 +185,17 @@ async def lifespan(app: FastAPI): # Ensure bucket exists if not minio_client.bucket_exists(MINIO_BUCKET): minio_client.make_bucket(MINIO_BUCKET) - print(f"Created MinIO bucket: {MINIO_BUCKET}") + logging.info("api startup: created MinIO bucket %s", MINIO_BUCKET) + logging.info( + "api started auth=%s temporal=%s minio=%s bucket=%s", + "disabled" if auth_cfg.disabled else "enabled", + temporal_host, + minio_endpoint, + MINIO_BUCKET, + ) yield + logging.info("api stopping") reset_logs() @@ -242,6 +251,7 @@ async def lifespan(app: FastAPI): allow_methods=["GET", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"], allow_headers=["*"], ) +request_logging.install(app) # Rate limiting configuration # Default: 100 requests/minute for general endpoints, 10/minute for uploads diff --git a/pipeline/api/request_logging.py b/pipeline/api/request_logging.py new file mode 100644 index 0000000..865328e --- /dev/null +++ b/pipeline/api/request_logging.py @@ -0,0 +1,76 @@ +from __future__ import annotations + +import logging +import time + +from fastapi import FastAPI, Request +from fastapi.exception_handlers import ( + http_exception_handler, + request_validation_exception_handler, +) +from fastapi.exceptions import RequestValidationError +from starlette.exceptions import HTTPException as StarletteHTTPException + +logger = logging.getLogger("pipeline.api.requests") + + +def install(app: FastAPI) -> None: + app.add_exception_handler(StarletteHTTPException, _record_http_error) + app.add_exception_handler(RequestValidationError, _record_validation_error) + app.middleware("http")(_log_request) + + +async def _record_http_error(request: Request, exc: StarletteHTTPException): + request.state.error = exc.detail + return await http_exception_handler(request, exc) + + +async def _record_validation_error(request: Request, exc: RequestValidationError): + request.state.error = [ + {"type": e.get("type"), "loc": e.get("loc"), "msg": e.get("msg")} for e in exc.errors() + ] + return await request_validation_exception_handler(request, exc) + + +def _uri(request: Request) -> str: + query = request.url.query + return f"{request.url.path}?{query}" if query else request.url.path + + +async def _log_request(request: Request, call_next): + started = time.perf_counter() + try: + response = await call_next(request) + except Exception as exc: + logger.exception( + "request method=%s uri=%s status=500 duration_ms=%d error=%r", + request.method, + _uri(request), + _elapsed_ms(started), + exc, + ) + raise + + status = response.status_code + if status < 400: + logger.info( + "request method=%s uri=%s status=%s duration_ms=%d", + request.method, + _uri(request), + status, + _elapsed_ms(started), + ) + else: + logger.warning( + "request method=%s uri=%s status=%s duration_ms=%d error=%r", + request.method, + _uri(request), + status, + _elapsed_ms(started), + getattr(request.state, "error", None), + ) + return response + + +def _elapsed_ms(started: float) -> int: + return int((time.perf_counter() - started) * 1000) diff --git a/pipeline/worker.py b/pipeline/worker.py index ca861c8..576ab12 100644 --- a/pipeline/worker.py +++ b/pipeline/worker.py @@ -26,6 +26,7 @@ run_ocr_and_store, update_document_state, ) +from .activity_logging import ActivityLogging from .network_group import validate_network_provider_config from .observability import configure_logs, reset_logs from .workflows import ( @@ -51,6 +52,7 @@ logging.getLogger("temporalio").setLevel(logging.INFO) TASK_QUEUE = "ocr-pipeline" +INTERCEPTORS = [ActivityLogging()] async def main(): @@ -68,17 +70,16 @@ async def _run(): translation_provider = os.environ.get("TRANSLATION_PROVIDER", "gemma_vllm").strip().lower() if translation_provider in {"gemma_vllm", "gemma4", "gemma"}: if not os.environ.get("TRANSLATION_VLLM_BASE_URL", "").strip(): - print("Error: TRANSLATION_VLLM_BASE_URL not set") + logging.error("worker startup: TRANSLATION_VLLM_BASE_URL not set, exiting") return # Initialize SQLite database - print("Initializing SQLite database...") + logging.info("worker startup: initializing SQLite database") db.init_db() temporal_host = os.environ.get("TEMPORAL_HOST", "localhost:7233") max_concurrent_activities = int(os.environ.get("TEMPORAL_MAX_CONCURRENT_ACTIVITIES", "4")) - print(f"Connecting to Temporal at {temporal_host}") - print(f"Worker activity concurrency: {max_concurrent_activities}") + logging.info("worker startup: connecting to Temporal at %s", temporal_host) client = await Client.connect(temporal_host) @@ -86,6 +87,7 @@ async def _run(): client, task_queue=TASK_QUEUE, max_concurrent_activities=max_concurrent_activities, + interceptors=INTERCEPTORS, workflows=[ DocumentPipelineWorkflow, ReingestionWorkflow, @@ -112,8 +114,12 @@ async def _run(): ], ) - print(f"Worker started on queue: {TASK_QUEUE}") - print("Waiting for workflows...") + logging.info( + "worker started queue=%s temporal=%s activity_concurrency=%s", + TASK_QUEUE, + temporal_host, + max_concurrent_activities, + ) await worker.run() diff --git a/tests/test_activity_logging.py b/tests/test_activity_logging.py new file mode 100644 index 0000000..7f5e078 --- /dev/null +++ b/tests/test_activity_logging.py @@ -0,0 +1,52 @@ +import logging +from unittest.mock import AsyncMock, MagicMock + +import pytest +from temporalio.testing import ActivityEnvironment + +from pipeline import activity_logging, worker + +LOGGER = "pipeline.activity_logging" + + +def _interceptor(next_result=None, error=None): + following = MagicMock() + following.execute_activity = AsyncMock(return_value=next_result, side_effect=error) + return activity_logging.ActivityLogging().intercept_activity(following) + + +def _messages(caplog): + return [(r.levelno, r.getMessage()) for r in caplog.records if r.name == LOGGER] + + +@pytest.mark.asyncio +async def test_logs_start_and_completion(caplog): + interceptor = _interceptor(next_result="done") + + with caplog.at_level(logging.INFO, logger=LOGGER): + result = await ActivityEnvironment().run(interceptor.execute_activity, MagicMock()) + + assert result == "done" + [(start_level, start), (end_level, end)] = _messages(caplog) + assert start_level == end_level == logging.INFO + assert start.startswith("activity started name=unknown") + assert "workflow_id=test" in start and "attempt=1" in start + assert end.startswith("activity completed name=unknown") and "duration_ms=" in end + + +@pytest.mark.asyncio +async def test_logs_failure_and_reraises(caplog): + interceptor = _interceptor(error=ValueError("bad page")) + + with caplog.at_level(logging.INFO, logger=LOGGER): + with pytest.raises(ValueError): + await ActivityEnvironment().run(interceptor.execute_activity, MagicMock()) + + [_, (level, failed)] = _messages(caplog) + assert level == logging.ERROR + assert failed.startswith("activity failed name=unknown") + assert "error=ValueError('bad page')" in failed + + +def test_worker_registers_the_interceptor(): + assert any(isinstance(i, activity_logging.ActivityLogging) for i in worker.INTERCEPTORS) diff --git a/tests/test_observability_wiring.py b/tests/test_observability_wiring.py index 9ae1510..1b29ee7 100644 --- a/tests/test_observability_wiring.py +++ b/tests/test_observability_wiring.py @@ -34,7 +34,7 @@ def test_worker_resets_logs_when_it_fails(monkeypatch, calls): assert calls == ["configure", "reset"] -def test_api_lifespan_configures_logs_first_and_resets_them_on_shutdown(monkeypatch, calls): +def test_api_lifespan_configures_logs_first_and_resets_them_on_shutdown(monkeypatch, calls, caplog): monkeypatch.setattr(api, "configure_logs", lambda: calls.append("configure")) monkeypatch.setattr(api, "reset_logs", lambda: calls.append("reset")) monkeypatch.setattr(api, "validate_network_provider_config", lambda: calls.append("validate")) @@ -48,6 +48,11 @@ async def run(): async with api.lifespan(api.app): calls.append("serving") - asyncio.run(run()) + monkeypatch.setenv("MINIO_ACCESS_KEY", "key") + monkeypatch.setenv("MINIO_SECRET_KEY", "secret") + + with caplog.at_level("INFO"): + asyncio.run(run()) assert calls == ["configure", "validate", "serving", "reset"] + assert any(r.getMessage().startswith("api started auth=disabled") for r in caplog.records) diff --git a/tests/test_request_logging.py b/tests/test_request_logging.py new file mode 100644 index 0000000..68bfe42 --- /dev/null +++ b/tests/test_request_logging.py @@ -0,0 +1,98 @@ +import logging + +import pytest +from fastapi import FastAPI, HTTPException +from fastapi.testclient import TestClient + +from pipeline.api import request_logging + +LOGGER = "pipeline.api.requests" + + +@pytest.fixture +def client(): + app = FastAPI() + request_logging.install(app) + + @app.get("/ok") + def ok(): + return {"secret": "body"} + + @app.get("/missing") + def missing(): + raise HTTPException(404, "Document not found") + + @app.get("/boom") + def boom(): + raise RuntimeError("database locked") + + @app.get("/typed/{n}") + def typed(n: int): + return {"n": n} + + return TestClient(app, raise_server_exceptions=False) + + +def _lines(caplog): + return [r for r in caplog.records if r.name == LOGGER] + + +def test_success_logs_method_uri_and_status_without_body(client, caplog): + with caplog.at_level(logging.INFO, logger=LOGGER): + client.get("/ok?index=documents") + + [record] = _lines(caplog) + message = record.getMessage() + assert record.levelno == logging.INFO + assert "method=GET uri=/ok?index=documents status=200" in message + assert "duration_ms=" in message + assert "secret" not in message + + +def test_error_response_logs_status_and_error_detail(client, caplog): + with caplog.at_level(logging.INFO, logger=LOGGER): + response = client.get("/missing") + + assert response.json() == {"detail": "Document not found"} + [record] = _lines(caplog) + assert record.levelno == logging.WARNING + assert "uri=/missing status=404" in record.getMessage() + assert "error='Document not found'" in record.getMessage() + + +def test_validation_error_logs_422_with_its_detail(client, caplog): + with caplog.at_level(logging.INFO, logger=LOGGER): + response = client.get("/typed/abc") + + assert response.status_code == 422 + [record] = _lines(caplog) + assert "status=422" in record.getMessage() + assert "int_parsing" in record.getMessage() + + +def test_unknown_route_logs_404(client, caplog): + with caplog.at_level(logging.INFO, logger=LOGGER): + client.get("/nowhere") + + [record] = _lines(caplog) + assert "uri=/nowhere status=404" in record.getMessage() + assert "error='Not Found'" in record.getMessage() + + +def test_unhandled_exception_logs_500_with_traceback(client, caplog): + with caplog.at_level(logging.INFO, logger=LOGGER): + response = client.get("/boom") + + assert response.status_code == 500 + [record] = _lines(caplog) + assert record.levelno == logging.ERROR + assert "uri=/boom status=500" in record.getMessage() + assert "error=RuntimeError('database locked')" in record.getMessage() + assert record.exc_info is not None + + +def test_the_app_logs_its_requests(test_client, caplog): + with caplog.at_level(logging.INFO, logger=LOGGER): + test_client.get("/health") + + assert any("uri=/health status=200" in r.getMessage() for r in _lines(caplog)) From 4dd0daffa7cd360ea5017e437ca62dc45e2243b7 Mon Sep 17 00:00:00 2001 From: Hariharan Thiagarajan Date: Thu, 5 Nov 2026 17:27:57 +0530 Subject: [PATCH 6/8] fix: address review on request, activity and crash logging [#241] - Request logging is pure ASGI: streamed responses timed to the last byte, mid-stream failures logged; 5xx at ERROR with the caught cause; 429 and reason-phrase fallback for missing detail; /health at DEBUG - Temporal interceptor also logs workflows, and cancellations of both; activity failures at WARNING (Temporal logs the traceback) - Log API startup failures and worker crashes, then flush - Log the three vector-delete handlers that swallowed exceptions Co-Authored-By: Claude Opus 5.5 --- docs/ADR/0007-ship-logs-over-otlp.md | 7 +- pipeline/activity_logging.py | 47 --------- pipeline/api/api.py | 23 ++++- pipeline/api/request_logging.py | 146 +++++++++++++++++---------- pipeline/temporal_logging.py | 87 ++++++++++++++++ pipeline/worker.py | 8 +- tests/test_activity_logging.py | 52 ---------- tests/test_observability_wiring.py | 53 ++++++++++ tests/test_request_logging.py | 100 ++++++++++++++++-- tests/test_temporal_logging.py | 129 +++++++++++++++++++++++ 10 files changed, 479 insertions(+), 173 deletions(-) delete mode 100644 pipeline/activity_logging.py create mode 100644 pipeline/temporal_logging.py delete mode 100644 tests/test_activity_logging.py create mode 100644 tests/test_temporal_logging.py diff --git a/docs/ADR/0007-ship-logs-over-otlp.md b/docs/ADR/0007-ship-logs-over-otlp.md index 5ff32a3..d787a6c 100644 --- a/docs/ADR/0007-ship-logs-over-otlp.md +++ b/docs/ADR/0007-ship-logs-over-otlp.md @@ -18,8 +18,9 @@ Decision: 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. Non-2xx responses log at WARNING with the error detail, and unhandled exceptions log at ERROR with the traceback. uvicorn's own access log doesn't reach the root logger, so it is not shipped and is not needed. -- Every activity logs start, completion or failure with name, workflow id, attempt and duration, through a worker interceptor (`pipeline/activity_logging.py`), so new activities are covered without per-activity code. -- API and worker startup messages are log lines, not `print`s, so they are shipped. +- 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/temporal_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. diff --git a/pipeline/activity_logging.py b/pipeline/activity_logging.py deleted file mode 100644 index 46ab6cb..0000000 --- a/pipeline/activity_logging.py +++ /dev/null @@ -1,47 +0,0 @@ -from __future__ import annotations - -import logging -import time -from typing import Any - -from temporalio import activity -from temporalio.worker import ( - ActivityInboundInterceptor, - ExecuteActivityInput, - Interceptor, -) - -logger = logging.getLogger(__name__) - - -class ActivityLogging(Interceptor): - def intercept_activity(self, next: ActivityInboundInterceptor) -> ActivityInboundInterceptor: - return _LogActivity(next) - - -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 Exception as exc: - logger.error( - "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 - - -def _elapsed_ms(started: float) -> int: - return int((time.perf_counter() - started) * 1000) diff --git a/pipeline/api/api.py b/pipeline/api/api.py index 25d328f..6b5c5f0 100644 --- a/pipeline/api/api.py +++ b/pipeline/api/api.py @@ -134,9 +134,21 @@ @asynccontextmanager async def lifespan(app: FastAPI): """Initialize Temporal and MinIO clients on startup.""" + configure_logs() + try: + await _start_services() + except Exception: + logging.exception("api startup failed") + reset_logs() + raise + yield + logging.info("api stopping") + reset_logs() + + +async def _start_services(): global temporal_client, minio_client, MINIO_BUCKET - configure_logs() validate_network_provider_config() auth_cfg = load_auth_config() @@ -194,9 +206,6 @@ async def lifespan(app: FastAPI): minio_endpoint, MINIO_BUCKET, ) - yield - logging.info("api stopping") - reset_logs() app = FastAPI( @@ -263,7 +272,7 @@ async def lifespan(app: FastAPI): limiter = Limiter(key_func=get_remote_address) app.state.limiter = limiter -app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler) +app.add_exception_handler(RateLimitExceeded, request_logging.recording(_rate_limit_exceeded_handler)) # Routers split out of this module, one per resource. Register them here. from .documents import router as documents_router # noqa: E402 @@ -569,6 +578,7 @@ def delete_single_chunk_from_index(document_id: str, chunk_num: int, index_name: resolved_index = index_name or get_default_index_name() return store.delete_chunk(resolved_index, document_id, chunk_num) except Exception as e: + logging.exception("vector chunk delete failed doc_id=%s chunk=%s", document_id, chunk_num) return {"deleted": False, "error": str(e)} @@ -584,6 +594,7 @@ def delete_chunks_from_index(document_id: str, index_name: str = "documents-inde result["doc_id"] = document_id return result except Exception as e: + logging.exception("vector document delete failed doc_id=%s", document_id) return {"deleted": 0, "doc_id": document_id, "error": str(e)} @@ -1019,6 +1030,7 @@ def _delete_document_points_from_prod(doc: dict) -> dict: store = QdrantVectorStore(client=client) return store.delete_by_doc_id(collection, doc_id) except Exception as exc: + logging.exception("prod vector delete failed doc_id=%s collection=%s", doc_id, collection) return {"deleted": False, "error": str(exc), "collection": collection} @@ -1048,6 +1060,7 @@ def _delete_scheme_points_from_prod(doc: dict) -> dict: store = QdrantVectorStore(client=client) return store.delete_by_doc_id(collection, doc_id) except Exception as exc: + logging.exception("prod vector delete failed doc_id=%s collection=%s", doc_id, collection) return {"deleted": False, "error": str(exc), "collection": collection} diff --git a/pipeline/api/request_logging.py b/pipeline/api/request_logging.py index 865328e..d5bc3ba 100644 --- a/pipeline/api/request_logging.py +++ b/pipeline/api/request_logging.py @@ -1,7 +1,10 @@ from __future__ import annotations +import inspect import logging import time +from http import HTTPStatus +from typing import Any, Callable from fastapi import FastAPI, Request from fastapi.exception_handlers import ( @@ -10,66 +13,101 @@ ) from fastapi.exceptions import RequestValidationError from starlette.exceptions import HTTPException as StarletteHTTPException +from starlette.types import ASGIApp, Message, Receive, Scope, Send logger = logging.getLogger("pipeline.api.requests") +QUIET_PATHS = frozenset({"/health"}) -def install(app: FastAPI) -> None: - app.add_exception_handler(StarletteHTTPException, _record_http_error) - app.add_exception_handler(RequestValidationError, _record_validation_error) - app.middleware("http")(_log_request) - - -async def _record_http_error(request: Request, exc: StarletteHTTPException): - request.state.error = exc.detail - return await http_exception_handler(request, exc) - - -async def _record_validation_error(request: Request, exc: RequestValidationError): - request.state.error = [ - {"type": e.get("type"), "loc": e.get("loc"), "msg": e.get("msg")} for e in exc.errors() - ] - return await request_validation_exception_handler(request, exc) - -def _uri(request: Request) -> str: - query = request.url.query - return f"{request.url.path}?{query}" if query else request.url.path - - -async def _log_request(request: Request, call_next): - started = time.perf_counter() +def install(app: FastAPI) -> None: + app.add_exception_handler(StarletteHTTPException, recording(http_exception_handler)) + app.add_exception_handler(RequestValidationError, recording(request_validation_exception_handler)) + app.add_middleware(RequestLogging) + + +def recording(handler: Callable) -> Callable: + async def record_then_handle(request: Request, exc: Exception): + request.state.error = _detail(exc) + request.state.error_cause = exc.__cause__ + result = handler(request, exc) + return await result if inspect.isawaitable(result) else result + + return record_then_handle + + +def _detail(exc: Exception) -> Any: + if isinstance(exc, RequestValidationError): + return [{"type": e.get("type"), "loc": e.get("loc"), "msg": e.get("msg")} for e in exc.errors()] + return getattr(exc, "detail", None) + + +class RequestLogging: + def __init__(self, app: ASGIApp) -> None: + self.app = app + + async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: + if scope["type"] != "http": + await self.app(scope, receive, send) + return + + started = time.perf_counter() + status: dict[str, int] = {} + + async def send_and_watch(message: Message) -> None: + if message["type"] == "http.response.start": + status["code"] = message["status"] + await send(message) + + try: + await self.app(scope, receive, send_and_watch) + except Exception as exc: + logger.error( + "request method=%s uri=%s status=%s duration_ms=%d error=%r", + scope["method"], + _uri(scope), + status.get("code", 500), + _elapsed_ms(started), + exc, + exc_info=True, + ) + raise + + _log_response(scope, status.get("code", 500), _elapsed_ms(started)) + + +def _log_response(scope: Scope, code: int, duration_ms: int) -> None: + method, uri = scope["method"], _uri(scope) + if code < 400: + level = logging.DEBUG if scope["path"] in QUIET_PATHS else logging.INFO + logger.log(level, "request method=%s uri=%s status=%s duration_ms=%d", method, uri, code, duration_ms) + return + + state = scope.get("state") or {} + error = state.get("error") or _reason(code) + cause = state.get("error_cause") + logger.log( + logging.ERROR if code >= 500 else logging.WARNING, + "request method=%s uri=%s status=%s duration_ms=%d error=%r", + method, + uri, + code, + duration_ms, + error, + exc_info=(type(cause), cause, cause.__traceback__) if code >= 500 and cause else None, + ) + + +def _uri(scope: Scope) -> str: + query = scope.get("query_string", b"").decode("latin-1") + return f"{scope['path']}?{query}" if query else scope["path"] + + +def _reason(code: int) -> str: try: - response = await call_next(request) - except Exception as exc: - logger.exception( - "request method=%s uri=%s status=500 duration_ms=%d error=%r", - request.method, - _uri(request), - _elapsed_ms(started), - exc, - ) - raise - - status = response.status_code - if status < 400: - logger.info( - "request method=%s uri=%s status=%s duration_ms=%d", - request.method, - _uri(request), - status, - _elapsed_ms(started), - ) - else: - logger.warning( - "request method=%s uri=%s status=%s duration_ms=%d error=%r", - request.method, - _uri(request), - status, - _elapsed_ms(started), - getattr(request.state, "error", None), - ) - return response + return HTTPStatus(code).phrase + except ValueError: + return str(code) def _elapsed_ms(started: float) -> int: diff --git a/pipeline/temporal_logging.py b/pipeline/temporal_logging.py new file mode 100644 index 0000000..6966bfa --- /dev/null +++ b/pipeline/temporal_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 TemporalLogging(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) diff --git a/pipeline/worker.py b/pipeline/worker.py index 576ab12..bdb6ae8 100644 --- a/pipeline/worker.py +++ b/pipeline/worker.py @@ -26,9 +26,9 @@ run_ocr_and_store, update_document_state, ) -from .activity_logging import ActivityLogging from .network_group import validate_network_provider_config from .observability import configure_logs, reset_logs +from .temporal_logging import TemporalLogging from .workflows import ( ChunkingOnlyWorkflow, DocumentPipelineWorkflow, @@ -52,7 +52,7 @@ logging.getLogger("temporalio").setLevel(logging.INFO) TASK_QUEUE = "ocr-pipeline" -INTERCEPTORS = [ActivityLogging()] +INTERCEPTORS = [TemporalLogging()] async def main(): @@ -60,7 +60,11 @@ async def main(): configure_logs() try: await _run() + except Exception: + logging.exception("worker crashed") + raise finally: + logging.info("worker stopping") reset_logs() diff --git a/tests/test_activity_logging.py b/tests/test_activity_logging.py deleted file mode 100644 index 7f5e078..0000000 --- a/tests/test_activity_logging.py +++ /dev/null @@ -1,52 +0,0 @@ -import logging -from unittest.mock import AsyncMock, MagicMock - -import pytest -from temporalio.testing import ActivityEnvironment - -from pipeline import activity_logging, worker - -LOGGER = "pipeline.activity_logging" - - -def _interceptor(next_result=None, error=None): - following = MagicMock() - following.execute_activity = AsyncMock(return_value=next_result, side_effect=error) - return activity_logging.ActivityLogging().intercept_activity(following) - - -def _messages(caplog): - return [(r.levelno, r.getMessage()) for r in caplog.records if r.name == LOGGER] - - -@pytest.mark.asyncio -async def test_logs_start_and_completion(caplog): - interceptor = _interceptor(next_result="done") - - with caplog.at_level(logging.INFO, logger=LOGGER): - result = await ActivityEnvironment().run(interceptor.execute_activity, MagicMock()) - - assert result == "done" - [(start_level, start), (end_level, end)] = _messages(caplog) - assert start_level == end_level == logging.INFO - assert start.startswith("activity started name=unknown") - assert "workflow_id=test" in start and "attempt=1" in start - assert end.startswith("activity completed name=unknown") and "duration_ms=" in end - - -@pytest.mark.asyncio -async def test_logs_failure_and_reraises(caplog): - interceptor = _interceptor(error=ValueError("bad page")) - - with caplog.at_level(logging.INFO, logger=LOGGER): - with pytest.raises(ValueError): - await ActivityEnvironment().run(interceptor.execute_activity, MagicMock()) - - [_, (level, failed)] = _messages(caplog) - assert level == logging.ERROR - assert failed.startswith("activity failed name=unknown") - assert "error=ValueError('bad page')" in failed - - -def test_worker_registers_the_interceptor(): - assert any(isinstance(i, activity_logging.ActivityLogging) for i in worker.INTERCEPTORS) diff --git a/tests/test_observability_wiring.py b/tests/test_observability_wiring.py index 1b29ee7..199815d 100644 --- a/tests/test_observability_wiring.py +++ b/tests/test_observability_wiring.py @@ -56,3 +56,56 @@ async def run(): assert calls == ["configure", "validate", "serving", "reset"] assert any(r.getMessage().startswith("api started auth=disabled") for r in caplog.records) + + +def test_worker_crash_is_logged_with_its_traceback(monkeypatch, calls, caplog): + monkeypatch.setattr(worker, "configure_logs", lambda: None) + monkeypatch.setattr(worker, "reset_logs", lambda: calls.append("reset")) + monkeypatch.setattr(worker, "_run", AsyncMock(side_effect=ConnectionError("temporal down"))) + + with caplog.at_level("INFO"), pytest.raises(ConnectionError): + asyncio.run(worker.main()) + + [crash] = [r for r in caplog.records if r.getMessage() == "worker crashed"] + assert crash.levelname == "ERROR" and isinstance(crash.exc_info[1], ConnectionError) + assert calls == ["reset"] + + +def test_worker_logs_startup_and_hands_its_interceptors_to_temporal(monkeypatch, caplog): + built = {} + + class FakeWorker: + def __init__(self, *args, **kwargs): + built.update(kwargs) + + async def run(self): + return None + + monkeypatch.setenv("TRANSLATION_PROVIDER", "none") + monkeypatch.setattr(worker, "validate_network_provider_config", lambda: None) + monkeypatch.setattr(worker.db, "init_db", lambda: None) + monkeypatch.setattr(worker.Client, "connect", AsyncMock(return_value=MagicMock())) + monkeypatch.setattr(worker, "Worker", FakeWorker) + + with caplog.at_level("INFO"): + asyncio.run(worker._run()) + + assert built["interceptors"] is worker.INTERCEPTORS + assert any(r.getMessage().startswith("worker started queue=ocr-pipeline") for r in caplog.records) + + +def test_api_startup_failure_is_logged_and_flushed(monkeypatch, calls, caplog): + monkeypatch.setattr(api, "configure_logs", lambda: None) + monkeypatch.setattr(api, "reset_logs", lambda: calls.append("reset")) + monkeypatch.setattr(api, "_start_services", AsyncMock(side_effect=RuntimeError("MINIO keys missing"))) + + async def run(): + async with api.lifespan(api.app): + calls.append("serving") + + with caplog.at_level("INFO"), pytest.raises(RuntimeError): + asyncio.run(run()) + + [failed] = [r for r in caplog.records if r.getMessage() == "api startup failed"] + assert failed.levelname == "ERROR" and isinstance(failed.exc_info[1], RuntimeError) + assert calls == ["reset"] diff --git a/tests/test_request_logging.py b/tests/test_request_logging.py index 68bfe42..d594a07 100644 --- a/tests/test_request_logging.py +++ b/tests/test_request_logging.py @@ -1,23 +1,39 @@ import logging import pytest -from fastapi import FastAPI, HTTPException +from fastapi import FastAPI, HTTPException, Request +from fastapi.responses import StreamingResponse from fastapi.testclient import TestClient +from pydantic import BaseModel +from slowapi import Limiter, _rate_limit_exceeded_handler +from slowapi.errors import RateLimitExceeded +from slowapi.util import get_remote_address from pipeline.api import request_logging LOGGER = "pipeline.api.requests" +class Upload(BaseModel): + pages: int + + @pytest.fixture def client(): app = FastAPI() + limiter = Limiter(key_func=get_remote_address) + app.state.limiter = limiter + app.add_exception_handler(RateLimitExceeded, request_logging.recording(_rate_limit_exceeded_handler)) request_logging.install(app) @app.get("/ok") def ok(): return {"secret": "body"} + @app.get("/health") + def health(): + return {"status": "ok"} + @app.get("/missing") def missing(): raise HTTPException(404, "Document not found") @@ -26,9 +42,29 @@ def missing(): def boom(): raise RuntimeError("database locked") - @app.get("/typed/{n}") - def typed(n: int): - return {"n": n} + @app.get("/wrapped") + def wrapped(): + try: + raise ConnectionError("qdrant unreachable") + except ConnectionError as exc: + raise HTTPException(503, "Search backend unavailable") from exc + + @app.post("/upload") + def upload(body: Upload): + return body + + @app.get("/limited") + @limiter.limit("1/minute") + def limited(request: Request): + return {"ok": True} + + @app.get("/stream") + def stream(): + def chunks(): + yield b"first" + raise RuntimeError("minio dropped") + + return StreamingResponse(chunks()) return TestClient(app, raise_server_exceptions=False) @@ -49,7 +85,15 @@ def test_success_logs_method_uri_and_status_without_body(client, caplog): assert "secret" not in message -def test_error_response_logs_status_and_error_detail(client, caplog): +def test_successful_health_checks_stay_at_debug(client, caplog): + with caplog.at_level(logging.DEBUG, logger=LOGGER): + client.get("/health") + + [record] = _lines(caplog) + assert record.levelno == logging.DEBUG + + +def test_client_error_logs_warning_with_detail(client, caplog): with caplog.at_level(logging.INFO, logger=LOGGER): response = client.get("/missing") @@ -60,14 +104,15 @@ def test_error_response_logs_status_and_error_detail(client, caplog): assert "error='Document not found'" in record.getMessage() -def test_validation_error_logs_422_with_its_detail(client, caplog): +def test_validation_error_logs_where_and_why_but_not_the_submitted_value(client, caplog): with caplog.at_level(logging.INFO, logger=LOGGER): - response = client.get("/typed/abc") + response = client.post("/upload", json={"pages": "farmer-typed-value"}) assert response.status_code == 422 [record] = _lines(caplog) assert "status=422" in record.getMessage() - assert "int_parsing" in record.getMessage() + assert "int_parsing" in record.getMessage() and "pages" in record.getMessage() + assert "farmer-typed-value" not in record.getMessage() def test_unknown_route_logs_404(client, caplog): @@ -79,6 +124,28 @@ def test_unknown_route_logs_404(client, caplog): assert "error='Not Found'" in record.getMessage() +def test_rate_limited_request_logs_429_with_its_limit(client, caplog): + client.get("/limited") + with caplog.at_level(logging.INFO, logger=LOGGER): + response = client.get("/limited") + + assert response.status_code == 429 + [record] = _lines(caplog) + assert "status=429" in record.getMessage() + assert "1 per 1 minute" in record.getMessage() + + +def test_server_error_from_a_caught_exception_keeps_its_cause(client, caplog): + with caplog.at_level(logging.INFO, logger=LOGGER): + response = client.get("/wrapped") + + assert response.status_code == 503 + [record] = _lines(caplog) + assert record.levelno == logging.ERROR + assert "error='Search backend unavailable'" in record.getMessage() + assert isinstance(record.exc_info[1], ConnectionError) + + def test_unhandled_exception_logs_500_with_traceback(client, caplog): with caplog.at_level(logging.INFO, logger=LOGGER): response = client.get("/boom") @@ -91,8 +158,21 @@ def test_unhandled_exception_logs_500_with_traceback(client, caplog): assert record.exc_info is not None +def test_failure_mid_stream_is_logged(client, caplog): + with caplog.at_level(logging.INFO, logger=LOGGER): + try: + client.get("/stream") + except Exception: + pass + + [record] = _lines(caplog) + assert record.levelno == logging.ERROR + assert "uri=/stream status=200" in record.getMessage() + assert "error=RuntimeError('minio dropped')" in record.getMessage() + + def test_the_app_logs_its_requests(test_client, caplog): with caplog.at_level(logging.INFO, logger=LOGGER): - test_client.get("/health") + test_client.get("/documents/summary") - assert any("uri=/health status=200" in r.getMessage() for r in _lines(caplog)) + assert any("uri=/documents/summary status=" in r.getMessage() for r in _lines(caplog)) diff --git a/tests/test_temporal_logging.py b/tests/test_temporal_logging.py new file mode 100644 index 0000000..f737889 --- /dev/null +++ b/tests/test_temporal_logging.py @@ -0,0 +1,129 @@ +import asyncio +import logging +import uuid +from datetime import timedelta +from unittest.mock import AsyncMock, MagicMock + +import pytest +from temporalio import activity, workflow +from temporalio.exceptions import ApplicationError +from temporalio.testing import ActivityEnvironment, WorkflowEnvironment +from temporalio.worker import Worker + +from pipeline import temporal_logging + +LOGGER = "pipeline.temporal_logging" + + +def _interceptor(next_result=None, error=None): + following = MagicMock() + following.execute_activity = AsyncMock(return_value=next_result, side_effect=error) + return temporal_logging.TemporalLogging().intercept_activity(following) + + +def _messages(caplog, name=LOGGER): + return [(r.levelno, r.getMessage()) for r in caplog.records if r.name == name] + + +@pytest.mark.asyncio +async def test_activity_logs_start_and_completion(caplog): + interceptor = _interceptor(next_result="done") + + with caplog.at_level(logging.INFO, logger=LOGGER): + result = await ActivityEnvironment().run(interceptor.execute_activity, MagicMock()) + + assert result == "done" + [(start_level, start), (end_level, end)] = _messages(caplog) + assert start_level == end_level == logging.INFO + assert start.startswith("activity started name=unknown") + assert "workflow_id=test" in start and "attempt=1" in start + assert end.startswith("activity completed name=unknown") and "duration_ms=" in end + + +@pytest.mark.asyncio +async def test_activity_failure_is_logged_and_reraised(caplog): + interceptor = _interceptor(error=ValueError("bad page")) + + with caplog.at_level(logging.INFO, logger=LOGGER): + with pytest.raises(ValueError): + await ActivityEnvironment().run(interceptor.execute_activity, MagicMock()) + + [_, (level, failed)] = _messages(caplog) + assert level == logging.WARNING + assert failed.startswith("activity failed name=unknown") + assert "error=ValueError('bad page')" in failed + + +@pytest.mark.asyncio +async def test_activity_cancellation_logs_an_end(caplog): + interceptor = _interceptor(error=asyncio.CancelledError()) + + with caplog.at_level(logging.INFO, logger=LOGGER): + with pytest.raises(asyncio.CancelledError): + await ActivityEnvironment().run(interceptor.execute_activity, MagicMock()) + + [_, (level, cancelled)] = _messages(caplog) + assert level == logging.WARNING + assert cancelled.startswith("activity cancelled name=unknown") + + +@activity.defn +async def _echo(value: str) -> str: + return value + + +@workflow.defn(sandboxed=False) +class _Succeeds: + @workflow.run + async def run(self) -> str: + return await workflow.execute_activity(_echo, "ok", start_to_close_timeout=timedelta(seconds=10)) + + +@workflow.defn +class _Fails: + @workflow.run + async def run(self) -> None: + raise ApplicationError("chunking rejected", non_retryable=True) + + +async def _run_workflow(workflow_cls): + async with await WorkflowEnvironment.start_time_skipping() as env: + async with Worker( + env.client, + task_queue="logging-test", + workflows=[workflow_cls], + activities=[_echo], + interceptors=[temporal_logging.TemporalLogging()], + ): + return await env.client.execute_workflow( + workflow_cls.run, id=f"wf-{uuid.uuid4()}", task_queue="logging-test" + ) + + +@pytest.mark.asyncio +@pytest.mark.workflow +async def test_workflow_and_its_activities_log_start_and_completion(caplog): + with caplog.at_level(logging.INFO): + assert await _run_workflow(_Succeeds) == "ok" + + lines = [r.getMessage() for r in caplog.records] + for prefix in ( + "workflow started name=_Succeeds", + "activity started name=_echo", + "activity completed name=_echo", + "workflow completed name=_Succeeds", + ): + assert any(line.startswith(prefix) for line in lines), prefix + + +@pytest.mark.asyncio +@pytest.mark.workflow +async def test_workflow_failure_is_logged_with_its_error(caplog): + with caplog.at_level(logging.INFO): + with pytest.raises(Exception): + await _run_workflow(_Fails) + + [failed] = [r for r in caplog.records if r.getMessage().startswith("workflow failed name=_Fails")] + assert failed.levelno == logging.ERROR + assert "chunking rejected" in failed.getMessage() + assert failed.exc_info is not None From 19eaf793d91bec4d930d47cd38297c0b9164feb8 Mon Sep 17 00:00:00 2001 From: Hariharan Thiagarajan Date: Thu, 5 Nov 2026 17:31:41 +0530 Subject: [PATCH 7/8] refactor: keep the interceptor in pipeline/activity_logging.py [#241] Co-Authored-By: Claude Opus 5.5 --- docs/ADR/0007-ship-logs-over-otlp.md | 2 +- pipeline/{temporal_logging.py => activity_logging.py} | 0 pipeline/worker.py | 2 +- ...{test_temporal_logging.py => test_activity_logging.py} | 8 ++++---- 4 files changed, 6 insertions(+), 6 deletions(-) rename pipeline/{temporal_logging.py => activity_logging.py} (100%) rename tests/{test_temporal_logging.py => test_activity_logging.py} (95%) diff --git a/docs/ADR/0007-ship-logs-over-otlp.md b/docs/ADR/0007-ship-logs-over-otlp.md index d787a6c..2e131c3 100644 --- a/docs/ADR/0007-ship-logs-over-otlp.md +++ b/docs/ADR/0007-ship-logs-over-otlp.md @@ -20,7 +20,7 @@ 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/temporal_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. +- 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. diff --git a/pipeline/temporal_logging.py b/pipeline/activity_logging.py similarity index 100% rename from pipeline/temporal_logging.py rename to pipeline/activity_logging.py diff --git a/pipeline/worker.py b/pipeline/worker.py index bdb6ae8..98558a2 100644 --- a/pipeline/worker.py +++ b/pipeline/worker.py @@ -26,9 +26,9 @@ run_ocr_and_store, update_document_state, ) +from .activity_logging import TemporalLogging from .network_group import validate_network_provider_config from .observability import configure_logs, reset_logs -from .temporal_logging import TemporalLogging from .workflows import ( ChunkingOnlyWorkflow, DocumentPipelineWorkflow, diff --git a/tests/test_temporal_logging.py b/tests/test_activity_logging.py similarity index 95% rename from tests/test_temporal_logging.py rename to tests/test_activity_logging.py index f737889..8cc66a5 100644 --- a/tests/test_temporal_logging.py +++ b/tests/test_activity_logging.py @@ -10,15 +10,15 @@ from temporalio.testing import ActivityEnvironment, WorkflowEnvironment from temporalio.worker import Worker -from pipeline import temporal_logging +from pipeline import activity_logging -LOGGER = "pipeline.temporal_logging" +LOGGER = "pipeline.activity_logging" def _interceptor(next_result=None, error=None): following = MagicMock() following.execute_activity = AsyncMock(return_value=next_result, side_effect=error) - return temporal_logging.TemporalLogging().intercept_activity(following) + return activity_logging.TemporalLogging().intercept_activity(following) def _messages(caplog, name=LOGGER): @@ -93,7 +93,7 @@ async def _run_workflow(workflow_cls): task_queue="logging-test", workflows=[workflow_cls], activities=[_echo], - interceptors=[temporal_logging.TemporalLogging()], + interceptors=[activity_logging.TemporalLogging()], ): return await env.client.execute_workflow( workflow_cls.run, id=f"wf-{uuid.uuid4()}", task_queue="logging-test" From 498b7f54dae50cc4fef00886efffaeca404b1a6e Mon Sep 17 00:00:00 2001 From: Hariharan Thiagarajan Date: Thu, 5 Nov 2026 17:35:43 +0530 Subject: [PATCH 8/8] refactor: rename the interceptor back to ActivityLogging [#241] Co-Authored-By: Claude Opus 5.5 --- pipeline/activity_logging.py | 2 +- pipeline/worker.py | 4 ++-- tests/test_activity_logging.py | 4 ++-- 3 files changed, 5 insertions(+), 5 deletions(-) diff --git a/pipeline/activity_logging.py b/pipeline/activity_logging.py index 6966bfa..67ffd9e 100644 --- a/pipeline/activity_logging.py +++ b/pipeline/activity_logging.py @@ -18,7 +18,7 @@ logger = logging.getLogger(__name__) -class TemporalLogging(Interceptor): +class ActivityLogging(Interceptor): def intercept_activity(self, next: ActivityInboundInterceptor) -> ActivityInboundInterceptor: return _LogActivity(next) diff --git a/pipeline/worker.py b/pipeline/worker.py index 98558a2..438e36f 100644 --- a/pipeline/worker.py +++ b/pipeline/worker.py @@ -26,7 +26,7 @@ run_ocr_and_store, update_document_state, ) -from .activity_logging import TemporalLogging +from .activity_logging import ActivityLogging from .network_group import validate_network_provider_config from .observability import configure_logs, reset_logs from .workflows import ( @@ -52,7 +52,7 @@ logging.getLogger("temporalio").setLevel(logging.INFO) TASK_QUEUE = "ocr-pipeline" -INTERCEPTORS = [TemporalLogging()] +INTERCEPTORS = [ActivityLogging()] async def main(): diff --git a/tests/test_activity_logging.py b/tests/test_activity_logging.py index 8cc66a5..ac93276 100644 --- a/tests/test_activity_logging.py +++ b/tests/test_activity_logging.py @@ -18,7 +18,7 @@ def _interceptor(next_result=None, error=None): following = MagicMock() following.execute_activity = AsyncMock(return_value=next_result, side_effect=error) - return activity_logging.TemporalLogging().intercept_activity(following) + return activity_logging.ActivityLogging().intercept_activity(following) def _messages(caplog, name=LOGGER): @@ -93,7 +93,7 @@ async def _run_workflow(workflow_cls): task_queue="logging-test", workflows=[workflow_cls], activities=[_echo], - interceptors=[activity_logging.TemporalLogging()], + interceptors=[activity_logging.ActivityLogging()], ): return await env.client.execute_workflow( workflow_cls.run, id=f"wf-{uuid.uuid4()}", task_queue="logging-test"