support async replication

internal/app/app.go

@@ -700,6 +700,11 @@
 		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
 }
 
@@ -2118,9 +2123,27 @@
 		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 {
+			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 %s and current lag on host %s is %s, so we don't wait for catch up any more",
+					app.config.ASyncAllowedLag, node.Host(), delay)
+				break
+			}
+		}
 		time.Sleep(sleep)
 		if time.Now().After(deadline) {
 			break
@@ -2224,6 +2247,15 @@
 	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 @@
 	}
 	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 @@
 	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:"a_sync" yaml:"a_sync"`
+	ASyncAllowedLag                       	int64                        `config:"a_sync_allowed_lag" yaml:"a_sync_allowed_lag"`
 }
 
 // DefaultConfig returns default configuration for MySync
@@ -164,6 +166,8 @@
 		ReplicationChannel:                      "",
 		ExternalReplicationChannel:              "external",
 		ExternalReplicationType:                 util.Disabled,
+		ASync:          						false,
+		ASyncAllowedLag: 0,
 	}
 	return config, nil
 }
@@ -205,5 +209,8 @@
 	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 @@
 	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 @@
 	}
 	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 @@
 	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 @@
 	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 UNIX_TIMESTAMP(CURRENT_TIMESTAMP(0))  - CAST(:ts AS DECIMAL(20,0)) AS delay`,
 }