Don't block on ctx.Done() if startup fails.

Operator startup errors aren't recoverable without restarting the
process, so it's critical that the operator process is allowed to
terminate in these cases.

Author: benluddy

pkg/lib/queueinformer/queueinformer_operator.go

@@ -169,13 +169,16 @@ func (o *operator) Run(ctx context.Context) {
 				close(o.atLevel)
 				close(o.done)
 			}()
-			o.start(ctx)
+			if err := o.start(ctx); err != nil {
+				o.logger.WithError(err).Error("error encountered during startup")
+				return
+			}
 			<-ctx.Done()
 		}()
 	})
 }
 
-func (o *operator) start(ctx context.Context) {
+func (o *operator) start(ctx context.Context) error {
 	defer close(o.ready)
 
 	// goroutine will be unnecessary after https://github.com/kubernetes/enhancements/pull/1503
@@ -197,21 +200,19 @@ func (o *operator) start(ctx context.Context) {
 	select {
 	case err := <-errs:
 		if err != nil {
-			o.logger.Infof("operator not ready: %s", err.Error())
-			return
+			return fmt.Errorf("operator not ready: %s", err.Error())
 		}
 		o.logger.Info("operator ready")
 	case <-ctx.Done():
-		return
+		return nil
 	}
 
 	o.logger.Info("starting informers...")
 	o.RunInformers(ctx)
 
 	o.logger.Info("waiting for caches to sync...")
 	if ok := cache.WaitForCacheSync(ctx.Done(), o.hasSynced); !ok {
-		o.logger.Info("failed to wait for caches to sync")
-		return
+		return fmt.Errorf("failed to wait for caches to sync")
 	}
 
 	o.logger.Info("starting workers...")
@@ -220,6 +221,8 @@ func (o *operator) start(ctx context.Context) {
 			go o.worker(ctx, queueInformer)
 		}
 	}
+
+	return nil
 }
 
 // worker runs a worker thread that just dequeues items, processes them, and marks them done.

pkg/lib/queueinformer/queueinformer_operator_test.go

@@ -18,7 +18,7 @@ func (f versionFunc) ServerVersion() (*version.Info, error) {
 	return (func() (*version.Info, error))(f)()
 }
 
-func TestOperatorRunReadyChannelClosed(t *testing.T) {
+func TestOperatorRunChannelClosure(t *testing.T) {
 	for _, tc := range []struct {
 		name string
 		// set up the operator under test and return a cleanup func to be invoked when the test completes
@@ -78,10 +78,16 @@ func TestOperatorRunReadyChannelClosed(t *testing.T) {
 
 			o.Run(ctx)
 
-			select {
-			case <-o.Ready():
-			case <-time.After(time.Second):
-				t.Error("timed out before ready channel closed")
+			timeout := time.After(time.Second)
+			for n, ch := range map[string]<-chan struct{}{
+				"ready": o.Ready(),
+				"done":  o.Done(),
+			} {
+				select {
+				case <-ch:
+				case <-timeout:
+					t.Errorf("timed out before %s channel closed", n)
+				}
 			}
 		})
 	}