Check if InFlightEvents is empty after scheduler_perf workload

pkg/scheduler/framework/events.go

@@ -105,6 +105,42 @@ var (
 	WildCardEvent = ClusterEvent{Resource: WildCard, ActionType: All, Label: "WildCardEvent"}
 	// UnschedulableTimeout is the event when a pod stays in unschedulable for longer than timeout.
 	UnschedulableTimeout = ClusterEvent{Resource: WildCard, ActionType: All, Label: "UnschedulableTimeout"}
+	// AllEvents contains all events defined above.
+	AllEvents = []ClusterEvent{
+		AssignedPodAdd,
+		NodeAdd,
+		NodeDelete,
+		AssignedPodUpdate,
+		UnscheduledPodAdd,
+		UnscheduledPodUpdate,
+		UnscheduledPodDelete,
+		assignedPodOtherUpdate,
+		AssignedPodDelete,
+		PodRequestScaledDown,
+		PodLabelChange,
+		PodTolerationChange,
+		PodSchedulingGateEliminatedChange,
+		NodeSpecUnschedulableChange,
+		NodeAllocatableChange,
+		NodeLabelChange,
+		NodeAnnotationChange,
+		NodeTaintChange,
+		NodeConditionChange,
+		PvAdd,
+		PvUpdate,
+		PvcAdd,
+		PvcUpdate,
+		StorageClassAdd,
+		StorageClassUpdate,
+		CSINodeAdd,
+		CSINodeUpdate,
+		CSIDriverAdd,
+		CSIDriverUpdate,
+		CSIStorageCapacityAdd,
+		CSIStorageCapacityUpdate,
+		WildCardEvent,
+		UnschedulableTimeout,
+	}
 )
 
 // PodSchedulingPropertiesChange interprets the update of a pod and returns corresponding UpdatePodXYZ event(s).

test/integration/scheduler_perf/scheduler_perf.go

@@ -52,10 +52,13 @@ import (
 	featuregatetesting "k8s.io/component-base/featuregate/testing"
 	logsapi "k8s.io/component-base/logs/api/v1"
 	"k8s.io/component-base/metrics/legacyregistry"
+	"k8s.io/component-base/metrics/testutil"
 	"k8s.io/klog/v2"
+	"k8s.io/kubernetes/pkg/features"
 	"k8s.io/kubernetes/pkg/scheduler/apis/config"
 	"k8s.io/kubernetes/pkg/scheduler/apis/config/scheme"
 	"k8s.io/kubernetes/pkg/scheduler/apis/config/validation"
+	schedframework "k8s.io/kubernetes/pkg/scheduler/framework"
 	"k8s.io/kubernetes/pkg/scheduler/framework/plugins/names"
 	frameworkruntime "k8s.io/kubernetes/pkg/scheduler/framework/runtime"
 	"k8s.io/kubernetes/pkg/scheduler/metrics"
@@ -927,6 +930,13 @@ func RunBenchmarkPerfScheduling(b *testing.B, outOfTreePluginRegistry frameworkr
 						}
 					}
 
+					if tc.FeatureGates[features.SchedulerQueueingHints] {
+						// In any case, we should make sure InFlightEvents is empty after running the scenario.
+						if err = checkEmptyInFlightEvents(); err != nil {
+							tCtx.Errorf("%s: %s", w.Name, err)
+						}
+					}
+
 					// Reset metrics to prevent metrics generated in current workload gets
 					// carried over to the next workload.
 					legacyregistry.Reset()
@@ -1027,6 +1037,23 @@ func compareMetricWithThreshold(items []DataItem, threshold float64, metricSelec
 	return nil
 }
 
+func checkEmptyInFlightEvents() error {
+	labels := []string{metrics.PodPoppedInFlightEvent}
+	for _, event := range schedframework.AllEvents {
+		labels = append(labels, event.Label)
+	}
+	for _, label := range labels {
+		value, err := testutil.GetGaugeMetricValue(metrics.InFlightEvents.WithLabelValues(label))
+		if err != nil {
+			return fmt.Errorf("failed to get InFlightEvents metric for label %s", label)
+		}
+		if value > 0 {
+			return fmt.Errorf("InFlightEvents for label %s should be empty, but has %v items", label, value)
+		}
+	}
+	return nil
+}
+
 func runWorkload(tCtx ktesting.TContext, tc *testCase, w *workload, informerFactory informers.SharedInformerFactory) []DataItem {
 	b, benchmarking := tCtx.TB().(*testing.B)
 	if benchmarking {

test/integration/scheduler_perf/scheduler_test.go

@@ -18,6 +18,9 @@ package benchmark
 
 import (
 	"testing"
+
+	"k8s.io/component-base/metrics/legacyregistry"
+	"k8s.io/kubernetes/pkg/features"
 )
 
 func TestScheduling(t *testing.T) {
@@ -43,6 +46,17 @@ func TestScheduling(t *testing.T) {
 					informerFactory, tCtx := setupTestCase(t, tc, nil, nil)
 
 					runWorkload(tCtx, tc, w, informerFactory)
+
+					if tc.FeatureGates[features.SchedulerQueueingHints] {
+						// In any case, we should make sure InFlightEvents is empty after running the scenario.
+						if err = checkEmptyInFlightEvents(); err != nil {
+							tCtx.Errorf("%s: %s", w.Name, err)
+						}
+					}
+
+					// Reset metrics to prevent metrics generated in current workload gets
+					// carried over to the next workload.
+					legacyregistry.Reset()
 				})
 			}
 		})