Test and Observe

"""Structured logging, correlation ids, metrics and alerts: knowing what the bot is doing without recording what people said.""" import contextlib import contextvars import hashlib import hmac import json import logging import re import secrets import time from datetime import datetime, timezone from bot.privacy import redact from bot.providers import Message, Provider, ProviderError, Reply correlation_id: contextvars.ContextVar[str | None] = contextvars.ContextVar("correlation_id", default=None) # --- correlation ids ---------------------------------------------------------------------------------------------- @contextlib.contextmanager def correlation(prefix: str = "msg"): """Give everything that happens inside this block (including awaited calls and tasks it starts) one shared id. If an id is already set, for example by the channel adapter, it is kept: one user interaction, one id.""" if correlation_id.get() is not None: yield correlation_id.get() return token = correlation_id.set(f"{prefix}-{secrets.token_hex(4)}") try: yield correlation_id.get() finally: correlation_id.reset(token) def pseudonym(value: str, key: bytes) -> str: """A stable, non-reversible label for an id: the same user always gets the same label under one key, so patterns are visible in the logs, but the label does not reveal the id. Change the key and the labels change (and cannot be linked).""" return "u_" + hmac.new(key, value.encode(), hashlib.sha256).hexdigest()[:12] # --- keeping secrets and personal data out of log lines ---------------------------------------------------------------- SECRET_PATTERNS = [ (re.compile(r"(?<!\d)\d{6,}:[A-Za-z0-9_-]{30,}"), "[telegram-token]"), # no \b: it must also match inside /bot<token>/getMe (re.compile(r"\bxox[a-zA-Z]-[A-Za-z0-9-]{10,}"), "[slack-token]"), (re.compile(r"\bsk-[A-Za-z0-9_-]{16,}"), "[api-key]"), (re.compile(r"\b[\w-]{23,28}\.[\w-]{6,7}\.[\w-]{27,}\b"), "[discord-token]"), (re.compile(r"(?i)\bbearer\s+[A-Za-z0-9._~+/=-]{8,}"), "Bearer [token]"), (re.compile(r"(?i)\bx-api-key\W+[A-Za-z0-9._-]{8,}"), "x-api-key [key]"), ] def scrub(text: str) -> str: """Replace anything that looks like a token or key, then the obvious personal identifiers (emails, long numbers).""" for pattern, replacement in SECRET_PATTERNS: text = pattern.sub(replacement, text) return redact(text) class ScrubFilter(logging.Filter): """Runs scrub() over the message and over every string field attached to a record, as a last line of defence.""" STANDARD = set(logging.LogRecord("", 0, "", 0, "", (), None).__dict__) | {"message", "asctime"} def filter(self, record: logging.LogRecord) -> bool: record.msg = scrub(record.getMessage()) record.args = () for name, value in list(record.__dict__.items()): if name not in self.STANDARD and isinstance(value, str): setattr(record, name, scrub(value)) return True class JsonFormatter(logging.Formatter): """One JSON object per line. JSON escapes newlines and control characters, so text cannot forge extra log lines.""" def format(self, record: logging.LogRecord) -> str: entry = {"time": datetime.fromtimestamp(record.created, timezone.utc).isoformat(timespec="milliseconds"), "level": record.levelname, "logger": record.name, "message": record.getMessage(), "correlation_id": correlation_id.get()} for name, value in record.__dict__.items(): if name not in ScrubFilter.STANDARD and not name.startswith("_"): entry[name] = value if record.exc_info: entry["exception"] = type(record.exc_info[1]).__name__ # the class, not the message (which may hold data) return json.dumps(entry, default=str) def setup_logging(handler: logging.Handler | None = None, level: int = logging.INFO, name: str = "chatbot") -> logging.Logger: """A logger that writes scrubbed JSON lines. Pass a RotatingFileHandler for a file; the default is the console.""" logger = logging.getLogger(name) logger.setLevel(level) logger.handlers.clear() handler = handler or logging.StreamHandler() handler.setFormatter(JsonFormatter()) handler.addFilter(ScrubFilter()) logger.addHandler(handler) logger.propagate = False return logger # --- events: the lists every chapter has been appending to ------------------------------------------------------------ class EventLog(list): """A list of events that tells its observers about each new one. Replaces the plain lists used so far, so the flags recorded all through the course (rate_limited, moderated:..., event_error:...) become log lines and counters.""" def __init__(self, *observers) -> None: super().__init__() self.observers = list(observers) def _notify(self, item) -> None: for observer in self.observers: try: observer(item) except Exception: # noqa: BLE001 - logging must never break the bot pass def append(self, item) -> None: super().append(item) self._notify(item) def extend(self, items) -> None: for item in items: self.append(item) def __iadd__(self, items): self.extend(items) return self def log_observer(logger: logging.Logger, key: bytes): """Turn events into log lines. Events from the Bot are (session, flag); adapters record just a flag.""" def observe(item) -> None: session, flag = item if isinstance(item, tuple) else (None, item) level = logging.ERROR if flag.split(":")[0] in {"event_error", "update_error", "stream_error", "send_failed"} else logging.INFO logger.log(level, "event", extra={"flag": flag, "session": pseudonym(session, key) if session else None}) return observe # --- metrics --------------------------------------------------------------------------------------------------------- class Metrics: """Counters and timing samples in memory. Enough for a small bot; a bigger one would export them to a metrics system.""" def __init__(self, keep: int = 5000) -> None: self.counters: dict[str, float] = {} self.samples: dict[str, list[float]] = {} self.keep = keep def inc(self, name: str, amount: float = 1.0) -> None: self.counters[name] = self.counters.get(name, 0.0) + amount def observe(self, name: str, value: float) -> None: samples = self.samples.setdefault(name, []) samples.append(value) if len(samples) > self.keep: del samples[: len(samples) - self.keep] @staticmethod def percentile(values: list[float], q: float) -> float: if not values: return 0.0 ordered = sorted(values) return ordered[min(len(ordered) - 1, int(q * len(ordered)))] def snapshot(self) -> dict: out = dict(self.counters) for name, values in self.samples.items(): out[f"{name}_p50"] = self.percentile(values, 0.50) out[f"{name}_p95"] = self.percentile(values, 0.95) out[f"{name}_count"] = len(values) return out def observer(self): def observe(item) -> None: flag = item[1] if isinstance(item, tuple) else item self.inc("events_total") self.inc("event:" + flag.split(":")[0]) return observe def check_alerts(metrics: Metrics, *, min_messages: int = 20, max_error_rate: float = 0.05, max_p95_seconds: float = 10.0, max_fallbacks: int = 5) -> list[str]: """A few rules worth waking someone for. Rates need enough traffic to mean anything, hence min_messages.""" snap, alerts = metrics.snapshot(), [] messages = snap.get("messages_total", 0) errors = snap.get("provider_errors_total", 0) if messages >= min_messages and errors / messages > max_error_rate: alerts.append(f"provider error rate {errors / messages:.0%} over {int(messages)} messages (limit {max_error_rate:.0%})") if snap.get("handle_seconds_p95", 0) > max_p95_seconds: alerts.append(f"p95 latency {snap['handle_seconds_p95']:.1f} s (limit {max_p95_seconds:g} s)") if snap.get("event:event_error", 0) + snap.get("event:update_error", 0) > 0: alerts.append("unexpected exceptions caught by an adapter (event_error / update_error)") if snap.get("event:global_budget", 0) > 0: alerts.append("the daily spending budget was reached") if snap.get("fallbacks_total", 0) > max_fallbacks: alerts.append(f"{int(snap['fallbacks_total'])} fallbacks to the backup provider (limit {max_fallbacks})") return alerts # --- a provider wrapper that logs every call ------------------------------------------------------------------------------ class LoggedProvider: """Logs one line per provider call: how long it took, which model, how many tokens, whether it failed and how. It never logs the messages or the reply.""" def __init__(self, inner: Provider, logger: logging.Logger, metrics: Metrics | None = None, clock=time.perf_counter) -> None: self.inner, self.logger, self.metrics, self.clock = inner, logger, metrics, clock self.name = inner.name def _record(self, started: float, reply: Reply | None, error: ProviderError | None, streamed: bool, first_piece: float | None = None) -> None: seconds = self.clock() - started fields = {"provider": self.inner.name, "seconds": round(seconds, 3), "streamed": streamed} if first_piece is not None: fields["first_piece_seconds"] = round(first_piece, 3) if reply is not None: fields.update(model=reply.model, input_tokens=reply.usage.input_tokens, output_tokens=reply.usage.output_tokens, stop_reason=reply.stop_reason, vendor_request_id=reply.request_id, served_by=reply.served_by) if error is not None: fields.update(error=type(error).__name__, status=error.status, retryable=error.retryable, vendor_request_id=error.request_id) self.logger.log(logging.WARNING if error else logging.INFO, "provider_call", extra=fields) if self.metrics: self.metrics.inc("provider_calls_total") self.metrics.observe("provider_seconds", seconds) if error: self.metrics.inc("provider_errors_total") if reply is not None: self.metrics.inc("tokens_total", reply.usage.input_tokens + reply.usage.output_tokens) async def complete(self, messages: list[Message], *, system: str | None = None, max_tokens: int = 1024) -> Reply: started = self.clock() try: reply = await self.inner.complete(messages, system=system, max_tokens=max_tokens) except ProviderError as e: self._record(started, None, e, False) raise self._record(started, reply, None, False) return reply async def stream(self, messages: list[Message], *, system: str | None = None, max_tokens: int = 1024): started, first = self.clock(), None final = None try: async for item in self.inner.stream(messages, system=system, max_tokens=max_tokens): if first is None: first = self.clock() - started if isinstance(item, Reply): final = item yield item except ProviderError as e: self._record(started, None, e, True, first) raise self._record(started, final, None, True, first)

async def handle(self, session_id: str, text: str, *, user_id: str | None = None) -> str: """Answer one message with one complete reply.""" user_id = user_id or session_id with correlation(): started, mark, outcome, state = time.perf_counter(), len(self.events), "error", {} try: reply = await self._handle(session_id, text, user_id, state) outcome = "answered" if state.get("model") else "local" # "local": answered without calling the model return reply finally: self._log_message(session_id, user_id, text, outcome, started, mark, streamed=False)

def _log_message(self, session_id: str, user_id: str, text: str, outcome: str, started: float, mark: int, *, streamed: bool) -> None: """One line per message: who (as pseudonyms), how big, how long, what happened. Never the text itself.""" seconds = time.perf_counter() - started flags = [flag for _, flag in self.events[mark:]] if self.metrics: self.metrics.inc("messages_total") self.metrics.inc(f"outcome:{outcome}") self.metrics.observe("handle_seconds", seconds) if self.logger: self.logger.info("message", extra={"session": pseudonym(session_id, self.log_key), "user": pseudonym(user_id, self.log_key), "input_chars": len(text), "outcome": outcome, "flags": flags, "seconds": round(seconds, 3), "streamed": streamed})