The podMetrics updates the targetPort by reading the pool from the datastore

Author: liu-cong

pkg/epp/backend/metrics/logger.go

@@ -41,7 +41,9 @@ type Datastore interface {
 	PodList(func(PodMetrics) bool) []PodMetrics
 }
 
-func LogMetricsPeriodically(ctx context.Context, datastore Datastore, refreshPrometheusMetricsInterval time.Duration) {
+// 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

pkg/epp/backend/metrics/pod_metrics.go

@@ -35,11 +35,11 @@ const (
 )
 
 type podMetrics struct {
-	pod        unsafe.Pointer // stores a *Pod
-	metrics    unsafe.Pointer // stores a *Metrics
-	pmc        PodMetricsClient
-	targetPort int32 // metrics endpoint port in the pod
-	interval   time.Duration
+	pod      unsafe.Pointer // stores a *Pod
+	metrics  unsafe.Pointer // stores a *Metrics
+	pmc      PodMetricsClient
+	ds       Datastore
+	interval time.Duration
 
 	parentCtx context.Context
 	once      sync.Once // ensure the StartRefreshLoop is only called once.
@@ -101,9 +101,14 @@ func (pm *podMetrics) startRefreshLoop() {
 }
 
 func (pm *podMetrics) refreshMetrics() error {
+	pool, err := pm.ds.PoolGet()
+	if err != nil {
+		// No inference pool or not initialize.
+		return err
+	}
 	ctx, cancel := context.WithTimeout(context.Background(), fetchMetricsTimeout)
 	defer cancel()
-	updated, err := pm.pmc.FetchMetrics(ctx, pm.GetPod(), pm.GetMetrics(), pm.targetPort)
+	updated, err := pm.pmc.FetchMetrics(ctx, pm.GetPod(), pm.GetMetrics(), pool.Spec.TargetPortNumber)
 	if err != nil {
 		// As refresher is running in the background, it's possible that the pod is deleted but
 		// the refresh goroutine doesn't read the done channel yet. In this case, we just return nil.

pkg/epp/backend/metrics/pod_metrics_test.go

@@ -26,6 +26,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
+	"sigs.k8s.io/gateway-api-inference-extension/api/v1alpha2"
 )
 
 var (
@@ -59,8 +60,9 @@ func TestMetricsRefresh(t *testing.T) {
 	ctx := context.Background()
 	pmc := &FakePodMetricsClient{}
 	pmf := NewPodMetricsFactory(pmc, time.Millisecond)
+
 	// The refresher is initialized with empty metrics.
-	pm := pmf.NewPodMetrics(ctx, pod1, 8000)
+	pm := pmf.NewPodMetrics(ctx, pod1, &fakeDataStore{})
 
 	namespacedName := types.NamespacedName{Name: pod1.Name, Namespace: pod1.Namespace}
 	// Use SetRes to simulate an update of metrics from the pod.
@@ -78,3 +80,17 @@ func TestMetricsRefresh(t *testing.T) {
 	// Still expect the same condition (no metrics update).
 	assert.EventuallyWithT(t, condition, time.Second, time.Millisecond)
 }
+
+type fakeDataStore struct{}
+
+func (f *fakeDataStore) PoolGet() (*v1alpha2.InferencePool, error) {
+	return &v1alpha2.InferencePool{Spec: v1alpha2.InferencePoolSpec{TargetPortNumber: 8000}}, nil
+}
+func (f *fakeDataStore) PodGetAll() []PodMetrics {
+	// Not implemented.
+	return nil
+}
+func (f *fakeDataStore) PodList(func(PodMetrics) bool) []PodMetrics {
+	// Not implemented.
+	return nil
+}

pkg/epp/backend/metrics/types.go

@@ -41,17 +41,17 @@ type PodMetricsFactory struct {
 	refreshMetricsInterval time.Duration
 }
 
-func (f *PodMetricsFactory) NewPodMetrics(parentCtx context.Context, in *corev1.Pod, port int32) PodMetrics {
+func (f *PodMetricsFactory) NewPodMetrics(parentCtx context.Context, in *corev1.Pod, ds Datastore) PodMetrics {
 	pm := &podMetrics{
-		pod:        unsafe.Pointer(toInternalPod(in)),
-		metrics:    unsafe.Pointer(newMetrics()),
-		pmc:        f.pmc,
-		targetPort: port,
-		interval:   f.refreshMetricsInterval,
-		parentCtx:  parentCtx,
-		once:       sync.Once{},
-		done:       make(chan struct{}),
-		logger:     log.FromContext(parentCtx),
+		pod:       unsafe.Pointer(toInternalPod(in)),
+		metrics:   unsafe.Pointer(newMetrics()),
+		pmc:       f.pmc,
+		ds:        ds,
+		interval:  f.refreshMetricsInterval,
+		parentCtx: parentCtx,
+		once:      sync.Once{},
+		done:      make(chan struct{}),
+		logger:    log.FromContext(parentCtx),
 	}
 	pm.startRefreshLoop()
 	return pm

pkg/epp/datastore/datastore.go

@@ -238,7 +238,7 @@ func (ds *datastore) PodUpdateOrAddIfNotExist(pod *corev1.Pod, pool *v1alpha2.In
 	var pm backendmetrics.PodMetrics
 	existing, ok := ds.pods.Load(namespacedName)
 	if !ok {
-		pm = ds.pmf.NewPodMetrics(ds.parentCtx, pod, pool.Spec.TargetPortNumber)
+		pm = ds.pmf.NewPodMetrics(ds.parentCtx, pod, ds)
 		ds.pods.Store(namespacedName, pm)
 	} else {
 		pm = existing.(backendmetrics.PodMetrics)

pkg/epp/server/runserver.go

@@ -121,7 +121,7 @@ func (r *ExtProcServerRunner) SetupWithManager(ctx context.Context, mgr ctrl.Man
 // The runnable implements LeaderElectionRunnable with leader election disabled.
 func (r *ExtProcServerRunner) AsRunnable(logger logr.Logger) manager.Runnable {
 	return runnable.NoLeaderElection(manager.RunnableFunc(func(ctx context.Context) error {
-		backendmetrics.LogMetricsPeriodically(ctx, r.Datastore, r.RefreshPrometheusMetricsInterval)
+		backendmetrics.StartMetricsLogger(ctx, r.Datastore, r.RefreshPrometheusMetricsInterval)
 		var srv *grpc.Server
 		if r.SecureServing {
 			var cert tls.Certificate