support referencing cluster authentication information from secrets

Signed-off-by: Iceber Gu <[email protected]>

Author: Iceber

cmd/apiserver/app/options/options.go

@@ -26,6 +26,8 @@ import (
 	storageoptions "github.com/clusterpedia-io/clusterpedia/pkg/storage/options"
 )
 
+const DefaultNamespace = "clusterpedia-system"
+
 type ClusterPediaServerOptions struct {
 	MaxRequestsInFlight         int
 	MaxMutatingRequestsInFlight int
@@ -41,6 +43,7 @@ type ClusterPediaServerOptions struct {
 	Traces         *genericoptions.TracingOptions
 	Metrics        *metrics.Options
 
+	RunInNamespace string
 	Storage        *storageoptions.StorageOptions
 	ResourceServer *kubeapiserver.Options
 }
@@ -72,6 +75,7 @@ func NewServerOptions() *ClusterPediaServerOptions {
 		Traces:         genericoptions.NewTracingOptions(),
 		Metrics:        metrics.NewOptions(),
 
+		RunInNamespace: DefaultNamespace,
 		Storage:        storageoptions.NewStorageOptions(),
 		ResourceServer: kubeapiserver.NewOptions(),
 	}
@@ -101,6 +105,7 @@ func (o *ClusterPediaServerOptions) Config() (*apiserver.Config, error) {
 	if err != nil {
 		return nil, err
 	}
+	resourceServerConfig.SecretNamespace = o.RunInNamespace
 
 	if err := o.SecureServing.MaybeDefaultWithSelfSignedCerts("localhost", nil, []net.IP{net.ParseIP("127.0.0.1")}); err != nil {
 		return nil, fmt.Errorf("error create self-signed certificates: %v", err)
@@ -171,6 +176,7 @@ func (o *ClusterPediaServerOptions) Flags() cliflag.NamedFlagSets {
 	var fss cliflag.NamedFlagSets
 
 	genericfs := fss.FlagSet("generic")
+	genericfs.StringVar(&o.RunInNamespace, "namespace", o.RunInNamespace, "The namespace in which the Pod is running.")
 	genericfs.IntVar(&o.MaxRequestsInFlight, "max-requests-inflight", o.MaxRequestsInFlight, ""+
 		"Otherwise, this flag limits the maximum number of non-mutating requests in flight, or a zero value disables the limit completely.")
 	genericfs.IntVar(&o.MaxMutatingRequestsInFlight, "max-mutating-requests-inflight", o.MaxMutatingRequestsInFlight, ""+

cmd/clustersynchro-manager/app/config/config.go

@@ -14,6 +14,7 @@ import (
 
 type Config struct {
 	Kubeconfig    *restclient.Config
+	Namespace     string
 	CRDClient     *crdclientset.Clientset
 	EventRecorder record.EventRecorder
 

cmd/clustersynchro-manager/app/options/options.go

@@ -31,6 +31,8 @@ import (
 
 const (
 	ClusterSynchroManagerUserAgent = "cluster-synchro-manager"
+
+	DefaultNamespace = "clusterpedia-system"
 )
 
 type Options struct {
@@ -45,6 +47,7 @@ type Options struct {
 	Metrics          *MetricsOptions
 	KubeStateMetrics *kubestatemetrics.Options
 
+	RunInNamespace          string
 	WorkerNumber            int // WorkerNumber is the number of worker goroutines
 	PageSizeForResourceSync int64
 	ShardingName            string
@@ -59,7 +62,7 @@ func NewClusterSynchroManagerOptions() (*Options, error) {
 	componentbaseconfigv1alpha1.RecommendedDefaultClientConnectionConfiguration(&clientConnection)
 
 	leaderElection.ResourceName = "clusterpedia-clustersynchro-manager"
-	leaderElection.ResourceNamespace = "clusterpedia-system"
+	leaderElection.ResourceNamespace = DefaultNamespace
 	leaderElection.ResourceLock = resourcelock.LeasesResourceLock
 
 	clientConnection.ContentType = runtime.ContentTypeJSON
@@ -78,6 +81,7 @@ func NewClusterSynchroManagerOptions() (*Options, error) {
 	options.Storage = storageoptions.NewStorageOptions()
 	options.Metrics = NewMetricsOptions()
 	options.KubeStateMetrics = kubestatemetrics.NewOptions()
+	options.RunInNamespace = DefaultNamespace
 
 	options.WorkerNumber = 5
 	return &options, nil
@@ -92,6 +96,7 @@ func (o *Options) Flags() cliflag.NamedFlagSets {
 	genericfs.Int32Var(&o.ClientConnection.Burst, "kube-api-burst", o.ClientConnection.Burst, "Burst to use while talking with kubernetes apiserver.")
 	genericfs.IntVar(&o.WorkerNumber, "worker-number", o.WorkerNumber, "The number of worker goroutines.")
 	genericfs.StringVar(&o.ShardingName, "sharding-name", o.ShardingName, "The sharding name of manager.")
+	genericfs.StringVar(&o.RunInNamespace, "namespace", o.RunInNamespace, "The namespace in which the Pod is running.")
 
 	syncfs := fss.FlagSet("resource sync")
 	syncfs.Int64Var(&o.PageSizeForResourceSync, "page-size", o.PageSizeForResourceSync, "The requested chunk size of initial and resync watch lists for resource sync")
@@ -169,8 +174,11 @@ func (o *Options) Config() (*config.Config, error) {
 	if o.ShardingName != "" {
 		o.LeaderElection.ResourceName = fmt.Sprintf("%s-%s", o.LeaderElection.ResourceName, o.ShardingName)
 	}
+	// Override the namespace for leader election resource.
+	o.LeaderElection.ResourceNamespace = o.RunInNamespace
 
 	return &config.Config{
+		Namespace:     o.RunInNamespace,
 		CRDClient:     crdclient,
 		Kubeconfig:    kubeconfig,
 		EventRecorder: eventRecorder,

pkg/kubeapiserver/apiserver.go

@@ -66,6 +66,7 @@ func NewDefaultConfig() *Config {
 }
 
 type ExtraConfig struct {
+	SecretNamespace                 string
 	AllowPediaClusterConfigReuse    bool
 	ExtraProxyRequestHeaderPrefixes []string
 	AllowedProxySubresources        map[schema.GroupResource]sets.Set[string]
@@ -103,6 +104,7 @@ func (c *Config) Complete() CompletedConfig {
 type completedConfig struct {
 	GenericConfig genericapiserver.CompletedConfig
 
+	SecretNamespace          string
 	StorageFactory           storage.StorageFactory
 	InformerFactory          informers.SharedInformerFactory
 	InitialAPIGroupResources []*restmapper.APIGroupResources
@@ -138,8 +140,9 @@ func (c completedConfig) New(delegationTarget genericapiserver.DelegationTarget)
 	restManager := NewRESTManager(c.GenericConfig.Serializer, runtime.ContentTypeJSON, c.StorageFactory, c.InitialAPIGroupResources)
 	discoveryManager := discovery.NewDiscoveryManager(c.GenericConfig.Serializer, restManager, delegate)
 
+	secretLister := c.GenericConfig.SharedInformerFactory.Core().V1().Secrets().Lister().Secrets(c.ExtraConfig.SecretNamespace)
 	clusterInformer := c.InformerFactory.Cluster().V1alpha2().PediaClusters()
-	connector := proxyrest.NewProxyConnector(clusterInformer.Lister(), c.ExtraConfig.AllowPediaClusterConfigReuse, c.ExtraConfig.ExtraProxyRequestHeaderPrefixes)
+	connector := proxyrest.NewProxyConnector(clusterInformer.Lister(), secretLister, c.ExtraConfig.AllowPediaClusterConfigReuse, c.ExtraConfig.ExtraProxyRequestHeaderPrefixes)
 
 	methodSet := sets.New("GET")
 	for _, rest := range proxyrest.GetSubresourceRESTs(connector) {

pkg/kubeapiserver/options.go

@@ -12,6 +12,8 @@ import (
 )
 
 type Options struct {
+	ClusterCertificationSecretNamespace string
+
 	// AllowPediaClusterConfigForProxyRequest controls all proxy requests.
 	// TODO(iceber): Perhaps we could add a separate setting specifically for subresource proxy request.
 	AllowPediaClusterConfigForProxyRequest bool

pkg/kubeapiserver/resourcerest/proxy/proxy_connector.go

@@ -12,6 +12,7 @@ import (
 
 	clusterlister "github.com/clusterpedia-io/clusterpedia/pkg/generated/listers/cluster/v1alpha2"
 	"github.com/clusterpedia-io/clusterpedia/pkg/utils"
+	corev1listers "k8s.io/client-go/listers/core/v1"
 )
 
 const DefaultProxyRequestHeaderPrefix = "X-Clusterpedia-Proxy-"
@@ -20,16 +21,18 @@ type Connector struct {
 	allowConfigReuse    bool
 	extraHeaderPrefixes []string
 	clusterLister       clusterlister.PediaClusterLister
+	secretLister        corev1listers.SecretNamespaceLister
 }
 
-func NewProxyConnector(clusterLister clusterlister.PediaClusterLister, allowPediaClusterConfigReuse bool, extraHeaderPrefixes []string) ClusterConnectionGetter {
+func NewProxyConnector(clusterLister clusterlister.PediaClusterLister, secretLister corev1listers.SecretNamespaceLister, allowPediaClusterConfigReuse bool, extraHeaderPrefixes []string) ClusterConnectionGetter {
 	if len(extraHeaderPrefixes) == 0 {
 		extraHeaderPrefixes = []string{DefaultProxyRequestHeaderPrefix}
 	}
 	return &Connector{
 		allowConfigReuse:    allowPediaClusterConfigReuse,
 		extraHeaderPrefixes: extraHeaderPrefixes,
 		clusterLister:       clusterLister,
+		secretLister:        secretLister,
 	}
 }
 
@@ -100,7 +103,7 @@ func (c *Connector) GetClusterConnection(ctx context.Context, name string, req *
 	}
 
 	if !authInHeader && c.allowConfigReuse {
-		config, err = utils.BuildClusterRestConfig(cluster)
+		config, err = utils.BuildClusterRestConfig(cluster, c.secretLister)
 		if err != nil {
 			return "", nil, err
 		}

pkg/synchromanager/clustersynchro_manager.go

@@ -9,12 +9,16 @@ import (
 	"sync"
 	"time"
 
+	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/equality"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/util/wait"
+	v1 "k8s.io/client-go/informers/core/v1"
+	"k8s.io/client-go/kubernetes"
+	corev1listers "k8s.io/client-go/listers/core/v1"
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/client-go/util/retry"
 	"k8s.io/client-go/util/workqueue"
@@ -45,6 +49,8 @@ type Manager struct {
 
 	clusterpediaclient crdclientset.Interface
 	informerFactory    externalversions.SharedInformerFactory
+	secretLister       corev1listers.SecretNamespaceLister
+	secretInformer     cache.SharedIndexInformer
 
 	shardingName               string
 	queue                      workqueue.RateLimitingInterface
@@ -57,18 +63,20 @@ type Manager struct {
 	synchrolock       sync.RWMutex
 	synchros          map[string]*clustersynchro.ClusterSynchro
 	synchroWaitGroup  wait.Group
+
+	clusterSecretsMap sync.Map
 }
 
 var _ kubestatemetrics.ClusterMetricsWriterListGetter = &Manager{}
 
-func NewManager(client crdclientset.Interface, storage storage.StorageFactory, syncConfig clustersynchro.ClusterSyncConfig, shardingName string) *Manager {
-	factory := externalversions.NewSharedInformerFactory(client, 0)
+func NewManager(client kubernetes.Interface, clusterpediaClient crdclientset.Interface, storage storage.StorageFactory, syncConfig clustersynchro.ClusterSyncConfig, shardingName string, secretNamespace string) *Manager {
+	factory := externalversions.NewSharedInformerFactory(clusterpediaClient, 0)
 	clusterinformer := factory.Cluster().V1alpha2().PediaClusters()
 	clusterSyncResourcesInformer := factory.Cluster().V1alpha2().ClusterSyncResources()
 
 	manager := &Manager{
 		informerFactory:    factory,
-		clusterpediaclient: client,
+		clusterpediaclient: clusterpediaClient,
 		shardingName:       shardingName,
 
 		storage:                    storage,
@@ -83,6 +91,23 @@ func NewManager(client crdclientset.Interface, storage storage.StorageFactory, s
 		synchros:          make(map[string]*clustersynchro.ClusterSynchro),
 	}
 
+	secretInformer := v1.NewSecretInformer(client, secretNamespace, 0, nil)
+	if _, err := secretInformer.AddEventHandler(
+		cache.ResourceEventHandlerFuncs{
+			AddFunc:    func(obj any) { manager.handleSecret(nil, obj.(*corev1.Secret)) },
+			UpdateFunc: func(older, newer any) { manager.handleSecret(older.(*corev1.Secret), newer.(*corev1.Secret)) },
+			DeleteFunc: func(obj any) {
+				objName, err := cache.DeletionHandlingObjectToName(obj)
+				if err != nil {
+					return
+				}
+				manager.handleDeletedSecret(objName.Name)
+			},
+		},
+	); err != nil {
+		klog.ErrorS(err, "error when adding event handler to informer")
+	}
+
 	if _, err := clusterinformer.Informer().AddEventHandler(
 		cache.ResourceEventHandlerFuncs{
 			AddFunc:    manager.addCluster,
@@ -130,6 +155,21 @@ func (manager *Manager) Run(workers int, stopCh <-chan struct{}) {
 
 	// informerFactory should not be controlled by stopCh
 	stopInformer := make(chan struct{})
+
+	// 优先启动 secret informer
+	manager.secretInformer.Run(stopInformer)
+	timeout := make(chan struct{})
+	go func() {
+		select {
+		case <-stopCh:
+		case <-time.After(60 * time.Second):
+		}
+		close(timeout)
+	}()
+	if !cache.WaitForCacheSync(timeout, manager.secretInformer.HasSynced) {
+		klog.Fatal("clustersynchro manager: wait for secret informer failed")
+	}
+
 	manager.informerFactory.Start(stopInformer)
 	if !cache.WaitForCacheSync(stopCh, manager.clusterInformer.HasSynced) {
 		klog.Fatal("clustersynchro manager: wait for informer factory failed")
@@ -159,6 +199,31 @@ func (manager *Manager) Run(workers int, stopCh <-chan struct{}) {
 	klog.Info("cluster synchro manager stopped.")
 }
 
+func (manager *Manager) handleSecret(old *corev1.Secret, obj *corev1.Secret) {
+	if old != nil && reflect.DeepEqual(old.Data, obj.Data) {
+		return
+	}
+
+	// TODO(Iceber): 通过字段 Key 来优化 cluster 的查找
+	manager.clusterSecretsMap.Range(func(k, v any) bool {
+		secrets := v.(map[string]struct{})
+		if _, ok := secrets[obj.Name]; ok {
+			manager.enqueue(k.(string))
+		}
+		return true
+	})
+}
+
+func (manager *Manager) handleDeletedSecret(name string) {
+	manager.clusterSecretsMap.Range(func(k, v any) bool {
+		secrets := v.(map[string]struct{})
+		if _, ok := secrets[name]; ok {
+			manager.enqueue(k.(string))
+		}
+		return true
+	})
+}
+
 func (manager *Manager) addCluster(obj interface{}) {
 	manager.enqueue(obj)
 }
@@ -328,7 +393,23 @@ func (manager *Manager) reconcileCluster(cluster *clusterv1alpha2.PediaCluster)
 	synchro := manager.synchros[cluster.Name]
 	manager.synchrolock.RUnlock()
 
-	config, err := utils.BuildClusterRestConfig(cluster)
+	// TODO(Iceber): Need to optimize and simplify the following code
+	secrets := make(map[string]struct{})
+	if source := cluster.Spec.CertificationFrom.Cert; source != nil {
+		secrets[source.Name] = struct{}{}
+	}
+	if source := cluster.Spec.CertificationFrom.Token; source != nil {
+		secrets[source.Name] = struct{}{}
+	}
+	if source := cluster.Spec.CertificationFrom.KubeConfig; source != nil {
+		secrets[source.Name] = struct{}{}
+	}
+	if source := cluster.Spec.CertificationFrom.Key; source != nil {
+		secrets[source.Name] = struct{}{}
+	}
+	manager.clusterSecretsMap.Store(cluster.Name, secrets)
+
+	config, err := utils.BuildClusterRestConfig(cluster, manager.secretLister)
 	if err != nil {
 		klog.ErrorS(err, "Failed to build cluster config", "cluster", cluster.Name)
 		manager.UpdateClusterAPIServerAndValidatedCondition(cluster.Name, cluster.Spec.APIServer, synchro, clusterv1alpha2.InvalidConfigReason,
@@ -453,6 +534,8 @@ func (manager *Manager) stopClusterSynchro(name string) {
 }
 
 func (manager *Manager) removeCluster(name string) error {
+	manager.clusterSecretsMap.Delete(name)
+
 	manager.synchrolock.Lock()
 	synchro := manager.synchros[name]
 	delete(manager.synchros, name)