feat(subscriber): read engine load via GetLoads gRPC (CAC-235)

.github/workflows/ci.yml

@@ -106,8 +106,8 @@ jobs:
     - name: Verify protobufs up to date
       run: |
         make proto-gen
-        if [ -n "$(git status --porcelain -- pkg/server/proto)" ]; then
-          git status --short -- pkg/server/proto
+        if [ -n "$(git status --porcelain -- pkg/server/proto pkg/adapters/engine/vllmengine)" ]; then
+          git status --short -- pkg/server/proto pkg/adapters/engine/vllmengine
           exit 1
         fi
 

Makefile

@@ -220,9 +220,19 @@ manifests: controller-gen ## Generate CRD, RBAC, and webhook manifests.
 	$(CONTROLLER_GEN) rbac:roleName=inference-cache-manager-role crd webhook paths="./..." output:crd:artifacts:config=config/crd/bases output:rbac:artifacts:config=config/rbac output:webhook:artifacts:config=config/webhook
 
 .PHONY: proto-gen
-proto-gen: protoc-gen-go ## Generate protobuf Go code.
+proto-gen: protoc-gen-go proto-gen-vendored ## Generate protobuf Go code (IC contract + vendored external stubs).
 	PATH="$(LOCALBIN):$$PATH" protoc -I proto --go_out=. --go_opt=module=$(MODULE) --go-grpc_out=. --go-grpc_opt=module=$(MODULE) $$(find proto -name '*.proto' | sort)
 
+# The vendored external engine contract (GetLoads only) lives OUTSIDE proto/ on
+# purpose — it is not part of IC's own gRPC API and is exempt from buf lint — so
+# `proto-gen` above (which only scans proto/) does not cover it. This target
+# regenerates its stubs from the same pinned generators; it is a prerequisite of
+# proto-gen so a plain `make proto-gen` keeps both in sync, and CI diffs the
+# output the same way. See pkg/adapters/engine/vllmengine/vllm_engine.proto.
+.PHONY: proto-gen-vendored
+proto-gen-vendored: protoc-gen-go ## Regenerate the vendored external vLLM engine stubs (pkg/adapters/engine/vllmengine).
+	PATH="$(LOCALBIN):$$PATH" protoc -I pkg/adapters/engine/vllmengine --go_out=. --go_opt=module=$(MODULE) --go-grpc_out=. --go-grpc_opt=module=$(MODULE) pkg/adapters/engine/vllmengine/vllm_engine.proto
+
 .PHONY: proto-lint
 proto-lint: buf ## Lint the gRPC contract with buf (lint-only; codegen stays on protoc).
 	$(BUF) lint
@@ -276,7 +286,7 @@ vulncheck: $(LOCALBIN) ## Scan dependencies + reachable code for known Go vulner
 COVER_MIN ?= 90
 COVER_PROFILE ?= cover.out
 COVER_PROFILE_LOGIC ?= cover.logic.out
-COVER_EXCLUDE := pkg/server/proto/|zz_generated|/cmd/|/hack/|pkg/testing/
+COVER_EXCLUDE := pkg/server/proto/|pkg/adapters/engine/vllmengine/|zz_generated|/cmd/|/hack/|pkg/testing/
 
 .PHONY: cover
 cover: ## Run tests with coverage and print the per-function report (logic packages, cross-package counted).
@@ -541,7 +551,7 @@ ci: verify-naming verify-no-internal-refs verify-syft-pin fmt-check vet ci-lint
 .PHONY: pre-pr
 pre-pr: ci ## Pre-PR gate: CI gate + generated-code drift check + sample admission check + review checklist.
 	@$(MAKE) --no-print-directory manifests generate proto-gen >/dev/null
-	@gen='config/crd config/rbac/role.yaml config/webhook/manifests.yaml api/v1alpha1/zz_generated.deepcopy.go pkg/server/proto'; \
+	@gen='config/crd config/rbac/role.yaml config/webhook/manifests.yaml api/v1alpha1/zz_generated.deepcopy.go pkg/server/proto pkg/adapters/engine/vllmengine'; \
 	if ! git diff --quiet -- $$gen; then \
 		echo "✗ generated-code drift — regenerate and commit these files:"; \
 		git --no-pager diff --name-only -- $$gen; \

cmd/kvevent-subscriber/main.go

@@ -6,10 +6,13 @@
 // Two independent paths share the gRPC client:
 //   - Event path: ZMQ → decoded EventBatch → ReportCacheState (prefix adds) +
 //     PublishEvent (removals/clears). Debounced on a short window.
-//   - Stats path: HTTP GET against the engine's Prometheus /metrics → derived
-//     ReplicaStats → ReportCacheState (stats-only CSU). Ticks on its own
-//     cadence (~10s), so the snapshot/CR status surface (cache_memory_bytes,
-//     hit_rate, pressure) lights up regardless of event rate.
+//   - Stats path: engine load → derived ReplicaStats → ReportCacheState
+//     (stats-only CSU). The load source is selectable: HTTP GET against the
+//     engine's Prometheus /metrics (default), or the VllmEngine GetLoads gRPC
+//     RPC when --engine-loads-grpc is set (vLLM gRPC engines expose no HTTP
+//     /metrics). Ticks on its own cadence (~10s), so the snapshot/CR status
+//     surface (cache_memory_bytes, hit_rate, pressure) lights up regardless of
+//     event rate.
 //
 // The two paths are independent failure domains — a scrape failure never
 // blocks the event stream, and an event-stream drop never delays a stats tick.
@@ -19,6 +22,7 @@ import (
 	"context"
 	"errors"
 	"flag"
+	"io"
 	"log/slog"
 	"net/http"
 	"os"
@@ -33,6 +37,50 @@ import (
 	icpb "github.com/cachebox-project/inference-cache/pkg/server/proto/inferencecache/v1alpha1"
 )
 
+// engineScraper is the load-source contract both scrapers satisfy — a local
+// mirror of the engine package's unexported statsScraper. NewStatsReporter
+// accepts either concrete type through it.
+type engineScraper interface {
+	Scrape(context.Context) (*icpb.ReplicaStats, error)
+}
+
+// scraperParams carries the load-source selection inputs for buildStatsScraper.
+type scraperParams struct {
+	loadsGRPC      string
+	metricsURL     string
+	engineModel    string
+	tier           engine.CacheTier
+	cacheSizeBytes int64
+	ceiling        int
+}
+
+// buildStatsScraper selects the engine load source: the GetLoads gRPC scraper
+// when loadsGRPC is non-empty (preferred for vLLM gRPC engines, which expose no
+// HTTP /metrics), else the HTTP /metrics scraper. It returns the scraper, an
+// optional closer (the gRPC scraper owns a client conn — the HTTP one owns
+// nothing, so closer is nil), and an error only from gRPC dial setup.
+func buildStatsScraper(p scraperParams, httpClient *http.Client, logger *slog.Logger) (engineScraper, io.Closer, error) {
+	if p.loadsGRPC != "" {
+		gs, err := engine.NewGRPCLoadsScraper(engine.GRPCLoadsScraperConfig{
+			Addr:                  p.loadsGRPC,
+			CacheSizeBytes:        p.cacheSizeBytes,
+			MaxConcurrencyCeiling: p.ceiling,
+		})
+		if err != nil {
+			return nil, nil, err
+		}
+		return gs, gs, nil
+	}
+	ms := engine.NewMetricsScraper(httpClient, engine.ScraperConfig{
+		URL:                   p.metricsURL,
+		Tier:                  p.tier,
+		ModelLabel:            p.engineModel,
+		CacheSizeBytes:        p.cacheSizeBytes,
+		MaxConcurrencyCeiling: p.ceiling,
+	}, logger)
+	return ms, nil, nil
+}
+
 func main() {
 	var (
 		endpoint           = flag.String("engine-endpoint", "tcp://127.0.0.1:5557", "engine KV-event ZMQ PUB endpoint")
@@ -44,6 +92,7 @@ func main() {
 		scheme             = flag.String("hash-scheme", "vllm", "engine prefix-hash scheme (required, non-empty)")
 		window             = flag.Duration("window", 100*time.Millisecond, "add-batching/debounce flush window")
 		metricsURL         = flag.String("engine-metrics-url", "http://127.0.0.1:8000/metrics", "engine Prometheus /metrics URL")
+		loadsGRPC          = flag.String("engine-loads-grpc", "", "engine VllmEngine gRPC address (host:port) to read load via the GetLoads RPC instead of scraping --engine-metrics-url. Preferred for vLLM gRPC engines, which expose no HTTP /metrics. Empty = use the HTTP scrape.")
 		statsInterval      = flag.Duration("stats-interval", 10*time.Second, "ReplicaStats scrape/emit cadence")
 		cacheSizeBytes     = flag.Int64("engine-cache-size-bytes", 0, "engine total KV-cache capacity in bytes (multiplied by usage_perc to derive cacheMemoryBytes; 0 emits cacheMemoryBytes=0)")
 		ceiling            = flag.Int("max-concurrency-ceiling", 256, "denominator for the pressure proxy = clamp01((num_requests_running+num_requests_waiting)/ceiling)")
@@ -91,17 +140,35 @@ func main() {
 		engine.WithIgnoreBlockRemoved(*ignoreBlockRemoved))
 	sub := engine.NewSubscriber(*endpoint, *topic, engine.WithSubscriberLogger(logger))
 
-	scraper := engine.NewMetricsScraper(
-		&http.Client{Timeout: 5 * time.Second},
-		engine.ScraperConfig{
-			URL:                   *metricsURL,
-			Tier:                  tier,
-			ModelLabel:            *engineModel,
-			CacheSizeBytes:        *cacheSizeBytes,
-			MaxConcurrencyCeiling: *ceiling,
-		},
-		logger,
-	)
+	// Load source: gRPC GetLoads (preferred for vLLM gRPC engines, which expose no
+	// HTTP /metrics) when --engine-loads-grpc is set, else the HTTP /metrics scrape.
+	// Both satisfy the same statsScraper, so the StatsReporter is identical.
+	scraper, scraperCloser, serr := buildStatsScraper(scraperParams{
+		loadsGRPC:      *loadsGRPC,
+		metricsURL:     *metricsURL,
+		engineModel:    *engineModel,
+		tier:           tier,
+		cacheSizeBytes: *cacheSizeBytes,
+		ceiling:        *ceiling,
+	}, &http.Client{Timeout: 5 * time.Second}, logger)
+	if serr != nil {
+		logger.Error("build stats scraper", "err", serr)
+		os.Exit(1)
+	}
+	if scraperCloser != nil {
+		defer func() {
+			if err := scraperCloser.Close(); err != nil {
+				logger.Warn("closing load-scraper connection", "err", err)
+			}
+		}()
+	}
+
+	loadSource, loadTarget := "HTTP /metrics scrape", *metricsURL
+	if *loadsGRPC != "" {
+		loadSource, loadTarget = "GetLoads gRPC", *loadsGRPC
+	}
+	logger.Info("engine load source", "source", loadSource, "target", loadTarget)
+
 	statsReporter := engine.NewStatsReporter(client, scraper, cfg,
 		engine.WithStatsInterval(*statsInterval),
 		engine.WithStatsLogger(logger),
@@ -135,7 +202,8 @@ func main() {
 		"server", *server,
 		"replica_id", *replica,
 		"model_id", *model,
-		"engine_metrics_url", *metricsURL,
+		"load_source", loadSource,
+		"load_target", loadTarget,
 		"stats_interval", statsInterval.String(),
 		"cache_tier", *cacheTier,
 		"engine_model_name", *engineModel,

cmd/kvevent-subscriber/main_test.go

@@ -0,0 +1,50 @@
+package main
+
+import (
+	"io"
+	"log/slog"
+	"net/http"
+	"testing"
+
+	"github.com/cachebox-project/inference-cache/pkg/adapters/engine"
+)
+
+// TestBuildStatsScraperSelectsSource pins the load-source control flow: a
+// non-empty --engine-loads-grpc must select the GetLoads gRPC scraper (and hand
+// back a closer for its conn), while an empty flag must preserve the HTTP
+// /metrics scraper (which owns no conn, so no closer). This is the crux of
+// Option B — the wrong branch silently leaves a vLLM engine served over gRPC reporting no
+// load — so it gets an explicit test.
+func TestBuildStatsScraperSelectsSource(t *testing.T) {
+	hc := &http.Client{}
+	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
+
+	t.Run("nonempty flag selects gRPC", func(t *testing.T) {
+		s, closer, err := buildStatsScraper(scraperParams{loadsGRPC: "127.0.0.1:50051"}, hc, logger)
+		if err != nil {
+			t.Fatalf("buildStatsScraper: %v", err)
+		}
+		if _, ok := s.(*engine.GRPCLoadsScraper); !ok {
+			t.Fatalf("selected %T, want *engine.GRPCLoadsScraper", s)
+		}
+		if closer == nil {
+			t.Fatal("gRPC scraper must return a closer to release its client conn")
+		}
+		if err := closer.Close(); err != nil {
+			t.Errorf("closer.Close: %v", err)
+		}
+	})
+
+	t.Run("empty flag preserves HTTP", func(t *testing.T) {
+		s, closer, err := buildStatsScraper(scraperParams{metricsURL: "http://127.0.0.1:8000/metrics"}, hc, logger)
+		if err != nil {
+			t.Fatalf("buildStatsScraper: %v", err)
+		}
+		if _, ok := s.(*engine.MetricsScraper); !ok {
+			t.Fatalf("selected %T, want *engine.MetricsScraper", s)
+		}
+		if closer != nil {
+			t.Fatal("HTTP scraper owns no conn; closer must be nil")
+		}
+	})
+}

docs/design/kvevent-subscriber-wiring.md

@@ -66,10 +66,14 @@ Concretely:
   admission picks it up once the operator sets the key), `--hash-scheme` ← the
   adapter's runtime convention (`"vllm"` or `"sglang"`), `--server` ← the policy-server
   in-cluster Service DNS (operator-configurable via a controller flag),
-  `--engine-endpoint` ← `tcp://127.0.0.1:<engine ZMQ port>`. The stats-path flags
-  (`--engine-metrics-url`, `--stats-interval`, etc.) are added by the adapter when the
-  shipped subscriber binary learns to scrape and emit `ReplicaStats`; passing flags the
-  binary doesn't recognise would crash the sidecar at startup. No operator-supplied
+  `--engine-endpoint` ← `tcp://127.0.0.1:<engine ZMQ port>`. The stats path runs on
+  the subscriber binary's built-in defaults: it scrapes the engine's HTTP Prometheus
+  `/metrics` (the `--engine-metrics-url` default) and emits `ReplicaStats` on its own
+  cadence. Explicit stats-path flags — `--stats-interval`, the cache-size/ceiling
+  hints, and `--engine-loads-grpc` (read load via the engine's GetLoads gRPC RPC
+  instead of HTTP; see the deferred note below) — are all recognised by the binary
+  but not yet set by the sidecar; wiring them (and deriving their values from the
+  CR) is the follow-up enrichment tracked below. No operator-supplied
   `--replica-id` / `--model-id` on the demo path.
 
 ### Why this combination
@@ -101,6 +105,16 @@ Concretely:
   ceilings, cache-size hint) — not added in this PR. The sidecar passes the
   `kvevent-subscriber` binary's flag defaults and can be enriched in a follow-up when an
   operator needs the knobs.
+* **Auto-injecting `--engine-loads-grpc` (GetLoads load source)** — the subscriber binary
+  can now read engine load over the VllmEngine `GetLoads` gRPC RPC instead of scraping HTTP
+  `/metrics` (a vLLM engine served over gRPC exposes no `/metrics` endpoint at all), selected by the
+  `--engine-loads-grpc` flag. The sidecar renderer does **not** pass that flag yet, so
+  auto-injected sidecars keep the HTTP default. Deferred deliberately: `GetLoads` requires a
+  newer engine floor (a gRPC server that implements `GetLoads`; vLLM ≥ 0.19) than the engine currently
+  deployed, so auto-wiring it now would point every sidecar at an unimplemented RPC and flip
+  the load signal to "stale" (see the `StatsReporter` stale-escalation path). It gets wired
+  alongside the stats-path knobs above once the engine floor moves; until then it is opt-in
+  via the binary flag for gRPC-only engine deployments.
 * **TLS subscriber → policy-server** — separate ticket.
 * **HA / multi-server policy-server target** — separate ticket.
 * **Readiness gating on first KV event observed** — natural follow-up once this lands.

pkg/adapters/engine/doc.go

@@ -5,10 +5,12 @@
 // Two independent paths share one gRPC client:
 //   - Event path: ZMQ → EventBatch → ReportCacheState (prefix adds) +
 //     PublishEvent (removals/clears), debounced on a short window.
-//   - Stats path: HTTP GET against the engine's Prometheus /metrics →
-//     MetricsScraper → StatsReporter → ReportCacheState (stats-only
-//     CacheStateUpdate populating cacheMemoryBytes / hitRate / pressure on its
-//     own cadence, default ~10s).
+//   - Stats path: engine load → a statsScraper → StatsReporter → ReportCacheState
+//     (stats-only CacheStateUpdate populating cacheMemoryBytes / hitRate / pressure
+//     on its own cadence, default ~10s). The load source is selectable: an HTTP GET
+//     against the engine's Prometheus /metrics (MetricsScraper, the default), or the
+//     VllmEngine GetLoads gRPC RPC (GRPCLoadsScraper) for gRPC engines that expose
+//     no /metrics endpoint.
 //
 // Metadata only — never KV tensors or prompt text. Fail-soft on both paths:
 // neither a ZMQ drop nor a scrape failure can stall the engine. The package is