Merge pull request #12910 from mfojtik/retry-rc-watch

Merged by openshift-bot

Author: OpenShift Bot

Diff for: pkg/deploy/registry/rest.go

@@ -6,17 +6,20 @@ import (
 	"time"
 
 	kapi "k8s.io/kubernetes/pkg/api"
+	"k8s.io/kubernetes/pkg/api/unversioned"
 	kcoreclient "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset/typed/core/internalversion"
 	"k8s.io/kubernetes/pkg/fields"
 	"k8s.io/kubernetes/pkg/watch"
 
+	"github.com/golang/glog"
 	"github.com/openshift/origin/pkg/deploy/api"
 	deployutil "github.com/openshift/origin/pkg/deploy/util"
 )
 
 var (
 	// ErrUnknownDeploymentPhase is returned for WaitForRunningDeployment if an unknown phase is returned.
 	ErrUnknownDeploymentPhase = errors.New("unknown deployment phase")
+	ErrTooOldResourceVersion  = errors.New("too old resource version")
 )
 
 // WaitForRunningDeployment waits until the specified deployment is no longer New or Pending. Returns true if
@@ -33,6 +36,15 @@ func WaitForRunningDeployment(rn kcoreclient.ReplicationControllersGetter, obser
 
 	if _, err := watch.Until(timeout, w, func(e watch.Event) (bool, error) {
 		if e.Type == watch.Error {
+			// When we send too old resource version in observed replication controller to
+			// watcher, restart the watch with latest available controller.
+			switch t := e.Object.(type) {
+			case *unversioned.Status:
+				if t.Reason == unversioned.StatusReasonGone {
+					glog.V(5).Infof("encountered error while watching for replication controller: %v (retrying)", t)
+					return false, ErrTooOldResourceVersion
+				}
+			}
 			return false, fmt.Errorf("encountered error while watching for replication controller: %v", e.Object)
 		}
 		obj, isController := e.Object.(*kapi.ReplicationController)
@@ -49,6 +61,13 @@ func WaitForRunningDeployment(rn kcoreclient.ReplicationControllersGetter, obser
 			return false, ErrUnknownDeploymentPhase
 		}
 	}); err != nil {
+		if err == ErrTooOldResourceVersion {
+			latestRC, err := rn.ReplicationControllers(observed.Namespace).Get(observed.Name)
+			if err != nil {
+				return observed, false, err
+			}
+			return WaitForRunningDeployment(rn, latestRC, timeout)
+		}
 		return observed, false, err
 	}
 

Diff for: pkg/deploy/registry/rest_test.go

@@ -0,0 +1,110 @@
+package registry
+
+import (
+	"testing"
+	"time"
+
+	kapi "k8s.io/kubernetes/pkg/api"
+	"k8s.io/kubernetes/pkg/api/unversioned"
+	"k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset/fake"
+	"k8s.io/kubernetes/pkg/client/testing/core"
+	"k8s.io/kubernetes/pkg/runtime"
+	"k8s.io/kubernetes/pkg/watch"
+
+	deployapi "github.com/openshift/origin/pkg/deploy/api"
+)
+
+func TestWaitForRunningDeploymentSuccess(t *testing.T) {
+	fakeController := &kapi.ReplicationController{}
+	fakeController.Name = "test-1"
+	fakeController.Namespace = "test"
+
+	kubeclient := fake.NewSimpleClientset([]runtime.Object{fakeController}...)
+	fakeWatch := watch.NewFake()
+	kubeclient.PrependWatchReactor("replicationcontrollers", core.DefaultWatchReactor(fakeWatch, nil))
+	stopChan := make(chan struct{})
+
+	go func() {
+		defer close(stopChan)
+		rc, ok, err := WaitForRunningDeployment(kubeclient.Core(), fakeController, 10*time.Second)
+		if err != nil {
+			t.Fatalf("unexpected error: %v", err)
+		}
+		if !ok {
+			t.Errorf("expected to return success")
+		}
+		if rc == nil {
+			t.Errorf("expected returned replication controller to not be nil")
+		}
+	}()
+
+	fakeController.Annotations = map[string]string{deployapi.DeploymentStatusAnnotation: string(deployapi.DeploymentStatusRunning)}
+	fakeWatch.Modify(fakeController)
+	<-stopChan
+}
+
+func TestWaitForRunningDeploymentRestartWatch(t *testing.T) {
+	fakeController := &kapi.ReplicationController{}
+	fakeController.Name = "test-1"
+	fakeController.Namespace = "test"
+
+	kubeclient := fake.NewSimpleClientset([]runtime.Object{fakeController}...)
+	fakeWatch := watch.NewFake()
+
+	watchCalledChan := make(chan struct{})
+	kubeclient.PrependWatchReactor("replicationcontrollers", func(action core.Action) (bool, watch.Interface, error) {
+		fakeWatch.Reset()
+		watchCalledChan <- struct{}{}
+		return core.DefaultWatchReactor(fakeWatch, nil)(action)
+	})
+
+	getReceivedChan := make(chan struct{})
+	kubeclient.PrependReactor("get", "replicationcontrollers", func(action core.Action) (bool, runtime.Object, error) {
+		close(getReceivedChan)
+		return true, fakeController, nil
+	})
+
+	stopChan := make(chan struct{})
+	go func() {
+		defer close(stopChan)
+		rc, ok, err := WaitForRunningDeployment(kubeclient.Core(), fakeController, 10*time.Second)
+		if err != nil {
+			t.Fatalf("unexpected error: %v", err)
+		}
+		if !ok {
+			t.Errorf("expected to return success")
+		}
+		if rc == nil {
+			t.Errorf("expected returned replication controller to not be nil")
+		}
+	}()
+
+	select {
+	case <-watchCalledChan:
+	case <-time.After(time.Second * 5):
+		t.Fatalf("timeout waiting for the watch to start")
+	}
+
+	// Send the StatusReasonGone error to watcher which should trigger the watch restart.
+	goneError := &unversioned.Status{Reason: unversioned.StatusReasonGone}
+	fakeWatch.Error(goneError)
+
+	// Make sure we observed the "get" action on replication controller, so the watch gets
+	// the latest resourceVersion.
+	select {
+	case <-getReceivedChan:
+	case <-time.After(time.Second * 5):
+		t.Fatalf("timeout waiting for get on replication controllers")
+	}
+
+	// Wait for the watcher to restart and then transition the replication controller to
+	// running state.
+	select {
+	case <-watchCalledChan:
+		fakeController.Annotations = map[string]string{deployapi.DeploymentStatusAnnotation: string(deployapi.DeploymentStatusRunning)}
+		fakeWatch.Modify(fakeController)
+		<-stopChan
+	case <-time.After(time.Second * 5):
+		t.Fatalf("timeout waiting for the watch restart")
+	}
+}