feat: add push-based engine load reporting and Router load monitor

.github/workflows/pr-test-sgl-router.yml

@@ -324,7 +324,7 @@ jobs:
         run: |
           docker build -t sgl-router-fake-worker:e2e \
             -f experimental/sgl-router/tests/e2e/k8s_integration/Dockerfile.fake_worker \
-            experimental/sgl-router/tests/e2e/k8s_integration/
+            .
       - name: Bootstrap kind + deploy
         run: bash experimental/sgl-router/tests/e2e/k8s_integration/setup.sh
       - name: Set up Python

.pre-commit-config.yaml

@@ -29,7 +29,7 @@ repos:
     rev: 7.0.0
     hooks:
       - id: isort
-        exclude: '^python/sglang/srt/grpc/.*_pb2\.py$|^python/sglang/srt/grpc/.*_pb2_grpc\.py$|^python/sglang/srt/grpc/.*_pb2\.pyi$|^python/sglang/srt/grpc/.*_pb2_grpc\.pyi$'
+        exclude: '^python/sglang/srt/grpc/.*_pb2\.py$|^python/sglang/srt/grpc/.*_pb2_grpc\.py$|^python/sglang/srt/grpc/.*_pb2\.pyi$|^python/sglang/srt/grpc/.*_pb2_grpc\.pyi$|^python/sglang/srt/load_reporter/proto/.*_pb2(_grpc)?\.py$'
   - repo: https://github.com/astral-sh/ruff-pre-commit
     rev: v0.15.1
     hooks:
@@ -46,12 +46,13 @@ repos:
           python/sglang/srt/grpc/.*_pb2_grpc\.py$|
           python/sglang/srt/grpc/.*_pb2\.pyi$|
           python/sglang/srt/grpc/.*_pb2_grpc\.pyi$|
+          python/sglang/srt/load_reporter/proto/.*_pb2(_grpc)?\.py$|
           )$
   - repo: https://github.com/psf/black
     rev: 26.1.0
     hooks:
       - id: black-jupyter
-        exclude: '^python/sglang/srt/grpc/.*_pb2\.py$|^python/sglang/srt/grpc/.*_pb2_grpc\.py$|^python/sglang/srt/grpc/.*_pb2\.pyi$|^python/sglang/srt/grpc/.*_pb2_grpc\.pyi$'
+        exclude: '^python/sglang/srt/grpc/.*_pb2\.py$|^python/sglang/srt/grpc/.*_pb2_grpc\.py$|^python/sglang/srt/grpc/.*_pb2\.pyi$|^python/sglang/srt/grpc/.*_pb2_grpc\.pyi$|^python/sglang/srt/load_reporter/proto/.*_pb2(_grpc)?\.py$'
   - repo: https://github.com/codespell-project/codespell
     rev: v2.4.1
     hooks:

docker/sgl-router.Dockerfile

@@ -37,6 +37,8 @@ RUN cargo install cargo-chef --locked --version ^0.1
 WORKDIR /work
 COPY experimental/sgl-router/Cargo.toml ./
 COPY experimental/sgl-router/rust-toolchain.toml ./
+COPY experimental/sgl-router/build.rs ./
+COPY experimental/sgl-router/proto ./proto
 # Stub a minimal src tree so cargo can resolve the workspace, generate
 # the lockfile (gitignored upstream), then prepare the chef recipe.
 RUN mkdir -p src && echo "fn main() {}" > src/main.rs \
@@ -69,6 +71,8 @@ RUN cargo chef cook --release --recipe-path recipe.json
 
 # Now bring in the real sources and the manifest they need.
 COPY experimental/sgl-router/Cargo.toml ./
+COPY experimental/sgl-router/build.rs ./
+COPY experimental/sgl-router/proto ./proto
 COPY experimental/sgl-router/src ./src
 
 # --locked is intentionally omitted: the lockfile is generated in-container

experimental/sgl-router/Cargo.toml

@@ -28,7 +28,7 @@ dynamo-tokenizers = { git = "https://github.com/ai-dynamo/dynamo", rev = "1efdd4
 dynamo-parsers = { git = "https://github.com/ai-dynamo/dynamo", rev = "1efdd4dcb901caeae636131321094090d252c8d6" }
 
 # Async runtime + http
-tokio = { version = "1.42", features = ["full"] }
+tokio = { version = "=1.48.0", features = ["full"] }
 axum = { version = "0.8", features = ["macros", "tracing"] }
 tower = { version = "0.5", features = ["full"] }
 tower-http = { version = "0.6", features = ["trace", "compression-gzip", "cors", "timeout", "request-id"] }
@@ -62,12 +62,15 @@ tracing-subscriber = { version = "0.3", features = ["env-filter", "json"] }
 futures = "0.3"
 bytes = "1"
 rand = "0.8"
-tokio-stream = "0.1"
+tokio-stream = { version = "0.1", features = ["net"] }
 dashmap = "6"
 kube = { version = "0.96", features = ["runtime", "derive"] }
 k8s-openapi = { version = "0.23", features = ["v1_31"] }
 tokio-util = "0.7"
 uuid = { version = "1", features = ["v4"] }
+prost = "0.13"
+prost-types = "0.13"
+tonic = { version = "0.12", features = ["transport"] }
 
 # KV-event subsystem — msgpack-encoded events over ZMQ and sha256-based
 # block hashing matching SGLang's `radix_cache`.  Wire format authority is
@@ -85,7 +88,7 @@ serde = { version = "1", features = ["derive"] }
 serde_json = "1"
 tempfile = "3"
 tower = { version = "0.5", features = ["util"] }
-tokio = { version = "1.42", features = ["test-util"] }
+tokio = { version = "=1.48.0", features = ["test-util"] }
 # Low-level msgpack encoder used to hand-construct wire bytes in
 # kv_events golden-bytes tests (decode-only path uses rmp-serde).
 rmp = "0.8"
@@ -95,6 +98,10 @@ rmp = "0.8"
 criterion = { version = "0.5", features = ["html_reports"] }
 rand = "0.8"
 
+[build-dependencies]
+protoc-bin-vendored = "3"
+tonic-build = "0.12"
+
 [[test]]
 name = "component"
 path = "tests/component/main.rs"

experimental/sgl-router/README.md

@@ -4,9 +4,9 @@ Slim, KV-aware, OpenAI-compatible router for SGLang workers.
 
 Serves a single model and routes across its workers. Exposes
 `/v1/tokenize`, `/v1/detokenize`, `/v1/models`, `/v1/chat/completions`
-(buffered and SSE), plus `/healthz` / `/readyz` and `/metrics`. Worker
-pools come from either a static URL list or Kubernetes EndpointSlice
-discovery.
+(buffered and SSE), plus `/healthz` / `/readyz`, `/metrics`, and the
+load-monitor diagnostic endpoint `/v1/load_monitor/snapshot`. Worker pools
+come from either a static URL list or Kubernetes EndpointSlice discovery.
 
 ## Building
 
@@ -50,6 +50,46 @@ Omit `--service-discovery-namespace` to watch all namespaces (requires
 cluster-wide RBAC). For prefill/decode disaggregation, replace `--selector`
 with `--prefill-selector` and `--decode-selector`.
 
+## Engine-reported load monitoring
+
+Load monitoring is disabled by default. When enabled, the Router first binds
+an independent gRPC listener, then asks every discovered worker to start or
+renew reporting through `/v1/start_reporting`. Port `0` is supported and the
+actual bound port is sent to the engine:
+
+```bash
+sgl-router \
+  --host 0.0.0.0 --port 30000 \
+  --model-id qwen3 \
+  --tokenizer-path /models/qwen3/tokenizer.json \
+  --worker-urls http://10.0.0.1:30000 http://10.0.0.2:30000 \
+  --policy load_based \
+  --load-monitor \
+  --load-monitor-bind-host 0.0.0.0 \
+  --load-monitor-bind-port 0 \
+  --load-monitor-report-ip 10.0.0.10
+```
+
+`--load-monitor-report-ip` is required and must be reachable from the engine.
+The first version uses a fixed 1-second report interval, 3-second freshness
+window, 15-second lease, and 2-second registration timeout. `load_based`,
+`power_of_two`, `cache_aware_zmq`, and sticky policies with a load-scored
+fallback require the monitor; round-robin and random can run without it.
+
+The Snapshot endpoint returns one immutable, versioned capture with worker
+freshness, source and sequence metadata, complete DP-rank values, and aggregate
+load. When monitoring is disabled it returns:
+
+```json
+{"enabled":false,"version":0,"captured_at":null,"workers":[]}
+```
+
+The Router intentionally sends no `Authorization` header to
+`/v1/start_reporting`. It is therefore compatible with an unauthenticated
+open-source or fake engine. Engine builds that enforce `ADMIN_FORCE` on this
+endpoint currently reject registration with 401/403; authenticated reporting
+is outside this Router-only change.
+
 ## License
 
 Apache-2.0.

experimental/sgl-router/benches/policy_select.rs

@@ -15,7 +15,7 @@ use sgl_router::discovery::{ModelId, WorkerId, WorkerMode, WorkerSpec};
 use sgl_router::policies::power_of_two::PowerOfTwoChoicesPolicy;
 use sgl_router::policies::random::RandomPolicy;
 use sgl_router::policies::round_robin::RoundRobinPolicy;
-use sgl_router::policies::{Policy, SelectionContext};
+use sgl_router::policies::{Policy, PolicyCandidate, SelectionContext};
 use sgl_router::workers::{Worker, WorkerRegistry};
 use std::sync::Arc;
 
@@ -39,6 +39,17 @@ fn bench_policy(c: &mut Criterion, name: &str, policy: Arc<dyn Policy>) {
     let mut group = c.benchmark_group(format!("policy_select::{name}"));
     for &n in &[4usize, 16, 64, 256] {
         let workers = workers(n, "tiny");
+        let candidates = workers
+            .iter()
+            .map(|worker| PolicyCandidate {
+                worker: Arc::clone(worker),
+                load: Some(sgl_router::load_monitor::AggregateLoad {
+                    max_total_num_tokens: 1,
+                    max_running_requests: 1,
+                    ..Default::default()
+                }),
+            })
+            .collect::<Vec<_>>();
         let model = ModelId("tiny".into());
         // Same body across iterations — measures the policy's per-call
         // cost rather than body-parsing overhead.
@@ -51,7 +62,7 @@ fn bench_policy(c: &mut Criterion, name: &str, policy: Arc<dyn Policy>) {
         group.bench_with_input(BenchmarkId::from_parameter(n), &n, |b, _| {
             b.iter(|| {
                 let ctx = SelectionContext::new(&model, Some(&body));
-                let chosen = policy.select(black_box(&workers), &ctx);
+                let chosen = policy.select(black_box(&candidates), &ctx);
                 black_box(chosen);
             });
         });

experimental/sgl-router/build.rs

@@ -0,0 +1,25 @@
+// SPDX-FileCopyrightText: Copyright (c) 2026 The SGLang Authors
+// SPDX-License-Identifier: Apache-2.0
+
+use std::error::Error;
+
+/// Generates Rust gRPC bindings from the Router-local load-monitor protocol.
+///
+/// The build uses a vendored `protoc`, so contributors and CI do not need a
+/// system protobuf compiler. The generated code is written to Cargo's normal
+/// `OUT_DIR` and included by `src/load_monitor/proto.rs`.
+///
+/// # Errors
+///
+/// Returns an error when the vendored compiler cannot be located or when the
+/// protobuf schema cannot be compiled.
+fn main() -> Result<(), Box<dyn Error>> {
+    let protoc = protoc_bin_vendored::protoc_bin_path()?;
+    std::env::set_var("PROTOC", protoc);
+    println!("cargo:rerun-if-changed=proto/load_monitor.proto");
+    tonic_build::configure()
+        .build_server(true)
+        .build_client(true)
+        .compile_protos(&["proto/load_monitor.proto"], &["proto"])?;
+    Ok(())
+}

experimental/sgl-router/proto/load_monitor.proto

@@ -0,0 +1,58 @@
+syntax = "proto3";
+
+package router.loadmonitor.v1;
+
+import "google/protobuf/empty.proto";
+
+service LoadMonitorService {
+  rpc Report(stream LoadReport) returns (google.protobuf.Empty);
+}
+
+enum WorkerType {
+  WORKER_TYPE_UNSPECIFIED = 0;
+  WORKER_TYPE_REGULAR = 1;
+  WORKER_TYPE_PREFILL = 2;
+  WORKER_TYPE_DECODE = 3;
+}
+
+enum ReportStatus {
+  REPORT_STATUS_UNSPECIFIED = 0;
+  REPORT_STATUS_HEALTHY = 1;
+  REPORT_STATUS_STALE = 2;
+  REPORT_STATUS_UNREACHABLE = 3;
+}
+
+message Worker {
+  string worker_addr = 1;
+  WorkerType worker_type = 2;
+  optional string model = 3;
+  optional string zone = 4;
+}
+
+message RankLoad {
+  int32 dp_rank = 1;
+  int64 snapshot_time_unix_ms = 2;
+  int64 num_running_reqs = 3;
+  int64 num_waiting_reqs = 4;
+  int64 num_waiting_uncached_tokens = 5;
+  int64 num_used_tokens = 6;
+  int64 num_total_tokens = 7;
+  int64 max_total_num_tokens = 8;
+  int64 max_running_requests = 9;
+  double token_usage = 10;
+  double gen_throughput = 11;
+  double cache_hit_rate = 12;
+  double utilization = 13;
+  // Completed uncached Prefill compute throughput in tokens per second.
+  double prefill_throughput = 14;
+}
+
+message LoadReport {
+  string source_instance_id = 1;
+  uint64 sequence_id = 2;
+  int64 report_time_unix_ms = 3;
+  Worker worker = 4;
+  ReportStatus status = 5;
+  optional string last_error = 6;
+  repeated RankLoad ranks = 7;
+}