// EnableRelay enables the functionality of start/watch/handle relay.
// According to relay schedule of DM-master, a source worker will enable relay in two scenarios: its bound source has
// `enable-relay: true` in config, or it has a UpstreamRelayWorkerKeyAdapter etcd KV.
// The paths to EnableRelay are:
// - source config `enable-relay: true`, which is checked in enableHandleSubtasks
//   - when DM-worker Server.Start
//   - when DM-worker Server watches a SourceBound change, which is to turn a free source worker to bound or notify a
//     bound worker that source config has changed
//   - when DM-worker Server fails watching and recovers from a snapshot
// - UpstreamRelayWorkerKeyAdapter
//   - when DM-worker Server.Start
//   - when DM-worker Server watches a UpstreamRelayWorkerKeyAdapter change
//   - when DM-worker Server fails watching and recovers from a snapshot
EnableRelay enables the functionality of start/watch/handle relay.
According to relay schedule of DM-master, a source worker will enable relay in two scenarios: its bound source has
`enable-relay: true` in config, or it has a UpstreamRelayWorkerKeyAdapter etcd KV.
The paths to EnableRelay are:
- source config `enable-relay: true`, which is checked in enableHandleSubtasks
- when DM-worker Server.Start
- when DM-worker Server watches a SourceBound change, which is to turn a free source worker to bound or notify a
bound worker that source config has changed
- when DM-worker Server fails watching and recovers from a snapshot
- UpstreamRelayWorkerKeyAdapter
- when DM-worker Server.Start
- when DM-worker Server watches a UpstreamRelayWorkerKeyAdapter change
- when DM-worker Server fails watching and recovers from a snapshot
// EnableRelay enables the functionality of start / watch / handle relay . // According to relay schedule of DM - master a source worker will enable relay in two scenarios : its bound source has // ` enable - relay : true ` in config or it has a UpstreamRelayWorkerKeyAdapter etcd KV . // The paths to EnableRelay are : // - source config ` enable - relay : true ` which is checked in enableHandleSubtasks // - when DM - worker Server . Start // - when DM - worker Server watches a SourceBound change which is to turn a free source worker to bound or notify a // bound worker that source config has changed // - when DM - worker Server fails watching and recovers from a snapshot // - UpstreamRelayWorkerKeyAdapter // - when DM - worker Server . Start // - when DM - worker Server watches a UpstreamRelayWorkerKeyAdapter change // - when DM - worker Server fails watching and recovers from a snapshot
func (w *SourceWorker) EnableRelay(startBySourceCfg bool) (err error) {
	w.l.Info("enter EnableRelay")
	w.Lock()
	defer w.Unlock()
	if w.relayEnabled.Load() {
		w.l.Warn("already enabled relay")
		return nil
	}
	w.startedRelayBySourceCfg = startBySourceCfg
	failpoint.Inject("MockGetSourceCfgFromETCD", func(_ failpoint.Value) {
		failpoint.Goto("bypass")
	})
	if refreshErr := w.refreshSourceCfg(); refreshErr != nil {
		return refreshErr
	}
	failpoint.Label("bypass")
	w.relayCtx, w.relayCancel = context.WithCancel(w.ctx)
	var subTaskCfgs map[string]config.SubTaskConfig
	_, _, subTaskCfgs, _, err = w.fetchSubTasksAndAdjust()
	if err != nil {
		return err
	}
	dctx, dcancel := context.WithTimeout(w.etcdClient.Ctx(), time.Duration(len(subTaskCfgs)*3)*time.Second)
	defer dcancel()
	minLoc, err1 := getMinLocInAllSubTasks(dctx, subTaskCfgs)
	if err1 != nil {
		w.l.Error("meet error when EnableRelay", zap.Error(err1))
	}
	if minLoc != nil {
		w.l.Info("get min location in all subtasks", zap.Stringer("location", *minLoc))
		w.cfg.RelayBinLogName = binlog.AdjustPosition(minLoc.Position).Name
		w.cfg.RelayBinlogGTID = minLoc.GTIDSetStr()
		w.cfg.UUIDSuffix, err = binlog.ExtractSuffix(minLoc.Position.Name)
		if err != nil {
			return err
		}
	} else {
		w.cfg.UUIDSuffix = binlog.MinUUIDSuffix
	}
	if w.relayDir != "" {
		workerRelayDir := filepath.Join(w.relayDir, w.name)
		log.L().Info("use worker's relay-dir", zap.String("RelayDir", workerRelayDir))
		w.cfg.RelayDir = workerRelayDir
	}
	w.relayHolder = NewRelayHolder(w.cfg)
	relayPurger, err := w.relayHolder.Init(w.relayCtx, []relay.PurgeInterceptor{
		w,
	})
	if err != nil {
		return err
	}
	w.relayPurger = relayPurger
	relayStage, revRelay, err := ha.GetRelayStage(w.etcdClient, w.cfg.SourceID)
	if err != nil {
		return err
	}
	startImmediately := !relayStage.IsDeleted && relayStage.Expect == pb.Stage_Running
	if startImmediately {
		w.l.Info("start relay for existing relay stage")
		w.relayHolder.Start()
		w.relayPurger.Start()
	}
	w.relayWg.Add(1)
	go func() {
		defer w.relayWg.Done()
		w.observeRelayStage(w.relayCtx, w.etcdClient, revRelay)
	}()
	w.relayEnabled.Store(true)
	w.l.Info("relay enabled")
	w.subTaskHolder.resetAllSubTasks(w.getRelayWithoutLock())
	return nil
}
func ( w * SourceWorker ) EnableRelay ( startBySourceCfg bool ) ( err error ) { w . l . Info ( " " ) w . Lock ( ) defer w . Unlock ( ) if w . relayEnabled . Load ( ) { w . l . Warn ( " " ) return nil } w . startedRelayBySourceCfg = startBySourceCfg failpoint . Inject ( " " , func ( _ failpoint . Value ) { failpoint . Goto ( " " ) } ) if refreshErr := w . refreshSourceCfg ( ) ; refreshErr != nil { return refreshErr } failpoint . Label ( " " ) w . relayCtx , w . relayCancel = context . WithCancel ( w . ctx ) var subTaskCfgs map [ string ] config . SubTaskConfig _ , _ , subTaskCfgs , _ , err = w . fetchSubTasksAndAdjust ( ) if err != nil { return err } dctx , dcancel := context . WithTimeout ( w . etcdClient . Ctx ( ) , time . Duration ( len ( subTaskCfgs ) * 3 ) * time . Second ) defer dcancel ( ) minLoc , err1 := getMinLocInAllSubTasks ( dctx , subTaskCfgs ) if err1 != nil { w . l . Error ( " " , zap . Error ( err1 ) ) } if minLoc != nil { w . l . Info ( " " , zap . Stringer ( " " , * minLoc ) ) w . cfg . RelayBinLogName = binlog . AdjustPosition ( minLoc . Position ) . Name w . cfg . RelayBinlogGTID = minLoc . GTIDSetStr ( ) w . cfg . UUIDSuffix , err = binlog . ExtractSuffix ( minLoc . Position . Name ) if err != nil { return err } } else { w . cfg . UUIDSuffix = binlog . MinUUIDSuffix } if w . relayDir != " " { workerRelayDir := filepath . Join ( w . relayDir , w . name ) log . L ( ) . Info ( " " , zap . String ( " " , workerRelayDir ) ) w . cfg . RelayDir = workerRelayDir } w . relayHolder = NewRelayHolder ( w . cfg ) relayPurger , err := w . relayHolder . Init ( w . relayCtx , [ ] relay . PurgeInterceptor { w , } ) if err != nil { return err } w . relayPurger = relayPurger relayStage , revRelay , err := ha . GetRelayStage ( w . etcdClient , w . cfg . SourceID ) if err != nil { return err } startImmediately := ! relayStage . IsDeleted && relayStage . Expect == pb . Stage_Running if startImmediately { w . l . Info ( " " ) w . relayHolder . Start ( ) w . relayPurger . Start ( ) } w . relayWg . Add ( 1 ) go func ( ) { defer w . relayWg . Done ( ) w . observeRelayStage ( w . relayCtx , w . etcdClient , revRelay ) } ( ) w . relayEnabled . Store ( true ) w . l . Info ( " " ) w . subTaskHolder . resetAllSubTasks ( w . getRelayWithoutLock ( ) ) return nil }
EnableRelay enables the functionality of start/watch/handle relay.
EnableRelay enables the functionality of start / watch / handle relay .
// we need update config from etcd first in case this cfg is updated by master // 1. adjust relay starting position, to the earliest of subtasks //nolint:dogsled // set UUIDSuffix when bound to a source // set UUIDSuffix even not checkpoint exist // so we will still remove relay dir // 2. initial relay holder, the cfg's password need decrypt // worker's relay-dir has higher priority than source's relay-dir // 3. get relay stage from etcd and check if need starting // we get the newest relay stages directly which will omit the relay stage PUT/DELETE event // because triggering these events is useless now // TODO: need retry // 4. watch relay stage // TODO: handle fatal error from observeRelayStage //nolint:errcheck