Sync groups members for pods that turns into terminated status (#7217) (#7247)

Once a Pod is updated to terminated state (succeeded or failed), the selectorItems
in GroupEntityIndex need to be notified so that they are excluded from any Network
Policy computations in appliedTo or address groups.

Signed-off-by: Dyanngg <[email protected]>

Author: Dyanngg

pkg/controller/grouping/controller.go

@@ -207,14 +207,14 @@ func (c *GroupEntityController) Run(stopCh <-chan struct{}) {
 
 func (c *GroupEntityController) addPod(obj interface{}) {
 	pod := obj.(*v1.Pod)
-	klog.V(2).Infof("Processing Pod %s/%s ADD event, labels: %v", pod.Namespace, pod.Name, pod.Labels)
+	klog.V(2).InfoS("Processing Pod ADD event", "pod", klog.KObj(pod), "labels", pod.Labels)
 	c.groupEntityIndex.AddPod(pod)
 	c.podAddEvents.Increment()
 }
 
 func (c *GroupEntityController) updatePod(_, curObj interface{}) {
 	curPod := curObj.(*v1.Pod)
-	klog.V(2).Infof("Processing Pod %s/%s UPDATE event, labels: %v", curPod.Namespace, curPod.Name, curPod.Labels)
+	klog.V(2).InfoS("Processing Pod UPDATE event", "pod", klog.KObj(curPod), "labels", curPod.Labels, "phase", curPod.Status.Phase)
 	c.groupEntityIndex.AddPod(curPod)
 }
 
@@ -237,14 +237,14 @@ func (c *GroupEntityController) deletePod(old interface{}) {
 
 func (c *GroupEntityController) addNamespace(obj interface{}) {
 	namespace := obj.(*v1.Namespace)
-	klog.V(2).Infof("Processing Namespace %s ADD event, labels: %v", namespace.Name, namespace.Labels)
+	klog.V(2).InfoS("Processing Namespace ADD event", "namespace", namespace.Name, "labels", namespace.Labels)
 	c.groupEntityIndex.AddNamespace(namespace)
 	c.namespaceAddEvents.Increment()
 }
 
 func (c *GroupEntityController) updateNamespace(_, curObj interface{}) {
 	curNamespace := curObj.(*v1.Namespace)
-	klog.V(2).Infof("Processing Namespace %s UPDATE event, labels: %v", curNamespace.Name, curNamespace.Labels)
+	klog.V(2).InfoS("Processing Namespace UPDATE event", "namespace", curNamespace.Name, "labels", curNamespace.Labels)
 	c.groupEntityIndex.AddNamespace(curNamespace)
 }
 

pkg/controller/grouping/controller_test.go

@@ -89,6 +89,7 @@ func TestGroupEntityControllerRun(t *testing.T) {
 			informerFactory := informers.NewSharedInformerFactory(client, informerDefaultResync)
 			crdInformerFactory := crdinformers.NewSharedInformerFactory(crdClient, informerDefaultResync)
 			stopCh := make(chan struct{})
+			defer close(stopCh)
 
 			c := NewGroupEntityController(index, informerFactory.Core().V1().Pods(), informerFactory.Core().V1().Namespaces(), crdInformerFactory.Crd().V1alpha2().ExternalEntities())
 			assert.False(t, index.HasSynced(), "GroupEntityIndex has been synced before starting InformerFactories")

pkg/controller/grouping/group_entity_index.go

@@ -29,6 +29,7 @@ import (
 
 	"antrea.io/antrea/pkg/apis/crd/v1alpha2"
 	"antrea.io/antrea/pkg/controller/types"
+	"antrea.io/antrea/pkg/util/k8s"
 	utilsets "antrea.io/antrea/pkg/util/sets"
 )
 
@@ -744,6 +745,9 @@ func entityAttrsUpdated(oldEntity, newEntity metav1.Object) bool {
 	switch oldValue := oldEntity.(type) {
 	case *v1.Pod:
 		// For Pod, we only care about PodIP and NodeName update.
+		// Also, when a Pod is updated to terminated state, the selectorItems need to be
+		// notified so that they are excluded from any Network Policy computations in
+		// appliedTo or address groups.
 		// Some other attributes we care about are immutable, e.g. the named ContainerPort.
 		newValue := newEntity.(*v1.Pod)
 		if oldValue.Status.PodIP != newValue.Status.PodIP {
@@ -752,6 +756,9 @@ func entityAttrsUpdated(oldEntity, newEntity metav1.Object) bool {
 		if oldValue.Spec.NodeName != newValue.Spec.NodeName {
 			return true
 		}
+		if k8s.IsPodTerminated(oldValue) != k8s.IsPodTerminated(newValue) {
+			return true
+		}
 		return false
 	case *v1alpha2.ExternalEntity:
 		newValue := newEntity.(*v1alpha2.ExternalEntity)

pkg/controller/grouping/group_entity_index_test.go

@@ -472,6 +472,39 @@ func TestGroupEntityIndexEventHandlers(t *testing.T) {
 			},
 			expectedGroupsCalled: map[GroupType][]string{},
 		},
+		{
+			name:           "update an existing pod's phase to running",
+			existingPods:   []*v1.Pod{podFoo1, podBar1, podFoo1InOtherNamespace},
+			existingGroups: []*group{groupPodFooType1, groupPodFooAllNamespaceType1, groupEEFooType1, groupPodAllNamespaceType1},
+			inputEvent: func(i *GroupEntityIndex) {
+				i.AddPod(copyAndMutatePod(podFoo1, func(pod *v1.Pod) {
+					pod.Status.Phase = v1.PodRunning
+				}))
+			},
+			expectedGroupsCalled: map[GroupType][]string{},
+		},
+		{
+			name:           "update an existing pod's phase to succeeded",
+			existingPods:   []*v1.Pod{podFoo1, podBar1, podFoo1InOtherNamespace},
+			existingGroups: []*group{groupPodFooType1, groupPodFooAllNamespaceType1, groupEEFooType1, groupPodAllNamespaceType1},
+			inputEvent: func(i *GroupEntityIndex) {
+				i.AddPod(copyAndMutatePod(podFoo1, func(pod *v1.Pod) {
+					pod.Status.Phase = v1.PodSucceeded
+				}))
+			},
+			expectedGroupsCalled: map[GroupType][]string{groupType1: {groupPodFooType1.groupName, groupPodFooAllNamespaceType1.groupName, groupPodAllNamespaceType1.groupName}},
+		},
+		{
+			name:           "update an existing pod's phase to failed",
+			existingPods:   []*v1.Pod{podFoo1, podBar1, podFoo1InOtherNamespace},
+			existingGroups: []*group{groupPodFooType1, groupPodFooAllNamespaceType1, groupPodBarType1, groupPodAllNamespaceType1},
+			inputEvent: func(i *GroupEntityIndex) {
+				i.AddPod(copyAndMutatePod(podBar1, func(pod *v1.Pod) {
+					pod.Status.Phase = v1.PodFailed
+				}))
+			},
+			expectedGroupsCalled: map[GroupType][]string{groupType1: {groupPodBarType1.groupName, groupPodAllNamespaceType1.groupName}},
+		},
 		{
 			name:                     "delete an existing pod",
 			existingPods:             []*v1.Pod{podFoo1, podBar1, podFoo1InOtherNamespace},