Spaces:
Build error
Build error
| """Request logger for the Headroom proxy. | |
| Logs requests to an in-memory deque and optionally to a JSONL file. | |
| Extracted from server.py for maintainability. | |
| """ | |
| from __future__ import annotations | |
| import json | |
| import sys | |
| from collections import deque | |
| from dataclasses import asdict | |
| from pathlib import Path | |
| from typing import TYPE_CHECKING | |
| if TYPE_CHECKING: | |
| from ..memory.tracker import ComponentStats | |
| from headroom.proxy.models import RequestLog | |
| class RequestLogger: | |
| """Log requests to JSONL file. | |
| Uses a deque with max 10,000 entries to prevent unbounded memory growth. | |
| """ | |
| MAX_LOG_ENTRIES = 10_000 | |
| def __init__(self, log_file: str | None = None, log_full_messages: bool = False): | |
| self.log_file = Path(log_file) if log_file else None | |
| self.log_full_messages = log_full_messages | |
| # Use deque with maxlen for automatic FIFO eviction | |
| self._logs: deque[RequestLog] = deque(maxlen=self.MAX_LOG_ENTRIES) | |
| if self.log_file: | |
| self.log_file.parent.mkdir(parents=True, exist_ok=True) | |
| def log(self, entry: RequestLog): | |
| """Log a request. Oldest entries are automatically removed when limit reached.""" | |
| self._logs.append(entry) | |
| if self.log_file: | |
| with open(self.log_file, "a") as f: | |
| log_dict = asdict(entry) | |
| if not self.log_full_messages: | |
| log_dict.pop("request_messages", None) | |
| log_dict.pop("response_content", None) | |
| f.write(json.dumps(log_dict) + "\n") | |
| def get_recent(self, n: int = 100) -> list[dict]: | |
| """Get recent log entries.""" | |
| # Convert deque to list for slicing (deque doesn't support slicing) | |
| entries = list(self._logs)[-n:] | |
| return [ | |
| { | |
| k: v | |
| for k, v in asdict(e).items() | |
| if k not in ("request_messages", "response_content") | |
| } | |
| for e in entries | |
| ] | |
| def stats(self) -> dict: | |
| """Get logging statistics.""" | |
| return { | |
| "total_logged": len(self._logs), | |
| "log_file": str(self.log_file) if self.log_file else None, | |
| } | |
| def get_memory_stats(self) -> ComponentStats: | |
| """Get memory statistics for the MemoryTracker. | |
| Returns: | |
| ComponentStats with current memory usage. | |
| """ | |
| from ..memory.tracker import ComponentStats | |
| # Calculate size | |
| size_bytes = sys.getsizeof(self._logs) | |
| for log_entry in self._logs: | |
| size_bytes += sys.getsizeof(log_entry) | |
| # Add string fields | |
| if log_entry.request_id: | |
| size_bytes += len(log_entry.request_id) | |
| if log_entry.provider: | |
| size_bytes += len(log_entry.provider) | |
| if log_entry.model: | |
| size_bytes += len(log_entry.model) | |
| if log_entry.error: | |
| size_bytes += len(log_entry.error) | |
| # Messages and response can be large | |
| if log_entry.request_messages: | |
| size_bytes += sys.getsizeof(log_entry.request_messages) | |
| if log_entry.response_content: | |
| size_bytes += len(log_entry.response_content) | |
| return ComponentStats( | |
| name="request_logger", | |
| entry_count=len(self._logs), | |
| size_bytes=size_bytes, | |
| budget_bytes=None, | |
| hits=0, | |
| misses=0, | |
| evictions=0, | |
| ) | |