support timeout and ratelimit (#637)

pkg/ingress/kube/annotations/annotations.go

@@ -56,10 +56,14 @@ type Ingress struct {
 
 	IPAccessControl *IPAccessControlConfig
 
+	Timeout *TimeoutConfig
+
 	Retry *RetryConfig
 
 	LoadBalance *LoadBalanceConfig
 
+	localRateLimit *localRateLimitConfig
+
 	Fallback *FallbackConfig
 
 	Auth *AuthConfig

pkg/ingress/kube/annotations/local_rate_limit.go

@@ -0,0 +1,110 @@
+// Copyright (c) 2022 Alibaba Group Holding Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package annotations
+
+import (
+	types "github.com/gogo/protobuf/types"
+
+	networking "istio.io/api/networking/v1alpha3"
+	"istio.io/istio/pilot/pkg/networking/core/v1alpha3/mseingress"
+)
+
+const (
+	limitRPM             = "route-limit-rpm"
+	limitRPS             = "route-limit-rps"
+	limitBurstMultiplier = "route-limit-burst-multiplier"
+
+	defaultBurstMultiplier = 5
+	defaultStatusCode      = 429
+)
+
+var (
+	_ Parser       = localRateLimit{}
+	_ RouteHandler = localRateLimit{}
+
+	second = &types.Duration{
+		Seconds: 1,
+	}
+
+	minute = &types.Duration{
+		Seconds: 60,
+	}
+)
+
+type localRateLimitConfig struct {
+	TokensPerFill uint32
+	MaxTokens     uint32
+	FillInterval  *types.Duration
+}
+
+type localRateLimit struct{}
+
+func (l localRateLimit) Parse(annotations Annotations, config *Ingress, _ *GlobalContext) error {
+	if !needLocalRateLimitConfig(annotations) {
+		return nil
+	}
+
+	var local *localRateLimitConfig
+	defer func() {
+		config.localRateLimit = local
+	}()
+
+	multiplier := defaultBurstMultiplier
+	if m, err := annotations.ParseIntForHigress(limitBurstMultiplier); err == nil {
+		multiplier = m
+	}
+
+	if rpm, err := annotations.ParseIntForHigress(limitRPM); err == nil {
+		local = &localRateLimitConfig{
+			MaxTokens:     uint32(rpm * multiplier),
+			TokensPerFill: uint32(rpm),
+			FillInterval:  minute,
+		}
+	} else if rps, err := annotations.ParseIntForHigress(limitRPS); err == nil {
+		local = &localRateLimitConfig{
+			MaxTokens:     uint32(rps * multiplier),
+			TokensPerFill: uint32(rps),
+			FillInterval:  second,
+		}
+	}
+
+	return nil
+}
+
+func (l localRateLimit) ApplyRoute(route *networking.HTTPRoute, config *Ingress) {
+	localRateLimitConfig := config.localRateLimit
+	if localRateLimitConfig == nil {
+		return
+	}
+
+	route.RouteHTTPFilters = append(route.RouteHTTPFilters, &networking.HTTPFilter{
+		Name: mseingress.LocalRateLimit,
+		Filter: &networking.HTTPFilter_LocalRateLimit{
+			LocalRateLimit: &networking.LocalRateLimit{
+				TokenBucket: &networking.TokenBucket{
+					MaxTokens:     localRateLimitConfig.MaxTokens,
+					TokensPefFill: localRateLimitConfig.TokensPerFill,
+					FillInterval:  localRateLimitConfig.FillInterval,
+				},
+				StatusCode: defaultStatusCode,
+			},
+		},
+	})
+}
+
+func needLocalRateLimitConfig(annotations Annotations) bool {
+	return annotations.HasHigress(limitRPM) ||
+		annotations.HasHigress(limitRPS)
+}

pkg/ingress/kube/annotations/local_rate_limit_test.go

@@ -0,0 +1,127 @@
+// Copyright (c) 2022 Alibaba Group Holding Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package annotations
+
+import (
+	"reflect"
+	"testing"
+
+	networking "istio.io/api/networking/v1alpha3"
+	"istio.io/istio/pilot/pkg/networking/core/v1alpha3/mseingress"
+)
+
+func TestLocalRateLimitParse(t *testing.T) {
+	localRateLimit := localRateLimit{}
+	inputCases := []struct {
+		input  map[string]string
+		expect *localRateLimitConfig
+	}{
+		{},
+		{
+			input: map[string]string{
+				buildHigressAnnotationKey(limitRPM): "2",
+			},
+			expect: &localRateLimitConfig{
+				MaxTokens:     10,
+				TokensPerFill: 2,
+				FillInterval:  minute,
+			},
+		},
+		{
+			input: map[string]string{
+				buildHigressAnnotationKey(limitRPM):             "2",
+				buildHigressAnnotationKey(limitRPS):             "3",
+				buildHigressAnnotationKey(limitBurstMultiplier): "10",
+			},
+			expect: &localRateLimitConfig{
+				MaxTokens:     20,
+				TokensPerFill: 2,
+				FillInterval:  minute,
+			},
+		},
+		{
+			input: map[string]string{
+				buildHigressAnnotationKey(limitRPS):             "3",
+				buildHigressAnnotationKey(limitBurstMultiplier): "10",
+			},
+			expect: &localRateLimitConfig{
+				MaxTokens:     30,
+				TokensPerFill: 3,
+				FillInterval:  second,
+			},
+		},
+	}
+
+	for _, inputCase := range inputCases {
+		t.Run("", func(t *testing.T) {
+			config := &Ingress{}
+			_ = localRateLimit.Parse(inputCase.input, config, nil)
+			if !reflect.DeepEqual(inputCase.expect, config.localRateLimit) {
+				t.Fatal("Should be equal")
+			}
+		})
+	}
+}
+
+func TestLocalRateLimitApplyRoute(t *testing.T) {
+	localRateLimit := localRateLimit{}
+	inputCases := []struct {
+		config *Ingress
+		input  *networking.HTTPRoute
+		expect *networking.HTTPRoute
+	}{
+		{
+			config: &Ingress{},
+			input:  &networking.HTTPRoute{},
+			expect: &networking.HTTPRoute{},
+		},
+		{
+			config: &Ingress{
+				localRateLimit: &localRateLimitConfig{
+					MaxTokens:     60,
+					TokensPerFill: 20,
+					FillInterval:  second,
+				},
+			},
+			input: &networking.HTTPRoute{},
+			expect: &networking.HTTPRoute{
+				RouteHTTPFilters: []*networking.HTTPFilter{
+					{
+						Name: mseingress.LocalRateLimit,
+						Filter: &networking.HTTPFilter_LocalRateLimit{
+							LocalRateLimit: &networking.LocalRateLimit{
+								TokenBucket: &networking.TokenBucket{
+									MaxTokens:     60,
+									TokensPefFill: 20,
+									FillInterval:  second,
+								},
+								StatusCode: defaultStatusCode,
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+
+	for _, inputCase := range inputCases {
+		t.Run("", func(t *testing.T) {
+			localRateLimit.ApplyRoute(inputCase.input, inputCase.config)
+			if !reflect.DeepEqual(inputCase.input, inputCase.expect) {
+				t.Fatal("Should be equal")
+			}
+		})
+	}
+}

pkg/ingress/kube/annotations/timeout.go

@@ -0,0 +1,62 @@
+// Copyright (c) 2022 Alibaba Group Holding Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//	http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package annotations
+
+import (
+	types "github.com/gogo/protobuf/types"
+
+	networking "istio.io/api/networking/v1alpha3"
+)
+
+const timeoutAnnotation = "timeout"
+
+var (
+	_ Parser       = timeout{}
+	_ RouteHandler = timeout{}
+)
+
+type TimeoutConfig struct {
+	time *types.Duration
+}
+
+type timeout struct{}
+
+func (t timeout) Parse(annotations Annotations, config *Ingress, _ *GlobalContext) error {
+	if !needTimeoutConfig(annotations) {
+		return nil
+	}
+
+	if time, err := annotations.ParseIntForHigress(timeoutAnnotation); err == nil {
+		config.Timeout = &TimeoutConfig{
+			time: &types.Duration{
+				Seconds: int64(time),
+			},
+		}
+	}
+	return nil
+}
+
+func (t timeout) ApplyRoute(route *networking.HTTPRoute, config *Ingress) {
+	timeout := config.Timeout
+	if timeout == nil || timeout.time == nil || timeout.time.Seconds == 0 {
+		return
+	}
+
+	route.Timeout = timeout.time
+}
+
+func needTimeoutConfig(annotations Annotations) bool {
+	return annotations.HasHigress(timeoutAnnotation)
+}

pkg/ingress/kube/annotations/timeout_test.go

@@ -0,0 +1,122 @@
+// Copyright (c) 2022 Alibaba Group Holding Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package annotations
+
+import (
+	"reflect"
+	"testing"
+
+	types "github.com/gogo/protobuf/types"
+
+	networking "istio.io/api/networking/v1alpha3"
+)
+
+func TestTimeoutParse(t *testing.T) {
+	timeout := timeout{}
+	inputCases := []struct {
+		input  map[string]string
+		expect *TimeoutConfig
+	}{
+		{},
+		{
+			input: map[string]string{
+				HigressAnnotationsPrefix + "/" + timeoutAnnotation: "",
+			},
+		},
+		{
+			input: map[string]string{
+				HigressAnnotationsPrefix + "/" + timeoutAnnotation: "0",
+			},
+			expect: &TimeoutConfig{
+				time: &types.Duration{},
+			},
+		},
+		{
+			input: map[string]string{
+				HigressAnnotationsPrefix + "/" + timeoutAnnotation: "10",
+			},
+			expect: &TimeoutConfig{
+				time: &types.Duration{
+					Seconds: 10,
+				},
+			},
+		},
+	}
+
+	for _, c := range inputCases {
+		t.Run("", func(t *testing.T) {
+			config := &Ingress{}
+			_ = timeout.Parse(c.input, config, nil)
+			if !reflect.DeepEqual(c.expect, config.Timeout) {
+				t.Fatalf("Should be equal.")
+			}
+		})
+	}
+}
+
+func TestTimeoutApplyRoute(t *testing.T) {
+	timeout := timeout{}
+	inputCases := []struct {
+		config *Ingress
+		input  *networking.HTTPRoute
+		expect *networking.HTTPRoute
+	}{
+		{
+			config: &Ingress{},
+			input:  &networking.HTTPRoute{},
+			expect: &networking.HTTPRoute{},
+		},
+		{
+			config: &Ingress{
+				Timeout: &TimeoutConfig{},
+			},
+			input:  &networking.HTTPRoute{},
+			expect: &networking.HTTPRoute{},
+		},
+		{
+			config: &Ingress{
+				Timeout: &TimeoutConfig{
+					time: &types.Duration{},
+				},
+			},
+			input:  &networking.HTTPRoute{},
+			expect: &networking.HTTPRoute{},
+		},
+		{
+			config: &Ingress{
+				Timeout: &TimeoutConfig{
+					time: &types.Duration{
+						Seconds: 10,
+					},
+				},
+			},
+			input: &networking.HTTPRoute{},
+			expect: &networking.HTTPRoute{
+				Timeout: &types.Duration{
+					Seconds: 10,
+				},
+			},
+		},
+	}
+
+	for _, inputCase := range inputCases {
+		t.Run("", func(t *testing.T) {
+			timeout.ApplyRoute(inputCase.input, inputCase.config)
+			if !reflect.DeepEqual(inputCase.input, inputCase.expect) {
+				t.Fatalf("Should be equal")
+			}
+		})
+	}
+}