Merge pull request kubernetes#21030 from janetkuo/deployment-label-adopted

Auto commit by PR queue bot

Author: k8s-merge-robot

pkg/controller/deployment/deployment_controller.go

@@ -665,6 +665,10 @@ func (dc *DeploymentController) getNewReplicaSet(deployment extensions.Deploymen
 	newRevision := strconv.FormatInt(maxOldRevision+1, 10)
 
 	existingNewRS, err := deploymentutil.GetNewReplicaSetFromList(deployment, dc.client,
+		func(namespace string, options api.ListOptions) (*api.PodList, error) {
+			podList, err := dc.podStore.Pods(namespace).List(options.LabelSelector)
+			return &podList, err
+		},
 		func(namespace string, options api.ListOptions) ([]extensions.ReplicaSet, error) {
 			return dc.rsStore.ReplicaSets(namespace).List(options.LabelSelector)
 		})

pkg/util/deployment/deployment.go

@@ -25,11 +25,14 @@ import (
 	"k8s.io/kubernetes/pkg/api/unversioned"
 	"k8s.io/kubernetes/pkg/apis/extensions"
 	clientset "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset"
+	unversionedcore "k8s.io/kubernetes/pkg/client/typed/generated/core/unversioned"
+	unversionedextensions "k8s.io/kubernetes/pkg/client/typed/generated/extensions/unversioned"
 	"k8s.io/kubernetes/pkg/labels"
 	"k8s.io/kubernetes/pkg/util/integer"
 	intstrutil "k8s.io/kubernetes/pkg/util/intstr"
 	labelsutil "k8s.io/kubernetes/pkg/util/labels"
 	podutil "k8s.io/kubernetes/pkg/util/pod"
+	"k8s.io/kubernetes/pkg/util/wait"
 )
 
 const (
@@ -55,28 +58,21 @@ func GetOldReplicaSets(deployment extensions.Deployment, c clientset.Interface)
 		})
 }
 
+// TODO: switch this to full namespacers
+type rsListFunc func(string, api.ListOptions) ([]extensions.ReplicaSet, error)
+type podListFunc func(string, api.ListOptions) (*api.PodList, error)
+
 // GetOldReplicaSetsFromLists returns two sets of old replica sets targeted by the given Deployment; get PodList and ReplicaSetList with input functions.
 // Note that the first set of old replica sets doesn't include the ones with no pods, and the second set of old replica sets include all old replica sets.
-func GetOldReplicaSetsFromLists(deployment extensions.Deployment, c clientset.Interface, getPodList func(string, api.ListOptions) (*api.PodList, error), getRSList func(string, api.ListOptions) ([]extensions.ReplicaSet, error)) ([]*extensions.ReplicaSet, []*extensions.ReplicaSet, error) {
-	namespace := deployment.ObjectMeta.Namespace
-	selector, err := unversioned.LabelSelectorAsSelector(deployment.Spec.Selector)
-	if err != nil {
-		return nil, nil, fmt.Errorf("invalid label selector: %v", err)
-	}
-
-	// 1. Find all pods whose labels match deployment.Spec.Selector
-	options := api.ListOptions{LabelSelector: selector}
-	podList, err := getPodList(namespace, options)
-	if err != nil {
-		return nil, nil, fmt.Errorf("error listing pods: %v", err)
-	}
-	// 2. Find the corresponding replica sets for pods in podList.
+func GetOldReplicaSetsFromLists(deployment extensions.Deployment, c clientset.Interface, getPodList podListFunc, getRSList rsListFunc) ([]*extensions.ReplicaSet, []*extensions.ReplicaSet, error) {
+	// Find all pods whose labels match deployment.Spec.Selector, and corresponding replica sets for pods in podList.
+	// All pods and replica sets are labeled with pod-template-hash to prevent overlapping
 	// TODO: Right now we list all replica sets and then filter. We should add an API for this.
 	oldRSs := map[string]extensions.ReplicaSet{}
 	allOldRSs := map[string]extensions.ReplicaSet{}
-	rsList, err := getRSList(namespace, options)
+	rsList, podList, err := rsAndPodsWithHashKeySynced(deployment, c, getRSList, getPodList)
 	if err != nil {
-		return nil, nil, fmt.Errorf("error listing replica sets: %v", err)
+		return nil, nil, fmt.Errorf("error labeling replica sets and pods with pod-template-hash: %v", err)
 	}
 	newRSTemplate := GetNewReplicaSetTemplate(deployment)
 	for _, pod := range podList.Items {
@@ -113,6 +109,9 @@ func GetOldReplicaSetsFromLists(deployment extensions.Deployment, c clientset.In
 // Returns nil if the new replica set doesn't exist yet.
 func GetNewReplicaSet(deployment extensions.Deployment, c clientset.Interface) (*extensions.ReplicaSet, error) {
 	return GetNewReplicaSetFromList(deployment, c,
+		func(namespace string, options api.ListOptions) (*api.PodList, error) {
+			return c.Core().Pods(namespace).List(options)
+		},
 		func(namespace string, options api.ListOptions) ([]extensions.ReplicaSet, error) {
 			rsList, err := c.Extensions().ReplicaSets(namespace).List(options)
 			return rsList.Items, err
@@ -121,14 +120,8 @@ func GetNewReplicaSet(deployment extensions.Deployment, c clientset.Interface) (
 
 // GetNewReplicaSetFromList returns a replica set that matches the intent of the given deployment; get ReplicaSetList with the input function.
 // Returns nil if the new replica set doesn't exist yet.
-func GetNewReplicaSetFromList(deployment extensions.Deployment, c clientset.Interface, getRSList func(string, api.ListOptions) ([]extensions.ReplicaSet, error)) (*extensions.ReplicaSet, error) {
-	namespace := deployment.ObjectMeta.Namespace
-	selector, err := unversioned.LabelSelectorAsSelector(deployment.Spec.Selector)
-	if err != nil {
-		return nil, fmt.Errorf("invalid label selector: %v", err)
-	}
-
-	rsList, err := getRSList(namespace, api.ListOptions{LabelSelector: selector})
+func GetNewReplicaSetFromList(deployment extensions.Deployment, c clientset.Interface, getPodList podListFunc, getRSList rsListFunc) (*extensions.ReplicaSet, error) {
+	rsList, _, err := rsAndPodsWithHashKeySynced(deployment, c, getRSList, getPodList)
 	if err != nil {
 		return nil, fmt.Errorf("error listing ReplicaSets: %v", err)
 	}
@@ -144,6 +137,166 @@ func GetNewReplicaSetFromList(deployment extensions.Deployment, c clientset.Inte
 	return nil, nil
 }
 
+// rsAndPodsWithHashKeySynced returns the RSs and pods the given deployment targets, with pod-template-hash information synced.
+func rsAndPodsWithHashKeySynced(deployment extensions.Deployment, c clientset.Interface, getRSList rsListFunc, getPodList podListFunc) ([]extensions.ReplicaSet, *api.PodList, error) {
+	namespace := deployment.Namespace
+	selector, err := unversioned.LabelSelectorAsSelector(deployment.Spec.Selector)
+	if err != nil {
+		return nil, nil, err
+	}
+	options := api.ListOptions{LabelSelector: selector}
+	rsList, err := getRSList(namespace, options)
+	if err != nil {
+		return nil, nil, err
+	}
+	syncedRSList := []extensions.ReplicaSet{}
+	for _, rs := range rsList {
+		// Add pod-template-hash information if it's not in the RS.
+		// Otherwise, new RS produced by Deployment will overlap with pre-existing ones
+		// that aren't constrained by the pod-template-hash.
+		syncedRS, err := addHashKeyToRSAndPods(deployment, c, rs, getPodList)
+		if err != nil {
+			return nil, nil, err
+		}
+		syncedRSList = append(syncedRSList, *syncedRS)
+	}
+	syncedPodList, err := getPodList(namespace, options)
+	if err != nil {
+		return nil, nil, err
+	}
+	return syncedRSList, syncedPodList, nil
+}
+
+// addHashKeyToRSAndPods adds pod-template-hash information to the given rs, if it's not already there, with the following steps:
+// 1. Add hash label to the rs's pod template, and make sure the controller sees this update so that no orphaned pods will be created
+// 2. Add hash label to all pods this rs owns
+// 3. Add hash label to the rs's label and selector
+func addHashKeyToRSAndPods(deployment extensions.Deployment, c clientset.Interface, rs extensions.ReplicaSet, getPodList podListFunc) (updatedRS *extensions.ReplicaSet, err error) {
+	// If the rs already has the new hash label in its selector, it's done syncing
+	namespace := deployment.Namespace
+	hash := fmt.Sprintf("%d", podutil.GetPodTemplateSpecHash(api.PodTemplateSpec{
+		ObjectMeta: rs.Spec.Template.ObjectMeta,
+		Spec:       rs.Spec.Template.Spec,
+	}))
+	if labelsutil.SelectorHasLabel(rs.Spec.Selector, extensions.DefaultDeploymentUniqueLabelKey) {
+		return &rs, nil
+	}
+	// 1. Add hash template label to the rs. This ensures that any newly created pods will have the new label.
+	if len(rs.Spec.Template.Labels[extensions.DefaultDeploymentUniqueLabelKey]) == 0 {
+		updatedRS, err = updateRSWithRetries(c.Extensions().ReplicaSets(namespace), &rs, func(updated *extensions.ReplicaSet) {
+			updated.Spec.Template.Labels = labelsutil.AddLabel(updated.Spec.Template.Labels, extensions.DefaultDeploymentUniqueLabelKey, hash)
+		})
+		if err != nil {
+			return nil, err
+		}
+	}
+	// Make sure rs pod template is updated so that it won't create pods without the new label (orphaned pods).
+	if updatedRS.Generation > updatedRS.Status.ObservedGeneration {
+		if err = waitForReplicaSetUpdated(c, updatedRS.Generation, namespace, rs.Name); err != nil {
+			return nil, err
+		}
+	}
+
+	// 2. Update all pods managed by the rs to have the new hash label, so they will be correctly adopted.
+	selector, err := unversioned.LabelSelectorAsSelector(rs.Spec.Selector)
+	if err != nil {
+		return nil, err
+	}
+	options := api.ListOptions{LabelSelector: selector}
+	podList, err := getPodList(namespace, options)
+	if err != nil {
+		return nil, err
+	}
+	if err = labelPodsWithHash(podList, c, namespace, hash); err != nil {
+		return nil, err
+	}
+
+	// 3. Update rs label and selector to include the new hash label
+	// Copy the old selector, so that we can scrub out any orphaned pods
+	if updatedRS, err = updateRSWithRetries(c.Extensions().ReplicaSets(namespace), &rs, func(updated *extensions.ReplicaSet) {
+		updated.Labels = labelsutil.AddLabel(updated.Labels, extensions.DefaultDeploymentUniqueLabelKey, hash)
+		updated.Spec.Selector = labelsutil.AddLabelToSelector(updated.Spec.Selector, extensions.DefaultDeploymentUniqueLabelKey, hash)
+	}); err != nil {
+		return nil, err
+	}
+
+	// TODO: look for orphaned pods and label them in the background somewhere else periodically
+
+	return updatedRS, nil
+}
+
+func waitForReplicaSetUpdated(c clientset.Interface, desiredGeneration int64, namespace, name string) error {
+	return wait.Poll(10*time.Millisecond, 1*time.Minute, func() (bool, error) {
+		rs, err := c.Extensions().ReplicaSets(namespace).Get(name)
+		if err != nil {
+			return false, err
+		}
+		return rs.Status.ObservedGeneration >= desiredGeneration, nil
+	})
+}
+
+// labelPodsWithHash labels all pods in the given podList with the new hash label.
+func labelPodsWithHash(podList *api.PodList, c clientset.Interface, namespace, hash string) error {
+	for _, pod := range podList.Items {
+		// Only label the pod that doesn't already have the new hash
+		if pod.Labels[extensions.DefaultDeploymentUniqueLabelKey] != hash {
+			if _, err := updatePodWithRetries(c.Core().Pods(namespace), &pod, func(updated *api.Pod) {
+				pod.Labels = labelsutil.AddLabel(pod.Labels, extensions.DefaultDeploymentUniqueLabelKey, hash)
+			}); err != nil {
+				return err
+			}
+		}
+	}
+	return nil
+}
+
+// TODO: use client library instead when it starts to support update retries
+//       see https://github.com/kubernetes/kubernetes/issues/21479
+type updateRSFunc func(rs *extensions.ReplicaSet)
+
+func updateRSWithRetries(rsClient unversionedextensions.ReplicaSetInterface, rs *extensions.ReplicaSet, applyUpdate updateRSFunc) (*extensions.ReplicaSet, error) {
+	var err error
+	oldRs := rs
+	err = wait.Poll(10*time.Millisecond, 1*time.Minute, func() (bool, error) {
+		// Apply the update, then attempt to push it to the apiserver.
+		applyUpdate(rs)
+		if rs, err = rsClient.Update(rs); err == nil {
+			// rs contains the latest controller post update
+			return true, nil
+		}
+		// Update the controller with the latest resource version, if the update failed we
+		// can't trust rs so use oldRs.Name.
+		if rs, err = rsClient.Get(oldRs.Name); err != nil {
+			// The Get failed: Value in rs cannot be trusted.
+			rs = oldRs
+		}
+		// The Get passed: rs contains the latest controller, expect a poll for the update.
+		return false, nil
+	})
+	// If the error is non-nil the returned controller cannot be trusted, if it is nil, the returned
+	// controller contains the applied update.
+	return rs, err
+}
+
+type updatePodFunc func(pod *api.Pod)
+
+func updatePodWithRetries(podClient unversionedcore.PodInterface, pod *api.Pod, applyUpdate updatePodFunc) (*api.Pod, error) {
+	var err error
+	oldPod := pod
+	err = wait.Poll(10*time.Millisecond, 1*time.Minute, func() (bool, error) {
+		// Apply the update, then attempt to push it to the apiserver.
+		applyUpdate(pod)
+		if pod, err = podClient.Update(pod); err == nil {
+			return true, nil
+		}
+		if pod, err = podClient.Get(oldPod.Name); err != nil {
+			pod = oldPod
+		}
+		return false, nil
+	})
+	return pod, err
+}
+
 // Returns the desired PodTemplateSpec for the new ReplicaSet corresponding to the given ReplicaSet.
 func GetNewReplicaSetTemplate(deployment extensions.Deployment) api.PodTemplateSpec {
 	// newRS will have the same template as in deployment spec, plus a unique label in some cases.