allow master flag to take multiple values for watching configmaps

Author: linki

kube/configmap.go

@@ -16,36 +16,42 @@ import (
 )
 
 type ConfigMapWatcher struct {
-	client    kubernetes.Interface
+	clients   []kubernetes.Interface
 	namespace string
 	selector  fields.Selector
 	configs   chan provider.EgressConfig
 }
 
-func NewConfigMapWatcher(client kubernetes.Interface, namespace, selectorStr string, configs chan provider.EgressConfig) (*ConfigMapWatcher, error) {
+func NewConfigMapWatcher(clients []kubernetes.Interface, namespace, selectorStr string, configs chan provider.EgressConfig) (*ConfigMapWatcher, error) {
 	selector, err := fields.ParseSelector(selectorStr)
 	if err != nil {
 		return nil, err
 	}
 
 	return &ConfigMapWatcher{
-		client:    client,
+		clients:   clients,
 		namespace: namespace,
 		selector:  selector,
 		configs:   configs,
 	}, nil
 }
 
 func (c *ConfigMapWatcher) Run(ctx context.Context) {
+	for _, client := range c.clients {
+		c.runForClient(ctx, client)
+	}
+}
+
+func (c *ConfigMapWatcher) runForClient(ctx context.Context, client kubernetes.Interface) {
 	informer := cache.NewSharedIndexInformer(
 		&cache.ListWatch{
 			ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
 				options.LabelSelector = c.selector.String()
-				return c.client.CoreV1().ConfigMaps(c.namespace).List(ctx, options)
+				return client.CoreV1().ConfigMaps(c.namespace).List(ctx, options)
 			},
 			WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
 				options.LabelSelector = c.selector.String()
-				return c.client.CoreV1().ConfigMaps(c.namespace).Watch(ctx, options)
+				return client.CoreV1().ConfigMaps(c.namespace).Watch(ctx, options)
 			},
 		},
 		&v1.ConfigMap{},
@@ -105,11 +111,23 @@ func (c *ConfigMapWatcher) del(obj interface{}) {
 }
 
 func (c *ConfigMapWatcher) ListConfigs(ctx context.Context) ([]provider.EgressConfig, error) {
+	egressConfigs := []provider.EgressConfig{}
+	for _, client := range c.clients {
+		configs, err := c.listConfigsForClient(ctx, client)
+		if err != nil {
+			return nil, err
+		}
+		egressConfigs = append(egressConfigs, configs...)
+	}
+	return egressConfigs, nil
+}
+
+func (c *ConfigMapWatcher) listConfigsForClient(ctx context.Context, client kubernetes.Interface) ([]provider.EgressConfig, error) {
 	opts := metav1.ListOptions{
 		LabelSelector: c.selector.String(),
 	}
 
-	configMaps, err := c.client.CoreV1().ConfigMaps(c.namespace).List(ctx, opts)
+	configMaps, err := client.CoreV1().ConfigMaps(c.namespace).List(ctx, opts)
 	if err != nil {
 		return nil, err
 	}

main.go

@@ -36,7 +36,7 @@ var (
 )
 
 type Config struct {
-	Master             string
+	Masters            []string
 	KubeConfig         string
 	DryRun             bool
 	LogFormat          string
@@ -62,7 +62,6 @@ type Config struct {
 }
 
 var defaultConfig = &Config{
-	Master:                     "",
 	KubeConfig:                 "",
 	VPCID:                      "",
 	ClusterID:                  "",
@@ -122,7 +121,7 @@ Example:
 	app.DefaultEnvars()
 
 	// Flags related to Kubernetes
-	app.Flag("master", "The Kubernetes API server to connect to (default: auto-detect)").Default(defaultConfig.Master).StringVar(&cfg.Master)
+	app.Flag("master", "The Kubernetes API server to connect to (default: auto-detect)").Default("").StringsVar(&cfg.Masters)
 	app.Flag("kubeconfig", "Retrieve target cluster configuration from a Kubernetes configuration file (default: auto-detect)").Default(defaultConfig.KubeConfig).StringVar(&cfg.KubeConfig)
 	app.Flag("use-platform-credentials", "Use Platform credentials (default: disabled)").BoolVar(&cfg.UsePlatformCredentials)
 	app.Flag("credentials-dir", "Directory where the Platform credentials are stored (default: /meta/credentials)").Default(auth.DefaultCredentialsDir).Envar(auth.CredentialsDirEnvar).StringVar(&cfg.CredentialsDir)
@@ -175,7 +174,7 @@ func main() {
 	}
 
 	configsChan := make(chan provider.EgressConfig)
-	cmWatcher, err := kube.NewConfigMapWatcher(newKubeClient(cfg), cfg.Namespace, "egress=static", configsChan)
+	cmWatcher, err := kube.NewConfigMapWatcher(newKubeClients(cfg), cfg.Namespace, "egress=static", configsChan)
 	if err != nil {
 		log.Fatalf("Failed to setup ConfigMap watcher: %v", err)
 	}
@@ -193,14 +192,23 @@ func main() {
 	controller.Run(ctx)
 }
 
-// newKubeClient returns a new Kubernetes client with the default config.
-func newKubeClient(cfg *Config) kubernetes.Interface {
+// newKubeClients returns multiple Kubernetes clients with the given config.
+func newKubeClients(cfg *Config) []kubernetes.Interface {
 	var kubeconfig string
 	if _, err := os.Stat(clientcmd.RecommendedHomeFile); err == nil {
 		kubeconfig = clientcmd.RecommendedHomeFile
 	}
 	log.Debugf("use config file %s", kubeconfig)
-	config, err := clientcmd.BuildConfigFromFlags(cfg.Master, kubeconfig)
+	var clients []kubernetes.Interface
+	for _, master := range cfg.Masters {
+		clients = append(clients, newKubeClient(cfg, master, kubeconfig))
+	}
+	return clients
+}
+
+// newKubeClient returns a new Kubernetes client with the given config.
+func newKubeClient(cfg *Config, master, kubeconfig string) kubernetes.Interface {
+	config, err := clientcmd.BuildConfigFromFlags(master, kubeconfig)
 	if err != nil {
 		log.Fatalf("build config failed: %v", err)
 	}