fix: prevent xdsIR updates from overwriting RateLimit configs from other xdsIR #3771
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,244 @@ | ||
| // Copyright Envoy Gateway Authors | ||
| // SPDX-License-Identifier: Apache-2.0 | ||
| // The full text of the Apache license is available in the LICENSE file at | ||
| // the root of the repo. | ||
|
|
||
| package runner | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
| "testing" | ||
| "time" | ||
|
|
||
| cachetypes "github.com/envoyproxy/go-control-plane/pkg/cache/types" | ||
| cachev3 "github.com/envoyproxy/go-control-plane/pkg/cache/v3" | ||
| resourcev3 "github.com/envoyproxy/go-control-plane/pkg/resource/v3" | ||
| rlsconfv3 "github.com/envoyproxy/go-control-plane/ratelimit/config/ratelimit/v3" | ||
| "github.com/google/go-cmp/cmp" | ||
| "github.com/google/go-cmp/cmp/cmpopts" | ||
| "github.com/stretchr/testify/assert" | ||
| "github.com/stretchr/testify/require" | ||
|
|
||
| egv1a1 "github.com/envoyproxy/gateway/api/v1alpha1" | ||
| "github.com/envoyproxy/gateway/internal/envoygateway/config" | ||
| "github.com/envoyproxy/gateway/internal/infrastructure/kubernetes/ratelimit" | ||
| "github.com/envoyproxy/gateway/internal/ir" | ||
| "github.com/envoyproxy/gateway/internal/message" | ||
| ) | ||
|
|
||
| func Test_subscribeAndTranslate(t *testing.T) { | ||
| t.Parallel() | ||
|
|
||
| testxds := func(gwName string) *ir.Xds { | ||
| return &ir.Xds{ | ||
| HTTP: []*ir.HTTPListener{ | ||
| { | ||
| CoreListenerDetails: ir.CoreListenerDetails{ | ||
| Name: fmt.Sprintf("default/%s/listener-0", gwName), | ||
| }, | ||
| Routes: []*ir.HTTPRoute{ | ||
| { | ||
| Name: "route-0", | ||
| Traffic: &ir.TrafficFeatures{ | ||
| RateLimit: &ir.RateLimit{ | ||
| Global: &ir.GlobalRateLimit{ | ||
| Rules: []*ir.RateLimitRule{ | ||
| { | ||
| HeaderMatches: []*ir.StringMatch{ | ||
| { | ||
| Name: "x-user-id", | ||
| Distinct: true, | ||
| }, | ||
| }, | ||
| Limit: ir.RateLimitValue{ | ||
| Requests: 100, | ||
| Unit: ir.RateLimitUnit(egv1a1.RateLimitUnitMinute), | ||
| }, | ||
| }, | ||
| { | ||
| HeaderMatches: []*ir.StringMatch{ | ||
| { | ||
| Name: "x-another-user-id", | ||
| Distinct: true, | ||
| }, | ||
| }, | ||
| Limit: ir.RateLimitValue{ | ||
| Requests: 10, | ||
| Unit: ir.RateLimitUnit(egv1a1.RateLimitUnitSecond), | ||
| }, | ||
| }, | ||
| }, | ||
| }, | ||
| }, | ||
| }, | ||
| }, | ||
| { | ||
| Name: "route-1", | ||
| Traffic: &ir.TrafficFeatures{ | ||
| RateLimit: &ir.RateLimit{ | ||
| Global: &ir.GlobalRateLimit{ | ||
| Rules: []*ir.RateLimitRule{ | ||
| { | ||
| HeaderMatches: []*ir.StringMatch{ | ||
| { | ||
| Name: "x-user-id", | ||
| Distinct: true, | ||
| }, | ||
| }, | ||
| Limit: ir.RateLimitValue{ | ||
| Requests: 100, | ||
| Unit: ir.RateLimitUnit(egv1a1.RateLimitUnitMinute), | ||
| }, | ||
| }, | ||
| }, | ||
| }, | ||
| }, | ||
| }, | ||
| }, | ||
| }, | ||
| }, | ||
| }, | ||
| } | ||
| } | ||
|
|
||
| testRateLimitConfig := func(gwName string) *rlsconfv3.RateLimitConfig { | ||
| return &rlsconfv3.RateLimitConfig{ | ||
| Name: fmt.Sprintf("default/%s/listener-0", gwName), | ||
| Domain: fmt.Sprintf("default/%s/listener-0", gwName), | ||
| Descriptors: []*rlsconfv3.RateLimitDescriptor{ | ||
| { | ||
| Key: "route-0", | ||
| Value: "route-0", | ||
| Descriptors: []*rlsconfv3.RateLimitDescriptor{ | ||
| { | ||
| Key: "rule-0-match-0", | ||
| RateLimit: &rlsconfv3.RateLimitPolicy{ | ||
| Unit: rlsconfv3.RateLimitUnit_MINUTE, | ||
| RequestsPerUnit: 100, | ||
| }, | ||
| }, | ||
| { | ||
| Key: "rule-1-match-0", | ||
| RateLimit: &rlsconfv3.RateLimitPolicy{ | ||
| Unit: rlsconfv3.RateLimitUnit_SECOND, | ||
| RequestsPerUnit: 10, | ||
| }, | ||
| }, | ||
| }, | ||
| }, | ||
| { | ||
| Key: "route-1", | ||
| Value: "route-1", | ||
| Descriptors: []*rlsconfv3.RateLimitDescriptor{ | ||
| { | ||
| Key: "rule-0-match-0", | ||
| RateLimit: &rlsconfv3.RateLimitPolicy{ | ||
| Unit: rlsconfv3.RateLimitUnit_MINUTE, | ||
| RequestsPerUnit: 100, | ||
| }, | ||
| }, | ||
| }, | ||
| }, | ||
| }, | ||
| } | ||
| } | ||
|
|
||
| testCases := []struct { | ||
| name string | ||
| // xdsIRs contains a list of xds updates that the runner will receive. | ||
| xdsIRs []message.Update[string, *ir.Xds] | ||
| wantRateLimitConfigs map[string]cachetypes.Resource | ||
| }{ | ||
| { | ||
| name: "one xds is added", | ||
| xdsIRs: []message.Update[string, *ir.Xds]{ | ||
| { | ||
| Key: "gw0", | ||
| Value: testxds("gw0"), | ||
| }, | ||
| }, | ||
| wantRateLimitConfigs: map[string]cachetypes.Resource{ | ||
| "default/gw0/listener-0": testRateLimitConfig("gw0"), | ||
| }, | ||
| }, | ||
| { | ||
| name: "two xds are added", | ||
| xdsIRs: []message.Update[string, *ir.Xds]{ | ||
| { | ||
| Key: "gw0", | ||
| Value: testxds("gw0"), | ||
| }, | ||
| { | ||
| Key: "gw1", | ||
| Value: testxds("gw1"), | ||
| }, | ||
| }, | ||
| wantRateLimitConfigs: map[string]cachetypes.Resource{ | ||
| "default/gw0/listener-0": testRateLimitConfig("gw0"), | ||
| "default/gw1/listener-0": testRateLimitConfig("gw1"), | ||
| }, | ||
| }, | ||
| { | ||
| name: "one xds is deleted", | ||
| xdsIRs: []message.Update[string, *ir.Xds]{ | ||
| { | ||
| Key: "gw0", | ||
| Value: testxds("gw0"), | ||
| }, | ||
| { | ||
| Key: "gw1", | ||
| Value: testxds("gw1"), | ||
| }, | ||
| { | ||
| Key: "gw0", | ||
| Delete: true, | ||
| }, | ||
| }, | ||
| wantRateLimitConfigs: map[string]cachetypes.Resource{ | ||
| "default/gw1/listener-0": testRateLimitConfig("gw1"), | ||
| }, | ||
| }, | ||
| } | ||
|
|
||
| for _, tt := range testCases { | ||
| tt := tt | ||
| t.Run(tt.name, func(t *testing.T) { | ||
| t.Parallel() | ||
|
|
||
| ctx, cancel := context.WithCancel(context.Background()) | ||
| defer cancel() | ||
| xdsIR := new(message.XdsIR) | ||
| defer xdsIR.Close() | ||
| cfg, err := config.New() | ||
| require.NoError(t, err) | ||
|
|
||
| r := New(&Config{ | ||
| Server: *cfg, | ||
| XdsIR: xdsIR, | ||
| cache: cachev3.NewSnapshotCache(false, cachev3.IDHash{}, nil), | ||
| }) | ||
|
|
||
| go r.subscribeAndTranslate(ctx) | ||
|
|
||
| for _, xds := range tt.xdsIRs { | ||
| if xds.Delete { | ||
| xdsIR.Delete(xds.Key) | ||
| continue | ||
| } | ||
| xdsIR.Store(xds.Key, xds.Value) | ||
| } | ||
|
|
||
| diff := "" | ||
| if !assert.Eventually(t, func() bool { | ||
| rs, err := r.cache.GetSnapshot(ratelimit.InfraName) | ||
| require.NoError(t, err) | ||
|
|
||
| diff = cmp.Diff(tt.wantRateLimitConfigs, rs.GetResources(resourcev3.RateLimitConfigType), cmpopts.IgnoreUnexported(rlsconfv3.RateLimitConfig{}, rlsconfv3.RateLimitDescriptor{}, rlsconfv3.RateLimitPolicy{})) | ||
| return diff == "" | ||
| }, time.Second*1, time.Millisecond*20) { | ||
| t.Fatalf("snapshot mismatch (-want +got):\n%s", diff) | ||
| } | ||
| }) | ||
| } | ||
| } |
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
is there a way to not maintain this cache here and instead create an API like
rCache := r.getSnapshot(ctx)instead ? this will allow us to reduce one copy in memoryThere was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Let me try; depending on the complexity that would give though, that looks like a better idea.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
The change would be like this sanposhiho@a9f0a19
So, either way, we have to have a cache to track which update.Key generates which rate limit configurations. (A cache size would be smaller than the current cache of entire configs, that's one good point though.)
The cache is required because otherwise we wouldn't know which part of snapshot we have to update. (A
message.Updatecomes fromwatchable.Snapshotdoesn't have a previous state of ir.)So, I'm not very motivated to proceed that way; For me, it looks like only a little gain (less memory usage) would come with a certain downside (complexity).
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
hey @sanposhiho thanks for writing up the quick diff, what I was suggesting was a little different, instead of holding the entire cache on the stack using
rateLimitConfigsCachewhich incurs a constant memory hit, can we retrieve the cache snapshot using an API likehttps://github.com/envoyproxy/go-control-plane/blob/1da4500d00e270d803caefbe0c20e4d3d162e586/pkg/cache/v3/snapshot.go#L124
every time we get a new new message, and then update it (add or delete key based on type of update) ?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Sorry, what's the difference from my diff sanposhiho@a9f0a19? It fetches the previous snapshot with GetSnapshot -> GetResources, and try to rebuild a new snapshot from there already.
My point is that we, either way, need
keyToRateLimitCfgcache to know whichir.Keyhas generated which part of snapshot so that we can calculate a correct snapshot after update.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I missed that 🙈
lets go ahead with diff to keep the code complexity low and we can revisit these if/when we hit a memory bottleneck