floraachy
/
console
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[migration] fix index source modification

This commit is contained in:
Kassian Sun 2023-04-17 12:33:23 +08:00
parent e41b2adfcf
commit 81c9df8ef4
2 changed files with 27 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
plugin/migration/model.go
View File

@ -136,7 +136,6 @@ type IndexMigrationSourceConfig struct {
// Parition configs // Parition configs
Start float64 `json:"start"` Start float64 `json:"start"`
End float64 `json:"end"` End float64 `json:"end"`
Docs int64 `json:"docs"`
DocCount int64 `json:"doc_count"` DocCount int64 `json:"doc_count"`
Step interface{} `json:"step"` Step interface{} `json:"step"`
PartitionId int `json:"partition_id"` PartitionId int `json:"partition_id"`

52
plugin/migration/pipeline.go
View File

@ -980,31 +980,6 @@ func (p *DispatcherProcessor) splitMajorMigrationTask(taskItem *task2.Task) erro
Compress: clusterMigrationTask.Settings.Bulk.Compress, Compress: clusterMigrationTask.Settings.Bulk.Compress,
}, },
} }
indexParameters := IndexMigrationTaskConfig{
Source: source,
Target: target,
}
indexMigrationTask := task2.Task{
ParentId: []string{taskItem.ID},
Cancellable: true,
Runnable: false,
Status: task2.StatusReady,
StartTimeInMillis: time.Now().UnixMilli(),
Metadata: task2.Metadata{
Type: "index_migration",
Labels: util.MapStr{
"business_id": "index_migration",
"source_cluster_id": clusterMigrationTask.Cluster.Source.Id,
"target_cluster_id": clusterMigrationTask.Cluster.Target.Id,
"partition_count": 1,
"index_name": index.Source.Name,
"unique_index_name": index.Source.GetUniqueIndexName(),
},
},
ConfigString: util.MustToJSON(indexParameters),
}
indexMigrationTask.ID = util.GetUUID()
if index.Partition != nil { if index.Partition != nil {
partitionQ := &elastic.PartitionQuery{ partitionQ := &elastic.PartitionQuery{
@ -1099,6 +1074,33 @@ func (p *DispatcherProcessor) splitMajorMigrationTask(taskItem *task2.Task) erro
} }
} else { } else {
source.DocCount = index.Source.Docs source.DocCount = index.Source.Docs
indexParameters := IndexMigrationTaskConfig{
Source: source,
Target: target,
}
indexMigrationTask := task2.Task{
ParentId: []string{taskItem.ID},
Cancellable: true,
Runnable: false,
Status: task2.StatusReady,
StartTimeInMillis: time.Now().UnixMilli(),
Metadata: task2.Metadata{
Type: "index_migration",
Labels: util.MapStr{
"business_id": "index_migration",
"source_cluster_id": clusterMigrationTask.Cluster.Source.Id,
"target_cluster_id": clusterMigrationTask.Cluster.Target.Id,
"partition_count": 1,
"index_name": index.Source.Name,
"unique_index_name": index.Source.GetUniqueIndexName(),
},
},
ConfigString: util.MustToJSON(indexParameters),
}
indexMigrationTask.ID = util.GetUUID()
err = orm.Create(nil, &indexMigrationTask) err = orm.Create(nil, &indexMigrationTask)
if err != nil { if err != nil {
return fmt.Errorf("store index migration task error: %w", err) return fmt.Errorf("store index migration task error: %w", err)