CircutBreaking config support

Ref: 56734b13bc06113b1432a951b5405ce977b61b76

enroute-dp/internal/dag/builder.go

@@ -759,6 +759,7 @@ func (b *builder) processRoutes(ir *gatewayhostv1.GatewayHost, visited []*gatewa
 			}
 
 			b.SetupRouteFilters(r, &route, ir.Namespace)
+			routeServiceFilters := b.RouteServiceFilters(r, &route, ir.Namespace)
 
 			if ir != nil && ir.Spec.VirtualHost != nil && logger.EL.ELogger != nil {
 				logger.EL.ELogger.Debugf("dag:builder:processRoutes() Valid: IR [%s] Walk through Route Services", ir.Spec.VirtualHost.Fqdn)
@@ -845,6 +846,7 @@ func (b *builder) processRoutes(ir *gatewayhostv1.GatewayHost, visited []*gatewa
 					HealthCheck:          service.HealthCheck,
 					UpstreamValidation:   uv,
 					ClientValidation:     cv,
+					ClusterFilters:       routeServiceFilters,
 					SNI:                  s.ExternalName,
 				})
 				if ir != nil && ir.Spec.VirtualHost != nil && logger.EL.ELogger != nil {

enroute-dp/internal/dag/builder_routefilters.go

@@ -4,6 +4,7 @@ import (
 	// "fmt"
 
 	gatewayhostv1 "github.com/saarasio/enroute/enroute-dp/apis/enroute/v1"
+	"github.com/saarasio/enroute/enroute-dp/saarasconfig"
 )
 
 func (b *builder) lookupRouteFilter(m RouteFilterMeta) *RouteFilter {
@@ -46,6 +47,30 @@ func (b *builder) addRouteFilter(rf_k8s *gatewayhostv1.RouteFilter, m RouteFilte
 	return &rf_dag
 }
 
+func (b *builder) RouteServiceFilters(dag_r *Route, k8s_r *gatewayhostv1.Route, ns string) []*RouteFilter {
+
+	var rfilters []*RouteFilter
+
+	if k8s_r != nil && k8s_r.Filters != nil {
+		if len(k8s_r.Filters) > 0 {
+			for _, f := range k8s_r.Filters {
+				if f.Type == saarasconfig.FILTER_TYPE_SERVICE_CIRCUITBREAKERS {
+					m := RouteFilterMeta{filter_type: f.Type, name: f.Name, namespace: ns}
+					rf := b.lookupHTTPRouteFilter(m)
+					if rf != nil && dag_r != nil {
+						if dag_r.RouteFilters == nil {
+							dag_r.RouteFilters = make([]*RouteFilter, 0)
+						}
+						rfilters = append(rfilters, rf)
+					}
+				}
+			}
+		}
+	}
+
+	return rfilters
+}
+
 func (b *builder) SetupRouteFilters(dag_r *Route, k8s_r *gatewayhostv1.Route, ns string) {
 
 	if k8s_r != nil && k8s_r.Filters != nil {

enroute-dp/internal/dag/dag.go

@@ -386,6 +386,8 @@ type Cluster struct {
 
 	// Set cluster SNI
 	SNI string
+
+	ClusterFilters []*RouteFilter
 }
 
 func (c Cluster) Visit(f func(Vertex)) {

enroute-dp/internal/envoy/cluster.go

@@ -32,6 +32,7 @@ import (
 	"github.com/saarasio/enroute/enroute-dp/internal/dag"
 	"github.com/saarasio/enroute/enroute-dp/internal/logger"
 	"github.com/saarasio/enroute/enroute-dp/internal/protobuf"
+	"github.com/saarasio/enroute/enroute-dp/saarasconfig"
 )
 
 // CACertificateKey stores the key for the TLS validation secret cert
@@ -162,6 +163,7 @@ func cluster(cluster *dag.Cluster, service *dag.TCPService) *envoy_config_cluste
 		c.CloseConnectionsOnHostHealthFailure = true
 	}
 
+	// honor annotations
 	if anyPositive(service.MaxConnections, service.MaxPendingRequests, service.MaxRequests, service.MaxRetries) {
 		c.CircuitBreakers = &envoy_config_cluster_v3.CircuitBreakers{
 			Thresholds: []*envoy_config_cluster_v3.CircuitBreakers_Thresholds{{
@@ -172,6 +174,19 @@ func cluster(cluster *dag.Cluster, service *dag.TCPService) *envoy_config_cluste
 			}},
 		}
 	}
+
+	// honor circuitbreaker filter
+	for _, f := range cluster.ClusterFilters {
+		if f.Filter.Filter_type == saarasconfig.FILTER_TYPE_SERVICE_CIRCUITBREAKERS {
+			cbc, err := saarasconfig.UnmarshalCircuitBreakerconfig(f.Filter.Filter_config)
+			if err != nil {
+			}
+			c.CircuitBreakers.Thresholds[0].MaxConnections = u32nil(cbc.MaxConnections)
+			c.CircuitBreakers.Thresholds[0].MaxPendingRequests = u32nil(cbc.MaxPendingRequests)
+			c.CircuitBreakers.Thresholds[0].MaxRequests = u32nil(cbc.MaxRequests)
+			c.CircuitBreakers.Thresholds[0].MaxRetries = u32nil(cbc.MaxRetries)
+		}
+	}
 	return c
 }
 

enroute-dp/saarasconfig/commonconfig.go

@@ -42,6 +42,7 @@ const FILTER_TYPE_HTTP_RATELIMIT string = "http_filter_ratelimit"
 const FILTER_TYPE_RT_RATELIMIT string = "route_filter_ratelimit"
 const FILTER_TYPE_HTTP_JWT string = "http_filter_jwt"
 const FILTER_TYPE_HTTP_ACCESSLOG string = "http_filter_accesslog"
+const FILTER_TYPE_SERVICE_CIRCUITBREAKERS string = "service_filter_circuitbreakers"
 
 const PROXY_CONFIG_RATELIMIT string = "globalconfig_ratelimit"
 const PROXY_CONFIG_ACCESSLOG string = "globalconfig_accesslog"
@@ -186,3 +187,35 @@ func UnmarshalRouteMatchCondition(route_config string) (RouteMatchConditions, er
 
 	return mc, err
 }
+
+type CircuitBreakerConfig struct {
+	// Circuit breaking limits
+
+	// Max connections is maximum number of connections
+	// that Envoy will make to the upstream cluster.
+	MaxConnections uint32
+
+	// MaxPendingRequests is maximum number of pending
+	// requests that Envoy will allow to the upstream cluster.
+	MaxPendingRequests uint32
+
+	// MaxRequests is the maximum number of parallel requests that
+	// Envoy will make to the upstream cluster.
+	MaxRequests uint32
+
+	// MaxRetries is the maximum number of parallel retries that
+	// Envoy will allow to the upstream cluster.
+	MaxRetries uint32
+}
+
+func UnmarshalCircuitBreakerconfig(cc_config string) (CircuitBreakerConfig, error) {
+	var cbc CircuitBreakerConfig
+	var err error
+
+	buf := strings.NewReader(cc_config)
+	if err = json.NewDecoder(buf).Decode(&cbc); err != nil {
+		errors.Wrap(err, "decoding response")
+	}
+
+	return cbc, err
+}