Coverage for gco / services / structured_logging.py: 100.00%
40 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-10-28 16:19 +0000
1"""
2Structured JSON logging for GCO services.
4Provides a JSON log formatter that outputs structured logs compatible with
5CloudWatch Logs Insights queries. Use ``configure_structured_logging()``
6at service startup to enable JSON output for all loggers.
8Example CloudWatch Insights query:
9 fields @timestamp, level, message, cluster_id, region
10 | filter level = "ERROR"
11 | sort @timestamp desc
13While tracing is active (see :mod:`gco.services.tracing`), a line logged
14inside a traced request also carries ``trace_id``, ``span_id`` and
15``trace_sampled`` -- the ids the request's spans carry in ``aws/spans`` --
16so one trace's log lines can be pulled from the service's log group:
17 fields @timestamp, message
18 | filter trace_id = "<32 hex digits>"
20Environment Variables:
21 LOG_FORMAT: "json" for structured logging, "text" for human-readable (default: json)
22 LOG_LEVEL: Logging level (default: INFO)
23"""
25import json
26import logging
27import os
28import traceback
29from datetime import UTC, datetime
30from typing import Any
32from gco.services import tracing
35class StructuredJsonFormatter(logging.Formatter):
36 """
37 Formats log records as single-line JSON objects.
39 Each log line includes:
40 - timestamp (ISO 8601)
41 - level
42 - logger (logger name)
43 - message
44 - The formatter's default fields (e.g., cluster_id, region)
45 - trace_id, span_id and trace_sampled when a traced span is current
46 - Any extra fields passed via the ``extra`` dict
48 Exceptions are serialized into an ``exception`` field with type,
49 message, and traceback.
50 """
52 def __init__(self, service_name: str = "gco", **default_fields: Any):
53 super().__init__()
54 self.service_name = service_name
55 self.default_fields = default_fields
57 def format(self, record: logging.LogRecord) -> str:
58 log_entry: dict[str, Any] = {
59 "timestamp": datetime.now(UTC).isoformat(),
60 "level": record.levelname,
61 "logger": record.name,
62 "message": record.getMessage(),
63 "service": self.service_name,
64 }
66 # Include default fields (e.g., cluster_id, region)
67 log_entry.update(self.default_fields)
69 # Correlate with the current span; empty (and cheap) while tracing is off
70 log_entry.update(tracing.current_trace_fields())
72 # Include any extra fields passed via logger.info("msg", extra={...})
73 # Filter out standard LogRecord attributes
74 standard_attrs = {
75 "name",
76 "msg",
77 "args",
78 "created",
79 "relativeCreated",
80 "exc_info",
81 "exc_text",
82 "stack_info",
83 "lineno",
84 "funcName",
85 "pathname",
86 "filename",
87 "module",
88 "thread",
89 "threadName",
90 "process",
91 "processName",
92 "levelname",
93 "levelno",
94 "msecs",
95 "message",
96 "taskName",
97 }
98 for key, value in record.__dict__.items():
99 if key not in standard_attrs and not key.startswith("_"):
100 log_entry[key] = value
102 # Serialize exception info
103 if record.exc_info and record.exc_info[1]:
104 exc_type, exc_value, exc_tb = record.exc_info
105 log_entry["exception"] = {
106 "type": exc_type.__name__ if exc_type else "Unknown",
107 "message": str(exc_value),
108 "traceback": traceback.format_exception(exc_type, exc_value, exc_tb),
109 }
111 return json.dumps(log_entry, default=str)
114def configure_structured_logging(
115 service_name: str = "gco",
116 level: str | None = None,
117 **default_fields: Any,
118) -> None:
119 """
120 Configure the root logger with structured JSON output.
122 Call this once at service startup. All existing loggers will inherit
123 the JSON formatter.
125 Args:
126 service_name: Name included in every log line (e.g., "health-monitor").
127 level: Log level override. Defaults to LOG_LEVEL env var or INFO.
128 **default_fields: Extra fields included in every log line
129 (e.g., cluster_id="...", region="us-east-1").
130 """
131 log_format = os.environ.get("LOG_FORMAT", "json").lower()
132 log_level = level or os.environ.get("LOG_LEVEL") or "INFO"
134 root_logger = logging.getLogger()
135 root_logger.setLevel(getattr(logging, log_level.upper(), logging.INFO))
137 # Remove existing handlers to avoid duplicate output
138 root_logger.handlers.clear()
140 handler = logging.StreamHandler()
142 if log_format == "json":
143 handler.setFormatter(StructuredJsonFormatter(service_name=service_name, **default_fields))
144 else:
145 # Human-readable fallback for local development
146 handler.setFormatter(
147 logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
148 )
150 root_logger.addHandler(handler)
153#: Unicode characters some log viewers and parsers render as line breaks.
154#: Neutralized alongside CR/LF so they cannot forge log entries either:
155#: U+0085 NEL, U+2028 LINE SEPARATOR, U+2029 PARAGRAPH SEPARATOR.
156_UNICODE_LINE_BREAKS = "\u0085\u2028\u2029"
159def sanitize_log_value(value: Any) -> str:
160 """
161 Sanitize a value for safe inclusion in log messages.
163 Neutralizes CR/LF (rendered as the literal two-character sequences
164 ``\\r``/``\\n``), Unicode line separators, and other control characters
165 that could be used for log injection (log forging), where an attacker
166 embeds line breaks or terminal escapes in user input to spoof log
167 entries. Printable text — including non-ASCII — passes through unchanged.
169 Always call this on untrusted inputs (HTTP path/query params, job IDs,
170 namespaces, etc.) before passing them to ``logger.info/warning/error``.
172 Args:
173 value: Any value; will be coerced to ``str``.
175 Returns:
176 A single-line-safe string suitable for plain-text log sinks.
177 """
178 text = str(value)
179 text = text.replace("\r", "\\r").replace("\n", "\\n")
180 return "".join(
181 ch if ord(ch) >= 32 and ord(ch) != 127 and ch not in _UNICODE_LINE_BREAKS else "?"
182 for ch in text
183 )