Merge pull request #171 from jonstacks/issue-132

Add Circuit Breaker Module support

api/v1alpha1/httpsedge_types.go

@@ -51,6 +51,9 @@ type HTTPSEdgeRouteSpec struct {
 	// +kubebuilder:validation:Required
 	Backend TunnelGroupBackend `json:"backend,omitempty"`
 
+	// CircuitBreaker is a circuit breaker configuration to apply to this route
+	CircuitBreaker *EndpointCircuitBreaker `json:"circuitBreaker,omitempty"`
+
 	// Compression is whether or not to enable compression for this route
 	Compression *EndpointCompression `json:"compression,omitempty"`
 

api/v1alpha1/ngrok_common.go

@@ -1,5 +1,7 @@
 package v1alpha1
 
+import "k8s.io/apimachinery/pkg/api/resource"
+
 // common ngrok API/Dashboard fields
 type ngrokAPICommon struct {
 	// Description is a human-readable description of the object in the ngrok API/Dashboard
@@ -71,3 +73,25 @@ type EndpointWebhookVerification struct {
 	// requests from the given provider. All providers except AWS SNS require a secret
 	SecretRef *SecretKeyRef `json:"secret,omitempty"`
 }
+
+type EndpointCircuitBreaker struct {
+	// Integer number of seconds after which the circuit is tripped to wait before
+	// re-evaluating upstream health
+	TrippedDuration uint32 `json:"trippedDuration,omitempty"`
+
+	// Integer number of seconds in the statistical rolling window that metrics are
+	// retained for.
+	RollingWindow uint32 `json:"rollingWindow,omitempty"`
+
+	// Integer number of buckets into which metrics are retained. Max 128.
+	//+kubebuilder:validation:Minimum=1
+	//+kubebuilder:validation:Maximum=128
+	NumBuckets uint32 `json:"numBuckets,omitempty"`
+
+	// Integer number of requests in a rolling window that will trip the circuit.
+	// Helpful if traffic volume is low.
+	VolumeThreshold uint32 `json:"volumeThreshold,omitempty"`
+
+	// Error threshold percentage should be between 0 - 1.0, not 0-100.0
+	ErrorThresholdPercentage resource.Quantity `json:"errorThresholdPercentage,omitempty"`
+}

api/v1alpha1/ngrokmoduleset_types.go

@@ -29,15 +29,17 @@ import (
 )
 
 type NgrokModuleSetModules struct {
-	// Compression configuration for this module
+	// CircuitBreaker configuration for this module set
+	CircuitBreaker *EndpointCircuitBreaker `json:"circuitBreaker,omitempty"`
+	// Compression configuration for this module set
 	Compression *EndpointCompression `json:"compression,omitempty"`
-	// Header configuration for this module
+	// Header configuration for this module set
 	Headers *EndpointHeaders `json:"headers,omitempty"`
-	// IPRestriction configuration for this module
+	// IPRestriction configuration for this module set
 	IPRestriction *EndpointIPPolicy `json:"ipRestriction,omitempty"`
-	// TLSTermination configuration for this module
+	// TLSTermination configuration for this module set
 	TLSTermination *EndpointTLSTerminationAtEdge `json:"tlsTermination,omitempty"`
-	// WebhookVerification configuration for this module
+	// WebhookVerification configuration for this module set
 	WebhookVerification *EndpointWebhookVerification `json:"webhookVerification,omitempty"`
 }
 
@@ -60,6 +62,9 @@ func (ms *NgrokModuleSet) Merge(o *NgrokModuleSet) {
 	msmod := &ms.Modules
 	omod := o.Modules
 
+	if omod.CircuitBreaker != nil {
+		msmod.CircuitBreaker = omod.CircuitBreaker
+	}
 	if omod.Compression != nil {
 		msmod.Compression = omod.Compression
 	}

internal/controllers/httpsedge_controller.go

@@ -440,6 +440,7 @@ type edgeRouteModuleUpdater struct {
 
 func (u *edgeRouteModuleUpdater) updateModulesForRoute(ctx context.Context, route *ngrok.HTTPSEdgeRoute, routeSpec *ingressv1alpha1.HTTPSEdgeRouteSpec) error {
 	funcs := []func(context.Context, *ngrok.HTTPSEdgeRoute, *ingressv1alpha1.HTTPSEdgeRouteSpec) error{
+		u.setEdgeRouteCircuitBreaker,
 		u.setEdgeRouteCompression,
 		u.setEdgeRouteIPRestriction,
 		u.setEdgeRouteRequestHeaders,
@@ -462,6 +463,42 @@ func (u *edgeRouteModuleUpdater) edgeRouteItem(route *ngrok.HTTPSEdgeRoute) *ngr
 	}
 }
 
+func (u *edgeRouteModuleUpdater) setEdgeRouteCircuitBreaker(ctx context.Context, route *ngrok.HTTPSEdgeRoute, routeSpec *ingressv1alpha1.HTTPSEdgeRouteSpec) error {
+	circuitBreaker := routeSpec.CircuitBreaker
+
+	client := u.clientset.CircuitBreaker()
+
+	// Early return if nothing to be done
+	if circuitBreaker == nil {
+		if route.CircuitBreaker == nil {
+			u.log.Info("CircuitBreaker matches desired state, skipping update")
+			return nil
+		}
+
+		return client.Delete(ctx, u.edgeRouteItem(route))
+	}
+
+	module := ngrok.EndpointCircuitBreaker{
+		TrippedDuration:          circuitBreaker.TrippedDuration,
+		RollingWindow:            circuitBreaker.RollingWindow,
+		NumBuckets:               circuitBreaker.NumBuckets,
+		VolumeThreshold:          circuitBreaker.VolumeThreshold,
+		ErrorThresholdPercentage: circuitBreaker.ErrorThresholdPercentage.AsApproximateFloat64(),
+	}
+
+	if reflect.DeepEqual(module, route.CircuitBreaker) {
+		u.log.Info("CircuitBreaker matches desired state, skipping update")
+		return nil
+	}
+
+	_, err := client.Replace(ctx, &ngrok.EdgeRouteCircuitBreakerReplace{
+		EdgeID: route.EdgeID,
+		ID:     route.ID,
+		Module: module,
+	})
+	return err
+}
+
 func (u *edgeRouteModuleUpdater) setEdgeRouteCompression(ctx context.Context, route *ngrok.HTTPSEdgeRoute, routeSpec *ingressv1alpha1.HTTPSEdgeRouteSpec) error {
 	compression := routeSpec.Compression
 

internal/store/driver.go

@@ -403,6 +403,7 @@ func (d *Driver) calculateHTTPSEdges() []ingressv1alpha1.HTTPSEdge {
 							Backend: ingressv1alpha1.TunnelGroupBackend{
 								Labels: backendToLabelMap(httpIngressPath.Backend, ingress.Namespace),
 							},
+							CircuitBreaker:      modSet.Modules.CircuitBreaker,
 							Compression:         modSet.Modules.Compression,
 							IPRestriction:       modSet.Modules.IPRestriction,
 							Headers:             modSet.Modules.Headers,