feat: add global sync functionality for cross-cluster synchronization

This commit introduces a comprehensive global synchronization feature that
enables synchronization of global objects between Doris clusters.

Features:
- Support for global snapshot creation with configurable options
- Flexible sync options: all, backup_privilege, backup_catalog, backup_workload_group
- Automatic backup completion monitoring with timeout handling
- Global snapshot information retrieval and processing
- SQL statement execution on destination cluster
- Comprehensive error handling and logging throughout the process

API Endpoint:
- POST /sync_global - Handles global synchronization requests

Request Structure:
- name: Unique identifier for the sync operation
- src: Source cluster specification
- dest: Destination cluster specification
- Sync options (at least one required if 'all' is not set):
  - all: Sync all global objects
  - backup_privilege: Sync user privileges
  - backup_catalog: Sync catalog information
  - backup_workload_group: Sync workload group settings

Workflow:
1. Create global snapshot on source cluster
2. Wait for backup completion with polling mechanism
3. Retrieve global snapshot details via RPC
4. Convert snapshot data to SQL statements
5. Execute SQL statements on destination cluster

Additional improvements:
- Internationalize all Chinese logs and comments to English
- Refactor validation logic for flexible option combinations
- Decompose syncGlobal function into smaller, focused functions

This feature enables seamless migration and synchronization of global
configurations between Doris clusters, supporting various deployment
scenarios including disaster recovery and cluster migration.

Author: carolinchen-docker

pkg/ccr/base/spec.go

@@ -83,7 +83,7 @@ func ParseBackupState(state string) BackupState {
 	}
 }
 
-// isSystemDatabase 判断是否为系统数据库，需要跳过
+// isSystemDatabase determines if it's a system database that should be skipped
 func isSystemDatabase(dbName string) bool {
 	systemDatabases := []string{
 		"information_schema",
@@ -501,7 +501,7 @@ func (s *Spec) GetAllDatabases() ([]string, error) {
 		if err := rows.Scan(&database); err != nil {
 			return nil, xerror.Wrapf(err, xerror.Normal, "scan database failed")
 		}
-		// 过滤系统数据库
+		// Filter system databases
 		if !isSystemDatabase(database) {
 			databases = append(databases, database)
 		}
@@ -823,20 +823,20 @@ func (s *Spec) CreateSnapshot(snapshotName string, tables []string) error {
 	return nil
 }
 
-// CreateGlobalSnapshot 创建全局快照，用于同步全局对象
-// 根据不同的选项设置不同的属性
+// CreateGlobalSnapshot creates a global snapshot for synchronizing global objects
+// Sets different properties based on different options
 func (s *Spec) CreateGlobalSnapshot(snapshotName string, backupPrivilege, backupCatalog, backupWorkloadGroup bool) error {
-	log.Infof("创建全局快照 %s", snapshotName)
+	log.Infof("Creating global snapshot %s", snapshotName)
 
 	db, err := s.Connect()
 	if err != nil {
 		return err
 	}
 
-	// 构建SQL语句
+	// Build SQL statement
 	sql := fmt.Sprintf("BACKUP GLOBAL SNAPSHOT %s TO `__keep_on_local__`", utils.FormatKeywordName(snapshotName))
 
-	// 根据不同的选项添加属性
+	// Add properties based on different options
 	properties := make([]string, 0)
 	if backupPrivilege {
 		properties = append(properties, "\"backup_privilege\" = \"true\"")
@@ -848,15 +848,15 @@ func (s *Spec) CreateGlobalSnapshot(snapshotName string, backupPrivilege, backup
 		properties = append(properties, "\"backup_workload_group\" = \"true\"")
 	}
 
-	// 如果有属性，添加PROPERTIES子句
+	// If there are properties, add PROPERTIES clause
 	if len(properties) > 0 {
 		sql += fmt.Sprintf(" PROPERTIES (%s)", strings.Join(properties, ", "))
 	}
 
-	log.Infof("创建全局快照SQL: %s", sql)
+	log.Infof("Creating global snapshot SQL: %s", sql)
 	_, err = db.Exec(sql)
 	if err != nil {
-		return xerror.Wrapf(err, xerror.Normal, "创建全局快照 %s 失败, SQL: %s", snapshotName, sql)
+		return xerror.Wrapf(err, xerror.Normal, "Failed to create global snapshot %s, SQL: %s", snapshotName, sql)
 	}
 
 	return nil
@@ -936,20 +936,20 @@ func (s *Spec) checkBackupFinished(snapshotName string) (BackupState, string, er
 	return BackupStateUnknown, "", xerror.Errorf(xerror.Normal, "no backup state found, sql: %s", sql)
 }
 
-// checkGlobalBackupFinished 检查全局备份状态
+// checkGlobalBackupFinished checks global backup status
 func (s *Spec) checkGlobalBackupFinished(snapshotName string) (BackupState, string, error) {
-	log.Tracef("检查全局备份状态 %s", snapshotName)
+	log.Tracef("Checking global backup status %s", snapshotName)
 
 	db, err := s.Connect()
 	if err != nil {
 		return BackupStateUnknown, "", err
 	}
 
 	sql := fmt.Sprintf("SHOW GLOBAL BACKUP WHERE SnapshotName = \"%s\"", snapshotName)
-	log.Infof("检查全局备份状态SQL: %s", sql)
+	log.Infof("Checking global backup status SQL: %s", sql)
 	rows, err := db.Query(sql)
 	if err != nil {
-		return BackupStateUnknown, "", xerror.Wrapf(err, xerror.Normal, "查询全局备份状态失败, SQL: %s", sql)
+		return BackupStateUnknown, "", xerror.Wrapf(err, xerror.Normal, "Failed to query global backup status, SQL: %s", sql)
 	}
 	defer rows.Close()
 
@@ -961,57 +961,57 @@ func (s *Spec) checkGlobalBackupFinished(snapshotName string) (BackupState, stri
 
 		stateStr, err := rowParser.GetString("State")
 		if err != nil {
-			return BackupStateUnknown, "", xerror.Wrap(err, xerror.Normal, "解析全局备份State失败")
+			return BackupStateUnknown, "", xerror.Wrap(err, xerror.Normal, "Failed to parse global backup State")
 		}
 
 		status, err := rowParser.GetString("Status")
 		if err != nil {
-			return BackupStateUnknown, "", xerror.Wrap(err, xerror.Normal, "解析全局备份Status失败")
+			return BackupStateUnknown, "", xerror.Wrap(err, xerror.Normal, "Failed to parse global backup Status")
 		}
 
-		log.Infof("检查全局快照 %s 备份状态: [%v]", snapshotName, stateStr)
+		log.Infof("Checking global snapshot %s backup status: [%v]", snapshotName, stateStr)
 		return ParseBackupState(stateStr), status, nil
 	}
 
 	if err := rows.Err(); err != nil {
-		return BackupStateUnknown, "", xerror.Wrapf(err, xerror.Normal, "检查全局备份状态, SQL: %s", sql)
+		return BackupStateUnknown, "", xerror.Wrapf(err, xerror.Normal, "Check global backup status, SQL: %s", sql)
 	}
 
-	return BackupStateUnknown, "", xerror.Errorf(xerror.Normal, "未找到全局备份状态, SQL: %s", sql)
+	return BackupStateUnknown, "", xerror.Errorf(xerror.Normal, "Global backup status not found, SQL: %s", sql)
 }
 
-// CheckGlobalBackupFinished 检查全局备份是否完成
+// CheckGlobalBackupFinished checks if global backup is finished
 func (s *Spec) CheckGlobalBackupFinished(snapshotName string) (bool, error) {
-	log.Tracef("检查全局备份是否完成, spec: %s, snapshot: %s", s.String(), snapshotName)
+	log.Tracef("Checking if global backup is finished, spec: %s, snapshot: %s", s.String(), snapshotName)
 
-	// 重试网络相关错误，避免在目标网络中断、进程重启时进行全量同步
+	// Retry network related errors to avoid full sync when target network is interrupted or process is restarted
 	if backupState, status, err := s.checkGlobalBackupFinished(snapshotName); err != nil && !isNetworkRelated(err) {
 		return false, err
 	} else if err == nil && backupState == BackupStateFinished {
 		return true, nil
 	} else if err == nil && backupState == BackupStateCancelled {
-		return false, xerror.Errorf(xerror.Normal, "全局备份失败或被取消, 备份状态: %s", status)
+		return false, xerror.Errorf(xerror.Normal, "Global backup failed or cancelled, backup status: %s", status)
 	} else {
-		// BackupStatePending, BackupStateUnknown 或网络相关错误
+		// BackupStatePending, BackupStateUnknown or network related errors
 		if err != nil {
-			log.Warnf("检查全局备份状态失败, spec: %s, snapshot: %s, err: %v", s.String(), snapshotName, err)
+			log.Warnf("Failed to check global backup status, spec: %s, snapshot: %s, err: %v", s.String(), snapshotName, err)
 		}
 		return false, nil
 	}
 }
 
 func (s *Spec) RestoreGlobalInfo(sqls string) error {
-	log.Infof("恢复全局信息")
+	log.Infof("Restoring global information")
 
 	db, err := s.Connect()
 	if err != nil {
 		return err
 	}
 
-	log.Infof("执行全局恢复SQL: %s", sqls)
+	log.Infof("Executing global restore SQL: %s", sqls)
 	_, err = db.Exec(sqls)
 	if err != nil {
-		return xerror.Wrapf(err, xerror.Normal, "恢复全局信息失败, SQL: %s", sqls)
+		return xerror.Wrapf(err, xerror.Normal, "Failed to restore global information, SQL: %s", sqls)
 	}
 
 	return nil

pkg/rpc/fe.go

@@ -472,12 +472,6 @@ func (rpc *FeRpc) GetSnapshot(spec *base.Spec, labelName string, compress bool)
 	return convertResult[festruct.TGetSnapshotResult_](result, err)
 }
 
-// func (rpc *FeRpc) GetGlobalSnapshot(spec *base.Spec, labelName string) (*festruct.TGetGlobalSnapshotResult_, error) {
-// 	// GetGlobalSnapshot 不支持主节点重定向，直接调用主节点客户端
-// 	// 这是因为 TGetGlobalSnapshotResult_ 结构体不包含 MasterAddress 字段
-// 	return rpc.masterClient.GetGlobalSnapshot(spec, labelName)
-// }
-
 func (rpc *FeRpc) GetGlobalSnapshot(spec *base.Spec, labelName string) (*festruct.TGetGlobalSnapshotResult_, error) {
 	caller := func(client IFeRpc) (resultType, error) {
 		return client.GetGlobalSnapshot(spec, labelName)
@@ -816,15 +810,17 @@ func (rpc *singleFeClient) GetSnapshot(spec *base.Spec, labelName string, compre
 	}
 }
 
-// struct TGetGlobalSnapshotRequest {
-//     1: optional string cluster
-//     2: optional string user
-//     3: optional string passwd
-//     4: optional string token
-//     5: optional string label_name
-//     6: optional string snapshot_name
-//     7: optional TSnapshotType snapshot_type
-// }
+//	struct TGetGlobalSnapshotRequest {
+//	    1: optional string cluster
+//	    2: optional string user
+//	    3: optional string passwd
+//	    4: optional string token
+//	    5: optional string label_name
+//	    6: optional string snapshot_name
+//	    7: optional TSnapshotType snapshot_type
+//	}
+//
+// Get Global Snapshot rpc
 func (rpc *singleFeClient) GetGlobalSnapshot(spec *base.Spec, labelName string) (*festruct.TGetGlobalSnapshotResult_, error) {
 	log.Tracef("Call GetGlobalSnapshot, addr: %s, spec: %s, label: %s", rpc.Address(), spec, labelName)