feat: Add cluster-level CCR sync and dynamic monitor interval configuration

CHANGELOG.md

@@ -1,6 +1,6 @@
 # 更新日志
 
-# 3.0.5/2.1.10
+# 3.0.6/2.1.10
 
 ### Fix
 
@@ -11,17 +11,20 @@
 - 修复 DROP INDEX 导致 UPSERT 找不到 index 的问题 (selectdb/ccr-syncer#490)
 - 修复 unknown column 导致的 VIEW 无法创建的问题 (selectdb/ccr-syncer#510)
 - 修复 gls 未释放导致的内存泄漏问题 (selectdb/ccr-syncer#576)
+- 使用 `force_replace` 修复 VIEW schema 不一致无法同步的问题 (selectdb/ccr-syncer#579)
+- 构建 table mapping 前检查 table 是否已经被删除 (selectdb/ccr-syncer#612)
 
 ### Feature
 
-- 支持 txn insert (selectdb/ccr-syncer#290)
+- 支持 txn insert (selectdb/ccr-syncer#290,selectdb/ccr-syncer#592)
 - 支持 lock binlog，提前释放不需要的 binlog，避免占用上游资源 (selectdb/ccr-syncer#399, selectdb/ccr-syncer#406, selectdb/ccr-syncer#407)
 - 支持一批获取多个 binlog (selectdb/ccr-syncer#400)
 - 增加 `metrics` 接口用于获取 ccr metrics (selectdb/ccr-syncer#402, selectdb/ccr-syncer#461)
 - 支持修改 view comment (selectdb/ccr-syncer#408)
 - 支持幂等性 (selectdb/ccr-syncer#409, selectdb/ccr-syncer#416, selectdb/ccr-syncer#415, selectdb/ccr-syncer#424, ...)
 - 增加 `desync.sh` 脚本 (selectdb/ccr-syncer#452)
 - 支持 pipline txn（并行 ingest，串行提交）(selectdb/ccr-syncer#585)
+- 增加 `/view` 接口，用于获取 JOB 状态（支持 terminal, table, html）(selectdb/ccr-syncer#588)
 
 ### Improve
 

CLUSTER_CCR_CHANGES.md

@@ -0,0 +1,140 @@
+# Cluster CCR Feature Changes
+
+## 概述
+本次更改为CCR Syncer添加了集群级别同步功能，允许用户一次性同步整个集群的所有数据库，并提供了动态配置监控间隔的功能。
+
+## 主要更改
+
+### 1. 新增ClusterSync参数
+
+#### 文件：`pkg/service/http_service.go`
+
+**新增字段：**
+```go
+type CreateCcrRequest struct {
+    // ... 其他字段
+    // Whether it's cluster-level sync, if true, will get all databases from source cluster and create sync task for each database
+    ClusterSync bool `json:"cluster_sync"`
+}
+```
+
+**新增函数：**
+- `createClusterCcr()` - 创建集群级别的CCR同步任务
+- `startDatabaseMonitor()` - 启动数据库监控守护进程
+- `monitorDatabaseChanges()` - 检测数据库变化并处理
+- `handleNewDatabases()` - 处理新增数据库
+- `handleDeletedDatabases()` - 处理删除的数据库
+- `getDatabaseList()` - 获取数据库列表
+
+**功能特性：**
+- 自动获取源集群所有数据库
+- 为每个数据库创建独立的同步任务
+- 支持动态监控新增/删除的数据库
+- 自动创建/删除对应的同步任务
+- 支持失败重试机制
+
+### 2. 动态监控间隔配置
+
+#### 新增全局变量：
+```go
+var (
+    databaseMonitorInterval time.Duration = 2 * time.Minute
+    intervalMutex           sync.RWMutex
+    intervalUpdateChan      = make(chan time.Duration, 1)
+)
+```
+
+#### 新增HTTP Handler：
+
+**`/update_monitor_interval` - 更新监控间隔**
+- 请求方法：POST
+- 请求格式：
+```json
+{
+  "interval_seconds": 300
+}
+```
+- 功能：动态更新数据库监控的检查间隔
+
+**`/get_monitor_interval` - 获取当前监控间隔**
+- 请求方法：GET
+- 响应格式：
+```json
+{
+  "success": true,
+  "interval_seconds": 120
+}
+```
+- 功能：获取当前的监控间隔设置
+
+### 3. 性能优化
+
+**锁优化：**
+- 使用channel通信替代频繁的锁操作
+- 减少了`startDatabaseMonitor`函数中的锁竞争
+- 提高了监控性能
+
+**实现细节：**
+- 使用`select`语句监听ticker和interval更新
+- 非阻塞的channel通信
+- 只在间隔真正改变时才重置ticker
+
+## API使用示例
+
+### 创建集群级同步任务
+```bash
+curl -X POST http://localhost:9190/create_ccr \
+  -H "Content-Type: application/json" \
+  -d '{
+    "name": "cluster_sync_job",
+    "src": {
+      "host": "source-cluster",
+      "port": 9030,
+      "user": "root",
+      "password": "password"
+    },
+    "dest": {
+      "host": "dest-cluster", 
+      "port": 9030,
+      "user": "root",
+      "password": "password"
+    },
+    "cluster_sync": true
+  }'
+```
+
+### 更新监控间隔
+```bash
+curl -X POST http://localhost:9190/update_monitor_interval \
+  -H "Content-Type: application/json" \
+  -d '{"interval_seconds": 300}'
+```
+
+### 获取监控间隔
+```bash
+curl -X GET http://localhost:9190/get_monitor_interval
+```
+
+## 技术特点
+
+1. **线程安全**：使用mutex保护全局变量
+2. **高性能**：优化锁使用，减少竞争
+3. **动态配置**：支持运行时修改监控间隔
+4. **容错性**：支持失败重试和错误处理
+5. **可观测性**：详细的日志记录
+6. **向后兼容**：不影响现有的单数据库同步功能
+
+## 测试建议
+
+1. 测试集群级同步功能
+2. 测试动态监控间隔更新
+3. 测试新增/删除数据库的自动处理
+4. 测试并发场景下的性能
+5. 测试错误恢复机制
+
+## 注意事项
+
+1. 集群级同步会为每个数据库创建独立的同步任务
+2. 任务命名格式：`{原任务名}_{数据库名}`
+3. 监控间隔最小值应大于0
+4. 建议在生产环境中谨慎设置监控间隔

pkg/ccr/base/spec.go

@@ -83,6 +83,22 @@ func ParseBackupState(state string) BackupState {
 	}
 }
 
+// isSystemDatabase 判断是否为系统数据库，需要跳过
+func isSystemDatabase(dbName string) bool {
+	systemDatabases := []string{
+		"information_schema",
+		"mysql",
+		"__internal_schema",
+	}
+
+	for _, sysDb := range systemDatabases {
+		if dbName == sysDb {
+			return true
+		}
+	}
+	return false
+}
+
 type RestoreState int
 
 const (
@@ -464,6 +480,40 @@ func (s *Spec) GetAllTables() ([]string, error) {
 	return tables, nil
 }
 
+func (s *Spec) GetAllDatabases() ([]string, error) {
+	log.Tracef("get all databases from cluster")
+
+	db, err := s.Connect()
+	if err != nil {
+		return nil, err
+	}
+
+	sql := "SHOW DATABASES"
+	rows, err := db.Query(sql)
+	if err != nil {
+		return nil, xerror.Wrapf(err, xerror.Normal, "query %s failed", sql)
+	}
+	defer rows.Close()
+
+	var databases []string
+	for rows.Next() {
+		var database string
+		if err := rows.Scan(&database); err != nil {
+			return nil, xerror.Wrapf(err, xerror.Normal, "scan database failed")
+		}
+		// 过滤系统数据库
+		if !isSystemDatabase(database) {
+			databases = append(databases, database)
+		}
+	}
+
+	if err := rows.Err(); err != nil {
+		return nil, xerror.Wrapf(err, xerror.Normal, "rows error")
+	}
+
+	return databases, nil
+}
+
 func (s *Spec) queryResult(querySQL string, queryColumn string, errMsg string) ([]string, error) {
 	db, err := s.Connect()
 	if err != nil {

pkg/ccr/handle/alter_job_v2.go

@@ -41,54 +41,16 @@ func (*AlterJobV2Handle) Handle(j *ccr.Job, commitSeq int64, alterJob *record.Al
 }
 
 func handleAlterRollup(j *ccr.Job, alterJob *record.AlterJobV2) error {
-	job_progress := j.GetJobProgress()
+	j.SaveAlterRollupShadowIndex(alterJob)
 	if !alterJob.IsFinished() {
-		switch alterJob.JobState {
-		case record.ALTER_JOB_STATE_PENDING:
-			// Once the rollup job step to WAITING_TXN, the upsert to the rollup index is allowed,
-			// but the dest index of the downstream cluster hasn't been created.
-			//
-			// To filter the upsert to the rollup index, save the shadow index ids here.
-			if job_progress.ShadowIndexes == nil {
-				job_progress.ShadowIndexes = make(map[int64]int64)
-			}
-			job_progress.ShadowIndexes[alterJob.RollupIndexId] = alterJob.BaseIndexId
-		case record.ALTER_JOB_STATE_CANCELLED:
-			// clear the shadow indexes
-			delete(job_progress.ShadowIndexes, alterJob.RollupIndexId)
-		}
 		return nil
 	}
 
-	// Once partial snapshot finished, the rollup indexes will be convert to normal index.
-	delete(job_progress.ShadowIndexes, alterJob.RollupIndexId)
-
 	return j.NewPartialSnapshot(alterJob.TableId, alterJob.TableName, nil, true, false)
 }
 
 func handleSchemaChange(j *ccr.Job, alterJob *record.AlterJobV2) error {
 	job_progress := j.GetJobProgress()
-	if !alterJob.IsFinished() {
-		switch alterJob.JobState {
-		case record.ALTER_JOB_STATE_PENDING:
-			// Once the schema change step to WAITING_TXN, the upsert to the shadow indexes is allowed,
-			// but the dest indexes of the downstream cluster hasn't been created.
-			//
-			// To filter the upsert to the shadow indexes, save the shadow index ids here.
-			if job_progress.ShadowIndexes == nil {
-				job_progress.ShadowIndexes = make(map[int64]int64)
-			}
-			for shadowIndexId, originIndexId := range alterJob.ShadowIndexes {
-				job_progress.ShadowIndexes[shadowIndexId] = originIndexId
-			}
-		case record.ALTER_JOB_STATE_CANCELLED:
-			// clear the shadow indexes
-			for shadowIndexId := range alterJob.ShadowIndexes {
-				delete(job_progress.ShadowIndexes, shadowIndexId)
-			}
-		}
-		return nil
-	}
 
 	// drop table dropTableSql
 	var destTableName string
@@ -98,12 +60,12 @@ func handleSchemaChange(j *ccr.Job, alterJob *record.AlterJobV2) error {
 		destTableName = alterJob.TableName
 	}
 
-	if ccr.FeatureSchemaChangePartialSync && alterJob.Type == record.ALTER_JOB_SCHEMA_CHANGE {
-		// Once partial snapshot finished, the shadow indexes will be convert to normal indexes.
-		for shadowIndexId := range alterJob.ShadowIndexes {
-			delete(job_progress.ShadowIndexes, shadowIndexId)
-		}
+	j.SaveSchemaChangeShadowIndexes(alterJob)
+	if !alterJob.IsFinished() {
+		return nil
+	}
 
+	if ccr.FeatureSchemaChangePartialSync && alterJob.Type == record.ALTER_JOB_SCHEMA_CHANGE {
 		return j.NewPartialSnapshot(alterJob.TableId, alterJob.TableName, nil, true, false)
 	}
 

pkg/ccr/handle/create_table.go

@@ -23,6 +23,11 @@ func (h *CreateTableHandle) Handle(j *ccr.Job, commitSeq int64, createTable *rec
 		return xerror.Errorf(xerror.Normal, "invalid sync type: %v", j.SyncType)
 	}
 
+	if createTable.IsCreateElasticSearch() {
+		log.Warnf("create table with elasticsearch is not supported yet, skip this binlog")
+		return nil
+	}
+
 	if createTable.IsCreateMaterializedView() {
 		log.Warnf("create async materialized view is not supported yet, skip this binlog")
 		return nil

pkg/ccr/handle/rename_partition.go

@@ -30,16 +30,16 @@ func (h *RenamePartitionHandle) IsBinlogCommitted(j *ccr.Job, record *record.Ren
 	}
 
 	for _, partition := range partitions {
-		if partition.Name == record.OldPartitionName {
-			log.Infof("partition %s is not renamed to %s in dest table %d, this binlog is not committed",
+		if partition.Name == record.NewPartitionName {
+			log.Infof("partition %s is not renamed to %s in dest table %d, this binlog is committed",
 				record.OldPartitionName, record.NewPartitionName, destTableId)
-			return false, nil
+			return true, nil
 		}
 	}
 
 	log.Infof("partition %s is renamed to %s in dest table %d, this binlog is not committed",
 		record.OldPartitionName, record.NewPartitionName, destTableId)
-	return true, nil
+	return false, nil
 }
 
 func (h *RenamePartitionHandle) IsIdempotent() bool {

pkg/ccr/ingest_binlog_job.go

@@ -543,8 +543,8 @@ func (j *IngestBinlogJob) preparePartition(srcTableId, destTableId int64,
 		srcIndexName := getSrcIndexName(job, srcIndexMeta)
 		if _, ok := destIndexNameMap[srcIndexName]; !ok {
 			j.setError(xerror.Errorf(xerror.Meta,
-				"index name %v not found in dest meta, is base index: %t, src index id: %d",
-				srcIndexName, srcIndexMeta.IsBaseIndex, indexId))
+				"index name %v not found in dest meta, is base index: %t, src index id: %d, shadow indexes %v",
+				srcIndexName, srcIndexMeta.IsBaseIndex, indexId, j.ccrJob.progress.ShadowIndexes))
 			return
 		}
 	}
@@ -806,6 +806,27 @@ func (j *IngestBinlogJob) applyDroppedBinlogs() {
 			return
 		}
 
+		if binlog.GetType() == festruct.TBinlogType_BARRIER {
+			barrierLog, err := record.NewBarrierLogFromJson(binlog.GetData())
+			if err != nil {
+				j.setError(err)
+				return
+			}
+
+			if barrierLog.Binlog != "" {
+				// keep compatible with old version
+				binlogType := festruct.TBinlogType(barrierLog.BinlogType)
+				newBinlog := festruct.NewTBinlog()
+				newBinlog.SetCommitSeq(utils.ThriftValueWrapper(binlog.GetCommitSeq()))
+				newBinlog.SetTimestamp(utils.ThriftValueWrapper(binlog.GetTimestamp()))
+				newBinlog.SetType(&binlogType)
+				newBinlog.SetDbId(utils.ThriftValueWrapper(binlog.GetDbId()))
+				newBinlog.SetData(&barrierLog.Binlog)
+				newBinlog.SetTableIds(binlog.GetTableIds())
+				binlog = newBinlog
+			}
+		}
+
 		if binlog.GetType() == festruct.TBinlogType_ALTER_JOB {
 			log.Infof("txn %d ingest binlog: trigger new partial snapshot by alter job binlog, commitSeq: %d", j.txnId, commitSeq)
 			alterJobRecord, err := record.NewAlterJobV2FromJson(binlog.GetData())