diff --git a/README.md b/README.md index 51526fe2..05d807b2 100644 --- a/README.md +++ b/README.md @@ -128,6 +128,8 @@ def expensive_api_call(user_id: int): *cachekit.io is in closed beta — [request access](https://cachekit.io) to get started.* +During the beta the backend reports hit counts and L1 hit latency to cachekit.io by default; `CACHEKIT_TELEMETRY=false` turns that off ([Telemetry](docs/backends/cachekitio.md#telemetry)). + How fresh a cachekit.io read is, and how long a deleted project's data stays readable: [Consistency and Deletion](https://docs.cachekit.io/concepts/consistency/). diff --git a/docs/api-reference.md b/docs/api-reference.md index 68b9c5a4..9db035f2 100644 --- a/docs/api-reference.md +++ b/docs/api-reference.md @@ -193,6 +193,7 @@ def critical_business_logic(): | `CACHEKIT_API_KEY` | **Yes** | — | API key (`ck_live_...`) for authentication | | `CACHEKIT_API_URL` | No | `https://api.cachekit.io` | API endpoint override | | `CACHEKIT_TIMEOUT` | No | `5.0` | HTTP request timeout in seconds | +| `CACHEKIT_TELEMETRY` | No | unset (on) | `false` turns off the usage telemetry headers ([Telemetry](backends/cachekitio.md#telemetry)) | #### Usage diff --git a/docs/backends/cachekitio.md b/docs/backends/cachekitio.md index fb8d360f..de741774 100644 --- a/docs/backends/cachekitio.md +++ b/docs/backends/cachekitio.md @@ -37,6 +37,7 @@ backend = CachekitIOBackend( api_key="ck_live_...", # default: CACHEKIT_API_KEY api_url="https://api.cachekit.io", # default: CACHEKIT_API_URL, then api.cachekit.io timeout=5.0, # default: CACHEKIT_TIMEOUT, then 5.0 seconds + telemetry=None, # False turns telemetry off; see Telemetry below ) ``` @@ -181,8 +182,48 @@ The API URL is validated on construction — HTTPS required, credentials in the CACHEKIT_API_KEY=ck_live_... # API key — or pass api_key= to CachekitIOBackend / @cache.io CACHEKIT_API_URL=https://api.cachekit.io # Optional — defaults to api.cachekit.io CACHEKIT_TIMEOUT=5.0 # Optional — request timeout in seconds +CACHEKIT_TELEMETRY=false # Optional — turns telemetry off (see Telemetry) ``` +## Telemetry + +While cachekit.io is in closed beta, a `CachekitIOBackend` sends usage telemetry to cachekit.io by default. It +becomes opt-in when the service launches. It goes only to the API URL the backend is configured with, as headers +on the cache requests the backend already makes; nothing is sent on its own, and the Redis, Memcached and File +backends and L1-only caches send none. + +| Headers | What they carry | +|---------|-----------------| +| `X-CacheKit-Session-ID`, `X-CacheKit-Session-Start` | A per-function session id, a random per-process id joined to the decorated function's `module.qualname` (so it carries the function's name), and its start time | +| `X-CacheKit-L1-Hits`, `X-CacheKit-L2-Hits`, `X-CacheKit-Misses` | The calling function's hit and miss counts this session | +| `X-CacheKit-L1-Hit-Rate` | L1 hits / (L1 hits + L2 hits + misses), three decimals | +| `X-CacheKit-L1-Latency-Buckets`, `-Sum-Ns`, `-One-In` | How long the L1 hits timed since the previous request took (below) | + +**L1 hit latency.** Each `@cache` call through the backend is timed with probability 1/256, chosen at random before +the call runs. A timed call that is an L1 hit adds its duration, from entering the wrapper to returning the value, +to a report: sixteen duration buckets doubling from 128 ns, and their sum in nanoseconds. A duration over 60 s is +dropped and counts toward neither. The next request to +cachekit.io carries the report and empties it, so each hit is reported once. A call that waited on another call's +miss for the same key, a miss, an L2 hit and a direct `backend.get()` are never timed, and the health check never +carries a report. A forked child starts with an empty report and its own random draws. Under uWSGI without +`--py-call-uwsgi-fork-hooks`, a worker draws the same sequence as its siblings until it builds a backend or makes its +first request or timed hit on one it inherited. Each call is still timed with probability 1/256, but those workers +time the same call positions. The flag removes the window. Hits timed after the last request are not sent. + +**Turning it off.** Pass `telemetry=False`, or set `CACHEKIT_TELEMETRY` (name and value in any case) to `false`, `0`, `off` or `no`. +Off from either wins over on from the other; both unset is on. An empty or blank value counts as unset, so it leaves +telemetry on. Any other value of the variable also turns telemetry off and logs one warning. Both are read once, when the backend is constructed: set the variable before creating it. +`@cache.io()` builds its own backend, so use the variable there. + +```python notest +from cachekit.backends.cachekitio import CachekitIOBackend + +backend = CachekitIOBackend(telemetry=False) +``` + +With telemetry off, requests carry none of the headers above. They still carry `X-CacheKit-L1-Status`, which the +API requires, and a `User-Agent` naming cachekit-py and its version. + ## When to Use **Use CachekitIOBackend when**: diff --git a/docs/configuration.md b/docs/configuration.md index fe19f636..e1c7f809 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -140,6 +140,10 @@ CACHEKIT_TIMEOUT=5.0 # The same variable sizes the Redis pool, whose default is 50 CACHEKIT_CONNECTION_POOL_SIZE=32 +# Optional: Turn telemetry off: false, 0, off or no (default: unset, which is on during the closed beta) +# Off here wins over CachekitIOBackend(telemetry=True). Read once, when the backend is constructed +CACHEKIT_TELEMETRY=false + # Optional: Allow custom API hostname - disables SSRF hostname allowlist (default: false) # Only set to true when pointing at a private test server CACHEKIT_ALLOW_CUSTOM_HOST=false @@ -154,6 +158,7 @@ CACHEKIT_ALLOW_CUSTOM_HOST=false | `CACHEKIT_TIMEOUT` | `float` | `5.0` | No | Per-request timeout in seconds | | `CACHEKIT_CONNECTION_POOL_SIZE` | `int` | `32` | No | HTTP connections the backend's one pool keeps, shared by every thread and by async calls. Each concurrent request uses one (HTTP/1.1); a request that finds all in use opens a short-lived extra connection, and urllib3 logs a pool-full warning. The same variable sizes the Redis pool, whose default is 50 | | `CACHEKIT_ALLOW_CUSTOM_HOST` | `bool` | `false` | No | Disable hostname allowlist (testing only) | +| `CACHEKIT_TELEMETRY` | on/off | unset (on) | No | `false`, `0`, `off` or `no` turns off the usage telemetry headers; `true`, `1`, `on` or `yes` leaves them on; an empty or blank value counts as unset (on); any other value is off. Off from this or from `CachekitIOBackend(telemetry=False)` wins. `CACHEKIT_TELEMETRY` is not bound to the `telemetry` field; the backend combines the two when it is constructed. The name matches in any case, like the other `CACHEKIT_*` variables. See [Telemetry](backends/cachekitio.md#telemetry) | **Security notes:** - `CACHEKIT_API_URL` must use HTTPS. HTTP is rejected at startup. diff --git a/docs/free-threading.md b/docs/free-threading.md index 5ec47eca..d455a6b6 100644 --- a/docs/free-threading.md +++ b/docs/free-threading.md @@ -66,6 +66,7 @@ covered, plus what the free-threaded CI lane surfaced and sites added since: | `decorators/single_flight.py` `AsyncFlights`, `ThreadFlights` (in-process single-flight of misses: one map per decorated function, used by `wrapper.py` and `local_wrapper.py`) | One `Lock` per map over its in-flight entries + PID-owner swap, checked by every method that takes the lock (`_FlightMap._owned`: `run`, `forget`, the settle paths). An async entry holds the call's task and its waiter count, which only that task's own loop touches. A sync waiter blocks on a per-call `Lock` its starter holds until the call settles | Safe. Lookup and insert run under the map lock, and an entry is removed only by the call that put it there (identity check), so a settling call never drops a newer one. A sync caller re-checks the cache under the map lock before it starts a call (L1-only: the object cache; backed: L1, through `L1Cache.peek`), so a call that settled between the caller's lookup and its map check is not run twice; a backed caller with L1 off has nothing to re-check and reads L2 on its own trip instead. The lock order is the map lock, then the cache's state lock; nothing that holds a cache lock takes a map lock. A joining caller's `on_join` runs outside the map lock, before it waits. An async caller joins only a task on its own running loop, never another thread's. An invalidation drops the key's entries under the map lock (`forget`); their callers keep waiting on the call they joined. An async starter inserts its call only if no `forget` ran since its lookup, so a call that an eager task factory ran up to its first await before an invalidation never reaches the map after it. A sync starter is settled in a `finally` that covers its insert, so an interrupt cannot leave its waiters blocked. The fork swap is the documented-benign shape: a forked child starts with an empty map, and a thread racing the swap starts its own call. Tests in `tests/unit/test_single_flight.py` | | `decorators/write_behind.py` `WriteBehind` (opt-in `write_behind=True`: one writer per decorated function, shared by its sync and async callers) | A `_State` holding the PID, two `Condition`s on one lock (`cond` for sends ending, `wake` for threads waiting for work), the queued and in-flight jobs, the byte count, the thread and waiting-thread counts and the closed flag, published as one reference. A process-wide count of waiting threads, keyed by PID, sits under the writers' lock. Every method binds it once (`_current`), and each writer thread is started with the state it serves and touches no other | Safe. Admission, coalescing, the thread count and the queue are read and written under the state's condition. A thread that finds nothing to send gives up its place in the same hold as that check, so a submit never counts a thread that is about to quit. A submit wakes a waiting thread in the same hold as its queueing, and a thread re-checks the queue under that lock before each wait, so no wake-up is lost. The process-wide count is only ever taken inside a state's lock, never the other way round. Two sends of one entry never overlap, and `cancel` waits on the jobs in flight at the call, by identity, so a later send of the same entry cannot hold it. Fork: a child's first caller takes over under a lock from `dict.setdefault(pid, Lock())`, re-checks, and publishes a fresh `_State` in one store, so no thread pairs the child's PID with the parent's lock or queue; the PID travels with the state, as in the `client.py` row. A child forked without at-fork hooks (the detection of the `l1_cache.py` row) starts no writer thread, and its misses write on the caller. Tests in `tests/unit/test_write_behind.py` | | `l1_cache.py` `L1CacheManager._take_over_if_forked` + `_empty_caches_after_fork` | Owner-PID fast path; on a PID change one thread takes over under a lock fetched by `dict.setdefault(pid, Lock())`. An `os.register_at_fork` hook gives every cache a fresh, empty state before the child's first `get()`, so a forked child starts with an empty L1; fork mechanics and the uWSGI limit are in the `_take_over_if_forked` docstring | Safe. `setdefault` is atomic under per-object locking, so every thread in a new child gets the same fresh lock, which no parent thread can have held at fork; the rest block on it and re-check. The fast path gates on `_owner_pid` alone, so a weak-memory reader may still see the inherited manager lock once: the same benign fork-swap race as the SWR row, since that lock is unheld unless a parent thread held it at fork. The hook runs while the child is single-threaded and replaces every cache's state, held lock or not, so it probes no lock, takes none and starts no thread. It does not free the inherited states, which would copy the parent's L1 pages into every child inside `fork()`: they wait in `_inherited_states` until the child's first put, whose take-over frees them in batches, giving up the GIL between batches, on the cleanup thread it restarts or, when the parent ran none, on a one-shot thread. Each state is taken off that list with one `list.pop()`, atomic under per-object locking, before it is freed, so two releasers running at once (the threads of two managers' take-overs) never take the same state and neither finds the list emptied under it. So a child bound for exec never pays. A child that uses its L1 pays once, up front: each free also writes the allocator's bookkeeping beside the entry, so its first put copies about 2.3 times the inherited L1 (36.5 MiB for a 16 MiB L1 in `tests/performance/test_l1_fork_memory.py`), and its own L1 then reuses that memory. Kept, those pages would cost nothing while the parent left them alone, as an idle preloaded master whose entries outlive its workers does. Freeing is the default anyway because a parent with a cleanup thread rewrites them within one L1 TTL, and from then on a child that kept them would hold a private copy nothing can reach, beside its own L1. A state whose lock is taken is only dropped: one a thread forked inside its critical section holds is freed with that thread's last reference, and one a parent thread held at fork can stay for the child's life, referenced by that thread's frame. Neither the hook nor the take-over clears shared state: each publishes a fresh `_L1State` (entries, byte count and lock) in one store, and a holder that is still running finishes on the state it bound (the forking thread, if it forked inside a critical section; without hooks, a live child thread holding a lock past the take-over's 1 s probe). Invalidations re-apply if a reset replaced the state meanwhile. Without hooks, the cost of a wrong guess is that namespace's L1 entries, once per child; `--py-call-uwsgi-fork-hooks` avoids it. Once it has run, the take-over skips its own timed cache-lock probe, which could otherwise drop a cache's entries under a live child thread. In a child the hook never reached (a fork made from C, as uWSGI does without `--py-call-uwsgi-fork-hooks`; any PID other than the importing process's and the one the hook last ran in), the take-over starts no thread and `start_background_cleanup` refuses to start one, including for a manager first built in that child: such a fork also skips CPython's own after-fork repair, so a thread started there can hang in `Thread.start()` or crash the interpreter. None of that logs, the take-over's lock-reset drop included, because the same fork skips `logging`'s at-fork lock reset, and a handler lock a parent thread held would hang the child. Expired entries in that child are evicted on read or under the memory bound instead. A process that first imports cachekit after a fork made from C (uWSGI `--lazy-apps` without `--py-call-uwsgi-fork-hooks`) is detected at that import (`cachekit/_fork.py`) when its parent ran Python threads at the fork: `threading` still lists them, under native thread ids that are no task of this process (`/proc/self/task`), where an `os.fork()` child's after-fork hook would have dropped them. Such a process has no import PID, so every PID of it counts as hookless and every gate above applies; an `os.fork()` child of it runs the hooks and is safe again. When its parent ran no threads it looks like a fresh process and starts the cleanup thread, which is safe: no parent thread held anything at the fork. The check is Linux only and sees only threads `threading` registered. When it cannot tell (no `/proc/self/task`, or the current thread's own id absent from it: another PID namespace, or a patched `get_native_id`) it takes the process for a fresh one. gevent green threads are listed under the native id of the thread they ran on, so a parent's green threads alone do not count. The current and main threads never count, nor any thread under either's native id, because after a fork the forking thread's object may keep the parent's native id: always after a fork made from C, and after `os.fork()` before Python 3.13. So a fork made from C from a thread other than the main one loses the parent's main thread as evidence; uWSGI forks from its main thread. It lists threads under `threading`'s `_active_limbo_lock`, so a parent thread that held that lock at the fork, which a thread does only while it starts, ends or lists threads, hangs the import, as `logging`'s module lock already can. Tests in `tests/unit/test_fork_import_signal.py` | +| `decorators/l1_latency.py` `L1LatencyReport` + `sampling_rng` | One `_State` (lock, PID, buckets, sum) per report, published by one assignment; `add`, `record` and `take` hold its lock. A PID change is taken over under a per-PID lock from `dict.setdefault`; every report construction and use reseeds the private sampling generator once per process under that lock; an `os.register_at_fork` hook reseeds it too | Safe. Racing first uses in a child agree on one new state, and no child thread takes a lock a parent thread may have held at fork. A fork made from C (uWSGI without `--py-call-uwsgi-fork-hooks`) replays the parent's draws until the child builds a backend or makes its first request or timed hit on an inherited one; each call keeps its 1/256 chance, but such workers time the same call positions until then: no per-draw PID check, since `os.getpid()` is a system call costing several draws on the hit path. Tests in `tests/unit/backends/test_cachekitio_telemetry.py` | | `backends/cachekitio/client.py` client cache + `CachekitIOBackend._own_lease` | One urllib3 connection pool per config, shared by every thread of a process (urllib3's pool is thread-safe: a `LifoQueue` of connections, each request on its own connection). The process-wide lease cache is a `WeakValueDictionary` under a `Lock`, both held by one `_Leases` object that carries the PID that built it; each lease records its PID too, and the backend re-leases on a mismatch, or once `close_http_clients()` has closed its client, checked per request; the new lease is in the cache, so a later close reaches it. Async methods send on the same client through `asyncio.to_thread` | Safe across threads, with and without the GIL: `tests/unit/backends/test_cachekitio_thread_share.py` runs 8 threads on one backend, sync and through `to_thread`, and expects zero failures on 3.14 and on 3.14t with the GIL off. Safe across fork. A forked child never sends on or reads from its parent's pooled connections, which share the parent's TLS sessions: its first request finds the cache's PID is not its own, replaces the cache and its lock with new ones, and re-leases a new client, whether it was forked by `os.fork()` or from C with no at-fork hooks (uWSGI). An owner-PID check rather than an at-fork hook for that reason. The PID travels with the cache and its lock in one published reference, and with the client in the lease (single-assignment publication), so no thread pairs a new PID with an inherited lock or client, and no child thread takes the parent's lock, which a parent thread may have held at fork. Inherited clients are dropped unclosed: `close()` takes the pool's queue lock, for the same reason, and garbage collection closes the child's copies of the sockets without sending anything. Two child threads racing the first request may each build a cache and a client; the one replaced is simply not shared, and each request holds its lease until it returns, so that replacement never closes a client under an in-flight request. A child runs async calls on an event loop of its own (`asyncio.run`): a child that keeps running its parent's loop object waits forever on that loop's inherited default executor, whose worker threads did not survive the fork, as every async decorator's L2 operation already did. In a child forked from C, any logging call whose level check is not already cached, urllib3's per-request `DEBUG` check included, waits on `logging`'s module lock, and hangs if a parent thread held it at fork; a logging reconfiguration clears those caches. `--py-call-uwsgi-fork-hooks` avoids this. Tests in `tests/unit/backends/test_cachekitio_fork.py` | | `invalidation.py` dispatch map, start lock and listener thread | `_evictors` (registry id → `WeakSet` of evictors) under a per-PID `Lock` from `dict.setdefault` (`_dispatch_locks`): decoration registers under it, and the listener thread takes a snapshot under it and runs the evictors outside it. The listener is redis-py's own `PubSubWorkerThread`, started behind an owner-PID check under a second per-PID lock (`_start_locks`) taken non-blocking, so a cache operation never waits on another thread's start | Safe. `setdefault` is atomic under per-object locking, so threads racing to create a new process's lock share one, and a child forked while a parent thread holds either lock gets its own instead of waiting on a lock nobody will release. Walking the live `WeakSet` while a decoration adds to it would raise; the snapshot under the lock rules that out. The owner-PID check reads `_listener_pid`, which the starting thread publishes last, so a stale read only skips or retries a start. Evictors call `L1Cache.invalidate` and `invalidate_many`, which lock per state as in the row above. A child forked without at-fork hooks starts no listener and logs nothing (the same detection as the row above). Tests in `tests/unit/test_invalidation_listener.py` | | `backends/redis/provider.py` `with_timeout` windows (`_open_windows`: pool → configured `socket_timeout` and the open windows' timeouts) | Per-PID `Lock` from `_pid_lock` (`_windows_locks`) around each window's open and close and around `listener_pool()`'s read of the pool's kwargs | Safe. Opening and closing each read and write the record and the pool's `socket_timeout` together, and a window can open and close on any thread's event loop over one shared pool, so without the lock a window closing the last record could drop one another thread just joined. No `await` runs under the lock. A child gets its own lock, as in the row above. A child forked while a window is open inherits that window but not the task holding it, so it could never close it: an `os.register_at_fork` hook drops every inherited record and puts each pool's configured timeout back, and each record carries the PID that opened it, so a fork made from C, which runs no hook, drops it on the pool's next window or listener clone instead; until then such a child opens its connections at the inherited window's timeout (no per-operation PID check, as in the object-cache row; `--py-call-uwsgi-fork-hooks` avoids it). A close restores the pool's timeout before it drops the record, so a fork between the two steps still leaves the hook a record to reset from. Close-order and fork tests (single-threaded) in `tests/unit/test_invalidation_listener.py` | diff --git a/docs/performance.md b/docs/performance.md index a02048fc..16f7bfde 100644 --- a/docs/performance.md +++ b/docs/performance.md @@ -320,8 +320,8 @@ The regression gate is the instruction budget: `make perf-ir` on your machine, a | Path | What one call does | CPython 3.12 | CPython 3.14 | |------|--------------------|-------------:|-------------:| | `l1_hit` | `@cache(backend=None)` L1 hit | 2,595 | 2,487 | -| `l1_hit_backed` | `@cache` with a backend configured, L1 hit (never reaches the in-process dict backend; decodes the verified MessagePack L1 holds after the first hit) | 83,015 | 83,882 | -| `l1_hit_backed_auto` | `l1_hit_backed` with `serializer="auto"` (decodes the verified MessagePack the auto serializer keeps) | 91,986 | 92,950 | +| `l1_hit_backed` | `@cache` with a backend configured, L1 hit (never reaches the in-process dict backend; decodes the verified MessagePack L1 holds after the first hit). The backend keeps an L1 hit-latency report, as a CachekitIO backend with telemetry on does, so each call pays the 1-in-256 timing draw and the timed hits' clock reads ([Telemetry](backends/cachekitio.md#telemetry)) | 80,413 | 81,213 | +| `l1_hit_backed_auto` | `@cache` with `serializer="auto"` and a backend that keeps no hit-latency report, L1 hit (decodes the verified MessagePack the auto serializer keeps) | 91,986 | 92,950 | | `minimal_l1_hit` | `@cache.minimal(backend=None)` L1 hit | 2,595 | 2,487 | | `l2_hit` | `@cache`, L1 disabled, L2 hit, as one in-process single-flight trip | 172,539 | 167,686 | | `miss` | `@cache`, L1 disabled, L2 miss, compute, L2 write, all as one in-process single-flight trip | 213,364 | 204,897 | diff --git a/src/cachekit/backends/cachekitio/backend.py b/src/cachekit/backends/cachekitio/backend.py index 981d110b..c2c90033 100644 --- a/src/cachekit/backends/cachekitio/backend.py +++ b/src/cachekit/backends/cachekitio/backend.py @@ -24,10 +24,11 @@ from cachekit.backends._uninterrupted import _await_uninterrupted, _clear_interrupted, _raised_during_request from cachekit.backends.cachekitio.client import ClientLease, lease_http_client -from cachekit.backends.cachekitio.config import CachekitIOBackendConfig +from cachekit.backends.cachekitio.config import CachekitIOBackendConfig, resolve_telemetry from cachekit.backends.cachekitio.error_handler import HTTPStatusError, HTTPTransportError, classify_http_error from cachekit.backends.errors import BackendError, BackendErrorType from cachekit.config.validation import ConfigurationError, hide_any_secret +from cachekit.decorators.l1_latency import L1LatencyReport, clock_qualifies from cachekit.decorators.stats_context import get_current_function_stats from cachekit.hash_utils import redact_cache_key, redact_error_for_log from cachekit.logging import get_structured_logger @@ -191,7 +192,7 @@ def _parse_fresh_for(value: str | None) -> int | None: ) -def _inject_metrics_headers(stats: _FunctionStats | None) -> dict[str, str]: +def _inject_metrics_headers(stats: _FunctionStats | None, telemetry: bool = True) -> dict[str, str]: """Extract cache metrics and format as HTTP headers. Extracts L1 hits, L2 hits, and misses from the provided function statistics, @@ -201,6 +202,8 @@ def _inject_metrics_headers(stats: _FunctionStats | None) -> dict[str, str]: Args: stats: Function statistics tracker, or None for graceful degradation. + telemetry: False sends L1-Status alone, none of the optional metrics headers + (spec/saas-api.md#telemetry-switch). Returns: dict[str, str]: Headers dictionary containing: @@ -227,8 +230,15 @@ def _inject_metrics_headers(stats: _FunctionStats | None) -> dict[str, str]: '2' >>> headers["X-CacheKit-L2-Hits"] '1' - >>> float(headers["X-CacheKit-L1-Hit-Rate"]) - 0.667 + >>> headers["X-CacheKit-L1-Hit-Rate"] # L1 / (L1 + L2 + misses) + '0.667' + >>> stats.record_miss() + >>> _inject_metrics_headers(stats)["X-CacheKit-L1-Hit-Rate"] + '0.500' + + >>> # Telemetry off: L1-Status alone + >>> _inject_metrics_headers(stats, telemetry=False) + {'X-CacheKit-L1-Status': 'miss'} >>> headers["X-CacheKit-L1-Status"] 'miss' @@ -256,22 +266,6 @@ def _inject_metrics_headers(stats: _FunctionStats | None) -> dict[str, str]: if stats is None: return {"X-CacheKit-L1-Status": "disabled"} - # Extract metrics from stats - info = stats.get_info() - l1_hits = info.l1_hits - l2_hits = info.l2_hits - misses = info.misses - - # Calculate L1 hit rate with zero-division guard - total_hits = l1_hits + l2_hits - if total_hits > 0: - l1_hit_rate = l1_hits / total_hits - else: - l1_hit_rate = 0.0 - - # Format L1 hit rate to 3 decimal places - l1_hit_rate_str = f"{l1_hit_rate:.3f}" - # Determine L1 status for rate limit classification # Conservative approach: report "miss" for enabled (counts as backend op) # or "disabled" if L1 is not enabled @@ -279,6 +273,18 @@ def _inject_metrics_headers(stats: _FunctionStats | None) -> dict[str, str]: l1_status = "miss" # Conservative: treat all as backend ops else: l1_status = "disabled" + if not telemetry: + return {"X-CacheKit-L1-Status": l1_status} + + # Extract metrics from stats + info = stats.get_info() + l1_hits = info.l1_hits + l2_hits = info.l2_hits + misses = info.misses + + # The share of counted operations L1 served, 0.000 with none counted (spec/saas-api.md#l1-hit-rate, API-78) + counted = l1_hits + l2_hits + misses + l1_hit_rate_str = f"{l1_hits / counted if counted else 0.0:.3f}" # Get session headers with exception safety try: @@ -337,6 +343,7 @@ def __init__( api_url: str | None = None, api_key: str | SecretStr | None = None, timeout: float | None = None, + telemetry: bool | None = None, ) -> None: """Initialize cachekit.io backend. @@ -344,6 +351,9 @@ def __init__( api_url: API endpoint URL. Default: ``CACHEKIT_API_URL``, then ``https://api.cachekit.io``. api_key: API key (``ck_live_...``). Default: ``CACHEKIT_API_KEY``. timeout: Request timeout in seconds. Default: ``CACHEKIT_TIMEOUT``, then 5.0. + telemetry: False turns telemetry off: no optional metrics header (session, counters, hit rate, + L1 hit latency) is sent. ``CACHEKIT_TELEMETRY=false`` does the same, and off from either wins; + both unset leaves it on. Both are read here, once. Each argument left as None is loaded from the environment via pydantic-settings, so ``CachekitIOBackend(api_key=...)`` alone is valid — an explicit argument wins, @@ -354,7 +364,7 @@ def __init__( validation (credentials in the URL, non-HTTPS, private address, host not in the allowlist). """ api_key = hide_any_secret(api_key) # a bytes key too: no local here holds it raw (CWE-532) - overrides: dict[str, Any] = {"api_url": api_url, "api_key": api_key, "timeout": timeout} + overrides: dict[str, Any] = {"api_url": api_url, "api_key": api_key, "timeout": timeout, "telemetry": telemetry} errors = None try: # The config takes a str key wrapped. A bytes key is unwrapped only here, inline, and its str field decodes it @@ -381,6 +391,12 @@ def __init__( # Each request re-leases after a fork or a close_http_clients() (see _own_lease). self._lease = lease_http_client(self._config) + # spec/saas-api.md#telemetry-switch: resolved once, for this instance's lifetime. + self._telemetry = resolve_telemetry(self._config.telemetry, os.environ) + # The wrappers configured with this backend time a sample of their L1 hits into this report, which rides + # this backend's next request. None, and nothing timed, with telemetry off or no qualifying clock (API-82). + self.l1_latency_report: L1LatencyReport | None = L1LatencyReport() if self._telemetry and clock_qualifies() else None + def _own_lease(self) -> ClientLease: """This process's open lease: re-leased after a fork, or once close_http_clients() closed its client. @@ -527,11 +543,14 @@ def _request_sync( Automatically injects cache metrics headers (L1/L2 hits, session ID) when called from within a @cache decorated function; with no stats in context, - only ``X-CacheKit-L1-Status: disabled``. + only ``X-CacheKit-L1-Status: disabled``. With telemetry off, only L1-Status. + + The L1 hit-latency report, if any, rides the first attempt alone: once sent it is + consumed, so the retry never carries it (spec API-86). """ url = f"/v1/cache/{endpoint}" - headers = self._request_headers(headers) - response = self._send(method, url, body, headers) + headers, first = self._request_headers(endpoint, headers) + response = self._send(method, url, body, first) if (delay := _write_retry_delay(method, response)) is not None: _logger.debug(f"CachekitIO {method} got HTTP 503; retrying once after Retry-After {delay}s") time.sleep(delay) @@ -553,20 +572,27 @@ async def _request_async( response, not the request: the attempt in flight still completes on its worker thread. """ url = f"/v1/cache/{endpoint}" - headers = self._request_headers(headers) - response = await asyncio.to_thread(self._send, method, url, body, headers) + headers, first = self._request_headers(endpoint, headers) + response = await asyncio.to_thread(self._send, method, url, body, first) if (delay := _write_retry_delay(method, response)) is not None: _logger.debug(f"CachekitIO {method} got HTTP 503; retrying once after Retry-After {delay}s") await asyncio.sleep(delay) response = await asyncio.to_thread(self._send, method, url, body, headers) return self._checked(method, response, miss_on_404) - @staticmethod - def _request_headers(headers: dict[str, str] | None) -> dict[str, str]: + def _request_headers(self, endpoint: str, headers: dict[str, str] | None) -> tuple[dict[str, str], dict[str, str]]: + """A request's headers for any attempt, and for its first: the same plus the L1 hit-latency report, if any. + + Taking the report empties it, so call this once per request. The health endpoint never takes one: the + server does not record it (API-84). + """ # Always — defaults to L1-Status: disabled when no stats. Built on the caller's thread: the stats # come from its context. - metrics_headers = _inject_metrics_headers(get_current_function_stats()) - return {**headers, **metrics_headers} if headers else metrics_headers + metrics_headers = _inject_metrics_headers(get_current_function_stats(), self._telemetry) + merged = {**headers, **metrics_headers} if headers else metrics_headers + report = self.l1_latency_report + taken = None if report is None or endpoint == "health" else report.take() + return merged, {**merged, **taken} if taken else merged # ==================== BaseBackend Protocol (Sync) ==================== # These sync methods send on the shared client on the calling thread (no event loop required) @@ -1195,4 +1221,5 @@ def with_timeout(self, timeout: float) -> CachekitIOBackend: api_url=self._config.api_url, api_key=self._config.api_key, timeout=timeout, + telemetry=self._telemetry, # resolved: a clone of an off backend stays off ) diff --git a/src/cachekit/backends/cachekitio/config.py b/src/cachekit/backends/cachekitio/config.py index 99a71e4d..c1c2df9f 100644 --- a/src/cachekit/backends/cachekitio/config.py +++ b/src/cachekit/backends/cachekitio/config.py @@ -5,13 +5,17 @@ from __future__ import annotations +import logging import re import socket +from collections.abc import Mapping from ipaddress import IPv4Address, IPv6Address, ip_address, ip_network +from typing import Any from urllib.parse import ParseResult, urlparse from pydantic import Field, SecretStr, field_validator -from pydantic_settings import SettingsConfigDict +from pydantic.fields import FieldInfo +from pydantic_settings import BaseSettings, PydanticBaseSettingsSource, SettingsConfigDict from urllib3.util import parse_url from cachekit.backends.base_config import BaseBackendConfig, inherit_config @@ -24,6 +28,59 @@ _BEARER_TOKEN = re.compile(r"[A-Za-z0-9\-._~+/]+=*") +_logger = logging.getLogger(__name__) + +# spec/saas-api.md#telemetry-switch: read case-insensitively after trimming; anything else is off. +TELEMETRY_ENV = "CACHEKIT_TELEMETRY" +_TELEMETRY_OFF = frozenset({"false", "0", "off", "no"}) +_TELEMETRY_ON = frozenset({"true", "1", "on", "yes"}) +_warned_bad_telemetry = False + + +def _parse_telemetry_env(raw: str | None) -> bool | None: + """``CACHEKIT_TELEMETRY`` as on (True), off (False) or unset (None). An unknown value is off, warned once. + + Examples: + >>> [_parse_telemetry_env(v) for v in (None, "", " ", " OFF ", "No", "TRUE", "1")] + [None, None, None, False, False, True, True] + """ + global _warned_bad_telemetry + value = (raw or "").strip().lower() + if not value: + return None + if value in _TELEMETRY_ON: + return True + if value not in _TELEMETRY_OFF and not _warned_bad_telemetry: + _warned_bad_telemetry = True + # The value is not echoed: an environment variable set by mistake can hold anything. + _logger.warning("%s is not one of true/false/1/0/on/off/yes/no; telemetry is off", TELEMETRY_ENV) + return False + + +def resolve_telemetry(setting: bool | None, environ: Mapping[str, str]) -> bool: + """Whether a CachekitIO backend sends the optional metrics headers: off from either control wins, both unset is on. + + The variable's name matches case-insensitively, as every other ``CACHEKIT_*`` variable loads + (``case_sensitive=False``). Where several casings are set, off from any of them wins. + + Examples: + >>> resolve_telemetry(None, {}) + True + >>> resolve_telemetry(True, {"CACHEKIT_TELEMETRY": "off"}) + False + >>> resolve_telemetry(False, {"CACHEKIT_TELEMETRY": "on"}) + False + >>> resolve_telemetry(None, {"cachekit_telemetry": "false"}) + False + >>> resolve_telemetry(None, {"CACHEKIT_TELEMETRY": "on", "Cachekit_Telemetry": "off"}) + False + """ + if setting is False: + return False + found = [_parse_telemetry_env(v) for k, v in environ.items() if k.upper() == TELEMETRY_ENV] + return False not in found + + # Loopback, private, link-local (cloud metadata), "this network" and unspecified addresses. An IPv4-mapped IPv6 # address is checked as the IPv4 address it maps to. _PRIVATE_NETWORKS = tuple( @@ -105,6 +162,20 @@ def _parse_api_url(url: str) -> tuple[str, ParseResult]: raise ValueError("Invalid API URL: could not be parsed") +class _WithoutTelemetry(PydanticBaseSettingsSource): + """A settings source with any ``telemetry`` value it found removed (``CachekitIOBackendConfig.telemetry``).""" + + def __init__(self, source: PydanticBaseSettingsSource) -> None: + super().__init__(source.settings_cls) + self._source = source + + def get_field_value(self, field: FieldInfo, field_name: str) -> tuple[Any, str, bool]: + return self._source.get_field_value(field, field_name) + + def __call__(self) -> dict[str, Any]: + return {k: v for k, v in self._source().items() if k != "telemetry"} + + class CachekitIOBackendConfig(BaseBackendConfig): """Configuration for cachekit.io backend. @@ -150,6 +221,24 @@ class CachekitIOBackendConfig(BaseBackendConfig): default=False, description="Allow custom API hostnames (disables SSRF hostname allowlist)", ) + # Set in code only. CACHEKIT_TELEMETRY is combined with it by resolve_telemetry, never bound to it: this + # config lets an argument override the environment, which would let on win over off (spec API-95). + telemetry: bool | None = Field( + default=None, + description="False turns the optional metrics headers off; None (unset) leaves CACHEKIT_TELEMETRY to decide", + ) + + @classmethod + def settings_customise_sources( + cls, + settings_cls: type[BaseSettings], + init_settings: PydanticBaseSettingsSource, + env_settings: PydanticBaseSettingsSource, + dotenv_settings: PydanticBaseSettingsSource, + file_secret_settings: PydanticBaseSettingsSource, + ) -> tuple[PydanticBaseSettingsSource, ...]: + """Every source as pydantic-settings orders them, but only the constructor's arguments can set ``telemetry``.""" + return (init_settings, *map(_WithoutTelemetry, (env_settings, dotenv_settings, file_secret_settings))) @field_validator("api_key") @classmethod diff --git a/src/cachekit/decorators/l1_latency.py b/src/cachekit/decorators/l1_latency.py new file mode 100644 index 00000000..9320ccf1 --- /dev/null +++ b/src/cachekit/decorators/l1_latency.py @@ -0,0 +1,204 @@ +"""The L1 hit-latency report a CachekitIO backend sends to the server (protocol ``spec/saas-api.md#l1-hit-latency``). + +A wrapper times a sample of its backed-mode calls, and adds each one that turns out to be an L1 hit to the +report of the backend its cache is configured with. The backend takes the report, whole, into the next request +it sends. Here, not in the backend package, because the wrapper imports it and must not load the HTTP client. +""" + +from __future__ import annotations + +import os +import random +import threading +import time +from collections.abc import Callable + +# Bucket i (0-based) holds the durations up to 2**(i + 7) ns, the last up to the 60 s ceiling (spec table). +BUCKETS = 16 +MAX_DURATION_NS = 60_000_000_000 +# Every integer on the wire is at most 2**53 - 1, the largest a JavaScript number holds exactly. +_MAX_WIRE_INT = 2**53 - 1 +# Each call is timed with probability 1/N, the same N for every backend: the wrapper's draw is SAMPLE_BITS random +# bits all zero, so N is derived from it and is always the probability the draw has. +SAMPLE_BITS = 8 +ONE_IN = 1 << SAMPLE_BITS +_ONE_IN_VALUE = str(ONE_IN) +# perf_counter qualifies only at 100 ns or finer (spec: Clock). +_MAX_CLOCK_RESOLUTION_S = 1e-7 + +BUCKETS_HEADER = "X-CacheKit-L1-Latency-Buckets" +SUM_HEADER = "X-CacheKit-L1-Latency-Sum-Ns" +ONE_IN_HEADER = "X-CacheKit-L1-Latency-One-In" + + +def clock_qualifies() -> bool: + """Whether ``time.perf_counter_ns`` resolves 100 ns or finer here; without it no latency header is sent (API-82).""" + return time.get_clock_info("perf_counter").resolution <= _MAX_CLOCK_RESOLUTION_S + + +def bucket_index(duration_ns: int) -> int: + """The 0-based bucket a duration falls in: above the previous bound, up to and including its own. + + Examples: + >>> [bucket_index(d) for d in (0, 128, 129, 256, 2_097_152, 2_097_153, MAX_DURATION_NS)] + [0, 0, 1, 1, 14, 15, 15] + """ + return min(max((duration_ns - 1).bit_length() - 7, 0), BUCKETS - 1) + + +class _State: + """One process's accumulator. Its ``pid`` never changes: a forked child gets a new ``_State``, never a reset one.""" + + __slots__ = ("buckets", "lock", "pid", "sum") + + def __init__(self) -> None: + self.lock = threading.Lock() + self.pid = os.getpid() + self.buckets = [0] * BUCKETS + self.sum = 0 + + +# One lock per process, for replacing an inherited state. dict.setdefault is atomic, so every thread of a process +# gets the same lock, and a child never uses one its parent's threads could hold. Reentrant: the reseed and the new +# state allocate under it, and a garbage-collection finalizer that calls cachekit on the same thread must not +# deadlock the process's first use. +_REPLACEMENT_LOCKS: dict[int, threading.RLock] = {} + + +def _replacement_lock(pid: int) -> threading.RLock: + return _REPLACEMENT_LOCKS.setdefault(pid, threading.RLock()) + + +# The wrappers' sampling draw (spec API-81). A private generator, never the module-level one: drawing from the +# user's own `random` stream would change the output of their seeded code, and a worker that re-seeds `random` per +# job would time the same call positions in every job. A forked child must not replay its parent's sequence, or +# sibling workers, and every job a prefork master forks from one state, would time the same call positions. The +# at-fork hook reseeds it; a fork made from C runs no hook (uWSGI without --py-call-uwsgi-fork-hooks), so the +# child's first report construction or use reseeds it too: a backend it builds, or its first request or first timed +# hit on one it inherited, whichever comes first. Until then such a worker draws the same sequence as its siblings: +# every call is still timed with probability 1/N, but the workers time the same call positions. A PID check on every +# draw would close that window, at the cost of an os.getpid() system call on every call. +sampling_rng = random.Random() # noqa: S311 (a sampling draw, not a secret) +_sampling_pid = os.getpid() + + +def _reseed_sampling(pid: int) -> None: + """Reseed the sampling draw once per process; the caller holds ``_replacement_lock(pid)`` or is the at-fork hook.""" + global _sampling_pid + if _sampling_pid != pid: + sampling_rng.seed() + _sampling_pid = pid + + +def _reseed_if_forked(pid: int) -> None: + """Reseed the draw if this process has not yet: the check every report construction and use makes, so a + report built inside a child forked from C, whose state never carries the parent's PID, reseeds it too.""" + if _sampling_pid != pid: + with _replacement_lock(pid): + _reseed_sampling(pid) + + +if hasattr(os, "register_at_fork"): + os.register_at_fork(after_in_child=lambda: _reseed_sampling(os.getpid())) + + +class L1LatencyReport: + """One CachekitIO backend's accumulator of timed L1 hits since its last report. + + ``add``, ``record`` and ``take`` hold one lock, so a report's buckets and sum always describe the same hits and + no two requests carry the same hit (API-85). A forked child starts empty (API-83): each use checks the state's + PID rather than relying on an at-fork hook, which uWSGI forks without running. The child never touches the + parent's state, whose lock a parent thread may have held at the fork: it builds a complete new state and + publishes it with one assignment, under a lock of the child's own, so threads racing their first use agree on + one state and none loses hits another counted. + + Examples: + >>> report = L1LatencyReport() + >>> for d in (90, 300, 300, 40_000): + ... report.add(d) + >>> report.take() + {'X-CacheKit-L1-Latency-Buckets': '1,0,2,0,0,0,0,0,0,1,0,0,0,0,0,0', 'X-CacheKit-L1-Latency-Sum-Ns': '40690', 'X-CacheKit-L1-Latency-One-In': '256'} + >>> report.take() is None + True + """ + + __slots__ = ("_state",) + + def __init__(self) -> None: + self._state = _State() + _reseed_if_forked(self._state.pid) + + def _current(self) -> _State: + state = self._state + pid = os.getpid() + if _sampling_pid != pid: # tested here, not in a call: this runs on every timed hit + _reseed_if_forked(pid) # a fork no hook reached: see sampling_rng + if state.pid != pid: + # A forked child's first use: threads racing their first use agree on one new state, so none + # discards hits another already counted into its own. The lock is this process's own, never one + # inherited from the parent, which a parent thread may have held at the fork. The new state is built + # before the lock: building it allocates, and a garbage-collection finalizer that calls a cached + # function on this thread may re-enter here meanwhile, publish its own state and count a hit into it. + # Only a state still stale under the lock is replaced, so that hit is kept. + fresh = _State() + with _replacement_lock(pid): + state = self._state + if state.pid != pid: + state = self._state = fresh + return state + + @staticmethod + def _count(state: _State, duration_ns: int) -> None: + """Count one duration into ``state``; the caller holds its lock. Above 60 s it is discarded (API-87). + + bucket_index, inlined without min and max: each timed hit runs this, and on CPython 3.12 the builtin calls + cost more than the rest of the count. + + Every step that can raise (each allocates an int) runs before the sum is stored, and storing it cannot raise, + so a failure leaves the buckets and the sum describing the same hits. + """ + if not 0 <= duration_ns <= MAX_DURATION_NS: + return + total = state.sum + duration_ns + if total > _MAX_WIRE_INT: # never reached in practice: about 150,000 hits of 60 s in one report + return + i = (duration_ns - 1).bit_length() - 7 + state.buckets[0 if i < 0 else BUCKETS - 1 if i >= BUCKETS else i] += 1 + state.sum = total + + def add(self, duration_ns: int) -> None: + """Count one timed L1 hit of ``duration_ns``.""" + state = self._current() + with state.lock: + self._count(state, duration_ns) + + def record(self, start_ns: int, clock: Callable[[], int]) -> None: + """Count one timed L1 hit that started at ``start_ns`` on ``clock`` and ends now. + + The clock is read once the lock is held, so the time spent waiting for it, which the caller pays before it + gets its value, is in the duration. + """ + state = self._current() + with state.lock: + self._count(state, clock() - start_ns) + + def take(self) -> dict[str, str] | None: + """The report's headers, emptying it in the same step; None while its duration sum is 0. + + A report whose sum is 0 stays: all-zero buckets are never attached. + """ + state = self._current() + with state.lock: + if state.sum < 1: + return None + buckets, total = state.buckets, state.sum + state.buckets = [0] * BUCKETS + state.sum = 0 + return { + BUCKETS_HEADER: ",".join(map(str, buckets)), + SUM_HEADER: str(total), + ONE_IN_HEADER: _ONE_IN_VALUE, + } + + +__all__ = ["ONE_IN", "SAMPLE_BITS", "L1LatencyReport", "bucket_index", "clock_qualifies"] diff --git a/src/cachekit/decorators/wrapper.py b/src/cachekit/decorators/wrapper.py index a65d51dd..4ad09de7 100644 --- a/src/cachekit/decorators/wrapper.py +++ b/src/cachekit/decorators/wrapper.py @@ -66,6 +66,7 @@ ) # Config import removed - using direct DecoratorConfig integration +from .l1_latency import SAMPLE_BITS, L1LatencyReport, sampling_rng from .orchestrator import FeatureOrchestrator from .orchestrator import logger as _operation_logger from .single_flight import AsyncFlights, ThreadFlights @@ -632,6 +633,33 @@ def logger(): # Slots per decorated function. A full memo is emptied rather than evicted slot by slot, so a hit # writes nothing to it. _MEMO_MAX_ENTRIES = 1024 +# The L1 hit-latency timer's draw, chance and clock. Globals, not closure cells: a wrapper's every free variable +# costs its every call, the memoised L1-only hit included, which never times anything. A test can replace them. +# getrandbits(k) == 0 holds with probability 2**-k: one draw, and a cached small int rather than a new float. +# The generator, and how a forked child reseeds it, are l1_latency.sampling_rng's. +_getrandbits = sampling_rng.getrandbits +_perf_counter_ns = time.perf_counter_ns + + +def _latency_report_of(backend: object) -> L1LatencyReport | None: + """The L1 hit-latency report a backend sends with its requests (a CachekitIO backend with telemetry on), or None. + + Read statically: a Mock or ``__getattr__`` proxy answers any attribute, and must not be read as one. + """ + report = inspect.getattr_static(backend, "l1_latency_report", None) + return report if isinstance(report, L1LatencyReport) else None + + +def _record_l1_latency(report: L1LatencyReport, timed_from: int) -> None: + """Record a timed L1 hit that has its value. Telemetry never costs a served hit: the call sits inside the hit's + ``try``, whose broad handler would read a failure here as a poisoned entry, invalidate it and go to L2, a second + outcome for one call. So a failure is logged and the hit's duration dropped.""" + try: + report.record(timed_from, _perf_counter_ns) + except Exception as e: # noqa: BLE001 (see the docstring) + _logger.debug("L1 hit latency not recorded: %s", type(e).__name__) + + # What HitMemo.get returns when it cannot answer a call. _MISS = object() @@ -939,6 +967,9 @@ def create_cache_wrapper( # If explicit backend provided, use it; otherwise get from provider on first use _backend = backend if backend is not None else None _handler_backend = None # the backend operation_handler's StandardCacheHandler was built for + # The report this wrapper times its backed-mode L1 hits into, set wherever _backend is: here for an explicit + # backend, in _install_cache_handler for one a call resolves, and in invalidate_cache/ainvalidate_cache. + _l1_latency = None if backend is None else _latency_report_of(backend) def _install_cache_handler(resolved: BaseBackend) -> None: """Install a StandardCacheHandler for the resolved backend, building it only when that backend changed. @@ -947,8 +978,9 @@ def _install_cache_handler(resolved: BaseBackend) -> None: function, so building one per call was throwaway work on every backed call. Keyed on the backend's identity, so a backend first resolved by invalidate_cache() still gets its handler. """ - nonlocal _handler_backend + nonlocal _handler_backend, _l1_latency if _handler_backend is not resolved: + _l1_latency = _latency_report_of(resolved) operation_handler.set_cache_handler( StandardCacheHandler( resolved, @@ -2065,8 +2097,10 @@ def _settle_zero_config() -> None: # Last: a caller that reads it False without the lock then reads the state set above. _zero_config_pending = False - def _sync_call(args: tuple[Any, ...], kwargs: dict[str, Any]) -> Any: # noqa: PLR0912 - """sync_wrapper's call past its lock-free hit.""" + def _sync_call(args: tuple[Any, ...], kwargs: dict[str, Any], timed_from: int | None) -> Any: # noqa: PLR0912 + """sync_wrapper's call past its lock-free hit. ``timed_from`` is the clock reading the call started at when + it is timed, else None: a backed-mode L1 hit records its duration in the step that produces its value + (spec/saas-api.md#l1-hit-latency), so a timed hit allocates nothing and adds no frame.""" # The stats context is set below the backed L1 hit: its only reader is the CachekitIO request, and # nothing above it reaches the backend. None until then, so a call that ends earlier resets nothing. token = None @@ -2220,6 +2254,8 @@ def _sync_call(args: tuple[Any, ...], kwargs: dict[str, Any]) -> Any: # noqa: P # Record L1 hit for cache_info() _stats.record_l1_hit() + if timed_from is not None and _l1_latency is not None: + _record_l1_latency(_l1_latency, timed_from) return l1_value except TenantResolutionError: # No tenant in the caller's context: nothing to decrypt as, and nothing wrong @@ -2581,10 +2617,13 @@ def sync_wrapper(*args: Any, **kwargs: Any) -> Any: if kwargs and "_bypass_cache" in kwargs: del kwargs["_bypass_cache"] return func(*args, **kwargs) - return _sync_call(args, kwargs) + # The L1 hit-latency draw, before the call's outcome is known (spec API-81). Only a backed-mode L1 hit whose + # backend keeps a report adds its duration; any other outcome discards it. A memoised hit, above, is L1-only + # mode, which has no backend to report to. + return _sync_call(args, kwargs, None if _getrandbits(SAMPLE_BITS) else _perf_counter_ns()) - async def _async_call(args: tuple[Any, ...], kwargs: dict[str, Any]) -> Any: - """async_wrapper's call past its lock-free hit.""" + async def _async_call(args: tuple[Any, ...], kwargs: dict[str, Any], timed_from: int | None) -> Any: + """async_wrapper's call past its lock-free hit; ``timed_from`` as in ``_sync_call``.""" token = None # the stats context is set below the backed L1 hit, as in _sync_call try: @@ -2702,6 +2741,8 @@ async def _async_call(args: tuple[Any, ...], kwargs: dict[str, Any]) -> Any: # Record L1 hit for cache_info() _stats.record_l1_hit() + if timed_from is not None and _l1_latency is not None: + _record_l1_latency(_l1_latency, timed_from) return l1_value except TenantResolutionError: @@ -2826,7 +2867,8 @@ async def async_wrapper(*args: Any, **kwargs: Any) -> Any: if kwargs and "_bypass_cache" in kwargs: del kwargs["_bypass_cache"] return await func(*args, **kwargs) - return await _async_call(args, kwargs) + # As in sync_wrapper. The clock is read here, in this coroutine's first step, not at its creation. + return await _async_call(args, kwargs, None if _getrandbits(SAMPLE_BITS) else _perf_counter_ns()) def _log_breaker_rejection(cache_key: str) -> None: """Log a call the circuit breaker rejected. Never recorded as a failure: a rejection is not one.""" @@ -3424,7 +3466,7 @@ def invalidate_cache(*args: Any, **kwargs: Any) -> InvalidationOutcome | None: Returns the call's InvalidationOutcome for one entry, and None for every entry. """ - nonlocal _backend + nonlocal _backend, _l1_latency if _zero_config_pending: _settle_zero_config() @@ -3435,6 +3477,7 @@ def invalidate_cache(*args: Any, **kwargs: Any) -> InvalidationOutcome | None: if not _l1_only_mode and _backend is None: try: _backend = _resolve_lazy_backend() + _l1_latency = _latency_report_of(_backend) # no handler is installed here (see _install_cache_handler) except Exception as e: # If backend creation fails, can't invalidate L2 _logger.debug("Failed to get backend for invalidation: %s", redact_error_for_log(e)) @@ -3460,7 +3503,7 @@ def invalidate_cache(*args: Any, **kwargs: Any) -> InvalidationOutcome | None: async def ainvalidate_cache(*args: Any, **kwargs: Any) -> InvalidationOutcome | None: """Async invalidate_cache: the same deletes, run off the event loop, and the same return value.""" - nonlocal _backend + nonlocal _backend, _l1_latency if _zero_config_pending: _settle_zero_config() @@ -3471,6 +3514,7 @@ async def ainvalidate_cache(*args: Any, **kwargs: Any) -> InvalidationOutcome | if not _l1_only_mode and _backend is None: try: _backend = _resolve_lazy_backend() + _l1_latency = _latency_report_of(_backend) # as in invalidate_cache except Exception as e: # If backend creation fails, can't invalidate L2 _logger.debug("Failed to get backend for async invalidation: %s", redact_error_for_log(e)) diff --git a/tests/performance/ir_baselines.json b/tests/performance/ir_baselines.json index 21dd8e7b..d85bd7e7 100644 --- a/tests/performance/ir_baselines.json +++ b/tests/performance/ir_baselines.json @@ -6,7 +6,7 @@ "byte_storage_store_64k": 720631, "file_set": 76035, "l1_hit": 2595, - "l1_hit_backed": 83015, + "l1_hit_backed": 80413, "l1_hit_backed_auto": 91986, "l2_hit": 172539, "l2_hit_async_metrics": 171976, @@ -31,7 +31,7 @@ "byte_storage_store_64k": 720751, "file_set": 74112, "l1_hit": 2487, - "l1_hit_backed": 83882, + "l1_hit_backed": 81213, "l1_hit_backed_auto": 92950, "l2_hit": 167686, "l2_hit_async_metrics": 167728, diff --git a/tests/performance/ir_workload.py b/tests/performance/ir_workload.py index 3697031d..9ee8e5bf 100644 --- a/tests/performance/ir_workload.py +++ b/tests/performance/ir_workload.py @@ -38,8 +38,13 @@ def build_workload(path: str) -> Callable[[], object]: import cachekit.decorators.orchestrator as orchestrator from cachekit import cache + from cachekit.decorators.l1_latency import L1LatencyReport, sampling_rng from cachekit.serializers import EncryptionWrapper, get_serializer + # The timing draw has its own generator, unseeded by pin_main_thread_clocks: seeded here, every run times the + # same calls of l1_hit_backed. + sampling_rng.seed(0) + value = {"id": 42, "name": "Ada Lovelace", "roles": ["admin", "ops"], "score": 97.5, "tags": list(range(16))} # Dict-heavy value (about 6 KB): the decoder's object_hook runs once per record. records = [ @@ -70,6 +75,14 @@ def exists(self, key: str) -> bool: def health_check(self) -> tuple[bool, dict[str, Any]]: return True, {} + class ReportingBackend(DictBackend): + """DictBackend with an L1 hit-latency report, as a CachekitIO backend has with telemetry on (its default): + every call draws whether it is timed, and about one hit in ``ONE_IN`` is timed into the report.""" + + def __init__(self) -> None: + super().__init__() + self.l1_latency_report = L1LatencyReport() + class MissBackend(DictBackend): """Never holds a value, so every call takes the full miss path: get, compute, serialize, set.""" @@ -118,7 +131,7 @@ def __init__(self, **kwargs: Any) -> None: # the default serializer and with the auto one. decorated: dict[str, Callable[[], Callable[..., dict]]] = { "l1_hit": lambda: cache(backend=None, ttl=300)(body), - "l1_hit_backed": lambda: cache(backend=DictBackend(), ttl=300)(body), + "l1_hit_backed": lambda: cache(backend=ReportingBackend(), ttl=300)(body), "l1_hit_backed_auto": lambda: cache(backend=DictBackend(), ttl=300, serializer="auto")(body), "minimal_l1_hit": lambda: cache.minimal(backend=None, ttl=300)(body), "l2_hit": l2, diff --git a/tests/unit/backends/test_cachekitio_telemetry.py b/tests/unit/backends/test_cachekitio_telemetry.py new file mode 100644 index 00000000..363e8bef --- /dev/null +++ b/tests/unit/backends/test_cachekitio_telemetry.py @@ -0,0 +1,907 @@ +"""L1 hit-latency telemetry and the telemetry switch (protocol spec/saas-api.md, Optional Metrics Headers). + +Each test names the requirement it checks. Requests go through the backend's real client on a fake pool +(tests/utils/cachekitio_fakes.py); the wrapper's draw and clock are replaced, so durations are exact. +""" + +from __future__ import annotations + +import asyncio +import os +import threading +import time +import uuid +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from cachekit import cache +from cachekit.backends.cachekitio.backend import CachekitIOBackend, _inject_metrics_headers +from cachekit.backends.cachekitio.config import CachekitIOBackendConfig, resolve_telemetry +from cachekit.backends.errors import BackendError +from cachekit.decorators import l1_latency, wrapper +from cachekit.decorators.l1_latency import ( + BUCKETS_HEADER, + MAX_DURATION_NS, + ONE_IN, + ONE_IN_HEADER, + SAMPLE_BITS, + SUM_HEADER, + L1LatencyReport, +) +from cachekit.decorators.wrapper import _FunctionStats, _latency_report_of +from tests.utils.cachekitio_fakes import TEST_API_KEY, FakePool, FakeRequest, fake_backend, response +from tests.utils.fork_helpers import child_outcome +from tests.utils.fork_helpers import report as report_to_parent + +SESSION_AND_COUNTERS = ( + "X-CacheKit-Session-ID", + "X-CacheKit-Session-Start", + "X-CacheKit-L1-Hits", + "X-CacheKit-L2-Hits", + "X-CacheKit-Misses", + "X-CacheKit-L1-Hit-Rate", +) +HEADERS = (BUCKETS_HEADER, SUM_HEADER, ONE_IN_HEADER) +OPTIONAL = (*SESSION_AND_COUNTERS, *HEADERS) + + +class _Store: + """A SaaS cache in a dict: GET/PUT/HEAD/DELETE on /v1/cache/, and the health endpoint.""" + + def __init__(self) -> None: + self.data: dict[str, bytes] = {} + + def __call__(self, request: FakeRequest): + if request.path == "/v1/cache/health": + return response(200, json={"status": "ok"}) + key = request.path.removeprefix("/v1/cache/") + if request.method == "PUT": + assert request.body is not None + self.data[key] = request.body + return response(200, json={"success": True}) + if request.method == "DELETE": + return response(200, json={"deleted": self.data.pop(key, None) is not None}) + if key not in self.data: + return response(404) + return response(200, self.data[key] if request.method == "GET" else b"") + + +def _ns() -> str: + """A namespace no other test shares: the L1 cache is process-wide.""" + return f"telemetry-{uuid.uuid4().hex}" + + +class _Clock: + """perf_counter_ns that advances ``step`` ns on each read.""" + + def __init__(self, step: int) -> None: + self.now = 1_000_000 + self.step = step + + def __call__(self) -> int: + self.now += self.step + return self.now + + +@pytest.fixture(autouse=True) +def _no_env(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("CACHEKIT_TELEMETRY", raising=False) + + +@pytest.fixture +def store() -> _Store: + return _Store() + + +@pytest.fixture +def timed(monkeypatch: pytest.MonkeyPatch) -> _Clock: + """Every call is drawn for timing, and each one takes 300 ns as the clock sees it.""" + clock = _Clock(300) + monkeypatch.setattr(wrapper, "_getrandbits", lambda k: 0) + monkeypatch.setattr(wrapper, "_perf_counter_ns", clock) + return clock + + +def _backend(store: _Store, **kwargs: object) -> tuple[CachekitIOBackend, FakePool]: + return fake_backend(store, **kwargs) + + +def _latency(request: FakeRequest) -> dict[str, str]: + return {name: request.headers[name] for name in HEADERS if name in request.headers} + + +def _sent(pool: FakePool) -> list[dict[str, str]]: + return [_latency(request) for request in pool.requests] + + +ONE_HIT_300 = { + "X-CacheKit-L1-Latency-Buckets": "0,0,1,0,0,0,0,0,0,0,0,0,0,0,0,0", + "X-CacheKit-L1-Latency-Sum-Ns": "300", + "X-CacheKit-L1-Latency-One-In": str(ONE_IN), +} + + +class TestWhatIsTimed: + def test_api79_sync_l1_hit_rides_the_next_request(self, store: _Store, timed: _Clock) -> None: + backend, pool = _backend(store) + + @cache(backend=backend, ttl=60, namespace=_ns()) + def f(x: int) -> int: + return x * 2 + + assert f(1) == 2 # miss: GET, PUT + assert f(1) == 2 # L1 hit, timed: 300 ns + assert _sent(pool) == [{}, {}] + assert f(2) == 4 # miss: its GET carries the report, its PUT nothing + assert _sent(pool)[2:] == [ONE_HIT_300, {}] + + async def test_api79_async_l1_hit_rides_the_next_request(self, store: _Store, timed: _Clock) -> None: + backend, pool = _backend(store) + + @cache(backend=backend, ttl=60, namespace=_ns()) + async def f(x: int) -> int: + return x * 2 + + assert await f(1) == 2 + coro = f(1) + timed.now += 10_000 # creating the coroutine starts no clock: its first step does + assert await coro == 2 + before = len(pool.requests) + assert await f(2) == 4 # a miss: its first request carries the report, nothing else does + assert [sent for sent in _sent(pool) if sent] == [ONE_HIT_300] + assert _sent(pool)[before] == ONE_HIT_300 + + def test_api79_the_clock_spans_key_building(self, store: _Store, monkeypatch: pytest.MonkeyPatch) -> None: + """The first clock read comes before the call builds its key, the last after the hit is served: the whole + call as the caller sees it, not only the L1 lookup.""" + events: list[str] = [] + monkeypatch.setattr(wrapper, "_getrandbits", lambda k: 0) + monkeypatch.setattr(wrapper, "_perf_counter_ns", lambda: events.append("clock") or 1_000 * len(events)) + backend, _ = _backend(store) + + def key(x: int) -> str: + events.append("key") + return f"telemetry-clock-{x}" + + @cache(backend=backend, ttl=60, namespace=_ns(), key=key) + def f(x: int) -> int: + return x + + f(1) + events.clear() + assert f(1) == 1 # hit + assert events == ["clock", "key", "clock"] + + @pytest.mark.parametrize("mode", ["sync", "async"]) + def test_a_failing_record_still_serves_the_hit_from_l1( + self, mode: str, store: _Store, timed: _Clock, monkeypatch: pytest.MonkeyPatch + ) -> None: + """The record runs inside the hit's try, whose broad handler treats an error as a poisoned entry. A recorder + failure must not reach it: the hit keeps its value and its one outcome, with no GET, and the entry stays.""" + backend, pool = _backend(store) + + def fail(self: L1LatencyReport, start_ns: int, clock: object) -> None: + raise MemoryError + + if mode == "sync": + + @cache(backend=backend, ttl=60, namespace=_ns()) + def f(x: int) -> int: + return x * 2 + + call = lambda: f(1) # noqa: E731 + else: + + @cache(backend=backend, ttl=60, namespace=_ns()) + async def g(x: int) -> int: + return x * 2 + + call = lambda: asyncio.run(g(1)) # noqa: E731 + + assert call() == 2 # miss: GET, PUT + sent = len(pool.requests) + monkeypatch.setattr(L1LatencyReport, "record", fail) + assert call() == 2 # L1 hit whose record fails + monkeypatch.undo() + assert call() == 2 # still an L1 hit: the entry was not invalidated + assert len(pool.requests) == sent # neither hit reached the backend + + @pytest.mark.parametrize("mode", ["sync", "async"]) + def test_api79_the_wait_for_the_reports_lock_is_in_the_duration( + self, mode: str, store: _Store, monkeypatch: pytest.MonkeyPatch + ) -> None: + """The caller pays for a contended report lock before it gets its value, so that wait is part of the hit.""" + monkeypatch.setattr(wrapper, "_getrandbits", lambda k: 0) + backend, _ = _backend(store) + report = backend.l1_latency_report + assert report is not None + + if mode == "sync": + + @cache(backend=backend, ttl=60, namespace=_ns()) + def f(x: int) -> int: + return x + + call = lambda: f(1) # noqa: E731 + else: + + @cache(backend=backend, ttl=60, namespace=_ns()) + async def g(x: int) -> int: + return x + + call = lambda: asyncio.run(g(1)) # noqa: E731 + + call() # miss + report.take() + hold_ns = 50_000_000 + state = report._state + state.lock.acquire() # another thread mid-record + releaser = threading.Timer(hold_ns / 1e9, state.lock.release) + releaser.start() + try: + call() # hit, blocked on the lock + finally: + releaser.join() + taken = report.take() + # The hold starts before the call (asyncio.run's setup eats into it), so not all of it is waited; without + # the wait in the interval the hit reads well under a millisecond. + assert taken is not None and int(taken["X-CacheKit-L1-Latency-Sum-Ns"]) >= hold_ns // 2 + + def test_api80_a_miss_and_an_l2_hit_are_never_timed(self, store: _Store, timed: _Clock) -> None: + backend, pool = _backend(store) + + @cache(backend=backend, ttl=60, l1_enabled=False, namespace=_ns()) + def f(x: int) -> int: + return x + + f(1) # miss + f(1) # L2 hit: this wrapper keeps no L1 + assert all(not sent for sent in _sent(pool)) + assert backend.l1_latency_report is not None and backend.l1_latency_report.take() is None + + def test_api80_a_joined_call_is_not_timed(self, store: _Store, timed: _Clock) -> None: + """A caller that joins another's in-flight miss counts an L1 hit, but its duration is the other call's.""" + backend, _pool = _backend(store) + started, release = threading.Event(), threading.Event() + + @cache(backend=backend, ttl=60, namespace=_ns()) + def f(x: int) -> int: + started.set() + release.wait(5) + return x + + results: list[int] = [] + first = threading.Thread(target=lambda: results.append(f(1))) + first.start() + assert started.wait(5) + second = threading.Thread(target=lambda: results.append(f(1))) + second.start() + time.sleep(0.2) # the second caller is parked on the first's flight + release.set() + first.join(5) + second.join(5) + assert results == [1, 1] + assert backend.l1_latency_report is not None + assert backend.l1_latency_report.take() is None + + def test_api80_existence_checks_and_reads_inside_a_call_are_not_timed(self, store: _Store, timed: _Clock) -> None: + """Backend reads made inside a wrapped call, under its stats context, time nothing: only the call itself is.""" + backend, pool = _backend(store) + backend.set("k", b"v") + + @cache(backend=backend, ttl=60, namespace=_ns()) + def f(x: int) -> bytes | None: + assert backend.exists("k") + return backend.get("k") + + before = len(pool.requests) + assert f(1) == b"v" # a miss: its function's exists and get run inside the call + inner = [r for r in pool.requests[before:] if r.path == "/v1/cache/k"] + assert [r.method for r in inner] == ["HEAD", "GET"] + assert all("X-CacheKit-L1-Hits" in r.headers for r in inner) # they ran under the call's context + assert backend.l1_latency_report is not None and backend.l1_latency_report.take() is None + + def test_api79_a_backend_first_resolved_by_invalidation_still_times_hits(self, store: _Store, timed: _Clock) -> None: + """invalidate_cache can resolve a lazy backend before any call does; an L1 entry another wrapper of the same + function wrote is then a hit before this wrapper ever installs a handler.""" + from cachekit.config.decorator import set_default_backend + + backend, pool = _backend(store) + ns = _ns() + + def f(x: int) -> int: + return x + + explicit = cache(backend=backend, ttl=60, namespace=ns)(f) + explicit(1) # miss: writes L1 + lazy = cache(ttl=60, namespace=ns)(f) # no default backend yet: resolved at first use + set_default_backend(backend) + try: + lazy.invalidate_cache(2) # resolves the default backend, installs no handler + assert lazy(1) == 1 # an L1 hit on the entry explicit wrote + finally: + set_default_backend(None) + assert backend.l1_latency_report is not None + taken = backend.l1_latency_report.take() + assert taken is not None and taken[SUM_HEADER] == "300" + + def test_api77_only_a_cachekitio_backend_gets_a_report(self, store: _Store, timed: _Clock) -> None: + """A Mock answers any attribute; a backend of another kind has none: neither is timed into.""" + backend, _ = _backend(store) + assert _latency_report_of(backend) is backend.l1_latency_report + assert _latency_report_of(MagicMock()) is None + assert _latency_report_of(object()) is None + + def test_api77_hits_ride_only_their_own_backends_requests(self, store: _Store, timed: _Clock) -> None: + a, pool_a = _backend(store) + b, pool_b = _backend(store) + + @cache(backend=a, ttl=60, namespace=_ns()) + def f(x: int) -> int: + return x + + f(1) + f(1) # hit on a + b.get("anything") + assert _sent(pool_b) == [{}] + a.get("anything") + assert _sent(pool_a)[-1] == ONE_HIT_300 + + +class TestSampling: + @pytest.mark.parametrize("backed", [False, True]) + def test_api81_the_draw_leaves_the_users_random_stream_alone( + self, backed: bool, store: _Store, monkeypatch: pytest.MonkeyPatch + ) -> None: + """Seeded user code must produce the same numbers with cached calls in between, and a worker that re-seeds + `random` per job must not time the same call positions in every job. cachekit's log sampling draws from + `random` below WARNING, which is not the draw under test, so its logger is pinned there.""" + import logging + import random + + monkeypatch.setattr(logging.getLogger("cachekit"), "level", logging.WARNING) + backend = _backend(store)[0] if backed else None + + @cache(backend=backend, ttl=60, namespace=_ns()) + def f(x: int) -> int: + return x + + random.seed(1234) + before = random.getstate() + for i in range(3): + f(i) + f(i) + assert random.getstate() == before + + def test_api81_a_child_forked_from_c_reseeds_at_its_first_report_use(self) -> None: + """A fork made from C runs no at-fork hook (uWSGI): the child replays its parent's draws until its first + report use (a request or a timed hit) reseeds them. Two children of one parent state then diverge.""" + import ctypes + + libc_fork = ctypes.PyDLL(None).fork + report = L1LatencyReport() + + def child() -> list[int]: + r, w = os.pipe() + pid = libc_fork() + if pid == 0: # pragma: no cover - child + os.close(r) + try: + replayed = [wrapper._getrandbits(32) for _ in range(4)] # still the parent's sequence + report.take() # the child's first request + outcome: object = (replayed, [wrapper._getrandbits(32) for _ in range(4)]) + except BaseException as exc: # never return into pytest from the child + outcome = f"child raised {exc!r}" + report_to_parent(w, outcome) + os.close(w) + return child_outcome(pid, r) # type: ignore[return-value] + + state = l1_latency.sampling_rng.getstate() + first = child() + l1_latency.sampling_rng.setstate(state) # a prefork master forks every worker from one state + second = child() + l1_latency.sampling_rng.setstate(state) + parent = [wrapper._getrandbits(32) for _ in range(4)] + assert first[0] == second[0] == parent # before the first report use: replayed, as documented + assert first[1] != second[1] # after it: each child draws its own sequence + + def test_api81_a_report_built_in_a_child_forked_from_c_reseeds(self) -> None: + """A worker that builds its backend after a fork made from C never takes over an inherited state, so its + report's construction must reseed the draw itself. Two such children of one parent state diverge at once.""" + import ctypes + + libc_fork = ctypes.PyDLL(None).fork + + def child() -> list[int]: + r, w = os.pipe() + pid = libc_fork() + if pid == 0: # pragma: no cover - child + os.close(r) + try: + L1LatencyReport() # the worker builds its backend, and so its report + outcome: object = [wrapper._getrandbits(32) for _ in range(4)] + except BaseException as exc: # never return into pytest from the child + outcome = f"child raised {exc!r}" + report_to_parent(w, outcome) + os.close(w) + return child_outcome(pid, r) # type: ignore[return-value] + + state = l1_latency.sampling_rng.getstate() + first = child() + l1_latency.sampling_rng.setstate(state) + second = child() + l1_latency.sampling_rng.setstate(state) + parent = [wrapper._getrandbits(32) for _ in range(4)] + assert isinstance(first, list) and isinstance(second, list) + assert first != second and first != parent and second != parent + + def test_api81_a_forked_child_draws_its_own_sequence(self) -> None: + r, w = os.pipe() + pid = os.fork() + if pid == 0: # pragma: no cover - child + os.close(r) + try: + outcome: object = [wrapper._getrandbits(32) for _ in range(4)] + except BaseException as exc: # never return into pytest from the child + outcome = f"child raised {exc!r}" + report_to_parent(w, outcome) + os.close(w) + child = child_outcome(pid, r) + parent = [wrapper._getrandbits(32) for _ in range(4)] + assert isinstance(child, list) and child != parent + + def test_api81_the_draw_is_made_on_every_backed_call_before_its_outcome( + self, store: _Store, monkeypatch: pytest.MonkeyPatch + ) -> None: + draws: list[int] = [] + monkeypatch.setattr(wrapper, "_getrandbits", lambda k: draws.append(k) or 1) + backend, pool = _backend(store) + + @cache(backend=backend, ttl=60, namespace=_ns()) + def f(x: int) -> int: + return x + + f(1) # miss + f(1) # hit + f(2) # miss + assert len(draws) == 3 + assert set(draws) == {SAMPLE_BITS} # 2**-SAMPLE_BITS = 1/ONE_IN + assert all(not sent for sent in _sent(pool)) # a non-zero draw times nothing + + def test_api81_probability_is_one_in_n_and_n_is_sent(self) -> None: + assert 1 <= ONE_IN <= 1_000_000 + report = L1LatencyReport() + report.add(1) + taken = report.take() + assert taken is not None and taken["X-CacheKit-L1-Latency-One-In"] == str(ONE_IN) + + def test_api81_a_real_draw_times_about_one_call_in_n(self, store: _Store) -> None: + backend, _ = _backend(store) + + @cache(backend=backend, ttl=60, namespace=_ns()) + def f(x: int) -> int: + return x + + f(1) + calls = 1_000 * ONE_IN + for _ in range(calls): + f(1) + report = backend.l1_latency_report + assert report is not None + taken = report.take() + assert taken is not None + timed = sum(map(int, taken["X-CacheKit-L1-Latency-Buckets"].split(","))) + # Binomial(1000 N, 1/N): mean 1000, sd about 31; 6 sd either way. + assert 800 < timed < 1200 + + +class TestClock: + def test_api82_no_report_without_a_qualifying_clock(self, store: _Store) -> None: + coarse = MagicMock(return_value=MagicMock(resolution=1e-6)) + with patch.object(l1_latency.time, "get_clock_info", coarse): + backend, _ = _backend(store) + assert backend.l1_latency_report is None + + +class TestReports: + def test_api83_a_forked_child_starts_empty(self) -> None: + """No at-fork hook is involved, as under uWSGI: the child's first use sees the inherited state's PID.""" + report = L1LatencyReport() + report.add(500) + r, w = os.pipe() + pid = os.fork() + if pid == 0: # pragma: no cover - child + os.close(r) + try: + inherited = report._state.pid != os.getpid() + first = report.take() + report.add(700) + outcome: object = (inherited, first, report.take()) + except BaseException as exc: # never return into pytest from the child + outcome = f"child raised {exc!r}" + report_to_parent(w, outcome) + os.close(w) + inherited, first, second = child_outcome(pid, r) # type: ignore[misc] + assert inherited is True # the child still held the parent's state until its first use + assert first is None + assert second == {BUCKETS_HEADER: "0,0,0,1,0,0,0,0,0,0,0,0,0,0,0,0", SUM_HEADER: "700", ONE_IN_HEADER: str(ONE_IN)} + taken = report.take() # the parent's hit is still the parent's + assert taken is not None and taken[SUM_HEADER] == "500" + + def test_api83_a_hit_counted_by_reentry_during_takeover_is_kept(self, monkeypatch: pytest.MonkeyPatch) -> None: + """A finalizer that runs while the takeover builds its new state, and records a hit on this thread, publishes + a state of its own. The outer takeover must keep that state, and the hit in it, rather than overwrite it.""" + report = L1LatencyReport() + child_pid = os.getpid() + 1_000_000 # the parent's state carries the real PID: a fork as far as it can tell + monkeypatch.setattr(l1_latency.os, "getpid", lambda: child_pid) + real_init = l1_latency._State.__init__ + reentered = False + + def init_with_reentry(state: object) -> None: + nonlocal reentered + real_init(state) # type: ignore[arg-type] + if not reentered: # as a finalizer would, mid-allocation, on this thread + reentered = True + report.add(700) + + monkeypatch.setattr(l1_latency._State, "__init__", init_with_reentry) + report.add(900) + taken = report.take() + assert taken is not None and taken[SUM_HEADER] == "1600" + + def test_api83_a_lock_held_at_the_fork_does_not_hang_the_child(self, monkeypatch: pytest.MonkeyPatch) -> None: + report = L1LatencyReport() + report._state.lock.acquire() # a parent thread mid-add when the process forked + monkeypatch.setattr(l1_latency.os, "getpid", lambda: -1) + report.add(200) + assert report.take() is not None + + def test_api83_racing_first_uses_in_a_hookless_child_keep_every_hit(self) -> None: + """Child threads that reach the inherited state at once agree on one new state: the parent's hit is never + sent, and no child hit is lost to another thread's replacement. _State construction is slowed so that every + thread is inside the replacement at once, the schedule that lost hits when each built its own.""" + report = L1LatencyReport() + report.add(12_345) + threads_n, hit_ns = 8, 700 + real_init = l1_latency._State.__init__ + + def slow_init(state: object) -> None: + real_init(state) # type: ignore[arg-type] + time.sleep(0.02) + + r, w = os.pipe() + pid = os.fork() + if pid == 0: # pragma: no cover - child + os.close(r) + try: + l1_latency._State.__init__ = slow_init # type: ignore[method-assign] + barrier = threading.Barrier(threads_n) + + def first_use() -> None: + barrier.wait() + report.add(hit_ns) + + workers = [threading.Thread(target=first_use) for _ in range(threads_n)] + for t in workers: + t.start() + for t in workers: + t.join() + outcome: object = report.take() + except BaseException as exc: # never return into pytest from the child + outcome = f"child raised {exc!r}" + report_to_parent(w, outcome) + os.close(w) + taken = child_outcome(pid, r) + assert taken == { + BUCKETS_HEADER: f"0,0,0,{threads_n},0,0,0,0,0,0,0,0,0,0,0,0", + SUM_HEADER: str(threads_n * hit_ns), + ONE_IN_HEADER: str(ONE_IN), + } + parent = report.take() + assert parent is not None and parent[SUM_HEADER] == "12345" + + def test_a_zero_sum_report_is_not_attached_and_stays(self) -> None: + report = L1LatencyReport() + report.add(0) + assert report.take() is None + report.add(5) + assert report.take() == { + "X-CacheKit-L1-Latency-Buckets": "2,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0", + "X-CacheKit-L1-Latency-Sum-Ns": "5", + "X-CacheKit-L1-Latency-One-In": str(ONE_IN), + } + + def test_api84_never_on_the_health_endpoint(self, store: _Store) -> None: + backend, pool = _backend(store) + assert backend.l1_latency_report is not None + backend.l1_latency_report.add(300) + assert backend.health_check()[0] + assert _sent(pool) == [{}] + backend.get("k") + assert _latency(pool.requests[-1])["X-CacheKit-L1-Latency-Sum-Ns"] == "300" + + async def test_api84_never_on_the_async_health_endpoint(self, store: _Store) -> None: + backend, pool = _backend(store) + assert backend.l1_latency_report is not None + backend.l1_latency_report.add(300) + assert (await backend.health_check_async())[0] + assert _sent(pool) == [{}] + + def test_api85_concurrent_adds_and_takes_lose_and_repeat_nothing(self) -> None: + report = L1LatencyReport() + taken: list[dict[str, str]] = [] + threads, per_thread = 8, 2_000 + done = threading.Event() + + def adder() -> None: + for _ in range(per_thread): + report.add(300) + + def taker() -> None: + while not done.is_set(): + if (t := report.take()) is not None: + taken.append(t) + + reader = threading.Thread(target=taker) + reader.start() + workers = [threading.Thread(target=adder) for _ in range(threads)] + for w in workers: + w.start() + for w in workers: + w.join() + done.set() + reader.join() + if (t := report.take()) is not None: + taken.append(t) + hits = [sum(map(int, t["X-CacheKit-L1-Latency-Buckets"].split(","))) for t in taken] + assert sum(hits) == threads * per_thread + # Every report's buckets and sum describe the same hits. + assert all(int(t["X-CacheKit-L1-Latency-Sum-Ns"]) == 300 * n for t, n in zip(taken, hits, strict=True)) + + @pytest.mark.parametrize("mode", ["sync", "async"]) + def test_api86_the_shed_write_retry_does_not_resend_the_report(self, mode: str) -> None: + answers = [response(503, headers={"Retry-After": "1"}), response(200, json={"success": True})] + backend, pool = fake_backend(lambda request: answers.pop(0)) + assert backend.l1_latency_report is not None + backend.l1_latency_report.add(300) + with patch.object(time, "sleep"), patch.object(asyncio, "sleep", AsyncMock()): + if mode == "sync": + backend.set("k", b"v", ttl=60) + else: + asyncio.run(backend.set_async("k", b"v", ttl=60)) + first, retry = _sent(pool) + assert first["X-CacheKit-L1-Latency-Sum-Ns"] == "300" + assert retry == {} + assert backend.l1_latency_report.take() is None # consumed, not added back + + def test_api86_a_failed_request_consumes_its_report(self) -> None: + backend, pool = fake_backend(lambda request: response(500)) + assert backend.l1_latency_report is not None + backend.l1_latency_report.add(300) + with pytest.raises(BackendError): + backend.get("k") + assert _sent(pool)[0]["X-CacheKit-L1-Latency-Sum-Ns"] == "300" + assert backend.l1_latency_report.take() is None + + def test_api86_the_client_makes_no_retry_of_its_own(self, store: _Store) -> None: + backend, pool = _backend(store) + backend.get("k") + assert pool.requests[0].options.get("retries") is False + + +class TestEncoding: + @pytest.mark.parametrize( + ("duration", "index"), + [(0, 0), (1, 0), (128, 0), (129, 1), (256, 1), (257, 2), (2_097_152, 14), (2_097_153, 15), (MAX_DURATION_NS, 15)] + # Every bound up to the ceiling: 2**k ns is the last duration of bucket k - 7, and 2**k + 1 the first of the next. + + [(d, min(k - 7 + (d > 2**k), 15)) for k in range(7, 36) for d in (2**k, 2**k + 1) if d <= MAX_DURATION_NS], + ) + def test_api87_bucket_bounds(self, duration: int, index: int) -> None: + """Checked twice: bucket_index states the rule, and a report counts by its own inlined copy of it.""" + assert l1_latency.bucket_index(duration) == index + report = L1LatencyReport() + report.add(duration) + assert report._state.buckets == [int(i == index) for i in range(l1_latency.BUCKETS)] + + def test_api87_a_duration_above_60s_is_discarded(self) -> None: + report = L1LatencyReport() + report.add(MAX_DURATION_NS + 1) + assert report.take() is None + report.add(MAX_DURATION_NS) + taken = report.take() + assert taken is not None and taken["X-CacheKit-L1-Latency-Buckets"].endswith(",1") + + @pytest.mark.parametrize("step", ["index", "increment"]) + def test_api85_a_count_that_fails_leaves_buckets_and_sum_consistent(self, step: str) -> None: + """An allocation can fail after the sum is computed. The report must still describe the same hits, or its next + header set fails the server's range check and the whole batch is lost.""" + + class FailingSub(int): # (d - 1) is the bucket index's first step + def __sub__(self, other: object) -> int: + raise MemoryError + + class FailingIncrement(list[int]): + def __setitem__(self, index: object, value: object) -> None: + raise MemoryError + + report = L1LatencyReport() + for _ in range(3): + report.add(8_192) + if step == "index": + duration: int = FailingSub(8_192) + else: + duration = 8_192 + report._state.buckets = FailingIncrement(report._state.buckets) + with pytest.raises(MemoryError): + report.add(duration) + assert report._state.sum == 3 * 8_192 + assert list(report._state.buckets) == [3 if i == 6 else 0 for i in range(l1_latency.BUCKETS)] + + def test_api87_the_sum_never_passes_2_pow_53(self) -> None: + report = L1LatencyReport() + report._state.sum = 2**53 - 10 + report.add(100) # would pass the limit: dropped + report.add(9) + taken = report.take() + assert taken is not None and taken["X-CacheKit-L1-Latency-Sum-Ns"] == str(2**53 - 1) + + +def _stats(l1: int, l2: int, misses: int) -> _FunctionStats: + stats = _FunctionStats(function_identifier="t.f", l1_enabled=True) + for _ in range(l1): + stats.record_l1_hit() + for _ in range(l2): + stats.record_l2_hit(1.0) + for _ in range(misses): + stats.record_miss() + return stats + + +class TestHitRate: + @pytest.mark.parametrize( + ("l1", "l2", "misses", "rate"), + [(0, 0, 0, "0.000"), (1, 1, 2, "0.250"), (2, 1, 0, "0.667"), (3, 0, 0, "1.000"), (0, 0, 5, "0.000"), (1, 0, 2, "0.333")], + ) + def test_api78_share_of_all_counted_operations(self, l1: int, l2: int, misses: int, rate: str) -> None: + assert _inject_metrics_headers(_stats(l1, l2, misses))["X-CacheKit-L1-Hit-Rate"] == rate + + +ON, OFF, UNSET = "on", "off", "unset" +_SETTING = {ON: True, OFF: False, UNSET: None} +_ENV = {ON: "true", OFF: "false", UNSET: None} +# spec/saas-api.md#telemetry-switch, row = setting, column = environment +_TABLE = { + (UNSET, UNSET): True, + (UNSET, ON): True, + (UNSET, OFF): False, + (ON, UNSET): True, + (ON, ON): True, + (ON, OFF): False, + (OFF, UNSET): False, + (OFF, ON): False, + (OFF, OFF): False, +} + + +class TestSwitch: + @pytest.mark.parametrize(("setting", "env"), list(_TABLE)) + def test_api94_95_truth_table(self, setting: str, env: str, store: _Store, monkeypatch: pytest.MonkeyPatch) -> None: + if _ENV[env] is not None: + monkeypatch.setenv("CACHEKIT_TELEMETRY", _ENV[env]) + backend, pool = _backend(store, telemetry=_SETTING[setting]) + on = _TABLE[setting, env] + assert (backend.l1_latency_report is not None) is on + + @cache(backend=backend, ttl=60, namespace=_ns()) + def f(x: int) -> int: + return x + + with patch.object(wrapper, "_getrandbits", lambda k: 0): + f(1) + f(1) + f(2) + last = pool.requests[2].headers # f(2)'s GET: inside a call, after a hit + if on: + assert all(name in last for name in OPTIONAL) + else: + assert not any(name in last for name in OPTIONAL) # API-96 + assert "X-CacheKit-L1-Status" in last + assert "User-Agent" in last or "user-agent" in {k.lower() for k in last} + + @pytest.mark.parametrize( + ("raw", "parsed"), + [ + ("false", False), + ("0", False), + ("off", False), + ("no", False), + (" OFF ", False), + ("True", True), + ("1", True), + ("on", True), + ("YES\n", True), + ("", None), + (" ", None), + ], + ) + def test_api95_environment_values(self, raw: str, parsed: bool | None) -> None: + from cachekit.backends.cachekitio.config import _parse_telemetry_env + + assert _parse_telemetry_env(raw) is parsed + + def test_api95_an_unknown_value_is_off_and_warns_once( + self, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture + ) -> None: + from cachekit.backends.cachekitio import config + + monkeypatch.setattr(config, "_warned_bad_telemetry", False) + assert resolve_telemetry(True, {"CACHEKIT_TELEMETRY": "maybe"}) is False + assert resolve_telemetry(None, {"CACHEKIT_TELEMETRY": "nope"}) is False + warnings = [r for r in caplog.records if "CACHEKIT_TELEMETRY" in r.getMessage()] + assert len(warnings) == 1 + assert "maybe" not in warnings[0].getMessage() + + def test_api95_no_loader_fills_the_setting_from_the_environment(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("CACHEKIT_API_KEY", TEST_API_KEY) + monkeypatch.setenv("CACHEKIT_TELEMETRY", "false") + assert CachekitIOBackendConfig().telemetry is None + assert CachekitIOBackendConfig(telemetry=True).telemetry is True + # and the argument does not override the environment's off + backend, _ = fake_backend(_Store(), telemetry=True) + assert backend._telemetry is False + + def test_api95_read_once_at_construction(self, store: _Store, monkeypatch: pytest.MonkeyPatch) -> None: + backend, pool = _backend(store) + monkeypatch.setenv("CACHEKIT_TELEMETRY", "off") + + @cache(backend=backend, ttl=60, namespace=_ns()) + def f(x: int) -> int: + return x + + f(1) + assert "X-CacheKit-Session-ID" in pool.requests[0].headers + off, off_pool = _backend(store) + monkeypatch.delenv("CACHEKIT_TELEMETRY") + + @cache(backend=off, ttl=60, namespace=_ns()) + def g(x: int) -> int: + return x + + g(1) + assert "X-CacheKit-Session-ID" not in off_pool.requests[0].headers + + def test_api95_a_clone_of_a_backend_the_environment_turned_off_stays_off( + self, store: _Store, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setenv("CACHEKIT_TELEMETRY", "off") + backend, _ = _backend(store) + monkeypatch.delenv("CACHEKIT_TELEMETRY") + assert backend.with_timeout(2.0)._telemetry is False + + @pytest.mark.parametrize("name", ["cachekit_telemetry", "Cachekit_Telemetry"]) + def test_api95_the_variable_name_matches_in_any_case( + self, name: str, store: _Store, monkeypatch: pytest.MonkeyPatch + ) -> None: + """As every other CACHEKIT_* variable loads (case_sensitive=False): a lowercase off must not fail open.""" + monkeypatch.setenv(name, "false") + backend, pool = _backend(store, telemetry=True) + assert backend.l1_latency_report is None + backend.get("k") + assert not any(header in pool.requests[0].headers for header in OPTIONAL) + + def test_api95_a_lowercase_on_is_on(self) -> None: + assert resolve_telemetry(None, {"cachekit_telemetry": "yes"}) is True + + def test_api96_off_accumulates_no_report(self, store: _Store, timed: _Clock) -> None: + backend, pool = _backend(store, telemetry=False) + assert backend.l1_latency_report is None + + @cache(backend=backend, ttl=60, namespace=_ns()) + def f(x: int) -> int: + return x + + f(1) + f(1) + f(2) + assert all(not sent for sent in _sent(pool))