Merge pull request #1277 from benluddy/fix-ready-not-closed

Bug 1798135: Fix cases where an operator's ready channel may never close.

Author: openshift-merge-robot

pkg/lib/queueinformer/config.go

@@ -161,7 +161,7 @@ func WithSyncer(syncer kubestate.Syncer) Option {
 }
 
 type operatorConfig struct {
-	discovery      discovery.DiscoveryInterface
+	serverVersion  discovery.ServerVersionInterface
 	queueInformers []*QueueInformer
 	informers      []cache.SharedIndexInformer
 	logger         *logrus.Logger
@@ -217,7 +217,7 @@ func WithNumWorkers(numWorkers int) OperatorOption {
 // validate returns an error if the config isn't valid.
 func (c *operatorConfig) validate() (err error) {
 	switch config := c; {
-	case config.discovery == nil:
+	case config.serverVersion == nil:
 		err = newInvalidOperatorConfigError("discovery client nil")
 	case config.numWorkers < 1:
 		err = newInvalidOperatorConfigError("must specify at least one worker per queue")

pkg/lib/queueinformer/queueinformer_operator.go

@@ -46,7 +46,7 @@ type Operator interface {
 }
 
 type operator struct {
-	discovery        discovery.DiscoveryInterface
+	serverVersion    discovery.ServerVersionInterface
 	queueInformers   []*QueueInformer
 	informers        []cache.SharedIndexInformer
 	hasSynced        cache.InformerSynced
@@ -161,26 +161,34 @@ func (o *operator) RunInformers(ctx context.Context) {
 // Run starts the operator's control loops.
 func (o *operator) Run(ctx context.Context) {
 	o.reconcileOnce.Do(func() {
-		go o.run(ctx)
+		go func() {
+			defer func() {
+				for _, queueInformer := range o.queueInformers {
+					queueInformer.queue.ShutDown()
+				}
+				close(o.atLevel)
+				close(o.done)
+			}()
+			o.start(ctx)
+			<-ctx.Done()
+		}()
 	})
 }
 
-func (o *operator) run(ctx context.Context) {
-	defer func() {
-		close(o.atLevel)
-		close(o.done)
-	}()
-
-	for _, queueInformer := range o.queueInformers {
-		defer queueInformer.queue.ShutDown()
-	}
+func (o *operator) start(ctx context.Context) {
+	defer close(o.ready)
 
+	// goroutine will be unnecessary after https://github.com/kubernetes/enhancements/pull/1503
 	errs := make(chan error)
 	go func() {
 		defer close(errs)
-		v, err := o.discovery.ServerVersion()
+		v, err := o.serverVersion.ServerVersion()
 		if err != nil {
-			errs <- errors.Wrap(err, "communicating with server failed")
+			select {
+			case errs <- errors.Wrap(err, "communicating with server failed"):
+			case <-ctx.Done():
+				// don't block send forever on cancellation
+			}
 			return
 		}
 		o.logger.Infof("connection established. cluster-version: %v", v)
@@ -212,9 +220,6 @@ func (o *operator) run(ctx context.Context) {
 			go o.worker(ctx, queueInformer)
 		}
 	}
-
-	close(o.ready)
-	<-ctx.Done()
 }
 
 // worker runs a worker thread that just dequeues items, processes them, and marks them done.
@@ -293,10 +298,10 @@ func (o *operator) processNextWorkItem(ctx context.Context, loop *QueueInformer)
 	return true
 }
 
-// NewOperator returns a new Operator configured to manage the cluster with the given discovery client.
-func NewOperator(disc discovery.DiscoveryInterface, options ...OperatorOption) (Operator, error) {
+// NewOperator returns a new Operator configured to manage the cluster with the given server version client.
+func NewOperator(sv discovery.ServerVersionInterface, options ...OperatorOption) (Operator, error) {
 	config := defaultOperatorConfig()
-	config.discovery = disc
+	config.serverVersion = sv
 	config.apply(options)
 	if err := config.validate(); err != nil {
 		return nil, err
@@ -306,14 +311,14 @@ func NewOperator(disc discovery.DiscoveryInterface, options ...OperatorOption) (
 
 }
 
-func newOperatorFromConfig(config *operatorConfig) (Operator, error) {
+func newOperatorFromConfig(config *operatorConfig) (*operator, error) {
 	op := &operator{
-		discovery:  config.discovery,
-		numWorkers: config.numWorkers,
-		logger:     config.logger,
-		ready:      make(chan struct{}),
-		done:       make(chan struct{}),
-		atLevel:    make(chan error, 25),
+		serverVersion: config.serverVersion,
+		numWorkers:    config.numWorkers,
+		logger:        config.logger,
+		ready:         make(chan struct{}),
+		done:          make(chan struct{}),
+		atLevel:       make(chan error, 25),
 	}
 	op.syncCh = op.atLevel
 

pkg/lib/queueinformer/queueinformer_operator_test.go

@@ -0,0 +1,88 @@
+package queueinformer
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/pkg/errors"
+	"k8s.io/apimachinery/pkg/version"
+)
+
+type versionFunc func() (*version.Info, error)
+
+func (f versionFunc) ServerVersion() (*version.Info, error) {
+	if f == nil {
+		return &version.Info{}, nil
+	}
+	return (func() (*version.Info, error))(f)()
+}
+
+func TestOperatorRunReadyChannelClosed(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
+		of func(cancel context.CancelFunc, o *operator) func()
+	}{
+		{
+			name: "error getting server version",
+			of: func(cancel context.CancelFunc, o *operator) func() {
+				o.serverVersion = versionFunc(func() (*version.Info, error) {
+					return nil, errors.New("test error")
+				})
+				return func() {}
+			},
+		},
+		{
+			name: "context cancelled while getting server version",
+			of: func(cancel context.CancelFunc, o *operator) func() {
+				done := make(chan struct{})
+				o.serverVersion = versionFunc(func() (*version.Info, error) {
+					defer func() {
+						<-done
+					}()
+					cancel()
+					return nil, errors.New("test error")
+				})
+				return func() {
+					close(done)
+				}
+			},
+		},
+		{
+			name: "context cancelled before cache sync",
+			of: func(cancel context.CancelFunc, o *operator) func() {
+				o.hasSynced = func() bool {
+					cancel()
+					return false
+				}
+				return func() {}
+			},
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
+			o, err := newOperatorFromConfig(defaultOperatorConfig())
+			if err != nil {
+				t.Fatalf("could not create operator from default config: %s", err)
+			}
+			o.serverVersion = versionFunc(nil)
+			o.hasSynced = func() bool { return true }
+
+			done := func() {}
+			if tc.of != nil {
+				done = tc.of(cancel, o)
+			}
+			defer done()
+
+			o.Run(ctx)
+
+			select {
+			case <-o.Ready():
+			case <-time.After(time.Second):
+				t.Error("timed out before ready channel closed")
+			}
+		})
+	}
+}