feat: standalone KV indexer runtime integration (#7295)

Author: tmonty12

components/src/dynamo/common/configuration/groups/kv_router_args.py

@@ -35,6 +35,7 @@
     "router_event_threads",
     "router_enable_cache_control",
     "router_queue_policy",
+    "remote_indexer_component",
 )
 
 
@@ -58,6 +59,7 @@ class KvRouterConfigBase(ConfigBase):
     router_event_threads: int
     router_enable_cache_control: bool
     router_queue_policy: str
+    remote_indexer_component: Optional[str]
 
     def kv_router_kwargs(self) -> dict:
         """Return a dict suitable for ``KvRouterConfig(**kwargs)``."""
@@ -269,3 +271,15 @@ def add_arguments(self, parser) -> None:
             arg_type=str,
             choices=["fcfs", "wspt"],
         )
+        add_argument(
+            g,
+            flag_name="--remote-indexer-component",
+            env_var="DYN_REMOTE_INDEXER_COMPONENT",
+            default=None,
+            help=(
+                "[EXPERIMENTAL] KV Router: Component name of a standalone KV indexer to use for overlap scoring. "
+                "When set, the router queries the standalone indexer via the request plane instead "
+                "of maintaining a local radix tree (e.g. 'kv-indexer')."
+            ),
+            arg_type=str,
+        )

components/src/dynamo/mocker/args.py

@@ -471,6 +471,13 @@ def parse_args() -> argparse.Namespace:
         default=os.environ.get("DYN_REQUEST_PLANE", "tcp"),
         help="Determines how requests are distributed from routers to workers. 'tcp' is fastest [nats|http|tcp]",
     )
+    parser.add_argument(
+        "--event-plane",
+        type=str,
+        choices=["nats", "zmq"],
+        default=os.environ.get("DYN_EVENT_PLANE", "nats"),
+        help="Determines how events are published [nats|zmq]",
+    )
 
     args = parser.parse_args()
     validate_worker_type_args(args)

components/src/dynamo/mocker/main.py

@@ -18,6 +18,7 @@
 
 os.environ.setdefault("DYN_COMPUTE_THREADS", "0")
 
+from dynamo.common.utils.runtime import create_runtime
 from dynamo.llm import (
     EngineType,
     EntrypointArgs,
@@ -26,7 +27,6 @@
     make_engine,
     run_input,
 )
-from dynamo.runtime import DistributedRuntime
 from dynamo.runtime.logging import configure_dynamo_logging
 
 from .args import create_temp_engine_args_file, parse_args, resolve_planner_profile_data
@@ -193,7 +193,6 @@ async def launch_workers(args: argparse.Namespace, extra_engine_args_path: Path)
     - Independent service registration and stats scraping
     - But still sharing the same tokio runtime (efficient)
     """
-    loop = asyncio.get_running_loop()
     futures = []
     runtimes = []
     per_worker_temp_files: list[Path] = []
@@ -227,10 +226,12 @@ async def launch_workers(args: argparse.Namespace, extra_engine_args_path: Path)
         logger.info(f"Creating mocker worker {worker_id + 1}/{args.num_workers}")
 
         # Create a separate DistributedRuntime for this worker (on same event loop)
-        runtime = DistributedRuntime(
-            loop,
+
+        runtime, loop = create_runtime(
             args.discovery_backend,
             args.request_plane,
+            args.event_plane,
+            True,  # statically set to True, just determines to enable_nats if event_plane is nats
         )
         runtimes.append(runtime)
 

docs/components/router/standalone-indexer.md

@@ -7,7 +7,10 @@ subtitle: Run the KV cache indexer as an independent HTTP service for querying b
 
 ## Overview
 
-The standalone KV indexer (`dynamo-kv-indexer`) is a lightweight HTTP binary that subscribes to ZMQ KV event streams from workers, maintains a radix tree of cached blocks, and exposes HTTP endpoints for querying and managing workers.
+The standalone KV indexer (`dynamo-kv-indexer`) is a lightweight binary that maintains a radix tree of cached blocks and exposes HTTP endpoints for querying and managing workers. It supports two operational modes:
+
+- **Standalone mode** (default): Subscribes to ZMQ KV event streams directly from workers. No Dynamo runtime dependencies required.
+- **Dynamo runtime mode** (`--dynamo-runtime`): Integrates with the Dynamo runtime for automatic worker discovery via MDC, KV event ingestion via the event plane (NATS or ZMQ), and serves indexer queries over the request plane for remote frontends.
 
 This is distinct from the [Standalone Router](../../../components/src/dynamo/router/README.md), which is a full routing service. The standalone indexer provides only the indexing and query layer without routing logic.
 
@@ -23,14 +26,17 @@ The indexer maintains one radix tree per `(model_name, tenant_id)` pair. Workers
 
 ## Compatibility
 
-The standalone indexer works with any engine that publishes KV cache events over ZMQ in the expected msgpack format. This includes bare vLLM and SGLang engines, which emit ZMQ KV events natively — no Dynamo-specific wrapper is required.
+In standalone mode, the indexer works with any engine that publishes KV cache events over ZMQ in the expected msgpack format. This includes bare vLLM and SGLang engines, which emit ZMQ KV events natively — no Dynamo-specific wrapper is required.
+
+In Dynamo runtime mode, the indexer discovers workers automatically via MDC and receives KV events through the event plane. It also registers a query endpoint on the request plane, allowing frontends to query overlap scores remotely without needing direct HTTP access.
 
 ## Use Cases
 
 - **Debugging**: Inspect the radix tree state to verify which blocks are cached on which workers.
 - **State verification**: Confirm that the indexer's view of KV cache state matches the router's internal state (used in integration tests).
 - **Custom routing**: Build external routing logic that queries the indexer for overlap scores and makes its own worker selection decisions.
 - **Monitoring**: Observe KV cache distribution across workers without running a full router.
+- **Remote indexing**: In Dynamo runtime mode, frontends can offload KV cache indexing to a dedicated service and query it over the request plane.
 
 ## P2P Recovery
 
@@ -75,18 +81,56 @@ Peers can be registered at startup via `--peers` or dynamically via the HTTP API
 
 ## Building
 
-The binary is a feature-gated target in the `dynamo-kv-router` crate:
+The binary is a feature-gated target in the `dynamo-kv-router` crate. The available cargo features control which capabilities are compiled in:
+
+| Feature | Description |
+|---------|-------------|
+| `standalone-indexer` | Core standalone indexer library (HTTP server, ZMQ listeners, P2P recovery) |
+| `metrics` | Prometheus metrics (`/metrics` endpoint, request/worker gauges) |
+| `indexer-bin` | CLI binary target |
+| `indexer-runtime` | Dynamo runtime integration (discovery, event plane, request plane) |
+| `test-endpoints` | Test-only endpoints (`/test/pause_listener`, `/test/resume_listener`) |
+
+### Standalone build (no runtime dependency)
 
 ```bash
 cargo build -p dynamo-kv-router --features indexer-bin --bin dynamo-kv-indexer
 ```
 
+This produces a binary with no `dynamo-runtime` dependency. It supports ZMQ event listeners, HTTP API, and P2P recovery.
+
+### Standalone build with metrics
+
+```bash
+cargo build -p dynamo-kv-router --features indexer-bin,metrics --bin dynamo-kv-indexer
+```
+
+Adds Prometheus metrics support (`/metrics` endpoint). Pulls in `dynamo-runtime` for the metrics implementation.
+
+### Runtime-enabled build
+
+```bash
+cargo build -p dynamo-kv-router --features indexer-bin,indexer-runtime --bin dynamo-kv-indexer
+```
+
+Enables the `--dynamo-runtime` CLI flag for MDC discovery, event plane subscription, and request plane query endpoint. Includes metrics.
+
 ## CLI
 
+### Standalone mode (default)
+
 ```bash
 dynamo-kv-indexer --port 8090 [--threads 4] [--block-size 16 --model-name my-model --tenant-id default --workers "1=tcp://host:5557,2:1=tcp://host:5558"] [--peers "http://peer1:8090,http://peer2:8091"]
 ```
 
+### Dynamo runtime mode (requires `indexer-runtime` feature)
+
+```bash
+dynamo-kv-indexer --dynamo-runtime --namespace default --component-name kv-indexer --worker-component backend --port 8090 [--threads 4]
+```
+
+In runtime mode, workers are discovered automatically via MDC. The `--workers` flag can still be used to register additional static workers alongside discovered ones.
+
 | Flag | Default | Description |
 |------|---------|-------------|
 | `--block-size` | (none) | KV cache block size for initial `--workers` (required when `--workers` is set) |
@@ -96,6 +140,10 @@ dynamo-kv-indexer --port 8090 [--threads 4] [--block-size 16 --model-name my-mod
 | `--model-name` | `default` | Model name for initial `--workers` |
 | `--tenant-id` | `default` | Tenant ID for initial `--workers` |
 | `--peers` | (none) | Comma-separated peer indexer URLs for P2P recovery on startup |
+| `--dynamo-runtime` | `false` | Enable Dynamo runtime integration (requires `indexer-runtime` feature) |
+| `--namespace` | `default` | Dynamo namespace to register the indexer component under (runtime mode) |
+| `--component-name` | `kv-indexer` | Component name for this indexer in the Dynamo runtime (runtime mode) |
+| `--worker-component` | `backend` | Component name that workers register under, for event plane subscription (runtime mode) |
 
 ## HTTP API
 
@@ -109,7 +157,7 @@ curl http://localhost:8090/health
 
 ### `GET /metrics` — Prometheus metrics
 
-Returns metrics in Prometheus text exposition format. Available when the binary is built with the `metrics` feature (enabled by default via `standalone-indexer`).
+Returns metrics in Prometheus text exposition format. Available when the binary is built with the `metrics` or `indexer-runtime` feature.
 
 ```bash
 curl http://localhost:8090/metrics
@@ -313,13 +361,44 @@ If no `replay_endpoint` is configured, gaps are logged as warnings but not recov
 
 The sequence counter (`last_seq`) persists across unregister/register cycles, so re-registering a worker after a gap will trigger replay on the first batch received by the new listener.
 
+## Dynamo Runtime Mode
+
+When started with `--dynamo-runtime`, the indexer integrates with the Dynamo distributed runtime:
+
+### Worker Discovery
+
+The indexer watches MDC (Model Discovery Catalog) for worker additions and removals. When a worker registers with MDC, the indexer automatically creates an indexer for its model and block size. Workers discovered via MDC are tracked separately from those registered via `--workers` or the `/register` HTTP API — a worker cannot be registered through both paths simultaneously.
+
+### Event Plane Subscription
+
+Instead of connecting directly to ZMQ PUB sockets on each worker, the indexer subscribes to KV events through the Dynamo event plane. The transport (NATS or ZMQ) is determined by the `DYNAMO_EVENT_TRANSPORT` environment variable. Events are routed to the appropriate indexer based on the worker ID.
+
+### Request Plane Query Endpoint
+
+The indexer registers a query endpoint on the Dynamo request plane, allowing frontends to send `IndexerQueryRequest` messages containing a model name, namespace, and block hashes. The indexer looks up the appropriate radix tree and returns overlap scores. This enables frontends to use a remote indexer for KV-aware routing without direct HTTP access.
+
+### Example
+
+```bash
+# Start the indexer with runtime integration
+dynamo-kv-indexer --dynamo-runtime \
+  --namespace my-namespace \
+  --component-name kv-indexer \
+  --worker-component backend \
+  --port 8090 --threads 4
+```
+
+The HTTP API remains fully available in runtime mode. Static workers can be added via `--workers` alongside discovered workers.
+
 ## Limitations
 
-- **ZMQ only**: Workers must publish KV events via ZMQ PUB sockets. The standalone indexer does not subscribe to NATS event streams.
+- **Standalone mode is ZMQ only**: In standalone mode, workers must publish KV events via ZMQ PUB sockets. Build with `indexer-runtime` and use `--dynamo-runtime` to receive events via the event plane (NATS or ZMQ).
 - **No routing logic**: The indexer only maintains the radix tree and answers queries. It does not track active blocks, manage request lifecycle, or perform worker selection.
 
 ## Architecture
 
+### Standalone Mode
+
 ```mermaid
 graph TD
     subgraph Workers
@@ -353,6 +432,62 @@ graph TD
     style CLIENT fill:#fff3e0,stroke:#333,color:#333
 ```
 
+### Dynamo Runtime Mode
+
+```mermaid
+graph TD
+    subgraph Workers
+        W1[Worker 1]
+        W2[Worker 2]
+    end
+
+    subgraph "Dynamo Runtime"
+        MDC[MDC Discovery]
+        EP[Event Plane<br/>NATS / ZMQ]
+        RP[Request Plane]
+    end
+
+    subgraph "Standalone Indexer"
+        DISC[Discovery Watcher]
+        SUB[Event Subscriber]
+        REG[Worker Registry]
+        IDX["Indexer Map<br/>(model, tenant) → Radix Tree"]
+        QE[Query Endpoint]
+        HTTP[HTTP API<br/>/query /dump /register]
+    end
+
+    FRONTEND[Frontend / Router]
+    CLIENT[External Client]
+
+    W1 -->|register| MDC
+    W2 -->|register| MDC
+    MDC -->|added/removed| DISC
+    DISC -->|add/remove workers| REG
+    W1 -->|KV events| EP
+    W2 -->|KV events| EP
+    EP -->|RouterEvent| SUB
+    SUB -->|apply events| IDX
+    FRONTEND -->|IndexerQueryRequest| RP
+    RP --> QE
+    QE -->|query| IDX
+    CLIENT -->|POST /query, GET /dump| HTTP
+    HTTP -->|query| IDX
+
+    style W1 fill:#f3e5f5,stroke:#333,color:#333
+    style W2 fill:#f3e5f5,stroke:#333,color:#333
+    style MDC fill:#e3f2fd,stroke:#333,color:#333
+    style EP fill:#e3f2fd,stroke:#333,color:#333
+    style RP fill:#e3f2fd,stroke:#333,color:#333
+    style IDX fill:#2e8b57,stroke:#333,color:#fff
+    style SUB fill:#2e8b57,stroke:#333,color:#fff
+    style DISC fill:#2e8b57,stroke:#333,color:#fff
+    style REG fill:#2e8b57,stroke:#333,color:#fff
+    style QE fill:#2e8b57,stroke:#333,color:#fff
+    style HTTP fill:#2e8b57,stroke:#333,color:#fff
+    style FRONTEND fill:#fff3e0,stroke:#333,color:#333
+    style CLIENT fill:#fff3e0,stroke:#333,color:#333
+```
+
 ### P2P Recovery Flow
 
 ```mermaid

lib/bindings/c/src/lib.rs

@@ -691,8 +691,9 @@ pub unsafe extern "C" fn create_routers(
             .kv_chooser_for(
                 &endpoint,
                 block_size,
-                Some(kv_router_config),
+                Some(kv_router_config.clone()),
                 WORKER_TYPE_DECODE,
+                Some(model_name.clone()),
             )
             .await
         {

lib/bindings/python/rust/llm/entrypoint.rs

@@ -40,21 +40,21 @@ pub enum EngineType {
 }
 
 #[pyclass]
-#[derive(Default, Clone, Debug, Copy)]
+#[derive(Default, Clone, Debug)]
 pub struct KvRouterConfig {
     inner: RsKvRouterConfig,
 }
 
 impl KvRouterConfig {
     pub fn inner(&self) -> RsKvRouterConfig {
-        self.inner
+        self.inner.clone()
     }
 }
 
 #[pymethods]
 impl KvRouterConfig {
     #[new]
-    #[pyo3(signature = (overlap_score_weight=1.0, router_temperature=0.0, use_kv_events=true, durable_kv_events=false, router_replica_sync=false, router_track_active_blocks=true, router_track_output_blocks=false, router_assume_kv_reuse=true, router_snapshot_threshold=1000000, router_reset_states=false, router_ttl_secs=120.0, router_max_tree_size=1048576, router_prune_target_ratio=0.8, router_queue_threshold=Some(2.0), router_event_threads=4, router_enable_cache_control=false, router_queue_policy="fcfs"))]
+    #[pyo3(signature = (overlap_score_weight=1.0, router_temperature=0.0, use_kv_events=true, durable_kv_events=false, router_replica_sync=false, router_track_active_blocks=true, router_track_output_blocks=false, router_assume_kv_reuse=true, router_snapshot_threshold=1000000, router_reset_states=false, router_ttl_secs=120.0, router_max_tree_size=1048576, router_prune_target_ratio=0.8, router_queue_threshold=Some(2.0), router_event_threads=4, router_enable_cache_control=false, router_queue_policy="fcfs", remote_indexer_component=None))]
     #[allow(clippy::too_many_arguments)]
     fn new(
         overlap_score_weight: f64,
@@ -74,6 +74,7 @@ impl KvRouterConfig {
         router_event_threads: u32,
         router_enable_cache_control: bool,
         router_queue_policy: &str,
+        remote_indexer_component: Option<String>,
     ) -> Self {
         KvRouterConfig {
             inner: RsKvRouterConfig {
@@ -96,6 +97,7 @@ impl KvRouterConfig {
                 router_queue_policy: router_queue_policy.parse().unwrap_or_else(|_| {
                     panic!("invalid router_queue_policy: {router_queue_policy:?}")
                 }),
+                remote_indexer_component,
             },
         }
     }