diff --git a/libnetwork/agent.go b/libnetwork/agent.go index aa8dc14b12..b68b15ac2d 100644 --- a/libnetwork/agent.go +++ b/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 + } } } diff --git a/libnetwork/networkdb/delegate.go b/libnetwork/networkdb/delegate.go index fff0ac78f0..c8071db072 100644 --- a/libnetwork/networkdb/delegate.go +++ b/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 } diff --git a/libnetwork/networkdb/networkdb.go b/libnetwork/networkdb/networkdb.go index fdc43f36b6..bf5bc547d7 100644 --- a/libnetwork/networkdb/networkdb.go +++ b/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 diff --git a/libnetwork/networkdb/watch_test.go b/libnetwork/networkdb/watch_test.go deleted file mode 100644 index 245a67d4ac..0000000000 --- a/libnetwork/networkdb/watch_test.go +++ /dev/null @@ -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, - }) - } -}