From b9676bc504ae2b304cd9216d71f6a1e8e4bfb987 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marcin=20Sza=C5=82owicz?= Date: Thu, 3 Apr 2025 10:25:38 +0200 Subject: [PATCH] New option to allow lower versions for replicas There may be a situation when one would need to be able to rearrange topology in such way thet server with higher version will be LEADER for REPLICA that has lower version. This patch allows that by introduction of AllowLowerVersionForReplication option which is turned off by default. Once enabled each time such change will be made, warning will be written into log file. --- RELEASE_VERSION | 2 +- go/config/config.go | 2 + go/http/api.go | 16 ++++-- go/inst/instance.go | 10 +++- go/inst/instance_test.go | 18 ++++++- go/inst/instance_topology.go | 92 ++++++++++++++++++++++++++++++----- go/inst/instance_utils.go | 2 +- go/logic/topology_recovery.go | 8 ++- 8 files changed, 128 insertions(+), 22 deletions(-) diff --git a/RELEASE_VERSION b/RELEASE_VERSION index 34cde5690..fbba6ba71 100644 --- a/RELEASE_VERSION +++ b/RELEASE_VERSION @@ -1 +1 @@ -3.2.6 +3.2.6-1 diff --git a/go/config/config.go b/go/config/config.go index dbd5567a5..6c170e9f1 100644 --- a/go/config/config.go +++ b/go/config/config.go @@ -151,6 +151,7 @@ type Configuration struct { BufferInstanceWrites bool // Set to 'true' for write-optimization on backend table (compromise: writes can be stale and overwrite non stale data) InstanceFlushIntervalMilliseconds int // Max interval between instance write buffer flushes SkipMaxScaleCheck bool // If you don't ever have MaxScale BinlogServer in your topology (and most people don't), set this to 'true' to save some pointless queries + AllowLowerVersionForReplication bool // Disable checking if MySQL version we're trying to replicate from is higher than the replica, warning is produced. UnseenInstanceForgetHours uint // Number of hours after which an unseen instance is forgotten SnapshotTopologiesIntervalHours uint // Interval in hour between snapshot-topologies invocation. Default: 0 (disabled) DiscoveryMaxConcurrency uint // Number of goroutines doing hosts discovery @@ -345,6 +346,7 @@ func newConfiguration() *Configuration { BufferInstanceWrites: false, InstanceFlushIntervalMilliseconds: 100, SkipMaxScaleCheck: true, + AllowLowerVersionForReplication: false, UnseenInstanceForgetHours: 240, SnapshotTopologiesIntervalHours: 0, DiscoverByShowSlaveHosts: false, diff --git a/go/http/api.go b/go/http/api.go index e329705e0..639662108 100644 --- a/go/http/api.go +++ b/go/http/api.go @@ -1520,11 +1520,15 @@ func (this *HttpAPI) CanReplicateFrom(params martini.Params, r render.Render, re } canReplicate, err := instance.CanReplicateFrom(belowInstance) - if err != nil { + if !canReplicate { Respond(r, &APIResponse{Code: ERROR, Message: err.Error()}) return } + if err != nil { + log.Warningf("CanReplicateFrom(): %v", err) + } + Respond(r, &APIResponse{Code: OK, Message: fmt.Sprintf("%t", canReplicate), Details: belowKey}) } @@ -1552,14 +1556,16 @@ func (this *HttpAPI) CanReplicateFromGTID(params martini.Params, r render.Render } canReplicate, err := instance.CanReplicateFrom(belowInstance) - if err != nil { + if !canReplicate { Respond(r, &APIResponse{Code: ERROR, Message: err.Error()}) return } - if !canReplicate { - Respond(r, &APIResponse{Code: OK, Message: fmt.Sprintf("%t", canReplicate), Details: belowKey}) - return + + if err != nil { + log.Warningf("CanReplicateFromGTID(): %v", err) + err = nil } + err = inst.CheckMoveViaGTID(instance, belowInstance) if err != nil { Respond(r, &APIResponse{Code: ERROR, Message: err.Error()}) diff --git a/go/inst/instance.go b/go/inst/instance.go index 9e6aa367e..61d2c09f2 100644 --- a/go/inst/instance.go +++ b/go/inst/instance.go @@ -448,6 +448,8 @@ func (this *Instance) IsDescendantOf(other *Instance) bool { // CanReplicateFrom uses heuristics to decide whether this instacne can practically replicate from other instance. // Checks are made to binlog format, version number, binary logs etc. func (this *Instance) CanReplicateFrom(other *Instance) (bool, error) { + var warningMsg error = nil + if this.Key.Equals(&other.Key) { return false, fmt.Errorf("instance cannot replicate from itself: %+v", this.Key) } @@ -462,7 +464,11 @@ func (this *Instance) CanReplicateFrom(other *Instance) (bool, error) { // Not OK for a replica, for it has to relay the logs. } if this.IsSmallerMajorVersion(other) && !this.IsBinlogServer() { - return false, fmt.Errorf("instance %+v has version %s, which is lower than %s on %+v ", this.Key, this.Version, other.Version, other.Key) + if !config.Config.AllowLowerVersionForReplication { + return false, fmt.Errorf("instance %+v has version %s, which is lower than %s on %+v ", this.Key, this.Version, other.Version, other.Key) + } else { + warningMsg = fmt.Errorf("instance %+v has version %s, which is lower than %s on %+v ", this.Key, this.Version, other.Version, other.Key) + } } if this.LogBinEnabled && this.LogReplicationUpdatesEnabled { if this.IsSmallerBinlogFormat(other) { @@ -483,7 +489,7 @@ func (this *Instance) CanReplicateFrom(other *Instance) (bool, error) { if this.SQLDelay < other.SQLDelay && int64(other.SQLDelay) > int64(config.Config.ReasonableMaintenanceReplicationLagSeconds) { return false, fmt.Errorf("%+v has higher SQL_Delay (%+v seconds) than %+v does (%+v seconds)", other.Key, other.SQLDelay, this.Key, this.SQLDelay) } - return true, nil + return true, warningMsg } // HasReasonableMaintenanceReplicationLag returns true when the replica lag is reasonable, and maintenance operations should have a green light to go. diff --git a/go/inst/instance_test.go b/go/inst/instance_test.go index 77a5fc9b5..cbfc330b5 100644 --- a/go/inst/instance_test.go +++ b/go/inst/instance_test.go @@ -17,10 +17,11 @@ package inst import ( + "testing" + "github.com/openark/golib/log" test "github.com/openark/golib/tests" "github.com/openark/orchestrator/go/config" - "testing" ) func init() { @@ -134,6 +135,21 @@ func TestCanReplicateFrom(t *testing.T) { canReplicate, _ = i55.CanReplicateFrom(&i56) test.S(t).ExpectFalse(canReplicate) + i80 := Instance{Key: key3, Version: "8.0"} + i80.LogBinEnabled = true + i80.LogReplicationUpdatesEnabled = true + i80.ServerID = 80 + + canReplicate, err = i56.CanReplicateFrom(&i80) + test.S(t).ExpectNotNil(err) + test.S(t).ExpectEquals(canReplicate, false) + + config.Config.AllowLowerVersionForReplication = true + canReplicate, err = i56.CanReplicateFrom(&i80) + test.S(t).ExpectNotNil(err) + test.S(t).ExpectEquals(canReplicate, true) + config.Config.AllowLowerVersionForReplication = false + iStatement := Instance{Key: key1, Binlog_format: "STATEMENT", ServerID: 1, Version: "5.5", LogBinEnabled: true, LogReplicationUpdatesEnabled: true} iRow := Instance{Key: key2, Binlog_format: "ROW", ServerID: 2, Version: "5.5", LogBinEnabled: true, LogReplicationUpdatesEnabled: true} canReplicate, err = iRow.CanReplicateFrom(&iStatement) diff --git a/go/inst/instance_topology.go b/go/inst/instance_topology.go index 91b78ecef..06f9ed779 100644 --- a/go/inst/instance_topology.go +++ b/go/inst/instance_topology.go @@ -299,9 +299,16 @@ func MoveUp(instanceKey *InstanceKey) (*Instance, error) { return instance, fmt.Errorf("master is not a replica itself: %+v", master.Key) } - if canReplicate, err := instance.CanReplicateFrom(master); canReplicate == false { + canReplicate, err := instance.CanReplicateFrom(master) + if !canReplicate { return instance, err } + + // There may be a situation when it's ok to replicate but err contains some information, we will log this as a warning then + if err != nil { + log.Warningf("MoveUp(): %v", err) + } + if master.IsBinlogServer() { // Quick solution via binlog servers return Repoint(instanceKey, &master.MasterKey, GTIDHintDeny) @@ -429,10 +436,16 @@ func MoveUpReplicas(instanceKey *InstanceKey, pattern string) ([](*Instance), *I var replicaErr error ExecuteOnTopology(func() { - if canReplicate, err := replica.CanReplicateFrom(instance); canReplicate == false || err != nil { + canReplicate, err := replica.CanReplicateFrom(instance) + if !canReplicate { replicaErr = err return } + + if err != nil { + log.Warningf("MoveUpReplicas(): :%v", err) + } + if instance.IsBinlogServer() { // Special case. Just repoint replica, err = Repoint(&replica.Key, instanceKey, GTIDHintDeny) @@ -527,9 +540,15 @@ func MoveBelow(instanceKey, siblingKey *InstanceKey) (*Instance, error) { return instance, fmt.Errorf("instances are not siblings: %+v, %+v", *instanceKey, *siblingKey) } - if canReplicate, err := instance.CanReplicateFrom(sibling); !canReplicate { + canReplicate, err := instance.CanReplicateFrom(sibling) + if !canReplicate { return instance, err } + + if err != nil { + log.Warningf("MoveBelow(): %v", err) + } + log.Infof("Will move %+v below %+v", instanceKey, siblingKey) if maintenanceToken, merr := BeginMaintenance(instanceKey, GetMaintenanceOwner(), fmt.Sprintf("move below %+v", *siblingKey)); merr != nil { @@ -629,9 +648,16 @@ func moveInstanceBelowViaGTID(instance, otherInstance *Instance) (*Instance, err return instance, merr } - if canReplicate, err := instance.CanReplicateFrom(otherInstance); !canReplicate { + var err error + canReplicate, err := instance.CanReplicateFrom(otherInstance) + if !canReplicate { return instance, err } + + if err != nil { + log.Warningf("moveInstanceBelowViaGTID(): %v", err) + } + if err := CheckMoveViaGTID(instance, otherInstance); err != nil { return instance, err } @@ -640,7 +666,6 @@ func moveInstanceBelowViaGTID(instance, otherInstance *Instance) (*Instance, err instanceKey := &instance.Key otherInstanceKey := &otherInstance.Key - var err error if maintenanceToken, merr := BeginMaintenance(instanceKey, GetMaintenanceOwner(), fmt.Sprintf("move below %+v", *otherInstanceKey)); merr != nil { err = fmt.Errorf("Cannot begin maintenance on %+v: %v", *instanceKey, merr) goto Cleanup @@ -812,10 +837,17 @@ func Repoint(instanceKey *InstanceKey, masterKey *InstanceKey, gtidHint Operatio return instance, err } } - if canReplicate, err := instance.CanReplicateFrom(master); !canReplicate { + + canReplicate, err := instance.CanReplicateFrom(master) + if !canReplicate { return instance, err } + if err != nil { + log.Warningf("Repoint(): %v", err) + err = nil + } + // if a binlog server check it is sufficiently up to date if master.IsBinlogServer() { // "Repoint" operation trusts the user. But only so much. Repoiting to a binlog server which is not yet there is strictly wrong. @@ -991,9 +1023,17 @@ func MakeCoMaster(instanceKey *InstanceKey) (*Instance, error) { } else if _, found, _ := ReadInstance(&master.MasterKey); found { return instance, fmt.Errorf("%+v is not a real master; it replicates from: %+v", master.Key, master.MasterKey) } - if canReplicate, err := master.CanReplicateFrom(instance); !canReplicate { + + canReplicate, err := master.CanReplicateFrom(instance) + if !canReplicate { return instance, err } + + if err != nil { + log.Warningf("MakeCoMaster(): %v", err) + err = nil + } + log.Infof("Will make %+v co-master of %+v", instanceKey, master.Key) var gitHint OperationGTIDHint = GTIDHintNeutral @@ -1584,9 +1624,16 @@ func MatchBelow(instanceKey, otherKey *InstanceKey, requireInstanceMaintenance b return instance, nil, merr } - if canReplicate, err := instance.CanReplicateFrom(otherInstance); !canReplicate { + canReplicate, err := instance.CanReplicateFrom(otherInstance) + if !canReplicate { return instance, nil, err } + + if err != nil { + log.Warningf("MatchBelow(): %v", err) + err = nil + } + var nextBinlogCoordinatesToMatch *BinlogCoordinates var countMatchedEvents int @@ -1791,9 +1838,16 @@ func TakeMaster(instanceKey *InstanceKey, allowTakingCoMaster bool) (*Instance, } log.Debugf("TakeMaster: will attempt making %+v take its master %+v, now resolved as %+v", *instanceKey, instance.MasterKey, masterInstance.Key) - if canReplicate, err := masterInstance.CanReplicateFrom(instance); canReplicate == false { + canReplicate, err := masterInstance.CanReplicateFrom(instance) + if !canReplicate { return instance, err } + + if err != nil { + log.Warningf("TakeMaster(): %v", err) + err = nil + } + // We begin masterInstance, err = StopReplication(&masterInstance.Key) if err != nil { @@ -2245,11 +2299,15 @@ func chooseCandidateReplica(replicas [](*Instance)) (candidateReplica *Instance, replicas = RemoveInstance(replicas, &candidateReplica.Key) for _, replica := range replicas { replica := replica - if canReplicate, err := replica.CanReplicateFrom(candidateReplica); !canReplicate { + + canReplicate, err := replica.CanReplicateFrom(candidateReplica) + + if !canReplicate { // lost due to inability to replicate cannotReplicateReplicas = append(cannotReplicateReplicas, replica) if err != nil { log.Errorf("chooseCandidateReplica(): error checking CanReplicateFrom(). replica: %v; error: %v", replica.Key, err) + err = nil // Zerro out the error so we don't log it again with warning } } else if replica.ExecBinlogCoordinates.SmallerThan(&candidateReplica.ExecBinlogCoordinates) { laterReplicas = append(laterReplicas, replica) @@ -2259,6 +2317,11 @@ func chooseCandidateReplica(replicas [](*Instance)) (candidateReplica *Instance, // lost due to being more advanced/ahead of chosen replica. aheadReplicas = append(aheadReplicas, replica) } + + if err != nil { + log.Warningf("chooseCandidateReplica(): replica: %v, warning: %+v", replica.Key, err) + err = nil + } } return candidateReplica, aheadReplicas, equalReplicas, laterReplicas, cannotReplicateReplicas, err } @@ -2658,9 +2721,16 @@ func RegroupReplicas(masterKey *InstanceKey, returnReplicaEvenOnFailureToRegroup // It may choose to use Pseudo-GTID, or normal binlog positions, or take advantage of binlog servers, // or it may combine any of the above in a multi-step operation. func relocateBelowInternal(instance, other *Instance) (*Instance, error) { - if canReplicate, err := instance.CanReplicateFrom(other); !canReplicate { + + canReplicate, err := instance.CanReplicateFrom(other) + if !canReplicate { return instance, log.Errorf("%+v cannot replicate from %+v. Reason: %+v", instance.Key, other.Key, err) } + + if err != nil { + log.Warningf("relocateBelowInternal(): %v", err) + } + // simplest: if InstanceIsMasterOf(other, instance) { // already the desired setup. diff --git a/go/inst/instance_utils.go b/go/inst/instance_utils.go index 5b514cc82..8094cba8d 100644 --- a/go/inst/instance_utils.go +++ b/go/inst/instance_utils.go @@ -221,7 +221,7 @@ func MajorVersion(version string) []string { // IsSmallerMajorVersion tests two versions against another and returns true if // the former is a smaller "major" varsion than the latter. -// e.g. 5.5.36 is NOT a smaller major version as comapred to 5.5.40, but IS as compared to 5.6.9 +// e.g. 5.5.36 is NOT a smaller major version as comapred to 5.5.40, but IS as compared to 6.0.0 func IsSmallerMajorVersion(version string, otherVersion string) bool { thisMajorVersion := MajorVersion(version) otherMajorVersion := MajorVersion(otherVersion) diff --git a/go/logic/topology_recovery.go b/go/logic/topology_recovery.go index adb281f64..676de65d8 100644 --- a/go/logic/topology_recovery.go +++ b/go/logic/topology_recovery.go @@ -1052,9 +1052,15 @@ func canTakeOverPromotedServerAsMaster(wantToTakeOver *inst.Instance, toBeTakenO if !wantToTakeOver.MasterKey.Equals(&toBeTakenOver.Key) { return false } - if canReplicate, _ := toBeTakenOver.CanReplicateFrom(wantToTakeOver); !canReplicate { + + canReplicate, err := toBeTakenOver.CanReplicateFrom(wantToTakeOver) + if !canReplicate { return false } + if err != nil { + log.Warningf("canTakeOverPromotedServerAsMaster(): %v", err) + } + return true }