forked from crossplane/crossplane
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
improve trace client performance using concurrent resource load
fixes crossplane#5707 Add a loader with configurable concurrency to load resources in concurrent manner. The xrm client delegates to the loader for resource load and supports a functional option to set the concurrency. Add a `--concurrency` flag for the `crank beta trace` command and configure the xrm client appropriately. Signed-off-by: gotwarlost <kananthmail-github@yahoo.com>
- Loading branch information
1 parent
6292789
commit b5d8a44
Showing
4 changed files
with
315 additions
and
20 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,129 @@ | ||
| /* | ||
| Copyright 2023 The Crossplane 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 xrm | ||
|
|
||
| import ( | ||
| "context" | ||
| "sync" | ||
|
|
||
| v1 "k8s.io/api/core/v1" | ||
|
|
||
| "github.com/crossplane/crossplane/cmd/crank/beta/trace/internal/resource" | ||
| ) | ||
|
|
||
| // channelCapacity is the buffer size of the processing channel, should be a high value | ||
| // so that there is no blocking. Correctness of processing does not depend on the channel capacity. | ||
| var channelCapacity = 1000 //nolint:gochecknoglobals // we treat this as constant only overrideable for tests. | ||
|
|
||
| // workItem maintains the relationship of a resource to be loaded with its parent | ||
| // such that the resource that is loaded can be added as a child. | ||
| type workItem struct { | ||
| parent *resource.Resource | ||
| child v1.ObjectReference | ||
| } | ||
|
|
||
| // resourceLoader is a delegate that loads resources and returns child resource refs. | ||
| type resourceLoader interface { | ||
| loadResource(ctx context.Context, ref *v1.ObjectReference) *resource.Resource | ||
| getResourceChildrenRefs(_ context.Context, r *resource.Resource) []v1.ObjectReference | ||
| } | ||
|
|
||
| // loader loads resources concurrently. | ||
| type loader struct { | ||
| root *resource.Resource // the root resource for which the tree is loaded | ||
| l resourceLoader // the resource loader | ||
| resourceLock sync.Mutex // lock when updating the children of any resource | ||
| processing sync.WaitGroup // "counter" to track requests in flight | ||
| ch chan workItem // processing channel | ||
| done chan struct{} // done channel, signaled when all resources are loaded | ||
| } | ||
|
|
||
| // newLoader creates a loader for the root resource. | ||
| func newLoader(root *resource.Resource, rl resourceLoader) *loader { | ||
| l := &loader{ | ||
| l: rl, | ||
| ch: make(chan workItem, channelCapacity), | ||
| done: make(chan struct{}), | ||
| root: root, | ||
| } | ||
| return l | ||
| } | ||
|
|
||
| // load loads the full resource tree in a concurrent manner. | ||
| func (l *loader) load(ctx context.Context, concurrency int) { | ||
| // make sure counters are incremented for root child refs before starting concurrent processing | ||
| refs := l.l.getResourceChildrenRefs(ctx, l.root) | ||
| l.addRefs(l.root, refs) | ||
|
|
||
| // signal the done channel after all items are processed | ||
| go func() { | ||
| l.processing.Wait() | ||
| close(l.done) | ||
| }() | ||
|
|
||
| if concurrency < 1 { | ||
| concurrency = defaultConcurrency | ||
| } | ||
| var wg sync.WaitGroup | ||
| for i := 0; i < concurrency; i++ { | ||
| wg.Add(1) | ||
| go func() { | ||
| defer wg.Done() | ||
| for { | ||
| select { | ||
| case <-l.done: | ||
| return | ||
| case item := <-l.ch: | ||
| l.processItem(ctx, item) | ||
| } | ||
| } | ||
| }() | ||
| } | ||
| wg.Wait() | ||
| } | ||
|
|
||
| // addRefs adds work items to the queue. | ||
| func (l *loader) addRefs(parent *resource.Resource, refs []v1.ObjectReference) { | ||
| // ensure counters are updated synchronously | ||
| l.processing.Add(len(refs)) | ||
| // free up the current processing routine even if the channel blocks. | ||
| go func() { | ||
| for _, ref := range refs { | ||
| l.ch <- workItem{ | ||
| parent: parent, | ||
| child: ref, | ||
| } | ||
| } | ||
| }() | ||
| } | ||
|
|
||
| // processItem processes a single work item in the queue and decrements the in-process counter | ||
| // after adding child references. | ||
| func (l *loader) processItem(ctx context.Context, item workItem) { | ||
| defer l.processing.Done() | ||
| res := l.l.loadResource(ctx, &item.child) | ||
| refs := l.l.getResourceChildrenRefs(ctx, res) | ||
| l.updateChild(item, res) | ||
| l.addRefs(res, refs) | ||
| } | ||
|
|
||
| // updateChild adds the supplied child resource to its parent. | ||
| func (l *loader) updateChild(item workItem, res *resource.Resource) { | ||
| l.resourceLock.Lock() | ||
| item.parent.Children = append(item.parent.Children, res) | ||
| l.resourceLock.Unlock() | ||
| } |
151 changes: 151 additions & 0 deletions
151
cmd/crank/beta/trace/internal/resource/xrm/loader_test.go
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,151 @@ | ||
| package xrm | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
| "math/rand" | ||
| "regexp" | ||
| "strconv" | ||
| "testing" | ||
|
|
||
| v1 "k8s.io/api/core/v1" | ||
| "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" | ||
|
|
||
| "github.com/crossplane/crossplane/cmd/crank/beta/trace/internal/resource" | ||
| ) | ||
|
|
||
| var reNum = regexp.MustCompile(`-(\d+)$`) | ||
|
|
||
| type simpleGenerator struct { | ||
| childDepth int | ||
| numItems int | ||
| } | ||
|
|
||
| func (d *simpleGenerator) createResource(apiVersion, kind, name string) *resource.Resource { | ||
| obj := map[string]any{ | ||
| "apiVersion": apiVersion, | ||
| "kind": kind, | ||
| "metadata": map[string]any{ | ||
| "name": name, | ||
| }, | ||
| } | ||
| return &resource.Resource{Unstructured: unstructured.Unstructured{Object: obj}} | ||
| } | ||
|
|
||
| func (d *simpleGenerator) createRefAtDepth(depth int) v1.ObjectReference { | ||
| prefix := "comp-res" | ||
| if depth == d.childDepth { | ||
| prefix = "managed-res" | ||
| } | ||
| return v1.ObjectReference{ | ||
| Kind: fmt.Sprintf("Depth%d", depth), | ||
| Name: fmt.Sprintf("%s-%d-%d", prefix, rand.Int(), depth), | ||
| APIVersion: "example.com/v1", | ||
| } | ||
| } | ||
|
|
||
| func (d *simpleGenerator) createResourceFromRef(ref *v1.ObjectReference) *resource.Resource { | ||
| return d.createResource(ref.APIVersion, ref.Kind, ref.Name) | ||
| } | ||
|
|
||
| func (d *simpleGenerator) loadResource(_ context.Context, ref *v1.ObjectReference) *resource.Resource { | ||
| return d.createResourceFromRef(ref) | ||
| } | ||
|
|
||
| func (d *simpleGenerator) depthFromResource(res *resource.Resource) int { | ||
| ret := 0 | ||
| matches := reNum.FindStringSubmatch(res.Unstructured.GetName()) | ||
| if len(matches) > 0 { | ||
| n, err := strconv.Atoi(matches[1]) | ||
| if err != nil { | ||
| panic(err) | ||
| } | ||
| ret = n | ||
| } | ||
| return ret | ||
| } | ||
|
|
||
| func (d *simpleGenerator) getResourceChildrenRefs(_ context.Context, r *resource.Resource) []v1.ObjectReference { | ||
| depth := d.depthFromResource(r) | ||
| if depth == d.childDepth { | ||
| return nil | ||
| } | ||
| var ret []v1.ObjectReference | ||
| for i := 0; i < d.numItems; i++ { | ||
| ret = append(ret, d.createRefAtDepth(depth+1)) | ||
| } | ||
| return ret | ||
| } | ||
|
|
||
| var _ resourceLoader = &simpleGenerator{} | ||
|
|
||
| func countItems(root *resource.Resource) int { | ||
| ret := 1 | ||
| for _, child := range root.Children { | ||
| ret += countItems(child) | ||
| } | ||
| return ret | ||
| } | ||
|
|
||
| func TestLoader(t *testing.T) { | ||
| tests := []struct { | ||
| name string | ||
| childDepth int | ||
| numItems int | ||
| channelCapacity int | ||
| concurrency int | ||
| expectedResources int | ||
| }{ | ||
| { | ||
| name: "simple", | ||
| childDepth: 3, | ||
| numItems: 3, | ||
| expectedResources: 1 + 3 + 9 + 27, | ||
| }, | ||
| { | ||
| name: "blocking buffer", | ||
| channelCapacity: 1, | ||
| concurrency: 1, | ||
| childDepth: 3, | ||
| numItems: 10, | ||
| expectedResources: 1 + 10 + 100 + 1000, | ||
| }, | ||
| { | ||
| name: "no children at root", | ||
| childDepth: 0, | ||
| numItems: 0, | ||
| expectedResources: 1, | ||
| }, | ||
| { | ||
| name: "uses default concurrency", | ||
| concurrency: -1, | ||
| childDepth: 3, | ||
| numItems: 3, | ||
| expectedResources: 1 + 3 + 9 + 27, | ||
| }, | ||
| } | ||
|
|
||
| for _, test := range tests { | ||
| t.Run(test.name, func(t *testing.T) { | ||
| orig := channelCapacity | ||
| defer func() { channelCapacity = orig }() | ||
|
|
||
| if test.channelCapacity > 0 { | ||
| channelCapacity = test.channelCapacity | ||
| } | ||
| concurrency := defaultConcurrency | ||
| if test.concurrency != 0 { | ||
| concurrency = test.concurrency | ||
| } | ||
| sg := &simpleGenerator{childDepth: test.childDepth, numItems: test.numItems} | ||
| rootRef := sg.createRefAtDepth(0) | ||
| root := sg.createResourceFromRef(&rootRef) | ||
| l := newLoader(root, sg) | ||
| l.load(context.Background(), concurrency) | ||
| n := countItems(root) | ||
| if test.expectedResources != n { | ||
| t.Errorf("resource count mismatch: want %d, got %d", test.expectedResources, n) | ||
| } | ||
| }) | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters