Fix

Signed-off-by: perdasilva <perdasilva@redhat.com>

Author: perdasilva

staging/operator-lifecycle-manager/cmd/catalog/main.go

@@ -47,6 +47,7 @@ func main() {
 }
 
 func (o *options) run(ctx context.Context, logger *logrus.Logger) error {
+	logger.Info("[CHECK] YES!!!!")
 	// If the catalogNamespaceEnvVarName environment variable is set, then  update the value of catalogNamespace.
 	if catalogNamespaceEnvVarValue := os.Getenv(catalogNamespaceEnvVarName); catalogNamespaceEnvVarValue != "" {
 		logger.Infof("%s environment variable is set. Updating Global Catalog Namespace to %s", catalogNamespaceEnvVarName, catalogNamespaceEnvVarValue)

staging/operator-lifecycle-manager/pkg/controller/operators/catalog/operator.go

@@ -463,9 +463,12 @@ func (o *Operator) syncSourceState(state grpc.SourceState) {
 	o.sourcesLastUpdate.Set(o.now().Time)
 
 	o.logger.Debugf("state.Key.Namespace=%s state.Key.Name=%s state.State=%s", state.Key.Namespace, state.Key.Name, state.State.String())
+	o.logger.Debug("[STATE] hello")
 	metrics.RegisterCatalogSourceState(state.Key.Name, state.Key.Namespace, state.State)
 
 	switch state.State {
+	case connectivity.Idle:
+		fallthrough
 	case connectivity.Ready:
 		o.resolverSourceProvider.Invalidate(resolvercache.SourceKey(state.Key))
 		if o.namespace == state.Key.Namespace {
@@ -765,6 +768,8 @@ func (o *Operator) syncConnection(logger *logrus.Entry, in *v1alpha1.CatalogSour
 	}
 
 	updateConnectionStateFunc := func(out *v1alpha1.CatalogSource, source *grpc.SourceMeta) {
+		logger.Infof("[TAG] Updating connection state: %s -> %s (%s)", source.Address, source.ConnectionState.String(), source.LastConnect.String())
+		fmt.Printf("[TAG] Updating connection state: %s -> %s (%s)\n", source.Address, source.ConnectionState.String(), source.LastConnect.String())
 		out.Status.GRPCConnectionState = &v1alpha1.GRPCConnectionState{
 			Address:           source.Address,
 			LastObservedState: source.ConnectionState.String(),
@@ -862,15 +867,19 @@ func (o *Operator) syncCatalogSources(obj interface{}) (syncError error) {
 	out, syncError := syncFunc(in, chain)
 
 	if out == nil {
+		logger.Debug("[TAG] out is nil -- skipping update")
 		return
 	}
 
 	if equalFunc(&catsrc.Status, &out.Status) {
+		logger.Debug("[TAG] no change in status -- skipping update")
 		return
 	}
 
+	logger.Debug("[TAG] Updating status")
 	updateErr := catalogsource.UpdateStatus(logger, o.client, out)
 	if syncError == nil && updateErr != nil {
+		logger.Debugf("[TAG] error updating: %s\n", updateErr)
 		syncError = updateErr
 	}
 

staging/operator-lifecycle-manager/pkg/controller/registry/grpc/source.go

@@ -169,7 +169,9 @@ func grpcConnection(address string) (*grpc.ClientConn, error) {
 func (s *SourceStore) Add(key registry.CatalogKey, address string) (*SourceConn, error) {
 	_ = s.Remove(key)
 
-	conn, err := grpcConnection(address)
+	s.logger.Debugf("[STATE] (%s) Adding new source with address: %s", key, address)
+	conn, err := grpc.Dial(address, grpc.WithInsecure())
+	// conn, err := grpcConnection(address)
 	if err != nil {
 		return nil, err
 	}
@@ -189,6 +191,7 @@ func (s *SourceStore) Add(key registry.CatalogKey, address string) (*SourceConn,
 	s.sources[key] = source
 	s.sourcesLock.Unlock()
 
+	s.logger.Debugf("[STATE] (%s) watching source with address: %s", key, address)
 	go s.watch(ctx, key, source)
 
 	return &source, nil
@@ -203,21 +206,29 @@ func (s *SourceStore) stateTimeout(state connectivity.State) time.Duration {
 
 func (s *SourceStore) watch(ctx context.Context, key registry.CatalogKey, source SourceConn) {
 	state := source.ConnectionState
+	s.logger.Debugf("[STATE] (%s) Watching for state changes", key)
 	for {
 		select {
 		case <-ctx.Done():
+			s.logger.Debugf("[STATE] (%s) Done", key)
 			return
 		default:
 			func() {
+				s.logger.Debugf("[STATE] (%s) Checking", key)
 				timer, cancel := context.WithTimeout(ctx, s.stateTimeout(state))
-				defer cancel()
+				defer func() {
+					s.logger.Debugf("[STATE] (%s) Cancelling", key)
+					cancel()
+				}()
 				if source.Conn.WaitForStateChange(timer, state) {
 					newState := source.Conn.GetState()
 					state = newState
+					s.logger.Debugf("[STATE] (%s) State: %s", key, state)
 
 					// update connection state
 					src := s.Get(key)
 					if src == nil {
+						s.logger.Debugf("[STATE] (%s) source not found -- clean up", key)
 						// source was removed, cleanup this goroutine
 						return
 					}
@@ -229,7 +240,10 @@ func (s *SourceStore) watch(ctx context.Context, key registry.CatalogKey, source
 					s.sourcesLock.Unlock()
 
 					// notify subscriber
+					s.logger.Debugf("[STATE] (%s) publishing state", key)
 					s.notify <- SourceState{Key: key, State: newState}
+				} else {
+					s.logger.Debugf("[STATE] no state change within timeout window")
 				}
 			}()
 		}

staging/operator-lifecycle-manager/pkg/lib/queueinformer/queueinformer_operator.go

@@ -195,6 +195,7 @@ func (o *operator) start(ctx context.Context) error {
 			return
 		}
 		o.logger.Infof("connection established. cluster-version: %v", v)
+		o.logger.Infof("connection is _really_ established: %v", v)
 	}()
 
 	select {