(logger g.LoggerType, jobId string, srcTaskDetail *models.SrcTaskDetail, destMySqlTaskDetail *models.MysqlDestTaskDetail, destKafkaTaskDetail *models.KafkaDestTaskDetail)
| 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) { |
| 786 | storeManager, err := common.NewStoreManager([]string{handler.ConsulAddr}, logger) |
| 787 | if err != nil { |
| 788 | return models.BasicTaskProfile{}, nil, fmt.Errorf("consul_addr=%v; connect to consul failed: %v", handler.ConsulAddr, err) |
| 789 | } |
| 790 | basicTaskProfile := models.BasicTaskProfile{} |
| 791 | // base info |
| 792 | { |
| 793 | consulJobItem, err := storeManager.GetJobInfo(jobId) |
| 794 | if err != nil { |
| 795 | return models.BasicTaskProfile{}, nil, fmt.Errorf("consul_addr=%v; get ket %v Job Item failed: %v", jobId, handler.ConsulAddr, err) |
| 796 | } |
| 797 | |
| 798 | basicTaskProfile.JobBaseInfo = models.JobBaseInfo{ |
| 799 | JobId: jobId, |
| 800 | SubscriptionTopic: consulJobItem.Topic, |
| 801 | JobStatus: consulJobItem.JobStatus, |
| 802 | JobCreateTime: consulJobItem.JobCreateTime, |
| 803 | JobSteps: consulJobItem.JobSteps, |
| 804 | Delay: 0, |
| 805 | } |
| 806 | |
| 807 | nomadJobMap, err := findJobMapFromNomad() |
| 808 | if err != nil { |
| 809 | return models.BasicTaskProfile{}, nil, fmt.Errorf("find nomad job list err %v", err) |
| 810 | } |
| 811 | if nomadJobItem, ok := nomadJobMap[consulJobItem.JobId]; ok { |
| 812 | if basicTaskProfile.JobBaseInfo.JobStatus == common.DtleJobStatusNonPaused { |
| 813 | basicTaskProfile.JobBaseInfo.JobStatus = nomadJobItem.Status |
| 814 | } |
| 815 | } |
| 816 | |
| 817 | } |
| 818 | // configuration |
| 819 | { |
| 820 | srcConfig := models.SrcConfig{ |
| 821 | SkipCreateDbTable: srcTaskDetail.TaskConfig.SkipCreateDbTable, |
| 822 | DropTableIfExists: srcTaskDetail.TaskConfig.DropTableIfExists, |
| 823 | GroupMaxSize: srcTaskDetail.TaskConfig.GroupMaxSize, |
| 824 | GroupTimeout: srcTaskDetail.TaskConfig.GroupTimeout, |
| 825 | ReplChanBufferSize: srcTaskDetail.TaskConfig.ReplChanBufferSize, |
| 826 | ChunkSize: srcTaskDetail.TaskConfig.ChunkSize, |
| 827 | SQLFilter: srcTaskDetail.TaskConfig.SQLFilter, |
| 828 | } |
| 829 | if srcTaskDetail.TaskConfig.MysqlSrcTaskConfig != nil { |
| 830 | srcConfig.MysqlSrcTaskConfig = &models.MysqlSrcTaskConfig{ |
| 831 | ExpandSyntaxSupport: srcTaskDetail.TaskConfig.MysqlSrcTaskConfig.ExpandSyntaxSupport, |
| 832 | Gtid: srcTaskDetail.TaskConfig.MysqlSrcTaskConfig.Gtid, |
| 833 | BinlogRelay: srcTaskDetail.TaskConfig.MysqlSrcTaskConfig.BinlogRelay, |
| 834 | WaitOnJob: srcTaskDetail.TaskConfig.MysqlSrcTaskConfig.WaitOnJob, |
| 835 | AutoGtid: srcTaskDetail.TaskConfig.MysqlSrcTaskConfig.AutoGtid, |
| 836 | DumpEntryLimit: srcTaskDetail.TaskConfig.MysqlSrcTaskConfig.DumpEntryLimit, |
| 837 | TwoWaySync: srcTaskDetail.TaskConfig.MysqlSrcTaskConfig.TwoWaySync, |
| 838 | TwoWaySyncGtid: srcTaskDetail.TaskConfig.MysqlSrcTaskConfig.TwoWaySyncGtid, |
| 839 | } |
| 840 | } else if srcTaskDetail.TaskConfig.OracleSrcTaskConfig != nil { |
| 841 | srcConfig.OracleSrcTaskConfig = &models.OracleSrcTaskConfig{ |
no test coverage detected