Merge pull request #1171 from openark/analysis-semi-sync

Better semi-sync analysis

go/db/generate_patches.go

@@ -546,4 +546,34 @@ var generateSQLPatches = []string{
 			database_instance
 			ADD COLUMN region varchar(32) CHARACTER SET ascii NOT NULL AFTER data_center
 	`,
+	`
+		ALTER TABLE
+			database_instance
+			ADD COLUMN semi_sync_master_timeout INT UNSIGNED NOT NULL DEFAULT 0 AFTER semi_sync_master_enabled
+	`,
+	`
+		ALTER TABLE
+			database_instance
+			ADD COLUMN semi_sync_master_wait_for_slave_count INT UNSIGNED NOT NULL DEFAULT 0 AFTER semi_sync_master_timeout
+	`,
+	`
+		ALTER TABLE
+			database_instance
+			ADD COLUMN semi_sync_master_status TINYINT UNSIGNED NOT NULL DEFAULT 0 AFTER semi_sync_master_wait_for_slave_count
+	`,
+	`
+		ALTER TABLE
+			database_instance
+			ADD COLUMN semi_sync_replica_status TINYINT UNSIGNED NOT NULL DEFAULT 0 AFTER semi_sync_master_status
+	`,
+	`
+		ALTER TABLE
+			database_instance
+			ADD COLUMN semi_sync_master_clients INT UNSIGNED NOT NULL DEFAULT 0 AFTER semi_sync_master_status
+	`,
+	`
+		ALTER TABLE
+			database_instance
+			ADD COLUMN semi_sync_available TINYINT UNSIGNED NOT NULL DEFAULT 0 AFTER semi_sync_enforced
+	`,
 }

go/inst/analysis.go

@@ -66,6 +66,8 @@ const (
 	ErrantGTIDStructureWarning                                               = "ErrantGTIDStructureWarning"
 	NoFailoverSupportStructureWarning                                        = "NoFailoverSupportStructureWarning"
 	NoWriteableMasterStructureWarning                                        = "NoWriteableMasterStructureWarning"
+	NoValidSemiSyncReplicasStructureWarning                                  = "NoValidSemiSyncReplicasStructureWarning"
+	NotEnoughValidSemiSyncReplicasStructureWarning                           = "NotEnoughValidSemiSyncReplicasStructureWarning"
 )
 
 type InstanceAnalysis struct {
@@ -140,6 +142,8 @@ type ReplicationAnalysis struct {
 	OracleGTIDImmediateTopology               bool
 	MariaDBGTIDImmediateTopology              bool
 	BinlogServerImmediateTopology             bool
+	SemiSyncMasterEnabled                     bool
+	CountSemiSyncReplicasEnabled              uint
 	CountLoggingReplicas                      uint
 	CountStatementBasedLoggingReplicas        uint
 	CountMixedBasedLoggingReplicas            uint

go/inst/analysis_dao.go

@@ -151,6 +151,12 @@ func GetReplicationAnalysis(clusterName string, hints *ReplicationAnalysisHints)
 			    	MIN(
 				    		master_instance.supports_oracle_gtid
 				    	) AS supports_oracle_gtid,
+						MIN(
+				    		master_instance.semi_sync_master_enabled
+				    	) AS semi_sync_master_enabled,
+						MIN(
+				    		master_instance.semi_sync_master_wait_for_slave_count
+				    	) AS semi_sync_master_wait_for_slave_count,
 						SUM(
 								replica_instance.is_co_master
 							) AS count_co_master_replicas,
@@ -166,6 +172,12 @@ func GetReplicationAnalysis(clusterName string, hints *ReplicationAnalysisHints)
 		        IFNULL(SUM(replica_instance.last_checked <= replica_instance.last_seen
                   AND replica_instance.binlog_server != 0),
               0) AS count_valid_binlog_server_replicas,
+						SUM(
+				    		replica_instance.semi_sync_replica_enabled
+							) AS count_semi_sync_replicas,
+				    IFNULL(SUM(replica_instance.last_checked <= replica_instance.last_seen
+				        AND replica_instance.semi_sync_replica_enabled != 0),
+				      0) AS count_valid_semi_sync_replicas,
 			    	MIN(
 				    		master_instance.mariadb_gtid
 				    	) AS is_mariadb_gtid,
@@ -295,6 +307,10 @@ func GetReplicationAnalysis(clusterName string, hints *ReplicationAnalysisHints)
 		countValidBinlogServerSlaves := m.GetUint("count_valid_binlog_server_replicas")
 		a.BinlogServerImmediateTopology = countValidBinlogServerSlaves == a.CountValidReplicas && a.CountValidReplicas > 0
 		a.PseudoGTIDImmediateTopology = m.GetBool("is_pseudo_gtid")
+		a.SemiSyncMasterEnabled = m.GetBool("semi_sync_master_enabled")
+		a.CountSemiSyncReplicasEnabled = m.GetUint("count_semi_sync_replicas")
+		countValidSemiSyncReplicasEnabled := m.GetUint("count_valid_semi_sync_replicas")
+		semiSyncMasterWaitForReplicaCount := m.GetUint("semi_sync_master_wait_for_slave_count")
 
 		a.MinReplicaGTIDMode = m.GetString("min_replica_gtid_mode")
 		a.MaxReplicaGTIDMode = m.GetString("max_replica_gtid_mode")
@@ -501,6 +517,13 @@ func GetReplicationAnalysis(clusterName string, hints *ReplicationAnalysisHints)
 				a.StructureAnalysis = append(a.StructureAnalysis, NoWriteableMasterStructureWarning)
 			}
 
+			if a.IsMaster && a.SemiSyncMasterEnabled && countValidSemiSyncReplicasEnabled == 0 {
+				a.StructureAnalysis = append(a.StructureAnalysis, NoValidSemiSyncReplicasStructureWarning)
+			}
+			if a.IsMaster && a.SemiSyncMasterEnabled && countValidSemiSyncReplicasEnabled > 0 && countValidSemiSyncReplicasEnabled < semiSyncMasterWaitForReplicaCount {
+				a.StructureAnalysis = append(a.StructureAnalysis, NotEnoughValidSemiSyncReplicasStructureWarning)
+			}
+
 		}
 		appendAnalysis(&a)
 

go/inst/instance.go

@@ -74,20 +74,26 @@ type Instance struct {
 
 	masterExecutedGtidSet string // Not exported
 
-	SlaveLagSeconds                 sql.NullInt64
-	SlaveHosts                      InstanceKeyMap
-	ClusterName                     string
-	SuggestedClusterAlias           string
-	DataCenter                      string
-	Region                          string
-	PhysicalEnvironment             string
-	ReplicationDepth                uint
-	IsCoMaster                      bool
-	HasReplicationCredentials       bool
-	ReplicationCredentialsAvailable bool
-	SemiSyncEnforced                bool
-	SemiSyncMasterEnabled           bool
-	SemiSyncReplicaEnabled          bool
+	SlaveLagSeconds                   sql.NullInt64
+	SlaveHosts                        InstanceKeyMap
+	ClusterName                       string
+	SuggestedClusterAlias             string
+	DataCenter                        string
+	Region                            string
+	PhysicalEnvironment               string
+	ReplicationDepth                  uint
+	IsCoMaster                        bool
+	HasReplicationCredentials         bool
+	ReplicationCredentialsAvailable   bool
+	SemiSyncAvailable                 bool // when both semi sync plugins (master & replica) are loaded
+	SemiSyncEnforced                  bool
+	SemiSyncMasterEnabled             bool
+	SemiSyncReplicaEnabled            bool
+	SemiSyncMasterTimeout             uint
+	SemiSyncMasterWaitForReplicaCount uint
+	SemiSyncMasterStatus              bool
+	SemiSyncMasterClients             uint
+	SemiSyncReplicaStatus             bool
 
 	LastSeenTimestamp    string
 	IsLastCheckValid     bool

go/inst/instance_dao.go

@@ -422,14 +422,41 @@ func ReadTopologyInstanceBufferable(instanceKey *InstanceKey, bufferWrites bool,
 			waitGroup.Add(1)
 			go func() {
 				defer waitGroup.Done()
-				err := sqlutils.QueryRowsMap(db, "show global status like 'rpl_semi_sync_%_status'", func(m sqlutils.RowMap) error {
-					if m.GetString("Variable_name") == "Rpl_semi_sync_master_status" {
+				semiSyncMasterPluginLoaded := false
+				semiSyncReplicaPluginLoaded := 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" {
 						instance.SemiSyncMasterEnabled = (m.GetString("Value") == "ON")
-					} else if m.GetString("Variable_name") == "Rpl_semi_sync_slave_status" {
+						semiSyncMasterPluginLoaded = true
+					} else if m.GetString("Variable_name") == "rpl_semi_sync_master_timeout" {
+						instance.SemiSyncMasterTimeout = m.GetUint("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.SemiSyncReplicaEnabled = (m.GetString("Value") == "ON")
+						semiSyncReplicaPluginLoaded = true
 					}
 					return nil
 				})
+				instance.SemiSyncAvailable = (semiSyncMasterPluginLoaded && semiSyncReplicaPluginLoaded)
+				errorChan <- err
+			}()
+		}
+		{
+			waitGroup.Add(1)
+			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" {
+						instance.SemiSyncMasterStatus = (m.GetString("Value") == "ON")
+					} else if m.GetString("Variable_name") == "Rpl_semi_sync_master_clients" {
+						instance.SemiSyncMasterClients = m.GetUint("Value")
+					} else if m.GetString("Variable_name") == "Rpl_semi_sync_slave_status" {
+						instance.SemiSyncReplicaStatus = (m.GetString("Value") == "ON")
+					}
+
+					return nil
+				})
 				errorChan <- err
 			}()
 		}
@@ -1110,8 +1137,14 @@ func readInstanceRow(m sqlutils.RowMap) *Instance {
 	instance.Region = m.GetString("region")
 	instance.PhysicalEnvironment = m.GetString("physical_environment")
 	instance.SemiSyncEnforced = m.GetBool("semi_sync_enforced")
+	instance.SemiSyncAvailable = m.GetBool("semi_sync_available")
 	instance.SemiSyncMasterEnabled = m.GetBool("semi_sync_master_enabled")
+	instance.SemiSyncMasterTimeout = m.GetUint("semi_sync_master_timeout")
+	instance.SemiSyncMasterWaitForReplicaCount = m.GetUint("semi_sync_master_wait_for_slave_count")
 	instance.SemiSyncReplicaEnabled = m.GetBool("semi_sync_replica_enabled")
+	instance.SemiSyncMasterStatus = m.GetBool("semi_sync_master_status")
+	instance.SemiSyncMasterClients = m.GetUint("semi_sync_master_clients")
+	instance.SemiSyncReplicaStatus = m.GetBool("semi_sync_replica_status")
 	instance.ReplicationDepth = m.GetUint("replication_depth")
 	instance.IsCoMaster = m.GetBool("is_co_master")
 	instance.ReplicationCredentialsAvailable = m.GetBool("replication_credentials_available")
@@ -2397,8 +2430,14 @@ func mkInsertOdkuForInstances(instances []*Instance, instanceWasActuallyFound bo
 		"has_replication_credentials",
 		"allow_tls",
 		"semi_sync_enforced",
+		"semi_sync_available",
 		"semi_sync_master_enabled",
+		"semi_sync_master_timeout",
+		"semi_sync_master_wait_for_slave_count",
 		"semi_sync_replica_enabled",
+		"semi_sync_master_status",
+		"semi_sync_master_clients",
+		"semi_sync_replica_status",
 		"instance_alias",
 		"last_discovery_latency",
 	}
@@ -2477,8 +2516,14 @@ func mkInsertOdkuForInstances(instances []*Instance, instanceWasActuallyFound bo
 		args = append(args, instance.HasReplicationCredentials)
 		args = append(args, instance.AllowTLS)
 		args = append(args, instance.SemiSyncEnforced)
+		args = append(args, instance.SemiSyncAvailable)
 		args = append(args, instance.SemiSyncMasterEnabled)
+		args = append(args, instance.SemiSyncMasterTimeout)
+		args = append(args, instance.SemiSyncMasterWaitForReplicaCount)
 		args = append(args, instance.SemiSyncReplicaEnabled)
+		args = append(args, instance.SemiSyncMasterStatus)
+		args = append(args, instance.SemiSyncMasterClients)
+		args = append(args, instance.SemiSyncReplicaStatus)
 		args = append(args, instance.InstanceAlias)
 		args = append(args, instance.LastDiscoveryLatency.Nanoseconds())
 	}