feat: log fields as attributes, not message text [#241]

AGENTS.md

@@ -78,7 +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
+- `observability.py` — bridges root-logger `INFO`+ lines to OTLP when `OTEL_EXPORTER_OTLP_ENDPOINT` is set (ClickHouse/HyperDX); no-op otherwise. See ADR 0007. Log fields go in `extra=` and export as flat attributes; the console prints them as `key=value`. Log fields go in `extra=` and export as flat attributes; the console prints them as `key=value`
 - `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

CONVENTIONS.md

@@ -110,11 +110,16 @@ Rebase merge only for new PR's. Each commit lands individually on `main` — ind
 
 ## Logging
 Always include `request_id`. Never log raw PII in newly added code.
+Keep the message short and fixed; put fields in `extra=`, so HyperDX can filter on them.
 ```python
 # Good
-logger.info("request_id=%s moderation_category=%s", request_id, category)
-logger.warning("request_id=%s pretranslation_failed=True falling_back=True", request_id)
+logger.info("search request", extra={"request_id": request_id, "index": index_name})
+
+# Bad — the fields are text inside the message, so nothing can filter on them
+logger.info("search request request_id=%s index=%s", request_id, index_name)
 ```
+Don't use a built-in log record name as a key (`name`, `msg`, `args`, `module`…); logging raises.
+The console still prints each extra as `key=value`, so greps keep working.
 Levels: `DEBUG` internal state · `INFO` turn lifecycle · `WARNING` recoverable failure · `ERROR` unrecoverable failure.
 
 In `pipeline`/`worker` code with no HTTP request in scope, use `workflow_id` as the correlation key instead of `request_id` — it's the identifier that actually threads through Temporal signals, activities, and the audit log for a given document.

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

@@ -18,6 +18,7 @@ Decision:
 
 Consequences:
 - Logs only. Traces and metrics are not emitted.
+- Each line is a short, fixed message. Its fields travel as attributes, so HyperDX can filter on them. A nested value, like Temporal's workflow context, is flattened into dotted keys (`temporal_activity.workflow_id`). Temporal no longer repeats that context inside the message.
 - Every API request logs one line on `pipeline.api.requests`: method, path with query string, status and duration, never a body. A pure ASGI middleware times the whole response, so streamed downloads are timed to their last byte and a failure mid-stream is logged.
 - Status below 400 logs at INFO (successful `/health` at DEBUG, since the UI polls it). 4xx logs at WARNING and 5xx at ERROR, with the error detail (validation errors keep only type, field and message, never the submitted value), or the HTTP reason when there is none. A 5xx raised from a caught exception carries that exception's traceback. Unhandled exceptions log at ERROR with their traceback. uvicorn's own access log doesn't reach the root logger and is not shipped.
 - A Temporal worker interceptor (`pipeline/activity_logging.py`) logs every workflow's and activity's start and its completion, failure or cancellation, so new workflows and activities are covered without extra code. Activity failures log at WARNING without a traceback, because each attempt may be retried and Temporal's own `Completing activity as failed` line already carries it. Workflow failures log at ERROR with the traceback.

pipeline/activities.py

@@ -1365,10 +1365,8 @@ async def promote_document_to_prod_qdrant(
         display_name=doc.get("display_name"),
     )
     activity.logger.info(
-        "Promoting %s records to PROD Qdrant collection %s (kind=%s)",
-        len(records),
-        prod_collection,
-        kind,
+        "promoting records to PROD Qdrant",
+        extra={"record_count": len(records), "collection": prod_collection, "kind": kind},
     )
     store = QdrantVectorStore(client=client)
     result = await asyncio.to_thread(
@@ -1581,7 +1579,7 @@ async def update_document_state(
     """Update document state in SQLite."""
     from . import db
 
-    activity.logger.info(f"Updating state for {workflow_id}: stage={stage}")
+    activity.logger.info("updating document stage", extra={"workflow_id": workflow_id, "stage": stage})
     db.update_document_stage(
         workflow_id=workflow_id,
         stage=stage,

pipeline/activity_logging.py

@@ -17,6 +17,11 @@
 
 logger = logging.getLogger(__name__)
 
+# Temporal appends its context to the message by default. It also sends it as
+# an attribute, which a log search can filter on; the message copy only clutters.
+activity.logger.activity_info_on_message = False
+workflow.logger.workflow_info_on_message = False
+
 
 class ActivityLogging(Interceptor):
     def intercept_activity(self, next: ActivityInboundInterceptor) -> ActivityInboundInterceptor:
@@ -31,55 +36,52 @@ def workflow_interceptor_class(
 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)
+        fields = {
+            "activity_type": info.activity_type,
+            "workflow_id": info.workflow_id,
+            "attempt": info.attempt,
+        }
+        logger.info("activity started", extra=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),
+                "activity cancelled", extra={**fields, "duration_ms": _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,
+                "activity failed",
+                extra={**fields, "duration_ms": _elapsed_ms(started), "error": repr(exc)},
             )
             raise
         logger.info(
-            "activity completed name=%s workflow_id=%s attempt=%s duration_ms=%d",
-            *fields,
-            _elapsed_ms(started),
+            "activity completed", extra={**fields, "duration_ms": _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)
+        fields = {
+            "workflow_type": info.workflow_type,
+            "workflow_id": info.workflow_id,
+            "run_id": info.run_id,
+        }
+        workflow.logger.info("workflow started", extra=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
-            )
+            workflow.logger.warning("workflow cancelled", extra=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,
+                "workflow failed", extra={**fields, "error": repr(exc)}, exc_info=True
             )
             raise
-        workflow.logger.info("workflow completed name=%s workflow_id=%s run_id=%s", *fields)
+        workflow.logger.info("workflow completed", extra=fields)
         return result
 
 

pipeline/api/api.py

@@ -101,7 +101,7 @@
 )
 from ..network_constants import RESULT_REJECTED
 from ..network_group import validate_network_provider_config
-from ..observability import configure_logs, reset_logs
+from ..observability import configure_console, configure_logs, reset_logs
 from ..workflows import (
     ChunkingOnlyWorkflow,
     DocumentPipelineWorkflow,
@@ -117,11 +117,7 @@
 # logging.info() calls in this module (e.g. /search) are silently dropped
 # without configuring the root logger too - same LOG_LEVEL convention as
 # pipeline/worker.py.
-logging.basicConfig(
-    level=LOG_LEVEL,
-    format='%(asctime)s [%(levelname)s] %(name)s: %(message)s',
-    datefmt='%H:%M:%S',
-)
+configure_console(LOG_LEVEL)
 
 TASK_QUEUE = "ocr-pipeline"
 
@@ -162,10 +158,12 @@ async def _start_services():
         )
     else:
         logging.info(
-            "Auth enabled: issuer=%s audience=%s jwks=%s",
-            auth_cfg.keycloak_issuer,
-            auth_cfg.keycloak_audience or "(none)",
-            auth_cfg.keycloak_jwks_url,
+            "auth enabled",
+            extra={
+                "issuer": auth_cfg.keycloak_issuer,
+                "audience": auth_cfg.keycloak_audience or "(none)",
+                "jwks": auth_cfg.keycloak_jwks_url,
+            },
         )
 
     # Initialize SQLite database
@@ -197,14 +195,16 @@ async def _start_services():
     # Ensure bucket exists
     if not minio_client.bucket_exists(MINIO_BUCKET):
         minio_client.make_bucket(MINIO_BUCKET)
-        logging.info("api startup: created MinIO bucket %s", MINIO_BUCKET)
+        logging.info("api startup: created MinIO bucket", extra={"bucket": 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,
+        "api started",
+        extra={
+            "auth": "disabled" if auth_cfg.disabled else "enabled",
+            "temporal": temporal_host,
+            "minio": minio_endpoint,
+            "bucket": MINIO_BUCKET,
+        },
     )
 
 
@@ -578,7 +578,9 @@ 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)
+        logging.exception(
+            "vector chunk delete failed", extra={"doc_id": document_id, "chunk": chunk_num}
+        )
         return {"deleted": False, "error": str(e)}
 
 
@@ -594,7 +596,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)
+        logging.exception("vector document delete failed", extra={"doc_id": document_id})
         return {"deleted": 0, "doc_id": document_id, "error": str(e)}
 
 
@@ -1030,7 +1032,9 @@ 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)
+        logging.exception(
+            "prod vector delete failed", extra={"doc_id": doc_id, "collection": collection}
+        )
         return {"deleted": False, "error": str(exc), "collection": collection}
 
 
@@ -1060,7 +1064,9 @@ 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)
+        logging.exception(
+            "prod vector delete failed", extra={"doc_id": doc_id, "collection": collection}
+        )
         return {"deleted": False, "error": str(exc), "collection": collection}
 
 

pipeline/api/request_logging.py

@@ -63,12 +63,11 @@ async def send_and_watch(message: Message) -> None:
             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,
+                "request failed",
+                extra={
+                    **_request_fields(scope, status.get("code", 500), _elapsed_ms(started)),
+                    "error": repr(exc),
+                },
                 exc_info=True,
             )
             raise
@@ -77,27 +76,33 @@ async def send_and_watch(message: Message) -> None:
 
 
 def _log_response(scope: Scope, code: int, duration_ms: int) -> None:
-    method, uri = scope["method"], _uri(scope)
+    fields = _request_fields(scope, code, duration_ms)
     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)
+        logger.log(level, "request completed", extra=fields)
         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,
+        "request failed",
+        extra={**fields, "error": str(error)},
         exc_info=(type(cause), cause, cause.__traceback__) if code >= 500 and cause else None,
     )
 
 
+def _request_fields(scope: Scope, code: int, duration_ms: int) -> dict[str, Any]:
+    return {
+        "method": scope["method"],
+        "uri": _uri(scope),
+        "path": scope["path"],
+        "status": code,
+        "duration_ms": duration_ms,
+    }
+
+
 def _uri(scope: Scope) -> str:
     query = scope.get("query_string", b"").decode("latin-1")
     return f"{scope['path']}?{query}" if query else scope["path"]

pipeline/api/search.py

@@ -194,10 +194,10 @@ async def run_search(payload: dict, user: RequireSearch):
     """
     from ..vector_store import get_default_index_name, get_vector_store
 
+    caller = user.username or user.email or user.user_id
     logging.info(
-        "search payload user=%s payload=%s",
-        user.username or user.email or user.user_id,
-        json.dumps(payload, ensure_ascii=False, default=str),
+        "search payload",
+        extra={"user": caller, "payload": json.dumps(payload, ensure_ascii=False, default=str)},
     )
 
     settings = db.get_search_settings()
@@ -207,11 +207,8 @@ async def run_search(payload: dict, user: RequireSearch):
         raise HTTPException(400, "query is required")
 
     logging.info(
-        "search request user=%s index=%s query=%r top_k=%s",
-        user.username or user.email or user.user_id,
-        index_name,
-        query,
-        payload.get("top_k"),
+        "search request",
+        extra={"user": caller, "index": index_name, "query": query, "top_k": payload.get("top_k")},
     )
 
     search_mode = (payload.get("search_mode") or settings.get("searchMethod") or "HYBRID").upper()
@@ -290,12 +287,14 @@ async def run_search(payload: dict, user: RequireSearch):
             break
 
     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,
-        index_name,
-        query,
-        len(hits),
-        len(final_hits),
+        "search response",
+        extra={
+            "user": caller,
+            "index": index_name,
+            "query": query,
+            "candidate_count": len(hits),
+            "final_count": len(final_hits),
+        },
     )
 
     return {

pipeline/auth/email_otp.py

@@ -431,8 +431,8 @@ def _rotate_password() -> str:
         return _token_response(payload)
 
     logging.error(
-        "email-otp: password-grant bridge failed for %s: status=%s body=%s",
-        email, status, payload,
+        "email-otp: password-grant bridge failed",
+        extra={"email": email, "status": status, "body": str(payload)},
     )
 
     error = str(payload.get("error") or "")