test(grpc): add basic connection event tests for source manager

Author: ecordell

cmd/olm/cleanup.go

@@ -37,6 +37,10 @@ func cleanup(logger *logrus.Logger, c operatorclient.ClientInterface, crc versio
 		logger.WithError(err).Fatal("couldn't clean previous release")
 	}
 
+	if err := waitForDelete(checkClusterServiceVersion(crc, "packageserver.v0.10.1"), deleteClusterServiceVersion(crc, "packageserver.v0.10.0")); err != nil {
+		logger.WithError(err).Fatal("couldn't clean previous release")
+	}
+
 	if err := waitForDelete(checkClusterServiceVersion(crc, "packageserver.v0.9.0"), deleteClusterServiceVersion(crc, "packageserver.v0.9.0")); err != nil {
 		logger.WithError(err).Fatal("couldn't clean previous release")
 	}

pkg/controller/operators/catalog/operator.go

@@ -136,20 +136,24 @@ func NewOperator(ctx context.Context, kubeconfigPath string, clock utilclock.Clo
 		serviceAccountQuerier:  scoped.NewUserDefinedServiceAccountQuerier(logger, crClient),
 		clientAttenuator:       scoped.NewClientAttenuator(logger, config, opClient, crClient),
 	}
-	op.sources = grpc.NewSourceStore(logger, op.syncSourceState)
+	op.sources = grpc.NewSourceStore(logger, 10*time.Second, 10*time.Minute, op.syncSourceState)
 	op.reconciler = reconciler.NewRegistryReconcilerFactory(lister, opClient, configmapRegistryImage, op.now)
 
 	// Set up syncing for namespace-scoped resources
 	for _, namespace := range watchedNamespaces {
-		// Wire OLM CR informers
+		// Wire OLM CR sharedIndexInformers
 		crInformerFactory := externalversions.NewSharedInformerFactoryWithOptions(op.client, resyncPeriod, externalversions.WithNamespace(namespace))
 
 		// Wire CSVs
 		csvInformer := crInformerFactory.Operators().V1alpha1().ClusterServiceVersions()
 		op.lister.OperatorsV1alpha1().RegisterClusterServiceVersionLister(namespace, csvInformer.Lister())
-		op.RegisterInformer(csvInformer.Informer())
+		if err := op.RegisterInformer(csvInformer.Informer()); err != nil {
+			return nil, err
+		}
 
-		csvInformer.Informer().AddIndexers(cache.Indexers{index.ProvidedAPIsIndexFuncKey: index.ProvidedAPIsIndexFunc})
+		if err := csvInformer.Informer().AddIndexers(cache.Indexers{index.ProvidedAPIsIndexFuncKey: index.ProvidedAPIsIndexFunc}); err != nil {
+			return nil, err
+		}
 		csvIndexer := csvInformer.Informer().GetIndexer()
 		op.csvProvidedAPIsIndexer[namespace] = csvIndexer
 
@@ -228,43 +232,43 @@ func NewOperator(ctx context.Context, kubeconfigPath string, clock utilclock.Clo
 			return nil, err
 		}
 
-		// Wire k8s informers
+		// Wire k8s sharedIndexInformers
 		k8sInformerFactory := informers.NewSharedInformerFactoryWithOptions(op.opClient.KubernetesInterface(), resyncPeriod, informers.WithNamespace(namespace))
-		informers := []cache.SharedIndexInformer{}
+		sharedIndexInformers := []cache.SharedIndexInformer{}
 
 		// Wire Roles
 		roleInformer := k8sInformerFactory.Rbac().V1().Roles()
 		op.lister.RbacV1().RegisterRoleLister(namespace, roleInformer.Lister())
-		informers = append(informers, roleInformer.Informer())
+		sharedIndexInformers = append(sharedIndexInformers, roleInformer.Informer())
 
 		// Wire RoleBindings
 		roleBindingInformer := k8sInformerFactory.Rbac().V1().RoleBindings()
 		op.lister.RbacV1().RegisterRoleBindingLister(namespace, roleBindingInformer.Lister())
-		informers = append(informers, roleBindingInformer.Informer())
+		sharedIndexInformers = append(sharedIndexInformers, roleBindingInformer.Informer())
 
 		// Wire ServiceAccounts
 		serviceAccountInformer := k8sInformerFactory.Core().V1().ServiceAccounts()
 		op.lister.CoreV1().RegisterServiceAccountLister(namespace, serviceAccountInformer.Lister())
-		informers = append(informers, serviceAccountInformer.Informer())
+		sharedIndexInformers = append(sharedIndexInformers, serviceAccountInformer.Informer())
 
 		// Wire Services
 		serviceInformer := k8sInformerFactory.Core().V1().Services()
 		op.lister.CoreV1().RegisterServiceLister(namespace, serviceInformer.Lister())
-		informers = append(informers, serviceInformer.Informer())
+		sharedIndexInformers = append(sharedIndexInformers, serviceInformer.Informer())
 
 		// Wire Pods
 		podInformer := k8sInformerFactory.Core().V1().Pods()
 		op.lister.CoreV1().RegisterPodLister(namespace, podInformer.Lister())
-		informers = append(informers, podInformer.Informer())
+		sharedIndexInformers = append(sharedIndexInformers, podInformer.Informer())
 
 		// Wire ConfigMaps
 		configMapInformer := k8sInformerFactory.Core().V1().ConfigMaps()
 		op.lister.CoreV1().RegisterConfigMapLister(namespace, configMapInformer.Lister())
-		informers = append(informers, configMapInformer.Informer())
+		sharedIndexInformers = append(sharedIndexInformers, configMapInformer.Informer())
 
 		// Generate and register QueueInformers for k8s resources
 		k8sSyncer := queueinformer.LegacySyncHandler(op.syncObject).ToSyncerWithDelete(op.handleDeletion)
-		for _, informer := range informers {
+		for _, informer := range sharedIndexInformers {
 			queueInformer, err := queueinformer.NewQueueInformer(
 				ctx,
 				queueinformer.WithLogger(op.logger),
@@ -294,7 +298,9 @@ func NewOperator(ctx context.Context, kubeconfigPath string, clock utilclock.Clo
 	if err != nil {
 		return nil, err
 	}
-	op.RegisterQueueInformer(crdQueueInformer)
+	if err := op.RegisterQueueInformer(crdQueueInformer); err != nil {
+		return nil, err
+	}
 
 	// Namespace sync for resolving subscriptions
 	namespaceInformer := informers.NewSharedInformerFactory(op.opClient.KubernetesInterface(), resyncPeriod).Core().V1().Namespaces()
@@ -328,7 +334,7 @@ func (o *Operator) syncSourceState(state grpc.SourceState) {
 
 	switch state.State {
 	case connectivity.Ready:
-		o.resolveNamespace(state.Key.Namespace)
+		o.nsResolveQueue.Add(state.Key.Namespace)
 	default:
 		if err := o.catsrcQueueSet.Requeue(state.Key.Namespace, state.Key.Name); err != nil {
 			o.logger.WithError(err).Info("couldn't requeue catalogsource from catalog status change")
@@ -509,12 +515,6 @@ func (o *Operator) syncCatalogSources(obj interface{}) (syncError error) {
 			return err
 		}
 
-		logger.Debug("updating catsrc status")
-		if _, err := o.client.OperatorsV1alpha1().CatalogSources(out.GetNamespace()).UpdateStatus(out); err != nil {
-			return err
-		}
-		logger.Debug("registry server recreated")
-
 		if err := o.sources.Remove(sourceKey); err != nil {
 			o.logger.WithError(err).Debug("error closing client connection")
 		}
@@ -672,10 +672,6 @@ func (o *Operator) syncSubscriptions(obj interface{}) error {
 	return nil
 }
 
-func (o *Operator) resolveNamespace(namespace string) {
-	o.nsResolveQueue.Add(namespace)
-}
-
 func (o *Operator) nothingToUpdate(logger *logrus.Entry, sub *v1alpha1.Subscription) bool {
 	// Only sync if catalog has been updated since last sync time
 	if o.sourcesLastUpdate.Before(&sub.Status.LastUpdated) && sub.Status.State != v1alpha1.SubscriptionStateNone && sub.Status.State != v1alpha1.SubscriptionStateUpgradeAvailable {

pkg/controller/operators/catalog/operator_test.go

@@ -787,12 +787,12 @@ func NewFakeOperator(ctx context.Context, namespace string, watchedNamespaces []
 				// 1 qps, 100 bucket size.  This is only for retry speed and its only the overall factor (not per item)
 				&workqueue.BucketRateLimiter{Limiter: rate.NewLimiter(rate.Limit(1), 100)},
 			), "resolver"),
-		resolver:              &fakes.FakeResolver{},
+		resolver:              config.resolver,
 		reconciler:            config.reconciler,
 		clientAttenuator:      scoped.NewClientAttenuator(logger, &rest.Config{}, opClientFake, clientFake),
 		serviceAccountQuerier: scoped.NewUserDefinedServiceAccountQuerier(logger, clientFake),
 	}
-	op.sources = grpc.NewSourceStore(config.logger, op.syncSourceState)
+	op.sources = grpc.NewSourceStore(config.logger, 1*time.Second, 5*time.Second, op.syncSourceState)
 	if op.reconciler == nil {
 		op.reconciler = reconciler.NewRegistryReconcilerFactory(lister, op.opClient, "test:pod", op.now)
 	}

pkg/controller/registry/grpc/source.go

@@ -35,35 +35,42 @@ type SourceConn struct {
 }
 
 type SourceStore struct {
-	sources     map[resolver.CatalogKey]SourceConn
-	sourcesLock sync.RWMutex
-	syncFn      func(SourceState)
-	logger      *logrus.Logger
-	notify      chan SourceState
+	sync.Once
+	sources      map[resolver.CatalogKey]SourceConn
+	sourcesLock  sync.RWMutex
+	syncFn       func(SourceState)
+	logger       *logrus.Logger
+	notify       chan SourceState
+	timeout      time.Duration
+	readyTimeout time.Duration
 }
 
-func NewSourceStore(logger *logrus.Logger, sync func(SourceState)) *SourceStore {
+func NewSourceStore(logger *logrus.Logger, timeout, readyTimeout time.Duration, sync func(SourceState)) *SourceStore {
 	return &SourceStore{
-		sources: make(map[resolver.CatalogKey]SourceConn),
-		notify:  make(chan SourceState),
-		syncFn:  sync,
-		logger:  logger,
+		sources:      make(map[resolver.CatalogKey]SourceConn),
+		notify:       make(chan SourceState),
+		syncFn:       sync,
+		logger:       logger,
+		timeout:      timeout,
+		readyTimeout: readyTimeout,
 	}
 }
 
 func (s *SourceStore) Start(ctx context.Context) {
-	s.logger.Warn("starting it")
+	s.logger.Debug("starting source manager")
 	go func() {
-		for {
-			select {
-			case <-ctx.Done():
-				s.logger.Warn("ending it")
-				return
-			case e := <-s.notify:
-				s.logger.Warnf("Got event: %#v", e)
-				s.syncFn(e)
+		s.Do(func() {
+			for {
+				select {
+				case <-ctx.Done():
+					s.logger.Debug("closing source manager")
+					return
+				case e := <-s.notify:
+					s.logger.Debugf("Got source event: %#v", e)
+					s.syncFn(e)
+				}
 			}
-		}
+		})
 	}()
 }
 
@@ -123,37 +130,39 @@ func (s *SourceStore) Add(key resolver.CatalogKey, address string) (*SourceConn,
 	return &source, nil
 }
 
+func (s *SourceStore) stateTimeout(state connectivity.State) time.Duration {
+	if state == connectivity.Ready {
+		return s.readyTimeout
+	}
+	return s.timeout
+}
+
 func (s *SourceStore) watch(ctx context.Context, key resolver.CatalogKey, source SourceConn) {
 	state := source.ConnectionState
 	for {
 		select {
 		case <-ctx.Done():
 			return
 		default:
-			s.logger.Warnf("source state: %s", state.String())
-			timeout := 10 * time.Second
-			if state == connectivity.Ready {
-				timeout = 10 * time.Minute
-			}
-			timer, _ := context.WithTimeout(context.Background(), timeout)
+			timer, _ := context.WithTimeout(ctx, s.stateTimeout(state))
 			if source.Conn.WaitForStateChange(timer, state) {
 				newState := source.Conn.GetState()
 				state = newState
-				s.logger.Warnf("source state changed: %s", newState.String())
 
 				// update connection state
-				if src := s.Get(key); src != nil {
-					src.LastConnect = metav1.Now()
-					src.ConnectionState = newState
-					s.logger.Warnf("setting state")
-					s.sourcesLock.Lock()
-					s.sources[key] = *src
-					s.sourcesLock.Unlock()
-					s.logger.Warnf("state set")
+				src := s.Get(key)
+				if src == nil {
+					// source was removed, cleanup this goroutine
+					return
 				}
 
+				src.LastConnect = metav1.Now()
+				src.ConnectionState = newState
+				s.sourcesLock.Lock()
+				s.sources[key] = *src
+				s.sourcesLock.Unlock()
+
 				// notify subscriber
-				s.logger.Warnf("notify")
 				s.notify <- SourceState{Key: key, State: newState}
 			}
 		}