Implement grpc

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

Author: qiujian16

cmd/registration/main.go

@@ -17,6 +17,7 @@ import (
 	"open-cluster-management.io/ocm/pkg/cmd/spoke"
 	"open-cluster-management.io/ocm/pkg/cmd/webhook"
 	"open-cluster-management.io/ocm/pkg/features"
+	"open-cluster-management.io/ocm/pkg/server/grpc"
 	"open-cluster-management.io/ocm/pkg/version"
 )
 
@@ -62,6 +63,7 @@ func newRegistrationCommand() *cobra.Command {
 	cmd.AddCommand(hub.NewRegistrationController())
 	cmd.AddCommand(spoke.NewRegistrationAgent())
 	cmd.AddCommand(webhook.NewRegistrationWebhook())
+	cmd.AddCommand(grpc.NewGRPCServer())
 
 	return cmd
 }

go.mod

@@ -8,6 +8,7 @@ require (
 	github.com/aws/aws-sdk-go-v2/service/eks v1.57.1
 	github.com/aws/aws-sdk-go-v2/service/iam v1.38.6
 	github.com/aws/smithy-go v1.22.2
+	github.com/cloudevents/sdk-go/v2 v2.15.3-0.20240911135016-682f3a9684e4
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc
 	github.com/evanphx/json-patch v5.9.0+incompatible
 	github.com/ghodss/yaml v1.0.0
@@ -24,6 +25,8 @@ require (
 	github.com/spf13/pflag v1.0.5
 	github.com/stretchr/testify v1.10.0
 	github.com/valyala/fasttemplate v1.2.2
+	golang.org/x/net v0.34.0
+	google.golang.org/grpc v1.67.0
 	gopkg.in/yaml.v2 v2.4.0
 	helm.sh/helm/v3 v3.16.3
 	k8s.io/api v0.31.4
@@ -44,6 +47,8 @@ require (
 	sigs.k8s.io/yaml v1.4.0
 )
 
+replace open-cluster-management.io/sdk-go => github.com/qiujian16/sdk-go v0.0.0-20250401062857-2c1e3b8c9c6b
+
 require (
 	cloud.google.com/go/compute/metadata v0.5.0 // indirect
 	dario.cat/mergo v1.0.1 // indirect
@@ -72,7 +77,6 @@ require (
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/cloudevents/sdk-go/protocol/kafka_confluent/v2 v2.0.0-20240413090539-7fef29478991 // indirect
 	github.com/cloudevents/sdk-go/protocol/mqtt_paho/v2 v2.0.0-20241008145627-6bcc075b5b6c // indirect
-	github.com/cloudevents/sdk-go/v2 v2.15.3-0.20240911135016-682f3a9684e4 // indirect
 	github.com/confluentinc/confluent-kafka-go/v2 v2.3.0 // indirect
 	github.com/coreos/go-semver v0.3.1 // indirect
 	github.com/coreos/go-systemd/v22 v22.5.0 // indirect
@@ -156,7 +160,6 @@ require (
 	go.uber.org/zap v1.27.0 // indirect
 	golang.org/x/crypto v0.32.0 // indirect
 	golang.org/x/exp v0.0.0-20241217172543-b2144cdd0a67 // indirect
-	golang.org/x/net v0.34.0 // indirect
 	golang.org/x/oauth2 v0.22.0 // indirect
 	golang.org/x/sync v0.10.0 // indirect
 	golang.org/x/sys v0.29.0 // indirect
@@ -168,7 +171,6 @@ require (
 	google.golang.org/genproto v0.0.0-20240123012728-ef4313101c80 // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20240814211410-ddb44dafa142 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20240903143218-8af14fe29dc1 // indirect
-	google.golang.org/grpc v1.67.0 // indirect
 	google.golang.org/protobuf v1.35.1 // indirect
 	gopkg.in/evanphx/json-patch.v4 v4.12.0 // indirect
 	gopkg.in/inf.v0 v0.9.1 // indirect

go.sum

@@ -291,6 +291,8 @@ github.com/prometheus/common v0.55.0 h1:KEi6DK7lXW/m7Ig5i47x0vRzuBsHuvJdi5ee6Y3G
 github.com/prometheus/common v0.55.0/go.mod h1:2SECS4xJG1kd8XF9IcM1gMX6510RAEL65zxzNImwdc8=
 github.com/prometheus/procfs v0.15.1 h1:YagwOFzUgYfKKHX6Dr+sHT7km/hxC76UB0learggepc=
 github.com/prometheus/procfs v0.15.1/go.mod h1:fB45yRUv8NstnjriLhBQLuOUt+WW4BsoGhij/e3PBqk=
+github.com/qiujian16/sdk-go v0.0.0-20250401062857-2c1e3b8c9c6b h1:jpIFO9RkBWKhIGXHdPCW5bmJTyVVLlw8yYnT+CrubAE=
+github.com/qiujian16/sdk-go v0.0.0-20250401062857-2c1e3b8c9c6b/go.mod h1:TyOjZC5YxyM5BRNgwTmLuTbHXX6xXqzYBXllrfoVp9w=
 github.com/robfig/cron v1.2.0 h1:ZjScXvvxeQ63Dbyxy76Fj3AT3Ut0aKsyd2/tl3DTMuQ=
 github.com/robfig/cron v1.2.0/go.mod h1:JGuDeoQd7Z6yL4zQhZ3OPEVHB7fL6Ka6skscFHfmt2k=
 github.com/rogpeppe/go-internal v1.12.0 h1:exVL4IDcn6na9z1rAb56Vxr+CgyK3nn3O+epU5NdKM8=
@@ -491,8 +493,6 @@ open-cluster-management.io/addon-framework v0.12.0 h1:5j7mpyk2ij0SLUZkwWk0KkNTWt
 open-cluster-management.io/addon-framework v0.12.0/go.mod h1:eReMWXrEHqtilwz5wzEpUrWw9Vfz0HJCH9pi3gOTZns=
 open-cluster-management.io/api v0.16.1 h1:mS+4UGxHLPQd7CRM0gdFQdVaz139Lo2bkLfqSE0CDNU=
 open-cluster-management.io/api v0.16.1/go.mod h1:9erZEWEn4bEqh0nIX2wA7f/s3KCuFycQdBrPrRzi0QM=
-open-cluster-management.io/sdk-go v0.16.1-0.20250327091909-6bd6228a47ad h1:37f9TEwX/U8esBjSJvPleDM3rcFpk9NY5e2ItjO6PcQ=
-open-cluster-management.io/sdk-go v0.16.1-0.20250327091909-6bd6228a47ad/go.mod h1:TyOjZC5YxyM5BRNgwTmLuTbHXX6xXqzYBXllrfoVp9w=
 sigs.k8s.io/apiserver-network-proxy/konnectivity-client v0.30.3 h1:2770sDpzrjjsAtVhSeUFseziht227YAWYHLGNM8QPwY=
 sigs.k8s.io/apiserver-network-proxy/konnectivity-client v0.30.3/go.mod h1:Ve9uj1L+deCXFrPOk1LpFXqTg7LCFzFso6PA48q/XZw=
 sigs.k8s.io/cluster-inventory-api v0.0.0-20240730014211-ef0154379848 h1:WYPi2PdQyZwZkHG648v2jQl6deyCgyjJ0fkLYgUJ618=

pkg/registration/hub/manager.go

@@ -44,6 +44,7 @@ import (
 	"open-cluster-management.io/ocm/pkg/registration/register"
 	awsirsa "open-cluster-management.io/ocm/pkg/registration/register/aws_irsa"
 	"open-cluster-management.io/ocm/pkg/registration/register/csr"
+	"open-cluster-management.io/ocm/pkg/registration/register/grpc"
 )
 
 // HubManagerOptions holds configuration for hub manager controller
@@ -56,6 +57,8 @@ type HubManagerOptions struct {
 	AutoApprovedCSRUsers       []string
 	AutoApprovedARNPatterns    []string
 	AwsResourceTags            []string
+	GRPCCAFile                 string
+	GRPCCAKeyFile              string
 }
 
 // NewHubManagerOptions returns a HubManagerOptions
@@ -88,6 +91,10 @@ func (m *HubManagerOptions) AddFlags(fs *pflag.FlagSet) {
 	fs.StringSliceVar(&m.AutoApprovedARNPatterns, "auto-approved-arn-patterns", m.AutoApprovedARNPatterns,
 		"A list of AWS EKS ARN patterns such that an EKS cluster will be auto approved if its ARN matches with any of the patterns")
 	fs.StringSliceVar(&m.AwsResourceTags, "aws-resource-tags", m.AwsResourceTags, "A list of tags to apply to AWS resources created through the OCM controllers")
+	fs.StringVar(&m.GRPCCAFile, "grpc-ca-file", m.GRPCCAFile,
+		"ca file to sign client cert for grpc")
+	fs.StringVar(&m.GRPCCAKeyFile, "grpc-key-file", m.GRPCCAKeyFile,
+		"ca key file to sign client cert for grpc")
 	m.ImportOption.AddFlags(fs)
 }
 
@@ -186,6 +193,13 @@ func (m *HubManagerOptions) RunControllerManagerWithInformers(
 				return err
 			}
 			drivers = append(drivers, awsIRSAHubDriver)
+		case "grpc":
+			grpcHubDriver, err := grpc.NewGRPCHubDriver(
+				kubeClient, kubeInformers, m.GRPCCAKeyFile, m.GRPCCAFile, 720*time.Hour, controllerContext.EventRecorder)
+			if err != nil {
+				return err
+			}
+			drivers = append(drivers, grpcHubDriver)
 		}
 	}
 	hubDriver := register.NewAggregatedHubDriver(drivers...)

pkg/registration/register/csr/certificate.go

@@ -33,7 +33,7 @@ import (
 // isCertificateValid return true if
 // 1) All certs in client certificate are not expired.
 // 2) At least one cert matches the given subject if specified
-func isCertificateValid(logger klog.Logger, certData []byte, subject *pkix.Name) (bool, error) {
+func IsCertificateValid(logger klog.Logger, certData []byte, subject *pkix.Name) (bool, error) {
 	certs, err := certutil.ParseCertsPEM(certData)
 	if err != nil {
 		return false, fmt.Errorf("unable to parse certificate: %v", err)

pkg/registration/register/csr/certificate_test.go

@@ -147,7 +147,7 @@ func TestIsCertificateValid(t *testing.T) {
 	for _, c := range cases {
 		t.Run(c.name, func(t *testing.T) {
 			logger, _ := ktesting.NewTestContext(t)
-			isValid, _ := isCertificateValid(logger, c.testCert.Cert, c.subject)
+			isValid, _ := IsCertificateValid(logger, c.testCert.Cert, c.subject)
 			if isValid != c.isValid {
 				t.Errorf("expected %t, but got %t", c.isValid, isValid)
 			}

pkg/registration/register/csr/csr.go

@@ -18,7 +18,6 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/client-go/informers"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/tools/cache"
@@ -287,7 +286,7 @@ func (c *CSRDriver) IsHubKubeConfigValid(ctx context.Context, secretOption regis
 		}
 	}
 
-	return isCertificateValid(logger, certData, nil)
+	return IsCertificateValid(logger, certData, nil)
 }
 
 func (c *CSRDriver) ManagedClusterDecorator(cluster *clusterv1.ManagedCluster) *clusterv1.ManagedCluster {
@@ -348,23 +347,30 @@ func (c *CSRDriver) BuildClients(ctx context.Context, secretOption register.Secr
 		return nil, fmt.Errorf("failed to create CSR control: %w", err)
 	}
 
-	err = csrControl.Informer().AddIndexers(cache.Indexers{
+	err = c.SetCSRControl(csrControl, secretOption.ClusterName)
+	if err != nil {
+		return nil, fmt.Errorf("failed to set CSR control: %w", err)
+	}
+	return clients, nil
+}
+
+func (c *CSRDriver) SetCSRControl(control CSRControl, clusterName string) error {
+	c.csrControl = control
+	err := control.Informer().AddIndexers(cache.Indexers{
 		indexByCluster: indexByClusterFunc,
 	})
 	if err != nil {
-		return nil, err
+		return err
 	}
 
-	err = csrControl.Informer().AddIndexers(cache.Indexers{
+	err = control.Informer().AddIndexers(cache.Indexers{
 		indexByAddon: indexByAddonFunc,
 	})
 	if err != nil {
-		utilruntime.HandleError(err)
+		return err
 	}
-
-	c.csrControl = csrControl
-	c.haltCSRCreation = haltCSRCreationFunc(csrControl.Informer().GetIndexer(), secretOption.ClusterName)
-	return clients, nil
+	c.haltCSRCreation = haltCSRCreationFunc(control.Informer().GetIndexer(), clusterName)
+	return nil
 }
 
 var _ register.RegisterDriver = &CSRDriver{}
@@ -424,7 +430,7 @@ func shouldCreateCSR(
 	additionalSecretData map[string][]byte) (bool, error) {
 	// create a csr to request new client certificate if
 	// a.there is no valid client certificate issued for the current cluster/agent
-	valid, err := isCertificateValid(logger, secret.Data[TLSCertFile], subject)
+	valid, err := IsCertificateValid(logger, secret.Data[TLSCertFile], subject)
 	if err != nil {
 		recorder.Eventf("CertificateValidationFailed", "Failed to validate client certificate for %s: %v", controllerName, err)
 		return true, nil

pkg/registration/register/csr/csr_test.go

@@ -75,7 +75,7 @@ func TestProcess(t *testing.T) {
 			validateActions: func(t *testing.T, hubActions []clienttesting.Action, secret *corev1.Secret) {
 				logger, _ := ktesting.NewTestContext(t)
 				testingcommon.AssertActions(t, hubActions, "get", "get")
-				valid, err := isCertificateValid(logger, secret.Data[TLSCertFile], testSubject)
+				valid, err := IsCertificateValid(logger, secret.Data[TLSCertFile], testSubject)
 				if err != nil {
 					t.Errorf("unexpected error: %v", err)
 				}

pkg/registration/register/factory/options.go

@@ -7,18 +7,21 @@ import (
 	"open-cluster-management.io/ocm/pkg/registration/register"
 	awsirsa "open-cluster-management.io/ocm/pkg/registration/register/aws_irsa"
 	"open-cluster-management.io/ocm/pkg/registration/register/csr"
+	"open-cluster-management.io/ocm/pkg/registration/register/grpc"
 )
 
 type Options struct {
 	RegistrationAuth string
 	CSROption        *csr.Option
 	AWSISRAOption    *awsirsa.AWSOption
+	GRPCOption       *grpc.Option
 }
 
 func NewOptions() *Options {
 	return &Options{
 		CSROption:     csr.NewCSROption(),
 		AWSISRAOption: awsirsa.NewAWSOption(),
+		GRPCOption:    grpc.NewOptions(),
 	}
 }
 
@@ -27,12 +30,15 @@ func (s *Options) AddFlags(fs *pflag.FlagSet) {
 		"The type of authentication to use to authenticate with hub.")
 	s.CSROption.AddFlags(fs)
 	s.AWSISRAOption.AddFlags(fs)
+	s.GRPCOption.AddFlags(fs)
 }
 
 func (s *Options) Validate() error {
 	switch s.RegistrationAuth {
 	case helpers.AwsIrsaAuthType:
 		return s.AWSISRAOption.Validate()
+	case "grpc":
+		return s.GRPCOption.Validate()
 	default:
 		return s.CSROption.Validate()
 	}
@@ -42,6 +48,8 @@ func (s *Options) Driver(secretOption register.SecretOption) register.RegisterDr
 	switch s.RegistrationAuth {
 	case helpers.AwsIrsaAuthType:
 		return awsirsa.NewAWSIRSADriver(s.AWSISRAOption, secretOption)
+	case "grpc":
+		return grpc.NewGRPCDriver(s.GRPCOption, s.CSROption, secretOption)
 	default:
 		return csr.NewCSRDriver(s.CSROption, secretOption)
 	}