Repository: kubernetes-sigs/gateway-api-inference-extension

Issue:
EPP crashing during scale test
Running scale testing for the EPP

QPS: 10k
length of prompt: 10k char
16 sim server backends

The EPP repeatedly hit fatal crashes a dump from the logs of one of the crashed pods shows 

[logs.txt](https://github.com/user-attachments/files/22014230/logs.txt)

Discussion:
@kfswain: Looks to be a prefix cache plugin issue, will continue to RCA and take a look into fixes@kfswain: Looking at the logs, the `requestID`  is always set to `""`. 

And the plugin state uses the request ID as the key: https://github.com/kubernetes-sigs/gateway-api-inference-extension/blob/229c1225313523d1c526fe827067f0e4e9afeadc/pkg/epp/plugins/plugin_state.go#L63

there is a sync map for the outer plugin state, but keyed on the requestID, so its essentially all contending for a single entry. 

Sync should still not have lock contention.. will continue to dig@kfswain: This line: https://github.com/kubernetes-sigs/gateway-api-inference-extension/blob/229c1225313523d1c526fe827067f0e4e9afeadc/pkg/epp/scheduling/framework/plugins/multi/prefix/indexer.go#L66

Is also a potential issue, we are adding to the map without locks, this could very likely cause the `fatal error: concurrent map writes` Will try rerunning without the prefix plugin to see if that fixes@nirrozenbaum: cc: @vMaroon @kfirtoledo please have a look also if you have free cycles@nirrozenbaum: found the bug. the following code is in the PluginState, used by the Prefix plugin:
https://github.com/kubernetes-sigs/gateway-api-inference-extension/blob/dc82925be1d18857f24b4bf9059a20929fb3f75c/pkg/epp/plugins/plugin_state.go#L78-L91

since RequestID is always `""`, different requests are getting the same `stateMap` in line 81, and then try to write concurrently different values to the same map in line 88. 

RequestID should always be set. I will push a fix.@nirrozenbaum: opened a PR #1490 with a fix. 
I'd be happy to get an update after repeating the test with this fix that it's not reproducible.@nirrozenbaum: > This line:
> 
> [gateway-api-inference-extension/pkg/epp/scheduling/framework/plugins/multi/prefix/indexer.go](https://github.com/kubernetes-sigs/gateway-api-inference-extension/blob/229c1225313523d1c526fe827067f0e4e9afeadc/pkg/epp/scheduling/framework/plugins/multi/prefix/indexer.go#L66)
> 
> Line 66 in [229c122](/kubernetes-sigs/gateway-api-inference-extension/commit/229c1225313523d1c526fe827067f0e4e9afeadc)
> 
>  for _, hash := range hashes { 
> Is also a potential issue, we are adding to the map without locks, this could very likely cause the `fatal error: concurrent map writes` Will try rerunning without the prefix plugin to see if that fixes

This looks like yet another bug. not sure what we really save here with the unlocking before the for loop and locking again right after. for loop should be very fast that this seems like a premature optimization. It's a different bug fix than #1490. 
will push in another PR@nirrozenbaum: > > This line:
> > [gateway-api-inference-extension/pkg/epp/scheduling/framework/plugins/multi/prefix/indexer.go](https://github.com/kubernetes-sigs/gateway-api-inference-extension/blob/229c1225313523d1c526fe827067f0e4e9afeadc/pkg/epp/scheduling/framework/plugins/multi/prefix/indexer.go#L66)
> > Line 66 in [229c122](/kubernetes-sigs/gateway-api-inference-extension/commit/229c1225313523d1c526fe827067f0e4e9afeadc)
> > for _, hash := range hashes {
> > Is also a potential issue, we are adding to the map without locks, this could very likely cause the `fatal error: concurrent map writes` Will try rerunning without the prefix plugin to see if that fixes
> 
> This looks like yet another bug. not sure what we really save here with the unlocking before the for loop and locking again right after. for loop should be very fast that this seems like a premature optimization. It's a different bug fix than [#1490](https://github.com/kubernetes-sigs/gateway-api-inference-extension/pull/1490). will push in another PR

after further analysis, this function seems to be fine. cache.Add uses internally a lock to protect the LRU cache per pod.
not only that no need to lock it, if we lock we hit deadlock (was trying to play with that).
current implementation of this function seems correct, so hopefully the fix in #1490 solves the probelm.@nirrozenbaum: @kfswain out of curiosity, can you provide details about the gateway you were using?
I assume GKE?
I remember reading envoy always generates the `X-Request-Id` header which is kind of a standard. does GKE generate that header? the fix handles the case it doesn't, but asking for my understanding.
@kfswain: > found the bug. the following code is in the PluginState, used by the Prefix plugin:
> 
> [gateway-api-inference-extension/pkg/epp/plugins/plugin_state.go](https://github.com/kubernetes-sigs/gateway-api-inference-extension/blob/dc82925be1d18857f24b4bf9059a20929fb3f75c/pkg/epp/plugins/plugin_state.go#L78-L91)
> 
> Lines 78 to 91 in [dc82925](/kubernetes-sigs/gateway-api-inference-extension/commit/dc82925be1d18857f24b4bf9059a20929fb3f75c)
> 
>  func (s *PluginState) Write(requestID string, key StateKey, val StateData) { 
>  	s.requestToLastAccessTime.Store(requestID, time.Now()) 
>  	var stateData map[StateKey]StateData 
>  	stateMap, ok := s.storage.Load(requestID) 
>  	if ok { 
>  		stateData = stateMap.(map[StateKey]StateData) 
>  	} else { 
>  		stateData = map[StateKey]StateData{} 
>  	} 
>   
>  	stateData[key] = val 
>   
>  	s.storage.Store(requestID, stateData) 
>  } 
> since RequestID is always `""`, different requests are getting the same `stateMap` in line 81, and then try to write concurrently different values to the same map in line 88.
> 
> RequestID should always be set. I will push a fix.

All these use sync.Maps under the hood, so even with the same key I think there shouldnt be concurrent writes... Even if the value is yet another map@kfswain: > [@kfswain[<img alt="" width="15" height="15" src="chrome-extension://dlebflppeeemcdpidccbiblndppbmjmh/ospo-chrome-ext-logo.png">](https://teams.googleplex.com/kfswain@google.com)](https://github.com/kfswain) out of curiosity, can you provide details about the gateway you were using? I assume GKE? I remember reading envoy always generates the `X-Request-Id` header which is kind of a standard. does GKE generate that header? the fix handles the case it doesn't, but asking for my understanding.

Yeah, GKE GW, and we do use Envoy under the hood. I can ask internally about that
@nirrozenbaum: > All these use sync.Maps under the hood, so even with the same key I think there shouldnt be concurrent writes... Even if the value is yet another map

@kfswain no, the above is not correct. 
PluginState uses a sync map from StateKey (string) to StateData (map[string]any). 
when two concurrent go routines read the same requestID (in our case the “” requestiD), both get a pointer to a map (Not Sync.Map!). 
both are writing concurrently to this map and this is the error we’re seeing in the attached log.@kfswain: Ahh yeah yeah, you're right https://github.com/kubernetes-sigs/gateway-api-inference-extension/blob/229c1225313523d1c526fe827067f0e4e9afeadc/pkg/epp/plugins/plugin_state.go#L70 is the line. I think i was crossing wires, another func was making the map as a local variable. I think you're right@nirrozenbaum: actually line 88 is the problematic line :)
https://github.com/kubernetes-sigs/gateway-api-inference-extension/blob/229c1225313523d1c526fe827067f0e4e9afeadc/pkg/epp/plugins/plugin_state.go#L88

@kfswain: Related to: https://github.com/kubernetes-sigs/gateway-api-inference-extension/issues/1458@kfswain: > actually line 88 is the problematic line :)
> 
> [gateway-api-inference-extension/pkg/epp/plugins/plugin_state.go](https://github.com/kubernetes-sigs/gateway-api-inference-extension/blob/229c1225313523d1c526fe827067f0e4e9afeadc/pkg/epp/plugins/plugin_state.go#L88)
> 
> Line 88 in [229c122](/kubernetes-sigs/gateway-api-inference-extension/commit/229c1225313523d1c526fe827067f0e4e9afeadc)
> 
>  stateData[key] = val

yes, apologies, multitasking my brain into a soft mush 😭 @nirrozenbaum: @kfswain are you planning to repeat the scale tests? can you update if you’re still seeing the crash or bug resolved?@kfswain: Yes, definitely to both

Solution patch:
diff --git a/api/v1/inferencepool_types.go b/api/v1/inferencepool_types.go
index af94582..2c7b527 100644
--- a/api/v1/inferencepool_types.go
+++ b/api/v1/inferencepool_types.go
@@ -96,6 +96,7 @@ type Port struct {
 
 // EndpointPickerRef specifies a reference to an Endpoint Picker extension and its
 // associated configuration.
+// +kubebuilder:validation:XValidation:rule="self.kind != 'Service' || has(self.port)",message="port is required when kind is 'Service' or unspecified (defaults to 'Service')"
 type EndpointPickerRef struct {
 	// Group is the group of the referent API object. When unspecified, the default value
 	// is "", representing the Core API group.
@@ -125,13 +126,15 @@ type EndpointPickerRef struct {
 	// +required
 	Name ObjectName `json:"name,omitempty"`
 
-	// PortNumber is the port number of the Endpoint Picker extension service. When unspecified,
-	// implementations SHOULD infer a default value of 9002 when the kind field is "Service" or
-	// unspecified (defaults to "Service").
+	// Port is the port of the Endpoint Picker extension service.
+	//
+	// Port is required when the referent is a Kubernetes Service. In this
+	// case, the port number is the service port number, not the target port.
+	// For other resources, destination port might be derived from the referent
+	// resource or this field.
 	//
 	// +optional
-	//nolint:kubeapilinter // ignore kubeapilinter here as we want to use pointer as zero means all ports in convention, we don't make to use 0 to indicate not set.
-	PortNumber *PortNumber `json:"portNumber,omitempty"`
+	Port *Port `json:"port,omitempty"`
 
 	// FailureMode configures how the parent handles the case when the Endpoint Picker extension
 	// is non-responsive. When unspecified, defaults to "FailClose".
diff --git a/api/v1/zz_generated.deepcopy.go b/api/v1/zz_generated.deepcopy.go
index b42573c..3b00363 100644
--- a/api/v1/zz_generated.deepcopy.go
+++ b/api/v1/zz_generated.deepcopy.go
@@ -33,9 +33,9 @@ func (in *EndpointPickerRef) DeepCopyInto(out *EndpointPickerRef) {
 		*out = new(Group)
 		**out = **in
 	}
-	if in.PortNumber != nil {
-		in, out := &in.PortNumber, &out.PortNumber
-		*out = new(PortNumber)
+	if in.Port != nil {
+		in, out := &in.Port, &out.Port
+		*out = new(Port)
 		**out = **in
 	}
 }
diff --git a/apix/v1alpha2/inferencepool_conversion.go b/apix/v1alpha2/inferencepool_conversion.go
index 31ea997..01520dc 100644
--- a/apix/v1alpha2/inferencepool_conversion.go
+++ b/apix/v1alpha2/inferencepool_conversion.go
@@ -254,7 +254,7 @@ func convertExtensionRefToV1(src *Extension) (v1.EndpointPickerRef, error) {
 	}
 	endpointPickerRef.Name = v1.ObjectName(src.Name)
 	if src.PortNumber != nil {
-		endpointPickerRef.PortNumber = ptr.To(v1.PortNumber(*src.PortNumber))
+		endpointPickerRef.Port = ptr.To(v1.Port{Number: v1.PortNumber(*src.PortNumber)})
 	}
 	if src.FailureMode != nil {
 		endpointPickerRef.FailureMode = v1.EndpointPickerFailureMode(*src.FailureMode)
@@ -275,8 +275,8 @@ func convertEndpointPickerRefFromV1(src *v1.EndpointPickerRef) (Extension, error
 		extension.Kind = ptr.To(Kind(src.Kind))
 	}
 	extension.Name = ObjectName(src.Name)
-	if src.PortNumber != nil {
-		extension.PortNumber = ptr.To(PortNumber(*src.PortNumber))
+	if src.Port != nil {
+		extension.PortNumber = ptr.To(PortNumber(src.Port.Number))
 	}
 	if src.FailureMode != "" {
 		extension.FailureMode = ptr.To(ExtensionFailureMode(src.FailureMode))
diff --git a/client-go/applyconfiguration/api/v1/endpointpickerref.go b/client-go/applyconfiguration/api/v1/endpointpickerref.go
index 8a9991b..0d78862 100644
--- a/client-go/applyconfiguration/api/v1/endpointpickerref.go
+++ b/client-go/applyconfiguration/api/v1/endpointpickerref.go
@@ -28,7 +28,7 @@ type EndpointPickerRefApplyConfiguration struct {
 	Group       *apiv1.Group                     `json:"group,omitempty"`
 	Kind        *apiv1.Kind                      `json:"kind,omitempty"`
 	Name        *apiv1.ObjectName                `json:"name,omitempty"`
-	PortNumber  *apiv1.PortNumber                `json:"portNumber,omitempty"`
+	Port        *PortApplyConfiguration          `json:"port,omitempty"`
 	FailureMode *apiv1.EndpointPickerFailureMode `json:"failureMode,omitempty"`
 }
 
@@ -62,11 +62,11 @@ func (b *EndpointPickerRefApplyConfiguration) WithName(value apiv1.ObjectName) *
 	return b
 }
 
-// WithPortNumber sets the PortNumber field in the declarative configuration to the given value
+// WithPort sets the Port field in the declarative configuration to the given value
 // and returns the receiver, so that objects can be built by chaining "With" function invocations.
-// If called multiple times, the PortNumber field is set to the value of the last call.
-func (b *EndpointPickerRefApplyConfiguration) WithPortNumber(value apiv1.PortNumber) *EndpointPickerRefApplyConfiguration {
-	b.PortNumber = &value
+// If called multiple times, the Port field is set to the value of the last call.
+func (b *EndpointPickerRefApplyConfiguration) WithPort(value *PortApplyConfiguration) *EndpointPickerRefApplyConfiguration {
+	b.Port = value
 	return b
 }
 
diff --git a/config/crd/bases/inference.networking.k8s.io_inferencepools.yaml b/config/crd/bases/inference.networking.k8s.io_inferencepools.yaml
index 66cbcec..8f60199 100644
--- a/config/crd/bases/inference.networking.k8s.io_inferencepools.yaml
+++ b/config/crd/bases/inference.networking.k8s.io_inferencepools.yaml
@@ -89,18 +89,33 @@ spec:
                     maxLength: 253
                     minLength: 1
                     type: string
-                  portNumber:
+                  port:
                     description: |-
-                      PortNumber is the port number of the Endpoint Picker extension service. When unspecified,
-                      implementations SHOULD infer a default value of 9002 when the kind field is "Service" or
-                      unspecified (defaults to "Service").
-                    format: int32
-                    maximum: 65535
-                    minimum: 1
-                    type: integer
+                      Port is the port of the Endpoint Picker extension service.
+
+                      Port is required when the referent is a Kubernetes Service. In this
+                      case, the port number is the service port number, not the target port.
+                      For other resources, destination port might be derived from the referent
+                      resource or this field.
+                    properties:
+                      number:
+                        description: |-
+                          Number defines the port number to access the selected model server Pods.
+                          The number must be in the range 1 to 65535.
+                        format: int32
+                        maximum: 65535
+                        minimum: 1
+                        type: integer
+                    required:
+                    - number
+                    type: object
                 required:
                 - name
                 type: object
+                x-kubernetes-validations:
+                - message: port is required when kind is 'Service' or unspecified
+                    (defaults to 'Service')
+                  rule: self.kind != 'Service' || has(self.port)
               selector:
                 description: |-
                   Selector determines which Pods are members of this inference pool.
diff --git a/config/manifests/inferencepool-resources.yaml b/config/manifests/inferencepool-resources.yaml
index 7c31da9..ffe1965 100644
--- a/config/manifests/inferencepool-resources.yaml
+++ b/config/manifests/inferencepool-resources.yaml
@@ -15,6 +15,9 @@ spec:
       app: vllm-llama3-8b-instruct
   endpointPickerRef:
     name: vllm-llama3-8b-instruct-epp
+    kind: Service
+    port:
+      number: 9002
 ---
 apiVersion: v1
 kind: Service
diff --git a/conformance/conformance.go b/conformance/conformance.go
index 6aebd69..5d94197 100644
--- a/conformance/conformance.go
+++ b/conformance/conformance.go
@@ -134,12 +134,7 @@ func DefaultOptions(t *testing.T) confsuite.ConformanceOptions {
 		*confflags.ImplementationContact,
 	)
 
-	// Inference Extension Specific Report Fields
-	inferenceExtensionVersion := "v0.3.0"
-	_ = inferenceExtensionVersion // Avoid unused variable error until implemented
-
 	baseManifestsValue := "resources/base.yaml"
-
 	opts := confsuite.ConformanceOptions{
 		Client:               c,
 		ClientOptions:        clientOptions,
@@ -166,7 +161,6 @@ func DefaultOptions(t *testing.T) confsuite.ConformanceOptions {
 		// TODO: Add the inference extension specific fields to ConformanceOptions struct if needed,
 		// or handle them during report generation.
 		// GatewayAPIInferenceExtensionChannel: inferenceExtensionChannel,
-		// GatewayAPIInferenceExtensionVersion: inferenceExtensionVersion,
 	}
 
 	// Populate SupportedFeatures based on the GatewayLayerProfile.
@@ -197,6 +191,8 @@ func RunConformance(t *testing.T) {
 // RunConformanceWithOptions runs the Inference Extension conformance tests with specific options.
 func RunConformanceWithOptions(t *testing.T, opts confsuite.ConformanceOptions) {
 	t.Helper()
+	ctx := context.Background()
+
 	t.Logf("Running Inference Extension conformance tests with GatewayClass %s", opts.GatewayClassName)
 	logDebugf(t, opts.Debug, "RunConformanceWithOptions: BaseManifests path being used by opts: %q", opts.BaseManifests)
 
@@ -209,19 +205,28 @@ func RunConformanceWithOptions(t *testing.T, opts confsuite.ConformanceOptions)
 	cSuite, err := confsuite.NewConformanceTestSuite(opts)
 	require.NoError(t, err, "error initializing conformance suite")
 
-	SetupConformanceTestSuite(t, cSuite, opts, tests.ConformanceTests)
-
+	installedCRDs := &apiextensionsv1.CustomResourceDefinitionList{}
+	err = opts.Client.List(ctx, installedCRDs)
+	require.NoError(t, err, "error getting installedCRDs")
+	apiVersion, err := getGatewayInferenceExtentionVersion(installedCRDs.Items)
+	if err != nil {
+		if opts.AllowCRDsMismatch {
+			apiVersion = "UNDEFINED"
+		} else {
+			require.NoError(t, err, "error getting the gateway ineference extension version")
+		}
+	}
+	SetupConformanceTestSuite(ctx, t, cSuite, opts, tests.ConformanceTests)
 	t.Log("Running Inference Extension conformance tests against all registered tests")
 	err = cSuite.Run(t, tests.ConformanceTests)
 	require.NoError(t, err, "error running conformance tests")
 
-	// Generate and write the report if requested.
 	if opts.ReportOutputPath != "" {
 		t.Log("Generating Inference Extension conformance report")
 		report, err := cSuite.Report() // Use the existing report generation logic.
 		require.NoError(t, err, "error generating conformance report")
 		inferenceReport := GatewayAPIInferenceExtensionConformanceReport{
-			GatewayAPIInferenceExtensionVersion: version.BundleVersion,
+			GatewayAPIInferenceExtensionVersion: apiVersion,
 			ConformanceReport:                   *report,
 		}
 		err = inferenceReport.WriteReport(t.Logf, opts.ReportOutputPath)
@@ -229,7 +234,7 @@ func RunConformanceWithOptions(t *testing.T, opts confsuite.ConformanceOptions)
 	}
 }
 
-func SetupConformanceTestSuite(t *testing.T, suite *confsuite.ConformanceTestSuite, opts confsuite.ConformanceOptions, tests []confsuite.ConformanceTest) {
+func SetupConformanceTestSuite(ctx context.Context, t *testing.T, suite *confsuite.ConformanceTestSuite, opts confsuite.ConformanceOptions, tests []confsuite.ConformanceTest) {
 	suite.Applier.ManifestFS = suite.ManifestFS
 	if suite.RunTest != "" {
 		idx := slices.IndexFunc(tests, func(t confsuite.ConformanceTest) bool {
@@ -257,13 +262,31 @@ func SetupConformanceTestSuite(t *testing.T, suite *confsuite.ConformanceTestSui
 	}
 	apikubernetes.NamespacesMustBeReady(t, suite.Client, suite.TimeoutConfig, namespaces)
 
-	ensureGatewayAvailableAndReady(t, suite.Client, opts, resources.PrimaryGatewayNN)
-	ensureGatewayAvailableAndReady(t, suite.Client, opts, resources.SecondaryGatewayNN)
+	ensureGatewayAvailableAndReady(ctx, t, suite.Client, opts, resources.PrimaryGatewayNN)
+	ensureGatewayAvailableAndReady(ctx, t, suite.Client, opts, resources.SecondaryGatewayNN)
+}
+
+func getGatewayInferenceExtentionVersion(crds []apiextensionsv1.CustomResourceDefinition) (string, error) {
+	var inferenceVersion string
+	for _, crd := range crds {
+		v, okv := crd.Annotations[version.BundleVersionAnnotation]
+		if !okv {
+			continue
+		}
+		if inferenceVersion != "" && v != inferenceVersion {
+			return "", errors.New("multiple gateway api inference extension CRDs versions detected")
+		}
+		inferenceVersion = v
+	}
+	if inferenceVersion == "" {
+		return "", errors.New("no gateway api inference extension CRDs with the proper annotations found in the cluster")
+	}
+	return inferenceVersion, nil
 }
 
 // ensureGatewayAvailableAndReady polls for the specified Gateway to exist and become ready
 // with an address and programmed condition.
-func ensureGatewayAvailableAndReady(t *testing.T, k8sClient client.Client, opts confsuite.ConformanceOptions, gatewayNN types.NamespacedName) {
+func ensureGatewayAvailableAndReady(ctx context.Context, t *testing.T, k8sClient client.Client, opts confsuite.ConformanceOptions, gatewayNN types.NamespacedName) {
 	t.Helper()
 
 	t.Logf("Attempting to fetch Gateway %s/%s.", gatewayNN.Namespace, gatewayNN.Name)
@@ -277,7 +300,6 @@ func ensureGatewayAvailableAndReady(t *testing.T, k8sClient client.Client, opts
 
 	logDebugf(t, opts.Debug, "Waiting up to %v for Gateway object %s/%s to appear after manifest application...", waitForGatewayCreationTimeout, gatewayNN.Namespace, gatewayNN.Name)
 
-	ctx := context.TODO()
 	pollErr := wait.PollUntilContextTimeout(ctx, extTimeoutConf.GatewayObjectPollInterval, waitForGatewayCreationTimeout, true, func(pollCtx context.Context) (bool, error) {
 		fetchErr := k8sClient.Get(pollCtx, gatewayNN, gw)
 		if fetchErr == nil {
diff --git a/conformance/reports/v0.5.1/gateway/kubvernor/README.md b/conformance/reports/v0.5.1/gateway/kubvernor/README.md
new file mode 100644
index 0000000..62b6545
--- /dev/null
+++ b/conformance/reports/v0.5.1/gateway/kubvernor/README.md
@@ -0,0 +1,12 @@
+# Kubvernor Rust Gateway
+
+## Table of Contents
+
+| Extension Version Tested | Profile Tested | Implementation Version | Mode    | Report                                                                |
+|--------------------------|----------------|------------------------|---------|-----------------------------------------------------------------------|
+| v0.5.1                   | Gateway        | [0.1.1](https://github.com/kubvernor/kubvernor/releases/tag/0.1.1)                  | default | [Conformance report](./kubvernor-inference-conformance-output-0.1.1.yaml) |
+
+## Reproduce
+
+To reproduce Kubvernor conformance report follow [README](https://github.com/kubvernor/kubvernor/blob/0.1.1/README.md)
+
diff --git a/conformance/reports/v0.5.1/gateway/kubvernor/kubvernor-inference-conformance-output-0.1.1.yaml b/conformance/reports/v0.5.1/gateway/kubvernor/kubvernor-inference-conformance-output-0.1.1.yaml
new file mode 100644
index 0000000..1595f49
--- /dev/null
+++ b/conformance/reports/v0.5.1/gateway/kubvernor/kubvernor-inference-conformance-output-0.1.1.yaml
@@ -0,0 +1,23 @@
+GatewayAPIInferenceExtensionVersion: v0.5.1
+apiVersion: gateway.networking.k8s.io/v1
+date: "2025-08-25T15:42:29+01:00"
+gatewayAPIChannel: standard
+gatewayAPIVersion: v1.2.1
+implementation:
+  contact:
+    - https://github.com/kubvernor/kubvernor
+  organization: kubvernor
+  project: kubvernor
+  url: https://github.com/kubvernor/kubvernor
+  version: 0.1.1
+kind: ConformanceReport
+mode: default
+profiles:
+  - core:
+      result: success
+      statistics:
+        Failed: 0
+        Passed: 9
+        Skipped: 0
+    name: Gateway
+    summary: Core tests succeeded.
diff --git a/conformance/resources/base.yaml b/conformance/resources/base.yaml
index 4fd3b38..7f4a7fa 100644
--- a/conformance/resources/base.yaml
+++ b/conformance/resources/base.yaml
@@ -159,6 +159,8 @@ spec:
     - number: 3000
   endpointPickerRef:
     name: primary-endpoint-picker-svc
+    port:
+      number: 9002
 ---
 # --- Primary Conformance EPP service Definition ---
 apiVersion: v1
@@ -255,6 +257,8 @@ spec:
   endpointPickerRef:
     name: secondary-endpoint-picker-svc
     failureMode: FailOpen
+    port:
+      number: 9002
 ---
 # --- Secondary Conformance EPP service Definition ---
 apiVersion: v1
diff --git a/pkg/epp/handlers/server.go b/pkg/epp/handlers/server.go
index 89e09f9..6a5c116 100644
--- a/pkg/epp/handlers/server.go
+++ b/pkg/epp/handlers/server.go
@@ -26,6 +26,7 @@ import (
 	extProcPb "github.com/envoyproxy/go-control-plane/envoy/service/ext_proc/v3"
 	envoyTypePb "github.com/envoyproxy/go-control-plane/envoy/type/v3"
 	"github.com/go-logr/logr"
+	"github.com/google/uuid"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 	"sigs.k8s.io/controller-runtime/pkg/log"
@@ -186,11 +187,18 @@ func (s *StreamingServer) Process(srv extProcPb.ExternalProcessor_ProcessServer)
 
 		switch v := req.Request.(type) {
 		case *extProcPb.ProcessingRequest_RequestHeaders:
-			if requestID := requtil.ExtractHeaderValue(v, requtil.RequestIdHeaderKey); len(requestID) > 0 {
-				logger = logger.WithValues(requtil.RequestIdHeaderKey, requestID)
-				loggerTrace = logger.V(logutil.TRACE)
-				ctx = log.IntoContext(ctx, logger)
+			requestID := requtil.ExtractHeaderValue(v, requtil.RequestIdHeaderKey)
+			// request ID is a must for maintaining a state per request in plugins that hold internal state and use PluginState.
+			// if request id was not supplied as a header, we generate it ourselves.
+			if len(requestID) == 0 {
+				requestID = uuid.NewString()
+				loggerTrace.Info("RequestID header is not found in the request, generated a request id")
+				reqCtx.Request.Headers[requtil.RequestIdHeaderKey] = requestID // update in headers so director can consume it
 			}
+			logger = logger.WithValues(requtil.RequestIdHeaderKey, requestID)
+			loggerTrace = logger.V(logutil.TRACE)
+			ctx = log.IntoContext(ctx, logger)
+
 			err = s.HandleRequestHeaders(reqCtx, v)
 		case *extProcPb.ProcessingRequest_RequestBody:
 			loggerTrace.Info("Incoming body chunk", "EoS", v.RequestBody.EndOfStream)
diff --git a/pkg/epp/plugins/plugin_state.go b/pkg/epp/plugins/plugin_state.go
index 2da3557..0281900 100644
--- a/pkg/epp/plugins/plugin_state.go
+++ b/pkg/epp/plugins/plugin_state.go
@@ -51,7 +51,7 @@ func NewPluginState(ctx context.Context) *PluginState {
 // Note: PluginState uses a sync.Map to back the storage, because it is thread safe.
 // It's aimed to optimize for the "write once and read many times" scenarios.
 type PluginState struct {
-	// key: RequestID, value: map[StateKey]StateData
+	// key: RequestID, value: sync.Map[StateKey]StateData
 	storage sync.Map
 	// key: RequestID, value: time.Time
 	requestToLastAccessTime sync.Map
@@ -66,9 +66,9 @@ func (s *PluginState) Read(requestID string, key StateKey) (StateData, error) {
 		return nil, ErrNotFound
 	}
 
-	stateData := stateMap.(map[StateKey]StateData)
-	if value, ok := stateData[key]; ok {
-		return value, nil
+	stateData := stateMap.(*sync.Map)
+	if value, ok := stateData.Load(key); ok {
+		return value.(StateData), nil
 	}
 
 	return nil, ErrNotFound
@@ -77,15 +77,15 @@ func (s *PluginState) Read(requestID string, key StateKey) (StateData, error) {
 // Write stores the given "val" in PluginState with the given "key" in the context of the given "requestID".
 func (s *PluginState) Write(requestID string, key StateKey, val StateData) {
 	s.requestToLastAccessTime.Store(requestID, time.Now())
-	var stateData map[StateKey]StateData
+	var stateData *sync.Map
 	stateMap, ok := s.storage.Load(requestID)
 	if ok {
-		stateData = stateMap.(map[StateKey]StateData)
+		stateData = stateMap.(*sync.Map)
 	} else {
-		stateData = map[StateKey]StateData{}
+		stateData = &sync.Map{}
 	}
 
-	stateData[key] = val
+	stateData.Store(key, val)
 
 	s.storage.Store(requestID, stateData)
 }
diff --git a/pkg/epp/scheduling/framework/plugins/multi/prefix/indexer.go b/pkg/epp/scheduling/framework/plugins/multi/prefix/indexer.go
index 0a209a8..bd9e2c9 100644
--- a/pkg/epp/scheduling/framework/plugins/multi/prefix/indexer.go
+++ b/pkg/epp/scheduling/framework/plugins/multi/prefix/indexer.go
@@ -38,15 +38,15 @@ type indexer struct {
 }
 
 // newIndexer initializes an indexer with size limits and starts cache size reporting.
-func newIndexer(maxLRUSize int) *indexer {
-	ix := &indexer{
+func newIndexer(ctx context.Context, maxLRUSize int) *indexer {
+	indexer := &indexer{
 		hashToPods: make(map[BlockHash]podSet),
 		podToLRU:   make(map[ServerID]*lru.Cache[BlockHash, struct{}]),
 		maxLRUSize: maxLRUSize,
 	}
 
-	go ix.ReportLRUSize(time.Second)
-	return ix
+	go indexer.reportLRUSize(ctx, time.Second)
+	return indexer
 }
 
 // Add adds a list of prefix hashes to the cache, tied to the server.
@@ -111,8 +111,8 @@ func (i *indexer) makeEvictionFn(pod ServerID) func(BlockHash, struct{}) {
 	}
 }
 
-// ReportLRUSize starts a goroutine that periodically reports the LRU cache size metric.
-func (i *indexer) ReportLRUSize(interval time.Duration) {
+// reportLRUSize starts a goroutine that periodically reports the LRU cache size metric.
+func (i *indexer) reportLRUSize(ctx context.Context, interval time.Duration) {
 	ticker := time.NewTicker(interval)
 	defer ticker.Stop()
 	for range ticker.C {
@@ -137,7 +137,7 @@ func (i *indexer) ReportLRUSize(interval time.Duration) {
 		}
 
 		metrics.RecordPrefixCacheSize(int64(totalEntries))
-		log.FromContext(context.TODO()).V(logutil.TRACE).Info("Prefix cache state",
+		log.FromContext(ctx).V(logutil.TRACE).Info("Prefix cache state",
 			"total entries", totalEntries,
 			"# pods", numPods,
 			"avg entries per pod", avg,
diff --git a/pkg/epp/scheduling/framework/plugins/multi/prefix/plugin.go b/pkg/epp/scheduling/framework/plugins/multi/prefix/plugin.go
index 8c7b52c..ef33521 100644
--- a/pkg/epp/scheduling/framework/plugins/multi/prefix/plugin.go
+++ b/pkg/epp/scheduling/framework/plugins/multi/prefix/plugin.go
@@ -148,7 +148,7 @@ func New(ctx context.Context, config Config) *Plugin {
 	capacity := config.LRUCapacityPerServer
 	if capacity <= 0 {
 		capacity = DefaultLRUCapacityPerServer
-		log.FromContext(context.TODO()).V(logutil.DEFAULT).Info(
+		log.FromContext(ctx).V(logutil.DEFAULT).Info(
 			"LRUCapacityPerServer is not positive, using default value",
 			"defaultCapacity", DefaultLRUCapacityPerServer,
 		)
@@ -158,7 +158,7 @@ func New(ctx context.Context, config Config) *Plugin {
 		typedName:   plugins.TypedName{Type: PrefixCachePluginType, Name: PrefixCachePluginType},
 		config:      config,
 		pluginState: plugins.NewPluginState(ctx),
-		indexer:     newIndexer(capacity),
+		indexer:     newIndexer(ctx, capacity),
 	}
 }
 
diff --git a/site-src/reference/spec.md b/site-src/reference/spec.md
index 12ce3d7..11080fe 100644
--- a/site-src/reference/spec.md
+++ b/site-src/reference/spec.md
@@ -51,7 +51,7 @@ _Appears in:_
 | `group` _[Group](#group)_ | Group is the group of the referent API object. When unspecified, the default value<br />is "", representing the Core API group. |  | MaxLength: 253 <br />MinLength: 0 <br />Pattern: `^$\|^[a-z0-9]([-a-z0-9]*[a-z0-9])?(\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*$` <br /> |
 | `kind` _[Kind](#kind)_ | Kind is the Kubernetes resource kind of the referent.<br />Required if the referent is ambiguous, e.g. service with multiple ports.<br />Defaults to "Service" when not specified.<br />ExternalName services can refer to CNAME DNS records that may live<br />outside of the cluster and as such are difficult to reason about in<br />terms of conformance. They also may not be safe to forward to (see<br />CVE-2021-25740 for more information). Implementations MUST NOT<br />support ExternalName Services. | Service | MaxLength: 63 <br />MinLength: 1 <br />Pattern: `^[a-zA-Z]([-a-zA-Z0-9]*[a-zA-Z0-9])?$` <br /> |
 | `name` _[ObjectName](#objectname)_ | Name is the name of the referent API object. |  | MaxLength: 253 <br />MinLength: 1 <br /> |
-| `portNumber` _[PortNumber](#portnumber)_ | PortNumber is the port number of the Endpoint Picker extension service. When unspecified,<br />implementations SHOULD infer a default value of 9002 when the kind field is "Service" or<br />unspecified (defaults to "Service"). |  | Maximum: 65535 <br />Minimum: 1 <br /> |
+| `port` _[Port](#port)_ | Port is the port of the Endpoint Picker extension service.<br />Port is required when the referent is a Kubernetes Service. In this<br />case, the port number is the service port number, not the target port.<br />For other resources, destination port might be derived from the referent<br />resource or this field. |  |  |
 | `failureMode` _[EndpointPickerFailureMode](#endpointpickerfailuremode)_ | FailureMode configures how the parent handles the case when the Endpoint Picker extension<br />is non-responsive. When unspecified, defaults to "FailClose". | FailClose | Enum: [FailOpen FailClose] <br /> |
 
 
@@ -340,6 +340,7 @@ Port defines the network port that will be exposed by this InferencePool.
 
 
 _Appears in:_
+- [EndpointPickerRef](#endpointpickerref)
 - [InferencePoolSpec](#inferencepoolspec)
 
 | Field | Description | Default | Validation |
@@ -358,7 +359,6 @@ _Validation:_
 - Minimum: 1
 
 _Appears in:_
-- [EndpointPickerRef](#endpointpickerref)
 - [Port](#port)
 
 



Test patch:
diff --git a/apix/v1alpha2/inferencepool_conversion_test.go b/apix/v1alpha2/inferencepool_conversion_test.go
index a4147f2..64b222f 100644
--- a/apix/v1alpha2/inferencepool_conversion_test.go
+++ b/apix/v1alpha2/inferencepool_conversion_test.go
@@ -21,6 +21,7 @@ import (
 
 	"github.com/google/go-cmp/cmp"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
 	v1 "sigs.k8s.io/gateway-api-inference-extension/api/v1"
 )
 
@@ -34,7 +35,7 @@ var (
 	v1Group       = v1.Group("my-group")
 	v1Kind        = v1.Kind("MyKind")
 	v1FailureMode = v1.EndpointPickerFailureMode("Deny")
-	v1PortNumber  = v1.PortNumber(9000)
+	v1Port        = v1.Port{Number: 9000}
 )
 
 func TestInferencePoolConvertTo(t *testing.T) {
@@ -110,7 +111,7 @@ func TestInferencePoolConvertTo(t *testing.T) {
 						Group:       &v1Group,
 						Kind:        v1Kind,
 						Name:        "my-epp-service",
-						PortNumber:  &v1PortNumber,
+						Port:        &v1Port,
 						FailureMode: v1FailureMode,
 					},
 				},
@@ -433,7 +434,7 @@ func TestInferencePoolConvertFrom(t *testing.T) {
 						Group:       &v1Group,
 						Kind:        v1Kind,
 						Name:        "my-epp-service",
-						PortNumber:  &v1PortNumber,
+						Port:        &v1Port,
 						FailureMode: v1FailureMode,
 					},
 				},
diff --git a/conformance/tests/inferencepool_invalid_epp_service.yaml b/conformance/tests/inferencepool_invalid_epp_service.yaml
index cdc2048..b3dc70e 100644
--- a/conformance/tests/inferencepool_invalid_epp_service.yaml
+++ b/conformance/tests/inferencepool_invalid_epp_service.yaml
@@ -11,6 +11,9 @@ spec:
   - number: 3000
   endpointPickerRef:
     name: non-existent-epp-svc
+    kind: Service
+    port:
+      number: 9002
 ---
 apiVersion: gateway.networking.k8s.io/v1
 kind: HTTPRoute
diff --git a/pkg/epp/scheduling/framework/plugins/multi/prefix/indexer_test.go b/pkg/epp/scheduling/framework/plugins/multi/prefix/indexer_test.go
index a151121..6d4fcc5 100644
--- a/pkg/epp/scheduling/framework/plugins/multi/prefix/indexer_test.go
+++ b/pkg/epp/scheduling/framework/plugins/multi/prefix/indexer_test.go
@@ -16,13 +16,14 @@ limitations under the License.
 package prefix
 
 import (
+	"context"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
 )
 
 func TestIndexer_AddAndGet(t *testing.T) {
-	i := newIndexer(2)
+	i := newIndexer(context.Background(), 2)
 
 	hash1 := BlockHash(1)
 	server := ServerID{Namespace: "default", Name: "server1"}
diff --git a/pkg/epp/server/server_test.go b/pkg/epp/server/server_test.go
index dfcc4bb..aff6d46 100644
--- a/pkg/epp/server/server_test.go
+++ b/pkg/epp/server/server_test.go
@@ -42,13 +42,10 @@ const (
 )
 
 func TestServer(t *testing.T) {
-	theHeaderValue := "body"
-	requestHeader := "x-test"
-
 	expectedRequestHeaders := map[string]string{metadata.DestinationEndpointKey: fmt.Sprintf("%s:%d", podAddress, poolPort),
-		"Content-Length": "42", ":method": "POST", requestHeader: theHeaderValue}
-	expectedResponseHeaders := map[string]string{"x-went-into-resp-headers": "true", ":method": "POST", requestHeader: theHeaderValue}
-	expectedSchedulerHeaders := map[string]string{":method": "POST", requestHeader: theHeaderValue}
+		"Content-Length": "42", ":method": "POST", "x-test": "body", "x-request-id": "test-request-id"}
+	expectedResponseHeaders := map[string]string{"x-went-into-resp-headers": "true", ":method": "POST", "x-test": "body"}
+	expectedSchedulerHeaders := map[string]string{":method": "POST", "x-test": "body", "x-request-id": "test-request-id"}
 
 	t.Run("server", func(t *testing.T) {
 		model := testutil.MakeInferenceObjective("v1").
@@ -66,9 +63,10 @@ func TestServer(t *testing.T) {
 
 		// Send request headers - no response expected
 		headers := utils.BuildEnvoyGRPCHeaders(map[string]string{
-			requestHeader:              theHeaderValue,
+			"x-test":                   "body",
 			":method":                  "POST",
 			metadata.FlowFairnessIDKey: "a-very-interesting-fairness-id",
+			"x-request-id":             "test-request-id",
 		}, true)
 		request := &pb.ProcessingRequest{
 			Request: &pb.ProcessingRequest_RequestHeaders{
@@ -130,9 +128,6 @@ func TestServer(t *testing.T) {
 		}
 
 		// Check headers passed to the scheduler
-		if len(director.requestHeaders) != 2 {
-			t.Errorf("Incorrect number of request headers %d instead of 2", len(director.requestHeaders))
-		}
 		for expectedKey, expectedValue := range expectedSchedulerHeaders {
 			got, ok := director.requestHeaders[expectedKey]
 			if !ok {
@@ -143,7 +138,7 @@ func TestServer(t *testing.T) {
 		}
 
 		// Send response headers
-		headers = utils.BuildEnvoyGRPCHeaders(map[string]string{requestHeader: theHeaderValue, ":method": "POST"}, false)
+		headers = utils.BuildEnvoyGRPCHeaders(map[string]string{"x-test": "body", ":method": "POST"}, false)
 		request = &pb.ProcessingRequest{
 			Request: &pb.ProcessingRequest_ResponseHeaders{
 				ResponseHeaders: headers,
diff --git a/test/cel/inferencepool_test.go b/test/cel/inferencepool_test.go
new file mode 100644
index 0000000..8b3ba3e
--- /dev/null
+++ b/test/cel/inferencepool_test.go
@@ -0,0 +1,114 @@
+/*
+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 main
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	v1 "sigs.k8s.io/gateway-api-inference-extension/api/v1"
+)
+
+func TestValidateInferencePool(t *testing.T) {
+	ctx := context.Background()
+
+	// baseInferencePool is a valid, InferencePool resource.
+	baseInferencePool := v1.InferencePool{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "base-pool",
+			Namespace: metav1.NamespaceDefault,
+		},
+		Spec: v1.InferencePoolSpec{
+			TargetPorts: []v1.Port{
+				{Number: 8000},
+			},
+			Selector: v1.LabelSelector{
+				MatchLabels: map[v1.LabelKey]v1.LabelValue{
+					"app": "model-server",
+				},
+			},
+			EndpointPickerRef: v1.EndpointPickerRef{
+				Name: "epp",
+				Kind: "Service",
+				Port: ptrTo(v1.Port{Number: 9002}),
+			},
+		},
+	}
+
+	testCases := []struct {
+		desc       string
+		mutate     func(ip *v1.InferencePool)
+		wantErrors []string
+	}{
+		{
+			desc: "passes validation with a valid configuration",
+			mutate: func(ip *v1.InferencePool) {
+			},
+			wantErrors: nil,
+		},
+		{
+			desc: "fails validation when kind is unset (defaults to Service) and port is missing",
+			mutate: func(ip *v1.InferencePool) {
+				// By setting Kind to an empty string, we rely on the API server's default value of "Service".
+				ip.Spec.EndpointPickerRef.Kind = ""
+				ip.Spec.EndpointPickerRef.Port = nil
+			},
+			wantErrors: []string{"port is required when kind is 'Service' or unspecified (defaults to 'Service')"},
+		},
+		{
+			desc: "fails validation when kind is explicitly 'Service' and port is missing",
+			mutate: func(ip *v1.InferencePool) {
+				ip.Spec.EndpointPickerRef.Kind = "Service"
+				ip.Spec.EndpointPickerRef.Port = nil
+			},
+			wantErrors: []string{"port is required when kind is 'Service' or unspecified (defaults to 'Service')"},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.desc, func(t *testing.T) {
+			ip := baseInferencePool.DeepCopy()
+			// Use a unique name for each test case to avoid conflicts.
+			ip.Name = fmt.Sprintf("test-pool-%v", time.Now().UnixNano())
+
+			if tc.mutate != nil {
+				tc.mutate(ip)
+			}
+			err := k8sClient.Create(ctx, ip)
+
+			// This is a boolean XOR. It's true if one is true, but not both.
+			// It ensures that an error is returned if and only if we expect one.
+			if (len(tc.wantErrors) != 0) != (err != nil) {
+				t.Fatalf("Unexpected response while creating InferencePool; got err=\n%v\n; want error=%v", err, tc.wantErrors != nil)
+			}
+
+			// If we got an error, check that it contains the expected substrings.
+			var missingErrorStrings []string
+			for _, wantError := range tc.wantErrors {
+				if !celErrorStringMatches(err.Error(), wantError) {
+					missingErrorStrings = append(missingErrorStrings, wantError)
+				}
+			}
+			if len(missingErrorStrings) != 0 {
+				t.Errorf("Unexpected response while creating InferencePool; got err=\n%v\n; missing strings within error=%q", err, missingErrorStrings)
+			}
+		})
+	}
+}
diff --git a/test/cel/main_test.go b/test/cel/main_test.go
new file mode 100644
index 0000000..2758a56
--- /dev/null
+++ b/test/cel/main_test.go
@@ -0,0 +1,130 @@
+/*
+Copyright 2023 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 main
+
+import (
+	"fmt"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/clientcmd"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/envtest"
+
+	inferencev1 "sigs.k8s.io/gateway-api-inference-extension/api/v1"
+	inferencev1alpha2 "sigs.k8s.io/gateway-api-inference-extension/apix/v1alpha2"
+)
+
+var k8sClient client.Client
+
+func TestMain(m *testing.M) {
+	scheme := runtime.NewScheme()
+	var restConfig *rest.Config
+	var testEnv *envtest.Environment
+	var err error
+
+	_ = inferencev1.Install(scheme)
+	_ = inferencev1alpha2.Install(scheme)
+
+	// Add core APIs in case we refer secrets, services and configmaps
+	_ = corev1.AddToScheme(scheme)
+
+	// If one wants to use a local cluster, a KUBECONFIG envvar should be passed,
+	// otherwise testenv will be used
+	kubeconfig := os.Getenv("KUBECONFIG")
+	if kubeconfig != "" {
+		restConfig, err = clientcmd.BuildConfigFromFlags("", kubeconfig)
+		if err != nil {
+			panic(fmt.Sprintf("Failed to get restConfig from BuildConfigFromFlags: %v", err))
+		}
+	} else {
+		// The version used here MUST reflect the available versions at
+		// controller-runtime repo: https://raw.githubusercontent.com/kubernetes-sigs/controller-tools/HEAD/envtest-releases.yaml
+		// If the envvar is not passed, the latest GA will be used
+		k8sVersion := os.Getenv("K8S_VERSION")
+		testEnv = &envtest.Environment{
+			Scheme:                      scheme,
+			ErrorIfCRDPathMissing:       true,
+			DownloadBinaryAssets:        true,
+			DownloadBinaryAssetsVersion: k8sVersion,
+			CRDInstallOptions: envtest.CRDInstallOptions{
+				Paths: []string{
+					filepath.Join("..", "..", "config", "crd", "bases"),
+				},
+				CleanUpAfterUse: true,
+			},
+		}
+
+		restConfig, err = testEnv.Start()
+		if err != nil {
+			panic(fmt.Sprintf("Error initializing test environment: %v", err))
+		}
+	}
+
+	k8sClient, err = client.New(restConfig, client.Options{
+		Scheme: scheme,
+	})
+	if err != nil {
+		panic(fmt.Sprintf("Error initializing Kubernetes client: %v", err))
+	}
+
+	rc := m.Run()
+	if testEnv != nil {
+		if err := testEnv.Stop(); err != nil {
+			panic(fmt.Sprintf("error stopping test environment: %v", err))
+		}
+	}
+
+	os.Exit(rc)
+}
+
+func ptrTo[T any](a T) *T {
+	return &a
+}
+
+func celErrorStringMatches(got, want string) bool {
+	gotL := strings.ToLower(got)
+	wantL := strings.ToLower(want)
+
+	// Starting in k8s v1.32, some CEL error messages changed to use "more" instead of "longer"
+	alternativeWantL := strings.ReplaceAll(wantL, "longer", "more")
+
+	// Starting in k8s v1.28, CEL error messages stopped adding spec and status prefixes to path names
+	wantLAdjusted := strings.ReplaceAll(wantL, "spec.", "")
+	wantLAdjusted = strings.ReplaceAll(wantLAdjusted, "status.", "")
+	alternativeWantL = strings.ReplaceAll(alternativeWantL, "spec.", "")
+	alternativeWantL = strings.ReplaceAll(alternativeWantL, "status.", "")
+
+	// Enum validation messages changed in k8s v1.28:
+	// Before: must be one of ['Exact', 'PathPrefix', 'RegularExpression']
+	// After: supported values: "Exact", "PathPrefix", "RegularExpression"
+	if strings.Contains(wantLAdjusted, "must be one of") {
+		r := strings.NewReplacer(
+			"must be one of", "supported values:",
+			"[", "",
+			"]", "",
+			"'", "\"",
+		)
+		wantLAdjusted = r.Replace(wantLAdjusted)
+	}
+	return strings.Contains(gotL, wantL) || strings.Contains(gotL, wantLAdjusted) || strings.Contains(gotL, alternativeWantL)
+}
diff --git a/test/integration/epp/hermetic_test.go b/test/integration/epp/hermetic_test.go
index 384ac5e..a215adc 100644
--- a/test/integration/epp/hermetic_test.go
+++ b/test/integration/epp/hermetic_test.go
@@ -76,6 +76,7 @@ import (
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/framework/plugins/scorer"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/server"
 	logutil "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/logging"
+	requtil "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/request"
 	epptestutil "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/testing"
 	integrationutils "sigs.k8s.io/gateway-api-inference-extension/test/integration"
 )
@@ -187,6 +188,12 @@ func TestFullDuplexStreamed_KubeInferenceObjectiveRequest(t *testing.T) {
 						RawValue: []byte("mom"),
 					},
 				},
+				&configPb.HeaderValueOption{
+					Header: &configPb.HeaderValue{
+						Key:      requtil.RequestIdHeaderKey,
+						RawValue: []byte("test-request-id"),
+					},
+				},
 			),
 		},
 		{
@@ -250,6 +257,12 @@ func TestFullDuplexStreamed_KubeInferenceObjectiveRequest(t *testing.T) {
 						RawValue: []byte("mom"),
 					},
 				},
+				&configPb.HeaderValueOption{
+					Header: &configPb.HeaderValue{
+						Key:      requtil.RequestIdHeaderKey,
+						RawValue: []byte("test-request-id"),
+					},
+				},
 			),
 		},
 		{
@@ -279,6 +292,12 @@ func TestFullDuplexStreamed_KubeInferenceObjectiveRequest(t *testing.T) {
 						RawValue: []byte("mom"),
 					},
 				},
+				&configPb.HeaderValueOption{
+					Header: &configPb.HeaderValue{
+						Key:      requtil.RequestIdHeaderKey,
+						RawValue: []byte("test-request-id"),
+					},
+				},
 			),
 		},
 		{
@@ -308,6 +327,12 @@ func TestFullDuplexStreamed_KubeInferenceObjectiveRequest(t *testing.T) {
 						RawValue: []byte("mom"),
 					},
 				},
+				&configPb.HeaderValueOption{
+					Header: &configPb.HeaderValue{
+						Key:      requtil.RequestIdHeaderKey,
+						RawValue: []byte("test-request-id"),
+					},
+				},
 			),
 		},
 		{
@@ -330,6 +355,10 @@ func TestFullDuplexStreamed_KubeInferenceObjectiveRequest(t *testing.T) {
 										Key:   metadata.ModelNameRewriteKey,
 										Value: modelSheddableTarget,
 									},
+									{
+										Key:   requtil.RequestIdHeaderKey,
+										Value: "test-request-id",
+									},
 								},
 							},
 						},
@@ -368,6 +397,12 @@ func TestFullDuplexStreamed_KubeInferenceObjectiveRequest(t *testing.T) {
 						RawValue: []byte("mom"),
 					},
 				},
+				&configPb.HeaderValueOption{
+					Header: &configPb.HeaderValue{
+						Key:      requtil.RequestIdHeaderKey,
+						RawValue: []byte("test-request-id"),
+					},
+				},
 			),
 		},
 		{
@@ -394,6 +429,10 @@ func TestFullDuplexStreamed_KubeInferenceObjectiveRequest(t *testing.T) {
 										Key:   metadata.ModelNameRewriteKey,
 										Value: modelDirect,
 									},
+									{
+										Key:   requtil.RequestIdHeaderKey,
+										Value: "test-request-id",
+									},
 								},
 							},
 						},
@@ -432,6 +471,12 @@ func TestFullDuplexStreamed_KubeInferenceObjectiveRequest(t *testing.T) {
 						RawValue: []byte("mom"),
 					},
 				},
+				&configPb.HeaderValueOption{
+					Header: &configPb.HeaderValue{
+						Key:      requtil.RequestIdHeaderKey,
+						RawValue: []byte("test-request-id"),
+					},
+				},
 			),
 		},
 		// Response flow tests
@@ -778,6 +823,12 @@ func TestFullDuplexStreamed_KubeInferenceObjectiveRequest(t *testing.T) {
 						RawValue: []byte("mom"),
 					},
 				},
+				&configPb.HeaderValueOption{
+					Header: &configPb.HeaderValue{
+						Key:      requtil.RequestIdHeaderKey,
+						RawValue: []byte("test-request-id"),
+					},
+				},
 			),
 		},
 		{
@@ -811,6 +862,12 @@ func TestFullDuplexStreamed_KubeInferenceObjectiveRequest(t *testing.T) {
 						RawValue: []byte("mom"),
 					},
 				},
+				&configPb.HeaderValueOption{
+					Header: &configPb.HeaderValue{
+						Key:      requtil.RequestIdHeaderKey,
+						RawValue: []byte("test-request-id"),
+					},
+				},
 			),
 		},
 		{
diff --git a/test/integration/util.go b/test/integration/util.go
index 0052449..d78b76e 100644
--- a/test/integration/util.go
+++ b/test/integration/util.go
@@ -31,6 +31,7 @@ import (
 
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/metadata"
 	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 (
@@ -130,6 +131,10 @@ func GenerateStreamedRequestSet(logger logr.Logger, prompt, model, targetModel s
 							Key:   metadata.ModelNameRewriteKey,
 							Value: targetModel,
 						},
+						{
+							Key:   requtil.RequestIdHeaderKey,
+							Value: "test-request-id",
+						},
 					},
 				},
 			},
diff --git a/test/testdata/inferencepool-e2e.yaml b/test/testdata/inferencepool-e2e.yaml
index 9fa5059..b8d7fb6 100644
--- a/test/testdata/inferencepool-e2e.yaml
+++ b/test/testdata/inferencepool-e2e.yaml
@@ -12,6 +12,9 @@ spec:
   endpointPickerRef:
     name: vllm-llama3-8b-instruct-epp
     namespace: $E2E_NS
+    kind: Service
+    port:
+      number: 9002
 ---
 apiVersion: v1
 kind: Service
diff --git a/test/testdata/inferencepool-leader-election-e2e.yaml b/test/testdata/inferencepool-leader-election-e2e.yaml
index 8c24347..9ba5dcb 100644
--- a/test/testdata/inferencepool-leader-election-e2e.yaml
+++ b/test/testdata/inferencepool-leader-election-e2e.yaml
@@ -10,6 +10,9 @@ spec:
   endpointPickerRef:
     name: vllm-llama3-8b-instruct-epp
     namespace: $E2E_NS
+    kind: Service
+    port:
+      number: 9002
 ---
 apiVersion: v1
 kind: Service
diff --git a/test/testdata/inferencepool-with-model-hermetic.yaml b/test/testdata/inferencepool-with-model-hermetic.yaml
index 794b7fc..df6ae30 100644
--- a/test/testdata/inferencepool-with-model-hermetic.yaml
+++ b/test/testdata/inferencepool-with-model-hermetic.yaml
@@ -11,6 +11,9 @@ spec:
       app: vllm-llama3-8b-instruct-pool
   endpointPickerRef:
     name: epp
+    kind: Service
+    port:
+      number: 9002
 ---
 apiVersion: inference.networking.x-k8s.io/v1alpha2
 kind: InferenceObjective