address code review comments

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

Author: nirrozenbaum

pkg/epp/handlers/server.go

@@ -308,7 +308,7 @@ func (s *StreamingServer) Process(srv extProcPb.ExternalProcessor_ProcessServer)
 		// Handle the err and fire an immediate response.
 		if err != nil {
 			logger.V(logutil.DEFAULT).Error(err, "Failed to process request", "request", req)
-			resp, err := BuildErrResponse(err)
+			resp, err := buildErrResponse(err)
 			if err != nil {
 				return err
 			}
@@ -389,7 +389,7 @@ func (r *RequestContext) updateStateAndSendIfNeeded(srv extProcPb.ExternalProces
 	return nil
 }
 
-func BuildErrResponse(err error) (*extProcPb.ProcessingResponse, error) {
+func buildErrResponse(err error) (*extProcPb.ProcessingResponse, error) {
 	var resp *extProcPb.ProcessingResponse
 
 	switch errutil.CanonicalCode(err) {
@@ -416,6 +416,17 @@ func BuildErrResponse(err error) (*extProcPb.ProcessingResponse, error) {
 				},
 			},
 		}
+	// This code can be returned by the director when there are no candidate pods for the request scheduling.
+	case errutil.ServiceUnavailable:
+		resp = &extProcPb.ProcessingResponse{
+			Response: &extProcPb.ProcessingResponse_ImmediateResponse{
+				ImmediateResponse: &extProcPb.ImmediateResponse{
+					Status: &envoyTypePb.HttpStatus{
+						Code: envoyTypePb.StatusCode_ServiceUnavailable,
+					},
+				},
+			},
+		}
 	// This code can be returned when users provide invalid json request.
 	case errutil.BadRequest:
 		resp = &extProcPb.ProcessingResponse{

pkg/epp/requestcontrol/director.go

@@ -143,9 +143,9 @@ func (d *Director) HandleRequest(ctx context.Context, reqCtx *handlers.RequestCo
 	}
 
 	// --- 3. Call Scheduler (with the relevant candidate pods) ---
-	candidatePods, err := d.getCandidatePodsForScheduling(reqCtx.Request.Metadata)
-	if err != nil {
-		return reqCtx, errutil.Error{Code: errutil.BadRequest, Msg: fmt.Errorf("failed to find candidate pods: %w", err).Error()}
+	candidatePods := d.getCandidatePodsForScheduling(ctx, reqCtx.Request.Metadata)
+	if len(candidatePods) == 0 {
+		return reqCtx, errutil.Error{Code: errutil.ServiceUnavailable, Msg: "failed to find candidate pods for serving the request"}
 	}
 	results, err := d.scheduler.Schedule(ctx, reqCtx.SchedulingRequest, candidatePods)
 	if err != nil {
@@ -190,18 +190,21 @@ func (d *Director) admitRequest(ctx context.Context, requestCriticality v1alpha2
 // Snapshot pod metrics from the datastore to:
 // 1. Reduce concurrent access to the datastore.
 // 2. Ensure consistent data during the scheduling operation of a request between all scheduling cycles.
-func (d *Director) getCandidatePodsForScheduling(requestMetadata map[string]any) ([]schedulingtypes.Pod, error) {
+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)
 	if !found {
-		return schedulingtypes.ToSchedulerPodMetrics(d.datastore.PodGetAll()), nil
+		return schedulingtypes.ToSchedulerPodMetrics(d.datastore.PodGetAll())
 	}
 
 	// Check if endpoint key is present in the subset map and ensure there is at least one value
 	endpointSubsetList, found := subsetMap[subsetHintKey].([]any)
 	if !found {
-		return schedulingtypes.ToSchedulerPodMetrics(d.datastore.PodGetAll()), nil
+		return schedulingtypes.ToSchedulerPodMetrics(d.datastore.PodGetAll())
 	} else if len(endpointSubsetList) == 0 {
-		return nil, fmt.Errorf("'%s' metadata cannot be empty", subsetHintKey)
+		loggerTrace.Info("found empty subset filter in request metadata, filtering all pods")
+		return []schedulingtypes.Pod{}
 	}
 
 	// Create a map of endpoint addresses for easy lookup
@@ -213,14 +216,18 @@ func (d *Director) getCandidatePodsForScheduling(requestMetadata map[string]any)
 		endpoints[epStr] = true
 	}
 
+	podTotalCount := 0
 	podFitleredList := d.datastore.PodList(func(pm backendmetrics.PodMetrics) bool {
+		podTotalCount++
 		if _, found := endpoints[pm.GetPod().Address]; found {
 			return true
 		}
 		return false
 	})
 
-	return schedulingtypes.ToSchedulerPodMetrics(podFitleredList), nil
+	loggerTrace.Info("filtered candidate pods by subset filtering", "podTotalCount", podTotalCount, "filteredCount", len(podFitleredList))
+
+	return schedulingtypes.ToSchedulerPodMetrics(podFitleredList)
 }
 
 // prepareRequest populates the RequestContext and calls the registered PreRequest plugins

pkg/epp/requestcontrol/director_test.go

@@ -442,7 +442,6 @@ func TestGetCandidatePodsForScheduling(t *testing.T) {
 		name     string
 		metadata map[string]any
 		output   []schedulingtypes.Pod
-		wantErr  bool
 	}{
 		{
 			name:     "SubsetFilter, filter not present — return all pods",
@@ -475,8 +474,7 @@ func TestGetCandidatePodsForScheduling(t *testing.T) {
 		{
 			name:     "SubsetFilter, filter present with empty list — return error",
 			metadata: makeFilterMetadata([]any{}),
-			output:   nil,
-			wantErr:  true,
+			output:   []schedulingtypes.Pod{},
 		},
 		{
 			name:     "SubsetFilter, subset with one matching pod",
@@ -519,15 +517,7 @@ func TestGetCandidatePodsForScheduling(t *testing.T) {
 		t.Run(test.name, func(t *testing.T) {
 			director := NewDirectorWithConfig(ds, &mockScheduler{}, &mockSaturationDetector{}, NewConfig())
 
-			got, err := director.getCandidatePodsForScheduling(test.metadata)
-
-			if test.wantErr && err == nil {
-				t.Fatalf("expected an error, but didn't receive")
-			}
-
-			if err != nil && !test.wantErr {
-				t.Fatalf("Unexpected error, got %v", err)
-			}
+			got := director.getCandidatePodsForScheduling(context.Background(), test.metadata)
 
 			diff := cmp.Diff(test.output, got, cmpopts.SortSlices(func(a, b schedulingtypes.Pod) bool {
 				return a.GetPod().NamespacedName.String() < b.GetPod().NamespacedName.String()

pkg/epp/util/error/error.go

@@ -30,6 +30,7 @@ const (
 	Unknown                        = "Unknown"
 	BadRequest                     = "BadRequest"
 	Internal                       = "Internal"
+	ServiceUnavailable             = "ServiceUnavailable"
 	ModelServerError               = "ModelServerError"
 	BadConfiguration               = "BadConfiguration"
 	InferencePoolResourceExhausted = "InferencePoolResourceExhausted"

pkg/epp/util/error/error_test.go

@@ -43,6 +43,14 @@ func TestError_Error(t *testing.T) {
 			},
 			want: "inference gateway: Internal - unexpected condition",
 		},
+		{
+			name: "ServiceUnavailable error",
+			err: Error{
+				Code: ServiceUnavailable,
+				Msg:  "service unavailable",
+			},
+			want: "inference gateway: ServiceUnavailable - service unavailable",
+		},
 		{
 			name: "ModelServerError",
 			err: Error{
@@ -124,6 +132,14 @@ func TestCanonicalCode(t *testing.T) {
 			},
 			want: Internal,
 		},
+		{
+			name: "Error type with ServiceUnavailable code",
+			err: Error{
+				Code: ServiceUnavailable,
+				Msg:  "Service unavailable error",
+			},
+			want: ServiceUnavailable,
+		},
 		{
 			name: "Error type with ModelServerError code",
 			err: Error{
@@ -205,6 +221,7 @@ func TestErrorConstants(t *testing.T) {
 		Unknown:                        "Unknown",
 		BadRequest:                     "BadRequest",
 		Internal:                       "Internal",
+		ServiceUnavailable:             "ServiceUnavailable",
 		ModelServerError:               "ModelServerError",
 		BadConfiguration:               "BadConfiguration",
 		InferencePoolResourceExhausted: "InferencePoolResourceExhausted",

test/integration/epp/hermetic_test.go

@@ -821,9 +821,9 @@ func TestFullDuplexStreamed_KubeInferenceModelRequest(t *testing.T) {
 					Response: &extProcPb.ProcessingResponse_ImmediateResponse{
 						ImmediateResponse: &extProcPb.ImmediateResponse{
 							Status: &envoyTypePb.HttpStatus{
-								Code: envoyTypePb.StatusCode_TooManyRequests,
+								Code: envoyTypePb.StatusCode_ServiceUnavailable,
 							},
-							Body: []byte("inference gateway: InferencePoolResourceExhausted - failed to find target pod: failed to run scheduler profile 'default'"),
+							Body: []byte("inference gateway: ServiceUnavailable - failed to find candidate pods for serving the request"),
 						},
 					},
 				},