github/client-go
1
0
Fork 0
mirror of https://github.com/kubernetes/client-go.git synced 2025-06-25 14:41:53 +00:00
Code Issues Projects Releases Wiki Activity

Merge pull request #129568 from deads2k/tighten-fifo-interface

Browse Source 
create RealFIFO implementation

Kubernetes-commit: a8ef3639f9db66d8c894a880fd35a9cd011b2cb9
This commit is contained in:
Kubernetes Publisher 2025-01-29 14:31:23 -08:00
commit 362c5e8de9
11 changed files with 1550 additions and 373 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
features/known_features.go
View File

@ -53,6 +53,12 @@ const (
// alpha: v1.30
InformerResourceVersion Feature = "InformerResourceVersion"
// owner: @deads2k
// beta: v1.33
//
// Refactor informers to deliver watch stream events in order instead of out of order.
InOrderInformers Feature = "InOrderInformers"
// owner: @p0lyn0mial
// beta: v1.30
//
@ -73,5 +79,6 @@ var defaultKubernetesFeatureGates = map[Feature]FeatureSpec{
ClientsAllowCBOR: {Default: false, PreRelease: Alpha},
ClientsPreferCBOR: {Default: false, PreRelease: Alpha},
InformerResourceVersion: {Default: false, PreRelease: Alpha},
InOrderInformers: {Default: true, PreRelease: Beta},
WatchListClient: {Default: false, PreRelease: Beta},
}

31
tools/cache/controller.go vendored
View File

@ -19,6 +19,7 @@ package cache
import (
"context"
"errors"
clientgofeaturegate "k8s.io/client-go/features"
"sync"
"time"
@ -72,13 +73,6 @@ type Config struct {
// resync.
ShouldResync ShouldResyncFunc
// If true, when Process() returns an error, re-enqueue the object.
// TODO: add interface to let you inject a delay/backoff or drop
// the object completely if desired. Pass the object in
// question to this interface as a parameter. This is probably moot
// now that this functionality appears at a higher level.
RetryOnError bool
// Called whenever the ListAndWatch drops the connection with an error.
//
// Contextual logging: WatchErrorHandlerWithContext should be used instead of WatchErrorHandler in code which supports contextual logging.
@ -213,15 +207,11 @@ func (c *controller) processLoop(ctx context.Context) {
// TODO: Plumb through the ctx so that this can
// actually exit when the controller is stopped. Or just give up on this stuff
// ever being stoppable.
obj, err := c.config.Queue.Pop(PopProcessFunc(c.config.Process))
_, err := c.config.Queue.Pop(PopProcessFunc(c.config.Process))
if err != nil {
if err == ErrFIFOClosed {
return
}
if c.config.RetryOnError {
// This is the safe way to re-enqueue.
c.config.Queue.AddIfNotPresent(obj)
}
}
}
}
@ -603,11 +593,17 @@ func newInformer(clientState Store, options InformerOptions) Controller {
// This will hold incoming changes. Note how we pass clientState in as a
// KeyLister, that way resync operations will result in the correct set
// of update/delete deltas.
fifo := NewDeltaFIFOWithOptions(DeltaFIFOOptions{
KnownObjects: clientState,
EmitDeltaTypeReplaced: true,
Transformer: options.Transform,
})
var fifo Queue
if clientgofeaturegate.FeatureGates().Enabled(clientgofeaturegate.InOrderInformers) {
fifo = NewRealFIFO(MetaNamespaceKeyFunc, clientState, options.Transform)
} else {
fifo = NewDeltaFIFOWithOptions(DeltaFIFOOptions{
KnownObjects: clientState,
EmitDeltaTypeReplaced: true,
Transformer: options.Transform,
})
}
cfg := &Config{
Queue: fifo,
@ -615,7 +611,6 @@ func newInformer(clientState Store, options InformerOptions) Controller {
ObjectType: options.ObjectType,
FullResyncPeriod: options.ResyncPeriod,
MinWatchTimeout: options.MinWatchTimeout,
RetryOnError: false,
Process: func(obj interface{}, isInInitialList bool) error {
if deltas, ok := obj.(Deltas); ok {

8
tools/cache/controller_test.go vendored
View File

@ -49,10 +49,7 @@ func Example() {
// This will hold incoming changes. Note how we pass downstream in as a
// KeyLister, that way resync operations will result in the correct set
// of update/delete deltas.
fifo := NewDeltaFIFOWithOptions(DeltaFIFOOptions{
KeyFunction: MetaNamespaceKeyFunc,
KnownObjects: downstream,
})
fifo := NewRealFIFO(MetaNamespaceKeyFunc, downstream, nil)
// Let's do threadsafe output to get predictable test results.
deletionCounter := make(chan string, 1000)
@ -62,7 +59,6 @@ func Example() {
ListerWatcher: source,
ObjectType: &v1.Pod{},
FullResyncPeriod: time.Millisecond * 100,
RetryOnError: false,
// Let's implement a simple controller that just deletes
// everything that comes in.
@ -88,7 +84,7 @@ func Example() {
// fifo's KeyOf is easiest, because it handles
// DeletedFinalStateUnknown markers.
key, err := fifo.KeyOf(newest.Object)
key, err := fifo.keyOf(newest.Object)
if err != nil {
return err
}

100
tools/cache/delta_fifo.go vendored
View File

@ -369,43 +369,6 @@ func (f *DeltaFIFO) Delete(obj interface{}) error {
return f.queueActionLocked(Deleted, obj)
}
// AddIfNotPresent inserts an item, and puts it in the queue. If the item is already
// present in the set, it is neither enqueued nor added to the set.
//
// This is useful in a single producer/consumer scenario so that the consumer can
// safely retry items without contending with the producer and potentially enqueueing
// stale items.
//
// Important: obj must be a Deltas (the output of the Pop() function). Yes, this is
// different from the Add/Update/Delete functions.
func (f *DeltaFIFO) AddIfNotPresent(obj interface{}) error {
deltas, ok := obj.(Deltas)
if !ok {
return fmt.Errorf("object must be of type deltas, but got: %#v", obj)
}
id, err := f.KeyOf(deltas)
if err != nil {
return KeyError{obj, err}
}
f.lock.Lock()
defer f.lock.Unlock()
f.addIfNotPresent(id, deltas)
return nil
}
// addIfNotPresent inserts deltas under id if it does not exist, and assumes the caller
// already holds the fifo lock.
func (f *DeltaFIFO) addIfNotPresent(id string, deltas Deltas) {
f.populated = true
if _, exists := f.items[id]; exists {
return
}
f.queue = append(f.queue, id)
f.items[id] = deltas
f.cond.Broadcast()
}
// re-listing and watching can deliver the same update multiple times in any
// order. This will combine the most recent two deltas if they are the same.
func dedupDeltas(deltas Deltas) Deltas {
@ -508,61 +471,6 @@ func (f *DeltaFIFO) queueActionInternalLocked(actionType, internalActionType Del
return nil
}
// List returns a list of all the items; it returns the object
// from the most recent Delta.
// You should treat the items returned inside the deltas as immutable.
func (f *DeltaFIFO) List() []interface{} {
f.lock.RLock()
defer f.lock.RUnlock()
return f.listLocked()
}
func (f *DeltaFIFO) listLocked() []interface{} {
list := make([]interface{}, 0, len(f.items))
for _, item := range f.items {
list = append(list, item.Newest().Object)
}
return list
}
// ListKeys returns a list of all the keys of the objects currently
// in the FIFO.
func (f *DeltaFIFO) ListKeys() []string {
f.lock.RLock()
defer f.lock.RUnlock()
list := make([]string, 0, len(f.queue))
for _, key := range f.queue {
list = append(list, key)
}
return list
}
// Get returns the complete list of deltas for the requested item,
// or sets exists=false.
// You should treat the items returned inside the deltas as immutable.
func (f *DeltaFIFO) Get(obj interface{}) (item interface{}, exists bool, err error) {
key, err := f.KeyOf(obj)
if err != nil {
return nil, false, KeyError{obj, err}
}
return f.GetByKey(key)
}
// GetByKey returns the complete list of deltas for the requested item,
// setting exists=false if that list is empty.
// You should treat the items returned inside the deltas as immutable.
func (f *DeltaFIFO) GetByKey(key string) (item interface{}, exists bool, err error) {
f.lock.RLock()
defer f.lock.RUnlock()
d, exists := f.items[key]
if exists {
// Copy item's slice so operations on this slice
// won't interfere with the object we return.
d = copyDeltas(d)
}
return d, exists, nil
}
// IsClosed checks if the queue is closed
func (f *DeltaFIFO) IsClosed() bool {
f.lock.Lock()
@ -576,9 +484,7 @@ func (f *DeltaFIFO) IsClosed() bool {
// is returned, so if you don't successfully process it, you need to add it back
// with AddIfNotPresent().
// process function is called under lock, so it is safe to update data structures
// in it that need to be in sync with the queue (e.g. knownKeys). The PopProcessFunc
// may return an instance of ErrRequeue with a nested error to indicate the current
// item should be requeued (equivalent to calling AddIfNotPresent under the lock).
// in it that need to be in sync with the queue (e.g. knownKeys).
// process should avoid expensive I/O operation so that other queue operations, i.e.
// Add() and Get(), won't be blocked for too long.
//
@ -625,10 +531,6 @@ func (f *DeltaFIFO) Pop(process PopProcessFunc) (interface{}, error) {
defer trace.LogIfLong(100 * time.Millisecond)
}
err := process(item, isInInitialList)
if e, ok := err.(ErrRequeue); ok {
f.addIfNotPresent(id, item)
err = e.Err
}
// Don't need to copyDeltas here, because we're transferring
// ownership to the caller.
return item, err

138
tools/cache/delta_fifo_test.go vendored
View File

@ -17,7 +17,6 @@ limitations under the License.
package cache
import (
"errors"
"fmt"
"reflect"
"runtime"
@ -25,6 +24,66 @@ import (
"time"
)
// List returns a list of all the items; it returns the object
// from the most recent Delta.
// You should treat the items returned inside the deltas as immutable.
// This function was moved here because it is not consistent with normal list semantics, but is used in unit testing.
func (f *DeltaFIFO) List() []interface{} {
f.lock.RLock()
defer f.lock.RUnlock()
return f.listLocked()
}
// This function was moved here because it is not consistent with normal list semantics, but is used in unit testing.
func (f *DeltaFIFO) listLocked() []interface{} {
list := make([]interface{}, 0, len(f.items))
for _, item := range f.items {
list = append(list, item.Newest().Object)
}
return list
}
// ListKeys returns a list of all the keys of the objects currently
// in the FIFO.
// This function was moved here because it is not consistent with normal list semantics, but is used in unit testing.
func (f *DeltaFIFO) ListKeys() []string {
f.lock.RLock()
defer f.lock.RUnlock()
list := make([]string, 0, len(f.queue))
for _, key := range f.queue {
list = append(list, key)
}
return list
}
// Get returns the complete list of deltas for the requested item,
// or sets exists=false.
// You should treat the items returned inside the deltas as immutable.
// This function was moved here because it is not consistent with normal list semantics, but is used in unit testing.
func (f *DeltaFIFO) Get(obj interface{}) (item interface{}, exists bool, err error) {
key, err := f.KeyOf(obj)
if err != nil {
return nil, false, KeyError{obj, err}
}
return f.GetByKey(key)
}
// GetByKey returns the complete list of deltas for the requested item,
// setting exists=false if that list is empty.
// You should treat the items returned inside the deltas as immutable.
// This function was moved here because it is not consistent with normal list semantics, but is used in unit testing.
func (f *DeltaFIFO) GetByKey(key string) (item interface{}, exists bool, err error) {
f.lock.RLock()
defer f.lock.RUnlock()
d, exists := f.items[key]
if exists {
// Copy item's slice so operations on this slice
// won't interfere with the object we return.
d = copyDeltas(d)
}
return d, exists, nil
}
// helper function to reduce stuttering
func testPop(f *DeltaFIFO) testFifoObject {
return Pop(f).(Deltas).Newest().Object.(testFifoObject)
@ -245,50 +304,6 @@ func TestDeltaFIFOW_ReplaceMakesDeletionsForObjectsOnlyInQueue(t *testing.T) {
}
}
func TestDeltaFIFO_requeueOnPop(t *testing.T) {
f := NewDeltaFIFOWithOptions(DeltaFIFOOptions{KeyFunction: testFifoObjectKeyFunc})
f.Add(mkFifoObj("foo", 10))
_, err := f.Pop(func(obj interface{}, isInInitialList bool) error {
if obj.(Deltas)[0].Object.(testFifoObject).name != "foo" {
t.Fatalf("unexpected object: %#v", obj)
}
return ErrRequeue{Err: nil}
})
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if _, ok, err := f.GetByKey("foo"); !ok || err != nil {
t.Fatalf("object should have been requeued: %t %v", ok, err)
}
_, err = f.Pop(func(obj interface{}, isInInitialList bool) error {
if obj.(Deltas)[0].Object.(testFifoObject).name != "foo" {
t.Fatalf("unexpected object: %#v", obj)
}
return ErrRequeue{Err: fmt.Errorf("test error")}
})
if err == nil || err.Error() != "test error" {
t.Fatalf("unexpected error: %v", err)
}
if _, ok, err := f.GetByKey("foo"); !ok || err != nil {
t.Fatalf("object should have been requeued: %t %v", ok, err)
}
_, err = f.Pop(func(obj interface{}, isInInitialList bool) error {
if obj.(Deltas)[0].Object.(testFifoObject).name != "foo" {
t.Fatalf("unexpected object: %#v", obj)
}
return nil
})
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if _, ok, err := f.GetByKey("foo"); ok || err != nil {
t.Fatalf("object should have been removed: %t %v", ok, err)
}
}
func TestDeltaFIFO_addUpdate(t *testing.T) {
f := NewDeltaFIFOWithOptions(DeltaFIFOOptions{KeyFunction: testFifoObjectKeyFunc})
f.Add(mkFifoObj("foo", 10))
@ -821,39 +836,6 @@ func TestDeltaFIFO_detectLineJumpers(t *testing.T) {
}
}
func TestDeltaFIFO_addIfNotPresent(t *testing.T) {
f := NewDeltaFIFOWithOptions(DeltaFIFOOptions{KeyFunction: testFifoObjectKeyFunc})
emptyDeltas := Deltas{}
if err := f.AddIfNotPresent(emptyDeltas); err == nil || !errors.Is(err, ErrZeroLengthDeltasObject) {
t.Errorf("Expected error '%v', got %v", ErrZeroLengthDeltasObject, err)
}
f.Add(mkFifoObj("b", 3))
b3 := Pop(f)
f.Add(mkFifoObj("c", 4))
c4 := Pop(f)
if e, a := 0, len(f.items); e != a {
t.Fatalf("Expected %v, got %v items in queue", e, a)
}
f.Add(mkFifoObj("a", 1))
f.Add(mkFifoObj("b", 2))
f.AddIfNotPresent(b3)
f.AddIfNotPresent(c4)
if e, a := 3, len(f.items); a != e {
t.Fatalf("expected queue length %d, got %d", e, a)
}
expectedValues := []int{1, 2, 4}
for _, expected := range expectedValues {
if actual := testPop(f).val; actual != expected {
t.Fatalf("expected value %d, got %d", expected, actual)
}
}
}
func TestDeltaFIFO_KeyOf(t *testing.T) {
f := DeltaFIFO{keyFunc: testFifoObjectKeyFunc}

97
tools/cache/fifo.go vendored
View File

@ -27,30 +27,16 @@ import (
// It is supposed to process the accumulator popped from the queue.
type PopProcessFunc func(obj interface{}, isInInitialList bool) error
// ErrRequeue may be returned by a PopProcessFunc to safely requeue
// the current item. The value of Err will be returned from Pop.
type ErrRequeue struct {
// Err is returned by the Pop function
Err error
}
// ErrFIFOClosed used when FIFO is closed
var ErrFIFOClosed = errors.New("DeltaFIFO: manipulating with closed queue")
func (e ErrRequeue) Error() string {
if e.Err == nil {
return "the popped item should be requeued without returning an error"
}
return e.Err.Error()
}
// Queue extends Store with a collection of Store keys to "process".
// Queue extends ReflectorStore with a collection of Store keys to "process".
// Every Add, Update, or Delete may put the object's key in that collection.
// A Queue has a way to derive the corresponding key given an accumulator.
// A Queue can be accessed concurrently from multiple goroutines.
// A Queue can be "closed", after which Pop operations return an error.
type Queue interface {
Store
ReflectorStore
// Pop blocks until there is at least one key to process or the
// Queue is closed. In the latter case Pop returns with an error.
@ -64,11 +50,6 @@ type Queue interface {
// Pop.
Pop(PopProcessFunc) (interface{}, error)
// AddIfNotPresent puts the given accumulator into the Queue (in
// association with the accumulator's key) if and only if that key
// is not already associated with a non-empty accumulator.
AddIfNotPresent(interface{}) error
// HasSynced returns true if the first batch of keys have all been
// popped. The first batch of keys are those of the first Replace
// operation if that happened before any Add, AddIfNotPresent,
@ -177,36 +158,6 @@ func (f *FIFO) Add(obj interface{}) error {
return nil
}
// AddIfNotPresent inserts an item, and puts it in the queue. If the item is already
// present in the set, it is neither enqueued nor added to the set.
//
// This is useful in a single producer/consumer scenario so that the consumer can
// safely retry items without contending with the producer and potentially enqueueing
// stale items.
func (f *FIFO) AddIfNotPresent(obj interface{}) error {
id, err := f.keyFunc(obj)
if err != nil {
return KeyError{obj, err}
}
f.lock.Lock()
defer f.lock.Unlock()
f.addIfNotPresent(id, obj)
return nil
}
// addIfNotPresent assumes the fifo lock is already held and adds the provided
// item to the queue under id if it does not already exist.
func (f *FIFO) addIfNotPresent(id string, obj interface{}) {
f.populated = true
if _, exists := f.items[id]; exists {
return
}
f.queue = append(f.queue, id)
f.items[id] = obj
f.cond.Broadcast()
}
// Update is the same as Add in this implementation.
func (f *FIFO) Update(obj interface{}) error {
return f.Add(obj)
@ -227,46 +178,6 @@ func (f *FIFO) Delete(obj interface{}) error {
return err
}
// List returns a list of all the items.
func (f *FIFO) List() []interface{} {
f.lock.RLock()
defer f.lock.RUnlock()
list := make([]interface{}, 0, len(f.items))
for _, item := range f.items {
list = append(list, item)
}
return list
}
// ListKeys returns a list of all the keys of the objects currently
// in the FIFO.
func (f *FIFO) ListKeys() []string {
f.lock.RLock()
defer f.lock.RUnlock()
list := make([]string, 0, len(f.items))
for key := range f.items {
list = append(list, key)
}
return list
}
// Get returns the requested item, or sets exists=false.
func (f *FIFO) Get(obj interface{}) (item interface{}, exists bool, err error) {
key, err := f.keyFunc(obj)
if err != nil {
return nil, false, KeyError{obj, err}
}
return f.GetByKey(key)
}
// GetByKey returns the requested item, or sets exists=false.
func (f *FIFO) GetByKey(key string) (item interface{}, exists bool, err error) {
f.lock.RLock()
defer f.lock.RUnlock()
item, exists = f.items[key]
return item, exists, nil
}
// IsClosed checks if the queue is closed
func (f *FIFO) IsClosed() bool {
f.lock.Lock()
@ -307,10 +218,6 @@ func (f *FIFO) Pop(process PopProcessFunc) (interface{}, error) {
}
delete(f.items, id)
err := process(item, isInInitialList)
if e, ok := err.(ErrRequeue); ok {
f.addIfNotPresent(id, item)
err = e.Err
}
return item, err
}
}

109
tools/cache/fifo_test.go vendored
View File

@ -17,13 +17,56 @@ limitations under the License.
package cache
import (
"fmt"
"reflect"
"runtime"
"testing"
"time"
)
// List returns a list of all the items.
// This function was moved here because it is not consistent with normal list semantics, but is used in unit testing.
func (f *FIFO) List() []interface{} {
f.lock.RLock()
defer f.lock.RUnlock()
list := make([]interface{}, 0, len(f.items))
for _, item := range f.items {
list = append(list, item)
}
return list
}
// ListKeys returns a list of all the keys of the objects currently
// in the FIFO.
// This function was moved here because it is not consistent with normal list semantics, but is used in unit testing.
func (f *FIFO) ListKeys() []string {
f.lock.RLock()
defer f.lock.RUnlock()
list := make([]string, 0, len(f.items))
for key := range f.items {
list = append(list, key)
}
return list
}
// Get returns the requested item, or sets exists=false.
// This function was moved here because it is not consistent with normal list semantics, but is used in unit testing.
func (f *FIFO) Get(obj interface{}) (item interface{}, exists bool, err error) {
key, err := f.keyFunc(obj)
if err != nil {
return nil, false, KeyError{obj, err}
}
return f.GetByKey(key)
}
// GetByKey returns the requested item, or sets exists=false.
// This function was moved here because it is not consistent with normal list semantics, but is used in unit testing.
func (f *FIFO) GetByKey(key string) (item interface{}, exists bool, err error) {
f.lock.RLock()
defer f.lock.RUnlock()
item, exists = f.items[key]
return item, exists, nil
}
func testFifoObjectKeyFunc(obj interface{}) (string, error) {
return obj.(testFifoObject).name, nil
}
@ -72,50 +115,6 @@ func TestFIFO_basic(t *testing.T) {
}
}
func TestFIFO_requeueOnPop(t *testing.T) {
f := NewFIFO(testFifoObjectKeyFunc)
f.Add(mkFifoObj("foo", 10))
_, err := f.Pop(func(obj interface{}, isInInitialList bool) error {
if obj.(testFifoObject).name != "foo" {
t.Fatalf("unexpected object: %#v", obj)
}
return ErrRequeue{Err: nil}
})
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if _, ok, err := f.GetByKey("foo"); !ok || err != nil {
t.Fatalf("object should have been requeued: %t %v", ok, err)
}
_, err = f.Pop(func(obj interface{}, isInInitialList bool) error {
if obj.(testFifoObject).name != "foo" {
t.Fatalf("unexpected object: %#v", obj)
}
return ErrRequeue{Err: fmt.Errorf("test error")}
})
if err == nil || err.Error() != "test error" {
t.Fatalf("unexpected error: %v", err)
}
if _, ok, err := f.GetByKey("foo"); !ok || err != nil {
t.Fatalf("object should have been requeued: %t %v", ok, err)
}
_, err = f.Pop(func(obj interface{}, isInInitialList bool) error {
if obj.(testFifoObject).name != "foo" {
t.Fatalf("unexpected object: %#v", obj)
}
return nil
})
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if _, ok, err := f.GetByKey("foo"); ok || err != nil {
t.Fatalf("object should have been removed: %t %v", ok, err)
}
}
func TestFIFO_addUpdate(t *testing.T) {
f := NewFIFO(testFifoObjectKeyFunc)
f.Add(mkFifoObj("foo", 10))
@ -204,26 +203,6 @@ func TestFIFO_detectLineJumpers(t *testing.T) {
}
}
func TestFIFO_addIfNotPresent(t *testing.T) {
f := NewFIFO(testFifoObjectKeyFunc)
f.Add(mkFifoObj("a", 1))
f.Add(mkFifoObj("b", 2))
f.AddIfNotPresent(mkFifoObj("b", 3))
f.AddIfNotPresent(mkFifoObj("c", 4))
if e, a := 3, len(f.items); a != e {
t.Fatalf("expected queue length %d, got %d", e, a)
}
expectedValues := []int{1, 2, 4}
for _, expected := range expectedValues {
if actual := Pop(f).(testFifoObject).val; actual != expected {
t.Fatalf("expected value %d, got %d", expected, actual)
}
}
}
func TestFIFO_HasSynced(t *testing.T) {
tests := []struct {
actions []func(f *FIFO)

34
tools/cache/reflector.go vendored
View File

@ -55,6 +55,28 @@ var (
defaultMinWatchTimeout = 5 * time.Minute
)
// ReflectorStore is the subset of cache.Store that the reflector uses
type ReflectorStore interface {
// Add adds the given object to the accumulator associated with the given object's key
Add(obj interface{}) error
// Update updates the given object in the accumulator associated with the given object's key
Update(obj interface{}) error
// Delete deletes the given object from the accumulator associated with the given object's key
Delete(obj interface{}) error
// Replace will delete the contents of the store, using instead the
// given list. Store takes ownership of the list, you should not reference
// it after calling this function.
Replace([]interface{}, string) error
// Resync is meaningless in the terms appearing here but has
// meaning in some implementations that have non-trivial
// additional behavior (e.g., DeltaFIFO).
Resync() error
}
// Reflector watches a specified resource and causes all changes to be reflected in the given store.
type Reflector struct {
// name identifies this reflector. By default, it will be a file:line if possible.
@ -72,7 +94,7 @@ type Reflector struct {
// The GVK of the object we expect to place in the store if unstructured.
expectedGVK *schema.GroupVersionKind
// The destination to sync up with the watch source
store Store
store ReflectorStore
// listerWatcher is used to perform lists and watches.
listerWatcher ListerWatcher
// backoff manages backoff of ListWatch
@ -189,13 +211,13 @@ func NewNamespaceKeyedIndexerAndReflector(lw ListerWatcher, expectedType interfa
// NewReflector creates a new Reflector with its name defaulted to the closest source_file.go:line in the call stack
// that is outside this package. See NewReflectorWithOptions for further information.
func NewReflector(lw ListerWatcher, expectedType interface{}, store Store, resyncPeriod time.Duration) *Reflector {
func NewReflector(lw ListerWatcher, expectedType interface{}, store ReflectorStore, resyncPeriod time.Duration) *Reflector {
return NewReflectorWithOptions(lw, expectedType, store, ReflectorOptions{ResyncPeriod: resyncPeriod})
}
// NewNamedReflector creates a new Reflector with the specified name. See NewReflectorWithOptions for further
// information.
func NewNamedReflector(name string, lw ListerWatcher, expectedType interface{}, store Store, resyncPeriod time.Duration) *Reflector {
func NewNamedReflector(name string, lw ListerWatcher, expectedType interface{}, store ReflectorStore, resyncPeriod time.Duration) *Reflector {
return NewReflectorWithOptions(lw, expectedType, store, ReflectorOptions{Name: name, ResyncPeriod: resyncPeriod})
}
@ -234,7 +256,7 @@ type ReflectorOptions struct {
// "yes". This enables you to use reflectors to periodically process
// everything as well as incrementally processing the things that
// change.
func NewReflectorWithOptions(lw ListerWatcher, expectedType interface{}, store Store, options ReflectorOptions) *Reflector {
func NewReflectorWithOptions(lw ListerWatcher, expectedType interface{}, store ReflectorStore, options ReflectorOptions) *Reflector {
reflectorClock := options.Clock
if reflectorClock == nil {
reflectorClock = clock.RealClock{}
@ -798,7 +820,7 @@ func handleWatch(
ctx context.Context,
start time.Time,
w watch.Interface,
store Store,
store ReflectorStore,
expectedType reflect.Type,
expectedGVK *schema.GroupVersionKind,
name string,
@ -826,7 +848,7 @@ func handleAnyWatch(
ctx context.Context,
start time.Time,
w watch.Interface,
store Store,
store ReflectorStore,
expectedType reflect.Type,
expectedGVK *schema.GroupVersionKind,
name string,

16
tools/cache/shared_informer.go vendored
View File

@ -540,11 +540,16 @@ func (s *sharedIndexInformer) RunWithContext(ctx context.Context) {
s.startedLock.Lock()
defer s.startedLock.Unlock()
fifo := NewDeltaFIFOWithOptions(DeltaFIFOOptions{
KnownObjects: s.indexer,
EmitDeltaTypeReplaced: true,
Transformer: s.transform,
})
var fifo Queue
if clientgofeaturegate.FeatureGates().Enabled(clientgofeaturegate.InOrderInformers) {
fifo = NewRealFIFO(MetaNamespaceKeyFunc, s.indexer, s.transform)
} else {
fifo = NewDeltaFIFOWithOptions(DeltaFIFOOptions{
KnownObjects: s.indexer,
EmitDeltaTypeReplaced: true,
Transformer: s.transform,
})
}
cfg := &Config{
Queue: fifo,
@ -552,7 +557,6 @@ func (s *sharedIndexInformer) RunWithContext(ctx context.Context) {
ObjectType: s.objectType,
ObjectDescription: s.objectDescription,
FullResyncPeriod: s.resyncCheckPeriod,
RetryOnError: false,
ShouldResync: s.processor.shouldResync,
Process: s.HandleDeltas,

407
tools/cache/the_real_fifo.go vendored Normal file
View File

@ -0,0 +1,407 @@
/*
Copyright 2025 The Kubernetes Authors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package cache
import (
"fmt"
utilruntime "k8s.io/apimachinery/pkg/util/runtime"
"k8s.io/apimachinery/pkg/util/sets"
utiltrace "k8s.io/utils/trace"
"sync"
"time"
)
// RealFIFO is a Queue in which every notification from the Reflector is passed
// in order to the Queue via Pop.
// This means that it
// 1. delivers notifications for items that have been deleted
// 2. delivers multiple notifications per item instead of simply the most recent value
type RealFIFO struct {
lock sync.RWMutex
cond sync.Cond
items []Delta
// populated is true if the first batch of items inserted by Replace() has been populated
// or Delete/Add/Update was called first.
populated bool
// initialPopulationCount is the number of items inserted by the first call of Replace()
initialPopulationCount int
// keyFunc is used to make the key used for queued item insertion and retrieval, and
// should be deterministic.
keyFunc KeyFunc
// knownObjects list keys that are "known" --- affecting Delete(),
// Replace(), and Resync()
knownObjects KeyListerGetter
// Indication the queue is closed.
// Used to indicate a queue is closed so a control loop can exit when a queue is empty.
// Currently, not used to gate any of CRUD operations.
closed bool
// Called with every object if non-nil.
transformer TransformFunc
}
var (
_ = Queue(&RealFIFO{}) // RealFIFO is a Queue
)
// Close the queue.
func (f *RealFIFO) Close() {
f.lock.Lock()
defer f.lock.Unlock()
f.closed = true
f.cond.Broadcast()
}
// KeyOf exposes f's keyFunc, but also detects the key of a Deltas object or
// DeletedFinalStateUnknown objects.
func (f *RealFIFO) keyOf(obj interface{}) (string, error) {
if d, ok := obj.(Deltas); ok {
if len(d) == 0 {
return "", KeyError{obj, ErrZeroLengthDeltasObject}
}
obj = d.Newest().Object
}
if d, ok := obj.(Delta); ok {
obj = d.Object
}
if d, ok := obj.(DeletedFinalStateUnknown); ok {
return d.Key, nil
}
return f.keyFunc(obj)
}
// HasSynced returns true if an Add/Update/Delete are called first,
// or the first batch of items inserted by Replace() has been popped.
func (f *RealFIFO) HasSynced() bool {
f.lock.Lock()
defer f.lock.Unlock()
return f.hasSynced_locked()
}
// ignoring lint to reduce delta to the original for review. It's ok adjust later.
//
//lint:file-ignore ST1003: should not use underscores in Go names
func (f *RealFIFO) hasSynced_locked() bool {
return f.populated && f.initialPopulationCount == 0
}
// addToItems_locked appends to the delta list.
func (f *RealFIFO) addToItems_locked(deltaActionType DeltaType, skipTransform bool, obj interface{}) error {
// we must be able to read the keys in order to determine whether the knownObjcts and the items
// in this FIFO overlap
_, err := f.keyOf(obj)
if err != nil {
return KeyError{obj, err}
}
// Every object comes through this code path once, so this is a good
// place to call the transform func.
//
// If obj is a DeletedFinalStateUnknown tombstone or the action is a Sync,
// then the object have already gone through the transformer.
//
// If the objects already present in the cache are passed to Replace(),
// the transformer must be idempotent to avoid re-mutating them,
// or coordinate with all readers from the cache to avoid data races.
// Default informers do not pass existing objects to Replace.
if f.transformer != nil {
_, isTombstone := obj.(DeletedFinalStateUnknown)
if !isTombstone && !skipTransform {
var err error
obj, err = f.transformer(obj)
if err != nil {
return err
}
}
}
f.items = append(f.items, Delta{
Type: deltaActionType,
Object: obj,
})
f.cond.Broadcast()
return nil
}
// Add inserts an item, and puts it in the queue. The item is only enqueued
// if it doesn't already exist in the set.
func (f *RealFIFO) Add(obj interface{}) error {
f.lock.Lock()
defer f.lock.Unlock()
f.populated = true
retErr := f.addToItems_locked(Added, false, obj)
return retErr
}
// Update is the same as Add in this implementation.
func (f *RealFIFO) Update(obj interface{}) error {
f.lock.Lock()
defer f.lock.Unlock()
f.populated = true
retErr := f.addToItems_locked(Updated, false, obj)
return retErr
}
// Delete removes an item. It doesn't add it to the queue, because
// this implementation assumes the consumer only cares about the objects,
// not the order in which they were created/added.
func (f *RealFIFO) Delete(obj interface{}) error {
f.lock.Lock()
defer f.lock.Unlock()
f.populated = true
retErr := f.addToItems_locked(Deleted, false, obj)
return retErr
}
// IsClosed checks if the queue is closed
func (f *RealFIFO) IsClosed() bool {
f.lock.Lock()
defer f.lock.Unlock()
return f.closed
}
// Pop waits until an item is ready and processes it. 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 processed.
// process function is called under lock, so it is safe
// update data structures in it that need to be in sync with the queue.
func (f *RealFIFO) Pop(process PopProcessFunc) (interface{}, error) {
f.lock.Lock()
defer f.lock.Unlock()
for len(f.items) == 0 {
// When the queue is empty, invocation of Pop() is blocked until new item is enqueued.
// When Close() is called, the f.closed is set and the condition is broadcasted.
// Which causes this loop to continue and return from the Pop().
if f.closed {
return nil, ErrFIFOClosed
}
f.cond.Wait()
}
isInInitialList := !f.hasSynced_locked()
item := f.items[0]
// The underlying array still exists and references this object, so the object will not be garbage collected unless we zero the reference.
f.items[0] = Delta{}
f.items = f.items[1:]
if f.initialPopulationCount > 0 {
f.initialPopulationCount--
}
// Only log traces if the queue depth is greater than 10 and it takes more than
// 100 milliseconds to process one item from the queue.
// Queue depth never goes high because processing an item is locking the queue,
// and new items can't be added until processing finish.
// https://github.com/kubernetes/kubernetes/issues/103789
if len(f.items) > 10 {
id, _ := f.keyOf(item)
trace := utiltrace.New("RealFIFO Pop Process",
utiltrace.Field{Key: "ID", Value: id},
utiltrace.Field{Key: "Depth", Value: len(f.items)},
utiltrace.Field{Key: "Reason", Value: "slow event handlers blocking the queue"})
defer trace.LogIfLong(100 * time.Millisecond)
}
// we wrap in Deltas here to be compatible with preview Pop functions and those interpreting the return value.
err := process(Deltas{item}, isInInitialList)
return Deltas{item}, err
}
// Replace
// 1. finds those items in f.items that are not in newItems and creates synthetic deletes for them
// 2. finds items in knownObjects that are not in newItems and creates synthetic deletes for them
// 3. adds the newItems to the queue
func (f *RealFIFO) Replace(newItems []interface{}, resourceVersion string) error {
f.lock.Lock()
defer f.lock.Unlock()
// determine the keys of everything we're adding. We cannot add the items until after the synthetic deletes have been
// created for items that don't existing in newItems
newKeys := sets.Set[string]{}
for _, obj := range newItems {
key, err := f.keyOf(obj)
if err != nil {
return KeyError{obj, err}
}
newKeys.Insert(key)
}
queuedItems := f.items
queuedKeys := []string{}
lastQueuedItemForKey := map[string]Delta{}
for _, queuedItem := range queuedItems {
queuedKey, err := f.keyOf(queuedItem.Object)
if err != nil {
return KeyError{queuedItem.Object, err}
}
if _, seen := lastQueuedItemForKey[queuedKey]; !seen {
queuedKeys = append(queuedKeys, queuedKey)
}
lastQueuedItemForKey[queuedKey] = queuedItem
}
// all the deletes already in the queue are important. There are two cases
// 1. queuedItems has delete for key/X and newItems has replace for key/X. This means the queued UID was deleted and a new one was created.
// 2. queuedItems has a delete for key/X and newItems does NOT have key/X. This means the queued item was deleted.
// Do deletion detection against objects in the queue.
for _, queuedKey := range queuedKeys {
if newKeys.Has(queuedKey) {
continue
}
// Delete pre-existing items not in the new list.
// This could happen if watch deletion event was missed while
// disconnected from apiserver.
lastQueuedItem := lastQueuedItemForKey[queuedKey]
// if we've already got the item marked as deleted, no need to add another delete
if lastQueuedItem.Type == Deleted {
continue
}
// if we got here, then the last entry we have for the queued item is *not* a deletion and we need to add a delete
deletedObj := lastQueuedItem.Object
retErr := f.addToItems_locked(Deleted, true, DeletedFinalStateUnknown{
Key: queuedKey,
Obj: deletedObj,
})
if retErr != nil {
return fmt.Errorf("couldn't enqueue object: %w", retErr)
}
}
// Detect deletions for objects not present in the queue, but present in KnownObjects
knownKeys := f.knownObjects.ListKeys()
for _, knownKey := range knownKeys {
if newKeys.Has(knownKey) { // still present
continue
}
if _, inQueuedItems := lastQueuedItemForKey[knownKey]; inQueuedItems { // already added delete for these
continue
}
deletedObj, exists, err := f.knownObjects.GetByKey(knownKey)
if err != nil {
deletedObj = nil
utilruntime.HandleError(fmt.Errorf("error during lookup, placing DeleteFinalStateUnknown marker without object: key=%q, err=%w", knownKey, err))
} else if !exists {
deletedObj = nil
utilruntime.HandleError(fmt.Errorf("key does not exist in known objects store, placing DeleteFinalStateUnknown marker without object: key=%q", knownKey))
}
retErr := f.addToItems_locked(Deleted, false, DeletedFinalStateUnknown{
Key: knownKey,
Obj: deletedObj,
})
if retErr != nil {
return fmt.Errorf("couldn't enqueue object: %w", retErr)
}
}
// now that we have the deletes we need for items, we can add the newItems to the items queue
for _, obj := range newItems {
retErr := f.addToItems_locked(Replaced, false, obj)
if retErr != nil {
return fmt.Errorf("couldn't enqueue object: %w", retErr)
}
}
if !f.populated {
f.populated = true
f.initialPopulationCount = len(f.items)
}
return nil
}
// Resync will ensure that every object in the Store has its key in the queue.
// This should be a no-op, because that property is maintained by all operations.
func (f *RealFIFO) Resync() error {
// TODO this cannot logically be done by the FIFO, it can only be done by the indexer
f.lock.Lock()
defer f.lock.Unlock()
if f.knownObjects == nil {
return nil
}
keysInQueue := sets.Set[string]{}
for _, item := range f.items {
key, err := f.keyOf(item.Object)
if err != nil {
return KeyError{item, err}
}
keysInQueue.Insert(key)
}
knownKeys := f.knownObjects.ListKeys()
for _, knownKey := range knownKeys {
// If we are doing Resync() and there is already an event queued for that object,
// we ignore the Resync for it. This is to avoid the race, in which the resync
// comes with the previous value of object (since queueing an event for the object
// doesn't trigger changing the underlying store <knownObjects>.
if keysInQueue.Has(knownKey) {
continue
}
knownObj, exists, err := f.knownObjects.GetByKey(knownKey)
if err != nil {
utilruntime.HandleError(fmt.Errorf("unable to queue object for sync: key=%q, err=%w", knownKey, err))
continue
} else if !exists {
utilruntime.HandleError(fmt.Errorf("key does not exist in known objects store, unable to queue object for sync: key=%q", knownKey))
continue
}
retErr := f.addToItems_locked(Sync, true, knownObj)
if retErr != nil {
return fmt.Errorf("couldn't queue object: %w", err)
}
}
return nil
}
// NewRealFIFO returns a Store which can be used to queue up items to
// process.
func NewRealFIFO(keyFunc KeyFunc, knownObjects KeyListerGetter, transformer TransformFunc) *RealFIFO {
if knownObjects == nil {
panic("coding error: knownObjects must be provided")
}
f := &RealFIFO{
items: make([]Delta, 0, 10),
keyFunc: keyFunc,
knownObjects: knownObjects,
transformer: transformer,
}
f.cond.L = &f.lock
return f
}

976
tools/cache/the_real_fifo_test.go vendored Normal file
View File

@ -0,0 +1,976 @@
/*
Copyright 2014 The Kubernetes Authors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package cache
import (
"fmt"
"reflect"
"runtime"
"testing"
"time"
)
func (f *RealFIFO) getItems() []Delta {
f.lock.Lock()
defer f.lock.Unlock()
ret := make([]Delta, len(f.items))
copy(ret, f.items)
return ret
}
const closedFIFOName = "FIFO WAS CLOSED"
func popN(queue Queue, count int) []interface{} {
result := []interface{}{}
for i := 0; i < count; i++ {
queue.Pop(func(obj interface{}, isInInitialList bool) error {
result = append(result, obj)
return nil
})
}
return result
}
// helper function to reduce stuttering
func testRealFIFOPop(f *RealFIFO) testFifoObject {
val := Pop(f)
if val == nil {
return testFifoObject{name: closedFIFOName}
}
return val.(Deltas).Newest().Object.(testFifoObject)
}
func emptyKnownObjects() KeyListerGetter {
return literalListerGetter(
func() []testFifoObject {
return []testFifoObject{}
},
)
}
func TestRealFIFO_basic(t *testing.T) {
f := NewRealFIFO(testFifoObjectKeyFunc, emptyKnownObjects(), nil)
const amount = 500
go func() {
for i := 0; i < amount; i++ {
f.Add(mkFifoObj(string([]rune{'a', rune(i)}), i+1))
}
}()
go func() {
for u := uint64(0); u < amount; u++ {
f.Add(mkFifoObj(string([]rune{'b', rune(u)}), u+1))
}
}()
lastInt := int(0)
lastUint := uint64(0)
for i := 0; i < amount*2; i++ {
switch obj := testRealFIFOPop(f).val.(type) {
case int:
if obj <= lastInt {
t.Errorf("got %v (int) out of order, last was %v", obj, lastInt)
}
lastInt = obj
case uint64:
if obj <= lastUint {
t.Errorf("got %v (uint) out of order, last was %v", obj, lastUint)
} else {
lastUint = obj
}
default:
t.Fatalf("unexpected type %#v", obj)
}
}
}
// TestRealFIFO_replaceWithDeleteDeltaIn tests that a `Sync` delta for an
// object `O` with ID `X` is added when .Replace is called and `O` is among the
// replacement objects even if the RealFIFO already stores in terminal position
// a delta of type `Delete` for ID `X`. Not adding the `Sync` delta causes
// SharedIndexInformers to miss `O`'s create notification, see https://github.com/kubernetes/kubernetes/issues/83810
// for more details.
func TestRealFIFO_replaceWithDeleteDeltaIn(t *testing.T) {
oldObj := mkFifoObj("foo", 1)
newObj := mkFifoObj("foo", 2)
f := NewRealFIFO(
testFifoObjectKeyFunc,
literalListerGetter(func() []testFifoObject {
return []testFifoObject{oldObj}
}),
nil,
)
f.Delete(oldObj)
f.Replace([]interface{}{newObj}, "")
actualDeltas := f.getItems()
expectedDeltas := []Delta{
{Type: Deleted, Object: oldObj},
{Type: Replaced, Object: newObj},
}
if !reflect.DeepEqual(expectedDeltas, actualDeltas) {
t.Errorf("expected %#v, got %#v", expectedDeltas, actualDeltas)
}
}
func TestRealFIFOW_ReplaceMakesDeletionsForObjectsOnlyInQueue(t *testing.T) {
obj := mkFifoObj("foo", 2)
objV2 := mkFifoObj("foo", 3)
table := []struct {
name string
operations func(f *RealFIFO)
expectedDeltas Deltas
}{
{
name: "Added object should be deleted on Replace",
operations: func(f *RealFIFO) {
f.Add(obj)
f.Replace([]interface{}{}, "0")
},
expectedDeltas: Deltas{
{Added, obj},
{Deleted, DeletedFinalStateUnknown{Key: "foo", Obj: obj}},
},
},
//{
// // ATTENTION: difference with delta_fifo_test, there is no option for emitDeltaTypeReplaced
// name: "Replaced object should have only a single Delete",
// operations: func(f *RealFIFO) {
// f.emitDeltaTypeReplaced = true
// f.Add(obj)
// f.Replace([]interface{}{obj}, "0")
// f.Replace([]interface{}{}, "0")
// },
// expectedDeltas: Deltas{
// {Added, obj},
// {Replaced, obj},
// {Deleted, DeletedFinalStateUnknown{Key: "foo", Obj: obj}},
// },
//},
{
name: "Deleted object should have only a single Delete",
operations: func(f *RealFIFO) {
f.Add(obj)
f.Delete(obj)
f.Replace([]interface{}{}, "0")
},
expectedDeltas: Deltas{
{Added, obj},
{Deleted, obj},
},
},
{
name: "Synced objects should have a single delete",
operations: func(f *RealFIFO) {
f.Add(obj)
f.Replace([]interface{}{obj}, "0")
f.Replace([]interface{}{obj}, "0")
f.Replace([]interface{}{}, "0")
},
expectedDeltas: Deltas{
{Added, obj},
{Replaced, obj},
{Replaced, obj},
{Deleted, DeletedFinalStateUnknown{Key: "foo", Obj: obj}},
},
},
{
name: "Added objects should have a single delete on multiple Replaces",
operations: func(f *RealFIFO) {
f.Add(obj)
f.Replace([]interface{}{}, "0")
f.Replace([]interface{}{}, "1")
},
expectedDeltas: Deltas{
{Added, obj},
{Deleted, DeletedFinalStateUnknown{Key: "foo", Obj: obj}},
},
},
{
name: "Added and deleted and added object should be deleted",
operations: func(f *RealFIFO) {
f.Add(obj)
f.Delete(obj)
f.Add(objV2)
f.Replace([]interface{}{}, "0")
},
expectedDeltas: Deltas{
{Added, obj},
{Deleted, obj},
{Added, objV2},
{Deleted, DeletedFinalStateUnknown{Key: "foo", Obj: objV2}},
},
},
}
for _, tt := range table {
tt := tt
t.Run(tt.name, func(t *testing.T) {
// Test with a RealFIFO with a backing KnownObjects
fWithKnownObjects := NewRealFIFO(
testFifoObjectKeyFunc,
literalListerGetter(func() []testFifoObject {
return []testFifoObject{}
}),
nil,
)
tt.operations(fWithKnownObjects)
actualDeltasWithKnownObjects := popN(fWithKnownObjects, len(fWithKnownObjects.getItems()))
actualAsDeltas := collapseDeltas(actualDeltasWithKnownObjects)
if !reflect.DeepEqual(tt.expectedDeltas, actualAsDeltas) {
t.Errorf("expected %#v, got %#v", tt.expectedDeltas, actualAsDeltas)
}
if len(fWithKnownObjects.items) != 0 {
t.Errorf("expected no extra deltas (empty map), got %#v", fWithKnownObjects.items)
}
// ATTENTION: difference with delta_fifo_test, there is no option without knownObjects
})
}
}
func collapseDeltas(ins []interface{}) Deltas {
ret := Deltas{}
for _, curr := range ins {
for _, delta := range curr.(Deltas) {
ret = append(ret, delta)
}
}
return ret
}
// ATTENTION: difference with delta_fifo_test, there is no requeue option anymore
// func TestDeltaFIFO_requeueOnPop(t *testing.T) {
func TestRealFIFO_addUpdate(t *testing.T) {
f := NewRealFIFO(
testFifoObjectKeyFunc,
emptyKnownObjects(),
nil,
)
f.Add(mkFifoObj("foo", 10))
f.Update(mkFifoObj("foo", 12))
f.Delete(mkFifoObj("foo", 15))
// ATTENTION: difference with delta_fifo_test, all items on the list. DeltaFIFO.List only showed newest, but Pop processed all.
expected1 := []Delta{
{
Type: Added,
Object: mkFifoObj("foo", 10),
},
{
Type: Updated,
Object: mkFifoObj("foo", 12),
},
{
Type: Deleted,
Object: mkFifoObj("foo", 15),
},
}
if e, a := expected1, f.getItems(); !reflect.DeepEqual(e, a) {
t.Errorf("Expected %+v, got %+v", e, a)
}
got := make(chan testFifoObject, 4)
done := make(chan struct{})
go func() {
defer close(done)
for {
obj := testRealFIFOPop(f)
if obj.name == closedFIFOName {
break
}
t.Logf("got a thing %#v", obj)
t.Logf("D len: %v", len(f.items))
got <- obj
}
}()
first := <-got
if e, a := 10, first.val; e != a {
t.Errorf("Didn't get updated value (%v), got %v", e, a)
}
second := <-got
if e, a := 12, second.val; e != a {
t.Errorf("Didn't get updated value (%v), got %v", e, a)
}
third := <-got
if e, a := 15, third.val; e != a {
t.Errorf("Didn't get updated value (%v), got %v", e, a)
}
select {
case unexpected := <-got:
t.Errorf("Got second value %v", unexpected.val)
case <-time.After(50 * time.Millisecond):
}
if e, a := 0, len(f.getItems()); e != a {
t.Errorf("Didn't get updated value (%v), got %v", e, a)
}
f.Close()
<-done
}
func TestRealFIFO_transformer(t *testing.T) {
mk := func(name string, rv int) testFifoObject {
return mkFifoObj(name, &rvAndXfrm{rv, 0})
}
xfrm := TransformFunc(func(obj interface{}) (interface{}, error) {
switch v := obj.(type) {
case testFifoObject:
v.val.(*rvAndXfrm).xfrm++
case DeletedFinalStateUnknown:
if x := v.Obj.(testFifoObject).val.(*rvAndXfrm).xfrm; x != 1 {
return nil, fmt.Errorf("object has been transformed wrong number of times: %#v", obj)
}
default:
return nil, fmt.Errorf("unexpected object: %#v", obj)
}
return obj, nil
})
must := func(err error) {
if err != nil {
t.Fatal(err)
}
}
mustTransform := func(obj interface{}) interface{} {
ret, err := xfrm(obj)
must(err)
return ret
}
f := NewRealFIFO(
testFifoObjectKeyFunc,
emptyKnownObjects(),
xfrm,
)
must(f.Add(mk("foo", 10)))
must(f.Add(mk("bar", 11)))
must(f.Update(mk("foo", 12)))
must(f.Delete(mk("foo", 15)))
must(f.Replace([]interface{}{}, ""))
must(f.Add(mk("bar", 16)))
must(f.Replace([]interface{}{}, ""))
// ATTENTION: difference with delta_fifo_test, without compression, we keep all the items, including bar being deleted multiple times.
// DeltaFIFO starts by checking keys, we start by checking types and keys
expected1 := []Delta{
{Type: Added, Object: mustTransform(mk("foo", 10))},
{Type: Added, Object: mustTransform(mk("bar", 11))},
{Type: Updated, Object: mustTransform(mk("foo", 12))},
{Type: Deleted, Object: mustTransform(mk("foo", 15))},
{Type: Deleted, Object: DeletedFinalStateUnknown{Key: "bar", Obj: mustTransform(mk("bar", 11))}},
{Type: Added, Object: mustTransform(mk("bar", 16))},
{Type: Deleted, Object: DeletedFinalStateUnknown{Key: "bar", Obj: mustTransform(mk("bar", 16))}},
}
actual1 := f.getItems()
if len(expected1) != len(actual1) {
t.Fatalf("Expected %+v, got %+v", expected1, actual1)
}
for i := 0; i < len(actual1); i++ {
e := expected1[i]
a := actual1[i]
if e.Type != a.Type {
t.Errorf("%d Expected %+v, got %+v", i, e, a)
}
eKey, err := f.keyOf(e)
if err != nil {
t.Fatal(err)
}
aKey, err := f.keyOf(a)
if err != nil {
t.Fatal(err)
}
if eKey != aKey {
t.Errorf("%d Expected %+v, got %+v", i, eKey, aKey)
}
}
for i := 0; i < len(expected1); i++ {
obj, err := f.Pop(func(o interface{}, isInInitialList bool) error { return nil })
if err != nil {
t.Fatalf("got nothing on try %v?", i)
}
a := obj.(Deltas)[0]
e := expected1[i]
if !reflect.DeepEqual(e, a) {
t.Errorf("%d Expected %+v, got %+v", i, e, a)
}
}
}
func TestRealFIFO_enqueueingNoLister(t *testing.T) {
f := NewRealFIFO(
testFifoObjectKeyFunc,
emptyKnownObjects(),
nil,
)
f.Add(mkFifoObj("foo", 10))
f.Update(mkFifoObj("bar", 15))
f.Add(mkFifoObj("qux", 17))
f.Delete(mkFifoObj("qux", 18))
// RealFIFO queues everything
f.Delete(mkFifoObj("baz", 20))
// ATTENTION: difference with delta_fifo_test, without compression every item is queued
expectList := []int{10, 15, 17, 18, 20}
for _, expect := range expectList {
if e, a := expect, testRealFIFOPop(f).val; e != a {
t.Errorf("Didn't get updated value (%v), got %v", e, a)
}
}
if e, a := 0, len(f.getItems()); e != a {
t.Errorf("queue unexpectedly not empty: %v != %v\n%#v", e, a, f.getItems())
}
}
func TestRealFIFO_enqueueingWithLister(t *testing.T) {
f := NewRealFIFO(
testFifoObjectKeyFunc,
literalListerGetter(func() []testFifoObject {
return []testFifoObject{mkFifoObj("foo", 5), mkFifoObj("bar", 6), mkFifoObj("baz", 7)}
}),
nil,
)
f.Add(mkFifoObj("foo", 10))
f.Update(mkFifoObj("bar", 15))
// This delete does enqueue the deletion, because "baz" is in the key lister.
f.Delete(mkFifoObj("baz", 20))
expectList := []int{10, 15, 20}
for _, expect := range expectList {
if e, a := expect, testRealFIFOPop(f).val; e != a {
t.Errorf("Didn't get updated value (%v), got %v", e, a)
}
}
if e, a := 0, len(f.items); e != a {
t.Errorf("queue unexpectedly not empty: %v != %v", e, a)
}
}
func TestRealFIFO_addReplace(t *testing.T) {
f := NewRealFIFO(
testFifoObjectKeyFunc,
emptyKnownObjects(),
nil,
)
f.Add(mkFifoObj("foo", 10))
f.Replace([]interface{}{mkFifoObj("foo", 15)}, "0")
got := make(chan testFifoObject, 3)
done := make(chan struct{})
go func() {
defer close(done)
for {
obj := testRealFIFOPop(f)
if obj.name == closedFIFOName {
break
}
got <- obj
}
}()
// ATTENTION: difference with delta_fifo_test, we get every event instead of the .Newest making us skip some for the test, but not at runtime.
curr := <-got
if e, a := 10, curr.val; e != a {
t.Errorf("Didn't get updated value (%v), got %v", e, a)
}
curr = <-got
if e, a := 15, curr.val; e != a {
t.Errorf("Didn't get updated value (%v), got %v", e, a)
}
select {
case unexpected := <-got:
t.Errorf("Got second value %v", unexpected.val)
case <-time.After(50 * time.Millisecond):
}
if items := f.getItems(); len(items) > 0 {
t.Errorf("item did not get removed")
}
f.Close()
<-done
}
func TestRealFIFO_ResyncNonExisting(t *testing.T) {
f := NewRealFIFO(
testFifoObjectKeyFunc,
literalListerGetter(func() []testFifoObject {
return []testFifoObject{mkFifoObj("foo", 5)}
}),
nil,
)
f.Delete(mkFifoObj("foo", 10))
f.Resync()
deltas := f.getItems()
if len(deltas) != 1 {
t.Fatalf("unexpected deltas length: %v", deltas)
}
if deltas[0].Type != Deleted {
t.Errorf("unexpected delta: %v", deltas[0])
}
}
func TestRealFIFO_Resync(t *testing.T) {
f := NewRealFIFO(
testFifoObjectKeyFunc,
literalListerGetter(func() []testFifoObject {
return []testFifoObject{mkFifoObj("foo", 5)}
}),
nil,
)
f.Resync()
deltas := f.getItems()
if len(deltas) != 1 {
t.Fatalf("unexpected deltas length: %v", deltas)
}
if deltas[0].Type != Sync {
t.Errorf("unexpected delta: %v", deltas[0])
}
}
func TestRealFIFO_DeleteExistingNonPropagated(t *testing.T) {
f := NewRealFIFO(
testFifoObjectKeyFunc,
emptyKnownObjects(),
nil,
)
f.Add(mkFifoObj("foo", 5))
f.Delete(mkFifoObj("foo", 6))
deltas := f.getItems()
if len(deltas) != 2 {
t.Fatalf("unexpected deltas length: %v", deltas)
}
if deltas[len(deltas)-1].Type != Deleted {
t.Errorf("unexpected delta: %v", deltas[len(deltas)-1])
}
}
func TestRealFIFO_ReplaceMakesDeletions(t *testing.T) {
// We test with only one pre-existing object because there is no
// promise about how their deletes are ordered.
// Try it with a pre-existing Delete
f := NewRealFIFO(
testFifoObjectKeyFunc,
literalListerGetter(func() []testFifoObject {
return []testFifoObject{mkFifoObj("foo", 5), mkFifoObj("bar", 6), mkFifoObj("baz", 7)}
}),
nil,
)
f.Delete(mkFifoObj("baz", 10))
f.Replace([]interface{}{mkFifoObj("foo", 5)}, "0")
expectedList := []Deltas{
{{Deleted, mkFifoObj("baz", 10)}},
// ATTENTION: difference with delta_fifo_test, logically the deletes of known items should happen BEFORE newItems are added, so this delete happens early now
// Since "bar" didn't have a delete event and wasn't in the Replace list
// it should get a tombstone key with the right Obj.
{{Deleted, DeletedFinalStateUnknown{Key: "bar", Obj: mkFifoObj("bar", 6)}}},
{{Replaced, mkFifoObj("foo", 5)}},
}
for _, expected := range expectedList {
cur := Pop(f).(Deltas)
if e, a := expected, cur; !reflect.DeepEqual(e, a) {
t.Errorf("Expected %#v, got %#v", e, a)
}
}
// Now try starting with an Add instead of a Delete
f = NewRealFIFO(
testFifoObjectKeyFunc,
literalListerGetter(func() []testFifoObject {
return []testFifoObject{mkFifoObj("foo", 5), mkFifoObj("bar", 6), mkFifoObj("baz", 7)}
}),
nil,
)
f.Add(mkFifoObj("baz", 10))
f.Replace([]interface{}{mkFifoObj("foo", 5)}, "0")
// ATTENTION: difference with delta_fifo_test, every event is its own Deltas with one item
expectedList = []Deltas{
{{Added, mkFifoObj("baz", 10)}},
{{Deleted, DeletedFinalStateUnknown{Key: "baz", Obj: mkFifoObj("baz", 10)}}},
// ATTENTION: difference with delta_fifo_test, logically the deletes of known items should happen BEFORE newItems are added, so this delete happens early now
// Since "bar" didn't have a delete event and wasn't in the Replace list
// it should get a tombstone key with the right Obj.
{{Deleted, DeletedFinalStateUnknown{Key: "bar", Obj: mkFifoObj("bar", 6)}}},
{{Replaced, mkFifoObj("foo", 5)}},
}
for _, expected := range expectedList {
cur := Pop(f).(Deltas)
if e, a := expected, cur; !reflect.DeepEqual(e, a) {
t.Errorf("Expected %#v, got %#v", e, a)
}
}
// Now try deleting and recreating the object in the queue, then delete it by a Replace call
f = NewRealFIFO(
testFifoObjectKeyFunc,
literalListerGetter(func() []testFifoObject {
return []testFifoObject{mkFifoObj("foo", 5), mkFifoObj("bar", 6), mkFifoObj("baz", 7)}
}),
nil,
)
f.Delete(mkFifoObj("bar", 6))
f.Add(mkFifoObj("bar", 100))
f.Replace([]interface{}{mkFifoObj("foo", 5)}, "0")
// ATTENTION: difference with delta_fifo_test, every event is its own Deltas with one item
expectedList = []Deltas{
{{Deleted, mkFifoObj("bar", 6)}},
{{Added, mkFifoObj("bar", 100)}},
// Since "bar" has a newer object in the queue than in the state,
// it should get a tombstone key with the latest object from the queue
{{Deleted, DeletedFinalStateUnknown{Key: "bar", Obj: mkFifoObj("bar", 100)}}},
// ATTENTION: difference with delta_fifo_test, logically the deletes of known items should happen BEFORE newItems are added, so this delete happens early now
{{Deleted, DeletedFinalStateUnknown{Key: "baz", Obj: mkFifoObj("baz", 7)}}},
{{Replaced, mkFifoObj("foo", 5)}},
}
for _, expected := range expectedList {
cur := Pop(f).(Deltas)
if e, a := expected, cur; !reflect.DeepEqual(e, a) {
t.Errorf("Expected %#v, got %#v", e, a)
}
}
// Now try syncing it first to ensure the delete use the latest version
f = NewRealFIFO(
testFifoObjectKeyFunc,
literalListerGetter(func() []testFifoObject {
return []testFifoObject{mkFifoObj("foo", 5), mkFifoObj("bar", 6), mkFifoObj("baz", 7)}
}),
nil,
)
f.Replace([]interface{}{mkFifoObj("bar", 100), mkFifoObj("foo", 5)}, "0")
f.Replace([]interface{}{mkFifoObj("foo", 5)}, "0")
// ATTENTION: difference with delta_fifo_test, every event is its own Deltas with one item
// ATTENTION: difference with delta_fifo_test, deltaFifo associated by key, but realFIFO orders across all keys, so this ordering changed
expectedList = []Deltas{
// ATTENTION: difference with delta_fifo_test, logically the deletes of known items should happen BEFORE newItems are added, so this delete happens early now
// Since "baz" didn't have a delete event and wasn't in the Replace list
{{Deleted, DeletedFinalStateUnknown{Key: "baz", Obj: mkFifoObj("baz", 7)}}},
{{Replaced, mkFifoObj("bar", 100)}},
{{Replaced, mkFifoObj("foo", 5)}},
// Since "bar" didn't have a delete event and wasn't in the Replace list
// it should get a tombstone key with the right Obj.
{{Deleted, DeletedFinalStateUnknown{Key: "bar", Obj: mkFifoObj("bar", 100)}}},
{{Replaced, mkFifoObj("foo", 5)}},
}
for i, expected := range expectedList {
cur := Pop(f).(Deltas)
if e, a := expected, cur; !reflect.DeepEqual(e, a) {
t.Errorf("%d Expected %#v, got %#v", i, e, a)
}
}
// Now try starting without an explicit KeyListerGetter
f = NewRealFIFO(
testFifoObjectKeyFunc,
emptyKnownObjects(),
nil,
)
f.Add(mkFifoObj("baz", 10))
f.Replace([]interface{}{mkFifoObj("foo", 5)}, "0")
expectedList = []Deltas{
{{Added, mkFifoObj("baz", 10)}},
{{Deleted, DeletedFinalStateUnknown{Key: "baz", Obj: mkFifoObj("baz", 10)}}},
{{Replaced, mkFifoObj("foo", 5)}},
}
for _, expected := range expectedList {
cur := Pop(f).(Deltas)
if e, a := expected, cur; !reflect.DeepEqual(e, a) {
t.Errorf("Expected %#v, got %#v", e, a)
}
}
}
// TestRealFIFO_ReplaceMakesDeletionsReplaced is the same as the above test, but
// ensures that a Replaced DeltaType is emitted.
func TestRealFIFO_ReplaceMakesDeletionsReplaced(t *testing.T) {
f := NewRealFIFO(
testFifoObjectKeyFunc,
literalListerGetter(func() []testFifoObject {
return []testFifoObject{mkFifoObj("foo", 5), mkFifoObj("bar", 6), mkFifoObj("baz", 7)}
}),
nil,
)
f.Delete(mkFifoObj("baz", 10))
f.Replace([]interface{}{mkFifoObj("foo", 6)}, "0")
expectedList := []Deltas{
{{Deleted, mkFifoObj("baz", 10)}},
// ATTENTION: difference with delta_fifo_test, logically the deletes of known items should happen BEFORE newItems are added, so this delete happens early now
// Since "bar" didn't have a delete event and wasn't in the Replace list
// it should get a tombstone key with the right Obj.
{{Deleted, DeletedFinalStateUnknown{Key: "bar", Obj: mkFifoObj("bar", 6)}}},
{{Replaced, mkFifoObj("foo", 6)}},
}
for _, expected := range expectedList {
cur := Pop(f).(Deltas)
if e, a := expected, cur; !reflect.DeepEqual(e, a) {
t.Errorf("Expected %#v, got %#v", e, a)
}
}
}
// ATTENTION: difference with delta_fifo_test, the previous value was hardcoded as use "Replace" so I've eliminated the option to set it differently
//func TestRealFIFO_ReplaceDeltaType(t *testing.T) {
func TestRealFIFO_UpdateResyncRace(t *testing.T) {
f := NewRealFIFO(
testFifoObjectKeyFunc,
literalListerGetter(func() []testFifoObject {
return []testFifoObject{mkFifoObj("foo", 5)}
}),
nil,
)
f.Update(mkFifoObj("foo", 6))
f.Resync()
expectedList := []Deltas{
{{Updated, mkFifoObj("foo", 6)}},
}
for _, expected := range expectedList {
cur := Pop(f).(Deltas)
if e, a := expected, cur; !reflect.DeepEqual(e, a) {
t.Errorf("Expected %#v, got %#v", e, a)
}
}
}
func TestRealFIFO_HasSyncedCorrectOnDeletion(t *testing.T) {
f := NewRealFIFO(
testFifoObjectKeyFunc,
literalListerGetter(func() []testFifoObject {
return []testFifoObject{mkFifoObj("foo", 5), mkFifoObj("bar", 6), mkFifoObj("baz", 7)}
}),
nil,
)
f.Replace([]interface{}{mkFifoObj("foo", 5)}, "0")
expectedList := []Deltas{
// ATTENTION: difference with delta_fifo_test, logically the deletes of known items should happen BEFORE newItems are added, so this delete happens early now
// Since "bar" didn't have a delete event and wasn't in the Replace list
// it should get a tombstone key with the right Obj.
{{Deleted, DeletedFinalStateUnknown{Key: "bar", Obj: mkFifoObj("bar", 6)}}},
{{Deleted, DeletedFinalStateUnknown{Key: "baz", Obj: mkFifoObj("baz", 7)}}},
{{Replaced, mkFifoObj("foo", 5)}},
}
for _, expected := range expectedList {
if f.HasSynced() {
t.Errorf("Expected HasSynced to be false")
}
cur, initial := pop2[Deltas](f)
if e, a := expected, cur; !reflect.DeepEqual(e, a) {
t.Errorf("Expected %#v, got %#v", e, a)
}
if initial != true {
t.Error("Expected initial list item")
}
}
if !f.HasSynced() {
t.Errorf("Expected HasSynced to be true")
}
}
func TestRealFIFO_detectLineJumpers(t *testing.T) {
f := NewRealFIFO(
testFifoObjectKeyFunc,
emptyKnownObjects(),
nil,
)
f.Add(mkFifoObj("foo", 10))
f.Add(mkFifoObj("bar", 1))
f.Add(mkFifoObj("foo", 11))
f.Add(mkFifoObj("foo", 13))
f.Add(mkFifoObj("zab", 30))
// ATTENTION: difference with delta_fifo_test, every event is delivered in order
if e, a := 10, testRealFIFOPop(f).val; a != e {
t.Fatalf("expected %d, got %d", e, a)
}
f.Add(mkFifoObj("foo", 14)) // ensure foo doesn't jump back in line
if e, a := 1, testRealFIFOPop(f).val; a != e {
t.Fatalf("expected %d, got %d", e, a)
}
if e, a := 11, testRealFIFOPop(f).val; a != e {
t.Fatalf("expected %d, got %d", e, a)
}
if e, a := 13, testRealFIFOPop(f).val; a != e {
t.Fatalf("expected %d, got %d", e, a)
}
if e, a := 30, testRealFIFOPop(f).val; a != e {
t.Fatalf("expected %d, got %d", e, a)
}
if e, a := 14, testRealFIFOPop(f).val; a != e {
t.Fatalf("expected %d, got %d", e, a)
}
}
func TestRealFIFO_KeyOf(t *testing.T) {
f := RealFIFO{keyFunc: testFifoObjectKeyFunc}
table := []struct {
obj interface{}
key string
}{
{obj: testFifoObject{name: "A"}, key: "A"},
{obj: DeletedFinalStateUnknown{Key: "B", Obj: nil}, key: "B"},
{obj: Deltas{{Object: testFifoObject{name: "C"}}}, key: "C"},
{obj: Deltas{{Object: DeletedFinalStateUnknown{Key: "D", Obj: nil}}}, key: "D"},
}
for _, item := range table {
got, err := f.keyOf(item.obj)
if err != nil {
t.Errorf("Unexpected error for %q: %v", item.obj, err)
continue
}
if e, a := item.key, got; e != a {
t.Errorf("Expected %v, got %v", e, a)
}
}
}
func TestRealFIFO_HasSynced(t *testing.T) {
tests := []struct {
actions []func(f *RealFIFO)
expectedSynced bool
}{
{
actions: []func(f *RealFIFO){},
expectedSynced: false,
},
{
actions: []func(f *RealFIFO){
func(f *RealFIFO) { f.Add(mkFifoObj("a", 1)) },
},
expectedSynced: true,
},
{
actions: []func(f *RealFIFO){
func(f *RealFIFO) { f.Replace([]interface{}{}, "0") },
},
expectedSynced: true,
},
{
actions: []func(f *RealFIFO){
func(f *RealFIFO) { f.Replace([]interface{}{mkFifoObj("a", 1), mkFifoObj("b", 2)}, "0") },
},
expectedSynced: false,
},
{
actions: []func(f *RealFIFO){
func(f *RealFIFO) { f.Replace([]interface{}{mkFifoObj("a", 1), mkFifoObj("b", 2)}, "0") },
func(f *RealFIFO) { Pop(f) },
},
expectedSynced: false,
},
{
actions: []func(f *RealFIFO){
func(f *RealFIFO) { f.Replace([]interface{}{mkFifoObj("a", 1), mkFifoObj("b", 2)}, "0") },
func(f *RealFIFO) { Pop(f) },
func(f *RealFIFO) { Pop(f) },
},
expectedSynced: true,
},
{
// 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;
// there cannot be duplicate keys in the list or apiserver is broken.
actions: []func(f *RealFIFO){
func(f *RealFIFO) { f.Replace([]interface{}{mkFifoObj("a", 1), mkFifoObj("a", 2)}, "0") },
func(f *RealFIFO) { Pop(f) },
// ATTENTION: difference with delta_fifo_test, every event is delivered, so a is listed twice and must be popped twice to remove both
func(f *RealFIFO) { Pop(f) },
},
expectedSynced: true,
},
}
for i, test := range tests {
f := NewRealFIFO(
testFifoObjectKeyFunc,
emptyKnownObjects(),
nil,
)
for _, action := range test.actions {
action(f)
}
if e, a := test.expectedSynced, f.HasSynced(); a != e {
t.Errorf("test case %v failed, expected: %v , got %v", i, e, a)
}
}
}
// TestRealFIFO_PopShouldUnblockWhenClosed checks that any blocking Pop on an empty queue
// should unblock and return after Close is called.
func TestRealFIFO_PopShouldUnblockWhenClosed(t *testing.T) {
f := NewRealFIFO(
testFifoObjectKeyFunc,
literalListerGetter(func() []testFifoObject {
return []testFifoObject{mkFifoObj("foo", 5)}
}),
nil,
)
c := make(chan struct{})
const jobs = 10
for i := 0; i < jobs; i++ {
go func() {
f.Pop(func(obj interface{}, isInInitialList bool) error {
return nil
})
c <- struct{}{}
}()
}
runtime.Gosched()
f.Close()
for i := 0; i < jobs; i++ {
select {
case <-c:
case <-time.After(500 * time.Millisecond):
t.Fatalf("timed out waiting for Pop to return after Close")
}
}
}