Implement --watch-namespace flag

railgun/manager/config.go

@@ -7,6 +7,7 @@ import (
  "github.com/kong/kubernetes-ingress-controller/pkg/annotations"
  "github.com/kong/kubernetes-ingress-controller/pkg/util"
  "github.com/spf13/pflag"
+ apiv1 "k8s.io/api/core/v1"
 )
 
 // -----------------------------------------------------------------------------
@@ -42,6 +43,7 @@ type Config struct {
  LeaderElectionID string
  Concurrency int
  FilterTag string
+ WatchNamespace string
 
  // Kubernetes API toggling
  IngressExtV1beta1Enabled util.EnablementStatus
@@ -105,6 +107,9 @@ func (c *Config) FlagSet() *pflag.FlagSet {
  flagSet.StringVar(&c.LeaderElectionID, "election-id", "5b374a9e.konghq.com", `Election id to use for status update.`)
  flagSet.StringVar(&c.FilterTag, "kong-filter-tag", "managed-by-railgun", "TODO")
  flagSet.IntVar(&c.Concurrency, "kong-concurrency", 10, "TODO")
+ flagSet.StringVar(&c.WatchNamespace, "watch-namespace", apiv1.NamespaceAll,
+ `Namespace(s) to watch for Kubernetes resources. Defaults to all namespaces. To watch multiple namespaces, use
+ a comma-separated list of namespaces.`)
 
  // Kubernetes API toggling
  flagSet.enablementStatusVar(&c.IngressNetV1Enabled, "controller-ingress-networkingv1", util.EnablementStatusEnabled, "Enable or disable the Ingress controller (using API version networking.k8s.io/v1)."+onOffUsage)

railgun/manager/run.go

@@ -5,6 +5,7 @@ import (
  "context"
  "fmt"
  "os"
+ "strings"
 
  "github.com/bombsimon/logrusr"
  "github.com/go-logr/logr"
@@ -13,6 +14,7 @@ import (
  clientgoscheme "k8s.io/client-go/kubernetes/scheme"
  "k8s.io/client-go/tools/clientcmd"
  ctrl "sigs.k8s.io/controller-runtime"
+ "sigs.k8s.io/controller-runtime/pkg/cache"
  "sigs.k8s.io/controller-runtime/pkg/healthz"
 
  "github.com/kong/kubernetes-ingress-controller/pkg/adminapi"
@@ -63,15 +65,28 @@ func Run(ctx context.Context, c *Config) error {
  // set "kubernetes.io/ingress.class" to be used by controllers (defaults to "kong")
  setupLog.Info(`the ingress class name has been set`, "value", c.IngressClassName)
 
- // build the controller manager
- mgr, err := ctrl.NewManager(kubeconfig, ctrl.Options{
+ controllerOpts := ctrl.Options{
  Scheme: scheme,
  MetricsBindAddress: c.MetricsAddr,
  Port: 9443,
  HealthProbeBindAddress: c.ProbeAddr,
  LeaderElection: c.EnableLeaderElection,
  LeaderElectionID: c.LeaderElectionID,
- })
+ }
+
+ // determine how to configure namespace watchers
+ if strings.Contains(c.WatchNamespace, ",") {
+ setupLog.Info("manager set up with multiple namespaces", "namespaces", c.WatchNamespace)
+ // this mode does not set the Namespace option, so the manager will default to watching all namespaces
+ // MultiNamespacedCacheBuilder imposes a filter on top of that watch to retrieve scoped resources
+ // from the watched namespaces only.
+ controllerOpts.NewCache = cache.MultiNamespacedCacheBuilder(strings.Split(c.WatchNamespace, ","))
+ } else {
+ controllerOpts.Namespace = c.WatchNamespace
+ }
+
+ // build the controller manager
+ mgr, err := ctrl.NewManager(kubeconfig, controllerOpts)
  if err != nil {
  setupLog.Error(err, "unable to start manager")
  return err

railgun/test/integration/ingress_test.go

@@ -309,3 +309,99 @@ func TestIngressClassNameSpec(t *testing.T) {
  }, ingressWait, waitTick)
 
 }
+
+func TestIngressNamespaces(t *testing.T) {
+ if useLegacyKIC() {
+ t.Skip("support for distinct namespace watches is not supported in legacy KIC")
+ }
+ ctx := context.Background()
+
+ elsewhere := "elsewhere"
+ nowhere := "nowhere"
+ nowhereNamespace := &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: nowhere}}
+ _, err := cluster.Client().CoreV1().Namespaces().Create(ctx, nowhereNamespace, metav1.CreateOptions{})
+ assert.NoError(t, err)
+ t.Log("deploying a minimal HTTP container deployment to test Ingress routes")
+ container := k8sgen.NewContainer("httpbin", httpBinImage, 80)
+ deployment := k8sgen.NewDeploymentForContainer(container)
+ elsewhereDeployment, err := cluster.Client().AppsV1().Deployments(elsewhere).Create(ctx, deployment, metav1.CreateOptions{})
+ assert.NoError(t, err)
+ nowhereDeployment, err := cluster.Client().AppsV1().Deployments(nowhere).Create(ctx, deployment, metav1.CreateOptions{})
+ assert.NoError(t, err)
+
+ defer func() {
+ t.Logf("cleaning up the deployment %s", elsewhereDeployment.Name)
+ assert.NoError(t, cluster.Client().AppsV1().Deployments(elsewhere).Delete(ctx, elsewhereDeployment.Name, metav1.DeleteOptions{}))
+ assert.NoError(t, cluster.Client().AppsV1().Deployments(nowhere).Delete(ctx, nowhereDeployment.Name, metav1.DeleteOptions{}))
+ }()
+
+ t.Logf("exposing deployment %s via service", deployment.Name)
+ service := k8sgen.NewServiceForDeployment(deployment, corev1.ServiceTypeLoadBalancer)
+ _, err = cluster.Client().CoreV1().Services(elsewhere).Create(ctx, service, metav1.CreateOptions{})
+ assert.NoError(t, err)
+ _, err = cluster.Client().CoreV1().Services(nowhere).Create(ctx, service, metav1.CreateOptions{})
+ assert.NoError(t, err)
+
+ defer func() {
+ t.Logf("cleaning up the service %s", service.Name)
+ assert.NoError(t, cluster.Client().CoreV1().Services(elsewhere).Delete(ctx, service.Name, metav1.DeleteOptions{}))
+ assert.NoError(t, cluster.Client().CoreV1().Services(nowhere).Delete(ctx, service.Name, metav1.DeleteOptions{}))
+ }()
+
+ t.Logf("creating an ingress for service %s with ingress.class %s", service.Name, ingressClass)
+ elsewhereIngress := k8sgen.NewIngressForService("/elsewhere", map[string]string{
+ annotations.IngressClassKey: ingressClass,
+ "konghq.com/strip-path": "true",
+ }, service)
+ nowhereIngress := k8sgen.NewIngressForService("/nowhere", map[string]string{
+ annotations.IngressClassKey: ingressClass,
+ "konghq.com/strip-path": "true",
+ }, service)
+ elsewhereIngress, err = cluster.Client().NetworkingV1().Ingresses(elsewhere).Create(ctx, elsewhereIngress, metav1.CreateOptions{})
+ assert.NoError(t, err)
+ nowhereIngress, err = cluster.Client().NetworkingV1().Ingresses(nowhere).Create(ctx, nowhereIngress, metav1.CreateOptions{})
+ assert.NoError(t, err)
+
+ defer func() {
+ t.Logf("ensuring that Ingress %s is cleaned up", elsewhereIngress.Name)
+ if err := cluster.Client().NetworkingV1().Ingresses(elsewhere).Delete(ctx, elsewhereIngress.Name, metav1.DeleteOptions{}); err != nil {
+ if !errors.IsNotFound(err) {
+ require.NoError(t, err)
+ }
+ }
+ if err := cluster.Client().NetworkingV1().Ingresses(nowhere).Delete(ctx, nowhereIngress.Name, metav1.DeleteOptions{}); err != nil {
+ if !errors.IsNotFound(err) {
+ require.NoError(t, err)
+ }
+ }
+ }()
+
+ t.Logf("waiting for routes from Ingress %s to be operational", elsewhereIngress.Name)
+ p := proxyReady()
+ assert.Eventually(t, func() bool {
+ resp, err := httpc.Get(fmt.Sprintf("%s/elsewhere", p.ProxyURL.String()))
+ if err != nil {
+ t.Logf("WARNING: error while waiting for %s: %v", p.ProxyURL.String(), err)
+ return false
+ }
+ defer resp.Body.Close()
+ if resp.StatusCode == http.StatusOK {
+ // now that the ingress backend is routable, make sure the contents we're getting back are what we expect
+ // Expected: "<title>httpbin.org</title>"
+ b := new(bytes.Buffer)
+ b.ReadFrom(resp.Body)
+ return strings.Contains(b.String(), "<title>httpbin.org</title>")
+ }
+ return false
+ }, ingressWait, waitTick)
+
+ assert.Eventually(t, func() bool {
+ resp, err := httpc.Get(fmt.Sprintf("%s/nowhere", p.ProxyURL.String()))
+ if err != nil {
+ t.Logf("WARNING: error while waiting for %s: %v", p.ProxyURL.String(), err)
+ return false
+ }
+ defer resp.Body.Close()
+ return expect404WithNoRoute(t, p.ProxyURL.String(), resp)
+ }, ingressWait, waitTick)
+}

railgun/test/integration/suite_test.go

@@ -12,6 +12,7 @@ import (
  "net/url"
  "os"
  "os/exec"
+ "strings"
  "testing"
  "time"
 
@@ -51,6 +52,9 @@ const (
 
  // ingressClass indicates the ingress class name which the tests will use for supported object reconcilation
  ingressClass = "kongtests"
+
+ // elsewhere is the name of an alternative namespace
+ elsewhere = "elsewhere"
 )
 
 // -----------------------------------------------------------------------------
@@ -66,6 +70,9 @@ var (
 
  // cluster is the object which contains a Kubernetes client for the testing cluster
  cluster ktfkind.Cluster
+
+ // watchNamespaces is a list of namespaces the controller watches
+ watchNamespaces = strings.Join([]string{elsewhere, corev1.NamespaceDefault}, ",")
 )
 
 // -----------------------------------------------------------------------------
@@ -138,6 +145,13 @@ func deployControllers(ctx context.Context, ready chan ktfkind.ProxyReadinessEve
  return err
  }
  }
+ // ensure the alternative namespace is created
+ elsewhereNS := &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "elsewhere"}}
+ if _, err := cluster.Client().CoreV1().Namespaces().Create(ctx, elsewhereNS, metav1.CreateOptions{}); err != nil {
+ if !errors.IsAlreadyExists(err) {
+ return err
+ }
+ }
 
  // run the controller in the background
  go func() {
@@ -210,6 +224,7 @@ func deployControllers(ctx context.Context, ready chan ktfkind.ProxyReadinessEve
  "--controller-kongplugin=disabled",
  "--controller-kongconsumer=disabled",
  "--election-id=integrationtests.konghq.com",
+ fmt.Sprintf("--watch-namespace=%s", watchNamespaces),
  fmt.Sprintf("--ingress-class=%s", ingressClass),
  "--log-level=trace",
  "--log-format=text",