controller: reconcile pods on startup

cmd/dynamic-networks-controller/networks-controller.go

@@ -121,10 +121,18 @@ func listenOnCoLocatedNode() v1coreinformerfactory.SharedInformerOption {
 	return v1coreinformerfactory.WithTweakListOptions(
 		func(options *v1.ListOptions) {
 			const (
-				filterKey           = "spec.nodeName"
 				nodeNameEnvVariable = "NODE_NAME"
 			)
-			options.FieldSelector = fields.OneTermEqualSelector(filterKey, os.Getenv(nodeNameEnvVariable)).String()
+			// The selector for the pods that this controller instance will watch/reconcile
+			selectorSet := fields.Set{
+				// select pods scheduled only on the node on which this controller instance is running
+				"spec.nodeName": os.Getenv(nodeNameEnvVariable),
+				// select pods with a phase Running to avoid interfering with the cni-plugin works
+				// when pods got created/deleted
+				// see https://kubernetes.io/docs/concepts/workloads/pods/pod-lifecycle/#pod-phase
+				"status.phase": string(corev1.PodRunning),
+			}
+			options.FieldSelector = fields.SelectorFromSet(selectorSet).String()
 		})
 }
 

pkg/controller/pod.go

@@ -7,6 +7,7 @@ import (
 	"time"
 
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/util/wait"
 	v1coreinformerfactory "k8s.io/client-go/informers"
@@ -119,6 +120,13 @@ func (pnc *PodNetworksController) Start(stopChan <-chan struct{}) {
 
 	if ok := cache.WaitForCacheSync(stopChan, pnc.arePodsSynched, pnc.areNetAttachDefsSynched); !ok {
 		klog.Infof("failed waiting for caches to sync")
+		return
+	}
+
+	// ensure that we didn't miss any updates before the cache sync completion
+	if err := pnc.reconcileOnStartup(); err != nil {
+		klog.Infof("failed to reconcile pods on startup: %v", err)
+		return
 	}
 
 	go wait.Until(pnc.worker, time.Second, stopChan)
@@ -131,6 +139,39 @@ func (pnc *PodNetworksController) worker() {
 	}
 }
 
+func (pnc *PodNetworksController) ignoreHostNetworkedPods(pod *corev1.Pod) bool {
+	// since there is no such "not has" relation in a field selector,
+	// filter out pods that are of no concern to the controller here
+	if pod.Spec.HostNetwork {
+		_, haveNetworkAttachments := pod.GetAnnotations()[nadv1.NetworkAttachmentAnnot]
+		namespacedName := annotations.NamespacedName(pod.GetNamespace(), pod.GetName())
+		if haveNetworkAttachments {
+			klog.Warningf("rejecting to add interfaces for host networked pod: %s", namespacedName)
+			pnc.Eventf(pod, corev1.EventTypeWarning, "InterfaceAddRejected", rejectInterfaceAddEventFormat(pod))
+		} else {
+			klog.V(logging.Debug).Infof("host networked pod [%s] got filtered out", namespacedName)
+		}
+		return true
+	}
+	return false
+}
+
+func (pnc *PodNetworksController) reconcileOnStartup() error {
+	pods, err := pnc.podsLister.List(labels.Everything())
+	if err != nil {
+		return fmt.Errorf("failed to list pods on current node: %v", err)
+	}
+	for _, pod := range pods {
+		if pnc.ignoreHostNetworkedPods(pod) {
+			continue
+		}
+		namespacedName := annotations.NamespacedName(pod.GetNamespace(), pod.GetName())
+		klog.V(logging.Debug).Infof("pod [%s] added to reconcile on startup", namespacedName)
+		pnc.workqueue.Add(&namespacedName)
+	}
+	return nil
+}
+
 func (pnc *PodNetworksController) processNextWorkItem() bool {
 	queueItem, shouldQuit := pnc.workqueue.Get()
 	if shouldQuit {
@@ -243,12 +284,7 @@ func (pnc *PodNetworksController) handlePodUpdate(oldObj interface{}, newObj int
 	oldPod := oldObj.(*corev1.Pod)
 	newPod := newObj.(*corev1.Pod)
 
-	if newPod.Spec.HostNetwork {
-		klog.Warningf(
-			"rejecting to add interfaces for host networked pod: %s",
-			annotations.NamespacedName(newPod.GetNamespace(), newPod.GetName()),
-		)
-		pnc.Eventf(newPod, corev1.EventTypeWarning, "InterfaceAddRejected", rejectInterfaceAddEventFormat(newPod))
+	if pnc.ignoreHostNetworkedPods(newPod) {
 		return
 	}
 	if !didNetworkSelectionElementsChange(oldPod, newPod) {