Merge pull request #1118 from zjj2wry/limit-rate

feat(#733)Support nginx bandwidth control

controllers/nginx/pkg/config/config.go

@@ -387,6 +387,8 @@ func NewDefault() Configuration {
 			CustomHTTPErrors:     []int{},
 			WhitelistSourceRange: []string{},
 			SkipAccessLogURLs:    []string{},
+			LimitRate:            0,
+			LimitRateAfter:       0,
 		},
 		UpstreamKeepaliveConnections: 0,
 		LimitConnZoneVariable:        defaultLimitConnZoneVariable,

controllers/nginx/pkg/template/template.go

@@ -404,6 +404,18 @@ func buildRateLimit(input interface{}) []string {
 		limits = append(limits, limit)
 	}
 
+	if loc.RateLimit.LimitRateAfter > 0 {
+		limit := fmt.Sprintf("limit_rate_after %vk;",
+			loc.RateLimit.LimitRateAfter)
+		limits = append(limits, limit)
+	}
+
+	if loc.RateLimit.LimitRate > 0 {
+		limit := fmt.Sprintf("limit_rate %vk;",
+			loc.RateLimit.LimitRate)
+		limits = append(limits, limit)
+	}
+
 	return limits
 }
 

core/pkg/ingress/annotations/ratelimit/main.go

@@ -22,12 +22,15 @@ import (
 	extensions "k8s.io/api/extensions/v1beta1"
 
 	"k8s.io/ingress/core/pkg/ingress/annotations/parser"
+	"k8s.io/ingress/core/pkg/ingress/resolver"
 )
 
 const (
-	limitIP  = "ingress.kubernetes.io/limit-connections"
-	limitRPS = "ingress.kubernetes.io/limit-rps"
-	limitRPM = "ingress.kubernetes.io/limit-rpm"
+	limitIP        = "ingress.kubernetes.io/limit-connections"
+	limitRPS       = "ingress.kubernetes.io/limit-rps"
+	limitRPM       = "ingress.kubernetes.io/limit-rpm"
+	limitRATE      = "ingress.kubernetes.io/limit-rate"
+	limitRATEAFTER = "ingress.kubernetes.io/limit-rate-after"
 
 	// allow 5 times the specified limit as burst
 	defBurst = 5
@@ -48,6 +51,10 @@ type RateLimit struct {
 	RPS Zone `json:"rps"`
 
 	RPM Zone `json:"rpm"`
+
+	LimitRate int `json:"limit-rate"`
+
+	LimitRateAfter int `json:"limit-rate-after"`
 }
 
 // Equal tests for equality between two RateLimit types
@@ -67,6 +74,12 @@ func (rt1 *RateLimit) Equal(rt2 *RateLimit) bool {
 	if !(&rt1.RPS).Equal(&rt2.RPS) {
 		return false
 	}
+	if rt1.LimitRate != rt2.LimitRate {
+		return false
+	}
+	if rt1.LimitRateAfter != rt2.LimitRateAfter {
+		return false
+	}
 
 	return true
 }
@@ -106,26 +119,38 @@ func (z1 *Zone) Equal(z2 *Zone) bool {
 }
 
 type ratelimit struct {
+	backendResolver resolver.DefaultBackend
 }
 
 // NewParser creates a new ratelimit annotation parser
-func NewParser() parser.IngressAnnotation {
-	return ratelimit{}
+func NewParser(br resolver.DefaultBackend) parser.IngressAnnotation {
+	return ratelimit{br}
 }
 
 // ParseAnnotations parses the annotations contained in the ingress
 // rule used to rewrite the defined paths
 func (a ratelimit) Parse(ing *extensions.Ingress) (interface{}, error) {
+	defBackend := a.backendResolver.GetDefaultBackend()
+	lr, err := parser.GetIntAnnotation(limitRATE, ing)
+	if err != nil {
+		lr = defBackend.LimitRate
+	}
+	lra, err := parser.GetIntAnnotation(limitRATEAFTER, ing)
+	if err != nil {
+		lra = defBackend.LimitRateAfter
+	}
 
 	rpm, _ := parser.GetIntAnnotation(limitRPM, ing)
 	rps, _ := parser.GetIntAnnotation(limitRPS, ing)
 	conn, _ := parser.GetIntAnnotation(limitIP, ing)
 
 	if rpm == 0 && rps == 0 && conn == 0 {
 		return &RateLimit{
-			Connections: Zone{},
-			RPS:         Zone{},
-			RPM:         Zone{},
+			Connections:    Zone{},
+			RPS:            Zone{},
+			RPM:            Zone{},
+			LimitRate:      lr,
+			LimitRateAfter: lra,
 		}, nil
 	}
 
@@ -150,5 +175,7 @@ func (a ratelimit) Parse(ing *extensions.Ingress) (interface{}, error) {
 			Burst:      rpm * defBurst,
 			SharedSize: defSharedSize,
 		},
+		LimitRate:      lr,
+		LimitRateAfter: lra,
 	}, nil
 }

core/pkg/ingress/annotations/ratelimit/main_test.go

@@ -24,6 +24,7 @@ import (
 	meta_v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	"k8s.io/apimachinery/pkg/util/intstr"
+	"k8s.io/ingress/core/pkg/ingress/defaults"
 )
 
 func buildIngress() *extensions.Ingress {
@@ -61,9 +62,19 @@ func buildIngress() *extensions.Ingress {
 	}
 }
 
+type mockBackend struct {
+}
+
+func (m mockBackend) GetDefaultBackend() defaults.Backend {
+	return defaults.Backend{
+		LimitRateAfter: 0,
+		LimitRate:      0,
+	}
+}
+
 func TestWithoutAnnotations(t *testing.T) {
 	ing := buildIngress()
-	_, err := NewParser().Parse(ing)
+	_, err := NewParser(mockBackend{}).Parse(ing)
 	if err != nil {
 		t.Error("unexpected error with ingress without annotations")
 	}
@@ -78,7 +89,7 @@ func TestBadRateLimiting(t *testing.T) {
 	data[limitRPM] = "0"
 	ing.SetAnnotations(data)
 
-	_, err := NewParser().Parse(ing)
+	_, err := NewParser(mockBackend{}).Parse(ing)
 	if err != nil {
 		t.Errorf("unexpected error with invalid limits (0)")
 	}
@@ -87,9 +98,12 @@ func TestBadRateLimiting(t *testing.T) {
 	data[limitIP] = "5"
 	data[limitRPS] = "100"
 	data[limitRPM] = "10"
+	data[limitRATEAFTER] = "100"
+	data[limitRATE] = "10"
+
 	ing.SetAnnotations(data)
 
-	i, err := NewParser().Parse(ing)
+	i, err := NewParser(mockBackend{}).Parse(ing)
 	if err != nil {
 		t.Errorf("unexpected error: %v", err)
 	}
@@ -106,4 +120,10 @@ func TestBadRateLimiting(t *testing.T) {
 	if rateLimit.RPM.Limit != 10 {
 		t.Errorf("expected 10 in limit by rpm but %v was returend", rateLimit.RPM)
 	}
+	if rateLimit.LimitRateAfter != 100 {
+		t.Errorf("expected 100 in limit by limitrateafter but %v was returend", rateLimit.LimitRateAfter)
+	}
+	if rateLimit.LimitRate != 10 {
+		t.Errorf("expected 10 in limit by limitrate but %v was returend", rateLimit.LimitRate)
+	}
 }

core/pkg/ingress/controller/annotations.go

@@ -62,7 +62,7 @@ func newAnnotationExtractor(cfg extractorConfig) annotationExtractor {
 			"Whitelist":            ipwhitelist.NewParser(cfg),
 			"UsePortInRedirects":   portinredirect.NewParser(cfg),
 			"Proxy":                proxy.NewParser(cfg),
-			"RateLimit":            ratelimit.NewParser(),
+			"RateLimit":            ratelimit.NewParser(cfg),
 			"Redirect":             rewrite.NewParser(cfg),
 			"SecureUpstream":       secureupstream.NewParser(cfg),
 			"ServiceUpstream":      serviceupstream.NewParser(),

core/pkg/ingress/defaults/main.go

@@ -88,4 +88,15 @@ type Backend struct {
 	// WhitelistSourceRange allows limiting access to certain client addresses
 	// http://nginx.org/en/docs/http/ngx_http_access_module.html
 	WhitelistSourceRange []string `json:"whitelist-source-range,-"`
+
+	// Limits the rate of response transmission to a client.
+	// The rate is specified in bytes per second. The zero value disables rate limiting.
+	// The limit is set per a request, and so if a client simultaneously opens two connections,
+	// the overall rate will be twice as much as the specified limit.
+	// http://nginx.org/en/docs/http/ngx_http_core_module.html#limit_rate
+	LimitRate int `json:"limit-rate"`
+
+    // Sets the initial amount after which the further transmission of a response to a client will be rate limited.
+	// http://nginx.org/en/docs/http/ngx_http_core_module.html#limit_rate_after
+	LimitRateAfter int `json:"limit-rate-after"`
 }