feat: decoupling apiserver and clustersynchro-manager by message midd…

…leware

Signed-off-by: zhouhaoA1 <[email protected]>

cmd/apiserver/app/options/options.go

@@ -20,6 +20,9 @@ import (
 	"github.com/clusterpedia-io/clusterpedia/pkg/apiserver"
 	"github.com/clusterpedia-io/clusterpedia/pkg/storage"
 	storageoptions "github.com/clusterpedia-io/clusterpedia/pkg/storage/options"
+	"github.com/clusterpedia-io/clusterpedia/pkg/watcher"
+	watchcomponents "github.com/clusterpedia-io/clusterpedia/pkg/watcher/components"
+	watchoptions "github.com/clusterpedia-io/clusterpedia/pkg/watcher/options"
 )
 
 type ClusterPediaServerOptions struct {
@@ -38,6 +41,8 @@ type ClusterPediaServerOptions struct {
 	//      Traces         *genericoptions.TracingOptions
 
 	Storage *storageoptions.StorageOptions
+
+	Subscriber *watchoptions.MiddlerwareOptions
 }
 
 func NewServerOptions() *ClusterPediaServerOptions {
@@ -64,7 +69,8 @@ func NewServerOptions() *ClusterPediaServerOptions {
 		Admission:      genericoptions.NewAdmissionOptions(),
 		//      Traces:         genericoptions.NewTracingOptions(),
 
-		Storage: storageoptions.NewStorageOptions(),
+		Storage:    storageoptions.NewStorageOptions(),
+		Subscriber: watchoptions.NewMiddlerwareOptions(),
 	}
 }
 
@@ -109,10 +115,19 @@ func (o *ClusterPediaServerOptions) Config() (*apiserver.Config, error) {
 		return nil, err
 	}
 
-	return &apiserver.Config{
+	config := &apiserver.Config{
 		GenericConfig:  genericConfig,
 		StorageFactory: storage,
-	}, nil
+	}
+
+	err = watcher.NewSubscriber(o.Subscriber)
+	watchcomponents.InitEventCacheSize(o.Subscriber.CacheSize)
+
+	if err != nil {
+		return nil, err
+	}
+
+	return config, nil
 }
 
 func (o *ClusterPediaServerOptions) genericOptionsApplyTo(config *genericapiserver.RecommendedConfig) error {
@@ -153,6 +168,8 @@ func (o *ClusterPediaServerOptions) Flags() cliflag.NamedFlagSets {
 	genericfs.IntVar(&o.MaxMutatingRequestsInFlight, "max-mutating-requests-inflight", o.MaxMutatingRequestsInFlight, ""+
 		"this flag limits the maximum number of mutating requests in flight, or a zero value disables the limit completely.")
 
+	//options.BindLeaderElectionFlags(&o.LeaderElection, genericfs)
+
 	o.CoreAPI.AddFlags(fss.FlagSet("global"))
 	o.SecureServing.AddFlags(fss.FlagSet("secure serving"))
 	o.Authentication.AddFlags(fss.FlagSet("authentication"))
@@ -165,6 +182,7 @@ func (o *ClusterPediaServerOptions) Flags() cliflag.NamedFlagSets {
 	// o.Traces.AddFlags(fss.FlagSet("traces"))
 
 	o.Storage.AddFlags(fss.FlagSet("storage"))
+	o.Subscriber.AddFlags(fss.FlagSet("middleware"))
 	return fss
 }
 

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

@@ -26,6 +26,8 @@ import (
 	"github.com/clusterpedia-io/clusterpedia/pkg/metrics"
 	"github.com/clusterpedia-io/clusterpedia/pkg/storage"
 	storageoptions "github.com/clusterpedia-io/clusterpedia/pkg/storage/options"
+	"github.com/clusterpedia-io/clusterpedia/pkg/watcher"
+	watchoptions "github.com/clusterpedia-io/clusterpedia/pkg/watcher/options"
 )
 
 const (
@@ -45,6 +47,7 @@ type Options struct {
 	KubeStateMetrics *kubestatemetrics.Options
 
 	WorkerNumber int // WorkerNumber is the number of worker goroutines
+	Publisher    *watchoptions.MiddlerwareOptions
 }
 
 func NewClusterSynchroManagerOptions() (*Options, error) {
@@ -77,6 +80,7 @@ func NewClusterSynchroManagerOptions() (*Options, error) {
 	options.KubeStateMetrics = kubestatemetrics.NewOptions()
 
 	options.WorkerNumber = 5
+	options.Publisher = watchoptions.NewMiddlerwareOptions()
 	return &options, nil
 }
 
@@ -96,10 +100,10 @@ func (o *Options) Flags() cliflag.NamedFlagSets {
 	fs.StringVar(&o.Kubeconfig, "kubeconfig", o.Kubeconfig, "Path to kubeconfig file with authorization and master location information.")
 
 	logsapi.AddFlags(o.Logs, fss.FlagSet("logs"))
-
 	o.Storage.AddFlags(fss.FlagSet("storage"))
 	o.Metrics.AddFlags(fss.FlagSet("metrics server"))
 	o.KubeStateMetrics.AddFlags(fss.FlagSet("kube state metrics"))
+	o.Publisher.AddFlags(fss.FlagSet("middleware"))
 	return fss
 }
 
@@ -125,6 +129,11 @@ func (o *Options) Config() (*config.Config, error) {
 		return nil, err
 	}
 
+	err = watcher.NewPulisher(o.Publisher)
+	if err != nil {
+		return nil, err
+	}
+
 	kubeconfig, err := clientcmd.BuildConfigFromFlags(o.Master, o.Kubeconfig)
 	if err != nil {
 		return nil, err

deploy/clusterpedia_apiserver_deployment.yaml

@@ -1,59 +1,75 @@
-apiVersion: v1
-kind: ServiceAccount
-metadata:
-  name: clusterpedia-apiserver
-  namespace: clusterpedia-system
----
-apiVersion: v1
-kind: Service
-metadata:
-  name: clusterpedia-apiserver
-  namespace: clusterpedia-system
-spec:
-  ports:
-  - port: 443
-    protocol: TCP
-    targetPort: 443
-  selector:
-    app: clusterpedia-apiserver
----
-apiVersion: apps/v1
-kind: Deployment
-metadata:
-  name: clusterpedia-apiserver
-  namespace: clusterpedia-system
-  labels:
-    app: clusterpedia-apiserver
-spec:
-  replicas: 1
-  selector:
-    matchLabels:
-      app: clusterpedia-apiserver
-  template:
-    metadata:
-      labels:
-        app: clusterpedia-apiserver
-    spec:
-      containers:
-      - name: apiserver
-        image: ghcr.io/clusterpedia-io/clusterpedia/apiserver:v0.7.0
-        command:
-        - /usr/local/bin/apiserver
-        - --secure-port=443
-        - --storage-config=/etc/clusterpedia/storage/internalstorage-config.yaml
-        - -v=3
-        env:
-        - name: DB_PASSWORD
-          valueFrom:
-            secretKeyRef:
-              name: internalstorage-password
-              key: password
-        volumeMounts:
-        - name: internalstorage-config
-          mountPath: /etc/clusterpedia/storage
-          readOnly: true
-      serviceAccountName: clusterpedia-apiserver
-      volumes:
-      - name: internalstorage-config
-        configMap:
-          name: clusterpedia-internalstorage
+apiVersion: v1
+kind: ServiceAccount
+metadata:
+  name: clusterpedia-apiserver
+  namespace: clusterpedia-system
+---
+apiVersion: v1
+kind: Service
+metadata:
+  name: clusterpedia-apiserver
+  namespace: clusterpedia-system
+spec:
+  ports:
+  - port: 6443
+    protocol: TCP
+    targetPort: 6443
+  selector:
+    app: clusterpedia-apiserver
+---
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: clusterpedia-apiserver
+  namespace: clusterpedia-system
+  labels:
+    app: clusterpedia-apiserver
+spec:
+  replicas: 3
+  selector:
+    matchLabels:
+      app: clusterpedia-apiserver
+  template:
+    metadata:
+      labels:
+        app: clusterpedia-apiserver
+    spec:
+      containers:
+      - name: apiserver
+        image: ghcr.io/clusterpedia-io/clusterpedia/apiserver:v0.7.0
+        command:
+        - /usr/local/bin/apiserver
+        - --secure-port=6443
+        - --storage-config=/etc/clusterpedia/storage/internalstorage-config.yaml
+        - -v=3
+        - --middleware-name=rabbitmq
+        - --middleware-serverIp=rabbitmq-cluster.rabbitmq-system.svc
+        - --middleware-user=$(RABBITMQ_USER)
+        - --middleware-serverPort=5672
+        - --middleware-password=$(RABBITMQ_PASSWD)
+        - --cache-size=200
+        env:
+        - name: DB_PASSWORD
+          valueFrom:
+            secretKeyRef:
+              name: internalstorage-password
+              key: password
+        - name: RABBITMQ_USER
+          valueFrom:
+            secretKeyRef:
+              name: rabbitmq-connect
+              key: user
+        - name: RABBITMQ_PASSWD
+          valueFrom:
+            secretKeyRef:
+              name: rabbitmq-connect
+              key: password
+        volumeMounts:
+        - name: internalstorage-config
+          mountPath: /etc/clusterpedia/storage
+          readOnly: true
+      serviceAccountName: clusterpedia-apiserver
+      volumes:
+      - name: internalstorage-config
+        configMap:
+          name: clusterpedia-internalstorage

deploy/clusterpedia_clustersynchro_manager_configmap.yaml

@@ -0,0 +1,7 @@
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: clustersynchro-manager-kubeconfig
+  namespace: clusterpedia-system
+data:
+  manager-kubeconfig: |

deploy/clusterpedia_clustersynchro_manager_deployment.yaml

@@ -1,45 +1,67 @@
-apiVersion: v1
-kind: ServiceAccount
-metadata:
-  name: clusterpedia-clustersynchro-manager
-  namespace: clusterpedia-system
----
-apiVersion: apps/v1
-kind: Deployment
-metadata:
-  name: clusterpedia-clustersynchro-manager
-  namespace: clusterpedia-system
-  labels:
-    app: clusterpedia-clustersynchro-manager
-spec:
-  replicas: 1
-  selector:
-    matchLabels:
-      app: clusterpedia-clustersynchro-manager
-  template:
-    metadata:
-      labels:
-        app: clusterpedia-clustersynchro-manager
-    spec:
-      containers:
-      - name: manager
-        image: ghcr.io/clusterpedia-io/clusterpedia/clustersynchro-manager:v0.7.0
-        command:
-        - /usr/local/bin/clustersynchro-manager
-        - --storage-config=/etc/clusterpedia/storage/internalstorage-config.yaml
-        - --feature-gates=PruneManagedFields=true,PruneLastAppliedConfiguration=true
-        env:
-        - name: DB_PASSWORD
-          valueFrom:
-            secretKeyRef:
-              name: internalstorage-password
-              key: password
-        volumeMounts:
-        - name: internalstorage-config
-          mountPath: /etc/clusterpedia/storage
-          readOnly: true
-      serviceAccountName: clusterpedia-clustersynchro-manager
-      volumes:
-      - name: internalstorage-config
-        configMap:
-          name: clusterpedia-internalstorage
+apiVersion: v1
+kind: ServiceAccount
+metadata:
+  name: clusterpedia-clustersynchro-manager
+  namespace: clusterpedia-system
+---
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: clusterpedia-clustersynchro-manager
+  namespace: clusterpedia-system
+  labels:
+    app: clusterpedia-clustersynchro-manager
+spec:
+  replicas: 3
+  selector:
+    matchLabels:
+      app: clusterpedia-clustersynchro-manager
+  template:
+    metadata:
+      labels:
+        app: clusterpedia-clustersynchro-manager
+    spec:
+      containers:
+      - name: manager
+        image: ghcr.io/clusterpedia-io/clusterpedia/clustersynchro-manager:v0.7.0
+        command:
+        - /usr/local/bin/clustersynchro-manager
+        - --storage-config=/etc/clusterpedia/storage/internalstorage-config.yaml
+        - --feature-gates=PruneManagedFields=true,PruneLastAppliedConfiguration=true
+        - --kubeconfig=/etc/clusterpedia/manager-kubeconfig
+        - --middleware-name=rabbitmq
+        - --middleware-serverIp=rabbitmq-cluster.rabbitmq-system.svc
+        - --middleware-user=$(RABBITMQ_USER)
+        - --middleware-serverPort=5672
+        - --middleware-password=$(RABBITMQ_PASSWD)
+        env:
+        - name: DB_PASSWORD
+          valueFrom:
+            secretKeyRef:
+              name: internalstorage-password
+              key: password
+        - name: RABBITMQ_USER
+          valueFrom:
+            secretKeyRef:
+              name: rabbitmq-connect
+              key: user
+        - name: RABBITMQ_PASSWD
+          valueFrom:
+            secretKeyRef:
+              name: rabbitmq-connect
+              key: password
+        volumeMounts:
+        - name: internalstorage-config
+          mountPath: /etc/clusterpedia/storage
+          readOnly: true
+        - name: manager-kubeconfig
+          mountPath: /etc/clusterpedia/
+          readOnly: true
+      serviceAccountName: clusterpedia-clustersynchro-manager
+      volumes:
+      - name: internalstorage-config
+        configMap:
+          name: clusterpedia-internalstorage
+      - name: manager-kubeconfig
+        configMap:
+          name: clustersynchro-manager-kubeconfig

deploy/clusterpedia_rabbitmq_secret.yaml

@@ -0,0 +1,8 @@
+apiVersion: v1
+kind: Secret
+metadata:
+  name: rabbitmq-connect
+  namespace: clusterpedia-system
+data:
+  user: YWRtaW4=
+  password: cyZQQDg3OTk4