feat: Filter out stale metrics

nayihz · nayihz · commit 6348fcb68ee5 · 2025-08-06T18:41:49.000+08:00
diff --git a/cmd/epp/runner/runner.go b/cmd/epp/runner/runner.go
@@ -284,7 +284,7 @@ func (r *Runner) Run(ctx context.Context) error {
 		ModelServerMetricsScheme: *modelServerMetricsScheme,
 		Client:                   metricsHttpClient,
 	},
-		*refreshMetricsInterval, *metricsStalenessThreshold)
+		*refreshMetricsInterval)
 
 	datastore := datastore.NewDatastore(ctx, pmf)
 
diff --git a/pkg/epp/backend/metrics/logger.go b/pkg/epp/backend/metrics/logger.go
@@ -68,9 +68,7 @@ func StartMetricsLogger(ctx context.Context, datastore Datastore, refreshPrometh
 					logger.V(logutil.DEFAULT).Info("Shutting down metrics logger thread")
 					return
 				case <-ticker.C:
-					podsWithFreshMetrics := datastore.PodList(func(pm PodMetrics) bool {
-						return time.Since(pm.GetMetrics().UpdateTime) <= metricsStalenessThreshold
-					})
+					podsWithFreshMetrics := datastore.PodList(PodsWithFreshMetrics(metricsStalenessThreshold))
 					podsWithStaleMetrics := datastore.PodList(func(pm PodMetrics) bool {
 						return time.Since(pm.GetMetrics().UpdateTime) > metricsStalenessThreshold
 					})
@@ -93,9 +91,7 @@ func refreshPrometheusMetrics(logger logr.Logger, datastore Datastore, metricsSt
 	var kvCacheTotal float64
 	var queueTotal int
 
-	podMetrics := datastore.PodList(func(pm PodMetrics) bool {
-		return time.Since(pm.GetMetrics().UpdateTime) <= metricsStalenessThreshold
-	})
+	podMetrics := datastore.PodList(PodsWithFreshMetrics(metricsStalenessThreshold))
 	logger.V(logutil.TRACE).Info("Refreshing Prometheus Metrics", "ReadyPods", len(podMetrics))
 	if len(podMetrics) == 0 {
 		return
diff --git a/pkg/epp/backend/metrics/pod_metrics_test.go b/pkg/epp/backend/metrics/pod_metrics_test.go
@@ -62,7 +62,7 @@ var (
 func TestMetricsRefresh(t *testing.T) {
 	ctx := context.Background()
 	pmc := &FakePodMetricsClient{}
-	pmf := NewPodMetricsFactory(pmc, time.Millisecond, time.Second*2)
+	pmf := NewPodMetricsFactory(pmc, time.Millisecond)
 
 	// The refresher is initialized with empty metrics.
 	pm := pmf.NewPodMetrics(ctx, pod1, &fakeDataStore{})
diff --git a/pkg/epp/backend/metrics/types.go b/pkg/epp/backend/metrics/types.go
@@ -29,21 +29,28 @@ import (
 )
 
 var (
-	AllPodPredicate = func(PodMetrics) bool { return true }
+	AllPodsPredicate = func(PodMetrics) bool { return true }
 )
 
-func NewPodMetricsFactory(pmc PodMetricsClient, refreshMetricsInterval, metricsStalenessThreshold time.Duration) *PodMetricsFactory {
+func PodsWithFreshMetrics(stalenessThreshold time.Duration) func(PodMetrics) bool {
+	return func(pm PodMetrics) bool {
+		if pm == nil {
+			return false // Skip nil pods
+		}
+		return time.Since(pm.GetMetrics().UpdateTime) <= stalenessThreshold
+	}
+}
+
+func NewPodMetricsFactory(pmc PodMetricsClient, refreshMetricsInterval time.Duration) *PodMetricsFactory {
 	return &PodMetricsFactory{
-		pmc:                       pmc,
-		refreshMetricsInterval:    refreshMetricsInterval,
-		metricsStalenessThreshold: metricsStalenessThreshold,
+		pmc:                    pmc,
+		refreshMetricsInterval: refreshMetricsInterval,
 	}
 }
 
 type PodMetricsFactory struct {
-	pmc                       PodMetricsClient
-	refreshMetricsInterval    time.Duration
-	metricsStalenessThreshold time.Duration
+	pmc                    PodMetricsClient
+	refreshMetricsInterval time.Duration
 }
 
 func (f *PodMetricsFactory) NewPodMetrics(parentCtx context.Context, in *corev1.Pod, ds Datastore) PodMetrics {
diff --git a/pkg/epp/controller/inferenceobjective_reconciler_test.go b/pkg/epp/controller/inferenceobjective_reconciler_test.go
@@ -198,7 +198,7 @@ func TestInferenceObjectiveReconciler(t *testing.T) {
 				WithObjects(initObjs...).
 				WithIndex(&v1alpha2.InferenceObjective{}, datastore.ModelNameIndexKey, indexInferenceObjectivesByModelName).
 				Build()
-			pmf := backendmetrics.NewPodMetricsFactory(&backendmetrics.FakePodMetricsClient{}, time.Second, time.Second*2)
+			pmf := backendmetrics.NewPodMetricsFactory(&backendmetrics.FakePodMetricsClient{}, time.Second)
 			ds := datastore.NewDatastore(t.Context(), pmf)
 			for _, m := range test.objectivessInStore {
 				ds.ObjectiveSetIfOlder(m)
diff --git a/pkg/epp/controller/inferencepool_reconciler_test.go b/pkg/epp/controller/inferencepool_reconciler_test.go
@@ -111,7 +111,7 @@ func TestInferencePoolReconciler(t *testing.T) {
 	req := ctrl.Request{NamespacedName: namespacedName}
 	ctx := context.Background()
 
-	pmf := backendmetrics.NewPodMetricsFactory(&backendmetrics.FakePodMetricsClient{}, time.Second, time.Second*2)
+	pmf := backendmetrics.NewPodMetricsFactory(&backendmetrics.FakePodMetricsClient{}, time.Second)
 	datastore := datastore.NewDatastore(ctx, pmf)
 	inferencePoolReconciler := &InferencePoolReconciler{Reader: fakeClient, Datastore: datastore, PoolGKNN: gknn}
 
@@ -186,7 +186,7 @@ func diffStore(datastore datastore.Datastore, params diffStoreParams) string {
 		params.wantPods = []string{}
 	}
 	gotPods := []string{}
-	for _, pm := range datastore.PodList(backendmetrics.AllPodPredicate) {
+	for _, pm := range datastore.PodList(backendmetrics.AllPodsPredicate) {
 		gotPods = append(gotPods, pm.GetPod().NamespacedName.Name)
 	}
 	if diff := cmp.Diff(params.wantPods, gotPods, cmpopts.SortSlices(func(a, b string) bool { return a < b })); diff != "" {
@@ -250,7 +250,7 @@ func TestXInferencePoolReconciler(t *testing.T) {
 	req := ctrl.Request{NamespacedName: namespacedName}
 	ctx := context.Background()
 
-	pmf := backendmetrics.NewPodMetricsFactory(&backendmetrics.FakePodMetricsClient{}, time.Second, time.Second*2)
+	pmf := backendmetrics.NewPodMetricsFactory(&backendmetrics.FakePodMetricsClient{}, time.Second)
 	datastore := datastore.NewDatastore(ctx, pmf)
 	inferencePoolReconciler := &InferencePoolReconciler{Reader: fakeClient, Datastore: datastore, PoolGKNN: gknn}
 
@@ -336,7 +336,7 @@ func xDiffStore(t *testing.T, datastore datastore.Datastore, params xDiffStorePa
 		params.wantPods = []string{}
 	}
 	gotPods := []string{}
-	for _, pm := range datastore.PodList(backendmetrics.AllPodPredicate) {
+	for _, pm := range datastore.PodList(backendmetrics.AllPodsPredicate) {
 		gotPods = append(gotPods, pm.GetPod().NamespacedName.Name)
 	}
 	if diff := cmp.Diff(params.wantPods, gotPods, cmpopts.SortSlices(func(a, b string) bool { return a < b })); diff != "" {
diff --git a/pkg/epp/controller/pod_reconciler_test.go b/pkg/epp/controller/pod_reconciler_test.go
@@ -44,7 +44,7 @@ var (
 	basePod3  = &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "pod3"}, Status: corev1.PodStatus{PodIP: "address-3"}}
 	basePod11 = &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "pod1"}, Status: corev1.PodStatus{PodIP: "address-11"}}
 	pmc       = &backendmetrics.FakePodMetricsClient{}
-	pmf       = backendmetrics.NewPodMetricsFactory(pmc, time.Second, time.Second*2)
+	pmf       = backendmetrics.NewPodMetricsFactory(pmc, time.Second)
 )
 
 func TestPodReconciler(t *testing.T) {
@@ -198,7 +198,7 @@ func TestPodReconciler(t *testing.T) {
 			}
 
 			var gotPods []*corev1.Pod
-			for _, pm := range store.PodList(backendmetrics.AllPodPredicate) {
+			for _, pm := range store.PodList(backendmetrics.AllPodsPredicate) {
 				pod := &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: pm.GetPod().NamespacedName.Name, Namespace: pm.GetPod().NamespacedName.Namespace}, Status: corev1.PodStatus{PodIP: pm.GetPod().Address}}
 				gotPods = append(gotPods, pod)
 			}
diff --git a/pkg/epp/datastore/datastore_test.go b/pkg/epp/datastore/datastore_test.go
@@ -82,7 +82,7 @@ func TestPool(t *testing.T) {
 			fakeClient := fake.NewClientBuilder().
 				WithScheme(scheme).
 				Build()
-			pmf := backendmetrics.NewPodMetricsFactory(&backendmetrics.FakePodMetricsClient{}, time.Second, time.Second*2)
+			pmf := backendmetrics.NewPodMetricsFactory(&backendmetrics.FakePodMetricsClient{}, time.Second)
 			datastore := NewDatastore(context.Background(), pmf)
 			_ = datastore.PoolSet(context.Background(), fakeClient, tt.inferencePool)
 			gotPool, gotErr := datastore.PoolGet()
@@ -214,7 +214,7 @@ func TestModel(t *testing.T) {
 	}
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			pmf := backendmetrics.NewPodMetricsFactory(&backendmetrics.FakePodMetricsClient{}, time.Second, time.Second*2)
+			pmf := backendmetrics.NewPodMetricsFactory(&backendmetrics.FakePodMetricsClient{}, time.Second)
 			ds := NewDatastore(t.Context(), pmf)
 			for _, m := range test.existingModels {
 				ds.ObjectiveSetIfOlder(m)
@@ -265,6 +265,11 @@ var (
 		},
 		WaitingModels: map[string]int{},
 	}
+	pod3 = &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "pod3",
+		},
+	}
 
 	pod1NamespacedName = types.NamespacedName{Name: pod1.Name, Namespace: pod1.Namespace}
 	pod2NamespacedName = types.NamespacedName{Name: pod2.Name, Namespace: pod2.Namespace}
@@ -281,6 +286,7 @@ func TestMetrics(t *testing.T) {
 		pmc       backendmetrics.PodMetricsClient
 		storePods []*corev1.Pod
 		want      []*backendmetrics.MetricsState
+		predict   func(backendmetrics.PodMetrics) bool
 	}{
 		{
 			name: "Probing metrics success",
@@ -326,6 +332,18 @@ func TestMetrics(t *testing.T) {
 				},
 			},
 		},
+		{
+			name: "Filter stale metrics",
+			pmc: &backendmetrics.FakePodMetricsClient{
+				Res: map[types.NamespacedName]*backendmetrics.MetricsState{
+					pod1NamespacedName: pod1Metrics,
+					pod2NamespacedName: pod2Metrics,
+				},
+			},
+			storePods: []*corev1.Pod{pod1, pod2, pod3},                          // We store pod3 but it has no metrics meaning it is stale.
+			want:      []*backendmetrics.MetricsState{pod1Metrics, pod2Metrics}, // pod3 metrics were stale and should not be included.
+			predict:   backendmetrics.PodsWithFreshMetrics(time.Second * 2),     // Only include pods with metrics updated in the last 2 seconds.
+		},
 	}
 
 	for _, test := range tests {
@@ -338,15 +356,18 @@ func TestMetrics(t *testing.T) {
 			fakeClient := fake.NewClientBuilder().
 				WithScheme(scheme).
 				Build()
-			pmf := backendmetrics.NewPodMetricsFactory(test.pmc, time.Millisecond, time.Second*2)
+			pmf := backendmetrics.NewPodMetricsFactory(test.pmc, time.Millisecond)
 			ds := NewDatastore(ctx, pmf)
 			_ = ds.PoolSet(ctx, fakeClient, inferencePool)
 			for _, pod := range test.storePods {
 				ds.PodUpdateOrAddIfNotExist(pod)
 			}
 			time.Sleep(1 * time.Second) // Give some time for the metrics to be fetched.
+			if test.predict == nil {
+				test.predict = backendmetrics.AllPodsPredicate
+			}
 			assert.EventuallyWithT(t, func(t *assert.CollectT) {
-				got := ds.PodList(backendmetrics.AllPodPredicate)
+				got := ds.PodList(test.predict)
 				metrics := []*backendmetrics.MetricsState{}
 				for _, one := range got {
 					metrics = append(metrics, one.GetMetrics())
@@ -432,15 +453,15 @@ func TestPods(t *testing.T) {
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 			ctx := context.Background()
-			pmf := backendmetrics.NewPodMetricsFactory(&backendmetrics.FakePodMetricsClient{}, time.Second, time.Second*2)
+			pmf := backendmetrics.NewPodMetricsFactory(&backendmetrics.FakePodMetricsClient{}, time.Second)
 			ds := NewDatastore(t.Context(), pmf)
 			for _, pod := range test.existingPods {
 				ds.PodUpdateOrAddIfNotExist(pod)
 			}
 
 			test.op(ctx, ds)
 			var gotPods []*corev1.Pod
-			for _, pm := range ds.PodList(backendmetrics.AllPodPredicate) {
+			for _, pm := range ds.PodList(backendmetrics.AllPodsPredicate) {
 				pod := &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: pm.GetPod().NamespacedName.Name, Namespace: pm.GetPod().NamespacedName.Namespace}, Status: corev1.PodStatus{PodIP: pm.GetPod().Address}}
 				gotPods = append(gotPods, pod)
 			}
diff --git a/pkg/epp/metrics/collectors/inference_pool.go b/pkg/epp/metrics/collectors/inference_pool.go
@@ -63,7 +63,7 @@ func (c *inferencePoolMetricsCollector) Collect(ch chan<- prometheus.Metric) {
 		return
 	}
 
-	podMetrics := c.ds.PodList(backendmetrics.AllPodPredicate)
+	podMetrics := c.ds.PodList(backendmetrics.AllPodsPredicate)
 	if len(podMetrics) == 0 {
 		return
 	}
diff --git a/pkg/epp/metrics/collectors/inference_pool_test.go b/pkg/epp/metrics/collectors/inference_pool_test.go
@@ -49,7 +49,7 @@ var (
 )
 
 func TestNoMetricsCollected(t *testing.T) {
-	pmf := backendmetrics.NewPodMetricsFactory(&backendmetrics.FakePodMetricsClient{}, time.Second, time.Second*2)
+	pmf := backendmetrics.NewPodMetricsFactory(&backendmetrics.FakePodMetricsClient{}, time.Second)
 	datastore := datastore.NewDatastore(context.Background(), pmf)
 
 	collector := &inferencePoolMetricsCollector{
@@ -67,7 +67,7 @@ func TestMetricsCollected(t *testing.T) {
 			pod1NamespacedName: pod1Metrics,
 		},
 	}
-	pmf := backendmetrics.NewPodMetricsFactory(pmc, time.Millisecond, time.Second*2)
+	pmf := backendmetrics.NewPodMetricsFactory(pmc, time.Millisecond)
 	ds := datastore.NewDatastore(context.Background(), pmf)
 
 	scheme := runtime.NewScheme()
diff --git a/pkg/epp/requestcontrol/director.go b/pkg/epp/requestcontrol/director.go
@@ -198,13 +198,13 @@ func (d *Director) getCandidatePodsForScheduling(ctx context.Context, requestMet
 
 	subsetMap, found := requestMetadata[subsetHintNamespace].(map[string]any)
 	if !found {
-		return d.toSchedulerPodMetrics(d.datastore.PodList(backendmetrics.AllPodPredicate))
+		return d.toSchedulerPodMetrics(d.datastore.PodList(backendmetrics.AllPodsPredicate))
 	}
 
 	// 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 d.toSchedulerPodMetrics(d.datastore.PodList(backendmetrics.AllPodPredicate))
+		return d.toSchedulerPodMetrics(d.datastore.PodList(backendmetrics.AllPodsPredicate))
 	} else if len(endpointSubsetList) == 0 {
 		loggerTrace.Info("found empty subset filter in request metadata, filtering all pods")
 		return []schedulingtypes.Pod{}
@@ -290,7 +290,7 @@ func (d *Director) HandleResponse(ctx context.Context, reqCtx *handlers.RequestC
 }
 
 func (d *Director) GetRandomPod() *backend.Pod {
-	pods := d.datastore.PodList(backendmetrics.AllPodPredicate)
+	pods := d.datastore.PodList(backendmetrics.AllPodsPredicate)
 	if len(pods) == 0 {
 		return nil
 	}
diff --git a/pkg/epp/requestcontrol/director_test.go b/pkg/epp/requestcontrol/director_test.go
@@ -93,7 +93,7 @@ func TestDirector_HandleRequest(t *testing.T) {
 		ObjRef()
 
 	// Datastore setup
-	pmf := backendmetrics.NewPodMetricsFactory(&backendmetrics.FakePodMetricsClient{}, time.Second, time.Second*2)
+	pmf := backendmetrics.NewPodMetricsFactory(&backendmetrics.FakePodMetricsClient{}, time.Second)
 	ds := datastore.NewDatastore(t.Context(), pmf)
 	ds.ObjectiveSetIfOlder(imFoodReview)
 	ds.ObjectiveSetIfOlder(imFoodReviewResolve)
@@ -531,7 +531,7 @@ func TestGetCandidatePodsForScheduling(t *testing.T) {
 		},
 	}
 
-	pmf := backendmetrics.NewPodMetricsFactory(&backendmetrics.FakePodMetricsClient{}, time.Second, time.Second*2)
+	pmf := backendmetrics.NewPodMetricsFactory(&backendmetrics.FakePodMetricsClient{}, time.Second)
 	ds := datastore.NewDatastore(t.Context(), pmf)
 	for _, testPod := range testInput {
 		ds.PodUpdateOrAddIfNotExist(testPod)
@@ -654,7 +654,7 @@ func TestGetRandomPod(t *testing.T) {
 
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			pmf := backendmetrics.NewPodMetricsFactory(&backendmetrics.FakePodMetricsClient{}, time.Millisecond, time.Second*2)
+			pmf := backendmetrics.NewPodMetricsFactory(&backendmetrics.FakePodMetricsClient{}, time.Millisecond)
 			ds := datastore.NewDatastore(t.Context(), pmf)
 			for _, pod := range test.storePods {
 				ds.PodUpdateOrAddIfNotExist(pod)
diff --git a/pkg/epp/saturationdetector/saturationdetector.go b/pkg/epp/saturationdetector/saturationdetector.go
@@ -108,16 +108,15 @@ func NewDetector(config *Config, datastore Datastore, logger logr.Logger) *Detec
 // (no capacity).
 func (d *Detector) IsSaturated(ctx context.Context) bool {
 	logger := log.FromContext(ctx).WithName(loggerName)
-	// TODO: filter out stale metrics here if needed.
-	allPodsMetrics := d.datastore.PodList(backendmetrics.AllPodPredicate)
-	if len(allPodsMetrics) == 0 {
+	allPodsFreshMetrics := d.datastore.PodList(backendmetrics.PodsWithFreshMetrics(d.config.MetricsStalenessThreshold))
+	if len(allPodsFreshMetrics) == 0 {
 		logger.V(logutil.VERBOSE).Info("No pods found in datastore; system is considered SATURATED (no capacity).")
 		// If there are no pods, there is no capacity to serve requests.
 		// Treat this as a saturated state to enable FlowController queuing.
 		return true
 	}
 
-	for _, podMetric := range allPodsMetrics {
+	for _, podMetric := range allPodsFreshMetrics {
 		metrics := podMetric.GetMetrics()
 		podNn := "unknown-pod"
 		if podMetric.GetPod() != nil {
@@ -130,13 +129,6 @@ func (d *Detector) IsSaturated(ctx context.Context) bool {
 			continue
 		}
 
-		// Check for metric staleness
-		if time.Since(metrics.UpdateTime) > d.config.MetricsStalenessThreshold {
-			logger.V(logutil.TRACE).Info("Pod metrics are stale, considered as not having good capacity",
-				"pod", podNn, "updateTime", metrics.UpdateTime, "stalenessThreshold", d.config.MetricsStalenessThreshold)
-			continue
-		}
-
 		// Check queue depth
 		if metrics.WaitingQueueSize > d.config.QueueDepthThreshold {
 			logger.V(logutil.TRACE).Info("Pod WaitingQueueSize is above threshold, considered as not having good capacity",
diff --git a/pkg/epp/saturationdetector/saturationdetector_test.go b/pkg/epp/saturationdetector/saturationdetector_test.go
@@ -41,7 +41,9 @@ type mockDatastore struct {
 func (fds *mockDatastore) PodList(predicate func(backendmetrics.PodMetrics) bool) []backendmetrics.PodMetrics {
 	pm := make([]backendmetrics.PodMetrics, 0, len(fds.pods))
 	for _, pod := range fds.pods {
-		pm = append(pm, pod)
+		if predicate(pod) {
+			pm = append(pm, pod)
+		}
 	}
 	return pm
 }
@@ -197,11 +199,9 @@ func TestDetector_IsSaturated(t *testing.T) {
 			expectedSaturat: true,
 		},
 		{
-			name:   "Single pod with nil metrics",
-			config: defaultConfig,
-			pods: []*backendmetrics.FakePodMetrics{
-				newMockPodMetrics("pod1", nil),
-			},
+			name:            "Single pod with nil metrics",
+			config:          defaultConfig,
+			pods:            []*backendmetrics.FakePodMetrics{},
 			expectedSaturat: true,
 		},
 		{
diff --git a/test/integration/epp/hermetic_test.go b/test/integration/epp/hermetic_test.go
@@ -977,7 +977,7 @@ func setUpHermeticServer(t *testing.T, podAndMetrics map[*backend.Pod]*backendme
 
 	// check if all pods are synced to datastore
 	assert.EventuallyWithT(t, func(t *assert.CollectT) {
-		assert.Len(t, serverRunner.Datastore.PodList(backendmetrics.AllPodPredicate), len(podAndMetrics), "Datastore not synced")
+		assert.Len(t, serverRunner.Datastore.PodList(backendmetrics.AllPodsPredicate), len(podAndMetrics), "Datastore not synced")
 	}, 10*time.Second, time.Second)
 
 	// Create a grpc connection
@@ -1086,7 +1086,7 @@ func BeforeSuite() func() {
 
 	serverRunner = server.NewDefaultExtProcServerRunner()
 	serverRunner.TestPodMetricsClient = &backendmetrics.FakePodMetricsClient{}
-	pmf := backendmetrics.NewPodMetricsFactory(serverRunner.TestPodMetricsClient, 10*time.Millisecond, time.Second*2)
+	pmf := backendmetrics.NewPodMetricsFactory(serverRunner.TestPodMetricsClient, 10*time.Millisecond)
 	// Adjust from defaults
 	serverRunner.PoolGKNN = common.GKNN{
 		NamespacedName: types.NamespacedName{Namespace: testNamespace, Name: testPoolName},
diff --git a/test/utils/server.go b/test/utils/server.go

Original file line number	Diff line number	Diff line change
`@@ -44,7 +44,7 @@ var (`
`44`	`44`	`basePod3 = &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "pod3"}, Status: corev1.PodStatus{PodIP: "address-3"}}`
`45`	`45`	`basePod11 = &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "pod1"}, Status: corev1.PodStatus{PodIP: "address-11"}}`
`46`	`46`	`pmc = &backendmetrics.FakePodMetricsClient{}`
`47`		`- pmf = backendmetrics.NewPodMetricsFactory(pmc, time.Second, time.Second*2)`
	`47`	`+ pmf = backendmetrics.NewPodMetricsFactory(pmc, time.Second)`
`48`	`48`	`)`
`49`	`49`
`50`	`50`	`func TestPodReconciler(t *testing.T) {`
`@@ -198,7 +198,7 @@ func TestPodReconciler(t *testing.T) {`
`198`	`198`	`}`
`199`	`199`
`200`	`200`	`var gotPods []*corev1.Pod`
`201`		`- for _, pm := range store.PodList(backendmetrics.AllPodPredicate) {`
	`201`	`+ for _, pm := range store.PodList(backendmetrics.AllPodsPredicate) {`
`202`	`202`	`pod := &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: pm.GetPod().NamespacedName.Name, Namespace: pm.GetPod().NamespacedName.Namespace}, Status: corev1.PodStatus{PodIP: pm.GetPod().Address}}`
`203`	`203`	`gotPods = append(gotPods, pod)`
`204`	`204`	`}`
Original file line number	Diff line number	Diff line change
`@@ -63,7 +63,7 @@ func (c *inferencePoolMetricsCollector) Collect(ch chan<- prometheus.Metric) {`
`63`	`63`	`return`
`64`	`64`	`}`
`65`	`65`
`66`		`- podMetrics := c.ds.PodList(backendmetrics.AllPodPredicate)`
	`66`	`+ podMetrics := c.ds.PodList(backendmetrics.AllPodsPredicate)`
`67`	`67`	`if len(podMetrics) == 0 {`
`68`	`68`	`return`
`69`	`69`	`}`
Original file line number	Diff line number	Diff line change
`@@ -49,7 +49,7 @@ var (`
`49`	`49`	`)`
`50`	`50`
`51`	`51`	`func TestNoMetricsCollected(t *testing.T) {`
`52`		`- pmf := backendmetrics.NewPodMetricsFactory(&backendmetrics.FakePodMetricsClient{}, time.Second, time.Second*2)`
	`52`	`+ pmf := backendmetrics.NewPodMetricsFactory(&backendmetrics.FakePodMetricsClient{}, time.Second)`
`53`	`53`	`datastore := datastore.NewDatastore(context.Background(), pmf)`
`54`	`54`
`55`	`55`	`collector := &inferencePoolMetricsCollector{`
`@@ -67,7 +67,7 @@ func TestMetricsCollected(t *testing.T) {`
`67`	`67`	`pod1NamespacedName: pod1Metrics,`
`68`	`68`	`},`
`69`	`69`	`}`
`70`		`- pmf := backendmetrics.NewPodMetricsFactory(pmc, time.Millisecond, time.Second*2)`
	`70`	`+ pmf := backendmetrics.NewPodMetricsFactory(pmc, time.Millisecond)`
`71`	`71`	`ds := datastore.NewDatastore(context.Background(), pmf)`
`72`	`72`
`73`	`73`	`scheme := runtime.NewScheme()`