Coverage for app/utils/logger.py: 100.00%
239 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-25 16:05 +0000
1"""Structured logging for deployment tasks.
3Three concerns are kept separate here:
5* **Buffer** — every log entry is appended to an in-memory list so the worker
6 task can hand the whole transcript back to the backend on completion.
7* **Console sink** — entries are also rendered once to the standard Python
8 logger so they appear in the worker container's stdout. Rendering lives in
9 a dedicated sink so the buffer and the Python logger stay independent.
10* **Event sink** — an optional callable that ships every entry to a
11 downstream consumer. The deploy task wires the Celery event bus here so
12 the backend's listener can stream entries to the browser.
14A single ``StructuredLogger`` orchestrates these. The class is threadsafe:
15``Popen`` line readers may write from a reader thread while the main task
16thread also writes phase markers, so buffer access goes through an ``RLock``.
17"""
19from __future__ import annotations
21import contextlib
22import json
23import logging
24import os
25import re
26import threading
27import time
28import traceback
29from collections.abc import Callable, Mapping
30from dataclasses import dataclass, field
31from datetime import UTC, datetime
32from enum import StrEnum
33from typing import Any
35# ----------------------------------------------------------------------------
36# Public enums
37# ----------------------------------------------------------------------------
40class LogLevel(StrEnum):
41 DEBUG = "DEBUG"
42 INFO = "INFO"
43 SUCCESS = "SUCCESS"
44 WARNING = "WARNING"
45 ERROR = "ERROR"
48class LogCategory(StrEnum):
49 PHASE = "phase"
50 OPERATION = "operation"
51 SYSTEM = "system"
52 STATUS = "status"
53 OUTPUT = "output"
54 ERROR = "error"
55 DEBUG = "debug"
56 PROGRESS = "progress"
57 # Kept for backwards compatibility with call sites that pass
58 # ``category=LogCategory.WARNING``. Treated identically to SYSTEM in
59 # rendering — the level (``warning``) carries the actual severity.
60 WARNING = "warning"
63# ----------------------------------------------------------------------------
64# Helpers
65# ----------------------------------------------------------------------------
67_ANSI_RE = re.compile(r"\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])")
68_LEVEL_ICONS: dict[LogLevel, str] = {
69 LogLevel.DEBUG: "🔍",
70 LogLevel.INFO: "ℹ️",
71 LogLevel.SUCCESS: "✓",
72 LogLevel.WARNING: "⚠️",
73 LogLevel.ERROR: "❌",
74}
77def _now_iso() -> str:
78 """Current UTC time as ISO-8601 with explicit ``Z`` suffix."""
79 return datetime.now(UTC).isoformat().replace("+00:00", "Z")
82def clean_text(text: str) -> str:
83 """Strip ANSI escapes and trim whitespace."""
84 return _ANSI_RE.sub("", text).strip()
87def truncate_text(text: str, max_lines: int = 50, max_chars: int = 5000) -> str:
88 """Truncate text intelligently for the buffered transcript.
90 For tool output the *tail* almost always matters more than the
91 head — Terraform/Packer print the error block last, and naively
92 cutting off the end (``text[:max_chars]``) systematically hides
93 the very thing we wanted to log. So the strategy is:
95 * Line-cap first: if there are more lines than ``max_lines``, keep
96 the first 15 and the last ``max_lines - 16`` so the trailing
97 error block survives.
98 * Char-cap second, but split the budget between head and tail
99 (40% / 60%) instead of dropping the tail entirely.
100 """
101 lines = text.split("\n")
102 if len(lines) > max_lines:
103 tail_keep = max_lines - 16
104 lines = lines[:15] + [f"... (truncated {len(lines) - 15 - tail_keep} lines) ..."] + lines[-tail_keep:]
105 text = "\n".join(lines)
107 if len(text) > max_chars:
108 head_budget = max_chars * 4 // 10
109 tail_budget = max_chars - head_budget - 64 # leave room for the marker
110 head = text[:head_budget]
111 tail = text[-tail_budget:]
112 text = head + f"\n... (truncated {len(text) - head_budget - tail_budget} characters) ...\n" + tail
114 return text
117def _scalar_or_str(value: Any) -> Any:
118 """Coerce one context value into a JSON-friendly scalar.
120 Tools like ``orjson`` would handle anything; we default to the stdlib
121 ``json`` module for portability, so we down-cast unknown types to their
122 repr.
123 """
124 if isinstance(value, str | int | float | bool) or value is None:
125 return value
126 if isinstance(value, dict):
127 return {k: _scalar_or_str(v) for k, v in value.items()}
128 if isinstance(value, list | tuple):
129 return [_scalar_or_str(v) for v in value]
130 return str(value)
133# ----------------------------------------------------------------------------
134# LogEntry
135# ----------------------------------------------------------------------------
138@dataclass(frozen=True)
139class LogEntry:
140 """A single log entry with explicit slots for common metadata.
142 Putting ``stdout``/``stderr``/``tool``/``phase`` etc. on the dataclass
143 instead of leaving everything in a generic context dict means consumers
144 (the frontend, the backend listener, the JSON exporter) can render those
145 fields specifically — e.g. monospace blocks for tool output rather than a
146 bag of strings inside ``extra``.
147 """
149 timestamp: str
150 level: LogLevel
151 category: LogCategory
152 message: str
153 correlation_id: str | None = None
154 operation: str | None = None
155 duration_ms: float | None = None
156 tool: str | None = None
157 returncode: int | None = None
158 stdout: str | None = None
159 stderr: str | None = None
160 phase: str | None = None
161 progress_pct: int | None = None
162 streaming: bool = False
163 truncated: bool = False
164 extra: Mapping[str, Any] = field(default_factory=dict)
166 def to_dict(self) -> dict[str, Any]:
167 """Flat JSON-serialisable representation; ``None`` fields are skipped."""
168 d: dict[str, Any] = {
169 "timestamp": self.timestamp,
170 "level": self.level.value,
171 "category": self.category.value,
172 "message": self.message,
173 }
174 for key in (
175 "correlation_id",
176 "operation",
177 "duration_ms",
178 "tool",
179 "returncode",
180 "stdout",
181 "stderr",
182 "phase",
183 "progress_pct",
184 ):
185 value = getattr(self, key)
186 if value is not None:
187 d[key] = round(value, 2) if key == "duration_ms" else value
188 if self.streaming:
189 d["streaming"] = True
190 if self.truncated:
191 d["truncated"] = True
192 if self.extra:
193 # Merge into top-level so consumers see context fields flat on
194 # the entry dict.
195 for k, v in self.extra.items():
196 if k not in d:
197 d[k] = v
198 return d
200 def __str__(self) -> str:
201 icon = _LEVEL_ICONS.get(self.level, "•")
202 duration = f" [{self.duration_ms:.2f}ms]" if self.duration_ms is not None else ""
203 return f"{icon} [{self.timestamp}] {self.message}{duration}"
206# ----------------------------------------------------------------------------
207# Sinks
208# ----------------------------------------------------------------------------
211class _Buffer:
212 """Threadsafe append-only list of entries.
214 ``RLock`` because export methods (``get_logs_dict``) iterate while a
215 second thread (``Popen`` reader) may still be appending. Using ``list``
216 + lock is plenty for our throughput; a queue/deque would be overkill.
217 """
219 def __init__(self) -> None:
220 self._entries: list[LogEntry] = []
221 self._lock = threading.RLock()
223 def append(self, entry: LogEntry) -> None:
224 with self._lock:
225 self._entries.append(entry)
227 def snapshot(self) -> list[LogEntry]:
228 with self._lock:
229 return list(self._entries)
231 def clear(self) -> None:
232 with self._lock:
233 self._entries.clear()
236class _ConsoleSink:
237 """Write each entry to the standard Python logger exactly once.
239 Disabled via ``WORKER_LOG_CONSOLE=0`` — useful in tests where pytest's
240 capture and our own buffer would otherwise both grow.
241 """
243 def __init__(self, name: str, enabled: bool) -> None:
244 self._logger = logging.getLogger(name)
245 self._enabled = enabled
246 self._method: dict[LogLevel, Callable[[str], None]] = {
247 LogLevel.DEBUG: self._logger.debug,
248 LogLevel.INFO: self._logger.info,
249 LogLevel.SUCCESS: self._logger.info,
250 LogLevel.WARNING: self._logger.warning,
251 LogLevel.ERROR: self._logger.error,
252 }
254 def write(self, entry: LogEntry) -> None:
255 if not self._enabled:
256 return
257 self._method.get(entry.level, self._logger.info)(str(entry))
260EventEmitter = Callable[[str, dict[str, Any]], None]
261"""Signature: ``emit(event_name, payload_dict) -> None``.
263The deploy task plugs Celery's ``self.send_event`` here so every log entry
264also turns into a ``task-log`` event on the bus. ``None`` disables it.
265"""
268# ----------------------------------------------------------------------------
269# StructuredLogger
270# ----------------------------------------------------------------------------
273class StructuredLogger:
274 """Buffer + console + optional event-bus emitter for one deployment task."""
276 LOG_EVENT_NAME = "task-log"
277 PROGRESS_EVENT_NAME = "task-progress"
279 def __init__(
280 self,
281 name: str,
282 *,
283 correlation_id: str | None = None,
284 console: bool = True,
285 event_emitter: EventEmitter | None = None,
286 track_timing: bool = True,
287 ) -> None:
288 self.name = name
289 self.correlation_id = correlation_id
290 self.track_timing = track_timing
291 self._buffer = _Buffer()
292 self._console = _ConsoleSink(name, console and os.environ.get("WORKER_LOG_CONSOLE", "1") != "0")
293 self._event_emitter = event_emitter
294 self._operation_stack: list[tuple[str, float]] = []
295 self._stack_lock = threading.Lock()
297 # ----- emitter wiring (settable post-construction) --------------------
299 def set_event_emitter(self, emitter: EventEmitter | None) -> None:
300 """Attach or detach the Celery-event sink at runtime.
302 ``tasks.py`` constructs the logger before it can capture ``self`` for
303 ``send_event``, so we need a setter.
304 """
305 self._event_emitter = emitter
307 # ----- core write-path ------------------------------------------------
309 def _record(self, entry: LogEntry, *, emit_event: bool = True) -> None:
310 self._buffer.append(entry)
311 self._console.write(entry)
312 if emit_event and self._event_emitter is not None:
313 try:
314 # Celery's EventReceiver injects its own ``timestamp``
315 # field (Unix-time float) and does arithmetic on it, so a
316 # payload carrying an ISO-string ``timestamp`` would crash
317 # it. Re-key ours as ``iso_timestamp``.
318 payload = entry.to_dict()
319 if "timestamp" in payload:
320 payload["iso_timestamp"] = payload.pop("timestamp")
321 self._event_emitter(self.LOG_EVENT_NAME, payload)
322 except Exception: # pragma: no cover — emitter is best-effort
323 # Never let log-shipping break the deployment; just swallow
324 # and continue. The buffered transcript still has the entry.
325 pass
327 def _build(
328 self,
329 message: str,
330 /,
331 *,
332 level: LogLevel,
333 category: LogCategory,
334 max_chars: int = 5000,
335 **fields: Any,
336 ) -> LogEntry:
337 # ``message`` is positional-only (note the ``/``) so a context
338 # kwarg literally named ``message`` (e.g. a git commit message via
339 # ``resource_info(..., message=...)``) lands in ``**fields`` rather
340 # than colliding with the parameter.
341 cleaned = clean_text(str(message))
342 truncated = False
343 if len(cleaned) > max_chars:
344 cleaned = truncate_text(cleaned, max_lines=50, max_chars=max_chars)
345 truncated = True
346 # Slot keys are recognised dataclass fields. Anything else lands
347 # in ``extra``.
348 slot_keys = {
349 "operation",
350 "duration_ms",
351 "tool",
352 "returncode",
353 "stdout",
354 "stderr",
355 "phase",
356 "progress_pct",
357 "streaming",
358 }
359 slots: dict[str, Any] = {k: v for k, v in fields.items() if k in slot_keys and v is not None}
360 extra: dict[str, Any] = {}
361 for k, v in fields.items():
362 if k in slot_keys:
363 continue
364 # Rename a context kwarg that collides with the top-level
365 # ``message`` slot so JSON consumers don't see two values
366 # for the same key after the dict is flattened.
367 key_out = "context_message" if k == "message" else k
368 extra[key_out] = _scalar_or_str(v)
369 return LogEntry(
370 timestamp=_now_iso(),
371 level=level,
372 category=category,
373 message=cleaned,
374 correlation_id=self.correlation_id,
375 truncated=truncated,
376 extra=extra,
377 **slots,
378 )
380 # ----- level shortcuts (kept for backwards compat) --------------------
382 def debug(self, message: str, category: LogCategory = LogCategory.DEBUG, **context: Any) -> None:
383 self._record(self._build(message, level=LogLevel.DEBUG, category=category, **context))
385 def info(self, message: str, category: LogCategory = LogCategory.SYSTEM, **context: Any) -> None:
386 self._record(self._build(message, level=LogLevel.INFO, category=category, **context))
388 def success(self, message: str, category: LogCategory = LogCategory.STATUS, **context: Any) -> None:
389 self._record(self._build(message, level=LogLevel.SUCCESS, category=category, **context))
391 def warning(self, message: str, category: LogCategory = LogCategory.SYSTEM, **context: Any) -> None:
392 self._record(self._build(message, level=LogLevel.WARNING, category=category, **context))
394 def error(self, message: str, category: LogCategory = LogCategory.ERROR, **context: Any) -> None:
395 self._record(self._build(message, level=LogLevel.ERROR, category=category, **context))
397 # ----- structured helpers --------------------------------------------
399 def phase(self, phase_name: str) -> None:
400 """Mark a major deployment phase.
402 Live progress is emitted separately via ``progress()`` so the
403 listener can update the DB ``progress_pct`` column. ``phase()`` only
404 adds a transcript marker.
405 """
406 self._record(
407 self._build(
408 f"=== {phase_name.upper()} ===",
409 level=LogLevel.INFO,
410 category=LogCategory.PHASE,
411 phase=phase_name,
412 )
413 )
415 def progress(
416 self,
417 phase_name: str,
418 idx: int,
419 total: int,
420 message: str = "",
421 phase_names: tuple[str, ...] | list[str] | None = None,
422 ) -> None:
423 """Send a progress update without buffering a per-step transcript entry.
425 The buffered transcript shouldn't grow by a progress marker per
426 step; we still emit the Celery custom event so the listener can
427 update the DB and the UI.
429 ``phase_names`` is the full ordered list of phases for this task.
430 When provided, the UI can render every stepper slot with its real
431 label immediately. The payload is small and safe to repeat on
432 every event — the listener overwrites its cached copy.
434 Do NOT put a ``timestamp`` field in the payload: Celery injects one
435 itself (as a Unix-time float) and would crash on an ISO string.
436 Pass the ISO time as ``iso_timestamp`` if needed downstream.
437 """
438 pct = max(0, min(100, round((idx / max(total, 1)) * 100)))
439 if self._event_emitter is not None:
440 payload: dict[str, Any] = {
441 "phase": phase_name,
442 "phase_index": idx,
443 "total_phases": total,
444 "progress_pct": pct,
445 "message": message,
446 "correlation_id": self.correlation_id,
447 "iso_timestamp": _now_iso(),
448 }
449 if phase_names is not None:
450 # Keep as a plain list — Celery uses JSON serialisation
451 # by default and tuples are coerced anyway.
452 payload["phase_names"] = list(phase_names)
453 with contextlib.suppress(Exception):
454 self._event_emitter(self.PROGRESS_EVENT_NAME, payload)
456 def operation_start(self, operation_name: str, **context: Any) -> None:
457 if self.track_timing:
458 with self._stack_lock:
459 self._operation_stack.append((operation_name, time.time()))
460 self._record(
461 self._build(
462 f"Starting: {operation_name}",
463 level=LogLevel.INFO,
464 category=LogCategory.OPERATION,
465 operation=operation_name,
466 **context,
467 )
468 )
470 def operation_end(self, operation_name: str, success: bool = True, **context: Any) -> None:
471 duration_ms: float | None = None
472 if self.track_timing:
473 with self._stack_lock:
474 if self._operation_stack:
475 last_op, start_time = self._operation_stack[-1]
476 if last_op == operation_name:
477 self._operation_stack.pop()
478 duration_ms = (time.time() - start_time) * 1000
479 status = "completed" if success else "failed"
480 self._record(
481 self._build(
482 f"{status.capitalize()}: {operation_name}",
483 level=LogLevel.INFO,
484 category=LogCategory.OPERATION,
485 operation=operation_name,
486 duration_ms=duration_ms,
487 **context,
488 )
489 )
491 def command_output(self, tool_name: str, output: str, returncode: int = 0, **context: Any) -> None:
492 """Buffer a captured command output block (post-completion).
494 For per-line streaming use ``tool_output_line`` instead — that one
495 emits each line as its own event without 5000-char framing.
496 """
497 level = LogLevel.ERROR if returncode != 0 else LogLevel.INFO
498 category = LogCategory.ERROR if returncode != 0 else LogCategory.OUTPUT
499 self._record(
500 self._build(
501 output,
502 level=level,
503 category=category,
504 tool=tool_name,
505 operation=tool_name,
506 returncode=returncode,
507 **context,
508 )
509 )
511 def tool_output_line(self, tool_name: str, line: str) -> None:
512 """One streaming line of subprocess output.
514 Skips the buffered transcript truncation entirely (the whole point of
515 streaming is to ship lines as they happen) and marks ``streaming=True``
516 so the frontend can render it as live tail without expecting the
517 usual operation framing.
518 """
519 cleaned = clean_text(line)
520 if not cleaned:
521 return
522 entry = LogEntry(
523 timestamp=_now_iso(),
524 level=LogLevel.INFO,
525 category=LogCategory.OUTPUT,
526 message=cleaned,
527 correlation_id=self.correlation_id,
528 tool=tool_name,
529 streaming=True,
530 )
531 self._record(entry)
533 def exception(self, message: str, exception: Exception | None = None, **context: Any) -> None:
534 """Log an exception with its real traceback.
536 Uses ``format_exception(type, exc, exc.__traceback__)`` rather than
537 ``format_exc()`` so it works outside an active ``except`` block —
538 ``format_exc()`` returned ``"NoneType: None\\n"`` in that case.
539 """
540 ctx: dict[str, Any] = dict(context)
541 if exception is not None:
542 ctx["exception_type"] = type(exception).__name__
543 ctx["exception_message"] = str(exception)
544 tb = "".join(traceback.format_exception(type(exception), exception, exception.__traceback__))
545 if len(tb) > 2000:
546 tb = truncate_text(tb, max_lines=30)
547 ctx["stack_trace"] = tb
548 else:
549 ctx.setdefault("exception_type", "Unknown")
550 self._record(self._build(message, level=LogLevel.ERROR, category=LogCategory.ERROR, **ctx))
552 def resource_info(self, resource_type: str, resource_name: str, **details: Any) -> None:
553 self._record(
554 self._build(
555 f"{resource_type}: {resource_name}",
556 level=LogLevel.INFO,
557 category=LogCategory.SYSTEM,
558 resource_type=resource_type,
559 resource_name=resource_name,
560 **details,
561 )
562 )
564 # ----- inspection / export -------------------------------------------
566 def get_logs_dict(self) -> list[dict[str, Any]]:
567 return [e.to_dict() for e in self._buffer.snapshot()]
569 def get_logs_json(self, pretty: bool = True) -> str:
570 return json.dumps(self.get_logs_dict(), indent=2 if pretty else None)
572 def get_logs_text(self) -> str:
573 return "\n".join(str(e) for e in self._buffer.snapshot())
575 def get_logs_by_category(self, category: LogCategory) -> list[dict[str, Any]]:
576 return [e.to_dict() for e in self._buffer.snapshot() if e.category == category]
578 def get_logs_by_level(self, level: LogLevel) -> list[dict[str, Any]]:
579 return [e.to_dict() for e in self._buffer.snapshot() if e.level == level]
581 def get_summary(self) -> dict[str, Any]:
582 entries = self._buffer.snapshot()
583 by_level: dict[str, int] = {}
584 by_category: dict[str, int] = {}
585 for e in entries:
586 by_level[e.level.value] = by_level.get(e.level.value, 0) + 1
587 by_category[e.category.value] = by_category.get(e.category.value, 0) + 1
588 return {
589 "total_entries": len(entries),
590 "by_level": by_level,
591 "by_category": by_category,
592 "timestamp_range": {
593 "first": entries[0].timestamp if entries else None,
594 "last": entries[-1].timestamp if entries else None,
595 },
596 }
598 def clear(self) -> None:
599 self._buffer.clear()
600 with self._stack_lock:
601 self._operation_stack.clear()
604# ----------------------------------------------------------------------------
605# Factory
606# ----------------------------------------------------------------------------
609def get_logger(name: str, *, correlation_id: str | None = None) -> StructuredLogger:
610 """Return a fresh ``StructuredLogger``.
612 Each deployment task gets its own instance — the per-task buffer is
613 state, so we deliberately don't memoise like ``logging.getLogger`` does.
614 """
615 return StructuredLogger(name, correlation_id=correlation_id)