Merge pull request #49674 from crimsonfaith91/rollout

hack/testdata/rollingupdate-statefulset-rv2.yaml

+apiVersion: apps/v1beta2
+kind: StatefulSet
+metadata:
+  name: nginx
+spec:
+  selector:
+    matchLabels:
+      app: nginx-statefulset
+  updateStrategy:
+    type: RollingUpdate
+  serviceName: "nginx"
+  replicas: 0
+  template:
+    metadata:
+      labels:
+        app: nginx-statefulset
+    spec:
+      terminationGracePeriodSeconds: 5
+      containers:
+      - name: nginx
+        image: gcr.io/google_containers/nginx-slim:0.8
+        ports:
+        - containerPort: 80
+          name: web
+        command:
+        - sh
+        - -c
+        - 'while true; do sleep 1; done'
+      - name: pause
+        image: gcr.io/google-containers/pause:2.0
+        ports:
+        - containerPort: 81
+          name: web-2

hack/testdata/nginx-statefulset.yaml → hack/testdata/rollingupdate-statefulset.yaml

-# A headless service to create DNS records
-apiVersion: v1
-kind: Service
+apiVersion: apps/v1beta2
+kind: StatefulSet
 metadata:
-  annotations:
-    service.alpha.kubernetes.io/tolerate-unready-endpoints: "true"
   name: nginx
-  labels:
-    app: nginx-statefulset
 spec:
-  ports:
-  - port: 80
-    name: web
-  # *.nginx.default.svc.cluster.local
-  clusterIP: None
   selector:
+    matchLabels:
       app: nginx-statefulset
----
-apiVersion: apps/v1beta1
-kind: StatefulSet
-metadata:
-  name: nginx
-spec:
+  updateStrategy:
+    type: RollingUpdate
   serviceName: "nginx"
   replicas: 0
   template:
@@ -28,7 +15,7 @@ spec:
       labels:
         app: nginx-statefulset
     spec:
-      terminationGracePeriodSeconds: 0
+      terminationGracePeriodSeconds: 5
       containers:
       - name: nginx
         image: gcr.io/google_containers/nginx-slim:0.7

pkg/controller/statefulset/stateful_set_control.go

@@ -256,11 +256,11 @@ func (ssc *defaultStatefulSetControl) updateStatefulSet(
 	collisionCount int32,
 	pods []*v1.Pod) (*apps.StatefulSetStatus, error) {
 	// get the current and update revisions of the set.
-	currentSet, err := applyRevision(set, currentRevision)
+	currentSet, err := ApplyRevision(set, currentRevision)
 	if err != nil {
 		return nil, err
 	}
-	updateSet, err := applyRevision(set, updateRevision)
+	updateSet, err := ApplyRevision(set, updateRevision)
 	if err != nil {
 		return nil, err
 	}

pkg/controller/statefulset/stateful_set_control_test.go

@@ -1281,7 +1281,7 @@ func TestStatefulSetControlRollback(t *testing.T) {
 			t.Fatalf("%s: %s", test.name, err)
 		}
 		history.SortControllerRevisions(revisions)
-		set, err = applyRevision(set, revisions[0])
+		set, err = ApplyRevision(set, revisions[0])
 		if err != nil {
 			t.Fatalf("%s: %s", test.name, err)
 		}

pkg/controller/statefulset/stateful_set_utils.go

@@ -17,6 +17,7 @@ limitations under the License.
 package statefulset
 
 import (
+	"bytes"
 	"encoding/json"
 	"fmt"
 	"regexp"
@@ -266,6 +267,15 @@ func newVersionedStatefulSetPod(currentSet, updateSet *apps.StatefulSet, current
 	return pod
 }
 
+// Match check if the given StatefulSet's template matches the template stored in the given history.
+func Match(ss *apps.StatefulSet, history *apps.ControllerRevision) (bool, error) {
+	patch, err := getPatch(ss)
+	if err != nil {
+		return false, err
+	}
+	return bytes.Equal(patch, history.Data.Raw), nil
+}
+
 // getPatch returns a strategic merge patch that can be applied to restore a StatefulSet to a
 // previous version. If the returned error is nil the patch is valid. The current state that we save is just the
 // PodSpecTemplate. We can modify this later to encompass more state (or less) and remain compatible with previously
@@ -319,9 +329,9 @@ func newRevision(set *apps.StatefulSet, revision int64, collisionCount *int32) (
 	return cr, nil
 }
 
-// applyRevision returns a new StatefulSet constructed by restoring the state in revision to set. If the returned error
+// ApplyRevision returns a new StatefulSet constructed by restoring the state in revision to set. If the returned error
 // is nil, the returned StatefulSet is valid.
-func applyRevision(set *apps.StatefulSet, revision *apps.ControllerRevision) (*apps.StatefulSet, error) {
+func ApplyRevision(set *apps.StatefulSet, revision *apps.ControllerRevision) (*apps.StatefulSet, error) {
 	obj, err := scheme.Scheme.DeepCopy(set)
 	if err != nil {
 		return nil, err

pkg/controller/statefulset/stateful_set_utils_test.go

@@ -258,7 +258,7 @@ func TestCreateApplyRevision(t *testing.T) {
 	key := "foo"
 	expectedValue := "bar"
 	set.Annotations[key] = expectedValue
-	restoredSet, err := applyRevision(set, revision)
+	restoredSet, err := ApplyRevision(set, revision)
 	if err != nil {
 		t.Fatal(err)
 	}

pkg/kubectl/BUILD

@@ -129,6 +129,7 @@ go_library(
         "//pkg/client/unversioned:go_default_library",
         "//pkg/controller/daemon:go_default_library",
         "//pkg/controller/deployment/util:go_default_library",
+        "//pkg/controller/statefulset:go_default_library",
         "//pkg/credentialprovider:go_default_library",
         "//pkg/kubectl/resource:go_default_library",
         "//pkg/kubectl/util:go_default_library",

pkg/kubectl/cmd/rollout/rollout.go

@@ -42,6 +42,7 @@ var (
 
 		   * deployments
 		   * daemonsets
+		   * statefulsets
 		`)
 )
 

pkg/kubectl/cmd/rollout/rollout_history.go

@@ -44,7 +44,7 @@ var (
 func NewCmdRolloutHistory(f cmdutil.Factory, out io.Writer) *cobra.Command {
 	options := &resource.FilenameOptions{}
 
-	validArgs := []string{"deployment", "daemonset"}
+	validArgs := []string{"deployment", "daemonset", "statefulset"}
 	argAliases := kubectl.ResourceAliases(validArgs)
 
 	cmd := &cobra.Command{

pkg/kubectl/cmd/rollout/rollout_status.go

@@ -50,7 +50,7 @@ var (
 func NewCmdRolloutStatus(f cmdutil.Factory, out io.Writer) *cobra.Command {
 	options := &resource.FilenameOptions{}
 
-	validArgs := []string{"deployment", "daemonset"}
+	validArgs := []string{"deployment", "daemonset", "statefulset"}
 	argAliases := kubectl.ResourceAliases(validArgs)
 
 	cmd := &cobra.Command{

pkg/kubectl/cmd/rollout/rollout_undo.go

@@ -64,7 +64,7 @@ var (
 func NewCmdRolloutUndo(f cmdutil.Factory, out io.Writer) *cobra.Command {
 	options := &UndoOptions{}
 
-	validArgs := []string{"deployment", "daemonset"}
+	validArgs := []string{"deployment", "daemonset", "statefulset"}
 	argAliases := kubectl.ResourceAliases(validArgs)
 
 	cmd := &cobra.Command{

pkg/kubectl/history.go

@@ -153,9 +153,9 @@ type DaemonSetHistoryViewer struct {
 // ViewHistory returns a revision-to-history map as the revision history of a deployment
 // TODO: this should be a describer
 func (h *DaemonSetHistoryViewer) ViewHistory(namespace, name string, revision int64) (string, error) {
-	versionedExtensionsClient := versionedExtensionsClientV1beta1(h.c)
 	versionedAppsClient := versionedAppsClientV1beta1(h.c)
-	ds, allHistory, err := controlledHistories(versionedExtensionsClient, versionedAppsClient, namespace, name)
+	versionedExtensionsClient := versionedExtensionsClientV1beta1(h.c)
+	versionedObj, allHistory, err := controlledHistories(versionedAppsClient, versionedExtensionsClient, namespace, name, "DaemonSet")
 	if err != nil {
 		return "", fmt.Errorf("unable to find history controlled by DaemonSet %s: %v", name, err)
 	}
@@ -175,7 +175,13 @@ func (h *DaemonSetHistoryViewer) ViewHistory(namespace, name string, revision in
 		if !ok {
 			return "", fmt.Errorf("unable to find the specified revision")
 		}
-		dsOfHistory, err := applyHistory(ds, history)
+
+		versionedDS, ok := versionedObj.(*extensionsv1beta1.DaemonSet)
+		if !ok {
+			return "", fmt.Errorf("unexpected non-DaemonSet object returned: %v", versionedDS)
+		}
+
+		dsOfHistory, err := applyHistory(versionedDS, history)
 		if err != nil {
 			return "", fmt.Errorf("unable to parse history %s", history.Name)
 		}
@@ -256,29 +262,51 @@ func (h *StatefulSetHistoryViewer) ViewHistory(namespace, name string, revision
 	})
 }
 
-// controlledHistories returns all ControllerRevisions controlled by the given DaemonSet
-func controlledHistories(extensions clientextensionsv1beta1.ExtensionsV1beta1Interface, apps clientappsv1beta1.AppsV1beta1Interface, namespace, name string) (*extensionsv1beta1.DaemonSet, []*appsv1beta1.ControllerRevision, error) {
+// controlledHistories returns all ControllerRevisions controlled by the given API object
+func controlledHistories(apps clientappsv1beta1.AppsV1beta1Interface, extensions clientextensionsv1beta1.ExtensionsV1beta1Interface, namespace, name, kind string) (runtime.Object, []*appsv1beta1.ControllerRevision, error) {
+	var obj runtime.Object
+	var labelSelector *metav1.LabelSelector
+
+	switch kind {
+	case "DaemonSet":
 		ds, err := extensions.DaemonSets(namespace).Get(name, metav1.GetOptions{})
 		if err != nil {
 			return nil, nil, fmt.Errorf("failed to retrieve DaemonSet %s: %v", name, err)
 		}
+		labelSelector = ds.Spec.Selector
+		obj = ds
+	case "StatefulSet":
+		ss, err := apps.StatefulSets(namespace).Get(name, metav1.GetOptions{})
+		if err != nil {
+			return nil, nil, fmt.Errorf("failed to retrieve StatefulSet %s: %v", name, err)
+		}
+		labelSelector = ss.Spec.Selector
+		obj = ss
+	default:
+		return nil, nil, fmt.Errorf("unsupported API object kind: %s", kind)
+	}
+
 	var result []*appsv1beta1.ControllerRevision
-	selector, err := metav1.LabelSelectorAsSelector(ds.Spec.Selector)
+	selector, err := metav1.LabelSelectorAsSelector(labelSelector)
 	if err != nil {
 		return nil, nil, err
 	}
-	historyList, err := apps.ControllerRevisions(ds.Namespace).List(metav1.ListOptions{LabelSelector: selector.String()})
+	historyList, err := apps.ControllerRevisions(namespace).List(metav1.ListOptions{LabelSelector: selector.String()})
 	if err != nil {
 		return nil, nil, err
 	}
+	accessor, err := meta.Accessor(obj)
+	if err != nil {
+		return nil, nil, fmt.Errorf("failed to obtain accessor for %s named %s: %v", kind, name, err)
+	}
 	for i := range historyList.Items {
 		history := historyList.Items[i]
-		// Only add history that belongs to the DaemonSet
-		if metav1.IsControlledBy(&history, ds) {
+		// Only add history that belongs to the API object
+		if metav1.IsControlledBy(&history, accessor) {
 			result = append(result, &history)
 		}
 	}
-	return ds, result, nil
+	return obj, result, nil
 }
 
 // applyHistory returns a specific revision of DaemonSet by applying the given history to a copy of the given DaemonSet

pkg/kubectl/rollback.go

@@ -39,6 +39,7 @@ import (
 	clientset "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset"
 	"k8s.io/kubernetes/pkg/controller/daemon"
 	deploymentutil "k8s.io/kubernetes/pkg/controller/deployment/util"
+	"k8s.io/kubernetes/pkg/controller/statefulset"
 	sliceutil "k8s.io/kubernetes/pkg/kubectl/util/slice"
 	printersinternal "k8s.io/kubernetes/pkg/printers/internalversion"
 )
@@ -59,6 +60,8 @@ func RollbackerFor(kind schema.GroupKind, c clientset.Interface) (Rollbacker, er
 		return &DeploymentRollbacker{c}, nil
 	case extensions.Kind("DaemonSet"), apps.Kind("DaemonSet"):
 		return &DaemonSetRollbacker{c}, nil
+	case apps.Kind("StatefulSet"):
+		return &StatefulSetRollbacker{c}, nil
 	}
 	return nil, fmt.Errorf("no rollbacker has been implemented for %q", kind)
 }
@@ -221,32 +224,22 @@ func (r *DaemonSetRollbacker) Rollback(obj runtime.Object, updatedAnnotations ma
 	if !ok {
 		return "", fmt.Errorf("passed object is not a DaemonSet: %#v", obj)
 	}
-	versionedExtensionsClient := versionedExtensionsClientV1beta1(r.c)
 	versionedAppsClient := versionedAppsClientV1beta1(r.c)
-	versionedDS, allHistory, err := controlledHistories(versionedExtensionsClient, versionedAppsClient, ds.Namespace, ds.Name)
+	versionedExtensionsClient := versionedExtensionsClientV1beta1(r.c)
+	versionedObj, allHistory, err := controlledHistories(versionedAppsClient, versionedExtensionsClient, ds.Namespace, ds.Name, "DaemonSet")
 	if err != nil {
 		return "", fmt.Errorf("unable to find history controlled by DaemonSet %s: %v", ds.Name, err)
 	}
+	versionedDS, ok := versionedObj.(*externalextensions.DaemonSet)
+	if !ok {
+		return "", fmt.Errorf("unexpected non-DaemonSet object returned: %v", versionedDS)
+	}
 
 	if toRevision == 0 && len(allHistory) <= 1 {
 		return "", fmt.Errorf("no last revision to roll back to")
 	}
 
-	// Find the history to rollback to
-	var toHistory *appsv1beta1.ControllerRevision
-	if toRevision == 0 {
-		// If toRevision == 0, find the latest revision (2nd max)
-		sort.Sort(historiesByRevision(allHistory))
-		toHistory = allHistory[len(allHistory)-2]
-	} else {
-		for _, h := range allHistory {
-			if h.Revision == toRevision {
-				// If toRevision != 0, find the history with matching revision
-				toHistory = h
-				break
-			}
-		}
-	}
+	toHistory := findHistory(toRevision, allHistory)
 	if toHistory == nil {
 		return "", revisionNotFoundErr(toRevision)
 	}
@@ -256,14 +249,7 @@ func (r *DaemonSetRollbacker) Rollback(obj runtime.Object, updatedAnnotations ma
 		if err != nil {
 			return "", err
 		}
-		content := bytes.NewBuffer([]byte{})
-		w := printersinternal.NewPrefixWriter(content)
-		internalTemplate := &api.PodTemplateSpec{}
-		if err := apiv1.Convert_v1_PodTemplateSpec_To_api_PodTemplateSpec(&appliedDS.Spec.Template, internalTemplate, nil); err != nil {
-			return "", fmt.Errorf("failed to convert podtemplate while printing: %v", err)
-		}
-		printersinternal.DescribePodTemplate(internalTemplate, w)
-		return fmt.Sprintf("will roll back to %s", content.String()), nil
+		return printPodTemplate(&appliedDS.Spec.Template)
 	}
 
 	// Skip if the revision already matches current DaemonSet
@@ -283,6 +269,104 @@ func (r *DaemonSetRollbacker) Rollback(obj runtime.Object, updatedAnnotations ma
 	return rollbackSuccess, nil
 }
 
+type StatefulSetRollbacker struct {
+	c clientset.Interface
+}
+
+// toRevision is a non-negative integer, with 0 being reserved to indicate rolling back to previous configuration
+func (r *StatefulSetRollbacker) Rollback(obj runtime.Object, updatedAnnotations map[string]string, toRevision int64, dryRun bool) (string, error) {
+	if toRevision < 0 {
+		return "", revisionNotFoundErr(toRevision)
+	}
+
+	ss, ok := obj.(*apps.StatefulSet)
+	if !ok {
+		return "", fmt.Errorf("passed object is not a StatefulSet: %#v", obj)
+	}
+	versionedAppsClient := versionedAppsClientV1beta1(r.c)
+	versionedExtensionsClient := versionedExtensionsClientV1beta1(r.c)
+	versionedObj, allHistory, err := controlledHistories(versionedAppsClient, versionedExtensionsClient, ss.Namespace, ss.Name, "StatefulSet")
+	if err != nil {
+		return "", fmt.Errorf("unable to find history controlled by StatefulSet %s: %v", ss.Name, err)
+	}
+
+	versionedSS, ok := versionedObj.(*appsv1beta1.StatefulSet)
+	if !ok {
+		return "", fmt.Errorf("unexpected non-StatefulSet object returned: %v", versionedSS)
+	}
+
+	if toRevision == 0 && len(allHistory) <= 1 {
+		return "", fmt.Errorf("no last revision to roll back to")
+	}
+
+	toHistory := findHistory(toRevision, allHistory)
+	if toHistory == nil {
+		return "", revisionNotFoundErr(toRevision)
+	}
+
+	if dryRun {
+		appliedSS, err := statefulset.ApplyRevision(versionedSS, toHistory)
+		if err != nil {
+			return "", err
+		}
+		return printPodTemplate(&appliedSS.Spec.Template)
+	}
+
+	// Skip if the revision already matches current StatefulSet
+	done, err := statefulset.Match(versionedSS, toHistory)
+	if err != nil {
+		return "", err
+	}
+	if done {
+		return fmt.Sprintf("%s (current template already matches revision %d)", rollbackSkipped, toRevision), nil
+	}
+
+	// Restore revision
+	if _, err = versionedAppsClient.StatefulSets(ss.Namespace).Patch(ss.Name, types.StrategicMergePatchType, toHistory.Data.Raw); err != nil {
+		return "", fmt.Errorf("failed restoring revision %d: %v", toRevision, err)
+	}
+
+	return rollbackSuccess, nil
+}
+
+// findHistory returns a controllerrevision of a specific revision from the given controllerrevisions.
+// It returns nil if no such controllerrevision exists.
+// If toRevision is 0, the last previously used history is returned.
+func findHistory(toRevision int64, allHistory []*appsv1beta1.ControllerRevision) *appsv1beta1.ControllerRevision {
+	if toRevision == 0 && len(allHistory) <= 1 {
+		return nil
+	}
+
+	// Find the history to rollback to
+	var toHistory *appsv1beta1.ControllerRevision
+	if toRevision == 0 {
+		// If toRevision == 0, find the latest revision (2nd max)
+		sort.Sort(historiesByRevision(allHistory))
+		toHistory = allHistory[len(allHistory)-2]
+	} else {
+		for _, h := range allHistory {
+			if h.Revision == toRevision {
+				// If toRevision != 0, find the history with matching revision
+				return h
+			}
+		}
+	}
+
+	return toHistory
+}
+
+// printPodTemplate converts a given pod template into a human-readable string.
+func printPodTemplate(specTemplate *v1.PodTemplateSpec) (string, error) {
+	content := bytes.NewBuffer([]byte{})
+	w := printersinternal.NewPrefixWriter(content)
+	internalTemplate := &api.PodTemplateSpec{}
+	if err := apiv1.Convert_v1_PodTemplateSpec_To_api_PodTemplateSpec(specTemplate, internalTemplate, nil); err != nil {
+		return "", fmt.Errorf("failed to convert podtemplate while printing: %v", err)
+	}
+	printersinternal.DescribePodTemplate(internalTemplate, w)
+	return fmt.Sprintf("will roll back to %s", content.String()), nil
+}
+
 func revisionNotFoundErr(r int64) error {
 	return fmt.Errorf("unable to find specified revision %v in history", r)
 }