review comments on use of client

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

Author: elevran

pkg/epp/datalayer/collector.go

@@ -21,6 +21,10 @@ import (
 	"errors"
 	"sync"
 	"time"
+
+	"sigs.k8s.io/controller-runtime/pkg/log"
+
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/logging"
 )
 
 // TODO:
@@ -37,7 +41,6 @@ const (
 // Ticker implements a time source for periodic invocation.
 // The Ticker is passed in as parameter a Collector to allow control over time
 // progress in tests, ensuring tests are deterministic and fast.
-
 type Ticker interface {
 	Channel() <-chan time.Time
 	Stop()
@@ -86,8 +89,11 @@ func (c *Collector) Start(ctx context.Context, ticker Ticker, ep Endpoint, sourc
 		started = true
 
 		go func(endpoint Endpoint, sources []DataSource) {
+			logger := log.FromContext(ctx).WithValues("endpoint", ep.GetPod().GetIPAddress())
+			logger.V(logging.DEFAULT).Info("starting collection")
+
 			defer func() {
-				// TODO: log end of collection for endpoint
+				logger.V(logging.DEFAULT).Info("terminating collection")
 				ticker.Stop()
 			}()
 
@@ -98,7 +104,7 @@ func (c *Collector) Start(ctx context.Context, ticker Ticker, ep Endpoint, sourc
 				case <-ticker.Channel():
 					for _, src := range sources {
 						ctx, cancel := context.WithTimeout(c.ctx, defaultCollectionTimeout)
-						_ = src.Collect(ctx, endpoint) // TODO: track errors per collector
+						_ = src.Collect(ctx, endpoint) // TODO: track errors per collector?
 						cancel()                       // release the ctx timeout resources
 					}
 				}

pkg/epp/datalayer/collector_test.go

@@ -24,6 +24,8 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/datalayer/mocks"
 )
@@ -41,10 +43,25 @@ func (d *DummySource) Collect(ctx context.Context, ep Endpoint) error {
 	return nil
 }
 
+func defaultEndpoint() Endpoint {
+	ms := NewEndpoint()
+	pod := &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "pod-name",
+			Namespace: "default",
+		},
+		Status: corev1.PodStatus{
+			PodIP: "1.2.3.4",
+		},
+	}
+	ms.UpdatePod(pod)
+	return ms
+}
+
 // --- Tests ---
 
 var (
-	endpoint = NewEndpoint()
+	endpoint = defaultEndpoint()
 	sources  = []DataSource{&DummySource{}}
 )
 

pkg/epp/datalayer/metrics/client.go

@@ -21,7 +21,6 @@ import (
 	"fmt"
 	"net/http"
 	"net/url"
-	"sync"
 	"time"
 
 	"github.com/prometheus/common/expfmt"
@@ -34,98 +33,36 @@ type Client interface {
 	Get(ctx context.Context, target *url.URL, ep datalayer.Addressable) (PrometheusMetricMap, error)
 }
 
-// ClientFactory returns a Client suitable for an endpoint.
-// Implementations may return a new Client each time, or use a cached
-// copy for optimized retrieval.
-type ClientFactory interface {
-	GetClientForEndpoint(ep datalayer.Addressable) (Client, error)
-}
-
-// GetDefaultClientFactory returns a default implementation of the
-// ClientFactory, which caches and reuses client across calls.
-func GetDefaultClientFactory() ClientFactory {
-	return defaultClientFactory
-}
-
 // -- package implementations --
-var (
-	cleanupTick          = 30 * time.Second
-	maxIdleTime          = time.Minute
-	defaultClientFactory = newClientFactory()
+const (
+	maxIdleConnections = 5000
+	maxIdleTime        = 10 * time.Second
+	timeout            = 10 * time.Second
 )
 
-type client struct {
-	cl       *http.Client
-	lastUsed time.Time
-}
-
-type clientmap struct {
-	cleanupOnce sync.Once
-	clients     sync.Map // key: target (Pod) IP address, value: (cached) HTTP client
-}
-
-func newClientFactory() *clientmap {
-	clm := &clientmap{}
-	clm.startCleanupGoroutine()
-	return clm
-}
-
-func (clm *clientmap) startCleanupGoroutine() {
-	clm.cleanupOnce.Do(func() {
-		go func() {
-			for {
-				time.Sleep(cleanupTick)
-				now := time.Now()
-
-				clm.clients.Range(func(key, value any) bool {
-					entry := value.(*client)
-					if now.Sub(entry.lastUsed) > maxIdleTime {
-						clm.clients.Delete(key)
-					}
-					return true
-				})
-			}
-		}()
-	})
-}
-
-func (clm *clientmap) GetClientForEndpoint(ep datalayer.Addressable) (Client, error) {
-	id := ep.GetIPAddress()
-
-	if value, found := clm.clients.Load(id); found {
-		cl, ok := value.(*client)
-		if !ok {
-			return nil, fmt.Errorf("invalid client stored for %s(%s)", id, ep.GetNamespacedName().String())
-		}
-		return cl, nil
-	}
-
-	value, _ := clm.clients.LoadOrStore(id, newClient()) // if stored, will return the new value
-	cl, ok := value.(*client)
-	if !ok {
-		return nil, fmt.Errorf("invalid client stored for %s(%s)", id, ep.GetNamespacedName().String())
-	}
-	return cl, nil
-}
-
-func newClient() *client {
-	return &client{
-		cl: &http.Client{
-			Timeout: 10 * time.Second,
+var (
+	defaultClient = &client{
+		Client: http.Client{
+			Timeout: timeout,
+			Transport: &http.Transport{
+				MaxIdleConns:        maxIdleConnections,
+				MaxIdleConnsPerHost: 4, // host is defined as scheme://host:port
+			},
 			// TODO: set additional timeouts, transport options, etc.
 		},
-		lastUsed: time.Now(),
 	}
+)
+
+type client struct {
+	http.Client
 }
 
 func (cl *client) Get(ctx context.Context, target *url.URL, ep datalayer.Addressable) (PrometheusMetricMap, error) {
-	cl.lastUsed = time.Now()
-
 	req, err := http.NewRequestWithContext(ctx, http.MethodGet, target.String(), nil)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create request: %v", err)
 	}
-	resp, err := cl.cl.Do(req)
+	resp, err := defaultClient.Do(req)
 	if err != nil {
 		return nil, fmt.Errorf("failed to fetch metrics from %s: %w", ep.GetNamespacedName(), err)
 	}

pkg/epp/datalayer/metrics/datasource.go

@@ -40,21 +40,21 @@ type DataSource struct {
 	metricsPort   atomic.Pointer[string] // target port to use in metrics URL
 	metricsPath   string                 // path to use in metrics URL
 
-	clients    ClientFactory
+	client     Client   // client (e.g. a wrapped http.Client) used to get metrics
 	extractors sync.Map // key: name, value: extractor
 }
 
 // NewDataSource returns a new MSP compliant metrics data source, configured with the provided
 // client factory. If ClientFactory is nil, a default factory is used.
-func NewDataSource(metricsScheme string, metricsPort int32, metricsPath string, clf ClientFactory) *DataSource {
-	if clf == nil {
-		clf = GetDefaultClientFactory()
+func NewDataSource(metricsScheme string, metricsPort int32, metricsPath string, cl Client) *DataSource {
+	if cl == nil {
+		cl = defaultClient
 	}
 
 	dataSrc := &DataSource{
 		metricsScheme: metricsScheme,
 		metricsPath:   metricsPath,
-		clients:       clf,
+		client:        cl,
 	}
 	dataSrc.SetPort(metricsPort)
 	return dataSrc
@@ -86,15 +86,10 @@ func (dataSrc *DataSource) AddExtractor(extractor datalayer.Extractor) error {
 // Collect is triggered by the data layer framework to fetch potentially new
 // MSP metrics data for an endpoint.
 func (dataSrc *DataSource) Collect(ctx context.Context, ep datalayer.Endpoint) error {
-	cl, err := dataSrc.clients.GetClientForEndpoint(ep.GetPod())
-	if cl == nil {
-		return err // TODO log error
-	}
-
 	target := dataSrc.getMetricsEndpoint(ep.GetPod())
-	families, err := cl.Get(ctx, target, ep.GetPod())
+	families, err := dataSrc.client.Get(ctx, target, ep.GetPod())
 
-	if err != nil { // TODO log error
+	if err != nil {
 		return err
 	}