add mysync-repl-mon feature

internal/app/app.go

@@ -256,6 +256,51 @@ func (app *App) externalCAFileChecker(ctx context.Context) {
 	}
 }
 
+func (app *App) replMonWriter(ctx context.Context) {
+	ticker := time.NewTicker(app.config.ReplMonWriteInterval)
+	for {
+		select {
+		case <-ticker.C:
+			localNode := app.cluster.Local()
+			sstatus, err := localNode.GetReplicaStatus()
+			if err != nil {
+				app.logger.Errorf("repl mon writer: got error %v while checking replica status", err)
+				time.Sleep(app.config.ReplMonErrorWaitInterval)
+				continue
+			}
+			if sstatus != nil {
+				app.logger.Infof("repl mon writer: host is replica")
+				time.Sleep(app.config.ReplMonSlaveWaitInterval)
+				continue
+			}
+			readOnly, _, err := localNode.IsReadOnly()
+			if err != nil {
+				app.logger.Errorf("repl mon writer: got error %v while checking read only status", err)
+				time.Sleep(app.config.ReplMonErrorWaitInterval)
+				continue
+			}
+			if readOnly {
+				app.logger.Infof("repl mon writer: host is read only")
+				time.Sleep(app.config.ReplMonSlaveWaitInterval)
+				continue
+			}
+			err = localNode.UpdateReplMonTable(app.config.ReplMonTableName)
+			if err != nil {
+				if mysql.IsErrorTableDoesNotExists(err) {
+					err = localNode.CreateReplMonTable(app.config.ReplMonTableName)
+					if err != nil {
+						app.logger.Errorf("repl mon writer: got error %v while creating repl mon table", err)
+					}
+					continue
+				}
+				app.logger.Errorf("repl mon writer: got error %v while writing in repl mon table", err)
+			}
+		case <-ctx.Done():
+			return
+		}
+	}
+}
+
 func (app *App) SetResetupStatus() {
 	err := app.setResetupStatus(app.cluster.Local().Host(), app.doesResetupFileExist())
 	if err != nil {
@@ -2285,6 +2330,9 @@ func (app *App) Run() int {
 	if app.config.ExternalReplicationType != util.Disabled {
 		go app.externalCAFileChecker(ctx)
 	}
+	if app.config.ASync {
+		go app.replMonWriter(ctx)
+	}
 
 	handlers := map[appState](func() appState){
 		stateFirstRun:    app.stateFirstRun,

internal/config/config.go

@@ -90,6 +90,11 @@ type Config struct {
 	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"`
+	ReplMon                                 bool                         `config:"repl_mon" yaml:"repl_mon"`
+	ReplMonTableName						string                       `config:"repl_mon_table_name" yaml:"repl_mon_table_name"`
+	ReplMonWriteInterval					time.Duration                `config:"repl_mon_write_interval" yaml:"repl_mon_write_interval"`
+	ReplMonErrorWaitInterval				time.Duration                `config:"repl_mon_error_wait_interval" yaml:"repl_mon_error_wait_interval"`
+	ReplMonSlaveWaitInterval				time.Duration                `config:"repl_mon_slave_wait_interval" yaml:"repl_mon_slave_wait_interval"`
 }
 
 // DefaultConfig returns default configuration for MySync
@@ -168,6 +173,11 @@ func DefaultConfig() (Config, error) {
 		ExternalReplicationType:                 util.Disabled,
 		ASync:                                   false,
 		AsyncAllowedLag:                         0,
+		ReplMon: 								 false,
+		ReplMonTableName: 						"mysql.mysync_repl_mon",
+		ReplMonWriteInterval:					 1 * time.Second,
+		ReplMonErrorWaitInterval:				 10 * time.Second,
+		ReplMonSlaveWaitInterval:				 10 * time.Second,
 	}
 	return config, nil
 }

internal/mysql/node.go

@@ -1016,3 +1016,13 @@ func (n *Node) CalcMdbReplMonTSDelay(ts string) (int64, error) {
 	err := n.queryRow(queryCalcMdbReplMonTSDelay, map[string]interface{}{"ts": ts}, result)
 	return result.Delay, err
 }
+
+func (n *Node) CreateReplMonTable(replMonTable string) error {
+	err := n.exec(queryCreateReplMonTable, map[string]interface{}{"replMonTable": replMonTable})
+	return err
+}
+
+func (n *Node) UpdateReplMonTable(replMonTable string) error {
+	err := n.exec(queryUpdateReplMon, map[string]interface{}{"replMonTable": replMonTable})
+	return err
+}

internal/mysql/queries.go

@@ -48,6 +48,8 @@ const (
 	queryGetReplicationSettings         = "get_replication_settings"
 	queryGetMdbReplMonTS                = "get_mdb_repl_mon_ts"
 	queryCalcMdbReplMonTSDelay          = "calc_mdb_repl_mon_ts_delay"
+	queryCreateReplMonTable				= "create_repl_mon_table"
+	queryUpdateReplMon					= "update_repl_mon"
 )
 
 var DefaultQueries = map[string]string{
@@ -129,4 +131,15 @@ var DefaultQueries = map[string]string{
 	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`,
+	queryCreateReplMonTable:			`CREATE TABLE IF NOT EXISTS :replMonTable (
+												id INT NOT NULL PRIMARY KEY,
+												ts TIMESTAMP(3)
+										)
+										ENGINE=INNODB`,
+	queryUpdateReplMon: 				`INSERT INTO :replMonTable (id, ts)
+										(
+											SELECT 1, CURRENT_TIMESTAMP(3)
+											WHERE @@read_only = 0
+										)
+									ON DUPLICATE KEY UPDATE ts = CURRENT_TIMESTAMP(3)`,
 }

tests/features/async_setting.feature

@@ -9,6 +9,7 @@ Feature: mysync async mode tests
       MYSYNC_FAILOVER=true
       MYSYNC_FAILOVER_DELAY=0s
       MYSYNC_FAILOVER_COOLDOWN=0s
+      REPL_MON=true
       """
     Given cluster is up and running
     When I wait for "10" seconds
@@ -23,25 +24,6 @@ Feature: mysync async mode tests
     And mysql host "mysql3" should have variable "rpl_semi_sync_master_enabled" set to "0"
     And mysql host "mysql3" should have variable "rpl_semi_sync_slave_enabled" set to "0"
 
-
-    When I run SQL on mysql host "mysql1"
-      """
-      CREATE TABLE mysql.mdb_repl_mon(
-          ts TIMESTAMP(3)
-      ) ENGINE=INNODB;
-      """
-    And I run SQL on mysql host "mysql1"
-      """
-      INSERT INTO mysql.mdb_repl_mon VALUES(CURRENT_TIMESTAMP(3));
-      """
-    And I run SQL on mysql host "mysql1"
-      """
-      CREATE EVENT mysql.mdb_repl_mon_event
-      ON SCHEDULE EVERY 1 SECOND
-      DO UPDATE mysql.mdb_repl_mon SET ts = CURRENT_TIMESTAMP(3);
-      """
-    Then mysql host "mysql1" should have event "mysql.mdb_repl_mon_event" in status "ENABLED"
-
     And I wait for "2" seconds
     And I run SQL on mysql host "mysql1"
       """
@@ -122,6 +104,7 @@ Feature: mysync async mode tests
       MYSYNC_FAILOVER=true
       MYSYNC_FAILOVER_DELAY=0s
       MYSYNC_FAILOVER_COOLDOWN=0s
+      REPL_MON=true
       """
     Given cluster is up and running
     When I wait for "10" seconds
@@ -244,6 +227,7 @@ Feature: mysync async mode tests
       MYSYNC_FAILOVER=true
       MYSYNC_FAILOVER_DELAY=0s
       MYSYNC_FAILOVER_COOLDOWN=0s
+      REPL_MON=true
       """
     Given cluster is up and running
     When I wait for "10" seconds
@@ -369,6 +353,7 @@ Feature: mysync async mode tests
       MYSYNC_FAILOVER=true
       MYSYNC_FAILOVER_DELAY=0s
       MYSYNC_FAILOVER_COOLDOWN=0s
+      REPL_MON=true
       """
     Given cluster is up and running
     When I wait for "10" seconds

tests/images/docker-compose.yaml

@@ -96,6 +96,7 @@ services:
       MYSYNC_REPLICATION_REPAIR_AGGRESSIVE_MODE:
       MYSYNC_SET_RO_TIMEOUT:
       MYSYNC_REPLICATION_LAG_QUERY:
+      REPL_MON:
     healthcheck:
       test: "mysql --user=admin --password=admin_pwd -e 'SELECT 1'"
       start_period: 30s
@@ -141,6 +142,7 @@ services:
       MYSYNC_REPLICATION_REPAIR_AGGRESSIVE_MODE:
       MYSYNC_SET_RO_TIMEOUT:
       MYSYNC_REPLICATION_LAG_QUERY:
+      REPL_MON:
     depends_on:
       mysql1:
         condition: service_healthy
@@ -180,6 +182,7 @@ services:
       MYSYNC_REPLICATION_REPAIR_AGGRESSIVE_MODE:
       MYSYNC_SET_RO_TIMEOUT:
       MYSYNC_REPLICATION_LAG_QUERY:
+      REPL_MON:
     depends_on:
       mysql1:
         condition: service_healthy

tests/images/mysql/mysync.yaml

@@ -60,3 +60,4 @@ replication_repair_aggressive_mode: ${MYSYNC_REPLICATION_REPAIR_AGGRESSIVE_MODE:
 test_filesystem_readonly_file: /tmp/readonly
 replication_channel: ''
 external_replication_type: 'external'
+repl_mon: ${REPL_MON:-false}