implement pod selector and use it in predication of pod reconciliations

Signed-off-by: Bruce Ma <[email protected]>

cmd/manager/main.go

@@ -21,6 +21,7 @@ import (
 	"fmt"
 	"os"
 
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	kubevirtv1 "kubevirt.io/api/core/v1"
 
 	"github.com/spf13/pflag"
@@ -34,6 +35,7 @@ import (
 	networkingv1 "github.com/alibaba/hybridnet/pkg/apis/networking/v1"
 	"github.com/alibaba/hybridnet/pkg/controllers/multicluster"
 	"github.com/alibaba/hybridnet/pkg/controllers/networking"
+	"github.com/alibaba/hybridnet/pkg/controllers/utils"
 	"github.com/alibaba/hybridnet/pkg/feature"
 	zapinit "github.com/alibaba/hybridnet/pkg/zap"
 )
@@ -56,13 +58,15 @@ func main() {
 		clientQPS             float32
 		clientBurst           int
 		metricsPort           int
+		selectorStr           string
 	)
 
 	// register flags
 	pflag.StringToIntVar(&controllerConcurrency, "controller-concurrency", map[string]int{}, "The specified concurrency of different controllers.")
 	pflag.Float32Var(&clientQPS, "kube-client-qps", 300, "The QPS limit of apiserver client.")
 	pflag.IntVar(&clientBurst, "kube-client-burst", 600, "The Burst limit of apiserver client.")
 	pflag.IntVar(&metricsPort, "metrics-port", 9899, "The port to listen on for prometheus metrics.")
+	pflag.StringVar(&selectorStr, "pod-label-selector", "", "The label selector to select specified pods for IPAM.")
 
 	// parse flags
 	pflag.CommandLine.AddGoFlagSet(flag.CommandLine)
@@ -82,6 +86,20 @@ func main() {
 	clientConfig.QPS = clientQPS
 	clientConfig.Burst = clientBurst
 
+	// initialize objects from flags
+	// if selector string is empty, it means select everything
+	labelSelector, err := metav1.ParseToLabelSelector(selectorStr)
+	if err != nil {
+		entryLog.Error(err, "unable to parse label selector")
+		os.Exit(1)
+	}
+
+	var podSelector utils.PodSelector
+	if podSelector, err = utils.LabelSelectorAsPodSelector(labelSelector); err != nil {
+		entryLog.Error(err, "unable to create pod selector")
+		os.Exit(1)
+	}
+
 	mgr, err := ctrl.NewManager(clientConfig, ctrl.Options{
 		Scheme:                  scheme,
 		Logger:                  ctrl.Log.WithName("manager"),
@@ -120,6 +138,7 @@ func main() {
 
 	if err = networking.RegisterToManager(globalContext, mgr, networking.RegisterOptions{
 		ConcurrencyMap: controllerConcurrency,
+		PodSelector:    podSelector,
 	}); err != nil {
 		entryLog.Error(err, "unable to register networking controllers")
 		os.Exit(1)

pkg/controllers/networking/manager.go

@@ -25,11 +25,13 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 
 	"github.com/alibaba/hybridnet/pkg/controllers/concurrency"
+	"github.com/alibaba/hybridnet/pkg/controllers/utils"
 )
 
 type RegisterOptions struct {
 	NewIPAMManager NewIPAMManagerFunction
 	ConcurrencyMap map[string]int
+	PodSelector    utils.PodSelector
 }
 
 func RegisterToManager(ctx context.Context, mgr manager.Manager, options RegisterOptions) error {
@@ -93,6 +95,7 @@ func RegisterToManager(ctx context.Context, mgr manager.Manager, options Registe
 		IPAMStore:             ipamStore,
 		IPAMManager:           ipamManager,
 		ControllerConcurrency: concurrency.ControllerConcurrency(options.ConcurrencyMap[ControllerPod]),
+		PodSelector:           options.PodSelector,
 	}).SetupWithManager(mgr); err != nil {
 		return fmt.Errorf("unable to inject controller %s: %v", ControllerPod, err)
 	}

pkg/controllers/networking/pod_controller.go

@@ -80,6 +80,7 @@ type PodReconciler struct {
 	IPAMManager IPAMManager
 
 	concurrency.ControllerConcurrency
+	PodSelector utils.PodSelector
 }
 
 //+kubebuilder:rbac:groups="",resources=pods,verbs=get;list;watch;create;update;patch;delete
@@ -773,6 +774,19 @@ func (r *PodReconciler) SetupWithManager(mgr ctrl.Manager) (err error) {
 			builder.WithPredicates(
 				&utils.IgnoreDeletePredicate{},
 				&predicate.ResourceVersionChangedPredicate{},
+				predicate.NewPredicateFuncs(func(obj client.Object) bool {
+					if r.PodSelector == nil {
+						return true
+					}
+
+					pod, ok := obj.(*corev1.Pod)
+					if !ok {
+						return false
+					}
+
+					// Only selected pods should be processed
+					return r.PodSelector.Matches(pod)
+				}),
 				predicate.NewPredicateFuncs(func(obj client.Object) bool {
 					pod, ok := obj.(*corev1.Pod)
 					if !ok {

pkg/controllers/utils/selector.go

@@ -0,0 +1,51 @@
+/*
+ Copyright 2024 The Hybridnet Authors.
+
+ Licensed 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 utils
+
+import (
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
+)
+
+type PodSelector interface {
+	Matches(pod *corev1.Pod) bool
+}
+
+type podSelector struct {
+	selector labels.Selector
+}
+
+func (p *podSelector) Matches(pod *corev1.Pod) bool {
+	// non-blocking during exception cases
+	if p == nil || pod == nil {
+		return true
+	}
+
+	return p.selector.Matches(labels.Set(pod.Labels))
+}
+
+func LabelSelectorAsPodSelector(labelSelector *metav1.LabelSelector) (PodSelector, error) {
+	selector, err := metav1.LabelSelectorAsSelector(labelSelector)
+	if err != nil {
+		return nil, err
+	}
+
+	return &podSelector{
+		selector: selector,
+	}, nil
+}