Fix several bugs in cloudevent library (#107)

1. set resource version in codec with string type
2. Do not check source in lister
3. allow setting token for grpc connection

Signed-off-by: Jian Qiu <[email protected]>

Author: qiujian16

pkg/cloudevents/clients/cluster/codec.go

@@ -2,8 +2,6 @@ package cluster
 
 import (
 	"fmt"
-	"strconv"
-
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	clusterv1 "open-cluster-management.io/api/cluster/v1"
@@ -42,11 +40,7 @@ func (c *ManagedClusterCodec) Encode(source string, eventType types.CloudEventsT
 		NewEvent()
 
 	if cluster.ResourceVersion != "" {
-		resourceVersion, err := strconv.ParseInt(cluster.ResourceVersion, 10, 64)
-		if err != nil {
-			return nil, fmt.Errorf("failed to parse the resourceversion for managedcluster %s, %v", cluster.Name, err)
-		}
-		evt.SetExtension(types.ExtensionResourceVersion, resourceVersion)
+		evt.SetExtension(types.ExtensionResourceVersion, cluster.ResourceVersion)
 	}
 
 	newCluster := cluster.DeepCopy()

pkg/cloudevents/clients/csr/client.go

@@ -10,6 +10,7 @@ import (
 	certificatev1 "k8s.io/api/certificates/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	utilrand "k8s.io/apimachinery/pkg/util/rand"
 	"k8s.io/apimachinery/pkg/watch"
 	"k8s.io/klog/v2"
 
@@ -43,6 +44,11 @@ func NewCSRClient(
 }
 
 func (c *CSRClient) Create(ctx context.Context, csr *certificatev1.CertificateSigningRequest, opts metav1.CreateOptions) (*certificatev1.CertificateSigningRequest, error) {
+	// generate csr name if name is not set
+	if csr.Name == "" && csr.GenerateName != "" {
+		csr.Name = csr.GenerateName + utilrand.String(5)
+		csr.GenerateName = ""
+	}
 	klog.V(4).Infof("creating CSR %s", csr.Name)
 	_, exists, err := c.watcherStore.Get("", csr.Name)
 	if err != nil {

pkg/cloudevents/clients/csr/client_test.go

@@ -2,6 +2,7 @@ package csr
 
 import (
 	"context"
+	v1 "open-cluster-management.io/api/cluster/v1"
 	"testing"
 	"time"
 
@@ -32,6 +33,9 @@ func TestCreate(t *testing.T) {
 			csr: &certificatev1.CertificateSigningRequest{
 				ObjectMeta: metav1.ObjectMeta{
 					Name: "cluster1",
+					Labels: map[string]string{
+						v1.ClusterNameLabelKey: "cluster1",
+					},
 				},
 			},
 			expectedErr: "",
@@ -42,6 +46,9 @@ func TestCreate(t *testing.T) {
 			csr: &certificatev1.CertificateSigningRequest{
 				ObjectMeta: metav1.ObjectMeta{
 					Name: "cluster1",
+					Labels: map[string]string{
+						v1.ClusterNameLabelKey: "cluster1",
+					},
 				},
 			},
 			expectedErr: "certificatesigningrequests.certificates.k8s.io \"cluster1\" already exists",

pkg/cloudevents/clients/csr/codec.go

@@ -2,10 +2,9 @@ package csr
 
 import (
 	"fmt"
-	"strconv"
-
 	cloudevents "github.com/cloudevents/sdk-go/v2"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	v1 "open-cluster-management.io/api/cluster/v1"
 
 	certificatev1 "k8s.io/api/certificates/v1"
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/types"
@@ -35,17 +34,21 @@ func (c *CSRCodec) Encode(source string, eventType types.CloudEventsType, csr *c
 		return nil, fmt.Errorf("unsupported cloudevents data type %s", eventType.CloudEventsDataType)
 	}
 
+	if len(csr.Labels) == 0 {
+		return nil, fmt.Errorf("no cluster label found for CSR")
+	}
+	cluster, ok := csr.Labels[v1.ClusterNameLabelKey]
+	if !ok {
+		return nil, fmt.Errorf("no cluster name found for CSR")
+	}
+
 	evt := types.NewEventBuilder(source, eventType).
 		WithResourceID(csr.Name).
-		WithClusterName(csr.Name).
+		WithClusterName(cluster).
 		NewEvent()
 
 	if csr.ResourceVersion != "" {
-		resourceVersion, err := strconv.ParseInt(csr.ResourceVersion, 10, 64)
-		if err != nil {
-			return nil, fmt.Errorf("failed to parse the resourceversion for csr %s, %v", csr.Name, err)
-		}
-		evt.SetExtension(types.ExtensionResourceVersion, resourceVersion)
+		evt.SetExtension(types.ExtensionResourceVersion, csr.ResourceVersion)
 	}
 
 	newCSR := csr.DeepCopy()

pkg/cloudevents/clients/store/lister.go

@@ -1,11 +1,8 @@
 package store
 
 import (
-	"fmt"
-
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
-	"open-cluster-management.io/sdk-go/pkg/cloudevents/clients/common"
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/generic"
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/types"
 )
@@ -25,9 +22,7 @@ func NewAgentWatcherStoreLister[T generic.ResourceObject](store ClientWatcherSto
 func (l *AgentWatcherStoreLister[T]) List(options types.ListOptions) ([]T, error) {
 	opts := metav1.ListOptions{}
 
-	if options.Source != types.SourceAll {
-		opts.LabelSelector = fmt.Sprintf("%s=%s", common.CloudEventsOriginalSourceLabelKey, options.Source)
-	}
+	// TODO we might want to specify source when list
 
 	list, err := l.store.List("", opts)
 	if err != nil {

pkg/cloudevents/clients/store/lister_test.go

@@ -32,7 +32,7 @@ func TestAgentLister(t *testing.T) {
 	}
 	if err := clusterStore.Add(&clusterv1.ManagedCluster{ObjectMeta: metav1.ObjectMeta{
 		Name:   "test2",
-		Labels: map[string]string{common.CloudEventsOriginalSourceLabelKey: "source2"},
+		Labels: map[string]string{common.CloudEventsOriginalSourceLabelKey: "source1"},
 	}}); err != nil {
 		t.Error(err)
 	}
@@ -42,7 +42,7 @@ func TestAgentLister(t *testing.T) {
 	if err != nil {
 		t.Error(err)
 	}
-	if len(clusters) != 1 {
+	if len(clusters) != 2 {
 		t.Error("unexpected clusters")
 	}
 }

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

@@ -30,7 +30,7 @@ type GRPCDialer struct {
 	URL              string
 	KeepAliveOptions KeepAliveOptions
 	TLSConfig        *tls.Config
-	TokenFile        string
+	Token            string
 	mu               sync.Mutex       // Mutex to protect the connection.
 	conn             *grpc.ClientConn // Cached gRPC client connection.
 }
@@ -66,15 +66,10 @@ func (d *GRPCDialer) Dial() (*grpc.ClientConn, error) {
 	if d.TLSConfig != nil {
 		// Enable TLS
 		dialOpts = append(dialOpts, grpc.WithTransportCredentials(credentials.NewTLS(d.TLSConfig)))
-		if len(d.TokenFile) != 0 {
-			// Use token-based authentication if token file is provided.
-			token, err := os.ReadFile(d.TokenFile)
-			if err != nil {
-				return nil, fmt.Errorf("failed to read token file %s, %v", d.TokenFile, err)
-			}
+		if len(d.Token) != 0 {
 			perRPCCred := oauth.TokenSource{
 				TokenSource: oauth2.StaticTokenSource(&oauth2.Token{
-					AccessToken: string(token),
+					AccessToken: d.Token,
 				})}
 			// Add per-RPC credentials to the dial options.
 			dialOpts = append(dialOpts, grpc.WithPerRPCCredentials(perRPCCred))
@@ -130,6 +125,8 @@ type GRPCConfig struct {
 	ClientKeyFile string `json:"clientKeyFile,omitempty" yaml:"clientKeyFile,omitempty"`
 	// TokenFile is the file path to a token file for authentication.
 	TokenFile string `json:"tokenFile,omitempty" yaml:"tokenFile,omitempty"`
+	// Token is the token for authentication
+	Token string `json:"token" yaml:"token"`
 	// keepalive options
 	KeepAliveConfig KeepAliveConfig `json:"keepAliveConfig,omitempty" yaml:"keepAliveConfig,omitempty"`
 }
@@ -175,14 +172,22 @@ func BuildGRPCOptionsFromFlags(configPath string) (*GRPCOptions, error) {
 	if config.ClientCertFile != "" && config.ClientKeyFile != "" && config.CAFile == "" {
 		return nil, fmt.Errorf("setting clientCertFile and clientKeyFile requires caFile")
 	}
-	if config.TokenFile != "" && config.CAFile == "" {
+	token := config.Token
+	if config.Token == "" && config.TokenFile != "" {
+		tokenBytes, err := os.ReadFile(config.TokenFile)
+		if err != nil {
+			return nil, fmt.Errorf("failed to read token file %s, %v", config.TokenFile, err)
+		}
+		token = string(tokenBytes)
+	}
+	if token != "" && config.CAFile == "" {
 		return nil, fmt.Errorf("setting tokenFile requires caFile")
 	}
 
 	options := &GRPCOptions{
 		Dialer: &GRPCDialer{
-			URL:       config.URL,
-			TokenFile: config.TokenFile,
+			URL:   config.URL,
+			Token: token,
 		},
 	}
 

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

@@ -34,7 +34,7 @@ func TestBuildGRPCOptionsFromFlags(t *testing.T) {
 		},
 		{
 			name:             "token config without caFile",
-			config:           "{\"url\":\"test\",\"tokenFile\":\"test\"}",
+			config:           "{\"url\":\"test\",\"token\":\"test\"}",
 			expectedErrorMsg: "setting tokenFile requires caFile",
 		},
 		{

pkg/cloudevents/server/grpc/server.go

@@ -245,10 +245,6 @@ func (bkr *GRPCBroker) respondResyncSpecRequest(ctx context.Context, eventDataTy
 	}
 
 	for _, obj := range objs {
-		// only respond with the resource of the resync type
-		if obj.Type() != eventDataType.String() {
-			continue
-		}
 		// respond with the deleting resource regardless of the resource version
 		if _, ok := obj.Extensions()[types.ExtensionDeletionTimestamp]; ok {
 			err = bkr.handleRes(ctx, obj, eventDataType, "delete_request")
@@ -259,8 +255,8 @@ func (bkr *GRPCBroker) respondResyncSpecRequest(ctx context.Context, eventDataTy
 		}
 
 		lastResourceVersion := findResourceVersion(obj.ID(), resourceVersions.Versions)
-		rv := obj.Extensions()[types.ExtensionResourceVersion].(string)
-		currentResourceVersion, err := strconv.ParseInt(rv, 10, 64)
+		currentResourceVersion, err := strconv.ParseInt(
+			obj.Extensions()[types.ExtensionResourceVersion].(string), 10, 64)
 		if err != nil {
 			log.V(4).Info("ignore the obj %v since it has a invalid resourceVersion, %v", obj, err)
 			continue

test/integration/cloudevents/util/grpc.go

@@ -3,6 +3,7 @@ package util
 import (
 	"crypto/tls"
 	"crypto/x509"
+	"os"
 	"time"
 
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options/grpc"
@@ -36,7 +37,11 @@ func newGRPCOptions(certPool *x509.CertPool, brokerURL, tokenFile string) *grpc.
 	}
 
 	if tokenFile != "" {
-		grpcOptions.Dialer.TokenFile = tokenFile
+		token, err := os.ReadFile(tokenFile)
+		if err != nil {
+			panic(err)
+		}
+		grpcOptions.Dialer.Token = string(token)
 	}
 
 	return grpcOptions