Add --k8s-namespace-whitelist setting that specifies namespaces to wa…

…tch.

Fixes #1181

Currently, Flux expects to have access to all namespaces, even if no manifests
in the repository reference another namespace, it will check all namespaces
for controllers to update.

This change adds a --k8s-namespace-whitelist setting which, if set, will restrict
Flux to only watch the specified namespaces and ignore all others.

Intended for clusters with large amounts of namespaces or restrictive RBAC
policies. If provided Flux will only monitor workloads in the given namespaces.
This significantly cuts the number of API calls made.

An empty list (i.e. not provided) yields the usual behaviour.

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 []string
 
 	mu sync.Mutex
 }
@@ -118,7 +119,8 @@ func NewCluster(clientset k8sclient.Interface,
 	ifclientset ifclient.Interface,
 	applier Applier,
 	sshKeyRing ssh.KeyRing,
-	logger log.Logger) *Cluster {
+	logger log.Logger,
+	nsWhitelist []string) *Cluster {
 
 	c := &Cluster{
 		client: extendedClient{
@@ -132,6 +134,7 @@ func NewCluster(clientset k8sclient.Interface,
 		applier:    applier,
 		logger:     logger,
 		sshKeyRing: sshKeyRing,
+		nsWhitelist: nsWhitelist,
 	}
 
 	return c
@@ -252,11 +255,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 +370,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 +407,32 @@ 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{}
+	validNamespaces := map[string]bool{}
+
+	namespaces, err := c.client.Namespaces().List(meta_v1.ListOptions{})
+	if err != nil {
+		return nsList, err
+	}
+
+	for _, namespace := range c.nsWhitelist {
+		validNamespaces[namespace] = true
+	}
+
+	for _, namespace := range namespaces.Items {
+		if len(validNamespaces) > 0 && ! validNamespaces[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**  |                               | |