You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@dubbo.apache.org by "Chen-BUPT (via GitHub)" <gi...@apache.org> on 2026/02/06 13:58:00 UTC
[PR] feat: Add KVCache offload funtion in PIXIU gateway [dubbo-go-pixiu]
Chen-BUPT opened a new pull request, #878:
URL: https://github.com/apache/dubbo-go-pixiu/pull/878
为了使网关具有智能路由的能力, 我在非/filter/ai文件夹增加了一些代码
具体效果和复现指南请看: https://www.chensadler.blog/article/kvcache3
给reviewer的指南: https://www.chensadler.blog/article/forreviewers
To enable intelligent routing capabilities in the gateway, I added some code under the non-/filter/ai directory.
For the specific behavior and reproduction steps, please refer to:
https://www.chensadler.blog/article/kvcache3
Reviewer guidelines:
https://www.chensadler.blog/article/forreviewers
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
Re: [PR] feat: Add KVCache offload funtion in PIXIU gateway [dubbo-go-pixiu]
Posted by "Chen-BUPT (via GitHub)" <gi...@apache.org>.
Chen-BUPT commented on code in PR #878:
URL: https://github.com/apache/dubbo-go-pixiu/pull/878#discussion_r2777080782
##########
pkg/filter/ai/kvcache/handlers.go:
##########
@@ -0,0 +1,183 @@
+package kvcache
+
+import (
+ "bytes"
+ "context"
+ "encoding/json"
+ "io"
+ "net/http"
+ "strings"
+ "time"
+)
+
+import (
+ "github.com/apache/dubbo-go-pixiu/pkg/common/constant"
+ contexthttp "github.com/apache/dubbo-go-pixiu/pkg/context/http"
+ "github.com/apache/dubbo-go-pixiu/pkg/logger"
+)
+
+func (f *Filter) manageCache(ctx context.Context, model string, prompt string, rawBody []byte, cacheStatus *LookupResponse, lookupDone bool) {
+ if ctx.Err() != nil {
+ return
+ }
+ tokens, err := f.tokenManager.GetTokens(ctx, model, prompt, rawBody)
+ if err != nil {
+ logger.Warnf("[KVCache] tokenize failed: %v", err)
+ return
+ }
+ if ctx.Err() != nil {
+ return
+ }
+ if !lookupDone || cacheStatus == nil {
+ cacheStatus, err = f.lmcacheClient.Lookup(ctx, &LookupRequest{Tokens: tokens})
+ if err != nil {
+ logger.Warnf("[KVCache] lookup failed: %v", err)
+ return
+ }
+ }
+ decision := f.cacheStrategy.MakeDecision(ctx, cacheStatus, model, prompt)
+ if ctx.Err() != nil {
+ return
+ }
+ if err := f.cacheStrategy.ExecuteDecision(ctx, decision, tokens); err != nil {
+ logger.Warnf("[KVCache] execute strategy failed: %v", err)
+ }
+}
+
+func readRequestBody(req *http.Request) ([]byte, error) {
+ if req == nil || req.Body == nil {
+ return nil, nil
+ }
+ if req.GetBody != nil {
+ reader, err := req.GetBody()
+ if err == nil {
+ defer reader.Close()
+ return io.ReadAll(reader)
+ }
+ }
+ bodyBytes, err := io.ReadAll(req.Body)
+ if err != nil {
+ return nil, err
+ }
+ req.Body.Close()
+ req.Body = io.NopCloser(bytes.NewReader(bodyBytes))
+ req.GetBody = func() (io.ReadCloser, error) {
+ return io.NopCloser(bytes.NewReader(bodyBytes)), nil
+ }
+ return bodyBytes, nil
+}
+
+func extractPromptAndModel(body []byte) (string, string, error) {
+ if len(body) == 0 {
+ return "", "", nil
+ }
+ var payload map[string]any
+ if err := json.Unmarshal(body, &payload); err != nil {
+ return "", "", err
+ }
+ model, _ := payload["model"].(string)
+ prompt := coercePrompt(payload["prompt"])
+ if prompt == "" {
+ prompt = extractPromptFromMessages(payload["messages"])
+ }
+ return strings.TrimSpace(prompt), model, nil
+}
+
+func coercePrompt(value any) string {
+ switch v := value.(type) {
+ case string:
+ return v
+ case []any:
+ parts := make([]string, 0, len(v))
+ for _, item := range v {
+ if str, ok := item.(string); ok {
+ parts = append(parts, str)
+ }
+ }
+ return strings.Join(parts, "\n")
+ default:
+ return ""
+ }
+}
+
+func extractPromptFromMessages(value any) string {
+ msgs, ok := value.([]any)
+ if !ok {
+ return ""
+ }
+ parts := make([]string, 0, len(msgs))
+ for _, msg := range msgs {
+ msgMap, ok := msg.(map[string]any)
+ if !ok {
+ continue
+ }
+ if content, ok := msgMap["content"].(string); ok {
+ parts = append(parts, content)
+ }
+ }
+ return strings.Join(parts, "\n")
+}
+
+func selectPreferredInstanceID(resp *LookupResponse) string {
+ if resp == nil || len(resp.LayoutInfo) == 0 {
+ return ""
+ }
+ var (
+ selected string
+ maxCount int
+ )
+ for instanceID, layout := range resp.LayoutInfo {
+ if layout.TokenCount > maxCount || selected == "" {
+ selected = instanceID
+ maxCount = layout.TokenCount
+ }
+ }
+ return selected
+}
+
+func effectiveTimeout(hc *contexthttp.HttpContext, cfg *Config) time.Duration {
+ if cfg == nil {
+ return 0
+ }
+ timeout := cfg.RequestTimeout
+ if hc != nil && hc.Timeout > 0 && (timeout <= 0 || hc.Timeout < timeout) {
+ timeout = hc.Timeout
+ }
+ if timeout <= 0 {
+ return 2 * time.Second
+ }
+ return timeout
+}
+
+func (f *Filter) tryRouteToCachedInstance(hc *contexthttp.HttpContext, model string, prompt string) (*LookupResponse, bool) {
+ if f == nil || f.tokenManager == nil || f.lmcacheClient == nil {
+ return nil, false
+ }
+ tokens, ok := f.tokenManager.GetCachedTokens(model, prompt)
+ if !ok || len(tokens) == 0 {
+ logger.Debugf("[KVCache] routing lookup skipped: token cache miss")
+ return nil, false
+ }
+ timeout := effectiveTimeout(hc, f.cfg)
+ if f.cfg != nil && f.cfg.LookupRoutingTimeout > 0 && f.cfg.LookupRoutingTimeout < timeout {
+ timeout = f.cfg.LookupRoutingTimeout
+ }
+ ctx, cancel := context.WithTimeout(hc.Ctx, timeout)
Review Comment:
fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
Re: [PR] feat: Add KVCache offload funtion in PIXIU gateway [dubbo-go-pixiu]
Posted by "Chen-BUPT (via GitHub)" <gi...@apache.org>.
Chen-BUPT commented on code in PR #878:
URL: https://github.com/apache/dubbo-go-pixiu/pull/878#discussion_r2777076007
##########
pkg/filter/ai/kvcache/token_manager.go:
##########
@@ -0,0 +1,305 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 kvcache
+
+import (
+ "context"
+ "encoding/json"
+ "fmt"
+ "strings"
+ "sync"
+ "sync/atomic"
+ "time"
+)
+
+import (
+ "github.com/go-resty/resty/v2"
+)
+
+type TokenManager struct {
+ httpClient *resty.Client
+ endpoint string
+ cache sync.Map
+ config TokenCacheConfig
+ circuitBreaker *CircuitBreaker
+
+ cacheSize int64
+ hitCount int64
+ missCount int64
+
+ hotWindow time.Duration
+ hotMax int
+ hotMu sync.Mutex
+ hotMap map[string][]time.Time
+}
+
+type TokenizeRequest struct {
+ Model string `json:"model,omitempty"`
+ Prompt string `json:"prompt"`
+}
+
+type TokenizeResponse struct {
+ Count int `json:"count"`
+ Tokens []int `json:"tokens"`
+ MaxLen int `json:"max_model_len"`
+}
+
+type tokenCacheEntry struct {
+ tokens []int
+ expiresAt time.Time
+}
+
+func NewTokenManager(endpoint string, httpClient *resty.Client, cfg TokenCacheConfig, cb *CircuitBreaker, hotWindow time.Duration, hotMax int) *TokenManager {
+ return &TokenManager{
+ httpClient: httpClient,
+ endpoint: endpoint,
+ config: cfg,
+ circuitBreaker: cb,
+ hotWindow: hotWindow,
+ hotMax: hotMax,
+ hotMap: make(map[string][]time.Time),
+ }
+}
+
+func (tm *TokenManager) GetTokens(ctx context.Context, model string, prompt string, rawBody []byte) ([]int, error) {
+ cacheKey := tm.cacheKey(model, prompt)
+ if tm.config.Enabled {
+ if tokens, ok := tm.loadCache(cacheKey); ok {
+ atomic.AddInt64(&tm.hitCount, 1)
+ return tokens, nil
+ }
+ atomic.AddInt64(&tm.missCount, 1)
+ }
+
+ tokens, err := tm.tokenize(ctx, model, prompt, rawBody)
+ if err != nil {
+ return nil, err
+ }
+
+ if tm.config.Enabled {
+ tm.storeCache(cacheKey, tokens)
+ }
+ return tokens, nil
+}
+
+func (tm *TokenManager) GetCachedTokens(model string, prompt string) ([]int, bool) {
+ if !tm.config.Enabled {
+ return nil, false
+ }
+ cacheKey := tm.cacheKey(model, prompt)
+ tokens, ok := tm.loadCache(cacheKey)
+ if ok {
+ atomic.AddInt64(&tm.hitCount, 1)
+ } else {
+ atomic.AddInt64(&tm.missCount, 1)
+ }
+ return tokens, ok
+}
+
+func (tm *TokenManager) InvalidateCache(model string, prompt string) {
+ cacheKey := tm.cacheKey(model, prompt)
+ if _, ok := tm.cache.Load(cacheKey); ok {
+ tm.cache.Delete(cacheKey)
+ atomic.AddInt64(&tm.cacheSize, -1)
+ }
+}
+
+func (tm *TokenManager) GetCacheStats() CacheStats {
+ size := atomic.LoadInt64(&tm.cacheSize)
+ hit := atomic.LoadInt64(&tm.hitCount)
+ miss := atomic.LoadInt64(&tm.missCount)
+ total := hit + miss
+ var hitRate float64
+ if total > 0 {
+ hitRate = float64(hit) / float64(total)
+ }
+ return CacheStats{
+ Size: int(size),
+ HitRate: hitRate,
+ HitCount: hit,
+ MissCount: miss,
+ }
+}
+
+func (tm *TokenManager) tokenize(ctx context.Context, model string, prompt string, rawBody []byte) ([]int, error) {
+ var tokens []int
+ err := tm.execute(ctx, func() error {
+ body, err := tm.buildTokenizeBody(model, prompt, rawBody)
+ if err != nil {
+ return err
+ }
+ resp, err := tm.doTokenizeRequest(ctx, body)
+ if err != nil {
+ return err
+ }
+ tokens = resp.Tokens
+ return nil
+ })
+ if err != nil {
+ return nil, err
+ }
+ return tokens, nil
+}
+
+func (tm *TokenManager) buildTokenizeBody(model string, prompt string, rawBody []byte) (any, error) {
+ if len(rawBody) > 0 {
+ return rawBody, nil
+ }
+ return TokenizeRequest{Model: model, Prompt: prompt}, nil
+}
+
+func (tm *TokenManager) doTokenizeRequest(ctx context.Context, body any) (*TokenizeResponse, error) {
+ tokenizeURL := strings.TrimRight(tm.endpoint, "/") + "/tokenize"
+ resp, err := tm.httpClient.R().
+ SetContext(ctx).
+ SetHeader("Content-Type", "application/json").
+ SetBody(body).
+ Post(tokenizeURL)
+ if err != nil {
+ return nil, fmt.Errorf("call tokenize: %w", err)
+ }
+ if resp.StatusCode() < 200 || resp.StatusCode() >= 300 {
+ return nil, fmt.Errorf("tokenize status %d: %s", resp.StatusCode(), strings.TrimSpace(string(resp.Body())))
+ }
+ var tokenResp TokenizeResponse
+ if err := json.Unmarshal(resp.Body(), &tokenResp); err != nil {
+ return nil, fmt.Errorf("decode tokenize response: %w", err)
+ }
+ return &tokenResp, nil
+}
+
+func (tm *TokenManager) RecordHot(model string, prompt string) {
+ if tm == nil || tm.hotWindow <= 0 || model == "" || prompt == "" {
+ return
+ }
+ now := time.Now()
+ key := tm.cacheKey(model, prompt)
+ tm.hotMu.Lock()
+ defer tm.hotMu.Unlock()
+ entries := tm.hotMap[key]
+ entries = append(entries, now)
+ entries = trimHotWindow(entries, now, tm.hotWindow)
+ if tm.hotMax > 0 && len(entries) > tm.hotMax {
+ entries = entries[len(entries)-tm.hotMax:]
+ }
+ if len(entries) == 0 {
+ delete(tm.hotMap, key)
+ return
+ }
+ tm.hotMap[key] = entries
+}
+
+func (tm *TokenManager) IsHot(model string, prompt string, threshold int) bool {
+ if tm == nil || tm.hotWindow <= 0 || threshold <= 0 || model == "" || prompt == "" {
+ return false
+ }
+ now := time.Now()
+ key := tm.cacheKey(model, prompt)
+ tm.hotMu.Lock()
+ defer tm.hotMu.Unlock()
+ entries := tm.hotMap[key]
+ if len(entries) == 0 {
+ return false
+ }
+ entries = trimHotWindow(entries, now, tm.hotWindow)
+ if tm.hotMax > 0 && len(entries) > tm.hotMax {
+ entries = entries[len(entries)-tm.hotMax:]
+ }
+ if len(entries) == 0 {
+ delete(tm.hotMap, key)
+ return false
+ }
+ tm.hotMap[key] = entries
+ return len(entries) >= threshold
+}
+
+func trimHotWindow(entries []time.Time, now time.Time, window time.Duration) []time.Time {
+ if window <= 0 || len(entries) == 0 {
+ return entries
+ }
+ cutoff := now.Add(-window)
+ idx := 0
+ for idx < len(entries) && entries[idx].Before(cutoff) {
+ idx++
+ }
+ if idx == 0 {
+ return entries
+ }
+ return append([]time.Time(nil), entries[idx:]...)
+}
+
+func (tm *TokenManager) execute(ctx context.Context, operation func() error) error {
+ if tm.circuitBreaker == nil {
+ return operation()
+ }
+ return tm.circuitBreaker.Execute(operation)
+}
+
+func (tm *TokenManager) cacheKey(model string, prompt string) string {
+ if model == "" {
+ return prompt
+ }
+ return model + ":" + prompt
+}
+
+func (tm *TokenManager) loadCache(key string) ([]int, bool) {
+ entryAny, ok := tm.cache.Load(key)
+ if !ok {
+ return nil, false
+ }
+ entry, ok := entryAny.(*tokenCacheEntry)
+ if !ok {
+ tm.cache.Delete(key)
+ atomic.AddInt64(&tm.cacheSize, -1)
+ return nil, false
+ }
+ if tm.config.TTL > 0 && time.Now().After(entry.expiresAt) {
+ tm.cache.Delete(key)
+ atomic.AddInt64(&tm.cacheSize, -1)
+ return nil, false
Review Comment:
solved
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
Re: [PR] feat: Add KVCache offload funtion in PIXIU gateway [dubbo-go-pixiu]
Posted by "sonarqubecloud[bot] (via GitHub)" <gi...@apache.org>.
sonarqubecloud[bot] commented on PR #878:
URL: https://github.com/apache/dubbo-go-pixiu/pull/878#issuecomment-3860657172
## [](https://sonarcloud.io/dashboard?id=apache_dubbo-go-pixiu&pullRequest=878) **Quality Gate passed**
Issues
 [7 New issues](https://sonarcloud.io/project/issues?id=apache_dubbo-go-pixiu&pullRequest=878&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)
 [0 Accepted issues](https://sonarcloud.io/project/issues?id=apache_dubbo-go-pixiu&pullRequest=878&issueStatuses=ACCEPTED)
Measures
 [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=apache_dubbo-go-pixiu&pullRequest=878&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)
 [0.0% Coverage on New Code](https://sonarcloud.io/component_measures?id=apache_dubbo-go-pixiu&pullRequest=878&metric=new_coverage&view=list)
 [0.0% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_dubbo-go-pixiu&pullRequest=878&metric=new_duplicated_lines_density&view=list)
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=apache_dubbo-go-pixiu&pullRequest=878)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
Re: [PR] feat: Add KVCache offload funtion in PIXIU gateway [dubbo-go-pixiu]
Posted by "Alanxtl (via GitHub)" <gi...@apache.org>.
Alanxtl commented on PR #878:
URL: https://github.com/apache/dubbo-go-pixiu/pull/878#issuecomment-3894466506
still need to write docs in dubbo-website and write samples in dubbo-go-pixiu-samples
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
Re: [PR] feat: Add KVCache offload funtion in PIXIU gateway [dubbo-go-pixiu]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #878:
URL: https://github.com/apache/dubbo-go-pixiu/pull/878#discussion_r2776695050
##########
pkg/filter/ai/kvcache/lmcache_client.go:
##########
@@ -0,0 +1,181 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 kvcache
+
+import (
+ "context"
+ "encoding/json"
+ "fmt"
+ "strings"
+ "time"
+)
+
+import (
+ "github.com/go-resty/resty/v2"
+)
+
+type LMCacheClient struct {
+ httpClient *resty.Client
+ baseURL string
+ retry RetryConfig
+ circuitBreaker *CircuitBreaker
+}
+
+type PinRequest struct {
+ Tokens []int `json:"tokens"`
+ InstanceID string `json:"instance_id"`
+ Location string `json:"location"`
+}
+
+type LookupRequest struct {
+ Tokens []int `json:"tokens"`
+}
+
+type CompressRequest struct {
+ Tokens []int `json:"tokens"`
+ InstanceID string `json:"instance_id"`
+ Location string `json:"location"`
+ Method string `json:"method"`
+}
+
+type EvictRequest struct {
+ Tokens []int `json:"tokens"`
+ InstanceID string `json:"instance_id"`
+}
+
+type PinResponse struct {
+ EventID string `json:"event_id"`
+ NumTokens int `json:"num_tokens"`
+}
+
+type CompressResponse struct {
+ EventID string `json:"event_id"`
+ NumTokens int `json:"num_tokens"`
+}
+
+type EvictResponse struct {
+ EventID string `json:"event_id"`
+ NumTokens int `json:"num_tokens"`
+}
+
+func NewLMCacheClient(baseURL string, httpClient *resty.Client, retry RetryConfig, cb *CircuitBreaker) *LMCacheClient {
+ return &LMCacheClient{
+ httpClient: httpClient,
+ baseURL: strings.TrimRight(baseURL, "/"),
+ retry: retry,
+ circuitBreaker: cb,
+ }
+}
+
+func (lc *LMCacheClient) Pin(ctx context.Context, req *PinRequest) (*PinResponse, error) {
+ var resp PinResponse
+ if err := lc.doRequestWithRetry(ctx, "/pin", req, &resp, "pin"); err != nil {
+ return nil, err
+ }
+ return &resp, nil
+}
+
+func (lc *LMCacheClient) Lookup(ctx context.Context, req *LookupRequest) (*LookupResponse, error) {
+ var resp LookupResponse
+ if err := lc.doRequestWithRetry(ctx, "/lookup", req, &resp, "lookup"); err != nil {
+ return nil, err
+ }
+ return &resp, nil
+}
+
+func (lc *LMCacheClient) Compress(ctx context.Context, req *CompressRequest) (*CompressResponse, error) {
+ var resp CompressResponse
+ if err := lc.doRequestWithRetry(ctx, "/compress", req, &resp, "compress"); err != nil {
+ return nil, err
+ }
+ return &resp, nil
+}
+
+func (lc *LMCacheClient) Evict(ctx context.Context, req *EvictRequest) (*EvictResponse, error) {
+ var resp EvictResponse
+ if err := lc.doRequestWithRetry(ctx, "/evict", req, &resp, "evict"); err != nil {
+ return nil, err
+ }
+ return &resp, nil
+}
+
+func (lc *LMCacheClient) doRequestWithRetry(ctx context.Context, path string, payload any, out any, op string) error {
+ var lastErr error
+ for attempt := 0; attempt < lc.retry.MaxAttempts; attempt++ {
Review Comment:
`retry.max_attempts` can be configured as 0 (Validate allows >= 0), but when it is 0 this retry loop does zero iterations and returns `lastErr` which is `nil`, causing callers to treat the request as successful without making any HTTP call. Ensure at least one attempt is executed (e.g., treat 0 as 1 or require `max_attempts >= 1`).
```suggestion
maxAttempts := lc.retry.MaxAttempts
if maxAttempts < 1 {
maxAttempts = 1
}
for attempt := 0; attempt < maxAttempts; attempt++ {
```
##########
pkg/filter/ai/kvcache/strategy.go:
##########
@@ -0,0 +1,113 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 kvcache
+
+import (
+ "context"
+)
+
+type CacheStrategy struct {
+ config CacheStrategyConfig
+ loadMonitor *LoadMonitor
+ lmcacheClient *LMCacheClient
+ tokenManager *TokenManager
+}
+
+type StrategyDecision struct {
+ ShouldCompress bool
+ ShouldPin bool
+ ShouldEvict bool
+ Reason string
+}
+
+func NewCacheStrategy(cfg CacheStrategyConfig, client *LMCacheClient, tokenManager *TokenManager) *CacheStrategy {
+ return &CacheStrategy{
+ config: cfg,
+ loadMonitor: NewLoadMonitor(),
+ lmcacheClient: client,
+ tokenManager: tokenManager,
+ }
+}
+
+func (cs *CacheStrategy) RecordRequest() {
+ if cs == nil || cs.loadMonitor == nil {
+ return
+ }
+ cs.loadMonitor.RecordRequest()
+}
+
+func (cs *CacheStrategy) MakeDecision(_ context.Context, cacheStatus *LookupResponse, model string, prompt string) *StrategyDecision {
+ if cs == nil {
+ return &StrategyDecision{}
+ }
+ decision := &StrategyDecision{}
+ metrics := cs.loadMonitor.Snapshot()
+
+ if cs.config.EnableCompression && cs.config.LoadThreshold > 0 &&
+ (metrics.CPUUsage >= cs.config.LoadThreshold || metrics.RequestRate >= cs.config.LoadThreshold) {
Review Comment:
`LoadThreshold` is validated as a 0..1 ratio, but `LoadMonitor.Snapshot()` sets `RequestRate` to requests/second. Comparing `metrics.RequestRate >= LoadThreshold` will almost always trip for any non-trivial traffic (e.g., threshold 0.7 means 0.7 req/s). Either change `LoadThreshold` semantics/validation to use req/s, or normalize `RequestRate` into 0..1 before comparing.
```suggestion
metrics.CPUUsage >= cs.config.LoadThreshold {
```
##########
pkg/filter/ai/kvcache/filter.go:
##########
@@ -0,0 +1,142 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 kvcache
+
+import (
+ "context"
+ "net/http"
+)
+
+import (
+ "github.com/go-resty/resty/v2"
+)
+
+import (
+ "github.com/apache/dubbo-go-pixiu/pkg/common/constant"
+ "github.com/apache/dubbo-go-pixiu/pkg/common/extension/filter"
+ contexthttp "github.com/apache/dubbo-go-pixiu/pkg/context/http"
+ "github.com/apache/dubbo-go-pixiu/pkg/logger"
+)
+
+const (
+ Kind = constant.AIKVCacheFilter
+)
+
+func init() {
+ filter.RegisterHttpFilter(&Plugin{})
+}
+
+type (
+ Plugin struct{}
+
+ FilterFactory struct {
+ cfg *Config
+ httpClient *http.Client
+ resty *resty.Client
+ tokenManager *TokenManager
+ lmcacheClient *LMCacheClient
+ cacheStrategy *CacheStrategy
+ }
+
+ Filter struct {
+ cfg *Config
+ tokenManager *TokenManager
+ lmcacheClient *LMCacheClient
+ cacheStrategy *CacheStrategy
+ }
+)
+
+func (p *Plugin) Kind() string { return Kind }
+
+func (p *Plugin) CreateFilterFactory() (filter.HttpFilterFactory, error) {
+ return &FilterFactory{cfg: &Config{}}, nil
+}
+
+func (factory *FilterFactory) Config() any { return factory.cfg }
+
+func (factory *FilterFactory) Apply() error {
+ factory.cfg.ApplyDefaults()
+ if err := factory.cfg.Validate(); err != nil {
+ return err
+ }
+ cfg := factory.cfg
+ factory.httpClient = &http.Client{
+ Timeout: cfg.RequestTimeout,
+ Transport: &http.Transport{
+ MaxIdleConns: cfg.MaxIdleConns,
+ MaxIdleConnsPerHost: cfg.MaxIdleConnsPerHost,
+ MaxConnsPerHost: cfg.MaxConnsPerHost,
+ },
+ }
+ factory.resty = resty.NewWithClient(factory.httpClient).
+ SetTimeout(cfg.RequestTimeout)
+
+ cbToken := NewCircuitBreaker(cfg.CircuitBreaker)
+ cbLMCache := NewCircuitBreaker(cfg.CircuitBreaker)
+ factory.tokenManager = NewTokenManager(cfg.VLLMEndpoint, factory.resty, cfg.TokenCache, cbToken, cfg.HotWindow, cfg.HotMaxRecords)
+ factory.lmcacheClient = NewLMCacheClient(cfg.LMCacheEndpoint, factory.resty, cfg.Retry, cbLMCache)
+ factory.cacheStrategy = NewCacheStrategy(cfg.CacheStrategy, factory.lmcacheClient, factory.tokenManager)
+ return nil
+}
+
+func (factory *FilterFactory) PrepareFilterChain(_ *contexthttp.HttpContext, chain filter.FilterChain) error {
+ f := &Filter{
+ cfg: factory.cfg,
+ tokenManager: factory.tokenManager,
+ lmcacheClient: factory.lmcacheClient,
+ cacheStrategy: factory.cacheStrategy,
+ }
+ chain.AppendDecodeFilters(f)
+ return nil
+}
+
+func (f *Filter) Decode(hc *contexthttp.HttpContext) filter.FilterStatus {
+ if f.cfg == nil || !f.cfg.Enabled {
+ return filter.Continue
+ }
+ if f.cacheStrategy != nil {
Review Comment:
This PR adds a new filter with non-trivial routing + async cache-management behavior, but there are no unit tests covering it. Adding focused tests (prompt/model extraction, preferred-endpoint routing override, LMCache retry edge cases) would help prevent regressions.
##########
pkg/filter/ai/kvcache/filter.go:
##########
@@ -0,0 +1,142 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 kvcache
+
+import (
+ "context"
+ "net/http"
+)
+
+import (
+ "github.com/go-resty/resty/v2"
+)
+
+import (
+ "github.com/apache/dubbo-go-pixiu/pkg/common/constant"
+ "github.com/apache/dubbo-go-pixiu/pkg/common/extension/filter"
+ contexthttp "github.com/apache/dubbo-go-pixiu/pkg/context/http"
+ "github.com/apache/dubbo-go-pixiu/pkg/logger"
+)
+
+const (
+ Kind = constant.AIKVCacheFilter
+)
+
+func init() {
+ filter.RegisterHttpFilter(&Plugin{})
+}
+
+type (
+ Plugin struct{}
+
+ FilterFactory struct {
+ cfg *Config
+ httpClient *http.Client
+ resty *resty.Client
+ tokenManager *TokenManager
+ lmcacheClient *LMCacheClient
+ cacheStrategy *CacheStrategy
+ }
+
+ Filter struct {
+ cfg *Config
+ tokenManager *TokenManager
+ lmcacheClient *LMCacheClient
+ cacheStrategy *CacheStrategy
+ }
+)
+
+func (p *Plugin) Kind() string { return Kind }
+
+func (p *Plugin) CreateFilterFactory() (filter.HttpFilterFactory, error) {
+ return &FilterFactory{cfg: &Config{}}, nil
+}
+
+func (factory *FilterFactory) Config() any { return factory.cfg }
+
+func (factory *FilterFactory) Apply() error {
+ factory.cfg.ApplyDefaults()
+ if err := factory.cfg.Validate(); err != nil {
+ return err
+ }
+ cfg := factory.cfg
+ factory.httpClient = &http.Client{
+ Timeout: cfg.RequestTimeout,
+ Transport: &http.Transport{
+ MaxIdleConns: cfg.MaxIdleConns,
+ MaxIdleConnsPerHost: cfg.MaxIdleConnsPerHost,
+ MaxConnsPerHost: cfg.MaxConnsPerHost,
+ },
+ }
+ factory.resty = resty.NewWithClient(factory.httpClient).
+ SetTimeout(cfg.RequestTimeout)
+
+ cbToken := NewCircuitBreaker(cfg.CircuitBreaker)
+ cbLMCache := NewCircuitBreaker(cfg.CircuitBreaker)
+ factory.tokenManager = NewTokenManager(cfg.VLLMEndpoint, factory.resty, cfg.TokenCache, cbToken, cfg.HotWindow, cfg.HotMaxRecords)
+ factory.lmcacheClient = NewLMCacheClient(cfg.LMCacheEndpoint, factory.resty, cfg.Retry, cbLMCache)
+ factory.cacheStrategy = NewCacheStrategy(cfg.CacheStrategy, factory.lmcacheClient, factory.tokenManager)
+ return nil
+}
+
+func (factory *FilterFactory) PrepareFilterChain(_ *contexthttp.HttpContext, chain filter.FilterChain) error {
+ f := &Filter{
+ cfg: factory.cfg,
+ tokenManager: factory.tokenManager,
+ lmcacheClient: factory.lmcacheClient,
+ cacheStrategy: factory.cacheStrategy,
+ }
+ chain.AppendDecodeFilters(f)
+ return nil
+}
+
+func (f *Filter) Decode(hc *contexthttp.HttpContext) filter.FilterStatus {
+ if f.cfg == nil || !f.cfg.Enabled {
+ return filter.Continue
+ }
+ if f.cacheStrategy != nil {
+ f.cacheStrategy.RecordRequest()
+ }
+ body, err := readRequestBody(hc.Request)
+ if err != nil {
+ logger.Warnf("[KVCache] read request body failed: %v", err)
+ return filter.Continue
+ }
+ prompt, model, err := extractPromptAndModel(body)
+ if err != nil {
+ logger.Warnf("[KVCache] parse request body failed: %v", err)
+ return filter.Continue
+ }
+ if prompt == "" {
+ return filter.Continue
+ }
+ if model == "" {
+ model = f.cfg.DefaultModel
+ }
+
+ f.tokenManager.RecordHot(model, prompt)
+
+ cacheStatus, routed := f.tryRouteToCachedInstance(hc, model, prompt)
+
+ ctx, cancel := context.WithTimeout(hc.Ctx, effectiveTimeout(hc, f.cfg))
+ go func() {
+ defer cancel()
+ f.manageCache(ctx, model, prompt, body, cacheStatus, routed)
+ }()
Review Comment:
`context.WithTimeout(hc.Ctx, ...)` uses `hc.Ctx`, which is set to `context.Background()` in the HTTP connection manager, so these background cache-management goroutines won't be canceled when the client request is canceled. Consider deriving this from `hc.Request.Context()` (or ensuring `hc.Ctx` is set from it) so work is aborted promptly and goroutine buildup under load is avoided.
##########
pkg/filter/ai/kvcache/token_manager.go:
##########
@@ -0,0 +1,305 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 kvcache
+
+import (
+ "context"
+ "encoding/json"
+ "fmt"
+ "strings"
+ "sync"
+ "sync/atomic"
+ "time"
+)
+
+import (
+ "github.com/go-resty/resty/v2"
+)
+
+type TokenManager struct {
+ httpClient *resty.Client
+ endpoint string
+ cache sync.Map
+ config TokenCacheConfig
+ circuitBreaker *CircuitBreaker
+
+ cacheSize int64
+ hitCount int64
+ missCount int64
+
+ hotWindow time.Duration
+ hotMax int
+ hotMu sync.Mutex
+ hotMap map[string][]time.Time
+}
+
+type TokenizeRequest struct {
+ Model string `json:"model,omitempty"`
+ Prompt string `json:"prompt"`
+}
+
+type TokenizeResponse struct {
+ Count int `json:"count"`
+ Tokens []int `json:"tokens"`
+ MaxLen int `json:"max_model_len"`
+}
+
+type tokenCacheEntry struct {
+ tokens []int
+ expiresAt time.Time
+}
+
+func NewTokenManager(endpoint string, httpClient *resty.Client, cfg TokenCacheConfig, cb *CircuitBreaker, hotWindow time.Duration, hotMax int) *TokenManager {
+ return &TokenManager{
+ httpClient: httpClient,
+ endpoint: endpoint,
+ config: cfg,
+ circuitBreaker: cb,
+ hotWindow: hotWindow,
+ hotMax: hotMax,
+ hotMap: make(map[string][]time.Time),
+ }
+}
+
+func (tm *TokenManager) GetTokens(ctx context.Context, model string, prompt string, rawBody []byte) ([]int, error) {
+ cacheKey := tm.cacheKey(model, prompt)
+ if tm.config.Enabled {
+ if tokens, ok := tm.loadCache(cacheKey); ok {
+ atomic.AddInt64(&tm.hitCount, 1)
+ return tokens, nil
+ }
+ atomic.AddInt64(&tm.missCount, 1)
+ }
+
+ tokens, err := tm.tokenize(ctx, model, prompt, rawBody)
+ if err != nil {
+ return nil, err
+ }
+
+ if tm.config.Enabled {
+ tm.storeCache(cacheKey, tokens)
+ }
+ return tokens, nil
+}
+
+func (tm *TokenManager) GetCachedTokens(model string, prompt string) ([]int, bool) {
+ if !tm.config.Enabled {
+ return nil, false
+ }
+ cacheKey := tm.cacheKey(model, prompt)
+ tokens, ok := tm.loadCache(cacheKey)
+ if ok {
+ atomic.AddInt64(&tm.hitCount, 1)
+ } else {
+ atomic.AddInt64(&tm.missCount, 1)
+ }
+ return tokens, ok
+}
+
+func (tm *TokenManager) InvalidateCache(model string, prompt string) {
+ cacheKey := tm.cacheKey(model, prompt)
+ if _, ok := tm.cache.Load(cacheKey); ok {
+ tm.cache.Delete(cacheKey)
+ atomic.AddInt64(&tm.cacheSize, -1)
+ }
+}
+
+func (tm *TokenManager) GetCacheStats() CacheStats {
+ size := atomic.LoadInt64(&tm.cacheSize)
+ hit := atomic.LoadInt64(&tm.hitCount)
+ miss := atomic.LoadInt64(&tm.missCount)
+ total := hit + miss
+ var hitRate float64
+ if total > 0 {
+ hitRate = float64(hit) / float64(total)
+ }
+ return CacheStats{
+ Size: int(size),
+ HitRate: hitRate,
+ HitCount: hit,
+ MissCount: miss,
+ }
+}
+
+func (tm *TokenManager) tokenize(ctx context.Context, model string, prompt string, rawBody []byte) ([]int, error) {
+ var tokens []int
+ err := tm.execute(ctx, func() error {
+ body, err := tm.buildTokenizeBody(model, prompt, rawBody)
+ if err != nil {
+ return err
+ }
+ resp, err := tm.doTokenizeRequest(ctx, body)
+ if err != nil {
+ return err
+ }
+ tokens = resp.Tokens
+ return nil
+ })
+ if err != nil {
+ return nil, err
+ }
+ return tokens, nil
+}
+
+func (tm *TokenManager) buildTokenizeBody(model string, prompt string, rawBody []byte) (any, error) {
+ if len(rawBody) > 0 {
+ return rawBody, nil
+ }
+ return TokenizeRequest{Model: model, Prompt: prompt}, nil
+}
+
+func (tm *TokenManager) doTokenizeRequest(ctx context.Context, body any) (*TokenizeResponse, error) {
+ tokenizeURL := strings.TrimRight(tm.endpoint, "/") + "/tokenize"
+ resp, err := tm.httpClient.R().
+ SetContext(ctx).
+ SetHeader("Content-Type", "application/json").
+ SetBody(body).
+ Post(tokenizeURL)
+ if err != nil {
+ return nil, fmt.Errorf("call tokenize: %w", err)
+ }
+ if resp.StatusCode() < 200 || resp.StatusCode() >= 300 {
+ return nil, fmt.Errorf("tokenize status %d: %s", resp.StatusCode(), strings.TrimSpace(string(resp.Body())))
+ }
+ var tokenResp TokenizeResponse
+ if err := json.Unmarshal(resp.Body(), &tokenResp); err != nil {
+ return nil, fmt.Errorf("decode tokenize response: %w", err)
+ }
+ return &tokenResp, nil
+}
+
+func (tm *TokenManager) RecordHot(model string, prompt string) {
+ if tm == nil || tm.hotWindow <= 0 || model == "" || prompt == "" {
+ return
+ }
+ now := time.Now()
+ key := tm.cacheKey(model, prompt)
+ tm.hotMu.Lock()
+ defer tm.hotMu.Unlock()
+ entries := tm.hotMap[key]
+ entries = append(entries, now)
+ entries = trimHotWindow(entries, now, tm.hotWindow)
+ if tm.hotMax > 0 && len(entries) > tm.hotMax {
+ entries = entries[len(entries)-tm.hotMax:]
+ }
+ if len(entries) == 0 {
+ delete(tm.hotMap, key)
+ return
+ }
+ tm.hotMap[key] = entries
+}
+
+func (tm *TokenManager) IsHot(model string, prompt string, threshold int) bool {
+ if tm == nil || tm.hotWindow <= 0 || threshold <= 0 || model == "" || prompt == "" {
+ return false
+ }
+ now := time.Now()
+ key := tm.cacheKey(model, prompt)
+ tm.hotMu.Lock()
+ defer tm.hotMu.Unlock()
+ entries := tm.hotMap[key]
+ if len(entries) == 0 {
+ return false
+ }
+ entries = trimHotWindow(entries, now, tm.hotWindow)
+ if tm.hotMax > 0 && len(entries) > tm.hotMax {
+ entries = entries[len(entries)-tm.hotMax:]
+ }
+ if len(entries) == 0 {
+ delete(tm.hotMap, key)
+ return false
+ }
+ tm.hotMap[key] = entries
+ return len(entries) >= threshold
+}
+
+func trimHotWindow(entries []time.Time, now time.Time, window time.Duration) []time.Time {
+ if window <= 0 || len(entries) == 0 {
+ return entries
+ }
+ cutoff := now.Add(-window)
+ idx := 0
+ for idx < len(entries) && entries[idx].Before(cutoff) {
+ idx++
+ }
+ if idx == 0 {
+ return entries
+ }
+ return append([]time.Time(nil), entries[idx:]...)
+}
+
+func (tm *TokenManager) execute(ctx context.Context, operation func() error) error {
+ if tm.circuitBreaker == nil {
+ return operation()
+ }
+ return tm.circuitBreaker.Execute(operation)
+}
+
+func (tm *TokenManager) cacheKey(model string, prompt string) string {
+ if model == "" {
+ return prompt
+ }
+ return model + ":" + prompt
Review Comment:
`cacheKey` concatenates the full `prompt` into the in-memory cache key. For large prompts this can significantly increase memory usage (duplicate prompt strings across cache/hotMap) and GC pressure. Consider using a bounded-length key (e.g., hash of model+prompt) while keeping collision risk acceptable.
##########
pkg/filter/ai/kvcache/handlers.go:
##########
@@ -0,0 +1,183 @@
+package kvcache
+
+import (
+ "bytes"
+ "context"
Review Comment:
This file is missing the standard Apache Software Foundation license header comment that appears at the top of other Go files in this repo. Please add the ASF header to keep licensing consistent.
##########
pkg/filter/ai/kvcache/load_monitor.go:
##########
@@ -0,0 +1,74 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 kvcache
+
+import (
+ "runtime"
+ "sync"
+ "time"
+)
+
+type LoadMonitor struct {
+ window time.Duration
+ last time.Time
+ count int64
+ rate float64
+ mutex sync.Mutex
+}
+
+func NewLoadMonitor() *LoadMonitor {
+ return &LoadMonitor{
+ window: time.Second,
+ last: time.Now(),
+ }
+}
+
+func (lm *LoadMonitor) RecordRequest() {
+ if lm == nil {
+ return
+ }
+ lm.mutex.Lock()
+ lm.count++
+ lm.mutex.Unlock()
+}
+
+func (lm *LoadMonitor) Snapshot() LoadMetrics {
+ if lm == nil {
+ return LoadMetrics{}
+ }
+ lm.mutex.Lock()
+ defer lm.mutex.Unlock()
+ now := time.Now()
+ elapsed := now.Sub(lm.last)
+ if elapsed >= lm.window && elapsed > 0 {
+ lm.rate = float64(lm.count) / elapsed.Seconds()
+ lm.count = 0
+ lm.last = now
+ }
+ var ms runtime.MemStats
+ runtime.ReadMemStats(&ms)
+ memUsage := 0.0
+ if ms.Sys > 0 {
+ memUsage = float64(ms.Alloc) / float64(ms.Sys)
+ }
+ return LoadMetrics{
+ CPUUsage: 0,
+ MemoryUsage: memUsage,
+ RequestRate: lm.rate,
+ }
Review Comment:
`Snapshot()` always returns `CPUUsage: 0`, but the strategy checks CPU usage against `LoadThreshold`. This makes CPU-based decisions impossible and can mislead operators configuring `load_threshold`. Either implement CPU utilization collection or remove CPU from the decision criteria/config.
##########
pkg/filter/ai/kvcache/handlers.go:
##########
@@ -0,0 +1,183 @@
+package kvcache
+
+import (
+ "bytes"
+ "context"
+ "encoding/json"
+ "io"
+ "net/http"
+ "strings"
+ "time"
+)
+
+import (
+ "github.com/apache/dubbo-go-pixiu/pkg/common/constant"
+ contexthttp "github.com/apache/dubbo-go-pixiu/pkg/context/http"
+ "github.com/apache/dubbo-go-pixiu/pkg/logger"
+)
+
+func (f *Filter) manageCache(ctx context.Context, model string, prompt string, rawBody []byte, cacheStatus *LookupResponse, lookupDone bool) {
+ if ctx.Err() != nil {
+ return
+ }
+ tokens, err := f.tokenManager.GetTokens(ctx, model, prompt, rawBody)
+ if err != nil {
+ logger.Warnf("[KVCache] tokenize failed: %v", err)
+ return
+ }
+ if ctx.Err() != nil {
+ return
+ }
+ if !lookupDone || cacheStatus == nil {
+ cacheStatus, err = f.lmcacheClient.Lookup(ctx, &LookupRequest{Tokens: tokens})
+ if err != nil {
+ logger.Warnf("[KVCache] lookup failed: %v", err)
+ return
+ }
+ }
+ decision := f.cacheStrategy.MakeDecision(ctx, cacheStatus, model, prompt)
+ if ctx.Err() != nil {
+ return
+ }
+ if err := f.cacheStrategy.ExecuteDecision(ctx, decision, tokens); err != nil {
+ logger.Warnf("[KVCache] execute strategy failed: %v", err)
+ }
+}
+
+func readRequestBody(req *http.Request) ([]byte, error) {
+ if req == nil || req.Body == nil {
+ return nil, nil
+ }
+ if req.GetBody != nil {
+ reader, err := req.GetBody()
+ if err == nil {
+ defer reader.Close()
+ return io.ReadAll(reader)
+ }
+ }
+ bodyBytes, err := io.ReadAll(req.Body)
+ if err != nil {
+ return nil, err
+ }
+ req.Body.Close()
+ req.Body = io.NopCloser(bytes.NewReader(bodyBytes))
+ req.GetBody = func() (io.ReadCloser, error) {
+ return io.NopCloser(bytes.NewReader(bodyBytes)), nil
+ }
+ return bodyBytes, nil
+}
+
+func extractPromptAndModel(body []byte) (string, string, error) {
+ if len(body) == 0 {
+ return "", "", nil
+ }
+ var payload map[string]any
+ if err := json.Unmarshal(body, &payload); err != nil {
+ return "", "", err
+ }
+ model, _ := payload["model"].(string)
+ prompt := coercePrompt(payload["prompt"])
+ if prompt == "" {
+ prompt = extractPromptFromMessages(payload["messages"])
+ }
+ return strings.TrimSpace(prompt), model, nil
+}
+
+func coercePrompt(value any) string {
+ switch v := value.(type) {
+ case string:
+ return v
+ case []any:
+ parts := make([]string, 0, len(v))
+ for _, item := range v {
+ if str, ok := item.(string); ok {
+ parts = append(parts, str)
+ }
+ }
+ return strings.Join(parts, "\n")
+ default:
+ return ""
+ }
+}
+
+func extractPromptFromMessages(value any) string {
+ msgs, ok := value.([]any)
+ if !ok {
+ return ""
+ }
+ parts := make([]string, 0, len(msgs))
+ for _, msg := range msgs {
+ msgMap, ok := msg.(map[string]any)
+ if !ok {
+ continue
+ }
+ if content, ok := msgMap["content"].(string); ok {
+ parts = append(parts, content)
+ }
+ }
+ return strings.Join(parts, "\n")
+}
+
+func selectPreferredInstanceID(resp *LookupResponse) string {
+ if resp == nil || len(resp.LayoutInfo) == 0 {
+ return ""
+ }
+ var (
+ selected string
+ maxCount int
+ )
+ for instanceID, layout := range resp.LayoutInfo {
+ if layout.TokenCount > maxCount || selected == "" {
+ selected = instanceID
+ maxCount = layout.TokenCount
+ }
+ }
+ return selected
+}
+
+func effectiveTimeout(hc *contexthttp.HttpContext, cfg *Config) time.Duration {
+ if cfg == nil {
+ return 0
+ }
+ timeout := cfg.RequestTimeout
+ if hc != nil && hc.Timeout > 0 && (timeout <= 0 || hc.Timeout < timeout) {
+ timeout = hc.Timeout
+ }
+ if timeout <= 0 {
+ return 2 * time.Second
+ }
+ return timeout
+}
+
+func (f *Filter) tryRouteToCachedInstance(hc *contexthttp.HttpContext, model string, prompt string) (*LookupResponse, bool) {
+ if f == nil || f.tokenManager == nil || f.lmcacheClient == nil {
+ return nil, false
+ }
+ tokens, ok := f.tokenManager.GetCachedTokens(model, prompt)
+ if !ok || len(tokens) == 0 {
+ logger.Debugf("[KVCache] routing lookup skipped: token cache miss")
+ return nil, false
+ }
+ timeout := effectiveTimeout(hc, f.cfg)
+ if f.cfg != nil && f.cfg.LookupRoutingTimeout > 0 && f.cfg.LookupRoutingTimeout < timeout {
+ timeout = f.cfg.LookupRoutingTimeout
+ }
+ ctx, cancel := context.WithTimeout(hc.Ctx, timeout)
Review Comment:
`context.WithTimeout(hc.Ctx, ...)` uses `hc.Ctx`, which is `context.Background()` in the HTTP manager, so lookup work won't be canceled when the client disconnects. Using `hc.Request.Context()` here (or setting `hc.Ctx` from it) would allow early cancellation and reduce unnecessary LMCache calls.
##########
pkg/filter/ai/kvcache/token_manager.go:
##########
@@ -0,0 +1,305 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 kvcache
+
+import (
+ "context"
+ "encoding/json"
+ "fmt"
+ "strings"
+ "sync"
+ "sync/atomic"
+ "time"
+)
+
+import (
+ "github.com/go-resty/resty/v2"
+)
+
+type TokenManager struct {
+ httpClient *resty.Client
+ endpoint string
+ cache sync.Map
+ config TokenCacheConfig
+ circuitBreaker *CircuitBreaker
+
+ cacheSize int64
+ hitCount int64
+ missCount int64
+
+ hotWindow time.Duration
+ hotMax int
+ hotMu sync.Mutex
+ hotMap map[string][]time.Time
+}
+
+type TokenizeRequest struct {
+ Model string `json:"model,omitempty"`
+ Prompt string `json:"prompt"`
+}
+
+type TokenizeResponse struct {
+ Count int `json:"count"`
+ Tokens []int `json:"tokens"`
+ MaxLen int `json:"max_model_len"`
+}
+
+type tokenCacheEntry struct {
+ tokens []int
+ expiresAt time.Time
+}
+
+func NewTokenManager(endpoint string, httpClient *resty.Client, cfg TokenCacheConfig, cb *CircuitBreaker, hotWindow time.Duration, hotMax int) *TokenManager {
+ return &TokenManager{
+ httpClient: httpClient,
+ endpoint: endpoint,
+ config: cfg,
+ circuitBreaker: cb,
+ hotWindow: hotWindow,
+ hotMax: hotMax,
+ hotMap: make(map[string][]time.Time),
+ }
+}
+
+func (tm *TokenManager) GetTokens(ctx context.Context, model string, prompt string, rawBody []byte) ([]int, error) {
+ cacheKey := tm.cacheKey(model, prompt)
+ if tm.config.Enabled {
+ if tokens, ok := tm.loadCache(cacheKey); ok {
+ atomic.AddInt64(&tm.hitCount, 1)
+ return tokens, nil
+ }
+ atomic.AddInt64(&tm.missCount, 1)
+ }
+
+ tokens, err := tm.tokenize(ctx, model, prompt, rawBody)
+ if err != nil {
+ return nil, err
+ }
+
+ if tm.config.Enabled {
+ tm.storeCache(cacheKey, tokens)
+ }
+ return tokens, nil
+}
+
+func (tm *TokenManager) GetCachedTokens(model string, prompt string) ([]int, bool) {
+ if !tm.config.Enabled {
+ return nil, false
+ }
+ cacheKey := tm.cacheKey(model, prompt)
+ tokens, ok := tm.loadCache(cacheKey)
+ if ok {
+ atomic.AddInt64(&tm.hitCount, 1)
+ } else {
+ atomic.AddInt64(&tm.missCount, 1)
+ }
+ return tokens, ok
+}
+
+func (tm *TokenManager) InvalidateCache(model string, prompt string) {
+ cacheKey := tm.cacheKey(model, prompt)
+ if _, ok := tm.cache.Load(cacheKey); ok {
+ tm.cache.Delete(cacheKey)
+ atomic.AddInt64(&tm.cacheSize, -1)
+ }
+}
+
+func (tm *TokenManager) GetCacheStats() CacheStats {
+ size := atomic.LoadInt64(&tm.cacheSize)
+ hit := atomic.LoadInt64(&tm.hitCount)
+ miss := atomic.LoadInt64(&tm.missCount)
+ total := hit + miss
+ var hitRate float64
+ if total > 0 {
+ hitRate = float64(hit) / float64(total)
+ }
+ return CacheStats{
+ Size: int(size),
+ HitRate: hitRate,
+ HitCount: hit,
+ MissCount: miss,
+ }
+}
+
+func (tm *TokenManager) tokenize(ctx context.Context, model string, prompt string, rawBody []byte) ([]int, error) {
+ var tokens []int
+ err := tm.execute(ctx, func() error {
+ body, err := tm.buildTokenizeBody(model, prompt, rawBody)
+ if err != nil {
+ return err
+ }
+ resp, err := tm.doTokenizeRequest(ctx, body)
+ if err != nil {
+ return err
+ }
+ tokens = resp.Tokens
+ return nil
+ })
+ if err != nil {
+ return nil, err
+ }
+ return tokens, nil
+}
+
+func (tm *TokenManager) buildTokenizeBody(model string, prompt string, rawBody []byte) (any, error) {
+ if len(rawBody) > 0 {
+ return rawBody, nil
+ }
+ return TokenizeRequest{Model: model, Prompt: prompt}, nil
+}
+
+func (tm *TokenManager) doTokenizeRequest(ctx context.Context, body any) (*TokenizeResponse, error) {
+ tokenizeURL := strings.TrimRight(tm.endpoint, "/") + "/tokenize"
+ resp, err := tm.httpClient.R().
+ SetContext(ctx).
+ SetHeader("Content-Type", "application/json").
+ SetBody(body).
+ Post(tokenizeURL)
+ if err != nil {
+ return nil, fmt.Errorf("call tokenize: %w", err)
+ }
+ if resp.StatusCode() < 200 || resp.StatusCode() >= 300 {
+ return nil, fmt.Errorf("tokenize status %d: %s", resp.StatusCode(), strings.TrimSpace(string(resp.Body())))
+ }
+ var tokenResp TokenizeResponse
+ if err := json.Unmarshal(resp.Body(), &tokenResp); err != nil {
+ return nil, fmt.Errorf("decode tokenize response: %w", err)
+ }
+ return &tokenResp, nil
+}
+
+func (tm *TokenManager) RecordHot(model string, prompt string) {
+ if tm == nil || tm.hotWindow <= 0 || model == "" || prompt == "" {
+ return
+ }
+ now := time.Now()
+ key := tm.cacheKey(model, prompt)
+ tm.hotMu.Lock()
+ defer tm.hotMu.Unlock()
+ entries := tm.hotMap[key]
+ entries = append(entries, now)
+ entries = trimHotWindow(entries, now, tm.hotWindow)
+ if tm.hotMax > 0 && len(entries) > tm.hotMax {
+ entries = entries[len(entries)-tm.hotMax:]
+ }
+ if len(entries) == 0 {
+ delete(tm.hotMap, key)
+ return
+ }
+ tm.hotMap[key] = entries
+}
+
+func (tm *TokenManager) IsHot(model string, prompt string, threshold int) bool {
+ if tm == nil || tm.hotWindow <= 0 || threshold <= 0 || model == "" || prompt == "" {
+ return false
+ }
+ now := time.Now()
+ key := tm.cacheKey(model, prompt)
+ tm.hotMu.Lock()
+ defer tm.hotMu.Unlock()
+ entries := tm.hotMap[key]
+ if len(entries) == 0 {
+ return false
+ }
+ entries = trimHotWindow(entries, now, tm.hotWindow)
+ if tm.hotMax > 0 && len(entries) > tm.hotMax {
+ entries = entries[len(entries)-tm.hotMax:]
+ }
+ if len(entries) == 0 {
+ delete(tm.hotMap, key)
+ return false
+ }
+ tm.hotMap[key] = entries
+ return len(entries) >= threshold
+}
+
+func trimHotWindow(entries []time.Time, now time.Time, window time.Duration) []time.Time {
+ if window <= 0 || len(entries) == 0 {
+ return entries
+ }
+ cutoff := now.Add(-window)
+ idx := 0
+ for idx < len(entries) && entries[idx].Before(cutoff) {
+ idx++
+ }
+ if idx == 0 {
+ return entries
+ }
+ return append([]time.Time(nil), entries[idx:]...)
+}
+
+func (tm *TokenManager) execute(ctx context.Context, operation func() error) error {
+ if tm.circuitBreaker == nil {
+ return operation()
+ }
+ return tm.circuitBreaker.Execute(operation)
+}
+
+func (tm *TokenManager) cacheKey(model string, prompt string) string {
+ if model == "" {
+ return prompt
+ }
+ return model + ":" + prompt
+}
+
+func (tm *TokenManager) loadCache(key string) ([]int, bool) {
+ entryAny, ok := tm.cache.Load(key)
+ if !ok {
+ return nil, false
+ }
+ entry, ok := entryAny.(*tokenCacheEntry)
+ if !ok {
+ tm.cache.Delete(key)
+ atomic.AddInt64(&tm.cacheSize, -1)
+ return nil, false
+ }
+ if tm.config.TTL > 0 && time.Now().After(entry.expiresAt) {
+ tm.cache.Delete(key)
+ atomic.AddInt64(&tm.cacheSize, -1)
+ return nil, false
Review Comment:
`cacheSize` is decremented on delete paths that can run concurrently (e.g., multiple goroutines can observe an entry as expired and both call `Delete` + `AddInt64(-1)`), which can make `cacheSize` drift/turn negative and break the max-size eviction loop. Consider tracking size with a guarded map (mutex + map) or only decrementing when you can prove a delete actually removed a present entry (e.g., `LoadAndDelete` in Go 1.20+).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
Re: [PR] feat: Add KVCache offload funtion in PIXIU gateway [dubbo-go-pixiu]
Posted by "Chen-BUPT (via GitHub)" <gi...@apache.org>.
Chen-BUPT commented on PR #878:
URL: https://github.com/apache/dubbo-go-pixiu/pull/878#issuecomment-3894702455
> still need to write docs in dubbo-website and write samples in dubbo-go-pixiu-samples
Noted.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
Re: [PR] feat: Add KVCache offload funtion in PIXIU gateway [dubbo-go-pixiu]
Posted by "Chen-BUPT (via GitHub)" <gi...@apache.org>.
Chen-BUPT commented on code in PR #878:
URL: https://github.com/apache/dubbo-go-pixiu/pull/878#discussion_r2777078238
##########
pkg/filter/ai/kvcache/load_monitor.go:
##########
@@ -0,0 +1,74 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 kvcache
+
+import (
+ "runtime"
+ "sync"
+ "time"
+)
+
+type LoadMonitor struct {
+ window time.Duration
+ last time.Time
+ count int64
+ rate float64
+ mutex sync.Mutex
+}
+
+func NewLoadMonitor() *LoadMonitor {
+ return &LoadMonitor{
+ window: time.Second,
+ last: time.Now(),
+ }
+}
+
+func (lm *LoadMonitor) RecordRequest() {
+ if lm == nil {
+ return
+ }
+ lm.mutex.Lock()
+ lm.count++
+ lm.mutex.Unlock()
+}
+
+func (lm *LoadMonitor) Snapshot() LoadMetrics {
+ if lm == nil {
+ return LoadMetrics{}
+ }
+ lm.mutex.Lock()
+ defer lm.mutex.Unlock()
+ now := time.Now()
+ elapsed := now.Sub(lm.last)
+ if elapsed >= lm.window && elapsed > 0 {
+ lm.rate = float64(lm.count) / elapsed.Seconds()
+ lm.count = 0
+ lm.last = now
+ }
+ var ms runtime.MemStats
+ runtime.ReadMemStats(&ms)
+ memUsage := 0.0
+ if ms.Sys > 0 {
+ memUsage = float64(ms.Alloc) / float64(ms.Sys)
+ }
+ return LoadMetrics{
+ CPUUsage: 0,
+ MemoryUsage: memUsage,
+ RequestRate: lm.rate,
+ }
Review Comment:
Fixed. Reworked cache deletion to use LoadAndDelete and only decrement cacheSize when deletion actually removed an existing key.
Changes:
- Added deleteCache(key) helper using sync.Map.LoadAndDelete
- Replaced delete paths in TTL expiry, type-mismatch cleanup, InvalidateCache, and evictOne with deleteCache
This prevents concurrent double-decrement on the same key and keeps cacheSize consistent for max-size eviction.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
Re: [PR] feat: Add KVCache offload funtion in PIXIU gateway [dubbo-go-pixiu]
Posted by "Chen-BUPT (via GitHub)" <gi...@apache.org>.
Chen-BUPT commented on code in PR #878:
URL: https://github.com/apache/dubbo-go-pixiu/pull/878#discussion_r2779290459
##########
pkg/common/constant/key.go:
##########
@@ -59,8 +59,11 @@ const (
LLMProxyFilter = "dgp.filter.llm.proxy"
LLMTokenizerFilter = "dgp.filter.llm.tokenizer"
+ AIKVCacheFilter = "dgp.filter.ai.kvcache"
MCPServerFilter = "dgp.filter.mcp.mcpserver"
+
+ LLMPreferredEndpointID = "llm_preferred_endpoint_id"
Review Comment:
已移到handlers.go和filter.go
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
Re: [PR] feat: Add KVCache offload funtion in PIXIU gateway [dubbo-go-pixiu]
Posted by "Similarityoung (via GitHub)" <gi...@apache.org>.
Similarityoung commented on code in PR #878:
URL: https://github.com/apache/dubbo-go-pixiu/pull/878#discussion_r2840557930
##########
pkg/filter/ai/kvcache/token_manager.go:
##########
@@ -0,0 +1,305 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 kvcache
+
+import (
+ "context"
+ "crypto/sha256"
+ "encoding/hex"
+ "encoding/json"
+ "fmt"
+ "strings"
+ "sync"
+ "sync/atomic"
+ "time"
+)
+
+import (
+ "github.com/go-resty/resty/v2"
+)
+
+type TokenManager struct {
+ httpClient *resty.Client
+ endpoint string
+ cache sync.Map
+ config TokenCacheConfig
+ circuitBreaker *CircuitBreaker
+
+ cacheSize int64
+ hitCount int64
+ missCount int64
+
+ hotWindow time.Duration
+ hotMax int
+ hotMu sync.Mutex
+ hotMap map[string][]time.Time
+}
+
+type TokenizeRequest struct {
+ Model string `json:"model,omitempty"`
+ Prompt string `json:"prompt"`
+}
+
+type TokenizeResponse struct {
+ Count int `json:"count"`
+ Tokens []int `json:"tokens"`
+ MaxLen int `json:"max_model_len"`
+}
+
+type tokenCacheEntry struct {
+ tokens []int
+ expiresAt time.Time
+}
+
+func NewTokenManager(endpoint string, httpClient *resty.Client, cfg TokenCacheConfig, cb *CircuitBreaker, hotWindow time.Duration, hotMax int) *TokenManager {
+ return &TokenManager{
+ httpClient: httpClient,
+ endpoint: endpoint,
+ config: cfg,
+ circuitBreaker: cb,
+ hotWindow: hotWindow,
+ hotMax: hotMax,
+ hotMap: make(map[string][]time.Time),
+ }
+}
+
+func (tm *TokenManager) GetTokens(ctx context.Context, model string, prompt string, rawBody []byte) ([]int, error) {
+ cacheKey := tm.cacheKey(model, prompt)
+ if tm.config.Enabled {
+ if tokens, ok := tm.loadCache(cacheKey); ok {
+ atomic.AddInt64(&tm.hitCount, 1)
+ return tokens, nil
+ }
+ atomic.AddInt64(&tm.missCount, 1)
+ }
+
+ tokens, err := tm.tokenize(ctx, model, prompt, rawBody)
+ if err != nil {
+ return nil, err
+ }
+
+ if tm.config.Enabled {
+ tm.storeCache(cacheKey, tokens)
+ }
+ return tokens, nil
+}
+
+func (tm *TokenManager) GetCachedTokens(model string, prompt string) ([]int, bool) {
+ if !tm.config.Enabled {
+ return nil, false
+ }
+ cacheKey := tm.cacheKey(model, prompt)
+ tokens, ok := tm.loadCache(cacheKey)
+ if ok {
+ atomic.AddInt64(&tm.hitCount, 1)
+ } else {
+ atomic.AddInt64(&tm.missCount, 1)
+ }
+ return tokens, ok
+}
+
+func (tm *TokenManager) InvalidateCache(model string, prompt string) {
+ cacheKey := tm.cacheKey(model, prompt)
+ tm.deleteCache(cacheKey)
+}
+
+func (tm *TokenManager) GetCacheStats() CacheStats {
+ size := atomic.LoadInt64(&tm.cacheSize)
+ hit := atomic.LoadInt64(&tm.hitCount)
+ miss := atomic.LoadInt64(&tm.missCount)
+ total := hit + miss
+ var hitRate float64
+ if total > 0 {
+ hitRate = float64(hit) / float64(total)
+ }
+ return CacheStats{
+ Size: int(size),
+ HitRate: hitRate,
+ HitCount: hit,
+ MissCount: miss,
+ }
+}
+
+func (tm *TokenManager) tokenize(ctx context.Context, model string, prompt string, rawBody []byte) ([]int, error) {
+ var tokens []int
+ err := tm.execute(ctx, func() error {
+ body, err := tm.buildTokenizeBody(model, prompt, rawBody)
+ if err != nil {
+ return err
+ }
+ resp, err := tm.doTokenizeRequest(ctx, body)
+ if err != nil {
+ return err
+ }
+ tokens = resp.Tokens
+ return nil
+ })
+ if err != nil {
+ return nil, err
+ }
+ return tokens, nil
+}
+
+func (tm *TokenManager) buildTokenizeBody(model string, prompt string, rawBody []byte) (any, error) {
+ if len(rawBody) > 0 {
+ return rawBody, nil
+ }
+ return TokenizeRequest{Model: model, Prompt: prompt}, nil
+}
+
+func (tm *TokenManager) doTokenizeRequest(ctx context.Context, body any) (*TokenizeResponse, error) {
+ tokenizeURL := strings.TrimRight(tm.endpoint, "/") + "/tokenize"
+ resp, err := tm.httpClient.R().
+ SetContext(ctx).
+ SetHeader("Content-Type", "application/json").
+ SetBody(body).
+ Post(tokenizeURL)
+ if err != nil {
+ return nil, fmt.Errorf("[kvcache] call tokenize: %w", err)
+ }
+ if resp.StatusCode() < 200 || resp.StatusCode() >= 300 {
+ return nil, fmt.Errorf("[kvcache] tokenize status %d: %s", resp.StatusCode(), strings.TrimSpace(string(resp.Body())))
+ }
+ var tokenResp TokenizeResponse
+ if err := json.Unmarshal(resp.Body(), &tokenResp); err != nil {
+ return nil, fmt.Errorf("[kvcache] decode tokenize response: %w", err)
+ }
+ return &tokenResp, nil
+}
+
+func (tm *TokenManager) RecordHot(model string, prompt string) {
+ if tm == nil || tm.hotWindow <= 0 || model == "" || prompt == "" {
+ return
+ }
+ now := time.Now()
+ key := tm.cacheKey(model, prompt)
+ tm.hotMu.Lock()
+ defer tm.hotMu.Unlock()
+ entries := tm.hotMap[key]
+ entries = append(entries, now)
+ entries = trimHotWindow(entries, now, tm.hotWindow)
+ if tm.hotMax > 0 && len(entries) > tm.hotMax {
+ entries = entries[len(entries)-tm.hotMax:]
+ }
+ if len(entries) == 0 {
+ delete(tm.hotMap, key)
+ return
+ }
+ tm.hotMap[key] = entries
+}
Review Comment:
hotMap 只在当前 key 被再次访问时才触发过期清理，未再次命中的 key 会永久留在 map 中。在高基数 prompt 场景下，hotMap 会随时间持续增长并带来长期内存膨胀风险，建议增加全局清理机制（周期扫描/容量上限+淘汰），而不是仅按访问路径被动清理。
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
Re: [PR] feat: Add KVCache offload funtion in PIXIU gateway [dubbo-go-pixiu]
Posted by "Chen-BUPT (via GitHub)" <gi...@apache.org>.
Chen-BUPT commented on code in PR #878:
URL: https://github.com/apache/dubbo-go-pixiu/pull/878#discussion_r2777089021
##########
pkg/filter/ai/kvcache/filter.go:
##########
@@ -0,0 +1,142 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 kvcache
+
+import (
+ "context"
+ "net/http"
+)
+
+import (
+ "github.com/go-resty/resty/v2"
+)
+
+import (
+ "github.com/apache/dubbo-go-pixiu/pkg/common/constant"
+ "github.com/apache/dubbo-go-pixiu/pkg/common/extension/filter"
+ contexthttp "github.com/apache/dubbo-go-pixiu/pkg/context/http"
+ "github.com/apache/dubbo-go-pixiu/pkg/logger"
+)
+
+const (
+ Kind = constant.AIKVCacheFilter
+)
+
+func init() {
+ filter.RegisterHttpFilter(&Plugin{})
+}
+
+type (
+ Plugin struct{}
+
+ FilterFactory struct {
+ cfg *Config
+ httpClient *http.Client
+ resty *resty.Client
+ tokenManager *TokenManager
+ lmcacheClient *LMCacheClient
+ cacheStrategy *CacheStrategy
+ }
+
+ Filter struct {
+ cfg *Config
+ tokenManager *TokenManager
+ lmcacheClient *LMCacheClient
+ cacheStrategy *CacheStrategy
+ }
+)
+
+func (p *Plugin) Kind() string { return Kind }
+
+func (p *Plugin) CreateFilterFactory() (filter.HttpFilterFactory, error) {
+ return &FilterFactory{cfg: &Config{}}, nil
+}
+
+func (factory *FilterFactory) Config() any { return factory.cfg }
+
+func (factory *FilterFactory) Apply() error {
+ factory.cfg.ApplyDefaults()
+ if err := factory.cfg.Validate(); err != nil {
+ return err
+ }
+ cfg := factory.cfg
+ factory.httpClient = &http.Client{
+ Timeout: cfg.RequestTimeout,
+ Transport: &http.Transport{
+ MaxIdleConns: cfg.MaxIdleConns,
+ MaxIdleConnsPerHost: cfg.MaxIdleConnsPerHost,
+ MaxConnsPerHost: cfg.MaxConnsPerHost,
+ },
+ }
+ factory.resty = resty.NewWithClient(factory.httpClient).
+ SetTimeout(cfg.RequestTimeout)
+
+ cbToken := NewCircuitBreaker(cfg.CircuitBreaker)
+ cbLMCache := NewCircuitBreaker(cfg.CircuitBreaker)
+ factory.tokenManager = NewTokenManager(cfg.VLLMEndpoint, factory.resty, cfg.TokenCache, cbToken, cfg.HotWindow, cfg.HotMaxRecords)
+ factory.lmcacheClient = NewLMCacheClient(cfg.LMCacheEndpoint, factory.resty, cfg.Retry, cbLMCache)
+ factory.cacheStrategy = NewCacheStrategy(cfg.CacheStrategy, factory.lmcacheClient, factory.tokenManager)
+ return nil
+}
+
+func (factory *FilterFactory) PrepareFilterChain(_ *contexthttp.HttpContext, chain filter.FilterChain) error {
+ f := &Filter{
+ cfg: factory.cfg,
+ tokenManager: factory.tokenManager,
+ lmcacheClient: factory.lmcacheClient,
+ cacheStrategy: factory.cacheStrategy,
+ }
+ chain.AppendDecodeFilters(f)
+ return nil
+}
+
+func (f *Filter) Decode(hc *contexthttp.HttpContext) filter.FilterStatus {
+ if f.cfg == nil || !f.cfg.Enabled {
+ return filter.Continue
+ }
+ if f.cacheStrategy != nil {
+ f.cacheStrategy.RecordRequest()
+ }
+ body, err := readRequestBody(hc.Request)
+ if err != nil {
+ logger.Warnf("[KVCache] read request body failed: %v", err)
+ return filter.Continue
+ }
+ prompt, model, err := extractPromptAndModel(body)
+ if err != nil {
+ logger.Warnf("[KVCache] parse request body failed: %v", err)
+ return filter.Continue
+ }
+ if prompt == "" {
+ return filter.Continue
+ }
+ if model == "" {
+ model = f.cfg.DefaultModel
+ }
+
+ f.tokenManager.RecordHot(model, prompt)
+
+ cacheStatus, routed := f.tryRouteToCachedInstance(hc, model, prompt)
+
+ ctx, cancel := context.WithTimeout(hc.Ctx, effectiveTimeout(hc, f.cfg))
+ go func() {
+ defer cancel()
+ f.manageCache(ctx, model, prompt, body, cacheStatus, routed)
+ }()
Review Comment:
fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
Re: [PR] feat: Add KVCache offload funtion in PIXIU gateway [dubbo-go-pixiu]
Posted by "Similarityoung (via GitHub)" <gi...@apache.org>.
Similarityoung commented on PR #878:
URL: https://github.com/apache/dubbo-go-pixiu/pull/878#issuecomment-3866647729
补点单测？
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
Re: [PR] feat: Add KVCache offload funtion in PIXIU gateway [dubbo-go-pixiu]
Posted by "Chen-BUPT (via GitHub)" <gi...@apache.org>.
Chen-BUPT commented on code in PR #878:
URL: https://github.com/apache/dubbo-go-pixiu/pull/878#discussion_r2779292763
##########
pkg/filter/ai/kvcache/config.go:
##########
@@ -0,0 +1,150 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 kvcache
+
+import (
+ "fmt"
+ "strings"
+ "time"
+)
+
+type Config struct {
+ Enabled bool `yaml:"enabled" json:"enabled" mapstructure:"enabled"`
+ VLLMEndpoint string `yaml:"vllm_endpoint" json:"vllm_endpoint" mapstructure:"vllm_endpoint"`
+ LMCacheEndpoint string `yaml:"lmcache_endpoint" json:"lmcache_endpoint" mapstructure:"lmcache_endpoint"`
+ DefaultModel string `yaml:"default_model" json:"default_model" mapstructure:"default_model"`
+ RequestTimeout time.Duration `yaml:"request_timeout" json:"request_timeout" mapstructure:"request_timeout"`
+ LookupRoutingTimeout time.Duration `yaml:"lookup_routing_timeout" json:"lookup_routing_timeout" mapstructure:"lookup_routing_timeout"`
+ HotWindow time.Duration `yaml:"hot_window" json:"hot_window" mapstructure:"hot_window"`
+ HotMaxRecords int `yaml:"hot_max_records" json:"hot_max_records" mapstructure:"hot_max_records"`
+ MaxIdleConns int `yaml:"max_idle_conns" json:"max_idle_conns" mapstructure:"max_idle_conns"`
+ MaxIdleConnsPerHost int `yaml:"max_idle_conns_per_host" json:"max_idle_conns_per_host" mapstructure:"max_idle_conns_per_host"`
+ MaxConnsPerHost int `yaml:"max_conns_per_host" json:"max_conns_per_host" mapstructure:"max_conns_per_host"`
+ TokenCache TokenCacheConfig `yaml:"token_cache" json:"token_cache" mapstructure:"token_cache"`
+ CacheStrategy CacheStrategyConfig `yaml:"cache_strategy" json:"cache_strategy" mapstructure:"cache_strategy"`
+ CircuitBreaker CircuitBreakerConfig `yaml:"circuit_breaker" json:"circuit_breaker" mapstructure:"circuit_breaker"`
+ Retry RetryConfig `yaml:"retry" json:"retry" mapstructure:"retry"`
+}
+
+type TokenCacheConfig struct {
+ MaxSize int `yaml:"max_size" json:"max_size" mapstructure:"max_size"`
+ TTL time.Duration `yaml:"ttl" json:"ttl" mapstructure:"ttl"`
+ Enabled bool `yaml:"enabled" json:"enabled" mapstructure:"enabled"`
+}
+
+type CacheStrategyConfig struct {
+ EnableCompression bool `yaml:"enable_compression" json:"enable_compression" mapstructure:"enable_compression"`
+ EnablePinning bool `yaml:"enable_pinning" json:"enable_pinning" mapstructure:"enable_pinning"`
+ EnableEviction bool `yaml:"enable_eviction" json:"enable_eviction" mapstructure:"enable_eviction"`
+ MemoryThreshold float64 `yaml:"memory_threshold" json:"memory_threshold" mapstructure:"memory_threshold"`
+ HotContentThreshold int `yaml:"hot_content_threshold" json:"hot_content_threshold" mapstructure:"hot_content_threshold"`
+ LoadThreshold float64 `yaml:"load_threshold" json:"load_threshold" mapstructure:"load_threshold"`
+ PinInstanceID string `yaml:"pin_instance_id" json:"pin_instance_id" mapstructure:"pin_instance_id"`
+ PinLocation string `yaml:"pin_location" json:"pin_location" mapstructure:"pin_location"`
+ CompressInstanceID string `yaml:"compress_instance_id" json:"compress_instance_id" mapstructure:"compress_instance_id"`
+ CompressLocation string `yaml:"compress_location" json:"compress_location" mapstructure:"compress_location"`
+ CompressMethod string `yaml:"compress_method" json:"compress_method" mapstructure:"compress_method"`
+ EvictInstanceID string `yaml:"evict_instance_id" json:"evict_instance_id" mapstructure:"evict_instance_id"`
+}
+
+type RetryConfig struct {
+ MaxAttempts int `yaml:"max_attempts" json:"max_attempts" mapstructure:"max_attempts"`
+ BaseBackoff time.Duration `yaml:"base_backoff" json:"base_backoff" mapstructure:"base_backoff"`
+ MaxBackoff time.Duration `yaml:"max_backoff" json:"max_backoff" mapstructure:"max_backoff"`
+}
+
+func (c *Config) Validate() error {
+ if !c.Enabled {
+ return nil
+ }
+ if strings.TrimSpace(c.VLLMEndpoint) == "" {
+ return fmt.Errorf("kvcache: vllm_endpoint is required when enabled")
+ }
+ if strings.TrimSpace(c.LMCacheEndpoint) == "" {
+ return fmt.Errorf("kvcache: lmcache_endpoint is required when enabled")
+ }
+ if c.TokenCache.MaxSize < 0 {
+ return fmt.Errorf("kvcache: token_cache.max_size must be >= 0")
+ }
+ if c.CacheStrategy.MemoryThreshold < 0 || c.CacheStrategy.MemoryThreshold > 1 {
+ return fmt.Errorf("kvcache: cache_strategy.memory_threshold must be between 0 and 1")
+ }
+ if c.CacheStrategy.LoadThreshold < 0 || c.CacheStrategy.LoadThreshold > 1 {
+ return fmt.Errorf("kvcache: cache_strategy.load_threshold must be between 0 and 1")
+ }
+ if c.CacheStrategy.HotContentThreshold < 0 {
+ return fmt.Errorf("kvcache: cache_strategy.hot_content_threshold must be >= 0")
+ }
+ if c.Retry.MaxAttempts < 0 {
+ return fmt.Errorf("kvcache: retry.max_attempts must be >= 0")
+ }
+ if c.Retry.BaseBackoff < 0 || c.Retry.MaxBackoff < 0 {
+ return fmt.Errorf("kvcache: retry backoff durations must be >= 0")
+ }
+ if c.HotWindow < 0 {
+ return fmt.Errorf("kvcache: hot_window must be >= 0")
+ }
+ if c.HotMaxRecords < 0 {
+ return fmt.Errorf("kvcache: hot_max_records must be >= 0")
+ }
+ return nil
+}
+
+func (c *Config) ApplyDefaults() {
+ if c.RequestTimeout <= 0 {
+ c.RequestTimeout = 2 * time.Second
Review Comment:
done
##########
pkg/filter/llm/proxy/filter.go:
##########
@@ -98,6 +98,18 @@ type (
}
)
+func getPreferredEndpointID(hc *contexthttp.HttpContext) string {
+ if hc == nil || hc.Params == nil {
+ return ""
+ }
+ val, ok := hc.Params[constant.LLMPreferredEndpointID]
+ if !ok {
+ return ""
+ }
+ endpointID, _ := val.(string)
+ return endpointID
Review Comment:
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
Re: [PR] feat: Add KVCache offload funtion in PIXIU gateway [dubbo-go-pixiu]
Posted by "Alanxtl (via GitHub)" <gi...@apache.org>.
Alanxtl commented on PR #878:
URL: https://github.com/apache/dubbo-go-pixiu/pull/878#issuecomment-3894465326
@nanjiek @Similarityoung check this
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
Re: [PR] feat: Add KVCache offload funtion in PIXIU gateway [dubbo-go-pixiu]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #878:
URL: https://github.com/apache/dubbo-go-pixiu/pull/878#issuecomment-3867053029
## [Codecov](https://app.codecov.io/gh/apache/dubbo-go-pixiu/pull/878?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:x: Patch coverage is `0%` with `652 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 19.94%. Comparing base ([`7d70213`](https://app.codecov.io/gh/apache/dubbo-go-pixiu/commit/7d702130f97bed1480e3cb086d933877cc7d009a?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`25c1f8d`](https://app.codecov.io/gh/apache/dubbo-go-pixiu/commit/25c1f8d58cf11fc387f20bb1e4a0b34d96dba8e5?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
| [Files with missing lines](https://app.codecov.io/gh/apache/dubbo-go-pixiu/pull/878?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [pkg/filter/ai/kvcache/token\_manager.go](https://app.codecov.io/gh/apache/dubbo-go-pixiu/pull/878?src=pr&el=tree&filepath=pkg%2Ffilter%2Fai%2Fkvcache%2Ftoken_manager.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGtnL2ZpbHRlci9haS9rdmNhY2hlL3Rva2VuX21hbmFnZXIuZ28=) | 0.00% | [165 Missing :warning: ](https://app.codecov.io/gh/apache/dubbo-go-pixiu/pull/878?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [pkg/filter/ai/kvcache/handlers.go](https://app.codecov.io/gh/apache/dubbo-go-pixiu/pull/878?src=pr&el=tree&filepath=pkg%2Ffilter%2Fai%2Fkvcache%2Fhandlers.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGtnL2ZpbHRlci9haS9rdmNhY2hlL2hhbmRsZXJzLmdv) | 0.00% | [122 Missing :warning: ](https://app.codecov.io/gh/apache/dubbo-go-pixiu/pull/878?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [pkg/filter/ai/kvcache/lmcache\_client.go](https://app.codecov.io/gh/apache/dubbo-go-pixiu/pull/878?src=pr&el=tree&filepath=pkg%2Ffilter%2Fai%2Fkvcache%2Flmcache_client.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGtnL2ZpbHRlci9haS9rdmNhY2hlL2xtY2FjaGVfY2xpZW50Lmdv) | 0.00% | [75 Missing :warning: ](https://app.codecov.io/gh/apache/dubbo-go-pixiu/pull/878?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [pkg/filter/ai/kvcache/load\_monitor.go](https://app.codecov.io/gh/apache/dubbo-go-pixiu/pull/878?src=pr&el=tree&filepath=pkg%2Ffilter%2Fai%2Fkvcache%2Fload_monitor.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGtnL2ZpbHRlci9haS9rdmNhY2hlL2xvYWRfbW9uaXRvci5nbw==) | 0.00% | [65 Missing :warning: ](https://app.codecov.io/gh/apache/dubbo-go-pixiu/pull/878?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [pkg/filter/ai/kvcache/filter.go](https://app.codecov.io/gh/apache/dubbo-go-pixiu/pull/878?src=pr&el=tree&filepath=pkg%2Ffilter%2Fai%2Fkvcache%2Ffilter.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGtnL2ZpbHRlci9haS9rdmNhY2hlL2ZpbHRlci5nbw==) | 0.00% | [59 Missing :warning: ](https://app.codecov.io/gh/apache/dubbo-go-pixiu/pull/878?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [pkg/filter/ai/kvcache/strategy.go](https://app.codecov.io/gh/apache/dubbo-go-pixiu/pull/878?src=pr&el=tree&filepath=pkg%2Ffilter%2Fai%2Fkvcache%2Fstrategy.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGtnL2ZpbHRlci9haS9rdmNhY2hlL3N0cmF0ZWd5Lmdv) | 0.00% | [55 Missing :warning: ](https://app.codecov.io/gh/apache/dubbo-go-pixiu/pull/878?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [pkg/filter/ai/kvcache/config.go](https://app.codecov.io/gh/apache/dubbo-go-pixiu/pull/878?src=pr&el=tree&filepath=pkg%2Ffilter%2Fai%2Fkvcache%2Fconfig.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGtnL2ZpbHRlci9haS9rdmNhY2hlL2NvbmZpZy5nbw==) | 0.00% | [52 Missing :warning: ](https://app.codecov.io/gh/apache/dubbo-go-pixiu/pull/878?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [pkg/filter/ai/kvcache/circuit\_breaker.go](https://app.codecov.io/gh/apache/dubbo-go-pixiu/pull/878?src=pr&el=tree&filepath=pkg%2Ffilter%2Fai%2Fkvcache%2Fcircuit_breaker.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGtnL2ZpbHRlci9haS9rdmNhY2hlL2NpcmN1aXRfYnJlYWtlci5nbw==) | 0.00% | [38 Missing :warning: ](https://app.codecov.io/gh/apache/dubbo-go-pixiu/pull/878?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [pkg/filter/llm/proxy/filter.go](https://app.codecov.io/gh/apache/dubbo-go-pixiu/pull/878?src=pr&el=tree&filepath=pkg%2Ffilter%2Fllm%2Fproxy%2Ffilter.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGtnL2ZpbHRlci9sbG0vcHJveHkvZmlsdGVyLmdv) | 0.00% | [11 Missing :warning: ](https://app.codecov.io/gh/apache/dubbo-go-pixiu/pull/878?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [pkg/server/cluster\_manager.go](https://app.codecov.io/gh/apache/dubbo-go-pixiu/pull/878?src=pr&el=tree&filepath=pkg%2Fserver%2Fcluster_manager.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGtnL3NlcnZlci9jbHVzdGVyX21hbmFnZXIuZ28=) | 0.00% | [10 Missing :warning: ](https://app.codecov.io/gh/apache/dubbo-go-pixiu/pull/878?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## develop #878 +/- ##
===========================================
- Coverage 20.63% 19.94% -0.70%
===========================================
Files 259 267 +8
Lines 19017 19669 +652
===========================================
- Hits 3925 3922 -3
- Misses 14707 15361 +654
- Partials 385 386 +1
```
| [Flag](https://app.codecov.io/gh/apache/dubbo-go-pixiu/pull/878/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [unittests](https://app.codecov.io/gh/apache/dubbo-go-pixiu/pull/878/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `19.94% <0.00%> (-0.70%)` | :arrow_down: |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
</details>
[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/apache/dubbo-go-pixiu/pull/878?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
Re: [PR] feat: Add KVCache offload funtion in PIXIU gateway [dubbo-go-pixiu]
Posted by "Chen-BUPT (via GitHub)" <gi...@apache.org>.
Chen-BUPT commented on code in PR #878:
URL: https://github.com/apache/dubbo-go-pixiu/pull/878#discussion_r2779261552
##########
pkg/filter/ai/kvcache/token_manager.go:
##########
@@ -0,0 +1,305 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 kvcache
+
+import (
+ "context"
+ "crypto/sha256"
+ "encoding/hex"
+ "encoding/json"
+ "fmt"
+ "strings"
+ "sync"
+ "sync/atomic"
+ "time"
+)
+
+import (
+ "github.com/go-resty/resty/v2"
+)
+
+type TokenManager struct {
+ httpClient *resty.Client
+ endpoint string
+ cache sync.Map
+ config TokenCacheConfig
+ circuitBreaker *CircuitBreaker
+
+ cacheSize int64
+ hitCount int64
+ missCount int64
+
+ hotWindow time.Duration
+ hotMax int
+ hotMu sync.Mutex
+ hotMap map[string][]time.Time
+}
+
+type TokenizeRequest struct {
+ Model string `json:"model,omitempty"`
+ Prompt string `json:"prompt"`
+}
+
+type TokenizeResponse struct {
+ Count int `json:"count"`
+ Tokens []int `json:"tokens"`
+ MaxLen int `json:"max_model_len"`
+}
+
+type tokenCacheEntry struct {
+ tokens []int
+ expiresAt time.Time
+}
+
+func NewTokenManager(endpoint string, httpClient *resty.Client, cfg TokenCacheConfig, cb *CircuitBreaker, hotWindow time.Duration, hotMax int) *TokenManager {
+ return &TokenManager{
+ httpClient: httpClient,
+ endpoint: endpoint,
+ config: cfg,
+ circuitBreaker: cb,
+ hotWindow: hotWindow,
+ hotMax: hotMax,
+ hotMap: make(map[string][]time.Time),
+ }
+}
+
+func (tm *TokenManager) GetTokens(ctx context.Context, model string, prompt string, rawBody []byte) ([]int, error) {
+ cacheKey := tm.cacheKey(model, prompt)
+ if tm.config.Enabled {
+ if tokens, ok := tm.loadCache(cacheKey); ok {
+ atomic.AddInt64(&tm.hitCount, 1)
+ return tokens, nil
+ }
+ atomic.AddInt64(&tm.missCount, 1)
+ }
+
+ tokens, err := tm.tokenize(ctx, model, prompt, rawBody)
+ if err != nil {
+ return nil, err
+ }
+
+ if tm.config.Enabled {
+ tm.storeCache(cacheKey, tokens)
+ }
+ return tokens, nil
+}
+
+func (tm *TokenManager) GetCachedTokens(model string, prompt string) ([]int, bool) {
+ if !tm.config.Enabled {
+ return nil, false
+ }
+ cacheKey := tm.cacheKey(model, prompt)
+ tokens, ok := tm.loadCache(cacheKey)
+ if ok {
+ atomic.AddInt64(&tm.hitCount, 1)
+ } else {
+ atomic.AddInt64(&tm.missCount, 1)
+ }
+ return tokens, ok
+}
+
+func (tm *TokenManager) InvalidateCache(model string, prompt string) {
+ cacheKey := tm.cacheKey(model, prompt)
+ tm.deleteCache(cacheKey)
+}
+
+func (tm *TokenManager) GetCacheStats() CacheStats {
+ size := atomic.LoadInt64(&tm.cacheSize)
+ hit := atomic.LoadInt64(&tm.hitCount)
+ miss := atomic.LoadInt64(&tm.missCount)
+ total := hit + miss
+ var hitRate float64
+ if total > 0 {
+ hitRate = float64(hit) / float64(total)
+ }
+ return CacheStats{
+ Size: int(size),
+ HitRate: hitRate,
+ HitCount: hit,
+ MissCount: miss,
+ }
+}
+
+func (tm *TokenManager) tokenize(ctx context.Context, model string, prompt string, rawBody []byte) ([]int, error) {
+ var tokens []int
+ err := tm.execute(ctx, func() error {
+ body, err := tm.buildTokenizeBody(model, prompt, rawBody)
+ if err != nil {
+ return err
+ }
+ resp, err := tm.doTokenizeRequest(ctx, body)
+ if err != nil {
+ return err
+ }
+ tokens = resp.Tokens
+ return nil
+ })
+ if err != nil {
+ return nil, err
+ }
+ return tokens, nil
+}
+
+func (tm *TokenManager) buildTokenizeBody(model string, prompt string, rawBody []byte) (any, error) {
+ if len(rawBody) > 0 {
+ return rawBody, nil
+ }
+ return TokenizeRequest{Model: model, Prompt: prompt}, nil
+}
+
+func (tm *TokenManager) doTokenizeRequest(ctx context.Context, body any) (*TokenizeResponse, error) {
Review Comment:
因为向下游进行智能路由等操作需要prompt对应的token 这里选择向推理引擎发/tokenizer请求来获得token
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
Re: [PR] feat: Add KVCache offload funtion in PIXIU gateway [dubbo-go-pixiu]
Posted by "Chen-BUPT (via GitHub)" <gi...@apache.org>.
Chen-BUPT commented on code in PR #878:
URL: https://github.com/apache/dubbo-go-pixiu/pull/878#discussion_r2779258887
##########
pkg/filter/ai/kvcache/config.go:
##########
@@ -0,0 +1,150 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 kvcache
+
+import (
+ "fmt"
+ "strings"
+ "time"
+)
+
+type Config struct {
+ Enabled bool `yaml:"enabled" json:"enabled" mapstructure:"enabled"`
+ VLLMEndpoint string `yaml:"vllm_endpoint" json:"vllm_endpoint" mapstructure:"vllm_endpoint"`
+ LMCacheEndpoint string `yaml:"lmcache_endpoint" json:"lmcache_endpoint" mapstructure:"lmcache_endpoint"`
+ DefaultModel string `yaml:"default_model" json:"default_model" mapstructure:"default_model"`
+ RequestTimeout time.Duration `yaml:"request_timeout" json:"request_timeout" mapstructure:"request_timeout"`
+ LookupRoutingTimeout time.Duration `yaml:"lookup_routing_timeout" json:"lookup_routing_timeout" mapstructure:"lookup_routing_timeout"`
+ HotWindow time.Duration `yaml:"hot_window" json:"hot_window" mapstructure:"hot_window"`
+ HotMaxRecords int `yaml:"hot_max_records" json:"hot_max_records" mapstructure:"hot_max_records"`
+ MaxIdleConns int `yaml:"max_idle_conns" json:"max_idle_conns" mapstructure:"max_idle_conns"`
+ MaxIdleConnsPerHost int `yaml:"max_idle_conns_per_host" json:"max_idle_conns_per_host" mapstructure:"max_idle_conns_per_host"`
+ MaxConnsPerHost int `yaml:"max_conns_per_host" json:"max_conns_per_host" mapstructure:"max_conns_per_host"`
+ TokenCache TokenCacheConfig `yaml:"token_cache" json:"token_cache" mapstructure:"token_cache"`
+ CacheStrategy CacheStrategyConfig `yaml:"cache_strategy" json:"cache_strategy" mapstructure:"cache_strategy"`
+ CircuitBreaker CircuitBreakerConfig `yaml:"circuit_breaker" json:"circuit_breaker" mapstructure:"circuit_breaker"`
+ Retry RetryConfig `yaml:"retry" json:"retry" mapstructure:"retry"`
+}
+
+type TokenCacheConfig struct {
+ MaxSize int `yaml:"max_size" json:"max_size" mapstructure:"max_size"`
+ TTL time.Duration `yaml:"ttl" json:"ttl" mapstructure:"ttl"`
+ Enabled bool `yaml:"enabled" json:"enabled" mapstructure:"enabled"`
+}
+
+type CacheStrategyConfig struct {
+ EnableCompression bool `yaml:"enable_compression" json:"enable_compression" mapstructure:"enable_compression"`
+ EnablePinning bool `yaml:"enable_pinning" json:"enable_pinning" mapstructure:"enable_pinning"`
+ EnableEviction bool `yaml:"enable_eviction" json:"enable_eviction" mapstructure:"enable_eviction"`
+ MemoryThreshold float64 `yaml:"memory_threshold" json:"memory_threshold" mapstructure:"memory_threshold"`
+ HotContentThreshold int `yaml:"hot_content_threshold" json:"hot_content_threshold" mapstructure:"hot_content_threshold"`
+ LoadThreshold float64 `yaml:"load_threshold" json:"load_threshold" mapstructure:"load_threshold"`
+ PinInstanceID string `yaml:"pin_instance_id" json:"pin_instance_id" mapstructure:"pin_instance_id"`
+ PinLocation string `yaml:"pin_location" json:"pin_location" mapstructure:"pin_location"`
+ CompressInstanceID string `yaml:"compress_instance_id" json:"compress_instance_id" mapstructure:"compress_instance_id"`
+ CompressLocation string `yaml:"compress_location" json:"compress_location" mapstructure:"compress_location"`
+ CompressMethod string `yaml:"compress_method" json:"compress_method" mapstructure:"compress_method"`
+ EvictInstanceID string `yaml:"evict_instance_id" json:"evict_instance_id" mapstructure:"evict_instance_id"`
+}
+
+type RetryConfig struct {
+ MaxAttempts int `yaml:"max_attempts" json:"max_attempts" mapstructure:"max_attempts"`
+ BaseBackoff time.Duration `yaml:"base_backoff" json:"base_backoff" mapstructure:"base_backoff"`
+ MaxBackoff time.Duration `yaml:"max_backoff" json:"max_backoff" mapstructure:"max_backoff"`
+}
Review Comment:
好的 我不知道
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
Re: [PR] feat: Add KVCache offload funtion in PIXIU gateway [dubbo-go-pixiu]
Posted by "sonarqubecloud[bot] (via GitHub)" <gi...@apache.org>.
sonarqubecloud[bot] commented on PR #878:
URL: https://github.com/apache/dubbo-go-pixiu/pull/878#issuecomment-3863670584
## [](https://sonarcloud.io/dashboard?id=apache_dubbo-go-pixiu&pullRequest=878) **Quality Gate passed**
Issues
 [7 New issues](https://sonarcloud.io/project/issues?id=apache_dubbo-go-pixiu&pullRequest=878&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)
 [0 Accepted issues](https://sonarcloud.io/project/issues?id=apache_dubbo-go-pixiu&pullRequest=878&issueStatuses=ACCEPTED)
Measures
 [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=apache_dubbo-go-pixiu&pullRequest=878&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)
 [0.0% Coverage on New Code](https://sonarcloud.io/component_measures?id=apache_dubbo-go-pixiu&pullRequest=878&metric=new_coverage&view=list)
 [0.0% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_dubbo-go-pixiu&pullRequest=878&metric=new_duplicated_lines_density&view=list)
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=apache_dubbo-go-pixiu&pullRequest=878)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
Re: [PR] feat: Add KVCache offload funtion in PIXIU gateway [dubbo-go-pixiu]
Posted by "sonarqubecloud[bot] (via GitHub)" <gi...@apache.org>.
sonarqubecloud[bot] commented on PR #878:
URL: https://github.com/apache/dubbo-go-pixiu/pull/878#issuecomment-3931745100
## [](https://sonarcloud.io/dashboard?id=apache_dubbo-go-pixiu&pullRequest=878) **Quality Gate passed**
Issues
 [10 New issues](https://sonarcloud.io/project/issues?id=apache_dubbo-go-pixiu&pullRequest=878&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)
 [0 Accepted issues](https://sonarcloud.io/project/issues?id=apache_dubbo-go-pixiu&pullRequest=878&issueStatuses=ACCEPTED)
Measures
 [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=apache_dubbo-go-pixiu&pullRequest=878&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)
 [0.0% Coverage on New Code](https://sonarcloud.io/component_measures?id=apache_dubbo-go-pixiu&pullRequest=878&metric=new_coverage&view=list)
 [0.0% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_dubbo-go-pixiu&pullRequest=878&metric=new_duplicated_lines_density&view=list)
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=apache_dubbo-go-pixiu&pullRequest=878)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
Re: [PR] feat: Add KVCache offload funtion in PIXIU gateway [dubbo-go-pixiu]
Posted by "Chen-BUPT (via GitHub)" <gi...@apache.org>.
Chen-BUPT commented on code in PR #878:
URL: https://github.com/apache/dubbo-go-pixiu/pull/878#discussion_r2777084129
##########
pkg/filter/ai/kvcache/token_manager.go:
##########
@@ -0,0 +1,305 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 kvcache
+
+import (
+ "context"
+ "encoding/json"
+ "fmt"
+ "strings"
+ "sync"
+ "sync/atomic"
+ "time"
+)
+
+import (
+ "github.com/go-resty/resty/v2"
+)
+
+type TokenManager struct {
+ httpClient *resty.Client
+ endpoint string
+ cache sync.Map
+ config TokenCacheConfig
+ circuitBreaker *CircuitBreaker
+
+ cacheSize int64
+ hitCount int64
+ missCount int64
+
+ hotWindow time.Duration
+ hotMax int
+ hotMu sync.Mutex
+ hotMap map[string][]time.Time
+}
+
+type TokenizeRequest struct {
+ Model string `json:"model,omitempty"`
+ Prompt string `json:"prompt"`
+}
+
+type TokenizeResponse struct {
+ Count int `json:"count"`
+ Tokens []int `json:"tokens"`
+ MaxLen int `json:"max_model_len"`
+}
+
+type tokenCacheEntry struct {
+ tokens []int
+ expiresAt time.Time
+}
+
+func NewTokenManager(endpoint string, httpClient *resty.Client, cfg TokenCacheConfig, cb *CircuitBreaker, hotWindow time.Duration, hotMax int) *TokenManager {
+ return &TokenManager{
+ httpClient: httpClient,
+ endpoint: endpoint,
+ config: cfg,
+ circuitBreaker: cb,
+ hotWindow: hotWindow,
+ hotMax: hotMax,
+ hotMap: make(map[string][]time.Time),
+ }
+}
+
+func (tm *TokenManager) GetTokens(ctx context.Context, model string, prompt string, rawBody []byte) ([]int, error) {
+ cacheKey := tm.cacheKey(model, prompt)
+ if tm.config.Enabled {
+ if tokens, ok := tm.loadCache(cacheKey); ok {
+ atomic.AddInt64(&tm.hitCount, 1)
+ return tokens, nil
+ }
+ atomic.AddInt64(&tm.missCount, 1)
+ }
+
+ tokens, err := tm.tokenize(ctx, model, prompt, rawBody)
+ if err != nil {
+ return nil, err
+ }
+
+ if tm.config.Enabled {
+ tm.storeCache(cacheKey, tokens)
+ }
+ return tokens, nil
+}
+
+func (tm *TokenManager) GetCachedTokens(model string, prompt string) ([]int, bool) {
+ if !tm.config.Enabled {
+ return nil, false
+ }
+ cacheKey := tm.cacheKey(model, prompt)
+ tokens, ok := tm.loadCache(cacheKey)
+ if ok {
+ atomic.AddInt64(&tm.hitCount, 1)
+ } else {
+ atomic.AddInt64(&tm.missCount, 1)
+ }
+ return tokens, ok
+}
+
+func (tm *TokenManager) InvalidateCache(model string, prompt string) {
+ cacheKey := tm.cacheKey(model, prompt)
+ if _, ok := tm.cache.Load(cacheKey); ok {
+ tm.cache.Delete(cacheKey)
+ atomic.AddInt64(&tm.cacheSize, -1)
+ }
+}
+
+func (tm *TokenManager) GetCacheStats() CacheStats {
+ size := atomic.LoadInt64(&tm.cacheSize)
+ hit := atomic.LoadInt64(&tm.hitCount)
+ miss := atomic.LoadInt64(&tm.missCount)
+ total := hit + miss
+ var hitRate float64
+ if total > 0 {
+ hitRate = float64(hit) / float64(total)
+ }
+ return CacheStats{
+ Size: int(size),
+ HitRate: hitRate,
+ HitCount: hit,
+ MissCount: miss,
+ }
+}
+
+func (tm *TokenManager) tokenize(ctx context.Context, model string, prompt string, rawBody []byte) ([]int, error) {
+ var tokens []int
+ err := tm.execute(ctx, func() error {
+ body, err := tm.buildTokenizeBody(model, prompt, rawBody)
+ if err != nil {
+ return err
+ }
+ resp, err := tm.doTokenizeRequest(ctx, body)
+ if err != nil {
+ return err
+ }
+ tokens = resp.Tokens
+ return nil
+ })
+ if err != nil {
+ return nil, err
+ }
+ return tokens, nil
+}
+
+func (tm *TokenManager) buildTokenizeBody(model string, prompt string, rawBody []byte) (any, error) {
+ if len(rawBody) > 0 {
+ return rawBody, nil
+ }
+ return TokenizeRequest{Model: model, Prompt: prompt}, nil
+}
+
+func (tm *TokenManager) doTokenizeRequest(ctx context.Context, body any) (*TokenizeResponse, error) {
+ tokenizeURL := strings.TrimRight(tm.endpoint, "/") + "/tokenize"
+ resp, err := tm.httpClient.R().
+ SetContext(ctx).
+ SetHeader("Content-Type", "application/json").
+ SetBody(body).
+ Post(tokenizeURL)
+ if err != nil {
+ return nil, fmt.Errorf("call tokenize: %w", err)
+ }
+ if resp.StatusCode() < 200 || resp.StatusCode() >= 300 {
+ return nil, fmt.Errorf("tokenize status %d: %s", resp.StatusCode(), strings.TrimSpace(string(resp.Body())))
+ }
+ var tokenResp TokenizeResponse
+ if err := json.Unmarshal(resp.Body(), &tokenResp); err != nil {
+ return nil, fmt.Errorf("decode tokenize response: %w", err)
+ }
+ return &tokenResp, nil
+}
+
+func (tm *TokenManager) RecordHot(model string, prompt string) {
+ if tm == nil || tm.hotWindow <= 0 || model == "" || prompt == "" {
+ return
+ }
+ now := time.Now()
+ key := tm.cacheKey(model, prompt)
+ tm.hotMu.Lock()
+ defer tm.hotMu.Unlock()
+ entries := tm.hotMap[key]
+ entries = append(entries, now)
+ entries = trimHotWindow(entries, now, tm.hotWindow)
+ if tm.hotMax > 0 && len(entries) > tm.hotMax {
+ entries = entries[len(entries)-tm.hotMax:]
+ }
+ if len(entries) == 0 {
+ delete(tm.hotMap, key)
+ return
+ }
+ tm.hotMap[key] = entries
+}
+
+func (tm *TokenManager) IsHot(model string, prompt string, threshold int) bool {
+ if tm == nil || tm.hotWindow <= 0 || threshold <= 0 || model == "" || prompt == "" {
+ return false
+ }
+ now := time.Now()
+ key := tm.cacheKey(model, prompt)
+ tm.hotMu.Lock()
+ defer tm.hotMu.Unlock()
+ entries := tm.hotMap[key]
+ if len(entries) == 0 {
+ return false
+ }
+ entries = trimHotWindow(entries, now, tm.hotWindow)
+ if tm.hotMax > 0 && len(entries) > tm.hotMax {
+ entries = entries[len(entries)-tm.hotMax:]
+ }
+ if len(entries) == 0 {
+ delete(tm.hotMap, key)
+ return false
+ }
+ tm.hotMap[key] = entries
+ return len(entries) >= threshold
+}
+
+func trimHotWindow(entries []time.Time, now time.Time, window time.Duration) []time.Time {
+ if window <= 0 || len(entries) == 0 {
+ return entries
+ }
+ cutoff := now.Add(-window)
+ idx := 0
+ for idx < len(entries) && entries[idx].Before(cutoff) {
+ idx++
+ }
+ if idx == 0 {
+ return entries
+ }
+ return append([]time.Time(nil), entries[idx:]...)
+}
+
+func (tm *TokenManager) execute(ctx context.Context, operation func() error) error {
+ if tm.circuitBreaker == nil {
+ return operation()
+ }
+ return tm.circuitBreaker.Execute(operation)
+}
+
+func (tm *TokenManager) cacheKey(model string, prompt string) string {
+ if model == "" {
+ return prompt
+ }
+ return model + ":" + prompt
Review Comment:
good idea. added
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
Re: [PR] feat: Add KVCache offload funtion in PIXIU gateway [dubbo-go-pixiu]
Posted by "Alanxtl (via GitHub)" <gi...@apache.org>.
Alanxtl commented on code in PR #878:
URL: https://github.com/apache/dubbo-go-pixiu/pull/878#discussion_r2779145903
##########
pkg/filter/ai/kvcache/config.go:
##########
@@ -0,0 +1,150 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 kvcache
+
+import (
+ "fmt"
+ "strings"
+ "time"
+)
+
+type Config struct {
+ Enabled bool `yaml:"enabled" json:"enabled" mapstructure:"enabled"`
+ VLLMEndpoint string `yaml:"vllm_endpoint" json:"vllm_endpoint" mapstructure:"vllm_endpoint"`
+ LMCacheEndpoint string `yaml:"lmcache_endpoint" json:"lmcache_endpoint" mapstructure:"lmcache_endpoint"`
+ DefaultModel string `yaml:"default_model" json:"default_model" mapstructure:"default_model"`
+ RequestTimeout time.Duration `yaml:"request_timeout" json:"request_timeout" mapstructure:"request_timeout"`
+ LookupRoutingTimeout time.Duration `yaml:"lookup_routing_timeout" json:"lookup_routing_timeout" mapstructure:"lookup_routing_timeout"`
+ HotWindow time.Duration `yaml:"hot_window" json:"hot_window" mapstructure:"hot_window"`
+ HotMaxRecords int `yaml:"hot_max_records" json:"hot_max_records" mapstructure:"hot_max_records"`
+ MaxIdleConns int `yaml:"max_idle_conns" json:"max_idle_conns" mapstructure:"max_idle_conns"`
+ MaxIdleConnsPerHost int `yaml:"max_idle_conns_per_host" json:"max_idle_conns_per_host" mapstructure:"max_idle_conns_per_host"`
+ MaxConnsPerHost int `yaml:"max_conns_per_host" json:"max_conns_per_host" mapstructure:"max_conns_per_host"`
+ TokenCache TokenCacheConfig `yaml:"token_cache" json:"token_cache" mapstructure:"token_cache"`
+ CacheStrategy CacheStrategyConfig `yaml:"cache_strategy" json:"cache_strategy" mapstructure:"cache_strategy"`
+ CircuitBreaker CircuitBreakerConfig `yaml:"circuit_breaker" json:"circuit_breaker" mapstructure:"circuit_breaker"`
+ Retry RetryConfig `yaml:"retry" json:"retry" mapstructure:"retry"`
+}
+
+type TokenCacheConfig struct {
+ MaxSize int `yaml:"max_size" json:"max_size" mapstructure:"max_size"`
+ TTL time.Duration `yaml:"ttl" json:"ttl" mapstructure:"ttl"`
+ Enabled bool `yaml:"enabled" json:"enabled" mapstructure:"enabled"`
+}
+
+type CacheStrategyConfig struct {
+ EnableCompression bool `yaml:"enable_compression" json:"enable_compression" mapstructure:"enable_compression"`
+ EnablePinning bool `yaml:"enable_pinning" json:"enable_pinning" mapstructure:"enable_pinning"`
+ EnableEviction bool `yaml:"enable_eviction" json:"enable_eviction" mapstructure:"enable_eviction"`
+ MemoryThreshold float64 `yaml:"memory_threshold" json:"memory_threshold" mapstructure:"memory_threshold"`
+ HotContentThreshold int `yaml:"hot_content_threshold" json:"hot_content_threshold" mapstructure:"hot_content_threshold"`
+ LoadThreshold float64 `yaml:"load_threshold" json:"load_threshold" mapstructure:"load_threshold"`
+ PinInstanceID string `yaml:"pin_instance_id" json:"pin_instance_id" mapstructure:"pin_instance_id"`
+ PinLocation string `yaml:"pin_location" json:"pin_location" mapstructure:"pin_location"`
+ CompressInstanceID string `yaml:"compress_instance_id" json:"compress_instance_id" mapstructure:"compress_instance_id"`
+ CompressLocation string `yaml:"compress_location" json:"compress_location" mapstructure:"compress_location"`
+ CompressMethod string `yaml:"compress_method" json:"compress_method" mapstructure:"compress_method"`
+ EvictInstanceID string `yaml:"evict_instance_id" json:"evict_instance_id" mapstructure:"evict_instance_id"`
+}
+
+type RetryConfig struct {
+ MaxAttempts int `yaml:"max_attempts" json:"max_attempts" mapstructure:"max_attempts"`
+ BaseBackoff time.Duration `yaml:"base_backoff" json:"base_backoff" mapstructure:"base_backoff"`
+ MaxBackoff time.Duration `yaml:"max_backoff" json:"max_backoff" mapstructure:"max_backoff"`
+}
+
+func (c *Config) Validate() error {
+ if !c.Enabled {
+ return nil
+ }
+ if strings.TrimSpace(c.VLLMEndpoint) == "" {
+ return fmt.Errorf("kvcache: vllm_endpoint is required when enabled")
+ }
+ if strings.TrimSpace(c.LMCacheEndpoint) == "" {
+ return fmt.Errorf("kvcache: lmcache_endpoint is required when enabled")
+ }
+ if c.TokenCache.MaxSize < 0 {
+ return fmt.Errorf("kvcache: token_cache.max_size must be >= 0")
+ }
+ if c.CacheStrategy.MemoryThreshold < 0 || c.CacheStrategy.MemoryThreshold > 1 {
+ return fmt.Errorf("kvcache: cache_strategy.memory_threshold must be between 0 and 1")
+ }
+ if c.CacheStrategy.LoadThreshold < 0 || c.CacheStrategy.LoadThreshold > 1 {
+ return fmt.Errorf("kvcache: cache_strategy.load_threshold must be between 0 and 1")
+ }
+ if c.CacheStrategy.HotContentThreshold < 0 {
+ return fmt.Errorf("kvcache: cache_strategy.hot_content_threshold must be >= 0")
+ }
+ if c.Retry.MaxAttempts < 0 {
+ return fmt.Errorf("kvcache: retry.max_attempts must be >= 0")
+ }
+ if c.Retry.BaseBackoff < 0 || c.Retry.MaxBackoff < 0 {
+ return fmt.Errorf("kvcache: retry backoff durations must be >= 0")
+ }
+ if c.HotWindow < 0 {
+ return fmt.Errorf("kvcache: hot_window must be >= 0")
+ }
+ if c.HotMaxRecords < 0 {
+ return fmt.Errorf("kvcache: hot_max_records must be >= 0")
+ }
+ return nil
+}
+
+func (c *Config) ApplyDefaults() {
+ if c.RequestTimeout <= 0 {
+ c.RequestTimeout = 2 * time.Second
Review Comment:
把常量都抽出去
##########
pkg/filter/llm/proxy/filter.go:
##########
@@ -98,6 +98,18 @@ type (
}
)
+func getPreferredEndpointID(hc *contexthttp.HttpContext) string {
+ if hc == nil || hc.Params == nil {
+ return ""
+ }
+ val, ok := hc.Params[constant.LLMPreferredEndpointID]
+ if !ok {
+ return ""
+ }
+ endpointID, _ := val.(string)
+ return endpointID
Review Comment:
```suggestion
if val, ok := hc.Params[constant.LLMPreferredEndpointID]; !ok {
return ""
}
if val, ok := val.(string); ok {
return val
}
return endpointID
```
##########
pkg/filter/ai/kvcache/circuit_breaker.go:
##########
@@ -0,0 +1,106 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 kvcache
+
+import (
+ "errors"
+ "sync"
+ "time"
+)
+
+var ErrCircuitBreakerOpen = errors.New("kvcache circuit breaker open")
+
+type CircuitState int
+
+const (
+ CircuitClosed CircuitState = iota
Review Comment:
not suggest using iota
##########
pkg/common/constant/key.go:
##########
@@ -59,8 +59,11 @@ const (
LLMProxyFilter = "dgp.filter.llm.proxy"
LLMTokenizerFilter = "dgp.filter.llm.tokenizer"
+ AIKVCacheFilter = "dgp.filter.ai.kvcache"
MCPServerFilter = "dgp.filter.mcp.mcpserver"
+
+ LLMPreferredEndpointID = "llm_preferred_endpoint_id"
Review Comment:
这个常量放到这里不合适吧
##########
pkg/filter/ai/kvcache/config.go:
##########
@@ -0,0 +1,150 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 kvcache
+
+import (
+ "fmt"
+ "strings"
+ "time"
+)
+
+type Config struct {
+ Enabled bool `yaml:"enabled" json:"enabled" mapstructure:"enabled"`
+ VLLMEndpoint string `yaml:"vllm_endpoint" json:"vllm_endpoint" mapstructure:"vllm_endpoint"`
+ LMCacheEndpoint string `yaml:"lmcache_endpoint" json:"lmcache_endpoint" mapstructure:"lmcache_endpoint"`
+ DefaultModel string `yaml:"default_model" json:"default_model" mapstructure:"default_model"`
+ RequestTimeout time.Duration `yaml:"request_timeout" json:"request_timeout" mapstructure:"request_timeout"`
+ LookupRoutingTimeout time.Duration `yaml:"lookup_routing_timeout" json:"lookup_routing_timeout" mapstructure:"lookup_routing_timeout"`
+ HotWindow time.Duration `yaml:"hot_window" json:"hot_window" mapstructure:"hot_window"`
+ HotMaxRecords int `yaml:"hot_max_records" json:"hot_max_records" mapstructure:"hot_max_records"`
+ MaxIdleConns int `yaml:"max_idle_conns" json:"max_idle_conns" mapstructure:"max_idle_conns"`
+ MaxIdleConnsPerHost int `yaml:"max_idle_conns_per_host" json:"max_idle_conns_per_host" mapstructure:"max_idle_conns_per_host"`
+ MaxConnsPerHost int `yaml:"max_conns_per_host" json:"max_conns_per_host" mapstructure:"max_conns_per_host"`
+ TokenCache TokenCacheConfig `yaml:"token_cache" json:"token_cache" mapstructure:"token_cache"`
+ CacheStrategy CacheStrategyConfig `yaml:"cache_strategy" json:"cache_strategy" mapstructure:"cache_strategy"`
+ CircuitBreaker CircuitBreakerConfig `yaml:"circuit_breaker" json:"circuit_breaker" mapstructure:"circuit_breaker"`
+ Retry RetryConfig `yaml:"retry" json:"retry" mapstructure:"retry"`
+}
+
+type TokenCacheConfig struct {
+ MaxSize int `yaml:"max_size" json:"max_size" mapstructure:"max_size"`
+ TTL time.Duration `yaml:"ttl" json:"ttl" mapstructure:"ttl"`
+ Enabled bool `yaml:"enabled" json:"enabled" mapstructure:"enabled"`
+}
+
+type CacheStrategyConfig struct {
+ EnableCompression bool `yaml:"enable_compression" json:"enable_compression" mapstructure:"enable_compression"`
+ EnablePinning bool `yaml:"enable_pinning" json:"enable_pinning" mapstructure:"enable_pinning"`
+ EnableEviction bool `yaml:"enable_eviction" json:"enable_eviction" mapstructure:"enable_eviction"`
+ MemoryThreshold float64 `yaml:"memory_threshold" json:"memory_threshold" mapstructure:"memory_threshold"`
+ HotContentThreshold int `yaml:"hot_content_threshold" json:"hot_content_threshold" mapstructure:"hot_content_threshold"`
+ LoadThreshold float64 `yaml:"load_threshold" json:"load_threshold" mapstructure:"load_threshold"`
+ PinInstanceID string `yaml:"pin_instance_id" json:"pin_instance_id" mapstructure:"pin_instance_id"`
+ PinLocation string `yaml:"pin_location" json:"pin_location" mapstructure:"pin_location"`
+ CompressInstanceID string `yaml:"compress_instance_id" json:"compress_instance_id" mapstructure:"compress_instance_id"`
+ CompressLocation string `yaml:"compress_location" json:"compress_location" mapstructure:"compress_location"`
+ CompressMethod string `yaml:"compress_method" json:"compress_method" mapstructure:"compress_method"`
+ EvictInstanceID string `yaml:"evict_instance_id" json:"evict_instance_id" mapstructure:"evict_instance_id"`
+}
+
+type RetryConfig struct {
+ MaxAttempts int `yaml:"max_attempts" json:"max_attempts" mapstructure:"max_attempts"`
+ BaseBackoff time.Duration `yaml:"base_backoff" json:"base_backoff" mapstructure:"base_backoff"`
+ MaxBackoff time.Duration `yaml:"max_backoff" json:"max_backoff" mapstructure:"max_backoff"`
+}
Review Comment:
这个retry能不能复用pixiu已有的retry呀
##########
pkg/filter/ai/kvcache/config.go:
##########
@@ -0,0 +1,150 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 kvcache
+
+import (
+ "fmt"
+ "strings"
+ "time"
+)
+
+type Config struct {
+ Enabled bool `yaml:"enabled" json:"enabled" mapstructure:"enabled"`
+ VLLMEndpoint string `yaml:"vllm_endpoint" json:"vllm_endpoint" mapstructure:"vllm_endpoint"`
+ LMCacheEndpoint string `yaml:"lmcache_endpoint" json:"lmcache_endpoint" mapstructure:"lmcache_endpoint"`
+ DefaultModel string `yaml:"default_model" json:"default_model" mapstructure:"default_model"`
+ RequestTimeout time.Duration `yaml:"request_timeout" json:"request_timeout" mapstructure:"request_timeout"`
+ LookupRoutingTimeout time.Duration `yaml:"lookup_routing_timeout" json:"lookup_routing_timeout" mapstructure:"lookup_routing_timeout"`
+ HotWindow time.Duration `yaml:"hot_window" json:"hot_window" mapstructure:"hot_window"`
+ HotMaxRecords int `yaml:"hot_max_records" json:"hot_max_records" mapstructure:"hot_max_records"`
+ MaxIdleConns int `yaml:"max_idle_conns" json:"max_idle_conns" mapstructure:"max_idle_conns"`
+ MaxIdleConnsPerHost int `yaml:"max_idle_conns_per_host" json:"max_idle_conns_per_host" mapstructure:"max_idle_conns_per_host"`
+ MaxConnsPerHost int `yaml:"max_conns_per_host" json:"max_conns_per_host" mapstructure:"max_conns_per_host"`
+ TokenCache TokenCacheConfig `yaml:"token_cache" json:"token_cache" mapstructure:"token_cache"`
+ CacheStrategy CacheStrategyConfig `yaml:"cache_strategy" json:"cache_strategy" mapstructure:"cache_strategy"`
+ CircuitBreaker CircuitBreakerConfig `yaml:"circuit_breaker" json:"circuit_breaker" mapstructure:"circuit_breaker"`
+ Retry RetryConfig `yaml:"retry" json:"retry" mapstructure:"retry"`
+}
+
+type TokenCacheConfig struct {
+ MaxSize int `yaml:"max_size" json:"max_size" mapstructure:"max_size"`
+ TTL time.Duration `yaml:"ttl" json:"ttl" mapstructure:"ttl"`
+ Enabled bool `yaml:"enabled" json:"enabled" mapstructure:"enabled"`
+}
+
+type CacheStrategyConfig struct {
+ EnableCompression bool `yaml:"enable_compression" json:"enable_compression" mapstructure:"enable_compression"`
+ EnablePinning bool `yaml:"enable_pinning" json:"enable_pinning" mapstructure:"enable_pinning"`
+ EnableEviction bool `yaml:"enable_eviction" json:"enable_eviction" mapstructure:"enable_eviction"`
+ MemoryThreshold float64 `yaml:"memory_threshold" json:"memory_threshold" mapstructure:"memory_threshold"`
+ HotContentThreshold int `yaml:"hot_content_threshold" json:"hot_content_threshold" mapstructure:"hot_content_threshold"`
+ LoadThreshold float64 `yaml:"load_threshold" json:"load_threshold" mapstructure:"load_threshold"`
+ PinInstanceID string `yaml:"pin_instance_id" json:"pin_instance_id" mapstructure:"pin_instance_id"`
+ PinLocation string `yaml:"pin_location" json:"pin_location" mapstructure:"pin_location"`
+ CompressInstanceID string `yaml:"compress_instance_id" json:"compress_instance_id" mapstructure:"compress_instance_id"`
+ CompressLocation string `yaml:"compress_location" json:"compress_location" mapstructure:"compress_location"`
+ CompressMethod string `yaml:"compress_method" json:"compress_method" mapstructure:"compress_method"`
+ EvictInstanceID string `yaml:"evict_instance_id" json:"evict_instance_id" mapstructure:"evict_instance_id"`
+}
+
+type RetryConfig struct {
+ MaxAttempts int `yaml:"max_attempts" json:"max_attempts" mapstructure:"max_attempts"`
+ BaseBackoff time.Duration `yaml:"base_backoff" json:"base_backoff" mapstructure:"base_backoff"`
+ MaxBackoff time.Duration `yaml:"max_backoff" json:"max_backoff" mapstructure:"max_backoff"`
+}
+
+func (c *Config) Validate() error {
+ if !c.Enabled {
+ return nil
+ }
+ if strings.TrimSpace(c.VLLMEndpoint) == "" {
+ return fmt.Errorf("kvcache: vllm_endpoint is required when enabled")
Review Comment:
把error的格式都改成这样的格式
```
[kvcache] vllm_endpoint is required when enabled
```
##########
pkg/filter/ai/kvcache/token_manager.go:
##########
@@ -0,0 +1,305 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 kvcache
+
+import (
+ "context"
+ "crypto/sha256"
+ "encoding/hex"
+ "encoding/json"
+ "fmt"
+ "strings"
+ "sync"
+ "sync/atomic"
+ "time"
+)
+
+import (
+ "github.com/go-resty/resty/v2"
+)
+
+type TokenManager struct {
+ httpClient *resty.Client
+ endpoint string
+ cache sync.Map
+ config TokenCacheConfig
+ circuitBreaker *CircuitBreaker
+
+ cacheSize int64
+ hitCount int64
+ missCount int64
+
+ hotWindow time.Duration
+ hotMax int
+ hotMu sync.Mutex
+ hotMap map[string][]time.Time
+}
+
+type TokenizeRequest struct {
+ Model string `json:"model,omitempty"`
+ Prompt string `json:"prompt"`
+}
+
+type TokenizeResponse struct {
+ Count int `json:"count"`
+ Tokens []int `json:"tokens"`
+ MaxLen int `json:"max_model_len"`
+}
+
+type tokenCacheEntry struct {
+ tokens []int
+ expiresAt time.Time
+}
+
+func NewTokenManager(endpoint string, httpClient *resty.Client, cfg TokenCacheConfig, cb *CircuitBreaker, hotWindow time.Duration, hotMax int) *TokenManager {
+ return &TokenManager{
+ httpClient: httpClient,
+ endpoint: endpoint,
+ config: cfg,
+ circuitBreaker: cb,
+ hotWindow: hotWindow,
+ hotMax: hotMax,
+ hotMap: make(map[string][]time.Time),
+ }
+}
+
+func (tm *TokenManager) GetTokens(ctx context.Context, model string, prompt string, rawBody []byte) ([]int, error) {
+ cacheKey := tm.cacheKey(model, prompt)
+ if tm.config.Enabled {
+ if tokens, ok := tm.loadCache(cacheKey); ok {
+ atomic.AddInt64(&tm.hitCount, 1)
+ return tokens, nil
+ }
+ atomic.AddInt64(&tm.missCount, 1)
+ }
+
+ tokens, err := tm.tokenize(ctx, model, prompt, rawBody)
+ if err != nil {
+ return nil, err
+ }
+
+ if tm.config.Enabled {
+ tm.storeCache(cacheKey, tokens)
+ }
+ return tokens, nil
+}
+
+func (tm *TokenManager) GetCachedTokens(model string, prompt string) ([]int, bool) {
+ if !tm.config.Enabled {
+ return nil, false
+ }
+ cacheKey := tm.cacheKey(model, prompt)
+ tokens, ok := tm.loadCache(cacheKey)
+ if ok {
+ atomic.AddInt64(&tm.hitCount, 1)
+ } else {
+ atomic.AddInt64(&tm.missCount, 1)
+ }
+ return tokens, ok
+}
+
+func (tm *TokenManager) InvalidateCache(model string, prompt string) {
+ cacheKey := tm.cacheKey(model, prompt)
+ tm.deleteCache(cacheKey)
+}
+
+func (tm *TokenManager) GetCacheStats() CacheStats {
+ size := atomic.LoadInt64(&tm.cacheSize)
+ hit := atomic.LoadInt64(&tm.hitCount)
+ miss := atomic.LoadInt64(&tm.missCount)
+ total := hit + miss
+ var hitRate float64
+ if total > 0 {
+ hitRate = float64(hit) / float64(total)
+ }
+ return CacheStats{
+ Size: int(size),
+ HitRate: hitRate,
+ HitCount: hit,
+ MissCount: miss,
+ }
+}
+
+func (tm *TokenManager) tokenize(ctx context.Context, model string, prompt string, rawBody []byte) ([]int, error) {
+ var tokens []int
+ err := tm.execute(ctx, func() error {
+ body, err := tm.buildTokenizeBody(model, prompt, rawBody)
+ if err != nil {
+ return err
+ }
+ resp, err := tm.doTokenizeRequest(ctx, body)
+ if err != nil {
+ return err
+ }
+ tokens = resp.Tokens
+ return nil
+ })
+ if err != nil {
+ return nil, err
+ }
+ return tokens, nil
+}
+
+func (tm *TokenManager) buildTokenizeBody(model string, prompt string, rawBody []byte) (any, error) {
+ if len(rawBody) > 0 {
+ return rawBody, nil
+ }
+ return TokenizeRequest{Model: model, Prompt: prompt}, nil
+}
+
+func (tm *TokenManager) doTokenizeRequest(ctx context.Context, body any) (*TokenizeResponse, error) {
Review Comment:
这个Tokenize是什么概念呀
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
Re: [PR] feat: Add KVCache offload funtion in PIXIU gateway [dubbo-go-pixiu]
Posted by "sonarqubecloud[bot] (via GitHub)" <gi...@apache.org>.
sonarqubecloud[bot] commented on PR #878:
URL: https://github.com/apache/dubbo-go-pixiu/pull/878#issuecomment-3867221983
## [](https://sonarcloud.io/dashboard?id=apache_dubbo-go-pixiu&pullRequest=878) **Quality Gate passed**
Issues
 [7 New issues](https://sonarcloud.io/project/issues?id=apache_dubbo-go-pixiu&pullRequest=878&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)
 [0 Accepted issues](https://sonarcloud.io/project/issues?id=apache_dubbo-go-pixiu&pullRequest=878&issueStatuses=ACCEPTED)
Measures
 [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=apache_dubbo-go-pixiu&pullRequest=878&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)
 [0.0% Coverage on New Code](https://sonarcloud.io/component_measures?id=apache_dubbo-go-pixiu&pullRequest=878&metric=new_coverage&view=list)
 [0.0% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_dubbo-go-pixiu&pullRequest=878&metric=new_duplicated_lines_density&view=list)
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=apache_dubbo-go-pixiu&pullRequest=878)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
Re: [PR] feat: Add KVCache offload funtion in PIXIU gateway [dubbo-go-pixiu]
Posted by "sonarqubecloud[bot] (via GitHub)" <gi...@apache.org>.
sonarqubecloud[bot] commented on PR #878:
URL: https://github.com/apache/dubbo-go-pixiu/pull/878#issuecomment-3860642267
## [](https://sonarcloud.io/dashboard?id=apache_dubbo-go-pixiu&pullRequest=878) **Quality Gate passed**
Issues
 [7 New issues](https://sonarcloud.io/project/issues?id=apache_dubbo-go-pixiu&pullRequest=878&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)
 [0 Accepted issues](https://sonarcloud.io/project/issues?id=apache_dubbo-go-pixiu&pullRequest=878&issueStatuses=ACCEPTED)
Measures
 [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=apache_dubbo-go-pixiu&pullRequest=878&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)
 [0.0% Coverage on New Code](https://sonarcloud.io/component_measures?id=apache_dubbo-go-pixiu&pullRequest=878&metric=new_coverage&view=list)
 [0.0% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_dubbo-go-pixiu&pullRequest=878&metric=new_duplicated_lines_density&view=list)
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=apache_dubbo-go-pixiu&pullRequest=878)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
Re: [PR] feat: Add KVCache offload funtion in PIXIU gateway [dubbo-go-pixiu]
Posted by "Alanxtl (via GitHub)" <gi...@apache.org>.
Alanxtl merged PR #878:
URL: https://github.com/apache/dubbo-go-pixiu/pull/878
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
Re: [PR] feat: Add KVCache offload funtion in PIXIU gateway [dubbo-go-pixiu]
Posted by "sonarqubecloud[bot] (via GitHub)" <gi...@apache.org>.
sonarqubecloud[bot] commented on PR #878:
URL: https://github.com/apache/dubbo-go-pixiu/pull/878#issuecomment-3928214689
## [](https://sonarcloud.io/dashboard?id=apache_dubbo-go-pixiu&pullRequest=878) **Quality Gate passed**
Issues
 [10 New issues](https://sonarcloud.io/project/issues?id=apache_dubbo-go-pixiu&pullRequest=878&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)
 [0 Accepted issues](https://sonarcloud.io/project/issues?id=apache_dubbo-go-pixiu&pullRequest=878&issueStatuses=ACCEPTED)
Measures
 [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=apache_dubbo-go-pixiu&pullRequest=878&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)
 [0.0% Coverage on New Code](https://sonarcloud.io/component_measures?id=apache_dubbo-go-pixiu&pullRequest=878&metric=new_coverage&view=list)
 [0.0% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_dubbo-go-pixiu&pullRequest=878&metric=new_duplicated_lines_density&view=list)
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=apache_dubbo-go-pixiu&pullRequest=878)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
Re: [PR] feat: Add KVCache offload funtion in PIXIU gateway [dubbo-go-pixiu]
Posted by "sonarqubecloud[bot] (via GitHub)" <gi...@apache.org>.
sonarqubecloud[bot] commented on PR #878:
URL: https://github.com/apache/dubbo-go-pixiu/pull/878#issuecomment-3860652386
## [](https://sonarcloud.io/dashboard?id=apache_dubbo-go-pixiu&pullRequest=878) **Quality Gate passed**
Issues
 [7 New issues](https://sonarcloud.io/project/issues?id=apache_dubbo-go-pixiu&pullRequest=878&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)
 [0 Accepted issues](https://sonarcloud.io/project/issues?id=apache_dubbo-go-pixiu&pullRequest=878&issueStatuses=ACCEPTED)
Measures
 [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=apache_dubbo-go-pixiu&pullRequest=878&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)
 [0.0% Coverage on New Code](https://sonarcloud.io/component_measures?id=apache_dubbo-go-pixiu&pullRequest=878&metric=new_coverage&view=list)
 [0.0% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_dubbo-go-pixiu&pullRequest=878&metric=new_duplicated_lines_density&view=list)
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=apache_dubbo-go-pixiu&pullRequest=878)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
Re: [PR] feat: Add KVCache offload funtion in PIXIU gateway [dubbo-go-pixiu]
Posted by "sonarqubecloud[bot] (via GitHub)" <gi...@apache.org>.
sonarqubecloud[bot] commented on PR #878:
URL: https://github.com/apache/dubbo-go-pixiu/pull/878#issuecomment-3889644955
## [](https://sonarcloud.io/dashboard?id=apache_dubbo-go-pixiu&pullRequest=878) **Quality Gate passed**
Issues
 [10 New issues](https://sonarcloud.io/project/issues?id=apache_dubbo-go-pixiu&pullRequest=878&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)
 [0 Accepted issues](https://sonarcloud.io/project/issues?id=apache_dubbo-go-pixiu&pullRequest=878&issueStatuses=ACCEPTED)
Measures
 [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=apache_dubbo-go-pixiu&pullRequest=878&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)
 [0.0% Coverage on New Code](https://sonarcloud.io/component_measures?id=apache_dubbo-go-pixiu&pullRequest=878&metric=new_coverage&view=list)
 [0.0% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_dubbo-go-pixiu&pullRequest=878&metric=new_duplicated_lines_density&view=list)
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=apache_dubbo-go-pixiu&pullRequest=878)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
Re: [PR] feat: Add KVCache offload funtion in PIXIU gateway [dubbo-go-pixiu]
Posted by "Chen-BUPT (via GitHub)" <gi...@apache.org>.
Chen-BUPT commented on code in PR #878:
URL: https://github.com/apache/dubbo-go-pixiu/pull/878#discussion_r2779274522
##########
pkg/filter/ai/kvcache/config.go:
##########
@@ -0,0 +1,150 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 kvcache
+
+import (
+ "fmt"
+ "strings"
+ "time"
+)
+
+type Config struct {
+ Enabled bool `yaml:"enabled" json:"enabled" mapstructure:"enabled"`
+ VLLMEndpoint string `yaml:"vllm_endpoint" json:"vllm_endpoint" mapstructure:"vllm_endpoint"`
+ LMCacheEndpoint string `yaml:"lmcache_endpoint" json:"lmcache_endpoint" mapstructure:"lmcache_endpoint"`
+ DefaultModel string `yaml:"default_model" json:"default_model" mapstructure:"default_model"`
+ RequestTimeout time.Duration `yaml:"request_timeout" json:"request_timeout" mapstructure:"request_timeout"`
+ LookupRoutingTimeout time.Duration `yaml:"lookup_routing_timeout" json:"lookup_routing_timeout" mapstructure:"lookup_routing_timeout"`
+ HotWindow time.Duration `yaml:"hot_window" json:"hot_window" mapstructure:"hot_window"`
+ HotMaxRecords int `yaml:"hot_max_records" json:"hot_max_records" mapstructure:"hot_max_records"`
+ MaxIdleConns int `yaml:"max_idle_conns" json:"max_idle_conns" mapstructure:"max_idle_conns"`
+ MaxIdleConnsPerHost int `yaml:"max_idle_conns_per_host" json:"max_idle_conns_per_host" mapstructure:"max_idle_conns_per_host"`
+ MaxConnsPerHost int `yaml:"max_conns_per_host" json:"max_conns_per_host" mapstructure:"max_conns_per_host"`
+ TokenCache TokenCacheConfig `yaml:"token_cache" json:"token_cache" mapstructure:"token_cache"`
+ CacheStrategy CacheStrategyConfig `yaml:"cache_strategy" json:"cache_strategy" mapstructure:"cache_strategy"`
+ CircuitBreaker CircuitBreakerConfig `yaml:"circuit_breaker" json:"circuit_breaker" mapstructure:"circuit_breaker"`
+ Retry RetryConfig `yaml:"retry" json:"retry" mapstructure:"retry"`
+}
+
+type TokenCacheConfig struct {
+ MaxSize int `yaml:"max_size" json:"max_size" mapstructure:"max_size"`
+ TTL time.Duration `yaml:"ttl" json:"ttl" mapstructure:"ttl"`
+ Enabled bool `yaml:"enabled" json:"enabled" mapstructure:"enabled"`
+}
+
+type CacheStrategyConfig struct {
+ EnableCompression bool `yaml:"enable_compression" json:"enable_compression" mapstructure:"enable_compression"`
+ EnablePinning bool `yaml:"enable_pinning" json:"enable_pinning" mapstructure:"enable_pinning"`
+ EnableEviction bool `yaml:"enable_eviction" json:"enable_eviction" mapstructure:"enable_eviction"`
+ MemoryThreshold float64 `yaml:"memory_threshold" json:"memory_threshold" mapstructure:"memory_threshold"`
+ HotContentThreshold int `yaml:"hot_content_threshold" json:"hot_content_threshold" mapstructure:"hot_content_threshold"`
+ LoadThreshold float64 `yaml:"load_threshold" json:"load_threshold" mapstructure:"load_threshold"`
+ PinInstanceID string `yaml:"pin_instance_id" json:"pin_instance_id" mapstructure:"pin_instance_id"`
+ PinLocation string `yaml:"pin_location" json:"pin_location" mapstructure:"pin_location"`
+ CompressInstanceID string `yaml:"compress_instance_id" json:"compress_instance_id" mapstructure:"compress_instance_id"`
+ CompressLocation string `yaml:"compress_location" json:"compress_location" mapstructure:"compress_location"`
+ CompressMethod string `yaml:"compress_method" json:"compress_method" mapstructure:"compress_method"`
+ EvictInstanceID string `yaml:"evict_instance_id" json:"evict_instance_id" mapstructure:"evict_instance_id"`
+}
+
+type RetryConfig struct {
+ MaxAttempts int `yaml:"max_attempts" json:"max_attempts" mapstructure:"max_attempts"`
+ BaseBackoff time.Duration `yaml:"base_backoff" json:"base_backoff" mapstructure:"base_backoff"`
+ MaxBackoff time.Duration `yaml:"max_backoff" json:"max_backoff" mapstructure:"max_backoff"`
+}
Review Comment:
这里的 retry 是给 filter 内部调用 外部 API（/lookup...）用的，pixiu 现有的 retry 不太能满足
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
Re: [PR] feat: Add KVCache offload funtion in PIXIU gateway [dubbo-go-pixiu]
Posted by "sonarqubecloud[bot] (via GitHub)" <gi...@apache.org>.
sonarqubecloud[bot] commented on PR #878:
URL: https://github.com/apache/dubbo-go-pixiu/pull/878#issuecomment-3863668351
## [](https://sonarcloud.io/dashboard?id=apache_dubbo-go-pixiu&pullRequest=878) **Quality Gate passed**
Issues
 [7 New issues](https://sonarcloud.io/project/issues?id=apache_dubbo-go-pixiu&pullRequest=878&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)
 [0 Accepted issues](https://sonarcloud.io/project/issues?id=apache_dubbo-go-pixiu&pullRequest=878&issueStatuses=ACCEPTED)
Measures
 [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=apache_dubbo-go-pixiu&pullRequest=878&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)
 [0.0% Coverage on New Code](https://sonarcloud.io/component_measures?id=apache_dubbo-go-pixiu&pullRequest=878&metric=new_coverage&view=list)
 [0.0% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_dubbo-go-pixiu&pullRequest=878&metric=new_duplicated_lines_density&view=list)
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=apache_dubbo-go-pixiu&pullRequest=878)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org