feat: Add GetActivePods functionto plugins handle and datastore

Signed-off-by: Kfir Toledo <[email protected]>

Author: kfirtoledo

cmd/epp/runner/runner.go

@@ -336,7 +336,7 @@ func (r *Runner) Run(ctx context.Context) error {
 		}
 	}
 
-	err = r.parsePluginsConfiguration(ctx)
+	err = r.parsePluginsConfiguration(ctx, datastore)
 	if err != nil {
 		setupLog.Error(err, "Failed to parse plugins configuration")
 		return err
@@ -411,7 +411,7 @@ func (r *Runner) registerInTreePlugins() {
 	plugins.Register(testfilter.HeaderBasedTestingFilterType, testfilter.HeaderBasedTestingFilterFactory)
 }
 
-func (r *Runner) parsePluginsConfiguration(ctx context.Context) error {
+func (r *Runner) parsePluginsConfiguration(ctx context.Context, ds datastore.Datastore) error {
 	if *configText == "" && *configFile == "" {
 		return nil // configuring through code, not through file
 	}
@@ -430,8 +430,9 @@ func (r *Runner) parsePluginsConfiguration(ctx context.Context) error {
 	}
 
 	r.registerInTreePlugins()
-	handle := plugins.NewEppHandle(ctx)
+	handle := plugins.NewEppHandle(ctx, ds.GetActivePods)
 	config, err := loader.LoadConfig(configBytes, handle, logger)
+
 	if err != nil {
 		return fmt.Errorf("failed to load the configuration - %w", err)
 	}

pkg/epp/datastore/datastore.go

@@ -61,6 +61,7 @@ type Datastore interface {
 	PodList(predicate func(backendmetrics.PodMetrics) bool) []backendmetrics.PodMetrics
 	PodUpdateOrAddIfNotExist(pod *corev1.Pod) bool
 	PodDelete(namespacedName types.NamespacedName)
+	GetActivePods() []types.NamespacedName
 
 	// Clears the store state, happens when the pool gets deleted.
 	Clear()
@@ -225,6 +226,16 @@ func (ds *datastore) PodUpdateOrAddIfNotExist(pod *corev1.Pod) bool {
 	return ok
 }
 
+// GetActivePods returns a list of all active pods.
+func (ds *datastore) GetActivePods() []types.NamespacedName {
+	var namespacedNames []types.NamespacedName
+	ds.pods.Range(func(k, _ any) bool {
+		namespacedNames = append(namespacedNames, k.(types.NamespacedName))
+		return true
+	})
+	return namespacedNames
+}
+
 func (ds *datastore) PodDelete(namespacedName types.NamespacedName) {
 	v, ok := ds.pods.LoadAndDelete(namespacedName)
 	if ok {

pkg/epp/plugins/handle.go

@@ -19,6 +19,8 @@ package plugins
 import (
 	"context"
 	"fmt"
+
+	"k8s.io/apimachinery/pkg/types"
 )
 
 // Handle provides plugins a set of standard data and tools to work with
@@ -27,6 +29,9 @@ type Handle interface {
 	Context() context.Context
 
 	HandlePlugins
+
+	// GetActivePods returns a list of all active pods
+	GetActivePods() []types.NamespacedName
 }
 
 // HandlePlugins defines a set of APIs to work with instantiated plugins
@@ -44,10 +49,14 @@ type HandlePlugins interface {
 	GetAllPluginsWithNames() map[string]Plugin
 }
 
+// GetActivePodsFunc is a function that returns a list of all active pods.
+type GetActivePodsFunc func() []types.NamespacedName
+
 // eppHandle is an implementation of the interface plugins.Handle
 type eppHandle struct {
 	ctx context.Context
 	HandlePlugins
+	getActivePods GetActivePodsFunc
 }
 
 // Context returns a context the plugins can use, if they need one
@@ -84,7 +93,12 @@ func (h *eppHandlePlugins) GetAllPluginsWithNames() map[string]Plugin {
 	return h.plugins
 }
 
-func NewEppHandle(ctx context.Context) Handle {
+// GetActivePods returns a function that returns a list of all active pods
+func (h *eppHandle) GetActivePods() []types.NamespacedName {
+	return h.getActivePods()
+}
+
+func NewEppHandle(ctx context.Context, getActivePods GetActivePodsFunc) Handle {
 	return &eppHandle{
 		ctx: ctx,
 		HandlePlugins: &eppHandlePlugins{

test/utils/handle.go

@@ -19,6 +19,8 @@ package utils
 import (
 	"context"
 
+	k8stypes "k8s.io/apimachinery/pkg/types"
+
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/plugins"
 )
 
@@ -33,6 +35,10 @@ func (h *testHandle) Context() context.Context {
 	return h.ctx
 }
 
+func (h *testHandle) GetActivePods() []k8stypes.NamespacedName {
+	return []k8stypes.NamespacedName{}
+}
+
 type testHandlePlugins struct {
 	plugins map[string]plugins.Plugin
 }