cmd/openshift-install/create: Retry watch connections

cmd/openshift-install/create.go

@@ -200,10 +200,21 @@ func destroyBootstrap(ctx context.Context, directory string) (err error) {
 		eventContext,
 		"",
 		func(sinceResourceVersion string) (watch.Interface, error) {
-			return events.Watch(metav1.ListOptions{
-				Watch:           true,
-				ResourceVersion: sinceResourceVersion,
-			})
+			for {
+				watcher, err := events.Watch(metav1.ListOptions{
+					ResourceVersion: sinceResourceVersion,
+				})
+				if err == nil {
+					return watcher, nil
+				}
+				select {
+				case <-eventContext.Done():
+					return watcher, err
+				default:
+					logrus.Warningf("Failed to connect events watcher: %s", err)
+					time.Sleep(2 * time.Second)
+				}
+			}
 		},
 		func(watchEvent watch.Event) (bool, error) {
 			event, ok := watchEvent.Object.(*corev1.Event)

cmd/openshift-install/watch.go

@@ -24,7 +24,6 @@ type RetryWatcher struct {
 	watcherFunc         WatcherFunc
 	resultChan          chan watch.Event
 	stopChan            chan struct{}
-	doneChan            chan struct{}
 }
 
 // Until is from https://github.com/kubernetes/kubernetes/pull/50102.
@@ -38,7 +37,6 @@ func NewRetryWatcher(initialResourceVersion string, watcherFunc WatcherFunc) *Re
 		lastResourceVersion: initialResourceVersion,
 		watcherFunc:         watcherFunc,
 		stopChan:            make(chan struct{}),
-		doneChan:            make(chan struct{}),
 		resultChan:          make(chan watch.Event, 0),
 	}
 	go rw.receive()
@@ -133,7 +131,7 @@ func (rw *RetryWatcher) doReceive() bool {
 }
 
 func (rw *RetryWatcher) receive() {
-	defer close(rw.doneChan)
+	defer close(rw.resultChan)
 
 	for {
 		select {
@@ -158,8 +156,3 @@ func (rw *RetryWatcher) ResultChan() <-chan watch.Event {
 func (rw *RetryWatcher) Stop() {
 	close(rw.stopChan)
 }
-
-// Done is from https://github.com/kubernetes/kubernetes/pull/50102.
-func (rw *RetryWatcher) Done() <-chan struct{} {
-	return rw.doneChan
-}