Update tests and code for new grpc library

Signed-off-by: perdasilva <[email protected]>
Upstream-repository: olm-perdasilva
Upstream-commit: 1ab67ce1f8ae6fb02fff045bef649216317c78a8

Author: perdasilva

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

@@ -474,6 +474,15 @@ func (o *Operator) syncSourceState(state grpc.SourceState) {
 	metrics.RegisterCatalogSourceState(state.Key.Name, state.Key.Namespace, state.State)
 
 	switch state.State {
+	case connectivity.Idle:
+		// If the connection is IDLE attempt to connect to attempt to bring it to a READY state
+		// A connection will transition to IDLE if:
+		// "No RPC activity on channel for IDLE_TIMEOUT (default: 5 mins) or upon receiving a GOAWAY while there are no pending RPCs."
+		// https://gitlab.uni-hannover.de/tci-gateway-module/grpc/-/blob/972e31165218b49d93e5e1f1a1e8bbcd3fa830d1/doc/connectivity-semantics-and-api.md
+		sourceConn := o.sources.Get(state.Key)
+		if sourceConn != nil && sourceConn.Conn != nil {
+			_, _ = registry.NewClientFromConn(sourceConn.Conn).HealthCheck(context.Background(), time.Second*5)
+		}
 	case connectivity.Ready:
 		o.resolverSourceProvider.Invalidate(resolvercache.SourceKey(state.Key))
 		if o.namespace == state.Key.Namespace {

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

@@ -9,7 +9,6 @@ import (
 	"time"
 
 	"github.com/operator-framework/operator-registry/pkg/client"
-
 	"github.com/sirupsen/logrus"
 	"golang.org/x/net/http/httpproxy"
 	"golang.org/x/net/proxy"

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

@@ -134,6 +134,7 @@ func TestConnectionEvents(t *testing.T) {
 				{Name: "test", Namespace: "test"}: {
 					connectivity.Connecting,
 					connectivity.Ready,
+					connectivity.Idle,
 				},
 			},
 		},
@@ -143,10 +144,12 @@ func TestConnectionEvents(t *testing.T) {
 				{Name: "test", Namespace: "test"}: {
 					connectivity.Connecting,
 					connectivity.Ready,
+					connectivity.Idle,
 				},
 				{Name: "test2", Namespace: "test2"}: {
 					connectivity.Connecting,
 					connectivity.Ready,
+					connectivity.Idle,
 				},
 			},
 		},

staging/operator-lifecycle-manager/pkg/package-server/provider/registry.go

@@ -226,6 +226,15 @@ func (p *RegistryProvider) syncSourceState(state registrygrpc.SourceState) {
 
 	var err error
 	switch state.State {
+	case connectivity.Idle:
+		// If the connection is IDLE attempt to connect to attempt to bring it to a READY state
+		// A connection will transition to IDLE if:
+		// "No RPC activity on channel for IDLE_TIMEOUT (default: 5 mins) or upon receiving a GOAWAY while there are no pending RPCs."
+		// https://gitlab.uni-hannover.de/tci-gateway-module/grpc/-/blob/972e31165218b49d93e5e1f1a1e8bbcd3fa830d1/doc/connectivity-semantics-and-api.md
+		sourceConn := p.sources.Get(state.Key)
+		if sourceConn != nil && sourceConn.Conn != nil {
+			_, _ = registry.NewClientFromConn(sourceConn.Conn).HealthCheck(context.Background(), time.Second*5)
+		}
 	case connectivity.Ready:
 		var client *registryClient
 		client, err = p.registryClient(key)

staging/operator-lifecycle-manager/test/e2e/subscription_e2e_test.go

@@ -16,6 +16,7 @@ import (
 	configv1 "github.com/openshift/api/config/v1"
 	configv1client "github.com/openshift/client-go/config/clientset/versioned/typed/config/v1"
 	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc/connectivity"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apiextensions-apiserver/pkg/apis/apiextensions"
@@ -2920,8 +2921,9 @@ func updateInternalCatalog(t GinkgoTInterface, c operatorclient.ClientInterface,
 	_, err = fetchCatalogSourceOnStatus(crc, catalogSourceName, namespace, func(catalog *operatorsv1alpha1.CatalogSource) bool {
 		before := fetchedInitialCatalog.Status.ConfigMapResource
 		after := catalog.Status.ConfigMapResource
+		lastObservedState := catalog.Status.GRPCConnectionState.LastObservedState
 		if after != nil && after.LastUpdateTime.After(before.LastUpdateTime.Time) && after.ResourceVersion != before.ResourceVersion &&
-			catalog.Status.GRPCConnectionState.LastConnectTime.After(after.LastUpdateTime.Time) && catalog.Status.GRPCConnectionState.LastObservedState == "READY" {
+			catalog.Status.GRPCConnectionState.LastConnectTime.After(after.LastUpdateTime.Time) && (lastObservedState == connectivity.Ready.String() || lastObservedState == connectivity.Idle.String()) {
 			fmt.Println("catalog updated")
 			return true
 		}

staging/operator-lifecycle-manager/test/e2e/util.go

@@ -13,6 +13,7 @@ import (
 	. "github.com/onsi/gomega"
 	"github.com/stretchr/testify/require"
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/connectivity"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apiextensions-apiserver/pkg/apis/apiextensions"
@@ -337,11 +338,11 @@ func registryPodHealthy(address string) bool {
 }
 
 func catalogSourceRegistryPodSynced(catalog *operatorsv1alpha1.CatalogSource) bool {
-	registry := catalog.Status.RegistryServiceStatus
+	registryStatus := catalog.Status.RegistryServiceStatus
 	connState := catalog.Status.GRPCConnectionState
-	if registry != nil && connState != nil && !connState.LastConnectTime.IsZero() && connState.LastObservedState == "READY" {
-		fmt.Printf("catalog %s pod with address %s\n", catalog.GetName(), registry.Address())
-		return registryPodHealthy(registry.Address())
+	if registryStatus != nil && connState != nil && !connState.LastConnectTime.IsZero() && (connState.LastObservedState == connectivity.Ready.String() || connState.LastObservedState == connectivity.Idle.String()) {
+		fmt.Printf("catalog %s pod with address %s\n", catalog.GetName(), registryStatus.Address())
+		return registryPodHealthy(registryStatus.Address())
 	}
 	state := "NO_CONNECTION"
 	if connState != nil {