Merge pull request #125884 from serathius/benchmark-storage

Benchmark storage

Kubernetes-commit: 3a849069043142b2bec8f45654b235ba0b660aad

go.mod

@@ -10,6 +10,7 @@ require (
 	github.com/coreos/go-systemd/v22 v22.5.0
 	github.com/emicklei/go-restful/v3 v3.11.0
 	github.com/fsnotify/fsnotify v1.7.0
+	github.com/go-logr/logr v1.4.2
 	github.com/gogo/protobuf v1.3.2
 	github.com/google/cel-go v0.20.1
 	github.com/google/gnostic-models v0.6.8
@@ -70,7 +71,6 @@ require (
 	github.com/dustin/go-humanize v1.0.1 // indirect
 	github.com/felixge/httpsnoop v1.0.4 // indirect
 	github.com/fxamacker/cbor/v2 v2.7.0 // indirect
-	github.com/go-logr/logr v1.4.2 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/go-openapi/jsonpointer v0.19.6 // indirect
 	github.com/go-openapi/jsonreference v0.20.2 // indirect

pkg/storage/cacher/cacher_test.go

@@ -22,6 +22,8 @@ import (
 	"testing"
 	"time"
 
+	"github.com/go-logr/logr"
+
 	apiequality "k8s.io/apimachinery/pkg/api/equality"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/fields"
@@ -37,7 +39,9 @@ import (
 	etcd3testing "k8s.io/apiserver/pkg/storage/etcd3/testing"
 	storagetesting "k8s.io/apiserver/pkg/storage/testing"
 	utilfeature "k8s.io/apiserver/pkg/util/feature"
+	"k8s.io/client-go/tools/cache"
 	featuregatetesting "k8s.io/component-base/featuregate/testing"
+	"k8s.io/klog/v2"
 	"k8s.io/utils/clock"
 )
 
@@ -396,6 +400,7 @@ type setupOptions struct {
 	resourcePrefix string
 	keyFunc        func(runtime.Object) (string, error)
 	indexerFuncs   map[string]storage.IndexerFunc
+	indexers       cache.Indexers
 	clock          clock.WithTicker
 }
 
@@ -425,6 +430,12 @@ func withSpecNodeNameIndexerFuncs(options *setupOptions) {
 			return pod.Spec.NodeName
 		},
 	}
+	options.indexers = map[string]cache.IndexFunc{
+		"f:spec.nodeName": func(obj interface{}) ([]string, error) {
+			pod := obj.(*example.Pod)
+			return []string{pod.Spec.NodeName}, nil
+		},
+	}
 }
 
 func testSetup(t *testing.T, opts ...setupOption) (context.Context, *Cacher, tearDownFunc) {
@@ -432,7 +443,7 @@ func testSetup(t *testing.T, opts ...setupOption) (context.Context, *Cacher, tea
 	return ctx, cacher, tearDown
 }
 
-func testSetupWithEtcdServer(t *testing.T, opts ...setupOption) (context.Context, *Cacher, *etcd3testing.EtcdTestServer, tearDownFunc) {
+func testSetupWithEtcdServer(t testing.TB, opts ...setupOption) (context.Context, *Cacher, *etcd3testing.EtcdTestServer, tearDownFunc) {
 	setupOpts := setupOptions{}
 	opts = append([]setupOption{withDefaults}, opts...)
 	for _, opt := range opts {
@@ -456,6 +467,7 @@ func testSetupWithEtcdServer(t *testing.T, opts ...setupOption) (context.Context
 		NewFunc:        newPod,
 		NewListFunc:    newPodList,
 		IndexerFuncs:   setupOpts.indexerFuncs,
+		Indexers:       &setupOpts.indexers,
 		Codec:          codecs.LegacyCodec(examplev1.SchemeGroupVersion),
 		Clock:          setupOpts.clock,
 	}
@@ -521,3 +533,24 @@ func (c *createWrapper) Create(ctx context.Context, key string, obj, out runtime
 		return true, nil
 	})
 }
+
+func BenchmarkStoreListCreate(b *testing.B) {
+	klog.SetLogger(logr.Discard())
+	b.Run("RV=NotOlderThan", func(b *testing.B) {
+		ctx, cacher, _, terminate := testSetupWithEtcdServer(b)
+		b.Cleanup(terminate)
+		storagetesting.RunBenchmarkStoreListCreate(ctx, b, cacher, metav1.ResourceVersionMatchNotOlderThan)
+	})
+	b.Run("RV=ExactMatch", func(b *testing.B) {
+		ctx, cacher, _, terminate := testSetupWithEtcdServer(b)
+		b.Cleanup(terminate)
+		storagetesting.RunBenchmarkStoreListCreate(ctx, b, cacher, metav1.ResourceVersionMatchExact)
+	})
+}
+
+func BenchmarkStoreList(b *testing.B) {
+	klog.SetLogger(logr.Discard())
+	ctx, cacher, _, terminate := testSetupWithEtcdServer(b, withSpecNodeNameIndexerFuncs)
+	b.Cleanup(terminate)
+	storagetesting.RunBenchmarkStoreList(ctx, b, cacher)
+}

pkg/storage/cacher/cacher_testing_utils_test.go

@@ -55,7 +55,7 @@ func init() {
 func newPod() runtime.Object     { return &example.Pod{} }
 func newPodList() runtime.Object { return &example.PodList{} }
 
-func newEtcdTestStorage(t *testing.T, prefix string) (*etcd3testing.EtcdTestServer, storage.Interface) {
+func newEtcdTestStorage(t testing.TB, prefix string) (*etcd3testing.EtcdTestServer, storage.Interface) {
 	server, _ := etcd3testing.NewUnsecuredEtcd3TestClientServer(t)
 	storage := etcd3.New(
 		server.V3Client,

pkg/storage/etcd3/store_test.go

@@ -27,6 +27,7 @@ import (
 	"sync/atomic"
 	"testing"
 
+	"github.com/go-logr/logr"
 	clientv3 "go.etcd.io/etcd/client/v3"
 	"go.etcd.io/etcd/server/v3/embed"
 	"google.golang.org/grpc/grpclog"
@@ -45,6 +46,7 @@ import (
 	"k8s.io/apiserver/pkg/storage/etcd3/testserver"
 	storagetesting "k8s.io/apiserver/pkg/storage/testing"
 	"k8s.io/apiserver/pkg/storage/value"
+	"k8s.io/klog/v2"
 )
 
 var scheme = runtime.NewScheme()
@@ -905,3 +907,20 @@ func BenchmarkStore_GetList(b *testing.B) {
 		})
 	}
 }
+
+func BenchmarkStoreListCreate(b *testing.B) {
+	klog.SetLogger(logr.Discard())
+	b.Run("RV=NotOlderThan", func(b *testing.B) {
+		ctx, store, _ := testSetup(b)
+		storagetesting.RunBenchmarkStoreListCreate(ctx, b, store, metav1.ResourceVersionMatchNotOlderThan)
+	})
+	b.Run("RV=ExactMatch", func(b *testing.B) {
+		ctx, store, _ := testSetup(b)
+		storagetesting.RunBenchmarkStoreListCreate(ctx, b, store, metav1.ResourceVersionMatchExact)
+	})
+}
+
+func BenchmarkStoreList(b *testing.B) {
+	ctx, store, _ := testSetup(b)
+	storagetesting.RunBenchmarkStoreList(ctx, b, store)
+}

pkg/storage/etcd3/testing/test_server.go

@@ -30,12 +30,12 @@ type EtcdTestServer struct {
 	V3Client *clientv3.Client
 }
 
-func (e *EtcdTestServer) Terminate(t *testing.T) {
+func (e *EtcdTestServer) Terminate(t testing.TB) {
 	// no-op, server termination moved to test cleanup
 }
 
 // NewUnsecuredEtcd3TestClientServer creates a new client and server for testing
-func NewUnsecuredEtcd3TestClientServer(t *testing.T) (*EtcdTestServer, *storagebackend.Config) {
+func NewUnsecuredEtcd3TestClientServer(t testing.TB) (*EtcdTestServer, *storagebackend.Config) {
 	server := &EtcdTestServer{}
 	server.V3Client = testserver.RunEtcd(t, nil)
 	config := &storagebackend.Config{

pkg/storage/testing/store_benchmarks.go

@@ -0,0 +1,261 @@
+/*
+Copyright 2024 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 testing
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"sync/atomic"
+	"testing"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/fields"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/util/rand"
+	"k8s.io/apiserver/pkg/apis/example"
+	"k8s.io/apiserver/pkg/storage"
+)
+
+func RunBenchmarkStoreListCreate(ctx context.Context, b *testing.B, store storage.Interface, match metav1.ResourceVersionMatch) {
+	objectCount := atomic.Uint64{}
+	pods := []*example.Pod{}
+	for i := 0; i < b.N; i++ {
+		name := rand.String(100)
+		pods = append(pods, &example.Pod{ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: name}})
+	}
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		pod := pods[i]
+		podOut := &example.Pod{}
+		err := store.Create(ctx, computePodKey(pod), pod, podOut, 0)
+		if err != nil {
+			panic(fmt.Sprintf("Unexpected error %s", err))
+		}
+		listOut := &example.PodList{}
+		err = store.GetList(ctx, "/pods", storage.ListOptions{
+			Recursive:            true,
+			ResourceVersion:      podOut.ResourceVersion,
+			ResourceVersionMatch: match,
+			Predicate: storage.SelectionPredicate{
+				Label: labels.Everything(),
+				Field: fields.Everything(),
+				Limit: 1,
+			},
+		}, listOut)
+		if err != nil {
+			panic(fmt.Sprintf("Unexpected error %s", err))
+		}
+		if len(listOut.Items) != 1 {
+			b.Errorf("Expected to get 1 element, got %d", len(listOut.Items))
+		}
+		objectCount.Add(uint64(len(listOut.Items)))
+	}
+	b.ReportMetric(float64(objectCount.Load())/float64(b.N), "objects/op")
+}
+
+func RunBenchmarkStoreList(ctx context.Context, b *testing.B, store storage.Interface) {
+	namespaceCount := 100
+	podPerNamespaceCount := 100
+	var paginateLimit int64 = 100
+	nodeCount := 100
+	namespacedNames, nodeNames := prepareBenchchmarkData(ctx, store, namespaceCount, podPerNamespaceCount, nodeCount)
+	b.ResetTimer()
+	maxRevision := 1 + namespaceCount*podPerNamespaceCount
+	cases := []struct {
+		name  string
+		match metav1.ResourceVersionMatch
+	}{
+		{
+			name:  "RV=Empty",
+			match: "",
+		},
+		{
+			name:  "RV=NotOlderThan",
+			match: metav1.ResourceVersionMatchNotOlderThan,
+		},
+		{
+			name:  "RV=MatchExact",
+			match: metav1.ResourceVersionMatchExact,
+		},
+	}
+
+	for _, c := range cases {
+		b.Run(c.name, func(b *testing.B) {
+			runBenchmarkStoreList(ctx, b, store, 0, maxRevision, c.match, false, nodeNames)
+		})
+	}
+	b.Run("Paginate", func(b *testing.B) {
+		for _, c := range cases {
+			b.Run(c.name, func(b *testing.B) {
+				runBenchmarkStoreList(ctx, b, store, paginateLimit, maxRevision, c.match, false, nodeNames)
+			})
+		}
+	})
+	b.Run("NodeIndexed", func(b *testing.B) {
+		for _, c := range cases {
+			b.Run(c.name, func(b *testing.B) {
+				runBenchmarkStoreList(ctx, b, store, 0, maxRevision, c.match, true, nodeNames)
+			})
+		}
+		b.Run("Paginate", func(b *testing.B) {
+			for _, c := range cases {
+				b.Run(c.name, func(b *testing.B) {
+					runBenchmarkStoreList(ctx, b, store, paginateLimit, maxRevision, c.match, true, nodeNames)
+				})
+			}
+		})
+	})
+	b.Run("Namespace", func(b *testing.B) {
+		for _, c := range cases {
+			b.Run(c.name, func(b *testing.B) {
+				runBenchmarkStoreListNamespace(ctx, b, store, maxRevision, c.match, namespacedNames)
+			})
+		}
+	})
+}
+
+func runBenchmarkStoreListNamespace(ctx context.Context, b *testing.B, store storage.Interface, maxRV int, match metav1.ResourceVersionMatch, namespaceNames []string) {
+	wg := sync.WaitGroup{}
+	objectCount := atomic.Uint64{}
+	pageCount := atomic.Uint64{}
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		wg.Add(1)
+		resourceVersion := ""
+		switch match {
+		case metav1.ResourceVersionMatchExact, metav1.ResourceVersionMatchNotOlderThan:
+			resourceVersion = fmt.Sprintf("%d", maxRV-99+i%100)
+		}
+		go func(resourceVersion string) {
+			defer wg.Done()
+			opts := storage.ListOptions{
+				Recursive:            true,
+				ResourceVersion:      resourceVersion,
+				ResourceVersionMatch: match,
+				Predicate:            storage.Everything,
+			}
+			for j := 0; j < len(namespaceNames); j++ {
+				objects, pages := paginate(ctx, store, "/pods/"+namespaceNames[j], opts)
+				objectCount.Add(uint64(objects))
+				pageCount.Add(uint64(pages))
+			}
+		}(resourceVersion)
+	}
+	wg.Wait()
+	b.ReportMetric(float64(objectCount.Load())/float64(b.N), "objects/op")
+	b.ReportMetric(float64(pageCount.Load())/float64(b.N), "pages/op")
+}
+
+func runBenchmarkStoreList(ctx context.Context, b *testing.B, store storage.Interface, limit int64, maxRV int, match metav1.ResourceVersionMatch, perNode bool, nodeNames []string) {
+	wg := sync.WaitGroup{}
+	objectCount := atomic.Uint64{}
+	pageCount := atomic.Uint64{}
+	for i := 0; i < b.N; i++ {
+		wg.Add(1)
+		resourceVersion := ""
+		switch match {
+		case metav1.ResourceVersionMatchExact, metav1.ResourceVersionMatchNotOlderThan:
+			resourceVersion = fmt.Sprintf("%d", maxRV-99+i%100)
+		}
+		go func(resourceVersion string) {
+			defer wg.Done()
+			opts := storage.ListOptions{
+				Recursive:            true,
+				ResourceVersion:      resourceVersion,
+				ResourceVersionMatch: match,
+				Predicate: storage.SelectionPredicate{
+					GetAttrs: podAttr,
+					Label:    labels.Everything(),
+					Field:    fields.Everything(),
+					Limit:    limit,
+				},
+			}
+			if perNode {
+				for _, nodeName := range nodeNames {
+					opts.Predicate.GetAttrs = podAttr
+					opts.Predicate.IndexFields = []string{"spec.nodeName"}
+					opts.Predicate.Field = fields.SelectorFromSet(fields.Set{"spec.nodeName": nodeName})
+					objects, pages := paginate(ctx, store, "/pods/", opts)
+					objectCount.Add(uint64(objects))
+					pageCount.Add(uint64(pages))
+				}
+			} else {
+				objects, pages := paginate(ctx, store, "/pods/", opts)
+				objectCount.Add(uint64(objects))
+				pageCount.Add(uint64(pages))
+			}
+		}(resourceVersion)
+	}
+	wg.Wait()
+	b.ReportMetric(float64(objectCount.Load())/float64(b.N), "objects/op")
+	b.ReportMetric(float64(pageCount.Load())/float64(b.N), "pages/op")
+}
+
+func paginate(ctx context.Context, store storage.Interface, key string, opts storage.ListOptions) (objectCount int, pageCount int) {
+	listOut := &example.PodList{}
+	err := store.GetList(ctx, key, opts, listOut)
+	if err != nil {
+		panic(fmt.Sprintf("Unexpected error %s", err))
+	}
+	opts.Predicate.Continue = listOut.Continue
+	opts.ResourceVersion = ""
+	opts.ResourceVersionMatch = ""
+	pageCount += 1
+	objectCount += len(listOut.Items)
+	for opts.Predicate.Continue != "" {
+		listOut := &example.PodList{}
+		err := store.GetList(ctx, key, opts, listOut)
+		if err != nil {
+			panic(fmt.Sprintf("Unexpected error %s", err))
+		}
+		opts.Predicate.Continue = listOut.Continue
+		pageCount += 1
+		objectCount += len(listOut.Items)
+	}
+	return objectCount, pageCount
+}
+
+func podAttr(obj runtime.Object) (labels.Set, fields.Set, error) {
+	pod := obj.(*example.Pod)
+	return nil, fields.Set{
+		"spec.nodeName": pod.Spec.NodeName,
+	}, nil
+}
+
+func prepareBenchchmarkData(ctx context.Context, store storage.Interface, namespaceCount, podPerNamespaceCount, nodeCount int) (namespaceNames, nodeNames []string) {
+	nodeNames = make([]string, nodeCount)
+	for i := 0; i < nodeCount; i++ {
+		nodeNames[i] = rand.String(100)
+	}
+	namespaceNames = make([]string, nodeCount)
+	out := &example.Pod{}
+	for i := 0; i < namespaceCount; i++ {
+		namespace := rand.String(100)
+		namespaceNames[i] = namespace
+		for j := 0; j < podPerNamespaceCount; j++ {
+			name := rand.String(100)
+			pod := &example.Pod{ObjectMeta: metav1.ObjectMeta{Namespace: namespace, Name: name}, Spec: example.PodSpec{NodeName: nodeNames[rand.Intn(nodeCount)]}}
+			err := store.Create(ctx, computePodKey(pod), pod, out, 0)
+			if err != nil {
+				panic(fmt.Sprintf("Unexpected error %s", err))
+			}
+		}
+	}
+	return namespaceNames, nodeNames
+}