feat: introduce telemetry module

config/default/kustomization.yaml

@@ -31,7 +31,7 @@ patchesStrategicMerge:
 # If you want your controller-manager to expose the /metrics
 # endpoint w/o any authn/z, please comment the following line.
 - manager_auth_proxy_patch.yaml
-
+- manager_config_patch.yaml
 
 
 # [WEBHOOK] To enable webhook, uncomment all the sections with [WEBHOOK] prefix including the one in

config/default/manager_auth_proxy_patch.yaml

@@ -8,22 +8,6 @@ metadata:
 spec:
   template:
     spec:
-      affinity:
-        nodeAffinity:
-          requiredDuringSchedulingIgnoredDuringExecution:
-            nodeSelectorTerms:
-              - matchExpressions:
-                - key: kubernetes.io/arch
-                  operator: In
-                  values:
-                    - amd64
-                    - arm64
-                    - ppc64le
-                    - s390x
-                - key: kubernetes.io/os
-                  operator: In
-                  values:
-                    - linux
       containers:
       - name: kube-rbac-proxy
         securityContext:

config/manager/kustomization.yaml

@@ -5,4 +5,4 @@ kind: Kustomization
 images:
 - name: controller
   newName: oceanbasedev/ob-operator
-  newTag: 2.0.1-alpha.1
+  newTag: 2.0.1-alpha.2

config/manager/manager.yaml

@@ -40,22 +40,28 @@ spec:
       # according to the platforms which are supported by your solution.
       # It is considered best practice to support multiple architectures. You can
       # build your manager image using the makefile target docker-buildx.
-      # affinity:
-      #   nodeAffinity:
-      #     requiredDuringSchedulingIgnoredDuringExecution:
-      #       nodeSelectorTerms:
-      #         - matchExpressions:
-      #           - key: kubernetes.io/arch
-      #             operator: In
-      #             values:
-      #               - amd64
-      #               - arm64
-      #               - ppc64le
-      #               - s390x
-      #           - key: kubernetes.io/os
-      #             operator: In
-      #             values:
-      #               - linux
+      affinity:
+        nodeAffinity:
+          requiredDuringSchedulingIgnoredDuringExecution:
+            nodeSelectorTerms:
+              - matchExpressions:
+                - key: kubernetes.io/arch
+                  operator: In
+                  values:
+                    - amd64
+                    - arm64
+                    - ppc64le
+                    - s390x
+                - key: kubernetes.io/os
+                  operator: In
+                  values:
+                    - linux
+                - key: kubernetes.io/hostname
+                  operator: In
+                  values:
+                    - sqaappnoxdnv62s2011161204053.sa128
+                - key: node-role.kubernetes.io/master
+                  operator: Exists
       securityContext:
         runAsNonRoot: true
         # TODO(user): For common cases that do not require escalating privileges
@@ -72,6 +78,11 @@ spec:
         - --leader-elect
         image: controller:latest
         name: manager
+        env:
+          # - name: TELEMETRY_DEBUG
+          #   value: "true"
+          - name: TELEMETRY_SIGNATURE
+            value: "dbe97393a695335d67de91dd4049ba"
         securityContext:
           allowPrivilegeEscalation: false
           capabilities:
@@ -93,7 +104,7 @@ spec:
         # More info: https://kubernetes.io/docs/concepts/configuration/manage-resources-containers/
         resources:
           limits:
-            cpu: 1
+            cpu: "1"
             memory: 1Gi
           requests:
             cpu: 10m

deploy/operator.yaml

@@ -11963,6 +11963,14 @@ rules:
   verbs:
   - create
   - patch
+- apiGroups:
+  - ""
+  resources:
+  - nodes
+  verbs:
+  - get
+  - list
+  - watch
 - apiGroups:
   - ""
   resources:
@@ -12614,6 +12622,12 @@ spec:
                 operator: In
                 values:
                 - linux
+              - key: kubernetes.io/hostname
+                operator: In
+                values:
+                - sqaappnoxdnv62s2011161204053.sa128
+              - key: node-role.kubernetes.io/master
+                operator: Exists
       containers:
       - args:
         - --health-probe-bind-address=:8081
@@ -12622,7 +12636,10 @@ spec:
         - --manager-namespace=oceanbase-system
         command:
         - /manager
-        image: oceanbasedev/ob-operator:2.0.1-alpha.1
+        env:
+        - name: TELEMETRY_SIGNATURE
+          value: dbe97393a695335d67de91dd4049ba
+        image: oceanbasedev/ob-operator:2.0.1-alpha.2
         livenessProbe:
           httpGet:
             path: /healthz
@@ -12642,7 +12659,7 @@ spec:
           periodSeconds: 10
         resources:
           limits:
-            cpu: 1
+            cpu: "1"
             memory: 1Gi
           requests:
             cpu: 10m

deploy/tenant.yaml

@@ -2,7 +2,7 @@ apiVersion: oceanbase.oceanbase.com/v1alpha1
 kind: OBTenant		
 metadata:		
   name: t1
-  namespace: oceanbase
+  # namespace: oceanbase
 spec:	
   obcluster: test
   tenantName: t1

deploy/tenant_restore.yaml

@@ -16,8 +16,6 @@ spec:
     standbyRo: t1s-ro
   source:
     restore:
-      # sourceUri: "file:///ob-backup/t1/data_backup_custom1,file:///ob-backup/t1/log_archive_custom1"
-      # path in xxxSource should 
       archiveSource:
         type: NFS
         path: "t1/log_archive_custom"

pkg/controller/obcluster_controller.go

@@ -29,6 +29,7 @@ import (
 
 	v1alpha1 "github.com/oceanbase/ob-operator/api/v1alpha1"
 	"github.com/oceanbase/ob-operator/pkg/resource"
+	"github.com/oceanbase/ob-operator/pkg/telemetry"
 )
 
 // OBClusterReconciler reconciles a OBCluster object
@@ -84,6 +85,7 @@ func (r *OBClusterReconciler) Reconcile(ctx context.Context, req ctrl.Request) (
 		Client:    r.Client,
 		Recorder:  r.Recorder,
 		Logger:    &logger,
+		Telemetry: telemetry.NewTelemetry(ctx, r.Recorder),
 	}
 	coordinator := resource.NewCoordinator(obclusterManager, &logger)
 	return coordinator.Coordinate()

pkg/controller/obparameter_controller.go

@@ -28,6 +28,7 @@ import (
 
 	v1alpha1 "github.com/oceanbase/ob-operator/api/v1alpha1"
 	"github.com/oceanbase/ob-operator/pkg/resource"
+	"github.com/oceanbase/ob-operator/pkg/telemetry"
 )
 
 // OBParameterReconciler reconciles a OBParameter object
@@ -72,6 +73,7 @@ func (r *OBParameterReconciler) Reconcile(ctx context.Context, req ctrl.Request)
 		Client:      r.Client,
 		Recorder:    r.Recorder,
 		Logger:      &logger,
+		Telemetry:   telemetry.NewTelemetry(ctx, r.Recorder),
 	}
 	coordinator := resource.NewCoordinator(obparameterManager, &logger)
 	return coordinator.Coordinate()

pkg/controller/observer_controller.go

@@ -32,6 +32,7 @@ import (
 
 	v1alpha1 "github.com/oceanbase/ob-operator/api/v1alpha1"
 	"github.com/oceanbase/ob-operator/pkg/resource"
+	"github.com/oceanbase/ob-operator/pkg/telemetry"
 )
 
 // OBServerReconciler reconciles a OBServer object
@@ -77,11 +78,12 @@ func (r *OBServerReconciler) Reconcile(ctx context.Context, req ctrl.Request) (c
 
 	// create observer manager
 	observerManager := &resource.OBServerManager{
-		Ctx:      ctx,
-		OBServer: observer,
-		Client:   r.Client,
-		Recorder: r.Recorder,
-		Logger:   &logger,
+		Ctx:       ctx,
+		OBServer:  observer,
+		Client:    r.Client,
+		Recorder:  r.Recorder,
+		Logger:    &logger,
+		Telemetry: telemetry.NewTelemetry(ctx, r.Recorder),
 	}
 
 	// execute finalizers

pkg/controller/obtenant_controller.go

@@ -27,6 +27,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/log"
 
 	"github.com/oceanbase/ob-operator/pkg/resource"
+	"github.com/oceanbase/ob-operator/pkg/telemetry"
 	"github.com/oceanbase/ob-operator/pkg/util/codec"
 
 	v1alpha1 "github.com/oceanbase/ob-operator/api/v1alpha1"
@@ -84,11 +85,12 @@ func (r *OBTenantReconciler) Reconcile(ctx context.Context, req ctrl.Request) (c
 
 	// create observer manager
 	obtenantManager := &resource.OBTenantManager{
-		Ctx:      ctx,
-		OBTenant: obtenant,
-		Client:   r.Client,
-		Recorder: r.Recorder,
-		Logger:   &logger,
+		Ctx:       ctx,
+		OBTenant:  obtenant,
+		Client:    r.Client,
+		Recorder:  r.Recorder,
+		Logger:    &logger,
+		Telemetry: telemetry.NewTelemetry(ctx, r.Recorder),
 	}
 
 	coordinator := resource.NewCoordinator(obtenantManager, &logger)

pkg/controller/obtenantbackup_controller.go

@@ -19,16 +19,19 @@ package controller
 import (
 	"context"
 	"fmt"
+	"sync"
 	"time"
 
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/tools/record"
+	"k8s.io/client-go/util/retry"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/log"
 
 	"github.com/oceanbase/ob-operator/pkg/resource"
+	"github.com/oceanbase/ob-operator/pkg/telemetry"
 
 	"github.com/pkg/errors"
 
@@ -41,10 +44,12 @@ import (
 // OBTenantBackupReconciler reconciles a OBTenantBackup object
 type OBTenantBackupReconciler struct {
 	client.Client
-	Scheme   *runtime.Scheme
-	Recorder record.EventRecorder
+	Scheme    *runtime.Scheme
+	Recorder  record.EventRecorder
+	Telemetry telemetry.Telemetry
 
-	con *operation.OceanbaseOperationManager
+	telemetryOnce sync.Once
+	con           *operation.OceanbaseOperationManager
 }
 
 //+kubebuilder:rbac:groups=oceanbase.oceanbase.com,resources=obtenantbackups,verbs=get;list;watch;create;update;patch;delete
@@ -124,23 +129,31 @@ func (r *OBTenantBackupReconciler) createBackupJobInOB(ctx context.Context, job
 		password, err := resource.ReadPassword(r.Client, job.Namespace, job.Spec.EncryptionSecret)
 		if err != nil {
 			logger.Error(err, "failed to read backup encryption secret")
-			r.Recorder.Event(job, "Warning", "ReadBackupEncryptionSecretFailed", err.Error())
+			r.getTelemetry(ctx).Event(job, "Warning", "ReadBackupEncryptionSecretFailed", err.Error())
 		} else if password != "" {
 			err = con.SetBackupPassword(password)
 			if err != nil {
 				logger.Error(err, "failed to set backup password")
-				r.Recorder.Event(job, "Warning", "SetBackupPasswordFailed", err.Error())
+				r.getTelemetry(ctx).Event(job, "Warning", "SetBackupPasswordFailed", err.Error())
 			}
 		}
 	}
 	latest, err := con.CreateAndReturnBackupJob(job.Spec.Type)
 	if err != nil {
 		logger.Error(err, "failed to create and return backup job")
+		r.getTelemetry(ctx).Event(job, "Warning", "CreateAndReturnBackupJobFailed", err.Error())
 		return err
 	}
 
 	job.Status.BackupJob = latest
-	return r.Status().Update(ctx, job)
+	err = r.retryUpdateStatus(ctx, job)
+	if err != nil {
+		logger.Error(err, "failed to update status")
+		r.getTelemetry(ctx).Event(job, "Warning", "UpdateStatusFailed", err.Error())
+		return err
+	}
+	r.getTelemetry(ctx).Event(job, "Create", "", "create backup job successfully")
+	return nil
 }
 
 // TODO: Calculate the progress of running jobs
@@ -187,7 +200,7 @@ func (r *OBTenantBackupReconciler) maintainRunningBackupJob(ctx context.Context,
 	case "CANCELED":
 		job.Status.Status = constants.BackupJobStatusCanceled
 	}
-	return r.Client.Status().Update(ctx, job)
+	return r.retryUpdateStatus(ctx, job)
 }
 
 func (r *OBTenantBackupReconciler) maintainRunningBackupCleanJob(ctx context.Context, job *v1alpha1.OBTenantBackup) error {
@@ -219,7 +232,7 @@ func (r *OBTenantBackupReconciler) maintainRunningBackupCleanJob(ctx context.Con
 		case "DOING":
 			job.Status.Status = constants.BackupJobStatusRunning
 		}
-		return r.Client.Status().Update(ctx, job)
+		return r.retryUpdateStatus(ctx, job)
 	}
 
 	return nil
@@ -252,7 +265,7 @@ func (r *OBTenantBackupReconciler) maintainRunningArchiveLogJob(ctx context.Cont
 		case "SUSPEND":
 			job.Status.Status = constants.BackupJobStatusSuspend
 		}
-		return r.Client.Status().Update(ctx, job)
+		return r.retryUpdateStatus(ctx, job)
 	}
 
 	return nil
@@ -279,3 +292,28 @@ func (r *OBTenantBackupReconciler) getObOperationClient(ctx context.Context, job
 	r.con = con
 	return con, nil
 }
+
+func (r *OBTenantBackupReconciler) retryUpdateStatus(ctx context.Context, job *v1alpha1.OBTenantBackup) error {
+	return retry.RetryOnConflict(retry.DefaultRetry, func() error {
+		newestJob := &v1alpha1.OBTenantBackup{}
+		err := r.Get(ctx, types.NamespacedName{
+			Namespace: job.GetNamespace(),
+			Name:      job.GetName(),
+		}, newestJob)
+		if err != nil {
+			return client.IgnoreNotFound(err)
+		}
+		newestJob.Status = job.Status
+		return r.Status().Update(ctx, newestJob)
+	})
+}
+
+func (r *OBTenantBackupReconciler) getTelemetry(ctx context.Context) telemetry.Telemetry {
+	if r.Telemetry != nil {
+		return r.Telemetry
+	}
+	r.telemetryOnce.Do(func() {
+		r.Telemetry = telemetry.NewTelemetry(ctx, r.Recorder)
+	})
+	return r.Telemetry
+}

pkg/controller/obtenantbackuppolicy_controller.go

@@ -27,6 +27,7 @@ import (
 
 	"github.com/oceanbase/ob-operator/api/v1alpha1"
 	"github.com/oceanbase/ob-operator/pkg/resource"
+	"github.com/oceanbase/ob-operator/pkg/telemetry"
 )
 
 // OBTenantBackupPolicyReconciler reconciles a OBTenantBackupPolicy object
@@ -74,6 +75,7 @@ func (r *OBTenantBackupPolicyReconciler) Reconcile(ctx context.Context, req ctrl
 		Client:       r.Client,
 		Recorder:     r.Recorder,
 		Logger:       &logger,
+		Telemetry:    telemetry.NewTelemetry(ctx, r.Recorder),
 	}
 
 	coordinator := resource.NewCoordinator(mgr, &logger)