feat: add label-selector for watching namespace

cmd/ingress/ingress.go

@@ -143,7 +143,8 @@ the apisix cluster and others are created`,
 	cmd.PersistentFlags().BoolVar(&cfg.EnableProfiling, "enable-profiling", true, "enable profiling via web interface host:port/debug/pprof")
 	cmd.PersistentFlags().StringVar(&cfg.Kubernetes.Kubeconfig, "kubeconfig", "", "Kubernetes configuration file (by default in-cluster configuration will be used)")
 	cmd.PersistentFlags().DurationVar(&cfg.Kubernetes.ResyncInterval.Duration, "resync-interval", time.Minute, "the controller resync (with Kubernetes) interval, the minimum resync interval is 30s")
-	cmd.PersistentFlags().StringSliceVar(&cfg.Kubernetes.AppNamespaces, "app-namespace", []string{config.NamespaceAll}, "namespaces that controller will watch for resources")
+	cmd.PersistentFlags().StringSliceVar(&cfg.Kubernetes.AppNamespaces, "app-namespace", []string{config.NamespaceAll}, "namespaces that controller will watch for resources, this is deprecated.")
+	cmd.PersistentFlags().StringSliceVar(&cfg.Kubernetes.NamespaceSelector, "namespace-selector", []string{""}, "labels that controller used to select namespaces which will watch for resources")
 	cmd.PersistentFlags().StringVar(&cfg.Kubernetes.IngressClass, "ingress-class", config.IngressClass, "the class of an Ingress object is set using the field IngressClassName in Kubernetes clusters version v1.18.0 or higher or the annotation \"kubernetes.io/ingress.class\" (deprecated)")
 	cmd.PersistentFlags().StringVar(&cfg.Kubernetes.ElectionID, "election-id", config.IngressAPISIXLeader, "election id used for campaign the controller leader")
 	cmd.PersistentFlags().StringVar(&cfg.Kubernetes.IngressVersion, "ingress-version", config.IngressNetworkingV1, "the supported ingress api group version, can be \"networking/v1beta1\", \"networking/v1\" (for Kubernetes version v1.19.0 or higher) and \"extensions/v1beta1\"")
@@ -155,5 +156,8 @@ the apisix cluster and others are created`,
 	cmd.PersistentFlags().StringVar(&cfg.APISIX.DefaultClusterAdminKey, "default-apisix-cluster-admin-key", "", "admin key used for the authorization of admin api / manager api for the default APISIX cluster")
 	cmd.PersistentFlags().StringVar(&cfg.APISIX.DefaultClusterName, "default-apisix-cluster-name", "default", "name of the default apisix cluster")
 
+	if err := cmd.PersistentFlags().MarkDeprecated("app-namespace", "use namespace-selector instead"); err != nil {
+		dief("failed to mark `app-namespace` as deprecated: %s", err)
+	}
 	return cmd
 }

conf/config-default.yaml

@@ -44,6 +44,10 @@ kubernetes:
                                        # and the minimal resync interval is 30s.
   app_namespaces: ["*"]                # namespace list that controller will watch for resources,
                                        # by default all namespaces (represented by "*") are watched.
+                                       # The `app_namespace` is deprecated, using `namespace_selector` instead since version 1.4.0
+  namespace_selector: [""]             # namespace_selector represent basis for selecting managed namespaces.
+                                       # the field is support since version 1.4.0
+                                       # For example, "apisix.ingress=watching", so ingress will watching the namespaces which labels "apisix.ingress=watching"
   election_id: "ingress-apisix-leader" # the election id for the controller leader campaign,
                                        # only the leader will watch and delivery resource changes,
                                        # other instances (as candidates) stand by.

docs/en/latest/practices/the-hard-way.md

@@ -623,6 +623,8 @@ data:
       resync_interval: "30s"
       app_namespaces:
       - "*"
+      namespace_selector:
+      - "apisix.ingress=watching"
       ingress_class: "apisix"
       ingress_version: "networking/v1"
       apisix_route_version: "apisix.apache.org/v2beta1"

pkg/config/config.go

@@ -77,6 +77,7 @@ type KubernetesConfig struct {
 	Kubeconfig          string             `json:"kubeconfig" yaml:"kubeconfig"`
 	ResyncInterval      types.TimeDuration `json:"resync_interval" yaml:"resync_interval"`
 	AppNamespaces       []string           `json:"app_namespaces" yaml:"app_namespaces"`
+	NamespaceSelector   []string           `json:"namespace_selector" yaml:"namespace_selector"`
 	ElectionID          string             `json:"election_id" yaml:"election_id"`
 	IngressClass        string             `json:"ingress_class" yaml:"ingress_class"`
 	IngressVersion      string             `json:"ingress_version" yaml:"ingress_version"`

pkg/ingress/compare.go

@@ -50,17 +50,16 @@ func (c *Controller) CompareResources(ctx context.Context) error {
 			log.Error(err.Error())
 			ctx.Done()
 		} else {
-			wns := make(map[string]struct{}, len(nsList.Items))
+			wns := new(sync.Map)
 			for _, v := range nsList.Items {
-				wns[v.Name] = struct{}{}
+				wns.Store(v.Name, struct{}{})
 			}
 			c.watchingNamespace = wns
 		}
 	}
-	if len(c.watchingNamespace) > 0 {
-		wg.Add(len(c.watchingNamespace))
-	}
-	for ns := range c.watchingNamespace {
+
+	c.watchingNamespace.Range(func(key, value interface{}) bool {
+		wg.Add(1)
 		go func(ns string) {
 			defer wg.Done()
 			// ApisixRoute
@@ -130,8 +129,9 @@ func (c *Controller) CompareResources(ctx context.Context) error {
 					}
 				}
 			}
-		}(ns)
-	}
+		}(key.(string))
+		return true
+	})
 	wg.Wait()
 
 	// 2.get all cache routes

pkg/ingress/controller.go

@@ -18,6 +18,7 @@ import (
 	"context"
 	"fmt"
 	"os"
+	"strings"
 	"sync"
 	"time"
 
@@ -70,7 +71,8 @@ type Controller struct {
 	namespace         string
 	cfg               *config.Config
 	wg                sync.WaitGroup
-	watchingNamespace map[string]struct{}
+	watchingNamespace *sync.Map
+	watchingLabels    types.Labels
 	apisix            apisix.APISIX
 	podCache          types.PodCache
 	translator        translation.Translator
@@ -90,6 +92,8 @@ type Controller struct {
 	leaderContextCancelFunc context.CancelFunc
 
 	// common informers and listers
+	namespaceInformer           cache.SharedIndexInformer
+	namespaceLister             listerscorev1.NamespaceLister
 	podInformer                 cache.SharedIndexInformer
 	podLister                   listerscorev1.PodLister
 	epInformer                  cache.SharedIndexInformer
@@ -112,6 +116,7 @@ type Controller struct {
 	apisixConsumerLister        listersv2alpha1.ApisixConsumerLister
 
 	// resource controllers
+	namespaceController     *namespaceController
 	podController           *podController
 	endpointsController     *endpointsController
 	endpointSliceController *endpointSliceController
@@ -148,12 +153,20 @@ func NewController(cfg *config.Config) (*Controller, error) {
 	}
 
 	var (
-		watchingNamespace map[string]struct{}
+		watchingNamespace = new(sync.Map)
+		watchingLabels    = make(map[string]string)
 	)
 	if len(cfg.Kubernetes.AppNamespaces) > 1 || cfg.Kubernetes.AppNamespaces[0] != v1.NamespaceAll {
-		watchingNamespace = make(map[string]struct{}, len(cfg.Kubernetes.AppNamespaces))
 		for _, ns := range cfg.Kubernetes.AppNamespaces {
-			watchingNamespace[ns] = struct{}{}
+			watchingNamespace.Store(ns, struct{}{})
+		}
+	}
+
+	// support namespace label-selector
+	if len(cfg.Kubernetes.NamespaceSelector) > 1 {
+		for _, labels := range cfg.Kubernetes.NamespaceSelector {
+			labelSlice := strings.Split(labels, "=")
+			watchingLabels[labelSlice[0]] = labelSlice[1]
 		}
 	}
 
@@ -171,6 +184,7 @@ func NewController(cfg *config.Config) (*Controller, error) {
 		metricsCollector:  metrics.NewPrometheusCollector(),
 		kubeClient:        kubeClient,
 		watchingNamespace: watchingNamespace,
+		watchingLabels:    watchingLabels,
 		secretSSLMap:      new(sync.Map),
 		recorder:          eventBroadcaster.NewRecorder(scheme.Scheme, v1.EventSource{Component: _component}),
 
@@ -188,6 +202,7 @@ func (c *Controller) initWhenStartLeading() {
 	kubeFactory := c.kubeClient.NewSharedIndexInformerFactory()
 	apisixFactory := c.kubeClient.NewAPISIXSharedIndexInformerFactory()
 
+	c.namespaceLister = kubeFactory.Core().V1().Namespaces().Lister()
 	c.podLister = kubeFactory.Core().V1().Pods().Lister()
 	c.epLister, c.epInformer = kube.NewEndpointListerAndInformer(kubeFactory, c.cfg.Kubernetes.WatchEndpointSlices)
 	c.svcLister = kubeFactory.Core().V1().Services().Lister()
@@ -236,6 +251,7 @@ func (c *Controller) initWhenStartLeading() {
 		apisixRouteInformer = apisixFactory.Apisix().V2beta2().ApisixRoutes().Informer()
 	}
 
+	c.namespaceInformer = kubeFactory.Core().V1().Namespaces().Informer()
 	c.podInformer = kubeFactory.Core().V1().Pods().Informer()
 	c.svcInformer = kubeFactory.Core().V1().Services().Informer()
 	c.ingressInformer = ingressInformer
@@ -251,6 +267,7 @@ func (c *Controller) initWhenStartLeading() {
 	} else {
 		c.endpointsController = c.newEndpointsController()
 	}
+	c.namespaceController = c.newNamespaceController()
 	c.podController = c.newPodController()
 	c.apisixUpstreamController = c.newApisixUpstreamController()
 	c.ingressController = c.newIngressController()
@@ -405,6 +422,11 @@ func (c *Controller) run(ctx context.Context) {
 
 	c.initWhenStartLeading()
 
+	// list namesapce and init watchingNamespace
+	if err := c.initWatchingNamespaceByLabels(ctx); err != nil {
+		ctx.Done()
+		return
+	}
 	// compare resources of k8s with objects of APISIX
 	if err = c.CompareResources(ctx); err != nil {
 		ctx.Done()
@@ -414,6 +436,9 @@ func (c *Controller) run(ctx context.Context) {
 	c.goAttach(func() {
 		c.checkClusterHealth(ctx, cancelFunc)
 	})
+	c.goAttach(func() {
+		c.namespaceInformer.Run(ctx.Done())
+	})
 	c.goAttach(func() {
 		c.podInformer.Run(ctx.Done())
 	})
@@ -445,6 +470,9 @@ func (c *Controller) run(ctx context.Context) {
 	c.goAttach(func() {
 		c.apisixConsumerInformer.Run(ctx.Done())
 	})
+	c.goAttach(func() {
+		c.namespaceController.run(ctx)
+	})
 	c.goAttach(func() {
 		c.podController.run(ctx)
 	})
@@ -502,7 +530,7 @@ func (c *Controller) namespaceWatching(key string) (ok bool) {
 		log.Warnf("resource %s was ignored since: %s", key, err)
 		return
 	}
-	_, ok = c.watchingNamespace[ns]
+	_, ok = c.watchingNamespace.Load(ns)
 	return
 }
 

pkg/ingress/namespace.go

@@ -0,0 +1,171 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements.  See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License.  You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+package ingress
+
+import (
+	"context"
+	"time"
+
+	"go.uber.org/zap"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/client-go/tools/cache"
+	"k8s.io/client-go/util/workqueue"
+
+	"github.com/apache/apisix-ingress-controller/pkg/log"
+	"github.com/apache/apisix-ingress-controller/pkg/types"
+)
+
+type namespaceController struct {
+	controller *Controller
+	workqueue  workqueue.RateLimitingInterface
+	workers    int
+}
+
+func (c *Controller) newNamespaceController() *namespaceController {
+	ctl := &namespaceController{
+		controller: c,
+		workqueue:  workqueue.NewNamedRateLimitingQueue(workqueue.NewItemFastSlowRateLimiter(1*time.Second, 60*time.Second, 5), "Namespace"),
+		workers:    1,
+	}
+	ctl.controller.namespaceInformer.AddEventHandler(
+		cache.ResourceEventHandlerFuncs{
+			AddFunc:    ctl.onAdd,
+			UpdateFunc: ctl.onUpdate,
+			DeleteFunc: ctl.onDelete,
+		},
+	)
+	return ctl
+}
+
+func (c *Controller) initWatchingNamespaceByLabels(ctx context.Context) error {
+	labelSelector := metav1.LabelSelector{MatchLabels: c.watchingLabels}
+	opts := metav1.ListOptions{
+		LabelSelector: labels.Set(labelSelector.MatchLabels).String(),
+	}
+	namespaces, err := c.kubeClient.Client.CoreV1().Namespaces().List(ctx, opts)
+	if err != nil {
+		return err
+	} else {
+		for _, ns := range namespaces.Items {
+			c.watchingNamespace.Store(ns.Name, struct{}{})
+		}
+	}
+	return nil
+}
+
+func (c *namespaceController) run(ctx context.Context) {
+	log.Info("namespace controller started")
+	defer log.Info("namespace controller exited")
+
+	if ok := cache.WaitForCacheSync(ctx.Done(), c.controller.namespaceInformer.HasSynced); !ok {
+		log.Error("informers sync failed")
+		return
+	}
+	for i := 0; i < c.workers; i++ {
+		go c.runWorker(ctx)
+	}
+	<-ctx.Done()
+}
+
+func (c *namespaceController) runWorker(ctx context.Context) {
+	for {
+		obj, quit := c.workqueue.Get()
+		if quit {
+			return
+		}
+		err := c.sync(ctx, obj.(*types.Event))
+		c.workqueue.Done(obj)
+		c.handleSyncErr(obj.(*types.Event), err)
+	}
+}
+
+func (c *namespaceController) sync(ctx context.Context, ev *types.Event) error {
+	if ev.Type != types.EventDelete {
+		// check the labels of specify namespace
+		namespace, err := c.controller.kubeClient.Client.CoreV1().Namespaces().Get(ctx, ev.Object.(string), metav1.GetOptions{})
+		if err != nil {
+			return err
+		} else {
+			// if labels of namespace contains the watchingLabels, the namespace should be set to controller.watchingNamespace
+			if c.controller.watchingLabels.IsSubsetOf(namespace.Labels) {
+				c.controller.watchingNamespace.Store(namespace.Name, struct{}{})
+			}
+		}
+	} else { // type == types.EventDelete
+		namespace := ev.Tombstone.(*corev1.Namespace)
+		if _, ok := c.controller.watchingNamespace.Load(namespace.Name); ok {
+			c.controller.watchingNamespace.Delete(namespace.Name)
+			// need to compare
+			err := c.controller.CompareResources(ctx)
+			if err != nil {
+				return err
+			}
+		}
+		// do nothing, if the namespace did not in controller.watchingNamespace
+	}
+	return nil
+}
+
+func (c *namespaceController) handleSyncErr(event *types.Event, err error) {
+	name := event.Object.(string)
+	if err != nil {
+		log.Warnw("sync namespace info failed, will retry",
+			zap.String("namespace", name),
+			zap.Error(err),
+		)
+		c.workqueue.AddRateLimited(event)
+	} else {
+		c.workqueue.Forget(event)
+	}
+}
+
+func (c *namespaceController) onAdd(obj interface{}) {
+	key, err := cache.MetaNamespaceKeyFunc(obj)
+	if err == nil {
+		log.Debugw(key)
+	}
+	c.workqueue.AddRateLimited(&types.Event{
+		Type:   types.EventAdd,
+		Object: key,
+	})
+}
+
+func (c *namespaceController) onUpdate(pre, cur interface{}) {
+	oldNamespace := pre.(*corev1.Namespace)
+	newNamespace := cur.(*corev1.Namespace)
+	if oldNamespace.ResourceVersion >= newNamespace.ResourceVersion {
+		return
+	}
+	key, err := cache.MetaNamespaceKeyFunc(cur)
+	if err != nil {
+		log.Errorf("found Namespace resource with error: %s", err)
+		return
+	}
+	c.workqueue.AddRateLimited(&types.Event{
+		Type:   types.EventUpdate,
+		Object: key,
+	})
+}
+
+func (c *namespaceController) onDelete(obj interface{}) {
+	namespace := obj.(*corev1.Namespace)
+	c.workqueue.AddRateLimited(&types.Event{
+		Type:      types.EventDelete,
+		Object:    namespace.Name,
+		Tombstone: namespace,
+	})
+}