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") + } + }) + } +}