Non test code updates due to PodInfo rename

Signed-off-by: Shmuel Kallner <[email protected]>

Author: shmuelk

pkg/epp/backend/metrics/pod_metrics.go

@@ -53,18 +53,18 @@ type PodMetricsClient interface {
 }
 
 func (pm *podMetrics) String() string {
-	return fmt.Sprintf("Pod: %v; Metrics: %v", pm.GetPod(), pm.GetMetrics())
+	return fmt.Sprintf("Pod: %v; Metrics: %v", pm.GetMetadata(), pm.GetMetrics())
 }
 
-func (pm *podMetrics) GetPod() *backend.Pod {
+func (pm *podMetrics) GetMetadata() *backend.Pod {
 	return pm.pod.Load()
 }
 
 func (pm *podMetrics) GetMetrics() *MetricsState {
 	return pm.metrics.Load()
 }
 
-func (pm *podMetrics) UpdatePod(pod *datalayer.PodInfo) {
+func (pm *podMetrics) UpdateMetadata(pod *datalayer.EndpointMetadata) {
 	pm.pod.Store(pod)
 }
 
@@ -73,7 +73,7 @@ func (pm *podMetrics) UpdatePod(pod *datalayer.PodInfo) {
 func (pm *podMetrics) startRefreshLoop(ctx context.Context) {
 	pm.startOnce.Do(func() {
 		go func() {
-			pm.logger.V(logutil.DEFAULT).Info("Starting refresher", "pod", pm.GetPod())
+			pm.logger.V(logutil.DEFAULT).Info("Starting refresher", "pod", pm.GetMetadata())
 			ticker := time.NewTicker(pm.interval)
 			defer ticker.Stop()
 			for {
@@ -84,7 +84,7 @@ func (pm *podMetrics) startRefreshLoop(ctx context.Context) {
 					return
 				case <-ticker.C: // refresh metrics periodically
 					if err := pm.refreshMetrics(); err != nil {
-						pm.logger.V(logutil.TRACE).Error(err, "Failed to refresh metrics", "pod", pm.GetPod())
+						pm.logger.V(logutil.TRACE).Error(err, "Failed to refresh metrics", "pod", pm.GetMetadata())
 					}
 				}
 			}
@@ -95,7 +95,7 @@ func (pm *podMetrics) startRefreshLoop(ctx context.Context) {
 func (pm *podMetrics) refreshMetrics() error {
 	ctx, cancel := context.WithTimeout(context.Background(), fetchMetricsTimeout)
 	defer cancel()
-	updated, err := pm.pmc.FetchMetrics(ctx, pm.GetPod(), pm.GetMetrics())
+	updated, err := pm.pmc.FetchMetrics(ctx, pm.GetMetadata(), pm.GetMetrics())
 	if err != nil {
 		pm.logger.V(logutil.TRACE).Info("Failed to refreshed metrics:", "err", err)
 	}
@@ -115,7 +115,7 @@ func (pm *podMetrics) refreshMetrics() error {
 }
 
 func (pm *podMetrics) stopRefreshLoop() {
-	pm.logger.V(logutil.DEFAULT).Info("Stopping refresher", "pod", pm.GetPod())
+	pm.logger.V(logutil.DEFAULT).Info("Stopping refresher", "pod", pm.GetMetadata())
 	pm.stopOnce.Do(func() {
 		close(pm.done)
 	})

pkg/epp/backend/metrics/types.go

@@ -52,7 +52,7 @@ type PodMetricsFactory struct {
 	refreshMetricsInterval time.Duration
 }
 
-func (f *PodMetricsFactory) NewEndpoint(parentCtx context.Context, pod *datalayer.PodInfo, ds datalayer.PoolInfo) PodMetrics {
+func (f *PodMetricsFactory) NewEndpoint(parentCtx context.Context, pod *datalayer.EndpointMetadata, ds datalayer.PoolInfo) PodMetrics {
 	pm := &podMetrics{
 		pmc:       f.pmc,
 		ds:        ds,

pkg/epp/backend/pod.go

@@ -20,4 +20,4 @@ import (
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/datalayer"
 )
 
-type Pod = datalayer.PodInfo
+type Pod = datalayer.EndpointMetadata

pkg/epp/datalayer/collector.go

@@ -88,7 +88,7 @@ func (c *Collector) Start(ctx context.Context, ticker Ticker, ep Endpoint, sourc
 	started := false
 
 	c.startOnce.Do(func() {
-		logger := log.FromContext(ctx).WithValues("endpoint", ep.GetPod().GetIPAddress())
+		logger := log.FromContext(ctx).WithValues("endpoint", ep.GetMetadata().GetIPAddress())
 		c.ctx, c.cancel = context.WithCancel(ctx)
 		started = true
 		ready = make(chan struct{})

pkg/epp/datalayer/endpoint.go

@@ -21,10 +21,10 @@ import (
 	"sync/atomic"
 )
 
-// EndpointPodState allows management of the Pod related attributes.
-type EndpointPodState interface {
-	GetPod() *PodInfo
-	UpdatePod(*PodInfo)
+// EndpointMetaState allows management of the EndpointMetadata related attributes.
+type EndpointMetaState interface {
+	GetMetadata() *EndpointMetadata
+	UpdateMetadata(*EndpointMetadata)
 	GetAttributes() *Attributes
 }
 
@@ -37,14 +37,14 @@ type EndpointMetricsState interface {
 // Endpoint represents an inference serving endpoint and its related attributes.
 type Endpoint interface {
 	fmt.Stringer
-	EndpointPodState
+	EndpointMetaState
 	EndpointMetricsState
 	AttributeMap
 }
 
 // ModelServer is an implementation of the Endpoint interface.
 type ModelServer struct {
-	pod        atomic.Pointer[PodInfo]
+	pod        atomic.Pointer[EndpointMetadata]
 	metrics    atomic.Pointer[Metrics]
 	attributes *Attributes
 }
@@ -68,14 +68,14 @@ func NewEndpoint(pod *PodInfo, metrics *Metrics) *ModelServer {
 // String returns a representation of the ModelServer. For brevity, only names of
 // extended attributes are returned and not their values.
 func (srv *ModelServer) String() string {
-	return fmt.Sprintf("Pod: %v; Metrics: %v; Attributes: %v", srv.GetPod(), srv.GetMetrics(), srv.Keys())
+	return fmt.Sprintf("Pod: %v; Metrics: %v; Attributes: %v", srv.GetMetadata(), srv.GetMetrics(), srv.Keys())
 }
 
-func (srv *ModelServer) GetPod() *PodInfo {
+func (srv *ModelServer) GetMetadata() *EndpointMetadata {
 	return srv.pod.Load()
 }
 
-func (srv *ModelServer) UpdatePod(pod *PodInfo) {
+func (srv *ModelServer) UpdateMetadata(pod *EndpointMetadata) {
 	srv.pod.Store(pod)
 }
 

pkg/epp/datalayer/factory.go

@@ -48,7 +48,7 @@ type PoolInfo interface {
 // providing methods to allocate and retire endpoints. This can potentially be used for
 // pooled memory or other management chores in the implementation.
 type EndpointFactory interface {
-	NewEndpoint(parent context.Context, inpod *PodInfo, poolinfo PoolInfo) Endpoint
+	NewEndpoint(parent context.Context, inEnpointMetadata *EndpointMetadata, poolinfo PoolInfo) Endpoint
 	ReleaseEndpoint(ep Endpoint)
 }
 
@@ -73,16 +73,16 @@ func NewEndpointFactory(sources []DataSource, refreshMetricsInterval time.Durati
 // NewEndpoint implements EndpointFactory.NewEndpoint.
 // Creates a new endpoint and starts its associated collector with its own ticker.
 // Guards against multiple concurrent calls for the same endpoint.
-func (lc *EndpointLifecycle) NewEndpoint(parent context.Context, inpod *PodInfo, _ PoolInfo) Endpoint {
-	key := types.NamespacedName{Namespace: inpod.GetNamespacedName().Namespace, Name: inpod.GetNamespacedName().Name}
+func (lc *EndpointLifecycle) NewEndpoint(parent context.Context, inEndpointMetadata *EndpointMetadata, _ PoolInfo) Endpoint {
+	key := types.NamespacedName{Namespace: inEndpointMetadata.GetNamespacedName().Namespace, Name: inEndpointMetadata.GetNamespacedName().Name}
 	logger := log.FromContext(parent).WithValues("pod", key)
 
 	if _, ok := lc.collectors.Load(key); ok {
 		logger.Info("collector already running for endpoint", "endpoint", key)
 		return nil
 	}
 
-	endpoint := NewEndpoint(inpod, nil)
+	endpoint := NewEndpoint(inEndpointMetadata, nil)
 	collector := NewCollector() // TODO or full backward compatibility, set the logger and poolinfo
 
 	if _, loaded := lc.collectors.LoadOrStore(key, collector); loaded {
@@ -104,7 +104,7 @@ func (lc *EndpointLifecycle) NewEndpoint(parent context.Context, inpod *PodInfo,
 // ReleaseEndpoint implements EndpointFactory.ReleaseEndpoint
 // Stops the collector and cleans up resources for the endpoint
 func (lc *EndpointLifecycle) ReleaseEndpoint(ep Endpoint) {
-	key := ep.GetPod().GetNamespacedName()
+	key := ep.GetMetadata().GetNamespacedName()
 
 	if value, ok := lc.collectors.LoadAndDelete(key); ok {
 		collector := value.(*Collector)

pkg/epp/datalayer/metrics/datasource.go

@@ -98,8 +98,8 @@ 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 {
-	target := dataSrc.getMetricsEndpoint(ep.GetPod())
-	families, err := dataSrc.client.Get(ctx, target, ep.GetPod())
+	target := dataSrc.getMetricsEndpoint(ep.GetMetadata())
+	families, err := dataSrc.client.Get(ctx, target, ep.GetMetadata())
 
 	if err != nil {
 		return err

pkg/epp/datastore/datastore.go

@@ -200,13 +200,13 @@ func (ds *datastore) ObjectiveGetAll() []*v1alpha2.InferenceObjective {
 // /// Pods/endpoints APIs ///
 // TODO: add a flag for callers to specify the staleness threshold for metrics.
 // ref: https://github.com/kubernetes-sigs/gateway-api-inference-extension/pull/1046#discussion_r2246351694
-func (ds *datastore) PodList(predicate func(backendmetrics.PodMetrics) bool) []backendmetrics.PodMetrics {
-	res := []backendmetrics.PodMetrics{}
+func (ds *datastore) PodList(predicate func(datalayer.Endpoint) bool) []datalayer.Endpoint {
+	res := []datalayer.Endpoint{}
 
 	ds.pods.Range(func(k, v any) bool {
-		pm := v.(backendmetrics.PodMetrics)
-		if predicate(pm) {
-			res = append(res, pm)
+		ep := v.(datalayer.Endpoint)
+		if predicate(ep) {
+			res = append(res, ep)
 		}
 		return true
 	})
@@ -228,14 +228,14 @@ func (ds *datastore) PodUpdateOrAddIfNotExist(pod *corev1.Pod) bool {
 	if len(ds.pool.Spec.TargetPorts) == 1 {
 		modelServerMetricsPort = int(ds.modelServerMetricsPort)
 	}
-	pods := []*datalayer.PodInfo{}
+	pods := []*datalayer.EndpointMetadata{}
 	for idx, port := range ds.pool.Spec.TargetPorts {
 		metricsPort := modelServerMetricsPort
 		if metricsPort == 0 {
 			metricsPort = int(port.Number)
 		}
 		pods = append(pods,
-			&datalayer.PodInfo{
+			&datalayer.EndpointMetadata{
 				NamespacedName: types.NamespacedName{
 					Name:      pod.Name + "-rank-" + strconv.Itoa(idx),
 					Namespace: pod.Namespace,
@@ -249,28 +249,28 @@ func (ds *datastore) PodUpdateOrAddIfNotExist(pod *corev1.Pod) bool {
 	}
 
 	result := true
-	for _, podInfo := range pods {
-		var pm backendmetrics.PodMetrics
-		existing, ok := ds.pods.Load(podInfo.NamespacedName)
+	for _, endpointMetadata := range pods {
+		var ep datalayer.Endpoint
+		existing, ok := ds.pods.Load(endpointMetadata.NamespacedName)
 		if !ok {
-			pm = ds.epf.NewEndpoint(ds.parentCtx, podInfo, ds)
-			ds.pods.Store(podInfo.NamespacedName, pm)
+			ep = ds.epf.NewEndpoint(ds.parentCtx, endpointMetadata, ds)
+			ds.pods.Store(endpointMetadata.NamespacedName, ep)
 			result = false
 		} else {
-			pm = existing.(backendmetrics.PodMetrics)
+			ep = existing.(backendmetrics.PodMetrics)
 		}
-		// Update pod properties if anything changed.
-		pm.UpdatePod(podInfo)
+		// Update endpoint properties if anything changed.
+		ep.UpdateMetadata(endpointMetadata)
 	}
 	return result
 }
 
 func (ds *datastore) PodDelete(podName string) {
 	ds.pods.Range(func(k, v any) bool {
-		pm := v.(backendmetrics.PodMetrics)
-		if pm.GetPod().PodName == podName {
+		ep := v.(datalayer.Endpoint)
+		if ep.GetMetadata().PodName == podName {
 			ds.pods.Delete(k)
-			ds.epf.ReleaseEndpoint(pm)
+			ds.epf.ReleaseEndpoint(ep)
 		}
 		return true
 	})
@@ -302,10 +302,10 @@ func (ds *datastore) podResyncAll(ctx context.Context, reader client.Reader) err
 
 	// Remove pods that don't belong to the pool or not ready any more.
 	ds.pods.Range(func(k, v any) bool {
-		pm := v.(backendmetrics.PodMetrics)
-		if exist := activePods[pm.GetPod().PodName]; !exist {
-			logger.V(logutil.VERBOSE).Info("Removing pod", "pod", pm.GetPod())
-			ds.PodDelete(pm.GetPod().PodName)
+		ep := v.(datalayer.Endpoint)
+		if exist := activePods[ep.GetMetadata().PodName]; !exist {
+			logger.V(logutil.VERBOSE).Info("Removing pod", "pod", ep.GetMetadata())
+			ds.PodDelete(ep.GetMetadata().PodName)
 		}
 		return true
 	})

pkg/epp/metrics/collectors/inference_pool.go

@@ -74,7 +74,7 @@ func (c *inferencePoolMetricsCollector) Collect(ch chan<- prometheus.Metric) {
 			prometheus.GaugeValue,
 			float64(pod.GetMetrics().WaitingQueueSize),
 			pool.Name,
-			pod.GetPod().NamespacedName.Name,
+			pod.GetMetadata().NamespacedName.Name,
 		)
 	}
 }

pkg/epp/requestcontrol/director.go

@@ -235,7 +235,7 @@ func (d *Director) getCandidatePodsForScheduling(ctx context.Context, requestMet
 	podTotalCount := 0
 	podFilteredList := d.datastore.PodList(func(pm backendmetrics.PodMetrics) bool {
 		podTotalCount++
-		if _, found := endpoints[pm.GetPod().GetIPAddress()]; found {
+		if _, found := endpoints[pm.GetMetadata().GetIPAddress()]; found {
 			return true
 		}
 		return false
@@ -279,9 +279,9 @@ func (d *Director) toSchedulerPodMetrics(pods []backendmetrics.PodMetrics) []sch
 	pm := make([]schedulingtypes.Pod, len(pods))
 	for i, pod := range pods {
 		if pod.GetAttributes() != nil {
-			pm[i] = &schedulingtypes.PodMetrics{Pod: pod.GetPod().Clone(), MetricsState: pod.GetMetrics().Clone(), AttributeMap: pod.GetAttributes().Clone()}
+			pm[i] = &schedulingtypes.PodMetrics{Pod: pod.GetMetadata().Clone(), MetricsState: pod.GetMetrics().Clone(), AttributeMap: pod.GetAttributes().Clone()}
 		} else {
-			pm[i] = &schedulingtypes.PodMetrics{Pod: pod.GetPod().Clone(), MetricsState: pod.GetMetrics().Clone(), AttributeMap: datalayer.NewAttributes()}
+			pm[i] = &schedulingtypes.PodMetrics{Pod: pod.GetMetadata().Clone(), MetricsState: pod.GetMetrics().Clone(), AttributeMap: datalayer.NewAttributes()}
 		}
 	}
 
@@ -338,7 +338,7 @@ func (d *Director) GetRandomPod() *backend.Pod {
 	}
 	number := rand.Intn(len(pods))
 	pod := pods[number]
-	return pod.GetPod()
+	return pod.GetMetadata()
 }
 
 func (d *Director) runPreRequestPlugins(ctx context.Context, request *schedulingtypes.LLMRequest,