Implements StatefulSet update

Implements history utilities for ControllerRevision in the controller/history package
StatefulSetStatus now has additional fields for consistency with DaemonSet and Deployment
StatefulSetStatus.Replicas now represents the current number of createdPods and StatefulSetStatus.ReadyReplicas is the current number of ready Pods

pkg/controller/statefulset/stateful_set_utils.go

@@ -17,15 +17,23 @@ limitations under the License.
 package statefulset
 
 import (
+	"encoding/json"
 	"fmt"
 	"regexp"
 	"strconv"
 
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/kubernetes/pkg/api"
 	"k8s.io/kubernetes/pkg/api/v1"
 	podutil "k8s.io/kubernetes/pkg/api/v1/pod"
 	apps "k8s.io/kubernetes/pkg/apis/apps/v1beta1"
 	"k8s.io/kubernetes/pkg/controller"
+	"k8s.io/kubernetes/pkg/controller/history"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/util/strategicpatch"
+
+	"k8s.io/client-go/kubernetes/scheme"
 
 	"github.com/golang/glog"
 )
@@ -36,6 +44,7 @@ const maxUpdateRetries = 10
 // updateConflictError is the error used to indicate that the maximum number of retries against the API server have
 // been attempted and we need to back off
 var updateConflictError = fmt.Errorf("aborting update after %d attempts", maxUpdateRetries)
+var patchCodec = api.Codecs.LegacyCodec(apps.SchemeGroupVersion)
 
 // overlappingStatefulSets sorts a list of StatefulSets by creation timestamp, using their names as a tie breaker.
 // Generally used to tie break between StatefulSets that have overlapping selectors.
@@ -246,6 +255,23 @@ func newControllerRef(set *apps.StatefulSet) *metav1.OwnerReference {
 	}
 }
 
+// setPodRevision sets the revision of Pod to revision by adding the StatefulSetRevisionLabel
+func setPodRevision(pod *v1.Pod, revision string) {
+	if pod.Labels == nil {
+		pod.Labels = make(map[string]string)
+	}
+	pod.Labels[apps.StatefulSetRevisionLabel] = revision
+}
+
+// getPodRevision gets the revision of Pod by inspecting the StatefulSetRevisionLabel. If pod has no revision the empty
+// string is returned.
+func getPodRevision(pod *v1.Pod) string {
+	if pod.Labels == nil {
+		return ""
+	}
+	return pod.Labels[apps.StatefulSetRevisionLabel]
+}
+
 // newStatefulSetPod returns a new Pod conforming to the set's Spec with an identity generated from ordinal.
 func newStatefulSetPod(set *apps.StatefulSet, ordinal int) *v1.Pod {
 	pod, _ := controller.GetPodFromTemplate(&set.Spec.Template, set, newControllerRef(set))
@@ -255,6 +281,122 @@ func newStatefulSetPod(set *apps.StatefulSet, ordinal int) *v1.Pod {
 	return pod
 }
 
+// newVersionedStatefulSetPod creates a new Pod for a StatefulSet. currentSet is the representation of the set at the
+// current revision. updateSet is the representation of the set at the updateRevision. currentRevision is the name of
+// the current revision. updateRevision is the name of the update revision. ordinal is the ordinal of the Pod. If the
+// returned error is nil, the returned Pod is valid.
+func newVersionedStatefulSetPod(currentSet, updateSet *apps.StatefulSet, currentRevision, updateRevision string, ordinal int) *v1.Pod {
+	if (currentSet.Spec.UpdateStrategy.Type == apps.RollingUpdateStatefulSetStrategyType &&
+		ordinal < int(currentSet.Status.CurrentReplicas)) ||
+		(currentSet.Spec.UpdateStrategy.Type == apps.PartitionStatefulSetStrategyType &&
+			ordinal < int(currentSet.Spec.UpdateStrategy.Partition.Ordinal)) {
+		pod := newStatefulSetPod(currentSet, ordinal)
+		setPodRevision(pod, currentRevision)
+		return pod
+	}
+	pod := newStatefulSetPod(updateSet, ordinal)
+	setPodRevision(pod, updateRevision)
+	return pod
+}
+
+// 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
+// recorded patches.
+func getPatch(set *apps.StatefulSet) ([]byte, error) {
+	str, err := runtime.Encode(patchCodec, set)
+	if err != nil {
+		return nil, err
+	}
+	var raw map[string]interface{}
+	json.Unmarshal([]byte(str), &raw)
+	objCopy := make(map[string]interface{})
+	specCopy := make(map[string]interface{})
+	spec := raw["spec"].(map[string]interface{})
+	template := spec["template"].(map[string]interface{})
+	specCopy["template"] = template
+	template["$patch"] = "replace"
+	objCopy["spec"] = specCopy
+	patch, err := json.Marshal(objCopy)
+	return patch, err
+}
+
+// newRevision creates a new ControllerRevision containing a patch that reapplies the target state of set.
+// The Revision of the returned ControllerRevision is set to revision. If the returned error is nil, the returned
+// ControllerRevision is valid. StatefulSet revisions are stored as patches that re-apply the current state of set
+// to a new StatefulSet using a strategic merge patch to replace the saved state of the new StatefulSet.
+func newRevision(set *apps.StatefulSet, revision int64) (*apps.ControllerRevision, error) {
+	patch, err := getPatch(set)
+	if err != nil {
+		return nil, err
+	}
+	selector, err := metav1.LabelSelectorAsSelector(set.Spec.Selector)
+	if err != nil {
+		return nil, err
+	}
+	return history.NewControllerRevision(set,
+		controllerKind,
+		selector,
+		runtime.RawExtension{Raw: patch},
+		revision)
+}
+
+// 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) {
+	obj, err := scheme.Scheme.DeepCopy(set)
+	if err != nil {
+		return nil, err
+	}
+	clone := obj.(*apps.StatefulSet)
+	patched, err := strategicpatch.StrategicMergePatch([]byte(runtime.EncodeOrDie(patchCodec, clone)), revision.Data.Raw, clone)
+	if err != nil {
+		return nil, err
+	}
+	err = json.Unmarshal(patched, clone)
+	if err != nil {
+		return nil, err
+	}
+	return clone, nil
+}
+
+// nextRevision finds the next valid revision number based on revisions. If the length of revisions
+// is 0 this is 1. Otherwise, it is 1 greater than the largest revision's Revision. This method
+// assumes that revisions has been sorted by Revision.
+func nextRevision(revisions []*apps.ControllerRevision) int64 {
+	count := len(revisions)
+	if count <= 0 {
+		return 1
+	}
+	return revisions[count-1].Revision + 1
+}
+
+// inconsistentStatus returns true if the ObservedGeneration of status is greater than set's
+// Generation or if any of the status's fields do not match those of set's status.
+func inconsistentStatus(set *apps.StatefulSet, status *apps.StatefulSetStatus) bool {
+	return set.Status.ObservedGeneration == nil ||
+		*status.ObservedGeneration > *set.Status.ObservedGeneration ||
+		status.Replicas != set.Status.Replicas ||
+		status.CurrentReplicas != set.Status.CurrentReplicas ||
+		status.ReadyReplicas != set.Status.ReadyReplicas ||
+		status.UpdatedReplicas != set.Status.UpdatedReplicas ||
+		status.CurrentRevision != set.Status.CurrentRevision ||
+		status.UpdateRevision != set.Status.UpdateRevision
+}
+
+// completeRollingUpdate completes a rolling update when all of set's replica Pods have been updated
+// to the updateRevision. status's currentRevision is set to updateRevision and its' updateRevision
+// is set to the empty string. status's currentReplicas is set to updateReplicas and its updateReplicas
+// are set to 0.
+func completeRollingUpdate(set *apps.StatefulSet, status *apps.StatefulSetStatus) {
+	if set.Spec.UpdateStrategy.Type == apps.RollingUpdateStatefulSetStrategyType &&
+		status.UpdatedReplicas == status.Replicas &&
+		status.ReadyReplicas == status.Replicas {
+		status.CurrentReplicas = status.UpdatedReplicas
+		status.CurrentRevision = status.UpdateRevision
+	}
+}
+
 // ascendingOrdinal is a sort.Interface that Sorts a list of Pods based on the ordinals extracted
 // from the Pod. Pod's that have not been constructed by StatefulSet's have an ordinal of -1, and are therefore pushed
 // to the front of the list.