Apply shedding upon saturation for priority below 0 (#1361)

* Apply sheding upon saturation for priority below 0

* Update log line

Author: ahg-g

pkg/epp/requestcontrol/director.go

@@ -164,19 +164,17 @@ func (d *Director) admitRequest(ctx context.Context, requestPriority int, fairne
 	logger.V(logutil.TRACE).Info("Entering Flow Control", "priority", requestPriority, "fairnessID", fairnessID)
 
 	// This will be removed in favor of a more robust implementation (Flow Control) in the very near future.
-	// For now we will keep similar behavior to the previous implementation.
 	// TODO: Make this a configurable value.
 	// Tracking issue https://github.com/kubernetes-sigs/gateway-api-inference-extension/issues/1347
-	if requestPriority >= 2 {
-		logger.V(logutil.DEBUG).Info("Critical request bypassing saturation check.")
+	if requestPriority >= 0 {
+		logger.V(logutil.TRACE).Info("Non-sheddable request bypassing saturation check.")
 		return nil
 	}
 
-	logger.V(logutil.DEBUG).Info("Performing saturation check for non-critical request.")
 	if d.saturationDetector.IsSaturated(ctx) { // Assuming non-nil Saturation Detector
 		return errutil.Error{
 			Code: errutil.InferencePoolResourceExhausted,
-			Msg:  "system saturated, non-critical request dropped",
+			Msg:  "system saturated, sheddable request dropped",
 		}
 	}
 

pkg/epp/requestcontrol/director_test.go

@@ -84,7 +84,7 @@ func TestDirector_HandleRequest(t *testing.T) {
 		ObjRef()
 	ioFoodReviewSheddable := testutil.MakeInferenceObjective("imFoodReviewSheddable").
 		CreationTimestamp(metav1.Unix(1000, 0)).
-		Priority(0).
+		Priority(-1).
 		ObjRef()
 	ioFoodReviewResolve := testutil.MakeInferenceObjective("imFoodReviewResolve").
 		CreationTimestamp(metav1.Unix(1000, 0)).
@@ -201,7 +201,7 @@ func TestDirector_HandleRequest(t *testing.T) {
 			targetModelName:        model,
 		},
 		{
-			name: "successful chat completions request (critical, saturation ignored)",
+			name: "successful chat completions request (default critical, saturation ignored)",
 			reqBodyMap: map[string]any{
 				"model": model,
 				"messages": []any{
@@ -211,21 +211,20 @@ func TestDirector_HandleRequest(t *testing.T) {
 					},
 				},
 			},
+			mockSaturationDetector: &mockSaturationDetector{isSaturated: true},
 			schedulerMockSetup: func(m *mockScheduler) {
 				m.scheduleResults = defaultSuccessfulScheduleResults
 			},
 			wantReqCtx: &handlers.RequestContext{
-				ObjectiveKey:    objectiveName,
 				TargetModelName: model,
 				TargetPod: &backend.Pod{
 					NamespacedName: types.NamespacedName{Namespace: "default", Name: "pod1"},
 					Address:        "192.168.1.100",
 				},
 				TargetEndpoint: "192.168.1.100:8000,192.168.2.100:8000,192.168.4.100:8000",
 			},
-			wantMutatedBodyModel:   model,
-			inferenceObjectiveName: objectiveName,
-			targetModelName:        model,
+			wantMutatedBodyModel: model,
+			targetModelName:      model,
 		},
 		{
 			name: "successful chat completions request with multiple messages (critical, saturation ignored)",
@@ -334,6 +333,7 @@ func TestDirector_HandleRequest(t *testing.T) {
 				"model":  modelSheddable,
 				"prompt": "sheddable prompt",
 			},
+			inferenceObjectiveName: objectiveNameSheddable,
 			mockSaturationDetector: &mockSaturationDetector{isSaturated: true},
 			wantErrCode:            errutil.InferencePoolResourceExhausted,
 		},

test/integration/epp/hermetic_test.go

@@ -282,42 +282,26 @@ func TestFullDuplexStreamed_KubeInferenceObjectiveRequest(t *testing.T) {
 			),
 		},
 		{
-			name:     "noncritical and all models past threshold, shed request",
-			requests: integrationutils.GenerateStreamedRequestSet(logger, "test4", modelSheddable, modelSQLLoraTarget, nil),
+			name:     "don't shed requests by default",
+			requests: integrationutils.GenerateStreamedRequestSet(logger, "test4", modelSQLLora, modelSQLLoraTarget, nil),
 			// pod 0: excluded; above queue size threshold
 			// pod 1: excluded; above KV cache threshold
 			// pod 2: excluded; above queue size threshold
 			pods: newPodStates(
-				podState{index: 0, queueSize: 6, kvCacheUsage: 0.2, activeModels: []string{"foo", "bar", modelSheddableTarget}},
-				podState{index: 1, queueSize: 0, kvCacheUsage: 0.85, activeModels: []string{"foo", modelSheddableTarget}},
-				podState{index: 2, queueSize: 10, kvCacheUsage: 0.9, activeModels: []string{"foo", modelSheddableTarget}},
-			),
-			wantErr:     false,
-			wantMetrics: map[string]string{},
-			wantResponses: integrationutils.NewImmediateErrorResponse(
-				envoyTypePb.StatusCode_TooManyRequests,
-				"inference gateway: InferencePoolResourceExhausted - system saturated, non-critical request dropped",
-			),
-		},
-		{
-			name:     "noncritical, but one server has capacity, do not shed",
-			requests: integrationutils.GenerateStreamedRequestSet(logger, "test5", modelSheddable, modelSheddableTarget, nil),
-			// Pod 1 will be picked because it has relatively low queue size and low KV cache.
-			pods: newPodStates(
-				podState{index: 0, queueSize: 4, kvCacheUsage: 0.2, activeModels: []string{"foo", "bar", modelSheddableTarget}},
-				podState{index: 1, queueSize: 4, kvCacheUsage: 0.85, activeModels: []string{"foo", modelSheddableTarget}},
-				podState{index: 2, queueSize: 10, kvCacheUsage: 0.9, activeModels: []string{"foo", modelSheddableTarget}},
+				podState{index: 0, queueSize: 6, kvCacheUsage: 0.2, activeModels: []string{"foo", "bar", modelSQLLoraTarget}},
+				podState{index: 1, queueSize: 0, kvCacheUsage: 0.85, activeModels: []string{"foo"}},
+				podState{index: 2, queueSize: 10, kvCacheUsage: 0.9, activeModels: []string{"foo"}},
 			),
 			wantMetrics: map[string]string{
 				"inference_model_request_total": inferenceObjectiveRequestTotal([]label{
-					{"model_name", modelSheddable},
-					{"target_model_name", modelSheddableTarget},
+					{"model_name", modelSQLLora},
+					{"target_model_name", modelSQLLoraTarget},
 				}),
 			},
 			wantErr: false,
 			wantResponses: integrationutils.NewRequestBufferedResponse(
 				"192.168.1.1:8000",
-				fmt.Sprintf(`{"max_tokens":100,"model":%q,"prompt":"test5","temperature":0}`, modelSheddableTarget),
+				fmt.Sprintf(`{"max_tokens":100,"model":%q,"prompt":"test4","temperature":0}`, modelSQLLoraTarget),
 				&configPb.HeaderValueOption{
 					Header: &configPb.HeaderValue{
 						Key:      "hi",