From 525c0aae00218adf8665d98549098b87133ff289 Mon Sep 17 00:00:00 2001 From: Elvand Lie Nababan Date: Sun, 18 Oct 2026 20:10:25 +0700 Subject: [PATCH] fix(keda): keep the autoscaled replica count on redeploy KEDA's HPA owns the live replica count, but redeploying replaces it with scale.min and can remove capacity under load. Preserve the count within new bounds while raw keeps its fixed count. Assisted-by: Codex (gpt-6.1-sol) Signed-off-by: Elvand Lie Nababan --- pkg/k8s/deployer.go | 25 ++++++ pkg/k8s/deployer_test.go | 26 ++++++ pkg/keda/deployer.go | 1 + pkg/keda/deployer_replicas_test.go | 125 +++++++++++++++++++++++++++++ 4 files changed, 177 insertions(+) create mode 100644 pkg/keda/deployer_replicas_test.go diff --git a/pkg/k8s/deployer.go b/pkg/k8s/deployer.go index b2696e588e..18a819b30a 100644 --- a/pkg/k8s/deployer.go +++ b/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 diff --git a/pkg/k8s/deployer_test.go b/pkg/k8s/deployer_test.go index b67fbdfa73..37d120fe49 100644 --- a/pkg/k8s/deployer_test.go +++ b/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) + } + }) + } +} diff --git a/pkg/keda/deployer.go b/pkg/keda/deployer.go index f327d80200..876deb1e85 100644 --- a/pkg/keda/deployer.go +++ b/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), ), } diff --git a/pkg/keda/deployer_replicas_test.go b/pkg/keda/deployer_replicas_test.go new file mode 100644 index 0000000000..90e38cb20a --- /dev/null +++ b/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") + } + }) + } +}