From aa2912e7c221bd06a2a7c421c64e9935e54a222a Mon Sep 17 00:00:00 2001 From: Ray Walker Date: Tue, 13 Oct 2026 00:39:17 +1100 Subject: [PATCH 1/5] fix(logging)!: write structured entries synchronously so a forked child keeps them StructuredLogger queued entries in a ring buffer that one cachekit-LogWriter thread per process drained. The thread started at import and nothing started it again after a fork, so every forked child (os.fork with hooks included: Gunicorn --preload, Celery prefork, multiprocessing's fork start method) buffered its entries and never wrote them, and each entry carried the parent's PID. The writer thread, the ring buffer and LogEntry are deleted. An entry is now serialised and logged on the caller's thread, after an isEnabledFor check on cachekit.logging, so a disabled level costs one level check and nothing is sampled or serialised. The pid field is read per entry. A value json cannot encode is written as its type name, never its str(), instead of the whole entry being dropped. Importing cachekit no longer starts a thread. The enable_structured_logging and Structured Logging docs described JSON output the flag does not produce; they now say what each kind of record is. BREAKING CHANGE: CachekitConfig.log_buffer_size, log_batch_size and log_flush_interval are removed, with their env vars CACHEKIT_LOG_BUFFER_SIZE, CACHEKIT_LOG_BATCH_SIZE and CACHEKIT_LOG_FLUSH_INTERVAL; they configured the deleted writer. Passing one to the constructor now raises ValidationError. A still-exported env var is ignored, so startup is unaffected. --- docs/api-reference.md | 11 +- src/cachekit/config/nested.py | 2 +- src/cachekit/config/settings.py | 15 -- src/cachekit/logging.py | 201 +++++------------------- tests/performance/ir_budget.py | 9 +- tests/unit/test_config_env_fallback.py | 3 + tests/unit/test_l1_memory_bounds.py | 4 - tests/unit/test_structured_logging.py | 208 ++++++++++++++++++++++++- tests/utils/tracemalloc_isolation.py | 11 +- 9 files changed, 260 insertions(+), 204 deletions(-) diff --git a/docs/api-reference.md b/docs/api-reference.md index 852b514c..430915b2 100644 --- a/docs/api-reference.md +++ b/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 records never are. Nothing is sampled or serialised below that logger's level. A forked child writes its own records, with its own `pid`. ## Best Practices diff --git a/src/cachekit/config/nested.py b/src/cachekit/config/nested.py index 853ccfda..4db5aea2 100644 --- a/src/cachekit/config/nested.py +++ b/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: diff --git a/src/cachekit/config/settings.py b/src/cachekit/config/settings.py index 6188bcee..82637657 100644 --- a/src/cachekit/config/settings.py +++ b/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( diff --git a/src/cachekit/logging.py b/src/cachekit/logging.py index 554c9efc..1d001dff 100644 --- a/src/cachekit/logging.py +++ b/src/cachekit/logging.py @@ -1,7 +1,8 @@ -"""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. """ import json @@ -11,8 +12,7 @@ import random import threading import time -from dataclasses import dataclass, field -from typing import Any, Optional +from typing import Any from cachekit.config import get_settings from cachekit.hash_utils import redact_error_for_log, redact_key_for_log @@ -21,145 +21,37 @@ 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, "WARNING": logging.WARNING, "ERROR": logging.ERROR} -@dataclass -class LogEntry: - """Lightweight log entry for ring buffer.""" - 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, - } - - -class LockFreeRingBuffer: - """Lock-free ring buffer for log entries. - - 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 _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 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 is 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 +71,26 @@ 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 = { + "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(), + **self._mask_pii(fields), + } + logger.log(levelno, json.dumps(entry, separators=(",", ":"), default=_type_name)) + 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 logger.isEnabledFor(levelno) and self._should_sample(): + self._write(level, levelno, message, kwargs) def debug(self, message: str, **kwargs): """Debug level logging.""" @@ -217,21 +105,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.""" + if logger.isEnabledFor(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.""" @@ -335,11 +211,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] = {} diff --git a/tests/performance/ir_budget.py b/tests/performance/ir_budget.py index a2e9ff93..452e8cd9 100644 --- a/tests/performance/ir_budget.py +++ b/tests/performance/ir_budget.py @@ -222,11 +222,10 @@ def measure(paths: list[str], jobs: int, child_timeout_s: float = CHILD_TIMEOUT_ "PYTHONNOUSERSITE": "1", "OPENBLAS_NUM_THREADS": "1", "OMP_NUM_THREADS": "1", # pyarrow converts on a thread pool otherwise: +-0.5% per op - # cachekit's log writer wakes every 1 s and its L1 sweeper every 30 s of real time. A woken - # thread takes the GIL at a load-dependent moment, and the sweeper compares expiry times - # stamped by the pinned main-thread clock against the real one, so it evicts live entries - # mid-run. A day-long interval keeps both asleep for the whole run. - "CACHEKIT_LOG_FLUSH_INTERVAL": "86400", + # cachekit's L1 sweeper wakes every 30 s of real time. A woken thread takes the GIL at a + # load-dependent moment, and the sweeper compares expiry times stamped by the pinned + # main-thread clock against the real one, so it evicts live entries mid-run. A day-long + # interval keeps it asleep for the whole run. "CACHEKIT_L1_CLEANUP_INTERVAL_SECONDS": "86400", } global _stopped diff --git a/tests/unit/test_config_env_fallback.py b/tests/unit/test_config_env_fallback.py index ca0538d3..7dc9c4e2 100644 --- a/tests/unit/test_config_env_fallback.py +++ b/tests/unit/test_config_env_fallback.py @@ -115,6 +115,9 @@ def test_removed_knob_env_vars_are_ignored(self, monkeypatch): "L1_ENABLED": "false", "ENABLE_PROMETHEUS_METRICS": "false", "BACKEND_PROVIDER_CLASS": "cachekit.backends.redis.provider.RedisBackendProvider", + "LOG_BUFFER_SIZE": "10000", + "LOG_BATCH_SIZE": "100", + "LOG_FLUSH_INTERVAL": "86400", } for name, value in removed.items(): monkeypatch.setenv(f"CACHEKIT_{name}", value) diff --git a/tests/unit/test_l1_memory_bounds.py b/tests/unit/test_l1_memory_bounds.py index c25eb7c2..b21876cf 100644 --- a/tests/unit/test_l1_memory_bounds.py +++ b/tests/unit/test_l1_memory_bounds.py @@ -333,7 +333,6 @@ def _c_fork_child_outcome() -> object: import faulthandler from cachekit import l1_cache - from cachekit import logging as structured_logging manager = L1CacheManager(default_max_memory_mb=10) cache = manager.get_cache("c-fork-ns") # captured pre-fork, like a decorator's _l1_cache @@ -343,9 +342,6 @@ def _c_fork_child_outcome() -> object: assert cleanup is not None manager._stop_cleanup.set() # ends it, still referenced: the parent ran cleanup cleanup.join() - for structured in structured_logging._logger_instances.values(): # their writers wake every flush interval - structured.writer.stop() - structured.writer.join() l1_logger = logging.getLogger("cachekit.l1_cache") l1_logger.addHandler(handler := logging.StreamHandler(open(os.devnull, "w"))) # noqa: SIM115 - ends with the process l1_logger.setLevel(logging.DEBUG) diff --git a/tests/unit/test_structured_logging.py b/tests/unit/test_structured_logging.py index fe98aa31..9e93ce76 100644 --- a/tests/unit/test_structured_logging.py +++ b/tests/unit/test_structured_logging.py @@ -1,7 +1,18 @@ """Unit tests for structured logging module.""" +from __future__ import annotations + +import ast +import json import logging -from unittest.mock import patch +import os +import signal +import subprocess +import sys +import threading +from pathlib import Path +from typing import NoReturn +from unittest.mock import MagicMock, patch import pytest @@ -9,6 +20,96 @@ StructuredLogger, get_structured_logger, ) +from tests.utils.fork_helpers import child_outcome, report + + +class _Capture(logging.Handler): + """Collects the JSON entries written to ``cachekit.logging``.""" + + def __init__(self) -> None: + super().__init__() + self.entries: list[dict] = [] + + def emit(self, record: logging.LogRecord) -> None: + self.entries.append(json.loads(record.getMessage())) + + +@pytest.fixture +def written(): + """Entries ``cachekit.logging`` writes while the test runs, with DEBUG enabled.""" + sink = logging.getLogger("cachekit.logging") + capture, level = _Capture(), sink.level + sink.addHandler(capture) + sink.setLevel(logging.DEBUG) + try: + yield capture.entries + finally: + sink.removeHandler(capture) + sink.setLevel(level) + + +def _c_fork_child_outcome() -> object: + """What a child forked from C reports; run only in a fresh interpreter (test_child_of_a_c_fork_starts_no_thread). + + Importing cachekit must leave this interpreter single-threaded: a thread waiting for the GIL at a + fork made from C leaves the child hung at its first GIL release, in no cachekit code. + """ + import ctypes + import faulthandler + + import cachekit # noqa: F401 - everything an application's import starts + + sink = logging.getLogger("cachekit.logging") + sink.addHandler(capture := _Capture()) + sink.setLevel(logging.DEBUG) + before = get_structured_logger("test.c_fork.parent") # built pre-fork, as a module-level _logger is + before._sampling_threshold = 100 # never sampled away + if threading.active_count() != 1: # not an assert: python -O strips those + raise RuntimeError(f"a thread would be alive at fork: {threading.enumerate()}") + faulthandler.register(signal.SIGALRM, chain=True) # the child inherits it: a hang dumps its stack + libc_fork = ctypes.PyDLL(None).fork # PyDLL keeps the GIL through the call + r, w = os.pipe() + child = libc_fork() + if child == 0: + try: + signal.alarm(5) + os.close(r) + requested: list[str] = [] + + class _RecordingThread(threading.Thread): # keeps the child single-threaded + def start(self) -> None: + requested.append(self.name) + + threading.Thread = _RecordingThread # the child's copy only; it never returns + after = get_structured_logger("test.c_fork.child") # first built in the child, as after --lazy-apps + before.debug("parent-built debug") + before.error("parent-built error") + after.error("child-built error") + outcome: object = { + "requested": requested, + "written": [(e["logger"], e["message"], e["pid"] == os.getpid()) for e in capture.entries], + } + except BaseException as e: + outcome = {"error": repr(e)} + try: + report(w, outcome) + finally: + os._exit(1) + os.close(w) + return child_outcome(child, r) + + +def _print_c_fork_child_outcome() -> NoReturn: + """Print _c_fork_child_outcome() and end without exit handlers, as its sibling in test_l1_memory_bounds does.""" + import traceback + + try: + print(repr(_c_fork_child_outcome()), flush=True) + except BaseException: + traceback.print_exc() + sys.stderr.flush() + os._exit(1) + os._exit(0) class TestStructuredLogger: @@ -86,6 +187,111 @@ def test_cache_operation_error_logging(self, mock_log, logger): assert extra["error_type"] == "TimeoutError" +class TestSynchronousEntries: + """Entries are written on the caller's thread: no writer thread, so none to lose after a fork.""" + + def test_error_is_written_before_returning_with_this_pid(self, written): + StructuredLogger("test.sync").error("boom", attempt=2) + + assert len(written) == 1 + entry = written[0] + assert (entry["level"], entry["message"], entry["logger"], entry["attempt"]) == ("ERROR", "boom", "test.sync", 2) + assert entry["pid"] == os.getpid() + + def test_pii_named_fields_are_masked(self, written): + StructuredLogger("test.mask").error("boom", api_token=12345, attempt=1) + + assert written[0]["api_token"] == "****" + assert written[0]["attempt"] == 1 + + def test_unencodable_value_is_written_as_its_type_only(self, written): + """json cannot encode it; its str() would print the key it carries (CWE-532), so only its type is written.""" + StructuredLogger("test.unencodable").error("boom", failure=ValueError("ns:tenant-42:func:app.f:args:x")) + + assert written[0]["failure"] == "" + assert "tenant-42" not in json.dumps(written) + + @pytest.mark.parametrize( + ("level", "sampled", "builds"), + [(logging.DEBUG, True, True), (logging.DEBUG, False, False), (logging.INFO, True, False)], + ids=["enabled-sampled", "enabled-sampled-out", "disabled"], + ) + def test_debug_is_built_only_when_enabled_and_sampled(self, monkeypatch, caplog, level, sampled, builds): + """Below the logger's level nothing is sampled or serialised: debug call sites cost a level check.""" + caplog.set_level(level, logger="cachekit.logging") + structured = StructuredLogger("test.gate") + sample = MagicMock(return_value=sampled) + write = MagicMock() + monkeypatch.setattr(structured, "_should_sample", sample) + monkeypatch.setattr(structured, "_write", write) + + structured.debug("detail", attempt=1) + + assert sample.called is (level == logging.DEBUG) + assert write.called is builds + + def test_error_is_never_sampled(self, monkeypatch, written): + structured = StructuredLogger("test.unsampled") + monkeypatch.setattr(structured, "_should_sample", lambda: False) + + structured.error("boom") + structured.warning("dropped") + + assert [e["message"] for e in written] == ["boom"] + + def test_constructing_a_logger_starts_no_thread(self, monkeypatch): + started: list[str] = [] + monkeypatch.setattr(threading.Thread, "start", lambda self: started.append(self.name)) + + StructuredLogger("test.threadless").error("boom") + + assert started == [] + + +@pytest.mark.skipif(not hasattr(os, "fork"), reason="fork() not available on this platform") +class TestEntriesAfterFork: + """A forked child writes what it logs, with its own PID, and starts no thread doing so.""" + + def test_os_fork_child_writes_its_error_with_its_own_pid(self, written): + structured = get_structured_logger("test.os_fork") # built pre-fork, as a module-level _logger is + r, w = os.pipe() + child = os.fork() + if child == 0: + try: + os.close(r) + written.clear() # the child's copy + structured.error("child-error") + outcome: object = [(e["message"], e["pid"]) for e in written] + except BaseException as e: + outcome = repr(e) + report(w, outcome) + os.close(w) + + assert child_outcome(child, r) == [("child-error", child)] + + def test_child_of_a_c_fork_starts_no_thread(self): + """A real fork from C: no at-fork hook runs. The child writes every entry, and requests no thread. + + The fork runs in a fresh interpreter (_c_fork_child_outcome): this one has threads it cannot end + (pytest-xdist's, other tests'), and one waiting for the GIL at fork hangs the child. + """ + code = "from tests.unit.test_structured_logging import _print_c_fork_child_outcome as f; f()" + root = Path(__file__).resolve().parents[2] # where `tests` imports from + run = subprocess.run( # noqa: S603 - trusted: sys.executable + literal code + [sys.executable, "-c", code], cwd=root, capture_output=True, text=True, timeout=60 + ) + + assert run.returncode == 0, run.stderr + assert ast.literal_eval(run.stdout) == { + "requested": [], + "written": [ + ("test.c_fork.parent", "parent-built debug", True), + ("test.c_fork.parent", "parent-built error", True), + ("test.c_fork.child", "child-built error", True), + ], + }, run.stderr # a hung child's stack, dumped on its alarm + + class TestFactoryFunction: """Test factory function.""" diff --git a/tests/utils/tracemalloc_isolation.py b/tests/utils/tracemalloc_isolation.py index 29eedf01..eb2b80bb 100644 --- a/tests/utils/tracemalloc_isolation.py +++ b/tests/utils/tracemalloc_isolation.py @@ -3,8 +3,8 @@ tracemalloc.start()/stop() swap the process-wide allocator hooks without synchronising with threads that are allocating at that moment, so on a free-threaded build a concurrent allocation can crash the process: https://github.com/python/cpython/issues/143143. The pytest -process is never single-threaded (``import cachekit`` starts the log writer; pytest-xdist adds -its I/O thread), so every tracemalloc start/stop in the test suite goes through here. +process is never single-threaded (pytest-xdist adds its I/O thread), so every tracemalloc +start/stop in the test suite goes through here. """ from __future__ import annotations @@ -19,14 +19,9 @@ import pytest -from cachekit import logging as ck_logging - def _run_alone(measure: Callable[[], Any]) -> Any: - """Subprocess side: stop cachekit's log writers, then measure with no other thread alive.""" - for structured in ck_logging._logger_instances.values(): - structured.writer.stop() - structured.writer.join() + """Subprocess side: measure with no other thread alive.""" assert threading.active_count() == 1, f"tracemalloc must not start or stop beside {threading.enumerate()}" result = measure() assert threading.active_count() == 1, f"a thread started while measuring: {threading.enumerate()}" From cd88b2de23412e0d29c2712afd90bf8105b8199e Mon Sep 17 00:00:00 2001 From: Ray Walker Date: Tue, 13 Oct 2026 01:02:25 +1100 Subject: [PATCH 2/5] fix(logging): log CRITICAL at its level and never raise on an unencodable field StructuredLogger.log() mapped only DEBUG/INFO/WARNING/ERROR, so CRITICAL, FATAL and WARN were written at INFO and sampled; the old writer resolved them with getattr(logging, name). They now map to their stdlib levels, and nothing at ERROR or above is sampled through log(), matching error(). json.dumps now runs on the caller's thread, so a field it raises on (a non-str dict key, a cycle, a too-deep nest) reached the caller and could replace the exception it was logging. Such a field is now written as its type name, like any other value json cannot encode, and the rest of the entry is kept. Caller fields are merged before the generated ones, so a pid= keyword can no longer make an entry report another process. --- docs/api-reference.md | 2 +- src/cachekit/logging.py | 36 +++++++++++++--- tests/unit/test_structured_logging.py | 62 ++++++++++++++++++++++++++- 3 files changed, 91 insertions(+), 9 deletions(-) diff --git a/docs/api-reference.md b/docs/api-reference.md index 430915b2..4fb8abf5 100644 --- a/docs/api-reference.md +++ b/docs/api-reference.md @@ -844,7 +844,7 @@ cachekit writes two kinds of structured record through Python's `logging`, so yo 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 records never are. Nothing is sampled or serialised below that logger's level. A forked child writes its own records, with its own `pid`. +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 `""`, never as its text. Nothing is sampled or serialised below that logger's level. A forked child writes its own records, with its own `pid`. ## Best Practices diff --git a/src/cachekit/logging.py b/src/cachekit/logging.py index 1d001dff..ba3aaf4b 100644 --- a/src/cachekit/logging.py +++ b/src/cachekit/logging.py @@ -27,7 +27,18 @@ # Performance and health thresholds HIGH_UTILIZATION_THRESHOLD = 0.9 # When to warn about high utilization -_LEVELS = {"DEBUG": logging.DEBUG, "INFO": logging.INFO, "WARNING": logging.WARNING, "ERROR": logging.ERROR} +_LEVELS = { + "DEBUG": logging.DEBUG, + "INFO": logging.INFO, + "WARN": logging.WARNING, + "WARNING": logging.WARNING, + "ERROR": logging.ERROR, + "CRITICAL": logging.CRITICAL, + "FATAL": logging.CRITICAL, +} + +# 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) def _type_name(value: object) -> str: @@ -35,6 +46,15 @@ def _type_name(value: object) -> str: return f"<{type(value).__name__}>" +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 + + class StructuredLogger: """Structured logger. @@ -42,7 +62,7 @@ class StructuredLogger: Features: - Nothing is sampled or serialised below that logger's level - - Sampling (10% default) of DEBUG/INFO/WARNING; ERROR is never sampled + - Sampling (10% default) of DEBUG/INFO/WARNING; ERROR and CRITICAL are never sampled - PII key-name masking (password/token/secret/key/auth kwargs) """ @@ -73,7 +93,8 @@ def _mask_pii(self, data: dict[str, Any]) -> dict[str, Any]: def _write(self, level: str, levelno: int, message: str, fields: dict[str, Any]) -> None: """Serialise one entry and log it now.""" - entry = { + 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, @@ -81,15 +102,18 @@ def _write(self, level: str, levelno: int, message: str, fields: dict[str, Any]) "hostname": self._hostname, "pid": os.getpid(), # read per entry: a forked child reports its own "thread_id": threading.get_ident(), - **self._mask_pii(fields), } - logger.log(levelno, json.dumps(entry, separators=(",", ":"), default=_type_name)) + 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.""" level = level.upper() levelno = _LEVELS.get(level, logging.INFO) - if logger.isEnabledFor(levelno) and self._should_sample(): + if logger.isEnabledFor(levelno) and (levelno >= logging.ERROR or self._should_sample()): self._write(level, levelno, message, kwargs) def debug(self, message: str, **kwargs): diff --git a/tests/unit/test_structured_logging.py b/tests/unit/test_structured_logging.py index 9e93ce76..2a3b8b5e 100644 --- a/tests/unit/test_structured_logging.py +++ b/tests/unit/test_structured_logging.py @@ -112,6 +112,19 @@ def _print_c_fork_child_outcome() -> NoReturn: os._exit(0) +def _cycle() -> list: + cycle: list = ["ns:tenant-42:func:app.f:args:x"] + cycle.append(cycle) + return cycle + + +def _too_deep() -> list: + nest: list = ["tenant-42"] + for _ in range(100_000): # past json's recursion limit on every supported interpreter; 10_000 is not, on 3.14 + nest = [nest] + return nest + + class TestStructuredLogger: """Test StructuredLogger functionality.""" @@ -211,6 +224,33 @@ def test_unencodable_value_is_written_as_its_type_only(self, written): assert written[0]["failure"] == "" assert "tenant-42" not in json.dumps(written) + @pytest.mark.parametrize( + ("build", "type_name"), + [ + (lambda: {(1, 2): "ns:tenant-42:func:app.f:args:x"}, ""), + (_cycle, ""), + (_too_deep, ""), + ], + ids=["tuple-key", "cycle", "too-deep"], + ) + def test_field_json_raises_on_is_written_as_its_type_only(self, written, build, type_name): + """json raises on it (TypeError, ValueError, RecursionError): the entry is still written, its other fields intact. + + Raising instead would replace the exception a caller is logging from inside its except block. + """ + StructuredLogger("test.unwalkable").error("boom", detail=build(), attempt=1) + + assert (written[0]["message"], written[0]["detail"], written[0]["attempt"]) == ("boom", type_name, 1) + assert "tenant-42" not in json.dumps(written) + + def test_generated_fields_win_over_same_named_caller_fields(self, written): + """A caller's pid= cannot make the entry report a process other than the one that wrote it.""" + StructuredLogger("test.reserved").error("boom", pid=1, thread_id=2, logger="spoofed", level="DEBUG") + + entry = written[0] + assert (entry["pid"], entry["thread_id"]) == (os.getpid(), threading.get_ident()) + assert (entry["logger"], entry["level"]) == ("test.reserved", "ERROR") + @pytest.mark.parametrize( ("level", "sampled", "builds"), [(logging.DEBUG, True, True), (logging.DEBUG, False, False), (logging.INFO, True, False)], @@ -230,14 +270,32 @@ def test_debug_is_built_only_when_enabled_and_sampled(self, monkeypatch, caplog, assert sample.called is (level == logging.DEBUG) assert write.called is builds - def test_error_is_never_sampled(self, monkeypatch, written): + def test_error_and_above_are_never_sampled(self, monkeypatch, written): structured = StructuredLogger("test.unsampled") monkeypatch.setattr(structured, "_should_sample", lambda: False) structured.error("boom") + structured.log("error", "via log") + structured.log("CRITICAL", "critical") structured.warning("dropped") - assert [e["message"] for e in written] == ["boom"] + assert [e["message"] for e in written] == ["boom", "via log", "critical"] + + @pytest.mark.parametrize( + ("name", "levelno"), + [("warn", logging.WARNING), ("CRITICAL", logging.CRITICAL), ("FATAL", logging.CRITICAL), ("NOTICE", logging.INFO)], + ) + def test_log_writes_a_level_name_at_its_stdlib_level(self, monkeypatch, caplog, name, levelno): + """Each stdlib level name maps to its own level, as the old writer's getattr(logging, name) did; others are INFO.""" + caplog.set_level(logging.DEBUG, logger="cachekit.logging") + structured = StructuredLogger("test.levels") + monkeypatch.setattr(structured, "_should_sample", lambda: True) + + structured.log(name, "boom") + + assert [(r.levelno, json.loads(r.getMessage())["level"]) for r in caplog.records if r.name == "cachekit.logging"] == [ + (levelno, name.upper()) + ] def test_constructing_a_logger_starts_no_thread(self, monkeypatch): started: list[str] = [] From 91c283814ea7242f5d69b7c22a43de40cb4fc5d2 Mon Sep 17 00:00:00 2001 From: Ray Walker Date: Tue, 13 Oct 2026 01:08:29 +1100 Subject: [PATCH 3/5] test(logging): trip RecursionError through json's pure-Python encoder The C encoder's depth limit is the thread's stack since 3.14, so a 100_000-deep nest encodes on a runner with an unlimited stack and the too-deep case failed on the free-threaded lane. The pure-Python encoder stops at sys.getrecursionlimit() on every supported interpreter. --- tests/unit/test_structured_logging.py | 18 ++++++++++++++---- 1 file changed, 14 insertions(+), 4 deletions(-) diff --git a/tests/unit/test_structured_logging.py b/tests/unit/test_structured_logging.py index 2a3b8b5e..b635b4d5 100644 --- a/tests/unit/test_structured_logging.py +++ b/tests/unit/test_structured_logging.py @@ -120,7 +120,7 @@ def _cycle() -> list: def _too_deep() -> list: nest: list = ["tenant-42"] - for _ in range(100_000): # past json's recursion limit on every supported interpreter; 10_000 is not, on 3.14 + for _ in range(sys.getrecursionlimit() + 100): nest = [nest] return nest @@ -229,12 +229,11 @@ def test_unencodable_value_is_written_as_its_type_only(self, written): [ (lambda: {(1, 2): "ns:tenant-42:func:app.f:args:x"}, ""), (_cycle, ""), - (_too_deep, ""), ], - ids=["tuple-key", "cycle", "too-deep"], + ids=["tuple-key", "cycle"], ) def test_field_json_raises_on_is_written_as_its_type_only(self, written, build, type_name): - """json raises on it (TypeError, ValueError, RecursionError): the entry is still written, its other fields intact. + """json raises on it (TypeError, ValueError): the entry is still written, its other fields intact. Raising instead would replace the exception a caller is logging from inside its except block. """ @@ -243,6 +242,17 @@ def test_field_json_raises_on_is_written_as_its_type_only(self, written, build, assert (written[0]["message"], written[0]["detail"], written[0]["attempt"]) == ("boom", type_name, 1) assert "tenant-42" not in json.dumps(written) + def test_field_nested_past_the_recursion_limit_is_written_as_its_type_only(self, monkeypatch, written): + """RecursionError, forced through json's pure-Python encoder, whose limit is sys.getrecursionlimit() everywhere. + + The C encoder's limit is the thread's stack: on an unlimited stack, 3.14 encodes a 100_000-deep nest. + """ + monkeypatch.setattr(json.encoder, "c_make_encoder", None) + + StructuredLogger("test.too_deep").error("boom", detail=_too_deep(), attempt=1) + + assert (written[0]["message"], written[0]["detail"], written[0]["attempt"]) == ("boom", "", 1) + def test_generated_fields_win_over_same_named_caller_fields(self, written): """A caller's pid= cannot make the entry report a process other than the one that wrote it.""" StructuredLogger("test.reserved").error("boom", pid=1, thread_id=2, logger="spoofed", level="DEBUG") From c3b0f6c7464dad1cde6550ace4efc564a3b82087 Mon Sep 17 00:00:00 2001 From: Ray Walker Date: Tue, 13 Oct 2026 02:01:57 +1100 Subject: [PATCH 4/5] fix(logging): write no structured entry in a child forked without at-fork hooks A fork made from C (uWSGI without --py-call-uwsgi-fork-hooks) skips logging's after-fork lock reset. If a parent thread held logging's module lock or a handler lock at the fork, the child's first structured entry hung: the level check takes the module lock when its level is not cached, and the handler takes its own. Before entries were written synchronously that child only queued them, so the hang is new with this branch. Such a child is detected the way the L1 take-over and the invalidation listener already detect it (l1_cache._forked_without_hooks), and writes nothing there, as those paths log nothing there. A child forked by Python (os.fork, multiprocessing, Gunicorn, Celery) is unaffected and writes its entries with its own pid. The C-fork test now holds logging's module lock and the handler's lock in threads that end before the fork, with no level check cached; without the guard the child hangs until its alarm. --- docs/api-reference.md | 2 +- src/cachekit/logging.py | 14 ++++++++-- tests/unit/test_structured_logging.py | 37 ++++++++++++--------------- 3 files changed, 29 insertions(+), 24 deletions(-) diff --git a/docs/api-reference.md b/docs/api-reference.md index 4fb8abf5..5314a695 100644 --- a/docs/api-reference.md +++ b/docs/api-reference.md @@ -844,7 +844,7 @@ cachekit writes two kinds of structured record through Python's `logging`, so yo 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 `""`, never as its text. Nothing is sampled or serialised below that logger's level. A forked child writes its own records, with its own `pid`. +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 `""`, 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 diff --git a/src/cachekit/logging.py b/src/cachekit/logging.py index ba3aaf4b..3ef33a70 100644 --- a/src/cachekit/logging.py +++ b/src/cachekit/logging.py @@ -3,6 +3,10 @@ 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. + +A child forked from C without at-fork hooks (uWSGI without ``--py-call-uwsgi-fork-hooks``) writes 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 @@ -14,6 +18,7 @@ import time 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 @@ -55,6 +60,11 @@ def _encodable(value: object) -> object: return value +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. @@ -113,7 +123,7 @@ def log(self, level: str, message: str, **kwargs): """Main logging method with sampling.""" level = level.upper() levelno = _LEVELS.get(level, logging.INFO) - if logger.isEnabledFor(levelno) and (levelno >= logging.ERROR or self._should_sample()): + if _enabled(levelno) and (levelno >= logging.ERROR or self._should_sample()): self._write(level, levelno, message, kwargs) def debug(self, message: str, **kwargs): @@ -130,7 +140,7 @@ def warning(self, message: str, **kwargs): def error(self, message: str, **kwargs): """Error level logging - never sampled.""" - if logger.isEnabledFor(logging.ERROR): + if _enabled(logging.ERROR): self._write("ERROR", logging.ERROR, message, kwargs) def cache_operation(self, operation: str, cache_key: str, **kwargs): diff --git a/tests/unit/test_structured_logging.py b/tests/unit/test_structured_logging.py index b635b4d5..8f474679 100644 --- a/tests/unit/test_structured_logging.py +++ b/tests/unit/test_structured_logging.py @@ -49,10 +49,12 @@ def written(): def _c_fork_child_outcome() -> object: - """What a child forked from C reports; run only in a fresh interpreter (test_child_of_a_c_fork_starts_no_thread). + """What a child forked from C reports; run only in a fresh interpreter (TestEntriesAfterFork's C-fork test). Importing cachekit must leave this interpreter single-threaded: a thread waiting for the GIL at a - fork made from C leaves the child hung at its first GIL release, in no cachekit code. + fork made from C leaves the child hung at its first GIL release, in no cachekit code. logging's + module lock and the handler's lock end held by threads the child lacks, and no level check is + cached: in the child, any call that takes either lock hangs, as no after-fork reset ran there. """ import ctypes import faulthandler @@ -64,6 +66,10 @@ def _c_fork_child_outcome() -> object: sink.setLevel(logging.DEBUG) before = get_structured_logger("test.c_fork.parent") # built pre-fork, as a module-level _logger is before._sampling_threshold = 100 # never sampled away + for lock in (logging._lock, capture.lock): # held at fork by a thread the child lacks + holder = threading.Thread(target=lock.acquire) # ends holding it + holder.start() + holder.join() if threading.active_count() != 1: # not an assert: python -O strips those raise RuntimeError(f"a thread would be alive at fork: {threading.enumerate()}") faulthandler.register(signal.SIGALRM, chain=True) # the child inherits it: a hang dumps its stack @@ -81,14 +87,9 @@ def start(self) -> None: requested.append(self.name) threading.Thread = _RecordingThread # the child's copy only; it never returns - after = get_structured_logger("test.c_fork.child") # first built in the child, as after --lazy-apps - before.debug("parent-built debug") - before.error("parent-built error") - after.error("child-built error") - outcome: object = { - "requested": requested, - "written": [(e["logger"], e["message"], e["pid"] == os.getpid()) for e in capture.entries], - } + before.debug("child debug") + before.error("child error") + outcome: object = {"requested": requested, "written": capture.entries} except BaseException as e: outcome = {"error": repr(e)} try: @@ -337,10 +338,11 @@ def test_os_fork_child_writes_its_error_with_its_own_pid(self, written): assert child_outcome(child, r) == [("child-error", child)] - def test_child_of_a_c_fork_starts_no_thread(self): - """A real fork from C: no at-fork hook runs. The child writes every entry, and requests no thread. + def test_child_of_a_c_fork_writes_nothing_and_starts_no_thread(self): + """A real fork from C: no at-fork hook runs, so logging's locks may be held by threads the child lacks. - The fork runs in a fresh interpreter (_c_fork_child_outcome): this one has threads it cannot end + The child's calls return without taking them: they write nothing and request no thread. The fork + runs in a fresh interpreter (_c_fork_child_outcome): this one has threads it cannot end (pytest-xdist's, other tests'), and one waiting for the GIL at fork hangs the child. """ code = "from tests.unit.test_structured_logging import _print_c_fork_child_outcome as f; f()" @@ -350,14 +352,7 @@ def test_child_of_a_c_fork_starts_no_thread(self): ) assert run.returncode == 0, run.stderr - assert ast.literal_eval(run.stdout) == { - "requested": [], - "written": [ - ("test.c_fork.parent", "parent-built debug", True), - ("test.c_fork.parent", "parent-built error", True), - ("test.c_fork.child", "child-built error", True), - ], - }, run.stderr # a hung child's stack, dumped on its alarm + assert ast.literal_eval(run.stdout) == {"requested": [], "written": []}, run.stderr # a hung child's stack class TestFactoryFunction: From df6e61405349e484bfa9cbd17a642b8f0fd3ab6f Mon Sep 17 00:00:00 2001 From: Ray Walker Date: Tue, 13 Oct 2026 02:59:17 +1100 Subject: [PATCH 5/5] fix(logging): drop the log writer from the performance docs; tidy StructuredLogger docs/performance.md and the instruction-budget harness still described the deleted log writer thread and its one-day flush interval. Both now name only the L1 cleanup worker, and the harness no longer says that import cachekit starts a thread. The main merge brought an SWR fork test that stopped the log writers before forking; there are none to stop now. The module docstring now says that log() and error() write no entry in a child forked without at-fork hooks, rather than every StructuredLogger method: cache_operation() and connection_pool_utilization() call logging as they did before. The C-fork test uses the shared run_fresh helper. cache_operation() logs through self.logger with the module's level table instead of a local import, a dead five-level map and a second getLogger. error() keeps its own _enabled() gate: routing it through log() would make a caller field named level or message a TypeError. --- docs/performance.md | 4 ++-- src/cachekit/logging.py | 24 +++++------------------ tests/performance/ir_budget.py | 4 ++-- tests/unit/test_structured_logging.py | 28 +++------------------------ tests/unit/test_swr_decorator.py | 4 ---- 5 files changed, 12 insertions(+), 52 deletions(-) diff --git a/docs/performance.md b/docs/performance.md index ff5f0969..94685136 100644 --- a/docs/performance.md +++ b/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. diff --git a/src/cachekit/logging.py b/src/cachekit/logging.py index 3ef33a70..a403d8a9 100644 --- a/src/cachekit/logging.py +++ b/src/cachekit/logging.py @@ -4,9 +4,9 @@ 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. -A child forked from C without at-fork hooks (uWSGI without ``--py-call-uwsgi-fork-hooks``) writes 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. +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 @@ -139,7 +139,7 @@ def warning(self, message: str, **kwargs): self.log("WARNING", message, **kwargs) def error(self, message: str, **kwargs): - """Error level logging - never sampled.""" + """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) @@ -203,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.""" diff --git a/tests/performance/ir_budget.py b/tests/performance/ir_budget.py index 452e8cd9..1c3bf2b6 100644 --- a/tests/performance/ir_budget.py +++ b/tests/performance/ir_budget.py @@ -3,8 +3,8 @@ Wall-clock benchmarks on a shared host cannot gate a 1% regression: run-to-run noise is several percent. Instruction counts can, once these sources of run-to-run noise are handled: -- ``import cachekit`` starts background threads (the log writer, the L1 cleanup worker) whose - counts swing by tens of percent between identical runs. Only the MAIN thread is counted +- cachekit runs background threads (the L1 cleanup worker) whose counts swing by tens of + percent between identical runs. Only the MAIN thread is counted (callgrind ``--separate-threads=yes``, file ``-01``). Work a library runs on its own threads is not counted either: about half of the real-data Arrow round trip runs on pyarrow's threads. - Interpreter startup is ~2 billion instructions. Each path runs at two loop sizes and the diff --git a/tests/unit/test_structured_logging.py b/tests/unit/test_structured_logging.py index 8f474679..036941dc 100644 --- a/tests/unit/test_structured_logging.py +++ b/tests/unit/test_structured_logging.py @@ -7,11 +7,8 @@ import logging import os import signal -import subprocess import sys import threading -from pathlib import Path -from typing import NoReturn from unittest.mock import MagicMock, patch import pytest @@ -20,7 +17,7 @@ StructuredLogger, get_structured_logger, ) -from tests.utils.fork_helpers import child_outcome, report +from tests.utils.fork_helpers import child_outcome, report, run_fresh class _Capture(logging.Handler): @@ -100,19 +97,6 @@ def start(self) -> None: return child_outcome(child, r) -def _print_c_fork_child_outcome() -> NoReturn: - """Print _c_fork_child_outcome() and end without exit handlers, as its sibling in test_l1_memory_bounds does.""" - import traceback - - try: - print(repr(_c_fork_child_outcome()), flush=True) - except BaseException: - traceback.print_exc() - sys.stderr.flush() - os._exit(1) - os._exit(0) - - def _cycle() -> list: cycle: list = ["ns:tenant-42:func:app.f:args:x"] cycle.append(cycle) @@ -341,15 +325,9 @@ def test_os_fork_child_writes_its_error_with_its_own_pid(self, written): def test_child_of_a_c_fork_writes_nothing_and_starts_no_thread(self): """A real fork from C: no at-fork hook runs, so logging's locks may be held by threads the child lacks. - The child's calls return without taking them: they write nothing and request no thread. The fork - runs in a fresh interpreter (_c_fork_child_outcome): this one has threads it cannot end - (pytest-xdist's, other tests'), and one waiting for the GIL at fork hangs the child. + The child's calls return without taking them: they write nothing and request no thread. """ - code = "from tests.unit.test_structured_logging import _print_c_fork_child_outcome as f; f()" - root = Path(__file__).resolve().parents[2] # where `tests` imports from - run = subprocess.run( # noqa: S603 - trusted: sys.executable + literal code - [sys.executable, "-c", code], cwd=root, capture_output=True, text=True, timeout=60 - ) + run = run_fresh(_c_fork_child_outcome) assert run.returncode == 0, run.stderr assert ast.literal_eval(run.stdout) == {"requested": [], "written": []}, run.stderr # a hung child's stack diff --git a/tests/unit/test_swr_decorator.py b/tests/unit/test_swr_decorator.py index b61d4c28..c278e52e 100644 --- a/tests/unit/test_swr_decorator.py +++ b/tests/unit/test_swr_decorator.py @@ -342,7 +342,6 @@ def _forked_stale_hit_outcome(fork: str) -> object: import types from cachekit import l1_cache, object_cache - from cachekit import logging as structured_logging from cachekit.config import L1CacheConfig backend = FakeSWRBackend() @@ -363,9 +362,6 @@ def l2_swr() -> int: if manager._cleanup_thread is not None: # ends it, still referenced: the parent ran cleanup manager._stop_cleanup.set() manager._cleanup_thread.join() - for structured in structured_logging._logger_instances.values(): # their writers wake every flush interval - structured.writer.stop() - structured.writer.join() if threading.active_count() != 1: # not an assert: python -O strips those raise RuntimeError(f"a thread would be alive at fork: {threading.enumerate()}") faulthandler.register(signal.SIGALRM, chain=True) # the child inherits it: a hang dumps its stack