Add --k8s-namespace-whitelist setting that specifies namespaces to watch.

cluster/kubernetes/kubernetes.go

@@ -104,11 +104,12 @@ func isAddon(obj k8sObject) bool {
 // Cluster is a handle to a Kubernetes API server.
 // (Typically, this code is deployed into the same cluster.)
 type Cluster struct {
-	client     extendedClient
-	applier    Applier
-	version    string // string response for the version command.
-	logger     log.Logger
-	sshKeyRing ssh.KeyRing
+	client      extendedClient
+	applier     Applier
+	version     string // string response for the version command.
+	logger      log.Logger
+	sshKeyRing  ssh.KeyRing
+	nsWhitelist map[string]bool
 
 	mu sync.Mutex
 }
@@ -118,7 +119,13 @@ func NewCluster(clientset k8sclient.Interface,
 	ifclientset ifclient.Interface,
 	applier Applier,
 	sshKeyRing ssh.KeyRing,
-	logger log.Logger) *Cluster {
+	logger log.Logger,
+	nsWhitelist []string) *Cluster {
+
+	nsWhitelistMap := map[string]bool{}
+	for _, namespace := range nsWhitelist {
+		nsWhitelistMap[namespace] = true
+	}
 
 	c := &Cluster{
 		client: extendedClient{
@@ -132,6 +139,7 @@ func NewCluster(clientset k8sclient.Interface,
 		applier:    applier,
 		logger:     logger,
 		sshKeyRing: sshKeyRing,
+		nsWhitelist: nsWhitelistMap,
 	}
 
 	return c
@@ -167,13 +175,13 @@ func (c *Cluster) SomeControllers(ids []flux.ResourceID) (res []cluster.Controll
 // AllControllers returns all controllers matching the criteria; that is, in
 // the namespace (or any namespace if that argument is empty)
 func (c *Cluster) AllControllers(namespace string) (res []cluster.Controller, err error) {
-	namespaces, err := c.client.Namespaces().List(meta_v1.ListOptions{})
+	namespaces, err := c.getNamespaces()
 	if err != nil {
 		return nil, errors.Wrap(err, "getting namespaces")
 	}
 
 	var allControllers []cluster.Controller
-	for _, ns := range namespaces.Items {
+	for _, ns := range namespaces {
 		if namespace != "" && ns.Name != namespace {
 			continue
 		}
@@ -252,11 +260,13 @@ func (c *Cluster) Ping() error {
 // Export exports cluster resources
 func (c *Cluster) Export() ([]byte, error) {
 	var config bytes.Buffer
-	list, err := c.client.Namespaces().List(meta_v1.ListOptions{})
+
+	namespaces, err := c.getNamespaces()
 	if err != nil {
 		return nil, errors.Wrap(err, "getting namespaces")
 	}
-	for _, ns := range list.Items {
+
+	for _, ns := range namespaces {
 		err := appendYAML(&config, "v1", "Namespace", ns)
 		if err != nil {
 			return nil, errors.Wrap(err, "marshalling namespace to YAML")
@@ -365,13 +375,13 @@ func mergeCredentials(c *Cluster, namespace string, podTemplate apiv1.PodTemplat
 func (c *Cluster) ImagesToFetch() registry.ImageCreds {
 	allImageCreds := make(registry.ImageCreds)
 
-	namespaces, err := c.client.Namespaces().List(meta_v1.ListOptions{})
+	namespaces, err := c.getNamespaces()
 	if err != nil {
 		c.logger.Log("err", errors.Wrap(err, "getting namespaces"))
 		return allImageCreds
 	}
 
-	for _, ns := range namespaces.Items {
+	for _, ns := range namespaces {
 		for kind, resourceKind := range resourceKinds {
 			podControllers, err := resourceKind.getPodControllers(c, ns.Name)
 			if err != nil {
@@ -402,3 +412,27 @@ func (c *Cluster) ImagesToFetch() registry.ImageCreds {
 
 	return allImageCreds
 }
+
+// getNamespaces returns a list of namespaces that the Flux instance is expected
+// to have access to and can look for resources inside of.
+// It returns a list of all namespaces unless a namespace whitelist has been set on the Cluster
+// instance, in which case it returns a list containing the namespaces from the whitelist
+// that exist in the cluster.
+func (c *Cluster) getNamespaces() ([]apiv1.Namespace, error) {
+	nsList := []apiv1.Namespace{}
+
+	namespaces, err := c.client.Namespaces().List(meta_v1.ListOptions{})
+	if err != nil {
+		return nsList, err
+	}
+
+	for _, namespace := range namespaces.Items {
+		if len(c.nsWhitelist) > 0 && ! c.nsWhitelist[namespace.ObjectMeta.Name] {
+			continue
+		}
+
+		nsList = append(nsList, namespace)
+  }
+
+	return nsList, nil
+}

cluster/kubernetes/kubernetes_test.go

@@ -0,0 +1,62 @@
+package kubernetes
+
+import (
+	apiv1 "k8s.io/api/core/v1"
+	meta_v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	fakekubernetes "k8s.io/client-go/kubernetes/fake"
+	"testing"
+	"reflect"
+)
+
+func newNamespace(name string) *apiv1.Namespace {
+	return &apiv1.Namespace{
+		ObjectMeta: meta_v1.ObjectMeta{
+			Name: name,
+		},
+		TypeMeta: meta_v1.TypeMeta{
+			APIVersion: "v1",
+			Kind: "Namespace",
+		},
+	}
+}
+
+func testGetNamespaces(t *testing.T, namespace []string, expected []string) {
+	clientset := fakekubernetes.NewSimpleClientset(newNamespace("default"),
+		newNamespace("kube-system"))
+
+	c := NewCluster(clientset, nil, nil, nil, nil, namespace)
+
+	namespaces, err := c.getNamespaces()
+	if err != nil {
+		t.Errorf("The error should be nil, not: %s", err)
+	}
+
+	result := []string{}
+	for _, namespace := range namespaces {
+		result = append(result, namespace.ObjectMeta.Name)
+	}
+
+	if reflect.DeepEqual(result, expected) != true {
+		t.Errorf("Unexpected namespaces: %v != %v.", result, expected)
+	}
+}
+
+func TestGetNamespacesDefault(t *testing.T) {
+	testGetNamespaces(t, []string{}, []string{"default","kube-system",})
+}
+
+func TestGetNamespacesNamespacesIsNil(t *testing.T) {
+	testGetNamespaces(t, nil, []string{"default","kube-system",})
+}
+
+func TestGetNamespacesNamespacesSet(t *testing.T) {
+	testGetNamespaces(t, []string{"default"}, []string{"default",})
+}
+
+func TestGetNamespacesNamespacesSetDoesNotExist(t *testing.T) {
+	testGetNamespaces(t, []string{"hello"}, []string{})
+}
+
+func TestGetNamespacesNamespacesMultiple(t *testing.T) {
+	testGetNamespaces(t, []string{"default","hello","kube-system"}, []string{"default","kube-system"})
+}

cmd/fluxd/main.go

@@ -114,6 +114,7 @@ func main() {
 		token       = fs.String("token", "", "Authentication token for upstream service")
 
 		dockerConfig = fs.String("docker-config", "", "path to a docker config to use for image registry credentials")
+		k8sNamespaceWhitelist = fs.StringSlice("k8s-namespace-whitelist", []string{}, "Optional, comma separated list of namespaces to monitor for workloads")
 	)
 
 	if err := fs.Parse(os.Args[1:]); err != nil {
@@ -241,7 +242,7 @@ func main() {
 		logger.Log("kubectl", kubectl)
 
 		kubectlApplier := kubernetes.NewKubectl(kubectl, restClientConfig)
-		k8sInst := kubernetes.NewCluster(clientset, ifclientset, kubectlApplier, sshKeyRing, logger)
+		k8sInst := kubernetes.NewCluster(clientset, ifclientset, kubectlApplier, sshKeyRing, logger, *k8sNamespaceWhitelist)
 
 		if err := k8sInst.Ping(); err != nil {
 			logger.Log("ping", err)

site/daemon.md

@@ -70,6 +70,9 @@ fluxd requires setup and offers customization though a multitude of flags.
 |--k8s-secret-name       | `flux-git-deploy`               | name of the k8s secret used to store the private SSH key|
 |--k8s-secret-volume-mount-path | `/etc/fluxd/ssh`         | mount location of the k8s secret storing the private SSH key|
 |--k8s-secret-data-key   | `identity`                      | data key holding the private SSH key within the k8s secret|
+|**k8s configuration**   |                            |  | |
+|--k8s-namespace-whitelist|                                | optional, comma separated list of namespaces to monitor for workloads (default: all namespaces)|
+|**upstream service**    |                            |  | |
 |--connect               |                               | connect to an upstream service e.g., Weave Cloud, at this base address|
 |--token                 |                               | authentication token for upstream service|
 |**SSH key generation**  |                               | |