Merge pull request #91968 from MikeSpreitzer/fix91709

Remove contemplation of invariant violations from delta_fifo.go

Kubernetes-commit: 95fcb55b416f7f68ba1707cb9e6d11a3ed63f5c7

Godeps/Godeps.json

@@ -444,7 +444,7 @@
 		},
 		{
 			"ImportPath": "k8s.io/apimachinery",
-			"Rev": "f3a77abeaa89"
+			"Rev": "f9108ab438d8"
 		},
 		{
 			"ImportPath": "k8s.io/gengo",

go.mod

@@ -27,7 +27,7 @@ require (
 	golang.org/x/oauth2 v0.0.0-20191202225959-858c2ad4c8b6
 	golang.org/x/time v0.0.0-20191024005414-555d28b269f0
 	k8s.io/api v0.0.0-20200827171543-af6c97a0fc38
-	k8s.io/apimachinery v0.0.0-20200827211417-f3a77abeaa89
+	k8s.io/apimachinery v0.0.0-20200828011407-f9108ab438d8
 	k8s.io/klog/v2 v2.2.0
 	k8s.io/utils v0.0.0-20200729134348-d5654de09c73
 	sigs.k8s.io/yaml v1.2.0
@@ -35,5 +35,5 @@ require (
 
 replace (
 	k8s.io/api => k8s.io/api v0.0.0-20200827171543-af6c97a0fc38
-	k8s.io/apimachinery => k8s.io/apimachinery v0.0.0-20200827211417-f3a77abeaa89
+	k8s.io/apimachinery => k8s.io/apimachinery v0.0.0-20200828011407-f9108ab438d8
 )

go.sum

@@ -334,7 +334,7 @@ honnef.co/go/tools v0.0.0-20190418001031-e561f6794a2a/go.mod h1:rf3lG4BRIbNafJWh
 honnef.co/go/tools v0.0.0-20190523083050-ea95bdfd59fc/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
 honnef.co/go/tools v0.0.1-2019.2.3/go.mod h1:a3bituU0lyd329TUQxRnasdCoJDkEUEAqEt0JzvZhAg=
 k8s.io/api v0.0.0-20200827171543-af6c97a0fc38/go.mod h1:ZPrpciYrawB63sVOSyRCIMMzmrE4ndsAt46dk7rQ6fU=
-k8s.io/apimachinery v0.0.0-20200827211417-f3a77abeaa89/go.mod h1:DnPGDnARWFvYa3pMHgSxtbZb7gpzzAZ1pTfaUNDVlmA=
+k8s.io/apimachinery v0.0.0-20200828011407-f9108ab438d8/go.mod h1:DnPGDnARWFvYa3pMHgSxtbZb7gpzzAZ1pTfaUNDVlmA=
 k8s.io/gengo v0.0.0-20200413195148-3a45101e95ac/go.mod h1:ezvh/TsK7cY6rbqRK0oQQ8IAqLxYwwyPxAX1Pzy0ii0=
 k8s.io/klog/v2 v2.0.0/go.mod h1:PBfzABfn139FHAV07az/IF9Wp1bkk3vpT2XSJ76fSDE=
 k8s.io/klog/v2 v2.2.0 h1:XRvcwJozkgZ1UQJmfMGpvRthQHOvihEhYtDfAaxMz/A=

tools/cache/delta_fifo.go

@@ -145,7 +145,7 @@ func NewDeltaFIFOWithOptions(opts DeltaFIFOOptions) *DeltaFIFO {
 // DeltaFIFO's Pop(), Get(), and GetByKey() methods return
 // interface{} to satisfy the Store/Queue interfaces, but they
 // will always return an object of type Deltas. List() returns
-// the newest objects currently in the FIFO.
+// the newest object from each accumulator in the FIFO.
 //
 // A DeltaFIFO's knownObjects KeyListerGetter provides the abilities
 // to list Store keys and to get objects by Store key.  The objects in
@@ -161,12 +161,13 @@ type DeltaFIFO struct {
 	lock sync.RWMutex
 	cond sync.Cond
 
-	// `items` maps keys to Deltas.
+	// `items` maps a key to a Deltas.
-	// `queue` maintains FIFO order of keys for consumption in Pop().
+	// Each such Deltas has at least one Delta.
-	// We maintain the property that keys in the `items` and `queue` are
-	// strictly 1:1 mapping, and that all Deltas in `items` should have
-	// at least one Delta.
 	items map[string]Deltas
+
+	// `queue` maintains FIFO order of keys for consumption in Pop().
+	// There are no duplicates in `queue`.
+	// A key is in `queue` if and only if it is in `items`.
 	queue []string
 
 	// populated is true if the first batch of items inserted by Replace() has been populated
@@ -376,8 +377,8 @@ func (f *DeltaFIFO) queueActionLocked(actionType DeltaType, obj interface{}) err
 	if err != nil {
 		return KeyError{obj, err}
 	}
-
+	oldDeltas := f.items[id]
-	newDeltas := append(f.items[id], Delta{actionType, obj})
+	newDeltas := append(oldDeltas, Delta{actionType, obj})
 	newDeltas = dedupDeltas(newDeltas)
 
 	if len(newDeltas) > 0 {
@@ -389,10 +390,14 @@ func (f *DeltaFIFO) queueActionLocked(actionType DeltaType, obj interface{}) err
 	} else {
 		// This never happens, because dedupDeltas never returns an empty list
 		// when given a non-empty list (as it is here).
-		// But if somehow it ever does return an empty list, then
+		// If somehow it happens anyway, deal with it but complain.
-		// We need to remove this from our map (extra items in the queue are
+		if oldDeltas == nil {
-		// ignored if they are not in the map).
+			klog.Errorf("Impossible dedupDeltas for id=%q: oldDeltas=%#+v, obj=%#+v; ignoring", id, oldDeltas, obj)
-		delete(f.items, id)
+			return nil
+		}
+		klog.Errorf("Impossible dedupDeltas for id=%q: oldDeltas=%#+v, obj=%#+v; breaking invariant by storing empty Deltas", id, oldDeltas, obj)
+		f.items[id] = newDeltas
+		return fmt.Errorf("Impossible dedupDeltas for id=%q: oldDeltas=%#+v, obj=%#+v; broke DeltaFIFO invariant by storing empty Deltas", id, oldDeltas, obj)
 	}
 	return nil
 }
@@ -459,7 +464,7 @@ func (f *DeltaFIFO) IsClosed() bool {
 	return f.closed
 }
 
-// Pop blocks until an item is added to the queue, and then returns it.  If
+// Pop blocks until the queue has some items, and then returns one.  If
 // multiple items are ready, they are returned in the order in which they were
 // added/updated. The item is removed from the queue (and the store) before it
 // is returned, so if you don't successfully process it, you need to add it back
@@ -494,7 +499,8 @@ func (f *DeltaFIFO) Pop(process PopProcessFunc) (interface{}, error) {
 		}
 		item, ok := f.items[id]
 		if !ok {
-			// Item may have been deleted subsequently.
+			// This should never happen
+			klog.Errorf("Inconceivable! %q was in f.queue but not f.items; ignoring.", id)
 			continue
 		}
 		delete(f.items, id)