feat: stub CNS Pod watcher (#2112)

* feat: cns watches pods

Signed-off-by: Evan Baker <[email protected]>

* indirect pod reconcile for more dynamic behavior

Signed-off-by: Evan Baker <[email protected]>

---------

Signed-off-by: Evan Baker <[email protected]>

Author: rbtr

cns/configuration/configuration.go

@@ -45,6 +45,7 @@ type CNSConfig struct {
 	CNIConflistFilepath         string
 	MellanoxMonitorIntervalSecs int
 	AZRSettings                 AZRSettings
+	WatchPods                   bool
 }
 
 type TelemetrySettings struct {

cns/kubecontroller/nodenetworkconfig/reconciler.go

@@ -164,6 +164,9 @@ func (r *Reconciler) SetupWithManager(mgr ctrl.Manager, node *v1.Node) error {
 		WithEventFilter(predicate.Funcs{
 			// check that the generation is the same - status changes don't update generation.
 			UpdateFunc: func(ue event.UpdateEvent) bool {
+				if ue.ObjectOld == nil || ue.ObjectNew == nil {
+					return false
+				}
 				return ue.ObjectOld.GetGeneration() == ue.ObjectNew.GetGeneration()
 			},
 		}).

cns/kubecontroller/podwatcher/podwatcher.go

@@ -0,0 +1,92 @@
+package podwatcher
+
+import (
+	"context"
+
+	"github.com/pkg/errors"
+	v1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/fields"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/event"
+	"sigs.k8s.io/controller-runtime/pkg/predicate"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+)
+
+type podcli interface {
+	List(ctx context.Context, list client.ObjectList, opts ...client.ListOption) error
+}
+
+type podListener interface {
+	Update([]v1.Pod)
+}
+
+type PodWatcher struct {
+	cli            podcli
+	listOpt        client.ListOption
+	ReconcileFuncs []reconcile.Func
+}
+
+func New(nodename string) *PodWatcher { //nolint:revive // private struct to force constructor
+	return &PodWatcher{
+		listOpt: &client.ListOptions{FieldSelector: fields.SelectorFromSet(fields.Set{"spec.nodeName": nodename})},
+	}
+}
+
+func (p *PodWatcher) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
+	for _, f := range p.ReconcileFuncs {
+		if _, err := f(ctx, req); err != nil {
+			return reconcile.Result{}, errors.Wrap(err, "failed to reconcile")
+		}
+	}
+	return reconcile.Result{}, nil
+}
+
+type PodFilter func([]v1.Pod) []v1.Pod
+
+var PodNetworkFilter PodFilter = func(pods []v1.Pod) []v1.Pod {
+	var filtered []v1.Pod
+	for _, pod := range pods {
+		if !pod.Spec.HostNetwork {
+			filtered = append(filtered, pod)
+		}
+	}
+	return filtered
+}
+
+func (p *PodWatcher) PodNotifierFunc(f PodFilter, listeners ...podListener) reconcile.Func {
+	return func(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
+		podList := &v1.PodList{}
+		if err := p.cli.List(ctx, podList, p.listOpt); err != nil {
+			return reconcile.Result{}, errors.Wrap(err, "failed to list pods")
+		}
+		pods := podList.Items
+		if f != nil {
+			pods = f(pods)
+		}
+		for _, l := range listeners {
+			l.Update(pods)
+		}
+		return reconcile.Result{}, nil
+	}
+}
+
+// SetupWithManager Sets up the reconciler with a new manager, filtering using NodeNetworkConfigFilter on nodeName.
+func (p *PodWatcher) SetupWithManager(mgr ctrl.Manager) error {
+	p.cli = mgr.GetClient()
+	err := ctrl.NewControllerManagedBy(mgr).
+		For(&v1.Pod{}).
+		WithEventFilter(predicate.Funcs{ // we only want create/delete events
+			UpdateFunc: func(event.UpdateEvent) bool {
+				return false
+			},
+			GenericFunc: func(event.GenericEvent) bool {
+				return false
+			},
+		}).
+		Complete(p)
+	if err != nil {
+		return errors.Wrap(err, "failed to set up pod watcher with manager")
+	}
+	return nil
+}

cns/service/main.go

@@ -32,6 +32,7 @@ import (
 	"github.com/Azure/azure-container-networking/cns/ipampool"
 	cssctrl "github.com/Azure/azure-container-networking/cns/kubecontroller/clustersubnetstate"
 	nncctrl "github.com/Azure/azure-container-networking/cns/kubecontroller/nodenetworkconfig"
+	"github.com/Azure/azure-container-networking/cns/kubecontroller/podwatcher"
 	"github.com/Azure/azure-container-networking/cns/logger"
 	"github.com/Azure/azure-container-networking/cns/multitenantcontroller"
 	"github.com/Azure/azure-container-networking/cns/multitenantcontroller/multitenantoperator"
@@ -54,6 +55,7 @@ import (
 	"github.com/avast/retry-go/v3"
 	"github.com/pkg/errors"
 	"go.uber.org/zap"
+	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/fields"
@@ -1189,6 +1191,9 @@ func InitializeCRDState(ctx context.Context, httpRestService cns.HTTPService, cn
 			&v1alpha.NodeNetworkConfig{}: {
 				Field: fields.SelectorFromSet(fields.Set{"metadata.name": nodeName}),
 			},
+			&corev1.Pod{}: {
+				Field: fields.SelectorFromSet(fields.Set{"spec.nodeName": nodeName}),
+			},
 		},
 	})
 
@@ -1259,6 +1264,14 @@ func InitializeCRDState(ctx context.Context, httpRestService cns.HTTPService, cn
 		}
 	}
 
+	// TODO: add pod listeners based on Swift V1 vs MT/V2 configuration
+	if cnsconfig.WatchPods {
+		pw := podwatcher.New(nodeName)
+		if err := pw.SetupWithManager(manager); err != nil {
+			return errors.Wrapf(err, "failed to setup pod watcher with manager")
+		}
+	}
+
 	// adding some routes to the root service mux
 	mux := httpRestServiceImplementation.Listener.GetMux()
 	mux.Handle("/readyz", http.StripPrefix("/readyz", &healthz.Handler{}))