Spaces:
Build error
Build error
File size: 3,498 Bytes
226e851 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 | """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,
)
|