feat: Support Reconciliation Tracing

apis/trace/v1/reconciliationtrace_types.go

@@ -149,35 +149,35 @@ type DryRunResult struct {
 	// SpecDiff describes the diff between the current spec and the final spec.
 	// The whole spec struct will be compared and an example SpecDiff looks like:
 	// {
-	//		Affinity: {
-	//    		PodAntiAffinity: "Preferred",
-	//    		Tenancy: "SharedNode",
-	//		},
+	//  	Affinity: {
+	//  		PodAntiAffinity: "Preferred",
+	//  		Tenancy: "SharedNode",
+	//  	},
 	//  	ComponentSpecs: {
-	//			{
+	//  		{
 	//  			ComponentDef: "postgresql",
-	//    			Name: "postgresql",
-	// -    		Replicas: 2,
-	// +    		Replicas: 3,
-	//    			Resources:
-	//				{
-	//      			Limits:
-	//					{
-	// -        			CPU: 500m,
-	// +        			CPU: 800m,
-	// -       				Memory: 512Mi,
-	// +       				Memory: 768Mi,
-	//					},
-	//      			Requests:
-	//					{
-	// -      				CPU: 500m,
-	// +       				CPU: 800m,
-	// -      				Memory: 512Mi,
-	// +       				Memory: 768Mi,
-	//					},
-	//				},
-	//			},
-	//		},
+	//  			Name: "postgresql",
+	// -			Replicas: 2,
+	// +			Replicas: 3,
+	//  			Resources:
+	//  			{
+	//  				Limits:
+	//  				{
+	// -					CPU: 500m,
+	// +					CPU: 800m,
+	// -					Memory: 512Mi,
+	// +					Memory: 768Mi,
+	//  				},
+	//  				Requests:
+	//  				{
+	// -					CPU: 500m,
+	// +					CPU: 800m,
+	// -					Memory: 512Mi,
+	// +					Memory: 768Mi,
+	//  				},
+	//  			},
+	//  		},
+	//  	},
 	// }
 	//
 	SpecDiff string `json:"specDiff"`

apis/workloads/v1/groupversion_info.go

@@ -38,4 +38,4 @@ var (
 	AddToScheme = SchemeBuilder.AddToScheme
 )
 
-const Kind = "InstanceSet"
+const InstanceSetKind = "InstanceSet"

cmd/dataprotection/main.go

@@ -207,7 +207,7 @@ func main() {
 	if len(managedNamespaces) > 0 {
 		setupLog.Info(fmt.Sprintf("managed namespaces: %s", managedNamespaces))
 	}
-	mgr, err := ctrl.NewManager(intctrlutil.GeKubeRestConfig(userAgent), ctrl.Options{
+	mgr, err := ctrl.NewManager(intctrlutil.GetKubeRestConfig(userAgent), ctrl.Options{
 		Scheme: scheme,
 		Metrics: server.Options{
 			BindAddress:   metricsAddr,

cmd/manager/main.go

@@ -319,7 +319,7 @@ func main() {
 	userAgent = viper.GetString(userAgentFlagKey.viperName())
 
 	setupLog.Info("golang runtime metrics.", "featureGate", intctrlutil.EnabledRuntimeMetrics())
-	mgr, err := ctrl.NewManager(intctrlutil.GeKubeRestConfig(userAgent), ctrl.Options{
+	mgr, err := ctrl.NewManager(intctrlutil.GetKubeRestConfig(userAgent), ctrl.Options{
 		Scheme: scheme,
 		Metrics: server.Options{
 			BindAddress:   metricsAddr,

config/crd/bases/trace.kubeblocks.io_reconciliationtraces.yaml

@@ -865,13 +865,14 @@ spec:
                   specDiff:
                     description: "SpecDiff describes the diff between the current
                       spec and the final spec.\nThe whole spec struct will be compared
-                      and an example SpecDiff looks like:\n{\n\t\tAffinity: {\n   \t\tPodAntiAffinity:
-                      \"Preferred\",\n   \t\tTenancy: \"SharedNode\",\n\t\t},\n \tComponentSpecs:
-                      {\n\t\t\t{\n \t\t\tComponentDef: \"postgresql\",\n   \t\t\tName:
-                      \"postgresql\",\n-    \t\tReplicas: 2,\n   \t\t\tResources:\n\t\t\t\t{\n
-                      \    \t\t\tLimits:\n\t\t\t\t\t{\n-        \t\t\tCPU: 500m,\n-
-                      \      \t\t\t\tMemory: 512Mi,\n\t\t\t\t\t},\n     \t\t\tRequests:\n\t\t\t\t\t{\n-
-                      \     \t\t\t\tCPU: 500m,\n-      \t\t\t\tMemory: 512Mi,\n\t\t\t\t\t},\n\t\t\t\t},\n\t\t\t},\n\t\t},\n}"
+                      and an example SpecDiff looks like:\n{\n \tAffinity: {\n \t\tPodAntiAffinity:
+                      \"Preferred\",\n \t\tTenancy: \"SharedNode\",\n \t},\n \tComponentSpecs:
+                      {\n \t\t{\n \t\t\tComponentDef: \"postgresql\",\n \t\t\tName:
+                      \"postgresql\",\n-\t\t\tReplicas: 2,\n \t\t\tResources:\n \t\t\t{\n
+                      \t\t\t\tLimits:\n \t\t\t\t{\n-\t\t\t\t\tCPU: 500m,\n-\t\t\t\t\tMemory:
+                      512Mi,\n \t\t\t\t},\n \t\t\t\tRequests:\n \t\t\t\t{\n-\t\t\t\t\tCPU:
+                      500m,\n-\t\t\t\t\tMemory: 512Mi,\n \t\t\t\t},\n \t\t\t},\n \t\t},\n
+                      \t},\n}"
                     type: string
                 required:
                 - desiredSpecRevision

controllers/apps/component_utils_test.go

@@ -69,7 +69,7 @@ var _ = Describe("Component Utils", func() {
 			)
 			pod := testapps.MockInstanceSetPod(&testCtx, nil, clusterName, compName, podName, role, mode)
 			ppod := testapps.NewPodFactory(testCtx.DefaultNamespace, "pod").
-				SetOwnerReferences(workloads.GroupVersion.String(), workloads.Kind, nil).
+				SetOwnerReferences(workloads.GroupVersion.String(), workloads.InstanceSetKind, nil).
 				AddAppInstanceLabel(clusterName).
 				AddAppComponentLabel(compName).
 				AddAppManagedByLabel().

controllers/apps/configuration/policy_util_test.go

@@ -42,14 +42,14 @@ import (
 
 var (
 	defaultNamespace = "default"
-	itsSchemaKind    = workloads.GroupVersion.WithKind(workloads.Kind)
+	itsSchemaKind    = workloads.GroupVersion.WithKind(workloads.InstanceSetKind)
 )
 
 func newMockInstanceSet(replicas int, name string, labels map[string]string) workloads.InstanceSet {
 	uid, _ := password.Generate(12, 12, 0, true, false)
 	return workloads.InstanceSet{
 		TypeMeta: metav1.TypeMeta{
-			Kind:       workloads.Kind,
+			Kind:       workloads.InstanceSetKind,
 			APIVersion: workloads.GroupVersion.String(),
 		},
 		ObjectMeta: metav1.ObjectMeta{

controllers/apps/transform_utils.go

@@ -203,7 +203,7 @@ func isOwnedByComp(obj client.Object) bool {
 // isOwnedByInstanceSet is used to judge if the obj is owned by the InstanceSet controller
 func isOwnedByInstanceSet(obj client.Object) bool {
 	for _, ref := range obj.GetOwnerReferences() {
-		if ref.Kind == workloads.Kind && ref.Controller != nil && *ref.Controller {
+		if ref.Kind == workloads.InstanceSetKind && ref.Controller != nil && *ref.Controller {
 			return true
 		}
 	}

controllers/apps/transform_utils_test.go

@@ -100,7 +100,7 @@ func TestIsOwnedByInstanceSet(t *testing.T) {
 
 	its.OwnerReferences = []metav1.OwnerReference{
 		{
-			Kind:       workloads.Kind,
+			Kind:       workloads.InstanceSetKind,
 			Controller: pointer.Bool(true),
 		},
 	}

controllers/apps/transformer_component_status.go

@@ -290,7 +290,7 @@ func (t *componentStatusTransformer) hasFailedPod() (bool, appsv1alpha1.Componen
 	hasFailedPod := meta.IsStatusConditionTrue(t.runningITS.Status.Conditions, string(workloads.InstanceFailure))
 	if hasFailedPod {
 		failureCondition := meta.FindStatusCondition(t.runningITS.Status.Conditions, string(workloads.InstanceFailure))
-		messages.SetObjectMessage(workloads.Kind, t.runningITS.Name, failureCondition.Message)
+		messages.SetObjectMessage(workloads.InstanceSetKind, t.runningITS.Name, failureCondition.Message)
 		return true, messages
 	}
 
@@ -309,7 +309,7 @@ func (t *componentStatusTransformer) hasFailedPod() (bool, appsv1alpha1.Componen
 	probeTimeoutDuration := time.Duration(defaultRoleProbeTimeoutAfterPodsReady) * time.Second
 	condition := meta.FindStatusCondition(t.runningITS.Status.Conditions, string(workloads.InstanceReady))
 	if time.Now().After(condition.LastTransitionTime.Add(probeTimeoutDuration)) {
-		messages.SetObjectMessage(workloads.Kind, t.runningITS.Name, "Role probe timeout, check whether the application is available")
+		messages.SetObjectMessage(workloads.InstanceSetKind, t.runningITS.Name, "Role probe timeout, check whether the application is available")
 		return true, messages
 	}
 

controllers/trace/change_capture_store.go

@@ -22,7 +22,6 @@ package trace
 import (
 	"sort"
 	"strconv"
-	"sync/atomic"
 
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
@@ -47,16 +46,16 @@ type changeCaptureStore struct {
 	scheme    *runtime.Scheme
 	formatter descriptionFormatter
 	store     map[model.GVKNObjKey]client.Object
-	clock     atomic.Int64
+	clock     int64
 	changes   []tracev1.ObjectChange
 }
 
 func (s *changeCaptureStore) Load(objects ...client.Object) error {
 	for _, object := range objects {
 		// sync the clock
 		revision := parseRevision(object.GetResourceVersion())
-		if revision > s.clock.Load() {
-			s.clock.Store(revision)
+		if revision > s.clock {
+			s.clock = revision
 		}
 		objectRef, err := getObjectRef(object, s.scheme)
 		if err != nil {
@@ -161,7 +160,8 @@ func newChangeCaptureStore(scheme *runtime.Scheme, formatter descriptionFormatte
 }
 
 func (s *changeCaptureStore) applyRevision() string {
-	return strconv.FormatInt(s.clock.Add(1), 10)
+	s.clock++
+	return strconv.FormatInt(s.clock, 10)
 }
 
 func (s *changeCaptureStore) captureCreation(objectRef *model.GVKNObjKey, object client.Object) {

controllers/trace/desired_state_handler_test.go

@@ -249,6 +249,11 @@ var _ = Describe("desired_state_handler test", func() {
 					list.Items = []kbappsv1.ComponentVersion{*componentVersion.DeepCopy()}
 					return nil
 				}).AnyTimes()
+			k8sMock.EXPECT().
+				List(gomock.Any(), &kbappsv1.SidecarDefinitionList{}, gomock.Any()).
+				DoAndReturn(func(_ context.Context, list *kbappsv1.SidecarDefinitionList, _ ...client.ListOption) error {
+					return nil
+				}).AnyTimes()
 			k8sMock.EXPECT().
 				List(gomock.Any(), &dpv1alpha1.BackupPolicyTemplateList{}, gomock.Any()).
 				DoAndReturn(func(_ context.Context, list *dpv1alpha1.BackupPolicyTemplateList, _ ...client.ListOption) error {

controllers/trace/dry_run_handler_test.go

@@ -227,6 +227,11 @@ var _ = Describe("dry_run_handler test", func() {
 					list.Items = []kbappsv1.ComponentVersion{*componentVersion.DeepCopy()}
 					return nil
 				}).AnyTimes()
+			k8sMock.EXPECT().
+				List(gomock.Any(), &kbappsv1.SidecarDefinitionList{}, gomock.Any()).
+				DoAndReturn(func(_ context.Context, list *kbappsv1.SidecarDefinitionList, _ ...client.ListOption) error {
+					return nil
+				}).AnyTimes()
 			k8sMock.EXPECT().
 				List(gomock.Any(), &dpv1alpha1.BackupPolicyTemplateList{}, gomock.Any()).
 				DoAndReturn(func(_ context.Context, list *dpv1alpha1.BackupPolicyTemplateList, _ ...client.ListOption) error {

controllers/trace/informer_manager.go

@@ -128,7 +128,8 @@ func (m *informerManager) processNextWorkItem() bool {
 	if evt, ok := object.(*corev1.Event); ok {
 		ro, err := m.scheme.New(evt.InvolvedObject.GroupVersionKind())
 		if err != nil {
-			return false
+			m.logger.Error(err, "new an event involved object failed")
+			return true
 		}
 		object, _ = ro.(client.Object)
 		err = m.cli.Get(context.Background(), client.ObjectKey{Namespace: evt.InvolvedObject.Namespace, Name: evt.InvolvedObject.Name}, object)

controllers/trace/mock_client.go

@@ -138,6 +138,9 @@ func (c *mockClient) Delete(ctx context.Context, obj client.Object, opts ...clie
 		object.SetDeletionTimestamp(&ts)
 		return c.store.Update(object)
 	}
+	if len(object.GetFinalizers()) != 0 {
+		return nil
+	}
 	return c.store.Delete(obj)
 }
 
@@ -175,8 +178,8 @@ func doPatch(obj client.Object, patch client.Patch, store ChangeCaptureStore, sc
 	if err != nil {
 		return err
 	}
-	newObj := oldObj.DeepCopyObject().(client.Object)
-	if err = applyPatch(newObj, patch.Type(), patchData); err != nil {
+	newObj, err := applyPatch(oldObj, patch.Type(), patchData)
+	if err != nil {
 		return err
 	}
 	metaChanged := checkMetadata(oldObj, newObj)
@@ -193,11 +196,11 @@ func doPatch(obj client.Object, patch client.Patch, store ChangeCaptureStore, sc
 	return nil
 }
 
-func applyPatch(obj client.Object, patchType types.PatchType, patchData []byte) error {
+func applyPatch(obj client.Object, patchType types.PatchType, patchData []byte) (client.Object, error) {
 	// Convert the object to JSON
 	originalJSON, err := json.Marshal(obj)
 	if err != nil {
-		return fmt.Errorf("failed to marshal original object: %w", err)
+		return nil, fmt.Errorf("failed to marshal original object: %w", err)
 	}
 
 	// Apply the patch
@@ -208,14 +211,32 @@ func applyPatch(obj client.Object, patchType types.PatchType, patchData []byte)
 	case types.MergePatchType:
 		patchedJSON, err = jsonpatch.MergePatch(originalJSON, patchData)
 	default:
-		return fmt.Errorf("unsupported patch type: %s", patchType)
+		return nil, fmt.Errorf("unsupported patch type: %s", patchType)
 	}
 	if err != nil {
-		return fmt.Errorf("failed to apply patch: %w", err)
+		return nil, fmt.Errorf("failed to apply patch: %w", err)
 	}
 
-	// Unmarshal the patched JSON back into the object
-	return json.Unmarshal(patchedJSON, obj)
+	// Unmarshal the patched JSON into a new clean object
+	newObj := obj.DeepCopyObject().(client.Object)
+	zeroOutStruct(newObj)
+	err = json.Unmarshal(patchedJSON, newObj)
+	return newObj, err
+}
+
+// zeroOutStruct zero out a struct
+func zeroOutStruct(i interface{}) {
+	v := reflect.ValueOf(i)
+	if v.Kind() != reflect.Ptr {
+		return // Must be a pointer
+	}
+	elem := v.Elem()
+	if elem.Kind() != reflect.Struct {
+		return // Must be pointing to a struct
+	}
+	// Create zero value of the same type and set it
+	zero := reflect.Zero(elem.Type())
+	elem.Set(zero)
 }
 
 func checkMetadata(oldObj client.Object, newObj client.Object) bool {
@@ -257,7 +278,7 @@ func checkStatus(oldObj client.Object, newObj client.Object) bool {
 }
 
 func (c *mockClient) DeleteAllOf(ctx context.Context, obj client.Object, opts ...client.DeleteAllOfOption) error {
-	panic("not implemented")
+	return fmt.Errorf("DeleteAllOf not implemented")
 }
 
 func (c *mockClient) Status() client.SubResourceWriter {

controllers/trace/object_revision_store.go

@@ -33,6 +33,10 @@ import (
 	"github.com/apecloud/kubeblocks/pkg/controller/model"
 )
 
+// ObjectRevisionStore defines an object store which can get the history revision.
+// WARN: This store is designed only for Reconciliation Trace Controller,
+// it's not thread-safe, it doesn't do a deep copy before returning the object.
+// Don't use it in other place.
 type ObjectRevisionStore interface {
 	Insert(object, reference client.Object) error
 	Get(objectRef *model.GVKNObjKey, revision int64) (client.Object, error)
@@ -67,15 +71,15 @@ func (s *objectRevisionStore) Insert(object, reference client.Object) error {
 	objectMap, ok := s.store[objectRef.GroupVersionKind]
 	if !ok {
 		objectMap = make(map[types.NamespacedName]map[int64]client.Object)
+		s.store[objectRef.GroupVersionKind] = objectMap
 	}
 	revisionMap, ok := objectMap[objectRef.ObjectKey]
 	if !ok {
 		revisionMap = make(map[int64]client.Object)
+		objectMap[objectRef.ObjectKey] = revisionMap
 	}
 	revision := parseRevision(object.GetResourceVersion())
 	revisionMap[revision] = object
-	objectMap[objectRef.ObjectKey] = revisionMap
-	s.store[objectRef.GroupVersionKind] = objectMap
 
 	// update reference counter
 	s.counterLock.Lock()
@@ -111,6 +115,7 @@ func (s *objectRevisionStore) Get(objectRef *model.GVKNObjKey, revision int64) (
 	if !ok {
 		return nil, apierrors.NewNotFound(objectRef.GroupVersion().WithResource(strings.ToLower(objectRef.Kind)).GroupResource(), objectRef.Name)
 	}
+	object = object.DeepCopyObject().(client.Object)
 	return object, nil
 }
 
@@ -122,7 +127,16 @@ func (s *objectRevisionStore) List(gvk *schema.GroupVersionKind) map[types.Names
 	if !ok {
 		return nil
 	}
-	return objectMap
+	objectMapCopy := make(map[types.NamespacedName]map[int64]client.Object, len(objectMap))
+	for name, revisionMap := range objectMap {
+		revisionMapCopy := make(map[int64]client.Object, len(revisionMap))
+		objectMapCopy[name] = revisionMapCopy
+		for revision, object := range revisionMap {
+			objectCopy := object.DeepCopyObject().(client.Object)
+			revisionMapCopy[revision] = objectCopy
+		}
+	}
+	return objectMapCopy
 }
 
 func (s *objectRevisionStore) Delete(objectRef *model.GVKNObjKey, reference client.Object, revision int64) {