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 new file mode 100644 index 0000000..e41b3c8 --- /dev/null +++ b/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) +} 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)