fix(keda): keep the autoscaled replica count on redeploy

pkg/k8s/deployer.go

@@ -109,6 +109,8 @@ type Deployer struct {
 	decorator deployer.DeployDecorator
 
 	exposer deployer.Exposer
+
+	autoscaledReplicas bool
 }
 
 func NewDeployer(opts ...DeployerOpt) *Deployer {
@@ -131,6 +133,13 @@ func WithDeployerVerbose(verbose bool) DeployerOpt {
 	}
 }
 
+// WithAutoscaledReplicas preserves live replicas within scale bounds on update when an autoscaler owns the count.
+func WithAutoscaledReplicas(autoscaledReplicas bool) DeployerOpt {
+	return func(d *Deployer) {
+		d.autoscaledReplicas = autoscaledReplicas
+	}
+}
+
 func WithDeployerDecorator(decorator deployer.DeployDecorator) DeployerOpt {
 	return func(d *Deployer) {
 		d.decorator = decorator
@@ -257,6 +266,9 @@ func (d *Deployer) Deploy(ctx context.Context, f fn.Function) (fn.DeploymentResu
 		if err := preserveDeploymentSelector(existingDeployment, deployment, f.Name); err != nil {
 			return fn.DeploymentResult{}, err
 		}
+		if d.autoscaledReplicas {
+			preserveAutoscaledReplicas(existingDeployment, deployment, f)
+		}
 
 		if _, err = deploymentClient.Update(ctx, deployment, metav1.UpdateOptions{}); err != nil {
 			return fn.DeploymentResult{}, fmt.Errorf("failed to update deployment: %w", err)
@@ -366,6 +378,19 @@ func preserveDeploymentSelector(existing, desired *appsv1.Deployment, fnName str
 	return nil
 }
 
+// preserveAutoscaledReplicas keeps the live replica count within the new scale bounds.
+// KEDA's HPA owns spec.replicas, so resetting it rescales the function.
+func preserveAutoscaledReplicas(existing, desired *appsv1.Deployment, f fn.Function) {
+	if existing == nil || existing.Spec.Replicas == nil {
+		return
+	}
+	replicas := max(*existing.Spec.Replicas, *desired.Spec.Replicas)
+	if scale := f.Scale; scale != nil && scale.Max != nil && *scale.Max >= 1 {
+		replicas = int32(min(int64(replicas), *scale.Max))
+	}
+	desired.Spec.Replicas = &replicas
+}
+
 // DeployNamespace is where a function will be deployed: the requested
 // namespace, or the one it is already deployed in. The wider arbitration
 // between kube context, flags, environment and global defaults is settled

pkg/k8s/deployer_test.go

@@ -1879,3 +1879,29 @@ func Test_createEnvVarSource_EmptyKey(t *testing.T) {
 		t.Fatal("expected error for empty source key")
 	}
 }
+
+func Test_preserveAutoscaledReplicas(t *testing.T) {
+	for _, tt := range []struct {
+		name     string
+		existing *appsv1.Deployment
+		desired  int32
+		max      *int64
+		want     int32
+	}{
+		{"live count", &appsv1.Deployment{Spec: appsv1.DeploymentSpec{Replicas: new(int32(6))}}, 1, nil, 6},
+		{"zero becomes one", &appsv1.Deployment{Spec: appsv1.DeploymentSpec{Replicas: new(int32(0))}}, 1, nil, 1},
+		{"raised minimum", &appsv1.Deployment{Spec: appsv1.DeploymentSpec{Replicas: new(int32(2))}}, 5, nil, 5},
+		{"lowered maximum", &appsv1.Deployment{Spec: appsv1.DeploymentSpec{Replicas: new(int32(8))}}, 1, new(int64(3)), 3},
+		{"zero maximum ignored", &appsv1.Deployment{Spec: appsv1.DeploymentSpec{Replicas: new(int32(6))}}, 1, new(int64(0)), 6},
+		{"nil replicas", &appsv1.Deployment{}, 5, nil, 5},
+		{"nil deployment", nil, 5, nil, 5},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			desired := &appsv1.Deployment{Spec: appsv1.DeploymentSpec{Replicas: &tt.desired}}
+			preserveAutoscaledReplicas(tt.existing, desired, fn.Function{Scale: &fn.ScaleOptions{Max: tt.max}})
+			if got := *desired.Spec.Replicas; got != tt.want {
+				t.Errorf("replicas = %d, want %d", got, tt.want)
+			}
+		})
+	}
+}

pkg/keda/deployer.go

@@ -41,6 +41,7 @@ func NewDeployer(opts ...DeployerOpt) *Deployer {
 		Deployer: *k8s.NewDeployer(
 			// init with the kedaDeployerDecorator to have the correct deployer labels&annotations
 			k8s.WithDeployerDecorator(&kedaDeployerDecorator{}),
+			k8s.WithAutoscaledReplicas(true),
 		),
 	}
 

pkg/keda/deployer_replicas_test.go

@@ -0,0 +1,125 @@
+package keda
+
+import (
+	"context"
+	"encoding/json"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	appsv1 "k8s.io/api/apps/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/client-go/kubernetes/scheme"
+	"k8s.io/client-go/tools/clientcmd"
+	clientcmdapi "k8s.io/client-go/tools/clientcmd/api"
+	fn "knative.dev/func/pkg/functions"
+	"knative.dev/func/pkg/k8s"
+)
+
+func TestDeploy_ReplicasOnRedeploy(t *testing.T) {
+	for _, tt := range []struct {
+		name  string
+		live  int32
+		scale *fn.ScaleOptions
+		raw   bool
+		kafka bool
+		want  int32
+	}{
+		{name: "keda preserves live count", live: 6, want: 6},
+		{name: "keda raises zero to one", live: 0, want: 1},
+		{name: "keda raises minimum", live: 2, scale: &fn.ScaleOptions{Min: new(int64(5))}, want: 5},
+		{name: "keda lowers maximum", live: 8, scale: &fn.ScaleOptions{Max: new(int64(3))}, want: 3},
+		{name: "kafka preserves live count", live: 6, kafka: true, want: 6},
+		{name: "raw uses minimum", live: 6, scale: &fn.ScaleOptions{Min: new(int64(2))}, raw: true, want: 2},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			updates := make(chan *appsv1.Deployment, 1)
+			server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				w.Header().Set("Content-Type", "application/json")
+				switch {
+				case r.Method == http.MethodGet && r.URL.Path == "/apis/apps/v1/namespaces/test/deployments/f":
+					_ = json.NewEncoder(w).Encode(&appsv1.Deployment{
+						TypeMeta:   metav1.TypeMeta{APIVersion: "apps/v1", Kind: "Deployment"},
+						ObjectMeta: metav1.ObjectMeta{Name: "f", Namespace: "test", ResourceVersion: "1"},
+						Spec:       appsv1.DeploymentSpec{Replicas: &tt.live},
+					})
+				case r.Method == http.MethodPut && r.URL.Path == "/apis/apps/v1/namespaces/test/deployments/f":
+					body, err := io.ReadAll(r.Body)
+					if err != nil {
+						t.Errorf("read deployment update: %v", err)
+						w.WriteHeader(http.StatusBadRequest)
+						return
+					}
+					deployment := &appsv1.Deployment{}
+					if err := runtime.DecodeInto(scheme.Codecs.UniversalDeserializer(), body, deployment); err != nil {
+						t.Errorf("decode deployment update: %v", err)
+						w.WriteHeader(http.StatusBadRequest)
+						return
+					}
+					updates <- deployment
+					_ = json.NewEncoder(w).Encode(deployment)
+				case r.Method == http.MethodPost && r.URL.Path == "/api/v1/namespaces/test/services":
+					w.WriteHeader(http.StatusInternalServerError)
+					_ = json.NewEncoder(w).Encode(&metav1.Status{
+						TypeMeta: metav1.TypeMeta{APIVersion: "v1", Kind: "Status"},
+						Status:   "Failure", Reason: metav1.StatusReasonInternalError, Code: 500,
+						Message: "stop before readiness",
+					})
+				default:
+					if r.Method != http.MethodGet {
+						t.Errorf("unexpected request: %s %s", r.Method, r.URL.Path)
+					}
+					w.WriteHeader(http.StatusNotFound)
+					_ = json.NewEncoder(w).Encode(&metav1.Status{
+						TypeMeta: metav1.TypeMeta{APIVersion: "v1", Kind: "Status"},
+						Status:   "Failure", Reason: metav1.StatusReasonNotFound, Code: 404,
+					})
+				}
+			}))
+			defer server.Close()
+
+			kubeconfig := filepath.Join(t.TempDir(), "config")
+			err := clientcmd.WriteToFile(clientcmdapi.Config{
+				Clusters:       map[string]*clientcmdapi.Cluster{"test": {Server: server.URL}},
+				Contexts:       map[string]*clientcmdapi.Context{"test": {Cluster: "test", Namespace: "test"}},
+				CurrentContext: "test",
+			}, kubeconfig)
+			if err != nil {
+				t.Fatal(err)
+			}
+			t.Setenv("KUBECONFIG", kubeconfig)
+
+			f := fn.Function{Name: "f", Namespace: "test", Scale: tt.scale,
+				Deploy: fn.DeploySpec{Image: "registry.example.com/f:latest"}}
+			if tt.kafka {
+				f.Scale = &fn.ScaleOptions{KEDA: &fn.KEDAScaleOptions{Triggers: []fn.KEDATrigger{{Type: "kafka"}}}}
+				f.Run.Kafka = &fn.KafkaConfig{Brokers: "broker:9092", Topic: "events", ConsumerGroup: "f"}
+			}
+			var d fn.Deployer = NewDeployer()
+			if tt.raw {
+				d = k8s.NewDeployer()
+			}
+			ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+			defer cancel()
+			if _, err := d.Deploy(ctx, f); err == nil || !strings.Contains(err.Error(), "stop before readiness") {
+				t.Fatalf("expected Service write failure, got %v", err)
+			}
+			select {
+			case deployment := <-updates:
+				if deployment.Spec.Replicas == nil {
+					t.Fatal("deployment update has nil replicas")
+				}
+				if got := *deployment.Spec.Replicas; got != tt.want {
+					t.Errorf("deployment update replicas = %d, want %d", got, tt.want)
+				}
+			default:
+				t.Fatal("deployment update was not reached")
+			}
+		})
+	}
+}