address review comments

Signed-off-by: Etai Lev Ran <[email protected]>

Author: elevran

cmd/epp/runner/runner.go

@@ -492,11 +492,16 @@ func setupMetricsV1(setupLog logr.Logger) (datalayer.EndpointFactory, error) {
 	return pmf, nil
 }
 
+// This function serves two (independent) purposes:
+// - creating data sources and configuring their extractors.
+// - configuring endpoint factory with the provided source.
+// In the future, data sources and extractors might be configured via
+// a file. Once done, this (and registering the sources with the
+// endpoint factory) should be moved accordingly.
+// Regardless, registration of all sources (e.g., if additional sources
+// are to be configured), must be done before the EndpointFactory is initialized.
 func setupDatalayer(logger logr.Logger) (datalayer.EndpointFactory, error) {
-	// create and register a metrics data source and extractor. In the future,
-	// data sources and extractors might be configured via a file. Once done,
-	// this (and registering the sources with the endpoint factory) should
-	// be moved accordingly.
+	// create and register a metrics data source and extractor.
 	source := dlmetrics.NewDataSource(*modelServerMetricsScheme,
 		*modelServerMetricsPath,
 		*modelServerMetricsHttpsInsecureSkipVerify,
@@ -515,6 +520,7 @@ func setupDatalayer(logger logr.Logger) (datalayer.EndpointFactory, error) {
 		return nil, err
 	}
 
+	// TODO: this could be moved to the configuration loading functions once ported over.
 	sources := datalayer.GetSources()
 	for _, src := range sources {
 		logger.Info("data layer configuration", "source", src.Name(), "extractors", src.Extractors())

pkg/epp/datalayer/collector_test.go

@@ -44,15 +44,14 @@ func (d *DummySource) Collect(ctx context.Context, ep Endpoint) error {
 }
 
 func defaultEndpoint() Endpoint {
-	ms := NewEndpoint()
 	pod := &PodInfo{
 		NamespacedName: types.NamespacedName{
 			Name:      "pod-name",
 			Namespace: "default",
 		},
 		Address: "1.2.3.4:5678",
 	}
-	ms.UpdatePod(pod)
+	ms := NewEndpoint(pod, nil)
 	return ms
 }
 

pkg/epp/datalayer/endpoint.go

@@ -48,16 +48,17 @@ type ModelServer struct {
 	attributes *Attributes
 }
 
-// NewEndpoint return a new (uninitialized) ModelServer.
-func NewEndpoint() *ModelServer {
-	return &ModelServer{
+// NewEndpoint returns a new ModelServer with the given PodInfo and Metrics.
+func NewEndpoint(pod *PodInfo, metrics *Metrics) *ModelServer {
+	if pod == nil {
+		pod = &PodInfo{}
+	}
+	if metrics == nil {
+		metrics = NewMetrics()
+	}
+	ep := &ModelServer{
 		attributes: NewAttributes(),
 	}
-}
-
-// NewInitializedEndpoint returns a new ModelServer with the given PodInfo and Metrics.
-func NewInitializedEndpoint(pod *PodInfo, metrics *Metrics) *ModelServer {
-	ep := NewEndpoint()
 	ep.UpdatePod(pod)
 	ep.UpdateMetrics(metrics)
 	return ep

pkg/epp/datalayer/factory.go

@@ -78,8 +78,8 @@ func (lc *EndpointLifecycle) NewEndpoint(parent context.Context, inpod *PodInfo,
 		return nil
 	}
 
-	endpoint := NewInitializedEndpoint(inpod, NewMetrics())
-	collector := NewCollector() // for full backward compatibility, set the logger and poolinfo
+	endpoint := NewEndpoint(inpod, nil)
+	collector := NewCollector() // TODO or full backward compatibility, set the logger and poolinfo
 
 	if _, loaded := lc.collectors.LoadOrStore(key, collector); loaded {
 		// another goroutine already created and stored a collector for this endpoint.

pkg/epp/datalayer/metrics/datasource.go

@@ -115,7 +115,6 @@ func (dataSrc *DataSource) Collect(ctx context.Context, ep datalayer.Endpoint) e
 		return true // continue iteration
 	})
 
-	// TODO: decide if to log only or return the errors
 	if len(errs) != 0 {
 		return errors.Join(errs...)
 	}

pkg/epp/datalayer/metrics/extractor.go

@@ -146,7 +146,7 @@ func (ext *Extractor) Extract(ctx context.Context, data any, ep datalayer.Endpoi
 	}
 
 	if len(errs) != 0 {
-		logger.V(logutil.TRACE).Info("Failed to refresh metrics:", "err", errors.Join(errs...))
+		return errors.Join(errs...)
 	}
 	return nil
 }

pkg/epp/datalayer/metrics/extractor_test.go

@@ -22,6 +22,7 @@ import (
 
 	"github.com/google/go-cmp/cmp"
 	dto "github.com/prometheus/client_model/go"
+	"k8s.io/utils/ptr"
 
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/datalayer"
 )
@@ -39,7 +40,7 @@ func TestExtractorExtract(t *testing.T) {
 		t.Fatalf("failed to create extractor: %v", err)
 	}
 
-	ep := datalayer.NewInitializedEndpoint(&datalayer.PodInfo{}, datalayer.NewMetrics())
+	ep := datalayer.NewEndpoint(nil, nil)
 	if ep == nil {
 		t.Fatal("expected non-nil endpoint")
 	}
@@ -59,8 +60,8 @@ func TestExtractorExtract(t *testing.T) {
 		{
 			name:    "empty PrometheusMetricMap",
 			data:    PrometheusMetricMap{},
-			wantErr: false, // no errors when metrics are missing
-			updated: false,
+			wantErr: true,  // errors when metrics are missing
+			updated: false, // and also not updated...
 		},
 		{
 			name: "single valid metric",
@@ -69,13 +70,13 @@ func TestExtractorExtract(t *testing.T) {
 					Type: dto.MetricType_GAUGE.Enum(),
 					Metric: []*dto.Metric{
 						{
-							Gauge: &dto.Gauge{Value: float64Ptr(5)},
+							Gauge: &dto.Gauge{Value: ptr.To(5.0)},
 						},
 					},
 				},
 			},
-			wantErr: false,
-			updated: true,
+			wantErr: true, // missing metrics can return an error
+			updated: true, // but should still update
 		},
 	}
 

pkg/epp/datalayer/metrics/logger.go

@@ -96,8 +96,8 @@ func printDebugMetrics(logger logr.Logger, datastore datalayer.PoolInfo, stalene
 	freshPods := datastore.PodList(podsWithFreshMetrics(stalenessThreshold))
 	stalePods := datastore.PodList(podsWithStaleMetrics(stalenessThreshold))
 
-	s := fmt.Sprintf("Current Pods and metrics gathered. Fresh metrics: %+v, Stale metrics: %+v", freshPods, stalePods)
-	logger.V(logutil.VERBOSE).Info(s)
+	logger.V(logutil.TRACE).Info("Current Pods and metrics gathered",
+		"Fresh metrics", fmt.Sprintf("%+v", freshPods), "Stale metrics", fmt.Sprintf("%+v", stalePods))
 }
 
 func refreshPrometheusMetrics(logger logr.Logger, datastore datalayer.PoolInfo, stalenessThreshold time.Duration) {

pkg/epp/datalayer/metrics/spec.go

@@ -149,9 +149,11 @@ func (spec *Spec) labelsMatch(metricLabels []*dto.LabelPair) bool {
 func extractValue(metric *dto.Metric) float64 {
 	if metric == nil {
 		return 0
-	} else if gauge := metric.GetGauge(); gauge != nil {
+	}
+	if gauge := metric.GetGauge(); gauge != nil {
 		return gauge.GetValue()
-	} else if counter := metric.GetCounter(); counter != nil {
+	}
+	if counter := metric.GetCounter(); counter != nil {
 		return counter.GetValue()
 	}
 	return 0

pkg/epp/datalayer/metrics/spec_test.go

@@ -26,14 +26,11 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"google.golang.org/protobuf/proto"
+	"k8s.io/utils/ptr"
 )
 
 // --- Test Helpers ---
 
-func float64Ptr(v float64) *float64 {
-	return &v
-}
-
 func makeMetric(labels map[string]string, value float64, timestampMs int64) *dto.Metric {
 	labelPairs := []*dto.LabelPair{}
 	for k, v := range labels {
@@ -503,14 +500,14 @@ func TestExtractValue(t *testing.T) {
 		{
 			name: "gauge metric",
 			metric: &dto.Metric{
-				Gauge: &dto.Gauge{Value: float64Ptr(42.5)},
+				Gauge: &dto.Gauge{Value: ptr.To(42.5)},
 			},
 			want: 42.5,
 		},
 		{
 			name: "counter metric",
 			metric: &dto.Metric{
-				Counter: &dto.Counter{Value: float64Ptr(99.9)},
+				Counter: &dto.Counter{Value: ptr.To(99.9)},
 			},
 			want: 99.9,
 		},