Merge pull request #2 from kamil-holubicki/DISTMYSQL-138-3

DISTMYSQL-138: MySQL Orchestrator does not support new semi-sync variables introduced in MySQL 8.0.26

Author: kamil-holubicki

go/http/api.go

@@ -75,6 +75,8 @@ var apiSynonyms = map[string]string{
 	"stop-slave-nice":            "stop-replica-nice",
 	"reset-slave":                "reset-replica",
 	"restart-slave-statements":   "restart-replica-statements",
+	"enable-semi-sync-master":    "enable-semi-sync-source",
+	"disable-semi-sync-master":   "disable-semi-sync-source",
 }
 
 var registeredPaths = []string{}

go/inst/instance.go

@@ -95,6 +95,8 @@ type Instance struct {
 	ReplicationCredentialsAvailable   bool
 	SemiSyncAvailable                 bool // when both semi sync plugins (master & replica) are loaded
 	SemiSyncPriority                  uint // higher value means higher priority, zero means async replica
+	SemiSyncMasterPluginNewVersion    bool // true for the plugin introduced with MySql 8.0.26
+	SemiSyncReplicaPluginNewVersion   bool // true for the plugin introduced with MySql 8.0.26
 	SemiSyncMasterEnabled             bool
 	SemiSyncReplicaEnabled            bool
 	SemiSyncMasterTimeout             uint64

go/inst/instance_dao.go

@@ -21,7 +21,6 @@ import (
 	"database/sql"
 	"errors"
 	"fmt"
-	"github.com/go-sql-driver/mysql"
 	"regexp"
 	"runtime"
 	"sort"
@@ -30,6 +29,8 @@ import (
 	"sync"
 	"time"
 
+	"github.com/go-sql-driver/mysql"
+
 	"github.com/openark/golib/log"
 	"github.com/openark/golib/math"
 	"github.com/openark/golib/sqlutils"
@@ -463,20 +464,54 @@ func ReadTopologyInstanceBufferable(instanceKey *InstanceKey, bufferWrites bool,
 				defer waitGroup.Done()
 				semiSyncMasterPluginLoaded := false
 				semiSyncReplicaPluginLoaded := false
+				instance.SemiSyncAvailable = false
+
 				err := sqlutils.QueryRowsMap(db, "show global variables like 'rpl_semi_sync_%'", func(m sqlutils.RowMap) error {
-					if m.GetString("Variable_name") == "rpl_semi_sync_master_enabled" {
+					variableName := m.GetString("Variable_name")
+					// Learn if semi-sync plugin is loaded and what is its version
+					if variableName == "rpl_semi_sync_master_enabled" {
 						instance.SemiSyncMasterEnabled = (m.GetString("Value") == "ON")
 						semiSyncMasterPluginLoaded = true
-					} else if m.GetString("Variable_name") == "rpl_semi_sync_master_timeout" {
-						instance.SemiSyncMasterTimeout = m.GetUint64("Value")
-					} else if m.GetString("Variable_name") == "rpl_semi_sync_master_wait_for_slave_count" {
-						instance.SemiSyncMasterWaitForReplicaCount = m.GetUint("Value")
-					} else if m.GetString("Variable_name") == "rpl_semi_sync_slave_enabled" {
+						instance.SemiSyncMasterPluginNewVersion = false
+					} else if variableName == "rpl_semi_sync_source_enabled" {
+						instance.SemiSyncMasterEnabled = (m.GetString("Value") == "ON")
+						semiSyncMasterPluginLoaded = true
+						instance.SemiSyncMasterPluginNewVersion = true
+					} else if variableName == "rpl_semi_sync_slave_enabled" {
 						instance.SemiSyncReplicaEnabled = (m.GetString("Value") == "ON")
 						semiSyncReplicaPluginLoaded = true
+						instance.SemiSyncReplicaPluginNewVersion = false
+					} else if variableName == "rpl_semi_sync_replica_enabled" {
+						instance.SemiSyncReplicaEnabled = (m.GetString("Value") == "ON")
+						semiSyncReplicaPluginLoaded = true
+						instance.SemiSyncReplicaPluginNewVersion = true
+					} else {
+						// additional info
+						matched, regexperr := regexp.MatchString("^rpl_semi_sync_(master|source)_timeout$", variableName)
+						if regexperr != nil {
+							return regexperr
+						}
+						if matched {
+							instance.SemiSyncMasterTimeout = m.GetUint64("Value")
+							return nil
+						}
+
+						matched, regexperr = regexp.MatchString("^rpl_semi_sync_(master|source)_wait_for_(slave|replica)_count$", variableName)
+						if regexperr != nil {
+							return regexperr
+						}
+						if matched {
+							instance.SemiSyncMasterWaitForReplicaCount = m.GetUint("Value")
+							return nil
+						}
 					}
 					return nil
 				})
+				if err != nil {
+					errorChan <- err
+					return
+				}
+
 				instance.SemiSyncAvailable = (semiSyncMasterPluginLoaded && semiSyncReplicaPluginLoaded)
 				errorChan <- err
 			}()
@@ -486,14 +521,32 @@ func ReadTopologyInstanceBufferable(instanceKey *InstanceKey, bufferWrites bool,
 			go func() {
 				defer waitGroup.Done()
 				err := sqlutils.QueryRowsMap(db, "show global status like 'rpl_semi_sync_%'", func(m sqlutils.RowMap) error {
-					if m.GetString("Variable_name") == "Rpl_semi_sync_master_status" {
+					variableName := m.GetString("Variable_name")
+					matched, regexperr := regexp.MatchString("^Rpl_semi_sync_(master|source)_status$", variableName)
+					if regexperr != nil {
+						return regexperr
+					}
+					if matched {
 						instance.SemiSyncMasterStatus = (m.GetString("Value") == "ON")
-					} else if m.GetString("Variable_name") == "Rpl_semi_sync_master_clients" {
+						return nil
+					}
+
+					matched, regexperr = regexp.MatchString("^Rpl_semi_sync_(master|source)_clients$", variableName)
+					if regexperr != nil {
+						return regexperr
+					}
+					if matched {
 						instance.SemiSyncMasterClients = m.GetUint("Value")
-					} else if m.GetString("Variable_name") == "Rpl_semi_sync_slave_status" {
-						instance.SemiSyncReplicaStatus = (m.GetString("Value") == "ON")
+						return nil
 					}
 
+					matched, regexperr = regexp.MatchString("^Rpl_semi_sync_(slave|replica)_status$", variableName)
+					if regexperr != nil {
+						return regexperr
+					}
+					if matched {
+						instance.SemiSyncReplicaStatus = (m.GetString("Value") == "ON")
+					}
 					return nil
 				})
 				errorChan <- err

go/inst/instance_topology_dao.go

@@ -223,7 +223,11 @@ func SetSemiSyncMaster(instanceKey *InstanceKey, enableMaster bool) (*Instance,
 	if err != nil {
 		return instance, err
 	}
-	if _, err := ExecInstance(instanceKey, "set @@global.rpl_semi_sync_master_enabled=?", enableMaster); err != nil {
+	query := "set @@global.rpl_semi_sync_master_enabled=?"
+	if instance.SemiSyncMasterPluginNewVersion {
+		query = "set @@global.rpl_semi_sync_source_enabled=?"
+	}
+	if _, err := ExecInstance(instanceKey, query, enableMaster); err != nil {
 		return instance, log.Errore(err)
 	}
 	return ReadTopologyInstance(instanceKey)
@@ -237,7 +241,13 @@ func SetSemiSyncReplica(instanceKey *InstanceKey, enableReplica bool) (*Instance
 	if instance.SemiSyncReplicaEnabled == enableReplica {
 		return instance, nil
 	}
-	if _, err := ExecInstance(instanceKey, "set @@global.rpl_semi_sync_slave_enabled=?", enableReplica); err != nil {
+
+	query := "set @@global.rpl_semi_sync_slave_enabled=?"
+	if instance.SemiSyncReplicaPluginNewVersion {
+		query = "set @@global.rpl_semi_sync_replica_enabled=?"
+	}
+
+	if _, err := ExecInstance(instanceKey, query, enableReplica); err != nil {
 		return instance, log.Errore(err)
 	}
 	if instance.ReplicationIOThreadRuning {

tests/system/semi-sync/01-semi-sync-all/expect_output

@@ -0,0 +1,4 @@
+127.0.0.1:10111   [0s,ok,VERSION,rw,ROW,>>,GTID,semi:master]
++ 127.0.0.1:10112 [0s,ok,VERSION,ro,ROW,>>,GTID,semi:replica]
++ 127.0.0.1:10113 [0s,ok,VERSION,ro,ROW,>>,GTID,semi:replica]
++ 127.0.0.1:10114 [0s,ok,VERSION,ro,ROW,>>,GTID,semi:replica]

tests/system/semi-sync/01-semi-sync-all/run

@@ -0,0 +1,2 @@
+# sed is to get rid of MySql version from the output
+orchestrator-client -c topology -i 127.0.0.1:10111 | sed -n "s/\(^.*0s,ok,\)\s*\S*\(,r.,.*$\)/\1VERSION\2/p"

tests/system/semi-sync/01-semi-sync-all/setup

@@ -0,0 +1,9 @@
+#!/bin/bash
+
+set -e
+
+orchestrator-client -c enable-semi-sync-master -i 127.0.0.1:10111
+orchestrator-client -c enable-semi-sync-replica -i 127.0.0.1:10112
+orchestrator-client -c enable-semi-sync-replica -i 127.0.0.1:10113
+orchestrator-client -c enable-semi-sync-replica -i 127.0.0.1:10114
+sleep 10

tests/system/semi-sync/skip_run

@@ -0,0 +1,8 @@
+#!/bin/bash
+
+orchestrator-client -c all-instances | while read i ; do
+  orchestrator-client -c disable-semi-sync-master -i $i
+  orchestrator-client -c disable-semi-sync-replica -i $i
+done
+
+sleep 5