put job config on dest instead of src #592-1

ffffwh · ffffwh · commit 2d4d3f67fb24 · 2022-07-19T17:04:06.000+08:00
diff --git a/driver/common/store.go b/driver/common/store.go
@@ -101,20 +101,25 @@ func (sm *StoreManager) GetGtidForJob(jobName string) (string, error) {
 	return string(p.Value), nil
 }
 
-func (sm *StoreManager) GetConfig(jobName string) (*MySQLDriverConfig, error) {
+func (sm *StoreManager) WatchConfig(jobName string, stopCh <-chan struct{}) (*MySQLDriverConfig, error) {
 	key := fmt.Sprintf("dtle/%v/Config", jobName)
 
-	kv, err := sm.consulStore.Get(key)
+	ch, err := sm.consulStore.Watch(key, stopCh)
 	if err != nil {
 		return nil, err
 	}
 
-	config := &MySQLDriverConfig{}
-	err = json.Unmarshal(kv.Value, config)
-	if err != nil {
-		return nil, err
+	pair := <-ch
+	if pair == nil {
+		return nil, fmt.Errorf("failed to get config")
+	} else {
+		config := &MySQLDriverConfig{}
+		err = json.Unmarshal(pair.Value, config)
+		if err != nil {
+			return nil, err
+		}
+		return config, nil
 	}
-	return config, nil
 }
 
 func (sm *StoreManager) GetSourceType(jobName string) (string, error) {
@@ -205,7 +210,7 @@ func (sm *StoreManager) DstPutNats(jobName string, natsAddr string, stopCh chan
 	return nil
 }
 
-func (sm *StoreManager) SrcWatchNats(jobName string, stopCh chan struct{},
+func (sm *StoreManager) SrcWatchNats(jobName string, stopCh <-chan struct{},
 	onWatchError func(error)) (natsAddr string, err error) {
 	sm.logger.Debug("SrcWatchNats")
 
diff --git a/driver/handle.go b/driver/handle.go
@@ -166,19 +166,64 @@ func (h *taskHandle) NewRunner(d *Driver) (runner DriverHandle, err error) {
 
 	switch common.TaskTypeFromString(h.taskConfig.Name) {
 	case common.TaskTypeSrc:
+		if h.driverConfig.ConnectionConfig != nil {
+			h.driverConfig.SrcConnectionConfig = h.driverConfig.ConnectionConfig
+		}
+		// get job config v1. Call `PutSourceType` before `SrcWatchNats`.
+		if h.driverConfig.OracleConfig != nil {
+			err = d.storeManager.PutSourceType(ctx.Subject, "oracle")
+			if err != nil {
+				return nil, errors.Wrap(err, "PutSourceType")
+			}
+		} else if h.driverConfig.SrcConnectionConfig != nil {
+			err = d.storeManager.PutSourceType(ctx.Subject, "mysql")
+			if err != nil {
+				return nil, errors.Wrap(err, "PutSourceType")
+			}
+		} else {
+			// get job config v2. No need to PutSourceType.
+		}
+
+		natsAddr, err := d.storeManager.SrcWatchNats(ctx.Subject, h.ctx.Done(), func(err error) {
+			d.logger.Error("**** SrcWatchNats error")
+		})
+		if err != nil {
+			return nil, errors.Wrap(err, "SrcWatchNats")
+		}
+
+		switch h.driverConfig.GetConfigFrom {
+		case "":
+			d.logger.Info("get job config from nomad config")
+		case "consul":
+			d.logger.Info("get job config from consul")
+			h.driverConfig, err = d.storeManager.WatchConfig(ctx.Subject, h.ctx.Done())
+			if err != nil {
+				return nil, errors.Wrap(err, "WatchConfig")
+			}
+		default:
+			return nil, fmt.Errorf("unrecognized GetConfigFrom %v", h.driverConfig.GetConfigFrom)
+		}
+
 		if h.driverConfig.OracleConfig != nil {
 			h.logger.Debug("found oracle src", "OracleConfig", h.driverConfig.OracleConfig)
-			runner, err = extractor.NewExtractorOracle(ctx, h.driverConfig, h.logger, d.storeManager, h.waitCh)
+			runner, err = extractor.NewExtractorOracle(ctx, h.driverConfig, h.logger, d.storeManager, h.waitCh, natsAddr)
 			if err != nil {
 				return nil, errors.Wrap(err, "NewExtractor")
 			}
-		} else {
-			runner, err = mysql.NewExtractor(ctx, h.driverConfig, h.logger, d.storeManager, h.waitCh, h.ctx)
+		} else if h.driverConfig.SrcConnectionConfig != nil {
+			runner, err = mysql.NewExtractor(ctx, h.driverConfig, h.logger, d.storeManager, h.waitCh, h.ctx, natsAddr)
 			if err != nil {
 				return nil, errors.Wrap(err, "NewOracleExtractor")
 			}
+		} else {
+			err = fmt.Errorf("no src connection config")
+			h.logger.Error(err.Error())
+			return nil, err
 		}
 	case common.TaskTypeDest:
+		if h.driverConfig.ConnectionConfig != nil {
+			h.driverConfig.DestConnectionConfig = h.driverConfig.ConnectionConfig
+		}
 		h.logger.Debug("found dest", "allConfig", h.driverConfig)
 		if h.driverConfig.KafkaConfig != nil {
 			h.logger.Debug("found kafka", "KafkaConfig", h.driverConfig.KafkaConfig)
diff --git a/driver/mysql/applier.go b/driver/mysql/applier.go
@@ -278,25 +278,24 @@ func (a *Applier) Run() {
 		return
 	}
 
-	switch a.mysqlContext.GetConfigFrom {
-	case "":
-		a.mysqlContext.DestConnectionConfig = a.mysqlContext.SrcConnectionConfig
-		a.logger.Info("get job config from nomad config")
-	case "consul":
-		a.logger.Info("get job config from consul")
-		a.mysqlContext, err = a.storeManager.GetConfig(a.subject)
+	err = a.storeManager.PutConfig(a.subject, a.mysqlContext)
+	if err != nil {
+		a.onError(common.TaskStateDead, errors.Wrap(err, "PutConfig"))
+		return
+	}
+
+	var sourceType string
+	if a.mysqlContext.SrcConnectionConfig != nil {
+		sourceType = "mysql"
+	} else if a.mysqlContext.OracleConfig != nil {
+		sourceType = "oracle"
+	} else {
+		// src get config from nomad job config
+		sourceType, err = a.storeManager.GetSourceType(a.subject)
 		if err != nil {
-			a.onError(common.TaskStateDead, errors.Wrap(err, "GetConfig"))
+			a.onError(common.TaskStateDead, errors.Wrap(err, "watchSourceType"))
 			return
 		}
-	default:
-		a.onError(common.TaskStateDead, fmt.Errorf("unrecognized GetConfigFrom %v", a.mysqlContext.GetConfigFrom))
-	}
-
-	sourceType, err := a.storeManager.GetSourceType(a.subject)
-	if err != nil {
-		a.onError(common.TaskStateDead, errors.Wrap(err, "watchSourceType"))
-		return
 	}
 
 	if sourceType == "mysql" {
diff --git a/driver/mysql/extractor.go b/driver/mysql/extractor.go
@@ -109,7 +109,9 @@ type Extractor struct {
 	targetGtid string
 }
 
-func NewExtractor(execCtx *common.ExecContext, cfg *common.MySQLDriverConfig, logger g.LoggerType, storeManager *common.StoreManager, waitCh chan *drivers.ExitResult, ctx context.Context) (*Extractor, error) {
+func NewExtractor(execCtx *common.ExecContext, cfg *common.MySQLDriverConfig, logger g.LoggerType,
+	storeManager *common.StoreManager, waitCh chan *drivers.ExitResult, ctx context.Context,
+	natsAddr string) (*Extractor, error) {
 	logger.Info("NewExtractor", "job", execCtx.Subject)
 
 	e := &Extractor{
@@ -130,6 +132,7 @@ func NewExtractor(execCtx *common.ExecContext, cfg *common.MySQLDriverConfig, lo
 		memory1:         new(int64),
 		memory2:         new(int64),
 		replicateDoDb:   map[string]*common.SchemaContext{},
+		natsAddr:        natsAddr,
 	}
 	e.dataChannel = make(chan *common.EntryContext, cfg.ReplChanBufferSize*4)
 	e.timestampCtx = NewTimestampContext(e.shutdownCh, e.logger, func() bool {
@@ -151,17 +154,6 @@ func NewExtractor(execCtx *common.ExecContext, cfg *common.MySQLDriverConfig, lo
 func (e *Extractor) Run() {
 	var err error
 
-	err = e.storeManager.PutSourceType(e.subject, "mysql")
-	if err != nil {
-		e.onError(common.TaskStateDead, errors.Wrap(err, "PutSourceType"))
-		return
-	}
-
-	err = e.storeManager.PutConfig(e.subject, e.mysqlContext)
-	if err != nil {
-		e.onError(common.TaskStateDead, errors.Wrap(err, "PutKey config"))
-	}
-
 	{
 		jobStatus, _ := e.storeManager.GetJobStatus(e.subject)
 		if jobStatus == common.TargetGtidFinished {
@@ -203,13 +195,6 @@ func (e *Extractor) Run() {
 		}
 		e.logger.Info("after WaitOnJob", "job2", e.mysqlContext.WaitOnJob, "firstWait", firstWait)
 	}
-	e.natsAddr, err = e.storeManager.SrcWatchNats(e.subject, e.shutdownCh, func(err error) {
-		e.onError(common.TaskStateDead, err)
-	})
-	if err != nil {
-		e.onError(common.TaskStateDead, errors.Wrap(err, "SrcWatchNats"))
-		return
-	}
 
 	err = common.GetGtidFromConsul(e.storeManager, e.subject, e.logger, e.mysqlContext)
 	if err != nil {
diff --git a/driver/oracle/extractor/extractor_oracle.go b/driver/oracle/extractor/extractor_oracle.go
@@ -120,7 +120,7 @@ type OracleSchemaInfo struct {
 	Tables map[string]*ast.CreateTableStmt
 }
 
-func NewExtractorOracle(execCtx *common.ExecContext, cfg *common.MySQLDriverConfig, logger g.LoggerType, storeManager *common.StoreManager, waitCh chan *drivers.ExitResult) (*ExtractorOracle, error) {
+func NewExtractorOracle(execCtx *common.ExecContext, cfg *common.MySQLDriverConfig, logger g.LoggerType, storeManager *common.StoreManager, waitCh chan *drivers.ExitResult, natsAddr string) (*ExtractorOracle, error) {
 	logger.Info("NewExtractorOracle", "job", execCtx.Subject)
 
 	e := &ExtractorOracle{
@@ -140,6 +140,7 @@ func NewExtractorOracle(execCtx *common.ExecContext, cfg *common.MySQLDriverConf
 		memory1:         new(int64),
 		memory2:         new(int64),
 		OracleContext:   new(OracleContext),
+		natsAddr:        natsAddr,
 	}
 	e.dataChannel = make(chan *common.EntryContext, cfg.ReplChanBufferSize*4)
 	e.timestampCtx = NewTimestampContext(e.shutdownCh, e.logger, func() bool {
@@ -154,20 +155,6 @@ func NewExtractorOracle(execCtx *common.ExecContext, cfg *common.MySQLDriverConf
 func (e *ExtractorOracle) Run() {
 	var err error
 
-	err = e.storeManager.PutSourceType(e.subject, "oracle")
-	if err != nil {
-		e.onError(common.TaskStateDead, errors.Wrap(err, "PutSourceType"))
-		return
-	}
-
-	e.logger.Info("src watch Nats")
-	e.natsAddr, err = e.storeManager.SrcWatchNats(e.subject, e.shutdownCh, func(err error) {
-		e.onError(common.TaskStateDead, err)
-	})
-	if err != nil {
-		e.onError(common.TaskStateDead, errors.Wrap(err, "SrcWatchNats"))
-		return
-	}
 	// init nats
 	e.logger.Info("initNatsPubClient")
 	e.logger.Debug("begin Connect nats server", "NatAddr", e.natsAddr)
