rate-limit: implement connection level local rate limiting (#4823)

Adds support for rate limiting L4 TCP connections.

Part of #2018

Signed-off-by: Shashank Ram <[email protected]>

Author: shashankram

pkg/envoy/lds/inmesh.go

@@ -3,16 +3,22 @@ package lds
 import (
 	"fmt"
 	"strings"
+	"time"
 
 	xds_core "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
 	xds_listener "github.com/envoyproxy/go-control-plane/envoy/config/listener/v3"
+	xds_local_ratelimit "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/network/local_ratelimit/v3"
 	xds_tcp_proxy "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/network/tcp_proxy/v3"
+	xds_type "github.com/envoyproxy/go-control-plane/envoy/type/v3"
 	"github.com/envoyproxy/go-control-plane/pkg/wellknown"
 	"github.com/golang/protobuf/ptypes/any"
 	"github.com/pkg/errors"
 	"google.golang.org/protobuf/types/known/anypb"
+	"google.golang.org/protobuf/types/known/durationpb"
 	"google.golang.org/protobuf/types/known/wrapperspb"
 
+	policyv1alpha1 "github.com/openservicemesh/osm/pkg/apis/policy/v1alpha1"
+
 	"github.com/openservicemesh/osm/pkg/constants"
 	"github.com/openservicemesh/osm/pkg/envoy"
 	"github.com/openservicemesh/osm/pkg/envoy/rds/route"
@@ -75,6 +81,15 @@ func (lb *listenerBuilder) getInboundHTTPFilters(trafficMatch *trafficpolicy.Tra
 		filters = append(filters, rbacFilter)
 	}
 
+	// Apply the network level local rate limit filter if configured for the TrafficMatch
+	if trafficMatch.RateLimit != nil && trafficMatch.RateLimit.Local != nil && trafficMatch.RateLimit.Local.TCP != nil {
+		rateLimitFilter, err := buildTCPLocalRateLimitFilter(trafficMatch.RateLimit.Local.TCP, trafficMatch.Name)
+		if err != nil {
+			return nil, err
+		}
+		filters = append(filters, rateLimitFilter)
+	}
+
 	// Build the HTTP Connection Manager filter from its options
 	inboundConnManager, err := httpConnManagerOptions{
 		direction:         inbound,
@@ -228,6 +243,15 @@ func (lb *listenerBuilder) getInboundTCPFilters(trafficMatch *trafficpolicy.Traf
 		filters = append(filters, rbacFilter)
 	}
 
+	// Apply the network level local rate limit filter if configured for the TrafficMatch
+	if trafficMatch.RateLimit != nil && trafficMatch.RateLimit.Local != nil && trafficMatch.RateLimit.Local.TCP != nil {
+		rateLimitFilter, err := buildTCPLocalRateLimitFilter(trafficMatch.RateLimit.Local.TCP, trafficMatch.Name)
+		if err != nil {
+			return nil, err
+		}
+		filters = append(filters, rateLimitFilter)
+	}
+
 	// Apply the TCP Proxy Filter
 	tcpProxy := &xds_tcp_proxy.TcpProxy{
 		StatPrefix:       fmt.Sprintf("%s.%s", inboundMeshTCPProxyStatPrefix, trafficMatch.Cluster),
@@ -248,6 +272,45 @@ func (lb *listenerBuilder) getInboundTCPFilters(trafficMatch *trafficpolicy.Traf
 	return filters, nil
 }
 
+func buildTCPLocalRateLimitFilter(config *policyv1alpha1.TCPLocalRateLimitSpec, statPrefix string) (*xds_listener.Filter, error) {
+	if config == nil {
+		return nil, nil
+	}
+
+	var fillInterval time.Duration
+	switch config.Unit {
+	case "second":
+		fillInterval = time.Second
+	case "minute":
+		fillInterval = time.Minute
+	case "hour":
+		fillInterval = time.Hour
+	default:
+		return nil, errors.Errorf("invalid unit %q for TCP connection rate limiting", config.Unit)
+	}
+
+	rateLimit := &xds_local_ratelimit.LocalRateLimit{
+		StatPrefix: statPrefix,
+		TokenBucket: &xds_type.TokenBucket{
+			MaxTokens:     config.Connections + config.Burst,
+			TokensPerFill: wrapperspb.UInt32(config.Connections),
+			FillInterval:  durationpb.New(fillInterval),
+		},
+	}
+
+	marshalledConfig, err := anypb.New(rateLimit)
+	if err != nil {
+		return nil, err
+	}
+
+	filter := &xds_listener.Filter{
+		Name:       wellknown.RateLimit,
+		ConfigType: &xds_listener.Filter_TypedConfig{TypedConfig: marshalledConfig},
+	}
+
+	return filter, nil
+}
+
 // getOutboundHTTPFilter returns an HTTP connection manager network filter used to filter outbound HTTP traffic for the given route configuration
 func (lb *listenerBuilder) getOutboundHTTPFilter(routeConfigName string) (*xds_listener.Filter, error) {
 	var marshalledFilter *any.Any

pkg/envoy/lds/inmesh_test.go

@@ -13,6 +13,7 @@ import (
 	"google.golang.org/protobuf/types/known/wrapperspb"
 
 	configv1alpha2 "github.com/openservicemesh/osm/pkg/apis/config/v1alpha2"
+	policyv1alpha1 "github.com/openservicemesh/osm/pkg/apis/policy/v1alpha1"
 
 	"github.com/openservicemesh/osm/pkg/auth"
 	"github.com/openservicemesh/osm/pkg/catalog"
@@ -271,6 +272,32 @@ func TestGetInboundMeshHTTPFilterChain(t *testing.T) {
 			expectedFilterNames: []string{wellknown.HTTPConnectionManager},
 			expectError:         false,
 		},
+		{
+			name:           "inbound HTTP filter chain with rate limiting enabled",
+			permissiveMode: true,
+			trafficMatch: &trafficpolicy.TrafficMatch{
+				Name:                "inbound_ns1/svc1_90_http",
+				DestinationPort:     90,
+				DestinationProtocol: "http",
+				ServerNames:         []string{"svc1.ns1.svc.cluster.local"},
+				RateLimit: &policyv1alpha1.RateLimitSpec{
+					Local: &policyv1alpha1.LocalRateLimitSpec{
+						TCP: &policyv1alpha1.TCPLocalRateLimitSpec{
+							Connections: 100,
+							Unit:        "minute",
+						},
+					},
+				},
+			},
+			expectedFilterChainMatch: &xds_listener.FilterChainMatch{
+				DestinationPort:      &wrapperspb.UInt32Value{Value: 90},
+				ServerNames:          []string{"svc1.ns1.svc.cluster.local"},
+				TransportProtocol:    "tls",
+				ApplicationProtocols: []string{"osm"},
+			},
+			expectedFilterNames: []string{wellknown.RateLimit, wellknown.HTTPConnectionManager},
+			expectError:         false,
+		},
 	}
 
 	trafficTargets := []trafficpolicy.TrafficTargetWithRoutes{
@@ -358,7 +385,6 @@ func TestGetInboundMeshTCPFilterChain(t *testing.T) {
 			expectedFilterNames: []string{wellknown.RoleBasedAccessControl, wellknown.TCPProxy},
 			expectError:         false,
 		},
-
 		{
 			name:           "inbound TCP filter chain with permissive mode enabled",
 			permissiveMode: true,
@@ -377,6 +403,32 @@ func TestGetInboundMeshTCPFilterChain(t *testing.T) {
 			expectedFilterNames: []string{wellknown.TCPProxy},
 			expectError:         false,
 		},
+		{
+			name:           "inbound TCP filter chain with local TCP rate limiting enabled",
+			permissiveMode: true,
+			trafficMatch: &trafficpolicy.TrafficMatch{
+				Name:                "inbound_ns1/svc1_90_http",
+				DestinationPort:     90,
+				DestinationProtocol: "tcp",
+				ServerNames:         []string{"svc1.ns1.svc.cluster.local"},
+				RateLimit: &policyv1alpha1.RateLimitSpec{
+					Local: &policyv1alpha1.LocalRateLimitSpec{
+						TCP: &policyv1alpha1.TCPLocalRateLimitSpec{
+							Connections: 100,
+							Unit:        "minute",
+						},
+					},
+				},
+			},
+			expectedFilterChainMatch: &xds_listener.FilterChainMatch{
+				DestinationPort:      &wrapperspb.UInt32Value{Value: 90},
+				ServerNames:          []string{"svc1.ns1.svc.cluster.local"},
+				TransportProtocol:    "tls",
+				ApplicationProtocols: []string{"osm"},
+			},
+			expectedFilterNames: []string{wellknown.RateLimit, wellknown.TCPProxy},
+			expectError:         false,
+		},
 	}
 
 	trafficTargets := []trafficpolicy.TrafficTargetWithRoutes{