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.NamespaceLabels, "namespace-labels", []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\"")

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"`
+	NamespaceLabels     []string           `json:"namespace_labels" yaml:"namespace_labels"`
 	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/controller.go

@@ -18,6 +18,7 @@ import (
 	"context"
 	"fmt"
 	"os"
+	"strings"
 	"sync"
 	"time"
 
@@ -71,6 +72,7 @@ type Controller struct {
 	cfg               *config.Config
 	wg                sync.WaitGroup
 	watchingNamespace map[string]struct{}
+	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
@@ -149,6 +154,7 @@ func NewController(cfg *config.Config) (*Controller, error) {
 
 	var (
 		watchingNamespace map[string]struct{}
+		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))
@@ -157,6 +163,14 @@ func NewController(cfg *config.Config) (*Controller, error) {
 		}
 	}
 
+	// support namespace label-selector
+	if len(cfg.Kubernetes.NamespaceLabels) > 1 {
+		for _, labels := range cfg.Kubernetes.NamespaceLabels {
+			labelSlice := strings.Split(labels, "=")
+			watchingLabels[labelSlice[0]] = labelSlice[1]
+		}
+	}
+
 	// recorder
 	utilruntime.Must(apisixscheme.AddToScheme(scheme.Scheme))
 	eventBroadcaster := record.NewBroadcaster()
@@ -171,6 +185,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 +203,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 +252,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 +268,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 +423,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 +437,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 +471,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)
 	})

pkg/ingress/namespace.go

@@ -0,0 +1,178 @@
+// 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[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[namespace.Name] = struct{}{}
+			}
+		}
+	} else { // type == types.EventDelete
+		namespace := ev.Tombstone.(*corev1.Namespace)
+		if _, ok := c.controller.watchingNamespace[namespace.Name]; ok {
+			if namespace, err := c.controller.kubeClient.Client.CoreV1().Namespaces().Get(ctx, ev.Object.(string), metav1.GetOptions{}); err == nil {
+				// if namespace is still in k8s cluster, the delete event is deprecated
+				if c.controller.watchingLabels.IsSubsetOf(namespace.Labels) {
+					c.controller.watchingNamespace[namespace.Name] = struct{}{}
+				}
+				return nil
+			}
+			delete(c.controller.watchingNamespace, 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,
+	})
+}