subscriber

type AllBlocksCleared ¶

type AllBlocksCleared struct{}

AllBlocksCleared reports that the engine flushed its entire KV cache.

type BlockRemoved ¶

type BlockRemoved struct {
	BlockHashes [][]byte
}

BlockRemoved reports KV blocks that were evicted. BlockHashes are opaque bytes (see BlockStored).

type BlockStored ¶

type BlockStored struct {
	BlockHashes     [][]byte
	ParentBlockHash []byte
	TokenIDs        []uint32
	BlockSize       int32
	LoRAID          *int64
}

BlockStored reports KV blocks that became resident. BlockSize is the number of tokens per block; a block covers BlockSize tokens of the prefix.

BlockHashes are the engine's opaque per-block hashes (int variants normalized to 8-byte big-endian). They serve only as a stable per-block identity for the reverse map; the index key itself is our own content fingerprint derived from TokenIDs (see positional.go).

ParentBlockHash is the engine hash of the block preceding this event's first block — nil for a sequence root or when absent — normalized like BlockHashes. It lets the subscriber chain its rolling prefix hash across events.

TokenIDs are the flat token IDs of this event's blocks (BlockSize tokens per block, in block order). They are hashed in-pod to derive the fingerprint and never leave the pod.

LoRAID is the engine's LoRA adapter id for these blocks — nil when the event carries no adapter (msgpack nil, a truncated 5-field tuple, or a base-model request). It is the engine's INTERNAL integer id, assigned in adapter load order, so Config.AdapterID maps it to the stable adapter identity that becomes the index partition; it is never mixed into the content fingerprint.

type CacheTier ¶

type CacheTier string

CacheTier selects which vLLM cache-usage gauge feeds cache_memory_bytes. vLLM 0.21+ exposes a single unified `vllm:kv_cache_usage_perc`; older vLLM exposed `vllm:gpu_cache_usage_perc` and (on some builds) `vllm:cpu_cache_usage_perc`. "auto" probes that fallback chain — kv → gpu → cpu — so the scraper degrades across vLLM releases without operator action.

const (
	CacheTierAuto CacheTier = "auto"
	CacheTierKV   CacheTier = "kv"  // vLLM 0.21+: vllm:kv_cache_usage_perc
	CacheTierGPU  CacheTier = "gpu" // legacy: vllm:gpu_cache_usage_perc
	CacheTierCPU  CacheTier = "cpu" // legacy: vllm:cpu_cache_usage_perc
)

func ValidCacheTierNames() []CacheTier

func (t CacheTier) IsValid() bool

type Config ¶

type Config struct {
	// ReplicaID identifies the engine replica these events come from. Required;
	// it is the index key the server attributes prefixes/stats to.
	ReplicaID string
	// ModelID is the served model identifier. Required.
	ModelID string
	// TenantID is the tenant namespace. Optional (empty = shared/default).
	TenantID string
	// HashScheme names the engine's prefix-hash domain (e.g. "vllm"). Required
	// and non-empty: an empty scheme fails open server-side (dropped on ingest),
	// so reporting with one would silently lose the data.
	HashScheme string
	// AdapterNames maps the engine's INTERNAL LoRA id (BlockStored.lora_id) to
	// the stable adapter identity used as the index partition — the same string
	// a gateway puts in LookupRouteRequest.adapter_id.
	//
	// The mapping is needed because the engine's id is a load-order integer, not
	// a name: vLLM assigns lora_int_id from the ordered --lora-modules list (and
	// increments it for adapters loaded at runtime), so the SAME integer can mean
	// different adapters on two replicas whose load order differs. The index is
	// shared across replicas, so partitioning on the raw integer would re-create
	// the aliasing this partition exists to prevent — across replicas instead of
	// within one.
	//
	// Only a NIL lora_id (base model / no LoRA) uses the default ("") partition,
	// and that path needs no configuration. A non-nil lora_id with NO mapping is
	// FAIL-CLOSED at ingest — AdapterID reports ok=false and the subscriber drops
	// the event rather than partitioning it under a replica-local "lora:<id>",
	// which could place different adapters in one partition across replicas whose
	// --lora-modules order differs (the very cross-replica alias this partition
	// exists to prevent).
	//
	// So LoRA caching REQUIRES this map: supply it from the same --lora-modules
	// ordering the engine gets, and send the matching adapter_id from the gateway
	// (whatever identity is ingested here MUST match the gateway's query id — the
	// same producer/consumer agreement HashScheme requires). An unmapped adapter
	// is neither aliased nor cached — its prefixes get no hint (a miss, never a
	// wrong replica) until it is mapped. Nil/empty AdapterNames is correct only
	// for base-model / non-LoRA traffic.
	AdapterNames map[int64]string
}

Config is the per-engine-replica identity the subscriber stamps onto every report. vLLM's KV-cache events carry none of this, so the subscriber must be told which replica/model/tenant it watches and which hash scheme the engine uses (so the server keeps different engines' hashes in disjoint domains).

func (c Config) AdapterID(loraID *int64) (string, bool)

func (c Config) ClearedEvent(tsSeconds float64) *icpb.CacheEvent

func (c Config) EvictedEvent(prefixHash []byte, adapterID string, tsSeconds float64) *icpb.CacheEvent

func (c Config) StatsUpdate(tsUs int64, stats *icpb.ReplicaStats) *icpb.CacheStateUpdate

func (c Config) Update(tsUs int64, prefixes []*icpb.PrefixEntry) *icpb.CacheStateUpdate

func (c Config) Validate() error

type Event ¶

type Event interface {
	// contains filtered or unexported methods
}

Event is one decoded KV-cache event. The concrete types are BlockStored, BlockRemoved, and AllBlocksCleared.

type EventBatch ¶

type EventBatch struct {
	// TimestampSeconds is the engine's batch timestamp (Unix seconds, float).
	TimestampSeconds float64
	Events           []Event
}

EventBatch is one decoded vLLM event batch.

func DecodeEventBatch(payload []byte) (*EventBatch, error)

type EvictedPrefix ¶

type EvictedPrefix struct {
	PrefixHash []byte
	AdapterID  string
	TokenCount int32
}

EvictedPrefix is one index key the engine evicted, resolved back to the entry we derived for it: our content-fingerprint prefix hash, the adapter partition it lives in, and its cumulative token count. The adapter scopes the eviction to the right partition (the fingerprint is token-only, so one prefix hash can be live under several adapters at once); the token count lets the caller re-report the prefix at a colder tier (T2) when a paired L2 store still holds it, instead of only being able to delete it.

type MetricsScraper ¶

type MetricsScraper struct {
	// contains filtered or unexported fields
}

MetricsScraper polls an engine's Prometheus /metrics endpoint and projects the payload into a ReplicaStats, using the per-engine metricProfile selected by ScraperConfig.Scheme. It is fail-soft: any HTTP/parse error returns a zero ReplicaStats + the error so the caller logs and retries on the next tick.

For vLLM, hit rate is a sliding signal — the per-scrape delta of (prefix_cache_hits_total / prefix_cache_queries_total) — so the very first scrape returns hit_rate=0 (no delta available) and the previous values are cached on the scraper for the next tick. A counter reset (engine restart) resets the delta state too. SGLang reads a direct gauge instead (see hitRate).

func NewMetricsScraper(httpClient *http.Client, cfg ScraperConfig, logger *slog.Logger) *MetricsScraper

func (s *MetricsScraper) Scrape(ctx context.Context) (*icpb.ReplicaStats, error)

type Reporter ¶

type Reporter struct {
	// contains filtered or unexported fields
}

Reporter forwards decoded KV-cache events to the policy server over gRPC.

Adds (BlockStored, tagged tier T1) are accumulated and flushed on a short window — this debounces high engine event rates. Each flush sends one CacheStateUpdate on a fresh, time-bounded ReportCacheState stream. A BlockRemoved is handled one of two ways depending on whether the engine has an L2 offload tier (WithIgnoreBlockRemoved): with one, the block moved tiers, so it is re-reported at T2 through the same ReportCacheState add path; without one it is gone, forwarded as PREFIX_EVICTED via a time-bounded unary PublishEvent. AllBlocksCleared → ALL_CLEARED, also via PublishEvent. Adds, T2 downgrades, and evictions all carry the entry's resolved adapter partition (see positionalIndex).

Every RPC uses its own bounded context, so a stalled or unreachable server can never block the loop for longer than rpcTimeout — the cache is an optimization and must never stall the engine. Errors are logged and dropped (soft state); Run only returns on context cancellation or input close.

func NewReporter(client icpb.InferenceCacheClient, cfg Config, opts ...ReporterOption) *Reporter

func (r *Reporter) Run(ctx context.Context, in <-chan *EventBatch) error

type ReporterOption ¶

type ReporterOption func(*Reporter)

ReporterOption configures a Reporter.

func WithIgnoreBlockRemoved(b bool) ReporterOption

func WithLogger(l *slog.Logger) ReporterOption

func WithRPCTimeout(d time.Duration) ReporterOption

func WithWindow(d time.Duration) ReporterOption

type ScraperConfig ¶

type ScraperConfig struct {
	// URL is the engine's Prometheus /metrics endpoint.
	URL string
	// Scheme selects the per-engine metric profile ("vllm" | "sglang"). Empty
	// defaults to vLLM. It MUST match the engine the URL points at: a mismatch
	// makes every metric family unrecognized and Scrape fails loud rather than
	// fabricate zeros. Wired from the subscriber's --hash-scheme.
	Scheme string
	// Tier selects which cache-usage gauge feeds cache_memory_bytes.
	// Defaults to CacheTierAuto.
	Tier CacheTier
	// ModelLabel filters every series by the `model_name` Prometheus label
	// (the one vLLM stamps on its metrics). When non-empty, only series whose
	// label matches participate in the scrape — so a /metrics endpoint that
	// happens to expose multiple models cannot attribute another model's
	// usage/load/hit-rate to the replica this scraper is configured for. When
	// empty, every series is included (the legacy aggregate behaviour).
	ModelLabel string
	// CacheSizeBytes is the engine's total KV-cache capacity, used to map a
	// usage_perc gauge (0..1) to bytes. When zero, cache_memory_bytes is
	// emitted as 0 — the ranker doesn't consume this field, so an honest
	// "unknown" is preferred over a fabricated number.
	CacheSizeBytes int64
	// MaxConcurrencyCeiling is the denominator for the pressure proxy:
	//   pressure = clamp01((num_requests_running + num_requests_waiting) / ceiling).
	// 0 disables pressure (it stays 0).
	MaxConcurrencyCeiling int
	// Timeout bounds each scrape; defaults to defaultScrapeTimeout when <= 0.
	Timeout time.Duration
}

ScraperConfig tunes the metrics scraper. URL is required.

type StatsReporter ¶

type StatsReporter struct {
	// contains filtered or unexported fields
}

StatsReporter periodically scrapes engine /metrics and emits a stats-only CacheStateUpdate via ReportCacheState. It runs alongside the event Reporter (different cadence, different data source) and shares the same gRPC client.

Failure independence is load-bearing: a scrape failure (engine /metrics down, HTTP timeout, parse error) logs and skips the tick — it never blocks the event path or kills the subscriber. The two paths are independent failure domains.

func NewStatsReporter(client icpb.InferenceCacheClient, scraper statsScraper, cfg Config, opts ...StatsReporterOption) *StatsReporter

func (r *StatsReporter) Run(ctx context.Context) error

type StatsReporterOption ¶

type StatsReporterOption func(*StatsReporter)

StatsReporterOption configures a StatsReporter.

func WithStatsInterval(d time.Duration) StatsReporterOption

func WithStatsLogger(l *slog.Logger) StatsReporterOption

func WithStatsRPCTimeout(d time.Duration) StatsReporterOption

type Subscriber ¶

type Subscriber struct {
	// contains filtered or unexported fields
}

Subscriber reads a vLLM KV-cache event stream from a ZMQ PUB endpoint, decodes each batch, and emits it. It reconnects with backoff and never returns except on context cancellation (fail-soft — the engine is unaffected by our outages).

func NewSubscriber(endpoint, topic string, opts ...SubscriberOption) *Subscriber

func (s *Subscriber) Run(ctx context.Context, out chan<- *EventBatch) error

type SubscriberOption ¶

type SubscriberOption func(*Subscriber)

SubscriberOption configures a Subscriber.

func WithSubscriberBackoff(d time.Duration) SubscriberOption

func WithSubscriberLogger(l *slog.Logger) SubscriberOption