Merge pull request #101316 from ravisantoshgudimetla/add-minReadySeconds-impl

Add min ready seconds impl

test/integration/statefulset/statefulset_test.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"fmt"
 	"testing"
+	"time"
 
 	appsv1 "k8s.io/api/apps/v1"
 	v1 "k8s.io/api/core/v1"
@@ -27,11 +28,21 @@ import (
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/util/json"
 	"k8s.io/apimachinery/pkg/util/wait"
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	"k8s.io/client-go/dynamic"
+	clientset "k8s.io/client-go/kubernetes"
+	featuregatetesting "k8s.io/component-base/featuregate/testing"
 	apiservertesting "k8s.io/kubernetes/cmd/kube-apiserver/app/testing"
+	podutil "k8s.io/kubernetes/pkg/api/v1/pod"
+	"k8s.io/kubernetes/pkg/features"
 	"k8s.io/kubernetes/test/integration/framework"
 )
 
+const (
+	interval = 100 * time.Millisecond
+	timeout  = 60 * time.Second
+)
+
 // TestVolumeTemplateNoopUpdate ensures embedded StatefulSet objects with embedded PersistentVolumes can be updated
 func TestVolumeTemplateNoopUpdate(t *testing.T) {
 	// Start the server with default storage setup
@@ -226,3 +237,115 @@ func TestDeletingAndFailedPods(t *testing.T) {
 		t.Fatalf("failed to verify deleting pod %s has been replaced with a new non-deleting pod: %v", deletingPod.Name, err)
 	}
 }
+
+func TestStatefulSetAvailable(t *testing.T) {
+	tests := []struct {
+		name           string
+		totalReplicas  int32
+		readyReplicas  int32
+		activeReplicas int32
+		enabled        bool
+	}{
+		{
+			name:           "When feature gate is enabled, only certain replicas would become active",
+			totalReplicas:  4,
+			readyReplicas:  3,
+			activeReplicas: 2,
+			enabled:        true,
+		},
+		{
+			name:           "When feature gate is disabled, all the ready replicas would become active",
+			totalReplicas:  4,
+			readyReplicas:  3,
+			activeReplicas: 3,
+			enabled:        false,
+		},
+	}
+	for _, test := range tests {
+		defer featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.StatefulSetMinReadySeconds, test.enabled)()
+		s, closeFn, rm, informers, c := scSetup(t)
+		defer closeFn()
+		ns := framework.CreateTestingNamespace("test-available-pods", s, t)
+		defer framework.DeleteTestingNamespace(ns, s, t)
+		stopCh := runControllerAndInformers(rm, informers)
+		defer close(stopCh)
+
+		labelMap := labelMap()
+		sts := newSTS("sts", ns.Name, 4)
+		sts.Spec.MinReadySeconds = int32(3600)
+		stss, _ := createSTSsPods(t, c, []*appsv1.StatefulSet{sts}, []*v1.Pod{})
+		sts = stss[0]
+		waitSTSStable(t, c, sts)
+
+		// Verify STS creates 4 pods
+		podClient := c.CoreV1().Pods(ns.Name)
+		pods := getPods(t, podClient, labelMap)
+		if len(pods.Items) != 4 {
+			t.Fatalf("len(pods) = %d, want 4", len(pods.Items))
+		}
+
+		// Separate 3 pods into their own list
+		firstPodList := &v1.PodList{Items: pods.Items[:1]}
+		secondPodList := &v1.PodList{Items: pods.Items[1:2]}
+		thirdPodList := &v1.PodList{Items: pods.Items[2:]}
+		// First pod: Running, but not Ready
+		// by setting the Ready condition to false with LastTransitionTime to be now
+		setPodsReadyCondition(t, c, firstPodList, v1.ConditionFalse, time.Now())
+		// Second pod: Running and Ready, but not Available
+		// by setting LastTransitionTime to now
+		setPodsReadyCondition(t, c, secondPodList, v1.ConditionTrue, time.Now())
+		// Third pod: Running, Ready, and Available
+		// by setting LastTransitionTime to more than 3600 seconds ago
+		setPodsReadyCondition(t, c, thirdPodList, v1.ConditionTrue, time.Now().Add(-120*time.Minute))
+
+		stsClient := c.AppsV1().StatefulSets(ns.Name)
+		if err := wait.PollImmediate(interval, timeout, func() (bool, error) {
+			newSts, err := stsClient.Get(context.TODO(), sts.Name, metav1.GetOptions{})
+			if err != nil {
+				return false, err
+			}
+			// Verify 4 pods exist, 3 pods are Ready, and 2 pods are Available
+			return newSts.Status.Replicas == test.totalReplicas && newSts.Status.ReadyReplicas == test.readyReplicas && newSts.Status.AvailableReplicas == test.activeReplicas, nil
+		}); err != nil {
+			t.Fatalf("Failed to verify number of Replicas, ReadyReplicas and AvailableReplicas of rs %s to be as expected: %v", sts.Name, err)
+		}
+	}
+}
+
+func setPodsReadyCondition(t *testing.T, clientSet clientset.Interface, pods *v1.PodList, conditionStatus v1.ConditionStatus, lastTransitionTime time.Time) {
+	replicas := int32(len(pods.Items))
+	var readyPods int32
+	err := wait.PollImmediate(interval, timeout, func() (bool, error) {
+		readyPods = 0
+		for i := range pods.Items {
+			pod := &pods.Items[i]
+			if podutil.IsPodReady(pod) {
+				readyPods++
+				continue
+			}
+			pod.Status.Phase = v1.PodRunning
+			_, condition := podutil.GetPodCondition(&pod.Status, v1.PodReady)
+			if condition != nil {
+				condition.Status = conditionStatus
+				condition.LastTransitionTime = metav1.Time{Time: lastTransitionTime}
+			} else {
+				condition = &v1.PodCondition{
+					Type:               v1.PodReady,
+					Status:             conditionStatus,
+					LastTransitionTime: metav1.Time{Time: lastTransitionTime},
+				}
+				pod.Status.Conditions = append(pod.Status.Conditions, *condition)
+			}
+			_, err := clientSet.CoreV1().Pods(pod.Namespace).UpdateStatus(context.TODO(), pod, metav1.UpdateOptions{})
+			if err != nil {
+				// When status fails to be updated, we continue to next pod
+				continue
+			}
+			readyPods++
+		}
+		return readyPods >= replicas, nil
+	})
+	if err != nil {
+		t.Fatalf("failed to mark all StatefulSet pods to ready: %v", err)
+	}
+}