delta_fifo_test.go

Documentation: k8s.io/client-go/tools/cache

     1  /*
     2  Copyright 2014 The Kubernetes Authors.
     3  
     4  Licensed under the Apache License, Version 2.0 (the "License");
     5  you may not use this file except in compliance with the License.
     6  You may obtain a copy of the License at
     7  
     8      http://www.apache.org/licenses/LICENSE-2.0
     9  
    10  Unless required by applicable law or agreed to in writing, software
    11  distributed under the License is distributed on an "AS IS" BASIS,
    12  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    13  See the License for the specific language governing permissions and
    14  limitations under the License.
    15  */
    16  
    17  package cache
    18  
    19  import (
    20  	"errors"
    21  	"fmt"
    22  	"reflect"
    23  	"runtime"
    24  	"testing"
    25  	"time"
    26  )
    27  
    28  // helper function to reduce stuttering
    29  func testPop(f *DeltaFIFO) testFifoObject {
    30  	return Pop(f).(Deltas).Newest().Object.(testFifoObject)
    31  }
    32  
    33  // literalListerGetter is a KeyListerGetter that is based on a
    34  // function that returns a slice of objects to list and get.
    35  // The function must list the same objects every time.
    36  type literalListerGetter func() []testFifoObject
    37  
    38  var _ KeyListerGetter = literalListerGetter(nil)
    39  
    40  // ListKeys just calls kl.
    41  func (kl literalListerGetter) ListKeys() []string {
    42  	result := []string{}
    43  	for _, fifoObj := range kl() {
    44  		result = append(result, fifoObj.name)
    45  	}
    46  	return result
    47  }
    48  
    49  // GetByKey returns the key if it exists in the list returned by kl.
    50  func (kl literalListerGetter) GetByKey(key string) (interface{}, bool, error) {
    51  	for _, v := range kl() {
    52  		if v.name == key {
    53  			return v, true, nil
    54  		}
    55  	}
    56  	return nil, false, nil
    57  }
    58  
    59  func TestDeltaFIFO_basic(t *testing.T) {
    60  	f := NewDeltaFIFOWithOptions(DeltaFIFOOptions{KeyFunction: testFifoObjectKeyFunc})
    61  	const amount = 500
    62  	go func() {
    63  		for i := 0; i < amount; i++ {
    64  			f.Add(mkFifoObj(string([]rune{'a', rune(i)}), i+1))
    65  		}
    66  	}()
    67  	go func() {
    68  		for u := uint64(0); u < amount; u++ {
    69  			f.Add(mkFifoObj(string([]rune{'b', rune(u)}), u+1))
    70  		}
    71  	}()
    72  
    73  	lastInt := int(0)
    74  	lastUint := uint64(0)
    75  	for i := 0; i < amount*2; i++ {
    76  		switch obj := testPop(f).val.(type) {
    77  		case int:
    78  			if obj <= lastInt {
    79  				t.Errorf("got %v (int) out of order, last was %v", obj, lastInt)
    80  			}
    81  			lastInt = obj
    82  		case uint64:
    83  			if obj <= lastUint {
    84  				t.Errorf("got %v (uint) out of order, last was %v", obj, lastUint)
    85  			} else {
    86  				lastUint = obj
    87  			}
    88  		default:
    89  			t.Fatalf("unexpected type %#v", obj)
    90  		}
    91  	}
    92  }
    93  
    94  // TestDeltaFIFO_replaceWithDeleteDeltaIn tests that a `Sync` delta for an
    95  // object `O` with ID `X` is added when .Replace is called and `O` is among the
    96  // replacement objects even if the DeltaFIFO already stores in terminal position
    97  // a delta of type `Delete` for ID `X`. Not adding the `Sync` delta causes
    98  // SharedIndexInformers to miss `O`'s create notification, see https://github.com/kubernetes/kubernetes/issues/83810
    99  // for more details.
   100  func TestDeltaFIFO_replaceWithDeleteDeltaIn(t *testing.T) {
   101  	oldObj := mkFifoObj("foo", 1)
   102  	newObj := mkFifoObj("foo", 2)
   103  
   104  	f := NewDeltaFIFOWithOptions(DeltaFIFOOptions{
   105  		KeyFunction: testFifoObjectKeyFunc,
   106  		KnownObjects: literalListerGetter(func() []testFifoObject {
   107  			return []testFifoObject{oldObj}
   108  		}),
   109  	})
   110  
   111  	f.Delete(oldObj)
   112  	f.Replace([]interface{}{newObj}, "")
   113  
   114  	actualDeltas := Pop(f)
   115  	expectedDeltas := Deltas{
   116  		Delta{Type: Deleted, Object: oldObj},
   117  		Delta{Type: Sync, Object: newObj},
   118  	}
   119  	if !reflect.DeepEqual(expectedDeltas, actualDeltas) {
   120  		t.Errorf("expected %#v, got %#v", expectedDeltas, actualDeltas)
   121  	}
   122  }
   123  
   124  func TestDeltaFIFOW_ReplaceMakesDeletionsForObjectsOnlyInQueue(t *testing.T) {
   125  	obj := mkFifoObj("foo", 2)
   126  	objV2 := mkFifoObj("foo", 3)
   127  	table := []struct {
   128  		name           string
   129  		operations     func(f *DeltaFIFO)
   130  		expectedDeltas Deltas
   131  	}{
   132  		{
   133  			name: "Added object should be deleted on Replace",
   134  			operations: func(f *DeltaFIFO) {
   135  				f.Add(obj)
   136  				f.Replace([]interface{}{}, "0")
   137  			},
   138  			expectedDeltas: Deltas{
   139  				{Added, obj},
   140  				{Deleted, DeletedFinalStateUnknown{Key: "foo", Obj: obj}},
   141  			},
   142  		},
   143  		{
   144  			name: "Replaced object should have only a single Delete",
   145  			operations: func(f *DeltaFIFO) {
   146  				f.emitDeltaTypeReplaced = true
   147  				f.Add(obj)
   148  				f.Replace([]interface{}{obj}, "0")
   149  				f.Replace([]interface{}{}, "0")
   150  			},
   151  			expectedDeltas: Deltas{
   152  				{Added, obj},
   153  				{Replaced, obj},
   154  				{Deleted, DeletedFinalStateUnknown{Key: "foo", Obj: obj}},
   155  			},
   156  		},
   157  		{
   158  			name: "Deleted object should have only a single Delete",
   159  			operations: func(f *DeltaFIFO) {
   160  				f.Add(obj)
   161  				f.Delete(obj)
   162  				f.Replace([]interface{}{}, "0")
   163  			},
   164  			expectedDeltas: Deltas{
   165  				{Added, obj},
   166  				{Deleted, obj},
   167  			},
   168  		},
   169  		{
   170  			name: "Synced objects should have a single delete",
   171  			operations: func(f *DeltaFIFO) {
   172  				f.Add(obj)
   173  				f.Replace([]interface{}{obj}, "0")
   174  				f.Replace([]interface{}{obj}, "0")
   175  				f.Replace([]interface{}{}, "0")
   176  			},
   177  			expectedDeltas: Deltas{
   178  				{Added, obj},
   179  				{Sync, obj},
   180  				{Sync, obj},
   181  				{Deleted, DeletedFinalStateUnknown{Key: "foo", Obj: obj}},
   182  			},
   183  		},
   184  		{
   185  			name: "Added objects should have a single delete on multiple Replaces",
   186  			operations: func(f *DeltaFIFO) {
   187  				f.Add(obj)
   188  				f.Replace([]interface{}{}, "0")
   189  				f.Replace([]interface{}{}, "1")
   190  			},
   191  			expectedDeltas: Deltas{
   192  				{Added, obj},
   193  				{Deleted, DeletedFinalStateUnknown{Key: "foo", Obj: obj}},
   194  			},
   195  		},
   196  		{
   197  			name: "Added and deleted and added object should be deleted",
   198  			operations: func(f *DeltaFIFO) {
   199  				f.Add(obj)
   200  				f.Delete(obj)
   201  				f.Add(objV2)
   202  				f.Replace([]interface{}{}, "0")
   203  			},
   204  			expectedDeltas: Deltas{
   205  				{Added, obj},
   206  				{Deleted, obj},
   207  				{Added, objV2},
   208  				{Deleted, DeletedFinalStateUnknown{Key: "foo", Obj: objV2}},
   209  			},
   210  		},
   211  	}
   212  	for _, tt := range table {
   213  		tt := tt
   214  
   215  		t.Run(tt.name, func(t *testing.T) {
   216  			// Test with a DeltaFIFO with a backing KnownObjects
   217  			fWithKnownObjects := NewDeltaFIFOWithOptions(DeltaFIFOOptions{
   218  				KeyFunction: testFifoObjectKeyFunc,
   219  				KnownObjects: literalListerGetter(func() []testFifoObject {
   220  					return []testFifoObject{}
   221  				}),
   222  			})
   223  			tt.operations(fWithKnownObjects)
   224  			actualDeltasWithKnownObjects := Pop(fWithKnownObjects)
   225  			if !reflect.DeepEqual(tt.expectedDeltas, actualDeltasWithKnownObjects) {
   226  				t.Errorf("expected %#v, got %#v", tt.expectedDeltas, actualDeltasWithKnownObjects)
   227  			}
   228  			if len(fWithKnownObjects.items) != 0 {
   229  				t.Errorf("expected no extra deltas (empty map), got %#v", fWithKnownObjects.items)
   230  			}
   231  
   232  			// Test with a DeltaFIFO without a backing KnownObjects
   233  			fWithoutKnownObjects := NewDeltaFIFOWithOptions(DeltaFIFOOptions{
   234  				KeyFunction: testFifoObjectKeyFunc,
   235  			})
   236  			tt.operations(fWithoutKnownObjects)
   237  			actualDeltasWithoutKnownObjects := Pop(fWithoutKnownObjects)
   238  			if !reflect.DeepEqual(tt.expectedDeltas, actualDeltasWithoutKnownObjects) {
   239  				t.Errorf("expected %#v, got %#v", tt.expectedDeltas, actualDeltasWithoutKnownObjects)
   240  			}
   241  			if len(fWithoutKnownObjects.items) != 0 {
   242  				t.Errorf("expected no extra deltas (empty map), got %#v", fWithoutKnownObjects.items)
   243  			}
   244  		})
   245  	}
   246  }
   247  
   248  func TestDeltaFIFO_requeueOnPop(t *testing.T) {
   249  	f := NewDeltaFIFOWithOptions(DeltaFIFOOptions{KeyFunction: testFifoObjectKeyFunc})
   250  
   251  	f.Add(mkFifoObj("foo", 10))
   252  	_, err := f.Pop(func(obj interface{}, isInInitialList bool) error {
   253  		if obj.(Deltas)[0].Object.(testFifoObject).name != "foo" {
   254  			t.Fatalf("unexpected object: %#v", obj)
   255  		}
   256  		return ErrRequeue{Err: nil}
   257  	})
   258  	if err != nil {
   259  		t.Fatalf("unexpected error: %v", err)
   260  	}
   261  	if _, ok, err := f.GetByKey("foo"); !ok || err != nil {
   262  		t.Fatalf("object should have been requeued: %t %v", ok, err)
   263  	}
   264  
   265  	_, err = f.Pop(func(obj interface{}, isInInitialList bool) error {
   266  		if obj.(Deltas)[0].Object.(testFifoObject).name != "foo" {
   267  			t.Fatalf("unexpected object: %#v", obj)
   268  		}
   269  		return ErrRequeue{Err: fmt.Errorf("test error")}
   270  	})
   271  	if err == nil || err.Error() != "test error" {
   272  		t.Fatalf("unexpected error: %v", err)
   273  	}
   274  	if _, ok, err := f.GetByKey("foo"); !ok || err != nil {
   275  		t.Fatalf("object should have been requeued: %t %v", ok, err)
   276  	}
   277  
   278  	_, err = f.Pop(func(obj interface{}, isInInitialList bool) error {
   279  		if obj.(Deltas)[0].Object.(testFifoObject).name != "foo" {
   280  			t.Fatalf("unexpected object: %#v", obj)
   281  		}
   282  		return nil
   283  	})
   284  	if err != nil {
   285  		t.Fatalf("unexpected error: %v", err)
   286  	}
   287  	if _, ok, err := f.GetByKey("foo"); ok || err != nil {
   288  		t.Fatalf("object should have been removed: %t %v", ok, err)
   289  	}
   290  }
   291  
   292  func TestDeltaFIFO_addUpdate(t *testing.T) {
   293  	f := NewDeltaFIFOWithOptions(DeltaFIFOOptions{KeyFunction: testFifoObjectKeyFunc})
   294  	f.Add(mkFifoObj("foo", 10))
   295  	f.Update(mkFifoObj("foo", 12))
   296  	f.Delete(mkFifoObj("foo", 15))
   297  
   298  	if e, a := []interface{}{mkFifoObj("foo", 15)}, f.List(); !reflect.DeepEqual(e, a) {
   299  		t.Errorf("Expected %+v, got %+v", e, a)
   300  	}
   301  	if e, a := []string{"foo"}, f.ListKeys(); !reflect.DeepEqual(e, a) {
   302  		t.Errorf("Expected %+v, got %+v", e, a)
   303  	}
   304  
   305  	got := make(chan testFifoObject, 2)
   306  	go func() {
   307  		for {
   308  			obj := testPop(f)
   309  			t.Logf("got a thing %#v", obj)
   310  			t.Logf("D len: %v", len(f.queue))
   311  			got <- obj
   312  		}
   313  	}()
   314  
   315  	first := <-got
   316  	if e, a := 15, first.val; e != a {
   317  		t.Errorf("Didn't get updated value (%v), got %v", e, a)
   318  	}
   319  	select {
   320  	case unexpected := <-got:
   321  		t.Errorf("Got second value %v", unexpected.val)
   322  	case <-time.After(50 * time.Millisecond):
   323  	}
   324  	_, exists, _ := f.Get(mkFifoObj("foo", ""))
   325  	if exists {
   326  		t.Errorf("item did not get removed")
   327  	}
   328  }
   329  
   330  type rvAndXfrm struct {
   331  	rv   int
   332  	xfrm int
   333  }
   334  
   335  func TestDeltaFIFO_transformer(t *testing.T) {
   336  	mk := func(name string, rv int) testFifoObject {
   337  		return mkFifoObj(name, &rvAndXfrm{rv, 0})
   338  	}
   339  	xfrm := TransformFunc(func(obj interface{}) (interface{}, error) {
   340  		switch v := obj.(type) {
   341  		case testFifoObject:
   342  			v.val.(*rvAndXfrm).xfrm++
   343  		case DeletedFinalStateUnknown:
   344  			if x := v.Obj.(testFifoObject).val.(*rvAndXfrm).xfrm; x != 1 {
   345  				return nil, fmt.Errorf("object has been transformed wrong number of times: %#v", obj)
   346  			}
   347  		default:
   348  			return nil, fmt.Errorf("unexpected object: %#v", obj)
   349  		}
   350  		return obj, nil
   351  	})
   352  
   353  	must := func(err error) {
   354  		if err != nil {
   355  			t.Fatal(err)
   356  		}
   357  	}
   358  
   359  	f := NewDeltaFIFOWithOptions(DeltaFIFOOptions{
   360  		KeyFunction: testFifoObjectKeyFunc,
   361  		Transformer: xfrm,
   362  	})
   363  	must(f.Add(mk("foo", 10)))
   364  	must(f.Add(mk("bar", 11)))
   365  	must(f.Update(mk("foo", 12)))
   366  	must(f.Delete(mk("foo", 15)))
   367  	must(f.Replace([]interface{}{}, ""))
   368  	must(f.Add(mk("bar", 16)))
   369  	must(f.Replace([]interface{}{}, ""))
   370  
   371  	// Should be empty
   372  	if e, a := []string{"foo", "bar"}, f.ListKeys(); !reflect.DeepEqual(e, a) {
   373  		t.Errorf("Expected %+v, got %+v", e, a)
   374  	}
   375  
   376  	for i := 0; i < 2; i++ {
   377  		obj, err := f.Pop(func(o interface{}, isInInitialList bool) error { return nil })
   378  		if err != nil {
   379  			t.Fatalf("got nothing on try %v?", i)
   380  		}
   381  		obj = obj.(Deltas).Newest().Object
   382  		switch v := obj.(type) {
   383  		case testFifoObject:
   384  			if v.name != "foo" {
   385  				t.Errorf("expected regular deletion of foo, got %q", v.name)
   386  			}
   387  			rx := v.val.(*rvAndXfrm)
   388  			if rx.rv != 15 {
   389  				t.Errorf("expected last message, got %#v", obj)
   390  			}
   391  			if rx.xfrm != 1 {
   392  				t.Errorf("obj %v transformed wrong number of times.", obj)
   393  			}
   394  		case DeletedFinalStateUnknown:
   395  			tf := v.Obj.(testFifoObject)
   396  			rx := tf.val.(*rvAndXfrm)
   397  			if tf.name != "bar" {
   398  				t.Errorf("expected tombstone deletion of bar, got %q", tf.name)
   399  			}
   400  			if rx.rv != 16 {
   401  				t.Errorf("expected last message, got %#v", obj)
   402  			}
   403  			if rx.xfrm != 1 {
   404  				t.Errorf("tombstoned obj %v transformed wrong number of times.", obj)
   405  			}
   406  		default:
   407  			t.Errorf("unknown item %#v", obj)
   408  		}
   409  	}
   410  }
   411  
   412  func TestDeltaFIFO_enqueueingNoLister(t *testing.T) {
   413  	f := NewDeltaFIFOWithOptions(DeltaFIFOOptions{KeyFunction: testFifoObjectKeyFunc})
   414  	f.Add(mkFifoObj("foo", 10))
   415  	f.Update(mkFifoObj("bar", 15))
   416  	f.Add(mkFifoObj("qux", 17))
   417  	f.Delete(mkFifoObj("qux", 18))
   418  
   419  	// This delete does not enqueue anything because baz doesn't exist.
   420  	f.Delete(mkFifoObj("baz", 20))
   421  
   422  	expectList := []int{10, 15, 18}
   423  	for _, expect := range expectList {
   424  		if e, a := expect, testPop(f).val; e != a {
   425  			t.Errorf("Didn't get updated value (%v), got %v", e, a)
   426  		}
   427  	}
   428  	if e, a := 0, len(f.items); e != a {
   429  		t.Errorf("queue unexpectedly not empty: %v != %v\n%#v", e, a, f.items)
   430  	}
   431  }
   432  
   433  func TestDeltaFIFO_enqueueingWithLister(t *testing.T) {
   434  	f := NewDeltaFIFOWithOptions(DeltaFIFOOptions{
   435  		KeyFunction: testFifoObjectKeyFunc,
   436  		KnownObjects: literalListerGetter(func() []testFifoObject {
   437  			return []testFifoObject{mkFifoObj("foo", 5), mkFifoObj("bar", 6), mkFifoObj("baz", 7)}
   438  		}),
   439  	})
   440  	f.Add(mkFifoObj("foo", 10))
   441  	f.Update(mkFifoObj("bar", 15))
   442  
   443  	// This delete does enqueue the deletion, because "baz" is in the key lister.
   444  	f.Delete(mkFifoObj("baz", 20))
   445  
   446  	expectList := []int{10, 15, 20}
   447  	for _, expect := range expectList {
   448  		if e, a := expect, testPop(f).val; e != a {
   449  			t.Errorf("Didn't get updated value (%v), got %v", e, a)
   450  		}
   451  	}
   452  	if e, a := 0, len(f.items); e != a {
   453  		t.Errorf("queue unexpectedly not empty: %v != %v", e, a)
   454  	}
   455  }
   456  
   457  func TestDeltaFIFO_addReplace(t *testing.T) {
   458  	f := NewDeltaFIFOWithOptions(DeltaFIFOOptions{KeyFunction: testFifoObjectKeyFunc})
   459  	f.Add(mkFifoObj("foo", 10))
   460  	f.Replace([]interface{}{mkFifoObj("foo", 15)}, "0")
   461  	got := make(chan testFifoObject, 2)
   462  	go func() {
   463  		for {
   464  			got <- testPop(f)
   465  		}
   466  	}()
   467  
   468  	first := <-got
   469  	if e, a := 15, first.val; e != a {
   470  		t.Errorf("Didn't get updated value (%v), got %v", e, a)
   471  	}
   472  	select {
   473  	case unexpected := <-got:
   474  		t.Errorf("Got second value %v", unexpected.val)
   475  	case <-time.After(50 * time.Millisecond):
   476  	}
   477  	_, exists, _ := f.Get(mkFifoObj("foo", ""))
   478  	if exists {
   479  		t.Errorf("item did not get removed")
   480  	}
   481  }
   482  
   483  func TestDeltaFIFO_ResyncNonExisting(t *testing.T) {
   484  	f := NewDeltaFIFOWithOptions(DeltaFIFOOptions{
   485  		KeyFunction: testFifoObjectKeyFunc,
   486  		KnownObjects: literalListerGetter(func() []testFifoObject {
   487  			return []testFifoObject{mkFifoObj("foo", 5)}
   488  		}),
   489  	})
   490  	f.Delete(mkFifoObj("foo", 10))
   491  	f.Resync()
   492  
   493  	deltas := f.items["foo"]
   494  	if len(deltas) != 1 {
   495  		t.Fatalf("unexpected deltas length: %v", deltas)
   496  	}
   497  	if deltas[0].Type != Deleted {
   498  		t.Errorf("unexpected delta: %v", deltas[0])
   499  	}
   500  }
   501  
   502  func TestDeltaFIFO_Resync(t *testing.T) {
   503  	f := NewDeltaFIFOWithOptions(DeltaFIFOOptions{
   504  		KeyFunction: testFifoObjectKeyFunc,
   505  		KnownObjects: literalListerGetter(func() []testFifoObject {
   506  			return []testFifoObject{mkFifoObj("foo", 5)}
   507  		}),
   508  	})
   509  	f.Resync()
   510  
   511  	deltas := f.items["foo"]
   512  	if len(deltas) != 1 {
   513  		t.Fatalf("unexpected deltas length: %v", deltas)
   514  	}
   515  	if deltas[0].Type != Sync {
   516  		t.Errorf("unexpected delta: %v", deltas[0])
   517  	}
   518  }
   519  
   520  func TestDeltaFIFO_DeleteExistingNonPropagated(t *testing.T) {
   521  	f := NewDeltaFIFOWithOptions(DeltaFIFOOptions{
   522  		KeyFunction: testFifoObjectKeyFunc,
   523  		KnownObjects: literalListerGetter(func() []testFifoObject {
   524  			return []testFifoObject{}
   525  		}),
   526  	})
   527  	f.Add(mkFifoObj("foo", 5))
   528  	f.Delete(mkFifoObj("foo", 6))
   529  
   530  	deltas := f.items["foo"]
   531  	if len(deltas) != 2 {
   532  		t.Fatalf("unexpected deltas length: %v", deltas)
   533  	}
   534  	if deltas[len(deltas)-1].Type != Deleted {
   535  		t.Errorf("unexpected delta: %v", deltas[len(deltas)-1])
   536  	}
   537  }
   538  
   539  func TestDeltaFIFO_ReplaceMakesDeletions(t *testing.T) {
   540  	// We test with only one pre-existing object because there is no
   541  	// promise about how their deletes are ordered.
   542  
   543  	// Try it with a pre-existing Delete
   544  	f := NewDeltaFIFOWithOptions(DeltaFIFOOptions{
   545  		KeyFunction: testFifoObjectKeyFunc,
   546  		KnownObjects: literalListerGetter(func() []testFifoObject {
   547  			return []testFifoObject{mkFifoObj("foo", 5), mkFifoObj("bar", 6), mkFifoObj("baz", 7)}
   548  		}),
   549  	})
   550  	f.Delete(mkFifoObj("baz", 10))
   551  	f.Replace([]interface{}{mkFifoObj("foo", 5)}, "0")
   552  
   553  	expectedList := []Deltas{
   554  		{{Deleted, mkFifoObj("baz", 10)}},
   555  		{{Sync, mkFifoObj("foo", 5)}},
   556  		// Since "bar" didn't have a delete event and wasn't in the Replace list
   557  		// it should get a tombstone key with the right Obj.
   558  		{{Deleted, DeletedFinalStateUnknown{Key: "bar", Obj: mkFifoObj("bar", 6)}}},
   559  	}
   560  
   561  	for _, expected := range expectedList {
   562  		cur := Pop(f).(Deltas)
   563  		if e, a := expected, cur; !reflect.DeepEqual(e, a) {
   564  			t.Errorf("Expected %#v, got %#v", e, a)
   565  		}
   566  	}
   567  
   568  	// Now try starting with an Add instead of a Delete
   569  	f = NewDeltaFIFOWithOptions(DeltaFIFOOptions{
   570  		KeyFunction: testFifoObjectKeyFunc,
   571  		KnownObjects: literalListerGetter(func() []testFifoObject {
   572  			return []testFifoObject{mkFifoObj("foo", 5), mkFifoObj("bar", 6), mkFifoObj("baz", 7)}
   573  		}),
   574  	})
   575  	f.Add(mkFifoObj("baz", 10))
   576  	f.Replace([]interface{}{mkFifoObj("foo", 5)}, "0")
   577  
   578  	expectedList = []Deltas{
   579  		{{Added, mkFifoObj("baz", 10)},
   580  			{Deleted, DeletedFinalStateUnknown{Key: "baz", Obj: mkFifoObj("baz", 10)}}},
   581  		{{Sync, mkFifoObj("foo", 5)}},
   582  		// Since "bar" didn't have a delete event and wasn't in the Replace list
   583  		// it should get a tombstone key with the right Obj.
   584  		{{Deleted, DeletedFinalStateUnknown{Key: "bar", Obj: mkFifoObj("bar", 6)}}},
   585  	}
   586  
   587  	for _, expected := range expectedList {
   588  		cur := Pop(f).(Deltas)
   589  		if e, a := expected, cur; !reflect.DeepEqual(e, a) {
   590  			t.Errorf("Expected %#v, got %#v", e, a)
   591  		}
   592  	}
   593  
   594  	// Now try deleting and recreating the object in the queue, then delete it by a Replace call
   595  	f = NewDeltaFIFOWithOptions(DeltaFIFOOptions{
   596  		KeyFunction: testFifoObjectKeyFunc,
   597  		KnownObjects: literalListerGetter(func() []testFifoObject {
   598  			return []testFifoObject{mkFifoObj("foo", 5), mkFifoObj("bar", 6), mkFifoObj("baz", 7)}
   599  		}),
   600  	})
   601  	f.Delete(mkFifoObj("bar", 6))
   602  	f.Add(mkFifoObj("bar", 100))
   603  	f.Replace([]interface{}{mkFifoObj("foo", 5)}, "0")
   604  
   605  	expectedList = []Deltas{
   606  		{
   607  			{Deleted, mkFifoObj("bar", 6)},
   608  			{Added, mkFifoObj("bar", 100)},
   609  			// Since "bar" has a newer object in the queue than in the state,
   610  			// it should get a tombstone key with the latest object from the queue
   611  			{Deleted, DeletedFinalStateUnknown{Key: "bar", Obj: mkFifoObj("bar", 100)}},
   612  		},
   613  		{{Sync, mkFifoObj("foo", 5)}},
   614  		{{Deleted, DeletedFinalStateUnknown{Key: "baz", Obj: mkFifoObj("baz", 7)}}},
   615  	}
   616  
   617  	for _, expected := range expectedList {
   618  		cur := Pop(f).(Deltas)
   619  		if e, a := expected, cur; !reflect.DeepEqual(e, a) {
   620  			t.Errorf("Expected %#v, got %#v", e, a)
   621  		}
   622  	}
   623  
   624  	// Now try syncing it first to ensure the delete use the latest version
   625  	f = NewDeltaFIFOWithOptions(DeltaFIFOOptions{
   626  		KeyFunction: testFifoObjectKeyFunc,
   627  		KnownObjects: literalListerGetter(func() []testFifoObject {
   628  			return []testFifoObject{mkFifoObj("foo", 5), mkFifoObj("bar", 6), mkFifoObj("baz", 7)}
   629  		}),
   630  	})
   631  	f.Replace([]interface{}{mkFifoObj("bar", 100), mkFifoObj("foo", 5)}, "0")
   632  	f.Replace([]interface{}{mkFifoObj("foo", 5)}, "0")
   633  
   634  	expectedList = []Deltas{
   635  		{
   636  			{Sync, mkFifoObj("bar", 100)},
   637  			// Since "bar" didn't have a delete event and wasn't in the Replace list
   638  			// it should get a tombstone key with the right Obj.
   639  			{Deleted, DeletedFinalStateUnknown{Key: "bar", Obj: mkFifoObj("bar", 100)}},
   640  		},
   641  		{
   642  			{Sync, mkFifoObj("foo", 5)},
   643  			{Sync, mkFifoObj("foo", 5)},
   644  		},
   645  		{{Deleted, DeletedFinalStateUnknown{Key: "baz", Obj: mkFifoObj("baz", 7)}}},
   646  	}
   647  
   648  	for _, expected := range expectedList {
   649  		cur := Pop(f).(Deltas)
   650  		if e, a := expected, cur; !reflect.DeepEqual(e, a) {
   651  			t.Errorf("Expected %#v, got %#v", e, a)
   652  		}
   653  	}
   654  
   655  	// Now try starting without an explicit KeyListerGetter
   656  	f = NewDeltaFIFOWithOptions(DeltaFIFOOptions{KeyFunction: testFifoObjectKeyFunc})
   657  	f.Add(mkFifoObj("baz", 10))
   658  	f.Replace([]interface{}{mkFifoObj("foo", 5)}, "0")
   659  
   660  	expectedList = []Deltas{
   661  		{{Added, mkFifoObj("baz", 10)},
   662  			{Deleted, DeletedFinalStateUnknown{Key: "baz", Obj: mkFifoObj("baz", 10)}}},
   663  		{{Sync, mkFifoObj("foo", 5)}},
   664  	}
   665  
   666  	for _, expected := range expectedList {
   667  		cur := Pop(f).(Deltas)
   668  		if e, a := expected, cur; !reflect.DeepEqual(e, a) {
   669  			t.Errorf("Expected %#v, got %#v", e, a)
   670  		}
   671  	}
   672  }
   673  
   674  // TestDeltaFIFO_ReplaceMakesDeletionsReplaced is the same as the above test, but
   675  // ensures that a Replaced DeltaType is emitted.
   676  func TestDeltaFIFO_ReplaceMakesDeletionsReplaced(t *testing.T) {
   677  	f := NewDeltaFIFOWithOptions(DeltaFIFOOptions{
   678  		KeyFunction: testFifoObjectKeyFunc,
   679  		KnownObjects: literalListerGetter(func() []testFifoObject {
   680  			return []testFifoObject{mkFifoObj("foo", 5), mkFifoObj("bar", 6), mkFifoObj("baz", 7)}
   681  		}),
   682  		EmitDeltaTypeReplaced: true,
   683  	})
   684  
   685  	f.Delete(mkFifoObj("baz", 10))
   686  	f.Replace([]interface{}{mkFifoObj("foo", 6)}, "0")
   687  
   688  	expectedList := []Deltas{
   689  		{{Deleted, mkFifoObj("baz", 10)}},
   690  		{{Replaced, mkFifoObj("foo", 6)}},
   691  		// Since "bar" didn't have a delete event and wasn't in the Replace list
   692  		// it should get a tombstone key with the right Obj.
   693  		{{Deleted, DeletedFinalStateUnknown{Key: "bar", Obj: mkFifoObj("bar", 6)}}},
   694  	}
   695  
   696  	for _, expected := range expectedList {
   697  		cur := Pop(f).(Deltas)
   698  		if e, a := expected, cur; !reflect.DeepEqual(e, a) {
   699  			t.Errorf("Expected %#v, got %#v", e, a)
   700  		}
   701  	}
   702  }
   703  
   704  // TestDeltaFIFO_ReplaceDeltaType checks that passing EmitDeltaTypeReplaced
   705  // means that Replaced is correctly emitted.
   706  func TestDeltaFIFO_ReplaceDeltaType(t *testing.T) {
   707  	f := NewDeltaFIFOWithOptions(DeltaFIFOOptions{
   708  		KeyFunction: testFifoObjectKeyFunc,
   709  		KnownObjects: literalListerGetter(func() []testFifoObject {
   710  			return []testFifoObject{mkFifoObj("foo", 5)}
   711  		}),
   712  		EmitDeltaTypeReplaced: true,
   713  	})
   714  	f.Replace([]interface{}{mkFifoObj("foo", 5)}, "0")
   715  
   716  	expectedList := []Deltas{
   717  		{{Replaced, mkFifoObj("foo", 5)}},
   718  	}
   719  
   720  	for _, expected := range expectedList {
   721  		cur := Pop(f).(Deltas)
   722  		if e, a := expected, cur; !reflect.DeepEqual(e, a) {
   723  			t.Errorf("Expected %#v, got %#v", e, a)
   724  		}
   725  	}
   726  }
   727  
   728  func TestDeltaFIFO_UpdateResyncRace(t *testing.T) {
   729  	f := NewDeltaFIFOWithOptions(DeltaFIFOOptions{
   730  		KeyFunction: testFifoObjectKeyFunc,
   731  		KnownObjects: literalListerGetter(func() []testFifoObject {
   732  			return []testFifoObject{mkFifoObj("foo", 5)}
   733  		}),
   734  	})
   735  	f.Update(mkFifoObj("foo", 6))
   736  	f.Resync()
   737  
   738  	expectedList := []Deltas{
   739  		{{Updated, mkFifoObj("foo", 6)}},
   740  	}
   741  
   742  	for _, expected := range expectedList {
   743  		cur := Pop(f).(Deltas)
   744  		if e, a := expected, cur; !reflect.DeepEqual(e, a) {
   745  			t.Errorf("Expected %#v, got %#v", e, a)
   746  		}
   747  	}
   748  }
   749  
   750  // pop2 captures both parameters, unlike Pop().
   751  func pop2[T any](queue Queue) (T, bool) {
   752  	var result interface{}
   753  	var isList bool
   754  	queue.Pop(func(obj interface{}, isInInitialList bool) error {
   755  		result = obj
   756  		isList = isInInitialList
   757  		return nil
   758  	})
   759  	return result.(T), isList
   760  }
   761  
   762  func TestDeltaFIFO_HasSyncedCorrectOnDeletion(t *testing.T) {
   763  	f := NewDeltaFIFOWithOptions(DeltaFIFOOptions{
   764  		KeyFunction: testFifoObjectKeyFunc,
   765  		KnownObjects: literalListerGetter(func() []testFifoObject {
   766  			return []testFifoObject{mkFifoObj("foo", 5), mkFifoObj("bar", 6), mkFifoObj("baz", 7)}
   767  		}),
   768  	})
   769  	f.Replace([]interface{}{mkFifoObj("foo", 5)}, "0")
   770  
   771  	expectedList := []Deltas{
   772  		{{Sync, mkFifoObj("foo", 5)}},
   773  		// Since "bar" didn't have a delete event and wasn't in the Replace list
   774  		// it should get a tombstone key with the right Obj.
   775  		{{Deleted, DeletedFinalStateUnknown{Key: "bar", Obj: mkFifoObj("bar", 6)}}},
   776  		{{Deleted, DeletedFinalStateUnknown{Key: "baz", Obj: mkFifoObj("baz", 7)}}},
   777  	}
   778  
   779  	for _, expected := range expectedList {
   780  		if f.HasSynced() {
   781  			t.Errorf("Expected HasSynced to be false")
   782  		}
   783  		cur, initial := pop2[Deltas](f)
   784  		if e, a := expected, cur; !reflect.DeepEqual(e, a) {
   785  			t.Errorf("Expected %#v, got %#v", e, a)
   786  		}
   787  		if initial != true {
   788  			t.Error("Expected initial list item")
   789  		}
   790  	}
   791  	if !f.HasSynced() {
   792  		t.Errorf("Expected HasSynced to be true")
   793  	}
   794  }
   795  
   796  func TestDeltaFIFO_detectLineJumpers(t *testing.T) {
   797  	f := NewDeltaFIFOWithOptions(DeltaFIFOOptions{KeyFunction: testFifoObjectKeyFunc})
   798  
   799  	f.Add(mkFifoObj("foo", 10))
   800  	f.Add(mkFifoObj("bar", 1))
   801  	f.Add(mkFifoObj("foo", 11))
   802  	f.Add(mkFifoObj("foo", 13))
   803  	f.Add(mkFifoObj("zab", 30))
   804  
   805  	if e, a := 13, testPop(f).val; a != e {
   806  		t.Fatalf("expected %d, got %d", e, a)
   807  	}
   808  
   809  	f.Add(mkFifoObj("foo", 14)) // ensure foo doesn't jump back in line
   810  
   811  	if e, a := 1, testPop(f).val; a != e {
   812  		t.Fatalf("expected %d, got %d", e, a)
   813  	}
   814  
   815  	if e, a := 30, testPop(f).val; a != e {
   816  		t.Fatalf("expected %d, got %d", e, a)
   817  	}
   818  
   819  	if e, a := 14, testPop(f).val; a != e {
   820  		t.Fatalf("expected %d, got %d", e, a)
   821  	}
   822  }
   823  
   824  func TestDeltaFIFO_addIfNotPresent(t *testing.T) {
   825  	f := NewDeltaFIFOWithOptions(DeltaFIFOOptions{KeyFunction: testFifoObjectKeyFunc})
   826  
   827  	emptyDeltas := Deltas{}
   828  	if err := f.AddIfNotPresent(emptyDeltas); err == nil || !errors.Is(err, ErrZeroLengthDeltasObject) {
   829  		t.Errorf("Expected error '%v', got %v", ErrZeroLengthDeltasObject, err)
   830  	}
   831  
   832  	f.Add(mkFifoObj("b", 3))
   833  	b3 := Pop(f)
   834  	f.Add(mkFifoObj("c", 4))
   835  	c4 := Pop(f)
   836  	if e, a := 0, len(f.items); e != a {
   837  		t.Fatalf("Expected %v, got %v items in queue", e, a)
   838  	}
   839  
   840  	f.Add(mkFifoObj("a", 1))
   841  	f.Add(mkFifoObj("b", 2))
   842  	f.AddIfNotPresent(b3)
   843  	f.AddIfNotPresent(c4)
   844  
   845  	if e, a := 3, len(f.items); a != e {
   846  		t.Fatalf("expected queue length %d, got %d", e, a)
   847  	}
   848  
   849  	expectedValues := []int{1, 2, 4}
   850  	for _, expected := range expectedValues {
   851  		if actual := testPop(f).val; actual != expected {
   852  			t.Fatalf("expected value %d, got %d", expected, actual)
   853  		}
   854  	}
   855  }
   856  
   857  func TestDeltaFIFO_KeyOf(t *testing.T) {
   858  	f := DeltaFIFO{keyFunc: testFifoObjectKeyFunc}
   859  
   860  	table := []struct {
   861  		obj interface{}
   862  		key string
   863  	}{
   864  		{obj: testFifoObject{name: "A"}, key: "A"},
   865  		{obj: DeletedFinalStateUnknown{Key: "B", Obj: nil}, key: "B"},
   866  		{obj: Deltas{{Object: testFifoObject{name: "C"}}}, key: "C"},
   867  		{obj: Deltas{{Object: DeletedFinalStateUnknown{Key: "D", Obj: nil}}}, key: "D"},
   868  	}
   869  
   870  	for _, item := range table {
   871  		got, err := f.KeyOf(item.obj)
   872  		if err != nil {
   873  			t.Errorf("Unexpected error for %q: %v", item.obj, err)
   874  			continue
   875  		}
   876  		if e, a := item.key, got; e != a {
   877  			t.Errorf("Expected %v, got %v", e, a)
   878  		}
   879  	}
   880  }
   881  
   882  func TestDeltaFIFO_HasSynced(t *testing.T) {
   883  	tests := []struct {
   884  		actions        []func(f *DeltaFIFO)
   885  		expectedSynced bool
   886  	}{
   887  		{
   888  			actions:        []func(f *DeltaFIFO){},
   889  			expectedSynced: false,
   890  		},
   891  		{
   892  			actions: []func(f *DeltaFIFO){
   893  				func(f *DeltaFIFO) { f.Add(mkFifoObj("a", 1)) },
   894  			},
   895  			expectedSynced: true,
   896  		},
   897  		{
   898  			actions: []func(f *DeltaFIFO){
   899  				func(f *DeltaFIFO) { f.Replace([]interface{}{}, "0") },
   900  			},
   901  			expectedSynced: true,
   902  		},
   903  		{
   904  			actions: []func(f *DeltaFIFO){
   905  				func(f *DeltaFIFO) { f.Replace([]interface{}{mkFifoObj("a", 1), mkFifoObj("b", 2)}, "0") },
   906  			},
   907  			expectedSynced: false,
   908  		},
   909  		{
   910  			actions: []func(f *DeltaFIFO){
   911  				func(f *DeltaFIFO) { f.Replace([]interface{}{mkFifoObj("a", 1), mkFifoObj("b", 2)}, "0") },
   912  				func(f *DeltaFIFO) { Pop(f) },
   913  			},
   914  			expectedSynced: false,
   915  		},
   916  		{
   917  			actions: []func(f *DeltaFIFO){
   918  				func(f *DeltaFIFO) { f.Replace([]interface{}{mkFifoObj("a", 1), mkFifoObj("b", 2)}, "0") },
   919  				func(f *DeltaFIFO) { Pop(f) },
   920  				func(f *DeltaFIFO) { Pop(f) },
   921  			},
   922  			expectedSynced: true,
   923  		},
   924  		{
   925  			// This test case won't happen in practice since a Reflector, the only producer for delta_fifo today, always passes a complete snapshot consistent in time;
   926  			// there cannot be duplicate keys in the list or apiserver is broken.
   927  			actions: []func(f *DeltaFIFO){
   928  				func(f *DeltaFIFO) { f.Replace([]interface{}{mkFifoObj("a", 1), mkFifoObj("a", 2)}, "0") },
   929  				func(f *DeltaFIFO) { Pop(f) },
   930  			},
   931  			expectedSynced: true,
   932  		},
   933  	}
   934  
   935  	for i, test := range tests {
   936  		f := NewDeltaFIFOWithOptions(DeltaFIFOOptions{KeyFunction: testFifoObjectKeyFunc})
   937  
   938  		for _, action := range test.actions {
   939  			action(f)
   940  		}
   941  		if e, a := test.expectedSynced, f.HasSynced(); a != e {
   942  			t.Errorf("test case %v failed, expected: %v , got %v", i, e, a)
   943  		}
   944  	}
   945  }
   946  
   947  // TestDeltaFIFO_PopShouldUnblockWhenClosed checks that any blocking Pop on an empty queue
   948  // should unblock and return after Close is called.
   949  func TestDeltaFIFO_PopShouldUnblockWhenClosed(t *testing.T) {
   950  	f := NewDeltaFIFOWithOptions(DeltaFIFOOptions{
   951  		KeyFunction: testFifoObjectKeyFunc,
   952  		KnownObjects: literalListerGetter(func() []testFifoObject {
   953  			return []testFifoObject{mkFifoObj("foo", 5)}
   954  		}),
   955  	})
   956  
   957  	c := make(chan struct{})
   958  	const jobs = 10
   959  	for i := 0; i < jobs; i++ {
   960  		go func() {
   961  			f.Pop(func(obj interface{}, isInInitialList bool) error {
   962  				return nil
   963  			})
   964  			c <- struct{}{}
   965  		}()
   966  	}
   967  
   968  	runtime.Gosched()
   969  	f.Close()
   970  
   971  	for i := 0; i < jobs; i++ {
   972  		select {
   973  		case <-c:
   974  		case <-time.After(500 * time.Millisecond):
   975  			t.Fatalf("timed out waiting for Pop to return after Close")
   976  		}
   977  	}
   978  }
   979  
   980  func BenchmarkDeltaFIFOListKeys(b *testing.B) {
   981  	f := NewDeltaFIFOWithOptions(DeltaFIFOOptions{KeyFunction: testFifoObjectKeyFunc})
   982  	const amount = 10000
   983  
   984  	for i := 0; i < amount; i++ {
   985  		f.Add(mkFifoObj(string([]rune{'a', rune(i)}), i+1))
   986  	}
   987  	for u := uint64(0); u < amount; u++ {
   988  		f.Add(mkFifoObj(string([]rune{'b', rune(u)}), u+1))
   989  	}
   990  
   991  	b.ResetTimer()
   992  	b.RunParallel(func(pb *testing.PB) {
   993  		for pb.Next() {
   994  			_ = f.ListKeys()
   995  		}
   996  	})
   997  	b.StopTimer()
   998  }
   999
View as plain text