fix(router): preserve limiter state across reconciliation events - #1014
Open
nXtCyberNet wants to merge 7 commits into
Open
fix(router): preserve limiter state across reconciliation events#1014nXtCyberNet wants to merge 7 commits into
nXtCyberNet wants to merge 7 commits into
Conversation
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Code Review
This pull request refactors the rate limiting implementation to preserve limiter state during configuration updates by introducing a LimiterConfig struct and a change detection mechanism. The review feedback highlights critical logic issues in the configChanged helper, specifically its failure to detect Redis configuration changes and the unnecessary coupling of input and output limit updates. Furthermore, the reviewer suggests centralizing and improving the Redis client initialization to correctly handle address updates and reduce code duplication.
Comment on lines
+160
to
+183
| configChanged := func(oldConfig *LimiterConfig) bool { | ||
| if oldConfig == nil { | ||
| return true // No existing config, so it's "changed" | ||
| } | ||
|
|
||
| // Create global rate limiters | ||
| if ratelimit.InputTokensPerUnit != nil { | ||
| r.inputLimiter[model] = NewGlobalRateLimiter( | ||
| r.redisClient, | ||
| "kthena:ratelimit", | ||
| model, | ||
| "input", | ||
| *ratelimit.InputTokensPerUnit, | ||
| ratelimit.Unit, | ||
| ) | ||
| if oldConfig.Unit != ratelimit.Unit { | ||
| return true | ||
| } | ||
| // Compare token limits | ||
| oldInput := oldConfig.InputTokensPerUnit | ||
| oldOutput := oldConfig.OutputTokensPerUnit | ||
| if (oldInput == nil) != (ratelimit.InputTokensPerUnit == nil) { | ||
| return true | ||
| } | ||
| if oldInput != nil && ratelimit.InputTokensPerUnit != nil && *oldInput != *ratelimit.InputTokensPerUnit { | ||
| return true | ||
| } | ||
| if (oldOutput == nil) != (ratelimit.OutputTokensPerUnit == nil) { | ||
| return true | ||
| } | ||
| if oldOutput != nil && ratelimit.OutputTokensPerUnit != nil && *oldOutput != *ratelimit.OutputTokensPerUnit { | ||
| return true | ||
| } | ||
| return false | ||
| } |
Contributor
There was a problem hiding this comment.
The configChanged helper function has two logic issues:
- Missing Redis Change Detection: It does not check for changes in the Redis configuration (e.g., switching between local and global modes, or changing the Redis address). This means such configuration updates will be ignored and the limiter state will not be reset when it should be.
- Coupled Input/Output Limits: It compares both
InputTokensPerUnitandOutputTokensPerUnit. Since this function is used to independently decide whether to recreate the input and output limiters, a change in the output limit will unnecessarily trigger a recreation (and state reset) of the input limiter, and vice versa.
I suggest refactoring it to take the specific limit being checked and to include the Redis address in the comparison.
// Helper function to check if config has changed for a specific limiter
configChanged := func(oldConfig *LimiterConfig, newLimit *uint32, isInput bool) bool {
if oldConfig == nil {
return true
}
if oldConfig.Unit != ratelimit.Unit {
return true
}
// Check if switching between local/global or changing redis address
newRedisAddr := ""
if useGlobal {
newRedisAddr = ratelimit.Global.Redis.Address
}
if oldConfig.GlobalRedisAddress != newRedisAddr {
return true
}
var oldLimit *uint32
if isInput {
oldLimit = oldConfig.InputTokensPerUnit
} else {
oldLimit = oldConfig.OutputTokensPerUnit
}
if (oldLimit == nil) != (newLimit == nil) {
return true
}
if oldLimit != nil && newLimit != nil && *oldLimit != *newLimit {
return true
}
return false
}| defer r.mutex.Unlock() | ||
|
|
||
| // Determine if we should use global or local rate limiting | ||
| useGlobal := ratelimit.Global != nil && ratelimit.Global.Redis != nil |
Contributor
There was a problem hiding this comment.
The Redis client initialization should be moved here and updated to handle address changes. The current implementation (lines 195 and 249) only initializes the client if it is nil, which means it won't pick up a new Redis address if the configuration is updated. Moving it here also removes duplication between the input and output limiter blocks.
useGlobal := ratelimit.Global != nil && ratelimit.Global.Redis != nil
if useGlobal {
addr := ratelimit.Global.Redis.Address
if r.redisClient == nil || r.redisClient.Options().Addr != addr {
r.redisClient = redis.NewClient(&redis.Options{
Addr: addr,
})
// Test connection
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
if err := r.redisClient.Ping(ctx).Err(); err != nil {
return fmt.Errorf("failed to connect to redis: %w", err)
}
}
}| // Process input rate limiter | ||
| if ratelimit.InputTokensPerUnit != nil { | ||
| inputConfig, exists := r.inputConfigs[model] | ||
| if !exists || configChanged(inputConfig) { |
Contributor
There was a problem hiding this comment.
Update the call to the refactored configChanged function to pass the input-specific limit and the isInput flag.
Suggested change
| if !exists || configChanged(inputConfig) { | |
| if !exists || configChanged(inputConfig, ratelimit.InputTokensPerUnit, true) { |
| // Process output rate limiter | ||
| if ratelimit.OutputTokensPerUnit != nil { | ||
| outputConfig, exists := r.outputConfigs[model] | ||
| if !exists || configChanged(outputConfig) { |
Contributor
There was a problem hiding this comment.
Update the call to the refactored configChanged function to pass the output-specific limit and the isInput flag.
Suggested change
| if !exists || configChanged(outputConfig) { | |
| if !exists || configChanged(outputConfig, ratelimit.OutputTokensPerUnit, false) { |
Contributor
There was a problem hiding this comment.
Pull request overview
Fixes router rate-limiter resets triggered by Kubernetes reconciliation by making limiter updates config-aware so that limiter instances (and their token-bucket state) are preserved when the effective rate-limit configuration is unchanged.
Changes:
- Introduces a
LimiterConfigwrapper to track limiter instances alongside their configuration for change detection. - Updates
TokenRateLimiterto storeinputConfigs/outputConfigsinstead of raw limiter maps. - Adds config-change detection in
AddOrUpdateLimiter()to avoid recreating limiters on no-op reconciliations.
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Comment on lines
+159
to
+182
| // Helper function to check if config has changed | ||
| configChanged := func(oldConfig *LimiterConfig) bool { | ||
| if oldConfig == nil { | ||
| return true // No existing config, so it's "changed" | ||
| } | ||
|
|
||
| // Create global rate limiters | ||
| if ratelimit.InputTokensPerUnit != nil { | ||
| r.inputLimiter[model] = NewGlobalRateLimiter( | ||
| r.redisClient, | ||
| "kthena:ratelimit", | ||
| model, | ||
| "input", | ||
| *ratelimit.InputTokensPerUnit, | ||
| ratelimit.Unit, | ||
| ) | ||
| if oldConfig.Unit != ratelimit.Unit { | ||
| return true | ||
| } | ||
| // Compare token limits | ||
| oldInput := oldConfig.InputTokensPerUnit | ||
| oldOutput := oldConfig.OutputTokensPerUnit | ||
| if (oldInput == nil) != (ratelimit.InputTokensPerUnit == nil) { | ||
| return true | ||
| } | ||
| if oldInput != nil && ratelimit.InputTokensPerUnit != nil && *oldInput != *ratelimit.InputTokensPerUnit { | ||
| return true | ||
| } | ||
| if (oldOutput == nil) != (ratelimit.OutputTokensPerUnit == nil) { | ||
| return true | ||
| } | ||
| if oldOutput != nil && ratelimit.OutputTokensPerUnit != nil && *oldOutput != *ratelimit.OutputTokensPerUnit { | ||
| return true | ||
| } | ||
| return false |
Comment on lines
+156
to
+167
| // Determine if we should use global or local rate limiting | ||
| useGlobal := ratelimit.Global != nil && ratelimit.Global.Redis != nil | ||
|
|
||
| if useGlobal { | ||
| // Initialize Redis client if not already done | ||
| if r.redisClient == nil { | ||
| r.redisClient = redis.NewClient(&redis.Options{ | ||
| Addr: ratelimit.Global.Redis.Address, | ||
| }) | ||
|
|
||
| // Test connection | ||
| ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) | ||
| defer cancel() | ||
| if err := r.redisClient.Ping(ctx).Err(); err != nil { | ||
| return fmt.Errorf("failed to connect to redis: %w", err) | ||
| } | ||
| // Helper function to check if config has changed | ||
| configChanged := func(oldConfig *LimiterConfig) bool { | ||
| if oldConfig == nil { | ||
| return true // No existing config, so it's "changed" | ||
| } | ||
|
|
||
| // Create global rate limiters | ||
| if ratelimit.InputTokensPerUnit != nil { | ||
| r.inputLimiter[model] = NewGlobalRateLimiter( | ||
| r.redisClient, | ||
| "kthena:ratelimit", | ||
| model, | ||
| "input", | ||
| *ratelimit.InputTokensPerUnit, | ||
| ratelimit.Unit, | ||
| ) | ||
| if oldConfig.Unit != ratelimit.Unit { | ||
| return true | ||
| } | ||
| // Compare token limits |
Comment on lines
+225
to
+231
| r.inputConfigs[model] = &LimiterConfig{ | ||
| Limiter: newLimiter, | ||
| InputTokensPerUnit: ratelimit.InputTokensPerUnit, | ||
| Unit: ratelimit.Unit, | ||
| GlobalRedisAddress: redisAddr, | ||
| LastUpdateTime: time.Now(), | ||
| } |
| OutputTokensPerUnit *uint32 | ||
| Unit networkingv1alpha1.RateLimitUnit | ||
| GlobalRedisAddress string // for detecting config changes | ||
| LastUpdateTime time.Time |
Comment on lines
149
to
152
| // AddOrUpdateLimiter adds or updates rate limiter for a model | ||
| // Only recreates limiters if the configuration has actually changed, | ||
| // preserving limiter state across reconciliation events | ||
| func (r *TokenRateLimiter) AddOrUpdateLimiter(model string, ratelimit *networkingv1alpha1.RateLimit) error { |
Comment on lines
+216
to
+238
| if r.currentRedisAddr != redisAddr { | ||
| // Redis address changed - need to reinitialize | ||
| r.redisClient = redis.NewClient(&redis.Options{ | ||
| Addr: ratelimit.Global.Redis.Address, | ||
| Addr: redisAddr, | ||
| }) | ||
|
|
||
| // Test connection | ||
| ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) | ||
| defer cancel() | ||
| if err := r.redisClient.Ping(ctx).Err(); err != nil { | ||
| return fmt.Errorf("failed to connect to redis: %w", err) | ||
| r.redisClient = nil | ||
| r.currentRedisAddr = "" | ||
| return fmt.Errorf("failed to connect to redis at %s: %w", redisAddr, err) | ||
| } | ||
| r.currentRedisAddr = redisAddr | ||
| } | ||
|
|
||
| // Create global rate limiters | ||
| if ratelimit.InputTokensPerUnit != nil { | ||
| r.inputLimiter[model] = NewGlobalRateLimiter( | ||
| r.redisClient, | ||
| "kthena:ratelimit", | ||
| model, | ||
| "input", | ||
| *ratelimit.InputTokensPerUnit, | ||
| ratelimit.Unit, | ||
| ) | ||
| } else { | ||
| // Switched from global to local - close Redis client | ||
| if r.redisClient != nil { | ||
| r.redisClient = nil | ||
| r.currentRedisAddr = "" | ||
| } | ||
| } |
Comment on lines
215
to
220
| if useGlobal { | ||
| // Initialize Redis client if not already done | ||
| if r.redisClient == nil { | ||
| if r.currentRedisAddr != redisAddr { | ||
| // Redis address changed - need to reinitialize | ||
| r.redisClient = redis.NewClient(&redis.Options{ | ||
| Addr: ratelimit.Global.Redis.Address, | ||
| Addr: redisAddr, | ||
| }) |
|
|
||
| // InputLimiterConfig wraps LimiterConfig for input-specific storage | ||
| type InputLimiterConfig struct { | ||
| *LimiterConfig |
Member
There was a problem hiding this comment.
Is this variable used in any other packages?
Contributor
Author
There was a problem hiding this comment.
no , i have tested it by running the full test suit , and each on passes ,
| // OutputLimiterConfig wraps LimiterConfig for output-specific storage | ||
| type OutputLimiterConfig struct { | ||
| *LimiterConfig | ||
| HasLimit bool // whether output limit is configured |
Member
There was a problem hiding this comment.
Incidentally, is this Boolean value equivalent to LimiterConfig != nil?
Contributor
Author
There was a problem hiding this comment.
Yes, you are correct. I added it because of the 0 token edge case, where the code can bypass ratelimiter entirely.
Instead of hardcoding 1 as the minimum check, I added this so that if anyone wants to configure a different minimum token limit later, they can do so.
If you think this is not required, I can remove it.
Thanks for the review.
Contributor
Author
There was a problem hiding this comment.
Sorry for the confusion. After thinking about it again, I agree that this is not really needed.
The 0 token case is unnecessary to handle with a separate boolean, and even if it passes through this path, it should not create a real problem here.
I will remove HasLimit and simplify the logic.
Contributor
Author
|
/retest |
Contributor
|
@nXtCyberNet: Cannot trigger testing until a trusted user reviews the PR and leaves an DetailsIn response to this:
Instructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes/test-infra repository. |
Comment on lines
+215
to
+237
| if r.currentRedisAddr != redisAddr { | ||
| // Redis address changed - need to reinitialize | ||
| r.redisClient = redis.NewClient(&redis.Options{ | ||
| Addr: ratelimit.Global.Redis.Address, | ||
| Addr: redisAddr, | ||
| }) | ||
|
|
||
| // Test connection | ||
| ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) | ||
| defer cancel() | ||
| if err := r.redisClient.Ping(ctx).Err(); err != nil { | ||
| return fmt.Errorf("failed to connect to redis: %w", err) | ||
| r.redisClient = nil | ||
| r.currentRedisAddr = "" | ||
| return fmt.Errorf("failed to connect to redis at %s: %w", redisAddr, err) | ||
| } | ||
| r.currentRedisAddr = redisAddr | ||
| } | ||
|
|
||
| // Create global rate limiters | ||
| if ratelimit.InputTokensPerUnit != nil { | ||
| r.inputLimiter[model] = NewGlobalRateLimiter( | ||
| r.redisClient, | ||
| "kthena:ratelimit", | ||
| model, | ||
| "input", | ||
| *ratelimit.InputTokensPerUnit, | ||
| ratelimit.Unit, | ||
| ) | ||
| } else { | ||
| // Switched from global to local - close Redis client | ||
| if r.redisClient != nil { | ||
| r.redisClient = nil | ||
| r.currentRedisAddr = "" | ||
| } | ||
| } |
Comment on lines
215
to
230
| if r.currentRedisAddr != redisAddr { | ||
| // Redis address changed - need to reinitialize | ||
| r.redisClient = redis.NewClient(&redis.Options{ | ||
| Addr: ratelimit.Global.Redis.Address, | ||
| Addr: redisAddr, | ||
| }) | ||
|
|
||
| // Test connection | ||
| ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) | ||
| defer cancel() | ||
| if err := r.redisClient.Ping(ctx).Err(); err != nil { | ||
| return fmt.Errorf("failed to connect to redis: %w", err) | ||
| r.redisClient = nil | ||
| r.currentRedisAddr = "" | ||
| return fmt.Errorf("failed to connect to redis at %s: %w", redisAddr, err) | ||
| } | ||
| r.currentRedisAddr = redisAddr | ||
| } |
Comment on lines
+75
to
+78
| type OutputLimiterConfig struct { | ||
| *LimiterConfig | ||
| HasLimit bool // whether output limit is configured | ||
| } |
Comment on lines
144
to
147
| if hasOutputLimit && outputConfig.HasLimit && outputConfig.Limiter != nil && | ||
| outputConfig.Limiter.Tokens() < 1.0 { | ||
| return &OutputRateLimitExceededError{} | ||
| } |
Comment on lines
+317
to
320
| } else { | ||
| // Output rate limit removed | ||
| delete(r.outputLimiters, model) | ||
| } |
Comment on lines
142
to
145
| if hasOutputLimit && outputConfig.HasLimit && outputConfig.Limiter != nil && | ||
| outputConfig.Limiter.Tokens() < 1.0 { | ||
| return &OutputRateLimitExceededError{} | ||
| } |
Comment on lines
+235
to
+261
| if useGlobal { | ||
| newLimiter = NewGlobalRateLimiter( | ||
| r.redisClient, | ||
| "kthena:ratelimit", | ||
| model, | ||
| "input", | ||
| *ratelimit.InputTokensPerUnit, | ||
| ratelimit.Unit, | ||
| ) | ||
| } else { | ||
| // Create local rate limiter | ||
| duration := getTimeUnitDuration(ratelimit.Unit) | ||
| newLimiter = NewLocalLimiter( | ||
| rate.Limit(float64(*ratelimit.InputTokensPerUnit)/duration.Seconds()), | ||
| int(*ratelimit.InputTokensPerUnit), | ||
| ) | ||
| } | ||
|
|
||
| r.inputLimiters[model] = &InputLimiterConfig{ | ||
| LimiterConfig: &LimiterConfig{ | ||
| Limiter: newLimiter, | ||
| TokensPerUnit: *ratelimit.InputTokensPerUnit, // copied value | ||
| Unit: ratelimit.Unit, | ||
| IsGlobal: useGlobal, | ||
| GlobalRedisAddress: redisAddr, | ||
| }, | ||
| } |
Contributor
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 1 out of 1 changed files in this pull request and generated 11 comments.
Comments suppressed due to low confidence (1)
pkg/kthena-router/filters/ratelimit/ratelimit.go:1
HasLimitappears redundant becauser.outputLimiters[model]entries are only created whenratelimit.OutputTokensPerUnit != niland deleted when the limit is removed. This adds extra state to keep consistent and forces additional checks in the hot path. Consider removingHasLimitand simplifying checks to justhasOutputLimit && outputConfig != nil && outputConfig.Limiter != nil(or equivalent), relying on presence/absence in the map to represent whether the output limit is configured.
/*
Comment on lines
+171
to
+175
| // Initialize Redis client if needed (only happens once per pod lifetime) | ||
| if useGlobal && r.redisClient == nil { | ||
| r.redisClient = redis.NewClient(&redis.Options{ | ||
| Addr: ratelimit.Global.Redis.Address, | ||
| }) |
Comment on lines
164
to
166
| func (r *TokenRateLimiter) AddOrUpdateLimiter(model string, ratelimit *networkingv1alpha1.RateLimit) error { | ||
| r.mutex.Lock() | ||
| defer r.mutex.Unlock() |
Comment on lines
+171
to
+172
| // Initialize Redis client if needed (only happens once per pod lifetime) | ||
| if useGlobal && r.redisClient == nil { |
Comment on lines
+74
to
+78
| // OutputLimiterConfig wraps LimiterConfig for output-specific storage | ||
| type OutputLimiterConfig struct { | ||
| *LimiterConfig | ||
| HasLimit bool // whether output limit is configured | ||
| } |
Comment on lines
+142
to
+143
| if hasOutputLimit && outputConfig.HasLimit && outputConfig.Limiter != nil && | ||
| outputConfig.Limiter.Tokens() < 1.0 { |
Comment on lines
+156
to
+157
| if exists && outputConfig.HasLimit && outputConfig.Limiter != nil { | ||
| outputConfig.Limiter.AllowN(time.Now(), tokenCount) |
Comment on lines
+294
to
+295
| r.outputLimiters[model] = &OutputLimiterConfig{ | ||
| LimiterConfig: &LimiterConfig{ |
Comment on lines
+180
to
+183
| if err := r.redisClient.Ping(ctx).Err(); err != nil { | ||
| r.redisClient = nil | ||
| return fmt.Errorf("failed to connect to redis: %w", err) | ||
| } |
Contributor
Author
|
/retest |
Contributor
|
@nXtCyberNet: Cannot trigger testing until a trusted user reviews the PR and leaves an DetailsIn response to this:
Instructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes/test-infra repository. |
Comment on lines
+176
to
+195
| // Initialize Redis client if needed (only happens once per pod lifetime) | ||
| if useGlobal && r.redisClient == nil { | ||
| r.redisClient = redis.NewClient(&redis.Options{ | ||
| Addr: ratelimit.Global.Redis.Address, | ||
| }) | ||
|
|
||
| // Test connection | ||
| ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) | ||
| defer cancel() | ||
| if err := r.redisClient.Ping(ctx).Err(); err != nil { | ||
| r.redisClient = nil | ||
| return fmt.Errorf("failed to connect to redis: %w", err) | ||
| } | ||
| } | ||
|
|
||
| // Derive redisAddr for comparison in config change detection | ||
| redisAddr := "" | ||
| if useGlobal { | ||
| // Initialize Redis client if not already done | ||
| if r.redisClient == nil { | ||
| r.redisClient = redis.NewClient(&redis.Options{ | ||
| Addr: ratelimit.Global.Redis.Address, | ||
| }) | ||
|
|
||
| // Test connection | ||
| ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) | ||
| defer cancel() | ||
| if err := r.redisClient.Ping(ctx).Err(); err != nil { | ||
| return fmt.Errorf("failed to connect to redis: %w", err) | ||
| } | ||
| redisAddr = ratelimit.Global.Redis.Address | ||
| } |
Comment on lines
140
to
147
| // Check output token rate limit | ||
| // We reserve 1 token to indicate this request will consume output tokens | ||
| // This actually RESERVES a token from the bucket, properly enforcing the rate limit | ||
| if hasOutputLimit && outputConfig.HasLimit && outputConfig.Limiter != nil { | ||
| if !outputConfig.Limiter.AllowN(time.Now(), 1) { | ||
| return &OutputRateLimitExceededError{} | ||
| } | ||
| } |
Comment on lines
158
to
163
| if exists && outputConfig.HasLimit && outputConfig.Limiter != nil { | ||
| // Reserve the remaining tokens (we already reserved 1 in the pre-check) | ||
| if tokenCount > 1 { | ||
| outputConfig.Limiter.AllowN(time.Now(), tokenCount-1) | ||
| } | ||
| } |
Comment on lines
170
to
171
| r.mutex.Lock() | ||
| defer r.mutex.Unlock() |
Comment on lines
+176
to
+189
| // Initialize Redis client if needed (only happens once per pod lifetime) | ||
| if useGlobal && r.redisClient == nil { | ||
| r.redisClient = redis.NewClient(&redis.Options{ | ||
| Addr: ratelimit.Global.Redis.Address, | ||
| }) | ||
|
|
||
| // Test connection | ||
| ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) | ||
| defer cancel() | ||
| if err := r.redisClient.Ping(ctx).Err(); err != nil { | ||
| r.redisClient = nil | ||
| return fmt.Errorf("failed to connect to redis: %w", err) | ||
| } | ||
| } |
| // Check output token rate limit | ||
| // We reserve 1 token to indicate this request will consume output tokens | ||
| // This actually RESERVES a token from the bucket, properly enforcing the rate limit | ||
| if hasOutputLimit && outputConfig.HasLimit && outputConfig.Limiter != nil { |
|
|
||
| if exists { | ||
| outputLimiter.AllowN(time.Now(), tokenCount) | ||
| if exists && outputConfig.HasLimit && outputConfig.Limiter != nil { |
Comment on lines
299
to
314
| r.outputLimiters[model] = &OutputLimiterConfig{ | ||
| LimiterConfig: &LimiterConfig{ | ||
| Limiter: newLimiter, | ||
| TokensPerUnit: *ratelimit.OutputTokensPerUnit, // copied value | ||
| Unit: ratelimit.Unit, | ||
| IsGlobal: useGlobal, | ||
| GlobalRedisAddress: redisAddr, | ||
| }, | ||
| HasLimit: true, | ||
| } | ||
| } | ||
| // If config hasn't changed, keep existing limiter with its state | ||
| } else { | ||
| // Output rate limit removed | ||
| delete(r.outputLimiters, model) | ||
| } |
Contributor
Author
|
@LiZhenCheng9527 any updates? |
Member
@nXtCyberNet If this is the intention, i doubt why make it so complex? My suggestion is trying to check old modelRouter vs newModelRoute's RateLimit in below snippet, if they are equal, do nothing. If not, call store.RegisterCallback("ModelRoute", func(data datastore.EventData) {
switch data.EventType {
case datastore.EventAdd, datastore.EventUpdate:
if data.ModelRoute == nil || data.ModelRoute.Spec.RateLimit == nil {
return
}
klog.Infof("add or update rate limit for model %s", data.ModelName)
// Configure the unified rate limiter for this model
if err := loadRateLimiter.AddOrUpdateLimiter(data.ModelName, data.ModelRoute.Spec.RateLimit); err != nil {
klog.Errorf("failed to configure rate limiter for model %s: %v", data.ModelName, err)
}
case datastore.EventDelete:
klog.Infof("delete rate limit for model %s", data.ModelName)
loadRateLimiter.DeleteLimiter(data.ModelName)
}
}) |
nXtCyberNet
force-pushed
the
issue/ratelimit
branch
from
July 10, 2026 15:41
4de795f to
991f5f2
Compare
| if useGlobal { | ||
| // Initialize Redis client if not already done | ||
| if r.redisClient == nil { | ||
| if r.redisClient != nil { |
Contributor
Author
There was a problem hiding this comment.
sorry i think this is redunctant code , i will remove it
hzxuzhonghu
reviewed
Aug 10, 2026
| } | ||
|
|
||
| appliedLimit := loadRateLimiter.GetAppliedLimiter(data.ModelName) | ||
| if reflect.DeepEqual(appliedLimit, data.ModelRoute.Spec.RateLimit) { |
Member
There was a problem hiding this comment.
Not sure if you can add a ut
Contributor
Author
There was a problem hiding this comment.
could you please take a look
Contributor
|
[APPROVALNOTIFIER] This PR is NOT APPROVED This pull-request has been approved by: The full list of commands accepted by this bot can be found here. DetailsNeeds approval from an approver in each of these files:Approvers can indicate their approval by writing |
Signed-off-by: nXtCyberNet <rohantech2005@gmail.com>
Signed-off-by: nXtCyberNet <rohantech2005@gmail.com>
Signed-off-by: nXtCyberNet <rohantech2005@gmail.com>
Signed-off-by: nXtCyberNet <rohantech2005@gmail.com>
Signed-off-by: nXtCyberNet <rohantech2005@gmail.com>
nXtCyberNet
force-pushed
the
issue/ratelimit
branch
from
August 10, 2026 13:45
25655fb to
9b77fa6
Compare
Contributor
Author
|
/retest |
Contributor
|
@nXtCyberNet: Cannot trigger testing until a trusted user reviews the PR and leaves an DetailsIn response to this:
Instructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes/test-infra repository. |
This file contains hidden or 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
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
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.
What type of PR is this?
/kind bug
What this PR does / why we need it:
This PR fixes a rate-limiter state reset issue caused by Kubernetes reconciliation events.
Previously,
AddOrUpdateLimiter()recreated limiter instances unconditionally on every reconciliation, even when theModelRouterate-limit configuration had not changed. This caused the active token bucket state to be discarded, leading to:The fix introduces config-aware reconciliation logic:
This makes reconciliation idempotent and preserves bucket state across non-config updates such as:
Which issue(s) this PR fixes:
Fixes #1013
Special notes for your reviewer:
The root issue was not timing itself but shared mutable limiter state between the reconciliation path and request path without change detection.
This PR intentionally preserves limiter state across reconciliation events unless the effective configuration changes.
Validated against:
VerifyRateLimitResetMechanismDoes this PR introduce a user-facing change?: