From ebeee57480a2bc5dcd3bc461e55265daec9d08f7 Mon Sep 17 00:00:00 2001 From: RickZ <121943721+rickisba@users.noreply.github.com> Date: Wed, 29 Jul 2026 15:36:57 +0800 Subject: [PATCH 1/2] docs: design unified v1 observability --- .../issue-141-unified-observability.md | 592 ++++++++++++++++++ 1 file changed, 592 insertions(+) create mode 100644 docs/research/issue-141-unified-observability.md diff --git a/docs/research/issue-141-unified-observability.md b/docs/research/issue-141-unified-observability.md new file mode 100644 index 0000000..6f56696 --- /dev/null +++ b/docs/research/issue-141-unified-observability.md @@ -0,0 +1,592 @@ +# Issue #141 research: unified v1 LMCache observability + +Status: design proposal only
+Baseline audited: `ad087d1` (current `main` lineage, including merged PR #154)
+Non-goals: production instrumentation, Gateway I/O, routing/queue changes, or a tracing backend. + +## 1. Executive decision record + +1. Put the dependency-light contracts in **`core/observability/`**. Scheduler, + Proxy, KDN, Gateway, and Instance already depend on `core`; placing them under + `kdn_server` would invert dependencies for Scheduler/Proxy/Instance, while a + new top-level distribution would add packaging work without providing more + isolation than a carefully constrained `core` package. +2. Make an exported trace an immutable, versioned snapshot. Build it by appending + immutable stages/measurements to a process-local collector; never mutate a + measurement from `predicted` into `actual`. +3. Use UTC timestamps for correlation and per-process monotonic nanoseconds for + durations. Never subtract monotonic readings produced by different processes. +4. Propagate a small internal trace envelope alongside the current serialized + `Request`, stripping it before the OpenAI/vLLM body. During migration, retain + `Request_ID`, `scheduler-request-id`, the free-form `ProxyTask.trace`, and + `cacheroute_meta` unchanged. +5. Treat Prometheus before/after deltas as endpoint aggregates. Only an isolated, + single-request validation run may label them `inferred`; concurrent deltas are + never request-correlated actuals. + +## 2. Current path and identity inventory + +```text +client payload + -> Scheduler allocates cyclic integer Request_ID (1..65535) + -> serialized core.request.Request + scheduler-request-id HTTP header + -> Proxy restores Request_ID; ProxyTask.request_id duplicates it + -> OpenAI-style instance_body (identity is dropped) + -> Instance forwards the body unchanged (identity remains absent) + -> vLLM/LMCache + +Proxy prepare path + -> KDN text lookup (no trace/correlation envelope) + -> Instance /v1/kv/inject_ready {request_id, ...} + -> KDN /knowledge/inject_ready_kv {request_id, ...} + +Versioned KDN/Gateway contracts (currently separate from serving path) + -> string request_id (default req_) + optional correlation_id + -> runtime_profile + created_at + -> targeted calls add compatibility_profile_id, endpoint_id, generation + -> Gateway responses preserve those fields +``` + +### Existing correlation fields + +| Field | Location/semantics | Finding | +|---|---|---| +| `Request.Request_ID` | Scheduler-assigned cyclic `int` | Not globally unique; reuse after 65,535 and across Scheduler processes/restarts. | +| `ProxyTask.request_id` | Optional `int`, copied from `Request_ID` | Useful local alias, but annotation conflicts with KDN v1 string IDs. | +| `scheduler-request-id` | Scheduler-to-Proxy HTTP header | Redundant with payload; not forwarded beyond Proxy. | +| KDN v1 `request_id` | `VersionedMessage`, non-empty string, defaults `req_` | Strong message identity, but the production serving path does not use these contracts yet. | +| KDN v1 `correlation_id` | Optional string | Correct place to retain an end-to-end correlation, but currently absent from `core.Request`. | +| Cache operation `task_id` | `cacheop_` | Operation identity, not a request identity. | +| `idempotency_key` | Cache operation and queue work | Logical deduplication identity; must not be overloaded as trace correlation. | +| artifact/observation/endpoint/work IDs | Canonical or UUID-backed domain IDs | Resource identity only. | + +`TraceContext` should therefore carry both an opaque globally unique `request_id` +and a `correlation_id`; the legacy Scheduler integer belongs in +`legacy_request_id`. A generated trace ID must not replace either. + +### Current `cacheroute_meta` + +Both streaming and non-streaming Proxy responses expose exactly: `request_id`, +the entire mutable `trace` dictionary, `kv_ack`, `kv_ready_kids`, +`text_only_kids`, `miss_kids`, and `error`. Streaming inserts an +`event: cacheroute_meta` immediately before `[DONE]`; completions add +`_cacheroute_meta` to parsed JSON (or the raw-response wrapper). This behavior is +a compatibility contract for the rollout. + +## 3. Complete `ProxyTask.trace` audit and classification + +Kinds below mean: **P** predicted, **O** observed snapshot/event, **A** actual +duration measured at the named boundary, **I** inferred/derived, **L** +legacy-projected compatibility, and **X** ambiguous or incorrectly named. +All `*_ms` timestamps are wall-clock epoch milliseconds today unless explicitly +described as durations. + +### Wall-clock event markers + +| Keys | Kind | Meaning / issue | +|---|---|---| +| `proxy_recv_ms`, `route_select_start_ms`, `route_select_end_ms` | O | Proxy receipt and local selection events. | +| `proxy_enqueue_ms`, `prepare_queue_enqueue_ms`, `prepare_dequeue_ms`, `prepare_start_ms` | O | Prepare queue lifecycle; first two currently receive the same reading. | +| `kdn_fetch_start_ms`, `kdn_fetch_end_ms` | O | Proxy-observed KDN fetch boundary. | +| `text_prefill_build_start_ms`, `text_prefill_build_end_ms`, `prompt_injected_ms` | O | Local body construction/injection markers; “prefill” is a misnomer because no vLLM prefill occurs here. | +| `kdn_link_wait_start_ms`, `kdn_link_wait_end_ms` | X | Wall-clock reservation model points, not necessarily observed wait start/end; `end` can be a predicted future time. | +| `kv_inject_queue_enqueue_ms`, `kv_inject_reserved_start_ms` | X | Queue marker plus predicted reservation time; names do not disclose differing value kinds. | +| `kv_inject_start_ms`, `kv_inject_end_ms`, `kv_ack_start_ms`, `kv_ack_end_ms` | O | Proxy/Instance-control-call boundary; injection completion is inferred from the acknowledgement, not LMCache consumption. | +| `prepare_self_done_ms`, `ready_enqueue_ms`, `ready_dequeue_ms` | O | Prepare completion/buffer release/ready dequeue. | +| `forward_wait_start_ms`, `forward_wait_end_ms`, `forward_start_ms`, `forward_end_ms` | O | Dispatch wait and Proxy-to-Instance response boundary. | +| `first_token_ms` | X | First non-empty downstream chunk, which may be SSE metadata rather than a decoded token. | +| `decode_start_ms`, `decode_end_ms` | X | Locally inferred state transition and forward end; neither directly measures vLLM decode. | +| `prepare_error_ms`, `ready_failed_before_forward_ms`, `stream_exception_ms` | O | Failure event timestamps. | +| `pred_forward_start_ts_ms`, `pred_first_token_ts_ms`, `pred_forward_end_ts_ms`, `pred_worker_free_ts_ms` | P | Predicted epoch times. They must remain predictions even after recomputation. | +| `kdn_link_free_before`, `kdn_link_free_after` | X | Predicted/shared model wall-clock values; units are hidden and “free” is not an observation of the remote link. | + +### Durations, counts, decisions, and outcomes + +| Keys | Kind | Meaning / issue | +|---|---|---| +| `predict_text_prefill_ms`, `predict_redis_kv_load_ms`, `predict_residual_prefill_ms` | P | Model estimates, not runtime measurements. | +| `predict_prepare_initial_ms`, `predict_prepare_queue_wait_ms`, `predict_kv_transfer_ms`, `predict_prepare_prefix_ms`, `predict_kv_prepare_service_ms`, `predict_prepare_model_ms`, `predict_prepare_corrected_ms` | P | Evolving prepare estimates. “corrected” is derived after completion and is no longer a prediction. | +| `predict_know_prepare_ms`, `predict_prepare_ms` | **X** | Initially predicted, then overwritten with wall-clock-derived actual prepare duration at ready release/success. This is the clearest kind violation. | +| `predict_queue_wait_ms`, `predict_compute_ms`, `predict_prefill_service_ms`, `predict_vllm_internal_ms`, `predict_decode_ms`, `predict_cold_start_extra_ms`, `predict_total_ms`, `predict_wait_ms` | P | Reservation/model outputs; `predict_wait_ms` is a legacy alias for knowledge preparation, not queue wait. | +| `predict_error_ms` | I | Signed residual `actual_total_ms - predict_total_ms`, not a prediction. | +| `actual_prepare_ms`, `actual_prepare_total_ms`, `prepare_buffer_wait_ms` | A | Wall-clock-derived local/control-call durations; susceptible to wall-clock adjustment. “prepare” has call-only and end-to-end meanings. | +| `actual_total_ms`, `actual_know_prepare_ms`, `actual_ready_queue_ms`, `actual_vllm_internal_ms` | A | Wall-clock-derived Proxy-observed durations to first non-empty chunk. `actual_vllm_internal_ms` includes network, Instance, vLLM queue, and prefill. | +| `actual_compute_ms` | L | Alias of `actual_vllm_internal_ms`; it is not compute-only. | +| `prepare_seq`, `ready_release_seq`, `ready_worker_idx`, `prepared_buffer_size`, `recompute_generation` (task field) | O | Local ordering/state observations. `recompute_generation` is not exported in `trace`. | +| `ready_release_blocked_seq` | O | Optional observed sequence; dictionary annotation does not allow `None`. | +| `ready_release_bypass`, `kv_link_reserved`, `ttft_observable` | O | Integer booleans. Typed v1 should serialize booleans. | +| `predict_length_tokens`, `predict_bs`, `predict_total_tokens`, `predict_reused_tokens`, `predict_residual_tokens` | P | Inputs/outputs of the model; token reuse is predicted rather than observed. | +| `predict_kv_bandwidth_mbps` | P/I | Link-snapshot/env/default input used by prediction; provenance determines whether observed or inferred. | +| `predict_kv_bandwidth_source` | P | Provenance string; violates `Dict[str, int]`. | +| `predict_stage` | P | Free-form reservation phase (`prefill`/`decode`), not an observed execution stage. | +| `injection_mode` | O | Chosen request mode, not proof that KV was used. | +| `text_actual_path` | X | Free-form outcome (`text_inject`, `no_rag_or_empty_knowledge`). | +| `kvcache_actual_path` | X | Free-form outcome (`kv_inject`, `kv_inject_failed_fallback_text`, `no_kv_ready_fallback_text`). It combines attempted path, failure, and fallback. | +| `ready_release_policy` | O | Policy label, violates `Dict[str, int]`; contextual attribute rather than measurement. | +| `prepare_failed_injection_mode` | O | Mode at failure, violates dictionary annotation. | +| `first_token_missing_reason` | X | Free-form outcome string, violates dictionary annotation. | + +The declared `trace: Dict[str, int]` is incorrect: current values include floats +(`predict_kv_bandwidth_mbps`), strings, `None`, and integer booleans. `mark()` +also forces arbitrary values to `int`, although most call sites bypass it. + +### Separately stored ProxyTask prediction/state fields + +`predict_stage`, `pred_slot_idx`, `pred_slot_ready_ts_ms`, +`pred_forward_start_ts_ms`, `pred_prefill_start_ts_ms`, +`pred_first_token_ts_ms`, `pred_decode_ms`, `pred_forward_end_ts_ms`, +`pred_worker_free_ts_ms`, and `pred_service_ms` are prediction state outside the +dictionary. `reservation_seq` and `recompute_generation` order reservation +updates; `has_started_forward` and `has_seen_first_token` are observed local +booleans. Several values are copied into `trace`, producing two mutable sources +of truth. `created_at` is a wall-clock float and is not currently exported. + +### Knowledge, injection, failure, and completion markers outside `trace` + +* `kv_ready_kids`, `text_only_kids`, and `miss_kids` are KDN-result + classifications, not proof of cache consumption. +* `kv_ready_meta` holds untyped KDN metadata and is not in `cacheroute_meta`. +* `kv_ack` exposes the Instance/KDN acknowledgement, including current + `payload_bytes`, `network_queue_ms`, `network_transfer_ms`, and + `network_total_ms` when supplied. Their producer clock/boundary is not encoded. +* `error` is a free-form `prepare_failed: ...`, `ready_failed: ...`, or + `stream_wrap_failed: ...` string. Exceptions can leak implementation detail. +* There is no direct vLLM prefill-start/end, decoded-token event, LMCache + load-start/end, or server completion event. Proxy `forward_end_ms` is response + consumption completion; non-streaming `first_token_ms` is only first body + chunk and explicitly sets `ttft_observable=0`. + +## 4. Other current metrics and attribution + +Proxy metrics are predictors rather than per-request telemetry: +`queue_predictor`/TTFT regressors estimate prefill time, +`decode_tpot_predictor` estimates decode service, and the Redis regressor +estimates pull time. Queue snapshots report aggregate instantaneous queue/active +counts plus reservation-derived backlog. Instance registration reports a frozen +capability object (model/tokenizer/adapters/KV layout/parallelism and detected +vLLM/LMCache versions), while heartbeats usually send only its fingerprint. +Topology reporting contains endpoint-level measured/inferred latency and link +bandwidth. None is a request-correlated execution measurement. + +### Validator metric-attribution table + +The validator sums every Prometheus series with each metric name, losing label +dimensions, then subtracts before from after. Classification assumes the current +unlabelled request path. + +| Metric | Request actual | Endpoint aggregate | Isolated-test inference | Concurrent attribution | +|---|---:|---:|---:|---:| +| `vllm:external_prefix_cache_queries_total` | No | Yes | Yes: query occurred | Unsupported | +| `vllm:external_prefix_cache_hits_total` | No | Yes | Yes: some external hit occurred | Unsupported | +| `vllm:prompt_tokens_cached_total` | No | Yes | Yes: cached-token delta | Unsupported | +| `vllm:prompt_tokens_total` | No | Yes | Yes: prompt-token delta | Unsupported | +| `lmcache_mp_lookup_requested_tokens_total` | No | Yes | Yes: requested tokens | Unsupported | +| `lmcache_mp_lookup_hit_tokens_total` | No | Yes | Yes: hit tokens/rate | Unsupported | +| `lmcache_mp_l2_prefetch_lookup_requests_total` | No | Yes | Yes | Unsupported | +| `lmcache_mp_l2_prefetch_lookup_objects_chunks_total` | No | Yes | Yes | Unsupported | +| `lmcache_mp_l2_prefetch_hit_chunks_total` | No | Yes | Yes | Unsupported | +| `lmcache_mp_l2_prefetch_load_submitted_requests_total` | No | Yes | Yes | Unsupported | +| `lmcache_mp_l2_prefetch_load_submitted_objects_chunks_total` | No | Yes | Yes | Unsupported | +| `lmcache_mp_l2_prefetch_load_completed_chunks_total` | No | Yes | Yes | Unsupported | +| `lmcache_mp_l2_load_completed_requests_total` | No | Yes | Yes | Unsupported | +| `lmcache_mp_l2_prefetch_failure_chunks_total` | No | Yes | Yes: absence/presence of failures | Unsupported | +| `lmcache_mp_l1_read_chunks_total` | No | Yes | Yes | Unsupported | +| `lmcache_mp_l1_write_chunks_total` | No | Yes | Yes | Unsupported | + +Even in an isolated test, record `value_kind=inferred`, the before/after scrape +times, endpoint, full label set (if retained), and the exclusivity assumption. +Counter reset, delayed emission, background work, and scrape races can invalidate +the inference. Future public LMCache events or request-labelled metrics can be +ingested through a small optional observer adapter and emitted as `observed` or +`actual` only when the upstream event specifies the request/operation identity +and measurement boundary. The core schema must not import or depend on a private +LMCache API. + +## 5. Where context is lost + +1. Client-to-Scheduler has no canonical correlation ID or trace controls. +2. The Scheduler integer ID is cyclic and process-local; it has no boot/session + namespace. +3. Proxy reconstructs `Request` but does not receive a typed trace context; + route-selection time begins only at Proxy receipt. +4. `build_body_for_instance()` intentionally produces an OpenAI body and drops + `Request_ID`; Proxy-to-Instance forwarding sends no trace headers. +5. Instance forwards the same body to vLLM, so it cannot correlate vLLM/LMCache + observation with the Scheduler request. +6. Text lookup calls do not propagate identity. KV injection propagates only the + legacy integer `request_id`, with no operation/trace/correlation ID. +7. Gateway v1 contracts have good request/correlation/target provenance but are + not connected to the production serving path. +8. Streaming byte forwarding identifies the first non-empty chunk, not the first + decoded token, and does not parse upstream IDs/events. +9. `cacheroute_meta` is assembled only at Proxy response export; stages from + Scheduler, KDN, Instance, vLLM, and LMCache cannot currently join it. + +## 6. Package-boundary evaluation + +| Candidate | Benefits | Problems | Decision | +|---|---|---|---| +| `kdn_server/observability/` | Close to #139/#140 types | Forces Proxy, Scheduler, Instance, and `core` to depend on a server package; likely dependency inversion and future import cycles. | Reject. | +| `core/observability/` | Existing common inward dependency; simple imports; same repository/version | Must enforce that it does not import request forwarding, config, FastAPI, or components. | **Recommend.** | +| standalone top-level package/distribution | Strong conceptual isolation | Packaging/versioning overhead, duplicate shared primitives, and uncertain installation in current entry points. | Revisit only if contracts become independently released. | + +`core/observability/` may import Python stdlib, Pydantic, and selected immutable +types from `kdn_server.domain`/#139 and outcome vocabulary from +`kdn_server.contracts.errors`/#140. To keep dependency direction cleaner, the +first PR should instead define trace-only enums in `core` and type resource IDs +as validated opaque strings; adapters attach existing domain/contract JSON +without importing Gateway implementations. A follow-up can move genuinely +shared primitives (UTC validation and safe outcome vocabulary) inward if +maintainers approve. It must never import FastAPI, Redis, vLLM, LMCache, GPU +libraries, `httpx`, network clients, component modules, or `core.fwd/config`. + +## 7. Stable v1 schemas + +All models use Pydantic `ConfigDict(frozen=True, extra="forbid")`, stable +`schema_version="cacheroute.trace.v1"`, JSON-mode ISO-8601 UTC timestamps, enum +wire values, arrays (not sets), and deterministic ordering. Export returns a new +frozen snapshot; collection is append-only and process-local. Unknown future +enum values require a version change rather than silently accepting route names. + +### `TraceContext` + +Required: `trace_id` (`trace_<32 lowercase hex>`), opaque non-empty `request_id`, +opaque non-empty `correlation_id`, `sampled: bool`, and `created_at` UTC. Optional: +`parent_trace_id`, `legacy_request_id: int`, and `expires_at` UTC. `request_id` +identifies one end-user generation request; `correlation_id` groups retries or +related requests; `trace_id` identifies one exported timeline. Validate expiry +after creation. Context contains no prompt, token content, credentials, Redis +keys, physical KV bytes/pointers/paths, or arbitrary baggage. + +### `TraceStageName` + +Stable values, ordered by a separate integer rank (not enum declaration order): + +```text +runtime_profile_resolution +knowledge_lookup +semantic_resolution +artifact_compatibility +capability_snapshot_discovery +token_lookup +artifact_lookup +cache_observation +cache_operation_queue +cache_prefetch_execution +cache_pin_execution +cache_unpin_execution +cache_clear_execution +cache_rebuild_execution +gateway_request +gateway_async_operation +tier_adapter_observation +instance_lmcache_load +proxy_prepare_queue +proxy_ready_queue +vllm_prefill +first_token +decode +completion +fallback +legacy_scan +legacy_dump +legacy_restore +legacy_inject +``` + +Queue operation is generic and carries `operation` from existing +`CacheOperationType`; execution stages are explicit because they have materially +different safety/latency semantics. Do not add `/route` strings, Python method +names, Redis calls, or LMCache private event names as stage values. + +### `TraceValueKind` + +Stable values: `predicted`, `observed`, `actual`, `inferred`, +`legacy_projected`. “Ambiguous” is an audit classification, not a valid new +measurement kind; adapters must choose a truthful kind or omit the measurement. + +### `TraceStageOutcome` + +Stable values: `pending`, `running`, `succeeded`, `failed`, `cancelled`, +`unsupported`, `stale`, `partial`, `skipped`, `fallback`. Map #140 outcomes +directly: `success -> succeeded`, all identical values remain identical, +`incompatible` is represented as `failed` with `outcome_code="incompatible"`, +`text_fallback -> fallback`, and `idempotency_conflict -> failed`. Maintainer +approval is needed on whether to add `incompatible` and `idempotency_conflict` +directly instead of keeping the #140 code alongside the stage outcome. + +### `TraceProvenance` + +Required: `source_component` (`client|scheduler|proxy|kdn|gateway|instance|vllm|lmcache|legacy_adapter|test`), +`runtime_profile`, and `captured_at` UTC. Optional: `source_endpoint` (logical +endpoint/route label, never credentials or URL query), `endpoint_id`, +`endpoint_generation`, `gateway_profile`, `compatibility_profile_id`, +`observation_source`, `profile_id`, `adapter`, `tier`, `source_version`, +`fresh_until`, and `legacy: bool`. Targeted Gateway observations require endpoint +ID and generation together; non-Legacy generation is positive, Legacy unknown is +zero. Freshness follows #139 observation semantics and cannot precede capture. + +### `TraceMeasurement` + +Required: `name` (a versioned allow-listed measurement name), `kind`, one and +only one of `duration_ns`, `count`, `bytes`, `tokens`, `ratio`, `boolean`, +`timestamp`, or safe scalar `value`, plus `provenance`. Optional: `unit` only for +generic scalar values, `uncertainty`, `sample_count`, `observed_at`, `expires_at`, +and `legacy_name`. Duration unit is structurally fixed to nanoseconds and must be +non-negative. Ratios are finite and `[0,1]`; counts/bytes/tokens are non-negative. +Predicted and actual values are separate objects and may share a semantic name. +Generic values reject mappings/lists to prevent arbitrary payload smuggling. + +### `TraceStage` + +Required: `stage_id` (`stage_<32 hex>`), `name`, `sequence` (non-negative, +process-assigned append sequence), `outcome`, `provenance`, and tuple +`measurements`. Optional: UTC `started_at`/`ended_at`, `duration_ns`, +`operation_id`, `artifact_id`, safe `outcome_code`, safe `error_code`, sanitized +`error_message`, `retryable`, `fallback_eligible`, `fallback_stage_id`, +`parent_stage_id`, `partial_reason`, and `legacy`. Start/end are correlation +markers only. If both exist, order them by UTC but do not derive the duration from +them. Duration is local monotonic elapsed time. A terminal stage is immutable; +an in-progress collector state is not exported as the same object. + +### `RequestTrace` + +Required: `context`, ordered tuple `stages`, `exported_at` UTC, `complete: bool`, +and schema version. Optional: `source_components`, sanitized top-level +`error_code`, `legacy_trace` (JSON-safe copy of current dictionary), and +`legacy_cacheroute_meta` during migration. Canonical ordering is +`(stage-rank, started_at-or-max, source_component, sequence, stage_id)`; sequence +breaks ties but does not imply cross-process causality. JSON uses enum strings, +UTC `Z` timestamps, integer nanoseconds, and stable tuple-to-array conversion. + +### `CacheOperationTrace` + +Required: existing `CacheOperationTask.task_id` as `operation_id`, operation +type, `trace_id`, ordered stages, current existing task state, and provenance. +Optional: artifact/endpoint IDs and generation, idempotency key hash (never the +raw secret-like key), created/updated/expiry UTC timestamps, sanitized error and +fallback fields, and legacy projection. It references rather than duplicates +`CacheArtifact`, `CacheReplicaObservation`, `LMCacheEndpoint`, or +`CacheOperationTask`. + +### `OperationWaiterLink` + +Required: `operation_id`, request `trace_id`, request/correlation IDs, +`attached_at`, monotonically allocated `attach_sequence`, and state +`waiting|completed|cancelled|detached|expired`. Optional: `detached_at`, safe +reason, and operation outcome observed at attachment. This is a correlation +record, not scheduler state and cannot influence dispatch. + +### Secret and physical-KV rejection + +All models use `extra="forbid"` plus recursive validators rejecting case-folded +field names such as `password`, `secret`, `authorization`, `cookie`, `token_value`, +`redis_key`, `kv_bytes`, `physical_kv`, `tensor`, `pointer`, `device_address`, and +absolute/file paths. `tokens` means a numeric count only. Prompts, generated +content, raw exceptions, HTTP bodies/headers, Redis keys, manifests, physical KV +locations, and credentials are never valid trace values. Legacy projection +allow-lists known scalar keys rather than embedding the whole acknowledgement. + +## 8. Clock design + +Inject a `TraceClock` with `utc_now()` and `monotonic_ns()`; production uses +timezone-aware `datetime.now(timezone.utc)` and `time.perf_counter_ns()`. +Tests use a deterministic fake. On local stage start, store both readings in the +mutable collector. On finish, calculate `duration_ns = end_mono - start_mono`, +reject negatives, and export UTC start/end plus the local duration. Never export +monotonic readings, compare them across processes, or derive durations by +subtracting epoch milliseconds. + +A stage spanning Proxy and Instance is represented as either (a) two linked +local stages with independently measured durations, or (b) one parent stage with +UTC correlation markers and child stages. Its end-to-end duration is measured by +one process that owns both send and response receipt (for example, Proxy Gateway +request duration). Remote work has its own Instance duration. Network/clock +offset must not be inferred as the difference between unrelated monotonic clocks. + +## 9. Compatibility-first propagation + +| Mechanism | Use | Decision | +|---|---|---| +| Add fields directly to current Request | Easy serialization, but changes a widely used dataclass/wire body | Do not use in Phase 1/2. | +| Dedicated internal envelope | Separates routing body and trace context; explicit filtering | Primary mechanism from Phase 3. | +| HTTP headers | Works at hops and avoids OpenAI JSON pollution; intermediaries may strip them | Mirror only allow-listed context IDs per internal hop. | +| Process-local collector reference | Efficient within one process, impossible across processes | Use only behind adapters; never serialize references. | + +Recommended envelope: + +```json +{ + "request": {"Request_ID": 42, "Prompt": {}, "Service": {}, "Task": {}}, + "_cacheroute_internal": { + "trace_context": { + "schema_version": "cacheroute.trace-context.v1", + "trace_id": "trace_...", + "request_id": "req_...", + "correlation_id": "corr_...", + "legacy_request_id": 42, + "sampled": true + } + } +} +``` + +Proxy accepts both bare legacy Request and envelope. It strips the internal +section before constructing `instance_body`, then sends only allow-listed +`x-cacheroute-trace-id`, `x-cacheroute-request-id`, and +`x-cacheroute-correlation-id` to Instance. Instance strips these before vLLM +unless a public adapter explicitly supports correlation. KDN/Gateway versioned +calls place IDs in their existing request fields; headers are hop diagnostics, +not competing identity. Streaming merges immutable snapshots at the final meta +event; non-streaming does so in `_cacheroute_meta`. Existing meta fields and +dictionary remain unchanged while an optional `request_trace_v1` is added. + +If tracing is disabled, no collector/stages are allocated and propagation may be +limited to existing IDs. If not sampled, propagate context IDs with +`sampled=false` for correlation but collect/export no detailed measurements. +Sampling is decided once at ingress using a deterministic policy and never +changed downstream. Any trace validation/export error is swallowed at the +observability boundary with a rate-limited diagnostic; it must never change +routing, forwarding, fallback, reuse, queue ordering, or responses. + +## 10. Shared-operation correlation + +Maintain a bounded `operation_id -> ordered waiter links` registry in the KDN or +Gateway orchestration layer, separate from `CacheOperationTask` and scheduling. + +* **Attach:** idempotent on `(operation_id, trace_id)`. Duplicate legacy request + IDs are allowed because trace ID is authoritative. Return the existing link on + retry; allocate a stable sequence only once. +* **Detach/cancel:** request cancellation transitions only its link; it does not + cancel shared work unless the existing operation owner independently decides + to do so. No scheduling-policy change is implied. +* **Completion races:** retain terminal operation summaries until TTL. A waiter + registering after completion receives a completed link and the same immutable + summary. Completion snapshots the then-current waiters; later links remain + deterministically ordered after them. +* **Expiry:** terminal task/link TTL is explicit. Active-task entries have a + maximum lifetime and emit `expired`; purge uses injected UTC/monotonic clocks. +* **Bounded memory:** configure maximum operations, waiters per operation, and + terminal TTL; evict expired terminal records first. If capacity remains full, + omit correlation with a diagnostic rather than affecting the operation. +* **Idempotency:** the existing operation idempotency key controls operation + deduplication; waiter idempotency is the pair above. Never use request ID alone. +* **Ordering:** export links by `(attach_sequence, trace_id)` and operation stages + by the trace canonical order, never hash/set iteration order. + +## 11. Migration map (no silent renames) + +| Current representation | v1 projection | +|---|---| +| `Request_ID` / `ProxyTask.request_id` | `TraceContext.legacy_request_id`; generate opaque request/trace IDs. | +| `*_start_ms`, `*_end_ms` | UTC `started_at`/`ended_at` observations plus separately measured local `duration_ns`. | +| `predict_*` | `TraceMeasurement(kind="predicted")`; never overwrite. | +| overwritten `predict_prepare_ms` / `predict_know_prepare_ms` | Preserve original dictionary; adapter emits the final value as `legacy_projected` because original kind is unknowable, and emits unambiguous `actual_*` separately when present. | +| `actual_compute_ms` | Legacy projection of Proxy-observed Instance round trip to first chunk, not compute. | +| `actual_vllm_internal_ms` | Actual at Proxy boundary with name `proxy_forward_to_first_chunk`; not vLLM-internal. | +| `first_token_ms` | `first_token` only when token parsing proves it; otherwise `gateway_request`/forward first-chunk observation. | +| `decode_start_ms/end_ms` | Legacy-projected decode approximation until vLLM events exist. | +| `text_actual_path`, `kvcache_actual_path` | Explicit stage outcome + error/fallback code; retain strings in legacy dictionary. | +| `error` / `first_token_missing_reason` | Sanitized stable `error_code` and optional safe message; retain original only in existing meta. | +| integer boolean flags | Typed booleans in stage fields. | +| `kv_ack.network_*` | Measurement with declared producer/boundary and kind; otherwise legacy-projected. | +| queue snapshots / Prometheus deltas | Endpoint-level `observed`, or `inferred` only under documented isolated-test assumptions. | + +## 12. Rollout and exact file plan + +### Phase 1 — contracts only + +Add: + +* `core/observability/__init__.py` — public, dependency-light exports. +* `core/observability/enums.py` — stage, kind, outcome, component vocabularies and + stable rank table. +* `core/observability/models.py` — frozen schemas and validators. +* `core/observability/clock.py` — clock protocol, production clock, test seam. +* `tests/core/observability/test_models.py` +* `tests/core/observability/test_clock.py` +* `tests/core/observability/test_security.py` +* `docs/observability/v1-contract.md` — stable public contract (not root README). + +Do not modify production component files. This is the independently reviewable +first PR: enums, models, deterministic CPU-only tests, and contract documentation +only. It must not include a collector, adapter, propagation, or I/O. + +### Phase 2 — compatibility adapter + +Add `core/observability/legacy_proxy_trace.py` and its tests. Modify only +`proxy/proxy.py` at `build_cacheroute_meta` to optionally append +`request_trace_v1`, and possibly correct `ProxyTask.trace`'s annotation to +`Dict[str, Any]` without changing values. Retain the dictionary and meta shape. + +### Phase 3 — component instrumentation/propagation + +Add `core/observability/collector.py` and internal-envelope helpers. Modify +`scheduler/scheduler.py`, `core/fwd.py`, `proxy/proxy.py`, +`proxy/queue/task.py`, `proxy/queue/manager.py`, `proxy/queue/instance_queues.py`, +`proxy/queue/knowledge.py`, KDN cache-service facade call sites, +`kdn_server/gateway/base.py`, `instance/instance_api.py`, +`instance/control_plane.py`, and `instance/kv_service.py`. Add component and Mock +Gateway timeline tests. Changes remain observational and additive. + +### Phase 4 — optional runtime observations + +Add optional adapter modules under `instance/observability/` (public event or +labelled-metric adapters only) and validator reporting changes. No core model +imports an LMCache/vLLM client. Keep unlabelled Prometheus delta classification +as endpoint aggregate/isolated inference. + +## 13. CPU-only test matrix + +| Test | Assertions | External requirements | +|---|---|---| +| Serialization | Exact enum strings, UTC JSON, ns integers, deterministic stage/measurement order, round trip | None | +| Frozen export | Exported context/stage/trace cannot mutate; collector append creates a new snapshot | None | +| Clock | Injected monotonic duration, zero allowed, negative rejected, no epoch-derived duration | None | +| Value kinds | Same semantic metric can coexist as predicted/observed/actual/inferred; no overwrite | None | +| Provenance | Runtime/Profile/endpoint pair and positive generation validation; Legacy generation zero only | None | +| Freshness | valid expiry; stale observation retained with stale outcome; bad ordering rejected | None | +| Outcomes | unsupported/partial/failed/skipped/cancelled/fallback serialization and safe errors | None | +| Shared task | multiple waiters, duplicates, late registration, cancellation, expiry, bounded capacity, stable order | None | +| Legacy projection | Every current trace key maps or remains allow-listed legacy; overwritten predictions marked ambiguous projection | None | +| Free-form compatibility | Current mixed int/float/string/None dictionary is preserved byte-for-value in existing meta | None | +| Disabled/sampled | disabled allocates no stages; unsampled propagates only context; neither changes callback results/order | None | +| Security | reject secrets, auth headers, raw tokens/prompts, Redis keys, file paths, tensors, KV bytes/pointers, extras | None | +| Mock Gateway timeline | request, async operation, observation, shared waiter, Instance load, completion combine deterministically | In-memory Mock Gateway only | +| Import isolation | importing `core.observability` does not load FastAPI/httpx/redis/vllm/lmcache/torch/CUDA modules | None | +| Environment independence | full suite runs with network disabled and without Redis, LMCache, vLLM, GPU, or tracing backend | None | + +Validation for the first PR should run from the repository root with +`pytest -q tests/core/observability` and an import-isolation subprocess test. +Failure is any nondeterministic order, mutable exported object, negative duration +acceptance, kind collapse, forbidden physical/secret field acceptance, external +service access, or behavior difference when tracing is disabled. + +## 14. Maintainer decisions and unresolved questions + +1. Is `core/observability/` accepted, or should #139/#140 common primitives first + move inward to avoid `core -> kdn_server.domain/contracts` imports? +2. Is the canonical request ID generated at Client when supplied or always at + Scheduler? What retry semantics should preserve it? +3. Should `correlation_id` default to `request_id`, or be mandatory and distinct? +4. Are `incompatible` and `idempotency_conflict` first-class stage outcomes, or + should the trace retain the smaller stage vocabulary plus #140 `outcome_code`? +5. Which measurement names warrant a closed v1 allow-list in the first PR? +6. May safe `source_endpoint` contain normalized route templates such as + `/v1/chat/completions`, or only logical endpoint IDs? Arbitrary method names + must not become stage names either way. +7. What are the default sampling rate, terminal operation TTL, maximum tracked + operations, and maximum waiters per operation? +8. Should an unsampled request export a minimal `RequestTrace`, or only preserve + context in existing `cacheroute_meta`? +9. Can upstream streaming adapters reliably distinguish first decoded token from + role/metadata chunks across supported vLLM versions? Until confirmed, keep the + existing marker legacy-projected. +10. Which `kv_ack` network durations are measured with monotonic clocks at their + producer, and which are estimates? They must remain Legacy/ambiguous until + provenance is confirmed. +11. Is preserving raw free-form `error` in legacy meta acceptable during the + transition, given that v1 exports will sanitize it? + +These decisions do not block a contract-only first PR if the initial schema +marks disputed fields optional and the PR contains no production instrumentation. From bd5daa1c26f5972aa78c787f619fd2ac8373aa2f Mon Sep 17 00:00:00 2001 From: RickZ <121943721+rickisba@users.noreply.github.com> Date: Wed, 29 Jul 2026 16:35:37 +0800 Subject: [PATCH 2/2] feat: add v1 observability contracts and collector --- cacheroute_observability/README.md | 48 ++ cacheroute_observability/__init__.py | 20 + cacheroute_observability/clock.py | 49 +++ cacheroute_observability/collector.py | 205 +++++++++ cacheroute_observability/enums.py | 92 ++++ .../legacy_proxy_trace.py | 200 +++++++++ cacheroute_observability/models.py | 415 ++++++++++++++++++ .../issue-141-unified-observability.md | 68 ++- scripts/demo_observability_v1.py | 100 +++++ test/observability/conftest.py | 22 + test/observability/test_clock.py | 24 + test/observability/test_collector.py | 72 +++ test/observability/test_demo.py | 23 + test/observability/test_enums.py | 29 ++ test/observability/test_import_isolation.py | 18 + test/observability/test_legacy_proxy_trace.py | 48 ++ test/observability/test_models.py | 93 ++++ test/observability/test_security.py | 26 ++ 18 files changed, 1517 insertions(+), 35 deletions(-) create mode 100644 cacheroute_observability/README.md create mode 100644 cacheroute_observability/__init__.py create mode 100644 cacheroute_observability/clock.py create mode 100644 cacheroute_observability/collector.py create mode 100644 cacheroute_observability/enums.py create mode 100644 cacheroute_observability/legacy_proxy_trace.py create mode 100644 cacheroute_observability/models.py rename {docs => doc}/research/issue-141-unified-observability.md (92%) create mode 100644 scripts/demo_observability_v1.py create mode 100644 test/observability/conftest.py create mode 100644 test/observability/test_clock.py create mode 100644 test/observability/test_collector.py create mode 100644 test/observability/test_demo.py create mode 100644 test/observability/test_enums.py create mode 100644 test/observability/test_import_isolation.py create mode 100644 test/observability/test_legacy_proxy_trace.py create mode 100644 test/observability/test_models.py create mode 100644 test/observability/test_security.py diff --git a/cacheroute_observability/README.md b/cacheroute_observability/README.md new file mode 100644 index 0000000..3788490 --- /dev/null +++ b/cacheroute_observability/README.md @@ -0,0 +1,48 @@ +# CacheRoute observability v1 + +`cacheroute_observability` is the dependency-light Phase 1 contract and collection +boundary for Issue #141. It imports only the Python standard library and Pydantic, +so Client, Scheduler, Proxy, KDN, Gateway, and Instance code can adopt it later +without loading the CacheRoute application graph. + +## Models and vocabulary + +Frozen, extra-forbidding contracts represent trace context, provenance, typed +measurements, ordered stages, request traces, cache-operation traces, and links +between one shared operation and multiple waiting requests. Stable enums never +encode route names, implementation methods, Redis commands, or private LMCache +events. `TraceStageState` describes lifecycle (`pending`, `running`, `completed`); +`TraceStageOutcome` independently describes the terminal result. + +## Clocks and collector + +UTC wall-clock values support cross-process correlation. Durations come only from +local `monotonic_ns()` readings. `ManualTraceClock` makes both sources deterministic +in CPU-only tests. `TraceCollector` starts stages with both readings, calculates a +local monotonic duration on finish, preserves append sequence (including repeated +stage names), and exports a new immutable snapshot. Disabled and unsampled +collectors allocate no stages. Optional instrumentation can use `safely()` so an +observability failure cannot affect caller behavior. + +## Legacy projection and security + +`project_legacy_proxy_trace()` is a pure allow-list adapter. It classifies known +Proxy fields, treats ambiguous and overwritten values as `legacy_projected`, uses +truthful Proxy-boundary names, sanitizes known fallback codes, and omits unknown +keys and raw error text. It never embeds the source trace or `kv_ack` mapping. +Contracts reject extras, containers as generic scalar values, secret/physical-KV +field names, raw content, paths, credentials, Redis keys, KV bytes, tensors, and +pointers. + +## Phase 1 limitations + +This package provides contracts, a process-local collector, Legacy projection, +and a CPU-only demonstration. It does **not** instrument production components, +propagate context, change `cacheroute_meta`, perform Gateway/network I/O, attribute +aggregate metrics to requests, or integrate with LMCache/vLLM. Those remain later +Issue #141 phases after review. + +The temporary design record is +[`doc/research/issue-141-unified-observability.md`](../doc/research/issue-141-unified-observability.md) +and will be removed after Issue #141 is completed and stable material is moved to +maintained documentation. diff --git a/cacheroute_observability/__init__.py b/cacheroute_observability/__init__.py new file mode 100644 index 0000000..2ae9381 --- /dev/null +++ b/cacheroute_observability/__init__.py @@ -0,0 +1,20 @@ +"""Dependency-light public exports for CacheRoute observability v1.""" +from .clock import ManualTraceClock, SystemTraceClock, TraceClock +from .collector import TraceCollector +from .enums import ( + OperationWaiterState, TraceComponent, TraceStageName, TraceStageOutcome, + TraceStageState, TraceValueKind, +) +from .legacy_proxy_trace import project_legacy_proxy_trace +from .models import ( + CacheOperationTrace, OperationWaiterLink, RequestTrace, TraceContext, + TraceMeasurement, TraceProvenance, TraceStage, +) + +__all__ = [ + "CacheOperationTrace", "ManualTraceClock", "OperationWaiterLink", + "OperationWaiterState", "RequestTrace", "SystemTraceClock", "TraceClock", + "TraceCollector", "TraceComponent", "TraceContext", "TraceMeasurement", + "TraceProvenance", "TraceStage", "TraceStageName", "TraceStageOutcome", + "TraceStageState", "TraceValueKind", "project_legacy_proxy_trace", +] diff --git a/cacheroute_observability/clock.py b/cacheroute_observability/clock.py new file mode 100644 index 0000000..c5bf2b0 --- /dev/null +++ b/cacheroute_observability/clock.py @@ -0,0 +1,49 @@ +"""Injectable clocks that keep correlation timestamps separate from durations.""" +from __future__ import annotations + +import time +from datetime import datetime, timedelta, timezone +from typing import Protocol, runtime_checkable + + +@runtime_checkable +class TraceClock(Protocol): + def utc_now(self) -> datetime: ... + def monotonic_ns(self) -> int: ... + + +class SystemTraceClock: + def utc_now(self) -> datetime: + return datetime.now(timezone.utc) + + def monotonic_ns(self) -> int: + return time.perf_counter_ns() + + +class ManualTraceClock: + def __init__(self, utc_value: datetime, monotonic_value: int = 0): + if utc_value.tzinfo is None or utc_value.utcoffset() != timedelta(0): + raise ValueError("utc_value must be timezone-aware UTC") + if monotonic_value < 0: + raise ValueError("monotonic_value must be non-negative") + self._utc_value = utc_value + self._monotonic_value = monotonic_value + + def utc_now(self) -> datetime: + return self._utc_value + + def monotonic_ns(self) -> int: + return self._monotonic_value + + def advance_ns(self, value: int) -> None: + if value < 0: + raise ValueError("monotonic advance must be non-negative") + self._monotonic_value += value + + def advance_time(self, value: timedelta) -> None: + if value < timedelta(0): + raise ValueError("wall-clock advance must be non-negative") + self._utc_value += value + + +__all__ = ["ManualTraceClock", "SystemTraceClock", "TraceClock"] diff --git a/cacheroute_observability/collector.py b/cacheroute_observability/collector.py new file mode 100644 index 0000000..55c5166 --- /dev/null +++ b/cacheroute_observability/collector.py @@ -0,0 +1,205 @@ +"""Process-local append-only trace collection with monotonic durations.""" +from __future__ import annotations + +from dataclasses import dataclass, field +from datetime import datetime +from typing import Callable +from uuid import uuid4 + +from .clock import SystemTraceClock, TraceClock +from .enums import TraceStageName, TraceStageOutcome, TraceStageState +from .models import RequestTrace, TraceContext, TraceMeasurement, TraceProvenance, TraceStage + + +@dataclass +class _OpenStage: + stage_id: str + sequence: int + name: TraceStageName + provenance: TraceProvenance + started_at: datetime + started_monotonic_ns: int + parent_stage_id: str | None + operation_id: str | None + artifact_id: str | None + measurements: list[TraceMeasurement] = field(default_factory=list) + + +class TraceCollector: + """Collect local stages; exported contracts never contain monotonic readings.""" + + def __init__( + self, + context: TraceContext, + *, + clock: TraceClock | None = None, + enabled: bool = True, + id_factory: Callable[[], str] | None = None, + ): + self.context = context + self.clock = clock or SystemTraceClock() + self.enabled = bool(enabled) + self._id_factory = id_factory or (lambda: f"stage_{uuid4().hex}") + self._next_sequence = 0 + self._open: dict[str, _OpenStage] = {} + self._timeline: list[str] = [] + self._completed: dict[str, TraceStage] = {} + + @classmethod + def create( + cls, + *, + request_id: str, + correlation_id: str, + legacy_request_id: int | None = None, + sampled: bool = True, + clock: TraceClock | None = None, + enabled: bool = True, + id_factory: Callable[[], str] | None = None, + ) -> "TraceCollector": + """Create a context at the injected clock and return its collector.""" + selected_clock = clock or SystemTraceClock() + context = TraceContext( + trace_id=f"trace_{uuid4().hex}", + request_id=request_id, + correlation_id=correlation_id, + legacy_request_id=legacy_request_id, + sampled=sampled, + created_at=selected_clock.utc_now(), + ) + return cls(context, clock=selected_clock, enabled=enabled, id_factory=id_factory) + + @property + def collecting(self) -> bool: + return self.enabled and self.context.sampled + + def start_stage( + self, + name: TraceStageName, + provenance: TraceProvenance, + *, + parent_stage_id: str | None = None, + operation_id: str | None = None, + artifact_id: str | None = None, + ) -> str | None: + if not self.collecting: + return None + stage_id = self._id_factory() + if stage_id in self._open or stage_id in self._completed: + raise ValueError("stage_id must be unique within a collector") + stage = _OpenStage( + stage_id=stage_id, + sequence=self._next_sequence, + name=TraceStageName(name), + provenance=provenance, + started_at=self.clock.utc_now(), + started_monotonic_ns=self.clock.monotonic_ns(), + parent_stage_id=parent_stage_id, + operation_id=operation_id, + artifact_id=artifact_id, + ) + self._next_sequence += 1 + self._open[stage_id] = stage + self._timeline.append(stage_id) + return stage_id + + def append_measurement(self, stage_id: str | None, measurement: TraceMeasurement) -> bool: + if not self.collecting or stage_id is None: + return False + stage = self._open.get(stage_id) + if stage is None: + if stage_id in self._completed: + raise ValueError("cannot append to a completed stage") + raise KeyError("unknown stage_id") + stage.measurements.append(measurement) + return True + + def finish_stage( + self, + stage_id: str | None, + outcome: TraceStageOutcome, + *, + outcome_code: str | None = None, + error_code: str | None = None, + error_message: str | None = None, + retryable: bool = False, + fallback_eligible: bool = False, + fallback_stage_id: str | None = None, + partial_reason: str | None = None, + ) -> TraceStage | None: + if not self.collecting or stage_id is None: + return None + if stage_id in self._completed: + raise ValueError("stage is already completed") + stage = self._open.get(stage_id) + if stage is None: + raise KeyError("unknown stage_id") + ended_at = self.clock.utc_now() + elapsed = self.clock.monotonic_ns() - stage.started_monotonic_ns + if elapsed < 0: + raise ValueError("monotonic clock moved backwards") + completed = TraceStage( + stage_id=stage.stage_id, + sequence=stage.sequence, + name=stage.name, + state=TraceStageState.COMPLETED, + outcome=outcome, + provenance=stage.provenance, + measurements=tuple(stage.measurements), + started_at=stage.started_at, + ended_at=ended_at, + duration_ns=elapsed, + parent_stage_id=stage.parent_stage_id, + operation_id=stage.operation_id, + artifact_id=stage.artifact_id, + outcome_code=outcome_code, + error_code=error_code, + error_message=error_message, + retryable=retryable, + fallback_eligible=fallback_eligible, + fallback_stage_id=fallback_stage_id, + partial_reason=partial_reason, + ) + del self._open[stage_id] + self._completed[stage_id] = completed + return completed + + def export(self, *, complete: bool = False, error_code: str | None = None) -> RequestTrace: + stages: list[TraceStage] = [] + for stage_id in self._timeline: + completed = self._completed.get(stage_id) + if completed is not None: + stages.append(completed) + continue + opened = self._open[stage_id] + stages.append(TraceStage( + stage_id=opened.stage_id, + sequence=opened.sequence, + name=opened.name, + state=TraceStageState.RUNNING, + provenance=opened.provenance, + measurements=tuple(opened.measurements), + started_at=opened.started_at, + parent_stage_id=opened.parent_stage_id, + operation_id=opened.operation_id, + artifact_id=opened.artifact_id, + )) + components = tuple(dict.fromkeys(stage.provenance.source_component for stage in stages)) + return RequestTrace( + context=self.context, + stages=tuple(stages), + exported_at=self.clock.utc_now(), + complete=complete, + source_components=components, + error_code=error_code, + ) + + def safely(self, operation: Callable[[], object], default=None): + """Run optional instrumentation without allowing failures into caller behavior.""" + try: + return operation() + except Exception: + return default + + +__all__ = ["TraceCollector"] diff --git a/cacheroute_observability/enums.py b/cacheroute_observability/enums.py new file mode 100644 index 0000000..e0b40b9 --- /dev/null +++ b/cacheroute_observability/enums.py @@ -0,0 +1,92 @@ +"""Stable string vocabularies for CacheRoute observability v1.""" +from enum import Enum + + +class _StringEnum(str, Enum): + pass + + +class TraceComponent(_StringEnum): + CLIENT = "client" + SCHEDULER = "scheduler" + PROXY = "proxy" + KDN = "kdn" + GATEWAY = "gateway" + INSTANCE = "instance" + VLLM = "vllm" + LMCACHE = "lmcache" + LEGACY_ADAPTER = "legacy_adapter" + TEST = "test" + + +class TraceValueKind(_StringEnum): + PREDICTED = "predicted" + OBSERVED = "observed" + ACTUAL = "actual" + INFERRED = "inferred" + LEGACY_PROJECTED = "legacy_projected" + + +class TraceStageState(_StringEnum): + PENDING = "pending" + RUNNING = "running" + COMPLETED = "completed" + + +class TraceStageOutcome(_StringEnum): + SUCCESS = "success" + UNSUPPORTED = "unsupported" + INCOMPATIBLE = "incompatible" + STALE = "stale" + PARTIAL = "partial" + FAILED = "failed" + CANCELLED = "cancelled" + SKIPPED = "skipped" + TEXT_FALLBACK = "text_fallback" + IDEMPOTENCY_CONFLICT = "idempotency_conflict" + + +class TraceStageName(_StringEnum): + RUNTIME_PROFILE_RESOLUTION = "runtime_profile_resolution" + KNOWLEDGE_LOOKUP = "knowledge_lookup" + SEMANTIC_RESOLUTION = "semantic_resolution" + ARTIFACT_COMPATIBILITY = "artifact_compatibility" + CAPABILITY_SNAPSHOT_DISCOVERY = "capability_snapshot_discovery" + TOKEN_LOOKUP = "token_lookup" + ARTIFACT_LOOKUP = "artifact_lookup" + CACHE_OBSERVATION = "cache_observation" + CACHE_OPERATION_QUEUE = "cache_operation_queue" + CACHE_PREFETCH_EXECUTION = "cache_prefetch_execution" + CACHE_PIN_EXECUTION = "cache_pin_execution" + CACHE_UNPIN_EXECUTION = "cache_unpin_execution" + CACHE_CLEAR_EXECUTION = "cache_clear_execution" + CACHE_REBUILD_EXECUTION = "cache_rebuild_execution" + GATEWAY_REQUEST = "gateway_request" + GATEWAY_ASYNC_OPERATION = "gateway_async_operation" + TIER_ADAPTER_OBSERVATION = "tier_adapter_observation" + INSTANCE_LMCACHE_LOAD = "instance_lmcache_load" + PROXY_PREPARE_QUEUE = "proxy_prepare_queue" + PROXY_READY_QUEUE = "proxy_ready_queue" + VLLM_PREFILL = "vllm_prefill" + FIRST_TOKEN = "first_token" + DECODE = "decode" + COMPLETION = "completion" + FALLBACK = "fallback" + LEGACY_SCAN = "legacy_scan" + LEGACY_DUMP = "legacy_dump" + LEGACY_RESTORE = "legacy_restore" + LEGACY_INJECT = "legacy_inject" + + +class OperationWaiterState(_StringEnum): + WAITING = "waiting" + COMPLETED = "completed" + CANCELLED = "cancelled" + DETACHED = "detached" + EXPIRED = "expired" + + +__all__ = [ + "OperationWaiterState", "TraceComponent", "TraceStageName", + "TraceStageOutcome", "TraceStageState", "TraceValueKind", +] diff --git a/cacheroute_observability/legacy_proxy_trace.py b/cacheroute_observability/legacy_proxy_trace.py new file mode 100644 index 0000000..3752135 --- /dev/null +++ b/cacheroute_observability/legacy_proxy_trace.py @@ -0,0 +1,200 @@ +"""Pure projection from the current free-form Proxy trace into safe v1 contracts.""" +from __future__ import annotations + +import hashlib +from datetime import datetime, timezone +from typing import Any, Mapping + +from .enums import TraceComponent, TraceStageName, TraceStageOutcome, TraceStageState, TraceValueKind +from .models import RequestTrace, TraceContext, TraceMeasurement, TraceProvenance, TraceStage + + +_STAGE_KEYS: tuple[tuple[TraceStageName, tuple[str, ...]], ...] = ( + (TraceStageName.RUNTIME_PROFILE_RESOLUTION, ("proxy_recv_ms", "route_select_start_ms", "route_select_end_ms")), + (TraceStageName.KNOWLEDGE_LOOKUP, ("kdn_fetch_start_ms", "kdn_fetch_end_ms")), + (TraceStageName.PROXY_PREPARE_QUEUE, ( + "proxy_enqueue_ms", "prepare_queue_enqueue_ms", "prepare_dequeue_ms", "prepare_start_ms", + "ready_enqueue_ms", "actual_prepare_ms", "actual_prepare_total_ms", "actual_know_prepare_ms", + "prepare_buffer_wait_ms", "predict_prepare_ms", "predict_know_prepare_ms", + "predict_prepare_queue_wait_ms", "predict_kv_transfer_ms", "predict_prepare_prefix_ms", + "predict_kv_prepare_service_ms", "predict_prepare_initial_ms", "predict_prepare_model_ms", + "predict_prepare_corrected_ms", "prepare_error_ms", + )), + (TraceStageName.INSTANCE_LMCACHE_LOAD, ( + "kv_inject_queue_enqueue_ms", "kv_inject_reserved_start_ms", "kv_inject_start_ms", + "kv_inject_end_ms", "kv_ack_start_ms", "kv_ack_end_ms", "kdn_link_wait_start_ms", + "kdn_link_wait_end_ms", "predict_redis_kv_load_ms", + )), + (TraceStageName.PROXY_READY_QUEUE, ( + "ready_dequeue_ms", "forward_wait_start_ms", "forward_wait_end_ms", "forward_start_ms", + "actual_ready_queue_ms", "predict_queue_wait_ms", + )), + (TraceStageName.VLLM_PREFILL, ( + "predict_text_prefill_ms", "predict_residual_prefill_ms", "predict_prefill_service_ms", + "predict_vllm_internal_ms", "actual_vllm_internal_ms", "actual_compute_ms", + )), + (TraceStageName.FIRST_TOKEN, ("pred_first_token_ts_ms", "first_token_ms", "ttft_observable")), + (TraceStageName.DECODE, ("decode_start_ms", "decode_end_ms", "predict_decode_ms")), + (TraceStageName.COMPLETION, ("forward_end_ms", "actual_total_ms", "predict_total_ms", "predict_error_ms")), +) + +_TIMESTAMP_KEYS = { + "proxy_recv_ms", "route_select_start_ms", "route_select_end_ms", "kdn_fetch_start_ms", + "kdn_fetch_end_ms", "proxy_enqueue_ms", "prepare_queue_enqueue_ms", "prepare_dequeue_ms", + "prepare_start_ms", "ready_enqueue_ms", "prepare_error_ms", "kv_inject_queue_enqueue_ms", + "kv_inject_reserved_start_ms", "kv_inject_start_ms", "kv_inject_end_ms", "kv_ack_start_ms", + "kv_ack_end_ms", "kdn_link_wait_start_ms", "kdn_link_wait_end_ms", "ready_dequeue_ms", + "forward_wait_start_ms", "forward_wait_end_ms", "forward_start_ms", "pred_first_token_ts_ms", + "first_token_ms", "decode_start_ms", "decode_end_ms", "forward_end_ms", +} +_OVERWRITTEN_PREDICTIONS = {"predict_prepare_ms", "predict_know_prepare_ms", "predict_prepare_corrected_ms"} +_CLEAR_ACTUAL_NAMES = { + "actual_prepare_ms": "proxy_kv_ack_duration", + "actual_prepare_total_ms": "proxy_prepare_total_duration", + "actual_know_prepare_ms": "proxy_prepare_to_ready_duration", + "actual_ready_queue_ms": "proxy_ready_queue_duration", + "actual_total_ms": "proxy_enqueue_to_first_chunk_duration", + "actual_vllm_internal_ms": "proxy_forward_to_first_chunk_duration", + "prepare_buffer_wait_ms": "proxy_prepare_buffer_wait_duration", +} +_LEGACY_ALIAS_NAMES = {"actual_compute_ms": "proxy_forward_to_first_chunk_legacy_compute_alias"} +_KV_ACK_NUMERIC = {"payload_bytes", "network_queue_ms", "network_transfer_ms", "network_total_ms"} +_PATH_CODES = { + "text_inject": (TraceStageOutcome.SUCCESS, "legacy_text_inject"), + "no_rag_or_empty_knowledge": (TraceStageOutcome.SKIPPED, "legacy_no_rag_or_empty_knowledge"), + "kv_inject": (TraceStageOutcome.SUCCESS, "legacy_kv_inject"), + "kv_inject_failed_fallback_text": (TraceStageOutcome.TEXT_FALLBACK, "legacy_kv_inject_failed_text_fallback"), + "no_kv_ready_fallback_text": (TraceStageOutcome.TEXT_FALLBACK, "legacy_no_kv_ready_text_fallback"), +} + + +def _hex_id(prefix: str, *parts: object) -> str: + digest = hashlib.sha256("\x1f".join(str(part) for part in parts).encode()).hexdigest()[:32] + return f"{prefix}_{digest}" + + +def _millis_timestamp(value: Any) -> datetime | None: + if isinstance(value, bool) or not isinstance(value, (int, float)) or value < 0: + return None + try: + return datetime.fromtimestamp(value / 1000, tz=timezone.utc) + except (OverflowError, OSError, ValueError): + return None + + +def _measurement(key: str, value: Any, provenance: TraceProvenance) -> TraceMeasurement | None: + if key in _TIMESTAMP_KEYS: + timestamp = _millis_timestamp(value) + if timestamp is None: + return None + kind = TraceValueKind.PREDICTED if key.startswith("pred_") else TraceValueKind.OBSERVED + if key in {"first_token_ms", "decode_start_ms", "decode_end_ms", "kv_inject_reserved_start_ms", "kdn_link_wait_end_ms"}: + kind = TraceValueKind.LEGACY_PROJECTED + return TraceMeasurement(name=key.removesuffix("_ms") + "_timestamp", kind=kind, + provenance=provenance, timestamp=timestamp, legacy_name=key) + if isinstance(value, bool) or not isinstance(value, (int, float)) or not __import__("math").isfinite(value): + return None + if value < 0 and key != "predict_error_ms": + return None + if key in _CLEAR_ACTUAL_NAMES: + return TraceMeasurement(name=_CLEAR_ACTUAL_NAMES[key], kind=TraceValueKind.ACTUAL, + provenance=provenance, duration_ns=int(value * 1_000_000), legacy_name=key) + if key in _LEGACY_ALIAS_NAMES: + return TraceMeasurement(name=_LEGACY_ALIAS_NAMES[key], kind=TraceValueKind.LEGACY_PROJECTED, + provenance=provenance, duration_ns=max(0, int(value * 1_000_000)), legacy_name=key) + kind = TraceValueKind.PREDICTED if key.startswith("predict_") else TraceValueKind.LEGACY_PROJECTED + if key in _OVERWRITTEN_PREDICTIONS: + kind = TraceValueKind.LEGACY_PROJECTED + if key == "ttft_observable": + return TraceMeasurement(name=key, kind=TraceValueKind.OBSERVED, provenance=provenance, + boolean=bool(value), legacy_name=key) + if key == "predict_error_ms": + return TraceMeasurement(name="prediction_error_ms", kind=TraceValueKind.INFERRED, + provenance=provenance, safe_scalar=value, legacy_name=key) + if key.endswith("_ms"): + return TraceMeasurement(name=key.removesuffix("_ms") + "_duration", kind=kind, + provenance=provenance, duration_ns=int(value * 1_000_000), legacy_name=key) + return None + + +def project_legacy_proxy_trace( + *, + request_id: str, + correlation_id: str, + legacy_request_id: int | None, + trace: Mapping[str, Any], + kv_ack: Mapping[str, Any] | None = None, + exported_at: datetime, + runtime_profile: str, +) -> RequestTrace: + """Project allow-listed scalar fields; raw inputs never enter the result.""" + trace_id = _hex_id("trace", request_id, correlation_id, legacy_request_id) + context = TraceContext( + trace_id=trace_id, request_id=request_id, correlation_id=correlation_id, + legacy_request_id=legacy_request_id, sampled=True, created_at=exported_at, + ) + provenance = TraceProvenance( + source_component=TraceComponent.LEGACY_ADAPTER, runtime_profile=runtime_profile, + captured_at=exported_at, legacy=True, + ) + stages: list[TraceStage] = [] + sequence = 0 + for stage_name, keys in _STAGE_KEYS: + measurements = tuple( + measurement for key in keys + if key in trace + for measurement in [_measurement(key, trace[key], provenance)] + if measurement is not None + ) + if not measurements: + continue + stages.append(TraceStage( + stage_id=_hex_id("stage", trace_id, sequence, stage_name.value), sequence=sequence, + name=stage_name, state=TraceStageState.COMPLETED, outcome=TraceStageOutcome.SUCCESS, + provenance=provenance, measurements=measurements, started_at=exported_at, + ended_at=exported_at, duration_ns=0, + )) + sequence += 1 + + if kv_ack: + measurements = [] + for key in sorted(_KV_ACK_NUMERIC): + value = kv_ack.get(key) + if isinstance(value, (int, float)) and not isinstance(value, bool) and value >= 0: + field = "bytes" if key == "payload_bytes" else "duration_ns" + converted = int(value) if field == "bytes" else int(value * 1_000_000) + measurements.append(TraceMeasurement( + name=f"legacy_kv_ack_{key}", kind=TraceValueKind.LEGACY_PROJECTED, + provenance=provenance, legacy_name=key, **{field: converted}, + )) + if measurements: + stages.append(TraceStage( + stage_id=_hex_id("stage", trace_id, sequence, "kv_ack"), sequence=sequence, + name=TraceStageName.INSTANCE_LMCACHE_LOAD, state=TraceStageState.COMPLETED, + outcome=TraceStageOutcome.SUCCESS, provenance=provenance, + measurements=tuple(measurements), started_at=exported_at, ended_at=exported_at, + duration_ns=0, outcome_code="legacy_kv_ack", + )) + sequence += 1 + + for path_key in ("text_actual_path", "kvcache_actual_path"): + path_value = trace.get(path_key) + classification = _PATH_CODES.get(path_value) if isinstance(path_value, str) else None + if classification is None: + continue + outcome, code = classification + stage_name = TraceStageName.FALLBACK if outcome is TraceStageOutcome.TEXT_FALLBACK else TraceStageName.PROXY_PREPARE_QUEUE + stages.append(TraceStage( + stage_id=_hex_id("stage", trace_id, sequence, path_key), sequence=sequence, + name=stage_name, state=TraceStageState.COMPLETED, outcome=outcome, + provenance=provenance, started_at=exported_at, ended_at=exported_at, + duration_ns=0, outcome_code=code, + )) + sequence += 1 + + components = (TraceComponent.LEGACY_ADAPTER,) if stages else () + return RequestTrace(context=context, stages=tuple(stages), exported_at=exported_at, + complete=True, source_components=components) + + +__all__ = ["project_legacy_proxy_trace"] diff --git a/cacheroute_observability/models.py b/cacheroute_observability/models.py new file mode 100644 index 0000000..36e8237 --- /dev/null +++ b/cacheroute_observability/models.py @@ -0,0 +1,415 @@ +"""Frozen, dependency-light CacheRoute observability contracts.""" +from __future__ import annotations + +import math +import re +from datetime import datetime, timedelta +from typing import Any, ClassVar + +from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator + +from .enums import ( + OperationWaiterState, TraceComponent, TraceStageName, TraceStageOutcome, + TraceStageState, TraceValueKind, +) + +_TRACE_RE = re.compile(r"^trace_[0-9a-f]{32}$") +_STAGE_RE = re.compile(r"^stage_[0-9a-f]{32}$") +_SAFE_CODE_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.:-]{0,127}$") +_FORBIDDEN_KEYS = { + "authorization", "cookie", "credentials", "device_address", "device_pointer", + "generated_content", "http_headers", "kv_bytes", "password", "physical_kv", + "physical_path", "prompt", "raw_exception", "redis_key", "request_body", + "secret", "tensor", "token_value", +} + + +def _utc(value: datetime, name: str) -> datetime: + if value.tzinfo is None or value.utcoffset() is None or value.utcoffset() != timedelta(0): + raise ValueError(f"{name} must be timezone-aware UTC") + return value + + +def _nonempty(value: str, name: str, maximum: int = 256) -> str: + if not value.strip() or len(value) > maximum: + raise ValueError(f"{name} must be non-empty and at most {maximum} characters") + return value + + +def _safe_code(value: str, name: str) -> str: + _nonempty(value, name, 128) + if not _SAFE_CODE_RE.fullmatch(value): + raise ValueError(f"{name} must be a safe logical code") + return value + + +def _scan_forbidden(value: Any) -> None: + if isinstance(value, dict): + for key, nested in value.items(): + normalized = str(key).strip().lower().replace("-", "_") + if normalized in _FORBIDDEN_KEYS: + raise ValueError(f"forbidden observability field: {key}") + _scan_forbidden(nested) + elif isinstance(value, (list, tuple, set)): + for nested in value: + _scan_forbidden(nested) + + +class ContractModel(BaseModel): + model_config = ConfigDict(frozen=True, extra="forbid") + + @model_validator(mode="before") + @classmethod + def reject_forbidden_fields(cls, value: Any) -> Any: + _scan_forbidden(value) + return value + + def model_copy(self, *, update=None, deep: bool = False): + """Copy through validation so frozen-contract invariants cannot be bypassed.""" + values = self.model_dump(mode="python") + values.update(update or {}) + return type(self).model_validate(values) + + +class TraceContext(ContractModel): + schema_version: str = "cacheroute.trace-context.v1" + trace_id: str + request_id: str + correlation_id: str + legacy_request_id: int | None = None + sampled: bool = True + created_at: datetime + expires_at: datetime | None = None + + @field_validator("schema_version") + @classmethod + def exact_version(cls, value: str) -> str: + if value != "cacheroute.trace-context.v1": + raise ValueError("unsupported trace context schema version") + return value + + @field_validator("trace_id") + @classmethod + def trace_id_format(cls, value: str) -> str: + if not _TRACE_RE.fullmatch(value): + raise ValueError("trace_id must use trace_<32 lowercase hex>") + return value + + @field_validator("request_id", "correlation_id") + @classmethod + def bounded_ids(cls, value: str, info) -> str: + return _nonempty(value, info.field_name, 256) + + @field_validator("created_at", "expires_at") + @classmethod + def utc_datetimes(cls, value: datetime | None, info): + return None if value is None else _utc(value, info.field_name) + + @model_validator(mode="after") + def expiry_order(self): + if self.expires_at is not None and self.expires_at <= self.created_at: + raise ValueError("expires_at must be later than created_at") + return self + + +class TraceProvenance(ContractModel): + source_component: TraceComponent + runtime_profile: str + captured_at: datetime + source_endpoint: str | None = None + endpoint_id: str | None = None + endpoint_generation: int | None = None + compatibility_profile_id: str | None = None + gateway_profile: str | None = None + adapter: str | None = None + tier: str | None = None + source_version: str | None = None + fresh_until: datetime | None = None + legacy: bool = False + + @field_validator("runtime_profile") + @classmethod + def profile(cls, value: str) -> str: + return _safe_code(value, "runtime_profile") + + @field_validator( + "source_endpoint", "endpoint_id", "compatibility_profile_id", "gateway_profile", + "adapter", "tier", "source_version", + ) + @classmethod + def safe_labels(cls, value: str | None, info): + return None if value is None else _safe_code(value, info.field_name) + + @field_validator("captured_at", "fresh_until") + @classmethod + def utc_datetimes(cls, value: datetime | None, info): + return None if value is None else _utc(value, info.field_name) + + @model_validator(mode="after") + def endpoint_and_freshness(self): + if (self.endpoint_id is None) != (self.endpoint_generation is None): + raise ValueError("endpoint_id and endpoint_generation must appear together") + if self.endpoint_generation is not None: + minimum = 0 if self.legacy else 1 + if self.endpoint_generation < minimum: + raise ValueError("endpoint_generation is invalid for provenance") + if not self.legacy and self.endpoint_generation == 0: + raise ValueError("endpoint_generation=0 is Legacy-only") + if self.fresh_until is not None and self.fresh_until < self.captured_at: + raise ValueError("fresh_until cannot be earlier than captured_at") + return self + + +Scalar = str | int | float | bool + + +class TraceMeasurement(ContractModel): + _VALUE_FIELDS: ClassVar[tuple[str, ...]] = ( + "duration_ns", "count", "bytes", "tokens", "ratio", "boolean", "timestamp", "safe_scalar", + ) + name: str + kind: TraceValueKind + provenance: TraceProvenance + duration_ns: int | None = Field(default=None, ge=0) + count: int | None = Field(default=None, ge=0) + bytes: int | None = Field(default=None, ge=0) + tokens: int | None = Field(default=None, ge=0) + ratio: float | None = None + boolean: bool | None = None + timestamp: datetime | None = None + safe_scalar: Scalar | None = None + observed_at: datetime | None = None + expires_at: datetime | None = None + uncertainty: float | None = Field(default=None, ge=0) + sample_count: int | None = Field(default=None, ge=0) + legacy_name: str | None = None + + @field_validator("name", "legacy_name") + @classmethod + def safe_names(cls, value: str | None, info): + return None if value is None else _safe_code(value, info.field_name) + + @field_validator("timestamp", "observed_at", "expires_at") + @classmethod + def utc_datetimes(cls, value: datetime | None, info): + return None if value is None else _utc(value, info.field_name) + + @field_validator("ratio") + @classmethod + def valid_ratio(cls, value: float | None): + if value is not None and (not math.isfinite(value) or not 0 <= value <= 1): + raise ValueError("ratio must be finite and between 0 and 1") + return value + + @field_validator("safe_scalar", mode="before") + @classmethod + def scalar_only(cls, value: Any): + if value is not None and (isinstance(value, (dict, list, set, tuple)) or not isinstance(value, (str, int, float, bool))): + raise ValueError("safe_scalar must be a scalar") + if isinstance(value, float) and not math.isfinite(value): + raise ValueError("safe_scalar float must be finite") + if isinstance(value, str): + _nonempty(value, "safe_scalar", 256) + return value + + @model_validator(mode="after") + def exactly_one_value(self): + if sum(getattr(self, field) is not None for field in self._VALUE_FIELDS) != 1: + raise ValueError("exactly one typed measurement value is required") + if self.expires_at is not None: + reference = self.observed_at or self.timestamp or self.provenance.captured_at + if self.expires_at <= reference: + raise ValueError("measurement expires_at must be later than its observation") + return self + + +class TraceStage(ContractModel): + stage_id: str + sequence: int = Field(ge=0) + name: TraceStageName + state: TraceStageState + outcome: TraceStageOutcome | None = None + provenance: TraceProvenance + measurements: tuple[TraceMeasurement, ...] = () + started_at: datetime | None = None + ended_at: datetime | None = None + duration_ns: int | None = Field(default=None, ge=0) + parent_stage_id: str | None = None + operation_id: str | None = None + artifact_id: str | None = None + outcome_code: str | None = None + error_code: str | None = None + error_message: str | None = None + retryable: bool = False + fallback_eligible: bool = False + fallback_stage_id: str | None = None + partial_reason: str | None = None + + @field_validator("stage_id", "parent_stage_id", "fallback_stage_id") + @classmethod + def stage_ids(cls, value: str | None): + if value is not None and not _STAGE_RE.fullmatch(value): + raise ValueError("stage IDs must use stage_<32 lowercase hex>") + return value + + @field_validator("operation_id", "artifact_id", "outcome_code", "error_code", "partial_reason") + @classmethod + def safe_codes(cls, value: str | None, info): + return None if value is None else _safe_code(value, info.field_name) + + @field_validator("error_message") + @classmethod + def sanitized_message(cls, value: str | None): + if value is None: + return None + _nonempty(value, "error_message", 256) + if any(char in value for char in ("\n", "\r", "\x00")): + raise ValueError("error_message must be a sanitized single line") + return value + + @field_validator("started_at", "ended_at") + @classmethod + def utc_datetimes(cls, value: datetime | None, info): + return None if value is None else _utc(value, info.field_name) + + @model_validator(mode="after") + def lifecycle(self): + if self.state is TraceStageState.COMPLETED and self.outcome is None: + raise ValueError("completed stages require an outcome") + if self.state is not TraceStageState.COMPLETED and self.outcome is not None: + raise ValueError("pending/running stages cannot claim a terminal outcome") + if self.ended_at is not None and self.started_at is None: + raise ValueError("ended_at requires started_at") + if self.started_at is not None and self.ended_at is not None and self.ended_at < self.started_at: + raise ValueError("ended_at cannot precede started_at") + if self.state is TraceStageState.COMPLETED and (self.ended_at is None or self.duration_ns is None): + raise ValueError("completed stages require ended_at and duration_ns") + return self + + +class RequestTrace(ContractModel): + schema_version: str = "cacheroute.trace.v1" + context: TraceContext + stages: tuple[TraceStage, ...] = () + exported_at: datetime + complete: bool = False + source_components: tuple[TraceComponent, ...] = () + error_code: str | None = None + + @field_validator("schema_version") + @classmethod + def exact_version(cls, value: str) -> str: + if value != "cacheroute.trace.v1": + raise ValueError("unsupported request trace schema version") + return value + + @field_validator("exported_at") + @classmethod + def exported_utc(cls, value: datetime): + return _utc(value, "exported_at") + + @field_validator("error_code") + @classmethod + def safe_error(cls, value: str | None): + return None if value is None else _safe_code(value, "error_code") + + @model_validator(mode="after") + def sequence_order(self): + sequences = [stage.sequence for stage in self.stages] + if sequences != sorted(sequences) or len(sequences) != len(set(sequences)): + raise ValueError("stages must have unique ascending sequence values") + expected = tuple(dict.fromkeys(stage.provenance.source_component for stage in self.stages)) + if self.source_components and self.source_components != expected: + raise ValueError("source_components must follow first stage occurrence") + return self + + +class CacheOperationTrace(ContractModel): + schema_version: str = "cacheroute.cache-operation-trace.v1" + operation_id: str + trace_id: str + operation: str + state: str + stages: tuple[TraceStage, ...] + provenance: TraceProvenance + artifact_id: str | None = None + endpoint_id: str | None = None + endpoint_generation: int | None = None + created_at: datetime + updated_at: datetime + completed_at: datetime | None = None + error_code: str | None = None + fallback_code: str | None = None + + @field_validator("operation_id", "operation", "state", "artifact_id", "endpoint_id", "error_code", "fallback_code") + @classmethod + def safe_values(cls, value: str | None, info): + return None if value is None else _safe_code(value, info.field_name) + + @field_validator("trace_id") + @classmethod + def trace_id_format(cls, value: str): + if not _TRACE_RE.fullmatch(value): + raise ValueError("trace_id must use trace_<32 lowercase hex>") + return value + + @field_validator("created_at", "updated_at", "completed_at") + @classmethod + def utc_datetimes(cls, value: datetime | None, info): + return None if value is None else _utc(value, info.field_name) + + @model_validator(mode="after") + def consistency(self): + if self.updated_at < self.created_at or (self.completed_at is not None and self.completed_at < self.created_at): + raise ValueError("operation timestamps are out of order") + if (self.endpoint_id is None) != (self.endpoint_generation is None): + raise ValueError("endpoint_id and endpoint_generation must appear together") + if self.endpoint_generation is not None: + minimum = 0 if self.provenance.legacy else 1 + if self.endpoint_generation < minimum or (not self.provenance.legacy and self.endpoint_generation == 0): + raise ValueError("endpoint_generation is invalid") + sequences = [stage.sequence for stage in self.stages] + if sequences != sorted(sequences) or len(sequences) != len(set(sequences)): + raise ValueError("operation stages must have unique ascending sequence values") + return self + + +class OperationWaiterLink(ContractModel): + operation_id: str + trace_id: str + request_id: str + correlation_id: str + attach_sequence: int = Field(ge=0) + attached_at: datetime + state: OperationWaiterState + detached_at: datetime | None = None + reason: str | None = None + + @field_validator("operation_id", "request_id", "correlation_id", "reason") + @classmethod + def safe_values(cls, value: str | None, info): + return None if value is None else _safe_code(value, info.field_name) if info.field_name in {"operation_id", "reason"} else _nonempty(value, info.field_name, 256) + + @field_validator("trace_id") + @classmethod + def trace_id_format(cls, value: str): + if not _TRACE_RE.fullmatch(value): + raise ValueError("trace_id must use trace_<32 lowercase hex>") + return value + + @field_validator("attached_at", "detached_at") + @classmethod + def utc_datetimes(cls, value: datetime | None, info): + return None if value is None else _utc(value, info.field_name) + + @model_validator(mode="after") + def detach_order(self): + if self.detached_at is not None and self.detached_at < self.attached_at: + raise ValueError("detached_at cannot precede attached_at") + return self + + +__all__ = [ + "CacheOperationTrace", "OperationWaiterLink", "RequestTrace", "TraceContext", + "TraceMeasurement", "TraceProvenance", "TraceStage", +] diff --git a/docs/research/issue-141-unified-observability.md b/doc/research/issue-141-unified-observability.md similarity index 92% rename from docs/research/issue-141-unified-observability.md rename to doc/research/issue-141-unified-observability.md index 6f56696..195f48b 100644 --- a/docs/research/issue-141-unified-observability.md +++ b/doc/research/issue-141-unified-observability.md @@ -2,15 +2,16 @@ Status: design proposal only
Baseline audited: `ad087d1` (current `main` lineage, including merged PR #154)
+Lifecycle: temporary design record for Issue #141. Remove this file after the Issue is completed and its stable content has been moved into maintained package and component documentation. + Non-goals: production instrumentation, Gateway I/O, routing/queue changes, or a tracing backend. ## 1. Executive decision record -1. Put the dependency-light contracts in **`core/observability/`**. Scheduler, - Proxy, KDN, Gateway, and Instance already depend on `core`; placing them under - `kdn_server` would invert dependencies for Scheduler/Proxy/Instance, while a - new top-level distribution would add packaging work without providing more - isolation than a carefully constrained `core` package. +1. Put the dependency-light contracts in **`cacheroute_observability/`**. Scheduler, + Proxy, KDN, Gateway, and Instance can import this standalone package without + loading `core` or another component; placing it under `kdn_server` would invert + dependencies for Scheduler/Proxy/Instance. 2. Make an exported trace an immutable, versioned snapshot. Build it by appending immutable stages/measurements to a process-local collector; never mutate a measurement from `predicted` into `actual`. @@ -227,18 +228,14 @@ LMCache API. | Candidate | Benefits | Problems | Decision | |---|---|---|---| | `kdn_server/observability/` | Close to #139/#140 types | Forces Proxy, Scheduler, Instance, and `core` to depend on a server package; likely dependency inversion and future import cycles. | Reject. | -| `core/observability/` | Existing common inward dependency; simple imports; same repository/version | Must enforce that it does not import request forwarding, config, FastAPI, or components. | **Recommend.** | -| standalone top-level package/distribution | Strong conceptual isolation | Packaging/versioning overhead, duplicate shared primitives, and uncertain installation in current entry points. | Revisit only if contracts become independently released. | - -`core/observability/` may import Python stdlib, Pydantic, and selected immutable -types from `kdn_server.domain`/#139 and outcome vocabulary from -`kdn_server.contracts.errors`/#140. To keep dependency direction cleaner, the -first PR should instead define trace-only enums in `core` and type resource IDs -as validated opaque strings; adapters attach existing domain/contract JSON -without importing Gateway implementations. A follow-up can move genuinely -shared primitives (UTC validation and safe outcome vocabulary) inward if -maintainers approve. It must never import FastAPI, Redis, vLLM, LMCache, GPU -libraries, `httpx`, network clients, component modules, or `core.fwd/config`. +| `cacheroute_observability/` | Standalone dependency-light boundary; simple imports; same repository/version | Must enforce that it imports only the standard library and Pydantic. | **Implemented.** | +| separately released distribution | Strong release isolation | Independent packaging/versioning overhead is unnecessary in Phase 1. | Revisit only if contracts need an independent release cycle. | + +`cacheroute_observability/` imports only Python stdlib and Pydantic. It defines +trace-only enums and uses validated opaque resource IDs instead of importing or +duplicating #139 domain payloads, #140 contracts, or Gateway implementations. It +must never import FastAPI, Redis, vLLM, LMCache, GPU libraries, `httpx`, network +clients, application components, configuration, or forwarding modules. ## 7. Stable v1 schemas @@ -494,34 +491,35 @@ Gateway orchestration layer, separate from `CacheOperationTask` and scheduling. ## 12. Rollout and exact file plan -### Phase 1 — contracts only +### Phase 1 — dependency-light vertical slice Add: -* `core/observability/__init__.py` — public, dependency-light exports. -* `core/observability/enums.py` — stage, kind, outcome, component vocabularies and +* `cacheroute_observability/__init__.py` — public, dependency-light exports. +* `cacheroute_observability/enums.py` — stage, kind, outcome, component vocabularies and stable rank table. -* `core/observability/models.py` — frozen schemas and validators. -* `core/observability/clock.py` — clock protocol, production clock, test seam. -* `tests/core/observability/test_models.py` -* `tests/core/observability/test_clock.py` -* `tests/core/observability/test_security.py` -* `docs/observability/v1-contract.md` — stable public contract (not root README). +* `cacheroute_observability/models.py` — frozen schemas and validators. +* `cacheroute_observability/clock.py` — clock protocol, production clock, test seam. +* `cacheroute_observability/collector.py` — process-local append-only collection. +* `cacheroute_observability/legacy_proxy_trace.py` — pure Legacy projection. +* `test/observability/` — CPU-only contract, collector, isolation, adapter, and demo tests. +* `scripts/demo_observability_v1.py` — deterministic executable demonstration. +* `cacheroute_observability/README.md` — maintained package documentation. Do not modify production component files. This is the independently reviewable -first PR: enums, models, deterministic CPU-only tests, and contract documentation -only. It must not include a collector, adapter, propagation, or I/O. +first PR: contracts, collector, Legacy adapter, deterministic CPU-only tests, and +package documentation. It includes no production instrumentation, propagation, +Gateway I/O, or external tracing backend. -### Phase 2 — compatibility adapter +### Phase 2 — production compatibility integration -Add `core/observability/legacy_proxy_trace.py` and its tests. Modify only -`proxy/proxy.py` at `build_cacheroute_meta` to optionally append +Modify `proxy/proxy.py` at `build_cacheroute_meta` to optionally append `request_trace_v1`, and possibly correct `ProxyTask.trace`'s annotation to `Dict[str, Any]` without changing values. Retain the dictionary and meta shape. ### Phase 3 — component instrumentation/propagation -Add `core/observability/collector.py` and internal-envelope helpers. Modify +Add internal-envelope helpers and modify `scheduler/scheduler.py`, `core/fwd.py`, `proxy/proxy.py`, `proxy/queue/task.py`, `proxy/queue/manager.py`, `proxy/queue/instance_queues.py`, `proxy/queue/knowledge.py`, KDN cache-service facade call sites, @@ -553,18 +551,18 @@ as endpoint aggregate/isolated inference. | Disabled/sampled | disabled allocates no stages; unsampled propagates only context; neither changes callback results/order | None | | Security | reject secrets, auth headers, raw tokens/prompts, Redis keys, file paths, tensors, KV bytes/pointers, extras | None | | Mock Gateway timeline | request, async operation, observation, shared waiter, Instance load, completion combine deterministically | In-memory Mock Gateway only | -| Import isolation | importing `core.observability` does not load FastAPI/httpx/redis/vllm/lmcache/torch/CUDA modules | None | +| Import isolation | importing `cacheroute_observability` does not load FastAPI/httpx/redis/vllm/lmcache/torch/CUDA modules | None | | Environment independence | full suite runs with network disabled and without Redis, LMCache, vLLM, GPU, or tracing backend | None | Validation for the first PR should run from the repository root with -`pytest -q tests/core/observability` and an import-isolation subprocess test. +`python3 -m pytest -q test/observability` and an import-isolation subprocess test. Failure is any nondeterministic order, mutable exported object, negative duration acceptance, kind collapse, forbidden physical/secret field acceptance, external service access, or behavior difference when tracing is disabled. ## 14. Maintainer decisions and unresolved questions -1. Is `core/observability/` accepted, or should #139/#140 common primitives first +1. Is `cacheroute_observability/` accepted, or should #139/#140 common primitives first move inward to avoid `core -> kdn_server.domain/contracts` imports? 2. Is the canonical request ID generated at Client when supplied or always at Scheduler? What retry semantics should preserve it? diff --git a/scripts/demo_observability_v1.py b/scripts/demo_observability_v1.py new file mode 100644 index 0000000..087de6b --- /dev/null +++ b/scripts/demo_observability_v1.py @@ -0,0 +1,100 @@ +#!/usr/bin/env python3 +"""Deterministic, CPU-only demonstration of CacheRoute observability v1.""" +from __future__ import annotations + +import json +import sys +from datetime import datetime, timedelta, timezone +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from cacheroute_observability import ( # noqa: E402 + CacheOperationTrace, ManualTraceClock, OperationWaiterLink, OperationWaiterState, + TraceCollector, TraceComponent, TraceContext, TraceMeasurement, TraceProvenance, + TraceStageName, TraceStageOutcome, TraceValueKind, project_legacy_proxy_trace, +) + + +def main() -> int: + clock = ManualTraceClock(datetime(2026, 1, 2, 3, 4, 5, tzinfo=timezone.utc), 1_000) + context = TraceContext( + trace_id="trace_00000000000000000000000000000001", request_id="req-demo-1", + correlation_id="corr-demo", legacy_request_id=42, sampled=True, + created_at=clock.utc_now(), expires_at=clock.utc_now() + timedelta(minutes=5), + ) + provenance = TraceProvenance(source_component=TraceComponent.TEST, + runtime_profile="test.mock", captured_at=clock.utc_now()) + stage_ids = iter(f"stage_{number:032x}" for number in range(1, 20)) + collector = TraceCollector(context, clock=clock, id_factory=lambda: next(stage_ids)) + + ordered = [ + (TraceStageName.RUNTIME_PROFILE_RESOLUTION, TraceStageOutcome.SUCCESS), + (TraceStageName.GATEWAY_REQUEST, TraceStageOutcome.UNSUPPORTED), + (TraceStageName.FALLBACK, TraceStageOutcome.TEXT_FALLBACK), + (TraceStageName.GATEWAY_REQUEST, TraceStageOutcome.SUCCESS), + (TraceStageName.PROXY_PREPARE_QUEUE, TraceStageOutcome.SUCCESS), + (TraceStageName.PROXY_READY_QUEUE, TraceStageOutcome.SUCCESS), + (TraceStageName.FIRST_TOKEN, TraceStageOutcome.SUCCESS), + (TraceStageName.DECODE, TraceStageOutcome.SUCCESS), + (TraceStageName.COMPLETION, TraceStageOutcome.SUCCESS), + ] + for index, (name, outcome) in enumerate(ordered): + stage_id = collector.start_stage(name, provenance) + if name is TraceStageName.PROXY_READY_QUEUE: + collector.append_measurement(stage_id, TraceMeasurement( + name="queue_wait", kind=TraceValueKind.PREDICTED, provenance=provenance, + duration_ns=2_000_000, + )) + collector.append_measurement(stage_id, TraceMeasurement( + name="queue_wait", kind=TraceValueKind.ACTUAL, provenance=provenance, + duration_ns=3_000_000, + )) + clock.advance_ns(100 + index) + clock.advance_time(timedelta(microseconds=1)) + collector.finish_stage(stage_id, outcome) + + request_trace = collector.export(complete=True) + operation = CacheOperationTrace( + operation_id="cacheop-demo", trace_id=context.trace_id, operation="prefetch", + state="succeeded", stages=request_trace.stages[1:4], provenance=provenance, + artifact_id="artifact-demo", endpoint_id="endpoint-demo", endpoint_generation=1, + created_at=context.created_at, updated_at=clock.utc_now(), completed_at=clock.utc_now(), + ) + waiters = tuple(OperationWaiterLink( + operation_id=operation.operation_id, trace_id=f"trace_{number:032x}", + request_id=f"req-waiter-{number}", correlation_id="corr-demo", + attach_sequence=number - 2, attached_at=clock.utc_now(), + state=OperationWaiterState.COMPLETED, + ) for number in (2, 3)) + legacy = project_legacy_proxy_trace( + request_id="req-legacy", correlation_id="corr-legacy", legacy_request_id=7, + trace={ + "predict_prepare_ms": 12, "actual_vllm_internal_ms": 8, + "first_token_ms": 1760000000000, "kvcache_actual_path": "kv_inject_failed_fallback_text", + "error": "raw exception must be omitted", "unknown_secretish_result": "omitted", + }, + kv_ack={"payload_bytes": 1024, "raw": "omitted"}, exported_at=clock.utc_now(), + runtime_profile="legacy", + ) + + output = { + "request_trace": request_trace.model_dump(mode="json"), + "stage_sequence": [stage.name.value for stage in request_trace.stages], + "value_kinds": [ + measurement.kind.value for stage in request_trace.stages + for measurement in stage.measurements + ], + "cache_operation": operation.model_dump(mode="json"), + "shared_operation_waiter_ids": [waiter.request_id for waiter in waiters], + "legacy_projection": legacy.model_dump(mode="json"), + } + print(json.dumps(output, indent=2, sort_keys=True)) + print("observability v1 demo: passed") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/test/observability/conftest.py b/test/observability/conftest.py new file mode 100644 index 0000000..836cf50 --- /dev/null +++ b/test/observability/conftest.py @@ -0,0 +1,22 @@ +from datetime import datetime, timezone + +import pytest + +from cacheroute_observability import TraceComponent, TraceContext, TraceProvenance + + +@pytest.fixture +def now(): + return datetime(2026, 1, 1, tzinfo=timezone.utc) + + +@pytest.fixture +def context(now): + return TraceContext(trace_id="trace_" + "1" * 32, request_id="req-1", + correlation_id="corr-1", sampled=True, created_at=now) + + +@pytest.fixture +def provenance(now): + return TraceProvenance(source_component=TraceComponent.TEST, + runtime_profile="test.mock", captured_at=now) diff --git a/test/observability/test_clock.py b/test/observability/test_clock.py new file mode 100644 index 0000000..bd5937f --- /dev/null +++ b/test/observability/test_clock.py @@ -0,0 +1,24 @@ +from datetime import timedelta + +import pytest + +from cacheroute_observability import ManualTraceClock, SystemTraceClock + + +def test_manual_clock_advances_independently(now): + clock = ManualTraceClock(now, 10) + clock.advance_ns(0) + assert clock.monotonic_ns() == 10 + clock.advance_ns(5) + clock.advance_time(timedelta(seconds=2)) + assert clock.monotonic_ns() == 15 + assert clock.utc_now() == now + timedelta(seconds=2) + + +def test_clock_rejects_invalid_values(now): + with pytest.raises(ValueError): + ManualTraceClock(now.replace(tzinfo=None)) + clock = ManualTraceClock(now) + with pytest.raises(ValueError): + clock.advance_ns(-1) + assert SystemTraceClock().monotonic_ns() >= 0 diff --git a/test/observability/test_collector.py b/test/observability/test_collector.py new file mode 100644 index 0000000..1ccd218 --- /dev/null +++ b/test/observability/test_collector.py @@ -0,0 +1,72 @@ +from datetime import timedelta + +import pytest +from pydantic import ValidationError + +from cacheroute_observability import ( + ManualTraceClock, TraceCollector, TraceMeasurement, TraceStageName, +) + + +def _ids(): + return iter(f"stage_{i:032x}" for i in range(1, 20)) + + +def test_order_repetition_monotonic_and_snapshot(context, provenance, now): + clock = ManualTraceClock(now, 100) + ids = _ids() + collector = TraceCollector(context, clock=clock, id_factory=lambda: next(ids)) + names = ["gateway_request", "fallback", "gateway_request", "completion"] + first_snapshot = None + for index, name in enumerate(names): + stage_id = collector.start_stage(name, provenance) + if index == 0: + collector.append_measurement(stage_id, TraceMeasurement( + name="latency", kind="predicted", provenance=provenance, duration_ns=4)) + collector.append_measurement(stage_id, TraceMeasurement( + name="latency", kind="actual", provenance=provenance, duration_ns=5)) + clock.advance_ns(index) + clock.advance_time(timedelta(seconds=20 if index == 0 else 0)) + collector.finish_stage(stage_id, "text_fallback" if name == "fallback" else "success") + if index == 0: + first_snapshot = collector.export() + final = collector.export(complete=True) + assert [stage.name.value for stage in final.stages] == names + assert [stage.duration_ns for stage in final.stages] == [0, 1, 2, 3] + assert len(first_snapshot.stages) == 1 + assert [m.kind.value for m in final.stages[0].measurements] == ["predicted", "actual"] + with pytest.raises(ValidationError): + final.stages += () + + +def test_finish_errors(context, provenance, now): + clock = ManualTraceClock(now, 10) + ids = _ids() + collector = TraceCollector(context, clock=clock, id_factory=lambda: next(ids)) + with pytest.raises(KeyError): + collector.finish_stage("stage_" + "9" * 32, "success") + stage_id = collector.start_stage(TraceStageName.COMPLETION, provenance) + collector.finish_stage(stage_id, "success") + with pytest.raises(ValueError): + collector.finish_stage(stage_id, "success") + + +def test_disabled_and_unsampled(context, provenance, now): + disabled = TraceCollector(context, enabled=False) + assert disabled.start_stage("completion", provenance) is None + assert disabled.export().stages == () + unsampled_context = context.model_copy(update={"sampled": False}) + unsampled = TraceCollector(unsampled_context) + assert unsampled.start_stage("completion", provenance) is None + assert unsampled.export().context.sampled is False + assert unsampled.safely(lambda: 1 / 0, "unchanged") == "unchanged" + + +def test_negative_monotonic_duration_rejected(context, provenance, now): + clock = ManualTraceClock(now, 10) + ids = _ids() + collector = TraceCollector(context, clock=clock, id_factory=lambda: next(ids)) + stage_id = collector.start_stage("completion", provenance) + clock._monotonic_value = 9 + with pytest.raises(ValueError): + collector.finish_stage(stage_id, "success") diff --git a/test/observability/test_demo.py b/test/observability/test_demo.py new file mode 100644 index 0000000..82f534f --- /dev/null +++ b/test/observability/test_demo.py @@ -0,0 +1,23 @@ +import json +import subprocess +import sys +from pathlib import Path + + +def test_demo_succeeds_and_shows_required_data(): + root = Path(__file__).resolve().parents[2] + result = subprocess.run([sys.executable, str(root / "scripts/demo_observability_v1.py")], + cwd=root, check=True, text=True, capture_output=True) + lines = result.stdout.splitlines() + assert lines[-1] == "observability v1 demo: passed" + payload = json.loads("\n".join(lines[:-1])) + assert payload["request_trace"]["schema_version"] == "cacheroute.trace.v1" + assert payload["request_trace"]["context"]["schema_version"] == "cacheroute.trace-context.v1" + assert payload["stage_sequence"][:4] == [ + "runtime_profile_resolution", "gateway_request", "fallback", "gateway_request", + ] + assert payload["value_kinds"] == ["predicted", "actual"] + assert payload["shared_operation_waiter_ids"] == ["req-waiter-2", "req-waiter-3"] + legacy_json = json.dumps(payload["legacy_projection"]) + assert "raw exception" not in legacy_json + assert "unknown_secretish_result" not in legacy_json diff --git a/test/observability/test_enums.py b/test/observability/test_enums.py new file mode 100644 index 0000000..5858433 --- /dev/null +++ b/test/observability/test_enums.py @@ -0,0 +1,29 @@ +from cacheroute_observability import ( + TraceComponent, TraceStageName, TraceStageOutcome, TraceStageState, TraceValueKind, +) + + +def test_exact_enum_wire_values(): + assert [x.value for x in TraceComponent] == [ + "client", "scheduler", "proxy", "kdn", "gateway", "instance", "vllm", + "lmcache", "legacy_adapter", "test", + ] + assert [x.value for x in TraceValueKind] == [ + "predicted", "observed", "actual", "inferred", "legacy_projected", + ] + assert [x.value for x in TraceStageState] == ["pending", "running", "completed"] + assert [x.value for x in TraceStageOutcome] == [ + "success", "unsupported", "incompatible", "stale", "partial", "failed", + "cancelled", "skipped", "text_fallback", "idempotency_conflict", + ] + assert [x.value for x in TraceStageName] == [ + "runtime_profile_resolution", "knowledge_lookup", "semantic_resolution", + "artifact_compatibility", "capability_snapshot_discovery", "token_lookup", + "artifact_lookup", "cache_observation", "cache_operation_queue", + "cache_prefetch_execution", "cache_pin_execution", "cache_unpin_execution", + "cache_clear_execution", "cache_rebuild_execution", "gateway_request", + "gateway_async_operation", "tier_adapter_observation", "instance_lmcache_load", + "proxy_prepare_queue", "proxy_ready_queue", "vllm_prefill", "first_token", + "decode", "completion", "fallback", "legacy_scan", "legacy_dump", + "legacy_restore", "legacy_inject", + ] diff --git a/test/observability/test_import_isolation.py b/test/observability/test_import_isolation.py new file mode 100644 index 0000000..bf57a36 --- /dev/null +++ b/test/observability/test_import_isolation.py @@ -0,0 +1,18 @@ +import subprocess +import sys + + +def test_import_isolation_in_fresh_process(): + code = r''' +import sys +import cacheroute_observability +forbidden = {"fastapi", "httpx", "aiohttp", "redis", "numpy", "torch", "vllm", + "lmcache", "core", "scheduler", "proxy", "instance", "kdn_server"} +loaded = {name.split(".", 1)[0] for name in sys.modules} +unexpected = forbidden & loaded +assert not unexpected, sorted(unexpected) +print("observability import isolation: passed") +''' + result = subprocess.run([sys.executable, "-c", code], check=True, text=True, + capture_output=True) + assert result.stdout.strip() == "observability import isolation: passed" diff --git a/test/observability/test_legacy_proxy_trace.py b/test/observability/test_legacy_proxy_trace.py new file mode 100644 index 0000000..118dd0f --- /dev/null +++ b/test/observability/test_legacy_proxy_trace.py @@ -0,0 +1,48 @@ +from cacheroute_observability import TraceStageName, TraceValueKind, project_legacy_proxy_trace + + +def _measurements(result): + return {m.legacy_name: m for stage in result.stages for m in stage.measurements} + + +def test_legacy_classification_omission_and_sanitization(now): + result = project_legacy_proxy_trace( + request_id="req", correlation_id="corr", legacy_request_id=9, + trace={ + "predict_total_ms": 10, + "predict_prepare_ms": 11, + "predict_know_prepare_ms": 12, + "actual_vllm_internal_ms": 13, + "actual_compute_ms": 14, + "first_token_ms": 1760000000000, + "decode_start_ms": 1760000000001, + "decode_end_ms": 1760000000002, + "kvcache_actual_path": "kv_inject_failed_fallback_text", + "error": "Traceback secret raw exception", + "unknown_key": {"anything": "must not copy"}, + }, + kv_ack={"payload_bytes": 42, "password": "must not copy"}, + exported_at=now, runtime_profile="legacy", + ) + measurements = _measurements(result) + assert measurements["predict_total_ms"].kind is TraceValueKind.PREDICTED + assert measurements["predict_prepare_ms"].kind is TraceValueKind.LEGACY_PROJECTED + assert measurements["predict_know_prepare_ms"].kind is TraceValueKind.LEGACY_PROJECTED + assert measurements["actual_vllm_internal_ms"].name == "proxy_forward_to_first_chunk_duration" + assert measurements["actual_vllm_internal_ms"].kind is TraceValueKind.ACTUAL + assert measurements["actual_compute_ms"].kind is TraceValueKind.LEGACY_PROJECTED + assert measurements["first_token_ms"].kind is TraceValueKind.LEGACY_PROJECTED + assert measurements["decode_start_ms"].kind is TraceValueKind.LEGACY_PROJECTED + assert "unknown_key" not in result.model_dump_json() + assert "Traceback" not in result.model_dump_json() + assert "password" not in result.model_dump_json() + fallback = [stage for stage in result.stages if stage.name is TraceStageName.FALLBACK] + assert fallback[0].outcome.value == "text_fallback" + assert [stage.sequence for stage in result.stages] == list(range(len(result.stages))) + + +def test_projection_is_deterministic(now): + kwargs = dict(request_id="req", correlation_id="corr", legacy_request_id=None, + trace={"actual_total_ms": 2, "unknown": 3}, exported_at=now, + runtime_profile="legacy") + assert project_legacy_proxy_trace(**kwargs) == project_legacy_proxy_trace(**kwargs) diff --git a/test/observability/test_models.py b/test/observability/test_models.py new file mode 100644 index 0000000..98e37f3 --- /dev/null +++ b/test/observability/test_models.py @@ -0,0 +1,93 @@ +from datetime import timedelta + +import pytest +from pydantic import ValidationError + +from cacheroute_observability import ( + CacheOperationTrace, OperationWaiterLink, OperationWaiterState, RequestTrace, + TraceComponent, TraceContext, TraceMeasurement, TraceProvenance, TraceStage, + TraceStageOutcome, TraceValueKind, +) + + +def test_context_is_frozen_utc_and_round_trips(context): + with pytest.raises(ValidationError): + context.request_id = "changed" + encoded = context.model_dump_json() + assert '"created_at":"2026-01-01T00:00:00Z"' in encoded + assert TraceContext.model_validate_json(encoded) == context + + +def test_context_validation(now): + with pytest.raises(ValidationError): + TraceContext(trace_id="bad", request_id="r", correlation_id="c", created_at=now) + with pytest.raises(ValidationError): + TraceContext(trace_id="trace_" + "0" * 32, request_id="r", correlation_id="c", + created_at=now.replace(tzinfo=None)) + with pytest.raises(ValidationError): + TraceContext(trace_id="trace_" + "0" * 32, request_id="r", correlation_id="c", + created_at=now, expires_at=now) + + +def test_endpoint_generation_and_freshness(now): + current = TraceProvenance(source_component="gateway", runtime_profile="v1", + captured_at=now, endpoint_id="endpoint-1", endpoint_generation=1, + fresh_until=now + timedelta(seconds=1)) + assert current.endpoint_generation == 1 + legacy = TraceProvenance(source_component="legacy_adapter", runtime_profile="legacy", + captured_at=now, endpoint_id="endpoint-old", endpoint_generation=0, + legacy=True) + assert legacy.endpoint_generation == 0 + for values in ( + {"endpoint_id": "endpoint-1"}, + {"endpoint_id": "endpoint-1", "endpoint_generation": 0}, + {"fresh_until": now - timedelta(microseconds=1)}, + ): + with pytest.raises(ValidationError): + TraceProvenance(source_component="gateway", runtime_profile="v1", captured_at=now, **values) + + +def test_measurement_exact_value_and_ranges(provenance): + predicted = TraceMeasurement(name="queue_wait", kind="predicted", provenance=provenance, + duration_ns=0) + actual = TraceMeasurement(name="queue_wait", kind="actual", provenance=provenance, + duration_ns=4) + assert (predicted.kind, actual.kind) == (TraceValueKind.PREDICTED, TraceValueKind.ACTUAL) + for values in ({}, {"count": 1, "tokens": 2}, {"duration_ns": -1}, {"ratio": float("nan")}, {"ratio": 1.1}): + with pytest.raises(ValidationError): + TraceMeasurement(name="bad", kind="observed", provenance=provenance, **values) + + +def test_stage_state_outcome_and_time_rules(now, provenance): + base = dict(stage_id="stage_" + "2" * 32, sequence=0, name="completion", + provenance=provenance, started_at=now) + with pytest.raises(ValidationError): + TraceStage(**base, state="completed", ended_at=now, duration_ns=0) + with pytest.raises(ValidationError): + TraceStage(**base, state="running", outcome="success") + with pytest.raises(ValidationError): + TraceStage(**base, state="completed", outcome="success", + ended_at=now - timedelta(seconds=1), duration_ns=1) + complete = TraceStage(**base, state="completed", outcome="success", ended_at=now, duration_ns=0) + assert complete.outcome is TraceStageOutcome.SUCCESS + + +def test_request_and_shared_operation_models(now, context, provenance): + stage = TraceStage(stage_id="stage_" + "3" * 32, sequence=0, name="cache_operation_queue", + state="completed", outcome="success", provenance=provenance, + started_at=now, ended_at=now, duration_ns=0) + trace = RequestTrace(context=context, stages=(stage,), exported_at=now, complete=True, + source_components=(TraceComponent.TEST,)) + assert RequestTrace.model_validate_json(trace.model_dump_json()) == trace + operation = CacheOperationTrace( + operation_id="cacheop-1", trace_id=context.trace_id, operation="prefetch", + state="succeeded", stages=(stage,), provenance=provenance, artifact_id="artifact-1", + endpoint_id="endpoint-1", endpoint_generation=1, created_at=now, updated_at=now, + completed_at=now, + ) + waiters = tuple(OperationWaiterLink( + operation_id=operation.operation_id, trace_id="trace_" + str(i) * 32, + request_id=f"req-{i}", correlation_id="corr", attach_sequence=i, + attached_at=now, state=OperationWaiterState.WAITING, + ) for i in (1, 2)) + assert [waiter.attach_sequence for waiter in waiters] == [1, 2] diff --git a/test/observability/test_security.py b/test/observability/test_security.py new file mode 100644 index 0000000..8b0fc35 --- /dev/null +++ b/test/observability/test_security.py @@ -0,0 +1,26 @@ +import pytest +from pydantic import ValidationError + +from cacheroute_observability import TraceContext, TraceMeasurement + + +@pytest.mark.parametrize("key", [ + "password", "authorization", "redis_key", "kv_bytes", "tensor", + "device_pointer", "physical_path", "request_body", "raw_exception", +]) +def test_recursive_secret_and_physical_field_rejection(now, key): + data = { + "trace_id": "trace_" + "1" * 32, "request_id": "r", "correlation_id": "c", + "created_at": now, "nested": {key: "bad"}, + } + with pytest.raises(ValidationError, match="forbidden observability field"): + TraceContext.model_validate(data) + + +def test_generic_container_and_extra_rejected(provenance): + with pytest.raises(ValidationError): + TraceMeasurement(name="unsafe", kind="observed", provenance=provenance, + safe_scalar={"safe": "looking"}) + with pytest.raises(ValidationError): + TraceMeasurement(name="unsafe", kind="observed", provenance=provenance, + safe_scalar="ok", prompt="secret")