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

feat: migration skip exists docs with bulk operation

This commit is contained in:
hardy 2024-02-04 17:29:12 +08:00 committed by hardy
parent 00dfa9ed5d
commit a338af5c9a
2 changed files with 18 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
plugin/task_manager/index_migration/index_migration.go
View File

@ -129,12 +129,12 @@ func (p *processor) handleSplitSubTask(taskItem *task.Task) error {
Processor: []util.MapStr{
{
"es_scroll": util.MapStr{
"remove_type": docType == "",
"slice_size": cfg.Source.SliceSize,
"batch_size": cfg.Source.BatchSize,
"indices": indexName,
"elasticsearch": sourceClusterID,
"bulk_create_operation": cfg.Target.Bulk.CreateOperation,
"remove_type": docType == "",
"slice_size": cfg.Source.SliceSize,
"batch_size": cfg.Source.BatchSize,
"indices": indexName,
"elasticsearch": sourceClusterID,
"bulk_operation": cfg.Target.Bulk.Operation,
"queue": util.MapStr{
"name": scrollID,
"labels": util.MapStr{
@ -187,11 +187,11 @@ func (p *processor) handleSplitSubTask(taskItem *task.Task) error {
"bulk_indexing": util.MapStr{
"detect_active_queue": false,
"bulk": util.MapStr{
"batch_size_in_mb": cfg.Target.Bulk.BatchSizeInMB,
"batch_size_in_docs": cfg.Target.Bulk.BatchSizeInDocs,
"invalid_queue": "bulk_indexing_400",
"compress": cfg.Target.Bulk.Compress,
"bulk_create_operation": cfg.Target.Bulk.CreateOperation,
"batch_size_in_mb": cfg.Target.Bulk.BatchSizeInMB,
"batch_size_in_docs": cfg.Target.Bulk.BatchSizeInDocs,
"invalid_queue": "bulk_indexing_400",
"compress": cfg.Target.Bulk.Compress,
"bulk_operation": cfg.Target.Bulk.Operation,
},
"max_worker_size": cfg.Target.Bulk.MaxWorkerSize,
"num_of_slices": cfg.Target.Bulk.SliceSize,

14
plugin/task_manager/model/migration.go
View File

@ -86,13 +86,13 @@ type IndexMigrationSourceConfig struct {
}
type IndexMigrationBulkConfig struct {
BatchSizeInDocs int `json:"batch_size_in_docs"`
BatchSizeInMB int `json:"batch_size_in_mb"`
MaxWorkerSize int `json:"max_worker_size"`
IdleTimeoutInSeconds int `json:"idle_timeout_in_seconds"`
SliceSize int `json:"slice_size"`
Compress bool `json:"compress"`
CreateOperation bool `json:"bulk_create_operation"`
BatchSizeInDocs int `json:"batch_size_in_docs"`
BatchSizeInMB int `json:"batch_size_in_mb"`
MaxWorkerSize int `json:"max_worker_size"`
IdleTimeoutInSeconds int `json:"idle_timeout_in_seconds"`
SliceSize int `json:"slice_size"`
Compress bool `json:"compress"`
Operation string `json:"operation"`
}
type IndexMigrationTargetConfig struct {