diff --git a/.coverage b/.coverage deleted file mode 100644 index 89085e7..0000000 Binary files a/.coverage and /dev/null differ diff --git a/.github/copilot-instructions.md b/.github/copilot-instructions.md index 3f9ac7b..c91f1c4 100644 --- a/.github/copilot-instructions.md +++ b/.github/copilot-instructions.md @@ -18,7 +18,9 @@ Python style - Use type aliases for complex types - Use built-in generics for type hints - Use `logging.getLogger(__name__)`, not print -- Lazy % formatting in logging: `logger.info("msg %s", var)` +- Structured logging: constant message + data in `extra`: `logger.info("Message accepted.", extra={"writer": name})` +- Do not re-log `topic`, `user`, `resource`, `http_method` or the Lambda context; they are bound per request in `src/utils/observability.py` +- `logger.exception()` only inside an `except` block; outside it pass `exc_info=exc` to `logger.error()` - F-strings in exceptions: `raise ValueError(f"Error {var}")` - All imports at top of file (never inside functions) - Apache 2.0 license header in every .py file (including `__init__.py`) @@ -27,6 +29,8 @@ Python style - Use single backticks in docstrings (`value`), never double backticks (`` ``value`` ``) - Do not use `# -----------` separator comments to divide sections - End all log messages with a period: `logger.info("Message.")` +- Every non-2xx response must produce exactly one log line explaining the cause +- Levels: TRACE payloads, DEBUG steps, INFO request outcomes, WARNING rejected requests, ERROR actionable failures Patterns - `__init__` methods must not raise exceptions; defer validation and connection to first use (lazy init) diff --git a/DEVELOPER.md b/DEVELOPER.md index 312c885..69f9241 100644 --- a/DEVELOPER.md +++ b/DEVELOPER.md @@ -214,6 +214,49 @@ View container logs in pytest output by increasing log level: pytest tests/integration/ -v --log-cli-level=DEBUG ``` +## Logging Conventions + +Logging is structured. `src/utils/observability.py` attaches the AWS Lambda Powertools JSON handler to the root logger, so modules keep using the standard library logger and inherit the format, the Lambda execution context and the request correlation id. + +```python +import logging + +logger = logging.getLogger(__name__) +``` + +Rules: + +- The message is a constant sentence ending with a period. Variable data goes into `extra`, never into the sentence. + ```python + logger.warning("Request rejected: unknown topic.", extra={"known_topics": sorted(known_topics)}) + ``` +- Do not add `topic`, `user`, `resource`, `http_method`, `correlation_id` or the Lambda context to `extra`; they are bound once per request by `bind_request_context()` and `append_request_context()`. +- Every non-2xx response must produce exactly one log line explaining the cause. +- Every request produces exactly one `INFO` line: `Request completed.`, emitted by `dispatch_request()`. Handlers do not emit their own `INFO` outcome lines; they attach outcome fields with `append_request_context()` (e.g. `writers_ok`, `message_key`, `row_count`) so the completion line carries them. +- Every failed request produces exactly one `ERROR` record (the aggregated dispatch failure); per-writer failure detail is logged at `WARNING`. This keeps `level = "ERROR"` metric filters counting failures, not log lines. The per-writer `WARNING` carries `exc_info=True`, because the aggregated `ERROR` is emitted outside the `except` block and can no longer reach the traceback. +- Levels: `TRACE` payloads, `DEBUG` steps, `INFO` request outcomes, `WARNING` rejected requests and soft failures, `ERROR` failures that need action. See the level table in [README](./README.md#logging--correlation). +- Never log tokens, passwords or full message payloads outside `TRACE`. `TRACE` payload logging goes through `log_payload_at_trace()`, which redacts and size caps the payload. +- `logger.exception()` is only valid inside an `except` block. Outside one, pass the captured exception: `logger.error("...", exc_info=exc)`. +- Durations are logged as milliseconds with an explicit key (`duration_ms`, `writer_duration_ms`, `query_duration_ms`). + +Assert on structured fields in tests, not on formatted strings: + +```python +def test_rejects_unknown_topic(caplog): + caplog.set_level(logging.WARNING) + ... + assert "Request rejected: unknown topic." == caplog.records[-1].message +``` + +Keys bound with `append_request_context()` live on the Powertools formatter rather than on the `LogRecord`. To assert on them, render the record with the Powertools logger (`registered_formatter` exists only there, not on a `logging.getLogger()` instance): + +```python +from src.utils.observability import logger as powertools_logger + +payload = json.loads(powertools_logger.registered_formatter.format(caplog.records[-1])) +assert "run-42" == payload["correlation_id"] +``` + ## Run All Quality Gates Run Black, Pylint, mypy, unit tests (with coverage), and integration tests in a single command: diff --git a/README.md b/README.md index f8323a9..02bb563 100644 --- a/README.md +++ b/README.md @@ -128,11 +128,52 @@ Supporting configs: - `topic_schemas/*.json` – each file contains a JSON Schema for a topic. In the current code these are explicitly loaded inside `event_gate_lambda.py`. (Future enhancement: auto-discover or index file.) Environment variables: -- `LOG_LEVEL` (optional) – defaults to `INFO`. +- `LOG_LEVEL` (optional) – defaults to `INFO`. Accepts `TRACE`, `DEBUG`, `INFO`, `WARNING`, `ERROR`. An unknown value falls back to `INFO` and is reported with a warning. +- `POWERTOOLS_LOG_LEVEL` (optional) – takes precedence over `LOG_LEVEL`. +- `POWERTOOLS_SERVICE_NAME` (optional) – value of the `service` key in every log line. Defaults to `eventgate`. +- `TRACE_REDACT_KEYS` (optional) – comma separated message keys redacted from `TRACE` payload logs. Defaults to `password,secret,token,key,apikey,api_key`. +- `TRACE_MAX_BYTES` (optional) – maximum size of a logged `TRACE` payload. Defaults to `10000`. - `CONF_DIR` (optional) – directory containing `config.json` and `access.json`. Defaults to `conf`. - `POSTGRES_SECRET_NAME` (optional) – AWS Secrets Manager secret name holding PostgreSQL connection credentials (host, port, database, user, password). Required for Postgres writer and stats reader. - `POSTGRES_SECRET_REGION` (optional) – AWS region of the Secrets Manager secret. Must be set together with `POSTGRES_SECRET_NAME`. +## Logging & Correlation + +Both lambdas emit structured JSON logs through [AWS Lambda Powertools](https://docs.aws.amazon.com/powertools/python/latest/core/logger/). Every line carries the service name, log level, `function_request_id`, `cold_start` and a `correlation_id`. Static execution facts (`function_name`, `function_memory_size`, `function_arn`) are logged once per container on the cold start line `Lambda execution context.` instead of being repeated on every record. + +The correlation id is resolved per request in this order: +1. The `X-Correlation-ID` request header, when it matches `^[A-Za-z0-9._:-]{1,128}$`. +2. The `X-Request-ID` request header, with the same constraint. +3. The API Gateway request id (`requestContext.requestId`). + +Callers that already have a run or job id should send it as `X-Correlation-ID` so their logs and EventGate logs can be joined. The resolved id is returned in the `X-Correlation-ID` response header of every response, including errors. + +Log levels used by the service: + +| Level | Content | +|-----------|-----------------------------------------------------------------------------------------------| +| `TRACE` | Full message payloads, redacted and size capped. Never enable by default. | +| `DEBUG` | Configuration loading, lazy initialization, per-writer send attempts, connection reuse. | +| `INFO` | Exactly one `Request completed.` line per request, carrying the outcome fields (`status_code`, `duration_ms`, and e.g. `writers_ok`, `message_key`, `row_count`); cold start initialization. | +| `WARNING` | Rejected requests (auth, authorization, validation), degraded health, Kafka flush retries, individual writer failures. | +| `ERROR` | Partial fan-out failures (one aggregated line per failed request), failed queries, unhandled request errors. | + +Every non-2xx response has exactly one log line explaining the cause, and every failed request produces exactly one `ERROR` record — per-writer failure detail is reported at `WARNING`. + +Example CloudWatch Logs Insights queries: + +```text +fields @timestamp, level, message, status_code, duration_ms +| filter correlation_id = "" +| sort @timestamp asc +``` + +```text +fields @timestamp, topic, user, message +| filter level = "WARNING" and status_code = 403 +| stats count() by user, topic +``` + ## Local Development & Testing | Purpose | Relative link | diff --git a/adr/000-template.md b/adr/000-template.md new file mode 100644 index 0000000..7d99c09 --- /dev/null +++ b/adr/000-template.md @@ -0,0 +1,27 @@ +# ADR-XYZ: + +## Status + +**** — + +## Context and Problem Statement + +Change me. + +## Decision Outcome + +Change me. + +## Alternatives Considered + +Change me. + +## Related Tickets + +* Ticket 1 +* Ticket 2 + +## References + +* Reference 1 +* Reference 2 \ No newline at end of file diff --git a/adr/002-observability/002-observability.md b/adr/002-observability/002-observability.md new file mode 100644 index 0000000..9aff976 --- /dev/null +++ b/adr/002-observability/002-observability.md @@ -0,0 +1,154 @@ +# ADR-002: Observability — structured logging and request correlation + +## Status + +**ACCEPTED** — August 2026 + +## Context and Problem Statement + +EventGate ran with 79 log statements: 52 at `DEBUG`, none at `INFO`. Production runs with `LOG_LEVEL=INFO`, so service emitted effectively nothing per request: no request line, outcome, or rejection reason. Raising level to `DEBUG` produced unusable output because root logger also enabled `botocore`, `urllib3`, and `s3transfer` noise. + +Operational failures: + +- Client `401`, `403`, or `400` responses produced no log line. Support could not answer why request was rejected. +- Partial fan-out failure—Kafka accepted, Postgres failed—returned `500` without recording sink state. Reconciliation required manual work. +- No invocation-wide or caller-to-service correlation. +- No durations; slow writer could not be identified. + +## Decision Outcome + +Adopt [AWS Lambda Powertools for Python](https://docs.aws.amazon.com/powertools/python/latest/core/logger/) as logging backend for both EventGate Lambdas. Emit JSON logs enriched with Lambda execution context and request correlation ID. Return correlation ID to caller. Do not adopt AWS X-Ray tracing now. + +### Integration approach + +Create Powertools `Logger` once in `src/utils/observability.py`. Attach its handler to root logger through `configure_root_logging()`. Modules continue using `logging.getLogger(__name__)`. + +This keeps existing call sites unchanged and sends JSON output through shared root handler. Persistent keys, including `correlation_id`, live on shared `LambdaPowertoolsFormatter`; appending once per request adds key to every module log line without threading logger through call stack. + +Do not use `copy_config_to_registered_loggers()`: it sets `propagate = False` on module loggers and breaks unit-test `caplog`. Root attachment avoids this. + +Do not use `inject_lambda_context` decorator: it raises `AttributeError` when Lambda context is `None`, as in unit and integration tests invoking `lambda_handler`. `bind_request_context()` supplies equivalent binding and tolerates missing context. The one thing the decorator does that binding must reproduce is the per-invocation sampling re-draw; `refresh_sampled_log_level()` covers it. + +### Logging strategy + +Guiding rule: **log volume tracks trouble, not traffic.** A request that works costs a fixed, small number of lines no matter how much work it did. A request that fails spends as many lines as needed to explain itself. Steady-state cost stays flat as traffic grows, and detail is bought on demand instead of paid for continuously. + +Three rules implement this on the happy path. + +**1. One `INFO` line per request.** `dispatch_request()` emits `Request completed.` with `status_code` and `duration_ms` after every invocation, success or failure. Handlers emit no `INFO` outcome line of their own. Count holds regardless of writer count, message size, or result-set size. + +**2. Outcome data rides on that line as fields, not as extra lines.** Handlers attach results through `append_request_context()`—`writers_ok`, `message_key`, `row_count`, `has_more`—so completion line carries them. Recording one more fact about a request must not cost one more line. Prefer new key over new log call. + +**3. Step detail is `DEBUG` and stays off in production.** Per-writer sends, connection reuse, configuration loading, token refresh, schema resolution. Answers "how did this request get here", which matters while investigating and is worthless at steady state. + +Failure path inverts the budget, and splits in two. The classes are disjoint: a request takes the shape of one or the other, never both. + +**Rejections (4xx).** Exactly one rejection `WARNING` names the cause—auth, authorization, validation, unknown route—reason always in fields. No `ERROR`. Support answers "why was my request rejected" from that single line without reproducing it. "Exactly one" counts rejection lines only; operational `WARNING`s—Kafka flush retries, missing key fields, token refresh failure—may legitimately co-occur on the same request. + +**Failures (5xx).** Exactly one `ERROR` per failed request: the record emitted by whichever frame converts the failure into the 5xx response. For fan-out that is the aggregated dispatch failure in `_post_topic_message()`, listing every failing writer in a single record. For a handler that builds its own 500, such as `handler_stats`, it is that handler's own `logger.exception`. For anything escaping to the boundary it is `dispatch_request()`. This is a discipline each handler upholds, not a property the boundary can enforce. + +Per-writer failure detail stays at `WARNING`, carrying `exc_info=True`, because the aggregate is emitted after `_write_to_all()` returns—outside the `except` block—and can no longer reach the traceback. + +The invariant worth alarming on follows from the split: + +> Among records emitted during a single invocation, the count of `ERROR` records equals the count of 5xx responses the service built to report a failed request. + +An alarm on `level = "ERROR"` then counts failed requests, not log lines. Both counts—the one `INFO` line and the `ERROR` total—are scoped to a single invocation, and four things sit outside them by design: + +- **Initialization and import-time errors** belong to no request. A failing cold start is retried by the Lambda runtime, so one broken deployment emits `ERROR`s at container rate, not request rate. `log_uncaught_exceptions=True` can add a second record for the same escaping exception. +- **`/terminate`** leaves through `SystemExit`, which the boundary deliberately does not catch: no completion line, no `ERROR`, and neither is wanted. An exception raised inside `bind_request_context()` escapes the same way, because the binding call runs before the `try`. +- **A degraded `/health` probe** returns `503` and logs one `WARNING`. The `503` is the probe answering correctly, not the request failing, and what deserves an alarm is the dependency it names. +- **Platform 5xx**—timeout, out-of-memory, throttling, API Gateway `502` from a malformed response shape—reach the caller without the service logging anything. + +An `ERROR` alarm therefore detects service-built request failures only. Cold-start breakage, dependency degradation, and platform faults each need their own signal; none is covered here. + +| Level | Content | Expected frequency | +|---|---|---| +| `TRACE` | Full message payloads, redacted and size-capped. | Never enabled by default. | +| `DEBUG` | Per-step detail: writer sends, config loading, connection reuse, token refresh, routing. | Off in production. | +| `INFO` | Request outcome (`Request completed.`); lifecycle events—lambda initialization, cold-start context, Postgres connection established, token public keys loaded. | Once per request, plus lifecycle lines that are tied to container or connection age, not to request count. | +| `WARNING` | Rejected requests—auth, authorization, validation—degraded health, per-writer failures, Kafka flush retries. | Once per rejection, plus once per failed sub-step. | +| `ERROR` | The record converting a failure into a 5xx: aggregated dispatch failure, handler-built 500, unhandled request errors. Plus misconfiguration preventing service, which is outside request scope. | Once per failed request. | + +Cap third-party loggers (`boto3`, `botocore`, `urllib3`, `s3transfer`, `aiosql`, `confluent_kafka`) at `WARNING` so `DEBUG` and `TRACE` remain readable. + +Retain custom numeric `TRACE` level (`5`). Register it with standard `logging` before constructing Powertools `Logger`; pass level numerically so Powertools validates it. Unit test pins behavior: unrecognized level falls back to `INFO` and logs one `WARNING` naming the rejected level, disabling payload logging. + +### Attaching tracebacks + +One rule governs every exception log, and it concerns ownership rather than level: + +> A traceback is logged exactly once, by the frame that converts the exception into a response or swallows it. A frame that wraps and re-raises does not log it above `DEBUG`. + +Nothing is lost by deferring: `raise WriteError(...) from exc` plus `exc_info=True` upstream formats the full `__cause__` chain. + +Given that rule, the call to use follows from where the code sits and which level the budget allows: + +| Situation | Call | +|---|---| +| Inside `except`, owning the failure, logging at `ERROR` | `logger.exception(msg)` | +| Inside `except`, owning the failure, logging at `WARNING` or `DEBUG` | `logger.warning(msg, exc_info=True)`, `logger.debug(msg, exc_info=True)` | +| Outside `except`, exception captured earlier | `logger.error(msg, exc_info=captured_exc)` | +| Error state found by inspection, no exception exists | plain `logger.error(msg, extra={...})`, no `exc_info` | + +`logger.exception()` hardcodes `ERROR`. That is why mixed usage across the codebase is not an inconsistency to normalize away: per-writer failures sit at `WARNING` precisely to keep the `ERROR` count equal to the failed-request count, and `WARNING` is unreachable through `logger.exception()`. + +Row three exists because `logger.exception()` outside an `except` block reads an empty `sys.exc_info()` and logs no traceback at all. It applies wherever a step collects failures across several `try` blocks and reports once at the end. Such a captured exception is legitimately `None`—a step can fail through a callback or a timeout rather than a raised exception—and `exc_info=None` correctly omits a traceback that does not exist, so it must not be "corrected" to `exc_info=True`. + +### Raising detail when something is wrong + +Production runs at `INFO`. When one line per request is not enough: + +- Set `LOG_LEVEL=DEBUG` on the function to replay step detail. Third-party capping keeps output readable, which is what made `DEBUG` unusable before this ADR. +- Set `LOG_LEVEL=TRACE` for redacted, size-capped payloads. Never leave enabled. +- Set `POWERTOOLS_LOGGER_SAMPLE_RATE` to collect `DEBUG` detail for a fraction of production traffic without paying for it on every request. + +Sampling needs two things wired, and `refresh_sampled_log_level()` supplies both from `bind_request_context()`. Powertools draws the lottery when the `Logger` is constructed and re-draws it only from `refresh_sample_rate_calculation()`, which ships inside the rejected `inject_lambda_context` decorator; calling it per invocation is what makes the draw per request instead of per container. The draw then sets the level of the Powertools logger alone, so the outcome is mirrored onto the root logger—otherwise module loggers, which resolve their level from root, drop their `DEBUG` records before the shared handler sees them. Mirroring takes the lower of the two levels, because sampling may only add detail: a run already at `TRACE` is left alone rather than pulled up to `DEBUG`. + +Recommended value: **`0`**, the default, as the standing production setting—one line per request plus the correlation ID answers most questions, and `LOG_LEVEL=DEBUG` covers the rest on demand. Use **`0.05`** while chasing an intermittent failure that cannot be reproduced on demand, which is the case sampling exists for. Cost is predictable: roughly 15 `DEBUG` lines per request on the fan-out `POST` path against a one-line `INFO` baseline puts average volume near `1 + 15r`, so `0.01` costs +15% and `0.05` costs +75%. + +Correlation ID is what makes a quiet default workable: one ID joins every line of an invocation and ties them to caller's own ID, so a request is reconstructed from few lines rather than many. + +### Correlation-ID contract + +Resolve ID for each request, in order: + +1. `X-Correlation-ID` request header. +2. `X-Request-ID` request header. +3. API Gateway request ID (`requestContext.requestId`). + +Accept headers only when matching `^[A-Za-z0-9._:-]{1,128}$`. Reject newlines and control characters to prevent log injection; cap length to prevent every log line from being bloated. + +Return resolved ID in `X-Correlation-ID` response header for every success and error response. Stamp it centrally in `dispatch_request()`, single point through which every response passes. + +### Consequences + +- Logs change from plain text to JSON. No downstream log parser exists; JSON is easier for human reading and CloudWatch Logs Insights queries. +- `dispatch_request()` catches `Exception` at boundary instead of fixed list of six exception types. Escaping `psycopg2.Error` or `KafkaException` now becomes structured error rather than runtime traceback and opaque API Gateway `502`. `SystemExit` still propagates; `/terminate` remains unaffected. +- Malformed request body on `POST /topics/{topic_name}` returns `400 validation`, not `500 internal`. +- Callers receive additive `X-Correlation-ID` response header; response body contract unchanged. +- CloudWatch cost rises, but by a bounded amount: `INFO` emits one line per invocation plus a few initialization lines per container, instead of nothing. Cost scales with request count only—not with message size, writer count, or result-set size—so it stays predictable as traffic grows. +- Adding a fact to a request is free in line count but not in line width. The completion line grows as handlers attach keys, which is the intended trade: one wider record beats several narrow ones for CloudWatch Logs Insights queries. + +### Not in scope + +- **X-Ray tracing.** If enabled later, Powertools `Tracer` plugs into `observability.py`; function needs `Tracing: Active` and `xray:PutTraceSegments`. Kafka spans need manual subsegments. +- **EMF metrics** for writer failures, auth failures, and cold starts. Track separately. +- **Downstream correlation propagation** through Kafka message headers and EventBridge `TraceHeader`. Add when consumer can read it. + +## Alternatives Considered + +1. **Plain standard-library logging with hand-written JSON formatter.** No new dependency, but Lambda context, cold-start detection, correlation-ID plumbing, and log sampling would require local implementation and maintenance. +2. **AWS Lambda Powertools (chosen).** Purpose-built for Lambda. Provides JSON formatter, `cold_start`, `function_request_id`, and correlation-ID handling. It also ships `DEBUG` sampling. Lambdas deploy as container images, so change is one `requirements.txt` entry with no layer ARN per region. Cost: one production dependency and roughly 2–4 MB image size, negligible next to librdkafka build. +3. **Powertools plus X-Ray tracing.** Deferred. Target-account X-Ray usage remains undecided; it adds cost; `confluent-kafka` cannot be patched by X-Ray SDK, so Kafka spans need manual instrumentation. Correlation ID meets immediate need: joining log lines within invocation and across services. + +## Related Tickets + +* [#193](https://github.com/AbsaOSS/EventGate/issues/193) +* [#219](https://github.com/AbsaOSS/EventGate/issues/219) +* [#220](https://github.com/AbsaOSS/EventGate/issues/220) + +## References + +* [AWS Lambda Powertools for Python Logger](https://docs.aws.amazon.com/powertools/python/latest/core/logger/) +* [AWS X-Ray SDK for Python](https://docs.aws.amazon.com/xray/latest/devguide/xray-sdk-python.html) diff --git a/adr/README.md b/adr/README.md new file mode 100644 index 0000000..52fa975 --- /dev/null +++ b/adr/README.md @@ -0,0 +1,10 @@ +# Architectural Decision Records (ADR) + +This module contains a collection of Architectural Decision Records (ADRs) which have been made during the +development of the project. It consists of a collection of markdown files, each of which describes a single +ADR - a single topic or a theme that we decide to review, investigate, discuss, and make decision upon. + +The ADRs are numbered sequentially, and are stored in directories (because we might include diagrams or other resources +along the markdown file) each dedicated to a major part of the project. + +The ADRs here are written for the project-specific practices and decisions. \ No newline at end of file diff --git a/requirements.txt b/requirements.txt index 89a4e29..bcb7426 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,3 +1,4 @@ +aws-lambda-powertools==3.31.1 cryptography==50.0.0 jsonschema==4.26.0 PyJWT==2.13.0 diff --git a/src/event_gate_lambda.py b/src/event_gate_lambda.py index 3988717..748d213 100644 --- a/src/event_gate_lambda.py +++ b/src/event_gate_lambda.py @@ -16,8 +16,9 @@ """AWS Lambda entry point for the EventGate service.""" -import os +import logging import sys +import time from typing import Any import boto3 @@ -27,22 +28,22 @@ from src.handlers.handler_health import HandlerHealth from src.handlers.handler_token import HandlerToken from src.handlers.handler_topic import HandlerTopic -from src.utils.conf_path import CONF_DIR, INVALID_CONF_ENV +from src.utils.conf_path import CONF_DIR, log_configuration_source from src.utils.config_loader import load_config from src.utils.constants import SSL_CA_BUNDLE_KEY -from src.utils.logging_levels import init_root_logger +from src.utils.observability import configured_log_level, init_lambda_logging from src.utils.utils import dispatch_request from src.writers.writer_eventbridge import WriterEventBridge from src.writers.writer_kafka import WriterKafka from src.writers.writer_postgres import WriterPostgres -logger = init_root_logger(__name__) -logger.debug("Initialized logger with level %s.", os.environ.get("LOG_LEVEL", "INFO")) +logger = init_lambda_logging(__name__) + +_INIT_STARTED_AT = time.perf_counter() +logger.info("Initializing EventGate lambda.", extra={"log_level": configured_log_level()}) # Load main configuration -logger.debug("Using CONF_DIR=%s.", CONF_DIR) -if INVALID_CONF_ENV: - logger.warning("CONF_DIR env var set to non-existent path: %s; fell back to %s.", INVALID_CONF_ENV, CONF_DIR) +log_configuration_source(logger) config = load_config(CONF_DIR) logger.debug("Loaded main configuration.") @@ -73,6 +74,15 @@ handler_health = HandlerHealth(writers) handler_api = HandlerApi().with_api_definition_loaded() +logger.info( + "EventGate lambda initialized.", + extra={ + "init_duration_ms": round((time.perf_counter() - _INIT_STARTED_AT) * 1000, 2), + "writers": sorted(writers), + "topics": sorted(handler_topic.topics), + }, +) + # Route to handler function mapping ROUTE_MAP: dict[str, Any] = { "/api": lambda _: handler_api.get_api(), @@ -85,12 +95,12 @@ } -def lambda_handler(event: dict[str, Any], _context: Any = None) -> dict[str, Any]: +def lambda_handler(event: dict[str, Any], context: Any = None) -> dict[str, Any]: """AWS Lambda entry point for EventGate. Dispatches based on API Gateway proxy resource field. Args: event: The event data from API Gateway. - _context: The mandatory context argument for AWS Lambda invocation (unused). + context: The AWS Lambda context object, used to enrich logs with the execution context. Returns: A dictionary compatible with API Gateway Lambda Proxy integration. """ - return dispatch_request(event, ROUTE_MAP, logger) + return dispatch_request(event, ROUTE_MAP, logger, context) diff --git a/src/event_stats_lambda.py b/src/event_stats_lambda.py index f8bc438..aab4e43 100644 --- a/src/event_stats_lambda.py +++ b/src/event_stats_lambda.py @@ -16,24 +16,25 @@ """AWS Lambda entry point for the EventStats service.""" -import os +import logging +import time from typing import Any from src.handlers.handler_health import HandlerHealth from src.handlers.handler_stats import HandlerStats from src.readers.reader_postgres import ReaderPostgres -from src.utils.conf_path import CONF_DIR, INVALID_CONF_ENV +from src.utils.conf_path import CONF_DIR, log_configuration_source from src.utils.config_loader import load_topic_names -from src.utils.logging_levels import init_root_logger +from src.utils.observability import configured_log_level, init_lambda_logging from src.utils.utils import dispatch_request -logger = init_root_logger(__name__) -logger.debug("Initialized EventStats logger with level %s.", os.environ.get("LOG_LEVEL", "INFO")) +logger = init_lambda_logging(__name__) + +_INIT_STARTED_AT = time.perf_counter() +logger.info("Initializing EventStats lambda.", extra={"log_level": configured_log_level()}) # Load main configuration -logger.debug("Using CONF_DIR=%s.", CONF_DIR) -if INVALID_CONF_ENV: - logger.warning("CONF_DIR env var set to non-existent path: %s; fell back to %s.", INVALID_CONF_ENV, CONF_DIR) +log_configuration_source(logger) # Load topic names. topic_names = load_topic_names(CONF_DIR) @@ -46,6 +47,14 @@ handler_stats = HandlerStats(topics, reader_postgres) handler_health = HandlerHealth({"postgres_reader": reader_postgres}) +logger.info( + "EventStats lambda initialized.", + extra={ + "init_duration_ms": round((time.perf_counter() - _INIT_STARTED_AT) * 1000, 2), + "topics": sorted(topics), + }, +) + # Route to handler function mapping ROUTE_MAP: dict[str, Any] = { "/stats/{topic_name}": handler_stats.handle_request, @@ -53,12 +62,12 @@ } -def lambda_handler(event: dict[str, Any], _context: Any = None) -> dict[str, Any]: +def lambda_handler(event: dict[str, Any], context: Any = None) -> dict[str, Any]: """AWS Lambda entry point for EventStats. Dispatches based on API Gateway proxy resource field. Args: event: The event data from API Gateway. - _context: The mandatory context argument for AWS Lambda invocation (unused). + context: The AWS Lambda context object, used to enrich logs with the execution context. Returns: A dictionary compatible with API Gateway Lambda Proxy integration. """ - return dispatch_request(event, ROUTE_MAP, logger) + return dispatch_request(event, ROUTE_MAP, logger, context) diff --git a/src/handlers/handler_api.py b/src/handlers/handler_api.py index beda89a..e37af5a 100644 --- a/src/handlers/handler_api.py +++ b/src/handlers/handler_api.py @@ -46,10 +46,10 @@ def with_api_definition_loaded(self) -> "HandlerApi": if not self.api_spec: raise ValueError("API specification file is empty") - logger.debug("Loaded API definition from %s.", api_path) + logger.debug("Loaded API definition.", extra={"api_path": api_path}) return self except (FileNotFoundError, PermissionError, ValueError) as exc: - logger.exception("Failed to load or read API specification from %s.", api_path) + logger.exception("Failed to load or read the API specification.", extra={"api_path": api_path}) raise RuntimeError("API specification initialization failed") from exc def get_api(self) -> dict[str, Any]: diff --git a/src/handlers/handler_health.py b/src/handlers/handler_health.py index d184ea2..eeab801 100644 --- a/src/handlers/handler_health.py +++ b/src/handlers/handler_health.py @@ -68,14 +68,14 @@ def get_health(self) -> dict[str, Any]: uptime_seconds = int((datetime.now(timezone.utc) - self.start_time).total_seconds()) if not failures: - logger.debug("Health check passed.") + logger.debug("Health check passed.", extra={"dependencies": statuses}) return { "statusCode": 200, "headers": {"Content-Type": "application/json"}, "body": json.dumps({"status": "ok", "uptime_seconds": uptime_seconds, "dependencies": statuses}), } - logger.debug("Health check degraded: %s.", failures) + logger.warning("Health check degraded.", extra={"failures": failures, "dependencies": statuses}) return { "statusCode": 503, "headers": {"Content-Type": "application/json"}, diff --git a/src/handlers/handler_stats.py b/src/handlers/handler_stats.py index 391d02d..b5eb0c5 100644 --- a/src/handlers/handler_stats.py +++ b/src/handlers/handler_stats.py @@ -18,11 +18,13 @@ import json import logging +import time from typing import Any from src.readers.reader_postgres import ReaderPostgres from src.utils.constants import POSTGRES_DEFAULT_LIMIT, SUPPORTED_STATS_TOPICS -from src.utils.utils import build_error_response +from src.utils.observability import append_request_context +from src.utils.utils import build_error_response, resolve_request_topic logger = logging.getLogger(__name__) @@ -38,6 +40,14 @@ def __init__( self.topics = topics self.reader_postgres = reader_postgres + @staticmethod + def _log_field_rejected(field_name: str) -> None: + """Log a rejected request body field. + Args: + field_name: Name of the field that failed validation. + """ + logger.warning("Request rejected: request body field is invalid.", extra={"field": field_name}) + def handle_request(self, event: dict[str, Any]) -> dict[str, Any]: """Handle POST /stats/{topic_name} requests. Args: @@ -45,27 +55,35 @@ def handle_request(self, event: dict[str, Any]) -> dict[str, Any]: Returns: API Gateway response dict. """ - path_params = event.get("pathParameters") or {} - topic_name = path_params.get("topic_name", "") - if not topic_name: - return build_error_response(400, "validation", "Missing path parameter 'topic_name'.") - topic_name = topic_name.lower() + topic_name, topic_error = resolve_request_topic(event) + if topic_error is not None: + return topic_error + + logger.debug("Handling POST topic stats.") if topic_name not in self.topics: + logger.warning("Request rejected: unknown topic.", extra={"known_topics": sorted(self.topics)}) return build_error_response(404, "topic", f"Topic '{topic_name}' not found.") if topic_name not in SUPPORTED_STATS_TOPICS: + logger.warning( + "Request rejected: stats are not supported for the topic.", + extra={"supported_topics": sorted(SUPPORTED_STATS_TOPICS)}, + ) return build_error_response( 400, "validation", f"Stats are only supported for topics '{', '.join(SUPPORTED_STATS_TOPICS)}'." ) - # Parse request body + # Parse request body. Every field is optional and defaulted, so an absent body is a valid + # "no filters" query - unlike `/topics`, where the body carries the message itself. try: body = json.loads(event.get("body") or "{}") except (json.JSONDecodeError, TypeError): + logger.warning("Request rejected: request body is not valid JSON.") return build_error_response(400, "validation", "Request body must be valid JSON.") if not isinstance(body, dict): + logger.warning("Request rejected: request body is not a JSON object.") return build_error_response(400, "validation", "Request body must be a JSON object.") timestamp_start = body.get("timestamp_start") @@ -74,15 +92,20 @@ def handle_request(self, event: dict[str, Any]) -> dict[str, Any]: limit: int = body.get("limit", POSTGRES_DEFAULT_LIMIT) if timestamp_start is not None and (isinstance(timestamp_start, bool) or not isinstance(timestamp_start, int)): + self._log_field_rejected("timestamp_start") return build_error_response(400, "validation", "Field 'timestamp_start' must be an integer (epoch ms).") if timestamp_end is not None and (isinstance(timestamp_end, bool) or not isinstance(timestamp_end, int)): + self._log_field_rejected("timestamp_end") return build_error_response(400, "validation", "Field 'timestamp_end' must be an integer (epoch ms).") if cursor is not None and (isinstance(cursor, bool) or not isinstance(cursor, int)): + self._log_field_rejected("cursor") return build_error_response(400, "validation", "Field 'cursor' must be an integer (internal_id).") if not isinstance(limit, int) or isinstance(limit, bool) or limit < 1: + self._log_field_rejected("limit") return build_error_response(400, "validation", "Field 'limit' must be a positive integer.") # Execute query + started_at = time.perf_counter() try: rows, pagination = self.reader_postgres.read_stats( timestamp_start=timestamp_start, @@ -91,9 +114,22 @@ def handle_request(self, event: dict[str, Any]) -> dict[str, Any]: limit=limit, ) except RuntimeError: - logger.exception("Stats query failed for topic %s.", topic_name) + logger.exception( + "Stats query failed.", + extra={"query_duration_ms": round((time.perf_counter() - started_at) * 1000, 2)}, + ) return build_error_response(500, "database", "Stats query failed.") + # The outcome fields are appended to the request context so the single INFO line emitted + # by `dispatch_request()` ("Request completed.") carries them; no second INFO line here. + append_request_context( + query_duration_ms=round((time.perf_counter() - started_at) * 1000, 2), + row_count=len(rows), + has_more=pagination.get("has_more"), + limit=pagination.get("limit"), + ) + logger.debug("Stats query completed.") + return { "statusCode": 200, "headers": {"Content-Type": "application/json"}, diff --git a/src/handlers/handler_token.py b/src/handlers/handler_token.py index 114b8a6..b97c00c 100644 --- a/src/handlers/handler_token.py +++ b/src/handlers/handler_token.py @@ -63,7 +63,10 @@ def _refresh_keys_if_needed(self) -> None: logger.debug("Token public keys are stale, refreshing now.") self.with_public_keys_queried() except RuntimeError: - logger.warning("Token public key refresh failed, using existing keys.") + logger.warning( + "Token public key refresh failed, using existing keys.", + extra={"public_key_count": len(self.public_keys)}, + ) def with_public_keys_queried(self) -> "HandlerToken": """Load token public keys from the configured URL. @@ -92,12 +95,15 @@ def with_public_keys_queried(self) -> "HandlerToken": self.public_keys = [ cast(RSAPublicKey, serialization.load_der_public_key(base64.b64decode(raw_key))) for raw_key in raw_keys ] - logger.debug("Loaded %d token public keys.", len(self.public_keys)) + logger.info("Loaded token public keys.", extra={"public_key_count": len(self.public_keys)}) self._last_loaded_at = datetime.now(timezone.utc) return self except (requests.RequestException, ValueError, KeyError, UnsupportedAlgorithm) as exc: - logger.exception("Failed to fetch or deserialize token public key.") + logger.exception( + "Failed to fetch or deserialize token public key.", + extra={"public_keys_url": self.public_keys_url}, + ) raise RuntimeError("Token public key initialization failed") from exc def decode_jwt(self, token_encoded: str) -> dict[str, Any]: @@ -111,12 +117,19 @@ def decode_jwt(self, token_encoded: str) -> dict[str, Any]: """ self._refresh_keys_if_needed() - logger.debug("Decoding JWT.") + logger.debug("Decoding JWT.", extra={"public_key_count": len(self.public_keys)}) for public_key in self.public_keys: try: return jwt.decode(token_encoded, public_key, algorithms=["RS256"]) except jwt.PyJWTError: continue + + # DEBUG on purpose: the rejection itself is logged once by the calling handler, this line + # only adds the key count for debugging key-rollover issues. + logger.debug( + "JWT verification failed for all public keys.", + extra={"public_key_count": len(self.public_keys)}, + ) raise jwt.PyJWTError("Verification failed for all public keys") def get_token_provider_info(self) -> dict[str, Any]: diff --git a/src/handlers/handler_topic.py b/src/handlers/handler_topic.py index 09b6661..373ed8c 100644 --- a/src/handlers/handler_topic.py +++ b/src/handlers/handler_topic.py @@ -19,6 +19,7 @@ import json import logging import os +import time from typing import Any import jwt @@ -30,7 +31,8 @@ from src.utils.conf_path import CONF_DIR from src.utils.config_loader import TopicAccessMap, TopicKeyMap, load_access_config, load_topic_keys_config from src.utils.constants import TOPIC_DLCHANGE, TOPIC_RUNS, TOPIC_STATUS_CHANGE, TOPIC_TEST -from src.utils.utils import build_error_response +from src.utils.observability import append_request_context +from src.utils.utils import build_error_response, resolve_request_topic from src.writers.writer import WriteError, Writer logger = logging.getLogger(__name__) @@ -68,7 +70,7 @@ def with_load_topic_schemas(self) -> "HandlerTopic": The current instance with loaded topic schemas. """ topic_schemas_dir = os.path.join(CONF_DIR, "topic_schemas") - logger.debug("Loading topic schemas from %s.", topic_schemas_dir) + logger.debug("Loading topic schemas.", extra={"topic_schemas_dir": topic_schemas_dir}) with open(os.path.join(topic_schemas_dir, "runs.json"), "r", encoding="utf-8") as file: self.topics[TOPIC_RUNS] = json.load(file) @@ -79,7 +81,7 @@ def with_load_topic_schemas(self) -> "HandlerTopic": with open(os.path.join(topic_schemas_dir, "status_change.json"), "r", encoding="utf-8") as file: self.topics[TOPIC_STATUS_CHANGE] = json.load(file) - logger.debug("Loaded topic schemas successfully.") + logger.debug("Loaded topic schemas.", extra={"topics": sorted(self.topics)}) return self def with_load_topic_keys_config(self) -> "HandlerTopic": @@ -109,17 +111,33 @@ def handle_request(self, event: dict[str, Any]) -> dict[str, Any]: Returns: API Gateway response. """ - topic_name = event["pathParameters"]["topic_name"].lower() + topic_name, topic_error = resolve_request_topic(event) + if topic_error is not None: + return topic_error + method = event.get("httpMethod") if method == "GET": return self._get_topic_schema(topic_name) if method == "POST": + try: + # A message is mandatory here, so an empty body must fail parsing. `/stats` differs + # on purpose: there the body only carries optional filters. + topic_message = json.loads(event.get("body") or "") + except (json.JSONDecodeError, TypeError): + logger.warning("Request rejected: message body is not valid JSON.") + return build_error_response(400, "validation", "Request body must be valid JSON.") + if not isinstance(topic_message, dict): + logger.warning("Request rejected: message body is not a JSON object.") + return build_error_response(400, "validation", "Request body must be a JSON object.") + return self._post_topic_message( topic_name, - json.loads(event["body"]), + topic_message, self.handler_token.extract_token(event.get("headers", {})), ) + + logger.warning("Request rejected: unsupported HTTP method.") return build_error_response(404, "route", "Resource not found") def _get_topic_schema(self, topic_name: str) -> dict[str, Any]: @@ -129,9 +147,10 @@ def _get_topic_schema(self, topic_name: str) -> dict[str, Any]: Returns: API Gateway response with topic schema or error. """ - logger.debug("Handling GET TopicSchema(%s).", topic_name) + logger.debug("Handling GET topic schema.") if topic_name not in self.topics: + logger.warning("Request rejected: unknown topic.", extra={"known_topics": sorted(self.topics)}) return build_error_response(404, "topic", f"Topic '{topic_name}' not found") return { @@ -153,7 +172,7 @@ def _post_topic_message(self, topic_name: str, topic_message: dict[str, Any], to jwt.PyJWTError: If token decoding fails. ValidationError: If message validation fails. """ - logger.debug("Handling POST TopicMessage(%s).", topic_name) + logger.debug("Handling POST topic message.") if not self.access_config: logger.error("Access configuration not loaded.") @@ -161,46 +180,61 @@ def _post_topic_message(self, topic_name: str, topic_message: dict[str, Any], to try: token: dict[str, Any] = self.handler_token.decode_jwt(token_encoded) - except jwt.PyJWTError: # type: ignore[attr-defined] + except jwt.PyJWTError as exc: # type: ignore[attr-defined] + logger.warning( + "Request rejected: token verification failed.", + extra={"auth_error": type(exc).__name__, "token_present": bool(token_encoded)}, + ) return build_error_response(401, "auth", "Invalid or missing token") if topic_name not in self.topics: + logger.warning("Request rejected: unknown topic.", extra={"known_topics": sorted(self.topics)}) return build_error_response(404, "topic", f"Topic '{topic_name}' not found") user = token.get("sub") + append_request_context(user=user) + authorized_user = self._resolve_authorized_user(topic_name, user) if authorized_user is None: + logger.warning("Request rejected: user is not authorized for the topic.") return build_error_response(403, "auth", f"User '{user}' is not authorized for topic '{topic_name}'") + # Log under the configured spelling of the user, since the token casing may differ. + append_request_context(user=authorized_user) + allowed, perm_error = self._validate_user_permissions(topic_name, authorized_user, topic_message) if not allowed: + logger.warning("Request rejected: user permissions do not allow the message.", extra={"reason": perm_error}) return build_error_response( 403, "permission", perm_error or f"Permission denied for user '{authorized_user}' for POST to topic '{topic_name}'", ) - logger.debug("Authorized user '%s' for topic '%s'.", authorized_user, topic_name) + logger.debug("User authorized for the topic.") try: validate(instance=topic_message, schema=self.topics[topic_name]) except ValidationError as exc: + logger.warning( + "Request rejected: message does not match the topic schema.", + extra={"validation_path": str(exc.json_path), "validator": str(exc.validator)}, + ) return build_error_response(400, "validation", exc.message) message_key = self._resolve_message_key(topic_name, topic_message) - errors = [] - for writer_name, writer in self.writers.items(): - try: - writer.write(topic_name, topic_message, message_key) - except WriteError as exc: - errors.append({"type": writer_name, "message": str(exc)}) + writers_ok, errors = self._write_to_all(topic_name, topic_message, message_key) if errors: logger.error( - "POST to topic '%s' failed: %d of %d writer(s) reported errors.", - topic_name, - len(errors), - len(self.writers), + "Message dispatch failed for at least one writer.", + extra={ + "writers_ok": writers_ok, + "writers_failed": [error["type"] for error in errors], + "writer_errors": errors, + "writer_count": len(self.writers), + "message_key": message_key, + }, ) return { "statusCode": 500, @@ -208,13 +242,64 @@ def _post_topic_message(self, topic_name: str, topic_message: dict[str, Any], to "body": json.dumps({"success": False, "statusCode": 500, "errors": errors}), } - logger.info("Message accepted for topic '%s' from user '%s'.", topic_name, authorized_user) + # The outcome fields are appended to the request context so the single INFO line emitted + # by `dispatch_request()` ("Request completed.") carries them; no second INFO line here. + append_request_context(message_key=message_key, writers_ok=writers_ok) + logger.debug("Message accepted.") return { "statusCode": 202, "headers": {"Content-Type": "application/json"}, "body": json.dumps({"success": True, "statusCode": 202}), } + def _write_to_all( + self, + topic_name: str, + topic_message: dict[str, Any], + message_key: str, + ) -> tuple[list[str], list[dict[str, str]]]: + """Dispatch a message to every configured writer, collecting per-writer outcomes. + Args: + topic_name: Target topic name. + topic_message: Message payload. + message_key: Resolved transport key. + Returns: + Tuple of (writers_ok, errors) where `writers_ok` lists the writers that accepted the + message and `errors` holds one entry per failing writer. + """ + writers_ok: list[str] = [] + errors: list[dict[str, str]] = [] + + for writer_name, writer in self.writers.items(): + started_at = time.perf_counter() + try: + writer.write(topic_name, topic_message, message_key) + writers_ok.append(writer_name) + logger.debug( + "Writer accepted the message.", + extra={ + "writer": writer_name, + "writer_duration_ms": round((time.perf_counter() - started_at) * 1000, 2), + }, + ) + except WriteError as exc: + errors.append({"type": writer_name, "message": str(exc)}) + # WARNING on purpose: the request-level ERROR is emitted once by the caller with + # every failed writer folded in, so an alert on ERROR counts one failure once. + # The traceback rides on this line because the caller runs outside the `except` + # block and can no longer reach it. + logger.warning( + "Writer failed to publish the message.", + exc_info=True, + extra={ + "writer": writer_name, + "writer_duration_ms": round((time.perf_counter() - started_at) * 1000, 2), + "writer_error": str(exc), + }, + ) + + return writers_ok, errors + def _resolve_authorized_user(self, topic_name: str, user: str | None) -> str | None: """Match a token user to a configured user for a topic, ignoring case. Args: @@ -273,11 +358,11 @@ def _resolve_message_key(self, topic_name: str, message: dict[str, Any]) -> str: key_value = message.get(key_field) if key_value is None: - logger.warning("Topic key field '%s' missing for topic '%s'.", key_field, topic_name) + logger.warning("Topic key field is missing from the message.", extra={"key_field": key_field}) return "" if isinstance(key_value, (dict, list)): - logger.warning("Topic key field '%s' for topic '%s' is not scalar.", key_field, topic_name) + logger.warning("Topic key field is not a scalar value.", extra={"key_field": key_field}) return "" return str(key_value) diff --git a/src/readers/reader_postgres.py b/src/readers/reader_postgres.py index 355f0b4..98444f8 100644 --- a/src/readers/reader_postgres.py +++ b/src/readers/reader_postgres.py @@ -113,6 +113,11 @@ def read_stats( ts_start = timestamp_start if timestamp_start is not None else (now_ms - POSTGRES_DEFAULT_WINDOW_MS) ts_end = timestamp_end if timestamp_end is not None else now_ms + started_at = time.perf_counter() + logger.debug( + "Running stats query.", + extra={"ts_start": ts_start, "ts_end": ts_end, "cursor": cursor, "limit": limit}, + ) try: col_names, raw_rows = self._execute_with_retry( lambda conn: self._run_stats_query(conn, ts_start, ts_end, cursor, limit) @@ -139,7 +144,14 @@ def read_stats( "limit": limit, } - logger.debug("Stats query returned %d rows.", len(rows)) + logger.debug( + "Stats query returned rows.", + extra={ + "row_count": len(rows), + "has_more": has_more, + "query_duration_ms": round((time.perf_counter() - started_at) * 1000, 2), + }, + ) return rows, pagination def _run_stats_query( diff --git a/src/utils/conf_path.py b/src/utils/conf_path.py index 0998987..62cb477 100644 --- a/src/utils/conf_path.py +++ b/src/utils/conf_path.py @@ -22,6 +22,7 @@ 4. Fallback to /conf even if missing (subsequent file operations will raise) """ +import logging import os @@ -72,4 +73,18 @@ def resolve_conf_dir(env_var: str = "CONF_DIR"): CONF_DIR, INVALID_CONF_ENV = resolve_conf_dir() -__all__ = ["resolve_conf_dir", "CONF_DIR", "INVALID_CONF_ENV"] + +def log_configuration_source(target_logger: logging.Logger) -> None: + """Report which configuration directory the service resolved. + Args: + target_logger: Logger of the calling entry point. + """ + target_logger.debug("Using configuration directory.", extra={"conf_dir": CONF_DIR}) + if INVALID_CONF_ENV: + target_logger.warning( + "CONF_DIR env var points to a non-existent path; falling back.", + extra={"invalid_conf_dir": INVALID_CONF_ENV, "conf_dir": CONF_DIR}, + ) + + +__all__ = ["resolve_conf_dir", "log_configuration_source", "CONF_DIR", "INVALID_CONF_ENV"] diff --git a/src/utils/config_loader.py b/src/utils/config_loader.py index 3093da4..fb59fdb 100644 --- a/src/utils/config_loader.py +++ b/src/utils/config_loader.py @@ -43,7 +43,7 @@ def load_config(conf_dir: str) -> dict[str, Any]: config_path = os.path.join(conf_dir, "config.json") with open(config_path, "r", encoding="utf-8") as file: config: dict[str, Any] = json.load(file) - logger.debug("Loaded main configuration from %s.", config_path) + logger.debug("Loaded main configuration.", extra={"config_path": config_path}) return config @@ -139,7 +139,7 @@ def load_access_config(config: dict[str, Any], aws_s3: ServiceResource) -> Topic Normalized mapping of topic names to per-user permission constraints. """ access_path: str = config["access_config"] - logger.debug("Loading access configuration from %s.", access_path) + logger.debug("Loading access configuration.", extra={"access_path": access_path}) access_data = _load_json_from_path(access_path, aws_s3) logger.debug("Loaded access configuration.") @@ -174,7 +174,7 @@ def load_topic_keys_config(config: dict[str, Any], aws_s3: ServiceResource) -> T logger.debug("No topic_keys_config configured. Using empty topic key mapping.") return {} - logger.debug("Loading topic key configuration from %s.", topic_keys_path) + logger.debug("Loading topic key configuration.", extra={"topic_keys_path": topic_keys_path}) topic_key_data = _load_json_from_path(topic_keys_path, aws_s3) logger.debug("Loaded topic key configuration.") @@ -203,5 +203,5 @@ def load_topic_names(conf_dir: str) -> list[str]: if os.path.isfile(schema_path): topics.append(topic_name) - logger.debug("Discovered %d topic(s) from %s.", len(topics), schemas_dir) + logger.debug("Discovered topics.", extra={"topic_count": len(topics), "schemas_dir": schemas_dir}) return topics diff --git a/src/utils/logging_levels.py b/src/utils/logging_levels.py index b21062f..0012310 100644 --- a/src/utils/logging_levels.py +++ b/src/utils/logging_levels.py @@ -14,12 +14,14 @@ # limitations under the License. # -"""Custom logging levels and shared logging setup. +"""Custom logging levels and log level resolution. -Adds a TRACE level below DEBUG for very verbose payload logging. +Adds a TRACE level below DEBUG for very verbose payload logging and resolves the +configured level from the environment. The level is registered with the standard +`logging` module, which is what makes `LOG_LEVEL=TRACE` resolvable by AWS Lambda +Powertools as well. """ -from __future__ import annotations import logging import os @@ -44,24 +46,26 @@ def trace(self: logging.Logger, message: str, *args, **kws): # type: ignore[ove logging.Logger.trace = trace # type: ignore[attr-defined] -def init_root_logger(module_name: str) -> logging.Logger: - """Initialize the root logger and return a module-level logger. - Args: - module_name: Typically `__name__` of the calling module. +def configured_log_level() -> str: + """Read the log level requested through the environment. Returns: - A logger instance for the calling module. + The upper-cased level name, defaulting to `INFO`. """ - root_logger = logging.getLogger() - if not root_logger.handlers: - root_logger.addHandler(logging.StreamHandler()) + return (os.environ.get("POWERTOOLS_LOG_LEVEL") or os.environ.get("LOG_LEVEL") or "INFO").strip().upper() - log_level = os.environ.get("LOG_LEVEL", "INFO").upper() - try: - root_logger.setLevel(log_level) - except ValueError: - root_logger.setLevel("INFO") - root_logger.warning("Invalid LOG_LEVEL %s; defaulting to INFO.", log_level) - return logging.getLogger(module_name) + +def resolve_log_level() -> tuple[int, str | None]: + """Resolve the configured log level to a numeric logging level. + Returns: + Tuple of (level, rejected). `level` is the numeric level for the configured name, falling + back to `logging.INFO` when the name is unknown; `rejected` is that unusable name, or + `None` when the configured level resolved cleanly. + """ + configured = configured_log_level() + level = logging.getLevelName(configured) + if isinstance(level, int): + return level, None + return logging.INFO, configured -__all__ = ["TRACE_LEVEL", "init_root_logger"] +__all__ = ["TRACE_LEVEL", "configured_log_level", "resolve_log_level"] diff --git a/src/utils/observability.py b/src/utils/observability.py new file mode 100644 index 0000000..7f4b632 --- /dev/null +++ b/src/utils/observability.py @@ -0,0 +1,234 @@ +# +# Copyright 2026 ABSA Group Limited +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +"""Structured logging and request correlation for the EventGate lambdas. + +Wraps AWS Lambda Powertools `Logger` so that every log line is emitted as JSON +enriched with the Lambda execution context and a request correlation id. +Module loggers keep using `logging.getLogger(__name__)`; the Powertools handler +is attached to the root logger, so their records are formatted and enriched too. +""" + +import logging +import os +import re +from typing import Any + +from aws_lambda_powertools import Logger + +from src.utils.logging_levels import TRACE_LEVEL, configured_log_level, resolve_log_level + +DEFAULT_SERVICE_NAME = "eventgate" + +# Attribute carrying the invocation id on the AWS Lambda context object. +AWS_REQUEST_ID = "aws_request_id" + +# Header accepted from callers so that an upstream request id can be carried through EventGate. +CORRELATION_ID_HEADER = "x-correlation-id" +REQUEST_ID_HEADER = "x-request-id" +# Header returned to the caller on every response, so a client can quote a single id in a ticket. +CORRELATION_ID_RESPONSE_HEADER = "X-Correlation-ID" + +# Inbound correlation ids end up in the log stream, so they must not carry newlines or control +# characters (log injection) and must stay small enough not to bloat every log line. +CORRELATION_ID_PATTERN = re.compile(r"^[A-Za-z0-9._:-]{1,128}$") + +# Third-party loggers that flood the log group when the service runs at DEBUG or TRACE. +NOISY_LOGGERS = ("boto3", "botocore", "urllib3", "s3transfer", "aiosql", "confluent_kafka") + +_INITIAL_LOG_LEVEL, _ = resolve_log_level() + +logger: Logger = Logger( + service=os.environ.get("POWERTOOLS_SERVICE_NAME", DEFAULT_SERVICE_NAME), + level=_INITIAL_LOG_LEVEL, + use_rfc3339=True, + utc=True, + log_uncaught_exceptions=True, +) + +# Names of the log keys bound for the duration of a single invocation; cleared at the start of +# the next one so a warm container does not leak context between requests. +_request_context_key_names: set[str] = set() + +# Mutable container so the flag can be flipped without a module level `global` statement. +_container_state: dict[str, bool] = {"cold_start": True} + + +def configure_root_logging() -> None: + """Route every logger in the process through the Powertools JSON handler. + + Replaces handlers installed by the Lambda runtime so that records are not emitted twice, + and caps noisy third-party loggers at WARNING so `LOG_LEVEL=DEBUG` stays readable. + """ + log_level, _ = resolve_log_level() + + root_logger = logging.getLogger() + root_logger.handlers.clear() + root_logger.addHandler(logger.registered_handler) + root_logger.setLevel(log_level) + + for noisy_logger in NOISY_LOGGERS: + logging.getLogger(noisy_logger).setLevel(max(log_level, logging.WARNING)) + + +def refresh_sampled_log_level() -> int: + """Re-draw the `DEBUG` sampling decision and mirror the outcome onto the root logger. + + Powertools draws the `POWERTOOLS_LOGGER_SAMPLE_RATE` lottery when the `Logger` is built and + re-draws it only from `refresh_sample_rate_calculation()`, which ships inside the + `inject_lambda_context` decorator this module deliberately does not use. Calling it per + invocation is what makes the draw per request rather than per container. + + The draw sets the level of the Powertools logger alone. Module loggers resolve their effective + level from the root logger, so the outcome has to be mirrored there or their `DEBUG` records + are dropped before the shared handler sees them. + + Returns: + The level now in force on the root logger. + """ + logger.refresh_sample_rate_calculation() + + configured_level, _ = resolve_log_level() + # Powertools sets `DEBUG` unconditionally when a request is sampled. Sampling may only add + # detail, so a run configured at `TRACE` must not be pulled up to `DEBUG`. + sampled_level = min(logger.level, configured_level) + + root_logger = logging.getLogger() + if root_logger.level != sampled_level: + root_logger.setLevel(sampled_level) + for noisy_logger in NOISY_LOGGERS: + logging.getLogger(noisy_logger).setLevel(max(sampled_level, logging.WARNING)) + + return sampled_level + + +def init_lambda_logging(module_name: str) -> logging.Logger: + """Configure logging for a lambda entry point. + Args: + module_name: Typically `__name__` of the calling entry point module. + Returns: + The module logger to use for the rest of the module. + """ + configure_root_logging() + module_logger = logging.getLogger(module_name) + + if rejected_level := resolve_log_level()[1]: + module_logger.warning( + "Invalid log level configured; defaulting to INFO.", + extra={"log_level": rejected_level}, + ) + + return module_logger + + +def resolve_correlation_id(event: dict[str, Any]) -> str: + """Resolve the correlation id for an invocation. + Args: + event: API Gateway proxy event. + Returns: + The caller supplied correlation id when present and well-formed, otherwise the + API Gateway request id, otherwise an empty string. + """ + headers = event.get("headers") or {} + lowered_headers = {str(key).lower(): value for key, value in headers.items()} + + for header in (CORRELATION_ID_HEADER, REQUEST_ID_HEADER): + header_value = lowered_headers.get(header) + if isinstance(header_value, str) and CORRELATION_ID_PATTERN.fullmatch(header_value.strip()): + return header_value.strip() + + request_context = event.get("requestContext") or {} + api_request_id = request_context.get("requestId") + return str(api_request_id) if api_request_id else "" + + +def append_request_context(**context_keys: Any) -> None: + """Append structured log context for the current invocation only. + Args: + **context_keys: Key/value pairs added to every subsequent log line of this invocation. + """ + if not context_keys: + return + _request_context_key_names.update(context_keys) + logger.append_keys(**context_keys) + + +def _clear_request_context() -> None: + """Drop the context appended during the previous invocation of a warm container.""" + if _request_context_key_names: + logger.remove_keys(list(_request_context_key_names)) + _request_context_key_names.clear() + + +def bind_request_context(event: dict[str, Any], context: Any = None) -> str: + """Bind Lambda and request context to the logger for the current invocation. + Args: + event: API Gateway proxy event. + context: AWS Lambda context object. May be `None` when invoked locally or from tests. + Returns: + The resolved correlation id (possibly an empty string). + """ + _clear_request_context() + refresh_sampled_log_level() + + correlation_id = resolve_correlation_id(event) + logger.set_correlation_id(correlation_id or None) + + cold_start = _container_state["cold_start"] + _container_state["cold_start"] = False + + append_request_context( + cold_start=cold_start, + resource=event.get("resource", ""), + http_method=event.get("httpMethod", ""), + ) + + if context is not None: + append_request_context(function_request_id=getattr(context, AWS_REQUEST_ID, None)) + if cold_start: + # Static per-deployment facts are logged once per container instead of being bound + # to every line; the log stream already identifies the function, so repeating them + # would only inflate every record. + logger.info( + "Lambda execution context.", + extra={ + "function_name": getattr(context, "function_name", None), + "function_memory_size": getattr(context, "memory_limit_in_mb", None), + "function_arn": getattr(context, "invoked_function_arn", None), + }, + ) + + return correlation_id + + +__all__ = [ + "AWS_REQUEST_ID", + "CORRELATION_ID_HEADER", + "CORRELATION_ID_PATTERN", + "CORRELATION_ID_RESPONSE_HEADER", + "DEFAULT_SERVICE_NAME", + "NOISY_LOGGERS", + "REQUEST_ID_HEADER", + "TRACE_LEVEL", + "append_request_context", + "bind_request_context", + "configure_root_logging", + "configured_log_level", + "init_lambda_logging", + "refresh_sampled_log_level", + "logger", + "resolve_correlation_id", +] diff --git a/src/utils/postgres_base.py b/src/utils/postgres_base.py index b474026..9b1ae5a 100644 --- a/src/utils/postgres_base.py +++ b/src/utils/postgres_base.py @@ -127,7 +127,10 @@ def _get_connection(self) -> Any: if options: connect_kwargs["options"] = options self._connection = psycopg2.connect(**connect_kwargs) - logger.debug("New PostgreSQL connection established.") + logger.info( + "New PostgreSQL connection established.", + extra={"database": pg_config["database"], "host": pg_config["host"], "port": pg_config["port"]}, + ) return self._connection def _close_connection(self) -> None: @@ -138,7 +141,7 @@ def _close_connection(self) -> None: try: conn_to_close.close() except (PsycopgError, OSError): - logger.debug("Failed to close PostgreSQL connection.") + logger.debug("Failed to close PostgreSQL connection.", exc_info=True) def _execute_with_retry[T](self, operation: Callable[..., T], *, retry: bool = True) -> T: """Run `operation(connection)` with one retry on `OperationalError`. @@ -161,5 +164,8 @@ def _execute_with_retry[T](self, operation: Callable[..., T], *, retry: bool = T last_exc = exc self._close_connection() if attempt < max_attempts - 1: - logger.warning("PostgreSQL connection lost, reconnecting.") + logger.warning( + "PostgreSQL connection lost, reconnecting.", + extra={"attempt": attempt + 1, "max_attempts": max_attempts}, + ) raise RuntimeError(f"Database connection failed after {max_attempts} attempts: {last_exc}") from last_exc diff --git a/src/utils/trace_logging.py b/src/utils/trace_logging.py index 232693a..c216465 100644 --- a/src/utils/trace_logging.py +++ b/src/utils/trace_logging.py @@ -26,13 +26,15 @@ from src.utils.safe_serialization import safe_serialize_for_log -def log_payload_at_trace(logger: logging.Logger, writer_name: str, topic_name: str, message: dict[str, Any]) -> None: +def log_payload_at_trace(logger: logging.Logger, writer_name: str, message: dict[str, Any]) -> None: """Log message payload at TRACE level with safe serialization. + The topic is not repeated here; it is already bound to every log line of the + request by `resolve_request_topic()`. + Args: logger: Logger instance to use for logging. writer_name: Name of the writer (e.g., "EventBridge", "Kafka", "Postgres"). - topic_name: Topic name being written to. message: Message payload to log. """ if not logger.isEnabledFor(TRACE_LEVEL): @@ -42,10 +44,10 @@ def log_payload_at_trace(logger: logging.Logger, writer_name: str, topic_name: s safe_payload = safe_serialize_for_log(message) if safe_payload: logger.trace( # type: ignore[attr-defined] - "%s payload topic=%s payload=%s", writer_name, topic_name, safe_payload + "Writer payload.", extra={"writer": writer_name, "payload": safe_payload} ) except (TypeError, ValueError): # pragma: no cover - defensive serialization guard - logger.trace("%s payload topic=%s ", writer_name, topic_name) # type: ignore[attr-defined] + logger.trace("Writer payload is not serializable.", extra={"writer": writer_name}) # type: ignore[attr-defined] __all__ = ["log_payload_at_trace"] diff --git a/src/utils/utils.py b/src/utils/utils.py index e19d494..3897d29 100644 --- a/src/utils/utils.py +++ b/src/utils/utils.py @@ -18,11 +18,14 @@ import json import logging +import time from collections.abc import Callable from typing import Any import boto3 +from src.utils.observability import CORRELATION_ID_RESPONSE_HEADER, append_request_context, bind_request_context + logger = logging.getLogger(__name__) @@ -48,37 +51,85 @@ def build_error_response(status: int, err_type: str, message: str) -> dict[str, } +def resolve_request_topic(event: dict[str, Any]) -> tuple[str, dict[str, Any] | None]: + """Resolve the `topic_name` path parameter and bind it as a request scoped log key. + Shared by every `/{...}/{topic_name}` handler so the rejection message, the status code and + the normalization stay identical across lambdas. + Args: + event: API Gateway proxy event. + Returns: + Tuple of (topic_name, error_response). On success `topic_name` is the lowercased topic and + `error_response` is `None`; when the path parameter is missing `topic_name` is empty and + `error_response` holds the 400 response to return. + """ + path_parameters = event.get("pathParameters") or {} + raw_topic_name = path_parameters.get("topic_name") + if not raw_topic_name: + logger.warning("Request rejected: path parameter 'topic_name' is missing.") + return "", build_error_response(400, "validation", "Missing path parameter 'topic_name'.") + + topic_name = str(raw_topic_name).lower() + append_request_context(topic=topic_name) + return topic_name, None + + +def with_correlation_header(response: dict[str, Any], correlation_id: str) -> dict[str, Any]: + """Add the correlation id header to a response so callers can quote it when reporting issues. + Args: + response: API Gateway response dictionary. + correlation_id: Correlation id of the current request. Ignored when empty. + Returns: + The response with the correlation header set. + """ + if not correlation_id or not isinstance(response, dict): + return response + + headers = dict(response.get("headers") or {}) + headers[CORRELATION_ID_RESPONSE_HEADER] = correlation_id + response["headers"] = headers + return response + + def dispatch_request( event: dict[str, Any], route_map: dict[str, Callable[..., dict[str, Any]]], request_logger: logging.Logger, + context: Any = None, ) -> dict[str, Any]: """Dispatch an API Gateway event to the matching route handler. + Binds the request context to the logger, logs the request outcome and stamps the response + with the correlation id. Args: event: API Gateway proxy event. route_map: Mapping of resource paths to handler callables. request_logger: Logger instance for error reporting. + context: AWS Lambda context object, when available. Returns: API Gateway response dictionary. """ + started_at = time.perf_counter() + correlation_id = bind_request_context(event, context) + resource = str(event.get("resource", "")).lower() + try: - resource = event.get("resource", "").lower() route_function = route_map.get(resource) + if route_function is None: + request_logger.warning("No route matched the requested resource.") + response = build_error_response(404, "route", "Resource not found.") + else: + request_logger.debug("Dispatching request to the route handler.") + response = route_function(event) + except Exception: # pylint: disable=broad-exception-caught + # Boundary handler: any escaping exception must become a stable 500 with a logged cause, + # otherwise API Gateway returns an opaque 502 and the traceback is unstructured. + request_logger.exception("Unhandled error while processing the request.") + response = build_error_response(500, "internal", "Unexpected server error.") + + duration_ms = round((time.perf_counter() - started_at) * 1000, 2) + status_code = response.get("statusCode") if isinstance(response, dict) else None + request_logger.info("Request completed.", extra={"status_code": status_code, "duration_ms": duration_ms}) - if route_function: - return route_function(event) - - return build_error_response(404, "route", "Resource not found.") - except ( - KeyError, - json.JSONDecodeError, - ValueError, - AttributeError, - TypeError, - RuntimeError, - ) as request_exc: - request_logger.exception("Request processing error: %s.", request_exc) - return build_error_response(500, "internal", "Unexpected server error.") + return with_correlation_header(response, correlation_id) def load_postgres_config(secret_name: str, secret_region: str) -> dict[str, Any]: diff --git a/src/writers/writer_eventbridge.py b/src/writers/writer_eventbridge.py index 076f007..be1c144 100644 --- a/src/writers/writer_eventbridge.py +++ b/src/writers/writer_eventbridge.py @@ -18,6 +18,7 @@ import json import logging +import time from typing import Any, Optional import boto3 @@ -66,10 +67,11 @@ def write(self, topic_name: str, message: dict[str, Any], message_key: str = "") self._client = boto3.client("events") logger.debug("EventBridge client initialized.") - log_payload_at_trace(logger, "EventBridge", topic_name, message) + log_payload_at_trace(logger, "EventBridge", message) + started_at = time.perf_counter() try: - logger.debug("Sending to EventBridge %s.", topic_name) + logger.debug("Sending message to EventBridge.") response = self._client.put_events( Entries=[ { @@ -80,15 +82,37 @@ def write(self, topic_name: str, message: dict[str, Any], message_key: str = "") } ] ) + duration_ms = round((time.perf_counter() - started_at) * 1000, 2) failed_count = response.get("FailedEntryCount", 0) if failed_count > 0: self._entries = response.get("Entries", []) failed_repr = self._format_failed_entries() msg = f"{failed_count} EventBridge entries failed: {failed_repr}" - logger.error(msg) + logger.error( + "EventBridge rejected entries.", + extra={ + "writer_duration_ms": duration_ms, + "failed_entry_count": failed_count, + "failed_entries": failed_repr, + }, + ) raise WriteError(msg) + + entries = response.get("Entries") or [{}] + logger.debug( + "EventBridge accepted the message.", + extra={ + "writer_duration_ms": duration_ms, + "event_id": entries[0].get("EventId"), + }, + ) except (BotoCoreError, ClientError) as err: # explicit AWS client-related errors - logger.exception("EventBridge put_events call failed.") + logger.exception( + "EventBridge put_events call failed.", + extra={ + "writer_duration_ms": round((time.perf_counter() - started_at) * 1000, 2), + }, + ) raise WriteError(str(err)) from err def check_health(self) -> str | None: diff --git a/src/writers/writer_kafka.py b/src/writers/writer_kafka.py index 58a9ad3..4492825 100644 --- a/src/writers/writer_kafka.py +++ b/src/writers/writer_kafka.py @@ -106,23 +106,38 @@ def write(self, topic_name: str, message: dict[str, Any], message_key: str = "") logger.debug("Kafka producer not initialized - skipping Kafka writer.") return - log_payload_at_trace(logger, "Kafka", topic_name, message) + log_payload_at_trace(logger, "Kafka", message) errors: list[str] = [] - has_exception = False + captured_exception: KafkaException | None = None + delivery: dict[str, Any] = {} + started_at = time.perf_counter() + + def delivery_report(err: Any, msg: Any) -> None: + """Collect the Kafka delivery outcome for logging and error reporting.""" + if err is not None: + errors.append(str(err)) + return + try: + delivery.update({"kafka_partition": msg.partition(), "kafka_offset": msg.offset()}) + except (AttributeError, TypeError): + # The delivery succeeded; only the partition/offset detail is unavailable, because + # not every producer implementation exposes message metadata. Swallowing this keeps + # a successful write from being reported as a failure over a missing log field. + logger.debug("Kafka delivery metadata unavailable.", exc_info=True) # Produce step try: - logger.debug("Sending to Kafka %s.", topic_name) + logger.debug("Sending message to Kafka.", extra={"message_key": message_key}) self._producer.produce( topic_name, key=message_key, value=json.dumps(message).encode("utf-8"), - callback=lambda err, msg: (errors.append(str(err)) if err is not None else None), + callback=delivery_report, ) except KafkaException as e: errors.append(f"Produce exception: {e}") - has_exception = True + captured_exception = e # Flush step (always attempted) remaining: int | None = None @@ -131,28 +146,50 @@ def write(self, topic_name: str, message: dict[str, Any], message_key: str = "") remaining = self._flush_with_timeout(_KAFKA_FLUSH_TIMEOUT_SEC) except KafkaException as e: errors.append(f"Flush exception: {e}") - has_exception = True + captured_exception = e # Treat None (flush returns None in some stubs) as success equivalent to 0 pending - if (remaining is None or remaining == 0) and not errors: + if remaining is None or remaining == 0: break if attempt < _MAX_RETRIES: logger.warning( - "Kafka flush pending (%s message(s) remain) attempt %d/%d.", remaining, attempt, _MAX_RETRIES + "Kafka flush pending, retrying.", + extra={ + "pending_messages": remaining, + "attempt": attempt, + "max_attempts": _MAX_RETRIES, + }, ) time.sleep(_RETRY_BACKOFF_SEC) # Warn if messages still pending after retries if isinstance(remaining, int) and remaining > 0: logger.warning( - "Kafka flush timeout after %ss: %d message(s) still pending.", _KAFKA_FLUSH_TIMEOUT_SEC, remaining + "Kafka flush timed out with messages still pending.", + extra={ + "pending_messages": remaining, + "flush_timeout_sec": _KAFKA_FLUSH_TIMEOUT_SEC, + }, ) + duration_ms = round((time.perf_counter() - started_at) * 1000, 2) + if errors: failure_text = "Kafka writer failed: " + "; ".join(errors) - (logger.exception if has_exception else logger.error)(failure_text) + # logger.exception() is only valid inside an except block; outside it the traceback is + # taken from sys.exc_info() and would be empty, so pass the captured exception instead. + logger.error( + "Kafka writer failed.", + exc_info=captured_exception, + extra={"writer_duration_ms": duration_ms, "writer_errors": errors}, + ) raise WriteError(failure_text) + logger.debug( + "Kafka accepted the message.", + extra={"writer_duration_ms": duration_ms, **delivery}, + ) + def check_health(self) -> str | None: """Check Kafka writer health. Returns: diff --git a/src/writers/writer_postgres.py b/src/writers/writer_postgres.py index fa50209..a8915eb 100644 --- a/src/writers/writer_postgres.py +++ b/src/writers/writer_postgres.py @@ -18,6 +18,7 @@ import json import logging +import time from dataclasses import dataclass from datetime import datetime, timezone from functools import cached_property @@ -84,7 +85,6 @@ def _insert_dlchange(self, cursor: Any, message: dict[str, Any]) -> None: cursor: Database cursor. message: Event payload. """ - logger.debug("Sending to Postgres - dlchange.") cursor.execute( self._queries.insert_dlchange, { @@ -112,7 +112,6 @@ def _insert_run(self, cursor: Any, message: dict[str, Any]) -> None: cursor: Database cursor. message: Event payload (includes jobs array). """ - logger.debug("Sending to Postgres - runs.") cursor.execute( self._queries.insert_run, { @@ -126,6 +125,7 @@ def _insert_run(self, cursor: Any, message: dict[str, Any]) -> None: "timestamp_end": message["timestamp_end"], }, ) + logger.debug("Inserting run jobs.", extra={"job_count": len(message["jobs"])}) for job in message["jobs"]: cursor.execute( self._queries.insert_run_job, @@ -147,7 +147,6 @@ def _insert_test(self, cursor: Any, message: dict[str, Any]) -> None: cursor: Database cursor. message: Event payload. """ - logger.debug("Sending to Postgres - test.") cursor.execute( self._queries.insert_test, { @@ -168,9 +167,11 @@ def _upsert_status_change(self, cursor: Any, message: dict[str, Any]) -> None: cursor: Database cursor. message: Event payload. """ - logger.debug("Sending to Postgres - status_change.") ts = datetime.fromtimestamp(message["timestamp_event"] / 1000.0, tz=timezone.utc) event_type = message["event_type"] + # An unrecognized event_type leaves every timestamp below unset and still upserts a row, + # so this line is the only trace of why the stored record looks empty. + logger.debug("Sending to Postgres - status_change.", extra={"event_type": event_type}) created_at: datetime | None = None started_at: datetime | None = None @@ -232,43 +233,50 @@ def write(self, topic_name: str, message: dict[str, Any], message_key: str = "") Raises: WriteError: If publishing fails. """ + # Checked first: a topic this writer does not persist must never fail the request because + # of a Postgres configuration problem it is not involved in. + if topic_name not in POSTGRES_WRITE_TOPICS: + logger.debug( + "Topic is not persisted by the Postgres writer - skipping.", + extra={"postgres_topics": sorted(POSTGRES_WRITE_TOPICS)}, + ) + return + try: pg_config = self._pg_config except (RuntimeError, BotoCoreError, ClientError, ValueError, KeyError) as e: err_msg = f"The Postgres writer failed with unknown error: {e!s}" - logger.exception(err_msg) + logger.exception("Postgres writer failed to load its configuration.") raise WriteError(err_msg) from e if not pg_config.get("database"): - logger.debug("No Postgres - skipping Postgres writer.") + logger.debug("No Postgres database configured - skipping Postgres writer.") return missing = [field for field in REQUIRED_CONNECTION_FIELDS if not pg_config.get(field)] if missing: msg = f"PostgreSQL connection field '{missing[0]}' not configured." - logger.error(msg) + logger.error("Postgres connection is not fully configured.", extra={"missing_fields": missing}) raise WriteError(msg) if not self._is_psycopg2_available(): raise WriteError("psycopg2 is not available for the configured Postgres writer.") - log_payload_at_trace(logger, "Postgres", topic_name, message) - - if topic_name not in POSTGRES_WRITE_TOPICS: - msg = f"Unknown topic for Postgres/{topic_name}" - logger.debug(msg) - return # no need to pollute the logs and no write should happen for these + log_payload_at_trace(logger, "Postgres", message) try: self._execute_with_retry(lambda conn: self._write_topic(conn, topic_name, message), retry=False) except (RuntimeError, PsycopgError, ValueError, KeyError) as e: self._close_connection() err_msg = f"The Postgres writer failed with unknown error: {e!s}" - logger.exception(err_msg) + logger.exception("Postgres writer failed while inserting the message.") raise WriteError(err_msg) from e def _write_topic(self, connection: Any, topic_name: str, message: dict[str, Any]) -> None: """Execute the insert for the given topic inside a transaction.""" + started_at = time.perf_counter() + logger.debug("Inserting message into Postgres.") + with connection.cursor() as cursor: if topic_name == TOPIC_DLCHANGE: self._insert_dlchange(cursor, message) @@ -280,6 +288,11 @@ def _write_topic(self, connection: Any, topic_name: str, message: dict[str, Any] self._upsert_status_change(cursor, message) connection.commit() + logger.debug( + "Postgres accepted the message.", + extra={"writer_duration_ms": round((time.perf_counter() - started_at) * 1000, 2)}, + ) + @staticmethod def _is_psycopg2_available() -> bool: """Check whether psycopg2 is importable.""" diff --git a/tests/integration/test_health_endpoint.py b/tests/integration/test_health_endpoint.py index 2229454..d743b59 100644 --- a/tests/integration/test_health_endpoint.py +++ b/tests/integration/test_health_endpoint.py @@ -17,6 +17,7 @@ import json +from src.utils.observability import CORRELATION_ID_RESPONSE_HEADER from tests.integration.conftest import EventGateTestClient @@ -44,3 +45,26 @@ def test_get_health_includes_uptime(self, eventgate_client: EventGateTestClient) assert "uptime_seconds" in body assert isinstance(body["uptime_seconds"], (int, float)) assert body["uptime_seconds"] >= 0 + + +class TestCorrelationId: + """Tests for the request correlation id contract.""" + + def test_caller_correlation_id_is_returned(self, eventgate_client: EventGateTestClient) -> None: + """Test a caller supplied correlation id is echoed back in the response headers.""" + response = eventgate_client.invoke("/health", "GET", headers={"X-Correlation-ID": "run-42"}) + + assert "run-42" == response["headers"][CORRELATION_ID_RESPONSE_HEADER] + + def test_malformed_correlation_id_is_not_echoed(self, eventgate_client: EventGateTestClient) -> None: + """Test a malformed correlation id is rejected instead of being written back.""" + response = eventgate_client.invoke("/health", "GET", headers={"X-Correlation-ID": "bad value\ninjected"}) + + assert CORRELATION_ID_RESPONSE_HEADER not in response["headers"] + + def test_error_responses_carry_the_correlation_id(self, eventgate_client: EventGateTestClient) -> None: + """Test the correlation id is present on error responses too.""" + response = eventgate_client.invoke("/unknown", "GET", headers={"X-Correlation-ID": "run-43"}) + + assert 404 == response["statusCode"] + assert "run-43" == response["headers"][CORRELATION_ID_RESPONSE_HEADER] diff --git a/tests/unit/handlers/test_handler_topic.py b/tests/unit/handlers/test_handler_topic.py index f82035f..d7c93e2 100644 --- a/tests/unit/handlers/test_handler_topic.py +++ b/tests/unit/handlers/test_handler_topic.py @@ -15,6 +15,7 @@ # import json +import logging import re from unittest.mock import patch, mock_open, MagicMock @@ -22,6 +23,7 @@ import pytest from src.handlers.handler_topic import HandlerTopic +from src.utils.observability import logger as powertools_logger from src.writers.writer import WriteError @@ -434,3 +436,163 @@ def test_post_permission_denied( body = json.loads(resp["body"]) assert "permission" == body["errors"][0]["type"] assert expected_fragment in body["errors"][0]["message"] + + +## request rejection logging +def logged_messages(caplog, level): + """Collect log messages captured at a single level.""" + return [record.message for record in caplog.records if record.levelno == level] + + +def test_post_missing_topic_path_parameter_is_rejected(event_gate_module, make_event, caplog): + caplog.set_level(logging.WARNING) + event = make_event("/topics/{topic_name}", method="POST", body={"a": 1}) + + resp = event_gate_module.lambda_handler(event) + + assert 400 == resp["statusCode"] + assert "Request rejected: path parameter 'topic_name' is missing." in logged_messages(caplog, logging.WARNING) + + +def test_post_non_object_body_is_rejected(event_gate_module, make_event, caplog): + caplog.set_level(logging.WARNING) + event = make_event("/topics/{topic_name}", method="POST", topic="public.cps.za.test", body="[1, 2]") + + resp = event_gate_module.lambda_handler(event) + + assert 400 == resp["statusCode"] + assert "Request rejected: message body is not a JSON object." in logged_messages(caplog, logging.WARNING) + + +def test_unsupported_method_is_rejected(event_gate_module, make_event, caplog): + caplog.set_level(logging.WARNING) + event = make_event("/topics/{topic_name}", method="PUT", topic="public.cps.za.test") + + resp = event_gate_module.lambda_handler(event) + + assert 404 == resp["statusCode"] + assert "Request rejected: unsupported HTTP method." in logged_messages(caplog, logging.WARNING) + + +def test_invalid_token_is_logged(event_gate_module, make_event, valid_payload, caplog): + caplog.set_level(logging.WARNING) + with patch.object(event_gate_module.handler_token, "decode_jwt", side_effect=jwt.PyJWTError("nope")): + event = make_event( + "/topics/{topic_name}", + method="POST", + topic="public.cps.za.test", + body=valid_payload, + headers={"Authorization": "Bearer token"}, + ) + resp = event_gate_module.lambda_handler(event) + + assert 401 == resp["statusCode"] + assert "Request rejected: token verification failed." in logged_messages(caplog, logging.WARNING) + + +def test_unauthorized_user_is_logged(event_gate_module, make_event, valid_payload, caplog): + caplog.set_level(logging.WARNING) + with patch.object(event_gate_module.handler_token, "decode_jwt", return_value={"sub": "UnknownUser"}): + event = make_event( + "/topics/{topic_name}", + method="POST", + topic="public.cps.za.test", + body=valid_payload, + headers={"Authorization": "Bearer token"}, + ) + resp = event_gate_module.lambda_handler(event) + + assert 403 == resp["statusCode"] + assert "Request rejected: user is not authorized for the topic." in logged_messages(caplog, logging.WARNING) + + +def test_schema_validation_failure_is_logged(event_gate_module, make_event, caplog): + caplog.set_level(logging.WARNING) + with patch.object(event_gate_module.handler_token, "decode_jwt", return_value={"sub": "TestUser"}): + event_gate_module.handler_topic.access_config["public.cps.za.test"] = {"TestUser": {}} + event = make_event( + "/topics/{topic_name}", + method="POST", + topic="public.cps.za.test", + body={"event_id": "e1"}, + headers={"Authorization": "Bearer token"}, + ) + resp = event_gate_module.lambda_handler(event) + + assert 400 == resp["statusCode"] + assert "Request rejected: message does not match the topic schema." in logged_messages(caplog, logging.WARNING) + + +def test_partial_writer_failure_reports_both_sides(event_gate_module, make_event, valid_payload, caplog): + caplog.set_level(logging.ERROR) + with patch.object(event_gate_module.handler_token, "decode_jwt", return_value={"sub": "TestUser"}): + event_gate_module.handler_topic.access_config["public.cps.za.test"] = {"TestUser": {}} + for name, writer in event_gate_module.handler_topic.writers.items(): + writer.write = MagicMock(side_effect=WriteError("down") if name == "postgres" else None) + + event = make_event( + "/topics/{topic_name}", + method="POST", + topic="public.cps.za.test", + body=valid_payload, + headers={"Authorization": "Bearer token"}, + ) + resp = event_gate_module.lambda_handler(event) + + dispatch_failures = [r for r in caplog.records if r.message == "Message dispatch failed for at least one writer."] + assert 500 == resp["statusCode"] + assert 1 == len(dispatch_failures) + assert ["postgres"] == dispatch_failures[0].writers_failed + assert ["eventbridge", "kafka"] == sorted(dispatch_failures[0].writers_ok) + assert ["down"] == [error["message"] for error in dispatch_failures[0].writer_errors] + + +def test_writer_failure_warning_carries_the_traceback(event_gate_module, make_event, valid_payload, caplog): + caplog.set_level(logging.WARNING) + with patch.object(event_gate_module.handler_token, "decode_jwt", return_value={"sub": "TestUser"}): + event_gate_module.handler_topic.access_config["public.cps.za.test"] = {"TestUser": {}} + for name, writer in event_gate_module.handler_topic.writers.items(): + writer.write = MagicMock(side_effect=WriteError("down") if name == "postgres" else None) + + event = make_event( + "/topics/{topic_name}", + method="POST", + topic="public.cps.za.test", + body=valid_payload, + headers={"Authorization": "Bearer token"}, + ) + event_gate_module.lambda_handler(event) + + writer_failures = [r for r in caplog.records if r.message == "Writer failed to publish the message."] + assert 1 == len(writer_failures) + # The caller emits its ERROR outside the `except` block, so this is the only line that can + # carry the traceback of the underlying writer failure. + assert writer_failures[0].exc_info is not None + assert WriteError is writer_failures[0].exc_info[0] + + +def test_accepted_message_is_logged(event_gate_module, make_event, valid_payload, caplog): + caplog.set_level(logging.INFO) + with patch.object(event_gate_module.handler_token, "decode_jwt", return_value={"sub": "TestUser"}): + event_gate_module.handler_topic.access_config["public.cps.za.test"] = {"TestUser": {}} + for writer in event_gate_module.handler_topic.writers.values(): + writer.write = MagicMock(return_value=None) + + event = make_event( + "/topics/{topic_name}", + method="POST", + topic="public.cps.za.test", + body=valid_payload, + headers={"Authorization": "Bearer token"}, + ) + resp = event_gate_module.lambda_handler(event) + + completed = [r for r in caplog.records if r.message == "Request completed."] + assert 202 == resp["statusCode"] + assert 1 == len(completed) + + # The outcome fields live on the formatter (request scoped keys), not on the LogRecord. + payload = json.loads(powertools_logger.registered_formatter.format(completed[0])) + assert 202 == payload["status_code"] + assert ["eventbridge", "kafka", "postgres"] == sorted(payload["writers_ok"]) + assert "message_key" in payload diff --git a/tests/unit/test_event_gate_lambda.py b/tests/unit/test_event_gate_lambda.py index 142f18f..14a5056 100644 --- a/tests/unit/test_event_gate_lambda.py +++ b/tests/unit/test_event_gate_lambda.py @@ -54,7 +54,7 @@ def test_internal_error_path(event_gate_module, make_event): def test_post_invalid_json_body(event_gate_module, make_event): - # Invalid JSON triggers json.loads exception and returns 500 internal error + # Invalid JSON is rejected as a client error, not swallowed as an internal error event = make_event( "/topics/{topic_name}", method="POST", @@ -63,9 +63,9 @@ def test_post_invalid_json_body(event_gate_module, make_event): headers={"Authorization": "Bearer token"}, ) resp = event_gate_module.lambda_handler(event) - assert 500 == resp["statusCode"] + assert 400 == resp["statusCode"] body = json.loads(resp["body"]) - assert any(e["type"] == "internal" for e in body["errors"]) # internal error path + assert any(e["type"] == "validation" for e in body["errors"]) # validation error path def test_boto3_s3_client_default_ssl_verification(): diff --git a/tests/unit/utils/test_conf_path.py b/tests/unit/utils/test_conf_path.py index 7098b0c..183e614 100644 --- a/tests/unit/utils/test_conf_path.py +++ b/tests/unit/utils/test_conf_path.py @@ -39,7 +39,7 @@ def test_env_var_invalid_directory_falls_back_parent(monkeypatch): conf_dir, invalid = conf_path_module.resolve_conf_dir() # Should fall back to repository conf directory /conf expected_root_conf = (Path(conf_path_module.__file__).resolve().parent.parent.parent / "conf").resolve() - assert Path(conf_dir).resolve() == expected_root_conf + assert expected_root_conf == Path(conf_dir).resolve() assert invalid == os.path.abspath(missing_path) @@ -78,7 +78,8 @@ def build(base: Path, code: str): mod = _load_isolated_conf_path(build) try: conf_dir, invalid = mod.resolve_conf_dir() - assert conf_dir.endswith("pkg/conf") # current directory conf chosen + expected_current_conf = (Path(mod.__file__).resolve().parent / "conf").resolve() + assert expected_current_conf == Path(conf_dir).resolve() # current directory conf chosen assert invalid is None finally: mod._tmp.cleanup() # type: ignore[attr-defined] @@ -97,7 +98,7 @@ def build(base: Path, code: str): conf_dir, invalid = mod.resolve_conf_dir() # Parent conf path returned even though it does not exist expected_root_conf = (Path(mod.__file__).resolve().parent.parent.parent / "conf").resolve() - assert Path(conf_dir).resolve() == expected_root_conf + assert expected_root_conf == Path(conf_dir).resolve() assert invalid is None finally: mod._tmp.cleanup() # type: ignore[attr-defined] @@ -118,7 +119,8 @@ def build(base: Path, code: str): bad_path = "/definitely/not/there/abc123" monkeypatch.setenv("CONF_DIR", bad_path) conf_dir, invalid = mod.resolve_conf_dir() - assert conf_dir.endswith("pkg_invalid_current/conf") + expected_current_conf = (Path(mod.__file__).resolve().parent / "conf").resolve() + assert expected_current_conf == Path(conf_dir).resolve() assert invalid == os.path.abspath(bad_path) finally: mod._tmp.cleanup() # type: ignore[attr-defined] @@ -139,7 +141,7 @@ def build(base: Path, code: str): monkeypatch.setenv("CONF_DIR", bad_path) conf_dir, invalid = mod.resolve_conf_dir() expected_root_conf = (Path(mod.__file__).resolve().parent.parent.parent / "conf").resolve() - assert Path(conf_dir).resolve() == expected_root_conf + assert expected_root_conf == Path(conf_dir).resolve() assert invalid == os.path.abspath(bad_path) finally: mod._tmp.cleanup() # type: ignore[attr-defined] diff --git a/tests/unit/utils/test_logging_levels.py b/tests/unit/utils/test_logging_levels.py new file mode 100644 index 0000000..59672b1 --- /dev/null +++ b/tests/unit/utils/test_logging_levels.py @@ -0,0 +1,69 @@ +# +# Copyright 2026 ABSA Group Limited +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +import logging + +import pytest +from aws_lambda_powertools import Logger + +from src.utils.logging_levels import TRACE_LEVEL, configured_log_level, resolve_log_level + + +@pytest.fixture(autouse=True) +def clear_log_level_env(monkeypatch): + """Start every test from an unset log level configuration.""" + monkeypatch.delenv("LOG_LEVEL", raising=False) + monkeypatch.delenv("POWERTOOLS_LOG_LEVEL", raising=False) + + +## configured_log_level() +def test_configured_log_level_defaults_to_info(): + assert "INFO" == configured_log_level() + + +def test_powertools_log_level_takes_precedence(monkeypatch): + monkeypatch.setenv("LOG_LEVEL", "DEBUG") + monkeypatch.setenv("POWERTOOLS_LOG_LEVEL", "warning") + + assert "WARNING" == configured_log_level() + + +## resolve_log_level() +@pytest.mark.parametrize( + "configured,expected", + [("TRACE", TRACE_LEVEL), ("debug", logging.DEBUG), ("INFO", logging.INFO), ("ERROR", logging.ERROR)], +) +def test_resolve_log_level_known_levels(monkeypatch, configured, expected): + monkeypatch.setenv("LOG_LEVEL", configured) + + assert (expected, None) == resolve_log_level() + + +def test_resolve_log_level_falls_back_to_info(monkeypatch): + monkeypatch.setenv("LOG_LEVEL", "NOT_A_LEVEL") + + assert (logging.INFO, "NOT_A_LEVEL") == resolve_log_level() + + +## Powertools compatibility +def test_powertools_logger_accepts_the_custom_trace_level(monkeypatch, capsys): + monkeypatch.setenv("LOG_LEVEL", "TRACE") + + powertools_logger = Logger(service="eventgate-trace-check", level=resolve_log_level()[0]) + powertools_logger.trace("Payload.") + + assert TRACE_LEVEL == powertools_logger.log_level + assert '"level":"TRACE"' in capsys.readouterr().out diff --git a/tests/unit/utils/test_observability.py b/tests/unit/utils/test_observability.py new file mode 100644 index 0000000..a4baab0 --- /dev/null +++ b/tests/unit/utils/test_observability.py @@ -0,0 +1,262 @@ +# +# Copyright 2026 ABSA Group Limited +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +import json +import logging + +import pytest + +from src.utils import observability +from src.utils.logging_levels import TRACE_LEVEL +from src.utils.observability import ( + append_request_context, + bind_request_context, + configure_root_logging, + logger, + resolve_correlation_id, +) + + +class FakeLambdaContext: + function_name = "eventgate" + aws_request_id = "aws-request-id-1" + memory_limit_in_mb = 512 + invoked_function_arn = "arn:aws:lambda:eu-west-1:000000000000:function:eventgate" + + +def formatted(record): + """Render a captured record the way the lambda emits it, so persistent keys are visible.""" + return json.loads(logger.registered_formatter.format(record)) + + +@pytest.fixture(autouse=True) +def clean_logger_state(): + """Keep the module level logger state from leaking between tests.""" + observability._container_state["cold_start"] = True + yield + bind_request_context({}) + logger.set_correlation_id(None) + + +## resolve_correlation_id() +def test_correlation_id_from_correlation_header(): + event = {"headers": {"X-Correlation-ID": "run-42_a.b:c"}} + + assert "run-42_a.b:c" == resolve_correlation_id(event) + + +def test_correlation_id_header_lookup_is_case_insensitive(): + event = {"headers": {"x-CORRELATION-id": " run-42 "}} + + assert "run-42" == resolve_correlation_id(event) + + +def test_correlation_id_from_request_id_header(): + event = {"headers": {"X-Request-ID": "req-7"}} + + assert "req-7" == resolve_correlation_id(event) + + +@pytest.mark.parametrize( + "value", + ["with space", "line\nbreak", "", "!@#$", "x" * 129], +) +def test_correlation_id_rejects_malformed_header(value): + event = {"headers": {"X-Correlation-ID": value}, "requestContext": {"requestId": "apigw-1"}} + + assert "apigw-1" == resolve_correlation_id(event) + + +def test_correlation_id_falls_back_to_api_gateway_request_id(): + event = {"requestContext": {"requestId": "apigw-1"}} + + assert "apigw-1" == resolve_correlation_id(event) + + +def test_correlation_id_empty_when_unavailable(): + assert "" == resolve_correlation_id({}) + + +## bind_request_context() +def test_bind_request_context_binds_correlation_id_and_request_keys(caplog): + caplog.set_level(logging.INFO) + event = {"headers": {"X-Correlation-ID": "run-42"}, "resource": "/topics/{topic_name}", "httpMethod": "POST"} + + correlation_id = bind_request_context(event, FakeLambdaContext()) + logging.getLogger("src.handlers.handler_topic").info("Message accepted.") + + payload = formatted(caplog.records[-1]) + assert "run-42" == correlation_id + assert "run-42" == payload["correlation_id"] + assert "/topics/{topic_name}" == payload["resource"] + assert "POST" == payload["http_method"] + assert "aws-request-id-1" == payload["function_request_id"] + assert payload["cold_start"] is True + + +def test_bind_request_context_reports_cold_start_once(caplog): + caplog.set_level(logging.INFO) + + bind_request_context({}, FakeLambdaContext()) + logging.getLogger("src.utils.utils").info("Request completed.") + first = formatted(caplog.records[-1]) + + bind_request_context({}, FakeLambdaContext()) + logging.getLogger("src.utils.utils").info("Request completed.") + second = formatted(caplog.records[-1]) + + assert first["cold_start"] is True + assert second["cold_start"] is False + + +def test_lambda_execution_context_is_logged_once_and_not_bound_per_line(caplog): + caplog.set_level(logging.INFO) + + bind_request_context({}, FakeLambdaContext()) + logging.getLogger("src.utils.utils").info("Request completed.") + + context_lines = [r for r in caplog.records if r.message == "Lambda execution context."] + assert 1 == len(context_lines) + assert "eventgate" == context_lines[0].function_name + assert 512 == context_lines[0].function_memory_size + assert FakeLambdaContext.invoked_function_arn == context_lines[0].function_arn + + payload = formatted(caplog.records[-1]) + assert "aws-request-id-1" == payload["function_request_id"] + assert "function_name" not in payload + assert "function_arn" not in payload + assert "function_memory_size" not in payload + + bind_request_context({}, FakeLambdaContext()) + assert 1 == len([r for r in caplog.records if r.message == "Lambda execution context."]) + + +def test_bind_request_context_tolerates_missing_lambda_context(caplog): + caplog.set_level(logging.INFO) + + bind_request_context({"resource": "/health"}, None) + logging.getLogger("src.utils.utils").info("Request completed.") + + payload = formatted(caplog.records[-1]) + assert "/health" == payload["resource"] + assert "function_request_id" not in payload + + +def test_request_log_keys_do_not_leak_into_the_next_invocation(caplog): + caplog.set_level(logging.INFO) + + bind_request_context({"resource": "/topics/{topic_name}"}, None) + append_request_context(topic="test", user="someone") + logging.getLogger("src.handlers.handler_topic").info("Message accepted.") + assert "test" == formatted(caplog.records[-1])["topic"] + + bind_request_context({"resource": "/health"}, None) + logging.getLogger("src.handlers.handler_health").info("Health check passed.") + + payload = formatted(caplog.records[-1]) + assert "topic" not in payload + assert "user" not in payload + + +## configure_root_logging() +def test_configure_root_logging_routes_root_through_powertools_handler(): + configure_root_logging() + + root_logger = logging.getLogger() + assert [logger.registered_handler] == root_logger.handlers + + +def test_configure_root_logging_caps_noisy_loggers(monkeypatch): + monkeypatch.setenv("LOG_LEVEL", "TRACE") + monkeypatch.delenv("POWERTOOLS_LOG_LEVEL", raising=False) + + configure_root_logging() + + assert TRACE_LEVEL == logging.getLogger().level + for noisy_logger in observability.NOISY_LOGGERS: + assert logging.WARNING == logging.getLogger(noisy_logger).level + + +## refresh_sampled_log_level() +@pytest.fixture +def restore_sampling_rate(): + """Restore the sampling rate the module level logger was constructed with.""" + original_rate = logger.sampling_rate + yield + logger.sampling_rate = original_rate + + +def test_unsampled_request_keeps_the_configured_level(monkeypatch, restore_sampling_rate): + monkeypatch.setenv("LOG_LEVEL", "INFO") + monkeypatch.delenv("POWERTOOLS_LOG_LEVEL", raising=False) + configure_root_logging() + logger.sampling_rate = 0 + + assert logging.INFO == observability.refresh_sampled_log_level() + assert logging.INFO == logging.getLogger().level + + +def test_sampled_request_lowers_the_root_level_to_debug(monkeypatch, restore_sampling_rate): + monkeypatch.setenv("LOG_LEVEL", "INFO") + monkeypatch.delenv("POWERTOOLS_LOG_LEVEL", raising=False) + configure_root_logging() + logger.sampling_rate = 1 + + # Module loggers resolve their level from the root logger, so the sampled outcome only + # reaches `src.*` records if it is mirrored there. + assert logging.DEBUG == observability.refresh_sampled_log_level() + assert logging.DEBUG == logging.getLogger().level + assert logging.getLogger("src.writers.writer_kafka").isEnabledFor(logging.DEBUG) + + +def test_sampling_never_raises_the_level_of_a_trace_run(monkeypatch, restore_sampling_rate): + monkeypatch.setenv("LOG_LEVEL", "TRACE") + monkeypatch.delenv("POWERTOOLS_LOG_LEVEL", raising=False) + configure_root_logging() + logger.sampling_rate = 1 + + # Powertools sets DEBUG unconditionally when a request is sampled; sampling may only add + # detail, so a TRACE run must not be pulled up to DEBUG. + assert TRACE_LEVEL == observability.refresh_sampled_log_level() + assert TRACE_LEVEL == logging.getLogger().level + + +def test_sampled_request_keeps_noisy_loggers_capped(monkeypatch, restore_sampling_rate): + monkeypatch.setenv("LOG_LEVEL", "INFO") + monkeypatch.delenv("POWERTOOLS_LOG_LEVEL", raising=False) + configure_root_logging() + logger.sampling_rate = 1 + + observability.refresh_sampled_log_level() + + for noisy_logger in observability.NOISY_LOGGERS: + assert logging.WARNING == logging.getLogger(noisy_logger).level + + +def test_bind_request_context_redraws_the_sample_per_invocation(monkeypatch, restore_sampling_rate): + monkeypatch.setenv("LOG_LEVEL", "INFO") + monkeypatch.delenv("POWERTOOLS_LOG_LEVEL", raising=False) + configure_root_logging() + + logger.sampling_rate = 1 + bind_request_context({"resource": "/topics/{topic_name}"}, None) + assert logging.DEBUG == logging.getLogger().level + + # The draw is per request, not per container: the next invocation of a warm container + # re-evaluates it instead of inheriting the previous outcome. + logger.sampling_rate = 0 + bind_request_context({"resource": "/topics/{topic_name}"}, None) + assert logging.INFO == logging.getLogger().level diff --git a/tests/unit/utils/test_trace_logging.py b/tests/unit/utils/test_trace_logging.py index 8ac7b00..d4d5ca9 100644 --- a/tests/unit/utils/test_trace_logging.py +++ b/tests/unit/utils/test_trace_logging.py @@ -29,7 +29,7 @@ def test_log_payload_skipped_when_trace_not_enabled(): logger = MagicMock() logger.isEnabledFor.return_value = False - log_payload_at_trace(logger, "TestWriter", "test.topic", {"key": "value"}) + log_payload_at_trace(logger, "TestWriter", {"key": "value"}) logger.isEnabledFor.assert_called_once_with(TRACE_LEVEL) logger.trace.assert_not_called() @@ -47,7 +47,9 @@ def test_trace_eventbridge(caplog): writer.write("topic.eb", {"k": 1}) - assert any("EventBridge payload" in rec.message for rec in caplog.records) + assert any( + "Writer payload." == rec.message and "EventBridge" == getattr(rec, "writer", None) for rec in caplog.records + ) def test_trace_kafka(caplog): @@ -69,7 +71,7 @@ def flush(self, timeout=None): writer.write("topic.kf", {"k": 2}) - assert any("Kafka payload" in rec.message for rec in caplog.records) + assert any("Writer payload." == rec.message and "Kafka" == getattr(rec, "writer", None) for rec in caplog.records) def test_trace_postgres(caplog, monkeypatch): @@ -117,4 +119,6 @@ def connect(self, **kwargs): message = {"event_id": "e", "tenant_id": "t", "source_app": "a", "environment": "dev", "timestamp": 1} writer.write("public.cps.za.test", message) - assert any("Postgres payload" in rec.message for rec in caplog.records) + assert any( + "Writer payload." == rec.message and "Postgres" == getattr(rec, "writer", None) for rec in caplog.records + ) diff --git a/tests/unit/utils/test_utils.py b/tests/unit/utils/test_utils.py index cc59bf4..5e8d7d9 100644 --- a/tests/unit/utils/test_utils.py +++ b/tests/unit/utils/test_utils.py @@ -15,8 +15,32 @@ # import json +import logging -from src.utils.utils import build_error_response +import pytest + +from src.utils.observability import CORRELATION_ID_RESPONSE_HEADER, bind_request_context +from src.utils.observability import logger as powertools_logger +from src.utils.utils import build_error_response, dispatch_request, resolve_request_topic + +logger = logging.getLogger(__name__) + + +def make_event(resource="/health", method="GET", correlation_id="run-42"): + """Build a minimal API Gateway proxy event.""" + return { + "resource": resource, + "httpMethod": method, + "headers": {"X-Correlation-ID": correlation_id}, + } + + +@pytest.fixture(autouse=True) +def clean_logger_state(): + """`append_request_context()` mutates module level logger state; drop it after every test.""" + yield + bind_request_context({}) + powertools_logger.set_correlation_id(None) ## build_error_response() @@ -33,3 +57,96 @@ def test_build_error_response_structure(): assert 1 == len(body["errors"]) assert "topic" == body["errors"][0]["type"] assert "Topic not found" == body["errors"][0]["message"] + + +## resolve_request_topic() +def test_resolve_request_topic_normalizes_and_binds_the_topic(caplog): + caplog.set_level(logging.INFO) + + topic_name, error = resolve_request_topic({"pathParameters": {"topic_name": "Public.CPS.ZA.Test"}}) + + assert "public.cps.za.test" == topic_name + assert error is None + + logger.info("Probe.") + payload = json.loads(powertools_logger.registered_formatter.format(caplog.records[-1])) + assert "public.cps.za.test" == payload["topic"] + + +@pytest.mark.parametrize( + "event", [{}, {"pathParameters": None}, {"pathParameters": {}}, {"pathParameters": {"topic_name": ""}}] +) +def test_resolve_request_topic_rejects_a_missing_path_parameter(caplog, event): + caplog.set_level(logging.WARNING) + + topic_name, error = resolve_request_topic(event) + + assert "" == topic_name + assert error is not None + assert 400 == error["statusCode"] + assert "validation" == json.loads(error["body"])["errors"][0]["type"] + assert any( + record.message == "Request rejected: path parameter 'topic_name' is missing." for record in caplog.records + ) + + +## dispatch_request() +def test_dispatch_request_routes_and_stamps_the_correlation_id(): + route_map = {"/health": lambda _: {"statusCode": 200, "headers": {"Content-Type": "application/json"}}} + + resp = dispatch_request(make_event(), route_map, logger) + + assert 200 == resp["statusCode"] + assert "run-42" == resp["headers"][CORRELATION_ID_RESPONSE_HEADER] + + +def test_dispatch_request_overwrites_a_handler_supplied_correlation_id(): + """The id bound to the request wins over one a handler put on its own response.""" + route_map = {"/health": lambda _: {"statusCode": 200, "headers": {CORRELATION_ID_RESPONSE_HEADER: "stale-id"}}} + + resp = dispatch_request(make_event(), route_map, logger) + + assert "run-42" == resp["headers"][CORRELATION_ID_RESPONSE_HEADER] + + +def test_dispatch_request_logs_the_request_outcome(caplog): + caplog.set_level(logging.INFO) + route_map = {"/health": lambda _: {"statusCode": 503, "headers": {}}} + + dispatch_request(make_event(), route_map, logger) + + completed = [record for record in caplog.records if record.message == "Request completed."] + assert 1 == len(completed) + assert 503 == completed[0].status_code + assert completed[0].duration_ms >= 0 + + +def test_dispatch_request_warns_on_unknown_route(caplog): + caplog.set_level(logging.WARNING) + + resp = dispatch_request(make_event(resource="/unknown"), {}, logger) + + assert 404 == resp["statusCode"] + assert any(record.message == "No route matched the requested resource." for record in caplog.records) + + +@pytest.mark.parametrize("error", [RuntimeError("boom"), OSError("connection reset"), IndexError("out of range")]) +def test_dispatch_request_converts_any_handler_error_into_a_logged_500(caplog, error): + caplog.set_level(logging.ERROR) + + def failing_route(_event): + raise error + + resp = dispatch_request(make_event(), {"/health": failing_route}, logger) + + assert 500 == resp["statusCode"] + assert "internal" == json.loads(resp["body"])["errors"][0]["type"] + assert "run-42" == resp["headers"][CORRELATION_ID_RESPONSE_HEADER] + assert any(record.message == "Unhandled error while processing the request." for record in caplog.records) + + +def test_dispatch_request_does_not_swallow_system_exit(): + route_map = {"/terminate": lambda _: (_ for _ in ()).throw(SystemExit("TERMINATING"))} + + with pytest.raises(SystemExit): + dispatch_request(make_event(resource="/terminate"), route_map, logger) diff --git a/tests/unit/writers/test_writer_kafka.py b/tests/unit/writers/test_writer_kafka.py index bce2baf..7a95d60 100644 --- a/tests/unit/writers/test_writer_kafka.py +++ b/tests/unit/writers/test_writer_kafka.py @@ -140,8 +140,8 @@ def test_write_flush_retries_until_success(monkeypatch, caplog): # It should break as soon as remaining == 0 (after flush call returning 0) assert producer.flush_calls == 5 # sequence consumed until 0 # Warnings logged for attempts before success (flush_calls -1) because last attempt didn't warn - warn_messages = [r.message for r in caplog.records if r.levelno == logging.WARNING] - assert any("attempt 1" in m or "attempt 2" in m for m in warn_messages) + warn_records = [r for r in caplog.records if r.levelno == logging.WARNING] + assert [1, 2, 3, 4] == [r.attempt for r in warn_records] def test_write_timeout_warning_when_remaining_after_retries(monkeypatch, caplog): @@ -152,9 +152,10 @@ def test_write_timeout_warning_when_remaining_after_retries(monkeypatch, caplog) writer._producer = producer writer.write("topic", {"f": 6}) timeout_warnings = [ - r.message for r in caplog.records if "timeout" in r.message - ] # final warning should mention timeout + r for r in caplog.records if r.message == "Kafka flush timed out with messages still pending." + ] # final warning reports the pending messages assert timeout_warnings, "Expected timeout warning logged" + assert 2 == timeout_warnings[0].pending_messages assert producer.flush_calls == 3 # retried 3 times diff --git a/tests/unit/writers/test_writer_postgres.py b/tests/unit/writers/test_writer_postgres.py index c7ae19d..dc6aee2 100644 --- a/tests/unit/writers/test_writer_postgres.py +++ b/tests/unit/writers/test_writer_postgres.py @@ -228,6 +228,19 @@ def test_write_unknown_topic_skips_silently(reset_env, monkeypatch): assert 0 == len(store) +def test_write_unknown_topic_ignores_broken_postgres_configuration(reset_env, monkeypatch): + """A topic Postgres does not persist must not fail because Postgres itself is misconfigured.""" + writer = WriterPostgres({}) + monkeypatch.setattr( + type(writer), + "_pg_config", + property(lambda self: (_ for _ in ()).throw(RuntimeError("secret unavailable"))), + ) + monkeypatch.setattr(pb, "psycopg2", None) + + writer.write("public.cps.za.unknown", {}) + + def test_write_success_known_topic(reset_env, monkeypatch): store = [] monkeypatch.setattr(pb, "psycopg2", DummyPsycopg(store))