agent_search_gateway.observability

  1"""Secret-safe structured observability primitives."""
  2
  3import json
  4import logging
  5import re
  6from collections.abc import Callable, Iterable
  7from contextlib import suppress
  8from logging.handlers import RotatingFileHandler
  9from pathlib import Path
 10from typing import TextIO
 11from urllib.parse import urlsplit, urlunsplit
 12
 13from .errors import ConfigFailure, ErrorCode
 14from .request_ids import current_request_id
 15
 16LOG_MAX_BYTES = 5 * 1024 * 1024
 17LOG_BACKUP_COUNT = 3
 18
 19FileHandlerFactory = Callable[..., logging.Handler]
 20
 21_FIELD_NAME = re.compile(r"[a-z][a-z0-9_]*\Z")
 22_PREFERRED_FIELD_ORDER = ("provider", "stage", "url", "event")
 23_EMERGENCY_MESSAGE = "agent-search-gateway debug logging sink failure\n"
 24
 25
 26class SecretValue:
 27    """A secret whose implicit string representations are always redacted."""
 28
 29    __slots__ = ("_value",)
 30
 31    def __init__(self, value: str) -> None:
 32        self._value = value
 33
 34    def reveal(self) -> str:
 35        return self._value
 36
 37    def __repr__(self) -> str:
 38        return "<redacted>"
 39
 40    def __str__(self) -> str:
 41        return "<redacted>"
 42
 43
 44class SecretRedactor:
 45    """Redact configured secrets from fully rendered logging output."""
 46
 47    def __init__(self, secrets: Iterable[SecretValue] = ()) -> None:
 48        self._secrets: set[str] = set()
 49        self._ordered_secrets: tuple[str, ...] = ()
 50        self.add_secrets(secrets)
 51
 52    def add_secrets(self, secrets: Iterable[SecretValue]) -> None:
 53        for secret in secrets:
 54            value = secret.reveal()
 55            if not value:
 56                continue
 57            self._secrets.add(value)
 58            escaped = json.dumps(value, ensure_ascii=False)[1:-1]
 59            if escaped:
 60                self._secrets.add(escaped)
 61        self._ordered_secrets = tuple(sorted(self._secrets, key=len, reverse=True))
 62
 63    def redact(self, rendered: str) -> str:
 64        for secret in self._ordered_secrets:
 65            rendered = rendered.replace(secret, "<redacted>")
 66        return rendered
 67
 68
 69class SecretRedactingFilter(logging.Filter):
 70    """Compatibility filter for existing provider tests and ad-hoc handlers."""
 71
 72    def __init__(self, secrets: Iterable[SecretValue]) -> None:
 73        super().__init__()
 74        self._redactor = SecretRedactor(secrets)
 75
 76    def filter(self, record: logging.LogRecord) -> bool:
 77        record.msg = self._redactor.redact(record.getMessage())
 78        record.args = ()
 79        return True
 80
 81
 82def normalize_log_reason(value: str, max_chars: int = 160) -> str:
 83    if max_chars <= 0:
 84        raise ValueError("max_chars must be positive")
 85    normalized = " ".join(value.split())
 86    if len(normalized) <= max_chars:
 87        return normalized
 88    if max_chars == 1:
 89        return "…"
 90    return f"{normalized[: max_chars - 1]}…"
 91
 92
 93def elapsed_ms(monotonic: Callable[[], float], started: float) -> int:
 94    """Convert a monotonic interval to a non-negative whole-millisecond duration."""
 95
 96    return max(0, int((monotonic() - started) * 1000))
 97
 98
 99def target_url_for_log(value: str) -> str:
100    """Strip URI userinfo while retaining target path/query/fragment diagnostics."""
101
102    return _url_for_log(value, keep_query=True, keep_fragment=True)
103
104
105def http_endpoint_for_log(value: str) -> str:
106    """Render an HTTP endpoint without userinfo or request-specific query data."""
107
108    return _url_for_log(value, keep_query=False, keep_fragment=False)
109
110
111def _url_for_log(value: str, *, keep_query: bool, keep_fragment: bool) -> str:
112    try:
113        parsed = urlsplit(value)
114    except (ValueError, UnicodeError):
115        return "<invalid-url>"
116    if parsed.scheme not in {"http", "https"} or not parsed.netloc:
117        return "<invalid-url>"
118    _, separator, hostport = parsed.netloc.rpartition("@")
119    netloc = hostport if separator else parsed.netloc
120    return urlunsplit(
121        (
122            parsed.scheme,
123            netloc,
124            parsed.path,
125            parsed.query if keep_query else "",
126            parsed.fragment if keep_fragment else "",
127        )
128    )
129
130
131def _render_string(value: str) -> str:
132    if (
133        value
134        and value.isprintable()
135        and not any(character.isspace() or character in {'"', "\\"} for character in value)
136    ):
137        return value
138    return json.dumps(value, ensure_ascii=False, separators=(",", ":"))
139
140
141def _render_scalar(value: object) -> str:
142    if value is None:
143        return "null"
144    if isinstance(value, bool):
145        return "true" if value else "false"
146    if isinstance(value, str):
147        return _render_string(value)
148    if isinstance(value, (int, float)):
149        return repr(value)
150    return _render_string(f"<{type(value).__name__}>")
151
152
153def _validated_field_name(name: str) -> str:
154    if _FIELD_NAME.fullmatch(name) is None:
155        raise ValueError(f"invalid log field name: {name}")
156    return name
157
158
159class KeyValueFormatter(logging.Formatter):
160    """Render structured events as deterministic, single physical lines."""
161
162    def __init__(self, redactor: SecretRedactor) -> None:
163        super().__init__()
164        self._redactor = redactor
165
166    def format(self, record: logging.LogRecord) -> str:
167        event = getattr(record, "gateway_event", None)
168        raw_fields = getattr(record, "gateway_fields", None)
169        if isinstance(event, str) and isinstance(raw_fields, dict):
170            fields = dict(raw_fields)
171        else:
172            event = "log"
173            fields = {"message": record.getMessage()}
174
175        values: dict[str, object] = {
176            _validated_field_name(str(key)): value for key, value in fields.items()
177        }
178        values["event"] = event
179        request_id = current_request_id() or "-"
180        parts = [record.levelname, f"request={request_id}"]
181        emitted: set[str] = set()
182        for name in _PREFERRED_FIELD_ORDER:
183            if name not in values:
184                continue
185            parts.append(f"{name}={_render_scalar(values[name])}")
186            emitted.add(name)
187        for name in sorted(values.keys() - emitted):
188            parts.append(f"{name}={_render_scalar(values[name])}")
189
190        if record.exc_info:
191            traceback_text = self.formatException(record.exc_info)
192            parts.append(f"traceback={_render_string(traceback_text)}")
193        return self._redactor.redact(" ".join(parts))
194
195
196def log_event(
197    logger: logging.Logger,
198    level: int,
199    event: str,
200    *,
201    exc_info: bool | BaseException = False,
202    **fields: object,
203) -> None:
204    if not logger.isEnabledFor(level):
205        return
206    _validated_field_name(event)
207    for name in fields:
208        _validated_field_name(name)
209    logger.log(
210        level,
211        "",
212        exc_info=exc_info,
213        extra={"gateway_event": event, "gateway_fields": dict(fields)},
214    )
215
216
217def _write_emergency(stderr: TextIO) -> None:
218    try:
219        stderr.write(_EMERGENCY_MESSAGE)
220        stderr.flush()
221    except Exception:
222        pass
223
224
225class SafeRotatingFileHandler(RotatingFileHandler):
226    """Rotating handler whose sink errors never escape into business code."""
227
228    def __init__(
229        self,
230        filename: Path,
231        *,
232        stderr: TextIO,
233        max_bytes: int,
234        backup_count: int,
235    ) -> None:
236        self._emergency_stderr = stderr
237        super().__init__(
238            filename,
239            mode="a",
240            maxBytes=max_bytes,
241            backupCount=backup_count,
242            encoding="utf-8",
243        )
244
245    def handleError(self, record: logging.LogRecord) -> None:  # noqa: N802
246        _write_emergency(self._emergency_stderr)
247
248
249class _SafeDelegatingHandler(logging.Handler):
250    """Contain exceptions from injected handlers used by failure-path tests."""
251
252    def __init__(self, delegate: logging.Handler, stderr: TextIO) -> None:
253        super().__init__(delegate.level)
254        self._delegate = delegate
255        self._stderr = stderr
256
257    def emit(self, record: logging.LogRecord) -> None:
258        try:
259            self._delegate.handle(record)
260        except Exception:
261            _write_emergency(self._stderr)
262
263    def close(self) -> None:
264        with suppress(Exception):
265            self._delegate.close()
266        super().close()
267
268
269class DebugLoggingSession:
270    """Own one debug logging configuration and restore prior logger state."""
271
272    def __init__(
273        self,
274        logger: logging.Logger,
275        handlers: tuple[logging.Handler, ...],
276        redactor: SecretRedactor,
277        *,
278        previous_level: int,
279        previous_propagate: bool,
280    ) -> None:
281        self._logger = logger
282        self._handlers = handlers
283        self._redactor = redactor
284        self._previous_level = previous_level
285        self._previous_propagate = previous_propagate
286        self._closed = False
287
288    def add_secrets(self, *secrets: SecretValue | Iterable[SecretValue]) -> None:
289        collected: list[SecretValue] = []
290        for item in secrets:
291            if isinstance(item, SecretValue):
292                collected.append(item)
293            else:
294                collected.extend(item)
295        self._redactor.add_secrets(collected)
296
297    def close(self) -> None:
298        if self._closed:
299            return
300        self._closed = True
301        for handler in self._handlers:
302            self._logger.removeHandler(handler)
303            with suppress(Exception):
304                handler.flush()
305            with suppress(Exception):
306                handler.close()
307        self._logger.setLevel(self._previous_level)
308        self._logger.propagate = self._previous_propagate
309
310
311def _default_file_handler(
312    log_file: Path,
313    *,
314    stderr: TextIO,
315    max_bytes: int,
316    backup_count: int,
317) -> logging.Handler:
318    return SafeRotatingFileHandler(
319        log_file,
320        stderr=stderr,
321        max_bytes=max_bytes,
322        backup_count=backup_count,
323    )
324
325
326def configure_debug_logging(
327    log_file: Path,
328    *,
329    stderr: TextIO,
330    max_bytes: int = LOG_MAX_BYTES,
331    backup_count: int = LOG_BACKUP_COUNT,
332    file_handler_factory: FileHandlerFactory | None = None,
333) -> DebugLoggingSession:
334    """Install project-only DEBUG handlers with rotation and final redaction."""
335
336    logger = logging.getLogger("agent_search_gateway")
337    previous_level = logger.level
338    previous_propagate = logger.propagate
339    handlers: list[logging.Handler] = []
340    try:
341        if max_bytes <= 0:
342            raise ValueError("max_bytes must be positive")
343        if backup_count < 0:
344            raise ValueError("backup_count must not be negative")
345        log_file.parent.mkdir(parents=True, exist_ok=True)
346        redactor = SecretRedactor()
347        formatter = KeyValueFormatter(redactor)
348
349        stderr_handler = logging.StreamHandler(stderr)
350        stderr_handler.setFormatter(formatter)
351        handlers.append(stderr_handler)
352
353        factory = file_handler_factory or _default_file_handler
354        file_handler = factory(
355            log_file,
356            stderr=stderr,
357            max_bytes=max_bytes,
358            backup_count=backup_count,
359        )
360        if not isinstance(file_handler, logging.Handler):
361            raise TypeError("file handler factory must return logging.Handler")
362        file_handler.setFormatter(formatter)
363        if file_handler_factory is not None:
364            file_handler = _SafeDelegatingHandler(file_handler, stderr)
365        handlers.append(file_handler)
366
367        for handler in handlers:
368            handler.__dict__["_agent_search_gateway_debug_owned"] = True
369            logger.addHandler(handler)
370        logger.setLevel(logging.DEBUG)
371        logger.propagate = False
372        return DebugLoggingSession(
373            logger,
374            tuple(handlers),
375            redactor,
376            previous_level=previous_level,
377            previous_propagate=previous_propagate,
378        )
379    except Exception as exc:
380        for handler in handlers:
381            logger.removeHandler(handler)
382            with suppress(Exception):
383                handler.close()
384        logger.setLevel(previous_level)
385        logger.propagate = previous_propagate
386        raise ConfigFailure(ErrorCode.CONFIG_ERROR, "Failed to configure debug logging") from exc