Use server-side filtering of events when looking for abnormalities

pkg/controller/directvolumemigration/rsync.go

@@ -144,8 +144,7 @@ func (t *Task) areRsyncTransferPodsRunning() (bool, error) {
 				migevent.LogAbnormalEventsForResource(
 					destClient, t.Log,
 					"Found abnormal event for Rsync transfer Pod on destination cluster",
-					types.NamespacedName{Namespace: pod.Namespace, Name: pod.Name},
-					"pod")
+					types.NamespacedName{Namespace: pod.Namespace, Name: pod.Name}, "Pod")
 
 				for _, podCond := range pod.Status.Conditions {
 					if podCond.Reason == corev1.PodReasonUnschedulable {
@@ -768,8 +767,7 @@ func (t *Task) areRsyncRoutesAdmitted() (bool, []string, error) {
 		migevent.LogAbnormalEventsForResource(
 			destClient, t.Log,
 			"Found abnormal event for Rsync Route on destination cluster",
-			types.NamespacedName{Namespace: route.Namespace, Name: route.Name},
-			"route")
+			types.NamespacedName{Namespace: route.Namespace, Name: route.Name}, "Route")
 
 		admitted := false
 		message := "no status condition available for the route"

pkg/controller/directvolumemigration/stunnel.go

@@ -618,7 +618,7 @@ func (t *Task) areStunnelClientPodsRunning() (bool, error) {
 					srcClient, t.Log,
 					"Found abnormal event for Stunnel Client Pod on source cluster",
 					types.NamespacedName{Namespace: pod.Namespace, Name: pod.Name},
-					"pod")
+					"Pod")
 
 				for _, podCond := range pod.Status.Conditions {
 					if podCond.Reason == corev1.PodReasonUnschedulable {

pkg/controller/migmigration/hooks.go

@@ -159,8 +159,7 @@ func (t *Task) ensureJob(job *batchv1.Job, hook migapi.MigPlanHook, migHook miga
 		migevent.LogAbnormalEventsForResource(
 			client, t.Log,
 			"Found abnormal event for Hook Job",
-			types.NamespacedName{Namespace: runningJob.Namespace, Name: runningJob.Name},
-			"job")
+			types.NamespacedName{Namespace: runningJob.Namespace, Name: runningJob.Name}, "Job")
 	}
 
 	if runningJob == nil && err == nil {

pkg/controller/migmigration/pod.go

@@ -152,8 +152,7 @@ func (t *Task) haveResticPodsStarted() (bool, error) {
 		migevent.LogAbnormalEventsForResource(
 			client, t.Log,
 			"Found abnormal event for Restic Pod",
-			types.NamespacedName{Namespace: pod.Namespace, Name: pod.Name},
-			"pod")
+			types.NamespacedName{Namespace: pod.Namespace, Name: pod.Name}, "Pod")
 
 		if pod.DeletionTimestamp != nil {
 			t.Log.Info("Deletion timestamp found on Restic Pod, "+
@@ -277,8 +276,7 @@ func (t *Task) haveVeleroPodsStarted() (bool, error) {
 			migevent.LogAbnormalEventsForResource(
 				client, t.Log,
 				"Found abnormal event for Velero Pod",
-				types.NamespacedName{Namespace: pod.Namespace, Name: pod.Name},
-				"pod")
+				types.NamespacedName{Namespace: pod.Namespace, Name: pod.Name}, "Pod")
 
 			if pod.DeletionTimestamp != nil {
 				t.Log.Info("Found Velero Pod with deletion timestamp."+

pkg/controller/migmigration/stage.go

@@ -523,8 +523,7 @@ func (t *Task) stagePodReport(client k8sclient.Client) (report PodStartReport, e
 		migevent.LogAbnormalEventsForResource(
 			client, t.Log,
 			"Found abnormal event for Stage Pod",
-			types.NamespacedName{Namespace: pod.Namespace, Name: pod.Name},
-			"pod")
+			types.NamespacedName{Namespace: pod.Namespace, Name: pod.Name}, "Pod")
 
 		initReady := true
 		for _, c := range pod.Status.InitContainerStatuses {

pkg/controller/migmigration/validation.go

@@ -300,13 +300,12 @@ func ensureRegistryHealth(c k8sclient.Client, migration *migapi.MigMigration) (i
 			return nEnsured, "", liberr.Wrap(err)
 		}
 
-		for _, registryPod := range registryPods.Items {
+		for _, pod := range registryPods.Items {
 			// Logs abnormal events for Registry Pods if any are found
 			migevent.LogAbnormalEventsForResource(
 				client, log,
 				"Found abnormal event for Registry Pod",
-				types.NamespacedName{Namespace: registryPod.Namespace, Name: registryPod.Name},
-				"pod")
+				types.NamespacedName{Namespace: pod.Namespace, Name: pod.Name}, "Pod")
 		}
 
 		registryPodCount := len(registryPods.Items)

pkg/event/event.go

@@ -2,8 +2,8 @@ package event
 
 import (
 	"context"
+	"fmt"
 	"path"
-	"strings"
 
 	"github.com/go-logr/logr"
 	corev1 "k8s.io/api/core/v1"
@@ -15,29 +15,24 @@ import (
 // GetAbnormalEventsForResource gets unique events of non-normal type for
 // a namespaced resource. Useful for logging the most relevant events
 // related to a resource we're waiting on.
+// Make sure to pass in the kind exactly as it is capitalized on the event, e.g. "Pod"
 func GetAbnormalEventsForResource(client client.Client,
 	nsName types.NamespacedName, resourceKind string) ([]corev1.Event, error) {
 	uniqueEventMap := make(map[string]corev1.Event)
 
 	eList := corev1.EventList{}
 	options := k8sclient.InNamespace(nsName.Namespace)
-	err := client.List(context.TODO(), options, &eList)
+	fieldSelector := fmt.Sprintf("involvedObject.name=%s,involvedObject.kind=%s,type!=Normal",
+		nsName.Name, resourceKind)
+	err := options.SetFieldSelector(fieldSelector)
+	if err != nil {
+		return nil, fmt.Errorf("field selector construction failed: fieldSelector=[%v]", fieldSelector)
+	}
+	err = client.List(context.TODO(), options, &eList)
 	if err != nil {
 		return nil, err
 	}
 	for _, event := range eList.Items {
-		// Only want events for the kind indicated
-		if strings.ToLower(event.InvolvedObject.Kind) != strings.ToLower(resourceKind) {
-			continue
-		}
-		// Only get events for the resource.name we're interested in
-		if event.InvolvedObject.Name != nsName.Name {
-			continue
-		}
-		// Only get abnormal events
-		if event.Type == "Normal" {
-			continue
-		}
 		// Check if same event reason has already been seen
 		eventFromMap, ok := uniqueEventMap[event.Reason]
 		if !ok {