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

This reverts commit c68671d908.

Signed-off-by: Rob Murray <rob.murray@docker.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()).WithError(err).Error("Failed to unmarshal service table value")
+		log.G(context.TODO()).Errorf("Failed to unmarshal service table value: %v", err)
 		return
 	}
 
@@ -924,54 +924,53 @@ 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 {
-		logger.Errorf("Invalid endpoint name/ip received while handling service table event %s", value)
+		log.G(context.TODO()).Errorf("Invalid endpoint name/ip received while handling service table event %s", value)
 		return
 	}
 
-	logger.Debug("handleEpTableEvent")
-
 	switch ev.(type) {
-	case networkdb.CreateEvent, networkdb.UpdateEvent:
+	case networkdb.CreateEvent:
+		log.G(context.TODO()).Debugf("handleEpTableEvent ADD %s R:%v", eid, epRec)
 		if svcID != "" {
 			// This is a remote task part of a service
-			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
-				}
+			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
 			}
 		} else {
 			// This is a remote container simply attached to an attachable network
 			if err := c.addContainerNameResolution(nid, eid, containerName, taskAliases, ip, "handleEpTableEvent"); err != nil {
-				logger.WithError(err).Errorf("failed adding container name resolution")
+				log.G(context.TODO()).Errorf("failed adding container name resolution for %s epRec:%v err:%v", eid, epRec, err)
 			}
 		}
 
 	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 {
-				logger.WithError(err).Error("failed removing service binding")
+				log.G(context.TODO()).Errorf("failed removing service binding for %s epRec:%v err:%v", eid, epRec, err)
 				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 {
-				logger.WithError(err).Errorf("failed removing container name resolution")
+				log.G(context.TODO()).Errorf("failed removing container name resolution for %s epRec:%v err:%v", eid, epRec, err)
 			}
 		}
+	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,13 +169,11 @@ func (nDB *NetworkDB) handleTableEvent(tEvent *TableEvent, isBulkSync bool) bool
 	}
 
 	nDB.Lock()
-	var entryPresent bool
-	prev, err := nDB.getEntry(tEvent.TableName, tEvent.NetworkID, tEvent.Key)
+	e, 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 prev.ltime >= tEvent.LTime {
+		if e.ltime >= tEvent.LTime {
 			nDB.Unlock()
 			return false
 		}
@@ -189,7 +187,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,
@@ -223,33 +221,18 @@ func (nDB *NetworkDB) handleTableEvent(tEvent *TableEvent, isBulkSync bool) bool
 	}
 
 	var op opType
-	value := tEvent.Value
 	switch tEvent.Type {
-	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.
+	case TableEventTypeCreate:
 		op = opCreate
-		if entryPresent && !prev.deleting {
-			op = opUpdate
-		}
+	case TableEventTypeUpdate:
+		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, value))
-SkipBroadcast:
+	nDB.broadcaster.Write(makeEvent(op, tEvent.TableName, tEvent.NetworkID, tEvent.Key, tEvent.Value))
 	return network.inSync
 }
 

libnetwork/networkdb/networkdb.go

@@ -252,27 +252,14 @@ 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
 
-	return &NetworkDB{
-		config: c,
-		indexes: map[int]*iradix.Tree[*entry]{
-			byTable:   iradix.New[*entry](),
-			byNetwork: iradix.New[*entry](),
-		},
+	nDB := &NetworkDB{
+		config:         c,
+		indexes:        make(map[int]*iradix.Tree[*entry]),
 		networks:       make(map[string]map[string]*network),
 		nodes:          make(map[string]*node),
 		failedNodes:    make(map[string]*node),
@@ -281,6 +268,16 @@ func new(c *Config) *NetworkDB {
 		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

@@ -1,129 +0,0 @@
-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,
-		})
-	}
-}