[migration] batch task by task type
This commit is contained in:
Kassian Sun
parent
492bd7dd66
commit
651c86feff
|
|
@ -111,68 +111,70 @@ func (p *DispatcherProcessor) Process(ctx *pipeline.Context) error {
|
||||||
if ctx.IsCanceled() {
|
if ctx.IsCanceled() {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
tasks, err := p.getMigrationTasks(p.config.TaskBatchSize)
|
for _, taskType := range []string{"cluster_migration", "index_migration", "pipeline"} {
|
||||||
if err != nil {
|
tasks, err := p.getMigrationTasks(taskType, p.config.TaskBatchSize)
|
||||||
log.Errorf("failed to get migration tasks, err: %v", err)
|
if err != nil {
|
||||||
return err
|
log.Errorf("failed to get migration tasks, err: %v", err)
|
||||||
}
|
return err
|
||||||
if len(tasks) == 0 {
|
}
|
||||||
return nil
|
if len(tasks) == 0 {
|
||||||
}
|
|
||||||
for _, t := range tasks {
|
|
||||||
if ctx.IsCanceled() {
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
if t.Metadata.Labels == nil {
|
for _, t := range tasks {
|
||||||
log.Errorf("got migration task [%s] with empty labels, skip handling", t.ID)
|
if ctx.IsCanceled() {
|
||||||
continue
|
return nil
|
||||||
}
|
}
|
||||||
log.Debugf("start handling task [%s] (type: %s, status: %s)", t.ID, t.Metadata.Type, t.Status)
|
if t.Metadata.Labels == nil {
|
||||||
switch t.Metadata.Type {
|
log.Errorf("got migration task [%s] with empty labels, skip handling", t.ID)
|
||||||
case "cluster_migration":
|
continue
|
||||||
// handle major task
|
}
|
||||||
switch t.Status {
|
log.Debugf("start handling task [%s] (type: %s, status: %s)", t.ID, t.Metadata.Type, t.Status)
|
||||||
case task2.StatusReady:
|
switch t.Metadata.Type {
|
||||||
err = p.handleReadyMajorTask(&t)
|
case "cluster_migration":
|
||||||
case task2.StatusRunning:
|
// handle major task
|
||||||
err = p.handleRunningMajorTask(&t)
|
switch t.Status {
|
||||||
case task2.StatusPendingStop:
|
case task2.StatusReady:
|
||||||
err = p.handlePendingStopMajorTask(&t)
|
err = p.handleReadyMajorTask(&t)
|
||||||
|
case task2.StatusRunning:
|
||||||
|
err = p.handleRunningMajorTask(&t)
|
||||||
|
case task2.StatusPendingStop:
|
||||||
|
err = p.handlePendingStopMajorTask(&t)
|
||||||
|
}
|
||||||
|
if err != nil {
|
||||||
|
log.Errorf("failed to handling major task [%s]: [%v]", t.ID, err)
|
||||||
|
}
|
||||||
|
case "index_migration":
|
||||||
|
// handle sub migration task
|
||||||
|
switch t.Status {
|
||||||
|
case task2.StatusReady:
|
||||||
|
// split sub task
|
||||||
|
err = p.handleReadySubTask(&t)
|
||||||
|
case task2.StatusRunning:
|
||||||
|
// check pipeline tasks status
|
||||||
|
err = p.handleRunningSubTask(&t)
|
||||||
|
case task2.StatusPendingStop:
|
||||||
|
// mark pipeline tasks as pending_stop
|
||||||
|
err = p.handlePendingStopSubTask(&t)
|
||||||
|
}
|
||||||
|
if err != nil {
|
||||||
|
log.Errorf("failed to handling sub task [%s]: [%v]", t.ID, err)
|
||||||
|
}
|
||||||
|
case "pipeline":
|
||||||
|
// handle pipeline task
|
||||||
|
err = p.pipelineTaskProcessor.Process(&t)
|
||||||
|
if err != nil {
|
||||||
|
log.Errorf("failed to handling pipeline task [%s]: [%v]", t.ID, err)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Errorf("failed to handling major task [%s]: [%v]", t.ID, err)
|
t.Status = task2.StatusError
|
||||||
|
tn := time.Now()
|
||||||
|
t.CompletedTime = &tn
|
||||||
|
p.saveTaskAndWriteLog(&t, "", &task2.TaskResult{
|
||||||
|
Success: false,
|
||||||
|
Error: err.Error(),
|
||||||
|
}, fmt.Sprintf("error handling task [%s]", t.ID))
|
||||||
}
|
}
|
||||||
case "index_migration":
|
|
||||||
// handle sub migration task
|
|
||||||
switch t.Status {
|
|
||||||
case task2.StatusReady:
|
|
||||||
// split sub task
|
|
||||||
err = p.handleReadySubTask(&t)
|
|
||||||
case task2.StatusRunning:
|
|
||||||
// check pipeline tasks status
|
|
||||||
err = p.handleRunningSubTask(&t)
|
|
||||||
case task2.StatusPendingStop:
|
|
||||||
// mark pipeline tasks as pending_stop
|
|
||||||
err = p.handlePendingStopSubTask(&t)
|
|
||||||
}
|
|
||||||
if err != nil {
|
|
||||||
log.Errorf("failed to handling sub task [%s]: [%v]", t.ID, err)
|
|
||||||
}
|
|
||||||
case "pipeline":
|
|
||||||
// handle pipeline task
|
|
||||||
err = p.pipelineTaskProcessor.Process(&t)
|
|
||||||
if err != nil {
|
|
||||||
log.Errorf("failed to handling pipeline task [%s]: [%v]", t.ID, err)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
if err != nil {
|
|
||||||
t.Status = task2.StatusError
|
|
||||||
tn := time.Now()
|
|
||||||
t.CompletedTime = &tn
|
|
||||||
p.saveTaskAndWriteLog(&t, "", &task2.TaskResult{
|
|
||||||
Success: false,
|
|
||||||
Error: err.Error(),
|
|
||||||
}, fmt.Sprintf("error handling task [%s]", t.ID))
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
//es index refresh
|
//es index refresh
|
||||||
|
|
@ -741,7 +743,7 @@ func (p *DispatcherProcessor) getPreferenceInstance(majorTaskID string) (instanc
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
func (p *DispatcherProcessor) getMigrationTasks(size int) ([]task2.Task, error) {
|
func (p *DispatcherProcessor) getMigrationTasks(taskType string, size int) ([]task2.Task, error) {
|
||||||
queryDsl := util.MapStr{
|
queryDsl := util.MapStr{
|
||||||
"size": size,
|
"size": size,
|
||||||
"sort": []util.MapStr{
|
"sort": []util.MapStr{
|
||||||
|
|
@ -759,6 +761,13 @@ func (p *DispatcherProcessor) getMigrationTasks(size int) ([]task2.Task, error)
|
||||||
"status": []string{task2.StatusReady, task2.StatusRunning, task2.StatusPendingStop},
|
"status": []string{task2.StatusReady, task2.StatusRunning, task2.StatusPendingStop},
|
||||||
},
|
},
|
||||||
},
|
},
|
||||||
|
{
|
||||||
|
"term": util.MapStr{
|
||||||
|
"metadata.type": util.MapStr{
|
||||||
|
"value": taskType,
|
||||||
|
},
|
||||||
|
},
|
||||||
|
},
|
||||||
},
|
},
|
||||||
},
|
},
|
||||||
},
|
},
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue