Integrate circuit breaker modules with reverse proxy

3 files changed, 36 insertions, 2 deletions

diff --git a/modules/caddyhttp/reverseproxy/healthchecks.go b/modules/caddyhttp/reverseproxy/healthchecks.go
index 0b46d04..673f7c4 100644
--- a/modules/caddyhttp/reverseproxy/healthchecks.go
+++ b/modules/caddyhttp/reverseproxy/healthchecks.go
@@ -64,6 +64,14 @@ type PassiveHealthChecks struct {
 	UnhealthyLatency      caddy.Duration `json:"unhealthy_latency,omitempty"`
 }
 
+// CircuitBreaker is a type that can act as an early-warning
+// system for the health checker when backends are getting
+// overloaded.
+type CircuitBreaker interface {
+	OK() bool
+	RecordMetric(statusCode int, latency time.Duration)
+}
+
 // activeHealthChecker runs active health checks on a
 // regular basis and blocks until
 // h.HealthChecks.Active.stopChan is closed.
@@ -202,7 +210,7 @@ func (h *Handler) doActiveHealthCheck(hostAddr string, host Host) error {
 // remembers 1 failure for upstream for the configured
 // duration. If passive health checks are disabled or
 // failure expiry is 0, this is a no-op.
-func (h Handler) countFailure(upstream *Upstream) {
+func (h *Handler) countFailure(upstream *Upstream) {
 	// only count failures if passive health checking is enabled
 	// and if failures are configured have a non-zero expiry
 	if h.HealthChecks == nil || h.HealthChecks.Passive == nil {
diff --git a/modules/caddyhttp/reverseproxy/hosts.go b/modules/caddyhttp/reverseproxy/hosts.go
index 5100936..b40e614 100644
--- a/modules/caddyhttp/reverseproxy/hosts.go
+++ b/modules/caddyhttp/reverseproxy/hosts.go
@@ -69,21 +69,29 @@ type Upstream struct {
 
 	healthCheckPolicy *PassiveHealthChecks
 	hostURL           *url.URL
+	cb                CircuitBreaker
 }
 
 // Available returns true if the remote host
-// is available to receive requests.
+// is available to receive requests. This is
+// the method that should be used by selection
+// policies, etc. to determine if a backend
+// should be able to be sent a request.
 func (u *Upstream) Available() bool {
 	return u.Healthy() && !u.Full()
 }
 
 // Healthy returns true if the remote host
 // is currently known to be healthy or "up".
+// It consults the circuit breaker, if any.
 func (u *Upstream) Healthy() bool {
 	healthy := !u.Host.Unhealthy()
 	if healthy && u.healthCheckPolicy != nil {
 		healthy = u.Host.Fails() < u.healthCheckPolicy.MaxFails
 	}
+	if healthy && u.cb != nil {
+		healthy = u.cb.OK()
+	}
 	return healthy
 }
 
diff --git a/modules/caddyhttp/reverseproxy/reverseproxy.go b/modules/caddyhttp/reverseproxy/reverseproxy.go
index ca54741..16d7f7a 100644
--- a/modules/caddyhttp/reverseproxy/reverseproxy.go
+++ b/modules/caddyhttp/reverseproxy/reverseproxy.go
@@ -37,12 +37,14 @@ func init() {
 // Handler implements a highly configurable and production-ready reverse proxy.
 type Handler struct {
 	TransportRaw  json.RawMessage `json:"transport,omitempty"`
+	CBRaw         json.RawMessage `json:"circuit_breaker,omitempty"`
 	LoadBalancing *LoadBalancing  `json:"load_balancing,omitempty"`
 	HealthChecks  *HealthChecks   `json:"health_checks,omitempty"`
 	Upstreams     UpstreamPool    `json:"upstreams,omitempty"`
 	FlushInterval caddy.Duration  `json:"flush_interval,omitempty"`
 
 	Transport http.RoundTripper `json:"-"`
+	CB        CircuitBreaker    `json:"-"`
 }
 
 // CaddyModule returns the Caddy module information.
@@ -55,6 +57,7 @@ func (Handler) CaddyModule() caddy.ModuleInfo {
 
 // Provision ensures that h is set up properly before use.
 func (h *Handler) Provision(ctx caddy.Context) error {
+	// start by loading modules
 	if h.TransportRaw != nil {
 		val, err := ctx.LoadModuleInline("protocol", "http.handlers.reverse_proxy.transport", h.TransportRaw)
 		if err != nil {
@@ -73,6 +76,14 @@ func (h *Handler) Provision(ctx caddy.Context) error {
 		h.LoadBalancing.SelectionPolicy = val.(Selector)
 		h.LoadBalancing.SelectionPolicyRaw = nil // allow GC to deallocate - TODO: Does this help?
 	}
+	if h.CBRaw != nil {
+		val, err := ctx.LoadModuleInline("type", "http.handlers.reverse_proxy.circuit_breakers", h.CBRaw)
+		if err != nil {
+			return fmt.Errorf("loading circuit breaker module: %s", err)
+		}
+		h.CB = val.(CircuitBreaker)
+		h.CBRaw = nil // allow GC to deallocate - TODO: Does this help?
+	}
 
 	if h.Transport == nil {
 		h.Transport = defaultTransport
@@ -123,6 +134,8 @@ func (h *Handler) Provision(ctx caddy.Context) error {
 	}
 
 	for _, upstream := range h.Upstreams {
+		upstream.cb = h.CB
+
 		// url parser requires a scheme
 		if !strings.Contains(upstream.Address, "://") {
 			upstream.Address = "http://" + upstream.Address
@@ -307,6 +320,11 @@ func (h *Handler) reverseProxy(rw http.ResponseWriter, req *http.Request, upstre
 		return err
 	}
 
+	// update circuit breaker on current conditions
+	if upstream.cb != nil {
+		upstream.cb.RecordMetric(res.StatusCode, latency)
+	}
+
 	// perform passive health checks (if enabled)
 	if h.HealthChecks != nil && h.HealthChecks.Passive != nil {
 		// strike if the status code matches one that is "bad"