support async replication

.github/workflows/docker-tests-8.0.yml

@@ -42,6 +42,7 @@ jobs:
         command:
           - 'VERSION=8.0 GODOG_FEATURE=active_nodes.feature make test'
           - 'VERSION=8.0 GODOG_FEATURE=async.feature make test'
+          - 'VERSION=8.0 GODOG_FEATURE=async_setting.feature make test'
           - 'VERSION=8.0 GODOG_FEATURE=cascade_replicas.feature make test'
           - 'VERSION=8.0 GODOG_FEATURE=CLI.feature make test'
           - 'VERSION=8.0 GODOG_FEATURE=crash_recovery.feature make test'

.github/workflows/docker-tests.yml

@@ -41,8 +41,8 @@ jobs:
       matrix:
         command:
           - 'GODOG_FEATURE=active_nodes.feature make test'                                                                        
-          - 'GODOG_FEATURE=async.feature make test'                                                                               
-          - 'GODOG_FEATURE=cascade_replicas.feature make test'                                                                    
+          - 'GODOG_FEATURE=async.feature make test'
+          - 'GODOG_FEATURE=cascade_replicas.feature make test'
           - 'GODOG_FEATURE=CLI.feature make test'                                                                                 
           - 'GODOG_FEATURE=crash_recovery.feature make test'                                                                      
           - 'GODOG_FEATURE=events_reenable.feature make test'                                                                     

internal/app/app.go

@@ -700,6 +700,11 @@ func (app *App) stateManager() appState {
 		app.logger.Errorf("failed to update active nodes in dcs: %v", err)
 	}
 
+	err = app.updateMdbReplMonTS(master)
+	if err != nil {
+		app.logger.Errorf("failed to update mdb_repl_mon timestamp: %v", err)
+	}
+
 	return stateManager
 }
 
@@ -1218,8 +1223,13 @@ func (app *App) performSwitchover(clusterState map[string]*NodeState, activeNode
 		newMaster = switchover.To
 	} else if switchover.From != "" {
 		positions2 := filterOutNodeFromPositions(positions, switchover.From)
+		PriorityChoiceMaxLag := app.config.PriorityChoiceMaxLag
+		AsyncAllowedLagTime := time.Duration(app.config.AsyncAllowedLag) * time.Second
+		if app.config.ASync && switchover.Cause == CauseAuto && AsyncAllowedLagTime > app.config.PriorityChoiceMaxLag {
+			PriorityChoiceMaxLag = AsyncAllowedLagTime
+		}
 		// we ignore splitbrain flag as it should be handled during searching most recent host
-		newMaster, err = getMostDesirableNode(app.logger, positions2, app.config.PriorityChoiceMaxLag)
+		newMaster, err = getMostDesirableNode(app.logger, positions2, PriorityChoiceMaxLag)
 		if err != nil {
 			return fmt.Errorf("switchover: error while looking for highest priority node: %s", switchover.From)
 		}
@@ -1270,6 +1280,10 @@ func (app *App) performSwitchover(clusterState map[string]*NodeState, activeNode
 
 	// turn slaves to the new master
 	app.logger.Info("switchover: phase 5: turn to the new master")
+	err = app.cluster.Get(newMaster).SetOnline()
+	if err != nil {
+		return fmt.Errorf("got error on setting new master %s online %v", newMaster, err)
+	}
 	errs = util.RunParallel(func(host string) error {
 		if host == newMaster || !clusterState[host].PingOk {
 			return nil
@@ -2118,9 +2132,28 @@ func (app *App) waitForCatchUp(node *mysql.Node, gtidset gtids.GTIDSet, timeout
 		if gtidExecuted.Contain(gtidset) {
 			return true, nil
 		}
-		if app.dcs.Get(pathCurrentSwitch, new(Switchover)) == dcs.ErrNotFound {
+		switchover := new(Switchover)
+		if app.dcs.Get(pathCurrentSwitch, switchover) == dcs.ErrNotFound {
 			return false, nil
 		}
+		if app.config.ASync && switchover.Cause == CauseAuto {
+			app.logger.Infof("async mode is active and this is auto switch so we checking new master delay")
+			ts, err := app.GetMdbReplMonTS()
+			if err != nil {
+				app.logger.Errorf("failed to get mdb repl mon ts: %v", err)
+				continue
+			}
+			delay, err := node.CalcMdbReplMonTSDelay(ts)
+			if err != nil {
+				app.logger.Errorf("failed to calc mdb repl mon ts: %v", err)
+				continue
+			}
+			if delay < app.config.AsyncAllowedLag {
+				app.logger.Infof("async allowed lag is %d and current lag on host %s is %d, so we don't wait for catch up any more",
+					app.config.AsyncAllowedLag, node.Host(), delay)
+				return true, nil
+			}
+		}
 		time.Sleep(sleep)
 		if time.Now().After(deadline) {
 			break
@@ -2224,6 +2257,15 @@ func (app *App) getNodePositions(activeNodes []string) ([]nodePosition, error) {
 	return positions, util.CombineErrors(errs)
 }
 
+func (app *App) updateMdbReplMonTS(master string) error {
+	masterNode := app.cluster.Get(master)
+	ts, err := masterNode.GetMdbReplMonTS()
+	if err != nil {
+		return fmt.Errorf("failed to get master mdb_repl_mon timestamp: %v", err)
+	}
+	return app.SetMdbReplMonTS(ts)
+}
+
 /*
 Run enters the main application loop
 When Run exits mysync process is over

internal/app/app_dcs.go

@@ -228,3 +228,24 @@ func (app *App) GetMasterHostFromDcs() (string, error) {
 	}
 	return "", nil
 }
+
+func (app *App) SetMdbReplMonTS(ts string) error {
+	err := app.dcs.Create(pathMasterReplMonTS, ts)
+	if err != nil && err != dcs.ErrExists {
+		return err
+	}
+	err = app.dcs.Set(pathMasterReplMonTS, ts)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+func (app *App) GetMdbReplMonTS() (string, error) {
+	var ts string
+	err := app.dcs.Get(pathMasterReplMonTS, &ts)
+	if errors.Is(err, dcs.ErrNotFound) {
+		return "", nil
+	}
+	return ts, err
+}

internal/app/data.go

@@ -67,6 +67,9 @@ const (
 	pathResetupStatus = "resetup_status"
 
 	pathLastShutdownNodeTime = "last_shutdown_node_time"
+
+	// last known timestamp from mysql.mdb_repl_mon
+	pathMasterReplMonTS = "master_repl_mon_ts"
 )
 
 var (

internal/config/config.go

@@ -88,6 +88,8 @@ type Config struct {
 	ReplicationChannel                      string                       `config:"replication_channel" yaml:"replication_channel"`
 	ExternalReplicationChannel              string                       `config:"external_replication_channel" yaml:"external_replication_channel"`
 	ExternalReplicationType                 util.ExternalReplicationType `config:"external_replication_type" yaml:"external_replication_type"`
+	ASync                                   bool                         `config:"async" yaml:"async"`
+	AsyncAllowedLag                         int64                        `config:"async_allowed_lag" yaml:"async_allowed_lag"`
 }
 
 // DefaultConfig returns default configuration for MySync
@@ -164,6 +166,8 @@ func DefaultConfig() (Config, error) {
 		ReplicationChannel:                      "",
 		ExternalReplicationChannel:              "external",
 		ExternalReplicationType:                 util.Disabled,
+		ASync:                                   false,
+		AsyncAllowedLag:                         0,
 	}
 	return config, nil
 }
@@ -205,5 +209,8 @@ func (cfg *Config) Validate() error {
 	if cfg.NotCriticalDiskUsage > cfg.CriticalDiskUsage {
 		return fmt.Errorf("not_critical_disk_usage should be <= critical_disk_usage")
 	}
+	if cfg.SemiSync && cfg.ASync {
+		return fmt.Errorf("can't run in both semisync and async mode")
+	}
 	return nil
 }

internal/mysql/data.go

@@ -268,6 +268,14 @@ type Version struct {
 	PatchVersion int `db:"PatchVersion"`
 }
 
+type MdbReplMonTS struct {
+	Timestamp string `db:"ts"`
+}
+
+type MdbReplMonTSDelay struct {
+	Delay int64 `db:"delay"`
+}
+
 const (
 	Version80Major              = 8
 	Version80Minor              = 0

internal/mysql/node.go

@@ -985,3 +985,15 @@ func (n *Node) SetDefaultReplicationSettings(masterNode *Node) error {
 	}
 	return nil
 }
+
+func (n *Node) GetMdbReplMonTS() (string, error) {
+	result := new(MdbReplMonTS)
+	err := n.queryRow(queryGetMdbReplMonTS, nil, result)
+	return result.Timestamp, err
+}
+
+func (n *Node) CalcMdbReplMonTSDelay(ts string) (int64, error) {
+	result := new(MdbReplMonTSDelay)
+	err := n.queryRow(queryCalcMdbReplMonTSDelay, map[string]interface{}{"ts": ts}, result)
+	return result.Delay, err
+}

internal/mysql/queries.go

@@ -45,6 +45,8 @@ const (
 	querySetInnodbFlushLogAtTrxCommit   = "set_innodb_flush_log_at_trx_commit"
 	querySetSyncBinlog                  = "set_sync_binlog"
 	queryGetReplicationSettings         = "get_replication_settings"
+	queryGetMdbReplMonTS                = "get_mdb_repl_mon_ts"
+	queryCalcMdbReplMonTSDelay          = "calc_mdb_repl_mon_ts_delay"
 )
 
 var DefaultQueries = map[string]string{
@@ -123,4 +125,6 @@ var DefaultQueries = map[string]string{
 	querySetInnodbFlushLogAtTrxCommit: `SET GLOBAL innodb_flush_log_at_trx_commit = :level`,
 	queryGetReplicationSettings:       `SELECT @@innodb_flush_log_at_trx_commit as InnodbFlushLogAtTrxCommit, @@sync_binlog as SyncBinlog`,
 	querySetSyncBinlog:                `SET GLOBAL sync_binlog = :sync_binlog`,
+	queryGetMdbReplMonTS:              `SELECT UNIX_TIMESTAMP(ts) AS ts FROM mysql.mdb_repl_mon`,
+	queryCalcMdbReplMonTSDelay:        `SELECT FLOOR(CAST(:ts AS DECIMAL(20,3)) - UNIX_TIMESTAMP(ts)) AS delay FROM mysql.mdb_repl_mon`,
 }