(logger g.LoggerType, jobParam *models.CreateOrUpdateMysqlToMysqlJobParamV2, user *common.User)
| 308 | } |
| 309 | |
| 310 | func createOrUpdateJob(logger g.LoggerType, jobParam *models.CreateOrUpdateMysqlToMysqlJobParamV2, |
| 311 | user *common.User) (*models.CreateOrUpdateMysqlToMysqlJobRespV2, error) { |
| 312 | |
| 313 | failover := g.PtrToBool(jobParam.Failover, true) |
| 314 | if jobParam.IsPasswordEncrypted { |
| 315 | realPwd, err := handler.DecryptPassword(jobParam.SrcTask.ConnectionConfig.Password, g.RsaPrivateKey) |
| 316 | if nil != err { |
| 317 | return nil, fmt.Errorf("decrypt src password failed: %v", err) |
| 318 | } |
| 319 | jobParam.SrcTask.ConnectionConfig.Password = realPwd |
| 320 | |
| 321 | realPwd, err = handler.DecryptPassword(jobParam.DestTask.ConnectionConfig.Password, g.RsaPrivateKey) |
| 322 | if nil != err { |
| 323 | return nil, fmt.Errorf("decrypt dest password failed: %v", err) |
| 324 | } |
| 325 | jobParam.DestTask.ConnectionConfig.Password = realPwd |
| 326 | } |
| 327 | |
| 328 | // set default |
| 329 | if jobParam.SrcTask.ReplChanBufferSize == 0 { |
| 330 | jobParam.SrcTask.ReplChanBufferSize = common.DefaultChannelBufferSize |
| 331 | } |
| 332 | if jobParam.SrcTask.GroupMaxSize == 0 { |
| 333 | jobParam.SrcTask.GroupMaxSize = common.DefaultSrcGroupMaxSize |
| 334 | } |
| 335 | if jobParam.SrcTask.ChunkSize == 0 { |
| 336 | jobParam.SrcTask.ChunkSize = common.DefaultChunkSize |
| 337 | } |
| 338 | if jobParam.SrcTask.GroupTimeout == 0 { |
| 339 | jobParam.SrcTask.GroupTimeout = common.DefaultSrcGroupTimeout |
| 340 | } |
| 341 | logger.Info("MysqlDestTaskConfig", "config", jobParam.DestTask.MysqlDestTaskConfig) |
| 342 | if jobParam.DestTask.MysqlDestTaskConfig != nil { |
| 343 | if jobParam.DestTask.MysqlDestTaskConfig.ParallelWorkers == 0 { |
| 344 | jobParam.DestTask.MysqlDestTaskConfig.ParallelWorkers = common.DefaultNumWorkers |
| 345 | } |
| 346 | if !jobParam.DestTask.MysqlDestTaskConfig.UseMySQLDependency && jobParam.DestTask.MysqlDestTaskConfig.DependencyHistorySize == 0 { |
| 347 | jobParam.DestTask.MysqlDestTaskConfig.DependencyHistorySize = common.DefaultDependencyHistorySize |
| 348 | } |
| 349 | } |
| 350 | |
| 351 | nomadJob, err := convertJobToNomadJob(failover, jobParam) |
| 352 | if nil != err { |
| 353 | return nil, fmt.Errorf("convert job param to nomad job request failed, error: %v", err) |
| 354 | } |
| 355 | |
| 356 | nomadJobreq := nomadApi.JobRegisterRequest{ |
| 357 | Job: nomadJob, |
| 358 | } |
| 359 | nomadJobReqByte, err := json.Marshal(nomadJobreq) |
| 360 | if nil != err { |
| 361 | return nil, fmt.Errorf("marshal nomad job request failed, error: %v", err) |
| 362 | } |
| 363 | url := handler.BuildUrl("/v1/jobs") |
| 364 | logger.Info("invoke nomad api begin", "url", url) |
| 365 | nomadResp := nomadApi.JobRegisterResponse{} |
| 366 | if err := handler.InvokePostApiWithJson(url, nomadJobReqByte, &nomadResp); nil != err { |
| 367 | return nil, fmt.Errorf("invoke nomad api %v failed: %v", url, err) |
no test coverage detected