remove protocol specifics from cmd-line flags (#1296)

* remove protocol specifics from cmd-line flags

Signed-off-by: Nir Rozenbaum <[email protected]>

* address code review comments

Signed-off-by: Nir Rozenbaum <[email protected]>

* godoc

Signed-off-by: Nir Rozenbaum <[email protected]>

* rebase + move ObjectiveKey to metadata pkg

Signed-off-by: Nir Rozenbaum <[email protected]>

---------

Signed-off-by: Nir Rozenbaum <[email protected]>

Author: nirrozenbaum

cmd/epp/runner/runner.go

@@ -109,20 +109,6 @@ var (
 		"The path to the certificate for secure serving. The certificate and private key files "+
 			"are assumed to be named tls.crt and tls.key, respectively. If not set, and secureServing is enabled, "+
 			"then a self-signed certificate is used.")
-	// header/metadata flags
-	destinationEndpointHintKey = flag.String(
-		"destination-endpoint-hint-key",
-		runserver.DefaultDestinationEndpointHintKey,
-		"Header and response metadata key used by Envoy to route to the appropriate pod. This must match Envoy configuration.")
-	destinationEndpointHintMetadataNamespace = flag.String(
-		"destination-endpoint-hint-metadata-namespace",
-		runserver.DefaultDestinationEndpointHintMetadataNamespace,
-		"The key for the outer namespace struct in the metadata field of the extproc response that is used to wrap the"+
-			"target endpoint. If not set, then an outer namespace struct should not be created.")
-	fairnessIDHeaderKey = flag.String(
-		"fairness-id-header-key",
-		runserver.DefaultFairnessIDHeaderKey,
-		"The header key used to pass the fairness ID to be used in Flow Control.")
 	// metric flags
 	totalQueuedRequestsMetric = flag.String(
 		"total-queued-requests-metric",
@@ -201,7 +187,6 @@ func bindEnvToFlags() {
 		"MODEL_SERVER_METRICS_PATH":                       "model-server-metrics-path",
 		"MODEL_SERVER_METRICS_SCHEME":                     "model-server-metrics-scheme",
 		"MODEL_SERVER_METRICS_HTTPS_INSECURE_SKIP_VERIFY": "model-server-metrics-https-insecure-skip-verify",
-		"DESTINATION_ENDPOINT_HINT_KEY":                   "destination-endpoint-hint-key",
 		"POOL_NAME":                                       "pool-name",
 		"POOL_NAMESPACE":                                  "pool-namespace",
 		// durations & bools work too; flag.Set expects the *string* form
@@ -352,20 +337,17 @@ func (r *Runner) Run(ctx context.Context) error {
 
 	// --- Setup ExtProc Server Runner ---
 	serverRunner := &runserver.ExtProcServerRunner{
-		GrpcPort:                                 *grpcPort,
-		DestinationEndpointHintMetadataNamespace: *destinationEndpointHintMetadataNamespace,
-		DestinationEndpointHintKey:               *destinationEndpointHintKey,
-		FairnessIDHeaderKey:                      *fairnessIDHeaderKey,
-		PoolNamespacedName:                       poolNamespacedName,
-		PoolGKNN:                                 poolGKNN,
-		Datastore:                                datastore,
-		SecureServing:                            *secureServing,
-		HealthChecking:                           *healthChecking,
-		CertPath:                                 *certPath,
-		RefreshPrometheusMetricsInterval:         *refreshPrometheusMetricsInterval,
-		MetricsStalenessThreshold:                *metricsStalenessThreshold,
-		Director:                                 director,
-		SaturationDetector:                       saturationDetector,
+		GrpcPort:                         *grpcPort,
+		PoolNamespacedName:               poolNamespacedName,
+		PoolGKNN:                         poolGKNN,
+		Datastore:                        datastore,
+		SecureServing:                    *secureServing,
+		HealthChecking:                   *healthChecking,
+		CertPath:                         *certPath,
+		RefreshPrometheusMetricsInterval: *refreshPrometheusMetricsInterval,
+		MetricsStalenessThreshold:        *metricsStalenessThreshold,
+		Director:                         director,
+		SaturationDetector:               saturationDetector,
 	}
 	if err := serverRunner.SetupWithManager(ctx, mgr); err != nil {
 		setupLog.Error(err, "Failed to setup EPP controllers")

pkg/epp/handlers/request.go

@@ -24,13 +24,10 @@ import (
 	extProcPb "github.com/envoyproxy/go-control-plane/envoy/service/ext_proc/v3"
 	"google.golang.org/protobuf/types/known/structpb"
 
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/metadata"
 	errutil "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/error"
 )
 
-const (
-	ObjectiveKey = "x-gateway-inference-objective"
-)
-
 func (s *StreamingServer) HandleRequestHeaders(reqCtx *RequestContext, req *extProcPb.ProcessingRequest_RequestHeaders) error {
 	reqCtx.RequestReceivedTimestamp = time.Now()
 
@@ -61,13 +58,13 @@ func (s *StreamingServer) HandleRequestHeaders(reqCtx *RequestContext, req *extP
 			reqCtx.Request.Headers[header.Key] = header.Value
 		}
 		switch header.Key {
-		case s.fairnessIDHeaderKey:
+		case metadata.FlowFairnessIDKey:
 			reqCtx.FairnessID = reqCtx.Request.Headers[header.Key]
 			// remove the fairness ID header from the request headers,
 			// this is not data that should be manipulated or sent to the backend.
 			// It is only used for flow control.
 			delete(reqCtx.Request.Headers, header.Key)
-		case ObjectiveKey:
+		case metadata.ObjectiveKey:
 			reqCtx.ObjectiveKey = reqCtx.Request.Headers[header.Key]
 			// remove the objective header from the request headers,
 			// this is not data that should be manipulated or sent to the backend.
@@ -117,7 +114,7 @@ func (s *StreamingServer) generateHeaders(reqCtx *RequestContext) []*configPb.He
 	headers := []*configPb.HeaderValueOption{
 		{
 			Header: &configPb.HeaderValue{
-				Key:      s.destinationEndpointHintKey,
+				Key:      metadata.DestinationEndpointKey,
 				RawValue: []byte(reqCtx.TargetEndpoint),
 			},
 		},
@@ -146,27 +143,21 @@ func (s *StreamingServer) generateHeaders(reqCtx *RequestContext) []*configPb.He
 }
 
 func (s *StreamingServer) generateMetadata(endpoint string) *structpb.Struct {
-	targetEndpointValue := &structpb.Struct{
+	return &structpb.Struct{
 		Fields: map[string]*structpb.Value{
-			s.destinationEndpointHintKey: {
-				Kind: &structpb.Value_StringValue{
-					StringValue: endpoint,
-				},
-			},
-		},
-	}
-	dynamicMetadata := targetEndpointValue
-	if s.destinationEndpointHintMetadataNamespace != "" {
-		// If a namespace is defined, wrap the selected endpoint with that.
-		dynamicMetadata = &structpb.Struct{
-			Fields: map[string]*structpb.Value{
-				s.destinationEndpointHintMetadataNamespace: {
-					Kind: &structpb.Value_StructValue{
-						StructValue: targetEndpointValue,
+			metadata.DestinationEndpointNamespace: {
+				Kind: &structpb.Value_StructValue{
+					StructValue: &structpb.Struct{
+						Fields: map[string]*structpb.Value{
+							metadata.DestinationEndpointKey: {
+								Kind: &structpb.Value_StringValue{
+									StringValue: endpoint,
+								},
+							},
+						},
 					},
 				},
 			},
-		}
+		},
 	}
-	return dynamicMetadata
 }

pkg/epp/handlers/request_test.go

@@ -21,15 +21,14 @@ import (
 
 	configPb "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
 	extProcPb "github.com/envoyproxy/go-control-plane/envoy/service/ext_proc/v3"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/metadata"
 )
 
 func TestHandleRequestHeaders(t *testing.T) {
 	t.Parallel()
 
 	// Setup a mock server and request context
-	server := &StreamingServer{
-		fairnessIDHeaderKey: "test-fairness-id",
-	}
+	server := &StreamingServer{}
 
 	reqCtx := &RequestContext{
 		Request: &Request{
@@ -46,7 +45,7 @@ func TestHandleRequestHeaders(t *testing.T) {
 						Value: "test-value",
 					},
 					{
-						Key:   "test-fairness-id",
+						Key:   metadata.FlowFairnessIDKey,
 						Value: "test-fairness-id-value",
 					},
 				},
@@ -63,7 +62,7 @@ func TestHandleRequestHeaders(t *testing.T) {
 	if reqCtx.FairnessID != "test-fairness-id-value" {
 		t.Errorf("expected fairness ID to be 'test-fairness-id-value', got %s", reqCtx.FairnessID)
 	}
-	if reqCtx.Request.Headers["test-fairness-id"] == "test-fairness-id-value" {
+	if reqCtx.Request.Headers[metadata.FlowFairnessIDKey] == "test-fairness-id-value" {
 		t.Errorf("expected fairness ID header to be removed from request headers, but it was not")
 	}
 }

pkg/epp/handlers/server.go

@@ -44,13 +44,10 @@ const (
 	bodyByteLimit = 62000
 )
 
-func NewStreamingServer(destinationEndpointHintMetadataNamespace, destinationEndpointHintKey, fairnessIDHeaderKey string, datastore Datastore, director Director) *StreamingServer {
+func NewStreamingServer(datastore Datastore, director Director) *StreamingServer {
 	return &StreamingServer{
-		destinationEndpointHintMetadataNamespace: destinationEndpointHintMetadataNamespace,
-		destinationEndpointHintKey:               destinationEndpointHintKey,
-		fairnessIDHeaderKey:                      fairnessIDHeaderKey,
-		director:                                 director,
-		datastore:                                datastore,
+		director:  director,
+		datastore: datastore,
 	}
 }
 
@@ -67,15 +64,8 @@ type Datastore interface {
 // Server implements the Envoy external processing server.
 // https://www.envoyproxy.io/docs/envoy/latest/api-v3/service/ext_proc/v3/external_processor.proto
 type StreamingServer struct {
-	// The key of the header to specify the target pod address. This value needs to match Envoy
-	// configuration.
-	destinationEndpointHintKey string
-	// The key acting as the outer namespace struct in the metadata extproc response to communicate
-	// back the picked endpoints.
-	destinationEndpointHintMetadataNamespace string
-	fairnessIDHeaderKey                      string
-	datastore                                Datastore
-	director                                 Director
+	datastore Datastore
+	director  Director
 }
 
 // RequestContext stores context information during the life time of an HTTP request.

pkg/epp/metadata/consts.go

@@ -0,0 +1,33 @@
+/*
+Copyright 2025 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package metadata
+
+const (
+	// SubsetFilterNamespace is the key for the outer namespace struct in the metadata field of the extproc request that is used to wrap the subset filter.
+	SubsetFilterNamespace = "envoy.lb.subset_hint"
+	// SubsetFilterKey is the metadata key used by Envoy to specify an array candidate pods for serving the request.
+	// If not specified, all the pods that are associated with the pool are candidates.
+	SubsetFilterKey = "x-gateway-destination-endpoint-subset"
+	// DestinationEndpointNamespace is the key for the outer namespace struct in the metadata field of the extproc response that is used to wrap the target endpoint.
+	DestinationEndpointNamespace = "envoy.lb"
+	// DestinationEndpointKey is the header and response metadata key used by Envoy to route to the appropriate pod.
+	DestinationEndpointKey = "x-gateway-destination-endpoint"
+	// FlowFairnessIDKey is the header key used to pass the fairness ID to be used in Flow Control.
+	FlowFairnessIDKey = "x-gateway-inference-fairness-id"
+	// ObjectiveKey is the header key used to specify the objective of an incoming request.
+	ObjectiveKey = "x-gateway-inference-objective"
+)

pkg/epp/requestcontrol/director.go

@@ -35,18 +35,14 @@ import (
 	backendmetrics "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/backend/metrics"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/datastore"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/handlers"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/metadata"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/metrics"
 	schedulingtypes "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/types"
 	errutil "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/error"
 	logutil "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/logging"
 	requtil "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/request"
 )
 
-const (
-	subsetHintNamespace = "envoy.lb.subset_hint"
-	subsetHintKey       = "x-gateway-destination-endpoint-subset"
-)
-
 // Scheduler defines the interface required by the Director for scheduling.
 type Scheduler interface {
 	Schedule(ctx context.Context, request *schedulingtypes.LLMRequest, candidatePods []schedulingtypes.Pod) (result *schedulingtypes.SchedulingResult, err error)
@@ -196,13 +192,13 @@ func (d *Director) admitRequest(ctx context.Context, requestCriticality v1alpha2
 func (d *Director) getCandidatePodsForScheduling(ctx context.Context, requestMetadata map[string]any) []schedulingtypes.Pod {
 	loggerTrace := log.FromContext(ctx).V(logutil.TRACE)
 
-	subsetMap, found := requestMetadata[subsetHintNamespace].(map[string]any)
+	subsetMap, found := requestMetadata[metadata.SubsetFilterNamespace].(map[string]any)
 	if !found {
 		return d.toSchedulerPodMetrics(d.datastore.PodList(backendmetrics.AllPodPredicate))
 	}
 
 	// Check if endpoint key is present in the subset map and ensure there is at least one value
-	endpointSubsetList, found := subsetMap[subsetHintKey].([]any)
+	endpointSubsetList, found := subsetMap[metadata.SubsetFilterKey].([]any)
 	if !found {
 		return d.toSchedulerPodMetrics(d.datastore.PodList(backendmetrics.AllPodPredicate))
 	} else if len(endpointSubsetList) == 0 {

pkg/epp/requestcontrol/director_test.go

@@ -39,6 +39,7 @@ import (
 	backendmetrics "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/backend/metrics"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/datastore"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/handlers"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/metadata"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/plugins"
 	schedulingtypes "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/types"
 	errutil "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/error"
@@ -435,8 +436,8 @@ func TestDirector_HandleRequest(t *testing.T) {
 func TestGetCandidatePodsForScheduling(t *testing.T) {
 	var makeFilterMetadata = func(data []any) map[string]any {
 		return map[string]any{
-			"envoy.lb.subset_hint": map[string]any{
-				"x-gateway-destination-endpoint-subset": data,
+			metadata.SubsetFilterNamespace: map[string]any{
+				metadata.SubsetFilterKey: data,
 			},
 		}
 	}
@@ -493,7 +494,7 @@ func TestGetCandidatePodsForScheduling(t *testing.T) {
 		},
 		{
 			name:     "SubsetFilter, namespace present filter not present — return all pods",
-			metadata: map[string]any{"envoy.lb.subset_hint": map[string]any{}},
+			metadata: map[string]any{metadata.SubsetFilterNamespace: map[string]any{}},
 			output: []schedulingtypes.Pod{
 				&schedulingtypes.PodMetrics{
 					Pod:          outputPod1,