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

[migration] add back instance init error check

This commit is contained in:
Kassian Sun 2023-04-22 14:58:59 +08:00
parent 28a6c55671
commit b263c480c7
1 changed files with 9 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
plugin/migration/pipeline.go
View File

@ -91,8 +91,12 @@ func newMigrationDispatcherProcessor(c *config.Config) (pipeline.Processor, erro
processor := DispatcherProcessor{ processor := DispatcherProcessor{
id: util.GetUUID(), id: util.GetUUID(),
config: &cfg, config: &cfg,
state: map[string]DispatcherState{},
}
err := processor.refreshInstanceJobsFromES()
if err != nil {
return nil, err
} }
processor.refreshInstanceJobsFromES()
processor.pipelineTaskProcessor = pipeline_task.NewProcessor(cfg.Elasticsearch, cfg.IndexName, cfg.LogIndexName) processor.pipelineTaskProcessor = pipeline_task.NewProcessor(cfg.Elasticsearch, cfg.IndexName, cfg.LogIndexName)
return &processor, nil return &processor, nil
@ -1311,7 +1315,7 @@ func (p *DispatcherProcessor) getInstanceState(instanceID string) DispatcherStat
return p.state[instanceID] return p.state[instanceID]
} }
func (p *DispatcherProcessor) refreshInstanceJobsFromES() { func (p *DispatcherProcessor) refreshInstanceJobsFromES() error {
log.Debug("refreshing instance state from ES") log.Debug("refreshing instance state from ES")
p.stateLock.Lock() p.stateLock.Lock()
defer p.stateLock.Unlock() defer p.stateLock.Unlock()
@ -1319,9 +1323,11 @@ func (p *DispatcherProcessor) refreshInstanceJobsFromES() {
state, err := p.getInstanceTaskState() state, err := p.getInstanceTaskState()
if err != nil { if err != nil {
log.Errorf("failed to get instance task state, err: %v", err) log.Errorf("failed to get instance task state, err: %v", err)
return return err
} }
p.state = state p.state = state
return nil
} }
func (p *DispatcherProcessor) cleanGatewayQueue(taskItem *task2.Task) { func (p *DispatcherProcessor) cleanGatewayQueue(taskItem *task2.Task) {