Merge branch 'main' into kaushikmitra/latency-predictor-encoder-cache-feature

Author: kaushikmitr

cmd/epp/runner/runner.go

@@ -85,6 +85,7 @@ import (
 	"github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/flowcontrol/saturationdetector/utilization"
 	"github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/flowcontrol/usagelimits"
 	"github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/flowcontrol/usagelimits/priorityholdback"
+	"github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/flowcontrol/usagelimits/softreflectiveceiling"
 	"github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/requestcontrol/admitter/latencyslo"
 	"github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/requestcontrol/admitter/probabilisticadmitter"
 	reqdataprodprefix "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/requestcontrol/dataproducer/approximateprefix"
@@ -610,6 +611,7 @@ func (r *Runner) registerInTreePlugins() {
 	fwkplugin.Register(slodeadline.SLODeadlineOrderingPolicyType, slodeadline.SLODeadlineOrderingPolicyFactory)
 	fwkplugin.Register(usagelimits.StaticUsageLimitPolicyType, usagelimits.StaticPolicyFactory)
 	fwkplugin.Register(priorityholdback.PolicyType, priorityholdback.PolicyFactory)
+	fwkplugin.Register(softreflectiveceiling.PolicyType, softreflectiveceiling.Factory)
 
 	// Register Request level data producer plugins as defaults for their respective data keys.
 	fwkplugin.RegisterAsDefaultProducer(reqdataprodprefix.ApproxPrefixCachePluginType, reqdataprodprefix.ApproxPrefixCacheFactory, attrprefix.PrefixCacheMatchInfoDataKey)

cmd/epp/runner/test_runner.go

@@ -19,6 +19,7 @@ package runner
 import (
 	"context"
 	"encoding/json"
+	"net"
 
 	"k8s.io/client-go/rest"
 	ctrl "sigs.k8s.io/controller-runtime"
@@ -33,8 +34,9 @@ import (
 // NewTestRunnerSetup creates a setup runner dedicated for integration tests. When mockDataSource is
 // non-nil, its plugin type is registered as a factory that returns the provided instance, so the
 // YAML config can reference it by type name and the runner wires it into the endpoint factory
-// automatically.
-func NewTestRunnerSetup(ctx context.Context, cfg *rest.Config, opts *runserver.Options, mockDataSource fwkdl.DataSource) (*Runner, ctrl.Manager, datastore.Datastore, error) {
+// automatically. When grpcListener is non-nil the ext_proc server serves on it and
+// opts.GRPCPort is ignored.
+func NewTestRunnerSetup(ctx context.Context, cfg *rest.Config, opts *runserver.Options, mockDataSource fwkdl.DataSource, grpcListener net.Listener) (*Runner, ctrl.Manager, datastore.Datastore, error) {
 	runner := NewRunner()
 
 	if mockDataSource != nil {
@@ -51,13 +53,17 @@ func NewTestRunnerSetup(ctx context.Context, cfg *rest.Config, opts *runserver.O
 	managerOverrides := []func(*ctrl.Options){
 		func(o *ctrl.Options) {
 			o.Controller.SkipNameValidation = &skipNameValidation
+			// The kernel assigns the port, so the bind cannot lose a race to
+			// another listener the way a port number chosen in advance can.
+			o.Metrics.BindAddress = "127.0.0.1:0"
 		},
 	}
 
 	manager, ds, err := runner.setup(ctx, cfg, opts, managerOverrides)
 	if err != nil {
 		return runner, manager, ds, err
 	}
+	runner.serverRunner.GrpcListener = grpcListener
 
 	// Production runs the ext_proc and health servers on a context that outlives
 	// the manager (see Runner.runWithGracefulShutdown). Integration tests drive

config/charts/README.md

@@ -52,6 +52,7 @@ helm install my-standalone-router ./config/charts/llm-d-router-standalone \
   --set router.proxy.mode=service \
   --set router.proxy.replicas=3
 ```
+
 ---
 
 ### 2. Gateway Mode (`llm-d-router-gateway`)
@@ -513,17 +514,17 @@ httpRoute:
 
 ### Standalone Mode Configuration
 
-Configures EPP to run with a proxy (Envoy proxy or Agentgateway proxy) that intercepts and routes client traffic directly to model servers (exclusive to `llm-d-router-standalone`). The proxy runs as a sidecar in the EPP pod by default, or as a separate horizontally scalable service when `router.proxy.mode=service` (Envoy only).
+Configures EPP to run with a proxy (Envoy proxy or Agentgateway proxy) that intercepts and routes client traffic directly to model servers (exclusive to `llm-d-router-standalone`). The proxy runs as a sidecar in the EPP pod by default, or as a separate horizontally scalable service when `router.proxy.mode=service`.
 
 #### Proxy Sidecar Parameters
 
 | **Parameter Name** | **Description** | **Default** |
 | :--- | :--- | :--- |
 | `router.proxy.enabled` | Enable the proxy (Envoy or Agentgateway) in front of EPP. | `false` |
 | `router.proxy.proxyType` | Type of proxy. Options: `[envoy, agentgateway]`. | `envoy` |
-| `router.proxy.mode` | Proxy deployment mode. `sidecar` runs the proxy in the EPP pod; `service` runs it as its own horizontally scalable Deployment and Service reaching EPP over the EPP Service (Envoy only). | `sidecar` |
+| `router.proxy.mode` | Proxy deployment mode. `sidecar` runs the proxy in the EPP pod; `service` runs it as its own horizontally scalable Deployment and Service reaching EPP over the EPP Service. | `sidecar` |
 | `router.proxy.replicas` | Replica count for the proxy Deployment when `mode=service`. | `2` |
-| `router.proxy.failOpen` | Whether the proxy passes traffic through (fail-open) when EPP is unreachable. | `true` |
+| `router.proxy.failOpen` | Whether the proxy passes traffic through (fail-open) when EPP is unreachable. Applies to `proxyType=envoy` only; Agentgateway exposes no fail-open setting and rejects requests (fails closed) when EPP is unreachable. | `true` |
 | `router.proxy.name` | Name of the sidecar container. | `""` |
 | `router.proxy.image` | Sidecar container image. | `""` |
 | `router.proxy.imagePullPolicy` | Sidecar container image pull policy. | `IfNotPresent` |

config/charts/llm-d-router-standalone/templates/_validations.tpl

@@ -33,10 +33,6 @@ standalone validations
   {{- if not $proxy.enabled -}}
     {{- fail ".Values.router.proxy.enabled must be true when .Values.router.proxy.mode=service" -}}
   {{- end -}}
-  {{- $proxyType := default "envoy" ($proxy.proxyType | default "envoy") | lower -}}
-  {{- if ne $proxyType "envoy" -}}
-    {{- fail (printf ".Values.router.proxy.mode=service currently supports only proxyType=envoy, got %q" $proxyType) -}}
-  {{- end -}}
   {{- $hasHTTP := false -}}
   {{- range $servicePort := (.Values.router.extraServicePorts | default (list)) -}}
     {{- if eq (toString (index $servicePort "name")) "http" -}}
@@ -62,7 +58,7 @@ standalone validations
     {{- $listenerPort := include "llm-d-router.standaloneProxyListenerPort" . -}}
     {{- $flags := .Values.router.epp.flags | default dict -}}
     {{- if and (hasKey $flags "secure-serving") (ne (toString (index $flags "secure-serving")) "false") -}}
-      {{- fail ".Values.router.epp.flags.secure-serving must be false when proxyType=agentgateway; standalone agentgateway uses plaintext gRPC to EPP over localhost" -}}
+      {{- fail ".Values.router.epp.flags.secure-serving must be false when proxyType=agentgateway; standalone agentgateway uses plaintext gRPC to EPP (over loopback in sidecar mode, or the EPP Service in service mode)" -}}
     {{- end -}}
   {{- end -}}
 {{- end -}}

config/charts/routerlib/templates/_helpers.tpl

@@ -415,7 +415,7 @@ binds:
         policies:
           inferenceRouting:
             endpointPicker:
-              host: {{ printf "127.0.0.1:%v" (.Values.router.epp.extProcPort | default 9002) | quote }}
+              host: {{ printf "%s:%v" (include "llm-d-router.proxy.extProcHost" .) (.Values.router.epp.extProcPort | default 9002) | quote }}
             destinationMode: passthrough
 services:
 - name: {{ $serviceName | quote }}

config/charts/routerlib/templates/_proxy-deployment.yaml

@@ -1,5 +1,8 @@
 {{- define "llm-d-epp.proxy-deployment" -}}
 {{- $proxy := include "llm-d-router.proxy" . | fromYaml | default dict -}}
+{{- $proxyType := include "llm-d-router.proxyType" . | trim -}}
+{{- $proxyConfigMap := index $proxy "configMap" | default dict -}}
+{{- $proxyConfigMapName := index $proxyConfigMap "name" | default "" -}}
 {{- if and $proxy.enabled (eq (include "llm-d-router.proxyMode" .) "service") -}}
 apiVersion: apps/v1
 kind: Deployment
@@ -62,14 +65,31 @@ spec:
           resources:
           {{- toYaml . | nindent 12 }}
         {{- end }}
-        {{- with $proxy.volumeMounts }}
+        {{- if or (and (eq $proxyType "agentgateway") $proxyConfigMapName) $proxy.volumeMounts }}
           volumeMounts:
+          {{- if and (eq $proxyType "agentgateway") $proxyConfigMapName }}
+            - name: agentgateway-config-template
+              mountPath: /config
+              readOnly: true
+          {{- end }}
+          {{- with $proxy.volumeMounts }}
           {{- toYaml . | nindent 12 }}
+          {{- end }}
         {{- end }}
-      {{- with $proxy.volumes }}
+      {{- if or (and (eq $proxyType "agentgateway") $proxyConfigMapName) $proxy.volumes }}
       volumes:
+      {{- if and (eq $proxyType "agentgateway") $proxyConfigMapName }}
+        - name: agentgateway-config-template
+          configMap:
+            name: {{ $proxyConfigMapName }}
+            items:
+              - key: config.yaml
+                path: config.yaml
+      {{- end }}
+      {{- with $proxy.volumes }}
       {{- tpl (toYaml .) $ | nindent 8 }}
       {{- end }}
+      {{- end }}
 ---
 {{- end -}}
 {{- end -}}

deploy/config/soft-reflective-ceiling-epp-config.yaml

@@ -0,0 +1,26 @@
+# EPP configuration for the soft-reflective-ceiling usage limit policy.
+#
+# The policy applies a graduated, priority-aware dispatch ceiling: the
+# highest-priority band is never gated, and progressively lower-priority
+# bands are gated first as pool saturation rises. Gated items remain in
+# their queues; the ceiling controls dispatch, not admission.
+apiVersion: llm-d.ai/v1alpha1
+kind: EndpointPickerConfig
+featureGates:
+- flowControl
+plugins:
+- type: queue-scorer
+- type: kv-cache-utilization-scorer
+- type: prefix-cache-scorer
+- type: soft-reflective-ceiling-policy
+schedulingProfiles:
+- name: default
+  plugins:
+  - pluginRef: queue-scorer
+    weight: 2.0
+  - pluginRef: kv-cache-utilization-scorer
+    weight: 2.0
+  - pluginRef: prefix-cache-scorer
+    weight: 3.0
+flowControl:
+  usageLimitPolicyPluginRef: soft-reflective-ceiling-policy

hack/verify-helm.sh

@@ -116,6 +116,7 @@ test_cases_llm_d_router_standalone["latency-predictor"]="--set router.latencyPre
 test_cases_llm_d_router_standalone["llm-d-router-gateway"]="--set router.inferencePool.create=true --set router.modelServers.matchLabels.app=llm-instance-gateway"
 test_cases_llm_d_router_standalone["agentgateway"]="--set router.proxy.proxyType=agentgateway --set router.modelServers.matchLabels.app=llm-instance-gateway --set router.inferencePool.create=false --set 'router.modelServers.targetPorts[0].number=8000'"
 test_cases_llm_d_router_standalone["proxy-service"]="--set router.modelServers.matchLabels.app=llm-instance-gateway --set router.inferencePool.create=false --set router.proxy.mode=service --set router.proxy.replicas=3"
+test_cases_llm_d_router_standalone["agentgateway-service"]="--set router.proxy.proxyType=agentgateway --set router.proxy.mode=service --set router.modelServers.matchLabels.app=llm-instance-gateway --set router.inferencePool.create=false --set 'router.modelServers.targetPorts[0].number=8000'"
 test_cases_llm_d_router_standalone["triton"]="--set router.modelServers.type=triton --set router.modelServers.matchLabels.app=llm-instance-gateway --set router.inferencePool.create=false"
 
 
@@ -190,13 +191,6 @@ if eval "${mismatched_agentgateway_listener_target_port_command}"; then
   exit 1
 fi
 
-unsupported_proxy_service_agentgateway_command="${HELM} template ${SCRIPT_ROOT}/config/charts/llm-d-router-standalone --set router.modelServers.matchLabels.app=llm-instance-gateway --set router.inferencePool.create=false --set router.proxy.mode=service --set router.proxy.proxyType=agentgateway --set router.proxy.agentgateway.service.name=llm-instance-gateway --set 'router.proxy.agentgateway.service.ports[0]=8000' --set 'router.modelServers.targetPorts[0].number=8000' >/dev/null"
-echo "Executing: ${unsupported_proxy_service_agentgateway_command}"
-if eval "${unsupported_proxy_service_agentgateway_command}"; then
-  echo "Helm template unexpectedly succeeded for proxy mode=service with proxyType=agentgateway"
-  exit 1
-fi
-
 invalid_failopen_command="${HELM} template ${SCRIPT_ROOT}/config/charts/llm-d-router-standalone --set router.modelServers.matchLabels.app=llm-instance-gateway --set router.inferencePool.create=false --set router.proxy.failOpen=notabool >/dev/null"
 echo "Executing: ${invalid_failopen_command}"
 if eval "${invalid_failopen_command}"; then
@@ -294,3 +288,33 @@ if ! grep -q -- 'llm-d-router-proxy: proxy-svc-proxy' "${proxy_service_render_ou
   echo "Proxy service mode did not render the dedicated proxy selector label"
   exit 1
 fi
+
+echo "Verifying llm-d-router-standalone agentgateway in service mode reaches EPP over the Service..."
+agentgateway_service_mode_output="${TEMP_DIR}/llm-d-router-standalone-agentgateway-service-render.yaml"
+agentgateway_service_mode_command="${HELM} template ag-svc ${SCRIPT_ROOT}/config/charts/llm-d-router-standalone --namespace ag-ns --set router.proxy.proxyType=agentgateway --set router.proxy.mode=service --set router.modelServers.matchLabels.app=llm-instance-gateway --set router.inferencePool.create=false --set 'router.modelServers.targetPorts[0].number=8000' > ${agentgateway_service_mode_output}"
+echo "Executing: ${agentgateway_service_mode_command}"
+eval "${agentgateway_service_mode_command}"
+if ! grep -q -- 'name: ag-svc-proxy' "${agentgateway_service_mode_output}"; then
+  echo "Agentgateway service mode did not render the separate proxy Deployment/Service named ag-svc-proxy"
+  exit 1
+fi
+# The agentgateway container lives only in the standalone proxy Deployment, not the EPP pod.
+agentgateway_container_count=$(grep -c -- '- name: agentgateway-proxy$' "${agentgateway_service_mode_output}")
+if [ "${agentgateway_container_count}" -ne 1 ]; then
+  echo "Agentgateway service mode rendered ${agentgateway_container_count} agentgateway-proxy containers, expected exactly 1 (in the proxy Deployment)"
+  exit 1
+fi
+# endpointPicker must target the EPP Service FQDN, not loopback.
+if ! grep -q -- 'host: "ag-svc-epp.ag-ns.svc.cluster.local:9002"' "${agentgateway_service_mode_output}"; then
+  echo "Agentgateway service mode did not point endpointPicker.host at the EPP Service FQDN"
+  exit 1
+fi
+if grep -q -- 'host: "127.0.0.1:9002"' "${agentgateway_service_mode_output}"; then
+  echo "Agentgateway service mode still points endpointPicker.host at loopback"
+  exit 1
+fi
+# The agentgateway config must be mounted in the proxy Deployment.
+if ! grep -q -- 'agentgateway-config-template' "${agentgateway_service_mode_output}"; then
+  echo "Agentgateway service mode did not mount the agentgateway config in the proxy Deployment"
+  exit 1
+fi

internal/runnable/grpc.go

@@ -21,6 +21,7 @@ import (
 	"fmt"
 	"net"
 
+	"github.com/go-logr/logr"
 	"google.golang.org/grpc"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
@@ -34,33 +35,48 @@ func GRPCServer(name string, srv *grpc.Server, port int) manager.Runnable {
 		log := ctrl.Log.WithValues("name", name)
 		log.Info("gRPC server starting")
 
-		// Start listening.
 		lis, err := net.Listen("tcp", fmt.Sprintf(":%d", port))
 		if err != nil {
 			return fmt.Errorf("gRPC server failed to listen - %w", err)
 		}
+		log.Info("gRPC server listening", "port", lis.Addr().(*net.TCPAddr).Port)
+		return serveGRPC(ctx, log, srv, lis)
+	})
+}
 
-		log.Info("gRPC server listening", "port", port)
-
-		// Shutdown on context closed.
-		// Terminate the server on context closed.
-		// Make sure the goroutine does not leak.
-		doneCh := make(chan struct{})
-		defer close(doneCh)
-		go func() {
-			select {
-			case <-ctx.Done():
-				log.Info("gRPC server shutting down")
-				srv.GracefulStop()
-			case <-doneCh:
-			}
-		}()
+// GRPCServerOnListener converts the given gRPC server into a runnable serving on
+// an already-bound listener. Mirrors manager.Server's Listener field: reserving
+// the port in advance of the runnable starting removes the window in which
+// another process can take a port that was selected but not yet bound.
+// Takes ownership of the listener: grpc.Server.Serve closes it on return.
+func GRPCServerOnListener(name string, srv *grpc.Server, lis net.Listener) manager.Runnable {
+	return manager.RunnableFunc(func(ctx context.Context) error {
+		log := ctrl.Log.WithValues("name", name)
+		log.Info("gRPC server starting")
+		log.Info("gRPC server listening", "address", lis.Addr().String())
+		return serveGRPC(ctx, log, srv, lis)
+	})
+}
 
-		// Keep serving until terminated.
-		if err := srv.Serve(lis); err != nil && err != grpc.ErrServerStopped {
-			return fmt.Errorf("gRPC server failed - %w", err)
+func serveGRPC(ctx context.Context, log logr.Logger, srv *grpc.Server, lis net.Listener) error {
+	// Shutdown on context closed.
+	// Terminate the server on context closed.
+	// Make sure the goroutine does not leak.
+	doneCh := make(chan struct{})
+	defer close(doneCh)
+	go func() {
+		select {
+		case <-ctx.Done():
+			log.Info("gRPC server shutting down")
+			srv.GracefulStop()
+		case <-doneCh:
 		}
-		log.Info("gRPC server terminated")
-		return nil
-	})
+	}()
+
+	// Keep serving until terminated.
+	if err := srv.Serve(lis); err != nil && err != grpc.ErrServerStopped {
+		return fmt.Errorf("gRPC server failed - %w", err)
+	}
+	log.Info("gRPC server terminated")
+	return nil
 }

pkg/epp/flowcontrol/integration_helpers_test.go

@@ -143,6 +143,43 @@ type dispatchResult struct {
 	err     error
 }
 
+// fillQueue enqueues n requests of byteSize bytes that stay queued (the caller supplies a
+// detector that blocks dispatch) and waits until settled reports the fill is observable in
+// registry stats. Sending overflow traffic only after the fill is observable is what makes a
+// subsequent capacity rejection deterministic; a fixed enqueue stagger would race with goroutine
+// scheduling under CI load. The returned release func cancels the queued requests' context,
+// drains their results, and asserts none was rejected for capacity.
+func (h *integrationHarness) fillQueue(
+	t *testing.T, key flowcontrol.FlowKey, n int, byteSize uint64, settled func() bool,
+) (release func()) {
+	t.Helper()
+	queuedCtx, queuedCancel := context.WithCancel(h.ctx)
+	t.Cleanup(queuedCancel)
+	queued := make(chan dispatchResult, n)
+	for i := 0; i < n; i++ {
+		id := fmt.Sprintf("queued-req-%d", i)
+		go func() {
+			req := &testRequest{id: id, key: key, byteSize: byteSize, ttl: 5 * time.Minute}
+			outcome, err := h.fc.EnqueueAndWait(queuedCtx, req)
+			queued <- dispatchResult{id: id, outcome: outcome, err: err}
+		}()
+	}
+	require.Eventually(t, settled, time.Second, time.Millisecond,
+		"%d filling requests should be observably queued before proceeding", n)
+	return func() {
+		queuedCancel()
+		for i := 0; i < n; i++ {
+			select {
+			case r := <-queued:
+				require.NotEqual(t, fcTypes.QueueOutcomeRejectedCapacity, r.outcome,
+					"queued request %s should have been admitted, not capacity-rejected", r.id)
+			case <-time.After(5 * time.Second):
+				t.Fatalf("timed out waiting for queued request %d after cancellation", i)
+			}
+		}
+	}
+}
+
 // --- Producer and Detector ---
 
 type producerAndDetector struct {