fix(logging)!: write structured entries synchronously so a forked child keeps them (LAB-9130)

docs/api-reference.md

@@ -110,7 +110,7 @@ def your_function(args):
 
 - **`monitoring`** (`MonitoringConfig`, default: `MonitoringConfig()`) - Observability configuration:
   - `collect_stats` (`bool`, default: `True`) - Collect cache hit/miss statistics
-  - `enable_structured_logging` (`bool`, default: `True`) - Enable structured JSON logging
+  - `enable_structured_logging` (`bool`, default: `True`) - Log cache operations at INFO, with their fields on each record ([Structured Logging](#structured-logging))
   - `enable_prometheus_metrics` (`bool`, default: `True`) - Export Prometheus metrics
 
 #### Encryption Parameters
@@ -840,10 +840,11 @@ query examples, and alerting rules.
 
 ### Structured Logging
 
-All operations include structured logging with:
-- Operation context (namespace, cache key, serializer)
-- Performance metrics (duration, cache hit/miss)
-- Error classification and recovery actions
+cachekit writes two kinds of structured record through Python's `logging`, so your handlers and levels decide what is kept.
+
+With `enable_structured_logging=True` (the default), these cache operations are logged at INFO on the `cachekit.decorators.orchestrator` logger: L1 and L2 hits of a sync function that has a backend, failed operations and circuit-breaker rejections. Each record's `structured` attribute holds the operation, the namespace and the redacted cache key, plus the duration, hit, serializer and error type where they apply. The message is plain text; your formatter decides whether those fields are rendered.
+
+SDK diagnostics, such as the CachekitIO backend's debug lines, are one JSON object per record on the `cachekit.logging` logger, with `timestamp`, `level`, `message`, `logger`, `hostname`, `pid` and `thread_id`. Each is written when it is logged, on the calling thread. DEBUG, INFO and WARNING records are sampled at `CACHEKIT_LOG_SAMPLING_RATE` (default `0.1`), and ERROR and CRITICAL records never are. A field value JSON cannot encode is written as its type name, such as `"<dict>"`, never as its text. Nothing is sampled or serialised below that logger's level. A child forked by Python (`os.fork()`, `multiprocessing`, Gunicorn, Celery) writes its own records, with its own `pid`. A child forked from C without Python's at-fork hooks (uWSGI without `--py-call-uwsgi-fork-hooks`) writes none: `logging`'s after-fork lock reset did not run there, so a lock a parent thread held at the fork would hang the call.
 
 ## Best Practices
 

docs/performance.md

@@ -215,9 +215,9 @@ The regression gate is the instruction budget, run locally with `make perf-ir`.
 `make perf-ir` counts the instructions each hot path executes per call and fails when a path costs **1% or more** above its committed budget. It warns from 0.2%. The orjson round trip fails at 2% and warns from 1% (see Limits). Instruction counts are deterministic where wall clock is not: two full runs agree within 0.08% per path (0.01% on CPython 3.12).
 
 **Method** (`tests/performance/ir_budget.py`; the measured process is `ir_workload.py`):
-- Each path runs under `valgrind --tool=callgrind --separate-threads=yes`, and only the main thread is counted. cachekit's background threads (log writer, L1 cleanup) vary by tens of percent between identical runs.
+- Each path runs under `valgrind --tool=callgrind --separate-threads=yes`, and only the main thread is counted. cachekit's background threads (L1 cleanup) vary by tens of percent between identical runs.
 - Per-call cost is `(Ir[3000 calls] - Ir[1000 calls]) / 2000`, so interpreter startup (~2 billion instructions) and warmup cancel. Interpreter teardown is skipped. The two real-data paths cost about 60 million instructions per call, so they run 10 and 30 calls after 5 warmup calls; startup still cancels, and two runs agreed within 0.03%.
-- The measured process has a fixed environment (`PYTHONHASHSEED=0`, one BLAS/OpenMP thread, a one-day log flush and L1 cleanup interval, nothing inherited), seeded log sampling, and main-thread clocks that advance 1μs per read. Code that records its own duration otherwise executes more instructions when it runs slower. The cleanup sweep reads the real clock, so inside a run it would evict entries stamped with the pinned one.
+- The measured process has a fixed environment (`PYTHONHASHSEED=0`, one BLAS/OpenMP thread, a one-day L1 cleanup interval, nothing inherited), seeded log sampling, and main-thread clocks that advance 1μs per read. Code that records its own duration otherwise executes more instructions when it runs slower. The cleanup sweep reads the real clock, so inside a run it would evict entries stamped with the pinned one.
 - Nothing that runs on real time lands in the measured loop: no background thread wakes, cyclic GC is off, and the GIL switch interval is long enough that the main thread gives up the GIL only where the code releases it.
 - Each path runs at five heap layouts (0 to 880 extra objects held before the workload) and its figure is the median. Any code change moves the layout, and the allocators take shorter or longer paths in some heap states: single layouts of the orjson round trip and the minimal L1 hit sat up to 2.4% from the rest. The median ignores one or two such layouts. The cheapest layout is not used, because a lucky one can sit 1% below the rest, and a budget recorded there makes every later run look like a regression.
 - `ir_workload.py`'s text is compiled into every measured process, so editing it moves the layout the budgets were recorded at: re-record after any change to it. The gate itself (`ir_budget.py`) is kept out of the measured process, so editing it does not move a budget.

src/cachekit/config/nested.py

@@ -234,7 +234,7 @@ class MonitoringConfig:
 
     Attributes:
         collect_stats: Collect cache hit/miss statistics (default: True)
-        enable_structured_logging: Enable structured JSON logging (default: True)
+        enable_structured_logging: Log cache operations at INFO, with their fields on each record (default: True)
         enable_prometheus_metrics: Export Prometheus metrics (default: True)
 
     Examples:

src/cachekit/config/settings.py

@@ -196,21 +196,6 @@ class CachekitConfig(RedactingSettings):
         le=1.0,
         description="Log sampling rate (0.0 to 1.0, default 10%)",
     )
-    log_buffer_size: int = Field(
-        default=10000,
-        gt=0,
-        description="Ring buffer size for async logging",
-    )
-    log_batch_size: int = Field(
-        default=100,
-        gt=0,
-        description="Batch size for async log writes",
-    )
-    log_flush_interval: float = Field(
-        default=1.0,
-        gt=0,
-        description="Flush interval for async logging in seconds",
-    )
 
     # Deployment and feature flags
     deployment_uuid: Optional[str] = Field(

src/cachekit/logging.py

@@ -1,7 +1,12 @@
-"""Structured logging with minimal overhead.
+"""Structured JSON logging with sampling.
 
-This module provides lock-free, sampling-based structured logging
-with asynchronous batch writes.
+Each entry is serialised and written on the caller's thread. There is no background writer: a thread
+started at import is not running in a forked child (Gunicorn ``--preload``, Celery prefork,
+``multiprocessing``'s fork start method), so everything a child buffered for it was lost.
+
+In a child forked from C without at-fork hooks (uWSGI without ``--py-call-uwsgi-fork-hooks``),
+``log()`` and ``error()`` write no entry: ``logging``'s own after-fork lock reset did not run there, so
+a level check or a handler lock a parent thread held at fork would hang the caller.
 """
 
 import json
@@ -11,155 +16,72 @@
 import random
 import threading
 import time
-from dataclasses import dataclass, field
-from typing import Any, Optional
+from typing import Any
 
+from cachekit import l1_cache
 from cachekit.config import get_settings
 from cachekit.hash_utils import redact_error_for_log, redact_key_for_log
 
 # Configure base logger
 logger = logging.getLogger(__name__)
 
 
-# Global configuration - loaded once at module import from the settings singleton
-_settings = get_settings()
-SAMPLING_RATE = _settings.log_sampling_rate
-RING_BUFFER_SIZE = _settings.log_buffer_size
-BATCH_SIZE = _settings.log_batch_size
-FLUSH_INTERVAL = _settings.log_flush_interval
-del _settings  # keep only the four scalars; don't pin a stale config past reset_settings()
+# Loaded once at module import from the settings singleton
+SAMPLING_RATE = get_settings().log_sampling_rate
 
 # Performance and health thresholds
 HIGH_UTILIZATION_THRESHOLD = 0.9  # When to warn about high utilization
 
+_LEVELS = {
+    "DEBUG": logging.DEBUG,
+    "INFO": logging.INFO,
+    "WARN": logging.WARNING,
+    "WARNING": logging.WARNING,
+    "ERROR": logging.ERROR,
+    "CRITICAL": logging.CRITICAL,
+    "FATAL": logging.CRITICAL,
+}
 
-@dataclass
-class LogEntry:
-    """Lightweight log entry for ring buffer."""
+# What json.dumps raises on a value its default= never sees: a non-str dict key, a cycle, a too-deep nest
+_UNENCODABLE = (TypeError, ValueError, RecursionError)
 
-    timestamp: float
-    level: str
-    message: str
-    extra: dict[str, Any] = field(default_factory=dict)
 
-    def to_dict(self) -> dict[str, Any]:
-        """Convert to dictionary for JSON serialization."""
-        return {
-            "timestamp": self.timestamp,
-            "level": self.level,
-            "message": self.message,
-            **self.extra,
-        }
+def _type_name(value: object) -> str:
+    """Stand in for a value JSON cannot encode: its type only, as its str() may embed a cache key (CWE-532)."""
+    return f"<{type(value).__name__}>"
 
 
-class LockFreeRingBuffer:
-    """Lock-free ring buffer for log entries.
+def _encodable(value: object) -> object:
+    """The value if json can encode it, else its type name."""
+    try:
+        json.dumps(value, default=_type_name)
+    except _UNENCODABLE:
+        return _type_name(value)
+    return value
 
-    Uses atomic operations to minimize contention.
-    """
 
-    def __init__(self, size: int = RING_BUFFER_SIZE):
-        self.size = size
-        self.buffer: list[Optional[LogEntry]] = [None] * size
-        self.write_pos = 0
-        self.read_pos = 0
-
-    def append(self, entry: LogEntry) -> bool:
-        """Append entry to buffer without locks.
-
-        Returns True if successful, False if buffer full.
-        """
-        # Simple atomic increment (in CPython, int ops are atomic)
-        pos = self.write_pos
-        next_pos = (pos + 1) % self.size
-
-        # Check if buffer is full (would overwrite unread data)
-        if next_pos == self.read_pos:
-            return False
-
-        # Write entry
-        self.buffer[pos] = entry
-        self.write_pos = next_pos
-        return True
-
-    def drain(self, max_items: int = BATCH_SIZE) -> list[LogEntry]:
-        """Drain up to max_items from buffer."""
-        items = []
-        count = 0
-
-        while count < max_items and self.read_pos != self.write_pos:
-            entry = self.buffer[self.read_pos]
-            if entry is not None:
-                items.append(entry)
-                self.buffer[self.read_pos] = None  # Clear reference
-            self.read_pos = (self.read_pos + 1) % self.size
-            count += 1
-
-        return items
-
-
-class AsyncLogWriter(threading.Thread):
-    """Background thread for async log writing."""
-
-    def __init__(self, buffer: LockFreeRingBuffer):
-        super().__init__(daemon=True, name="cachekit-LogWriter")
-        self.buffer = buffer
-        self.running = True
-        self._stop_event = threading.Event()
-
-    def run(self):
-        """Main loop - batch write logs periodically."""
-        while self.running:
-            try:
-                # Wait for interval or stop signal
-                if self._stop_event.wait(timeout=FLUSH_INTERVAL):
-                    break
-
-                # Drain and write batch
-                entries = self.buffer.drain()
-                if entries:
-                    self._write_batch(entries)
-
-            except Exception as e:
-                logger.error(f"Error in async log writer: {redact_error_for_log(e)}")
-
-    def stop(self):
-        """Stop the writer thread."""
-        self.running = False
-        self._stop_event.set()
-
-    def _write_batch(self, entries: list[LogEntry]):
-        """Write a batch of entries to the logger."""
-        for entry in entries:
-            try:
-                # Convert to JSON for structured logging
-                msg = json.dumps(entry.to_dict(), separators=(",", ":"))
-                logger.log(getattr(logging, entry.level, logging.INFO), msg)
-            except Exception:
-                # Silently drop malformed entries
-                pass
+def _enabled(levelno: int) -> bool:
+    """Whether to write an entry at levelno: never in a child forked without at-fork hooks (module docstring)."""
+    return not l1_cache._forked_without_hooks() and logger.isEnabledFor(levelno)
 
 
 class StructuredLogger:
     """Structured logger.
 
+    Each entry is one JSON line on the ``cachekit.logging`` logger, written synchronously.
+
     Features:
-    - Lock-free ring buffer
-    - Sampling (10% default)
-    - Async batch writes
+    - Nothing is sampled or serialised below that logger's level
+    - Sampling (10% default) of DEBUG/INFO/WARNING; ERROR and CRITICAL are never sampled
     - PII key-name masking (password/token/secret/key/auth kwargs)
     """
 
     def __init__(self, name: str):
         self.name = name
-        self.buffer = LockFreeRingBuffer()
-        self.writer = AsyncLogWriter(self.buffer)
-        self.writer.start()
 
         # Pre-computed values for performance
         self._sampling_threshold = int(SAMPLING_RATE * 100)
         self._hostname = platform.node()
-        self._pid = os.getpid()
 
         # PII patterns to mask (pre-compiled for speed)
         self._pii_keys = {"password", "token", "secret", "key", "auth"}
@@ -179,30 +101,30 @@ def _mask_pii(self, data: dict[str, Any]) -> dict[str, Any]:
                 data[key] = "****"
         return data
 
+    def _write(self, level: str, levelno: int, message: str, fields: dict[str, Any]) -> None:
+        """Serialise one entry and log it now."""
+        entry: dict[str, Any] = {
+            **self._mask_pii(fields),  # first, so a caller's pid= cannot replace the real one
+            "timestamp": time.time(),
+            "level": level,
+            "message": message,
+            "logger": self.name,
+            "hostname": self._hostname,
+            "pid": os.getpid(),  # read per entry: a forked child reports its own
+            "thread_id": threading.get_ident(),
+        }
+        try:
+            line = json.dumps(entry, separators=(",", ":"), default=_type_name)
+        except _UNENCODABLE:  # raising would replace the exception a caller may be logging
+            line = json.dumps({k: _encodable(v) for k, v in entry.items()}, separators=(",", ":"), default=_type_name)
+        logger.log(levelno, line)
+
     def log(self, level: str, message: str, **kwargs):
         """Main logging method with sampling."""
-        # Fast path - skip if not sampled
-        if not self._should_sample():
-            return
-
-        # Build entry
-        entry = LogEntry(
-            timestamp=time.time(),
-            level=level.upper(),
-            message=message,
-            extra={
-                "logger": self.name,
-                "hostname": self._hostname,
-                "pid": self._pid,
-                "thread_id": threading.get_ident(),
-                **self._mask_pii(kwargs),
-            },
-        )
-
-        # Append to lock-free buffer
-        if not self.buffer.append(entry):
-            # Buffer full - drop entry (non-blocking behavior)
-            pass
+        level = level.upper()
+        levelno = _LEVELS.get(level, logging.INFO)
+        if _enabled(levelno) and (levelno >= logging.ERROR or self._should_sample()):
+            self._write(level, levelno, message, kwargs)
 
     def debug(self, message: str, **kwargs):
         """Debug level logging."""
@@ -217,21 +139,9 @@ def warning(self, message: str, **kwargs):
         self.log("WARNING", message, **kwargs)
 
     def error(self, message: str, **kwargs):
-        """Error level logging - always sampled."""
-        # Errors bypass sampling
-        entry = LogEntry(
-            timestamp=time.time(),
-            level="ERROR",
-            message=message,
-            extra={
-                "logger": self.name,
-                "hostname": self._hostname,
-                "pid": self._pid,
-                "thread_id": threading.get_ident(),
-                **self._mask_pii(kwargs),
-            },
-        )
-        self.buffer.append(entry)
+        """Error level logging - never sampled. Not via log(): a caller field named level or message would collide."""
+        if _enabled(logging.ERROR):
+            self._write("ERROR", logging.ERROR, message, kwargs)
 
     def cache_operation(self, operation: str, cache_key: str, **kwargs):
         """Log cache operation with standard fields."""
@@ -293,21 +203,7 @@ def cache_operation(self, operation: str, cache_key: str, **kwargs):
             # Silently ignore if Prometheus metrics not available
             pass
 
-        # For compatibility with tests that expect standard logging
-        import logging as std_logging
-
-        # Map string level to logging constant
-        level_map = {
-            "DEBUG": std_logging.DEBUG,
-            "INFO": std_logging.INFO,
-            "WARNING": std_logging.WARNING,
-            "ERROR": std_logging.ERROR,
-            "CRITICAL": std_logging.CRITICAL,
-        }
-        log_level = level_map.get(level.upper(), std_logging.INFO)
-
-        logger = std_logging.getLogger(self.name)
-        logger.log(log_level, "cache_operation", extra={"structured": context})
+        self.logger.log(_LEVELS[level], "cache_operation", extra={"structured": context})
 
     def connection_pool_utilization(self, utilization: float, **kwargs):
         """Log connection pool metrics."""
@@ -335,11 +231,6 @@ def _get_context(self) -> dict[str, Any]:
             "thread_id": threading.get_ident(),
         }
 
-    def __del__(self):
-        """Cleanup on deletion."""
-        if hasattr(self, "writer"):
-            self.writer.stop()
-
 
 # Global logger instances cache
 _logger_instances: dict[str, StructuredLogger] = {}