libn/networkdb: b'cast watch events from local POV

NetworkDB gossips changes to table entries to other nodes using distinct
CREATE, UPDATE and DELETE events. It is unfortunate that the wire
protocol distinguishes CREATEs from UPDATEs as nothing useful can be
done with this information. Newer events for an entry invalidate older
ones, so there is no guarantee that a CREATE event is broadcast to any
node before an UPDATE is broadcast. And due to the nature of gossip
protocols, even if the CREATE event is broadcast from the originating
node, there is no guarantee that any particular node will receive the
CREATE before an UPDATE. Any code which handles an UPDATE event
differently from a CREATE event is therefore going to behave in
unexpected ways in less than perfect conditions.

NetworkDB table watchers also receive CREATE, UPDATE and DELETE events.
Since the watched tables are local to the node, the events could all
have well-defined meanings that are actually useful. Unfortunately
NetworkDB is just bubbling up the wire-protocol event types to the
watchers. Redefine the table-watch events such that a CREATE event is
broadcast when an entry pops into existence in the local NetworkDB, an
UPDATE event is broadcast when an entry which was already present in the
NetworkDB state is modified, and a DELETE event is broadcast when an
entry which was already present in the NetworkDB state is marked for
deletion. DELETE events are broadcast with the same value as the most
recent CREATE or UPDATE event for the entry.

The handler for endpoint table events in the libnetwork agent assumed
incorrectly that CREATE events always correspond to adding a new active
endpoint and that UPDATE events always correspond to disabling an
endpoint. Fix up the handler to handle CREATE and UPDATE events using
the same code path, checking the table entry's ServiceDisabled flag to
determine which action to take.

Signed-off-by: Cory Snider <csnider@mirantis.com>

libnetwork/agent.go

@@ -911,7 +911,7 @@ func (c *Controller) handleEpTableEvent(ev events.Event) {
 
 	err := proto.Unmarshal(value, &epRec)
 	if err != nil {
-		log.G(context.TODO()).Errorf("Failed to unmarshal service table value: %v", err)
+		log.G(context.TODO()).WithError(err).Error("Failed to unmarshal service table value")
 		return
 	}
 
@@ -924,53 +924,54 @@ func (c *Controller) handleEpTableEvent(ev events.Event) {
 	serviceAliases := epRec.Aliases
 	taskAliases := epRec.TaskAliases
 
+	logger := log.G(context.TODO()).WithFields(log.Fields{
+		"nid": nid,
+		"eid": eid,
+		"T":   fmt.Sprintf("%T", ev),
+		"R":   epRec,
+	})
+
 	if containerName == "" || ip == nil {
-		log.G(context.TODO()).Errorf("Invalid endpoint name/ip received while handling service table event %s", value)
+		logger.Errorf("Invalid endpoint name/ip received while handling service table event %s", value)
 		return
 	}
 
+	logger.Debug("handleEpTableEvent")
+
 	switch ev.(type) {
-	case networkdb.CreateEvent:
-		log.G(context.TODO()).Debugf("handleEpTableEvent ADD %s R:%v", eid, epRec)
+	case networkdb.CreateEvent, networkdb.UpdateEvent:
 		if svcID != "" {
 			// This is a remote task part of a service
-			if err := c.addServiceBinding(svcName, svcID, nid, eid, containerName, vip, ingressPorts, serviceAliases, taskAliases, ip, "handleEpTableEvent"); err != nil {
-				log.G(context.TODO()).Errorf("failed adding service binding for %s epRec:%v err:%v", eid, epRec, err)
-				return
+			if epRec.ServiceDisabled {
+				if err := c.rmServiceBinding(svcName, svcID, nid, eid, containerName, vip, ingressPorts, serviceAliases, taskAliases, ip, "handleEpTableEvent", true, false); err != nil {
+					logger.WithError(err).Error("failed disabling service binding")
+					return
+				}
+			} else {
+				if err := c.addServiceBinding(svcName, svcID, nid, eid, containerName, vip, ingressPorts, serviceAliases, taskAliases, ip, "handleEpTableEvent"); err != nil {
+					logger.WithError(err).Error("failed adding service binding")
+					return
+				}
 			}
 		} else {
 			// This is a remote container simply attached to an attachable network
 			if err := c.addContainerNameResolution(nid, eid, containerName, taskAliases, ip, "handleEpTableEvent"); err != nil {
-				log.G(context.TODO()).Errorf("failed adding container name resolution for %s epRec:%v err:%v", eid, epRec, err)
+				logger.WithError(err).Errorf("failed adding container name resolution")
 			}
 		}
 
 	case networkdb.DeleteEvent:
-		log.G(context.TODO()).Debugf("handleEpTableEvent DEL %s R:%v", eid, epRec)
 		if svcID != "" {
 			// This is a remote task part of a service
 			if err := c.rmServiceBinding(svcName, svcID, nid, eid, containerName, vip, ingressPorts, serviceAliases, taskAliases, ip, "handleEpTableEvent", true, true); err != nil {
-				log.G(context.TODO()).Errorf("failed removing service binding for %s epRec:%v err:%v", eid, epRec, err)
+				logger.WithError(err).Error("failed removing service binding")
 				return
 			}
 		} else {
 			// This is a remote container simply attached to an attachable network
 			if err := c.delContainerNameResolution(nid, eid, containerName, taskAliases, ip, "handleEpTableEvent"); err != nil {
-				log.G(context.TODO()).Errorf("failed removing container name resolution for %s epRec:%v err:%v", eid, epRec, err)
+				logger.WithError(err).Errorf("failed removing container name resolution")
 			}
 		}
-	case networkdb.UpdateEvent:
-		log.G(context.TODO()).Debugf("handleEpTableEvent UPD %s R:%v", eid, epRec)
-		// We currently should only get these to inform us that an endpoint
-		// is disabled.  Report if otherwise.
-		if svcID == "" || !epRec.ServiceDisabled {
-			log.G(context.TODO()).Errorf("Unexpected update table event for %s epRec:%v", eid, epRec)
-			return
-		}
-		// This is a remote task that is part of a service that is now disabled
-		if err := c.rmServiceBinding(svcName, svcID, nid, eid, containerName, vip, ingressPorts, serviceAliases, taskAliases, ip, "handleEpTableEvent", true, false); err != nil {
-			log.G(context.TODO()).Errorf("failed disabling service binding for %s epRec:%v err:%v", eid, epRec, err)
-			return
-		}
 	}
 }

libnetwork/networkdb/delegate.go

@@ -169,11 +169,13 @@ func (nDB *NetworkDB) handleTableEvent(tEvent *TableEvent, isBulkSync bool) bool
 	}
 
 	nDB.Lock()
-	e, err := nDB.getEntry(tEvent.TableName, tEvent.NetworkID, tEvent.Key)
+	var entryPresent bool
+	prev, err := nDB.getEntry(tEvent.TableName, tEvent.NetworkID, tEvent.Key)
 	if err == nil {
+		entryPresent = true
 		// We have the latest state. Ignore the event
 		// since it is stale.
-		if e.ltime >= tEvent.LTime {
+		if prev.ltime >= tEvent.LTime {
 			nDB.Unlock()
 			return false
 		}
@@ -187,7 +189,7 @@ func (nDB *NetworkDB) handleTableEvent(tEvent *TableEvent, isBulkSync bool) bool
 		return false
 	}
 
-	e = &entry{
+	e := &entry{
 		ltime:    tEvent.LTime,
 		node:     tEvent.NodeName,
 		value:    tEvent.Value,
@@ -221,18 +223,33 @@ func (nDB *NetworkDB) handleTableEvent(tEvent *TableEvent, isBulkSync bool) bool
 	}
 
 	var op opType
+	value := tEvent.Value
 	switch tEvent.Type {
-	case TableEventTypeCreate:
+	case TableEventTypeCreate, TableEventTypeUpdate:
+		// Gossip messages could arrive out-of-order so it is possible
+		// for an entry's UPDATE event to be received before its CREATE
+		// event. The local watchers should not need to care about such
+		// nuances. Broadcast events to watchers based only on what
+		// changed in the local NetworkDB state.
 		op = opCreate
-	case TableEventTypeUpdate:
-		op = opUpdate
+		if entryPresent && !prev.deleting {
+			op = opUpdate
+		}
 	case TableEventTypeDelete:
+		if !entryPresent || prev.deleting {
+			goto SkipBroadcast
+		}
 		op = opDelete
+		// Broadcast the value most recently observed by watchers,
+		// which may be different from the value in the DELETE event
+		// (e.g. if the DELETE event was received out-of-order).
+		value = prev.value
 	default:
 		// TODO(thaJeztah): make switch exhaustive; add networkdb.TableEventTypeInvalid
 	}
 
-	nDB.broadcaster.Write(makeEvent(op, tEvent.TableName, tEvent.NetworkID, tEvent.Key, tEvent.Value))
+	nDB.broadcaster.Write(makeEvent(op, tEvent.TableName, tEvent.NetworkID, tEvent.Key, value))
+SkipBroadcast:
 	return network.inSync
 }
 

libnetwork/networkdb/networkdb.go

@@ -252,14 +252,27 @@ func DefaultConfig() *Config {
 // New creates a new instance of NetworkDB using the Config passed by
 // the caller.
 func New(c *Config) (*NetworkDB, error) {
+	nDB := new(c)
+	log.G(context.TODO()).Infof("New memberlist node - Node:%v will use memberlist nodeID:%v with config:%+v", c.Hostname, c.NodeID, c)
+	if err := nDB.clusterInit(); err != nil {
+		return nil, err
+	}
+
+	return nDB, nil
+}
+
+func new(c *Config) *NetworkDB {
 	// The garbage collection logic for entries leverage the presence of the network.
 	// For this reason the expiration time of the network is put slightly higher than the entry expiration so that
 	// there is at least 5 extra cycle to make sure that all the entries are properly deleted before deleting the network.
 	c.reapNetworkInterval = c.reapEntryInterval + 5*reapPeriod
 
-	nDB := &NetworkDB{
-		config:         c,
-		indexes:        make(map[int]*iradix.Tree[*entry]),
+	return &NetworkDB{
+		config: c,
+		indexes: map[int]*iradix.Tree[*entry]{
+			byTable:   iradix.New[*entry](),
+			byNetwork: iradix.New[*entry](),
+		},
 		networks:       make(map[string]map[string]*network),
 		nodes:          make(map[string]*node),
 		failedNodes:    make(map[string]*node),
@@ -268,16 +281,6 @@ func New(c *Config) (*NetworkDB, error) {
 		bulkSyncAckTbl: make(map[string]chan struct{}),
 		broadcaster:    events.NewBroadcaster(),
 	}
-
-	nDB.indexes[byTable] = iradix.New[*entry]()
-	nDB.indexes[byNetwork] = iradix.New[*entry]()
-
-	log.G(context.TODO()).Infof("New memberlist node - Node:%v will use memberlist nodeID:%v with config:%+v", c.Hostname, c.NodeID, c)
-	if err := nDB.clusterInit(); err != nil {
-		return nil, err
-	}
-
-	return nDB, nil
 }
 
 // Join joins this NetworkDB instance with a list of peer NetworkDB

libnetwork/networkdb/watch_test.go

@@ -0,0 +1,129 @@
+package networkdb
+
+import (
+	"net"
+	"testing"
+	"time"
+
+	"github.com/docker/go-events"
+	"github.com/hashicorp/memberlist"
+	"github.com/hashicorp/serf/serf"
+	"gotest.tools/v3/assert"
+	is "gotest.tools/v3/assert/cmp"
+)
+
+func TestWatch_out_of_order(t *testing.T) {
+	nDB := new(DefaultConfig())
+	nDB.networkBroadcasts = &memberlist.TransmitLimitedQueue{}
+	nDB.nodeBroadcasts = &memberlist.TransmitLimitedQueue{}
+	assert.Assert(t, nDB.JoinNetwork("network1"))
+
+	(&eventDelegate{nDB}).NotifyJoin(&memberlist.Node{
+		Name: "node1",
+		Addr: net.IPv4(1, 2, 3, 4),
+	})
+
+	d := &delegate{nDB}
+
+	msgs := messageBuffer{t: t}
+	appendTableEvent := tableEventHelper(&msgs, "node1", "network1", "table1")
+	msgs.Append(MessageTypeNetworkEvent, &NetworkEvent{
+		Type:      NetworkEventTypeJoin,
+		LTime:     1,
+		NodeName:  "node1",
+		NetworkID: "network1",
+	})
+	appendTableEvent(1, TableEventTypeCreate, "tombstone1", []byte("a"))
+	appendTableEvent(2, TableEventTypeDelete, "tombstone1", []byte("b"))
+	appendTableEvent(3, TableEventTypeCreate, "key1", []byte("value1"))
+	d.NotifyMsg(msgs.Compound())
+	msgs.Reset()
+
+	nDB.CreateEntry("table1", "network1", "local1", []byte("should not see me in watch events"))
+	watch, cancel := nDB.Watch("table1", "network1")
+	defer cancel()
+
+	// Receive events from node1, with events not received or received out of order
+	// Create, (hidden update), delete
+	appendTableEvent(4, TableEventTypeCreate, "key2", []byte("a"))
+	appendTableEvent(6, TableEventTypeDelete, "key2", []byte("b"))
+	// (Hidden recreate), delete
+	appendTableEvent(8, TableEventTypeDelete, "key2", []byte("c"))
+	// (Hidden recreate), update
+	appendTableEvent(10, TableEventTypeUpdate, "key2", []byte("d"))
+
+	// Update, create
+	appendTableEvent(11, TableEventTypeUpdate, "key3", []byte("b"))
+	appendTableEvent(10, TableEventTypeCreate, "key3", []byte("a"))
+
+	// (Hidden create), update, update
+	appendTableEvent(13, TableEventTypeUpdate, "key4", []byte("b"))
+	appendTableEvent(14, TableEventTypeUpdate, "key4", []byte("c"))
+
+	d.NotifyMsg(msgs.Compound())
+	msgs.Reset()
+
+	got := drainChannel(watch.C)
+	assert.Check(t, is.DeepEqual(got, []events.Event{
+		CreateEvent(event{Table: "table1", NetworkID: "network1", Key: "key2", Value: []byte("a")}),
+		// Delete value should match last observed value,
+		// irrespective of the content of the delete event over the wire.
+		DeleteEvent(event{Table: "table1", NetworkID: "network1", Key: "key2", Value: []byte("a")}),
+		// Updates to previously-deleted keys should be observed as creates.
+		CreateEvent(event{Table: "table1", NetworkID: "network1", Key: "key2", Value: []byte("d")}),
+
+		// Out-of-order update events should be observed as creates.
+		CreateEvent(event{Table: "table1", NetworkID: "network1", Key: "key3", Value: []byte("b")}),
+		CreateEvent(event{Table: "table1", NetworkID: "network1", Key: "key4", Value: []byte("b")}),
+		UpdateEvent(event{Table: "table1", NetworkID: "network1", Key: "key4", Value: []byte("c")}),
+	}))
+}
+
+func drainChannel(ch <-chan events.Event) []events.Event {
+	var events []events.Event
+	for {
+		select {
+		case ev := <-ch:
+			events = append(events, ev)
+		case <-time.After(time.Second):
+			return events
+		}
+	}
+}
+
+type messageBuffer struct {
+	t    *testing.T
+	msgs [][]byte
+}
+
+func (mb *messageBuffer) Append(typ MessageType, msg any) {
+	mb.t.Helper()
+	buf, err := encodeMessage(typ, msg)
+	if err != nil {
+		mb.t.Fatalf("failed to encode message: %v", err)
+	}
+	mb.msgs = append(mb.msgs, buf)
+}
+
+func (mb *messageBuffer) Compound() []byte {
+	return makeCompoundMessage(mb.msgs)
+}
+
+func (mb *messageBuffer) Reset() {
+	mb.msgs = nil
+}
+
+func tableEventHelper(mb *messageBuffer, nodeName, networkID, tableName string) func(ltime serf.LamportTime, typ TableEvent_Type, key string, value []byte) {
+	return func(ltime serf.LamportTime, typ TableEvent_Type, key string, value []byte) {
+		mb.t.Helper()
+		mb.Append(MessageTypeTableEvent, &TableEvent{
+			Type:      typ,
+			LTime:     ltime,
+			NodeName:  nodeName,
+			NetworkID: networkID,
+			TableName: tableName,
+			Key:       key,
+			Value:     value,
+		})
+	}
+}