diff --git a/daemon/libnetwork/agent.go b/daemon/libnetwork/agent.go index d7798f20ae..bf68a7a1a9 100644 --- a/daemon/libnetwork/agent.go +++ b/daemon/libnetwork/agent.go @@ -233,12 +233,7 @@ func (c *Controller) agentSetup(clusterProvider cluster.Provider) error { bindAddr := clusterProvider.GetLocalAddress() advAddr := clusterProvider.GetAdvertiseAddress() dataAddr := clusterProvider.GetDataPathAddress() - remoteList := clusterProvider.GetRemoteAddressList() - remoteAddrList := make([]string, 0, len(remoteList)) - for _, remote := range remoteList { - addr, _, _ := net.SplitHostPort(remote) - remoteAddrList = append(remoteAddrList, addr) - } + remoteAddrList := remoteAddrs(clusterProvider) listen := clusterProvider.GetListenAddress() listenAddr, _, _ := net.SplitHostPort(listen) @@ -330,6 +325,10 @@ func (c *Controller) agentInit(listenAddr, bindAddrOrInterface, advertiseAddr, d netDBConf.BindAddr = listenAddr netDBConf.AdvertiseAddr = advertiseAddr netDBConf.Keys = keys + // Answered afresh every time a rejoin is considered, so that a node which + // has to bootstrap its way back into the gossip cluster aims at the + // managers the cluster has now rather than the ones it had at startup. + netDBConf.BootstrapPeers = c.remoteAddrList if c.Config().NetworkControlPlaneMTU != 0 { // Consider the MTU remove the IP hdr (IPv4 or IPv6) and the TCP/UDP hdr. // To be on the safe side let's cut 100 bytes @@ -383,6 +382,43 @@ func (c *Controller) agentInit(listenAddr, bindAddrOrInterface, advertiseAddr, d return nil } +// remoteAddrList is [remoteAddrs] for whichever cluster provider is current, +// so that a NetworkDB holding on to it as a callback asks the cluster the node +// is in now rather than the one it was in when the agent came up. +// +// It reports nil once the node has left the cluster and the provider has gone, +// which NetworkDB reads as "no bootstrap nodes known" and skips its rejoin. +func (c *Controller) remoteAddrList() []string { + c.mu.Lock() + cp := c.cfg.ClusterProvider + c.mu.Unlock() + if cp == nil { + return nil + } + // Deliberately outside the lock above: this calls out into the cluster, + // which takes locks of its own. + return remoteAddrs(cp) +} + +// remoteAddrs is the addresses cp reports for the cluster's other managers, as +// NetworkDB wants them: bare IPs, with the port the cluster provider reports +// dropped, because a peer is assumed to gossip on the same port we do. +func remoteAddrs(cp cluster.Provider) []string { + remoteList := cp.GetRemoteAddressList() + remoteAddrList := make([]string, 0, len(remoteList)) + for _, remote := range remoteList { + addr, _, err := net.SplitHostPort(remote) + if err != nil { + // Nothing to drop: the provider named an address without a port. + // Pass it through rather than the empty string SplitHostPort + // leaves behind, which is not an address at all. + addr = remote + } + remoteAddrList = append(remoteAddrList, addr) + } + return remoteAddrList +} + func (c *Controller) agentJoin(remoteAddrList []string) error { agent := c.getAgent() if agent == nil { diff --git a/daemon/libnetwork/agent_test.go b/daemon/libnetwork/agent_test.go index 8015ee5075..4528c699f1 100644 --- a/daemon/libnetwork/agent_test.go +++ b/daemon/libnetwork/agent_test.go @@ -5,11 +5,15 @@ import ( "net" "net/netip" "slices" + "sync" "testing" "github.com/gogo/protobuf/proto" "gotest.tools/v3/assert" + is "gotest.tools/v3/assert/cmp" + "github.com/moby/moby/v2/daemon/libnetwork/cluster" + "github.com/moby/moby/v2/daemon/libnetwork/config" "github.com/moby/moby/v2/daemon/libnetwork/networkdb" ) @@ -414,3 +418,68 @@ func TestHandleEPTableEvent(t *testing.T) { }) } } + +// clusterProviderStub answers GetRemoteAddressList and nothing else: every +// other method of the interface is left to the nil embedded value, so a test +// which reaches one panics rather than quietly passing. +type clusterProviderStub struct { + cluster.Provider + mu sync.Mutex + remotes []string + calls int +} + +func (p *clusterProviderStub) GetRemoteAddressList() []string { + p.mu.Lock() + defer p.mu.Unlock() + p.calls++ + return slices.Clone(p.remotes) +} + +func TestRemoteAddrs(t *testing.T) { + for _, tc := range []struct { + name string + reported []string + want []string + }{ + {name: "none", reported: nil, want: []string{}}, + {name: "host and port", reported: []string{"10.0.0.1:2377"}, want: []string{"10.0.0.1"}}, + // The daemon's own providers report these, and SplitHostPort leaves + // the empty string behind for them, which is not an address at all. + {name: "no port", reported: []string{"10.0.0.1"}, want: []string{"10.0.0.1"}}, + {name: "ipv6", reported: []string{"[fd00::1]:2377"}, want: []string{"fd00::1"}}, + {name: "mixed", reported: []string{"10.0.0.1:2377", "10.0.0.2"}, want: []string{"10.0.0.1", "10.0.0.2"}}, + } { + t.Run(tc.name, func(t *testing.T) { + assert.Check(t, is.DeepEqual(remoteAddrs(&clusterProviderStub{remotes: tc.reported}), tc.want)) + }) + } +} + +// TestRemoteAddrListIsLastKnownGood pins the half of the bootstrap-peer +// contract which is libnetwork's: the list is asked for afresh on every call +// and passed on as the provider reported it, with nothing cached, dropped or +// second-guessed here. A node which has lost the Swarm control plane therefore +// keeps bootstrapping off the managers it last saw, which is what lets gossip +// carry on without one. +// +// Swarmkit holds up the other half: its manager list lives in memory and on +// disk, a failed connection only re-weights an entry, and the one path which +// removes one runs off a dispatcher session message -- so an outage freezes +// the list rather than draining it. +func TestRemoteAddrListIsLastKnownGood(t *testing.T) { + cp := &clusterProviderStub{remotes: []string{"10.0.0.1:2377", "10.0.0.2:2377"}} + c := &Controller{cfg: &config.Config{ClusterProvider: cp}} + + want := []string{"10.0.0.1", "10.0.0.2"} + for i := range 3 { + assert.Check(t, is.DeepEqual(c.remoteAddrList(), want), "call %d", i) + } + assert.Check(t, is.Equal(cp.calls, 3), "the peers must be asked for every time, not snapshotted") + + // Once the node leaves the cluster the provider goes with it, and NetworkDB + // is told there is nothing to bootstrap from rather than being handed a + // list which can no longer be answered for. + c.cfg.ClusterProvider = nil + assert.Check(t, is.Nil(c.remoteAddrList())) +} diff --git a/daemon/libnetwork/networkdb/cluster.go b/daemon/libnetwork/networkdb/cluster.go index 43d29f949a..040eb4f0ec 100644 --- a/daemon/libnetwork/networkdb/cluster.go +++ b/daemon/libnetwork/networkdb/cluster.go @@ -174,22 +174,25 @@ func (nDB *NetworkDB) clusterInit() error { return nil } +// retryJoin joins members, retrying every retryInterval until it succeeds or +// ctx is done. The first attempt is made straight away: waiting out a tick +// first would make the whole call a no-op whenever ctx expires within +// retryInterval, and it is the caller who chooses that budget. func (nDB *NetworkDB) retryJoin(ctx context.Context, members []string) { t := time.NewTicker(retryInterval) defer t.Stop() - for { + for ctx.Err() == nil { + if _, err := nDB.memberlist.Join(members); err != nil { + log.G(ctx).Errorf("Failed to join memberlist %s on retry: %v", members, err) + } else if err := nDB.sendNodeEvent(NodeEventTypeJoin); err != nil { + log.G(ctx).Errorf("failed to send node join on retry: %v", err) + } else { + return + } + select { case <-t.C: - if _, err := nDB.memberlist.Join(members); err != nil { - log.G(ctx).Errorf("Failed to join memberlist %s on retry: %v", members, err) - continue - } - if err := nDB.sendNodeEvent(NodeEventTypeJoin); err != nil { - log.G(ctx).Errorf("failed to send node join on retry: %v", err) - continue - } - return case <-ctx.Done(): return } @@ -200,9 +203,12 @@ func (nDB *NetworkDB) clusterJoin(members []string) error { mlist := nDB.memberlist if _, err := mlist.Join(members); err != nil { - // In case of failure, we no longer need to explicitly call retryJoin. - // rejoinClusterBootStrap, which runs every nDB.config.rejoinClusterInterval, - // will retryJoin for nDB.config.rejoinClusterDuration. + // No explicit retryJoin here: rejoinClusterBootStrap runs every + // nDB.config.rejoinClusterInterval and will retryJoin for + // nDB.config.rejoinClusterDuration. Note that it works from + // Config.BootstrapPeers, not from members, so it only picks this + // join up again if the caller also names these addresses there -- + // and a NetworkDB configured without BootstrapPeers never retries. return fmt.Errorf("could not join node to memberlist: %v", err) } @@ -273,56 +279,94 @@ func (nDB *NetworkDB) reapDeadNode() { } } -// rejoinClusterBootStrap is called periodically to check if all bootStrap nodes are active in the cluster, -// if not, call the cluster join to merge 2 separate clusters that are formed when all managers -// stopped/started at the same time -func (nDB *NetworkDB) rejoinClusterBootStrap() { +// bootstrapPeersToRejoin is the subset of peers which are not currently in the +// cluster, and so are worth rejoining. Ourselves is counted as in it: a +// bootstrap address which is our own is one we have no way to rejoin and no +// reason to, rather than an absent peer. +// +// An address which names no port is read as being on the port we gossip on, +// and not config.BindPort: that is only a request, left zero by every caller +// which is happy with memberlist's default. Addresses are compared unmapped, +// because memberlist knows its peers by whatever form their address arrived +// in, so a 4-in-6 address has to match the same address written as IPv4. +func (nDB *NetworkDB) bootstrapPeersToRejoin(peers []string) []string { nDB.RLock() - if len(nDB.bootStrapIP) == 0 { - nDB.RUnlock() - return - } + defer nDB.RUnlock() myself, ok := nDB.nodes[nDB.config.NodeID] if !ok { - nDB.RUnlock() log.G(context.TODO()).Warnf("rejoinClusterBootstrap unable to find local node info using ID:%v", nDB.config.NodeID) - return + return nil } - bootStrapIPs := make([]string, 0, len(nDB.bootStrapIP)) - for _, bootIP := range nDB.bootStrapIP { + gossipPort := myself.Port + + joined := make(map[netip.AddrPort]struct{}, len(nDB.nodes)) + for _, node := range nDB.nodes { + nodeIP, _ := netip.AddrFromSlice(node.Addr) + joined[netip.AddrPortFrom(nodeIP.Unmap(), node.Port)] = struct{}{} + } + + rejoin := make([]string, 0, len(peers)) + for _, bootIP := range peers { // bootstrap IPs are usually IP:port from the Join bootstrapIP, err := netip.ParseAddrPort(bootIP) if err != nil { // try to parse it as an IP without port // Note this seems to be the case for swarm that do not specify any port addr, err := netip.ParseAddr(bootIP) - if err == nil { - bootstrapIP = netip.AddrPortFrom(addr, uint16(nDB.config.BindPort)) + if err != nil { + continue } + // A peer is assumed to gossip on the same port we do. + bootstrapIP = netip.AddrPortFrom(addr, gossipPort) } - if bootstrapIP.IsValid() { - for _, node := range nDB.nodes { - if node == myself { - continue - } - nodeIP, _ := netip.AddrFromSlice(node.Addr) - if bootstrapIP == netip.AddrPortFrom(nodeIP.Unmap(), node.Port) { - // One of the bootstrap nodes (and not myself) is part of the cluster, return - nDB.RUnlock() - return - } - } - bootStrapIPs = append(bootStrapIPs, bootIP) + if _, ok := joined[netip.AddrPortFrom(bootstrapIP.Addr().Unmap(), bootstrapIP.Port())]; ok { + // This bootstrap node is already part of the cluster, so it + // needs no rejoining. The others still might. + continue } + rejoin = append(rejoin, bootIP) } - nDB.RUnlock() - if len(bootStrapIPs) == 0 { - // this will also avoid to call the Join with an empty list erasing the current bootstrap ip list - log.G(context.TODO()).Debug("rejoinClusterBootStrap did not find any valid IP") + return rejoin +} + +// rejoinClusterBootStrap is called periodically to check if all bootStrap nodes are active in the cluster, +// if not, call the cluster join to merge 2 separate clusters that are formed when all managers +// stopped/started at the same time +// +// Each bootstrap address is considered on its own: being attached to one of +// them says nothing about the others, and memberlist.Join reports an error only +// when every address it was given fails, so a rejoin which reached part of the +// list still returns success. Only the addresses which are not already peers +// are worth rejoining, and they stay worth rejoining for as long as they are +// absent. +// +// That matters most for a partition which outlives node reaping. reconnectNode +// retries the peers it saw fail, but reapDeadNode forgets them after +// nodeReapInterval, leaving this function as the only thing which can still +// merge the two halves. With a bootstrap node left on each side, each half is +// attached to one bootstrap address throughout, so it has to keep trying the +// ones it cannot see or the split outlives the network fault which caused it. +func (nDB *NetworkDB) rejoinClusterBootStrap() { + // Ask who the bootstrap nodes are before taking the lock: the answer comes + // from the caller, whose own locks must not be taken beneath ours. + if nDB.config.BootstrapPeers == nil { return } - // None of the bootStrap nodes are in the cluster, call memberlist join + peers := nDB.config.BootstrapPeers() + if len(peers) == 0 { + return + } + + bootStrapIPs := nDB.bootstrapPeersToRejoin(peers) + if len(bootStrapIPs) == 0 { + // Every bootstrap node is already a peer, or none of them parsed. + // memberlist.Join reports success on an empty list, so without this + // retryJoin would announce a join it never made. + log.G(context.TODO()).Debug("rejoinClusterBootStrap has no bootstrap node to rejoin") + return + } + // Some of the bootStrap nodes are not in the cluster, call memberlist join log.G(context.TODO()).Debugf("rejoinClusterBootStrap, calling cluster join with bootStrap %v", bootStrapIPs) ctx, cancel := context.WithTimeout(nDB.ctx, nDB.config.rejoinClusterDuration) defer cancel() diff --git a/daemon/libnetwork/networkdb/cluster_test.go b/daemon/libnetwork/networkdb/cluster_test.go index 0df489cdcd..3e2d291d1e 100644 --- a/daemon/libnetwork/networkdb/cluster_test.go +++ b/daemon/libnetwork/networkdb/cluster_test.go @@ -1,19 +1,25 @@ package networkdb import ( + "context" "iter" "maps" "math" "math/bits" "math/rand/v2" + "net" "slices" + "strconv" "strings" "testing" "testing/synctest" "time" + "github.com/hashicorp/memberlist" + "github.com/moby/moby/v2/daemon/internal/stringid" "gotest.tools/v3/assert" is "gotest.tools/v3/assert/cmp" + "gotest.tools/v3/poll" "pgregory.net/rapid" ) @@ -258,3 +264,90 @@ func distributionStats(vals iter.Seq[int]) (mean, stdev, minv, maxv float64) { stdev = math.Sqrt(sumSq/float64(n) - mean*mean) return mean, stdev, minv, maxv } + +// TestBootstrapPeersToRejoin covers how a bootstrap address is read before it +// is compared with the cluster's peers. Both readings are load-bearing and +// neither is exercised by the tests which drive a real cluster: those name +// their bootstrap nodes as ip:port and set BindPort explicitly, so the address +// arrives in the one form which needs no interpretation. +func TestBootstrapPeersToRejoin(t *testing.T) { + // A node gossiping on 7946 with config.BindPort left zero, which is what + // the daemon does: it never sets BindPort, so memberlist picks its own + // default and the port we gossip on is not the one in the config. + db := newNetworkDB(DefaultConfig()) + defer db.broadcaster.Close() + assert.Assert(t, is.Equal(db.config.BindPort, 0), "the point of the test is that BindPort is unset") + + db.config.NodeID = "self" + // net.ParseIP yields the 4-in-6 form, which is how memberlist holds an + // IPv4 peer address. + db.nodes["self"] = &node{Node: memberlist.Node{Name: "self", Addr: net.ParseIP("192.0.2.1"), Port: 7946}} + db.nodes["peer"] = &node{Node: memberlist.Node{Name: "peer", Addr: net.ParseIP("192.0.2.2"), Port: 7946}} + + for _, tc := range []struct { + name string + peers []string + want []string + }{ + {name: "none known", peers: nil, want: []string{}}, + {name: "peer by ip:port", peers: []string{"192.0.2.2:7946"}, want: []string{}}, + // Read on the port we gossip on. Read on config.BindPort instead and + // this is 192.0.2.2:0, which matches nothing and is rejoined forever. + {name: "peer without a port", peers: []string{"192.0.2.2"}, want: []string{}}, + // The peer side is unmapped before it is compared, so the bootstrap + // side has to be too, or these never match. + {name: "peer as 4-in-6 with port", peers: []string{"[::ffff:192.0.2.2]:7946"}, want: []string{}}, + {name: "peer as 4-in-6 without a port", peers: []string{"::ffff:192.0.2.2"}, want: []string{}}, + {name: "ourselves", peers: []string{"192.0.2.1"}, want: []string{}}, + {name: "unparseable", peers: []string{"not-an-address"}, want: []string{}}, + {name: "an absent bootstrap node", peers: []string{"192.0.2.9:7946"}, want: []string{"192.0.2.9:7946"}}, + {name: "a peer on another port", peers: []string{"192.0.2.2:9999"}, want: []string{"192.0.2.2:9999"}}, + // Being attached to one bootstrap node says nothing about the others. + { + name: "only the ones which are absent", + peers: []string{"192.0.2.2", "192.0.2.9:7946", "192.0.2.1"}, + want: []string{"192.0.2.9:7946"}, + }, + } { + t.Run(tc.name, func(t *testing.T) { + assert.Check(t, is.DeepEqual(db.bootstrapPeersToRejoin(tc.peers), tc.want)) + }) + } +} + +// TestRetryJoinAttemptsBeforeTicking pins that retryJoin makes its first +// attempt straight away. Waiting out a tick first makes the whole call a no-op +// for any caller whose budget is shorter than retryInterval, and no test which +// drives a real cluster would notice: they all allow a budget several ticks +// long. +func TestRetryJoinAttemptsBeforeTicking(t *testing.T) { + conf := DefaultConfig() + conf.BindAddr = "127.0.0.1" + conf.AdvertiseAddr = conf.BindAddr + + launch := func(name string) *NetworkDB { + t.Helper() + c := *conf + c.Hostname = name + c.NodeID = stringid.TruncateID(stringid.GenerateRandomID()) + c.BindPort = int(dbPort.Add(1)) + return launchNode(t, c) + } + // Deliberately not joined to each other: retryJoin is the only thing which + // puts them in touch. + a, b := launch("node1"), launch("node2") + defer closeNetworkDBInstances(t, []*NetworkDB{a, b}) + + ctx, cancel := context.WithTimeout(t.Context(), retryInterval/10) + defer cancel() + a.retryJoin(ctx, []string{net.JoinHostPort(b.config.AdvertiseAddr, strconv.Itoa(b.config.BindPort))}) + + poll.WaitOn(t, func(t poll.LogT) poll.Result { + a.RLock() + defer a.RUnlock() + if _, ok := a.nodes[b.config.NodeID]; ok { + return poll.Success() + } + return poll.Continue("waiting for the join made inside the budget to have taken") + }, poll.WithDelay(100*time.Millisecond), poll.WithTimeout(30*time.Second)) +} diff --git a/daemon/libnetwork/networkdb/networkdb.go b/daemon/libnetwork/networkdb/networkdb.go index a72cb6130d..edf8c8bcb8 100644 --- a/daemon/libnetwork/networkdb/networkdb.go +++ b/daemon/libnetwork/networkdb/networkdb.go @@ -107,10 +107,6 @@ type NetworkDB struct { // Reference to the memberlist's keyring to add & remove keys keyring *memberlist.Keyring - // bootStrapIP is the list of IPs that can be used to bootstrap - // the gossip. - bootStrapIP []string - // lastStatsTimestamp is the last timestamp when the stats got printed lastStatsTimestamp time.Time @@ -214,7 +210,9 @@ type Config struct { AdvertiseAddr string // BindPort is the local node's port to which we bind to for - // cluster communication. + // cluster communication. Zero leaves the choice to memberlist, so it is + // not the port this node ends up gossiping on: for that, ask memberlist + // what it advertises. BindPort int // Keys to be added to the Keyring of the memberlist. Key at index @@ -284,6 +282,21 @@ type Config struct { // Only set by tests. rngSeed *[32]byte + // BootstrapPeers reports the addresses this node should try to rejoin the + // cluster through, or nil if none are known. It is consulted every time a + // rejoin is considered, so a caller which can name the current peers keeps + // this node from bootstrapping off addresses which have since gone away. + // + // It is called from the rejoin timer with no NetworkDB lock held, so it + // must be safe for concurrent use, and it must not block for long. + // + // When nil this node never bootstraps its way back. Join still gets it + // into the cluster, and reconnectNode goes on retrying the peers it saw + // fail, so a node which falls out can still return by itself -- but once + // reapDeadNode has forgotten those peers there is nothing left to rejoin + // through. + BootstrapPeers func() []string + // StatsPrintPeriod the period to use to print queue stats // Default is 5min StatsPrintPeriod time.Duration @@ -383,10 +396,6 @@ func newNetworkDB(c *Config) *NetworkDB { // Join joins this NetworkDB instance with a list of peer NetworkDB // instances passed by the caller in the form of addr:port func (nDB *NetworkDB) Join(members []string) error { - nDB.Lock() - nDB.bootStrapIP = append([]string(nil), members...) - log.G(context.TODO()).Infof("The new bootstrap node list is:%v", nDB.bootStrapIP) - nDB.Unlock() return nDB.clusterJoin(members) } diff --git a/daemon/libnetwork/networkdb/networkdb_test.go b/daemon/libnetwork/networkdb/networkdb_test.go index ae77d8c116..7cd513e134 100644 --- a/daemon/libnetwork/networkdb/networkdb_test.go +++ b/daemon/libnetwork/networkdb/networkdb_test.go @@ -7,6 +7,7 @@ import ( "os" "strconv" "strings" + "sync" "sync/atomic" "testing" "text/tabwriter" @@ -91,6 +92,44 @@ func closeNetworkDBInstances(t TestingT, dbs []*NetworkDB) { } } +// addrOf is db's gossip address, in the ip:port form a bootstrap list takes. +func addrOf(db *NetworkDB) string { + return net.JoinHostPort(db.config.AdvertiseAddr, strconv.Itoa(db.config.BindPort)) +} + +// peerList answers [Config.BootstrapPeers] with something a test can change as +// it goes, so a node is only told where to rejoin once the test wants it to +// try. Its zero value answers "no bootstrap nodes known". +type peerList struct { + mu sync.Mutex + addrs []string +} + +// get is the [Config.BootstrapPeers] callback. It is called from the rejoin +// timer, so it copies rather than handing out the slice set below. +func (p *peerList) get() []string { + p.mu.Lock() + defer p.mu.Unlock() + return append([]string(nil), p.addrs...) +} + +func (p *peerList) set(addrs ...string) { + p.mu.Lock() + defer p.mu.Unlock() + p.addrs = addrs +} + +// waitFor polls cond until it holds, failing the test if it never does. +func waitFor(t TestingT, what string, cond func() bool) { + t.Helper() + poll.WaitOn(t, func(t poll.LogT) poll.Result { + if cond() { + return poll.Success() + } + return poll.Continue("waiting for %s", what) + }, poll.WithDelay(100*time.Millisecond), poll.WithTimeout(60*time.Second)) +} + func (nDB *NetworkDB) verifyNodeExistence(t *testing.T, node string, present bool) { t.Helper() for range 80 { @@ -210,6 +249,7 @@ func TestNetworkDBSimple(t *testing.T) { func TestNetworkDBJoinLeaveNetwork(t *testing.T) { dbs := createNetworkDBInstances(t, 2, "node", DefaultConfig()) + defer closeNetworkDBInstances(t, dbs) err := dbs[0].JoinNetwork("network1") assert.NilError(t, err) @@ -220,11 +260,11 @@ func TestNetworkDBJoinLeaveNetwork(t *testing.T) { assert.NilError(t, err) dbs[1].verifyNetworkExistence(t, dbs[0].config.NodeID, "network1", false) - closeNetworkDBInstances(t, dbs) } func TestNetworkDBJoinLeaveNetworks(t *testing.T) { dbs := createNetworkDBInstances(t, 2, "node", DefaultConfig()) + defer closeNetworkDBInstances(t, dbs) n := 10 for i := 1; i <= n; i++ { @@ -263,11 +303,11 @@ func TestNetworkDBJoinLeaveNetworks(t *testing.T) { dbs[0].verifyNetworkExistence(t, dbs[1].config.NodeID, fmt.Sprintf("network1%d", i), false) } - closeNetworkDBInstances(t, dbs) } func TestNetworkDBCRUDTableEntry(t *testing.T) { dbs := createNetworkDBInstances(t, 3, "node", DefaultConfig()) + defer closeNetworkDBInstances(t, dbs) err := dbs[0].JoinNetwork("network1") assert.NilError(t, err) @@ -293,7 +333,6 @@ func TestNetworkDBCRUDTableEntry(t *testing.T) { dbs[1].verifyEntryExistence(t, "test_table", "network1", "test_key", "", false) - closeNetworkDBInstances(t, dbs) } // TestCreateEntryOverTombstone checks that a key can be created again while @@ -377,6 +416,7 @@ func (nDB *NetworkDB) dumpTable(t *testing.T, tname string) { func TestNetworkDBCRUDTableEntries(t *testing.T) { dbs := createNetworkDBInstances(t, 2, "node", DefaultConfig()) + defer closeNetworkDBInstances(t, dbs) err := dbs[0].JoinNetwork("network1") assert.NilError(t, err) @@ -450,11 +490,15 @@ func TestNetworkDBCRUDTableEntries(t *testing.T) { dbs[n].dumpTable(t, "test_table") } - closeNetworkDBInstances(t, dbs) } func TestNetworkDBNodeLeave(t *testing.T) { dbs := createNetworkDBInstances(t, 2, "node", DefaultConfig()) + // Closing is what this test does rather than how it tidies up, so the + // teardown tracks what is still running: NetworkDB.Close is not + // idempotent, and a second call panics inside memberlist.Leave. + live := append([]*NetworkDB(nil), dbs...) + defer func() { closeNetworkDBInstances(t, live) }() err := dbs[0].JoinNetwork("network1") assert.NilError(t, err) @@ -468,12 +512,15 @@ func TestNetworkDBNodeLeave(t *testing.T) { dbs[1].verifyEntryExistence(t, "test_table", "network1", "test_key", "test_value", true) dbs[0].Close() + live = []*NetworkDB{dbs[1]} dbs[1].verifyEntryExistence(t, "test_table", "network1", "test_key", "test_value", false) dbs[1].Close() + live = nil } func TestNetworkDBWatch(t *testing.T) { dbs := createNetworkDBInstances(t, 2, "node", DefaultConfig()) + defer closeNetworkDBInstances(t, dbs) err := dbs[0].JoinNetwork("network1") assert.NilError(t, err) @@ -498,11 +545,11 @@ func TestNetworkDBWatch(t *testing.T) { testWatch(t, ch.C, "test_table", "network1", "test_key", "test_updated_value", "") cancel() - closeNetworkDBInstances(t, dbs) } func TestNetworkDBBulkSync(t *testing.T) { dbs := createNetworkDBInstances(t, 2, "node", DefaultConfig()) + defer closeNetworkDBInstances(t, dbs) err := dbs[0].JoinNetwork("network1") assert.NilError(t, err) @@ -529,7 +576,6 @@ func TestNetworkDBBulkSync(t *testing.T) { assert.NilError(t, err) } - closeNetworkDBInstances(t, dbs) } // TestBulkSyncWithFailedPeer checks that a bulk sync is still sent to a peer @@ -613,6 +659,7 @@ func TestNetworkDBCRUDMediumCluster(t *testing.T) { n := 5 dbs := createNetworkDBInstances(t, n, "node", DefaultConfig()) + defer closeNetworkDBInstances(t, dbs) // Shake out any data races. done := make(chan struct{}) @@ -678,11 +725,11 @@ func TestNetworkDBCRUDMediumCluster(t *testing.T) { assert.Check(t, is.Contains(err.Error(), "deleted and pending garbage collection"), err) } - closeNetworkDBInstances(t, dbs) } func TestNetworkDBNodeJoinLeaveIteration(t *testing.T) { dbs := createNetworkDBInstances(t, 2, "node", DefaultConfig()) + defer closeNetworkDBInstances(t, dbs) dbChangeWitness := func(nDB *NetworkDB) func(network string, expectNodeCount int) { staleNetworkTime := nDB.networkClock.Time() @@ -754,7 +801,6 @@ func TestNetworkDBNodeJoinLeaveIteration(t *testing.T) { dbs[1].verifyNetworkExistence(t, dbs[0].config.NodeID, "network1", true) witness1("network1", 2) - closeNetworkDBInstances(t, dbs) } func TestNetworkDBGarbageCollection(t *testing.T) { @@ -764,6 +810,7 @@ func TestNetworkDBGarbageCollection(t *testing.T) { config.StatsPrintPeriod = 15 * time.Second dbs := createNetworkDBInstances(t, 3, "node", config) + defer closeNetworkDBInstances(t, dbs) // 2 Nodes join network err := dbs[0].JoinNetwork("network1") @@ -818,7 +865,6 @@ func TestNetworkDBGarbageCollection(t *testing.T) { dbs[i].Unlock() } - closeNetworkDBInstances(t, dbs) } func checkNodeIsForgotten(t *testing.T, db *NetworkDB, nodeName string, msgAndArgs ...any) { @@ -920,6 +966,7 @@ func TestNodeReincarnation(t *testing.T) { func TestParallelCreate(t *testing.T) { dbs := createNetworkDBInstances(t, 1, "node", DefaultConfig()) + defer closeNetworkDBInstances(t, dbs) startCh := make(chan int) doneCh := make(chan error) @@ -944,11 +991,11 @@ func TestParallelCreate(t *testing.T) { // Only 1 write should have succeeded assert.Check(t, is.Equal(uint32(1), success.Load())) - closeNetworkDBInstances(t, dbs) } func TestParallelDelete(t *testing.T) { dbs := createNetworkDBInstances(t, 1, "node", DefaultConfig()) + defer closeNetworkDBInstances(t, dbs) err := dbs[0].CreateEntry("testTable", "testNetwork", "key", []byte("value")) assert.NilError(t, err) @@ -976,12 +1023,9 @@ func TestParallelDelete(t *testing.T) { // Only 1 write should have succeeded assert.Check(t, is.Equal(uint32(1), success.Load())) - closeNetworkDBInstances(t, dbs) } func TestNetworkDBIslands(t *testing.T) { - t.Skip("FIXME: flaky test; see https://github.com/moby/moby/issues/42459") - pollTimeout := func() time.Duration { const defaultTimeout = 120 * time.Second dl, ok := t.Deadline() @@ -994,23 +1038,36 @@ func TestNetworkDBIslands(t *testing.T) { return defaultTimeout } + var peers peerList + _ = log.SetLevel(log.DebugLevel) conf := DefaultConfig() // Shorten durations to speed up test execution. conf.rejoinClusterDuration = conf.rejoinClusterDuration / 10 conf.rejoinClusterInterval = conf.rejoinClusterInterval / 10 + // Answered empty until the bootstrap nodes are gone, so that nothing tries + // to rejoin anything while the cluster is still whole. + conf.BootstrapPeers = peers.get dbs := createNetworkDBInstances(t, 5, "node", conf) + // The first three leave part-way through and are relaunched, so the + // teardown tracks what is still running rather than closing the whole + // slice: NetworkDB.Close is not idempotent, and a second call panics + // inside memberlist.Leave. + live := append([]*NetworkDB(nil), dbs...) + defer func() { closeNetworkDBInstances(t, live) }() - // Get the node IP used currently - node := dbs[0].nodes[dbs[0].config.NodeID] - baseIPStr := node.Addr.String() + // Get the node IP used currently. The memberlist event delegate writes + // this map from its own goroutines while the cluster gossips, so it is + // not ours to read unlocked. + dbs[0].RLock() + baseIPStr := dbs[0].nodes[dbs[0].config.NodeID].Addr.String() + dbs[0].RUnlock() // Node 0,1,2 are going to be the 3 bootstrap nodes members := []string{ fmt.Sprintf("%s:%d", baseIPStr, dbs[0].config.BindPort), fmt.Sprintf("%s:%d", baseIPStr, dbs[1].config.BindPort), fmt.Sprintf("%s:%d", baseIPStr, dbs[2].config.BindPort), } - // Rejoining will update the list of the bootstrap members for i := 3; i < 5; i++ { t.Logf("Re-joining: %d", i) assert.Check(t, dbs[i].Join(members)) @@ -1025,6 +1082,7 @@ func TestNetworkDBIslands(t *testing.T) { departed[i] = dbs[i].config.NodeID dbs[i].Close() } + live = []*NetworkDB{dbs[3], dbs[4]} checkDBs := make(map[string]*NetworkDB) for i := 3; i < 5; i++ { @@ -1056,34 +1114,224 @@ func TestNetworkDBIslands(t *testing.T) { } poll.WaitOn(t, check, poll.WithDelay(time.Second), poll.WithTimeout(pollTimeout())) + // Only nodes 3 and 4 are left, and only they know where the bootstrap + // nodes live, so naming them now makes those two the only thing which can + // bring the cluster back together. + peers.set(members...) + // Spawn again the first 3 nodes with different names but same IP:port for i := range 3 { log.G(t.Context()).Infof("node %d coming back", i) conf := *dbs[i].config conf.NodeID = stringid.TruncateID(stringid.GenerateRandomID()) + // A returning node is given no bootstrap peers of its own: it does no + // join, exactly as the comment below says, so the reconvergence has to + // come from the two which stayed. + conf.BootstrapPeers = nil dbs[i] = launchNode(t, conf) + live = append(live, dbs[i]) } - // Give some time for the reconnect routine to run, it runs every 6s. + // Every node has to end up seeing all five of the current nodes as + // active, which is what reconverging means and what the relaunched three + // have no way to bring about for themselves: they did no join. + // + // Asserted by identity rather than by counting the node lists. A node + // which starts after another has departed can still be told of that + // departed incarnation by gossip already in flight, and it sits in + // failedNodes until something supersedes it at its address. Which nodes + // carry such a straggler is a matter of timing, so counting them tests + // the schedule rather than the convergence. + current := make([]string, 0, len(dbs)) + for _, db := range dbs { + current = append(current, db.config.NodeID) + } check = func(t poll.LogT) poll.Result { - // Verify that the cluster is again all connected. Note that the 3 previous node did not do any join. - // The two groups converge on the same state now: the survivors have - // nothing left to remember the departed incarnations by, so there is - // no longer an asymmetry between them and the nodes which came back. - for i := range 5 { - db := dbs[i] + for _, db := range dbs { db.RLock() - nNodes, nFailed := len(db.nodes), len(db.failedNodes) - db.RUnlock() - if nNodes != 5 { - return poll.Continue("%s:Waiting to connect to all nodes", dbs[i].config.Hostname) + var missing []string + for _, id := range current { + if _, ok := db.nodes[id]; !ok { + missing = append(missing, id) + } } - if nFailed != 0 { - return poll.Continue("%s:Waiting for 0 failedNodes", dbs[i].config.Hostname) + db.RUnlock() + if len(missing) > 0 { + return poll.Continue("%s: waiting to see %v active", db.config.Hostname, missing) } } return poll.Success() } poll.WaitOn(t, check, poll.WithDelay(time.Second), poll.WithTimeout(pollTimeout())) - closeNetworkDBInstances(t, dbs) +} + +// TestRejoinClusterBootStrapUsesCurrentPeers checks that a rejoin targets the +// peers the caller names now, not the ones it named when the node started. +// +// The bootstrap list was a snapshot taken once, at agent init, and nothing +// refreshed it. A swarm which replaces a manager at a new address therefore +// left every node holding an address which would never answer again, and no +// address for the manager which replaced it -- so the one path which can still +// merge a split cluster was aimed where the cluster used to be. +// +// Both bootstrap nodes are put beyond reach here and a replacement brought up +// somewhere the cluster has never seen, so gossip cannot carry the news: naming +// it in the live list is the only way the survivor can find it. +func TestRejoinClusterBootStrapUsesCurrentPeers(t *testing.T) { + var peers peerList + + conf := DefaultConfig() + // Shortened so the test is not sitting through the default minute-long + // rejoin interval. + conf.rejoinClusterInterval = 2 * time.Second + conf.rejoinClusterDuration = 3 * time.Second + conf.BootstrapPeers = peers.get + + dbs := createNetworkDBInstances(t, 3, "node", conf) + survivor := dbs[2] + + // Closed nodes are not put back here, so the teardown tracks what is still + // running rather than closing the whole slice: NetworkDB.Close is not + // idempotent, and a second call panics inside memberlist.Leave. + live := append([]*NetworkDB(nil), dbs...) + defer func() { closeNetworkDBInstances(t, live) }() + + peers.set(addrOf(dbs[0]), addrOf(dbs[1])) + + dbs[0].Close() + dbs[1].Close() + live = []*NetworkDB{survivor} + waitFor(t, "the survivor to be left on its own", func() bool { + survivor.RLock() + defer survivor.RUnlock() + return len(survivor.nodes) == 1 + }) + + // The replacement has no bootstrap list of its own -- no callback, and it + // joins nothing -- so it cannot reach out. Only the survivor can close the + // gap, and only if it is told where to look. + plain := DefaultConfig() + plain.rejoinClusterInterval = conf.rejoinClusterInterval + plain.rejoinClusterDuration = conf.rejoinClusterDuration + plain.Hostname = "replacement" + plain.NodeID = stringid.TruncateID(stringid.GenerateRandomID()) + plain.BindPort = int(dbPort.Add(1)) + plain.BindAddr = "127.0.0.1" + plain.AdvertiseAddr = plain.BindAddr + replacement := launchNode(t, *plain) + live = append(live, replacement) + + peers.set(addrOf(replacement)) + waitFor(t, "the survivor to rejoin through the replacement", func() bool { + survivor.RLock() + defer survivor.RUnlock() + _, ok := survivor.nodes[replacement.config.NodeID] + return ok + }) +} + +// TestRejoinClusterBootStrapStragglers checks that a bootstrap node which is +// already in the cluster does not suppress the rejoin of the ones which are +// not. +// +// rejoinClusterBootStrap used to abandon the whole bootstrap list as soon as +// any one address in it was found in the cluster, which is the wrong question +// to ask: being attached to one bootstrap node says nothing about the others. +// memberlist.Join compounds it, reporting failure only when every address +// fails, so even an attempted rejoin can leave a node attached to part of its +// list and satisfying the check from then on. +// +// That turns permanent once a partition outlives node reaping: reconnectNode +// retries the peers it saw fail, reapDeadNode forgets them after +// nodeReapInterval, and this function is all that is left to merge the halves +// -- latched, on both sides, on the bootstrap node each can still see. It is +// also the root cause of the TestNetworkDBIslands flakes, which hit it whenever +// a rejoin lands between two of that test's three relaunches. +// +// Here the two bootstrap nodes are brought back one at a time, with a wait in +// between for the survivor to reconnect to the first. That wait is what makes +// the fault deterministic: reaching it means a rejoin has already run and +// latched, so before the fix the second node could never be picked up. +func TestRejoinClusterBootStrapStragglers(t *testing.T) { + var peers peerList + + conf := DefaultConfig() + // Shortened so the test is not sitting through the default minute-long + // rejoin interval twice. + conf.rejoinClusterInterval = 2 * time.Second + conf.rejoinClusterDuration = 3 * time.Second + // Left empty while the cluster forms, so no node rejoins anything before + // the departures below give it something to do. + conf.BootstrapPeers = peers.get + + dbs := createNetworkDBInstances(t, 3, "node", conf) + survivor := dbs[2] + + // The two below are closed and only later replaced, so the teardown tracks + // what is still running rather than closing the whole slice: NetworkDB.Close + // is not idempotent, and a second call panics inside memberlist.Leave -- + // which would bury the failure that got us here. + live := append([]*NetworkDB(nil), dbs...) + defer func() { closeNetworkDBInstances(t, live) }() + + // Both of the others are bootstrap nodes, so that a rejoin has something + // to be selective about. + peers.set(addrOf(dbs[0]), addrOf(dbs[1])) + + sees := func(target *NetworkDB) bool { + survivor.RLock() + defer survivor.RUnlock() + _, ok := survivor.nodes[target.config.NodeID] + return ok + } + + departed := []string{dbs[0].config.NodeID, dbs[1].config.NodeID} + dbs[0].Close() + dbs[1].Close() + live = []*NetworkDB{survivor} + waitFor(t, "the survivor to be left on its own", func() bool { + survivor.RLock() + defer survivor.RUnlock() + return len(survivor.nodes) == 1 + }) + + // Both left cleanly, so nothing will try to reconnect them on the + // survivor's behalf. If memberlist had instead declared them failed, + // reconnectNode would pull them back and this test would pass whether or + // not the bug is present, so assert the precondition rather than assume + // it. + survivor.RLock() + for _, id := range departed { + _, failed := survivor.failedNodes[id] + assert.Check(t, !failed, "node %s should have left cleanly, not failed", id) + } + survivor.RUnlock() + + relaunch := func(i int) *NetworkDB { + t.Helper() + conf := *dbs[i].config + conf.NodeID = stringid.TruncateID(stringid.GenerateRandomID()) + // The returning node is given no way to reach out, so the survivor + // is the only thing that can close the gap. + conf.BootstrapPeers = nil + dbs[i] = launchNode(t, conf) + live = append(live, dbs[i]) + return dbs[i] + } + + // First one back, at the address it had before. Only the survivor can + // reattach it: the returning node has no bootstrap peers of its own. + // + // That one-sided arrangement is the point. It leaves the survivor in the + // position every node is in once a partition has outlived reapDeadNode -- + // the peers it lost are forgotten, reconnectNode has nothing left to + // retry, and the bootstrap list is the only way back -- without needing a + // day of waiting to get there. + first := relaunch(0) + waitFor(t, "the survivor to rejoin the first bootstrap node", func() bool { return sees(first) }) + + // Second one back, now that a rejoin has definitely run with the first + // already in the cluster. + second := relaunch(1) + waitFor(t, "the survivor to rejoin the second bootstrap node", func() bool { return sees(second) }) }