Merge pull request #34 from whalecold/chore/limit

fix:  use the correct envoy field to set the limit policy

README.md

@@ -260,6 +260,7 @@ spec:
             token_bucket:
               # the qps limit
               max_tokens: 4
+              tokens_per_fill: 10
   workloadSelector:
     labels:
       # the label of the service pod.

README_CN.md

@@ -250,6 +250,7 @@ spec:
             token_bucket:
               # 限流参数
               max_tokens: 4
+              tokens_per_fill: 10
   workloadSelector:
     labels:
       # 服务实例 pod 的标签，根据实际情况填写

core/manager/manager_test.go

@@ -315,13 +315,15 @@ func TestRegisterXDSUpdateHandler(t *testing.T) {
 			{
 				RoutePort: 80,
 				InlineRouteConfig: &xdsresource.RouteConfigResource{
-					MaxTokens: 100,
+					MaxTokens:     100,
+					TokensPerFill: 100,
 				},
 			},
 			{
 				RoutePort: 0,
 				InlineRouteConfig: &xdsresource.RouteConfigResource{
-					MaxTokens: 1000,
+					MaxTokens:     1000,
+					TokensPerFill: 100,
 				},
 			},
 		},

core/xdsresource/lds.go

@@ -207,7 +207,7 @@ func unmarshallHTTPConnectionManager(rawResources *any.Any) (string, *RouteConfi
 	if err := proto.Unmarshal(rawResources.GetValue(), httpConnMng); err != nil {
 		return "", nil, fmt.Errorf("unmarshal HttpConnectionManager failed: %s", err)
 	}
-	maxTokens, err := getLocalRateLimitFromHttpConnectionManager(httpConnMng)
+	maxTokens, tokensPerfill, err := getLocalRateLimitFromHttpConnectionManager(httpConnMng)
 	if err != nil {
 		return "", nil, err
 	}
@@ -223,7 +223,8 @@ func unmarshallHTTPConnectionManager(rawResources *any.Any) (string, *RouteConfi
 			return "", nil, fmt.Errorf("no route config Name")
 		}
 		return httpConnMng.GetRds().GetRouteConfigName(), &RouteConfigResource{
-			MaxTokens: maxTokens,
+			MaxTokens:     maxTokens,
+			TokensPerFill: tokensPerfill,
 		}, nil
 	case *v3httppb.HttpConnectionManager_RouteConfig:
 		var rcfg *v3routepb.RouteConfiguration
@@ -235,12 +236,13 @@ func unmarshallHTTPConnectionManager(rawResources *any.Any) (string, *RouteConfi
 			return "", nil, err
 		}
 		inlineRouteConfig.MaxTokens = maxTokens
+		inlineRouteConfig.TokensPerFill = tokensPerfill
 		return httpConnMng.GetRouteConfig().GetName(), inlineRouteConfig, nil
 	}
 	return "", nil, nil
 }
 
-func getLocalRateLimitFromHttpConnectionManager(hcm *v3httppb.HttpConnectionManager) (uint32, error) {
+func getLocalRateLimitFromHttpConnectionManager(hcm *v3httppb.HttpConnectionManager) (uint32, uint32, error) {
 	for _, filter := range hcm.HttpFilters {
 		switch filter.ConfigType.(type) {
 		case *v3httppb.HttpFilter_TypedConfig:
@@ -252,16 +254,16 @@ func getLocalRateLimitFromHttpConnectionManager(hcm *v3httppb.HttpConnectionMana
 			case RateLimitTypeURL:
 				lrl := &ratelimitv3.LocalRateLimit{}
 				if err := proto.Unmarshal(typedConfig, lrl); err != nil {
-					return 0, fmt.Errorf("unmarshal LocalRateLimit failed: %s", err)
+					return 0, 0, fmt.Errorf("unmarshal LocalRateLimit failed: %s", err)
 				}
 				if lrl.TokenBucket != nil {
-					return lrl.TokenBucket.MaxTokens, nil
+					return lrl.TokenBucket.MaxTokens, lrl.TokenBucket.TokensPerFill.GetValue(), nil
 				}
 			case TypedStructTypeURL:
 				// ratelimit may be configured with udpa struct.
 				ts := &udpatypev1.TypedStruct{}
 				if err := proto.Unmarshal(typedConfig, ts); err != nil {
-					return 0, fmt.Errorf("unmarshal TypedStruct failed: %s", err)
+					return 0, 0, fmt.Errorf("unmarshal TypedStruct failed: %s", err)
 				}
 				tokenBucket, ok := ts.GetValue().GetFields()["token_bucket"]
 				if !ok {
@@ -271,10 +273,14 @@ func getLocalRateLimitFromHttpConnectionManager(hcm *v3httppb.HttpConnectionMana
 				if !ok {
 					continue
 				}
-				return uint32(maxTokens.GetNumberValue()), nil
+				tokensPerfill, ok := tokenBucket.GetStructValue().GetFields()["tokens_per_fill"]
+				if !ok {
+					continue
+				}
+				return uint32(maxTokens.GetNumberValue()), uint32(tokensPerfill.GetNumberValue()), nil
 			}
 		}
-		return 0, nil
+		return 0, 0, nil
 	}
-	return 0, nil
+	return 0, 0, nil
 }

core/xdsresource/lds_test.go

@@ -27,6 +27,7 @@ import (
 	v3 "github.com/envoyproxy/go-control-plane/envoy/type/v3"
 	"github.com/golang/protobuf/ptypes/any"
 	_struct "github.com/golang/protobuf/ptypes/struct"
+	wrappers "github.com/golang/protobuf/ptypes/wrappers"
 	"github.com/stretchr/testify/assert"
 	"google.golang.org/protobuf/types/known/structpb"
 )
@@ -82,6 +83,7 @@ func TestUnmarshalLDSHttpConnectionManager(t *testing.T) {
 	assert.Equal(t, RouteConfigName1, lis1.NetworkFilters[0].RouteConfigName)
 	assert.Equal(t, uint32(80), lis1.NetworkFilters[0].RoutePort)
 	assert.Equal(t, uint32(10), lis1.NetworkFilters[0].InlineRouteConfig.MaxTokens)
+	assert.Equal(t, uint32(101), lis1.NetworkFilters[0].InlineRouteConfig.TokensPerFill)
 
 	// inline route config
 	lis2 := res[ReturnedLisName2]
@@ -145,6 +147,9 @@ func TestGetLocalRateLimitFromHttpConnectionManager(t *testing.T) {
 	rateLimit := &ratelimitv3.LocalRateLimit{
 		TokenBucket: &v3.TokenBucket{
 			MaxTokens: 10,
+			TokensPerFill: &wrappers.UInt32Value{
+				Value: 10,
+			},
 		},
 	}
 	hcm := &v3httppb.HttpConnectionManager{
@@ -161,9 +166,10 @@ func TestGetLocalRateLimitFromHttpConnectionManager(t *testing.T) {
 			},
 		},
 	}
-	token, err := getLocalRateLimitFromHttpConnectionManager(hcm)
+	token, tokensPerfill, err := getLocalRateLimitFromHttpConnectionManager(hcm)
 	assert.Equal(t, err, nil)
 	assert.Equal(t, token, uint32(10))
+	assert.Equal(t, tokensPerfill, uint32(10))
 
 	// ---------------------------------- struct ratelimit ------------------------------------
 	structLimit := &udpatypev1.TypedStruct{
@@ -179,6 +185,11 @@ func TestGetLocalRateLimitFromHttpConnectionManager(t *testing.T) {
 										NumberValue: 100,
 									},
 								},
+								"tokens_per_fill": {
+									Kind: &structpb.Value_NumberValue{
+										NumberValue: 100,
+									},
+								},
 							},
 						},
 					},
@@ -201,7 +212,8 @@ func TestGetLocalRateLimitFromHttpConnectionManager(t *testing.T) {
 			},
 		},
 	}
-	token, err = getLocalRateLimitFromHttpConnectionManager(hcm1)
+	token, tokensPerfill, err = getLocalRateLimitFromHttpConnectionManager(hcm1)
 	assert.Equal(t, err, nil)
 	assert.Equal(t, token, uint32(100))
+	assert.Equal(t, tokensPerfill, uint32(100))
 }

core/xdsresource/rds.go

@@ -34,6 +34,7 @@ type RouteConfigResource struct {
 	HTTPRouteConfig   *HTTPRouteConfig
 	ThriftRouteConfig *ThriftRouteConfig
 	MaxTokens         uint32
+	TokensPerFill     uint32
 }
 
 type HTTPRouteConfig struct {

core/xdsresource/testutil.go

@@ -107,6 +107,9 @@ var (
 	rateLimit = &ratelimitv3.LocalRateLimit{
 		TokenBucket: &typedv3.TokenBucket{
 			MaxTokens: 10,
+			TokensPerFill: &wrappers.UInt32Value{
+				Value: 101,
+			},
 		},
 	}
 	// Rds

xdssuite/circuitbreak_test.go

@@ -25,7 +25,7 @@ import (
 	"github.com/kitex-contrib/xds/core/xdsresource"
 )
 
-func cbConifg(conf interface{}) interface{} {
+func cbConfig(conf interface{}) interface{} {
 	m := conf.(map[string]interface{})
 	m = m["cb_config"].(map[string]interface{})
 	return m["service"]
@@ -48,7 +48,7 @@ func TestCircuitBreaker(t *testing.T) {
 			},
 		},
 	})
-	assert.Equal(t, cbConifg(cb.cb.Dump()), map[string]interface{}{
+	assert.Equal(t, cbConfig(cb.cb.Dump()), map[string]interface{}{
 		"c1": circuitbreak.CBConfig{
 			Enable:    true,
 			MinSample: 100,
@@ -64,7 +64,7 @@ func TestCircuitBreaker(t *testing.T) {
 			},
 		},
 	})
-	assert.Equal(t, cbConifg(cb.cb.Dump()), map[string]interface{}{
+	assert.Equal(t, cbConfig(cb.cb.Dump()), map[string]interface{}{
 		"c1": circuitbreak.CBConfig{
 			Enable: false,
 		},
@@ -81,7 +81,7 @@ func TestCircuitBreaker(t *testing.T) {
 			},
 		},
 	})
-	assert.Equal(t, cbConifg(cb.cb.Dump()), map[string]interface{}{
+	assert.Equal(t, cbConfig(cb.cb.Dump()), map[string]interface{}{
 		"c1": circuitbreak.CBConfig{
 			Enable: false,
 		},
@@ -98,7 +98,7 @@ func TestCircuitBreaker(t *testing.T) {
 			},
 		},
 	})
-	assert.Equal(t, cbConifg(cb.cb.Dump()), map[string]interface{}{
+	assert.Equal(t, cbConfig(cb.cb.Dump()), map[string]interface{}{
 		"c1": circuitbreak.CBConfig{
 			Enable: false,
 		},

xdssuite/limiter.go

@@ -52,13 +52,13 @@ func getLimiterPolicy(up map[string]xdsresource.Resource) map[uint32]uint32 {
 	if lds == nil {
 		return nil
 	}
-	maxTokens := make(map[uint32]uint32)
+	tpfs := make(map[uint32]uint32)
 	for _, lis := range lds.NetworkFilters {
 		if lis.InlineRouteConfig != nil {
-			maxTokens[lis.RoutePort] = lis.InlineRouteConfig.MaxTokens
+			tpfs[lis.RoutePort] = lis.InlineRouteConfig.TokensPerFill
 		}
 	}
-	return maxTokens
+	return tpfs
 }
 
 type limiter struct {

xdssuite/limiter_test.go

@@ -43,7 +43,7 @@ func TestLimiter(t *testing.T) {
 				{
 					RoutePort: 80,
 					InlineRouteConfig: &xdsresource.RouteConfigResource{
-						MaxTokens: 100,
+						TokensPerFill: 100,
 					},
 				},
 			},
@@ -58,7 +58,7 @@ func TestLimiter(t *testing.T) {
 				{
 					RoutePort: 80,
 					InlineRouteConfig: &xdsresource.RouteConfigResource{
-						MaxTokens: 100,
+						TokensPerFill: 100,
 					},
 				},
 			},
@@ -73,7 +73,7 @@ func TestLimiter(t *testing.T) {
 				{
 					RoutePort: 80,
 					InlineRouteConfig: &xdsresource.RouteConfigResource{
-						MaxTokens: 1000,
+						TokensPerFill: 1000,
 					},
 				},
 			},
@@ -88,7 +88,7 @@ func TestLimiter(t *testing.T) {
 				{
 					RoutePort: 0,
 					InlineRouteConfig: &xdsresource.RouteConfigResource{
-						MaxTokens: 999,
+						TokensPerFill: 999,
 					},
 				},
 			},
@@ -103,7 +103,7 @@ func TestLimiter(t *testing.T) {
 				{
 					RoutePort: 8080,
 					InlineRouteConfig: &xdsresource.RouteConfigResource{
-						MaxTokens: 999,
+						TokensPerFill: 999,
 					},
 				},
 			},