(logger g.LoggerType, jobParam *models.CreateOrUpdateMysqlToKafkaJobParamV2, user *common.User)
| 480 | } |
| 481 | |
| 482 | func buildKafkaJobListItem(logger g.LoggerType, jobParam *models.CreateOrUpdateMysqlToKafkaJobParamV2, |
| 483 | user *common.User) error { |
| 484 | // add data to consul |
| 485 | storeManager, err := common.NewStoreManager([]string{handler.ConsulAddr}, logger) |
| 486 | if err != nil { |
| 487 | return fmt.Errorf("consul_addr=%v; connect to consul failed: %v", handler.ConsulAddr, err) |
| 488 | } |
| 489 | jobInfo := common.JobListItemV2{ |
| 490 | JobId: jobParam.JobId, |
| 491 | JobStatus: common.DtleJobStatusNonPaused, |
| 492 | Topic: jobParam.DestTask.Topic, |
| 493 | JobCreateTime: time.Now().In(time.Local).Format(time.RFC3339), |
| 494 | SrcAddrList: []string{net.JoinHostPort(jobParam.SrcTask.ConnectionConfig.Host, |
| 495 | strconv.Itoa(int(jobParam.SrcTask.ConnectionConfig.Port)))}, |
| 496 | DstAddrList: jobParam.DestTask.BrokerAddrs, |
| 497 | User: fmt.Sprintf("%s:%s", user.Tenant, user.Username), |
| 498 | JobSteps: nil, |
| 499 | } |
| 500 | if jobParam.TaskStepName == "all" { |
| 501 | jobInfo.JobSteps = append(jobInfo.JobSteps, common.NewJobStep(mysql.JobFullCopy), common.NewJobStep(mysql.JobIncrCopy)) |
| 502 | } else if jobParam.TaskStepName == mysql.JobFullCopy { |
| 503 | jobInfo.JobSteps = append(jobInfo.JobSteps, common.NewJobStep(mysql.JobFullCopy)) |
| 504 | } else if jobParam.TaskStepName == mysql.JobIncrCopy { |
| 505 | jobInfo.JobSteps = append(jobInfo.JobSteps, common.NewJobStep(mysql.JobIncrCopy)) |
| 506 | } |
| 507 | err = storeManager.SaveJobInfo(jobInfo) |
| 508 | if nil != err { |
| 509 | return fmt.Errorf("consul_addr=%v ; sava job info list failed: %v", handler.ConsulAddr, err) |
| 510 | } |
| 511 | return nil |
| 512 | } |
| 513 | |
| 514 | func buildNomadTaskGroupItem(dtleTaskconfig map[string]interface{}, taskName, nodeId string, failover bool, retryTimes int) (*nomadApi.TaskGroup, string, error) { |
| 515 | dataCenter := "" |
no test coverage detected