🌱 single connection for grpc options.

pkg/cloudevents/generic/options/grpc/options.go

@@ -5,6 +5,7 @@ import (
 	"crypto/tls"
 	"fmt"
 	"os"
+	"sync"
 	"time"
 
 	"golang.org/x/oauth2"
@@ -30,7 +31,8 @@ type GRPCDialer struct {
 	KeepAliveOptions KeepAliveOptions
 	TLSConfig        *tls.Config
 	TokenFile        string
-	conn             *grpc.ClientConn
+	mu               sync.Mutex       // Mutex to protect the connection.
+	conn             *grpc.ClientConn // Cached gRPC client connection.
 }
 
 // KeepAliveOptions holds the keepalive options for the gRPC client.
@@ -43,6 +45,15 @@ type KeepAliveOptions struct {
 
 // Dial connects to the gRPC server and returns a gRPC client connection.
 func (d *GRPCDialer) Dial() (*grpc.ClientConn, error) {
+	// Return the cached connection if it exists and is ready.
+	// Should not return a nil or unready connection, otherwise the caller (cloudevents client)
+	// will not use the connection for sending and receiving events in reconnect scenarios.
+	// lock the connection to ensure the connnection is not created by multiple goroutines concurrently.
+	d.mu.Lock()
+	defer d.mu.Unlock()
+	if d.conn != nil && (d.conn.GetState() == connectivity.Connecting || d.conn.GetState() == connectivity.Ready) {
+		return d.conn, nil
+	}
 	// Prepare gRPC dial options.
 	dialOpts := []grpc.DialOption{}
 	if d.KeepAliveOptions.Enable {
@@ -94,6 +105,8 @@ func (d *GRPCDialer) Dial() (*grpc.ClientConn, error) {
 
 // Close closes the gRPC client connection.
 func (d *GRPCDialer) Close() error {
+	d.mu.Lock()
+	defer d.mu.Unlock()
 	if d.conn != nil {
 		return d.conn.Close()
 	}

pkg/cloudevents/generic/options/grpc/options_test.go

@@ -5,7 +5,8 @@ import (
 	"testing"
 	"time"
 
-	"k8s.io/apimachinery/pkg/api/equality"
+	"github.com/google/go-cmp/cmp"
+	"github.com/google/go-cmp/cmp/cmpopts"
 	clienttesting "open-cluster-management.io/sdk-go/pkg/testing"
 )
 
@@ -98,7 +99,7 @@ func TestBuildGRPCOptionsFromFlags(t *testing.T) {
 				}
 			}
 
-			if !equality.Semantic.DeepEqual(options, c.expectedOptions) {
+			if !cmp.Equal(options, c.expectedOptions, cmpopts.IgnoreUnexported(GRPCDialer{})) {
 				t.Errorf("unexpected options %v", options)
 			}
 		})

pkg/cloudevents/generic/optionsbuilder_test.go

@@ -7,8 +7,8 @@ import (
 	"testing"
 	"time"
 
-	"k8s.io/apimachinery/pkg/api/equality"
-
+	"github.com/google/go-cmp/cmp"
+	"github.com/google/go-cmp/cmp/cmpopts"
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options/grpc"
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options/mqtt"
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/types"
@@ -97,7 +97,7 @@ func assertOptions(t *testing.T, c buildingCloudEventsOptionTestCase) {
 		t.Errorf("unexpected error %v", err)
 	}
 
-	if !equality.Semantic.DeepEqual(config, c.expectedOptions) {
+	if !cmp.Equal(config, c.expectedOptions, cmpopts.IgnoreUnexported(mqtt.MQTTDialer{}, grpc.GRPCDialer{})) {
 		t.Errorf("unexpected config %v, %v", config, c.expectedOptions)
 	}
 

test/integration/cloudevents/options_racing_grpc_test.go

@@ -0,0 +1,195 @@
+package cloudevents
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"time"
+
+	"github.com/onsi/ginkgo"
+	"github.com/onsi/gomega"
+	"k8s.io/apimachinery/pkg/api/meta"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/rand"
+
+	"open-cluster-management.io/sdk-go/pkg/cloudevents/generic"
+	"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options/grpc"
+	"open-cluster-management.io/sdk-go/pkg/cloudevents/work/agent/codec"
+	"open-cluster-management.io/sdk-go/test/integration/cloudevents/agent"
+	"open-cluster-management.io/sdk-go/test/integration/cloudevents/source"
+	"open-cluster-management.io/sdk-go/test/integration/cloudevents/store"
+	"open-cluster-management.io/sdk-go/test/integration/cloudevents/util"
+
+	workv1client "open-cluster-management.io/api/client/work/clientset/versioned/typed/work/v1"
+)
+
+var _ = ginkgo.Describe("CloudEvents Options Racing Test - GRPC", func() {
+	ginkgo.Context("GRPC options racing test", func() {
+		var err error
+		var ctx context.Context
+		var cancel context.CancelFunc
+		var sourceID string
+		var resourceName string
+		var agentOptions *grpc.GRPCOptions
+		var sourceStore *store.MemoryStore
+		var sourceCloudEventsClient generic.CloudEventsClient[*store.Resource]
+
+		ginkgo.BeforeEach(func() {
+			ctx, cancel = context.WithCancel(context.Background())
+			sourceID = fmt.Sprintf("cloudevents-test-%s", rand.String(5))
+			resourceName = fmt.Sprintf("resource-%s", rand.String(5))
+			agentOptions = util.NewGRPCAgentOptions(grpcBrokerHost)
+			sourceStore = store.NewMemoryStore()
+
+			// forward the resource to agent and listen for the status from grpc broker
+			go func() {
+				for {
+					select {
+					case <-ctx.Done():
+						return
+					case res := <-grpcServer.ResourceChan():
+						if err := grpcBroker.UpdateResourceSpec(res); err != nil {
+							log.Printf("failed to update resource spec via gRPC broker %s, %v", res.ResourceID, err)
+						}
+					case res := <-grpcBroker.ResourceStatusChan():
+						// replace the source id with the original source id
+						res.Source = sourceID
+						if err := grpcServer.UpdateResourceStatus(res); err != nil {
+							log.Printf("failed to update resource status %s, %v", res.ResourceID, err)
+						}
+					}
+				}
+			}()
+
+			sourceOptions := grpc.NewSourceOptions(util.NewGRPCSourceOptions(certPool, grpcServerHost, tokenFile), sourceID)
+			sourceCloudEventsClient, err = source.StartResourceSourceClient(
+				ctx,
+				sourceOptions,
+				sourceID,
+				source.NewResourceLister(sourceStore),
+			)
+			gomega.Expect(err).ToNot(gomega.HaveOccurred())
+
+		})
+		ginkgo.AfterEach(func() {
+			// cancel the context to gracefully shutdown the agent
+			cancel()
+		})
+
+		ginkgo.It("Start two work agents to test the racing condition of grpc connection", func() {
+			var agentWorkClient1, agentWorkClient2 workv1client.ManifestWorkInterface
+			clusterName1 := fmt.Sprintf("cluster1-%s", rand.String(5))
+			clusterName2 := fmt.Sprintf("cluster2-%s", rand.String(5))
+			ctx1, cancel1 := context.WithCancel(ctx)
+			go func() {
+				ginkgo.By("start a work agent")
+				clientHolder, _, err := agent.StartWorkAgent(ctx1, clusterName1, agentOptions, codec.NewManifestBundleCodec())
+				gomega.Expect(err).ToNot(gomega.HaveOccurred())
+				agentWorkClient1 = clientHolder.ManifestWorks(clusterName1)
+			}()
+
+			go func() {
+				ginkgo.By("start another work agent")
+				clientHolder, _, err := agent.StartWorkAgent(ctx, clusterName2, agentOptions, codec.NewManifestBundleCodec())
+				gomega.Expect(err).ToNot(gomega.HaveOccurred())
+				agentWorkClient2 = clientHolder.ManifestWorks(clusterName2)
+			}()
+
+			time.Sleep(3 * time.Second) // sleep for the agents are subscribing to the broker
+
+			ginkgo.By("create resource from source")
+			resourceVersion := 1
+			resource1 := store.NewResource(clusterName1, resourceName, int64(resourceVersion))
+			sourceStore.Add(resource1)
+			err = sourceCloudEventsClient.Publish(ctx, createRequest, resource1)
+			gomega.Expect(err).ToNot(gomega.HaveOccurred())
+			resource2 := store.NewResource(clusterName2, resourceName, int64(resourceVersion))
+			sourceStore.Add(resource2)
+			err = sourceCloudEventsClient.Publish(ctx, createRequest, resource2)
+			gomega.Expect(err).ToNot(gomega.HaveOccurred())
+
+			ginkgo.By("ensure the work can be got by work agent")
+			gomega.Eventually(func() error {
+				workName1 := store.ResourceID(clusterName1, resourceName)
+				_, err := agentWorkClient1.Get(ctx, workName1, metav1.GetOptions{})
+				if err != nil {
+					return err
+				}
+
+				workName2 := store.ResourceID(clusterName2, resourceName)
+				_, err = agentWorkClient2.Get(ctx, workName2, metav1.GetOptions{})
+				if err != nil {
+					return err
+				}
+
+				return nil
+			}, 10*time.Second, 1*time.Second).Should(gomega.Succeed())
+
+			ginkgo.By("close the first work agent")
+			cancel1()
+			time.Sleep(1 * time.Second) // sleep for grpc connection is closed
+
+			ginkgo.By("update the resource status by the second work agent")
+			gomega.Eventually(func() error {
+				workName2 := store.ResourceID(clusterName2, resourceName)
+				if err := util.AddWorkFinalizer(ctx, agentWorkClient2, workName2); err != nil {
+					return err
+				}
+
+				if err := util.AssertWorkFinalizers(ctx, agentWorkClient2, workName2); err != nil {
+					return err
+				}
+
+				if err := util.UpdateWorkStatus(ctx, agentWorkClient2, workName2, util.WorkCreatedCondition); err != nil {
+					return err
+				}
+
+				return nil
+			}, 10*time.Second, 1*time.Second).Should(gomega.Succeed())
+
+			ginkgo.By("ensure the work status subscribed by source")
+			gomega.Eventually(func() error {
+				resource2, err = sourceStore.Get(store.ResourceID(clusterName2, resourceName))
+				if err != nil {
+					return err
+				}
+
+				if !meta.IsStatusConditionTrue(resource2.Status.Conditions, "Created") {
+					return fmt.Errorf("unexpected status %v", resource2.Status.Conditions)
+				}
+
+				return nil
+			}, 10*time.Second, 1*time.Second).Should(gomega.Succeed())
+
+			ginkgo.By("mark the resource deleting by source")
+			resource2, err = sourceStore.Get(store.ResourceID(clusterName2, resourceName))
+			gomega.Expect(err).ToNot(gomega.HaveOccurred())
+			resource2.DeletionTimestamp = &metav1.Time{Time: time.Now()}
+			err = sourceStore.Update(resource2)
+			gomega.Expect(err).ToNot(gomega.HaveOccurred())
+			err = sourceCloudEventsClient.Publish(ctx, deleteRequest, resource2)
+			gomega.Expect(err).ToNot(gomega.HaveOccurred())
+
+			ginkgo.By("delete the resource from agent")
+			gomega.Eventually(func() error {
+				workName := store.ResourceID(clusterName2, resourceName)
+				return util.RemoveWorkFinalizer(ctx, agentWorkClient2, workName)
+			}, 10*time.Second, 1*time.Second).Should(gomega.Succeed())
+
+			ginkgo.By("delete the resource from source")
+			gomega.Eventually(func() error {
+				resourceID := store.ResourceID(clusterName2, resourceName)
+				resource2, err = sourceStore.Get(resourceID)
+				if err != nil {
+					return err
+				}
+
+				if meta.IsStatusConditionTrue(resource2.Status.Conditions, "Deleted") {
+					sourceStore.Delete(resourceID)
+				}
+
+				return nil
+			}, 10*time.Second, 1*time.Second).Should(gomega.Succeed())
+		})
+	})
+})