Adding integration with pod scraping source and e2e tests

Signed-off-by: Braulio Dumba <Braulio.Dumba@ibm.com>

Author: dumb0002

charts/workload-variant-autoscaler/templates/rbac/role.yaml

@@ -50,6 +50,8 @@ rules:
   - secrets
   verbs:
   - get
+  - list
+  - watch
 - apiGroups:
   - apps
   resources:

cmd/main.go

@@ -426,10 +426,11 @@ func main() {
 	}
 	// +kubebuilder:scaffold:builder
 
+	// Create InferencePool reconciler
 	// Create InferencePool reconciler
 	inferencePoolReconciler := &controller.InferencePoolReconciler{
 		Datastore: ds,
-		Reader:    mgr.GetClient(),
+		Client:    mgr.GetClient(),
 		PoolGKNN:  poolutil.DefaultPoolGKNN(),
 	}
 

internal/actuator/direct_actuator.go

@@ -19,7 +19,7 @@ package actuator
 import (
 	"context"
 
-	poolutil "github.com/llm-d-incubation/workload-variant-autoscaler/internal/engines/scalefromzero"
+	poolutil "github.com/llm-d-incubation/workload-variant-autoscaler/internal/utils/pool"
 	autoscalingv1 "k8s.io/api/autoscaling/v1"
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"

internal/controller/inferencepool_reconciler.go

@@ -33,7 +33,7 @@ import (
 )
 
 type InferencePoolReconciler struct {
-	client.Reader
+	client.Client
 	Datastore datastore.Datastore
 	PoolGKNN  common.GKNN
 }
@@ -80,12 +80,12 @@ func (c *InferencePoolReconciler) Reconcile(ctx context.Context, req ctrl.Reques
 
 	switch pool := obj.(type) {
 	case *v1.InferencePool:
-		endpointPool, err = poolutils.InferencePoolToEndpointPool(ctx, c.Reader, pool)
+		endpointPool, err = poolutils.InferencePoolToEndpointPool(ctx, c.Client, pool)
 		if err != nil {
 			return ctrl.Result{}, fmt.Errorf("failed to convert InferencePool v1 to EndPointPool - %w", err)
 		}
 	case *v1alpha2.InferencePool:
-		endpointPool, err = poolutils.AlphaInferencePoolToEndpointPool(ctx, c.Reader, pool)
+		endpointPool, err = poolutils.AlphaInferencePoolToEndpointPool(ctx, c.Client, pool)
 		if err != nil {
 			return ctrl.Result{}, fmt.Errorf("failed to convert InferencePool v1alpha2 to EndPointPool - %w", err)
 		}
@@ -94,7 +94,7 @@ func (c *InferencePoolReconciler) Reconcile(ctx context.Context, req ctrl.Reques
 	}
 
 	if endpointPool != nil {
-		if err := c.Datastore.PoolSet(endpointPool); err != nil {
+		if err := c.Datastore.PoolSet(ctx, c.Client, endpointPool); err != nil {
 			return ctrl.Result{}, fmt.Errorf("failed to add endpoint into the datastore: - %w", err)
 		}
 	}

internal/controller/inferencepool_reconciler_test.go

@@ -101,7 +101,7 @@ func TestInferencePoolReconcile(t *testing.T) {
 			ctx := context.Background()
 
 			ds := datastore.NewDatastore()
-			inferencePoolReconciler := &InferencePoolReconciler{Reader: fakeClient, Datastore: ds, PoolGKNN: gknn}
+			inferencePoolReconciler := &InferencePoolReconciler{Client: fakeClient, Datastore: ds, PoolGKNN: gknn}
 
 			if _, err := inferencePoolReconciler.Reconcile(ctx, req); err != nil {
 				t.Errorf("Unexpected InferencePool reconcile error: %v", err)
@@ -222,7 +222,7 @@ func TestAlphaInferencePoolReconcile(t *testing.T) {
 			ctx := context.Background()
 
 			ds := datastore.NewDatastore()
-			inferencePoolReconciler := &InferencePoolReconciler{Reader: fakeClient, Datastore: ds, PoolGKNN: gknn}
+			inferencePoolReconciler := &InferencePoolReconciler{Client: fakeClient, Datastore: ds, PoolGKNN: gknn}
 
 			if _, err := inferencePoolReconciler.Reconcile(ctx, req); err != nil {
 				t.Errorf("Unexpected InferencePool reconcile error: %v", err)

internal/datastore/datastore.go

@@ -17,10 +17,14 @@ limitations under the License.
 package datastore
 
 import (
+	"context"
 	"errors"
 	"sync"
 
+	"github.com/llm-d-incubation/workload-variant-autoscaler/internal/collector/source"
+	"github.com/llm-d-incubation/workload-variant-autoscaler/internal/collector/source/pod"
 	poolutil "github.com/llm-d-incubation/workload-variant-autoscaler/internal/utils/pool"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
 var (
@@ -31,32 +35,59 @@ var (
 // The datastore is a local cache of relevant data for the given InferencePool (currently all pulled from k8s-api)
 type Datastore interface {
 	// InferencePool operations
-	PoolSet(pool *poolutil.EndpointPool) error
+	PoolSet(ctx context.Context, client client.Client, pool *poolutil.EndpointPool) error
 	PoolGet(name string) (*poolutil.EndpointPool, error)
+	PoolGetMetricsSource(name string) source.MetricsSource
 	PoolList() []*poolutil.EndpointPool
 	PoolGetFromLabels(labels map[string]string) (*poolutil.EndpointPool, error)
-	PoolDelete(poolName string)
+	PoolDelete(name string)
 
 	// Clears the store state, happens when the pool gets deleted.
 	Clear()
 }
 
 func NewDatastore() Datastore {
 	store := &datastore{
-		pools: &sync.Map{},
+		pools:    &sync.Map{},
+		registry: source.NewSourceRegistry(),
 	}
 	return store
 }
 
 type datastore struct {
-	pools *sync.Map
+	pools    *sync.Map
+	registry *source.SourceRegistry
 }
 
 // Datastore operations
-func (ds *datastore) PoolSet(pool *poolutil.EndpointPool) error {
+func (ds *datastore) PoolSet(ctx context.Context, client client.Client, pool *poolutil.EndpointPool) error {
 	if pool == nil {
 		return errPoolIsNull
 	}
+
+	if ds.registry.Get(pool.Name) == nil {
+		// Create pod source
+		config := pod.PodScrapingSourceConfig{
+			ServiceName:             pool.EndpointPicker.ServiceName,
+			ServiceNamespace:        pool.EndpointPicker.Namespace,
+			MetricsPort:             pool.EndpointPicker.MetricsPortNumber,
+			MetricsReaderSecretName: "metrics-reader-secret",
+			MetricsReaderSecretKey:  "token",
+		}
+
+		podSource, err := pod.NewPodScrapingSource(ctx, client, config)
+		if err != nil {
+			return err
+		}
+
+		// Register in registry
+		// TODO: We need to be able to update or delete a pod source object in the registry at internal/collector/source/registry.go
+		if err := ds.registry.Register(pool.Name, podSource); err != nil {
+			return err
+		}
+	}
+
+	// Store in the datastore
 	ds.pools.Store(pool.Name, pool)
 	return nil
 }
@@ -72,6 +103,11 @@ func (ds *datastore) PoolGet(name string) (*poolutil.EndpointPool, error) {
 	return epp, nil
 }
 
+func (ds *datastore) PoolGetMetricsSource(name string) source.MetricsSource {
+	source := ds.registry.Get(name)
+	return source
+}
+
 func (ds *datastore) PoolGetFromLabels(labels map[string]string) (*poolutil.EndpointPool, error) {
 	exist := false
 	var ep *poolutil.EndpointPool

internal/datastore/datastore_test.go

@@ -94,7 +94,7 @@ func TestDatastore(t *testing.T) {
 			}
 
 			// Test PoolSet
-			gotErr := ds.PoolSet(ep)
+			gotErr := ds.PoolSet(ctx, fakeClient, ep)
 			if diff := cmp.Diff(tt.wantErr, gotErr, cmpopts.EquateErrors()); diff != "" {
 				t.Errorf("Unexpected error diff (+got/-want): %s", diff)
 			}
@@ -136,7 +136,7 @@ func TestDatastore(t *testing.T) {
 				ds.PoolDelete(ep.Name)
 				assert.Equal(t, len(ds.PoolList()), tt.clearDeleteResultLen, "Pools map should have the expected length after item deleted")
 
-				if err := ds.PoolSet(ep); err != nil {
+				if err := ds.PoolSet(ctx, fakeClient, ep); err != nil {
 					t.Errorf("failed to add endpoint into the datastore: %v", err)
 				}
 				assert.Equal(t, len(ds.PoolList()), tt.listResultLen, "Pools map should have the expected length after item added")

internal/engines/scalefromzero/engine.go

@@ -19,6 +19,8 @@ package scalefromzero
 import (
 	"context"
 	"errors"
+	"fmt"
+	"strconv"
 	"sync"
 	"time"
 
@@ -29,22 +31,26 @@ import (
 	"k8s.io/client-go/rest"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/event"
 
 	wvav1alpha1 "github.com/llm-d-incubation/workload-variant-autoscaler/api/v1alpha1"
+	"github.com/llm-d-incubation/workload-variant-autoscaler/internal/actuator"
+	"github.com/llm-d-incubation/workload-variant-autoscaler/internal/collector/source"
 	"github.com/llm-d-incubation/workload-variant-autoscaler/internal/datastore"
+	"github.com/llm-d-incubation/workload-variant-autoscaler/internal/engines/common"
 	"github.com/llm-d-incubation/workload-variant-autoscaler/internal/engines/executor"
+	"github.com/llm-d-incubation/workload-variant-autoscaler/internal/interfaces"
 	"github.com/llm-d-incubation/workload-variant-autoscaler/internal/logging"
 	"github.com/llm-d-incubation/workload-variant-autoscaler/internal/utils"
+	poolutil "github.com/llm-d-incubation/workload-variant-autoscaler/internal/utils/pool"
 )
 
-// NOTE: This is a placeholder for the scale-from-zero engine implementation.
-// The actual logic for the scale-from-zero engine should be implemented here.
-
 type Engine struct {
 	client        client.Client
 	executor      executor.Executor
 	Datastore     datastore.Datastore
 	DynamicClient dynamic.Interface
+	Actuator      *actuator.DirectActuator
 	Mapper        meta.RESTMapper
 }
 
@@ -55,10 +61,16 @@ func NewEngine(client client.Client, mapper meta.RESTMapper, config *rest.Config
 		return nil, err
 	}
 
+	actuator, err := actuator.NewDirectActuator(config)
+	if err != nil {
+		return nil, err
+	}
+
 	engine := Engine{
 		client:        client,
 		Datastore:     ds,
 		DynamicClient: dynamicClient,
+		Actuator:      actuator,
 		Mapper:        mapper,
 	}
 
@@ -110,7 +122,7 @@ func (e *Engine) optimize(ctx context.Context) error {
 				defer wg.Done()
 				defer func() { <-sem }()
 
-				err := e.processInactiveVariant(ctx, va)
+				err := e.processInactiveVariant(ctx, va, 1)
 				if err != nil {
 					ctrl.Log.V(logging.DEBUG).Error(err, "Error Processing variant", "name", va.Name)
 					errorCh <- err
@@ -141,13 +153,13 @@ func (e *Engine) optimize(ctx context.Context) error {
 }
 
 // ProcessInactiveVariant processes a single inactive VariantAutoscaling resource.
-func (e *Engine) processInactiveVariant(ctx context.Context, va wvav1alpha1.VariantAutoscaling) error {
+func (e *Engine) processInactiveVariant(ctx context.Context, va wvav1alpha1.VariantAutoscaling, targetWorkloadReplicas int) error {
 	objAPI := va.GetScaleTargetAPI()
 	objKind := va.GetScaleTargetKind()
 	objName := va.GetScaleTargetName()
 
 	// Parse Group, Version, Kind, Resource
-	gvr, err := GetResourceForKind(e.Mapper, objAPI, objKind)
+	gvr, err := poolutil.GetResourceForKind(e.Mapper, objAPI, objKind)
 	if err != nil {
 		return err
 	}
@@ -173,16 +185,111 @@ func (e *Engine) processInactiveVariant(ctx context.Context, va wvav1alpha1.Vari
 		return err
 	}
 
-	epp := pool.EndpointPicker
+	// Use EPP source from registry
+	eppSource := e.Datastore.PoolGetMetricsSource(pool.Name)
+	if eppSource == nil {
+		return errors.New("endpointpicker metrics source not found in datastore")
+	}
+
+	results, err := eppSource.Refresh(ctx, source.RefreshSpec{})
+	if err != nil {
+		return err
+	}
+
+	// Check if there are pending request in the EPP flowcontrol queue for target workload VA modelID
+	result := results["all_metrics"]
+	pendingRequestExist := false
+	for _, value := range result.Values {
+		// Check for pending requests using queue size metrics
+		metricName := value.Labels["__name__"]
+		if metricName == "inference_pool_average_queue_size" && value.Value > 0 {
+			if value.Labels["target_model_name"] == va.Spec.ModelID {
+				ctrl.Log.Info(
+					"Target workload has pending requests, not scaling up", "metricName", metricName,
+					"metric", value.Labels, "value", value.Value)
+				pendingRequestExist = true
+				break
+			}
+		}
+	}
 
-	// For Tests only (REMOVE LATER)
-	ctrl.Log.V(logging.DEBUG).Info(
-		"Target EndpointPicker resolved for inactive variant",
-		"service", epp.ServiceName,
-		"namespace", epp.Namespace,
-		"metricsPort", epp.MetricsPortNumber,
-	)
+	if !pendingRequestExist {
+		ctrl.Log.Info("No pending requests found in the flowcontrol queue - skipping scaling up from zero")
+		return nil
+	}
+
+	// 1.  Scale up from zero to one
+	// TODO: Right now we are scaling all the VA for the same target model. We need to scale only the VA that has the lowest cost.
+	err = e.Actuator.ScaleTargetObject(ctx, unstructuredObj, int32(targetWorkloadReplicas))
+	if err != nil {
+		ctrl.Log.Error(err, "Error scaling up Target Workload", "variant", va.Name, "target VA model", va.Spec.ModelID)
+		return err
+	} else {
+		ctrl.Log.Info("Successfully scaled up Target Workload", "variant", va.Name, "target VA model", va.Spec.ModelID, "inferencepool", pool.EndpointPicker.ServiceName)
+	}
+
+	// 2. Create or update VariantDecision
+	reason := "Pending request in the inferencePool for target variant model"
+	decision, hasDecision := common.DecisionCache.Get(va.Name, va.Namespace)
+	if !hasDecision {
+		cost, err := strconv.ParseFloat(va.Spec.VariantCost, 64)
+		if err != nil {
+			return err
+		}
+		common.DecisionCache.Set(va.Name, va.Namespace, interfaces.VariantDecision{
+			VariantName:        va.Name,
+			Namespace:          va.Namespace,
+			ModelID:            va.Spec.ModelID,
+			Cost:               cost,
+			TargetReplicas:     targetWorkloadReplicas, // Scale up to 1 replica
+			CurrentReplicas:    targetWorkloadReplicas,
+			DesiredReplicas:    targetWorkloadReplicas,
+			LastRunTime:        metav1.Now(),
+			SaturationBased:    false,
+			SafetyOverride:     false,
+			ModelBasedDecision: false,
+			Reason:             reason, // Reason for scaling up
+		})
+	} else {
+		if decision.CurrentReplicas == 0 {
+			decision.TargetReplicas = targetWorkloadReplicas
+			decision.CurrentReplicas = targetWorkloadReplicas
+			decision.DesiredReplicas = targetWorkloadReplicas
+			decision.LastRunTime = metav1.Now()
+			decision.SaturationBased = false
+			decision.SafetyOverride = false
+			decision.ModelBasedDecision = false
+			decision.Reason = reason
+			common.DecisionCache.Set(va.Name, va.Namespace, decision)
+		} else {
+			ctrl.Log.Info("WARNING: Target variant decision.CurrentReplicas is not zero", "value", decision.CurrentReplicas)
+		}
+	}
+
+	// 3. Updates VA status.
+	// Fetch latest version from API server to avoid conflicts
+	var updateVa wvav1alpha1.VariantAutoscaling
+	if err := utils.GetVariantAutoscalingWithBackoff(ctx, e.client, va.Name, va.Namespace, &updateVa); err != nil {
+		ctrl.Log.Error(err, "Failed to get latest VA from API server", "name", va.Name)
+	}
+	// Update DesiredOptimizedAlloc
+	updateVa.Status.DesiredOptimizedAlloc = wvav1alpha1.OptimizedAlloc{
+		NumReplicas: targetWorkloadReplicas,
+		LastRunTime: metav1.Now(),
+	}
+	updateVa.Status.Actuation.Applied = true // Reset applied status until Actuator handles it (if needed)
+
+	// Set condition based on decision characteristics
+	wvav1alpha1.SetCondition(&updateVa,
+		wvav1alpha1.TypeOptimizationReady,
+		metav1.ConditionTrue,
+		wvav1alpha1.ReasonOptimizationSucceeded,
+		fmt.Sprintf("scalefromzero decision: %s", reason))
+
+	// 4. Trigger Reconciler
+	common.DecisionTrigger <- event.GenericEvent{
+		Object: &updateVa,
+	}
 
-	// TODO: Create EPP source and query metrics port
 	return nil
 }