working state, latency prediction

Author: BenjaminBraunDev

cmd/epp/runner/runner.go

@@ -19,6 +19,7 @@ package runner
 import (
 "context"
 "crypto/tls"
+"encoding/json"
 "errors"
 "flag"
 "fmt"
@@ -42,6 +43,7 @@ import (
 "sigs.k8s.io/controller-runtime/pkg/manager"
 "sigs.k8s.io/controller-runtime/pkg/metrics/filters"
 metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server"
+
 "sigs.k8s.io/gateway-api-inference-extension/internal/runnable"
 "sigs.k8s.io/gateway-api-inference-extension/pkg/common"
 backendmetrics "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/backend/metrics"
@@ -243,12 +245,6 @@ func (r *Runner) Run(ctx context.Context) error {
 runtime.SetBlockProfileRate(1)
 }
 
-err = r.parsePluginsConfiguration(ctx)
-if err != nil {
-setupLog.Error(err, "Failed to parse the configuration")
-return err
-}
-
 // ===================================================================
 // == Latency Predictor Integration
 // ===================================================================
@@ -267,8 +263,14 @@ func (r *Runner) Run(ctx context.Context) error {
 setupLog.Info("Latency predictor is disabled.")
 predictor = nil // This will be a true nil interface
 }
-
 // ===================================================================
+
+err = r.parsePluginsConfiguration(ctx, predictor, datastore)
+if err != nil {
+setupLog.Error(err, "Failed to parse the configuration")
+return err
+}
+
 // --- Initialize Core EPP Components ---
 if r.schedulerConfig == nil {
 err := errors.New("scheduler config must be set either by config api or through code")
@@ -282,10 +284,6 @@ func (r *Runner) Run(ctx context.Context) error {
 
 saturationDetector := saturationdetector.NewDetector(sdConfig, setupLog)
 
-if *enableLatencyPredictor {
-r.requestControlConfig.AddPlugins(slorequest.New(datastore, predictor))
-}
-
 director := requestcontrol.NewDirectorWithConfig(datastore, scheduler, saturationDetector, r.requestControlConfig)
 
 // --- Setup ExtProc Server Runner ---
@@ -315,11 +313,13 @@ func (r *Runner) Run(ctx context.Context) error {
 return err
 }
 
+// Register ext-proc server.
 if err := registerExtProcServer(mgr, serverRunner, ctrl.Log.WithName("ext-proc")); err != nil {
 return err
 }
 
 // --- Start Manager ---
+// This blocks until a signal is received.
 setupLog.Info("Controller manager starting")
 if err := mgr.Start(ctx); err != nil {
 setupLog.Error(err, "Error starting controller manager")
@@ -342,7 +342,18 @@ func (r *Runner) registerInTreePlugins() {
 plugins.Register(testfilter.HeaderBasedTestingFilterType, testfilter.HeaderBasedTestingFilterFactory)
 }
 
-func (r *Runner) parsePluginsConfiguration(ctx context.Context) error {
+func (r *Runner) registerLatencyPredictorPlugins(predictor latencypredictor.PredictorInterface, datastore datastore.Datastore) {
+// Register the SLO request tracker and scorer plugin, these plugins need access to the predictor and datastore.
+// We have to specify a custom factory function to create the plugins with the correct dependencies.
+plugins.Register(slorequest.SLORequestTrackerPluginType, func(name string, _ json.RawMessage, _ plugins.Handle) (plugins.Plugin, error) {
+return slorequest.New(predictor, datastore).WithName(name), nil
+})
+plugins.Register(scorer.SLOScorerPluginType, func(name string, _ json.RawMessage, _ plugins.Handle) (plugins.Plugin, error) {
+return scorer.NewSLOScorer(predictor, datastore).WithName(name), nil
+})
+}
+
+func (r *Runner) parsePluginsConfiguration(ctx context.Context, predictor latencypredictor.PredictorInterface, datastore datastore.Datastore) error {
 if *configText == "" && *configFile == "" {
 return nil // configuring through code, not through file
 }
@@ -361,6 +372,12 @@ func (r *Runner) parsePluginsConfiguration(ctx context.Context) error {
 }
 
 r.registerInTreePlugins()
+// If we have a latency predictor enabled and predictor and datastore are not nil,
+// register the latency predictor plugins (currently just the SLO scorer).
+if *enableLatencyPredictor && predictor != nil && datastore != nil {
+setupLog.Info("Registering latency predictor plugins")
+r.registerLatencyPredictorPlugins(predictor, datastore)
+}
 handle := plugins.NewEppHandle(ctx)
 config, err := loader.LoadConfig(configBytes, handle, logger)
 if err != nil {
@@ -477,13 +494,15 @@ func (r *Runner) parseConfiguration(ctx context.Context) error {
 }
 
 func initLogging(opts *zap.Options) {
+// Unless -zap-log-level is explicitly set, use -v
 useV := true
 flag.Visit(func(f *flag.Flag) {
 if f.Name == "zap-log-level" {
 useV = false
 }
 })
 if useV {
+// See https://pkg.go.dev/sigs.k8s.io/controller-runtime/pkg/log/zap#Options.Level
 lvl := -1 * (*logVerbosity)
 opts.Level = uberzap.NewAtomicLevelAt(zapcore.Level(int8(lvl)))
 }
@@ -543,11 +562,10 @@ func verifyMetricMapping(mapping backendmetrics.MetricMapping, logger logr.Logge
 if mapping.LoraRequestInfo == nil {
 logger.Info("Not scraping metric: LoraRequestInfo")
 }
-if mapping.TotalRunningRequests == nil {
-logger.Info("Not scraping metric: TotalRunningRequests")
-}
 }
 
+// setupPprofHandlers only implements the pre-defined profiles:
+// https://cs.opensource.google/go/go/+/refs/tags/go1.24.4:src/runtime/pprof/pprof.go;l=108
 func setupPprofHandlers(mgr ctrl.Manager) error {
 var err error
 profiles := []string{

config/manifests/gateway/gke/gcp-backend-policy.yaml

@@ -4,7 +4,7 @@ metadata:
  name: inferencepool-backend-policy
 spec:
  targetRef:
- group: "inference.networking.k8s.io"
+ group: "inference.networking.x-k8s.io"
  kind: InferencePool
  name: vllm-llama3-8b-instruct
  default:

config/manifests/gateway/gke/healthcheck.yaml

@@ -5,7 +5,7 @@ metadata:
  namespace: default
 spec:
  targetRef:
- group: "inference.networking.k8s.io"
+ group: "inference.networking.x-k8s.io"
  kind: InferencePool
  name: vllm-llama3-8b-instruct
  default:

config/manifests/gateway/gke/httproute.yaml

@@ -9,7 +9,7 @@ spec:
  name: inference-gateway
  rules:
  - backendRefs:
- - group: inference.networking.k8s.io
+ - group: inference.networking.x-k8s.io
  kind: InferencePool
  name: vllm-llama3-8b-instruct
  matches:

config/manifests/inferencepool-resources-lp.yaml

@@ -17,7 +17,6 @@ data:
  LATENCY_TPOT_SCALER_PATH: "/models/tpot_scaler.joblib"
  LATENCY_MODEL_TYPE: "xgboost"
  LATENCY_MAX_TRAINING_DATA_SIZE_PER_BUCKET: "5000"
-
 ---
 apiVersion: v1
 kind: ConfigMap
@@ -31,7 +30,6 @@ data:
  LOCAL_TPOT_MODEL_PATH: "/server_models/tpot.joblib"
  LOCAL_TTFT_SCALER_PATH: "/server_models/ttft_scaler.joblib"
  LOCAL_TPOT_SCALER_PATH: "/server_models/tpot_scaler.joblib"
-
 ---
 # --- InferencePool ---
 apiVersion: inference.networking.x-k8s.io/v1alpha2
@@ -44,7 +42,6 @@ spec:
  app: vllm-llama3-8b-instruct
  extensionRef:
  name: vllm-llama3-8b-instruct-epp
-
 ---
 # --- EPP Service ---
 apiVersion: v1
@@ -82,7 +79,12 @@ spec:
  port: 9090
  targetPort: 9090
  type: LoadBalancer 
-
+---
+apiVersion: v1
+kind: ServiceAccount
+metadata:
+ name: vllm-llama3-8b-instruct-epp
+ namespace: default
 ---
 # --- EPP Deployment with Individual Container Volumes ---
 apiVersion: apps/v1
@@ -102,6 +104,7 @@ spec:
  labels:
  app: vllm-llama3-8b-instruct-epp
  spec:
+ serviceAccountName: vllm-llama3-8b-instruct-epp
  # Conservatively, this timeout should mirror the longest grace period of the pods within the pool
  terminationGracePeriodSeconds: 130
  containers:
@@ -110,18 +113,22 @@ spec:
  image: us-central1-docker.pkg.dev/benjaminbraun-gke-dev/slo-routing/slo-routing-epp-exp
  imagePullPolicy: Always
  args:
- - -poolName
+ - -pool-name
  - "vllm-llama3-8b-instruct"
- - "-poolNamespace"
+ - "-pool-namespace"
  - "default"
+ - --pool-group
+ - "inference.networking.x-k8s.io"
  - -v
  - "4"
  - --zap-encoder
  - "json"
- - -grpcPort
+ - -grpc-port
  - "9002"
- - -grpcHealthPort
+ - -grpc-health-port
  - "9003"
+ - "--config-file"
+ - "/config/default-plugins.yaml"
  - "-enable-latency-predictor"
  env:
  - name: PREDICTION_SERVER_URL
@@ -147,6 +154,9 @@ spec:
  service: inference-extension
  initialDelaySeconds: 5
  periodSeconds: 10
+ volumeMounts:
+ - name: plugins-config-volume
+ mountPath: "/config"
  # Training Server Sidecar Container
  - name: training-server
  image: us-central1-docker.pkg.dev/benjaminbraun-gke-dev/slo-routing/latency_training:latest
@@ -337,23 +347,66 @@ spec:
  - name: prediction-server-3-storage
  emptyDir: 
  sizeLimit: "10Gi" # Dedicated volume for prediction server 3
-
+ - name: plugins-config-volume
+ configMap:
+ name: plugins-config
+---
+apiVersion: v1
+kind: ConfigMap
+metadata:
+ name: plugins-config
+ namespace: default
+data:
+ default-plugins.yaml: |
+ apiVersion: inference.networking.x-k8s.io/v1alpha1
+ kind: EndpointPickerConfig
+ plugins:
+ - type: prefix-cache-scorer
+ - type: slo-request-tracker
+ - type: slo-scorer
+ schedulingProfiles:
+ - name: default
+ plugins:
+ - pluginRef: prefix-cache-scorer
+ - pluginRef: slo-request-tracker
+ - pluginRef: slo-scorer
 ---
 # --- RBAC ---
-kind: ClusterRole
+kind: Role
+apiVersion: rbac.authorization.k8s.io/v1
+metadata:
+ name: pod-read
+ namespace: default
+rules:
+- apiGroups: [ "inference.networking.x-k8s.io" ]
+ resources: [ "inferenceobjectives", "inferencepools" ]
+ verbs: [ "get", "watch", "list" ]
+- apiGroups: [ "inference.networking.k8s.io" ]
+ resources: [ "inferencepools" ]
+ verbs: [ "get", "watch", "list" ]
+- apiGroups: [ "" ]
+ resources: [ "pods" ]
+ verbs: [ "get", "watch", "list" ]
+---
+kind: RoleBinding
 apiVersion: rbac.authorization.k8s.io/v1
 metadata:
+ name: pod-read-binding
+ namespace: default
+subjects:
+- kind: ServiceAccount
+ name: vllm-llama3-8b-instruct-epp
+ namespace: default
+roleRef:
+ apiGroup: rbac.authorization.k8s.io
+ kind: Role
  name: pod-read
+---
+kind: ClusterRole
+apiVersion: rbac.authorization.k8s.io/v1
+metadata:
+ name: auth-reviewer
 rules:
-- apiGroups: ["inference.networking.x-k8s.io"]
- resources: ["inferencepools"]
- verbs: ["get", "watch", "list"]
-- apiGroups: ["inference.networking.x-k8s.io"]
- resources: ["inferencemodels"]
- verbs: ["get", "watch", "list"]
-- apiGroups: [""]
- resources: ["pods"]
- verbs: ["get", "watch", "list"]
 - apiGroups:
  - authentication.k8s.io
  resources:
@@ -366,17 +419,16 @@ rules:
  - subjectaccessreviews
  verbs:
  - create
-
---- 
+---
 kind: ClusterRoleBinding
 apiVersion: rbac.authorization.k8s.io/v1
 metadata:
- name: pod-read-binding
+ name: auth-reviewer-binding
 subjects:
 - kind: ServiceAccount
- name: default
+ name: vllm-llama3-8b-instruct-epp
  namespace: default
 roleRef:
  apiGroup: rbac.authorization.k8s.io
  kind: ClusterRole
- name: pod-read
+ name: auth-reviewer

config/manifests/vllm/gpu-deployment.yaml

@@ -26,8 +26,6 @@ spec:
  - "8000"
  - "--max-num-seq"
  - "1024"
- - "--compilation-config"
- - "3"
  - "--enable-lora"
  - "--max-loras"
  - "2"
@@ -49,6 +47,8 @@ spec:
  key: token
  - name: VLLM_ALLOW_RUNTIME_LORA_UPDATING
  value: "true"
+ - name: LD_LIBRARY_PATH
+ value: "/usr/local/nvidia/lib64"
  ports:
  - containerPort: 8000
  name: http

pkg/epp/backend/metrics/metrics.go

@@ -37,9 +37,6 @@ const (
 LoraInfoMaxAdaptersMetricName = "max_lora"
 )
 
-// Updated to match the interface defined above - this implementation is now
-// in the main interface file and uses atomic.Value for thread safety
-
 type PodMetricsClientImpl struct {
 MetricMapping *MetricMapping
 ModelServerMetricsPort int32
@@ -100,15 +97,6 @@ func (p *PodMetricsClientImpl) promToPodMetrics(
 }
 }
 
-if p.MetricMapping.TotalRunningRequests != nil {
-queued, err := p.getMetric(metricFamilies, *p.MetricMapping.TotalRunningRequests)
-if err == nil {
-updated.RunningQueueSize = int(queued.GetGauge().GetValue())
-} else {
-errs = multierr.Append(errs, err)
-}
-}
-
 if p.MetricMapping.KVCacheUtilization != nil {
 usage, err := p.getMetric(metricFamilies, *p.MetricMapping.KVCacheUtilization)
 if err == nil {