Implement hash collision avoidance mechanism

pkg/controller/deployment/deployment_controller.go

@@ -574,7 +574,6 @@ func (dc *DeploymentController) syncDeployment(key string) error {
 		return nil
 	}
 	if err != nil {
-		utilruntime.HandleError(fmt.Errorf("Unable to retrieve deployment %v from store: %v", key, err))
 		return err
 	}
 

pkg/controller/deployment/util/deployment_util_test.go

@@ -31,6 +31,7 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/intstr"
 	core "k8s.io/client-go/testing"
+	"k8s.io/kubernetes/pkg/api"
 	"k8s.io/kubernetes/pkg/api/v1"
 	extensions "k8s.io/kubernetes/pkg/apis/extensions/v1beta1"
 	"k8s.io/kubernetes/pkg/client/clientset_generated/clientset/fake"
@@ -183,7 +184,8 @@ func newDControllerRef(d *extensions.Deployment) *metav1.OwnerReference {
 
 // generateRS creates a replica set, with the input deployment's template as its template
 func generateRS(deployment extensions.Deployment) extensions.ReplicaSet {
-	template := GetNewReplicaSetTemplate(&deployment)
+	cp, _ := api.Scheme.DeepCopy(deployment.Spec.Template)
+	template := cp.(v1.PodTemplateSpec)
 	return extensions.ReplicaSet{
 		ObjectMeta: metav1.ObjectMeta{
 			UID:             randomUID(),
@@ -192,7 +194,7 @@ func generateRS(deployment extensions.Deployment) extensions.ReplicaSet {
 			OwnerReferences: []metav1.OwnerReference{*newDControllerRef(&deployment)},
 		},
 		Spec: extensions.ReplicaSetSpec{
-			Replicas: func() *int32 { i := int32(0); return &i }(),
+			Replicas: new(int32),
 			Template: template,
 			Selector: &metav1.LabelSelector{MatchLabels: template.Labels},
 		},

pkg/controller/deployment/util/hash_test.go

@@ -110,7 +110,7 @@ func TestPodTemplateSpecHash(t *testing.T) {
 		specJson := strings.Replace(podSpec, "@@VERSION@@", strconv.Itoa(i), 1)
 		spec := v1.PodTemplateSpec{}
 		json.Unmarshal([]byte(specJson), &spec)
-		hash := GetPodTemplateSpecHash(spec)
+		hash := GetPodTemplateSpecHash(&spec, nil)
 		if v, ok := seenHashes[hash]; ok {
 			t.Errorf("Hash collision, old: %d new: %d", v, i)
 			break
@@ -139,6 +139,6 @@ func BenchmarkFnv(b *testing.B) {
 	json.Unmarshal([]byte(podSpec), &spec)
 
 	for i := 0; i < b.N; i++ {
-		GetPodTemplateSpecHash(spec)
+		GetPodTemplateSpecHash(&spec, nil)
 	}
 }

pkg/controller/deployment/util/pod_util.go

@@ -17,6 +17,7 @@ limitations under the License.
 package util
 
 import (
+	"encoding/binary"
 	"hash/fnv"
 
 	"github.com/golang/glog"
@@ -30,9 +31,17 @@ import (
 	hashutil "k8s.io/kubernetes/pkg/util/hash"
 )
 
-func GetPodTemplateSpecHash(template v1.PodTemplateSpec) uint32 {
+func GetPodTemplateSpecHash(template *v1.PodTemplateSpec, uniquifier *int64) uint32 {
 	podTemplateSpecHasher := fnv.New32a()
-	hashutil.DeepHashObject(podTemplateSpecHasher, template)
+	hashutil.DeepHashObject(podTemplateSpecHasher, *template)
+
+	// Add uniquifier in the hash if it exists.
+	if uniquifier != nil {
+		uniquifierBytes := make([]byte, 8)
+		binary.LittleEndian.PutUint64(uniquifierBytes, uint64(*uniquifier))
+		podTemplateSpecHasher.Write(uniquifierBytes)
+	}
+
 	return podTemplateSpecHasher.Sum32()
 }
 

pkg/controller/deployment/util/pod_util_test.go

+/*
+Copyright 2017 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package util
+
+import (
+	"math"
+	"testing"
+
+	"k8s.io/kubernetes/pkg/api/v1"
+)
+
+func int64P(num int64) *int64 {
+	return &num
+}
+
+func TestGetPodTemplateSpecHash(t *testing.T) {
+	tests := []struct {
+		name                string
+		template            *v1.PodTemplateSpec
+		collisionCount      *int64
+		otherCollisionCount *int64
+	}{
+		{
+			name:                "simple",
+			template:            &v1.PodTemplateSpec{},
+			collisionCount:      int64P(1),
+			otherCollisionCount: int64P(2),
+		},
+		{
+			name:                "using math.MaxInt64",
+			template:            &v1.PodTemplateSpec{},
+			collisionCount:      nil,
+			otherCollisionCount: int64P(int64(math.MaxInt64)),
+		},
+	}
+
+	for _, test := range tests {
+		hash := GetPodTemplateSpecHash(test.template, test.collisionCount)
+		otherHash := GetPodTemplateSpecHash(test.template, test.otherCollisionCount)
+
+		if hash == otherHash {
+			t.Errorf("expected different hashes but got the same: %d", hash)
+		}
+	}
+}

pkg/controller/deployment/util/replicaset_util.go

@@ -69,11 +69,12 @@ func UpdateRSWithRetries(rsClient unversionedextensions.ReplicaSetInterface, rsL
 }
 
 // GetReplicaSetHash returns the pod template hash of a ReplicaSet's pod template space
-func GetReplicaSetHash(rs *extensions.ReplicaSet) string {
-	meta := rs.Spec.Template.ObjectMeta
-	meta.Labels = labelsutil.CloneAndRemoveLabel(meta.Labels, extensions.DefaultDeploymentUniqueLabelKey)
-	return fmt.Sprintf("%d", GetPodTemplateSpecHash(v1.PodTemplateSpec{
-		ObjectMeta: meta,
-		Spec:       rs.Spec.Template.Spec,
-	}))
+func GetReplicaSetHash(rs *extensions.ReplicaSet, uniquifier *int64) (string, error) {
+	template, err := api.Scheme.DeepCopy(rs.Spec.Template)
+	if err != nil {
+		return "", err
+	}
+	rsTemplate := template.(v1.PodTemplateSpec)
+	rsTemplate.Labels = labelsutil.CloneAndRemoveLabel(rsTemplate.Labels, extensions.DefaultDeploymentUniqueLabelKey)
+	return fmt.Sprintf("%d", GetPodTemplateSpecHash(&rsTemplate, uniquifier)), nil
 }

test/e2e/deployment.go

@@ -21,8 +21,10 @@ import (
 	"math/rand"
 	"time"
 
+	"github.com/davecgh/go-spew/spew"
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
+
 	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
@@ -108,6 +110,9 @@ var _ = framework.KubeDescribe("Deployment", func() {
 	It("test Deployment ReplicaSet orphaning and adoption regarding controllerRef", func() {
 		testDeploymentsControllerRef(f)
 	})
+	It("deployment can avoid hash collisions", func() {
+		testDeploymentHashCollisionAvoidance(f)
+	})
 	// TODO: add tests that cover deployment.Spec.MinReadySeconds once we solved clock-skew issues
 	// See https://github.com/kubernetes/kubernetes/issues/29229
 })
@@ -1359,3 +1364,47 @@ func orphanDeploymentReplicaSets(c clientset.Interface, d *extensions.Deployment
 	deleteOptions.Preconditions = metav1.NewUIDPreconditions(string(d.UID))
 	return c.Extensions().Deployments(d.Namespace).Delete(d.Name, deleteOptions)
 }
+
+func testDeploymentHashCollisionAvoidance(f *framework.Framework) {
+	ns := f.Namespace.Name
+	c := f.ClientSet
+
+	deploymentName := "test-hash-collision"
+	framework.Logf("Creating Deployment %q", deploymentName)
+	podLabels := map[string]string{"name": nginxImageName}
+	d := framework.NewDeployment(deploymentName, int32(0), podLabels, nginxImageName, nginxImage, extensions.RollingUpdateDeploymentStrategyType)
+	deployment, err := c.Extensions().Deployments(ns).Create(d)
+	Expect(err).NotTo(HaveOccurred())
+	err = framework.WaitForDeploymentRevisionAndImage(c, ns, deploymentName, "1", nginxImage)
+	Expect(err).NotTo(HaveOccurred())
+
+	// TODO: Switch this to do a non-cascading deletion of the Deployment, mutate the ReplicaSet
+	// once it has no owner reference, then recreate the Deployment if we ever proceed with
+	// https://github.com/kubernetes/kubernetes/issues/44237
+	framework.Logf("Mock a hash collision")
+	newRS, err := deploymentutil.GetNewReplicaSet(deployment, c)
+	Expect(err).NotTo(HaveOccurred())
+	var nilRs *extensions.ReplicaSet
+	Expect(newRS).NotTo(Equal(nilRs))
+	_, err = framework.UpdateReplicaSetWithRetries(c, ns, newRS.Name, func(update *extensions.ReplicaSet) {
+		*update.Spec.Template.Spec.TerminationGracePeriodSeconds = int64(5)
+	})
+	Expect(err).NotTo(HaveOccurred())
+
+	framework.Logf("Expect deployment collision counter to increment")
+	if err := wait.PollImmediate(time.Second, time.Minute, func() (bool, error) {
+		d, err := c.Extensions().Deployments(ns).Get(deploymentName, metav1.GetOptions{})
+		if err != nil {
+			framework.Logf("cannot get deployment %q: %v", deploymentName, err)
+			return false, nil
+		}
+		framework.Logf(spew.Sprintf("deployment status: %#v", d.Status))
+		return d.Status.CollisionCount != nil && *d.Status.CollisionCount == int64(1), nil
+	}); err != nil {
+		framework.Failf("Failed to increment collision counter for deployment %q: %v", deploymentName, err)
+	}
+
+	framework.Logf("Expect a new ReplicaSet to be created")
+	err = framework.WaitForDeploymentRevisionAndImage(c, ns, deploymentName, "2", nginxImage)
+	Expect(err).NotTo(HaveOccurred())
+}

test/e2e/framework/util.go

@@ -3348,9 +3348,10 @@ func WatchRecreateDeployment(c clientset.Interface, d *extensions.Deployment) er
 }
 
 // WaitForDeploymentRevisionAndImage waits for the deployment's and its new RS's revision and container image to match the given revision and image.
-// Note that deployment revision and its new RS revision should be updated shortly, so we only wait for 1 minute here to fail early.
+// Note that deployment revision and its new RS revision should be updated shortly most of the time, but an overwhelmed RS controller
+// may result in taking longer to relabel a RS.
 func WaitForDeploymentRevisionAndImage(c clientset.Interface, ns, deploymentName string, revision, image string) error {
-	return testutil.WaitForDeploymentRevisionAndImage(c, ns, deploymentName, revision, image, Logf, Poll, pollShortTimeout)
+	return testutil.WaitForDeploymentRevisionAndImage(c, ns, deploymentName, revision, image, Logf, Poll, pollLongTimeout)
 }
 
 // CheckNewRSAnnotations check if the new RS's annotation is as expected
@@ -3486,16 +3487,17 @@ func WaitForPartialEvents(c clientset.Interface, ns string, objOrRef runtime.Obj
 
 type updateDeploymentFunc func(d *extensions.Deployment)
 
-func UpdateDeploymentWithRetries(c clientset.Interface, namespace, name string, applyUpdate updateDeploymentFunc) (deployment *extensions.Deployment, err error) {
-	deployments := c.Extensions().Deployments(namespace)
+func UpdateDeploymentWithRetries(c clientset.Interface, namespace, name string, applyUpdate updateDeploymentFunc) (*extensions.Deployment, error) {
+	var deployment *extensions.Deployment
 	var updateErr error
-	pollErr := wait.Poll(10*time.Millisecond, 1*time.Minute, func() (bool, error) {
-		if deployment, err = deployments.Get(name, metav1.GetOptions{}); err != nil {
+	pollErr := wait.PollImmediate(1*time.Second, 1*time.Minute, func() (bool, error) {
+		var err error
+		if deployment, err = c.Extensions().Deployments(namespace).Get(name, metav1.GetOptions{}); err != nil {
 			return false, err
 		}
 		// Apply the update, then attempt to push it to the apiserver.
 		applyUpdate(deployment)
-		if deployment, err = deployments.Update(deployment); err == nil {
+		if deployment, err = c.Extensions().Deployments(namespace).Update(deployment); err == nil {
 			Logf("Updating deployment %s", name)
 			return true, nil
 		}
@@ -3513,7 +3515,7 @@ type updateRsFunc func(d *extensions.ReplicaSet)
 func UpdateReplicaSetWithRetries(c clientset.Interface, namespace, name string, applyUpdate updateRsFunc) (*extensions.ReplicaSet, error) {
 	var rs *extensions.ReplicaSet
 	var updateErr error
-	pollErr := wait.PollImmediate(10*time.Millisecond, 1*time.Minute, func() (bool, error) {
+	pollErr := wait.PollImmediate(1*time.Second, 1*time.Minute, func() (bool, error) {
 		var err error
 		if rs, err = c.Extensions().ReplicaSets(namespace).Get(name, metav1.GetOptions{}); err != nil {
 			return false, err