Implements kubectl rollout status and history for StatefulSet

pkg/kubectl/history.go

@@ -52,6 +52,8 @@ func HistoryViewerFor(kind schema.GroupKind, c clientset.Interface) (HistoryView
 	switch kind {
 	case extensions.Kind("Deployment"), apps.Kind("Deployment"):
 		return &DeploymentHistoryViewer{c}, nil
+	case apps.Kind("StatefulSet"):
+		return &StatefulSetHistoryViewer{c}, nil
 	case extensions.Kind("DaemonSet"):
 		return &DaemonSetHistoryViewer{c}, nil
 	}
@@ -200,6 +202,58 @@ func (h *DaemonSetHistoryViewer) ViewHistory(namespace, name string, revision in
 	})
 }
 
+type StatefulSetHistoryViewer struct {
+	c clientset.Interface
+}
+
+func getOwner(revision apps.ControllerRevision) *metav1.OwnerReference {
+	ownerRefs := revision.GetOwnerReferences()
+	for i := range ownerRefs {
+		owner := &ownerRefs[i]
+		if owner.Controller != nil && *owner.Controller == true {
+			return owner
+		}
+	}
+	return nil
+}
+
+// ViewHistory returns a list of the revision history of a statefulset
+// TODO: this should be a describer
+// TODO: needs to implement detailed revision view
+func (h *StatefulSetHistoryViewer) ViewHistory(namespace, name string, revision int64) (string, error) {
+
+	sts, err := h.c.Apps().StatefulSets(namespace).Get(name, metav1.GetOptions{})
+	if err != nil {
+		return "", fmt.Errorf("failed to retrieve statefulset %s", err)
+	}
+	selector, err := metav1.LabelSelectorAsSelector(sts.Spec.Selector)
+	if err != nil {
+		return "", fmt.Errorf("failed to retrieve statefulset history %s", err)
+	}
+	revisions, err := h.c.Apps().ControllerRevisions(namespace).List(metav1.ListOptions{LabelSelector: selector.String()})
+	if err != nil {
+		return "", fmt.Errorf("failed to retrieve statefulset history %s", err)
+	}
+	if len(revisions.Items) <= 0 {
+		return "No rollout history found.", nil
+	}
+	revisionNumbers := make([]int64, len(revisions.Items))
+	for i := range revisions.Items {
+		if owner := getOwner(revisions.Items[i]); owner != nil && owner.UID == sts.UID {
+			revisionNumbers[i] = revisions.Items[i].Revision
+		}
+	}
+	sliceutil.SortInts64(revisionNumbers)
+
+	return tabbedString(func(out io.Writer) error {
+		fmt.Fprintf(out, "REVISION\n")
+		for _, r := range revisionNumbers {
+			fmt.Fprintf(out, "%d\n", r)
+		}
+		return nil
+	})
+}
+
 // controlledHistories returns all ControllerRevisions controlled by the given DaemonSet
 // TODO: Use external version DaemonSet instead when #3955 is fixed
 func controlledHistories(c clientset.Interface, ds *extensions.DaemonSet) ([]*appsv1beta1.ControllerRevision, error) {

pkg/kubectl/rollout_status.go

@@ -24,6 +24,7 @@ import (
 	"k8s.io/kubernetes/pkg/apis/apps"
 	"k8s.io/kubernetes/pkg/apis/extensions"
 	"k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset"
+	appsclient "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset/typed/apps/internalversion"
 	extensionsclient "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset/typed/extensions/internalversion"
 	"k8s.io/kubernetes/pkg/controller/deployment/util"
 )
@@ -39,6 +40,8 @@ func StatusViewerFor(kind schema.GroupKind, c internalclientset.Interface) (Stat
 		return &DeploymentStatusViewer{c.Extensions()}, nil
 	case extensions.Kind("DaemonSet"):
 		return &DaemonSetStatusViewer{c.Extensions()}, nil
+	case apps.Kind("StatefulSet"):
+		return &StatefulSetStatusViewer{c.Apps()}, nil
 	}
 	return nil, fmt.Errorf("no status viewer has been implemented for %v", kind)
 }
@@ -51,6 +54,10 @@ type DaemonSetStatusViewer struct {
 	c extensionsclient.DaemonSetsGetter
 }
 
+type StatefulSetStatusViewer struct {
+	c appsclient.StatefulSetsGetter
+}
+
 // Status returns a message describing deployment status, and a bool value indicating if the status is considered done
 func (s *DeploymentStatusViewer) Status(namespace, name string, revision int64) (string, bool, error) {
 	deployment, err := s.c.Deployments(namespace).Get(name, metav1.GetOptions{})
@@ -107,3 +114,34 @@ func (s *DaemonSetStatusViewer) Status(namespace, name string, revision int64) (
 	}
 	return fmt.Sprintf("Waiting for daemon set spec update to be observed...\n"), false, nil
 }
+
+// Status returns a message describing statefulset status, and a bool value indicating if the status is considered done
+func (s *StatefulSetStatusViewer) Status(namespace, name string, revision int64) (string, bool, error) {
+	sts, err := s.c.StatefulSets(namespace).Get(name, metav1.GetOptions{})
+	if err != nil {
+		return "", false, err
+	}
+	if sts.Spec.UpdateStrategy.Type == apps.OnDeleteStatefulSetStrategyType {
+		return "", true, fmt.Errorf("%s updateStrategy does not have a Status`", apps.OnDeleteStatefulSetStrategyType)
+	}
+	if sts.Status.ObservedGeneration == nil || sts.Generation > *sts.Status.ObservedGeneration {
+		return "Waiting for statefulset spec update to be observed...\n", false, nil
+	}
+	if sts.Status.ReadyReplicas < sts.Spec.Replicas {
+		return fmt.Sprintf("Waiting for %d pods to be ready...\n", sts.Spec.Replicas-sts.Status.ReadyReplicas), false, nil
+	}
+	if sts.Spec.UpdateStrategy.Type == apps.PartitionStatefulSetStrategyType {
+		if sts.Status.UpdatedReplicas < (sts.Spec.Replicas - sts.Spec.UpdateStrategy.Partition.Ordinal) {
+			return fmt.Sprintf("Waiting for partitioned roll out to finish: %d out of %d new pods have been updated...\n",
+				sts.Status.UpdatedReplicas, (sts.Spec.Replicas - sts.Spec.UpdateStrategy.Partition.Ordinal)), false, nil
+		}
+		return fmt.Sprintf("partitioned roll out complete: %d new pods have been updated...\n",
+			sts.Status.UpdatedReplicas), false, nil
+	}
+	if sts.Status.UpdateRevision != sts.Status.CurrentRevision {
+		return fmt.Sprintf("waiting for statefulset rolling update to complete %d pods at revision %s...\n",
+			sts.Status.UpdatedReplicas, sts.Status.UpdateRevision), false, nil
+	}
+	return fmt.Sprintf("statefulset rolling update complete %d pods at revision %s...\n", sts.Status.CurrentReplicas, sts.Status.CurrentRevision), true, nil
+
+}

pkg/kubectl/rollout_status_test.go

@@ -17,9 +17,12 @@ limitations under the License.
 package kubectl
 
 import (
+	"fmt"
 	"testing"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/kubernetes/pkg/api"
+	"k8s.io/kubernetes/pkg/apis/apps"
 	"k8s.io/kubernetes/pkg/apis/extensions"
 	"k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset/fake"
 )
@@ -231,6 +234,163 @@ func TestDaemonSetStatusViewerStatus(t *testing.T) {
 	}
 }
 
+func TestStatefulSetStatusViewerStatus(t *testing.T) {
+	tests := []struct {
+		name       string
+		generation int64
+		strategy   apps.StatefulSetUpdateStrategy
+		status     apps.StatefulSetStatus
+		msg        string
+		done       bool
+		err        bool
+	}{
+		{
+			name:       "on delete returns an error",
+			generation: 1,
+			strategy:   apps.StatefulSetUpdateStrategy{Type: apps.OnDeleteStatefulSetStrategyType},
+			status: apps.StatefulSetStatus{
+				ObservedGeneration: func() *int64 {
+					generation := int64(1)
+					return &generation
+				}(),
+				Replicas:        0,
+				ReadyReplicas:   1,
+				CurrentReplicas: 0,
+				UpdatedReplicas: 0,
+			},
+
+			msg:  "",
+			done: true,
+			err:  true,
+		},
+		{
+			name:       "unobserved update is not complete",
+			generation: 2,
+			strategy:   apps.StatefulSetUpdateStrategy{Type: apps.RollingUpdateStatefulSetStrategyType},
+			status: apps.StatefulSetStatus{
+				ObservedGeneration: func() *int64 {
+					generation := int64(1)
+					return &generation
+				}(),
+				Replicas:        3,
+				ReadyReplicas:   3,
+				CurrentReplicas: 3,
+				UpdatedReplicas: 0,
+			},
+
+			msg:  "Waiting for statefulset spec update to be observed...\n",
+			done: false,
+			err:  false,
+		},
+		{
+			name:       "if all pods are not ready the update is not complete",
+			generation: 1,
+			strategy:   apps.StatefulSetUpdateStrategy{Type: apps.RollingUpdateStatefulSetStrategyType},
+			status: apps.StatefulSetStatus{
+				ObservedGeneration: func() *int64 {
+					generation := int64(2)
+					return &generation
+				}(),
+				Replicas:        3,
+				ReadyReplicas:   2,
+				CurrentReplicas: 3,
+				UpdatedReplicas: 0,
+			},
+
+			msg:  fmt.Sprintf("Waiting for %d pods to be ready...\n", 1),
+			done: false,
+			err:  false,
+		},
+		{
+			name:       "partition update completes when all replicas above the partition are updated",
+			generation: 1,
+			strategy: apps.StatefulSetUpdateStrategy{Type: apps.PartitionStatefulSetStrategyType,
+				Partition: func() *apps.PartitionStatefulSetStrategy {
+					return &apps.PartitionStatefulSetStrategy{Ordinal: 2}
+				}()},
+			status: apps.StatefulSetStatus{
+				ObservedGeneration: func() *int64 {
+					generation := int64(2)
+					return &generation
+				}(),
+				Replicas:        3,
+				ReadyReplicas:   3,
+				CurrentReplicas: 2,
+				UpdatedReplicas: 1,
+			},
+
+			msg:  fmt.Sprintf("partitioned roll out complete: %d new pods have been updated...\n", 1),
+			done: true,
+			err:  false,
+		},
+		{
+			name:       "partition update is in progress if all pods above the partition have not been updated",
+			generation: 1,
+			strategy: apps.StatefulSetUpdateStrategy{Type: apps.PartitionStatefulSetStrategyType,
+				Partition: func() *apps.PartitionStatefulSetStrategy {
+					return &apps.PartitionStatefulSetStrategy{Ordinal: 2}
+				}()},
+			status: apps.StatefulSetStatus{
+				ObservedGeneration: func() *int64 {
+					generation := int64(2)
+					return &generation
+				}(),
+				Replicas:        3,
+				ReadyReplicas:   3,
+				CurrentReplicas: 3,
+				UpdatedReplicas: 0,
+			},
+
+			msg:  fmt.Sprintf("Waiting for partitioned roll out to finish: %d out of %d new pods have been updated...\n", 0, 1),
+			done: true,
+			err:  false,
+		},
+		{
+			name:       "update completes when all replicas are current",
+			generation: 1,
+			strategy:   apps.StatefulSetUpdateStrategy{Type: apps.RollingUpdateStatefulSetStrategyType},
+			status: apps.StatefulSetStatus{
+				ObservedGeneration: func() *int64 {
+					generation := int64(2)
+					return &generation
+				}(),
+				Replicas:        3,
+				ReadyReplicas:   3,
+				CurrentReplicas: 3,
+				UpdatedReplicas: 3,
+				CurrentRevision: "foo",
+				UpdateRevision:  "foo",
+			},
+
+			msg:  fmt.Sprintf("statefulset rolling update complete %d pods at revision %s...\n", 3, "foo"),
+			done: true,
+			err:  false,
+		},
+	}
+	for i := range tests {
+		test := tests[i]
+		s := newStatefulSet(3)
+		s.Status = test.status
+		s.Spec.UpdateStrategy = test.strategy
+		s.Generation = test.generation
+		client := fake.NewSimpleClientset(s).Apps()
+		dsv := &StatefulSetStatusViewer{c: client}
+		msg, done, err := dsv.Status(s.Namespace, s.Name, 0)
+		if test.err && err == nil {
+			t.Fatalf("%s: expected error", test.name)
+		}
+		if !test.err && err != nil {
+			t.Fatalf("%s: %s", test.name, err)
+		}
+		if done && !test.done {
+			t.Errorf("%s: want done %v got %v", test.name, done, test.done)
+		}
+		if msg != test.msg {
+			t.Errorf("%s: want message %s got %s", test.name, test.msg, msg)
+		}
+	}
+}
+
 func TestDaemonSetStatusViewerStatusWithWrongUpdateStrategyType(t *testing.T) {
 	d := &extensions.DaemonSet{
 		ObjectMeta: metav1.ObjectMeta{
@@ -252,3 +412,36 @@ func TestDaemonSetStatusViewerStatusWithWrongUpdateStrategyType(t *testing.T) {
 		t.Errorf("Status for daemon sets with UpdateStrategy type different than RollingUpdate should return error. Instead got: msg: %s\ndone: %t\n err: %v", msg, done, err)
 	}
 }
+
+func newStatefulSet(replicas int32) *apps.StatefulSet {
+	return &apps.StatefulSet{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "foo",
+			Namespace: metav1.NamespaceDefault,
+			Labels:    map[string]string{"a": "b"},
+		},
+		Spec: apps.StatefulSetSpec{
+			PodManagementPolicy: apps.OrderedReadyPodManagement,
+			Selector:            &metav1.LabelSelector{MatchLabels: map[string]string{"a": "b"}},
+			Template: api.PodTemplateSpec{
+				ObjectMeta: metav1.ObjectMeta{
+					Labels: map[string]string{"a": "b"},
+				},
+				Spec: api.PodSpec{
+					Containers: []api.Container{
+						{
+							Name:            "test",
+							Image:           "test_image",
+							ImagePullPolicy: api.PullIfNotPresent,
+						},
+					},
+					RestartPolicy: api.RestartPolicyAlways,
+					DNSPolicy:     api.DNSClusterFirst,
+				},
+			},
+			Replicas:       replicas,
+			UpdateStrategy: apps.StatefulSetUpdateStrategy{Type: apps.RollingUpdateStatefulSetStrategyType},
+		},
+		Status: apps.StatefulSetStatus{},
+	}
+}