Added configmap watcher

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"

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 {

kubernetes/configmap_watcher.go

@@ -0,0 +1,63 @@
+package kubernetes
+
+import (
+	"context"
+	"log"
+
+	"github.com/SwaDeshiTech/kubesync/client"
+	"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
+	Namespace     string
+	ConfigMapName string
+	Broker        *Broker
+}
+
+func (configMapWatcher *ConfigMapWatcher) Watcher() {
+	// Watch for changes in configmap
+	watcher, err := configMapWatcher.ClientSet.CoreV1().ConfigMaps(configMapWatcher.ConfigMapName).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:
+		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)
+}

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)
 }

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 {

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,

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)