grpc server

Signed-off-by: Wei Liu <[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.63.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.11+incompatible
 	github.com/ghodss/yaml v1.0.0
@@ -25,6 +26,7 @@ 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.38.0
 	gopkg.in/yaml.v2 v2.4.0
 	helm.sh/helm/v3 v3.17.3
 	k8s.io/api v0.32.4
@@ -39,7 +41,7 @@ require (
 	k8s.io/utils v0.0.0-20241104100929-3ea5e8cea738
 	open-cluster-management.io/addon-framework v1.0.0
 	open-cluster-management.io/api v1.0.0
-	open-cluster-management.io/sdk-go v1.0.0
+	open-cluster-management.io/sdk-go v1.0.1-0.20250625021536-7eeddd26d3e2
 	sigs.k8s.io/about-api v0.0.0-20250131010323-518069c31c03
 	sigs.k8s.io/cluster-inventory-api v0.0.0-20240730014211-ef0154379848
 	sigs.k8s.io/controller-runtime v0.20.2
@@ -76,7 +78,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
@@ -155,7 +156,6 @@ require (
 	go.uber.org/zap v1.27.0 // indirect
 	golang.org/x/crypto v0.37.0 // indirect
 	golang.org/x/exp v0.0.0-20241217172543-b2144cdd0a67 // indirect
-	golang.org/x/net v0.38.0 // indirect
 	golang.org/x/oauth2 v0.28.0 // indirect
 	golang.org/x/sync v0.13.0 // indirect
 	golang.org/x/sys v0.32.0 // indirect

go.sum

@@ -493,8 +493,8 @@ open-cluster-management.io/addon-framework v1.0.0 h1:ejTk4hPAJnwCSxQhY/tVDPg3SeH
 open-cluster-management.io/addon-framework v1.0.0/go.mod h1:Gw9zRGvuNJJ3XhTYanIuA7FFFw0EjtoE74l5OBZCZf8=
 open-cluster-management.io/api v1.0.0 h1:54QllH9DTudCk6VrGt0q8CDsE3MghqJeTaTN4RHZpE0=
 open-cluster-management.io/api v1.0.0/go.mod h1:/OeqXycNBZQoe3WG6ghuWsMgsKGuMZrK8ZpsU6gWL0Y=
-open-cluster-management.io/sdk-go v1.0.0 h1:pIiAHM/hzV3rEw3LSMgZuAUiNgkBrn8hLxFvJM5frw0=
-open-cluster-management.io/sdk-go v1.0.0/go.mod h1:vkLwIDN9W+WBlrHgHxMl5ZoHRT+H5qOq3cXAk7U5AJc=
+open-cluster-management.io/sdk-go v1.0.1-0.20250625021536-7eeddd26d3e2 h1:IIu3Udg7Y4SCpy2Rzk1CvQguyWgI5L6jbjh/NVjWikU=
+open-cluster-management.io/sdk-go v1.0.1-0.20250625021536-7eeddd26d3e2/go.mod h1:vkLwIDN9W+WBlrHgHxMl5ZoHRT+H5qOq3cXAk7U5AJc=
 sigs.k8s.io/about-api v0.0.0-20250131010323-518069c31c03 h1:1ShFiMjGQOR/8jTBkmZrk1gORxnvMwm1nOy2/DbHg4U=
 sigs.k8s.io/about-api v0.0.0-20250131010323-518069c31c03/go.mod h1:F1pT4mK53U6F16/zuaPSYpBaR7x5Kjym6aKJJC0/DHU=
 sigs.k8s.io/apiserver-network-proxy/konnectivity-client v0.31.0 h1:CPT0ExVicCzcpeN4baWEV2ko2Z/AsiZgEdwgcfwLgMo=

pkg/registration/hub/manager.go

@@ -46,6 +46,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
@@ -59,6 +60,8 @@ type HubManagerOptions struct {
 	AutoApprovedARNPatterns    []string
 	AwsResourceTags            []string
 	Labels                     string
+	GRPCCAFile                 string
+	GRPCCAKeyFile              string
 }
 
 // NewHubManagerOptions returns a HubManagerOptions
@@ -93,6 +96,10 @@ func (m *HubManagerOptions) AddFlags(fs *pflag.FlagSet) {
 	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.Labels, "labels", m.Labels,
 		"Labels to be added to the resources created by registration controller. The format is key1=value1,key2=value2.")
+	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)
 }
 
@@ -195,6 +202,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/hub/manifests/rbac/managedcluster-registration-clusterrole.yaml

@@ -16,7 +16,7 @@ rules:
   # TODO for backward compatible, we do not limit the resource name
   # remove this after we no longer support lower versions kubernetes (less than 1.14)
   #resourceNames: ["managed-cluster-lease"]
-  verbs: ["get", "update"]
+  verbs: ["get", "list", "watch", "update"]
 # Allow agent to get/list/watch managed cluster addons
 - apiGroups: ["addon.open-cluster-management.io"]
   resources: ["managedclusteraddons"]
@@ -28,4 +28,4 @@ rules:
 # Allow agent to send events to the hub
 - apiGroups: ["events.k8s.io"]
   resources: ["events"]
-  verbs: ["create"]
+  verbs: ["list", "watch", "create"]

pkg/registration/hub/manifests/rbac/managedcluster-work-clusterrole.yaml

@@ -13,7 +13,7 @@ rules:
 # Allow work agent to send event to hub
 - apiGroups: ["", "events.k8s.io"]
   resources: ["events"]
-  verbs: ["create", "patch", "update"]
+  verbs: ["list", "watch", "create", "patch", "update"]
 # Allow work agent to get/list/watch/update manifestworks
 - apiGroups: ["work.open-cluster-management.io"]
   resources: ["manifestworks"]

pkg/registration/register/csr/csr.go

@@ -9,17 +9,16 @@ import (
 	"math/rand"
 	"os"
 	"path"
-	"reflect"
 	"strings"
 	"time"
 
 	"github.com/openshift/library-go/pkg/controller/factory"
 	"github.com/openshift/library-go/pkg/operator/events"
 	certificates "k8s.io/api/certificates/v1"
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/equality"
 	"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"
@@ -81,6 +80,7 @@ func (c *CSRDriver) Process(
 	recorder events.Recorder) (*corev1.Secret, *metav1.Condition, error) {
 	logger := klog.FromContext(ctx)
 
+	logger.Info("existing csr name", "csr", c.csrName)
 	// reconcile pending csr if exists
 	if len(c.csrName) > 0 {
 		// build a secret data map if the csr is approved
@@ -233,6 +233,8 @@ func (c *CSRDriver) Process(
 		}, err
 	}
 
+	logger.Info("set csr name to", "csr", createdCSRName)
+
 	c.keyData = keyData
 	c.csrName = createdCSRName
 	return nil, nil, nil
@@ -296,6 +298,9 @@ func (c *CSRDriver) ManagedClusterDecorator(cluster *clusterv1.ManagedCluster) *
 }
 
 func (c *CSRDriver) Fork(addonName string, secretOption register.SecretOption) register.RegisterDriver {
+	if len(secretOption.Signer) == 0 {
+		secretOption.Signer = certificates.KubeAPIServerClientSignerName
+	}
 	csrOption := &CSROption{
 		ObjectMeta: metav1.ObjectMeta{
 			GenerateName: fmt.Sprintf("addon-%s-%s-", secretOption.ClusterName, addonName),
@@ -349,23 +354,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{}
@@ -447,7 +459,7 @@ func shouldCreateCSR(
 
 	// b.client certificate is sensitive to the additional secret data and the data changes
 	if err := hasAdditionalSecretData(additionalSecretData, secret); err != nil {
-		recorder.Eventf("AdditonalSecretDataChanged", "The additional secret data is changed for %v. Re-create the client certificate for %s", err, controllerName)
+		recorder.Eventf("AdditionalSecretDataChanged", "The additional secret data is changed for %v. Re-create the client certificate for %s", err, controllerName)
 		return true, nil
 	}
 
@@ -481,7 +493,7 @@ func hasAdditionalSecretData(additionalSecretData map[string][]byte, secret *cor
 			return fmt.Errorf("key %q not found in secret %q", k, secret.Namespace+"/"+secret.Name)
 		}
 
-		if !reflect.DeepEqual(v, value) {
+		if !equality.Semantic.DeepEqual(v, value) {
 			return fmt.Errorf("key %q in secret %q does not match the expected value",
 				k, secret.Namespace+"/"+secret.Name)
 		}

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.RegisterD
 	switch s.RegistrationAuth {
 	case helpers.AwsIrsaAuthType:
 		return awsirsa.NewAWSIRSADriver(s.AWSISRAOption, secretOption), nil
+	case "grpc":
+		return grpc.NewGRPCDriver(s.GRPCOption, s.CSROption, secretOption)
 	default:
 		return csr.NewCSRDriver(s.CSROption, secretOption)
 	}