Merge pull request #13279 from mfojtik/retry-scale-in-deploy

Merged by openshift-bot

Author: OpenShift Bot

pkg/deploy/cmd/test/support.go

@@ -3,6 +3,8 @@ package test
 import (
 	"fmt"
 
+	"k8s.io/client-go/pkg/api/errors"
+	"k8s.io/client-go/pkg/api/unversioned"
 	"k8s.io/kubernetes/pkg/kubectl"
 )
 
@@ -23,3 +25,23 @@ func (t *FakeScaler) Scale(namespace, name string, newSize uint, preconditions *
 func (t *FakeScaler) ScaleSimple(namespace, name string, preconditions *kubectl.ScalePrecondition, newSize uint) (string, error) {
 	return "", fmt.Errorf("unexpected call to ScaleSimple")
 }
+
+type FakeLaggedScaler struct {
+	Events     []ScaleEvent
+	RetryCount int
+}
+
+func (t *FakeLaggedScaler) Scale(namespace, name string, newSize uint, preconditions *kubectl.ScalePrecondition, retry, wait *kubectl.RetryParams) error {
+	if t.RetryCount != 2 {
+		t.RetryCount += 1
+		// This is faking a real error from the
+		// "k8s.io/kubernetes/plugin/pkg/admission/namespace/lifecycle" package.
+		return errors.NewForbidden(unversioned.GroupResource{Resource: "ReplicationController"}, name, fmt.Errorf("%s: not yet ready to handle request", name))
+	}
+	t.Events = append(t.Events, ScaleEvent{name, newSize})
+	return nil
+}
+
+func (t *FakeLaggedScaler) ScaleSimple(namespace, name string, preconditions *kubectl.ScalePrecondition, newSize uint) (string, error) {
+	return "", nil
+}

pkg/deploy/strategy/recreate/recreate.go

@@ -5,8 +5,10 @@ import (
 	"io"
 	"io/ioutil"
 	"os"
+	"strings"
 	"time"
 
+	"k8s.io/client-go/pkg/api/errors"
 	kapi "k8s.io/kubernetes/pkg/api"
 	kclientset "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset"
 	kcoreclient "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset/typed/core/internalversion"
@@ -231,11 +233,29 @@ func (s *RecreateDeploymentStrategy) DeployWithAcceptor(from *kapi.ReplicationCo
 	return nil
 }
 
-func (s *RecreateDeploymentStrategy) scaleAndWait(deployment *kapi.ReplicationController, replicas int, retry *kubectl.RetryParams, wait *kubectl.RetryParams) (*kapi.ReplicationController, error) {
+func (s *RecreateDeploymentStrategy) scaleAndWait(deployment *kapi.ReplicationController, replicas int, retry *kubectl.RetryParams, retryParams *kubectl.RetryParams) (*kapi.ReplicationController, error) {
 	if int32(replicas) == deployment.Spec.Replicas && int32(replicas) == deployment.Status.Replicas {
 		return deployment, nil
 	}
-	if err := s.scaler.Scale(deployment.Namespace, deployment.Name, uint(replicas), &kubectl.ScalePrecondition{Size: -1, ResourceVersion: ""}, retry, wait); err != nil {
+	var scaleErr error
+	err := wait.PollImmediate(1*time.Second, 30*time.Second, func() (bool, error) {
+		scaleErr = s.scaler.Scale(deployment.Namespace, deployment.Name, uint(replicas), &kubectl.ScalePrecondition{Size: -1, ResourceVersion: ""}, retry, retryParams)
+		if scaleErr == nil {
+			return true, nil
+		}
+		// This error is returned when the lifecycle admission plugin cache is not fully
+		// synchronized. In that case the scaling should be retried.
+		//
+		// FIXME: The error returned from admission should not be forbidden but come-back-later error.
+		if errors.IsForbidden(scaleErr) && strings.Contains(scaleErr.Error(), "not yet ready to handle request") {
+			return false, nil
+		}
+		return false, scaleErr
+	})
+	if err == wait.ErrWaitTimeout {
+		return nil, fmt.Errorf("%v: %v", err, scaleErr)
+	}
+	if err != nil {
 		return nil, err
 	}
 

pkg/deploy/strategy/recreate/recreate_test.go

@@ -293,6 +293,55 @@ func TestRecreate_acceptorSuccess(t *testing.T) {
 	}
 }
 
+func TestRecreate_acceptorSuccessWithColdCaches(t *testing.T) {
+	var deployment *kapi.ReplicationController
+	scaler := &cmdtest.FakeLaggedScaler{}
+
+	strategy := &RecreateDeploymentStrategy{
+		out:         &bytes.Buffer{},
+		errOut:      &bytes.Buffer{},
+		eventClient: fake.NewSimpleClientset().Core(),
+		decoder:     kapi.Codecs.UniversalDecoder(),
+		retryPeriod: 1 * time.Millisecond,
+		scaler:      scaler,
+	}
+
+	acceptorCalled := false
+	acceptor := &testAcceptor{
+		acceptFn: func(deployment *kapi.ReplicationController) error {
+			acceptorCalled = true
+			return nil
+		},
+	}
+
+	oldDeployment, _ := deployutil.MakeDeployment(deploytest.OkDeploymentConfig(1), kapi.Codecs.LegacyCodec(registered.GroupOrDie(kapi.GroupName).GroupVersions[0]))
+	deployment, _ = deployutil.MakeDeployment(deploytest.OkDeploymentConfig(2), kapi.Codecs.LegacyCodec(registered.GroupOrDie(kapi.GroupName).GroupVersions[0]))
+	strategy.rcClient = &fakeControllerClient{deployment: deployment}
+	strategy.podClient = &fakePodClient{deployerName: deployutil.DeployerPodNameForDeployment(deployment.Name)}
+
+	err := strategy.DeployWithAcceptor(oldDeployment, deployment, 2, acceptor)
+	if err != nil {
+		t.Fatalf("unexpected deploy error: %#v", err)
+	}
+
+	if !acceptorCalled {
+		t.Fatalf("expected acceptor to be called")
+	}
+
+	if e, a := 2, len(scaler.Events); e != a {
+		t.Fatalf("expected %d scale calls, got %d", e, a)
+	}
+	if e, a := uint(1), scaler.Events[0].Size; e != a {
+		t.Errorf("expected scale down to %d, got %d", e, a)
+	}
+	if e, a := uint(2), scaler.Events[1].Size; e != a {
+		t.Errorf("expected scale up to %d, got %d", e, a)
+	}
+	if scaler.RetryCount != 2 {
+		t.Errorf("expected retry when the caches are not initialized")
+	}
+}
+
 func TestRecreate_acceptorFail(t *testing.T) {
 	var deployment *kapi.ReplicationController
 	scaler := &cmdtest.FakeScaler{}