Final drop of the scale-from-zero implementation

charts/workload-variant-autoscaler/templates/manager/wva-configmap.yaml

@@ -27,6 +27,9 @@ data:
   # PROMETHEUS_BEARER_TOKEN: "your-token-here"           # Direct bearer token (development/testing)
   # PROMETHEUS_TOKEN_PATH: "/path/to/token/file"        # Path to bearer token file (production with mounted secrets)
 
+  # configuration from scalefromzero
+  EPP_METRIC_READER_BEARER_TOKEN: ""
+
   # Optimization configuration
   GLOBAL_OPT_INTERVAL: {{ .Values.wva.reconcileInterval | quote }}
 

charts/workload-variant-autoscaler/templates/manager/wva-deployment-controller-manager.yaml

@@ -64,6 +64,11 @@ spec:
         image: "{{ .Values.wva.image.repository }}:{{ .Values.wva.image.tag }}"
         imagePullPolicy: "{{ .Values.wva.imagePullPolicy }}"
         env:
+          - name: EPP_METRIC_READER_BEARER_TOKEN
+            valueFrom:
+              configMapKeyRef:
+                name: {{ include "workload-variant-autoscaler.fullname" . }}-variantautoscaling-config
+                key: EPP_METRIC_READER_BEARER_TOKEN
           - name: LOG_LEVEL
             value: {{ if .Values.wva.logging }}{{ .Values.wva.logging.level | default "info" | quote }}{{ else }}"info"{{ end }}
           - name: CONFIG_MAP_NAME

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

@@ -50,6 +50,8 @@ rules:
   - secrets
   verbs:
   - get
+  - list
+  - watch
 - apiGroups:
   - apps
   resources:
@@ -117,4 +119,9 @@ rules:
   verbs:
   - get
   - update
+- nonResourceURLs:
+  - /metrics
+  - /debug/pprof/*
+  verbs:
+  - get
 {{- end }}

cmd/main.go

@@ -425,7 +425,7 @@ func main() {
 	// Create InferencePool reconciler
 	inferencePoolReconciler := &controller.InferencePoolReconciler{
 		Datastore: ds,
-		Reader:    mgr.GetClient(),
+		Client:    mgr.GetClient(),
 		PoolGKNN:  poolutil.DefaultPoolGKNN(),
 	}
 

deploy/install.sh

@@ -54,6 +54,7 @@ EXAMPLE_DIR=${EXAMPLE_DIR:-"$WVA_PROJECT/$LLM_D_PROJECT/guides/$WELL_LIT_PATH_NA
 LLM_D_MODELSERVICE_VALUES=${LLM_D_MODELSERVICE_VALUES:-"$EXAMPLE_DIR/ms-$WELL_LIT_PATH_NAME/values.yaml"}
 ITL_AVERAGE_LATENCY_MS=${ITL_AVERAGE_LATENCY_MS:-20}
 TTFT_AVERAGE_LATENCY_MS=${TTFT_AVERAGE_LATENCY_MS:-200}
+ENABLE_SCALE_TO_ZERO=${ENABLE_SCALE_TO_ZERO:-true}
 
 # Gateway Configuration
 GATEWAY_PROVIDER=${GATEWAY_PROVIDER:-"istio"} # Options: kgateway, istio
@@ -451,7 +452,7 @@ deploy_wva_controller() {
 
     # Wait for WVA to be ready
     log_info "Waiting for WVA controller to be ready..."
-    kubectl wait --for=condition=Ready pod -l app.kubernetes.io/name=workload-variant-autoscaler -n $WVA_NS --timeout=30s || \
+    kubectl wait --for=condition=Ready pod -l app.kubernetes.io/name=workload-variant-autoscaler -n $WVA_NS --timeout=60s || \
         log_warning "WVA controller is not ready yet - check 'kubectl get pods -n $WVA_NS'"
 
     log_success "WVA deployment complete"
@@ -748,7 +749,7 @@ deploy_llm_d_infrastructure() {
                  .prefill.containers[0].args = [\"--time-to-first-token=$TTFT_AVERAGE_LATENCY_MS\", \"--inter-token-latency=$ITL_AVERAGE_LATENCY_MS\"]" \
                  -i "$LLM_D_MODELSERVICE_VALUES"
     else
-      log_info "Skipping llm-d-inference-simulator deployment (DEPLOY_LLM_D_INFERENCE_SIM=false)"
+        log_info "Skipping llm-d-inference-simulator deployment (DEPLOY_LLM_D_INFERENCE_SIM=false)"
     fi
 
     # Configure vLLM max-num-seqs if set (useful for e2e testing to force saturation)
@@ -762,17 +763,40 @@ deploy_llm_d_infrastructure() {
     helmfile apply -e $GATEWAY_PROVIDER -n ${LLMD_NS}
     kubectl apply -f httproute.yaml -n ${LLMD_NS}
 
-    if [ "$GATEWAY_PROVIDER" == "kgateway" ]; then
-        log_info "Patching kgateway service to NodePort"
-        export GATEWAY_NAME="infra-inference-scheduling-inference-gateway"
-        kubectl patch gatewayparameters.gateway.kgateway.dev $GATEWAY_NAME \
-        -n $LLMD_NS \
-        --type='merge' \
-        -p '{"spec":{"kube":{"service":{"type":"NodePort"}}}}'
+    # if [ "$GATEWAY_PROVIDER" == "kgateway" ]; then
+    #     log_info "Patching kgateway service to NodePort"
+    #     export GATEWAY_NAME="infra-inference-scheduling-inference-gateway"
+    #     kubectl patch gatewayparameters.gateway.kgateway.dev $GATEWAY_NAME \
+    #     -n $LLMD_NS \
+    #     --type='merge' \
+    #     -p '{"spec":{"kube":{"service":{"type":"NodePort"}}}}'
+    # fi
+
+    # Patch llm-d-inference-simulator deployment if scale-to-zero is enabled
+    if [ "$ENABLE_SCALE_TO_ZERO" == "true" ]; then
+        # Patch llm-d-inference-simulator deployment to use the correct image
+        log_info "Patching llm-d-inference-simulator deployment to enable flowcontrol and use a new image"
+        export DEPLOYMENT_NAME="gaie-sim-epp"
+        export NEW_IMAGE="ghcr.io/llm-d/llm-d-inference-scheduler:v0.5.0-rc.1"
+        kubectl patch deployment $DEPLOYMENT_NAME -n $LLMD_NS --type='json' -p='[
+            {
+                "op": "replace",
+                "path": "/spec/template/spec/containers/0/image",
+                "value": "'$NEW_IMAGE'"
+            },
+            {
+                "op": "add",
+                "path": "/spec/template/spec/containers/0/env/-",
+                "value": {
+                "name": "ENABLE_EXPERIMENTAL_FLOW_CONTROL_LAYER",
+                "value": "true"
+                }
+            }
+        ]'
     fi
 
     log_info "Waiting for llm-d components to initialize..."
-    kubectl wait --for=condition=Available deployment --all -n $LLMD_NS --timeout=30s || \
+    kubectl wait --for=condition=Available deployment --all -n $LLMD_NS --timeout=60s || \
         log_warning "llm-d components are not ready yet - check 'kubectl get pods -n $LLMD_NS'"
 
     # Deploy second model infrastructure for multi-model testing (limiter e2e tests)

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,15 @@ limitations under the License.
 package datastore
 
 import (
+	"context"
 	"errors"
+	"os"
 	"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 +36,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
+		token := os.Getenv("EPP_METRIC_READER_BEARER_TOKEN")
+		config := pod.PodScrapingSourceConfig{
+			ServiceName:      pool.EndpointPicker.ServiceName,
+			ServiceNamespace: pool.EndpointPicker.Namespace,
+			MetricsPort:      pool.EndpointPicker.MetricsPortNumber,
+			BearerToken:      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 +104,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")