// Start starts working, but the functionalities should be turned on separately.
Start starts working, but the functionalities should be turned on separately.
// Start starts working but the functionalities should be turned on separately .
func (w *SourceWorker) Start() {
	if w.cfg.Checker.CheckEnable {
		w.taskStatusChecker.Start()
	}
	var err error
	w.sourceDB, err = conn.DefaultDBProvider.Apply(&w.cfg.DecryptPassword().From)
	if err != nil {
		w.l.Error("can't connected to upstream", zap.Error(err))
	}
	w.wg.Add(1)
	defer w.wg.Done()
	w.l.Info("start running")
	printTaskInterval := 30 * time.Second
	failpoint.Inject("PrintStatusCheckSeconds", func(val failpoint.Value) {
		if seconds, ok := val.(int); ok {
			printTaskInterval = time.Duration(seconds) * time.Second
			log.L().Info("set printStatusInterval",
				zap.String("failpoint", "PrintStatusCheckSeconds"),
				zap.Int("value", seconds))
		}
	})
	ticker := time.NewTicker(printTaskInterval)
	w.closed.Store(false)
	defer ticker.Stop()
	for {
		select {
		case <-w.ctx.Done():
			w.l.Info("status print process exits!")
			return
		case <-ticker.C:
			old := w.sourceStatus.Load()
			if old != nil {
				status := old.(*binlog.SourceStatus)
				if time.Since(status.UpdateTime) < printTaskInterval/2 {
					w.l.Info("we just updated the source status, skip once",
						zap.Time("last update time", status.UpdateTime))
					continue
				}
			}
			if err2 := w.updateSourceStatus(w.ctx); err2 != nil {
				if terror.ErrNoMasterStatus.Equal(err2) {
					w.l.Warn("This source's bin_log is OFF, so it only supports full_mode.", zap.String("sourceID", w.cfg.SourceID), zap.Error(err2))
				} else {
					w.l.Error("failed to update source status", zap.Error(err2))
				}
				continue
			}
			sourceStatus := w.sourceStatus.Load().(*binlog.SourceStatus)
			if w.l.Core().Enabled(zap.DebugLevel) {
				w.l.Debug("runtime status", zap.String("status", w.GetUnitAndSourceStatusJSON("", sourceStatus)))
			}
			w.Status("", sourceStatus)
		}
	}
}
func ( w * SourceWorker ) Start ( ) { if w . cfg . Checker . CheckEnable { w . taskStatusChecker . Start ( ) } var err error w . sourceDB , err = conn . DefaultDBProvider . Apply ( & w . cfg . DecryptPassword ( ) . From ) if err != nil { w . l . Error ( " " , zap . Error ( err ) ) } w . wg . Add ( 1 ) defer w . wg . Done ( ) w . l . Info ( " " ) printTaskInterval := 30 * time . Second failpoint . Inject ( " " , func ( val failpoint . Value ) { if seconds , ok := val . ( int ) ; ok { printTaskInterval = time . Duration ( seconds ) * time . Second log . L ( ) . Info ( " " , zap . String ( " " , " " ) , zap . Int ( " " , seconds ) ) } } ) ticker := time . NewTicker ( printTaskInterval ) w . closed . Store ( false ) defer ticker . Stop ( ) for { select { case <- w . ctx . Done ( ) : w . l . Info ( " " ) return case <- ticker . C : old := w . sourceStatus . Load ( ) if old != nil { status := old . ( * binlog . SourceStatus ) if time . Since ( status . UpdateTime ) < printTaskInterval / 2 { w . l . Info ( " " , zap . Time ( " " , status . UpdateTime ) ) continue } } if err2 := w . updateSourceStatus ( w . ctx ) ; err2 != nil { if terror . ErrNoMasterStatus . Equal ( err2 ) { w . l . Warn ( " " , zap . String ( " " , w . cfg . SourceID ) , zap . Error ( err2 ) ) } else { w . l . Error ( " " , zap . Error ( err2 ) ) } continue } sourceStatus := w . sourceStatus . Load ( ) . ( * binlog . SourceStatus ) if w . l . Core ( ) . Enabled ( zap . DebugLevel ) { w . l . Debug ( " " , zap . String ( " " , w . GetUnitAndSourceStatusJSON ( " " , sourceStatus ) ) ) } w . Status ( " " , sourceStatus ) } } }
Start starts working, but the functionalities should be turned on separately.
Start starts working but the functionalities should be turned on separately .
// start task status checker // periodically print the status and update metrics