return: progress, delay, stage
(logger g.LoggerType, header http.Header, taskLogs []models.TaskLog, jobId string)
| 757 | |
| 758 | // return: progress, delay, stage |
| 759 | func getTaskProgress(logger g.LoggerType, header http.Header, taskLogs []models.TaskLog, jobId string) (*models.DumpProgress, int64, string) { |
| 760 | storeManager, err := common.NewStoreManager([]string{handler.ConsulAddr}, logger) |
| 761 | if err != nil { |
| 762 | logger.Warn("getTaskProgress: failed to connect to consul", "ConsulAddr", handler.ConsulAddr, "err", err) |
| 763 | return nil, 0, "" |
| 764 | } |
| 765 | |
| 766 | execRowCount, totalRowCount, err := storeManager.GetDumpProgress(jobId) |
| 767 | progress := &models.DumpProgress{ |
| 768 | ExecRowCount: execRowCount, |
| 769 | TotalRowCount: totalRowCount, |
| 770 | } |
| 771 | if err != nil { |
| 772 | logger.Warn("getTaskProgress: failed to GetJobStatus", "ConsulAddr", handler.ConsulAddr, "err", err) |
| 773 | return nil, 0, "" |
| 774 | } |
| 775 | |
| 776 | stage, err := storeManager.GetJobStage(jobId) |
| 777 | if err != nil { |
| 778 | logger.Warn("getTaskProgress: failed to GetJobStatus", "ConsulAddr", handler.ConsulAddr, "err", err) |
| 779 | return nil, 0, "" |
| 780 | } |
| 781 | return progress, 0, stage |
| 782 | } |
| 783 | |
| 784 | func buildBasicTaskProfile(logger g.LoggerType, jobId string, srcTaskDetail *models.SrcTaskDetail, |
| 785 | destMySqlTaskDetail *models.MysqlDestTaskDetail, destKafkaTaskDetail *models.KafkaDestTaskDetail) (models.BasicTaskProfile, []models.TaskLog, error) { |
no test coverage detected