[pluggable parser] add AppProtocol validation in the InferencePool Reconcile loop

cmd/epp/runner/health.go

@@ -27,15 +27,18 @@ import (
 	healthPb "google.golang.org/grpc/health/grpc_health_v1"
 	"google.golang.org/grpc/status"
 
+	v1 "sigs.k8s.io/gateway-api-inference-extension/api/v1"
 	logutil "sigs.k8s.io/gateway-api-inference-extension/pkg/common/observability/logging"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/datastore"
+	fwkrh "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/interface/requesthandling"
 )
 
 type healthServer struct {
 	logger                logr.Logger
 	datastore             datastore.Datastore
 	isLeader              *atomic.Bool
 	leaderElectionEnabled bool
+	supporter             fwkrh.AppProtocolSupporter
 }
 
 const (
@@ -45,11 +48,12 @@ const (
 
 func (s *healthServer) Check(ctx context.Context, in *healthPb.HealthCheckRequest) (*healthPb.HealthCheckResponse, error) {
 	isLive := s.datastore.PoolHasSynced()
+	protocolMatches := s.checkProtocolSupport(isLive)
 
 	// If leader election is disabled, use current logic: all checks are based on whether the pool has synced.
 	if !s.leaderElectionEnabled {
-		if !isLive {
-			s.logger.V(logutil.DEFAULT).Info("gRPC health check not serving (leader election disabled)", "service", in.Service)
+		if !isLive || !protocolMatches {
+			s.logger.V(logutil.DEFAULT).Info("gRPC health check not serving (leader election disabled)", "service", in.Service, "isLive", isLive, "protocolMatches", protocolMatches)
 			return &healthPb.HealthCheckResponse{Status: healthPb.HealthCheckResponse_NOT_SERVING}, nil
 		}
 		s.logger.V(logutil.TRACE).Info("gRPC health check serving (leader election disabled)", "service", in.Service)
@@ -64,12 +68,12 @@ func (s *healthServer) Check(ctx context.Context, in *healthPb.HealthCheckReques
 	switch in.Service {
 	case ReadinessCheckService:
 		checkName = "readiness"
-		isPassing = isLive && s.isLeader.Load()
+		isPassing = isLive && s.isLeader.Load() && protocolMatches
 	case "": // Handle overall server health for load balancers that use an empty service name.
 		checkName = "empty service name (considered as overall health)"
 		// The overall health for a load balancer should reflect readiness to accept traffic,
 		// which is true only for the leader pod that has synced its data.
-		isPassing = isLive && s.isLeader.Load()
+		isPassing = isLive && s.isLeader.Load() && protocolMatches
 	case LivenessCheckService:
 		checkName = "liveness"
 		// Any pod that is running and can respond to this gRPC check is considered "live".
@@ -79,7 +83,7 @@ func (s *healthServer) Check(ctx context.Context, in *healthPb.HealthCheckReques
 	case extProcPb.ExternalProcessor_ServiceDesc.ServiceName:
 		// The main service is considered ready only on the leader.
 		checkName = "ext_proc"
-		isPassing = isLive && s.isLeader.Load()
+		isPassing = isLive && s.isLeader.Load() && protocolMatches
 	default:
 		s.logger.V(logutil.DEFAULT).Info("gRPC health check requested unknown service", "available-services", []string{LivenessCheckService, ReadinessCheckService, extProcPb.ExternalProcessor_ServiceDesc.ServiceName}, "requested-service", in.Service)
 		return &healthPb.HealthCheckResponse{Status: healthPb.HealthCheckResponse_SERVICE_UNKNOWN}, nil
@@ -94,6 +98,34 @@ func (s *healthServer) Check(ctx context.Context, in *healthPb.HealthCheckReques
 	return &healthPb.HealthCheckResponse{Status: healthPb.HealthCheckResponse_SERVING}, nil
 }
 
+func (s *healthServer) checkProtocolSupport(isLive bool) bool {
+	if !isLive {
+		// If the pool is not synced, we should skip checking the protocol support.
+		return true
+	}
+	pool, err := s.datastore.PoolGet()
+	if err != nil {
+		return true
+	}
+	if s.supporter == nil {
+		return true
+	}
+	supported := s.supporter.SupportedAppProtocols()
+	if len(supported) == 0 {
+		return true
+	}
+	appProtocol := pool.AppProtocol
+	if appProtocol == "" {
+		appProtocol = v1.AppProtocolHTTP
+	}
+	for _, p := range supported {
+		if p == appProtocol {
+			return true
+		}
+	}
+	return false
+}
+
 func (s *healthServer) List(ctx context.Context, _ *healthPb.HealthListRequest) (*healthPb.HealthListResponse, error) {
 	statuses := make(map[string]*healthPb.HealthCheckResponse)
 

cmd/epp/runner/health_test.go

@@ -0,0 +1,193 @@
+/*
+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 runner
+
+import (
+	"context"
+	"sync/atomic"
+	"testing"
+
+	extProcPb "github.com/envoyproxy/go-control-plane/envoy/service/ext_proc/v3"
+	"github.com/go-logr/logr/testr"
+	healthPb "google.golang.org/grpc/health/grpc_health_v1"
+
+	v1 "sigs.k8s.io/gateway-api-inference-extension/api/v1"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/datalayer"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/datastore"
+	fwkrh "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/interface/requesthandling"
+)
+
+// Mock Datastore
+type mockDatastore struct {
+	datastore.Datastore
+	hasSynced bool
+	pool      *datalayer.EndpointPool
+	err       error
+}
+
+func (m *mockDatastore) PoolHasSynced() bool {
+	return m.hasSynced
+}
+
+func (m *mockDatastore) PoolGet() (*datalayer.EndpointPool, error) {
+	return m.pool, m.err
+}
+
+// Mock AppProtocolSupporter
+type mockSupporter struct {
+	protocols []v1.AppProtocol
+}
+
+func (m *mockSupporter) SupportedAppProtocols() []v1.AppProtocol {
+	return m.protocols
+}
+
+func TestHealthServer_Check(t *testing.T) {
+	tests := []struct {
+		name                  string
+		leaderElectionEnabled bool
+		isLeader              bool
+		hasSynced             bool
+		pool                  *datalayer.EndpointPool
+		poolErr               error
+		supporter             *mockSupporter
+		service               string
+		wantStatus            healthPb.HealthCheckResponse_ServingStatus
+	}{
+		{
+			name:                  "LeaderElectionDisabled_Live_ProtocolMatches",
+			leaderElectionEnabled: false,
+			hasSynced:             true,
+			pool:                  &datalayer.EndpointPool{AppProtocol: v1.AppProtocolHTTP},
+			wantStatus:            healthPb.HealthCheckResponse_SERVING,
+		},
+		{
+			name:                  "LeaderElectionDisabled_NotLive",
+			leaderElectionEnabled: false,
+			hasSynced:             false,
+			wantStatus:            healthPb.HealthCheckResponse_NOT_SERVING,
+		},
+		{
+			name:                  "LeaderElectionDisabled_ProtocolMismatch",
+			leaderElectionEnabled: false,
+			hasSynced:             true,
+			pool:                  &datalayer.EndpointPool{AppProtocol: v1.AppProtocolH2C},
+			supporter:             &mockSupporter{protocols: []v1.AppProtocol{v1.AppProtocolHTTP}},
+			wantStatus:            healthPb.HealthCheckResponse_NOT_SERVING,
+		},
+		{
+			name:                  "LeaderElectionEnabled_Liveness_AlwaysServing",
+			leaderElectionEnabled: true,
+			service:               LivenessCheckService,
+			wantStatus:            healthPb.HealthCheckResponse_SERVING,
+		},
+		{
+			name:                  "LeaderElectionEnabled_Readiness_Live_Leader_ProtocolMatches",
+			leaderElectionEnabled: true,
+			isLeader:              true,
+			hasSynced:             true,
+			pool:                  &datalayer.EndpointPool{AppProtocol: v1.AppProtocolHTTP},
+			service:               ReadinessCheckService,
+			wantStatus:            healthPb.HealthCheckResponse_SERVING,
+		},
+		{
+			name:                  "LeaderElectionEnabled_Readiness_NotLive",
+			leaderElectionEnabled: true,
+			isLeader:              true,
+			hasSynced:             false,
+			service:               ReadinessCheckService,
+			wantStatus:            healthPb.HealthCheckResponse_NOT_SERVING,
+		},
+		{
+			name:                  "LeaderElectionEnabled_Readiness_NotLeader",
+			leaderElectionEnabled: true,
+			isLeader:              false,
+			hasSynced:             true,
+			pool:                  &datalayer.EndpointPool{AppProtocol: v1.AppProtocolHTTP},
+			service:               ReadinessCheckService,
+			wantStatus:            healthPb.HealthCheckResponse_NOT_SERVING,
+		},
+		{
+			name:                  "LeaderElectionEnabled_Readiness_ProtocolMismatch",
+			leaderElectionEnabled: true,
+			isLeader:              true,
+			hasSynced:             true,
+			pool:                  &datalayer.EndpointPool{AppProtocol: v1.AppProtocolH2C},
+			supporter:             &mockSupporter{protocols: []v1.AppProtocol{v1.AppProtocolHTTP}},
+			service:               ReadinessCheckService,
+			wantStatus:            healthPb.HealthCheckResponse_NOT_SERVING,
+		},
+		{
+			name:                  "LeaderElectionEnabled_EmptyService_ReflectsReadiness_Serving",
+			leaderElectionEnabled: true,
+			isLeader:              true,
+			hasSynced:             true,
+			pool:                  &datalayer.EndpointPool{AppProtocol: v1.AppProtocolHTTP},
+			service:               "",
+			wantStatus:            healthPb.HealthCheckResponse_SERVING,
+		},
+		{
+			name:                  "LeaderElectionEnabled_ExtProc_ReflectsReadiness_Serving",
+			leaderElectionEnabled: true,
+			isLeader:              true,
+			hasSynced:             true,
+			pool:                  &datalayer.EndpointPool{AppProtocol: v1.AppProtocolHTTP},
+			service:               extProcPb.ExternalProcessor_ServiceDesc.ServiceName,
+			wantStatus:            healthPb.HealthCheckResponse_SERVING,
+		},
+		{
+			name:                  "LeaderElectionEnabled_UnknownService",
+			leaderElectionEnabled: true,
+			service:               "unknown",
+			wantStatus:            healthPb.HealthCheckResponse_SERVICE_UNKNOWN,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			logger := testr.New(t)
+			ds := &mockDatastore{
+				hasSynced: tt.hasSynced,
+				pool:      tt.pool,
+				err:       tt.poolErr,
+			}
+			var isLeader atomic.Bool
+			isLeader.Store(tt.isLeader)
+
+			var supporter fwkrh.AppProtocolSupporter
+			if tt.supporter != nil {
+				supporter = tt.supporter
+			}
+
+			s := &healthServer{
+				logger:                logger,
+				datastore:             ds,
+				isLeader:              &isLeader,
+				leaderElectionEnabled: tt.leaderElectionEnabled,
+				supporter:             supporter,
+			}
+
+			resp, err := s.Check(context.Background(), &healthPb.HealthCheckRequest{Service: tt.service})
+			if err != nil {
+				t.Fatalf("Check failed: %v", err)
+			}
+			if resp.Status != tt.wantStatus {
+				t.Errorf("Check() status = %v, want %v", resp.Status, tt.wantStatus)
+			}
+		})
+	}
+}

cmd/epp/runner/runner.go

@@ -387,7 +387,11 @@ func (r *Runner) setup(ctx context.Context, cfg *rest.Config, opts *runserver.Op
 
 	// --- Add Runnables to Manager ---
 	// Register health server.
-	if err := registerHealthServer(mgr, ctrl.Log.WithName("health"), ds, opts.GRPCHealthPort, isLeader, opts.EnableLeaderElection); err != nil {
+	supporter, ok := r.parser.(fwkrh.AppProtocolSupporter)
+	if !ok {
+		setupLog.Info("Parser does not implement AppProtocolSupporter; protocol checks will be skipped in health server")
+	}
+	if err := registerHealthServer(mgr, ctrl.Log.WithName("health"), ds, opts.GRPCHealthPort, isLeader, opts.EnableLeaderElection, supporter); err != nil {
 		return nil, nil, err
 	}
 
@@ -631,13 +635,14 @@ func registerExtProcServer(mgr manager.Manager, runner *runserver.ExtProcServerR
 }
 
 // registerHealthServer adds the Health gRPC server as a Runnable to the given manager.
-func registerHealthServer(mgr manager.Manager, logger logr.Logger, ds datastore.Datastore, port int, isLeader *atomic.Bool, leaderElectionEnabled bool) error {
+func registerHealthServer(mgr manager.Manager, logger logr.Logger, ds datastore.Datastore, port int, isLeader *atomic.Bool, leaderElectionEnabled bool, supporter fwkrh.AppProtocolSupporter) error {
 	srv := grpc.NewServer()
 	healthPb.RegisterHealthServer(srv, &healthServer{
 		logger:                logger,
 		datastore:             ds,
 		isLeader:              isLeader,
 		leaderElectionEnabled: leaderElectionEnabled,
+		supporter:             supporter,
 	})
 	if err := mgr.Add(
 		runnable.NoLeaderElection(runnable.GRPCServer("health", srv, port))); err != nil {

hack/verify-framework-imports.go

@@ -127,6 +127,15 @@ var currentCodeExceptionMap = map[string][]string{
 	"pkg/epp/framework/plugins/requestcontrol/dataproducer/approximateprefix/plugin.go": {
 		"pkg/epp/metrics",
 	},
+	"pkg/epp/framework/interface/requesthandling/plugins.go": {
+		"api/v1",
+	},
+	"pkg/epp/framework/plugins/requesthandling/parsers/vllmgrpc/vllmgrpc.go": {
+		"api/v1",
+	},
+	"pkg/epp/framework/plugins/requesthandling/parsers/vllmgrpc/vllmgrpc_test.go": {
+		"api/v1",
+	},
 }
 
 func init() {

pkg/epp/datalayer/endpoint_pool.go

@@ -16,11 +16,16 @@ limitations under the License.
 
 package datalayer
 
+import (
+	v1 "sigs.k8s.io/gateway-api-inference-extension/api/v1"
+)
+
 type EndpointPool struct {
 	Selector    map[string]string
 	TargetPorts []int
 	Namespace   string
 	Name        string
+	AppProtocol v1.AppProtocol
 }
 
 // NewEndpointPool creates and returns a new empty instance of EndpointPool.

pkg/epp/framework/interface/requesthandling/plugins.go

@@ -19,6 +19,7 @@ package requesthandle
 import (
 	"context"
 
+	v1 "sigs.k8s.io/gateway-api-inference-extension/api/v1"
 	fwkplugin "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/interface/plugin"
 
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/interface/requestcontrol"
@@ -43,3 +44,9 @@ type ParsedResponse struct {
 	// Usage is only populate when the raw response has usage.
 	Usage *requestcontrol.Usage
 }
+
+// AppProtocolSupporter is an optional interface that parsers can implement to specify
+// the list of supported protocols.
+type AppProtocolSupporter interface {
+	SupportedAppProtocols() []v1.AppProtocol
+}

pkg/epp/framework/plugins/requesthandling/parsers/vllmgrpc/vllmgrpc.go

@@ -25,6 +25,7 @@ import (
 
 	"google.golang.org/protobuf/proto"
 	"sigs.k8s.io/controller-runtime/pkg/log"
+	v1 "sigs.k8s.io/gateway-api-inference-extension/api/v1"
 	logutil "sigs.k8s.io/gateway-api-inference-extension/pkg/common/observability/logging"
 	fwkplugin "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/interface/plugin"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/interface/requestcontrol"
@@ -45,6 +46,7 @@ const (
 
 // compile-time type validation
 var _ fwkrh.Parser = &VllmGRPCParser{}
+var _ fwkrh.AppProtocolSupporter = &VllmGRPCParser{}
 
 // VllmGRPCParser implements the fwkrh.Parser interface for vLLM gRPC.
 type VllmGRPCParser struct {
@@ -75,6 +77,10 @@ func (p *VllmGRPCParser) TypedName() fwkplugin.TypedName {
 	return p.typedName
 }
 
+func (p *VllmGRPCParser) SupportedAppProtocols() []v1.AppProtocol {
+	return []v1.AppProtocol{v1.AppProtocolH2C}
+}
+
 // ParseRequest parses the gRPC request body and headers and returns an LLMRequestBody.
 func (p *VllmGRPCParser) ParseRequest(ctx context.Context, body []byte, headers map[string]string) (*scheduling.LLMRequestBody, error) {
 	logger := log.FromContext(ctx)