Merge pull request #3580 from Thushani-Jayasekera/rateLimitingWS

Add circuit breaking configs for websocket rate limiting

adapter/config/default_config.go

@@ -139,6 +139,24 @@ var defaultConfig = &Config{
 					},
 				},
 			},
+			CircuitBreakers: []upstreamCircuitBreaker{
+				{
+					Organizations:      "*",
+					CircuitBreakerName: "BasicCircuitBreaker",
+					MaxConnections:     3,
+					MaxRequests:        3,
+					MaxPendingRequests: 1,
+					MaxRetries:         3,
+				},
+				{
+					Organizations:      "*",
+					CircuitBreakerName: "EnhancedCircuitBreaker",
+					MaxConnections:     50,
+					MaxRequests:        50,
+					MaxPendingRequests: 1,
+					MaxRetries:         50,
+				},
+			},
 		},
 		Connection: connection{
 			Timeouts: connectionTimeouts{

adapter/config/types.go

@@ -235,11 +235,12 @@ type globalCors struct {
 // Envoy Upstream Related Configurations
 type envoyUpstream struct {
 	// UpstreamTLS related Configuration
-	TLS      upstreamTLS
-	Timeouts upstreamTimeout
-	Health   upstreamHealth
-	DNS      upstreamDNS
-	Retry    upstreamRetry
+	TLS             upstreamTLS
+	Timeouts        upstreamTimeout
+	Health          upstreamHealth
+	DNS             upstreamDNS
+	Retry           upstreamRetry
+	CircuitBreakers []upstreamCircuitBreaker
 }
 
 type upstreamTLS struct {
@@ -314,6 +315,16 @@ type upstreamRetry struct {
 	StatusCodes   []uint32
 }
 
+type upstreamCircuitBreaker struct {
+	Organizations      string
+	CircuitBreakerName string
+	MaxConnections     uint32
+	MaxPendingRequests uint32
+	MaxRequests        uint32
+	MaxConnectionPools uint32
+	MaxRetries         uint32
+}
+
 type security struct {
 	TokenService []tokenService
 	AuthHeader   authHeader

adapter/internal/discovery/xds/server.go

@@ -362,14 +362,14 @@ func UpdateAPI(vHost string, apiProject mgw.ProjectAPI, deployedEnvironments []*
 	apiHashValue := generateHashValue(apiYaml.Name, apiYaml.Version)
 
 	if mgwSwagger.GetProdEndpoints() != nil {
-		mgwSwagger.GetProdEndpoints().SetEndpointsConfig(apiYaml.EndpointConfig.ProductionEndpoints)
+		mgwSwagger.GetProdEndpoints().SetEndpointsConfig(apiYaml.EndpointConfig.ProductionEndpoints, apiYaml.EndpointConfig.EndpointType, apiYaml.OrganizationID)
 		if !mgwSwagger.GetProdEndpoints().SecurityConfig.Enabled && apiYaml.EndpointConfig.APIEndpointSecurity.Production.Enabled {
 			mgwSwagger.GetProdEndpoints().SecurityConfig = apiYaml.EndpointConfig.APIEndpointSecurity.Production
 		}
 	}
 
 	if mgwSwagger.GetSandEndpoints() != nil {
-		mgwSwagger.GetSandEndpoints().SetEndpointsConfig(apiYaml.EndpointConfig.SandBoxEndpoints)
+		mgwSwagger.GetSandEndpoints().SetEndpointsConfig(apiYaml.EndpointConfig.SandBoxEndpoints, apiYaml.EndpointConfig.EndpointType, apiYaml.OrganizationID)
 		if !mgwSwagger.GetSandEndpoints().SecurityConfig.Enabled && apiYaml.EndpointConfig.APIEndpointSecurity.Sandbox.Enabled {
 			mgwSwagger.GetSandEndpoints().SecurityConfig = apiYaml.EndpointConfig.APIEndpointSecurity.Sandbox
 		}

adapter/internal/oasparser/model/mgw_swagger.go

@@ -169,6 +169,9 @@ type InterceptEndpoint struct {
 
 const prototypedAPI = "prototyped"
 
+// BasicCircuitBreaker is the name for free tier cluster level circuit breaker
+const BasicCircuitBreaker = "BasicCircuitBreaker"
+
 // GetCorsConfig returns the CorsConfiguration Object.
 func (swagger *MgwSwagger) GetCorsConfig() *CorsConfig {
 	return swagger.xWso2Cors
@@ -631,7 +634,7 @@ func (swagger *MgwSwagger) setXWso2Endpoints() error {
 }
 
 // SetEndpointsConfig set configs for Endpoints sent by api.yaml
-func (endpointCluster *EndpointCluster) SetEndpointsConfig(endpointInfos []EndpointInfo) error {
+func (endpointCluster *EndpointCluster) SetEndpointsConfig(endpointInfos []EndpointInfo, apiType string, orgID string) error {
 	if endpointInfos == nil || len(endpointInfos) == 0 {
 		return nil
 	}
@@ -666,9 +669,57 @@ func (endpointCluster *EndpointCluster) SetEndpointsConfig(endpointInfos []Endpo
 			endpointCluster.Config.RetryConfig = retryConfig
 		}
 	}
+
+	if endpointCluster.Config.CircuitBreakers == nil && apiType == "WS" {
+		logger.LoggerOasparser.Debug("Adding CircuitBreakers for the endpoint cluster", endpointInfos[0].Endpoint)
+		conf, _ := config.ReadConfigs()
+		var selectedCircuitBreaker *CircuitBreakers
+
+		for _, circuitBreaker := range conf.Envoy.Upstream.CircuitBreakers {
+			if utills.GetIsOrganizationInList(orgID, circuitBreaker.Organizations) {
+				selectedCircuitBreaker = createCircuitBreaker(
+					circuitBreaker.MaxConnections,
+					circuitBreaker.MaxPendingRequests,
+					circuitBreaker.MaxRequests,
+					circuitBreaker.MaxRetries,
+					circuitBreaker.MaxConnectionPools,
+				)
+				break
+			}
+		}
+		if selectedCircuitBreaker == nil {
+			for _, circuitBreaker := range conf.Envoy.Upstream.CircuitBreakers {
+				// breaks from the first iteration
+				if circuitBreaker.CircuitBreakerName == BasicCircuitBreaker {
+					selectedCircuitBreaker = createCircuitBreaker(
+						circuitBreaker.MaxConnections,
+						circuitBreaker.MaxPendingRequests,
+						circuitBreaker.MaxRequests,
+						circuitBreaker.MaxRetries,
+						circuitBreaker.MaxConnectionPools,
+					)
+					break
+				}
+
+			}
+		}
+		endpointCluster.Config.CircuitBreakers = selectedCircuitBreaker
+	}
 	return nil
 }
 
+// Helper function to create a CircuitBreakers instance
+func createCircuitBreaker(maxConnections uint32, maxPendingRequests uint32,
+	maxRequests uint32, maxRetries uint32, maxConnectionPools uint32) *CircuitBreakers {
+	return &CircuitBreakers{
+		MaxConnections:     int32(maxConnections),
+		MaxConnectionPools: int32(maxConnectionPools),
+		MaxPendingRequests: int32(maxPendingRequests),
+		MaxRequests:        int32(maxRequests),
+		MaxRetries:         int32(maxRetries),
+	}
+}
+
 func (swagger *MgwSwagger) setXWso2ThrottlingTier() {
 	tier := ResolveThrottlingTier(swagger.vendorExtensions)
 	if tier != "" {

adapter/internal/oasparser/utills/mgw_swagger_utils.go

@@ -0,0 +1,36 @@
+/*
+ *  Copyright (c) 2024, WSO2 LLC. (http://www.wso2.org) All Rights Reserved.
+ *
+ *  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 utills holds the implementation for common utility functions
+package utills
+
+import (
+	"strings"
+)
+
+// GetIsOrganizationInList checks whether the orgID is in the cbOrganization list
+func GetIsOrganizationInList(orgID string, cbOrganizationList string) bool {
+	cbOrganizationsList := strings.Split(cbOrganizationList, ",")
+
+	for _, org := range cbOrganizationsList {
+		if org == orgID {
+			return true
+		}
+	}
+
+	return false
+}

resources/conf/config.toml.template

@@ -201,6 +201,24 @@ retainKeys = ["self_validate_jwt", "issuer", "claim_mappings", "consumer_key_cla
   # If the connection is an HTTP/2 downstream connection a drain sequence will occur prior to closing the connection
   idleTimeoutInSeconds = 3600
 
+# Envoy configs relevant to circuit breaking. We can define multiple circuitBreakers for different organizations
+# This should be the 1st circuit breaker
+[[router.upstream.circuitBreakers]]
+organizations = "*"
+circuitBreakerName = "BasicCircuitBreaker"
+maxConnections = 3
+maxRequests = 3
+maxPendingRequests = 0
+maxConnectionPools = 3
+
+[[router.upstream.circuitBreakers]]
+organizations = "e0682456-2ba6-4c5f-8f36-3c5b6dc46913,4b9afefb-4bcc-4e63-85d3-ddd593841012,d3a7dfea-fb10-4371-b21d-85d1bc28667b"
+circuitBreakerName = "EnhancedCircuitBreaker"
+maxConnections = 50
+maxRequests = 50
+maxPendingRequests = 0
+maxConnectionPools = 50
+
 # Configs relevant to the envoy rate-limit service
 [router.ratelimit]
   # Enable/disable envoy rate-limit service