File: //proc/self/root/opt/go/pkg/mod/go.mongodb.org/
[email protected]/x/mongo/driver/topology/fsm.go
// Copyright (C) MongoDB, Inc. 2017-present.
//
// Licensed under the Apache License, Version 2.0 (the "License"); you may
// not use this file except in compliance with the License. You may obtain
// a copy of the License at http://www.apache.org/licenses/LICENSE-2.0
package topology
import (
"bytes"
"fmt"
"sync/atomic"
"go.mongodb.org/mongo-driver/bson/primitive"
"go.mongodb.org/mongo-driver/internal/ptrutil"
"go.mongodb.org/mongo-driver/mongo/address"
"go.mongodb.org/mongo-driver/mongo/description"
)
var (
// MinSupportedMongoDBVersion is the version string for the lowest MongoDB version supported by the driver.
MinSupportedMongoDBVersion = "3.6"
// SupportedWireVersions is the range of wire versions supported by the driver.
SupportedWireVersions = description.NewVersionRange(6, 21)
)
type fsm struct {
description.Topology
maxElectionID primitive.ObjectID
maxSetVersion uint32
compatible atomic.Value
compatibilityErr error
}
func newFSM() *fsm {
f := fsm{}
f.compatible.Store(true)
return &f
}
// selectFSMSessionTimeout selects the timeout to return for the topology's
// finite state machine. If the logicalSessionTimeoutMinutes on the FSM exists
// and the server is data-bearing, then we determine this value by returning
//
// min{server timeout, FSM timeout}
//
// where a "nil" value is considered less than 0.
//
// Otherwise, if the FSM's logicalSessionTimeoutMinutes exist, then this
// function returns the FSM timeout.
//
// In the case where the FSM timeout DNE, we check all servers to see if any
// still do not have a timeout. This function chooses the lowest of the existing
// timeouts.
func selectFSMSessionTimeout(f *fsm, s description.Server) *int64 {
oldMinutes := f.SessionTimeoutMinutesPtr
comp := ptrutil.CompareInt64(oldMinutes, s.SessionTimeoutMinutesPtr)
// If the server is data-bearing and the current timeout exists and is
// either:
//
// 1. larger than the server timeout, or
// 2. non-nil while the server timeout is nil
//
// then return the server timeout.
if s.DataBearing() && (comp == 1 || comp == 2) {
return s.SessionTimeoutMinutesPtr
}
// If the current timeout exists and the server is not data-bearing OR
// min{server timeout, current timeout} = current timeout, then return
// the current timeout.
if oldMinutes != nil {
return oldMinutes
}
timeout := s.SessionTimeoutMinutesPtr
for _, server := range f.Servers {
// If the server is not data-bearing, then we do not consider
// it's timeout whether set or not.
if !server.DataBearing() {
continue
}
srvTimeout := server.SessionTimeoutMinutesPtr
comp := ptrutil.CompareInt64(timeout, srvTimeout)
if comp <= 0 { // timeout <= srvTimout
continue
}
timeout = server.SessionTimeoutMinutesPtr
}
return timeout
}
// apply takes a new server description and modifies the FSM's topology description based on it. It returns the
// updated topology description as well as a server description. The returned server description is either the same
// one that was passed in, or a new one in the case that it had to be changed.
//
// apply should operation on immutable descriptions so we don't have to lock for the entire time we're applying the
// server description.
func (f *fsm) apply(s description.Server) (description.Topology, description.Server) {
newServers := make([]description.Server, len(f.Servers))
copy(newServers, f.Servers)
// Reset the logicalSessionTimeoutMinutes to the minimum of the FSM
// and the description.server/f.servers.
serverTimeoutMinutes := selectFSMSessionTimeout(f, s)
f.Topology = description.Topology{
Kind: f.Kind,
Servers: newServers,
SetName: f.SetName,
}
f.Topology.SessionTimeoutMinutesPtr = serverTimeoutMinutes
if serverTimeoutMinutes != nil {
f.SessionTimeoutMinutes = uint32(*serverTimeoutMinutes)
}
if _, ok := f.findServer(s.Addr); !ok {
return f.Topology, s
}
updatedDesc := s
switch f.Kind {
case description.Unknown:
updatedDesc = f.applyToUnknown(s)
case description.Sharded:
updatedDesc = f.applyToSharded(s)
case description.ReplicaSetNoPrimary:
updatedDesc = f.applyToReplicaSetNoPrimary(s)
case description.ReplicaSetWithPrimary:
updatedDesc = f.applyToReplicaSetWithPrimary(s)
case description.Single:
updatedDesc = f.applyToSingle(s)
}
for _, server := range f.Servers {
if server.WireVersion != nil {
if server.WireVersion.Max < SupportedWireVersions.Min {
f.compatible.Store(false)
f.compatibilityErr = fmt.Errorf(
"server at %s reports wire version %d, but this version of the Go driver requires "+
"at least %d (MongoDB %s)",
server.Addr.String(),
server.WireVersion.Max,
SupportedWireVersions.Min,
MinSupportedMongoDBVersion,
)
f.Topology.CompatibilityErr = f.compatibilityErr
return f.Topology, s
}
if server.WireVersion.Min > SupportedWireVersions.Max {
f.compatible.Store(false)
f.compatibilityErr = fmt.Errorf(
"server at %s requires wire version %d, but this version of the Go driver only supports up to %d",
server.Addr.String(),
server.WireVersion.Min,
SupportedWireVersions.Max,
)
f.Topology.CompatibilityErr = f.compatibilityErr
return f.Topology, s
}
}
}
f.compatible.Store(true)
f.compatibilityErr = nil
return f.Topology, updatedDesc
}
func (f *fsm) applyToReplicaSetNoPrimary(s description.Server) description.Server {
switch s.Kind {
case description.Standalone, description.Mongos:
f.removeServerByAddr(s.Addr)
case description.RSPrimary:
f.updateRSFromPrimary(s)
case description.RSSecondary, description.RSArbiter, description.RSMember:
f.updateRSWithoutPrimary(s)
case description.Unknown, description.RSGhost:
f.replaceServer(s)
}
return s
}
func (f *fsm) applyToReplicaSetWithPrimary(s description.Server) description.Server {
switch s.Kind {
case description.Standalone, description.Mongos:
f.removeServerByAddr(s.Addr)
f.checkIfHasPrimary()
case description.RSPrimary:
f.updateRSFromPrimary(s)
case description.RSSecondary, description.RSArbiter, description.RSMember:
f.updateRSWithPrimaryFromMember(s)
case description.Unknown, description.RSGhost:
f.replaceServer(s)
f.checkIfHasPrimary()
}
return s
}
func (f *fsm) applyToSharded(s description.Server) description.Server {
switch s.Kind {
case description.Mongos, description.Unknown:
f.replaceServer(s)
case description.Standalone, description.RSPrimary, description.RSSecondary, description.RSArbiter, description.RSMember, description.RSGhost:
f.removeServerByAddr(s.Addr)
}
return s
}
func (f *fsm) applyToSingle(s description.Server) description.Server {
switch s.Kind {
case description.Unknown:
f.replaceServer(s)
case description.Standalone, description.Mongos:
if f.SetName != "" {
f.removeServerByAddr(s.Addr)
return s
}
f.replaceServer(s)
case description.RSPrimary, description.RSSecondary, description.RSArbiter, description.RSMember, description.RSGhost:
// A replica set name can be provided when creating a direct connection. In this case, if the set name returned
// by the hello response doesn't match up with the one provided during configuration, the server description
// is replaced with a default Unknown description.
//
// We create a new server description rather than doing s.Kind = description.Unknown because the other fields,
// such as RTT, need to be cleared for Unknown descriptions as well.
if f.SetName != "" && f.SetName != s.SetName {
s = description.Server{
Addr: s.Addr,
Kind: description.Unknown,
}
}
f.replaceServer(s)
}
return s
}
func (f *fsm) applyToUnknown(s description.Server) description.Server {
switch s.Kind {
case description.Mongos:
f.setKind(description.Sharded)
f.replaceServer(s)
case description.RSPrimary:
f.updateRSFromPrimary(s)
case description.RSSecondary, description.RSArbiter, description.RSMember:
f.setKind(description.ReplicaSetNoPrimary)
f.updateRSWithoutPrimary(s)
case description.Standalone:
f.updateUnknownWithStandalone(s)
case description.Unknown, description.RSGhost:
f.replaceServer(s)
}
return s
}
func (f *fsm) checkIfHasPrimary() {
if _, ok := f.findPrimary(); ok {
f.setKind(description.ReplicaSetWithPrimary)
} else {
f.setKind(description.ReplicaSetNoPrimary)
}
}
// hasStalePrimary returns true if the topology has a primary that is "stale".
func hasStalePrimary(fsm fsm, srv description.Server) bool {
// Compare the election ID values of the server and the topology lexicographically.
compRes := bytes.Compare(srv.ElectionID[:], fsm.maxElectionID[:])
if wireVersion := srv.WireVersion; wireVersion != nil && wireVersion.Max >= 17 {
// In the Post-6.0 case, a primary is considered "stale" if the server's election ID is greater than the
// topology's max election ID. In these versions, the primary is also considered "stale" if the server's
// election ID is LTE to the topologies election ID and the server's "setVersion" is less than the topology's
// max "setVersion".
return compRes == -1 || (compRes != 1 && srv.SetVersion < fsm.maxSetVersion)
}
// If the server's election ID is less than the topology's max election ID, the primary is considered
// "stale". Similarly, if the server's "setVersion" is less than the topology's max "setVersion", the
// primary is considered stale.
return compRes == -1 || fsm.maxSetVersion > srv.SetVersion
}
// transferEVTuple will transfer the ("ElectionID", "SetVersion") tuple from the description server to the topology.
// If the primary is stale, the tuple will not be transferred, the topology will update it's "Kind" value, and this
// routine will return "false".
func transferEVTuple(srv description.Server, fsm *fsm) bool {
stalePrimary := hasStalePrimary(*fsm, srv)
if wireVersion := srv.WireVersion; wireVersion != nil && wireVersion.Max >= 17 {
if stalePrimary {
fsm.checkIfHasPrimary()
return false
}
fsm.maxElectionID = srv.ElectionID
fsm.maxSetVersion = srv.SetVersion
return true
}
if srv.SetVersion != 0 && !srv.ElectionID.IsZero() {
if stalePrimary {
fsm.replaceServer(description.Server{
Addr: srv.Addr,
LastError: fmt.Errorf(
"was a primary, but its set version or election id is stale"),
})
fsm.checkIfHasPrimary()
return false
}
fsm.maxElectionID = srv.ElectionID
}
if srv.SetVersion > fsm.maxSetVersion {
fsm.maxSetVersion = srv.SetVersion
}
return true
}
func (f *fsm) updateRSFromPrimary(srv description.Server) {
if f.SetName == "" {
f.SetName = srv.SetName
} else if f.SetName != srv.SetName {
f.removeServerByAddr(srv.Addr)
f.checkIfHasPrimary()
return
}
if ok := transferEVTuple(srv, f); !ok {
return
}
if j, ok := f.findPrimary(); ok {
f.setServer(j, description.Server{
Addr: f.Servers[j].Addr,
LastError: fmt.Errorf("was a primary, but a new primary was discovered"),
})
}
f.replaceServer(srv)
for j := len(f.Servers) - 1; j >= 0; j-- {
found := false
for _, member := range srv.Members {
if member == f.Servers[j].Addr {
found = true
break
}
}
if !found {
f.removeServer(j)
}
}
for _, member := range srv.Members {
if _, ok := f.findServer(member); !ok {
f.addServer(member)
}
}
f.checkIfHasPrimary()
}
func (f *fsm) updateRSWithPrimaryFromMember(s description.Server) {
if f.SetName != s.SetName {
f.removeServerByAddr(s.Addr)
f.checkIfHasPrimary()
return
}
if s.Addr != s.CanonicalAddr {
f.removeServerByAddr(s.Addr)
f.checkIfHasPrimary()
return
}
f.replaceServer(s)
if _, ok := f.findPrimary(); !ok {
f.setKind(description.ReplicaSetNoPrimary)
}
}
func (f *fsm) updateRSWithoutPrimary(s description.Server) {
if f.SetName == "" {
f.SetName = s.SetName
} else if f.SetName != s.SetName {
f.removeServerByAddr(s.Addr)
return
}
for _, member := range s.Members {
if _, ok := f.findServer(member); !ok {
f.addServer(member)
}
}
if s.Addr != s.CanonicalAddr {
f.removeServerByAddr(s.Addr)
return
}
f.replaceServer(s)
}
func (f *fsm) updateUnknownWithStandalone(s description.Server) {
if len(f.Servers) > 1 {
f.removeServerByAddr(s.Addr)
return
}
f.setKind(description.Single)
f.replaceServer(s)
}
func (f *fsm) addServer(addr address.Address) {
f.Servers = append(f.Servers, description.Server{
Addr: addr.Canonicalize(),
})
}
func (f *fsm) findPrimary() (int, bool) {
for i, s := range f.Servers {
if s.Kind == description.RSPrimary {
return i, true
}
}
return 0, false
}
func (f *fsm) findServer(addr address.Address) (int, bool) {
canon := addr.Canonicalize()
for i, s := range f.Servers {
if canon == s.Addr {
return i, true
}
}
return 0, false
}
func (f *fsm) removeServer(i int) {
f.Servers = append(f.Servers[:i], f.Servers[i+1:]...)
}
func (f *fsm) removeServerByAddr(addr address.Address) {
if i, ok := f.findServer(addr); ok {
f.removeServer(i)
}
}
func (f *fsm) replaceServer(s description.Server) {
if i, ok := f.findServer(s.Addr); ok {
f.setServer(i, s)
}
}
func (f *fsm) setServer(i int, s description.Server) {
f.Servers[i] = s
}
func (f *fsm) setKind(k description.TopologyKind) {
f.Kind = k
}