diff --git a/AGENTS.md b/AGENTS.md index 406cb6e..786b052 100644 --- a/AGENTS.md +++ b/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 diff --git a/CONVENTIONS.md b/CONVENTIONS.md index 3f5c537..a333b3c 100644 --- a/CONVENTIONS.md +++ b/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. diff --git a/docs/ADR/0007-ship-logs-over-otlp.md b/docs/ADR/0007-ship-logs-over-otlp.md index 2e131c3..fd0f0a6 100644 --- a/docs/ADR/0007-ship-logs-over-otlp.md +++ b/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. diff --git a/pipeline/activities.py b/pipeline/activities.py index af119b2..ecbdc76 100644 --- a/pipeline/activities.py +++ b/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, diff --git a/pipeline/activity_logging.py b/pipeline/activity_logging.py index 67ffd9e..a92ed01 100644 --- a/pipeline/activity_logging.py +++ b/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,30 +36,28 @@ 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 @@ -62,24 +65,23 @@ async def execute_activity(self, input: ExecuteActivityInput) -> Any: 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 diff --git a/pipeline/api/api.py b/pipeline/api/api.py index 6b5c5f0..9b8c94c 100644 --- a/pipeline/api/api.py +++ b/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} diff --git a/pipeline/api/request_logging.py b/pipeline/api/request_logging.py index d5bc3ba..b97946d 100644 --- a/pipeline/api/request_logging.py +++ b/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,10 +76,10 @@ 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 {} @@ -88,16 +87,22 @@ def _log_response(scope: Scope, code: int, duration_ms: int) -> None: 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"] diff --git a/pipeline/api/search.py b/pipeline/api/search.py index 70beb7c..3225b86 100644 --- a/pipeline/api/search.py +++ b/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 { diff --git a/pipeline/auth/email_otp.py b/pipeline/auth/email_otp.py index 944abee..b0b3673 100644 --- a/pipeline/auth/email_otp.py +++ b/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 "") diff --git a/pipeline/discovery_publish_service.py b/pipeline/discovery_publish_service.py index 175fdb7..7c27d1b 100644 --- a/pipeline/discovery_publish_service.py +++ b/pipeline/discovery_publish_service.py @@ -67,9 +67,8 @@ def publish( """ if not is_publishable_kind(document_kind): logger.info( - "workflow_id=%s document_kind=%s network_publish_skipped=True", - workflow_id, - document_kind, + "network publish skipped", + extra={"workflow_id": workflow_id, "document_kind": document_kind}, ) return self._skipped() @@ -103,14 +102,18 @@ def publish( }, } url = f"{self.endpoint.rstrip('/')}/publish" + publish_fields = { + "workflow_id": workflow_id, + "catalog_id": catalog["id"], + "transaction_id": transaction_id, + } logger.info( - "workflow_id=%s catalog_id=%s transaction_id=%s message_id=%s " - "network_publish_url=%s", - workflow_id, - catalog["id"], - transaction_id, - envelope["context"]["messageId"], - url, + "network publish sent", + extra={ + **publish_fields, + "message_id": envelope["context"]["messageId"], + "network_publish_url": url, + }, ) logger.debug("Request to %s with \n body %s", url, envelope) try: @@ -121,31 +124,17 @@ def publish( # The activity records no artifact when publish raises, so this is # the only trace of a transport failure. logger.error( - "workflow_id=%s catalog_id=%s network_publish_failed=True transaction_id=%s " - "error=%s response_body=%s", - workflow_id, - catalog["id"], - transaction_id, - exc, - exc.response.text, + "network publish failed", + extra={**publish_fields, "error": str(exc), "response_body": exc.response.text}, ) raise except httpx.HTTPError as exc: - logger.error( - "workflow_id=%s catalog_id=%s network_publish_failed=True transaction_id=%s error=%s", - workflow_id, - catalog["id"], - transaction_id, - exc, - ) + logger.error("network publish failed", extra={**publish_fields, "error": str(exc)}) raise logger.info( - "workflow_id=%s catalog_id=%s network_publish_status=%s transaction_id=%s", - workflow_id, - catalog["id"], - response.status_code, - transaction_id + "network publish answered", + extra={**publish_fields, "network_publish_status": response.status_code}, ) result_status, errors = self._read_result(response, catalog["id"], workflow_id) @@ -188,11 +177,12 @@ def _read_result( raise ValueError("no results") except Exception: logger.warning( - "workflow_id=%s catalog_id=%s network_publish_result=unreadable " - "response_body=%.200s", - workflow_id, - catalog_id, - response.text, + "network publish result unreadable", + extra={ + "workflow_id": workflow_id, + "catalog_id": catalog_id, + "response_body": response.text[:200], + }, ) return None, [] @@ -205,11 +195,13 @@ def _read_result( if result_status != RESULT_ACCEPTED: logger.warning( - "workflow_id=%s catalog_id=%s network_publish_result=%s errors=%s", - workflow_id, - catalog_id, - result_status, - errors, + "network publish not accepted", + extra={ + "workflow_id": workflow_id, + "catalog_id": catalog_id, + "network_publish_result": result_status, + "errors": str(errors), + }, ) return result_status, errors diff --git a/pipeline/master_catalog_pg.py b/pipeline/master_catalog_pg.py index beeccde..417cc38 100644 --- a/pipeline/master_catalog_pg.py +++ b/pipeline/master_catalog_pg.py @@ -256,8 +256,14 @@ def sync_catalog_entry(workflow_id: str, status: str) -> Optional[int]: version = row[0] if row else None logger.info( - "master_catalog_pg: synced code=%s name=%s status=%s workflow_id=%s version=%s", - code, name, status, workflow_id, version, + "master catalog entry synced", + extra={ + "code": code, + "scheme_name": name, + "status": status, + "workflow_id": workflow_id, + "version": version, + }, ) _push_snapshots_to_redis(conn, version, now) return version @@ -298,8 +304,8 @@ def remove_catalog_entry(workflow_id: str) -> Optional[int]: if not deleted: logger.info( - "master_catalog_pg: no catalog row for workflow_id=%s (code=%s), nothing to remove", - workflow_id, code, + "master catalog has no row to remove", + extra={"workflow_id": workflow_id, "code": code}, ) return None @@ -317,8 +323,8 @@ def remove_catalog_entry(workflow_id: str) -> Optional[int]: version = row[0] if row else None logger.info( - "master_catalog_pg: removed code=%s workflow_id=%s version=%s", - code, workflow_id, version, + "master catalog entry removed", + extra={"code": code, "workflow_id": workflow_id, "version": version}, ) _push_snapshots_to_redis(conn, version, now) return version @@ -411,8 +417,13 @@ def _push_snapshots_to_redis(conn, version: Optional[int], updated_at: str) -> N } client.set(f"{_REDIS_KEY_PREFIX}:{tier}:snapshot", json.dumps(payload), ex=ttl) logger.info( - "master_catalog_pg: pushed %s:%s snapshot (%d entries, version=%s)", - _REDIS_KEY_PREFIX, tier, len(rows), version, + "master catalog snapshot pushed", + extra={ + "redis_key": f"{_REDIS_KEY_PREFIX}:{tier}:snapshot", + "tier": tier, + "entry_count": len(rows), + "version": version, + }, ) except Exception as exc: - logger.warning("master_catalog_pg: Redis push failed: %s", exc) + logger.warning("master catalog Redis push failed", extra={"error": str(exc)}) diff --git a/pipeline/observability.py b/pipeline/observability.py index 6a58f67..61b7383 100644 --- a/pipeline/observability.py +++ b/pipeline/observability.py @@ -2,7 +2,7 @@ import logging import os -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Any if TYPE_CHECKING: from opentelemetry.sdk._logs import LoggerProvider @@ -19,12 +19,64 @@ _NOT_EXPORTED = ("opentelemetry",) +# What every record has before `extra=` adds to it. +_STANDARD_KEYS = frozenset(vars(logging.makeLogRecord({}))) | {"message", "asctime"} + +CONSOLE_FORMAT = "%(asctime)s [%(levelname)s] %(name)s: %(message)s" +CONSOLE_DATE_FORMAT = "%H:%M:%S" + class _SkipUnexported(logging.Filter): def filter(self, record: logging.LogRecord) -> bool: return not record.name.startswith(_NOT_EXPORTED) +class _FlattenForExport(logging.Filter): + """Turns each extra into a flat, primitive attribute HyperDX can filter on. + + - A dict, like Temporal's `temporal_activity`, becomes dotted keys. + - Anything else that is not a primitive becomes text, or it is dropped. + """ + + def filter(self, record: logging.LogRecord) -> bool: + for key in _extra_keys(record): + value = vars(record).pop(key) + vars(record).update(_flatten(key, value)) + return True + + +class KeyValueFormatter(logging.Formatter): + """The base line, then each extra as `key=value`, so the console stays greppable.""" + + def format(self, record: logging.LogRecord) -> str: + fields = [f"{key}={vars(record)[key]}" for key in _extra_keys(record)] + return " ".join([super().format(record), *fields]) + + +def configure_console(level: int) -> None: + """Like `logging.basicConfig`, and a no-op too when root already has handlers.""" + + handler = logging.StreamHandler() + handler.setFormatter(KeyValueFormatter(CONSOLE_FORMAT, datefmt=CONSOLE_DATE_FORMAT)) + logging.basicConfig(level=level, handlers=[handler]) + + +def _extra_keys(record: logging.LogRecord) -> list[str]: + return [key for key in vars(record) if key not in _STANDARD_KEYS] + + +def _flatten(key: str, value: Any) -> dict[str, str | int | float | bool]: + if isinstance(value, dict): + return { + flat_key: flat_value + for child_key, child_value in value.items() + for flat_key, flat_value in _flatten(f"{key}.{child_key}", child_value).items() + } + if isinstance(value, (str, int, float, bool)): + return {key: value} + return {key: str(value)} + + def is_otel_handler(handler: logging.Handler) -> bool: return getattr(handler, _MARKER, False) @@ -65,6 +117,7 @@ def configure_logs(*, logger_provider: LoggerProvider | None = None) -> None: handler = LoggingHandler(level=logging.INFO, logger_provider=logger_provider) handler.addFilter(_SkipUnexported()) + handler.addFilter(_FlattenForExport()) setattr(handler, _MARKER, True) logging.getLogger().addHandler(handler) diff --git a/pipeline/worker.py b/pipeline/worker.py index 438e36f..8ee76ee 100644 --- a/pipeline/worker.py +++ b/pipeline/worker.py @@ -28,7 +28,7 @@ ) from .activity_logging import ActivityLogging 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, @@ -40,12 +40,7 @@ LOG_LEVEL = getattr(logging, os.environ.get("LOG_LEVEL", "INFO").strip().upper(), logging.INFO) -# Configure verbose logging -logging.basicConfig( - level=LOG_LEVEL, - format='%(asctime)s [%(levelname)s] %(name)s: %(message)s', - datefmt='%H:%M:%S' -) +configure_console(LOG_LEVEL) # Set Temporal SDK logging to INFO regardless of LOG_LEVEL - its own DEBUG # output is noisy and rarely what LOG_LEVEL=DEBUG is set to see. @@ -119,10 +114,12 @@ async def _run(): ) logging.info( - "worker started queue=%s temporal=%s activity_concurrency=%s", - TASK_QUEUE, - temporal_host, - max_concurrent_activities, + "worker started", + extra={ + "queue": TASK_QUEUE, + "temporal": temporal_host, + "activity_concurrency": max_concurrent_activities, + }, ) await worker.run() diff --git a/tests/test_activity_logging.py b/tests/test_activity_logging.py index ac93276..31d892c 100644 --- a/tests/test_activity_logging.py +++ b/tests/test_activity_logging.py @@ -21,8 +21,8 @@ def _interceptor(next_result=None, error=None): return activity_logging.ActivityLogging().intercept_activity(following) -def _messages(caplog, name=LOGGER): - return [(r.levelno, r.getMessage()) for r in caplog.records if r.name == name] +def _records(caplog, name=LOGGER): + return [r for r in caplog.records if r.name == name] @pytest.mark.asyncio @@ -33,11 +33,12 @@ async def test_activity_logs_start_and_completion(caplog): 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 + [start, end] = _records(caplog) + assert start.levelno == end.levelno == logging.INFO + assert start.getMessage() == "activity started" + assert (start.activity_type, start.workflow_id, start.attempt) == ("unknown", "test", 1) + assert end.getMessage() == "activity completed" + assert isinstance(end.duration_ms, int) @pytest.mark.asyncio @@ -48,10 +49,10 @@ async def test_activity_failure_is_logged_and_reraised(caplog): 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 + [_, failed] = _records(caplog) + assert failed.levelno == logging.WARNING + assert failed.getMessage() == "activity failed" + assert failed.error == "ValueError('bad page')" @pytest.mark.asyncio @@ -62,9 +63,9 @@ async def test_activity_cancellation_logs_an_end(caplog): 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") + [_, cancelled] = _records(caplog) + assert cancelled.levelno == logging.WARNING + assert cancelled.getMessage() == "activity cancelled" @activity.defn @@ -86,6 +87,13 @@ async def run(self) -> None: raise ApplicationError("chunking rejected", non_retryable=True) +@workflow.defn +class _WaitsForever: + @workflow.run + async def run(self) -> None: + await workflow.wait_condition(lambda: False) + + async def _run_workflow(workflow_cls): async with await WorkflowEnvironment.start_time_skipping() as env: async with Worker( @@ -106,14 +114,17 @@ 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", + logged = { + (r.getMessage(), getattr(r, "workflow_type", None) or getattr(r, "activity_type", None)) + for r in caplog.records + } + for expected in ( + ("workflow started", "_Succeeds"), + ("activity started", "_echo"), + ("activity completed", "_echo"), + ("workflow completed", "_Succeeds"), ): - assert any(line.startswith(prefix) for line in lines), prefix + assert expected in logged, expected @pytest.mark.asyncio @@ -123,7 +134,37 @@ async def test_workflow_failure_is_logged_with_its_error(caplog): with pytest.raises(Exception): await _run_workflow(_Fails) - [failed] = [r for r in caplog.records if r.getMessage().startswith("workflow failed name=_Fails")] + [failed] = [r for r in caplog.records if r.getMessage().startswith("workflow failed")] assert failed.levelno == logging.ERROR - assert "chunking rejected" in failed.getMessage() + assert failed.workflow_type == "_Fails" + assert "chunking rejected" in failed.error assert failed.exc_info is not None + + +@pytest.mark.asyncio +@pytest.mark.workflow +async def test_workflow_cancellation_logs_an_end(caplog): + async with await WorkflowEnvironment.start_time_skipping() as env: + async with Worker( + env.client, + task_queue="logging-test", + workflows=[_WaitsForever], + interceptors=[activity_logging.ActivityLogging()], + ): + with caplog.at_level(logging.INFO): + handle = await env.client.start_workflow( + _WaitsForever.run, id=f"wf-{uuid.uuid4()}", task_queue="logging-test" + ) + await handle.cancel() + with pytest.raises(Exception): + await handle.result() + + [cancelled] = [r for r in caplog.records if r.getMessage().startswith("workflow cancelled")] + assert cancelled.levelno == logging.WARNING + assert cancelled.workflow_type == "_WaitsForever" + + +def test_temporal_context_stays_out_of_the_message(): + # It still arrives as an attribute; in the message it only clutters the body. + assert activity.logger.activity_info_on_message is False + assert workflow.logger.workflow_info_on_message is False diff --git a/tests/test_discovery_publish_service.py b/tests/test_discovery_publish_service.py index 3539abf..94fcb5b 100644 --- a/tests/test_discovery_publish_service.py +++ b/tests/test_discovery_publish_service.py @@ -387,15 +387,12 @@ def test_logs_the_request_and_the_response_status(self, monkeypatch, caplog): transaction_id="txn-log", document_kind="advisory", workflow_id="wf-log" ) - messages = [r.getMessage() for r in caplog.records] - assert any( - "network_publish_url=https://discovery.example.com/publish" in m - and "workflow_id=wf-log" in m - and "transaction_id=txn-log" in m - and "catalog_id=cat-oan-knowledge-provider-advisories" in m - for m in messages - ) - assert any("network_publish_status=200" in m for m in messages) + [sent] = [r for r in caplog.records if r.getMessage() == "network publish sent"] + assert sent.network_publish_url == "https://discovery.example.com/publish" + assert (sent.workflow_id, sent.transaction_id) == ("wf-log", "txn-log") + assert sent.catalog_id == "cat-oan-knowledge-provider-advisories" + [answered] = [r for r in caplog.records if r.getMessage() == "network publish answered"] + assert answered.network_publish_status == 200 @pytest.mark.unit def test_logs_an_error_when_the_call_fails(self, monkeypatch, caplog): @@ -413,13 +410,9 @@ def test_logs_an_error_when_the_call_fails(self, monkeypatch, caplog): transaction_id="txn-log", document_kind="advisory", workflow_id="wf-log" ) - errors = [r.getMessage() for r in caplog.records if r.levelno == logging.ERROR] - assert any( - "network_publish_failed=True" in m - and "workflow_id=wf-log" in m - and "transaction_id=txn-log" in m - for m in errors - ) + [failed] = [r for r in caplog.records if r.levelno == logging.ERROR] + assert failed.getMessage() == "network publish failed" + assert (failed.workflow_id, failed.transaction_id) == ("wf-log", "txn-log") @pytest.mark.unit def test_skipped_publish_logs_no_request(self, monkeypatch, caplog): @@ -434,8 +427,8 @@ def test_skipped_publish_logs_no_request(self, monkeypatch, caplog): ) messages = [r.getMessage() for r in caplog.records] - assert any("network_publish_skipped=True" in m for m in messages) - assert not any("network_publish_url=" in m for m in messages) + assert "network publish skipped" in messages + assert "network publish sent" not in messages def _use_group_providers(monkeypatch, tmp_path, content): diff --git a/tests/test_master_catalog_pg.py b/tests/test_master_catalog_pg.py index cc7a747..5b1f55a 100644 --- a/tests/test_master_catalog_pg.py +++ b/tests/test_master_catalog_pg.py @@ -227,6 +227,23 @@ def test_sync_catalog_entry_dev_upserts_and_pushes_redis(catalog_pg): assert fake_redis.ttls["master-catalog:dev:snapshot"] == 172800 +def test_a_failed_redis_push_is_logged_and_does_not_fail_the_sync(catalog_pg, monkeypatch, caplog): + # Postgres is the source of truth; the warning is the only sign the mirror is stale. + db, pg, store, fake_redis = catalog_pg + _make_scheme_doc(db, workflow_id="wf-1", scheme_code="pm-ddky") + + def refuse(*args, **kwargs): + raise ConnectionError("redis down") + + monkeypatch.setattr(fake_redis, "set", refuse) + with caplog.at_level("WARNING", logger=pg.logger.name): + version = pg.sync_catalog_entry("wf-1", "dev") + + assert version == 1 + [failed] = [r for r in caplog.records if r.getMessage() == "master catalog Redis push failed"] + assert failed.error == "redis down" + + def test_sync_catalog_entry_live_appears_in_both_tiers(catalog_pg): db, pg, store, fake_redis = catalog_pg _make_scheme_doc(db, workflow_id="wf-2", scheme_code="mif", tool_routing="legacy") diff --git a/tests/test_observability.py b/tests/test_observability.py index d63d951..7d2769f 100644 --- a/tests/test_observability.py +++ b/tests/test_observability.py @@ -55,6 +55,40 @@ def test_info_lines_from_root_and_module_loggers_are_exported(exported): assert _bodies(exported) == ["search request query='cotton pest'", "ocr retry"] +def test_extras_are_exported_as_attributes(exported): + # Attributes, not body text, are what HyperDX can filter on. + logging.info("search request", extra={"query": "cotton pest", "top_k": 5}) + + [item] = exported.get_finished_logs() + assert item.log_record.body == "search request" + assert item.log_record.attributes["query"] == "cotton pest" + assert item.log_record.attributes["top_k"] == 5 + + +def test_a_dict_extra_is_flattened_into_dotted_attributes(exported): + # Temporal sends its context as one dict; flat keys let a search filter on workflow_id. + logging.info("ocr done", extra={"temporal_activity": {"workflow_id": "wf-1", "attempt": 2}}) + + [item] = exported.get_finished_logs() + assert item.log_record.attributes["temporal_activity.workflow_id"] == "wf-1" + assert item.log_record.attributes["temporal_activity.attempt"] == 2 + assert "temporal_activity" not in item.log_record.attributes + + +def test_a_non_primitive_extra_is_exported_as_text(exported): + logging.info("publish rejected", extra={"errors": ["bad catalog"]}) + + [item] = exported.get_finished_logs() + assert item.log_record.attributes["errors"] == "['bad catalog']" + + +def test_console_prints_extras_as_key_value(): + formatter = observability.KeyValueFormatter("%(message)s") + record = logging.makeLogRecord({"msg": "worker started", "queue": "ocr-pipeline"}) + + assert formatter.format(record) == "worker started queue=ocr-pipeline" + + def test_exported_line_keeps_its_level(exported): logging.getLogger("pipeline.activities").error("ingest failed") diff --git a/tests/test_observability_wiring.py b/tests/test_observability_wiring.py index 199815d..49cc54d 100644 --- a/tests/test_observability_wiring.py +++ b/tests/test_observability_wiring.py @@ -55,7 +55,8 @@ async def run(): asyncio.run(run()) assert calls == ["configure", "validate", "serving", "reset"] - assert any(r.getMessage().startswith("api started auth=disabled") for r in caplog.records) + [started] = [r for r in caplog.records if r.getMessage() == "api started"] + assert started.auth == "disabled" def test_worker_crash_is_logged_with_its_traceback(monkeypatch, calls, caplog): @@ -91,7 +92,8 @@ async def run(self): 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) + [started] = [r for r in caplog.records if r.getMessage() == "worker started"] + assert started.queue == "ocr-pipeline" def test_api_startup_failure_is_logged_and_flushed(monkeypatch, calls, caplog): @@ -109,3 +111,4 @@ async def 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 d594a07..8b09aac 100644 --- a/tests/test_request_logging.py +++ b/tests/test_request_logging.py @@ -78,11 +78,11 @@ def test_success_logs_method_uri_and_status_without_body(client, caplog): 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 + assert (record.method, record.uri, record.status) == ("GET", "/ok?index=documents", 200) + assert record.path == "/ok" + assert isinstance(record.duration_ms, int) + assert "secret" not in str(vars(record)) def test_successful_health_checks_stay_at_debug(client, caplog): @@ -100,8 +100,8 @@ def test_client_error_logs_warning_with_detail(client, caplog): 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() + assert (record.uri, record.status) == ("/missing", 404) + assert record.error == "Document not found" def test_validation_error_logs_where_and_why_but_not_the_submitted_value(client, caplog): @@ -110,9 +110,9 @@ def test_validation_error_logs_where_and_why_but_not_the_submitted_value(client, assert response.status_code == 422 [record] = _lines(caplog) - assert "status=422" in record.getMessage() - assert "int_parsing" in record.getMessage() and "pages" in record.getMessage() - assert "farmer-typed-value" not in record.getMessage() + assert record.status == 422 + assert "int_parsing" in record.error and "pages" in record.error + assert "farmer-typed-value" not in str(vars(record)) def test_unknown_route_logs_404(client, caplog): @@ -120,8 +120,8 @@ def test_unknown_route_logs_404(client, caplog): client.get("/nowhere") [record] = _lines(caplog) - assert "uri=/nowhere status=404" in record.getMessage() - assert "error='Not Found'" in record.getMessage() + assert (record.uri, record.status) == ("/nowhere", 404) + assert record.error == "Not Found" def test_rate_limited_request_logs_429_with_its_limit(client, caplog): @@ -131,8 +131,8 @@ def test_rate_limited_request_logs_429_with_its_limit(client, caplog): assert response.status_code == 429 [record] = _lines(caplog) - assert "status=429" in record.getMessage() - assert "1 per 1 minute" in record.getMessage() + assert record.status == 429 + assert "1 per 1 minute" in record.error def test_server_error_from_a_caught_exception_keeps_its_cause(client, caplog): @@ -142,7 +142,7 @@ def test_server_error_from_a_caught_exception_keeps_its_cause(client, caplog): assert response.status_code == 503 [record] = _lines(caplog) assert record.levelno == logging.ERROR - assert "error='Search backend unavailable'" in record.getMessage() + assert record.error == "Search backend unavailable" assert isinstance(record.exc_info[1], ConnectionError) @@ -153,8 +153,8 @@ def test_unhandled_exception_logs_500_with_traceback(client, caplog): 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.uri, record.status) == ("/boom", 500) + assert record.error == "RuntimeError('database locked')" assert record.exc_info is not None @@ -167,12 +167,12 @@ def test_failure_mid_stream_is_logged(client, caplog): [record] = _lines(caplog) assert record.levelno == logging.ERROR - assert "uri=/stream status=200" in record.getMessage() - assert "error=RuntimeError('minio dropped')" in record.getMessage() + assert (record.uri, record.status) == ("/stream", 200) + assert record.error == "RuntimeError('minio dropped')" def test_the_app_logs_its_requests(test_client, caplog): with caplog.at_level(logging.INFO, logger=LOGGER): test_client.get("/documents/summary") - assert any("uri=/documents/summary status=" in r.getMessage() for r in _lines(caplog)) + assert any(r.uri == "/documents/summary" for r in _lines(caplog))