moby/daemon/libnetwork/networkdb/networkdb_property_test.go

//go:build slowtests

package networkdb

import (
	"fmt"
	"maps"
	"os"
	"slices"
	"strconv"
	"strings"
	"testing"
	"time"

	"github.com/google/go-cmp/cmp"
	"gotest.tools/v3/poll"
	"pgregory.net/rapid"
)

func TestNetworkDBAlwaysConverges(t *testing.T) {
	rapid.Check(t, testConvergence)
}

func testConvergence(t *rapid.T) {
	numNodes := rapid.IntRange(2, 25).Draw(t, "numNodes")
	numNetworks := rapid.IntRange(1, 5).Draw(t, "numNetworks")

	fsm := &networkDBFSM{
		nDB:      createNetworkDBInstances(t, numNodes, "node", DefaultConfig()),
		state:    make([]map[string]map[string]string, numNodes),
		keysUsed: make(map[string]map[string]bool),
	}
	defer closeNetworkDBInstances(t, fsm.nDB)
	for i := range fsm.state {
		fsm.state[i] = make(map[string]map[string]string)
	}
	for i := range numNetworks {
		nw := "nw" + strconv.Itoa(i)
		fsm.networks = append(fsm.networks, nw)
		fsm.keysUsed[nw] = make(map[string]bool)
	}
	// Drive the NetworkDB instances with a sequence of actions in random order.
	// We do not check for convergence until afterwards as NetworkDB is an
	// eventually consistent system.
	t.Repeat(rapid.StateMachineActions(fsm))

	// Take the union of all entries in all networks owned by all nodes.
	converged := make(map[string]map[string]string)
	for _, state := range fsm.state {
		for network, entries := range state {
			if converged[network] == nil {
				converged[network] = make(map[string]string)
			}
			maps.Copy(converged[network], entries)
		}
	}
	expected := make(map[string]map[string]map[string]string, numNodes)
	for i, st := range fsm.state {
		exp := make(map[string]map[string]string)
		for k := range st {
			exp[k] = converged[k]
		}
		expected[fsm.nDB[i].config.NodeID] = exp
	}

	t.Logf("Waiting for NetworkDB state to converge to %#v", converged)
	for i, st := range fsm.state {
		t.Logf("Node #%d (%s): %v", i, fsm.nDB[i].config.NodeID, slices.Collect(maps.Keys(st)))
	}
	t.Log("Mutations:")
	for _, m := range fsm.mutations {
		t.Log(m)
	}
	t.Log("---------------------------")

	poll.WaitOn(t, func(t poll.LogT) poll.Result {
		actualState := make(map[string]map[string]map[string]string, numNodes)
		for _, nDB := range fsm.nDB {
			actual := make(map[string]map[string]string)
			for k, nw := range nDB.thisNodeNetworks {
				if !nw.leaving {
					actual[k] = make(map[string]string)
				}
			}
			actualState[nDB.config.NodeID] = actual
		}
		tableContent := make([]string, len(fsm.nDB))
		for i, nDB := range fsm.nDB {
			tableContent[i] = fmt.Sprintf("Node #%d (%s):\n%v", i, nDB.config.NodeID, nDB.DebugDumpTable("some_table"))
			nDB.WalkTable("some_table", func(network, key string, value []byte, deleting bool) bool {
				if deleting {
					return false
				}
				if actualState[nDB.config.NodeID][network] == nil {
					actualState[nDB.config.NodeID][network] = make(map[string]string)
				}
				actualState[nDB.config.NodeID][network][key] = string(value)
				return false
			})
		}
		diff := cmp.Diff(expected, actualState)
		if diff != "" {
			return poll.Continue("NetworkDB state has not converged:\n%v\n%v", diff, strings.Join(tableContent, "\n\n"))
		}
		return poll.Success()
	}, poll.WithTimeout(5*time.Minute), poll.WithDelay(200*time.Millisecond))

	convergenceTime := time.Since(fsm.lastMutation)
	t.Logf("NetworkDB state converged in %v", convergenceTime)

	// Log the convergence time to disk for later statistical analysis.

	if err := os.Mkdir("testdata", 0o755); err != nil && !os.IsExist(err) {
		t.Logf("Could not log convergence time to disk: failed to create testdata directory: %v", err)
		return
	}
	f, err := os.OpenFile("testdata/convergence_time.csv", os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o644)
	if err != nil {
		t.Logf("Could not log convergence time to disk: failed to open file: %v", err)
		return
	}
	defer func() {
		if err := f.Close(); err != nil {
			t.Logf("Could not log convergence time to disk: error closing file: %v", err)
		}
	}()
	if st, err := f.Stat(); err != nil {
		t.Logf("Could not log convergence time to disk: failed to stat file: %v", err)
		return
	} else if st.Size() == 0 {
		f.WriteString("Nodes,Networks,#Mutations,Convergence(ns)\n")
	}
	if _, err := fmt.Fprintf(f, "%v,%v,%v,%d\n", numNodes, numNetworks, len(fsm.mutations), convergenceTime); err != nil {
		t.Logf("Could not log convergence time to disk: failed to write to file: %v", err)
		return
	}
}

// networkDBFSM is a [rapid.StateMachine] providing the set of actions available
// for rapid to drive NetworkDB with in tests. See also
// [rapid.StateMachineActions] and [rapid.Repeat].
type networkDBFSM struct {
	nDB      []*NetworkDB
	networks []string // list of networks which can be joined
	// node -> joined-network -> key -> value
	state []map[string]map[string]string

	// Remember entry keys that have been used before to avoid trying to
	// create colliding keys. Due to how quickly the FSM runs, it is
	// possible for a node to not have learned that the previous generation
	// of the key was deleted before we try to create it again.
	// network -> key -> true
	keysUsed map[string]map[string]bool

	// Timestamp of the most recent state-machine action which perturbed the
	// system state.
	lastMutation time.Time
	mutations    []string
}

func (u *networkDBFSM) mutated(nodeidx int, action, network, key, value string) {
	u.lastMutation = time.Now()
	desc := fmt.Sprintf("  [%v] #%d(%v):%v(%s", u.lastMutation, nodeidx, u.nDB[nodeidx].config.NodeID, action, network)
	if key != "" {
		desc += fmt.Sprintf(", %s=%s", key, value)
	}
	desc += ")"
	u.mutations = append(u.mutations, desc)
}

func (u *networkDBFSM) Check(t *rapid.T) {
	// This method is required to implement the [rapid.StateMachine]
	// interface. But there is nothing much to check stepwise as we are
	// testing an eventually consistent system. The checks happen after
	// rapid is done randomly driving the FSM.
}

func (u *networkDBFSM) JoinNetwork(t *rapid.T) {
	// Pick a node that has not joined all networks...
	var nodes []int
	for i, s := range u.state {
		if len(s) < len(u.networks) {
			nodes = append(nodes, i)
		}
	}
	if len(nodes) == 0 {
		t.Skip("All nodes are already joined to all networks")
	}
	nodeidx := rapid.SampledFrom(nodes).Draw(t, "node")

	// ... and a network to join.
	networks := slices.DeleteFunc(slices.Clone(u.networks), func(n string) bool {
		_, ok := u.state[nodeidx][n]
		return ok
	})
	nw := rapid.SampledFrom(networks).Draw(t, "network")

	if err := u.nDB[nodeidx].JoinNetwork(nw); err != nil {
		t.Errorf("Node %v failed to join network %s: %v", nodeidx, nw, err)
	} else {
		u.state[nodeidx][nw] = make(map[string]string)
		u.mutated(nodeidx, "JoinNetwork", nw, "", "")
	}
}

// drawJoinedNode returns a random node that has joined at least one network.
func (u *networkDBFSM) drawJoinedNodeAndNetwork(t *rapid.T) (nodeidx int, nw string) {
	var nodes []int
	for i, s := range u.state {
		if len(s) > 0 {
			nodes = append(nodes, i)
		}
	}
	if len(nodes) == 0 {
		t.Skip("No node is joined to any network")
	}
	nodeidx = rapid.SampledFrom(nodes).Draw(t, "node")

	nw = rapid.SampledFrom(slices.Collect(maps.Keys(u.state[nodeidx]))).Draw(t, "network")
	return nodeidx, nw
}

func (u *networkDBFSM) LeaveNetwork(t *rapid.T) {
	nodeidx, nw := u.drawJoinedNodeAndNetwork(t)
	if err := u.nDB[nodeidx].LeaveNetwork(nw); err != nil {
		t.Errorf("Node %v failed to leave network %s: %v", nodeidx, nw, err)
	} else {
		delete(u.state[nodeidx], nw)
		u.mutated(nodeidx, "LeaveNetwork", nw, "", "")
	}
}

func (u *networkDBFSM) CreateEntry(t *rapid.T) {
	nodeidx, nw := u.drawJoinedNodeAndNetwork(t)
	key := rapid.StringMatching(`[a-z]{3,25}`).
		Filter(func(s string) bool { return !u.keysUsed[nw][s] }).
		Draw(t, "key")
	value := rapid.StringMatching(`[a-z]{5,20}`).Draw(t, "value")

	if err := u.nDB[nodeidx].CreateEntry("some_table", nw, key, []byte(value)); err != nil {
		t.Errorf("Node %v failed to create entry %s=%s in network %s: %v", nodeidx, key, value, nw, err)
	} else {
		u.state[nodeidx][nw][key] = value
		u.keysUsed[nw][key] = true
		u.mutated(nodeidx, "CreateEntry", nw, key, value)
	}
}

// drawOwnedDBKey returns a random key in nw owned by the node at nodeidx.
func (u *networkDBFSM) drawOwnedDBKey(t *rapid.T, nodeidx int, nw string) string {
	keys := slices.Collect(maps.Keys(u.state[nodeidx][nw]))
	if len(keys) == 0 {
		t.Skipf("Node %v owns no entries in network %s", nodeidx, nw)
		panic("unreachable")
	}
	return rapid.SampledFrom(keys).Draw(t, "key")
}

func (u *networkDBFSM) UpdateEntry(t *rapid.T) {
	nodeidx, nw := u.drawJoinedNodeAndNetwork(t)
	key := u.drawOwnedDBKey(t, nodeidx, nw)
	value := rapid.StringMatching(`[a-z]{5,20}`).Draw(t, "value")

	if err := u.nDB[nodeidx].UpdateEntry("some_table", nw, key, []byte(value)); err != nil {
		t.Errorf("Node %v failed to update entry %s=%s in network %s: %v", nodeidx, key, value, nw, err)
	} else {
		u.state[nodeidx][nw][key] = value
		u.mutated(nodeidx, "UpdateEntry", nw, key, value)
	}
}

func (u *networkDBFSM) DeleteEntry(t *rapid.T) {
	nodeidx, nw := u.drawJoinedNodeAndNetwork(t)
	key := u.drawOwnedDBKey(t, nodeidx, nw)

	if err := u.nDB[nodeidx].DeleteEntry("some_table", nw, key); err != nil {
		t.Errorf("Node %v failed to delete entry %s in network %s: %v", nodeidx, key, nw, err)
	} else {
		delete(u.state[nodeidx][nw], key)
		u.mutated(nodeidx, "DeleteEntry", nw, key, "")
	}
}

func (u *networkDBFSM) Sleep(t *rapid.T) {
	duration := time.Duration(rapid.IntRange(10, 500).Draw(t, "duration")) * time.Millisecond
	time.Sleep(duration)
}