Merge branch 'master' into optimize_test_process_in_case_of_cancellation

Author: noname0443

.github/workflows/docker-tests.yml

@@ -65,7 +65,7 @@ jobs:
           - 'GODOG_FEATURE=offline_mode.feature make test'
           - 'GODOG_FEATURE=priority.feature make test'
           - 'GODOG_FEATURE=readonly_filesystem.feature make test'
-          - 'GODOG_FEATURE=recovery.feature make test'
+          - 'GODOG_FEATURE=recovery.57.feature make test'
           - 'GODOG_FEATURE=repair.feature make test'
           - 'GODOG_FEATURE=repl_mon.feature make test'
           - 'GODOG_FEATURE=statefile.feature make test'

internal/app/app.go

@@ -20,6 +20,7 @@ import (
 
 	nodestate "github.com/yandex/mysync/internal/app/node_state"
 	"github.com/yandex/mysync/internal/app/optimization"
+	"github.com/yandex/mysync/internal/app/resetup"
 	"github.com/yandex/mysync/internal/config"
 	"github.com/yandex/mysync/internal/dcs"
 	"github.com/yandex/mysync/internal/log"
@@ -46,6 +47,7 @@ type App struct {
 	switchHelper         mysql.ISwitchHelper
 	lostQuorumTime       time.Time
 	replicationOptimizer optimization.ReplicationOpitimizer
+	lagResetupper        *resetup.LagResetupper
 }
 
 // NewApp returns new App. Suddenly.
@@ -88,6 +90,7 @@ func NewApp(configFile, logLevel string, interactive bool) (*App, error) {
 		logger,
 		config.OptimizationConfig,
 	)
+
 	switchHelper := mysql.NewSwitchHelper(config)
 	app := &App{
 		state:                stateFirstRun,
@@ -101,6 +104,10 @@ func NewApp(configFile, logLevel string, interactive bool) (*App, error) {
 		switchHelper:         switchHelper,
 		replicationOptimizer: replicationOptimizer,
 	}
+
+	// TODO: appDcs should be separate entity
+	app.lagResetupper = resetup.NewLagResetupper(logger, app, config.ResetupHostLag.Seconds())
+
 	logger.Info("app created")
 	return app, nil
 }
@@ -159,9 +166,9 @@ func (app *App) writeEmergeFile(msg string) {
 	}
 }
 
-func (app *App) writeResetupFile(msg string) {
+func (app *App) writeResetupFile() {
 	app.logger.Warn("touch resetup file")
-	err := os.WriteFile(app.config.Resetupfile, []byte(msg), 0644)
+	err := os.WriteFile(app.config.Resetupfile, []byte{}, 0644)
 	if err != nil {
 		app.logger.Errorf("failed to write resetup file: %v", err)
 	}
@@ -1581,20 +1588,20 @@ func (app *App) repairOfflineMode(clusterState map[string]*nodestate.NodeState,
 		if !state.PingOk {
 			continue
 		}
-		node := app.cluster.Get(host)
 		if host == master {
-			app.repairMasterOfflineMode(host, node, state)
+			app.repairMasterOfflineMode(host, state)
 		} else {
-			app.repairSlaveOfflineMode(host, node, state, masterNode, clusterState[master])
+			app.repairSlaveOfflineMode(host, state, masterNode, clusterState[master])
 		}
 	}
 }
 
-func (app *App) repairMasterOfflineMode(host string, node *mysql.Node, state *nodestate.NodeState) {
+func (app *App) repairMasterOfflineMode(host string, state *nodestate.NodeState) {
 	if state.IsOffline {
 		if app.IsRecoveryNeeded(host) {
 			return
 		}
+		node := app.cluster.Get(host)
 		err := node.SetOnline()
 		if err != nil {
 			app.logger.Errorf("repair: failed to set master %s online: %s", host, err)
@@ -1604,73 +1611,86 @@ func (app *App) repairMasterOfflineMode(host string, node *mysql.Node, state *no
 	}
 }
 
-func (app *App) repairSlaveOfflineMode(host string, node *mysql.Node, state *nodestate.NodeState, masterNode *mysql.Node, masterState *nodestate.NodeState) {
-	if state.SlaveState != nil && state.SlaveState.ReplicationLag != nil {
-		replPermBroken, _ := state.IsReplicationPermanentlyBroken()
-		if state.IsOffline && *state.SlaveState.ReplicationLag <= app.config.OfflineModeDisableLag.Seconds() {
-			if replPermBroken {
-				app.logger.Infof("repair: replica %s is permanently broken, won't set online", host)
-				return
-			}
-			resetupStatus, err := app.GetResetupStatus(host)
-			if err != nil {
-				app.logger.Errorf("repair: failed to get resetup status from host %s: %v", host, err)
-				return
-			}
-			startupTime, err := node.GetStartupTime()
-			if err != nil {
-				app.logger.Errorf("repair: failed to get mysql startup time from host %s: %v", host, err)
-				return
-			}
-			if resetupStatus.Status || resetupStatus.UpdateTime.Before(startupTime) {
-				app.logger.Errorf("repair: should not turn slave to online until get actual resetup status")
-				return
-			}
-			err = node.SetDefaultReplicationSettings(masterNode)
-			if err != nil {
-				app.logger.Errorf("repair: failed to set default replication settings on slave %s: %s", host, err)
-			}
-			err = node.SetOnline()
-			if err != nil {
-				app.logger.Errorf("repair: failed to set slave %s online: %s", host, err)
-			} else {
-				app.logger.Infof("repair: slave %s set online, because ReplicationLag (%f s) <= OfflineModeDisableLag (%v)",
-					host, *state.SlaveState.ReplicationLag, app.config.OfflineModeDisableLag)
-			}
+// nolint: gocyclo
+func (app *App) repairSlaveOfflineMode(host string, state *nodestate.NodeState, masterNode *mysql.Node, masterState *nodestate.NodeState) {
+	if state.SlaveState == nil || state.SlaveState.ReplicationLag == nil {
+		return
+	}
+
+	replPermBroken, _ := state.IsReplicationPermanentlyBroken()
+	node := app.cluster.Get(host)
+	// offline => online, if lag has decreased
+	if state.IsOffline && *state.SlaveState.ReplicationLag <= app.config.OfflineModeDisableLag.Seconds() {
+		if replPermBroken {
+			app.logger.Infof("repair: replica %s is permanently broken, won't set online", host)
+			return
 		}
-		if !state.IsOffline && !masterState.IsReadOnly && *state.SlaveState.ReplicationLag > app.config.OfflineModeEnableLag.Seconds() {
-			err := node.SetOffline()
-			if err != nil {
-				app.logger.Errorf("repair: failed to set slave %s offline: %s", host, err)
-			} else {
-				app.logger.Infof("repair: slave %s set offline, because ReplicationLag (%f s) >= OfflineModeEnableLag (%v)",
-					host, *state.SlaveState.ReplicationLag, app.config.OfflineModeEnableLag)
-				err = app.replicationOptimizer.EnableNodeOptimization(node)
-				if err != nil {
-					app.logger.Errorf("repair: failed to set optimize replication settings on slave %s: %s", host, err)
-				}
-			}
+		resetupStatus, err := app.GetResetupStatus(host)
+		if err != nil {
+			app.logger.Errorf("repair: failed to get resetup status from host %s: %v", host, err)
+			return
 		}
-		// gradual transfer of permanently broken nodes to offline
-		lastShutdownNodeTime, err := app.GetLastShutdownNodeTime()
+		startupTime, err := node.GetStartupTime()
 		if err != nil {
-			app.logger.Errorf("repair: failed to get last shutdown node time: %s", err)
+			app.logger.Errorf("repair: failed to get mysql startup time from host %s: %v", host, err)
 			return
 		}
-		setOfflineIsPossible := time.Since(lastShutdownNodeTime) > app.config.OfflineModeEnableInterval
-		if !state.IsOffline && replPermBroken && setOfflineIsPossible {
-			err = app.UpdateLastShutdownNodeTime()
-			if err != nil {
-				app.logger.Errorf("repair: failed to update last shutdown node time: %s", err)
-			}
-			err = node.SetOffline()
+		if resetupStatus.Status || resetupStatus.UpdateTime.Before(startupTime) {
+			app.logger.Errorf("repair: should not turn slave to online until get actual resetup status")
+			return
+		}
+		err = node.SetDefaultReplicationSettings(masterNode)
+		if err != nil {
+			app.logger.Errorf("repair: failed to set default replication settings on slave %s: %s", host, err)
+		}
+		err = node.SetOnline()
+		if err != nil {
+			app.logger.Errorf("repair: failed to set slave %s online: %s", host, err)
+		} else {
+			app.logger.Infof("repair: slave %s set online, because ReplicationLag (%f s) <= OfflineModeDisableLag (%v)",
+				host, *state.SlaveState.ReplicationLag, app.config.OfflineModeDisableLag)
+		}
+		return
+	}
+	// online => offline, if lag has increased
+	if !state.IsOffline && !masterState.IsReadOnly && *state.SlaveState.ReplicationLag > app.config.OfflineModeEnableLag.Seconds() {
+		err := node.SetOffline()
+		if err != nil {
+			app.logger.Errorf("repair: failed to set slave %s offline: %s", host, err)
+		} else {
+			app.logger.Infof("repair: slave %s set offline, because ReplicationLag (%f s) >= OfflineModeEnableLag (%v)",
+				host, *state.SlaveState.ReplicationLag, app.config.OfflineModeEnableLag)
+			err = app.replicationOptimizer.EnableNodeOptimization(node)
 			if err != nil {
-				app.logger.Errorf("repair: failed to set slave %s offline: %s", host, err)
-			} else {
-				app.logger.Infof("repair: slave %s set offline, because replication permanently broken", host)
+				app.logger.Errorf("repair: failed to set optimize replication settings on slave %s: %s", host, err)
 			}
 		}
 	}
+
+	// nothing to do here
+	if !replPermBroken {
+		return
+	}
+
+	// gradual transfer of permanently broken nodes to offline
+	lastShutdownNodeTime, err := app.GetLastShutdownNodeTime()
+	if err != nil {
+		app.logger.Errorf("repair: failed to get last shutdown node time: %s", err)
+		return
+	}
+	setOfflineIsPossible := time.Since(lastShutdownNodeTime) > app.config.OfflineModeEnableInterval
+	if !state.IsOffline && replPermBroken && setOfflineIsPossible {
+		err = app.UpdateLastShutdownNodeTime()
+		if err != nil {
+			app.logger.Errorf("repair: failed to update last shutdown node time: %s", err)
+		}
+		err = node.SetOffline()
+		if err != nil {
+			app.logger.Errorf("repair: failed to set slave %s offline: %s", host, err)
+		} else {
+			app.logger.Infof("repair: slave %s set offline, because replication permanently broken", host)
+		}
+	}
 }
 
 // nolint: gocyclo
@@ -2485,6 +2505,7 @@ func (app *App) Run() int {
 
 	go app.healthChecker(ctx)
 	go app.recoveryChecker(ctx)
+	go app.replicationLagChecker(ctx)
 	go app.stateFileHandler(ctx)
 	if app.config.ExternalReplicationType != util.Disabled {
 		go app.externalCAFileChecker(ctx)

internal/app/recovery.go

@@ -89,7 +89,7 @@ func (app *App) checkRecovery() {
 		}
 
 		app.logger.Infof("recovery: new master is found %s, and current node is stuck for more than %v. Writing resetup file", master, StuckWaitTime)
-		app.writeResetupFile("")
+		app.writeResetupFile()
 		app.t.Clean(MasterStuckAt, localNode.Host())
 		return
 	}
@@ -110,7 +110,7 @@ func (app *App) checkRecovery() {
 			app.logger.Errorf("recovery: error getting replica status: %v", err)
 		}
 		app.logger.Errorf("recovery: local node %s is NOT behind the master %s, need RESETUP", localNode.Host(), masterNode)
-		app.writeResetupFile("")
+		app.writeResetupFile()
 	} else {
 		readOnly, _, err := localNode.IsReadOnly()
 		if err != nil {
@@ -164,5 +164,5 @@ func (app *App) checkCrashRecovery() {
 		return
 	}
 	app.logger.Errorf("recovery: local node %s is running after crash recovery %v, need RESETUP", localNode.Host(), ds.RecoveryTime)
-	app.writeResetupFile("")
+	app.writeResetupFile()
 }

internal/app/replication.go

@@ -25,6 +25,28 @@ type ReplicationRepairState struct {
 	LastGTIDExecuted string
 }
 
+// separated gorutine for checking local mysql lag
+func (app *App) replicationLagChecker(ctx context.Context) {
+	// 30s
+	// TODO: we should use 2 tikers - fast (recovery, ticker, etc) and slow - lag check
+	ticker := time.NewTicker(6 * app.config.RecoveryCheckInterval)
+	for {
+		select {
+		case <-ticker.C:
+			if app.doesResetupFileExist() {
+				app.logger.Infof("lag check: resetup file exists, waiting for resetup to complete")
+				return
+			}
+
+			if app.lagResetupper.CheckNeedResetup(app.cluster) {
+				app.writeResetupFile()
+			}
+		case <-ctx.Done():
+			return
+		}
+	}
+}
+
 func (app *App) MarkReplicationRunning(node *mysql.Node, channel string) {
 	var replState *ReplicationRepairState
 	key := app.makeReplStateKey(node, channel)

internal/app/resetup/resetup_lag.go

@@ -0,0 +1,88 @@
+package resetup
+
+import (
+	"github.com/yandex/mysync/internal/log"
+	"github.com/yandex/mysync/internal/mysql"
+)
+
+const (
+	prefix = "lag check"
+)
+
+type LagResetupperDcs interface {
+	GetMasterHostFromDcs() (string, error)
+}
+
+type LagResetupper struct {
+	logger         *log.Logger
+	dcs            LagResetupperDcs
+	resetupHostLag float64
+}
+
+func NewLagResetupper(l *log.Logger, dcs LagResetupperDcs, lag float64) *LagResetupper {
+	return &LagResetupper{
+		logger:         l,
+		dcs:            dcs,
+		resetupHostLag: lag,
+	}
+}
+
+func (r *LagResetupper) CheckNeedResetup(cluster *mysql.Cluster) bool {
+	err := cluster.UpdateHostsInfo()
+	if err != nil {
+		r.logger.Errorf("%s: failed to update hosts info: %v", prefix, err)
+		return false
+	}
+	localNode := cluster.Local()
+	sstatus, err := localNode.GetReplicaStatus()
+	if err != nil {
+		r.logger.Errorf("%s: host %s failed to get slave status %v", prefix, localNode.Host(), err)
+		return false
+	}
+
+	// definitely not a replica
+	if sstatus == nil {
+		return false
+	}
+
+	var master string
+	master, err = r.dcs.GetMasterHostFromDcs()
+	if err != nil {
+		r.logger.Errorf("%s: failed to get current master from dcs: %v", prefix, err)
+		return false
+	}
+
+	// probably not a replica, we need to wait
+	if master == localNode.Host() {
+		return false
+	}
+
+	lag := sstatus.GetReplicationLag()
+	// We have another problems, or small lag
+	if !lag.Valid || lag.Float64 <= r.resetupHostLag {
+		return false
+	}
+
+	// so, we clearly have replica with large lag
+	ifOffline, err := localNode.IsOffline()
+	if err != nil {
+		r.logger.Errorf("%s: failed to check local node if offline: %v", prefix, err)
+		return false
+	}
+
+	masterNode := cluster.Get(master)
+	masterRO, _, err := masterNode.IsReadOnly()
+	if err != nil {
+		r.logger.Errorf("%s: failed to check master (%s) if RO: %v", prefix, master, err)
+		return false
+	}
+
+	if ifOffline && !masterRO {
+		r.logger.Infof("%s: local host set to resetup, because ReplicationLag (%f s) > ResetupHostLag (%v)",
+			prefix, lag.Float64, r.resetupHostLag)
+
+		return true
+	}
+
+	return false
+}

internal/config/config.go

@@ -76,6 +76,7 @@ type Config struct {
 	OfflineModeEnableInterval               time.Duration                `config:"offline_mode_enable_interval" yaml:"offline_mode_enable_interval"`
 	OfflineModeEnableLag                    time.Duration                `config:"offline_mode_enable_lag" yaml:"offline_mode_enable_lag"`
 	OfflineModeDisableLag                   time.Duration                `config:"offline_mode_disable_lag" yaml:"offline_mode_disable_lag"`
+	ResetupHostLag                          time.Duration                `config:"resetup_host_lag" yaml:"resetup_host_lag"`
 	DisableSetReadonlyOnLost                bool                         `config:"disable_set_readonly_on_lost" yaml:"disable_set_readonly_on_lost"`
 	ResetupCrashedHosts                     bool                         `config:"resetup_crashed_hosts" yaml:"resetup_crashed_hosts"`
 	StreamFromReasonableLag                 time.Duration                `config:"stream_from_reasonable_lag" yaml:"stream_from_reasonable_lag"`
@@ -175,6 +176,7 @@ func DefaultConfig() (Config, error) {
 		OfflineModeEnableInterval:               15 * time.Minute,
 		OfflineModeEnableLag:                    24 * time.Hour,
 		OfflineModeDisableLag:                   30 * time.Second,
+		ResetupHostLag:                          25 * time.Hour,
 		StreamFromReasonableLag:                 5 * time.Minute,
 		PriorityChoiceMaxLag:                    60 * time.Second,
 		TestDiskUsageFile:                       "", // fake disk usage, only for docker tests