From 72de0b1440ce20c8b4ee04921a80d617e7593820 Mon Sep 17 00:00:00 2001 From: Elvand Lie Nababan Date: Sat, 10 Oct 2026 02:14:27 +0700 Subject: [PATCH] fix(k8s): tolerate scale-to-zero during deployment readiness Judge zero replicas from reconciled Deployment status without looking for an active ReplicaSet. Retry a missing pod-template-hash when scaler updates make Deployment and ReplicaSet reads inconsistent. API failures still propagate and unresolved state still times out. Exercise the exported name and selector waits with a real Clientset and an HTTP test server, including stale status, scale-down between reads, healthy nonzero deployments, cancellation and API failures. Fixes knative/func#4068 Assisted-by: OpenAI Codex and Claude Code Signed-off-by: Elvand Lie Nababan --- pkg/k8s/wait.go | 17 ++- pkg/k8s/wait_poll_test.go | 216 ++++++++++++++++++++++++++++++++++++++ pkg/k8s/wait_test.go | 99 +++++++++++++++++ 3 files changed, 331 insertions(+), 1 deletion(-) create mode 100644 pkg/k8s/wait_poll_test.go create mode 100644 pkg/k8s/wait_test.go diff --git a/pkg/k8s/wait.go b/pkg/k8s/wait.go index 4d46736331..cc82e3ba80 100644 --- a/pkg/k8s/wait.go +++ b/pkg/k8s/wait.go @@ -19,6 +19,9 @@ import ( // - All replicas are updated to the latest version // - There are no unavailable replicas // - All pods associated with the deployment are running +// A Deployment with zero desired replicas is available once the +// controller has observed the current generation and reports all-zero +// counts: no pods are wanted, so none are required to be ready. func WaitForDeploymentAvailable(ctx context.Context, clientset *kubernetes.Clientset, namespace, deploymentName string, timeout time.Duration) error { return wait.PollUntilContextTimeout(ctx, 1*time.Second, timeout, true, func(ctx context.Context) (bool, error) { deployment, err := clientset.AppsV1().Deployments(namespace).Get(ctx, deploymentName, metav1.GetOptions{}) @@ -69,6 +72,17 @@ func checkIfDeploymentIsAvailable(ctx context.Context, clientset *kubernetes.Cli desiredReplicas := *deployment.Spec.Replicas + // Scalers can request zero replicas while readiness is being polled. + // Require reconciled status, including replicas from older ReplicaSets. + if desiredReplicas == 0 { + return deployment.Status.ObservedGeneration >= deployment.Generation && + deployment.Status.Replicas == 0 && + deployment.Status.UpdatedReplicas == 0 && + deployment.Status.ReadyReplicas == 0 && + deployment.Status.AvailableReplicas == 0 && + deployment.Status.UnavailableReplicas == 0, nil + } + // Check if deployment is available for _, condition := range deployment.Status.Conditions { if condition.Type == appsv1.DeploymentAvailable && condition.Status == corev1.ConditionTrue { @@ -99,7 +113,8 @@ func checkIfDeploymentIsAvailable(ctx context.Context, clientset *kubernetes.Cli } if currentPodTemplateHash == "" { - return false, fmt.Errorf("could not find current pod-template-hash for deployment %s", deployment.Name) + // A scaler may change ReplicaSets after the Deployment read. + return false, nil } // Verify all pods are from the current ReplicaSet and are running diff --git a/pkg/k8s/wait_poll_test.go b/pkg/k8s/wait_poll_test.go new file mode 100644 index 0000000000..5065b830b3 --- /dev/null +++ b/pkg/k8s/wait_poll_test.go @@ -0,0 +1,216 @@ +package k8s_test + +import ( + "context" + "encoding/json" + "errors" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + "time" + + appsv1 "k8s.io/api/apps/v1" + corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/kubernetes" + "k8s.io/client-go/rest" + + "knative.dev/func/pkg/k8s" +) + +func TestWaitForDeploymentAvailable_ScaleToZero(t *testing.T) { + zero := waitDeployment(0) + waking := waitDeployment(1) + waking.Generation = 6 + waking.Status.ObservedGeneration = 6 + waking.Status.ReadyReplicas = 0 + waking.Status.AvailableReplicas = 0 + waking.Status.Conditions[0].Status = corev1.ConditionFalse + stale := zero.DeepCopy() + stale.Status.ObservedGeneration-- + remaining := zero.DeepCopy() + remaining.Status.Replicas = 1 + unavailable := zero.DeepCopy() + unavailable.Status.UnavailableReplicas = 1 + + for _, tt := range []struct { + name string + states []*appsv1.Deployment + bySelector bool + }{ + {name: "reconciled zero without an active ReplicaSet", states: []*appsv1.Deployment{zero}}, + {name: "selector wait", states: []*appsv1.Deployment{zero}, bySelector: true}, + {name: "scaler zeroes a waking deployment", states: []*appsv1.Deployment{waking, stale, zero}}, + {name: "remaining old replica", states: []*appsv1.Deployment{remaining, zero}}, + {name: "unavailable replica", states: []*appsv1.Deployment{unavailable, zero}}, + } { + t.Run(tt.name, func(t *testing.T) { + api := &waitAPI{deployments: tt.states} + client := api.client(t) + var err error + if tt.bySelector { + err = k8s.WaitForDeploymentAvailableBySelector(t.Context(), client, "ns", "app=fn", 10*time.Second) + } else { + err = k8s.WaitForDeploymentAvailable(t.Context(), client, "ns", "fn", 10*time.Second) + } + if err != nil { + t.Fatal(err) + } + if got := api.deploymentReads.Load(); got != int32(len(tt.states)) { + t.Errorf("deployment reads = %d, want %d", got, len(tt.states)) + } + if api.replicaSetReads.Load() != 0 || api.podReads.Load() != 0 { + t.Fatal("zero-replica readiness queried ReplicaSets or pods") + } + }) + } +} + +func TestWaitForDeploymentAvailable_ReplicaSetChangesBetweenReads(t *testing.T) { + for _, tt := range []struct { + name string + states []*appsv1.Deployment + }{ + {name: "scaler zeroes the ReplicaSet", states: []*appsv1.Deployment{waitDeployment(1), waitDeployment(0)}}, + {name: "active ReplicaSet appears on next poll", states: []*appsv1.Deployment{waitDeployment(1), waitDeployment(1)}}, + } { + t.Run(tt.name, func(t *testing.T) { + api := &waitAPI{deployments: tt.states, activeAfter: 2} + err := k8s.WaitForDeploymentAvailable(t.Context(), api.client(t), "ns", "fn", 5*time.Second) + if err != nil { + t.Fatal(err) + } + if api.deploymentReads.Load() != 2 { + t.Fatal("inconsistent ReplicaSet read was not retried") + } + }) + } +} + +func TestWaitForDeploymentAvailable_Errors(t *testing.T) { + stale := waitDeployment(0) + stale.Status.ObservedGeneration-- + for _, tt := range []struct { + name string + deployment *appsv1.Deployment + replicaStatus int + }{ + {name: "stale zero times out", deployment: stale}, + {name: "missing active ReplicaSet times out", deployment: waitDeployment(1)}, + {name: "ReplicaSet API error", deployment: waitDeployment(1), replicaStatus: http.StatusForbidden}, + } { + t.Run(tt.name, func(t *testing.T) { + api := &waitAPI{deployments: []*appsv1.Deployment{tt.deployment}, replicaStatus: tt.replicaStatus} + timeout := 100 * time.Millisecond + if tt.replicaStatus != 0 { + timeout = 5 * time.Second + } + err := k8s.WaitForDeploymentAvailable(t.Context(), api.client(t), "ns", "fn", timeout) + if tt.replicaStatus != 0 { + if !apierrors.IsForbidden(err) { + t.Fatalf("error = %v, want forbidden", err) + } + } else if !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("error = %v, want deadline exceeded", err) + } + }) + } +} + +func TestWaitForDeploymentAvailable_Cancelled(t *testing.T) { + stale := waitDeployment(0) + stale.Status.ObservedGeneration-- + api := &waitAPI{deployments: []*appsv1.Deployment{stale}} + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + timer := time.AfterFunc(100*time.Millisecond, cancel) + defer timer.Stop() + if err := k8s.WaitForDeploymentAvailable(ctx, api.client(t), "ns", "fn", 5*time.Second); !errors.Is(err, context.Canceled) { + t.Fatalf("error = %v, want canceled", err) + } +} + +func TestWaitForDeploymentAvailable_NonzeroReady(t *testing.T) { + api := &waitAPI{deployments: []*appsv1.Deployment{waitDeployment(1)}, activeAfter: 1} + if err := k8s.WaitForDeploymentAvailable(t.Context(), api.client(t), "ns", "fn", time.Second); err != nil { + t.Fatal(err) + } + if api.podReads.Load() != 1 { + t.Fatal("nonzero readiness did not check pods") + } +} + +func waitDeployment(replicas int32) *appsv1.Deployment { + return &appsv1.Deployment{ + ObjectMeta: metav1.ObjectMeta{Name: "fn", Namespace: "ns", Generation: 7}, + Spec: appsv1.DeploymentSpec{Replicas: &replicas, Selector: &metav1.LabelSelector{MatchLabels: map[string]string{"app": "fn"}}}, + Status: appsv1.DeploymentStatus{ + ObservedGeneration: 7, Replicas: replicas, UpdatedReplicas: replicas, ReadyReplicas: replicas, AvailableReplicas: replicas, + Conditions: []appsv1.DeploymentCondition{{Type: appsv1.DeploymentAvailable, Status: corev1.ConditionTrue}}, + }, + } +} + +// Deployment and ReplicaSet reads can observe different controller snapshots. +type waitAPI struct { + deployments []*appsv1.Deployment + activeAfter int32 + replicaStatus int + deploymentReads atomic.Int32 + replicaSetReads atomic.Int32 + podReads atomic.Int32 +} + +func (a *waitAPI) client(t *testing.T) *kubernetes.Clientset { + t.Helper() + server := httptest.NewServer(a) + t.Cleanup(server.Close) + client, err := kubernetes.NewForConfig(&rest.Config{Host: server.URL}) + if err != nil { + t.Fatal(err) + } + return client +} + +func (a *waitAPI) ServeHTTP(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + var response any + switch r.URL.Path { + case "/apis/apps/v1/namespaces/ns/deployments/fn", "/apis/apps/v1/namespaces/ns/deployments": + index := min(int(a.deploymentReads.Add(1))-1, len(a.deployments)-1) + d := *a.deployments[index] + if strings.HasSuffix(r.URL.Path, "/fn") { + response = d + } else { + response = appsv1.DeploymentList{Items: []appsv1.Deployment{d}} + } + case "/apis/apps/v1/namespaces/ns/replicasets": + a.replicaSetReads.Add(1) + if a.replicaStatus != 0 { + w.WriteHeader(a.replicaStatus) + response = metav1.Status{TypeMeta: metav1.TypeMeta{Kind: "Status", APIVersion: "v1"}, Status: "Failure", Reason: metav1.StatusReasonForbidden, Code: int32(a.replicaStatus), Message: "forbidden"} + break + } + replicas := int32(0) + if a.activeAfter != 0 && a.deploymentReads.Load() >= a.activeAfter { + replicas = 1 + } + response = appsv1.ReplicaSetList{Items: []appsv1.ReplicaSet{{ + ObjectMeta: metav1.ObjectMeta{Labels: map[string]string{"pod-template-hash": "abc"}}, + Spec: appsv1.ReplicaSetSpec{Replicas: &replicas}, + }}} + case "/api/v1/namespaces/ns/pods": + a.podReads.Add(1) + response = corev1.PodList{Items: []corev1.Pod{{ + ObjectMeta: metav1.ObjectMeta{Labels: map[string]string{"pod-template-hash": "abc"}}, + Status: corev1.PodStatus{Conditions: []corev1.PodCondition{{Type: corev1.PodReady, Status: corev1.ConditionTrue}}}, + }}} + default: + http.NotFound(w, r) + return + } + _ = json.NewEncoder(w).Encode(response) +} diff --git a/pkg/k8s/wait_test.go b/pkg/k8s/wait_test.go new file mode 100644 index 0000000000..8b37fa0cf1 --- /dev/null +++ b/pkg/k8s/wait_test.go @@ -0,0 +1,99 @@ +package k8s + +import ( + "context" + "testing" + + appsv1 "k8s.io/api/apps/v1" +) + +// zeroDeployment returns a Deployment wanting zero replicas at generation 2 +// with a fully reconciled all-zero status, unless mutated. The clientset is +// always nil in these tests: the zero-replica path must decide from status +// alone and never touch the API server (a nil dereference panics, so a pass +// proves no client use). +func zeroDeployment(mut func(*appsv1.Deployment)) *appsv1.Deployment { + zero := int32(0) + d := &appsv1.Deployment{ + Spec: appsv1.DeploymentSpec{ + Replicas: &zero, + }, + Status: appsv1.DeploymentStatus{ + ObservedGeneration: 2, + Replicas: 0, + UpdatedReplicas: 0, + ReadyReplicas: 0, + AvailableReplicas: 0, + UnavailableReplicas: 0, + }, + } + d.Name = "fn" + d.Namespace = "default" + d.Generation = 2 + if mut != nil { + mut(d) + } + return d +} + +// TestCheckIfDeploymentIsAvailable_ZeroReplicas covers issue #4068: a live +// Deployment at 0 desired replicas (e.g. a KEDA scaler idling a function to +// zero during the post-redeploy wait) must be judged on its reconciled +// status, not fail looking for an active ReplicaSet. +func TestCheckIfDeploymentIsAvailable_ZeroReplicas(t *testing.T) { + ctx := context.Background() + + t.Run("reconciled zero is ready without any client use", func(t *testing.T) { + ready, err := checkIfDeploymentIsAvailable(ctx, nil, zeroDeployment(nil)) + if err != nil { + t.Fatalf("expected no error, got %v", err) + } + if !ready { + t.Error("expected ready=true for a reconciled zero-replica Deployment") + } + }) + + t.Run("stale observedGeneration is not ready", func(t *testing.T) { + // A scaler changed replicas; status still describes generation 2. + d := zeroDeployment(func(d *appsv1.Deployment) { + d.Generation = 3 + d.Status.ObservedGeneration = 2 + }) + ready, err := checkIfDeploymentIsAvailable(ctx, nil, d) + if err != nil { + t.Fatalf("expected no error, got %v", err) + } + if ready { + t.Error("expected ready=false when status does not describe this generation") + } + }) + + t.Run("remaining replica from an older ReplicaSet is not ready", func(t *testing.T) { + // Scale-down not finished: one pod still counted in status.replicas + // (it belongs to an older ReplicaSet; the other counts are already + // zero). Must retry, not succeed. + d := zeroDeployment(func(d *appsv1.Deployment) { + d.Status.Replicas = 1 + }) + ready, err := checkIfDeploymentIsAvailable(ctx, nil, d) + if err != nil { + t.Fatalf("expected no error, got %v", err) + } + if ready { + t.Error("expected ready=false while a replica remains") + } + }) + + t.Run("unavailable replica is not ready", func(t *testing.T) { + d := zeroDeployment(func(d *appsv1.Deployment) { + d.Status.UnavailableReplicas = 1 + }) + ready, err := checkIfDeploymentIsAvailable(ctx, nil, d) + if err != nil { + t.Fatalf("expected no error, got %v", err) + } + if ready { + t.Error("expected ready=false with an unavailable replica") + } + }) +}