Each pod has independent loops to refresh metrics

Makefile

@@ -119,7 +119,7 @@ vet: ## Run go vet against code.
 
 .PHONY: test
 test: manifests generate fmt vet envtest ## Run tests.
-	KUBEBUILDER_ASSETS="$(shell $(ENVTEST) use $(ENVTEST_K8S_VERSION) --bin-dir $(LOCALBIN) -p path)" go test $$(go list ./... | grep -v /e2e) -coverprofile cover.out
+	KUBEBUILDER_ASSETS="$(shell $(ENVTEST) use $(ENVTEST_K8S_VERSION) --bin-dir $(LOCALBIN) -p path)" go test $$(go list ./... | grep -v /e2e) -race -coverprofile cover.out
 
 .PHONY: test-integration
 test-integration: manifests generate fmt vet envtest ## Run tests.

cmd/epp/main.go

@@ -37,7 +37,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 	"sigs.k8s.io/controller-runtime/pkg/metrics/filters"
 	"sigs.k8s.io/gateway-api-inference-extension/internal/runnable"
-	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/backend"
+	backendmetrics "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/backend/metrics"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/backend/vllm"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/datastore"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/metrics"
@@ -143,22 +143,20 @@ func run() error {
 
 	ctx := ctrl.SetupSignalHandler()
 
+	pmf := backendmetrics.NewPodMetricsFactory(&vllm.PodMetricsClientImpl{}, *refreshMetricsInterval)
 	// Setup runner.
-	datastore := datastore.NewDatastore()
-	provider := backend.NewProvider(&vllm.PodMetricsClientImpl{}, datastore)
+	datastore := datastore.NewDatastore(ctx, pmf)
 	serverRunner := &runserver.ExtProcServerRunner{
 		GrpcPort:                                 *grpcPort,
 		DestinationEndpointHintMetadataNamespace: *destinationEndpointHintMetadataNamespace,
 		DestinationEndpointHintKey:               *destinationEndpointHintKey,
 		PoolName:                                 *poolName,
 		PoolNamespace:                            *poolNamespace,
-		RefreshMetricsInterval:                   *refreshMetricsInterval,
-		RefreshPrometheusMetricsInterval:         *refreshPrometheusMetricsInterval,
 		Datastore:                                datastore,
 		SecureServing:                            *secureServing,
 		CertPath:                                 *certPath,
-		Provider:                                 provider,
 		UseStreaming:                             useStreamingServer,
+		RefreshPrometheusMetricsInterval:         *refreshPrometheusMetricsInterval,
 	}
 	if err := serverRunner.SetupWithManager(ctx, mgr); err != nil {
 		setupLog.Error(err, "Failed to setup ext-proc controllers")

pkg/epp/backend/metrics/fake.go

@@ -0,0 +1,90 @@
+/*
+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 metrics
+
+import (
+	"context"
+	"fmt"
+	"sync"
+
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"sigs.k8s.io/controller-runtime/pkg/log"
+	"sigs.k8s.io/gateway-api-inference-extension/api/v1alpha2"
+	logutil "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/logging"
+)
+
+// FakePodMetrics is an implementation of PodMetrics that doesn't run the async refresh loop.
+type FakePodMetrics struct {
+	Pod     *Pod
+	Metrics *Metrics
+}
+
+func (fpm *FakePodMetrics) GetPod() *Pod {
+	return fpm.Pod
+}
+func (fpm *FakePodMetrics) GetMetrics() *Metrics {
+	return fpm.Metrics
+}
+func (fpm *FakePodMetrics) UpdatePod(pod *corev1.Pod) {
+	fpm.Pod = toInternalPod(pod)
+}
+func (fpm *FakePodMetrics) StopRefreshLoop() {} // noop
+
+type FakePodMetricsClient struct {
+	errMu sync.RWMutex
+	Err   map[types.NamespacedName]error
+	resMu sync.RWMutex
+	Res   map[types.NamespacedName]*Metrics
+}
+
+func (f *FakePodMetricsClient) FetchMetrics(ctx context.Context, pod *Pod, existing *Metrics, port int32) (*Metrics, error) {
+	f.errMu.RLock()
+	err, ok := f.Err[pod.NamespacedName]
+	f.errMu.RUnlock()
+	if ok {
+		return nil, err
+	}
+	f.resMu.RLock()
+	res, ok := f.Res[pod.NamespacedName]
+	f.resMu.RUnlock()
+	if !ok {
+		return nil, fmt.Errorf("no pod found: %v", pod.NamespacedName)
+	}
+	log.FromContext(ctx).V(logutil.VERBOSE).Info("Fetching metrics for pod", "existing", existing, "new", res)
+	return res.Clone(), nil
+}
+
+func (f *FakePodMetricsClient) SetRes(new map[types.NamespacedName]*Metrics) {
+	f.resMu.Lock()
+	defer f.resMu.Unlock()
+	f.Res = new
+}
+
+func (f *FakePodMetricsClient) SetErr(new map[types.NamespacedName]error) {
+	f.errMu.Lock()
+	defer f.errMu.Unlock()
+	f.Err = new
+}
+
+type FakeDataStore struct {
+	Res map[string]*v1alpha2.InferenceModel
+}
+
+func (fds *FakeDataStore) FetchModelData(modelName string) (returnModel *v1alpha2.InferenceModel) {
+	return fds.Res[modelName]
+}

pkg/epp/backend/metrics/logger.go

@@ -0,0 +1,111 @@
+/*
+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 metrics
+
+import (
+	"context"
+	"time"
+
+	"github.com/go-logr/logr"
+	"sigs.k8s.io/controller-runtime/pkg/log"
+	"sigs.k8s.io/gateway-api-inference-extension/api/v1alpha2"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/metrics"
+	logutil "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/logging"
+)
+
+const (
+	// Note currently the EPP treats stale metrics same as fresh.
+	// TODO: https://github.com/kubernetes-sigs/gateway-api-inference-extension/issues/336
+	metricsValidityPeriod = 5 * time.Second
+)
+
+type Datastore interface {
+	PoolGet() (*v1alpha2.InferencePool, error)
+	// PodMetrics operations
+	// PodGetAll returns all pods and metrics, including fresh and stale.
+	PodGetAll() []PodMetrics
+	PodList(func(PodMetrics) bool) []PodMetrics
+}
+
+// StartMetricsLogger starts goroutines to 1) Print metrics debug logs if the DEBUG log level is
+// enabled; 2) flushes Prometheus metrics about the backend servers.
+func StartMetricsLogger(ctx context.Context, datastore Datastore, refreshPrometheusMetricsInterval time.Duration) {
+	logger := log.FromContext(ctx)
+
+	// Periodically flush prometheus metrics for inference pool
+	go func() {
+		for {
+			select {
+			case <-ctx.Done():
+				logger.V(logutil.DEFAULT).Info("Shutting down prometheus metrics thread")
+				return
+			default:
+				time.Sleep(refreshPrometheusMetricsInterval)
+				flushPrometheusMetricsOnce(logger, datastore)
+			}
+		}
+	}()
+
+	// Periodically print out the pods and metrics for DEBUGGING.
+	if logger := logger.V(logutil.DEBUG); logger.Enabled() {
+		go func() {
+			for {
+				select {
+				case <-ctx.Done():
+					logger.V(logutil.DEFAULT).Info("Shutting down metrics logger thread")
+					return
+				default:
+					time.Sleep(5 * time.Second)
+					podsWithFreshMetrics := datastore.PodList(func(pm PodMetrics) bool {
+						return time.Since(pm.GetMetrics().UpdateTime) <= metricsValidityPeriod
+					})
+					podsWithStaleMetrics := datastore.PodList(func(pm PodMetrics) bool {
+						return time.Since(pm.GetMetrics().UpdateTime) > metricsValidityPeriod
+					})
+					logger.Info("Current Pods and metrics gathered", "fresh metrics", podsWithFreshMetrics, "stale metrics", podsWithStaleMetrics)
+				}
+			}
+		}()
+	}
+}
+
+func flushPrometheusMetricsOnce(logger logr.Logger, datastore Datastore) {
+	pool, err := datastore.PoolGet()
+	if err != nil {
+		// No inference pool or not initialize.
+		logger.V(logutil.VERBOSE).Info("pool is not initialized, skipping flushing metrics")
+		return
+	}
+
+	var kvCacheTotal float64
+	var queueTotal int
+
+	podMetrics := datastore.PodGetAll()
+	logger.V(logutil.VERBOSE).Info("Flushing Prometheus Metrics", "ReadyPods", len(podMetrics))
+	if len(podMetrics) == 0 {
+		return
+	}
+
+	for _, pod := range podMetrics {
+		kvCacheTotal += pod.GetMetrics().KVCacheUsagePercent
+		queueTotal += pod.GetMetrics().WaitingQueueSize
+	}
+
+	podTotalCount := len(podMetrics)
+	metrics.RecordInferencePoolAvgKVCache(pool.Name, kvCacheTotal/float64(podTotalCount))
+	metrics.RecordInferencePoolAvgQueueSize(pool.Name, float64(queueTotal/podTotalCount))
+}