diff --git a/.design_docs/issue-242-design-decisions.md b/.design_docs/issue-242-design-decisions.md deleted file mode 100644 index 2ec97c5..0000000 --- a/.design_docs/issue-242-design-decisions.md +++ /dev/null @@ -1,188 +0,0 @@ -# Issue #242: Design Decisions Log - -This document records the design questions and decisions that shaped the implementation plan for issue #242 (programmatic control of precompute engine pipelines). It is intended to explain *why* the plan looks the way it does, not just *what* it does. - ---- - -## Problem Statement - -`asap-query-engine` acts as a proxy between Grafana and Prometheus. Currently, the `precompute_engine` inside it must be given a `streaming_config` (which metrics to precompute and how) at **startup time** via a static config file. - -The goal: start `asap-query-engine` with no precompute config, intercept queries from Grafana, call `asap-planner` on them to generate a `streaming_config` and `inference_config`, and then configure the running precompute engine with the result — no manual intervention required. - ---- - -## What Already Exists - -After exploring the codebase: - -- **Query interception** already exists — the HTTP proxy in `drivers/query/servers/http.rs` already intercepts and can record every query. -- **Planner integration** already exists — `LocalPlannerClient` and `QueryTracker` already collect queries and call the planner on a periodic loop. -- **The planner output is discarded** — `tracker.rs` logs the result and throws it away. This is the core gap. -- **`precompute_engine` has no runtime reconfiguration** — `agg_configs` are built from `streaming_config` at startup inside `run()` and never touched again. There is no API to add, remove, or update pipelines. - ---- - -## Decision 1: When to trigger planning - -**Question:** "Once on startup" — but at startup there are no intercepted queries yet. What does the planner plan? - -**Options considered:** -- (a) Call planner at startup with no queries, letting it do pure metric discovery from Prometheus. -- (b) Wait for the first observation window to elapse, collect real queries during that window, then plan once and configure. - -**Decision: (b)** — wait for the first real observation window, plan with actual queries. More meaningful input to the planner. During the observation window, all queries fall through to Prometheus via the existing fallback mechanism, so users see no gap in Grafana. - -**Future:** Subsequent observation windows will eventually trigger replanning (repeated reconfiguration). The design accommodates this without structural changes. - ---- - -## Decision 2: One-shot vs. repeated loop - -**Question:** After the first plan is applied, should the `QueryTracker` loop stop or keep running? - -**Decision:** Keep the loop running, but only *apply* the config on the first successful plan (via an `AtomicBool applied` flag). This keeps the tracker alive for observability and makes it trivial to extend to repeated reconfiguration later — just remove the flag check. - ---- - -## Decision 3: Runtime reconfiguration strategy (Option A vs. B) - -**Question:** How should the engine be reconfigured at runtime? - -**Option A — Lazy initialization (two-phase startup):** -- Start HTTP server with no engine; all queries fall back to Prometheus. -- After first window + plan, construct all components fresh from planner output. -- Install the new engine into the HTTP server via `Arc>>`. -- For *repeated* reconfiguration: tear down old engine and swap in a new one each time — store's accumulated sketch data is lost on every reconfiguration. - -**Option B — Start with empty configs, hot-swap internals:** -- Start all components immediately with empty `streaming_config` / `inference_config`. -- After planning, update shared state in-place: push new `agg_configs` to workers via message, swap `IngestState.agg_configs` via `ArcSwap`, update `SimpleEngine` and `SimpleMapStore` via `RwLock`. -- Workers are **not restarted** — they lazily pick up new aggregation configs as samples arrive. -- For *repeated* reconfiguration: same mechanism — update configs, workers adapt. Store's precomputed history is preserved across reconfigurations. - -**Decision: Option B** — chosen because it is more amenable to future repeated reconfiguration. Option A loses all precomputed sketch data on every config update, which is wasteful once the engine has been running for multiple windows. Option B preserves historical precomputed data and allows incremental updates. - ---- - -## Decision 4: Config replacement vs. merging - -**Question:** When new planner output arrives, does it replace the existing config entirely, or merge with it? - -**Decision: Replace** — the new planner output becomes the complete truth. However, the **input** to the planner includes the current `streaming_config` and `inference_config` (as `ControllerConfig.existing_*` fields), so the planner has the context it needs to make coherent decisions across windows. - -This means: if a metric was being precomputed and the planner decides it still should be, it will appear in the new config. If it doesn't appear, it is dropped — data for that aggregation expires naturally via the existing cleanup policy (see Decision 8). - ---- - -## Decision 5: Passing existing configs to the planner - -**Question:** `ControllerConfig` (in `asap-planner-rs`) currently has no fields for existing configs. The planner only knows about new query observations. For repeated reconfiguration, the planner needs context about what is already running to make coherent decisions. - -**Decision:** Add `existing_streaming_config: Option` and `existing_inference_config: Option` to `ControllerConfig` **now**, even though the planner does not yet use them. This wires the information through so the planner can use it in the future without a second round of type-signature changes. The fields are clearly marked with a `NOTE` comment. - ---- - -## Decision 6: Who applies the planner result - -**Question:** `QueryTracker` calls the planner but has no reference to the engine components. Who applies the result? - -**Options considered:** -- (a) **Channel/callback** — tracker sends `PlannerResult` via a `tokio::sync::watch` channel; a separate task in `main.rs` owns the receiver and applies it to all engine components. -- (b) **Tracker owns engine handles** — pass `Arc` references of engine components into `QueryTracker`; it applies the result directly. - -**Decision: (a)** — keeps `QueryTracker` decoupled. It only produces results; it does not know about engine internals. The applier task in `main.rs` is the single place that knows about all components and applies updates to them. - ---- - -## Decision 7: Atomic vs. non-atomic config update - -**Question:** `SimpleEngine` (inference) and `PrecomputeEngine` (streaming) are updated by separate calls. There is a brief window where one has the new config and the other does not. - -**Decision:** Accept the brief inconsistency. During the transition window, a query might be matched by the old inference config against data computed by the new streaming config (or vice versa), which could produce a miss and fall back to Prometheus. This is acceptable — it is transient and self-correcting within one query cycle. - -**Implementation note:** The inconsistency window is marked with a `NOTE` comment at the application site in `main.rs` and in `SimpleEngine`. - ---- - -## Decision 8: Stale precomputed data and in-flight state after config replace - -**Question:** When `streaming_config` is replaced, how is in-flight precompute data handled, and what happens to stale state in both the store and the workers? - -### In-flight data in precompute engine workers - -`UpdateAggConfigs` replaces each worker's `agg_configs` map but **does not touch `group_states`**. Existing `GroupState` entries (keyed by `(agg_id, group_key)`) for old aggregation IDs are left in the worker's HashMap. - -Consequences: -- **Data is not silently dropped.** The flush loop iterates over *all* `group_states`, so old groups still close their open windows and emit accumulated sketch data to the `SimpleMapStore` — under the old aggregation IDs. -- **No new samples reach old group states.** The ingest router only routes samples to agg IDs present in the new `agg_configs`, so old groups stop receiving input but still flush whatever they have accumulated. -- **Memory is never reclaimed.** Old `GroupState` entries are never evicted from the worker's HashMap. For the one-shot first-plan scenario this is a fixed-size overhead. For repeated reconfigurations this is a latent memory leak — stale group states accumulate in worker memory until process restart. - -The data committed to the store under old agg IDs becomes **orphaned**: the query engine now uses new inference/streaming configs pointing to new agg IDs, so those store entries are never queried and eventually expire via the cleanup policy. - -**Implementation:** On `UpdateAggConfigs`, workers now prune `group_states` entries whose `agg_id` is absent from the new config, after a final forced flush. Specifically: -1. Identify removed agg IDs (old `agg_configs` keys not present in `new_configs`). -2. For each removed group state, run the window-close logic with `effective_wm = i64::MAX` — since no new samples will arrive, all open windows are treated as due regardless of watermark. -3. Emit accumulated data from those windows to the store, then remove the entry from `group_states` and update `group_count`. -4. Swap in `new_configs`. - -The final flush happens **before** `agg_configs` is swapped so that `GroupState.config` (an `Arc`) is still valid during window-bound calculations. The emitted data is still written under old agg IDs (and will not be queried), but this avoids silently discarding partially-accumulated windows and correctly frees the worker memory. - -### Stale data already committed to `SimpleMapStore` - -**Decision:** Let it expire naturally via the existing cleanup policy. No active purge. - -**Rationale:** The cleanup policy already handles TTL-based eviction. Implementing an active purge would require iterating over potentially large store state and coordinating with in-progress queries. The natural expiry path is safe and requires no new code. - -**Implementation note:** A `NOTE` comment in `update_streaming_config` on the store marks the alternative (active purge) for future reference. - ---- - -## Decision 9: Worker update mechanism - -**Question:** Workers each hold their own `HashMap>`. To give them new configs without restarting, the options are: -- Send `UpdateAggConfigs` messages via the existing `WorkerMessage` channel. -- Use shared mutable state (e.g. `Arc>>`) that workers read on every sample. - -**Decision:** Message-passing (`WorkerMessage::UpdateAggConfigs`). This fits the existing actor-like architecture (workers already process typed messages) and avoids adding a lock acquisition on every sample's hot path. - -Workers lazily create `WindowManager` instances the first time they see a new `agg_id`, so new aggregations are picked up automatically as samples arrive after the update — no special initialization needed. - ---- - -## Decision 11: Synchronization primitive for `IngestState.agg_configs` - -**Question:** `IngestState` is behind `Arc` (immutable). The ingest handler reads `agg_configs` on the hot path — once per HTTP request, iterating over all configs for every sample. The applier task needs to swap in a new vec. What interior-mutability primitive to use? - -**Options considered:** -- `RwLock>` — readers hold the lock during the full iteration over configs. Lock held on the hot path. -- `RwLock>>` — readers briefly lock to clone the Arc, then iterate without holding the lock. -- `ArcSwap>` — truly lock-free reads via atomic pointer swap; no locking on the read path at all. - -**Decision: `ArcSwap>>`** — the ingest handler is on the hot path and should not pay any lock cost on reads. `ArcSwap::load()` is lock-free; the applier calls `ArcSwap::store(Arc::new(new_vec))` for the atomic swap. - -**Note:** This is distinct from `SimpleMapStore.streaming_config`, where `RwLock>` is used. The store's `streaming_config` is accessed only during batch inserts (less frequent, not per-sample), and the sequential ordering in the applier task means there is no real concurrent write race for the store. `RwLock>` (brief lock to clone the pointer, then use without lock) is sufficient there. - -**Note:** `SimpleEngine.inference_config` uses `RwLock` — queries are the read path, which is less frequent than ingest. Holding the read lock for the duration of a query lookup is acceptable. - ---- - -## Decision 10: Kafka / OTLP consumers during observation window - -**Question:** Kafka consumers and OTLP receivers also feed data into the precompute engine. With Option B (empty configs at startup), they will be active during the observation window but producing no useful work (no aggregation configs to match against). - -**Decision:** No special handling needed. With empty `agg_configs`, the ingest handler simply drops all incoming samples (no matching aggregation found). Once the first plan is applied, subsequent samples are processed correctly. This is the natural behavior and requires no additional code. - ---- - -## Summary of Key Structural Changes - -| Component | Before | After | -|---|---|---| -| `streaming_config` in store | `Arc` | `RwLock>` (brief lock to clone pointer, iterate without lock) | -| `inference_config` in `SimpleEngine` | `InferenceConfig` (owned) | `RwLock` (SimpleEngine is already behind Arc) | -| `agg_configs` in `IngestState` | `Vec>` | `ArcSwap>>` (lock-free reads on hot path) | -| Worker config updates | impossible | `WorkerMessage::UpdateAggConfigs` | -| Planner output | logged and discarded | sent via `watch` channel, applied by `main.rs` task | -| `PrecomputeEngine::run()` | creates channels internally, consumes self | channels created in `new()`, `handle()` extracted before `run()` | -| `ControllerConfig` | queries only | queries + `existing_streaming_config` + `existing_inference_config` | diff --git a/.design_docs/issue-242-plan.md b/.design_docs/issue-242-plan.md deleted file mode 100644 index c46f767..0000000 --- a/.design_docs/issue-242-plan.md +++ /dev/null @@ -1,265 +0,0 @@ -# Issue #242: Programmatic Control of Precompute Engine Pipelines - -## Goal - -Enable `asap-query-engine` to configure its precompute engine pipelines at runtime without manual intervention. Specifically: intercept queries from Grafana, call `asap-planner` on them after one observation window, and apply the resulting `streaming_config` + `inference_config` to the running engine — no static config files required. - -**Scope for this PR:** trigger once, after the first observation window. Repeated reconfiguration is future work, but the design is intentionally amenable to it. - ---- - -## Data Flow - -``` -Grafana → HTTP proxy → record query in QueryTracker - ↓ (all queries fall back to Prometheus during observation window) - -[after observation_window_secs] - -QueryTracker::evaluate() - → reads current streaming_config + inference_config (empty on first run) - → builds ControllerConfig (queries + existing configs) - → PlannerClient::plan() - → sends PlannerResult via watch::Sender> - → sets applied = true (won't send again until repeated-reconfig is implemented) - -main.rs applier task (watches the receiver) - → PrecomputeEngineHandle::update_streaming_config() - → SimpleEngine::update_inference_config() - → SimpleMapStore::update_streaming_config() -``` - ---- - -## Changes Required - -### 1. `asap-planner-rs/src/config/input.rs` — `ControllerConfig` - -Add two optional fields for existing configs. The planner ignores them for now; they are wired through so that future repeated-reconfig can pass the current state as context to the planner. - -```rust -// NOTE: reserved for future repeated-reconfig — planner does not yet use these -pub existing_streaming_config: Option, -pub existing_inference_config: Option, -``` - ---- - -### 2. `query_tracker/tracker.rs` — `QueryTracker` - -- Add `Arc>` and `Arc>` fields — read-only, used to populate `ControllerConfig.existing_*` before calling the planner. -- Add `applied: AtomicBool` — only sends the result the first time. The background loop keeps running (for observability / future extension) but subsequent results are dropped until repeated-reconfig is wired up. -- `start_background_loop` gains a `tokio::sync::watch::Sender>` parameter. - -```rust -pub struct QueryTracker { - entries: Mutex>, - config: QueryTrackerConfig, - streaming_config: Arc>, // read-only reference - inference_config: Arc>, // read-only reference - applied: AtomicBool, -} -``` - -On first successful plan: send `Some(result)` over the watch channel and set `applied = true`. - ---- - -### 3. `precompute_engine/series_router.rs` — `WorkerMessage` - -Add a new variant so workers can receive config updates without restarting: - -```rust -pub enum WorkerMessage { - // ... existing variants ... - UpdateAggConfigs(HashMap>), -} -``` - ---- - -### 4. `precompute_engine/engine.rs` — `PrecomputeEngine` + new `PrecomputeEngineHandle` - -Currently `run()` creates channels and workers internally and consumes `self`, making post-start updates impossible. Restructure: - -- **Move channel creation to `new()`** — senders and receivers created at construction, stored on the struct. -- Add `PrecomputeEngineHandle`: - -```rust -pub struct PrecomputeEngineHandle { - worker_senders: Vec>, - ingest_agg_configs: Arc>>>, -} - -impl PrecomputeEngineHandle { - /// Update the ingest handler's agg_configs and broadcast new configs to all workers. - pub async fn update_streaming_config(&self, config: &StreamingConfig) { - let agg_configs_map: HashMap> = config - .get_all_aggregation_configs() - .iter() - .map(|(&id, cfg)| (id, Arc::new(cfg.clone()))) - .collect(); - let agg_configs_vec: Vec> = - agg_configs_map.values().cloned().collect(); - - // Lock-free atomic swap — ingest handler readers see new configs immediately - self.ingest_agg_configs.store(Arc::new(agg_configs_vec)); - - for sender in &self.worker_senders { - let _ = sender - .send(WorkerMessage::UpdateAggConfigs(agg_configs_map.clone())) - .await; - } - } -} -``` - -- `PrecomputeEngine::handle() -> Arc` — callable before `run()`. -- `run()` uses the handle's senders and `ingest_agg_configs` rather than creating its own. - ---- - -### 5. `precompute_engine/ingest_handler.rs` — `IngestState` - -```rust -// Before -agg_configs: Vec>, - -// After -agg_configs: ArcSwap>>, -``` - -`IngestState` is behind `Arc` (immutable), so `agg_configs` needs interior mutability. `ArcSwap` is used because the ingest handler reads `agg_configs` on the hot path (once per request, iterates over all configs per sample). `ArcSwap::load()` is lock-free for readers; the applier does a single atomic pointer swap via `ArcSwap::store()`. - -```rust -// Reading (hot path, per request): -let configs = state.agg_configs.load(); -for config in configs.iter() { ... } - -// Writing (once, from PrecomputeEngineHandle::update_streaming_config): -self.ingest_agg_configs.store(Arc::new(new_vec)); -``` - ---- - -### 6. `precompute_engine/worker.rs` — handle `UpdateAggConfigs` - -Workers process `WorkerMessage::UpdateAggConfigs` in their run loop: - -```rust -WorkerMessage::UpdateAggConfigs(new_configs) => { - self.agg_configs = new_configs; -} -``` - -Workers lazily create `WindowManager` instances the first time they see a new `agg_id`, so adding aggregations at runtime works without additional changes. - ---- - -### 7. `engines/simple_engine/mod.rs` — `SimpleEngine` - -`SimpleEngine` is already behind `Arc` at the call site, so no extra `Arc` wrapper is needed on the field — just a `RwLock` for interior mutability. - -```rust -// Before -inference_config: InferenceConfig, - -// After -inference_config: RwLock, -``` - -Add update method: - -```rust -pub fn update_inference_config(&self, new_config: InferenceConfig) { - *self.inference_config.write().unwrap() = new_config; -} -``` - -> **NOTE:** `streaming_config` and `inference_config` are applied to their respective components independently, not atomically. There is a brief window where the precompute engine has a new `streaming_config` but `SimpleEngine` is still using the old `inference_config`. This is acceptable for the current use case. - ---- - -### 8. `stores/simple_map_store/*.rs` — `SimpleMapStore` (all variants) - -The store variants are behind `Arc` at the call site. The `RwLock` goes around the `Arc` (not around the config itself) so that readers briefly lock to clone the Arc pointer, then use it without holding the lock during lookups. - -```rust -// Before -streaming_config: Arc, - -// After -streaming_config: RwLock>, -``` - -Add update method on the store trait and each implementation: - -```rust -pub fn update_streaming_config(&self, new_config: StreamingConfig) { - *self.streaming_config.write().unwrap() = Arc::new(new_config); -} -``` - -Readers clone the Arc cheaply under a brief shared lock, then look up aggregation configs from it without holding the lock: - -```rust -let config = self.streaming_config.read().unwrap().clone(); -config.get_aggregation_config(aggregation_id) -``` - -> **NOTE:** Precomputed data for aggregations removed by the new config is not purged immediately. It expires naturally via the existing cleanup policy. If immediate purge is ever needed, `update_streaming_config` is the right place to add it. - ---- - -### 9. `main.rs` — wiring - -```rust -// Start with empty configs -let streaming_config = Arc::new(RwLock::new(StreamingConfig::empty())); -let inference_config = Arc::new(RwLock::new(InferenceConfig::empty())); - -// All components share Arc clones -let store = SimpleMapStore::new(streaming_config.clone(), cleanup_policy); -let engine = SimpleEngine::new(inference_config.clone(), streaming_config.clone(), ...); -let precompute = PrecomputeEngine::new(config, streaming_config.clone(), sink); -let pe_handle = precompute.handle(); // extract before run() consumes self -tokio::spawn(precompute.run()); - -// Tracker reads current configs, sends first result via watch channel -let (plan_tx, mut plan_rx) = tokio::sync::watch::channel(None::); -let tracker = Arc::new(QueryTracker::new( - tracker_config, - streaming_config.clone(), - inference_config.clone(), -)); -tracker.start_background_loop(planner_client, plan_tx); - -// Applier task: watches for first plan result and applies it -// NOTE: streaming_config and inference_config are not applied atomically — see SimpleEngine note -tokio::spawn(async move { - loop { - if plan_rx.changed().await.is_err() { break; } - if let Some(result) = plan_rx.borrow().clone() { - pe_handle.update_streaming_config(&result.streaming_config).await; - engine.update_inference_config(result.inference_config); - store.update_streaming_config(result.streaming_config); - } - } -}); -``` - ---- - -## Files Touched - -| File | Change | -|---|---| -| `asap-planner-rs/src/config/input.rs` | Add `existing_streaming_config`, `existing_inference_config` to `ControllerConfig` | -| `query_tracker/tracker.rs` | Add config refs, `AtomicBool`, accept watch sender in `start_background_loop` | -| `precompute_engine/series_router.rs` | Add `UpdateAggConfigs` variant to `WorkerMessage` | -| `precompute_engine/engine.rs` | Move channel creation to `new()`, add `PrecomputeEngineHandle`, expose `handle()` | -| `precompute_engine/ingest_handler.rs` | `agg_configs: ArcSwap>` (lock-free reads on hot path) | -| `precompute_engine/worker.rs` | Handle `UpdateAggConfigs` message | -| `engines/simple_engine/mod.rs` | `RwLock`, add `update_inference_config` | -| `stores/simple_map_store/*.rs` | `RwLock>`, add `update_streaming_config` | -| `main.rs` | Wire watch channel, spawn applier task |