Add benchmarks for adding objects with slow handlers

staging/src/k8s.io/client-go/tools/cache/controller_bench_test.go

@@ -0,0 +1,110 @@
+/*
+Copyright 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 (
+	"context"
+	"errors"
+	"fmt"
+	"testing"
+	"time"
+
+	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	fcache "k8s.io/client-go/tools/cache/testing"
+)
+
+const handlerWaitTime = time.Millisecond
+
+func BenchmarkAddWithSlowHandlers(b *testing.B) {
+	for _, unlockWhileProcessing := range []bool{false, true} {
+		b.Run(fmt.Sprintf("unlockWhileProcessing=%t", unlockWhileProcessing), func(b *testing.B) {
+			ctx, cancel := context.WithCancel(context.Background())
+			source := fcache.NewFakeControllerSource()
+			b.Cleanup(func() {
+				cancel()
+				source.Shutdown()
+			})
+			testIDs := []string{"a-hello"}
+			source.Add(&v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: testIDs[0]}})
+
+			store := NewStore(DeletionHandlingMetaNamespaceKeyFunc)
+			fifoOptions := RealFIFOOptions{}
+			if unlockWhileProcessing {
+				fifoOptions.UnlockWhileProcessing = true
+				fifoOptions.AtomicEvents = true
+			} else {
+				fifoOptions.KnownObjects = store
+			}
+			fifo := NewRealFIFOWithOptions(fifoOptions)
+			handler := ResourceEventHandlerFuncs{
+				UpdateFunc: func(oldObj, newObj interface{}) {
+					time.Sleep(handlerWaitTime)
+				},
+			}
+
+			cfg := &Config{
+				Queue:            fifo,
+				ListerWatcher:    source,
+				ObjectType:       &v1.Pod{},
+				FullResyncPeriod: 0,
+
+				Process: func(obj interface{}, isInInitialList bool) error {
+					if deltas, ok := obj.(Deltas); ok {
+						return processDeltas(handler, store, deltas, isInInitialList, DeletionHandlingMetaNamespaceKeyFunc)
+					}
+					return errors.New("object given as Process argument is not Deltas")
+				},
+				ProcessBatch: func(deltaList []Delta, isInInitialList bool) error {
+					return processDeltasInBatch(handler, store, deltaList, isInInitialList, DeletionHandlingMetaNamespaceKeyFunc)
+				},
+			}
+			c := New(cfg)
+			go c.RunWithContext(ctx)
+			if !WaitForCacheSync(ctx.Done(), c.HasSynced) {
+				b.Fatal("Timed out waiting for cache sync")
+			}
+
+			// Producer: Modify object as fast as the handler can process it. This should ensure that the process func is always running.
+			go func() {
+				// Stop when the benchmark context is cancelled.
+				ticker := time.NewTicker(handlerWaitTime)
+				defer ticker.Stop()
+				for {
+					select {
+					case <-ctx.Done():
+						return
+					case <-ticker.C:
+						source.Modify(&v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: testIDs[0]}})
+					}
+				}
+			}()
+			benchmarkReflectorWithSlowHandlers(b, fifo)
+		})
+	}
+}
+
+func benchmarkReflectorWithSlowHandlers(b *testing.B, fifo *RealFIFO) {
+	b.ResetTimer()
+	// Try adding an object to the queue, while the controller is processing other events.
+	for i := 0; i < b.N; i++ {
+		if err := fifo.Add(&v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "b-hello"}}); err != nil {
+			b.Fatal(err)
+		}
+	}
+	b.StopTimer()
+}