add the disaster function code

Author: acekingke

api/v1alpha1/mysqlcluster_types.go

@@ -117,7 +117,8 @@ type MysqlClusterSpec struct {
 	// Bootstraping from remote data source
 	// +optional
 	SourceConfig *corev1.SecretProjection `json:"sourceConfig,omitempty"`
-
+	// remote replica source
+	RemoteCluster *RemoteSourceStruct `json:"remoteCluster,omitempty"`
 	// Leader as follower represents if make leader use as follower to read
 	// +optional
 	// +kubebuilder:default:=false
@@ -140,6 +141,11 @@ type ReadOnlyType struct {
 	Tolerations []corev1.Toleration `json:"tolerations,omitempty"`
 }
 
+type RemoteSourceStruct struct {
+	Name      string `json:"name"`
+	NameSpace string `json:"namespace"`
+}
+
 // MysqlOpts defines the options of MySQL container.
 type MysqlOpts struct {
 	// Specifies mysql image to use.
@@ -366,7 +372,18 @@ const (
 
 // ClusterConditionType defines type for cluster condition type.
 type ClusterConditionType string
+type ClusterConditionsIndex uint8
 
+const (
+	ClIndexInit ClusterConditionsIndex = iota
+	ClIndexUpdate
+	ClIndexReady
+	ClIndexClose
+	ClIndexError
+	ClIndexScaleIn
+	ClIndexScaleOut
+	ClIndexRemoteSlave
+)
 const (
 	// ConditionInit indicates whether the cluster is initializing.
 	ConditionInit ClusterConditionType = "Initializing"
@@ -382,6 +399,8 @@ const (
 	ConditionScaleIn ClusterConditionType = "ScaleIn"
 	// ConditionScaleOut indicates whether the cluster replicas is increasing.
 	ConditionScaleOut ClusterConditionType = "ScaleOut"
+	// is it a slave for Remote cluster?
+	ConditionRemoteSlave ClusterConditionType = "RemoteSlave"
 )
 
 // ClusterCondition defines type for cluster conditions.

api/v1alpha1/zz_generated.deepcopy.go

@@ -98,6 +98,9 @@ func Convert_v1alpha1_MysqlClusterSpec_To_v1beta1_MysqlClusterSpec(in *v1alpha1.
 	if in.SourceConfig != nil {
 		out.DataSource.Remote.SourceConfig = in.SourceConfig
 	}
+	if in.RemoteCluster != nil {
+		out.DataSource.Remote.RemoteCluster = (*RemoteSourceStruct)(in.RemoteCluster)
+	}
 	if len(in.NFSServerAddress) != 0 {
 		ipStr := strings.Split(in.NFSServerAddress, ":")
 		out.DataSource.RestorePoint = in.RestorePoint
@@ -165,6 +168,9 @@ func Convert_v1beta1_MysqlClusterSpec_To_v1alpha1_MysqlClusterSpec(in *MysqlClus
 	if in.DataSource.Remote.SourceConfig != nil {
 		out.SourceConfig = in.DataSource.Remote.SourceConfig
 	}
+	if in.DataSource.Remote.RemoteCluster != nil {
+		out.RemoteCluster = (*v1alpha1.RemoteSourceStruct)(in.DataSource.Remote.RemoteCluster)
+	}
 	if len(in.DataSource.S3Backup.Name) != 0 {
 		out.RestoreFrom = in.DataSource.S3Backup.Name
 		out.BackupSecretName = in.DataSource.S3Backup.SecretName

api/v1beta1/mysqlcluster_conversion.go

@@ -276,6 +276,9 @@ const (
 	ConditionScaleIn ClusterConditionType = "ScaleIn"
 	// ConditionScaleOut indicates whether the cluster replicas is increasing.
 	ConditionScaleOut ClusterConditionType = "ScaleOut"
+
+	// is it a slave for Remote cluster?
+	ConditionRemoteSlave ClusterConditionType = "RemoteSlave"
 )
 
 // ClusterCondition defines type for cluster conditions.
@@ -438,10 +441,15 @@ type DataSource struct {
 	// +optional
 	RestorePoint string `json:"restorePoint"`
 }
-
+type RemoteSourceStruct struct {
+	Name      string `json:"name"`
+	NameSpace string `json:"namespace"`
+}
 type RemoteDataSource struct {
-	//
+	// xtrabackup remote source
 	SourceConfig *corev1.SecretProjection `json:"sourceConfig,omitempty"`
+	// remote replica source
+	RemoteCluster *RemoteSourceStruct `json:"remoteCluster,omitempty"`
 }
 
 type S3BackupDataSource struct {

api/v1beta1/mysqlcluster_types.go

@@ -2246,6 +2246,17 @@ spec:
                 required:
                 - num
                 type: object
+              remoteCluster:
+                description: remote replica source
+                properties:
+                  name:
+                    type: string
+                  namespace:
+                    type: string
+                required:
+                - name
+                - namespace
+                type: object
               replicas:
                 default: 3
                 description: Replicas is the number of pods.
@@ -3503,12 +3514,19 @@ spec:
                   remote:
                     description: Bootstraping from remote data source
                     properties:
+                      remoteCluster:
+                        description: remote replica source
+                        properties:
+                          name:
+                            type: string
+                          namespace:
+                            type: string
+                        required:
+                        - name
+                        - namespace
+                        type: object
                       sourceConfig:
-                        description: "Adapts a secret into a projected volume. \n
-                          The contents of the target Secret's Data field will be presented
-                          in a projected volume as files using the keys in the Data
-                          field as the file names. Note that this is identical to
-                          a secret volume source without the default mode."
+                        description: xtrabackup remote source
                         properties:
                           items:
                             description: If unspecified, each key-value pair in the

api/v1beta1/zz_generated.conversion.go

@@ -164,6 +164,10 @@ func (r *MysqlClusterReconciler) Reconcile(ctx context.Context, req ctrl.Request
 			clustersyncer.NewXenonCMSyncer(r.Client, instance),
 		)
 	}
+
+	if instance.Spec.RemoteCluster != nil {
+		syncers = append(syncers, clustersyncer.NewRemoteClusterCMSyncer(r.Client, instance))
+	}
 	if instance.Spec.MetricsOpts.Enabled {
 		syncers = append(syncers, clustersyncer.NewMetricsSVCSyncer(r.Client, instance))
 	}

api/v1beta1/zz_generated.deepcopy.go

@@ -117,7 +117,9 @@ type sqlRunner struct {
 type SQLRunner interface {
 	QueryExec(query Query) error
 	QueryRow(query Query, dest ...interface{}) error
+	QueryRowContext(ctx context.Context, query Query, dest ...interface{}) error
 	QueryRows(query Query) (*sql.Rows, error)
+	QueryRowsContext(ctx context.Context, query Query) (*sql.Rows, error)
 }
 
 type closeFunc func()
@@ -168,10 +170,24 @@ func (s sqlRunner) QueryExec(query Query) error {
 	return nil
 }
 
+// QueryExec used to run the query with args.
+func (s sqlRunner) ExecContext(ctx context.Context, query Query) error {
+	if _, err := s.db.ExecContext(ctx, query.String(), query.args...); err != nil {
+		return err
+	}
+
+	return nil
+}
+
 func (s sqlRunner) QueryRow(query Query, dest ...interface{}) error {
 	return s.db.QueryRow(query.escapedQuery, query.args...).Scan(dest...)
 }
 
+// QueryExec used to run the query with args.
+func (s sqlRunner) QueryRowContext(ctx context.Context, query Query, dest ...interface{}) error {
+	return s.db.QueryRowContext(ctx, query.String(), query.args...).Scan(dest...)
+}
+
 func (s sqlRunner) QueryRows(query Query) (*sql.Rows, error) {
 	rows, err := s.db.Query(query.escapedQuery, query.args...)
 	if err != nil {
@@ -181,6 +197,15 @@ func (s sqlRunner) QueryRows(query Query) (*sql.Rows, error) {
 	return rows, rows.Err()
 }
 
+func (s sqlRunner) QueryRowsContext(ctx context.Context, query Query) (*sql.Rows, error) {
+	rows, err := s.db.QueryContext(ctx, query.escapedQuery, query.args...)
+	if err != nil {
+		return nil, err
+	}
+
+	return rows, rows.Err()
+}
+
 // CheckSlaveStatusWithRetry check the slave status with retry time.
 func CheckSlaveStatusWithRetry(sqlRunner SQLRunner, retry uint32, replicaLag *int32) (isLagged, isReplicating corev1.ConditionStatus, err error) {
 	for {
@@ -202,8 +227,11 @@ func CheckSlaveStatusWithRetry(sqlRunner SQLRunner, retry uint32, replicaLag *in
 // CheckSlaveStatus check the slave status.
 func CheckSlaveStatus(sqlRunner SQLRunner, ReplicaLag *int32) (isLagged, isReplicating corev1.ConditionStatus, err error) {
 	var rows *sql.Rows
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
 	isLagged, isReplicating = corev1.ConditionUnknown, corev1.ConditionUnknown
-	rows, err = sqlRunner.QueryRows(NewQuery("show slave status;"))
+	rows, err = sqlRunner.QueryRowsContext(ctx, NewQuery("show slave status;"))
 	if err != nil {
 		return
 	}
@@ -287,12 +315,16 @@ func CheckReadOnly(sqlRunner SQLRunner) (corev1.ConditionStatus, error) {
 
 // GetGlobalVariable used to get the global variable by param.
 func GetGlobalVariable(sqlRunner SQLRunner, param string, val interface{}) error {
-	return sqlRunner.QueryRow(NewQuery("select @@global.?", param), val)
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	return sqlRunner.QueryRowContext(ctx, NewQuery("select @@global.?", param), val)
 }
 
 func CheckProcesslist(sqlRunner SQLRunner) (bool, error) {
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
 	var rows *sql.Rows
-	rows, err := sqlRunner.QueryRows(NewQuery("show processlist;"))
+	rows, err := sqlRunner.QueryRowsContext(ctx, NewQuery("show processlist;"))
 	if err != nil {
 		return false, err
 	}

config/crd/bases/mysql.radondb.com_mysqlclusters.yaml

@@ -153,6 +153,18 @@ func (c *initSidecar) getEnvVars() []corev1.EnvVar {
 			Value: "1",
 		})
 	}
+
+	if c.Spec.RemoteCluster != nil {
+		envs = append(envs, corev1.EnvVar{
+			Name:  "REMOTE_CLUSTER_NAME",
+			Value: c.Spec.RemoteCluster.Name,
+		}, corev1.EnvVar{
+			Name:  "REMOTE_CLUSTER_NAMESPACE",
+			Value: c.Spec.RemoteCluster.NameSpace,
+		})
+
+	}
+
 	return envs
 }
 
@@ -264,5 +276,12 @@ func (c *initSidecar) getVolumeMounts() []corev1.VolumeMount {
 			MountPath: utils.RemoteSourcePath,
 		})
 	}
+	if c.Spec.RemoteCluster != nil {
+		volumeMounts = append(volumeMounts,
+			corev1.VolumeMount{
+				Name:      utils.RemoteClusterCMVolumeName,
+				MountPath: utils.RemoteClusterCMMountPath,
+			})
+	}
 	return volumeMounts
 }