chore: merge role probe event handling (#10282)

Author: leon-ape

cmd/manager/main.go

@@ -433,7 +433,22 @@ func main() {
 		os.Exit(1)
 	}
 
-	if viper.GetBool(appsFlagKey.viperName()) {
+	appsEnabled := viper.GetBool(appsFlagKey.viperName())
+	workloadsEnabled := viper.GetBool(workloadsFlagKey.viperName())
+	if appsEnabled || workloadsEnabled {
+		if err = (&k8scorecontrollers.EventReconciler{
+			Client:           mgr.GetClient(),
+			Scheme:           mgr.GetScheme(),
+			Recorder:         mgr.GetEventRecorderFor("event-controller"),
+			AppsEnabled:      appsEnabled,
+			WorkloadsEnabled: workloadsEnabled,
+		}).SetupWithManager(mgr); err != nil {
+			setupLog.Error(err, "unable to create controller", "controller", "Event")
+			os.Exit(1)
+		}
+	}
+
+	if appsEnabled {
 		if err = (&appscontrollers.ClusterDefinitionReconciler{
 			Client:   mgr.GetClient(),
 			Scheme:   mgr.GetScheme(),
@@ -507,15 +522,6 @@ func main() {
 			os.Exit(1)
 		}
 
-		if err = (&k8scorecontrollers.EventReconciler{
-			Client:   mgr.GetClient(),
-			Scheme:   mgr.GetScheme(),
-			Recorder: mgr.GetEventRecorderFor("event-controller"),
-		}).SetupWithManager(mgr); err != nil {
-			setupLog.Error(err, "unable to create controller", "controller", "Event")
-			os.Exit(1)
-		}
-
 		if err = (&rollout.RolloutReconciler{
 			Client:   mgr.GetClient(),
 			Scheme:   mgr.GetScheme(),
@@ -526,7 +532,7 @@ func main() {
 		}
 	}
 
-	if viper.GetBool(workloadsFlagKey.viperName()) {
+	if workloadsEnabled {
 		if err = (&workloadscontrollers.InstanceSetReconciler{
 			Client:   mgr.GetClient(),
 			Scheme:   mgr.GetScheme(),
@@ -554,14 +560,6 @@ func main() {
 			os.Exit(1)
 		}
 
-		if err = (&workloadscontrollers.InstanceEventReconciler{
-			Client:   mgr.GetClient(),
-			Scheme:   mgr.GetScheme(),
-			Recorder: mgr.GetEventRecorderFor("instance-event-controller"),
-		}).SetupWithManager(mgr); err != nil {
-			setupLog.Error(err, "unable to create controller", "controller", "InstanceEvent")
-			os.Exit(1)
-		}
 	}
 
 	if viper.GetBool(operationsFlagKey.viperName()) {

controllers/apps/component/suite_test.go

@@ -183,9 +183,10 @@ var _ = BeforeSuite(func() {
 	Expect(err).ToNot(HaveOccurred())
 
 	err = (&k8score.EventReconciler{
-		Client:   k8sManager.GetClient(),
-		Scheme:   k8sManager.GetScheme(),
-		Recorder: k8sManager.GetEventRecorderFor("event-controller"),
+		Client:      k8sManager.GetClient(),
+		Scheme:      k8sManager.GetScheme(),
+		Recorder:    k8sManager.GetEventRecorderFor("event-controller"),
+		AppsEnabled: true,
 	}).SetupWithManager(k8sManager)
 	Expect(err).ToNot(HaveOccurred())
 

controllers/k8score/event_controller.go

@@ -32,9 +32,9 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/controller"
 	"sigs.k8s.io/controller-runtime/pkg/log"
 
+	"github.com/apecloud/kubeblocks/controllers/workloads"
 	"github.com/apecloud/kubeblocks/pkg/constant"
 	"github.com/apecloud/kubeblocks/pkg/controller/component"
-	"github.com/apecloud/kubeblocks/pkg/controller/instanceset"
 	intctrlutil "github.com/apecloud/kubeblocks/pkg/controllerutil"
 	viper "github.com/apecloud/kubeblocks/pkg/viperx"
 )
@@ -44,14 +44,16 @@ const (
 )
 
 type eventHandler interface {
-	Handle(cli client.Client, reqCtx intctrlutil.RequestCtx, recorder record.EventRecorder, event *corev1.Event) error
+	Handle(cli client.Client, reqCtx intctrlutil.RequestCtx, recorder record.EventRecorder, event *corev1.Event) (bool, error)
 }
 
 // EventReconciler reconciles an Event object
 type EventReconciler struct {
 	client.Client
-	Scheme   *runtime.Scheme
-	Recorder record.EventRecorder
+	Scheme           *runtime.Scheme
+	Recorder         record.EventRecorder
+	AppsEnabled      bool
+	WorkloadsEnabled bool
 }
 
 // events API only allows ready-only, create, patch
@@ -80,15 +82,17 @@ func (r *EventReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl
 		return intctrlutil.Reconciled()
 	}
 
-	handlers := []eventHandler{
-		&instanceset.PodRoleEventHandler{},
-		&component.AvailableEventHandler{},
-		&component.KBAgentTaskEventHandler{},
-	}
-	for _, handler := range handlers {
-		if err := handler.Handle(r.Client, reqCtx, r.Recorder, event); err != nil && !apierrors.IsNotFound(err) {
+	handled := false
+	for _, handler := range r.handlers() {
+		ok, err := handler.Handle(r.Client, reqCtx, r.Recorder, event)
+		if err != nil && !apierrors.IsNotFound(err) {
 			return intctrlutil.RequeueWithError(err, reqCtx.Log, "handleEventError")
 		}
+		handled = handled || ok
+	}
+
+	if !handled {
+		return intctrlutil.Reconciled()
 	}
 
 	if err := r.eventHandled(ctx, event); err != nil {
@@ -107,6 +111,20 @@ func (r *EventReconciler) SetupWithManager(mgr ctrl.Manager) error {
 		Complete(r)
 }
 
+func (r *EventReconciler) handlers() []eventHandler {
+	handlers := make([]eventHandler, 0, 3)
+	if r.AppsEnabled {
+		handlers = append(handlers,
+			&component.AvailableEventHandler{},
+			&component.KBAgentTaskEventHandler{},
+		)
+	}
+	if r.WorkloadsEnabled {
+		handlers = append(handlers, &workloads.RoleEventHandler{})
+	}
+	return handlers
+}
+
 func (r *EventReconciler) isEventHandled(event *corev1.Event) bool {
 	count := fmt.Sprintf("%d", event.Count)
 	annotations := event.GetAnnotations()

controllers/k8score/event_controller_test.go

@@ -22,8 +22,6 @@ package k8score
 import (
 	"encoding/json"
 	"fmt"
-	"strconv"
-	"time"
 
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
@@ -33,45 +31,25 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
-	appsv1 "github.com/apecloud/kubeblocks/apis/apps/v1"
-	workloads "github.com/apecloud/kubeblocks/apis/workloads/v1"
-	"github.com/apecloud/kubeblocks/pkg/constant"
 	"github.com/apecloud/kubeblocks/pkg/controller/builder"
-	"github.com/apecloud/kubeblocks/pkg/controller/instanceset"
 	"github.com/apecloud/kubeblocks/pkg/generics"
 	"github.com/apecloud/kubeblocks/pkg/kbagent/proto"
 	testapps "github.com/apecloud/kubeblocks/pkg/testutil/apps"
 )
 
 var _ = Describe("Event Controller", func() {
 	cleanEnv := func() {
-		// must wait till resources deleted and no longer existed before the testcases start,
-		// otherwise if later it needs to create some new resource objects with the same name,
-		// in race conditions, it will find the existence of old objects, resulting failure to
-		// create the new objects.
 		By("clean resources")
 
 		testapps.ClearClusterResources(&testCtx)
 
-		// delete rest mocked objects
 		inNS := client.InNamespace(testCtx.DefaultNamespace)
 		ml := client.HasLabels{testCtx.TestObjLabelKey}
-		// namespaced
 		testapps.ClearResources(&testCtx, generics.EventSignature, inNS, ml)
 		testapps.ClearResources(&testCtx, generics.PodSignature, inNS, ml)
 	}
 
-	const (
-		// roleChangedAnnotKey is used to mark the role change event has been handled.
-		roleChangedAnnotKey = "role.kubeblocks.io/event-handled"
-	)
-
-	var (
-		beforeLastTS = time.Date(2021, time.January, 1, 12, 0, 0, 0, time.UTC)
-		initLastTS   = time.Date(2022, time.January, 1, 12, 0, 0, 0, time.UTC)
-		afterLastTS  = time.Date(2023, time.January, 1, 12, 0, 0, 0, time.UTC)
-		eventSeq     = 0
-	)
+	var eventSeq int
 
 	createRoleChangedEvent := func(podName, role string, podUid types.UID) *corev1.Event {
 		eventSeq++
@@ -95,157 +73,30 @@ var _ = Describe("Event Controller", func() {
 			SetMessage(string(message)).
 			SetReason("roleProbe").
 			SetType(corev1.EventTypeNormal).
-			SetFirstTimestamp(metav1.NewTime(initLastTS)).
-			SetLastTimestamp(metav1.NewTime(initLastTS)).
-			SetEventTime(metav1.NewMicroTime(initLastTS)).
+			SetEventTime(metav1.NewMicroTime(metav1.Now().Time)).
 			SetReportingController(proto.ProbeEventReportingController).
 			SetReportingInstance(podName).
 			SetAction("mock-create-event-action").
 			GetObject()
 	}
 
-	createInvolvedPod := func(name, clusterName, componentName, itsName string) *corev1.Pod {
-		return builder.NewPodBuilder(testCtx.DefaultNamespace, name).
-			AddLabels(constant.AppInstanceLabelKey, clusterName).
-			AddLabels(constant.KBAppComponentLabelKey, componentName).
-			AddLabels(instanceset.WorkloadsInstanceLabelKey, itsName).
-			SetContainers([]corev1.Container{
-				{
-					Image: "foo",
-					Name:  "bar",
-				},
-			}).
-			GetObject()
-	}
-
 	BeforeEach(cleanEnv)
 
 	AfterEach(cleanEnv)
 
-	Context("When receiving role changed event", func() {
-		It("should handle it properly", func() {
-			By("create cluster & compdef")
-			compDefName := "test-compdef"
-			clusterName := "test-cluster"
-			defaultCompName := "mysql"
-			compDefObj := testapps.NewComponentDefinitionFactory(compDefName).
-				SetDefaultSpec().
-				Create(&testCtx).
-				GetObject()
-			clusterObj := testapps.NewClusterFactory(testCtx.DefaultNamespace, clusterName, "").
-				WithRandomName().
-				AddComponent(defaultCompName, compDefObj.GetName()).
-				Create(&testCtx).GetObject()
-			Eventually(testapps.CheckObjExists(&testCtx, client.ObjectKeyFromObject(clusterObj), &appsv1.Cluster{}, true)).Should(Succeed())
-
-			itsName := fmt.Sprintf("%s-%s", clusterObj.Name, defaultCompName)
-			its := testapps.NewInstanceSetFactory(clusterObj.Namespace, itsName, clusterObj.Name, defaultCompName).
-				SetReplicas(int32(3)).
-				AddContainer(corev1.Container{Name: testapps.DefaultMySQLContainerName, Image: testapps.ApeCloudMySQLImage}).
-				Create(&testCtx).GetObject()
-			Expect(testapps.GetAndChangeObj(&testCtx, client.ObjectKeyFromObject(its), func(tmpITS *workloads.InstanceSet) {
-				tmpITS.Spec.Roles = []workloads.ReplicaRole{
-					{
-						Name:                 "leader",
-						ParticipatesInQuorum: true,
-						UpdatePriority:       5,
-					},
-					{
-						Name:                 "follower",
-						ParticipatesInQuorum: true,
-						UpdatePriority:       4,
-					},
-				}
-			})()).Should(Succeed())
-			By("create involved pod")
-			var uid types.UID
-			podName := fmt.Sprintf("%s-%d", itsName, 0)
-			pod := createInvolvedPod(podName, clusterObj.Name, defaultCompName, itsName)
-			Expect(testCtx.CreateObj(ctx, pod)).Should(Succeed())
-			Eventually(func() error {
-				p := &corev1.Pod{}
-				defer func() {
-					uid = p.UID
-				}()
-				return k8sClient.Get(ctx, types.NamespacedName{
-					Namespace: pod.Namespace,
-					Name:      pod.Name,
-				}, p)
-			}).Should(Succeed())
-			Expect(uid).ShouldNot(BeNil())
-
-			By("send role changed event")
-			role := "leader"
-			sndEvent := createRoleChangedEvent(podName, role, uid)
+	Context("When receiving role probe event", func() {
+		It("should leave role probe events for the workloads role event reconciler", func() {
+			sndEvent := createRoleChangedEvent("pod-0", "leader", types.UID("pod-uid"))
 			Expect(testCtx.CreateObj(ctx, sndEvent)).Should(Succeed())
-			Eventually(func() string {
+
+			Consistently(func(g Gomega) {
 				event := &corev1.Event{}
-				if err := k8sClient.Get(ctx, types.NamespacedName{
+				g.Expect(k8sClient.Get(ctx, types.NamespacedName{
 					Namespace: sndEvent.Namespace,
 					Name:      sndEvent.Name,
-				}, event); err != nil {
-					return err.Error()
-				}
-				return event.InvolvedObject.Name
-			}).Should(Equal(sndEvent.InvolvedObject.Name))
-
-			Eventually(testapps.CheckObj(&testCtx, client.ObjectKeyFromObject(pod), func(g Gomega, p *corev1.Pod) {
-				g.Expect(p).ShouldNot(BeNil())
-				g.Expect(p.Labels).ShouldNot(BeNil())
-				g.Expect(p.Labels[constant.RoleLabelKey]).Should(Equal(role))
-				g.Expect(p.Annotations[constant.LastRoleEventVersionAnnotationKey]).Should(Equal(strconv.FormatInt(sndEvent.EventTime.UnixMicro(), 10)))
-			})).Should(Succeed())
-
-			Eventually(testapps.CheckObj(&testCtx, client.ObjectKeyFromObject(sndEvent), func(g Gomega, e *corev1.Event) {
-				g.Expect(e).ShouldNot(BeNil())
-				g.Expect(e.Annotations).ShouldNot(BeNil())
-				g.Expect(e.Annotations[roleChangedAnnotKey]).Should(Equal("count-0"))
-			})).Should(Succeed())
-
-			By("send role changed event with beforeLastTS earlier than pod last role changes event timestamp annotation should not be update successfully")
-			role = "follower"
-			sndInvalidEvent := createRoleChangedEvent(podName, role, uid)
-			sndInvalidEvent.EventTime = metav1.NewMicroTime(beforeLastTS)
-			Expect(testCtx.CreateObj(ctx, sndInvalidEvent)).Should(Succeed())
-			Eventually(func() string {
-				event := &corev1.Event{}
-				if err := k8sClient.Get(ctx, types.NamespacedName{
-					Namespace: sndInvalidEvent.Namespace,
-					Name:      sndInvalidEvent.Name,
-				}, event); err != nil {
-					return err.Error()
-				}
-				return event.InvolvedObject.Name
-			}).Should(Equal(sndInvalidEvent.InvolvedObject.Name))
-			Eventually(testapps.CheckObj(&testCtx, client.ObjectKeyFromObject(pod), func(g Gomega, p *corev1.Pod) {
-				g.Expect(p).ShouldNot(BeNil())
-				g.Expect(p.Labels).ShouldNot(BeNil())
-				g.Expect(p.Labels[constant.RoleLabelKey]).ShouldNot(Equal(role))
-				g.Expect(p.Annotations[constant.LastRoleEventVersionAnnotationKey]).ShouldNot(Equal(strconv.FormatInt(sndInvalidEvent.EventTime.UnixMicro(), 10)))
-			})).Should(Succeed())
-
-			By("send role changed event with afterLastTS later than pod last role changes event timestamp annotation should be update successfully")
-			role = "follower"
-			sndValidEvent := createRoleChangedEvent(podName, role, uid)
-			sndValidEvent.LastTimestamp = metav1.NewTime(afterLastTS)
-			sndValidEvent.EventTime = metav1.NewMicroTime(afterLastTS)
-			Expect(testCtx.CreateObj(ctx, sndValidEvent)).Should(Succeed())
-			Eventually(func() string {
-				event := &corev1.Event{}
-				if err := k8sClient.Get(ctx, types.NamespacedName{
-					Namespace: sndValidEvent.Namespace,
-					Name:      sndValidEvent.Name,
-				}, event); err != nil {
-					return err.Error()
-				}
-				return event.InvolvedObject.Name
-			}).Should(Equal(sndValidEvent.InvolvedObject.Name))
-			Eventually(testapps.CheckObj(&testCtx, client.ObjectKeyFromObject(pod), func(g Gomega, p *corev1.Pod) {
-				g.Expect(p).ShouldNot(BeNil())
-				g.Expect(p.Labels).ShouldNot(BeNil())
-				g.Expect(p.Labels[constant.RoleLabelKey]).Should(Equal(role))
-				g.Expect(p.Annotations[constant.LastRoleEventVersionAnnotationKey]).Should(Equal(strconv.FormatInt(sndValidEvent.LastTimestamp.UnixMicro(), 10)))
-			})).Should(Succeed())
+				}, event)).Should(Succeed())
+				g.Expect(event.Annotations).ShouldNot(HaveKey(eventHandledAnnotationKey))
+			}).Should(Succeed())
 		})
 	})
 })