Fix inconsistent update types from dedupe buffer.

If the dedupe buffer sees "create", "delete", "create", and it
elides the "delete" then it could send duplicate "new" events for
the same key resulting in miscounting downstream.

Author: fasaxc

felix/daemon/daemon.go

@@ -592,7 +592,7 @@ configRetry:
 		}
 		healthAggregator.Report(healthName, &health.HealthReport{Live: true, Ready: true})
 		// Up-to-data Typha client will refuse to connect unless Typha signals
-		// that it supports nore resource updates.
+		// that it supports node resource updates.
 		configParams.SetUseNodeResourceUpdates(true)
 
 		go func() {

libcalico-go/lib/backend/api/api.go

@@ -191,6 +191,21 @@ const (
 	UpdateTypeKVDeleted
 )
 
+func (u UpdateType) String() string {
+	switch u {
+	case UpdateTypeKVUnknown:
+		return "unknown"
+	case UpdateTypeKVNew:
+		return "new"
+	case UpdateTypeKVUpdated:
+		return "updated"
+	case UpdateTypeKVDeleted:
+		return "deleted"
+	default:
+		return fmt.Sprintf("Unknown<%v>", uint8(u))
+	}
+}
+
 // Interface can be implemented by anything that knows how to watch and report changes.
 type WatchInterface interface {
 	// Stops watching. Will close the channel returned by ResultChan(). Releases

libcalico-go/lib/backend/syncersv1/dedupebuffer/dedupe_buffer.go

@@ -192,6 +192,19 @@ func (d *DedupeBuffer) OnUpdatesKeysKnown(updates []api.Update, keys []string) {
 
 func (d *DedupeBuffer) queueUpdate(key string, u api.Update) {
 	debug := log.IsLevelEnabled(log.DebugLevel)
+
+	if u.Value != nil {
+		// A new KV or an update. Since we dedupe sequences of updates for the
+		// same key, we need to recalculate the update type to make sense to the
+		// downstream receiver.  We do this even if the update is not on the
+		// queue in order to handle resyncs with Typha.
+		if d.liveResourceKeys.Contains(key) {
+			u.UpdateType = api.UpdateTypeKVUpdated
+		} else {
+			u.UpdateType = api.UpdateTypeKVNew
+		}
+	}
+
 	if element, ok := d.keyToPendingUpdate[key]; ok {
 		// Already got an in-flight update for this key.
 		if u.Value == nil && !d.liveResourceKeys.Contains(key) {
@@ -208,6 +221,7 @@ func (d *DedupeBuffer) queueUpdate(key string, u api.Update) {
 			if debug {
 				log.WithField("key", key).Debug("Key updated before being sent.")
 			}
+
 			usk := element.Value.(updateWithStringKey)
 			usk.update = u
 			element.Value = usk