🌱 Use base controller in sdk-go (#1251)

* Use base controller in sdk-go

We can leverage contextual logger in base controller.

Signed-off-by: Jian Qiu <jqiu@redhat.com>

* Fix integration test error

Signed-off-by: Jian Qiu <jqiu@redhat.com>

---------

Signed-off-by: Jian Qiu <jqiu@redhat.com>

.github/workflows/cloudevents-integration.yml

@@ -19,19 +19,18 @@ permissions:
   contents: read
 
 jobs:
-  # TODO enable this after mqtt integration is stable
-  # mqtt-work-integration:
-  #   name: mqtt-work-integration
-  #   runs-on: ubuntu-latest
-  #   steps:
-  #     - name: checkout code
-  #       uses: actions/checkout@v5
-  #     - name: install Go
-  #       uses: actions/setup-go@v6
-  #       with:
-  #         go-version: ${{ env.GO_VERSION }}
-  #     - name: integration
-  #       run: make test-cloudevents-work-mqtt-integration
+  mqtt-work-integration:
+    name: mqtt-work-integration
+    runs-on: ubuntu-latest
+    steps:
+      - name: checkout code
+        uses: actions/checkout@v5
+      - name: install Go
+        uses: actions/setup-go@v6
+        with:
+          go-version: ${{ env.GO_VERSION }}
+      - name: integration
+        run: make test-cloudevents-work-mqtt-integration
 
   grpc-work-integration:
     name: grpc-work-integration

go.mod

@@ -41,7 +41,7 @@ require (
 	k8s.io/utils v0.0.0-20241210054802-24370beab758
 	open-cluster-management.io/addon-framework v1.1.0
 	open-cluster-management.io/api v1.1.1-0.20251112045944-3e1bb92b69e3
-	open-cluster-management.io/sdk-go v1.1.1-0.20251117075350-a9794783fa67
+	open-cluster-management.io/sdk-go v1.1.1-0.20251120064629-eb2d8ba7fdd3
 	sigs.k8s.io/about-api v0.0.0-20250131010323-518069c31c03
 	sigs.k8s.io/cluster-inventory-api v0.0.0-20240730014211-ef0154379848
 	sigs.k8s.io/controller-runtime v0.21.0

go.sum

@@ -517,8 +517,8 @@ open-cluster-management.io/addon-framework v1.1.0 h1:GoPbg5Q9KEI+Vvgs9PUs2IjIoU/
 open-cluster-management.io/addon-framework v1.1.0/go.mod h1:KPdLM+CfUKgwVuVE9Tyu2nOuD6LgDmx94HOCnJwLIdo=
 open-cluster-management.io/api v1.1.1-0.20251112045944-3e1bb92b69e3 h1:pJl/jwiUBO0D4PrL+G6JASKC8PDpPoxItLa6cTcj8TM=
 open-cluster-management.io/api v1.1.1-0.20251112045944-3e1bb92b69e3/go.mod h1:lEc5Wkc9ON5ym/qAtIqNgrE7NW7IEOCOC611iQMlnKM=
-open-cluster-management.io/sdk-go v1.1.1-0.20251117075350-a9794783fa67 h1:G4w5+FI1VpgLLJcijm4lGwSMXev0377iJ3Jlx62VKCY=
-open-cluster-management.io/sdk-go v1.1.1-0.20251117075350-a9794783fa67/go.mod h1:oQzZFphlr1hfzRGrMa24OYCFg9ZmMTJov3mb8OLVOaM=
+open-cluster-management.io/sdk-go v1.1.1-0.20251120064629-eb2d8ba7fdd3 h1:2SsJbXoUDakJuOyNMljxzwZv1DfANjCLrBDOEiphrWM=
+open-cluster-management.io/sdk-go v1.1.1-0.20251120064629-eb2d8ba7fdd3/go.mod h1:oQzZFphlr1hfzRGrMa24OYCFg9ZmMTJov3mb8OLVOaM=
 sigs.k8s.io/about-api v0.0.0-20250131010323-518069c31c03 h1:1ShFiMjGQOR/8jTBkmZrk1gORxnvMwm1nOy2/DbHg4U=
 sigs.k8s.io/about-api v0.0.0-20250131010323-518069c31c03/go.mod h1:F1pT4mK53U6F16/zuaPSYpBaR7x5Kjym6aKJJC0/DHU=
 sigs.k8s.io/apiserver-network-proxy/konnectivity-client v0.31.2 h1:jpcvIRr3GLoUoEKRkHKSmGjxb6lWwrBlJsXc+eUYQHM=

pkg/common/queue/queuekey.go

@@ -7,7 +7,7 @@ import (
 	"k8s.io/client-go/tools/cache"
 )
 
-func FileterByLabel(key string) factory.EventFilterFunc {
+func FileterByLabel(key string) func(obj interface{}) bool {
 	return func(obj interface{}) bool {
 		accessor, _ := meta.Accessor(obj)
 		return len(accessor.GetLabels()) > 0 && len(accessor.GetLabels()[key]) > 0

pkg/common/recorder/event_recorder.go

@@ -1,4 +1,4 @@
-package helpers
+package recorder
 
 import (
 	"context"
@@ -14,7 +14,7 @@ func NewEventRecorder(ctx context.Context, scheme *runtime.Scheme,
 	broadcaster := kevents.NewBroadcaster(&kevents.EventSinkImpl{Interface: eventsClient})
 	err := broadcaster.StartRecordingToSinkWithContext(ctx)
 	if err != nil {
-		return nil, nil
+		return nil, err
 	}
 	broadcaster.StartStructuredLogging(0)
 	recorder := broadcaster.NewRecorder(scheme, controllerName)

pkg/common/recorder/event_recorder_test.go

@@ -1,4 +1,4 @@
-package helpers
+package recorder
 
 import (
 	"context"

pkg/common/recorder/logging_recorder.go

@@ -0,0 +1,63 @@
+package recorder
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/openshift/library-go/pkg/operator/events"
+	"k8s.io/klog/v2"
+)
+
+// ContextualLoggingEventRecorder implements a recorder with contextual logging
+type ContextualLoggingEventRecorder struct {
+	component string
+	ctx       context.Context
+}
+
+func (r *ContextualLoggingEventRecorder) WithContext(ctx context.Context) events.Recorder {
+	newRecorder := *r
+	newRecorder.ctx = ctx
+	return &newRecorder
+}
+
+// NewContextualLoggingEventRecorder provides event recorder that will log all recorded events via klog.
+func NewContextualLoggingEventRecorder(component string) events.Recorder {
+	return &ContextualLoggingEventRecorder{
+		component: component,
+		ctx:       context.Background(),
+	}
+}
+
+func (r *ContextualLoggingEventRecorder) ComponentName() string {
+	return r.component
+}
+
+func (r *ContextualLoggingEventRecorder) ForComponent(component string) events.Recorder {
+	newRecorder := *r
+	newRecorder.component = component
+	return &newRecorder
+}
+
+func (r *ContextualLoggingEventRecorder) Shutdown() {}
+
+func (r *ContextualLoggingEventRecorder) WithComponentSuffix(suffix string) events.Recorder {
+	return r.ForComponent(fmt.Sprintf("%s-%s", r.ComponentName(), suffix))
+}
+
+func (r *ContextualLoggingEventRecorder) Event(reason, message string) {
+	logger := klog.FromContext(r.ctx)
+	logger.Info(fmt.Sprintf("INFO: %s", message), "component", r.component, "reason", reason)
+}
+
+func (r *ContextualLoggingEventRecorder) Eventf(reason, messageFmt string, args ...interface{}) {
+	r.Event(reason, fmt.Sprintf(messageFmt, args...))
+}
+
+func (r *ContextualLoggingEventRecorder) Warning(reason, message string) {
+	logger := klog.FromContext(r.ctx)
+	logger.Info(fmt.Sprintf("WARNING: %s", message), "component", r.component, "reason", reason)
+}
+
+func (r *ContextualLoggingEventRecorder) Warningf(reason, messageFmt string, args ...interface{}) {
+	r.Warning(reason, fmt.Sprintf(messageFmt, args...))
+}

pkg/common/testing/fake_sync_context.go

@@ -25,3 +25,19 @@ func NewFakeSyncContext(t *testing.T, clusterName string) *FakeSyncContext {
 		queue:     workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter()),
 	}
 }
+
+type FakeSDKSyncContext struct {
+	spokeName string
+	recorder  events.Recorder
+	queue     workqueue.TypedRateLimitingInterface[string]
+}
+
+func (f FakeSDKSyncContext) Queue() workqueue.TypedRateLimitingInterface[string] { return f.queue }
+
+func NewFakeSDKSyncContext(t *testing.T, clusterName string) *FakeSDKSyncContext {
+	return &FakeSDKSyncContext{
+		spokeName: clusterName,
+		recorder:  eventstesting.NewTestingEventRecorder(t),
+		queue:     workqueue.NewTypedRateLimitingQueue[string](workqueue.DefaultTypedControllerRateLimiter[string]()),
+	}
+}

pkg/placement/controllers/manager.go

@@ -14,7 +14,7 @@ import (
 	clusterscheme "open-cluster-management.io/api/client/cluster/clientset/versioned/scheme"
 	clusterinformers "open-cluster-management.io/api/client/cluster/informers/externalversions"
 
-	"open-cluster-management.io/ocm/pkg/common/helpers"
+	"open-cluster-management.io/ocm/pkg/common/recorder"
 	"open-cluster-management.io/ocm/pkg/placement/controllers/metrics"
 	"open-cluster-management.io/ocm/pkg/placement/controllers/scheduling"
 	"open-cluster-management.io/ocm/pkg/placement/debugger"
@@ -44,7 +44,7 @@ func RunControllerManagerWithInformers(
 	clusterClient clusterclient.Interface,
 	clusterInformers clusterinformers.SharedInformerFactory,
 ) error {
-	recorder, err := helpers.NewEventRecorder(ctx, clusterscheme.Scheme, kubeClient.EventsV1(), "placement-controller")
+	recorder, err := recorder.NewEventRecorder(ctx, clusterscheme.Scheme, kubeClient.EventsV1(), "placement-controller")
 	if err != nil {
 		return err
 	}

pkg/registration/hub/lease/controller_test.go

@@ -22,7 +22,7 @@ import (
 	clusterv1 "open-cluster-management.io/api/cluster/v1"
 	"open-cluster-management.io/sdk-go/pkg/patcher"
 
-	"open-cluster-management.io/ocm/pkg/common/helpers"
+	"open-cluster-management.io/ocm/pkg/common/recorder"
 	testingcommon "open-cluster-management.io/ocm/pkg/common/testing"
 	testinghelpers "open-cluster-management.io/ocm/pkg/registration/helpers/testing"
 )
@@ -174,7 +174,7 @@ func TestSync(t *testing.T) {
 
 			ctx := context.TODO()
 			syncCtx := testingcommon.NewFakeSyncContext(t, testinghelpers.TestManagedClusterName)
-			mcEventRecorder, err := helpers.NewEventRecorder(ctx, clusterscheme.Scheme, hubClient.EventsV1(), "test")
+			mcEventRecorder, err := recorder.NewEventRecorder(ctx, clusterscheme.Scheme, hubClient.EventsV1(), "test")
 			if err != nil {
 				t.Fatal(err)
 			}
@@ -288,7 +288,7 @@ func TestRequeueTime(t *testing.T) {
 			}
 
 			ctx := context.TODO()
-			mcEventRecorder, err := helpers.NewEventRecorder(ctx, clusterscheme.Scheme, hubClient.EventsV1(), "test")
+			mcEventRecorder, err := recorder.NewEventRecorder(ctx, clusterscheme.Scheme, hubClient.EventsV1(), "test")
 			if err != nil {
 				t.Fatal(err)
 			}

pkg/registration/hub/manager.go

@@ -29,7 +29,7 @@ import (
 	ocmfeature "open-cluster-management.io/api/feature"
 	operatorv1 "open-cluster-management.io/api/operator/v1"
 
-	commonhelpers "open-cluster-management.io/ocm/pkg/common/helpers"
+	commonhelpers "open-cluster-management.io/ocm/pkg/common/recorder"
 	"open-cluster-management.io/ocm/pkg/features"
 	"open-cluster-management.io/ocm/pkg/registration/hub/addon"
 	"open-cluster-management.io/ocm/pkg/registration/hub/clusterprofile"

pkg/registration/spoke/managedcluster/claim_reconcile_test.go

@@ -25,7 +25,7 @@ import (
 	clusterv1alpha1 "open-cluster-management.io/api/cluster/v1alpha1"
 	ocmfeature "open-cluster-management.io/api/feature"
 
-	"open-cluster-management.io/ocm/pkg/common/helpers"
+	"open-cluster-management.io/ocm/pkg/common/recorder"
 	testingcommon "open-cluster-management.io/ocm/pkg/common/testing"
 	"open-cluster-management.io/ocm/pkg/features"
 	testinghelpers "open-cluster-management.io/ocm/pkg/registration/helpers/testing"
@@ -199,7 +199,7 @@ func TestSync(t *testing.T) {
 
 			fakeHubClient := kubefake.NewClientset()
 			ctx := context.TODO()
-			hubEventRecorder, err := helpers.NewEventRecorder(ctx,
+			hubEventRecorder, err := recorder.NewEventRecorder(ctx,
 				clusterscheme.Scheme, fakeHubClient.EventsV1(), "test")
 			if err != nil {
 				t.Fatal(err)
@@ -578,7 +578,7 @@ func TestExposeClaims(t *testing.T) {
 
 			fakeHubClient := kubefake.NewClientset()
 			ctx := context.TODO()
-			hubEventRecorder, err := helpers.NewEventRecorder(ctx,
+			hubEventRecorder, err := recorder.NewEventRecorder(ctx,
 				clusterscheme.Scheme, fakeHubClient.EventsV1(), "test")
 			if err != nil {
 				t.Fatal(err)

pkg/registration/spoke/managedcluster/joining_controller_test.go

@@ -20,7 +20,7 @@ import (
 	clusterinformers "open-cluster-management.io/api/client/cluster/informers/externalversions"
 	clusterv1 "open-cluster-management.io/api/cluster/v1"
 
-	"open-cluster-management.io/ocm/pkg/common/helpers"
+	"open-cluster-management.io/ocm/pkg/common/recorder"
 	testingcommon "open-cluster-management.io/ocm/pkg/common/testing"
 	testinghelpers "open-cluster-management.io/ocm/pkg/registration/helpers/testing"
 )
@@ -86,7 +86,7 @@ func TestSyncManagedCluster(t *testing.T) {
 
 			fakeHubClient := kubefake.NewSimpleClientset()
 			ctx := context.TODO()
-			hubEventRecorder, err := helpers.NewEventRecorder(ctx,
+			hubEventRecorder, err := recorder.NewEventRecorder(ctx,
 				clusterscheme.Scheme, fakeHubClient.EventsV1(), "test")
 			if err != nil {
 				t.Fatal(err)

pkg/registration/spoke/managedcluster/resource_reconcile_test.go

@@ -26,7 +26,7 @@ import (
 	clusterinformers "open-cluster-management.io/api/client/cluster/informers/externalversions"
 	clusterv1 "open-cluster-management.io/api/cluster/v1"
 
-	"open-cluster-management.io/ocm/pkg/common/helpers"
+	"open-cluster-management.io/ocm/pkg/common/recorder"
 	testingcommon "open-cluster-management.io/ocm/pkg/common/testing"
 	testinghelpers "open-cluster-management.io/ocm/pkg/registration/helpers/testing"
 )
@@ -320,7 +320,7 @@ func TestHealthCheck(t *testing.T) {
 			fakeHubClient := kubefake.NewSimpleClientset()
 
 			ctx := context.TODO()
-			hubEventRecorder, err := helpers.NewEventRecorder(ctx,
+			hubEventRecorder, err := recorder.NewEventRecorder(ctx,
 				clusterscheme.Scheme, fakeHubClient.EventsV1(), "test")
 			if err != nil {
 				t.Fatal(err)

pkg/registration/spoke/spokeagent.go

@@ -27,8 +27,8 @@ import (
 	clusterv1informers "open-cluster-management.io/api/client/cluster/informers/externalversions"
 	ocmfeature "open-cluster-management.io/api/feature"
 
-	"open-cluster-management.io/ocm/pkg/common/helpers"
 	commonoptions "open-cluster-management.io/ocm/pkg/common/options"
+	eventrecorder "open-cluster-management.io/ocm/pkg/common/recorder"
 	"open-cluster-management.io/ocm/pkg/features"
 	"open-cluster-management.io/ocm/pkg/registration/register"
 	"open-cluster-management.io/ocm/pkg/registration/spoke/addon"
@@ -358,7 +358,7 @@ func (o *SpokeAgentConfig) RunSpokeAgentWithSpokeInformers(ctx context.Context,
 		recorder,
 	)
 
-	hubEventRecorder, err := helpers.NewEventRecorder(ctx, clusterscheme.Scheme, hubClient.EventsClient, "klusterlet-agent")
+	hubEventRecorder, err := eventrecorder.NewEventRecorder(ctx, clusterscheme.Scheme, hubClient.EventsClient, "klusterlet-agent")
 	if err != nil {
 		return fmt.Errorf("failed to create event recorder: %w", err)
 	}

pkg/server/grpc/options.go

@@ -51,9 +51,6 @@ func (o *GRPCServerOptions) Run(ctx context.Context, controllerContext *controll
 		return err
 	}
 
-	// start clients
-	go clients.Run(ctx)
-
 	// initlize grpc broker and register services
 	grpcEventServer := cloudeventsgrpc.NewGRPCBroker()
 	grpcEventServer.RegisterService(ctx, clusterce.ManagedClusterEventDataType,
@@ -69,6 +66,9 @@ func (o *GRPCServerOptions) Run(ctx context.Context, controllerContext *controll
 	grpcEventServer.RegisterService(ctx, payload.ManifestBundleEventDataType,
 		work.NewWorkService(clients.WorkClient, clients.WorkInformers.Work().V1().ManifestWorks()))
 
+	// start clients
+	go clients.Run(ctx)
+
 	// initlize and run grpc server
 	authorizer := grpcauthz.NewSARAuthorizer(clients.KubeClient)
 	return sdkgrpc.NewGRPCServer(serverOptions).

pkg/server/services/addon/addon.go

@@ -7,6 +7,7 @@ import (
 	cloudevents "github.com/cloudevents/sdk-go/v2"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
+	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/klog/v2"
 
@@ -91,31 +92,32 @@ func (s *AddonService) HandleStatusUpdate(ctx context.Context, evt *cloudevents.
 }
 
 func (s *AddonService) RegisterHandler(ctx context.Context, handler server.EventHandler) {
-	if _, err := s.addonInformer.Informer().AddEventHandler(s.EventHandlerFuncs(handler)); err != nil {
-		klog.Errorf("failed to register addon informer event handler, %v", err)
+	logger := klog.FromContext(ctx)
+	if _, err := s.addonInformer.Informer().AddEventHandler(s.EventHandlerFuncs(ctx, handler)); err != nil {
+		logger.Error(err, "failed to register addon informer event handler")
 	}
 }
 
-func (s *AddonService) EventHandlerFuncs(handler server.EventHandler) *cache.ResourceEventHandlerFuncs {
+func (s *AddonService) EventHandlerFuncs(ctx context.Context, handler server.EventHandler) *cache.ResourceEventHandlerFuncs {
 	return &cache.ResourceEventHandlerFuncs{
 		AddFunc: func(obj interface{}) {
 			key, err := cache.MetaNamespaceKeyFunc(obj)
 			if err != nil {
-				klog.Errorf("failed to get key for addon %v", err)
+				utilruntime.HandleErrorWithContext(ctx, err, "failed to get key for addon")
 				return
 			}
-			if err := handler.OnCreate(context.Background(), addonce.ManagedClusterAddOnEventDataType, key); err != nil {
-				klog.Error(err)
+			if err := handler.OnCreate(ctx, addonce.ManagedClusterAddOnEventDataType, key); err != nil {
+				utilruntime.HandleErrorWithContext(ctx, err, "failed to create addon", "key", key)
 			}
 		},
 		UpdateFunc: func(oldObj, newObj interface{}) {
 			key, err := cache.MetaNamespaceKeyFunc(newObj)
 			if err != nil {
-				klog.Errorf("failed to get key for addon %v", err)
+				utilruntime.HandleErrorWithContext(ctx, err, "failed to get key for addon")
 				return
 			}
-			if err := handler.OnUpdate(context.Background(), addonce.ManagedClusterAddOnEventDataType, key); err != nil {
-				klog.Error(err)
+			if err := handler.OnUpdate(ctx, addonce.ManagedClusterAddOnEventDataType, key); err != nil {
+				utilruntime.HandleErrorWithContext(ctx, err, "failed to update addon", "key", key)
 			}
 		},
 	}

pkg/server/services/addon/addon_test.go

@@ -208,7 +208,7 @@ func TestHandleStatusUpdate(t *testing.T) {
 func TestEventHandlerFuncs(t *testing.T) {
 	handler := &addOnHandler{}
 	service := &AddonService{}
-	eventHandlerFuncs := service.EventHandlerFuncs(handler)
+	eventHandlerFuncs := service.EventHandlerFuncs(context.Background(), handler)
 
 	addon := &addonv1alpha1.ManagedClusterAddOn{
 		ObjectMeta: metav1.ObjectMeta{Name: "test-addon", Namespace: "test-namespace"},

pkg/server/services/cluster/cluster.go

@@ -8,6 +8,7 @@ import (
 	"k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/klog/v2"
 
@@ -65,6 +66,8 @@ func (c *ClusterService) List(listOpts types.ListOptions) ([]*cloudevents.Event,
 }
 
 func (c *ClusterService) HandleStatusUpdate(ctx context.Context, evt *cloudevents.Event) error {
+	logger := klog.FromContext(ctx)
+
 	eventType, err := types.ParseCloudEventsType(evt.Type())
 	if err != nil {
 		return fmt.Errorf("failed to parse cloud event type %s, %v", evt.Type(), err)
@@ -74,7 +77,8 @@ func (c *ClusterService) HandleStatusUpdate(ctx context.Context, evt *cloudevent
 		return err
 	}
 
-	klog.V(4).Infof("cluster %s %s %s", cluster.Name, eventType.SubResource, eventType.Action)
+	logger.V(4).Info("handle cluster event",
+		"clusterName", cluster.Name, "subResource", eventType.SubResource, "action", eventType.Action)
 
 	switch eventType.Action {
 	case types.CreateRequestAction:
@@ -94,31 +98,32 @@ func (c *ClusterService) HandleStatusUpdate(ctx context.Context, evt *cloudevent
 }
 
 func (c *ClusterService) RegisterHandler(ctx context.Context, handler server.EventHandler) {
-	if _, err := c.clusterInformer.Informer().AddEventHandler(c.EventHandlerFuncs(handler)); err != nil {
-		klog.Errorf("failed to register cluster informer event handler, %v", err)
+	logger := klog.FromContext(ctx)
+	if _, err := c.clusterInformer.Informer().AddEventHandler(c.EventHandlerFuncs(ctx, handler)); err != nil {
+		logger.Error(err, "failed to register cluster informer event handler")
 	}
 }
 
-func (c *ClusterService) EventHandlerFuncs(handler server.EventHandler) *cache.ResourceEventHandlerFuncs {
+func (c *ClusterService) EventHandlerFuncs(ctx context.Context, handler server.EventHandler) *cache.ResourceEventHandlerFuncs {
 	return &cache.ResourceEventHandlerFuncs{
 		AddFunc: func(obj interface{}) {
 			accessor, err := meta.Accessor(obj)
 			if err != nil {
-				klog.Errorf("failed to get accessor for cluster %v", err)
+				utilruntime.HandleErrorWithContext(ctx, err, "failed to get accessor for cluster")
 				return
 			}
-			if err := handler.OnCreate(context.Background(), clusterce.ManagedClusterEventDataType, accessor.GetName()); err != nil {
-				klog.Error(err)
+			if err := handler.OnCreate(ctx, clusterce.ManagedClusterEventDataType, accessor.GetName()); err != nil {
+				utilruntime.HandleErrorWithContext(ctx, err, "failed to create cluster", "clusterName", accessor.GetName())
 			}
 		},
 		UpdateFunc: func(oldObj, newObj interface{}) {
 			accessor, err := meta.Accessor(newObj)
 			if err != nil {
-				klog.Errorf("failed to get accessor for cluster %v", err)
+				utilruntime.HandleErrorWithContext(ctx, err, "failed to get accessor for cluster")
 				return
 			}
-			if err := handler.OnUpdate(context.Background(), clusterce.ManagedClusterEventDataType, accessor.GetName()); err != nil {
-				klog.Error(err)
+			if err := handler.OnUpdate(ctx, clusterce.ManagedClusterEventDataType, accessor.GetName()); err != nil {
+				utilruntime.HandleErrorWithContext(ctx, err, "failed to update cluster", "clusterName", accessor.GetName())
 			}
 		},
 	}

pkg/server/services/cluster/cluster_test.go

@@ -253,7 +253,7 @@ func TestHandleStatusUpdate(t *testing.T) {
 func TestEventHandlerFuncs(t *testing.T) {
 	handler := &clusterHandler{}
 	service := &ClusterService{}
-	eventHandlerFuncs := service.EventHandlerFuncs(handler)
+	eventHandlerFuncs := service.EventHandlerFuncs(context.Background(), handler)
 
 	cluster := &clusterv1.ManagedCluster{
 		ObjectMeta: metav1.ObjectMeta{Name: "test-cluster"},

pkg/server/services/csr/csr.go

@@ -9,6 +9,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/selection"
+	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	certificatesv1informers "k8s.io/client-go/informers/certificates/v1"
 	"k8s.io/client-go/kubernetes"
 	certificatesv1listers "k8s.io/client-go/listers/certificates/v1"
@@ -99,31 +100,32 @@ func (c *CSRService) HandleStatusUpdate(ctx context.Context, evt *cloudevents.Ev
 }
 
 func (c *CSRService) RegisterHandler(ctx context.Context, handler server.EventHandler) {
-	if _, err := c.csrInformer.Informer().AddEventHandler(c.EventHandlerFuncs(handler)); err != nil {
-		klog.Errorf("failed to register csr informer event handler, %v", err)
+	logger := klog.FromContext(ctx)
+	if _, err := c.csrInformer.Informer().AddEventHandler(c.EventHandlerFuncs(ctx, handler)); err != nil {
+		logger.Error(err, "failed to register csr informer event handler")
 	}
 }
 
-func (c *CSRService) EventHandlerFuncs(handler server.EventHandler) *cache.ResourceEventHandlerFuncs {
+func (c *CSRService) EventHandlerFuncs(ctx context.Context, handler server.EventHandler) *cache.ResourceEventHandlerFuncs {
 	return &cache.ResourceEventHandlerFuncs{
 		AddFunc: func(obj interface{}) {
 			accessor, err := meta.Accessor(obj)
 			if err != nil {
-				klog.Errorf("failed to get accessor for csr %v", err)
+				utilruntime.HandleErrorWithContext(ctx, err, "failed to get accessor for csr")
 				return
 			}
-			if err := handler.OnCreate(context.Background(), csrce.CSREventDataType, accessor.GetName()); err != nil {
-				klog.Error(err)
+			if err := handler.OnCreate(ctx, csrce.CSREventDataType, accessor.GetName()); err != nil {
+				utilruntime.HandleErrorWithContext(ctx, err, "failed to create csr", "csrName", accessor.GetName())
 			}
 		},
 		UpdateFunc: func(oldObj, newObj interface{}) {
 			accessor, err := meta.Accessor(newObj)
 			if err != nil {
-				klog.Errorf("failed to get accessor for csr %v", err)
+				utilruntime.HandleErrorWithContext(ctx, err, "failed to get accessor for csr")
 				return
 			}
-			if err := handler.OnUpdate(context.Background(), csrce.CSREventDataType, accessor.GetName()); err != nil {
-				klog.Error(err)
+			if err := handler.OnUpdate(ctx, csrce.CSREventDataType, accessor.GetName()); err != nil {
+				utilruntime.HandleErrorWithContext(ctx, err, "failed to update csr", "csrName", accessor.GetName())
 			}
 		},
 	}

pkg/server/services/csr/csr_test.go

@@ -213,7 +213,7 @@ func TestHandleStatusUpdate(t *testing.T) {
 func TestEventHandlerFuncs(t *testing.T) {
 	handler := &csrOnHandler{}
 	service := &CSRService{}
-	eventHandlerFuncs := service.EventHandlerFuncs(handler)
+	eventHandlerFuncs := service.EventHandlerFuncs(context.Background(), handler)
 
 	csr := &certificatesv1.CertificateSigningRequest{
 		ObjectMeta: metav1.ObjectMeta{Name: "test-csr"},

pkg/server/services/event/event.go

@@ -37,6 +37,8 @@ func (e *EventService) List(listOpts types.ListOptions) ([]*cloudevents.Event, e
 }
 
 func (e *EventService) HandleStatusUpdate(ctx context.Context, evt *cloudevents.Event) error {
+	logger := klog.FromContext(ctx)
+
 	eventType, err := types.ParseCloudEventsType(evt.Type())
 	if err != nil {
 		return fmt.Errorf("failed to parse cloud event type %s, %v", evt.Type(), err)
@@ -46,7 +48,9 @@ func (e *EventService) HandleStatusUpdate(ctx context.Context, evt *cloudevents.
 		return err
 	}
 
-	klog.V(4).Infof("event %s/%s %s %s", event.Namespace, event.Name, eventType.SubResource, eventType.Action)
+	logger.V(4).Info("handle event",
+		"eventNamespace", event.Namespace, "eventName", event.Name,
+		"subResource", eventType.SubResource, "actionType", eventType.Action)
 
 	switch eventType.Action {
 	case types.CreateRequestAction:
@@ -73,6 +77,6 @@ func (e *EventService) HandleStatusUpdate(ctx context.Context, evt *cloudevents.
 	}
 }
 
-func (e *EventService) RegisterHandler(ctx context.Context, handler server.EventHandler) {
+func (e *EventService) RegisterHandler(_ context.Context, _ server.EventHandler) {
 	// do nothing
 }

pkg/server/services/lease/lease.go

@@ -7,6 +7,7 @@ import (
 	cloudevents "github.com/cloudevents/sdk-go/v2"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
+	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	leasev1 "k8s.io/client-go/informers/coordination/v1"
 	"k8s.io/client-go/kubernetes"
 	leaselister "k8s.io/client-go/listers/coordination/v1"
@@ -90,31 +91,32 @@ func (l *LeaseService) HandleStatusUpdate(ctx context.Context, evt *cloudevents.
 }
 
 func (l *LeaseService) RegisterHandler(ctx context.Context, handler server.EventHandler) {
-	if _, err := l.informer.Informer().AddEventHandler(l.EventHandlerFuncs(handler)); err != nil {
-		klog.Errorf("failed to register lease informer event handler, %v", err)
+	logger := klog.FromContext(ctx)
+	if _, err := l.informer.Informer().AddEventHandler(l.EventHandlerFuncs(ctx, handler)); err != nil {
+		logger.Error(err, "failed to register lease informer event handler")
 	}
 }
 
-func (l *LeaseService) EventHandlerFuncs(handler server.EventHandler) *cache.ResourceEventHandlerFuncs {
+func (l *LeaseService) EventHandlerFuncs(ctx context.Context, handler server.EventHandler) *cache.ResourceEventHandlerFuncs {
 	return &cache.ResourceEventHandlerFuncs{
 		AddFunc: func(obj interface{}) {
 			key, err := cache.MetaNamespaceKeyFunc(obj)
 			if err != nil {
-				klog.Errorf("failed to get key for lease %v", err)
+				utilruntime.HandleErrorWithContext(ctx, err, "failed to get key for lease")
 				return
 			}
-			if err := handler.OnCreate(context.Background(), leasece.LeaseEventDataType, key); err != nil {
-				klog.Error(err)
+			if err := handler.OnCreate(ctx, leasece.LeaseEventDataType, key); err != nil {
+				utilruntime.HandleErrorWithContext(ctx, err, "failed to create lease", "key", key)
 			}
 		},
 		UpdateFunc: func(oldObj, newObj interface{}) {
 			key, err := cache.MetaNamespaceKeyFunc(newObj)
 			if err != nil {
-				klog.Errorf("failed to get key for lease %v", err)
+				utilruntime.HandleErrorWithContext(ctx, err, "failed to get key for lease")
 				return
 			}
-			if err := handler.OnUpdate(context.Background(), leasece.LeaseEventDataType, key); err != nil {
-				klog.Error(err)
+			if err := handler.OnUpdate(ctx, leasece.LeaseEventDataType, key); err != nil {
+				utilruntime.HandleErrorWithContext(ctx, err, "failed to update lease", "key", key)
 			}
 		},
 	}

pkg/server/services/lease/lease_test.go

@@ -214,7 +214,7 @@ func TestHandleStatusUpdate(t *testing.T) {
 func TestEventHandlerFuncs(t *testing.T) {
 	handler := &leaseHandler{}
 	service := &LeaseService{}
-	eventHandlerFuncs := service.EventHandlerFuncs(handler)
+	eventHandlerFuncs := service.EventHandlerFuncs(context.Background(), handler)
 
 	lease := &coordinationv1.Lease{
 		ObjectMeta: metav1.ObjectMeta{Name: "test-lease", Namespace: "test-lease-namespace"},

pkg/server/services/work/work.go

@@ -7,10 +7,11 @@ import (
 	cloudevents "github.com/cloudevents/sdk-go/v2"
 	cloudeventstypes "github.com/cloudevents/sdk-go/v2/types"
 	"github.com/google/go-cmp/cmp"
-	"k8s.io/apimachinery/pkg/api/errors"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/meta"
 	"k8s.io/apimachinery/pkg/labels"
 	kubetypes "k8s.io/apimachinery/pkg/types"
+	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/klog/v2"
 
@@ -86,6 +87,8 @@ func (w *WorkService) List(listOpts types.ListOptions) ([]*cloudevents.Event, er
 }
 
 func (w *WorkService) HandleStatusUpdate(ctx context.Context, evt *cloudevents.Event) error {
+	logger := klog.FromContext(ctx)
+
 	eventType, err := types.ParseCloudEventsType(evt.Type())
 	if err != nil {
 		return fmt.Errorf("failed to parse cloud event type %s, %v", evt.Type(), err)
@@ -107,7 +110,7 @@ func (w *WorkService) HandleStatusUpdate(ctx context.Context, evt *cloudevents.E
 	}
 
 	last, err := w.getWorkByUID(clusterName, work.UID)
-	if errors.IsNotFound(err) {
+	if apierrors.IsNotFound(err) {
 		// work not found, could have been deleted, do nothing.
 		return nil
 	}
@@ -115,7 +118,9 @@ func (w *WorkService) HandleStatusUpdate(ctx context.Context, evt *cloudevents.E
 		return err
 	}
 
-	klog.V(4).Infof("work %s/%s %s %s", last.Namespace, last.Name, eventType.SubResource, eventType.Action)
+	logger.V(4).Info("handle work event",
+		"manifestWorkNamespace", last.Namespace, "manifestWorkName", last.Name,
+		"subResource", eventType.SubResource, "actionType", eventType.Action)
 
 	workPatcher := patcher.NewPatcher[
 		*workv1.ManifestWork, workv1.ManifestWorkSpec, workv1.ManifestWorkStatus](
@@ -146,61 +151,17 @@ func (w *WorkService) HandleStatusUpdate(ctx context.Context, evt *cloudevents.E
 }
 
 func (w *WorkService) RegisterHandler(ctx context.Context, handler server.EventHandler) {
-	if _, err := w.workInformer.Informer().AddEventHandler(w.EventHandlerFuncs(handler)); err != nil {
-		klog.Errorf("failed to register work informer event handler, %v", err)
+	logger := klog.FromContext(ctx)
+	if _, err := w.workInformer.Informer().AddEventHandler(w.EventHandlerFuncs(ctx, handler)); err != nil {
+		logger.Error(err, "failed to register work informer event handler")
 	}
 }
 
-func (w *WorkService) EventHandlerFuncs(handler server.EventHandler) *cache.ResourceEventHandlerFuncs {
+func (w *WorkService) EventHandlerFuncs(ctx context.Context, handler server.EventHandler) *cache.ResourceEventHandlerFuncs {
 	return &cache.ResourceEventHandlerFuncs{
-		AddFunc: func(obj interface{}) {
-			accessor, err := meta.Accessor(obj)
-			if err != nil {
-				klog.Errorf("failed to get accessor for work %v", err)
-				return
-			}
-			id := accessor.GetNamespace() + "/" + accessor.GetName()
-			if err := handler.OnCreate(context.Background(), payload.ManifestBundleEventDataType, id); err != nil {
-				klog.Error(err)
-			}
-		},
-		UpdateFunc: func(oldObj, newObj interface{}) {
-			oldAccessor, err := meta.Accessor(oldObj)
-			if err != nil {
-				klog.Errorf("failed to get accessor for work %v", err)
-				return
-			}
-
-			newAccessor, err := meta.Accessor(newObj)
-			if err != nil {
-				klog.Errorf("failed to get accessor for work %v", err)
-				return
-			}
-
-			// the manifestwork is not changed and is not deleting
-			if cmp.Equal(oldAccessor.GetLabels(), newAccessor.GetLabels()) &&
-				cmp.Equal(oldAccessor.GetAnnotations(), newAccessor.GetAnnotations()) &&
-				oldAccessor.GetGeneration() == newAccessor.GetGeneration() &&
-				newAccessor.GetDeletionTimestamp().IsZero() {
-				return
-			}
-
-			id := newAccessor.GetNamespace() + "/" + newAccessor.GetName()
-			if err := handler.OnUpdate(context.Background(), payload.ManifestBundleEventDataType, id); err != nil {
-				klog.Error(err)
-			}
-		},
-		DeleteFunc: func(obj interface{}) {
-			accessor, err := meta.Accessor(obj)
-			if err != nil {
-				klog.Errorf("failed to get accessor for work %v", err)
-				return
-			}
-			id := accessor.GetNamespace() + "/" + accessor.GetName()
-			if err := handler.OnDelete(context.Background(), payload.ManifestBundleEventDataType, id); err != nil {
-				klog.Error(err)
-			}
-		},
+		AddFunc:    handleOnCreateFunc(ctx, handler),
+		UpdateFunc: handleOnUpdateFunc(ctx, handler),
+		DeleteFunc: handleOnDeleteFunc(ctx, handler),
 	}
 }
 
@@ -216,5 +177,64 @@ func (w *WorkService) getWorkByUID(clusterName string, uid kubetypes.UID) (*work
 		}
 	}
 
-	return nil, errors.NewNotFound(common.ManifestWorkGR, string(uid))
+	return nil, apierrors.NewNotFound(common.ManifestWorkGR, string(uid))
+}
+
+func handleOnCreateFunc(ctx context.Context, handler server.EventHandler) func(obj interface{}) {
+	return func(obj interface{}) {
+		accessor, err := meta.Accessor(obj)
+		if err != nil {
+			utilruntime.HandleErrorWithContext(ctx, err, "failed to get accessor for work")
+			return
+		}
+		id := accessor.GetNamespace() + "/" + accessor.GetName()
+		if err := handler.OnCreate(ctx, payload.ManifestBundleEventDataType, id); err != nil {
+			utilruntime.HandleErrorWithContext(ctx, err, "failed to create work",
+				"manifestWork", accessor.GetName(), "manifestWorkNamespace", accessor.GetNamespace())
+		}
+	}
+}
+
+func handleOnUpdateFunc(ctx context.Context, handler server.EventHandler) func(oldObj, newObj interface{}) {
+	return func(oldObj, newObj interface{}) {
+		oldAccessor, err := meta.Accessor(oldObj)
+		if err != nil {
+			utilruntime.HandleErrorWithContext(ctx, err, "failed to get accessor for work")
+			return
+		}
+		newAccessor, err := meta.Accessor(newObj)
+		if err != nil {
+			utilruntime.HandleErrorWithContext(ctx, err, "failed to get accessor for work")
+			return
+		}
+
+		// the manifestwork is not changed and is not deleting
+		if cmp.Equal(oldAccessor.GetLabels(), newAccessor.GetLabels()) &&
+			cmp.Equal(oldAccessor.GetAnnotations(), newAccessor.GetAnnotations()) &&
+			oldAccessor.GetGeneration() >= newAccessor.GetGeneration() &&
+			newAccessor.GetDeletionTimestamp().IsZero() {
+			return
+		}
+
+		id := newAccessor.GetNamespace() + "/" + newAccessor.GetName()
+		if err := handler.OnUpdate(ctx, payload.ManifestBundleEventDataType, id); err != nil {
+			utilruntime.HandleErrorWithContext(ctx, err, "failed to update work",
+				"manifestWork", newAccessor.GetName(), "manifestWorkNamespace", newAccessor.GetNamespace())
+		}
+	}
+}
+
+func handleOnDeleteFunc(ctx context.Context, handler server.EventHandler) func(obj interface{}) {
+	return func(obj interface{}) {
+		accessor, err := meta.Accessor(obj)
+		if err != nil {
+			utilruntime.HandleErrorWithContext(ctx, err, "failed to get accessor for work")
+			return
+		}
+		id := accessor.GetNamespace() + "/" + accessor.GetName()
+		if err := handler.OnDelete(ctx, payload.ManifestBundleEventDataType, id); err != nil {
+			utilruntime.HandleErrorWithContext(ctx, err, "failed to delete work",
+				"manifestWork", accessor.GetName(), "manifestWorkNamespace", accessor.GetNamespace())
+		}
+	}
 }

pkg/server/services/work/work_test.go

@@ -340,7 +340,7 @@ func TestHandleStatusUpdate(t *testing.T) {
 func TestEventHandlerFuncs(t *testing.T) {
 	handler := &workHandler{}
 	service := &WorkService{}
-	eventHandlerFuncs := service.EventHandlerFuncs(handler)
+	eventHandlerFuncs := service.EventHandlerFuncs(context.Background(), handler)
 
 	work := &workv1.ManifestWork{
 		ObjectMeta: metav1.ObjectMeta{
@@ -408,10 +408,273 @@ func TestEventHandlerFuncs(t *testing.T) {
 	}
 }
 
+func TestHandleOnCreateFunc(t *testing.T) {
+	cases := []struct {
+		name              string
+		obj               interface{}
+		expectedCallCount int
+		expectError       bool
+	}{
+		{
+			name: "successful create",
+			obj: &workv1.ManifestWork{
+				ObjectMeta: metav1.ObjectMeta{Name: "test-work", Namespace: "test-namespace"},
+			},
+			expectedCallCount: 1,
+		},
+		{
+			name:              "invalid object type",
+			obj:               "invalid-object",
+			expectedCallCount: 0,
+			expectError:       true,
+		},
+	}
+
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			handler := &workHandler{}
+			createFunc := handleOnCreateFunc(context.Background(), handler)
+			createFunc(c.obj)
+			if handler.onCreateCallCount != c.expectedCallCount {
+				t.Errorf("expected %d onCreate calls, got %d", c.expectedCallCount, handler.onCreateCallCount)
+			}
+		})
+	}
+}
+
+func TestHandleOnUpdateFunc(t *testing.T) {
+	cases := []struct {
+		name              string
+		oldObj            interface{}
+		newObj            interface{}
+		expectedCallCount int
+		description       string
+	}{
+		{
+			name: "generation increased",
+			oldObj: &workv1.ManifestWork{
+				ObjectMeta: metav1.ObjectMeta{Name: "test-work", Namespace: "test-namespace", Generation: 1},
+			},
+			newObj: &workv1.ManifestWork{
+				ObjectMeta: metav1.ObjectMeta{Name: "test-work", Namespace: "test-namespace", Generation: 2},
+			},
+			expectedCallCount: 1,
+			description:       "should call OnUpdate when generation increases",
+		},
+		{
+			name: "generation same - no update",
+			oldObj: &workv1.ManifestWork{
+				ObjectMeta: metav1.ObjectMeta{Name: "test-work", Namespace: "test-namespace", Generation: 1},
+			},
+			newObj: &workv1.ManifestWork{
+				ObjectMeta: metav1.ObjectMeta{Name: "test-work", Namespace: "test-namespace", Generation: 1},
+			},
+			expectedCallCount: 0,
+			description:       "should not call OnUpdate when generation stays same and no label/annotation changes",
+		},
+		{
+			name: "generation decreased - no update",
+			oldObj: &workv1.ManifestWork{
+				ObjectMeta: metav1.ObjectMeta{Name: "test-work", Namespace: "test-namespace", Generation: 2},
+			},
+			newObj: &workv1.ManifestWork{
+				ObjectMeta: metav1.ObjectMeta{Name: "test-work", Namespace: "test-namespace", Generation: 1},
+			},
+			expectedCallCount: 0,
+			description:       "should not call OnUpdate when generation decreases and no label/annotation changes",
+		},
+		{
+			name: "deletion timestamp set",
+			oldObj: &workv1.ManifestWork{
+				ObjectMeta: metav1.ObjectMeta{Name: "test-work", Namespace: "test-namespace", Generation: 1},
+			},
+			newObj: &workv1.ManifestWork{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:              "test-work",
+					Namespace:         "test-namespace",
+					Generation:        1,
+					DeletionTimestamp: &metav1.Time{Time: time.Now()},
+				},
+			},
+			expectedCallCount: 1,
+			description:       "should call OnUpdate when deletion timestamp is set",
+		},
+		{
+			name: "labels changed",
+			oldObj: &workv1.ManifestWork{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:       "test-work",
+					Namespace:  "test-namespace",
+					Generation: 1,
+					Labels:     map[string]string{"key1": "value1"},
+				},
+			},
+			newObj: &workv1.ManifestWork{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:       "test-work",
+					Namespace:  "test-namespace",
+					Generation: 1,
+					Labels:     map[string]string{"key1": "value2"},
+				},
+			},
+			expectedCallCount: 1,
+			description:       "should call OnUpdate when labels change",
+		},
+		{
+			name: "labels added",
+			oldObj: &workv1.ManifestWork{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:       "test-work",
+					Namespace:  "test-namespace",
+					Generation: 1,
+				},
+			},
+			newObj: &workv1.ManifestWork{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:       "test-work",
+					Namespace:  "test-namespace",
+					Generation: 1,
+					Labels:     map[string]string{"key1": "value1"},
+				},
+			},
+			expectedCallCount: 1,
+			description:       "should call OnUpdate when labels are added",
+		},
+		{
+			name: "annotations changed",
+			oldObj: &workv1.ManifestWork{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:        "test-work",
+					Namespace:   "test-namespace",
+					Generation:  1,
+					Annotations: map[string]string{"key1": "value1"},
+				},
+			},
+			newObj: &workv1.ManifestWork{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:        "test-work",
+					Namespace:   "test-namespace",
+					Generation:  1,
+					Annotations: map[string]string{"key1": "value2"},
+				},
+			},
+			expectedCallCount: 1,
+			description:       "should call OnUpdate when annotations change",
+		},
+		{
+			name: "annotations added",
+			oldObj: &workv1.ManifestWork{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:       "test-work",
+					Namespace:  "test-namespace",
+					Generation: 1,
+				},
+			},
+			newObj: &workv1.ManifestWork{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:        "test-work",
+					Namespace:   "test-namespace",
+					Generation:  1,
+					Annotations: map[string]string{"key1": "value1"},
+				},
+			},
+			expectedCallCount: 1,
+			description:       "should call OnUpdate when annotations are added",
+		},
+		{
+			name: "labels and generation changed",
+			oldObj: &workv1.ManifestWork{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:       "test-work",
+					Namespace:  "test-namespace",
+					Generation: 1,
+					Labels:     map[string]string{"key1": "value1"},
+				},
+			},
+			newObj: &workv1.ManifestWork{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:       "test-work",
+					Namespace:  "test-namespace",
+					Generation: 2,
+					Labels:     map[string]string{"key1": "value2"},
+				},
+			},
+			expectedCallCount: 1,
+			description:       "should call OnUpdate when both labels and generation change",
+		},
+		{
+			name:   "invalid old object type",
+			oldObj: "invalid-object",
+			newObj: &workv1.ManifestWork{
+				ObjectMeta: metav1.ObjectMeta{Name: "test-work", Namespace: "test-namespace", Generation: 1},
+			},
+			expectedCallCount: 0,
+			description:       "should not call OnUpdate when old object is invalid",
+		},
+		{
+			name: "invalid new object type",
+			oldObj: &workv1.ManifestWork{
+				ObjectMeta: metav1.ObjectMeta{Name: "test-work", Namespace: "test-namespace", Generation: 1},
+			},
+			newObj:            "invalid-object",
+			expectedCallCount: 0,
+			description:       "should not call OnUpdate when new object is invalid",
+		},
+	}
+
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			handler := &workHandler{}
+			updateFunc := handleOnUpdateFunc(context.Background(), handler)
+			updateFunc(c.oldObj, c.newObj)
+			if handler.onUpdateCallCount != c.expectedCallCount {
+				t.Errorf("%s: expected %d OnUpdate calls, got %d", c.description, c.expectedCallCount, handler.onUpdateCallCount)
+			}
+		})
+	}
+}
+
+func TestHandleOnDeleteFunc(t *testing.T) {
+	cases := []struct {
+		name              string
+		obj               interface{}
+		expectedCallCount int
+		expectError       bool
+	}{
+		{
+			name: "successful delete",
+			obj: &workv1.ManifestWork{
+				ObjectMeta: metav1.ObjectMeta{Name: "test-work", Namespace: "test-namespace"},
+			},
+			expectedCallCount: 1,
+		},
+		{
+			name:              "invalid object type",
+			obj:               "invalid-object",
+			expectedCallCount: 0,
+			expectError:       true,
+		},
+	}
+
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			handler := &workHandler{}
+			deleteFunc := handleOnDeleteFunc(context.Background(), handler)
+			deleteFunc(c.obj)
+			if handler.onDeleteCallCount != c.expectedCallCount {
+				t.Errorf("expected %d onDelete calls, got %d", c.expectedCallCount, handler.onDeleteCallCount)
+			}
+		})
+	}
+}
+
 type workHandler struct {
-	onCreateCalled bool
-	onUpdateCalled bool
-	onDeleteCalled bool
+	onCreateCalled    bool
+	onUpdateCalled    bool
+	onDeleteCalled    bool
+	onCreateCallCount int
+	onUpdateCallCount int
+	onDeleteCallCount int
 }
 
 func (m *workHandler) OnCreate(ctx context.Context, t types.CloudEventsDataType, resourceID string) error {
@@ -422,6 +685,7 @@ func (m *workHandler) OnCreate(ctx context.Context, t types.CloudEventsDataType,
 		return fmt.Errorf("expected %v, got %v", "test-namespace/test-work", resourceID)
 	}
 	m.onCreateCalled = true
+	m.onCreateCallCount++
 	return nil
 }
 
@@ -433,6 +697,7 @@ func (m *workHandler) OnUpdate(ctx context.Context, t types.CloudEventsDataType,
 		return fmt.Errorf("expected %v, got %v", "test-namespace/test-work", resourceID)
 	}
 	m.onUpdateCalled = true
+	m.onUpdateCallCount++
 	return nil
 }
 
@@ -444,5 +709,6 @@ func (m *workHandler) OnDelete(ctx context.Context, t types.CloudEventsDataType,
 		return fmt.Errorf("expected %v, got %v", "test-namespace/test-work", resourceID)
 	}
 	m.onDeleteCalled = true
+	m.onDeleteCallCount++
 	return nil
 }

pkg/work/helper/helpers.go

@@ -10,7 +10,6 @@ import (
 	"strings"
 	"time"
 
-	"github.com/openshift/library-go/pkg/controller/factory"
 	"github.com/openshift/library-go/pkg/operator/resource/resourcehelper"
 	"github.com/openshift/library-go/pkg/operator/resource/resourcemerge"
 	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
@@ -30,6 +29,7 @@ import (
 	clusterv1beta1 "open-cluster-management.io/api/cluster/v1beta1"
 	workapiv1 "open-cluster-management.io/api/work/v1"
 	clustersdkv1beta1 "open-cluster-management.io/sdk-go/pkg/apis/cluster/v1beta1"
+	"open-cluster-management.io/sdk-go/pkg/basecontroller/factory"
 )
 
 const (
@@ -298,14 +298,14 @@ func GuessObjectGroupVersionKind(object runtime.Object) (*schema.GroupVersionKin
 }
 
 // AppliedManifestworkQueueKeyFunc return manifestwork key from appliedmanifestwork
-func AppliedManifestworkQueueKeyFunc(hubhash string) factory.ObjectQueueKeyFunc {
-	return func(obj runtime.Object) string {
+func AppliedManifestworkQueueKeyFunc(hubhash string) factory.ObjectQueueKeysFunc {
+	return func(obj runtime.Object) []string {
 		accessor, _ := meta.Accessor(obj)
 		if !strings.HasPrefix(accessor.GetName(), hubhash) {
-			return ""
+			return []string{}
 		}
 
-		return strings.TrimPrefix(accessor.GetName(), hubhash+"-")
+		return []string{strings.TrimPrefix(accessor.GetName(), hubhash+"-")}
 	}
 }
 

pkg/work/hub/controllers/manifestworkgarbagecollection/controller.go

@@ -5,7 +5,6 @@ import (
 	"fmt"
 	"time"
 
-	"github.com/openshift/library-go/pkg/controller/factory"
 	"github.com/openshift/library-go/pkg/operator/events"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/meta"
@@ -18,6 +17,7 @@ import (
 	workinformers "open-cluster-management.io/api/client/work/informers/externalversions/work/v1"
 	worklisters "open-cluster-management.io/api/client/work/listers/work/v1"
 	workapiv1 "open-cluster-management.io/api/work/v1"
+	"open-cluster-management.io/sdk-go/pkg/basecontroller/factory"
 
 	"open-cluster-management.io/ocm/pkg/common/queue"
 )
@@ -45,12 +45,11 @@ func NewManifestWorkGarbageCollectionController(
 			manifestWorkInformer.Informer(),
 		).
 		WithSync(controller.sync).
-		ToController("ManifestWorkGarbageCollectionController", recorder)
+		ToController("ManifestWorkGarbageCollectionController")
 }
 
 // sync is the main reconcile loop for completed ManifestWork TTL
-func (c *ManifestWorkGarbageCollectionController) sync(ctx context.Context, controllerContext factory.SyncContext) error {
-	key := controllerContext.QueueKey()
+func (c *ManifestWorkGarbageCollectionController) sync(ctx context.Context, controllerContext factory.SyncContext, key string) error {
 	logger := klog.FromContext(ctx)
 	logger.V(4).Info("Reconciling ManifestWork for TTL processing", "key", key)
 

pkg/work/hub/controllers/manifestworkgarbagecollection/controller_test.go

@@ -89,8 +89,8 @@ func TestManifestWorkGarbageCollectionController(t *testing.T) {
 			workInformerFactory.Start(ctx.Done())
 			workInformerFactory.WaitForCacheSync(ctx.Done())
 
-			syncContext := testingcommon.NewFakeSyncContext(t, "default/test")
-			err := controller.sync(ctx, syncContext)
+			syncContext := testingcommon.NewFakeSDKSyncContext(t, "default/test")
+			err := controller.sync(ctx, syncContext, "default/test")
 			if err != nil {
 				t.Errorf("Unexpected error: %v", err)
 			}

pkg/work/hub/controllers/manifestworkreplicasetcontroller/manifestworkreplicaset_controller.go

@@ -7,7 +7,6 @@ import (
 	"strings"
 	"time"
 
-	"github.com/openshift/library-go/pkg/controller/factory"
 	"github.com/openshift/library-go/pkg/operator/events"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/meta"
@@ -28,6 +27,7 @@ import (
 	workapiv1 "open-cluster-management.io/api/work/v1"
 	workapiv1alpha1 "open-cluster-management.io/api/work/v1alpha1"
 	workapplier "open-cluster-management.io/sdk-go/pkg/apis/work/v1/applier"
+	"open-cluster-management.io/sdk-go/pkg/basecontroller/factory"
 	"open-cluster-management.io/sdk-go/pkg/patcher"
 
 	"open-cluster-management.io/ocm/pkg/common/helpers"
@@ -101,23 +101,23 @@ func NewManifestWorkReplicaSetController(
 
 	return factory.New().
 		WithInformersQueueKeysFunc(queue.QueueKeyByMetaNamespaceName, manifestWorkReplicaSetInformer.Informer()).
-		WithFilteredEventsInformersQueueKeyFunc(func(obj runtime.Object) string {
+		WithFilteredEventsInformersQueueKeysFunc(func(obj runtime.Object) []string {
 			accessor, _ := meta.Accessor(obj)
 			labelValue, ok := accessor.GetLabels()[ManifestWorkReplicaSetControllerNameLabelKey]
 			if !ok {
-				return ""
+				return []string{}
 			}
 			keys := strings.Split(labelValue, ".")
 			if len(keys) != 2 {
-				return ""
+				return []string{}
 			}
-			return fmt.Sprintf("%s/%s", keys[0], keys[1])
+			return []string{fmt.Sprintf("%s/%s", keys[0], keys[1])}
 		},
 			queue.FileterByLabel(ManifestWorkReplicaSetControllerNameLabelKey),
 			manifestWorkInformer.Informer()).
 		WithInformersQueueKeysFunc(controller.placementDecisionQueueKeysFunc, placeDecisionInformer.Informer()).
 		WithInformersQueueKeysFunc(controller.placementQueueKeysFunc, placementInformer.Informer()).
-		WithSync(controller.sync).ToController("ManifestWorkReplicaSetController", recorder)
+		WithSync(controller.sync).ToController("ManifestWorkReplicaSetController")
 }
 
 func newController(
@@ -154,9 +154,10 @@ func newController(
 }
 
 // sync is the main reconcile loop for ManifestWorkReplicaSet. It is triggered every 15sec
-func (m *ManifestWorkReplicaSetController) sync(ctx context.Context, controllerContext factory.SyncContext) error {
-	key := controllerContext.QueueKey()
-	klog.V(4).Infof("Reconciling ManifestWorkReplicaSet %q", key)
+func (m *ManifestWorkReplicaSetController) sync(ctx context.Context, controllerContext factory.SyncContext, key string) error {
+	logger := klog.FromContext(ctx).WithValues("manifestWorkReplicaSet", key)
+
+	logger.V(5).Info("Reconciling ManifestWorkReplicaSet")
 
 	namespace, name, err := cache.SplitMetaNamespaceKey(key)
 	if err != nil {

pkg/work/hub/controllers/manifestworkreplicasetcontroller/manifestworkreplicaset_controllers_test.go

@@ -245,8 +245,8 @@ func TestManifestWorkReplicaSetControllerPatchStatus(t *testing.T) {
 				clusterInformers.Cluster().V1beta1().PlacementDecisions(),
 			)
 
-			controllerContext := testingcommon.NewFakeSyncContext(t, c.mwrSet.Namespace+"/"+c.mwrSet.Name)
-			err = ctrl.sync(context.TODO(), controllerContext)
+			controllerContext := testingcommon.NewFakeSDKSyncContext(t, c.mwrSet.Namespace+"/"+c.mwrSet.Name)
+			err = ctrl.sync(context.TODO(), controllerContext, c.mwrSet.Namespace+"/"+c.mwrSet.Name)
 			if err != nil {
 				t.Error(err)
 			}

pkg/work/spoke/auth/cache/auth.go

@@ -6,7 +6,6 @@ import (
 	"fmt"
 	"time"
 
-	"github.com/openshift/library-go/pkg/controller/factory"
 	"github.com/openshift/library-go/pkg/operator/events"
 	"k8s.io/apimachinery/pkg/api/meta"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
@@ -18,6 +17,7 @@ import (
 
 	worklister "open-cluster-management.io/api/client/work/listers/work/v1"
 	workapiv1 "open-cluster-management.io/api/work/v1"
+	"open-cluster-management.io/sdk-go/pkg/basecontroller/factory"
 
 	"open-cluster-management.io/ocm/pkg/work/spoke/auth/basic"
 	"open-cluster-management.io/ocm/pkg/work/spoke/auth/store"

pkg/work/spoke/auth/cache/executor_cache_controller.go

@@ -6,7 +6,6 @@ import (
 	"sync"
 	"time"
 
-	"github.com/openshift/library-go/pkg/controller/factory"
 	"github.com/openshift/library-go/pkg/operator/events"
 	rbacapiv1 "k8s.io/api/rbac/v1"
 	"k8s.io/apimachinery/pkg/api/meta"
@@ -18,6 +17,7 @@ import (
 	"k8s.io/klog/v2"
 
 	workapiv1 "open-cluster-management.io/api/work/v1"
+	"open-cluster-management.io/sdk-go/pkg/basecontroller/factory"
 
 	"open-cluster-management.io/ocm/pkg/work/spoke/auth/store"
 )
@@ -108,7 +108,7 @@ func newControllerInner(controller *CacheController,
 	}
 
 	cacheControllerName := "ManifestWorkExecutorCache"
-	syncCtx := factory.NewSyncContext(cacheControllerName, recorder)
+	syncCtx := factory.NewSyncContext(cacheControllerName)
 
 	_, err = rbInformer.Informer().AddEventHandler(&roleBindingEventHandler{
 		enqueueUpsertFunc: controller.bindingResourceUpsertEnqueueFn(syncCtx),
@@ -137,7 +137,7 @@ func newControllerInner(controller *CacheController,
 		WithBareInformers(rbInformer.Informer(), crbInformer.Informer()).
 		WithSync(controller.sync).
 		ResyncEvery(ResyncInterval). // cleanup unnecessary cache every ResyncInterval
-		ToController(cacheControllerName, recorder)
+		ToController(cacheControllerName)
 }
 
 func (c *CacheController) roleEnqueueFu(rbIndexer cache.Indexer) func(runtime.Object) []string {
@@ -254,14 +254,15 @@ func getInterestedExecutors(subjects []rbacapiv1.Subject, executorCaches *store.
 
 // sync is the main reconcile loop for executors. It is triggered when RBAC resources(
 // role, rolebinding, clusterrole, clusterrolebinding) for the executor changed
-func (c *CacheController) sync(ctx context.Context, controllerContext factory.SyncContext) error {
-	executorKey := controllerContext.QueueKey()
-	klog.V(4).Infof("Executor cache sync, executorKey: %v", executorKey)
+func (c *CacheController) sync(ctx context.Context, _ factory.SyncContext, executorKey string) error {
+	logger := klog.FromContext(ctx).WithValues("executorKey", executorKey)
+	ctx = klog.NewContext(ctx, logger)
+	logger.V(4).Info("Executor cache sync")
 	if executorKey == "key" {
 		// cleanup unnecessary cache
-		klog.V(4).Infof("There are %v cache items before cleanup", c.executorCaches.Count())
+		logger.V(4).Info("Cache items before cleanup", "count", c.executorCaches.Count())
 		c.cleanupUnnecessaryCache()
-		klog.V(4).Infof("There are %v cache items after cleanup", c.executorCaches.Count())
+		logger.V(4).Info("Cache items after cleanup", "count", c.executorCaches.Count())
 		return nil
 	}
 
@@ -277,6 +278,7 @@ func (c *CacheController) sync(ctx context.Context, controllerContext factory.Sy
 
 func (c *CacheController) iterateCacheItemsFn(ctx context.Context,
 	executorKey, saNamespace, saName string) func(v store.CacheValue) error {
+	logger := klog.FromContext(ctx)
 	return func(v store.CacheValue) error {
 		err := c.sarCheckerFn(ctx, &workapiv1.ManifestWorkSubjectServiceAccount{
 			Namespace: saNamespace,
@@ -288,8 +290,7 @@ func (c *CacheController) iterateCacheItemsFn(ctx context.Context,
 		},
 			v.Dimension.Namespace, v.Dimension.Name, store.GetOwnedByWork(v.Dimension.ExecuteAction))
 
-		klog.V(4).Infof("Update executor cache for executorKey: %s, dimension: %+v result: %v",
-			executorKey, v.Dimension, err)
+		logger.V(4).Info("Update executor cache", "dimension", v.Dimension, "error", err)
 		updateSARCheckResultToCache(c.executorCaches, executorKey, v.Dimension, err)
 		return nil
 	}

pkg/work/spoke/auth/factory.go

@@ -55,7 +55,8 @@ func NewFactory(
 }
 
 func (f *validatorFactory) NewExecutorValidator(ctx context.Context, isCacheValidator bool) ExecutorValidator {
-	klog.Infof("Executor caches enabled: %v", isCacheValidator)
+	logger := klog.FromContext(ctx)
+	logger.Info("Executor caches enabled", "cacheValidator", isCacheValidator)
 	sarValidator := basic.NewSARValidator(f.config, f.kubeClient)
 	if !isCacheValidator {
 		return sarValidator

pkg/work/spoke/controllers/finalizercontroller/add_finalizer_controller.go

@@ -3,7 +3,6 @@ package finalizercontroller
 import (
 	"context"
 
-	"github.com/openshift/library-go/pkg/controller/factory"
 	"github.com/openshift/library-go/pkg/operator/events"
 	"k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/klog/v2"
@@ -12,6 +11,7 @@ import (
 	workinformer "open-cluster-management.io/api/client/work/informers/externalversions/work/v1"
 	worklister "open-cluster-management.io/api/client/work/listers/work/v1"
 	workapiv1 "open-cluster-management.io/api/work/v1"
+	"open-cluster-management.io/sdk-go/pkg/basecontroller/factory"
 	"open-cluster-management.io/sdk-go/pkg/patcher"
 
 	"open-cluster-management.io/ocm/pkg/common/queue"
@@ -42,13 +42,11 @@ func NewAddFinalizerController(
 
 	return factory.New().
 		WithInformersQueueKeysFunc(queue.QueueKeyByMetaName, manifestWorkInformer.Informer()).
-		WithSync(controller.sync).ToController(manifestWorkAddFinalizerController, recorder)
+		WithSync(controller.sync).ToController(manifestWorkAddFinalizerController)
 }
 
-func (m *AddFinalizerController) sync(ctx context.Context, controllerContext factory.SyncContext) error {
-	manifestWorkName := controllerContext.QueueKey()
-	logger := klog.FromContext(ctx).WithName(manifestWorkAddFinalizerController).
-		WithValues("manifestWorkName", manifestWorkName)
+func (m *AddFinalizerController) sync(ctx context.Context, _ factory.SyncContext, manifestWorkName string) error {
+	logger := klog.FromContext(ctx).WithValues("manifestWorkName", manifestWorkName)
 	logger.V(5).Info("Reconciling ManifestWork")
 
 	manifestWork, err := m.manifestWorkLister.Get(manifestWorkName)

pkg/work/spoke/controllers/finalizercontroller/appliedmanifestwork_finalize_controller.go

@@ -5,7 +5,6 @@ import (
 	"fmt"
 	"time"
 
-	"github.com/openshift/library-go/pkg/controller/factory"
 	"github.com/openshift/library-go/pkg/operator/events"
 	"k8s.io/apimachinery/pkg/api/errors"
 	utilerrors "k8s.io/apimachinery/pkg/util/errors"
@@ -17,6 +16,7 @@ import (
 	workinformer "open-cluster-management.io/api/client/work/informers/externalversions/work/v1"
 	worklister "open-cluster-management.io/api/client/work/listers/work/v1"
 	workapiv1 "open-cluster-management.io/api/work/v1"
+	"open-cluster-management.io/sdk-go/pkg/basecontroller/factory"
 	"open-cluster-management.io/sdk-go/pkg/patcher"
 
 	commonhelper "open-cluster-management.io/ocm/pkg/common/helpers"
@@ -56,13 +56,11 @@ func NewAppliedManifestWorkFinalizeController(
 	return factory.New().
 		WithFilteredEventsInformersQueueKeysFunc(queue.QueueKeyByMetaName,
 			helper.AppliedManifestworkAgentIDFilter(agentID), appliedManifestWorkInformer.Informer()).
-		WithSync(controller.sync).ToController(appliedManifestWorkFinalizer, recorder)
+		WithSync(controller.sync).ToController(appliedManifestWorkFinalizer)
 }
 
-func (m *AppliedManifestWorkFinalizeController) sync(ctx context.Context, controllerContext factory.SyncContext) error {
-	appliedManifestWorkName := controllerContext.QueueKey()
-	logger := klog.FromContext(ctx).WithName(appliedManifestWorkFinalizer).
-		WithValues("appliedManifestWorkName", appliedManifestWorkName)
+func (m *AppliedManifestWorkFinalizeController) sync(ctx context.Context, controllerContext factory.SyncContext, appliedManifestWorkName string) error {
+	logger := klog.FromContext(ctx).WithValues("appliedManifestWorkName", appliedManifestWorkName)
 	logger.V(5).Info("Reconciling AppliedManifestWork")
 	ctx = klog.NewContext(ctx, logger)
 

pkg/work/spoke/controllers/finalizercontroller/appliedmanifestwork_finalize_controller_test.go

@@ -194,7 +194,7 @@ func TestFinalize(t *testing.T) {
 				rateLimiter:        workqueue.NewTypedItemExponentialFailureRateLimiter[string](0, 1*time.Second),
 			}
 
-			controllerContext := testingcommon.NewFakeSyncContext(t, testingWork.Name)
+			controllerContext := testingcommon.NewFakeSDKSyncContext(t, testingWork.Name)
 			err := controller.syncAppliedManifestWork(context.TODO(), controllerContext, testingWork)
 			if err != nil {
 				t.Fatal(err)

pkg/work/spoke/controllers/finalizercontroller/manifestwork_finalize_controller.go

@@ -5,7 +5,6 @@ import (
 	"fmt"
 	"time"
 
-	"github.com/openshift/library-go/pkg/controller/factory"
 	"github.com/openshift/library-go/pkg/operator/events"
 	"k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/meta"
@@ -17,6 +16,7 @@ import (
 	workinformer "open-cluster-management.io/api/client/work/informers/externalversions/work/v1"
 	worklister "open-cluster-management.io/api/client/work/listers/work/v1"
 	workapiv1 "open-cluster-management.io/api/work/v1"
+	"open-cluster-management.io/sdk-go/pkg/basecontroller/factory"
 	"open-cluster-management.io/sdk-go/pkg/patcher"
 
 	"open-cluster-management.io/ocm/pkg/common/queue"
@@ -58,15 +58,14 @@ func NewManifestWorkFinalizeController(
 
 	return factory.New().
 		WithInformersQueueKeysFunc(queue.QueueKeyByMetaName, manifestWorkInformer.Informer()).
-		WithFilteredEventsInformersQueueKeyFunc(
+		WithFilteredEventsInformersQueueKeysFunc(
 			helper.AppliedManifestworkQueueKeyFunc(hubHash),
 			helper.AppliedManifestworkHubHashFilter(hubHash),
 			appliedManifestWorkInformer.Informer()).
-		WithSync(controller.sync).ToController(manifestWorkFinalizer, recorder)
+		WithSync(controller.sync).ToController(manifestWorkFinalizer)
 }
 
-func (m *ManifestWorkFinalizeController) sync(ctx context.Context, controllerContext factory.SyncContext) error {
-	manifestWorkName := controllerContext.QueueKey()
+func (m *ManifestWorkFinalizeController) sync(ctx context.Context, controllerContext factory.SyncContext, manifestWorkName string) error {
 	appliedManifestWorkName := fmt.Sprintf("%s-%s", m.hubHash, manifestWorkName)
 
 	logger := klog.FromContext(ctx).WithName(appliedManifestWorkFinalizer).

pkg/work/spoke/controllers/finalizercontroller/manifestwork_finalize_controller_test.go

@@ -204,8 +204,8 @@ func TestSyncManifestWorkController(t *testing.T) {
 				rateLimiter:               workqueue.NewItemExponentialFailureRateLimiter(0, 1*time.Second),
 			}
 
-			controllerContext := testingcommon.NewFakeSyncContext(t, c.workName)
-			err := controller.sync(context.TODO(), controllerContext)
+			controllerContext := testingcommon.NewFakeSDKSyncContext(t, c.workName)
+			err := controller.sync(context.TODO(), controllerContext, c.workName)
 			if err != nil {
 				t.Errorf("Expect no sync error, but got %v", err)
 			}

pkg/work/spoke/controllers/finalizercontroller/unmanaged_appliedmanifestwork_controller.go

@@ -6,7 +6,6 @@ import (
 	"strings"
 	"time"
 
-	"github.com/openshift/library-go/pkg/controller/factory"
 	"github.com/openshift/library-go/pkg/operator/events"
 	"k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/meta"
@@ -19,6 +18,7 @@ import (
 	workinformer "open-cluster-management.io/api/client/work/informers/externalversions/work/v1"
 	worklister "open-cluster-management.io/api/client/work/listers/work/v1"
 	workapiv1 "open-cluster-management.io/api/work/v1"
+	"open-cluster-management.io/sdk-go/pkg/basecontroller/factory"
 	"open-cluster-management.io/sdk-go/pkg/patcher"
 
 	"open-cluster-management.io/ocm/pkg/common/queue"
@@ -77,19 +77,17 @@ func NewUnManagedAppliedWorkController(
 	}
 
 	return factory.New().
-		WithInformersQueueKeyFunc(func(obj runtime.Object) string {
+		WithInformersQueueKeysFunc(func(obj runtime.Object) []string {
 			accessor, _ := meta.Accessor(obj)
-			return fmt.Sprintf("%s-%s", hubHash, accessor.GetName())
+			return []string{fmt.Sprintf("%s-%s", hubHash, accessor.GetName())}
 		}, manifestWorkInformer.Informer()).
 		WithFilteredEventsInformersQueueKeysFunc(
 			queue.QueueKeyByMetaName,
 			helper.AppliedManifestworkAgentIDFilter(agentID), appliedManifestWorkInformer.Informer()).
-		WithSync(controller.sync).ToController(unManagedAppliedManifestWork, recorder)
+		WithSync(controller.sync).ToController(unManagedAppliedManifestWork)
 }
 
-func (m *unmanagedAppliedWorkController) sync(ctx context.Context, controllerContext factory.SyncContext) error {
-	appliedManifestWorkName := controllerContext.QueueKey()
-
+func (m *unmanagedAppliedWorkController) sync(ctx context.Context, controllerContext factory.SyncContext, appliedManifestWorkName string) error {
 	logger := klog.FromContext(ctx).WithName(appliedManifestWorkFinalizer).
 		WithValues("appliedManifestWorkName", appliedManifestWorkName)
 	ctx = klog.NewContext(ctx, logger)

pkg/work/spoke/controllers/finalizercontroller/unmanaged_appliedmanifestwork_controller_test.go

@@ -286,8 +286,8 @@ func TestSyncUnamanagedAppliedWork(t *testing.T) {
 				rateLimiter:               workqueue.NewItemExponentialFailureRateLimiter(0, c.evictionGracePeriod),
 			}
 
-			controllerContext := testingcommon.NewFakeSyncContext(t, c.appliedManifestWorkName)
-			if err := controller.sync(context.TODO(), controllerContext); err != nil {
+			controllerContext := testingcommon.NewFakeSDKSyncContext(t, c.appliedManifestWorkName)
+			if err := controller.sync(context.TODO(), controllerContext, c.appliedManifestWorkName); err != nil {
 				t.Errorf("Expect no sync error, but got %v", err)
 			}
 

pkg/work/spoke/controllers/manifestcontroller/appliedmanifestwork_reconciler.go

@@ -5,7 +5,6 @@ import (
 	"fmt"
 	"sort"
 
-	"github.com/openshift/library-go/pkg/controller/factory"
 	"k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -17,6 +16,7 @@ import (
 	"k8s.io/klog/v2"
 
 	workapiv1 "open-cluster-management.io/api/work/v1"
+	"open-cluster-management.io/sdk-go/pkg/basecontroller/factory"
 
 	commonhelper "open-cluster-management.io/ocm/pkg/common/helpers"
 	"open-cluster-management.io/ocm/pkg/work/helper"
@@ -29,7 +29,7 @@ type appliedManifestWorkReconciler struct {
 
 func (m *appliedManifestWorkReconciler) reconcile(
 	ctx context.Context,
-	controllerContext factory.SyncContext,
+	_ factory.SyncContext,
 	manifestWork *workapiv1.ManifestWork,
 	appliedManifestWork *workapiv1.AppliedManifestWork,
 	results []applyResult) (*workapiv1.ManifestWork, *workapiv1.AppliedManifestWork, []applyResult, error) {

pkg/work/spoke/controllers/manifestcontroller/appliedmanifestwork_reconciler_test.go

@@ -8,7 +8,6 @@ import (
 	"time"
 
 	"github.com/davecgh/go-spew/spew"
-	"github.com/openshift/library-go/pkg/controller/factory"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
@@ -20,6 +19,7 @@ import (
 	fakeworkclient "open-cluster-management.io/api/client/work/clientset/versioned/fake"
 	workinformers "open-cluster-management.io/api/client/work/informers/externalversions"
 	workapiv1 "open-cluster-management.io/api/work/v1"
+	"open-cluster-management.io/sdk-go/pkg/basecontroller/factory"
 	"open-cluster-management.io/sdk-go/pkg/patcher"
 
 	testingcommon "open-cluster-management.io/ocm/pkg/common/testing"
@@ -297,8 +297,8 @@ func TestSyncManifestWork(t *testing.T) {
 				hubHash: "test",
 			}
 
-			controllerContext := testingcommon.NewFakeSyncContext(t, testingWork.Name)
-			err := controller.sync(context.TODO(), controllerContext)
+			controllerContext := testingcommon.NewFakeSDKSyncContext(t, testingWork.Name)
+			err := controller.sync(context.TODO(), controllerContext, testingWork.Name)
 			if err != nil {
 				t.Fatal(err)
 			}

pkg/work/spoke/controllers/manifestcontroller/manifestwork_controller.go

@@ -5,7 +5,6 @@ import (
 	"fmt"
 	"time"
 
-	"github.com/openshift/library-go/pkg/controller/factory"
 	"github.com/openshift/library-go/pkg/operator/events"
 	"github.com/pkg/errors"
 	apiextensionsclient "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset"
@@ -26,6 +25,7 @@ import (
 	workinformer "open-cluster-management.io/api/client/work/informers/externalversions/work/v1"
 	worklister "open-cluster-management.io/api/client/work/listers/work/v1"
 	workapiv1 "open-cluster-management.io/api/work/v1"
+	"open-cluster-management.io/sdk-go/pkg/basecontroller/factory"
 	"open-cluster-management.io/sdk-go/pkg/patcher"
 
 	commonhelper "open-cluster-management.io/ocm/pkg/common/helpers"
@@ -77,7 +77,7 @@ func NewManifestWorkController(
 	restMapper meta.RESTMapper,
 	validator auth.ExecutorValidator) factory.Controller {
 
-	syncCtx := factory.NewSyncContext("manifestwork-controller", recorder)
+	syncCtx := factory.NewSyncContext("manifestwork-controller")
 
 	controller := &ManifestWorkController{
 		manifestWorkPatcher: patcher.NewPatcher[
@@ -117,15 +117,14 @@ func NewManifestWorkController(
 			appliedManifestWorkInformer.Informer(),
 		).
 		WithSyncContext(syncCtx).
-		WithSync(controller.sync).ToController(controllerName, recorder)
+		WithSync(controller.sync).ToController(controllerName)
 }
 
 // sync is the main reconcile loop for manifest work. It is triggered in two scenarios
 // 1. ManifestWork API changes
 // 2. Resources defined in manifest changed on spoke
-func (m *ManifestWorkController) sync(ctx context.Context, controllerContext factory.SyncContext) error {
-	manifestWorkName := controllerContext.QueueKey()
-	logger := klog.FromContext(ctx).WithName(controllerName).WithValues("manifestWorkName", manifestWorkName)
+func (m *ManifestWorkController) sync(ctx context.Context, controllerContext factory.SyncContext, manifestWorkName string) error {
+	logger := klog.FromContext(ctx).WithValues("manifestWorkName", manifestWorkName)
 	logger.V(5).Info("Reconciling ManifestWork")
 	ctx = klog.NewContext(ctx, logger)
 
@@ -229,7 +228,7 @@ func (m *ManifestWorkController) applyAppliedManifestWork(ctx context.Context, w
 	return appliedManifestWork, err
 }
 
-func onAddFunc(queue workqueue.RateLimitingInterface) func(obj interface{}) {
+func onAddFunc(queue workqueue.TypedRateLimitingInterface[string]) func(obj interface{}) {
 	return func(obj interface{}) {
 		accessor, err := meta.Accessor(obj)
 		if err != nil {
@@ -242,7 +241,7 @@ func onAddFunc(queue workqueue.RateLimitingInterface) func(obj interface{}) {
 	}
 }
 
-func onUpdateFunc(queue workqueue.RateLimitingInterface) func(oldObj, newObj interface{}) {
+func onUpdateFunc(queue workqueue.TypedRateLimitingInterface[string]) func(oldObj, newObj interface{}) {
 	return func(oldObj, newObj interface{}) {
 		newWork, ok := newObj.(*workapiv1.ManifestWork)
 		if !ok {

pkg/work/spoke/controllers/manifestcontroller/manifestwork_reconciler.go

@@ -5,7 +5,6 @@ import (
 	"errors"
 	"fmt"
 
-	"github.com/openshift/library-go/pkg/controller/factory"
 	"github.com/openshift/library-go/pkg/operator/events"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/meta"
@@ -18,8 +17,10 @@ import (
 	"k8s.io/klog/v2"
 
 	workapiv1 "open-cluster-management.io/api/work/v1"
+	"open-cluster-management.io/sdk-go/pkg/basecontroller/factory"
 
 	commonhelper "open-cluster-management.io/ocm/pkg/common/helpers"
+	"open-cluster-management.io/ocm/pkg/common/recorder"
 	"open-cluster-management.io/ocm/pkg/work/helper"
 	"open-cluster-management.io/ocm/pkg/work/spoke/apply"
 	"open-cluster-management.io/ocm/pkg/work/spoke/auth"
@@ -46,6 +47,7 @@ func (m *manifestworkReconciler) reconcile(
 	appliedManifestWork *workapiv1.AppliedManifestWork,
 	_ []applyResult) (*workapiv1.ManifestWork, *workapiv1.AppliedManifestWork, []applyResult, error) {
 	logger := klog.FromContext(ctx)
+	eventRecorder := recorder.NewContextualLoggingEventRecorder(controllerName).WithContext(ctx)
 	// We creat a ownerref instead of controller ref since multiple controller can declare the ownership of a manifests
 	owner := helper.NewAppliedManifestWorkOwner(appliedManifestWork)
 
@@ -54,7 +56,7 @@ func (m *manifestworkReconciler) reconcile(
 	resourceResults := make([]applyResult, len(manifestWork.Spec.Workload.Manifests))
 	err := retry.RetryOnConflict(retry.DefaultBackoff, func() error {
 		resourceResults = m.applyManifests(
-			ctx, manifestWork.Spec.Workload.Manifests, manifestWork.Spec, manifestWork.Status, controllerContext.Recorder(), *owner, resourceResults)
+			ctx, manifestWork.Spec.Workload.Manifests, manifestWork.Spec, manifestWork.Status, eventRecorder, *owner, resourceResults)
 
 		for _, result := range resourceResults {
 			if apierrors.IsConflict(result.Error) {

pkg/work/spoke/controllers/manifestcontroller/manifestwork_reconciler_test.go

@@ -375,8 +375,8 @@ func TestSync(t *testing.T) {
 			controller := newController(t, work, nil, spoketesting.NewFakeRestMapper()).
 				withKubeObject(c.spokeObject...).
 				withUnstructuredObject(c.spokeDynamicObject...)
-			syncContext := testingcommon.NewFakeSyncContext(t, workKey)
-			err := controller.toController().sync(context.TODO(), syncContext)
+			syncContext := testingcommon.NewFakeSDKSyncContext(t, workKey)
+			err := controller.toController().sync(context.TODO(), syncContext, work.Name)
 			if err != nil {
 				t.Errorf("Should be success with no err: %v", err)
 			}
@@ -418,8 +418,8 @@ func TestFailedToApplyResource(t *testing.T) {
 
 		return true, &corev1.Secret{}, fmt.Errorf("fake error")
 	})
-	syncContext := testingcommon.NewFakeSyncContext(t, workKey)
-	err := controller.toController().sync(context.TODO(), syncContext)
+	syncContext := testingcommon.NewFakeSDKSyncContext(t, workKey)
+	err := controller.toController().sync(context.TODO(), syncContext, work.Name)
 	if err == nil {
 		t.Errorf("Should return an err")
 	}
@@ -553,8 +553,8 @@ func TestUpdateStrategy(t *testing.T) {
 						"v1", "NewObject", "ns1", "n1",
 						map[string]interface{}{"spec": map[string]interface{}{"key1": "val1"}}), nil
 				})
-			syncContext := testingcommon.NewFakeSyncContext(t, workKey)
-			err := controller.toController().sync(context.TODO(), syncContext)
+			syncContext := testingcommon.NewFakeSDKSyncContext(t, workKey)
+			err := controller.toController().sync(context.TODO(), syncContext, work.Name)
 			if err != nil {
 				t.Errorf("Should be success with no err: %v", err)
 			}
@@ -590,8 +590,8 @@ func TestServerSideApplyConflict(t *testing.T) {
 	controller.dynamicClient.PrependReactor("patch", "newobjects", func(action clienttesting.Action) (handled bool, ret runtime.Object, err error) {
 		return true, nil, errors.NewConflict(schema.GroupResource{Resource: "newobjects"}, "n1", fmt.Errorf("conflict error"))
 	})
-	syncContext := testingcommon.NewFakeSyncContext(t, workKey)
-	err := controller.toController().sync(context.TODO(), syncContext)
+	syncContext := testingcommon.NewFakeSDKSyncContext(t, workKey)
+	err := controller.toController().sync(context.TODO(), syncContext, work.Name)
 	if err != nil {
 		t.Errorf("Should be success with no err: %v", err)
 	}
@@ -897,7 +897,7 @@ func TestOnAddFunc(t *testing.T) {
 
 	for _, c := range cases {
 		t.Run(c.name, func(t *testing.T) {
-			queue := workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter())
+			queue := workqueue.NewTypedRateLimitingQueue[string](workqueue.DefaultTypedControllerRateLimiter[string]())
 			addFunc := onAddFunc(queue)
 
 			addFunc(c.obj)
@@ -1060,7 +1060,7 @@ func TestOnUpdateFunc(t *testing.T) {
 
 	for _, c := range cases {
 		t.Run(c.name, func(t *testing.T) {
-			queue := workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter())
+			queue := workqueue.NewTypedRateLimitingQueue[string](workqueue.DefaultTypedControllerRateLimiter[string]())
 			updateFunc := onUpdateFunc(queue)
 
 			updateFunc(c.oldObj, c.newObj)

pkg/work/spoke/controllers/statuscontroller/availablestatus_controller.go

@@ -5,7 +5,6 @@ import (
 	"fmt"
 	"time"
 
-	"github.com/openshift/library-go/pkg/controller/factory"
 	"github.com/openshift/library-go/pkg/operator/events"
 	"k8s.io/apimachinery/pkg/api/equality"
 	"k8s.io/apimachinery/pkg/api/errors"
@@ -22,6 +21,7 @@ import (
 	workinformer "open-cluster-management.io/api/client/work/informers/externalversions/work/v1"
 	worklister "open-cluster-management.io/api/client/work/listers/work/v1"
 	workapiv1 "open-cluster-management.io/api/work/v1"
+	"open-cluster-management.io/sdk-go/pkg/basecontroller/factory"
 	"open-cluster-management.io/sdk-go/pkg/patcher"
 
 	commonhelper "open-cluster-management.io/ocm/pkg/common/helpers"
@@ -78,13 +78,11 @@ func NewAvailableStatusController(
 
 	return factory.New().
 		WithInformersQueueKeysFunc(queue.QueueKeyByMetaName, manifestWorkInformer.Informer()).
-		WithSync(controller.sync).ToController(controllerName, recorder), nil
+		WithSync(controller.sync).ToController(controllerName), nil
 }
 
-func (c *AvailableStatusController) sync(ctx context.Context, controllerContext factory.SyncContext) error {
-	manifestWorkName := controllerContext.QueueKey()
-
-	logger := klog.FromContext(ctx).WithName(controllerName).WithValues("manifestWorkName", manifestWorkName)
+func (c *AvailableStatusController) sync(ctx context.Context, controllerContext factory.SyncContext, manifestWorkName string) error {
+	logger := klog.FromContext(ctx).WithValues("manifestWorkName", manifestWorkName)
 	logger.V(4).Info("Reconciling ManifestWork")
 	ctx = klog.NewContext(ctx, logger)
 

test/integration-test.mk

@@ -30,8 +30,8 @@ build-work-integration:
 	go test -c ./test/integration/work -o ./work-integration.test
 
 test-registration-integration: ensure-kubebuilder-tools
-	go test -c ./test/integration/registration -o ./registration-integration.test
-	./registration-integration.test -ginkgo.slow-spec-threshold=15s -ginkgo.v -ginkgo.fail-fast ${ARGS}
+	go test -c ./test/integration/registration -o ./registration-integration.test -mod=vendor
+	./registration-integration.test -ginkgo.slow-spec-threshold=15s -ginkgo.v -ginkgo.fail-fast ${ARGS} -v=5
 .PHONY: test-registration-integration
 
 test-work-integration: ensure-kubebuilder-tools build-work-integration

test/integration/registration/integration_suite_test.go

@@ -15,6 +15,7 @@ import (
 	"k8s.io/client-go/kubernetes/scheme"
 	"k8s.io/client-go/rest"
 	"k8s.io/client-go/transport"
+	"k8s.io/klog/v2"
 	"sigs.k8s.io/controller-runtime/pkg/envtest"
 	logf "sigs.k8s.io/controller-runtime/pkg/log"
 	"sigs.k8s.io/controller-runtime/pkg/log/zap"
@@ -103,6 +104,11 @@ func runAgentWithContext(ctx context.Context, name string, agentConfig *spoke.Sp
 	}()
 }
 
+func init() {
+	klog.InitFlags(nil)
+	klog.SetOutput(ginkgo.GinkgoWriter)
+}
+
 func TestIntegration(t *testing.T) {
 	gomega.RegisterFailHandler(ginkgo.Fail)
 	ginkgo.RunSpecs(t, "Integration Suite")

test/integration/registration/spokecluster_grpc_test.go

@@ -83,8 +83,6 @@ var _ = ginkgo.Describe("Registration using GRPC", ginkgo.Ordered, ginkgo.Label(
 		hook, err := util.NewGRPCServerRegistrationHook(hubCfg)
 		gomega.Expect(err).NotTo(gomega.HaveOccurred())
 
-		go hook.Run(grpcServerCtx)
-
 		grpcEventServer := cloudeventsgrpc.NewGRPCBroker()
 		grpcEventServer.RegisterService(grpcServerCtx, clusterce.ManagedClusterEventDataType,
 			cluster.NewClusterService(hook.ClusterClient, hook.ClusterInformers.Cluster().V1().ManagedClusters()))
@@ -97,6 +95,11 @@ var _ = ginkgo.Describe("Registration using GRPC", ginkgo.Ordered, ginkgo.Label(
 		grpcEventServer.RegisterService(grpcServerCtx, leasece.LeaseEventDataType,
 			lease.NewLeaseService(hook.KubeClient, hook.KubeInformers.Coordination().V1().Leases()))
 
+		go func() {
+			defer ginkgo.GinkgoRecover()
+			hook.Run(grpcServerCtx)
+		}()
+
 		authorizer := util.NewMockAuthorizer()
 		server := sdkgrpc.NewGRPCServer(gRPCServerOptions).
 			WithAuthenticator(grpcauthn.NewMtlsAuthenticator()).

test/integration/work/suite_test.go

@@ -274,15 +274,16 @@ func startGRPCServer(ctx context.Context, temp string, cfg *rest.Config) (string
 
 	hook, err := util.NewGRPCServerWorkHook(cfg)
 	gomega.Expect(err).NotTo(gomega.HaveOccurred())
-	go func() {
-		defer ginkgo.GinkgoRecover()
-		hook.Run(ctx)
-	}()
 
 	grpcEventServer := cloudeventsgrpc.NewGRPCBroker()
 	grpcEventServer.RegisterService(ctx, payload.ManifestBundleEventDataType,
 		serviceswork.NewWorkService(hook.WorkClient, hook.WorkInformers.Work().V1().ManifestWorks()))
 
+	go func() {
+		defer ginkgo.GinkgoRecover()
+		hook.Run(ctx)
+	}()
+
 	authorizer := util.NewMockAuthorizer()
 	server := sdkgrpc.NewGRPCServer(gRPCServerOptions).
 		WithUnaryAuthorizer(authorizer).

vendor/modules.txt

@@ -1802,7 +1802,7 @@ open-cluster-management.io/api/operator/v1
 open-cluster-management.io/api/utils/work/v1/workapplier
 open-cluster-management.io/api/work/v1
 open-cluster-management.io/api/work/v1alpha1
-# open-cluster-management.io/sdk-go v1.1.1-0.20251117075350-a9794783fa67
+# open-cluster-management.io/sdk-go v1.1.1-0.20251120064629-eb2d8ba7fdd3
 ## explicit; go 1.24.0
 open-cluster-management.io/sdk-go/pkg/apis/cluster/v1alpha1
 open-cluster-management.io/sdk-go/pkg/apis/cluster/v1beta1
@@ -1846,6 +1846,7 @@ open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options/grpc/protobuf/
 open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options/grpc/protocol
 open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options/mqtt
 open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options/v2/grpc
+open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options/v2/mqtt
 open-cluster-management.io/sdk-go/pkg/cloudevents/generic/payload
 open-cluster-management.io/sdk-go/pkg/cloudevents/generic/types
 open-cluster-management.io/sdk-go/pkg/cloudevents/generic/utils

vendor/open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options/builder/optionsbuilder.go

@@ -8,6 +8,7 @@ import (
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options/grpc"
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options/mqtt"
 	grpcv2 "open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options/v2/grpc"
+	mqttv2 "open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options/v2/mqtt"
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/types"
 )
 
@@ -56,7 +57,7 @@ func BuildCloudEventsSourceOptions(config any,
 	clientId, sourceId string, dataType types.CloudEventsDataType) (*options.CloudEventsSourceOptions, error) {
 	switch config := config.(type) {
 	case *mqtt.MQTTOptions:
-		return mqtt.NewSourceOptions(config, clientId, sourceId), nil
+		return mqttv2.NewSourceOptions(config, clientId, sourceId), nil
 	case *grpc.GRPCOptions:
 		return grpcv2.NewSourceOptions(config, sourceId, dataType), nil
 	default:
@@ -69,7 +70,7 @@ func BuildCloudEventsAgentOptions(config any,
 	clusterName, clientId string, dataType types.CloudEventsDataType) (*options.CloudEventsAgentOptions, error) {
 	switch config := config.(type) {
 	case *mqtt.MQTTOptions:
-		return mqtt.NewAgentOptions(config, clusterName, clientId), nil
+		return mqttv2.NewAgentOptions(config, clusterName, clientId), nil
 	case *grpc.GRPCOptions:
 		return grpcv2.NewAgentOptions(config, clusterName, clientId, dataType), nil
 	default:

vendor/open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options/mqtt/agentoptions.go

@@ -24,6 +24,7 @@ type mqttAgentTransport struct {
 	agentID           string
 }
 
+// Deprecated: use v2.mqtt.NewAgentOptions instead
 func NewAgentOptions(mqttOptions *MQTTOptions, clusterName, agentID string) *options.CloudEventsAgentOptions {
 	mqttAgentOptions := &mqttAgentTransport{
 		MQTTOptions: *mqttOptions,
@@ -40,8 +41,6 @@ func NewAgentOptions(mqttOptions *MQTTOptions, clusterName, agentID string) *opt
 }
 
 func (o *mqttAgentTransport) WithContext(ctx context.Context, evtCtx cloudevents.EventContext) (context.Context, error) {
-	logger := klog.FromContext(ctx)
-
 	topic, err := getAgentPubTopic(ctx)
 	if err != nil {
 		return nil, err
@@ -51,59 +50,18 @@ func (o *mqttAgentTransport) WithContext(ctx context.Context, evtCtx cloudevents
 		return cloudeventscontext.WithTopic(ctx, string(*topic)), nil
 	}
 
-	eventType, err := types.ParseCloudEventsType(evtCtx.GetType())
-	if err != nil {
-		return nil, fmt.Errorf("unsupported event type %s, %v", eventType, err)
-	}
-
-	originalSource, err := evtCtx.GetExtension(types.ExtensionOriginalSource)
+	pubTopic, err := AgentPubTopic(ctx, &o.MQTTOptions, o.clusterName, evtCtx)
 	if err != nil {
 		return nil, err
 	}
 
-	// agent request to sync resource spec from all sources
-	if eventType.Action == types.ResyncRequestAction && originalSource == types.SourceAll {
-		if len(o.Topics.AgentBroadcast) == 0 {
-			logger.Info("the agent broadcast topic not set, fall back to the agent events topic")
-
-			// TODO after supporting multiple sources, we should list each source
-			eventsTopic := replaceLast(o.Topics.AgentEvents, "+", o.clusterName)
-			return cloudeventscontext.WithTopic(ctx, eventsTopic), nil
-		}
-
-		resyncTopic := strings.Replace(o.Topics.AgentBroadcast, "+", o.clusterName, 1)
-		return cloudeventscontext.WithTopic(ctx, resyncTopic), nil
-	}
-
-	topicSource, err := getSourceFromEventsTopic(o.Topics.AgentEvents)
-	if err != nil {
-		return nil, err
-	}
-
-	// agent publishes status events or spec resync events
-	eventsTopic := replaceLast(o.Topics.AgentEvents, "+", o.clusterName)
-	eventsTopic = replaceLast(eventsTopic, "+", topicSource)
-	return cloudeventscontext.WithTopic(ctx, eventsTopic), nil
+	return cloudeventscontext.WithTopic(ctx, pubTopic), nil
 }
 
 func (o *mqttAgentTransport) Connect(ctx context.Context) error {
-	subscribe := &paho.Subscribe{
-		Subscriptions: []paho.SubscribeOptions{
-			{
-				// TODO support multiple sources, currently the client require the source events topic has a sourceID, in
-				// the future, client may need a source list, it will subscribe to each source
-				// receiving the sources events
-				Topic: replaceLast(o.Topics.SourceEvents, "+", o.clusterName), QoS: byte(o.SubQoS),
-			},
-		},
-	}
-
-	// receiving status resync events from all sources
-	if len(o.Topics.SourceBroadcast) != 0 {
-		subscribe.Subscriptions = append(subscribe.Subscriptions, paho.SubscribeOptions{
-			Topic: o.Topics.SourceBroadcast,
-			QoS:   byte(o.SubQoS),
-		})
+	subscribe, err := AgentSubscribe(&o.MQTTOptions, o.clusterName)
+	if err != nil {
+		return err
 	}
 
 	protocol, err := o.GetCloudEventsProtocol(
@@ -157,3 +115,67 @@ func (o *mqttAgentTransport) Close(ctx context.Context) error {
 func (o *mqttAgentTransport) ErrorChan() <-chan error {
 	return o.errorChan
 }
+
+func AgentPubTopic(ctx context.Context, o *MQTTOptions, clusterName string, evtCtx cloudevents.EventContext) (string, error) {
+	logger := klog.FromContext(ctx)
+
+	ceType := evtCtx.GetType()
+	eventType, err := types.ParseCloudEventsType(ceType)
+	if err != nil {
+		return "", fmt.Errorf("unsupported event type %q, %v", ceType, err)
+	}
+
+	originalSourceVal, err := evtCtx.GetExtension(types.ExtensionOriginalSource)
+	if err != nil {
+		return "", err
+	}
+
+	originalSource, ok := originalSourceVal.(string)
+	if !ok {
+		return "", fmt.Errorf("originalsource extension must be a string, got %T", originalSourceVal)
+	}
+
+	// agent request to sync resource spec from all sources
+	if eventType.Action == types.ResyncRequestAction && originalSource == types.SourceAll {
+		if len(o.Topics.AgentBroadcast) == 0 {
+			logger.Info("the agent broadcast topic not set, fall back to the agent events topic")
+
+			// TODO after supporting multiple sources, we should list each source
+			return replaceLast(o.Topics.AgentEvents, "+", clusterName), nil
+		}
+
+		return strings.Replace(o.Topics.AgentBroadcast, "+", clusterName, 1), nil
+	}
+
+	topicSource, err := getSourceFromEventsTopic(o.Topics.AgentEvents)
+	if err != nil {
+		return "", err
+	}
+
+	// agent publishes status events or spec resync events
+	eventsTopic := replaceLast(o.Topics.AgentEvents, "+", clusterName)
+	return replaceLast(eventsTopic, "+", topicSource), nil
+}
+
+func AgentSubscribe(o *MQTTOptions, clusterName string) (*paho.Subscribe, error) {
+	subscribe := &paho.Subscribe{
+		Subscriptions: []paho.SubscribeOptions{
+			{
+				// TODO support multiple sources, currently the client require the source events topic has a sourceID, in
+				// the future, client may need a source list, it will subscribe to each source
+				// receiving the sources events
+				Topic: replaceLast(o.Topics.SourceEvents, "+", clusterName), QoS: byte(o.SubQoS),
+			},
+		},
+	}
+
+	// receiving status resync events from all sources
+	if len(o.Topics.SourceBroadcast) != 0 {
+		subscribe.Subscriptions = append(subscribe.Subscriptions, paho.SubscribeOptions{
+			Topic: o.Topics.SourceBroadcast,
+			QoS:   byte(o.SubQoS),
+		})
+	}
+
+	return subscribe, nil
+}

vendor/open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options/mqtt/logger.go

@@ -2,6 +2,7 @@ package mqtt
 
 import (
 	"fmt"
+
 	"github.com/eclipse/paho.golang/paho/log"
 	"k8s.io/klog/v2"
 )
@@ -14,8 +15,13 @@ type PahoDebugLogger struct {
 	logger klog.Logger
 }
 
-var _ log.Logger = &PahoErrorLogger{}
-var _ log.Logger = &PahoDebugLogger{}
+func NewPahoErrorLogger(logger klog.Logger) log.Logger {
+	return &PahoErrorLogger{logger: logger}
+}
+
+func NewPahoDebugLogger(logger klog.Logger) log.Logger {
+	return &PahoDebugLogger{logger: logger}
+}
 
 func (l *PahoErrorLogger) Println(v ...interface{}) {
 	l.logger.Error(fmt.Errorf("get err %s", fmt.Sprint(v...)), "MQTT error message")

vendor/open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options/mqtt/options.go

@@ -4,13 +4,14 @@ import (
 	"context"
 	"crypto/tls"
 	"fmt"
-	"k8s.io/klog/v2"
 	"net"
 	"os"
 	"regexp"
 	"strings"
 	"time"
 
+	"k8s.io/klog/v2"
+
 	cloudeventsmqtt "github.com/cloudevents/sdk-go/protocol/mqtt_paho/v2"
 	"github.com/eclipse/paho.golang/packets"
 	"github.com/eclipse/paho.golang/paho"
@@ -235,8 +236,8 @@ func (o *MQTTOptions) GetCloudEventsProtocol(
 
 	opts := []cloudeventsmqtt.Option{
 		cloudeventsmqtt.WithConnect(o.GetMQTTConnectOption(clientID)),
-		cloudeventsmqtt.WithDebugLogger(&PahoDebugLogger{logger: logger}),
-		cloudeventsmqtt.WithErrorLogger(&PahoErrorLogger{logger: logger}),
+		cloudeventsmqtt.WithDebugLogger(NewPahoDebugLogger(logger)),
+		cloudeventsmqtt.WithErrorLogger(NewPahoErrorLogger(logger)),
 	}
 	opts = append(opts, clientOpts...)
 	return cloudeventsmqtt.New(ctx, config, opts...)

vendor/open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options/mqtt/sourceoptions.go

@@ -23,6 +23,7 @@ type mqttSourceTransport struct {
 	clientID          string
 }
 
+// Deprecated: use v2.mqtt.NewSourceOptions instead
 func NewSourceOptions(mqttOptions *MQTTOptions, clientID, sourceID string) *options.CloudEventsSourceOptions {
 	mqttSourceOptions := &mqttSourceTransport{
 		MQTTOptions: *mqttOptions,
@@ -47,58 +48,20 @@ func (o *mqttSourceTransport) WithContext(ctx context.Context, evtCtx cloudevent
 		return cloudeventscontext.WithTopic(ctx, string(*topic)), nil
 	}
 
-	eventType, err := types.ParseCloudEventsType(evtCtx.GetType())
-	if err != nil {
-		return nil, fmt.Errorf("unsupported event type %s, %v", eventType, err)
-	}
-
-	clusterName, err := evtCtx.GetExtension(types.ExtensionClusterName)
+	pubTopic, err := SourcePubTopic(ctx, &o.MQTTOptions, o.sourceID, evtCtx)
 	if err != nil {
 		return nil, err
 	}
 
-	if eventType.Action == types.ResyncRequestAction && clusterName == types.ClusterAll {
-		// source request to get resources status from all agents
-		if len(o.Topics.SourceBroadcast) == 0 {
-			return nil, fmt.Errorf("the source broadcast topic not set")
-		}
-
-		resyncTopic := strings.Replace(o.Topics.SourceBroadcast, "+", o.sourceID, 1)
-		return cloudeventscontext.WithTopic(ctx, resyncTopic), nil
-	}
-
-	// source publishes spec events or status resync events
-	eventsTopic := strings.Replace(o.Topics.SourceEvents, "+", fmt.Sprintf("%s", clusterName), 1)
-	return cloudeventscontext.WithTopic(ctx, eventsTopic), nil
+	return cloudeventscontext.WithTopic(ctx, pubTopic), nil
 }
 
 func (o *mqttSourceTransport) Connect(ctx context.Context) error {
-	topicSource, err := getSourceFromEventsTopic(o.Topics.AgentEvents)
+	subscribe, err := SourceSubscribe(&o.MQTTOptions, o.sourceID)
 	if err != nil {
 		return err
 	}
 
-	if topicSource != o.sourceID {
-		return fmt.Errorf("the topic source %q does not match with the client sourceID %q",
-			o.Topics.AgentEvents, o.sourceID)
-	}
-
-	subscribe := &paho.Subscribe{
-		Subscriptions: []paho.SubscribeOptions{
-			{
-				Topic: o.Topics.AgentEvents, QoS: byte(o.SubQoS),
-			},
-		},
-	}
-
-	if len(o.Topics.AgentBroadcast) != 0 {
-		// receiving spec resync events from all agents
-		subscribe.Subscriptions = append(subscribe.Subscriptions, paho.SubscribeOptions{
-			Topic: o.Topics.AgentBroadcast,
-			QoS:   byte(o.SubQoS),
-		})
-	}
-
 	protocol, err := o.GetCloudEventsProtocol(
 		ctx,
 		o.clientID,
@@ -151,3 +114,65 @@ func (o *mqttSourceTransport) Close(ctx context.Context) error {
 func (o *mqttSourceTransport) ErrorChan() <-chan error {
 	return o.errorChan
 }
+
+func SourcePubTopic(ctx context.Context, o *MQTTOptions, sourceID string, evtCtx cloudevents.EventContext) (string, error) {
+	ceType := evtCtx.GetType()
+	eventType, err := types.ParseCloudEventsType(ceType)
+	if err != nil {
+		return "", fmt.Errorf("unsupported event type %q, %v", ceType, err)
+	}
+
+	clusterNameVal, err := evtCtx.GetExtension(types.ExtensionClusterName)
+	if err != nil {
+		return "", err
+	}
+
+	clusterName, ok := clusterNameVal.(string)
+	if !ok {
+		return "", fmt.Errorf("clustername extension must be a string, got %T", clusterNameVal)
+	}
+
+	if eventType.Action == types.ResyncRequestAction && clusterName == types.ClusterAll {
+		// source request to get resources status from all agents
+		if len(o.Topics.SourceBroadcast) == 0 {
+			return "", fmt.Errorf("the source broadcast topic not set")
+		}
+
+		resyncTopic := strings.Replace(o.Topics.SourceBroadcast, "+", sourceID, 1)
+		return resyncTopic, nil
+	}
+
+	// source publishes spec events or status resync events
+	eventsTopic := strings.Replace(o.Topics.SourceEvents, "+", clusterName, 1)
+	return eventsTopic, nil
+}
+
+func SourceSubscribe(o *MQTTOptions, sourceID string) (*paho.Subscribe, error) {
+	topicSource, err := getSourceFromEventsTopic(o.Topics.AgentEvents)
+	if err != nil {
+		return nil, err
+	}
+
+	if topicSource != sourceID {
+		return nil, fmt.Errorf("the topic source %q does not match the client sourceID %q",
+			topicSource, sourceID)
+	}
+
+	subscribe := &paho.Subscribe{
+		Subscriptions: []paho.SubscribeOptions{
+			{
+				Topic: o.Topics.AgentEvents, QoS: byte(o.SubQoS),
+			},
+		},
+	}
+
+	if len(o.Topics.AgentBroadcast) != 0 {
+		// receiving spec resync events from all agents
+		subscribe.Subscriptions = append(subscribe.Subscriptions, paho.SubscribeOptions{
+			Topic: o.Topics.AgentBroadcast,
+			QoS:   byte(o.SubQoS),
+		})
+	}
+
+	return subscribe, nil
+}

vendor/open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options/v2/mqtt/options.go

@@ -0,0 +1,44 @@
+package mqtt
+
+import (
+	"context"
+	"fmt"
+
+	v2 "github.com/cloudevents/sdk-go/v2"
+	"github.com/eclipse/paho.golang/paho"
+	"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options"
+	"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options/mqtt"
+)
+
+func NewAgentOptions(opts *mqtt.MQTTOptions, clusterName, agentID string) *options.CloudEventsAgentOptions {
+	return &options.CloudEventsAgentOptions{
+		CloudEventsTransport: newTransport(
+			fmt.Sprintf("%s-client", agentID),
+			opts,
+			func(ctx context.Context, e v2.Event) (string, error) {
+				return mqtt.AgentPubTopic(ctx, opts, clusterName, e.Context)
+			},
+			func() (*paho.Subscribe, error) {
+				return mqtt.AgentSubscribe(opts, clusterName)
+			},
+		),
+		AgentID:     agentID,
+		ClusterName: clusterName,
+	}
+}
+
+func NewSourceOptions(opts *mqtt.MQTTOptions, clientID, sourceID string) *options.CloudEventsSourceOptions {
+	return &options.CloudEventsSourceOptions{
+		CloudEventsTransport: newTransport(
+			clientID,
+			opts,
+			func(ctx context.Context, e v2.Event) (string, error) {
+				return mqtt.SourcePubTopic(ctx, opts, sourceID, e.Context)
+			},
+			func() (*paho.Subscribe, error) {
+				return mqtt.SourceSubscribe(opts, sourceID)
+			},
+		),
+		SourceID: sourceID,
+	}
+}

vendor/open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options/v2/mqtt/transport.go

@@ -0,0 +1,228 @@
+package mqtt
+
+import (
+	"context"
+	"fmt"
+	"sync"
+
+	cloudeventsmqtt "github.com/cloudevents/sdk-go/protocol/mqtt_paho/v2"
+	cloudevents "github.com/cloudevents/sdk-go/v2"
+	"github.com/cloudevents/sdk-go/v2/binding"
+	"github.com/eclipse/paho.golang/paho"
+	"k8s.io/klog/v2"
+	"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options"
+	"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options/mqtt"
+)
+
+type pubTopicGetter func(context.Context, cloudevents.Event) (string, error)
+type subscribeGetter func() (*paho.Subscribe, error)
+
+type mqttTransport struct {
+	opts *mqtt.MQTTOptions
+
+	mu         sync.RWMutex
+	subscribed bool
+	closeChan  chan struct{}
+	errorChan  chan error
+	msgChan    chan *paho.Publish
+
+	clientID string
+
+	getPublishTopic pubTopicGetter
+	getSubscribe    subscribeGetter
+
+	client *paho.Client
+}
+
+func newTransport(clientID string, opts *mqtt.MQTTOptions, pubTopicGetter pubTopicGetter, subscribeGetter subscribeGetter) *mqttTransport {
+	return &mqttTransport{
+		opts:            opts,
+		clientID:        clientID,
+		errorChan:       make(chan error, 1),
+		getPublishTopic: pubTopicGetter,
+		getSubscribe:    subscribeGetter,
+	}
+}
+
+func (t *mqttTransport) Connect(ctx context.Context) error {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+
+	logger := klog.FromContext(ctx)
+	tcpConn, err := t.opts.Dialer.Dial()
+	if err != nil {
+		return err
+	}
+
+	config := paho.ClientConfig{
+		ClientID: t.clientID,
+		Conn:     tcpConn,
+		OnClientError: func(err error) {
+			select {
+			case t.errorChan <- err:
+			default:
+				logger.Error(err, "mqtt client error")
+			}
+		},
+	}
+
+	t.client = paho.NewClient(config)
+	t.client.SetDebugLogger(mqtt.NewPahoDebugLogger(logger))
+	t.client.SetErrorLogger(mqtt.NewPahoErrorLogger(logger))
+
+	connAck, err := t.client.Connect(ctx, t.opts.GetMQTTConnectOption(t.clientID))
+	if err != nil {
+		return err
+	}
+	if connAck.ReasonCode != 0 {
+		return fmt.Errorf("failed to establish the connection: %s", connAck.String())
+	}
+
+	// Initialize closeChan and msgChan to support reconnect cycles
+	t.closeChan = make(chan struct{})
+	// TODO consider to make the channel size configurable
+	t.msgChan = make(chan *paho.Publish, 100)
+
+	logger.Info("mqtt is connected", "brokerHost", t.opts.Dialer.BrokerHost)
+
+	return nil
+}
+
+func (t *mqttTransport) Send(ctx context.Context, evt cloudevents.Event) error {
+	t.mu.RLock()
+	defer t.mu.RUnlock()
+
+	if t.client == nil {
+		return fmt.Errorf("transport not connected")
+	}
+
+	topic, err := t.getPublishTopic(ctx, evt)
+	if err != nil {
+		return err
+	}
+
+	msg := &paho.Publish{
+		QoS:   byte(t.opts.PubQoS),
+		Topic: topic,
+	}
+
+	if err := cloudeventsmqtt.WritePubMessage(ctx, (*binding.EventMessage)(&evt), msg); err != nil {
+		return err
+	}
+
+	if _, err := t.client.Publish(ctx, msg); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (t *mqttTransport) Subscribe(ctx context.Context) error {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+
+	if t.client == nil {
+		return fmt.Errorf("transport not connected")
+	}
+
+	if t.subscribed {
+		return fmt.Errorf("transport has already subscribed")
+	}
+
+	subscribe, err := t.getSubscribe()
+	if err != nil {
+		return err
+	}
+
+	t.client.AddOnPublishReceived(func(m paho.PublishReceived) (bool, error) {
+		select {
+		case t.msgChan <- m.Packet:
+			return true, nil
+		case <-t.getCloseChan():
+			// Only drop messages if we're shutting down
+			return false, fmt.Errorf("transport closed")
+		}
+		// No default case - will block until channel has space (no message loss)
+	})
+
+	if _, err := t.client.Subscribe(ctx, subscribe); err != nil {
+		return err
+	}
+
+	t.subscribed = true
+
+	logger := klog.FromContext(ctx)
+	for _, sub := range subscribe.Subscriptions {
+		logger.Info("subscribed to mqtt broker", "topic", sub.Topic, "QoS", sub.QoS)
+	}
+
+	return nil
+}
+
+// Receive starts receiving events and invokes the provided handler for each event.
+// This is a BLOCKING call that runs an event loop until the context is cancelled.
+func (t *mqttTransport) Receive(ctx context.Context, handleFn options.ReceiveHandlerFn) error {
+	t.mu.RLock()
+	if !t.subscribed {
+		t.mu.RUnlock()
+		return fmt.Errorf("transport not subscribed")
+	}
+	t.mu.RUnlock()
+
+	logger := klog.FromContext(ctx)
+	for {
+		select {
+		case <-ctx.Done():
+			return nil
+		case <-t.getCloseChan():
+			return nil
+		case m, ok := <-t.msgChan:
+			if !ok {
+				return nil
+			}
+			evt, err := binding.ToEvent(ctx, cloudeventsmqtt.NewMessage(m))
+			if err != nil {
+				logger.Error(err, "invalid event")
+				continue
+			}
+
+			handleFn(ctx, *evt)
+		}
+	}
+}
+
+func (t *mqttTransport) ErrorChan() <-chan error {
+	return t.errorChan
+}
+
+func (t *mqttTransport) Close(ctx context.Context) error {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	klog.FromContext(ctx).Info("close mqtt transport")
+
+	if t.client == nil {
+		// no client, do nothing
+		return nil
+	}
+
+	// Guard against double-close panic - check if already closed
+	if t.closeChan != nil {
+		select {
+		case <-t.closeChan:
+			// already closed
+		default:
+			close(t.closeChan)
+		}
+	}
+
+	t.subscribed = false
+
+	return t.client.Disconnect(&paho.Disconnect{ReasonCode: 0})
+}
+
+func (t *mqttTransport) getCloseChan() chan struct{} {
+	t.mu.RLock()
+	defer t.mu.RUnlock()
+
+	return t.closeChan
+}