From 6392a8a30b4497ff0c48f706b4f97f3a9ac165ce Mon Sep 17 00:00:00 2001 From: harry6396 Date: Thu, 31 Oct 2024 00:35:23 +0530 Subject: [PATCH 1/2] Added configmap watcher --- kubernetes/configmap_watcher.go | 43 +++++++++++++++++++++++++++++++++ 1 file changed, 43 insertions(+) create mode 100644 kubernetes/configmap_watcher.go diff --git a/kubernetes/configmap_watcher.go b/kubernetes/configmap_watcher.go new file mode 100644 index 0000000..9043e1f --- /dev/null +++ b/kubernetes/configmap_watcher.go @@ -0,0 +1,43 @@ +package kubernetes + +import ( + "context" + "fmt" + "log" + "slices" + + "github.com/SwaDeshiTech/kubesync/config" + v1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/watch" + "k8s.io/client-go/kubernetes" +) + +type ConfigMapWatcher struct { + ClientSet *kubernetes.Clientset +} + +func (configMapWatcher *ConfigMapWatcher) Watcher() { + // Watch for changes in namespaces + watcher, err := configMapWatcher.ClientSet.CoreV1().ConfigMaps("").Watch(context.Background(), metav1.ListOptions{}) + if err != nil { + log.Fatal(err) + } + + // Loop through the watch events + for { + event := <-watcher.ResultChan() + switch event.Type { + case watch.Added: + namespaceName := event.Object.(*v1.Namespace).Name + if slices.Contains(config.GetConfig().WhitelistedNamespace, namespaceName) { + fmt.Printf("New namespace created: %s\n", namespaceName) + syncResource := SyncResource{ + DestinationNameSpace: namespaceName, + SourceNameSpace: "kubesync", + } + syncResource.SyncResources() + } + } + } +} From c4158187b734c8c0490fee77cef85d5e0950c70b Mon Sep 17 00:00:00 2001 From: harry6396 Date: Thu, 23 Apr 2026 17:59:39 +0530 Subject: [PATCH 2/2] refactor: restructure syncer configuration and implement configmap change monitoring via broker system --- conf/syncer.yml | 12 ++++++--- config/syncer_config.go | 18 ++++++++++--- kubernetes/configmap_watcher.go | 46 +++++++++++++++++++++++---------- kubernetes/namespace_watcher.go | 2 +- kubernetes/subscriber.go | 13 +++++++++- kubernetes/sync.go | 4 +-- main.go | 7 +++++ 7 files changed, 77 insertions(+), 25 deletions(-) diff --git a/conf/syncer.yml b/conf/syncer.yml index 71e26e4..cbb32fc 100644 --- a/conf/syncer.yml +++ b/conf/syncer.yml @@ -1,7 +1,11 @@ syncer: - name: "syncer1" - sourceNamespace: "kubesync" - destinationNamespace: ["simbha"] - configMapList: ["confs"] - secretList: ["prodhub"] + configMap: + list: ["confs"] + sourceNamespace: "kubesync" + destinationNamespace: ["simbha"] + secret: + list: ["prodhub"] + sourceNamespace: "kubesync" + destinationNamespace: ["simbha"] k8sClusterName: "dev-cluster" \ No newline at end of file diff --git a/config/syncer_config.go b/config/syncer_config.go index 9cbd9cc..5f9af7e 100644 --- a/config/syncer_config.go +++ b/config/syncer_config.go @@ -15,12 +15,22 @@ type SyncerConfig struct { } type Syncer struct { - Name string `yaml:"name"` + Name string `yaml:"name"` + ConfigMap ConfigMap `yaml:"configMap"` + Secret Secret `yaml:"secret"` + K8sClusterName string `yaml:"k8sClusterName"` +} + +type ConfigMap struct { + List []string `yaml:"list"` + SourceNamespace string `yaml:"sourceNamespace"` + DestinationNamespace []string `yaml:"destinationNamespace"` +} + +type Secret struct { + List []string `yaml:"list"` SourceNamespace string `yaml:"sourceNamespace"` DestinationNamespace []string `yaml:"destinationNamespace"` - ConfigMapList []string `yaml:"configMapList"` - SecretList []string `yaml:"secretList"` - K8sClusterName string `yaml:"k8sClusterName"` } func ReadSyncerConfig() error { diff --git a/kubernetes/configmap_watcher.go b/kubernetes/configmap_watcher.go index 9043e1f..e41b3c8 100644 --- a/kubernetes/configmap_watcher.go +++ b/kubernetes/configmap_watcher.go @@ -2,10 +2,9 @@ package kubernetes import ( "context" - "fmt" "log" - "slices" + "github.com/SwaDeshiTech/kubesync/client" "github.com/SwaDeshiTech/kubesync/config" v1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -14,12 +13,15 @@ import ( ) type ConfigMapWatcher struct { - ClientSet *kubernetes.Clientset + ClientSet *kubernetes.Clientset + Namespace string + ConfigMapName string + Broker *Broker } func (configMapWatcher *ConfigMapWatcher) Watcher() { - // Watch for changes in namespaces - watcher, err := configMapWatcher.ClientSet.CoreV1().ConfigMaps("").Watch(context.Background(), metav1.ListOptions{}) + // Watch for changes in configmap + watcher, err := configMapWatcher.ClientSet.CoreV1().ConfigMaps(configMapWatcher.ConfigMapName).Watch(context.Background(), metav1.ListOptions{}) if err != nil { log.Fatal(err) } @@ -29,15 +31,33 @@ func (configMapWatcher *ConfigMapWatcher) Watcher() { event := <-watcher.ResultChan() switch event.Type { case watch.Added: - namespaceName := event.Object.(*v1.Namespace).Name - if slices.Contains(config.GetConfig().WhitelistedNamespace, namespaceName) { - fmt.Printf("New namespace created: %s\n", namespaceName) - syncResource := SyncResource{ - DestinationNameSpace: namespaceName, - SourceNameSpace: "kubesync", - } - syncResource.SyncResources() + case watch.Modified: + configMapName := event.Object.(*v1.ConfigMap).Name + go configMapWatcher.Broker.Publish("configmap", configMapName) + } + } +} + +func SubscribeToConfigMapChange() { + + syncerConfigs := config.GetSyncerConfig().SyncerConfigs + + for _, itr := range syncerConfigs { + for _, configMap := range itr.ConfigMap.List { + configMapWatcher := ConfigMapWatcher{ + ClientSet: client.K8sClientSetMap[itr.K8sClusterName], + Namespace: itr.ConfigMap.SourceNamespace, + ConfigMapName: configMap, } + configMapWatcher.Watcher() } } } + +func SubscribeToConfigMapChannel(broker *Broker, syncResource SyncResource) { + + subscriber := broker.AddSubscriber() + broker.Subscribe(subscriber, "configmap") + + go subscriber.ListenConfigMap(syncResource) +} diff --git a/kubernetes/namespace_watcher.go b/kubernetes/namespace_watcher.go index bd1284e..7ebea87 100644 --- a/kubernetes/namespace_watcher.go +++ b/kubernetes/namespace_watcher.go @@ -38,5 +38,5 @@ func SubscribeToNamespaceChannel(broker *Broker, syncResource SyncResource) { subscriber := broker.AddSubscriber() broker.Subscribe(subscriber, "namespace") - go subscriber.Listen(syncResource) + go subscriber.ListenNamespace(syncResource) } diff --git a/kubernetes/subscriber.go b/kubernetes/subscriber.go index 990540d..53a9183 100644 --- a/kubernetes/subscriber.go +++ b/kubernetes/subscriber.go @@ -73,7 +73,18 @@ func (s *Subscriber) Signal(msg *Message) { } } -func (s *Subscriber) Listen(syncResource SyncResource) { +func (s *Subscriber) ListenNamespace(syncResource SyncResource) { + // Listens to the message channel, prints once received. + for { + if msg, ok := <-s.messages; ok { + fmt.Printf("Subscriber %s, received: %s from topic: %s\n", s.id, msg.GetMessageBody(), msg.GetTopic()) + syncResource.DestinationNameSpace = msg.body + syncResource.SyncResources() + } + } +} + +func (s *Subscriber) ListenConfigMap(syncResource SyncResource) { // Listens to the message channel, prints once received. for { if msg, ok := <-s.messages; ok { diff --git a/kubernetes/sync.go b/kubernetes/sync.go index 5485279..24df0a7 100644 --- a/kubernetes/sync.go +++ b/kubernetes/sync.go @@ -37,7 +37,7 @@ func (syncResource *SyncResource) SyncResources() { log.Printf("----Executing syncer %s syncing resource from namespace %s to %s----", syncResource.SyncerConfig.Name, syncResource.SourceNameSpace, syncResource.DestinationNameSpace) - for _, configMapSyncer := range syncResource.SyncerConfig.ConfigMapList { + for _, configMapSyncer := range syncResource.SyncerConfig.ConfigMap.List { configMapSyncer := SyncK8s{ ClientSet: syncResource.K8sClient, SourceNameSpace: syncResource.SourceNameSpace, @@ -47,7 +47,7 @@ func (syncResource *SyncResource) SyncResources() { configMapSyncer.SyncConfigMap() } - for _, secretSyncer := range syncResource.SyncerConfig.SecretList { + for _, secretSyncer := range syncResource.SyncerConfig.Secret.List { secretSyncer := SyncK8s{ ClientSet: syncResource.K8sClient, SourceNameSpace: syncResource.SourceNameSpace, diff --git a/main.go b/main.go index 20742f4..3f4ffdd 100644 --- a/main.go +++ b/main.go @@ -36,9 +36,11 @@ func main() { cron.InitializeCrons() }() } + // construct new broker. broker := kubernetes.NewBroker() + //namespace watcher go func() { log.Println("----Starting namespace watcher----") @@ -54,6 +56,11 @@ func main() { } }() + //configmap watcher + go func() { + kubernetes.SubscribeToConfigMapChange() + }() + go func() { log.Println("----Subscribing sync resources to watcher----") kubernetes.SubscribeSyncResourcesToWatcher(broker)