Merge pull request #2446 from Tharsanan1/aireatelimitter

Add backend&subscription based AI ratelimit

adapter/go.mod

@@ -3,9 +3,9 @@ module github.com/wso2/apk/adapter
 go 1.22
 
 require (
-	github.com/envoyproxy/go-control-plane v0.12.0
+	github.com/envoyproxy/go-control-plane v0.13.0
 	github.com/fsnotify/fsnotify v1.7.0
-	github.com/golang/protobuf v1.5.3
+	github.com/golang/protobuf v1.5.4
 	github.com/google/uuid v1.6.0
 	github.com/onsi/ginkgo/v2 v2.14.0
 	github.com/onsi/gomega v1.30.0
@@ -14,7 +14,7 @@ require (
 	github.com/sirupsen/logrus v1.9.0
 	github.com/wso2/apk/common-go-libs v0.0.0-20231208100153-24bee7b4bd81
 	golang.org/x/exp v0.0.0-20231206192017-f3f8817b8deb
-	google.golang.org/grpc v1.62.0
+	google.golang.org/grpc v1.65.0
 	google.golang.org/protobuf v1.34.1
 	gopkg.in/yaml.v2 v2.4.0
 	k8s.io/api v0.29.2
@@ -24,11 +24,12 @@ require (
 )
 
 require (
+	cel.dev/expr v0.15.0 // indirect
 	github.com/agnivade/levenshtein v1.1.1 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/census-instrumentation/opencensus-proto v0.4.1 // indirect
-	github.com/cespare/xxhash/v2 v2.2.0 // indirect
-	github.com/cncf/xds/go v0.0.0-20231128003011-0fa0005c9caa // indirect
+	github.com/cespare/xxhash/v2 v2.3.0 // indirect
+	github.com/cncf/xds/go v0.0.0-20240423153145-555b57ec207b // indirect
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
 	github.com/emicklei/go-restful/v3 v3.11.0 // indirect
 	github.com/envoyproxy/protoc-gen-validate v1.0.4 // indirect
@@ -56,9 +57,10 @@ require (
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
+	github.com/planetscale/vtprotobuf v0.6.1-0.20240319094008-0393e58bdf10 // indirect
 	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
 	github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c // indirect
-	github.com/prometheus/client_model v0.5.0 // indirect
+	github.com/prometheus/client_model v0.6.0 // indirect
 	github.com/prometheus/common v0.45.0 // indirect
 	github.com/prometheus/procfs v0.12.0 // indirect
 	github.com/shirou/gopsutil/v3 v3.24.2 // indirect
@@ -70,17 +72,15 @@ require (
 	github.com/yusufpapurcu/wmi v1.2.4 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
 	go.uber.org/zap v1.26.0 // indirect
-	golang.org/x/net v0.21.0 // indirect
-	golang.org/x/oauth2 v0.16.0 // indirect
-	golang.org/x/term v0.17.0 // indirect
-	golang.org/x/text v0.14.0 // indirect
+	golang.org/x/net v0.25.0 // indirect
+	golang.org/x/oauth2 v0.20.0 // indirect
+	golang.org/x/term v0.20.0 // indirect
+	golang.org/x/text v0.15.0 // indirect
 	golang.org/x/time v0.3.0 // indirect
 	golang.org/x/tools v0.16.1 // indirect
 	gomodules.xyz/jsonpatch/v2 v2.4.0 // indirect
-	google.golang.org/appengine v1.6.8 // indirect
-	google.golang.org/genproto v0.0.0-20240123012728-ef4313101c80 // indirect
-	google.golang.org/genproto/googleapis/api v0.0.0-20240123012728-ef4313101c80 // indirect
-	google.golang.org/genproto/googleapis/rpc v0.0.0-20240123012728-ef4313101c80 // indirect
+	google.golang.org/genproto/googleapis/api v0.0.0-20240528184218-531527333157 // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20240528184218-531527333157 // indirect
 	gopkg.in/inf.v0 v0.9.1 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 	k8s.io/apiextensions-apiserver v0.29.2 // indirect
@@ -99,8 +99,8 @@ replace github.com/wso2/apk/common-go-libs => ../common-go-libs
 
 require (
 	github.com/ghodss/yaml v1.0.0
-	github.com/stretchr/testify v1.8.4
-	golang.org/x/sys v0.17.0 // indirect
+	github.com/stretchr/testify v1.9.0
+	golang.org/x/sys v0.20.0 // indirect
 	gopkg.in/natefinch/lumberjack.v2 v2.2.1
 	sigs.k8s.io/gateway-api v1.0.0
 )

adapter/internal/discovery/xds/server.go

@@ -97,9 +97,10 @@ var (
 
 	// todo(amali) there can be multiple vhosts for one EnvoyInternalAPI so handle this apiuuid+sand/prod should be the key
 
-	orgAPIMap                map[string]map[string]*EnvoyInternalAPI                      // organizationID -> Vhost:API_UUID -> EnvoyInternalAPI struct map
-	orgIDLatestAPIVersionMap map[string]map[string]map[string]semantic_version.SemVersion // organizationID -> Vhost:APIName -> VersionRange(vx/vx.x; x is int) -> Latest API Version
-
+	orgAPIMap                       map[string]map[string]*EnvoyInternalAPI                      // organizationID -> Vhost:API_UUID -> EnvoyInternalAPI struct map
+	orgIDLatestAPIVersionMap        map[string]map[string]map[string]semantic_version.SemVersion // organizationID -> Vhost:APIName -> VersionRange(vx/vx.x; x is int) -> Latest API Version
+	vHostToSubscriptionBasedAIRLMap map[string]bool
+	vHostToSubscriptionBasedRLMap   map[string]bool
 	// Envoy Label as map key
 	// TODO(amali) use this without generating all again.
 	gatewayLabelConfigMap map[string]*EnvoyGatewayConfig // GW-Label -> EnvoyGatewayConfig struct map
@@ -152,6 +153,8 @@ func init() {
 	gatewayLabelConfigMap = make(map[string]*EnvoyGatewayConfig)
 	orgAPIMap = make(map[string]map[string]*EnvoyInternalAPI)
 	orgIDLatestAPIVersionMap = make(map[string]map[string]map[string]semantic_version.SemVersion)
+	vHostToSubscriptionBasedAIRLMap = make(map[string]bool)
+	vHostToSubscriptionBasedRLMap = make(map[string]bool)
 
 	enforcerLabelMap = make(map[string]*EnforcerInternalAPI)
 	// currently subscriptions, configs, applications, applicationPolicies, subscriptionPolicies,
@@ -330,7 +333,7 @@ func GenerateEnvoyResoucesForGateway(gatewayName string) ([]types.Resource,
 			if found {
 				// Prepare the route config name based on the gateway listener section name.
 				routeConfigName := common.GetEnvoyRouteConfigName(listener.Name, string(listenerSection.Name))
-				routesConfig := oasParser.GetRouteConfigs(map[string][]*routev3.Route{vhost: routes}, routeConfigName, envoyGatewayConfig.customRateLimitPolicies)
+				routesConfig := oasParser.GetRouteConfigs(map[string][]*routev3.Route{vhost: routes}, routeConfigName, envoyGatewayConfig.customRateLimitPolicies, vHostToSubscriptionBasedAIRLMap, vHostToSubscriptionBasedRLMap)
 
 				routeConfigMatched, alreadyExistsInRouteConfigList := routeConfigs[routeConfigName]
 				if alreadyExistsInRouteConfigList {
@@ -353,7 +356,7 @@ func GenerateEnvoyResoucesForGateway(gatewayName string) ([]types.Resource,
 			var vhostToRouteArrayFilteredMapForSystemEndpoints = make(map[string][]*routev3.Route)
 			vhostToRouteArrayFilteredMapForSystemEndpoints[systemHost] = vhostToRouteArrayMap[systemHost]
 			routeConfigName := common.GetEnvoyRouteConfigName(common.GetEnvoyListenerName(string(listener.Protocol), uint32(listener.Port)), string(listener.Name))
-			systemRoutesConfig := oasParser.GetRouteConfigs(vhostToRouteArrayFilteredMapForSystemEndpoints, routeConfigName, envoyGatewayConfig.customRateLimitPolicies)
+			systemRoutesConfig := oasParser.GetRouteConfigs(vhostToRouteArrayFilteredMapForSystemEndpoints, routeConfigName, envoyGatewayConfig.customRateLimitPolicies, vHostToSubscriptionBasedAIRLMap, vHostToSubscriptionBasedRLMap)
 			routeConfigs[routeConfigName] = systemRoutesConfig
 		}
 	}
@@ -560,6 +563,14 @@ func PopulateInternalMaps(adapterInternalAPI *model.AdapterInternalAPI, labels,
 	}
 
 	err := UpdateOrgAPIMap(vHosts, labels, listenerName, sectionName, adapterInternalAPI)
+	for vhost := range vHosts {
+		if adapterInternalAPI.AIProvider.Enabled && adapterInternalAPI.GetSubscriptionValidation() {
+			vHostToSubscriptionBasedAIRLMap[vhost] = true
+		}
+		if adapterInternalAPI.GetSubscriptionValidation() {
+			vHostToSubscriptionBasedRLMap[vhost] = true
+		}
+	}
 	if err != nil {
 		logger.LoggerXds.ErrorC(logging.PrintError(logging.Error1415, logging.MAJOR,
 			"Error updating the API : %s:%s in vhosts: %s, API_UUID: %v. %v",

adapter/internal/oasparser/config_generator.go

@@ -85,8 +85,8 @@ func GetProductionListener(gateway *gwapiv1.Gateway, resolvedListenerCerts map[s
 //
 // The RouteConfiguration is named as "default"
 func GetRouteConfigs(vhostToRouteArrayMap map[string][]*routev3.Route, routeConfigName string,
-	customRateLimitPolicies []*model.CustomRateLimitPolicy) *routev3.RouteConfiguration {
-	vHosts := envoy.CreateVirtualHosts(vhostToRouteArrayMap, customRateLimitPolicies)
+	customRateLimitPolicies []*model.CustomRateLimitPolicy, vhostToSubscriptionAIRL map[string]bool, vhostToSubscriptionRL map[string]bool) *routev3.RouteConfiguration {
+	vHosts := envoy.CreateVirtualHosts(vhostToRouteArrayMap, customRateLimitPolicies, vhostToSubscriptionAIRL, vhostToSubscriptionRL)
 	routeConfig := envoy.CreateRoutesConfigForRds(vHosts, routeConfigName)
 	return routeConfig
 }
@@ -119,12 +119,6 @@ func GetCacheResources(endpoints []*corev3.Address, clusters []*clusterv3.Cluste
 	return listenerRes, clusterRes, routeConfigRes, endpointRes
 }
 
-// UpdateRoutesConfig updates the existing routes configuration with the provided map of vhost to array of routes.
-// All the already existing routes (within the routeConfiguration) will be removed.
-func UpdateRoutesConfig(routeConfig *routev3.RouteConfiguration, vhostToRouteArrayMap map[string][]*routev3.Route) {
-	routeConfig.VirtualHosts = envoy.CreateVirtualHosts(vhostToRouteArrayMap, nil)
-}
-
 // GetEnforcerAPI retrieves the ApiDS object model for a given swagger definition
 // along with the vhost to deploy the API.
 func GetEnforcerAPI(adapterInternalAPI *model.AdapterInternalAPI, vhost string) *api.Api {

adapter/internal/oasparser/envoyconf/constants.go

@@ -29,6 +29,8 @@ const (
 const (
 	httpConManagerStartPrefix  string = "ingress_http"
 	extAuthzPerRouteName       string = "type.googleapis.com/envoy.extensions.filters.http.ext_authz.v3.ExtAuthzPerRoute"
+	extProcPerRouteName        string = "type.googleapis.com/envoy.extensions.filters.http.ext_proc.v3.ExtProcPerRoute"
+	ratelimitPerRouteName          string = "type.googleapis.com/envoy.extensions.filters.http.ratelimit.v3.RateLimitPerRoute"
 	luaPerRouteName            string = "type.googleapis.com/envoy.extensions.filters.http.lua.v3.LuaPerRoute"
 	corsFilterName             string = "type.googleapis.com/envoy.extensions.filters.http.cors.v3.Cors"
 	localRateLimitPerRouteName string = "type.googleapis.com/envoy.extensions.filters.http.local_ratelimit.v3.LocalRateLimit"

adapter/internal/oasparser/envoyconf/http_filters.go

@@ -28,6 +28,7 @@ import (
 	envoy_config_ratelimit_v3 "github.com/envoyproxy/go-control-plane/envoy/config/ratelimit/v3"
 	cors_filter_v3 "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/http/cors/v3"
 	ext_authv3 "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/http/ext_authz/v3"
+	ext_process "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/http/ext_proc/v3"
 	luav3 "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/http/lua/v3"
 	ratelimit "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/http/ratelimit/v3"
 	routerv3 "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/http/router/v3"
@@ -47,9 +48,16 @@ import (
 	"github.com/golang/protobuf/ptypes/any"
 )
 
+
+// HTTPExternalProcessor HTTP filter
+const HTTPExternalProcessor = "envoy.filters.http.ext_proc"
+// RatelimitFilterName Ratelimit filter name
+const RatelimitFilterName = "envoy.filters.http.ratelimit"
+
 // getHTTPFilters generates httpFilter configuration
 func getHTTPFilters(globalLuaScript string) []*hcmv3.HttpFilter {
 	extAuth := getExtAuthzHTTPFilter()
+	extProcessor := getExtProcessHTTPFilter()
 	router := getRouterHTTPFilter()
 	luaLocal := getLuaFilter(LuaLocal, `
 function envoy_on_request(request_handle)
@@ -64,6 +72,7 @@ end`)
 		extAuth,
 		luaLocal,
 		luaGlobal,
+		extProcessor,
 	}
 	conf := config.ReadConfigs()
 	if conf.Envoy.RateLimit.Enabled {
@@ -162,6 +171,10 @@ func getRateLimitFilter() *hcmv3.HttpFilter {
 		Domain:                  RateLimiterDomain,
 		FailureModeDeny:         conf.Envoy.RateLimit.FailureModeDeny,
 		EnableXRatelimitHeaders: enableXRatelimitHeaders,
+		Timeout: &durationpb.Duration{
+			Nanos:   (int32(conf.Envoy.RateLimit.RequestTimeoutInMillis) % 1000) * 1000000,
+			Seconds: conf.Envoy.RateLimit.RequestTimeoutInMillis / 1000,
+		},
 		RateLimitService: &envoy_config_ratelimit_v3.RateLimitServiceConfig{
 			TransportApiVersion: corev3.ApiVersion_V3,
 			GrpcService: &corev3.GrpcService{
@@ -190,6 +203,43 @@ func getRateLimitFilter() *hcmv3.HttpFilter {
 	return &rlFilter
 }
 
+// getExtProcessHTTPFilter gets ExtAauthz http filter.
+func getExtProcessHTTPFilter() *hcmv3.HttpFilter {
+	// conf := config.ReadConfigs()
+	externalProcessor := &ext_process.ExternalProcessor{
+		GrpcService: &corev3.GrpcService{
+			TargetSpecifier: &corev3.GrpcService_EnvoyGrpc_{
+				EnvoyGrpc: &corev3.GrpcService_EnvoyGrpc{
+					ClusterName: extAuthzClusterName,
+				},
+			},
+		},
+		ProcessingMode: &ext_process.ProcessingMode{
+			ResponseBodyMode: ext_process.ProcessingMode_BUFFERED,
+			RequestHeaderMode: ext_process.ProcessingMode_SKIP,
+			ResponseHeaderMode: ext_process.ProcessingMode_SKIP,
+		},
+		MetadataOptions: &ext_process.MetadataOptions{
+			ForwardingNamespaces: &ext_process.MetadataOptions_MetadataNamespaces{
+				Untyped: []string{"envoy.filters.http.ext_authz", "envoy.filters.http.ext_proc"},
+			},
+		},
+		RequestAttributes: []string{"xds.route_metadata"},
+		ResponseAttributes: []string{"xds.route_metadata"},
+	}
+	ext, err2 := anypb.New(externalProcessor)
+	if err2 != nil {
+		logger.LoggerOasparser.Error(err2)
+	}
+	extProcessFilter := hcmv3.HttpFilter{
+		Name: HTTPExternalProcessor,
+		ConfigType: &hcmv3.HttpFilter_TypedConfig{
+			TypedConfig: ext,
+		},
+	}
+	return &extProcessFilter
+}
+
 // getExtAuthzHTTPFilter gets ExtAauthz http filter.
 func getExtAuthzHTTPFilter() *hcmv3.HttpFilter {
 	conf := config.ReadConfigs()

adapter/internal/oasparser/envoyconf/internal_dtos.go

@@ -45,6 +45,7 @@ type routeCreateParams struct {
 	environment                  string
 	envType                      string
 	mirrorClusterNames           map[string][]string
+	isAiAPI                      bool
 }
 
 // RatelimitCriteria criterias of rate limiting