Skip to content

Commit 467e7a2

Browse files
kmabdamaiqueb
authored andcommitted
controller: reconcile pods on startup
ensure that we don't miss updated made to pods' networks before the controller get started Signed-off-by: Abdallah Chatila <[email protected]>
1 parent a09c185 commit 467e7a2

File tree

2 files changed

+52
-8
lines changed

2 files changed

+52
-8
lines changed

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

Lines changed: 10 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -121,10 +121,18 @@ func listenOnCoLocatedNode() v1coreinformerfactory.SharedInformerOption {
121121
return v1coreinformerfactory.WithTweakListOptions(
122122
func(options *v1.ListOptions) {
123123
const (
124-
filterKey = "spec.nodeName"
125124
nodeNameEnvVariable = "NODE_NAME"
126125
)
127-
options.FieldSelector = fields.OneTermEqualSelector(filterKey, os.Getenv(nodeNameEnvVariable)).String()
126+
// The selector for the pods that this controller instance will watch/reconcile
127+
selectorSet := fields.Set{
128+
// select pods scheduled only on the node on which this controller instance is running
129+
"spec.nodeName": os.Getenv(nodeNameEnvVariable),
130+
// select pods with a phase Running to avoid interfering with the cni-plugin works
131+
// when pods got created/deleted
132+
// see https://kubernetes.io/docs/concepts/workloads/pods/pod-lifecycle/#pod-phase
133+
"status.phase": string(corev1.PodRunning),
134+
}
135+
options.FieldSelector = fields.SelectorFromSet(selectorSet).String()
128136
})
129137
}
130138

pkg/controller/pod.go

Lines changed: 42 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@ import (
77
"time"
88

99
corev1 "k8s.io/api/core/v1"
10+
"k8s.io/apimachinery/pkg/labels"
1011
"k8s.io/apimachinery/pkg/runtime"
1112
"k8s.io/apimachinery/pkg/util/wait"
1213
v1coreinformerfactory "k8s.io/client-go/informers"
@@ -119,6 +120,13 @@ func (pnc *PodNetworksController) Start(stopChan <-chan struct{}) {
119120

120121
if ok := cache.WaitForCacheSync(stopChan, pnc.arePodsSynched, pnc.areNetAttachDefsSynched); !ok {
121122
klog.Infof("failed waiting for caches to sync")
123+
return
124+
}
125+
126+
// ensure that we didn't miss any updates before the cache sync completion
127+
if err := pnc.reconcileOnStartup(); err != nil {
128+
klog.Infof("failed to reconcile pods on startup: %v", err)
129+
return
122130
}
123131

124132
go wait.Until(pnc.worker, time.Second, stopChan)
@@ -131,6 +139,39 @@ func (pnc *PodNetworksController) worker() {
131139
}
132140
}
133141

142+
func (pnc *PodNetworksController) ignoreHostNetworkedPods(pod *corev1.Pod) bool {
143+
// since there is no such "not has" relation in a field selector,
144+
// filter out pods that are of no concern to the controller here
145+
if pod.Spec.HostNetwork {
146+
_, haveNetworkAttachments := pod.GetAnnotations()[nadv1.NetworkAttachmentAnnot]
147+
namespacedName := annotations.NamespacedName(pod.GetNamespace(), pod.GetName())
148+
if haveNetworkAttachments {
149+
klog.Warningf("rejecting to add interfaces for host networked pod: %s", namespacedName)
150+
pnc.Eventf(pod, corev1.EventTypeWarning, "InterfaceAddRejected", rejectInterfaceAddEventFormat(pod))
151+
} else {
152+
klog.V(logging.Debug).Infof("host networked pod [%s] got filtered out", namespacedName)
153+
}
154+
return true
155+
}
156+
return false
157+
}
158+
159+
func (pnc *PodNetworksController) reconcileOnStartup() error {
160+
pods, err := pnc.podsLister.List(labels.Everything())
161+
if err != nil {
162+
return fmt.Errorf("failed to list pods on current node: %v", err)
163+
}
164+
for _, pod := range pods {
165+
if pnc.ignoreHostNetworkedPods(pod) {
166+
continue
167+
}
168+
namespacedName := annotations.NamespacedName(pod.GetNamespace(), pod.GetName())
169+
klog.V(logging.Debug).Infof("pod [%s] added to reconcile on startup", namespacedName)
170+
pnc.workqueue.Add(&namespacedName)
171+
}
172+
return nil
173+
}
174+
134175
func (pnc *PodNetworksController) processNextWorkItem() bool {
135176
queueItem, shouldQuit := pnc.workqueue.Get()
136177
if shouldQuit {
@@ -243,12 +284,7 @@ func (pnc *PodNetworksController) handlePodUpdate(oldObj interface{}, newObj int
243284
oldPod := oldObj.(*corev1.Pod)
244285
newPod := newObj.(*corev1.Pod)
245286

246-
if newPod.Spec.HostNetwork {
247-
klog.Warningf(
248-
"rejecting to add interfaces for host networked pod: %s",
249-
annotations.NamespacedName(newPod.GetNamespace(), newPod.GetName()),
250-
)
251-
pnc.Eventf(newPod, corev1.EventTypeWarning, "InterfaceAddRejected", rejectInterfaceAddEventFormat(newPod))
287+
if pnc.ignoreHostNetworkedPods(newPod) {
252288
return
253289
}
254290
if !didNetworkSelectionElementsChange(oldPod, newPod) {

0 commit comments

Comments
 (0)