It compiles, but not sure it works!

api_definition_manager.go

@@ -6,6 +6,7 @@ import (
 	"errors"
 	"github.com/gorilla/context"
 	"github.com/lonelycode/tykcommon"
+	"github.com/rubyist/circuitbreaker"
 	"io/ioutil"
 	"labix.org/v2/mgo"
 	"labix.org/v2/mgo/bson"
@@ -41,6 +42,7 @@ const (
 	HeaderInjectedResponse URLStatus = 7
 	TransformedResponse    URLStatus = 8
 	HardTimeout            URLStatus = 9
+	CircuitBreaker         URLStatus = 10
 )
 
 // RequestStatus is a custom type to avoid collisions
@@ -65,6 +67,7 @@ const (
 	StatusActionRedirect           RequestStatus = "Found an Action, changing route"
 	StatusRedirectFlowByReply      RequestStatus = "Exceptional action requested, redirecting flow!"
 	StatusHardTimeout              RequestStatus = "Hard Timeout enforced on path"
+	StatusCircuitBreaker           RequestStatus = "Circuit breaker enforced"
 )
 
 // URLSpec represents a flattened specification for URLs, used to check if a proxy URL
@@ -79,13 +82,19 @@ type URLSpec struct {
 	InjectHeaders           tykcommon.HeaderInjectionMeta
 	InjectHeadersResponse   tykcommon.HeaderInjectionMeta
 	HardTimeout             tykcommon.HardTimeoutMeta
+	CircuitBreaker          ExtendedCircuitBreakerMeta
 }
 
 type TransformSpec struct {
 	tykcommon.TemplateMeta
 	Template *textTemplate.Template
 }
 
+type ExtendedCircuitBreakerMeta struct {
+	tykcommon.CircuitBreakerMeta
+	CB *circuit.Breaker
+}
+
 // APISpec represents a path specification for an API, to avoid enumerating multiple nested lists, a single
 // flattened URL list is checked for matching paths and then it's status evaluated if found.
 type APISpec struct {
@@ -188,7 +197,7 @@ func (a *APIDefinitionLoader) MakeSpec(thisAppConfig tykcommon.APIDefinition) AP
 
 		// If we have transitioned to extended path specifications, we should use these now
 		if v.UseExtendedPaths {
-			pathSpecs, whiteListSpecs = a.getExtendedPathSpecs(v)
+			pathSpecs, whiteListSpecs = a.getExtendedPathSpecs(v, &newAppSpec)
 
 		} else {
 			log.Warning("Path-based version path list settings are being deprecated, please upgrade your defintitions to the new standard as soon as spossible")
@@ -494,7 +503,51 @@ func (a *APIDefinitionLoader) compileTimeoutPathSpec(paths []tykcommon.HardTimeo
 	return thisURLSpec
 }
 
-func (a *APIDefinitionLoader) getExtendedPathSpecs(apiVersionDef tykcommon.VersionInfo) ([]URLSpec, bool) {
+func (a *APIDefinitionLoader) compileCircuitBreakerPathSpec(paths []tykcommon.CircuitBreakerMeta, stat URLStatus, apiSpec *APISpec) []URLSpec {
+
+	// transform an extended configuration URL into an array of URLSpecs
+	// This way we can iterate the whole array once, on match we break with status
+	thisURLSpec := []URLSpec{}
+
+	for _, stringSpec := range paths {
+		newSpec := URLSpec{}
+		a.generateRegex(stringSpec.Path, &newSpec, stat)
+		// Extend with method actions
+		newSpec.CircuitBreaker = ExtendedCircuitBreakerMeta{CircuitBreakerMeta: stringSpec}
+		newSpec.CircuitBreaker.CB = circuit.NewRateBreaker(stringSpec.ThresholdPercent, stringSpec.Samples)
+
+		events := newSpec.CircuitBreaker.CB.Subscribe()
+		go func() {
+			path := stringSpec.Path
+			spec := apiSpec
+			breakerPtr := newSpec.CircuitBreaker.CB
+			for {
+				e := <-events
+				spec.FireEvent(EVENT_BreakerTriggered,
+					EVENT_CurcuitBreakerMeta{
+						EventMetaDefault: EventMetaDefault{Message: "Breaker Tripped"},
+						CircuitEvent:     e,
+						Path:             path,
+						APIID:            spec.APIID,
+					})
+				switch e {
+				case circuit.BreakerTripped:
+					// Start a timer function
+					go func(timeout int, breaker *circuit.Breaker) {
+						time.Sleep(time.Duration(timeout) * time.Second)
+						breaker.Reset()
+					}(newSpec.CircuitBreaker.ReturnToServiceAfter, breakerPtr)
+				}
+			}
+		}()
+
+		thisURLSpec = append(thisURLSpec, newSpec)
+	}
+
+	return thisURLSpec
+}
+
+func (a *APIDefinitionLoader) getExtendedPathSpecs(apiVersionDef tykcommon.VersionInfo, apiSpec *APISpec) ([]URLSpec, bool) {
 	// TODO: New compiler here, needs to put data into a different structure
 
 	ignoredPaths := a.compileExtendedPathSpec(apiVersionDef.ExtendedPaths.Ignored, Ignored)
@@ -506,6 +559,7 @@ func (a *APIDefinitionLoader) getExtendedPathSpecs(apiVersionDef tykcommon.Versi
 	headerTransformPaths := a.compileInjectedHeaderSpec(apiVersionDef.ExtendedPaths.TransformHeader, HeaderInjected)
 	headerTransformPathsOnResponse := a.compileInjectedHeaderSpec(apiVersionDef.ExtendedPaths.TransformResponseHeader, HeaderInjectedResponse)
 	hardTimeouts := a.compileTimeoutPathSpec(apiVersionDef.ExtendedPaths.HardTimeouts, HardTimeout)
+	circuitBreakers := a.compileCircuitBreakerPathSpec(apiVersionDef.ExtendedPaths.CircuitBreaker, CircuitBreaker, apiSpec)
 
 	combinedPath := []URLSpec{}
 	combinedPath = append(combinedPath, ignoredPaths...)
@@ -517,6 +571,7 @@ func (a *APIDefinitionLoader) getExtendedPathSpecs(apiVersionDef tykcommon.Versi
 	combinedPath = append(combinedPath, headerTransformPaths...)
 	combinedPath = append(combinedPath, headerTransformPathsOnResponse...)
 	combinedPath = append(combinedPath, hardTimeouts...)
+	combinedPath = append(combinedPath, circuitBreakers...)
 
 	if len(whiteListPaths) > 0 {
 		return combinedPath, true
@@ -552,6 +607,8 @@ func (a *APISpec) getURLStatus(stat URLStatus) RequestStatus {
 		return StatusTransformResponse
 	case HardTimeout:
 		return StatusHardTimeout
+	case CircuitBreaker:
+		return StatusCircuitBreaker
 	default:
 		log.Error("URL Status was not one of Ignored, Blacklist or WhiteList! Blocking.")
 		return EndPointNotAllowed
@@ -656,6 +713,11 @@ func (a *APISpec) CheckSpecMatchesStatus(url string, method interface{}, RxPaths
 					if method != nil && method.(string) == v.HardTimeout.Method {
 						return true, &v.HardTimeout.TimeOut
 					}
+				case CircuitBreaker:
+					if method != nil && method.(string) == v.CircuitBreaker.Method {
+						return true, &v.CircuitBreaker
+					}
+
 				}
 
 			}

event_system.go

@@ -7,6 +7,7 @@ import (
 	"errors"
 	"fmt"
 	"github.com/lonelycode/tykcommon"
+	"github.com/rubyist/circuitbreaker"
 	"labix.org/v2/mgo/bson"
 	"net/http"
 	"time"
@@ -26,6 +27,7 @@ const (
 	EVENT_VersionFailure    tykcommon.TykEvent = "VersionFailure"
 	EVENT_OrgQuotaExceeded  tykcommon.TykEvent = "OrgQuotaExceeded"
 	EVENT_TriggerExceeded   tykcommon.TykEvent = "TriggerExceeded"
+	EVENT_BreakerTriggered  tykcommon.TykEvent = "BreakerTriggered"
 )
 
 // EventMetaDefault is a standard embedded struct to be used with custom event metadata types, gives an interface for
@@ -59,6 +61,14 @@ type EVENT_AuthFailureMeta struct {
 	Key    string
 }
 
+// EVENT_CurcuitBreakerMeta is the event status for a circuit breaker tripping
+type EVENT_CurcuitBreakerMeta struct {
+	EventMetaDefault
+	Path         string
+	APIID        string
+	CircuitEvent circuit.BreakerEvent
+}
+
 // EVENT_KeyExpiredMeta is the metadata structure for an auth failure (EVENT_KeyExpired)
 type EVENT_KeyExpiredMeta struct {
 	EventMetaDefault
@@ -161,6 +171,25 @@ func (t TykMiddleware) FireEvent(eventName tykcommon.TykEvent, eventMetaData int
 	}
 }
 
+func (s APISpec) FireEvent(eventName tykcommon.TykEvent, eventMetaData interface{}) {
+
+	log.Debug("EVENT FIRED")
+	handlers, handlerExists := s.EventPaths[eventName]
+
+	if handlerExists {
+		log.Debug("FOUND EVENT HANDLERS")
+		eventMessage := EventMessage{}
+		eventMessage.EventMetaData = eventMetaData
+		eventMessage.EventType = eventName
+		eventMessage.TimeStamp = time.Now().Local().String()
+
+		for _, handler := range handlers {
+			log.Debug("FIRING HANDLER")
+			go handler.HandleEvent(eventMessage)
+		}
+	}
+}
+
 // LogMessageEventHandler is a sample Event Handler
 type LogMessageEventHandler struct {
 	conf map[string]interface{}

tyk_reverse_proxy_clone.go

@@ -371,6 +371,26 @@ func (p *ReverseProxy) CheckHardTimeoutEnforced(spec *APISpec, req *http.Request
 	return false, 0
 }
 
+func (p *ReverseProxy) CheckCircuitBreakerEnforced(spec *APISpec, req *http.Request) (bool, *ExtendedCircuitBreakerMeta) {
+	var stat RequestStatus
+	var meta interface{}
+	var found bool
+
+	_, versionPaths, _, _ := spec.GetVersionData(req)
+	found, meta = spec.CheckSpecMatchesStatus(req.URL.Path, req.Method, versionPaths, CircuitBreaker)
+	if found {
+		stat = StatusCircuitBreaker
+	}
+
+	if stat == StatusCircuitBreaker {
+		thisMeta := meta.(*ExtendedCircuitBreakerMeta)
+		log.Warning("CIRCUIT BREAKER ENFORCED: ", *thisMeta)
+		return true, thisMeta
+	}
+
+	return false, nil
+}
+
 func (p *ReverseProxy) WrappedServeHTTP(rw http.ResponseWriter, req *http.Request, withCache bool) *http.Response {
 	transport := p.Transport
 	if transport == nil {
@@ -424,7 +444,28 @@ func (p *ReverseProxy) WrappedServeHTTP(rw http.ResponseWriter, req *http.Reques
 		outreq.Header.Set("X-Forwarded-For", clientIP)
 	}
 
-	res, err := transport.RoundTrip(outreq)
+	// Circuit breaker
+	breakerEnforced, breakerConf := p.CheckCircuitBreakerEnforced(p.TykAPISpec, req)
+	// TODO:
+	// 1. If the circuit breaker is active - wrap the RoundTrip call with a breaker function
+	// 2. when we init the APISpec, we need to create CBs for each monitored endpoint, this means extending the APISpec so we can store pointers
+	// 3. Set up monitoring functions and hook them up to the event handler
+
+	var res *http.Response
+	var err error
+	if breakerEnforced {
+		if breakerConf.CB.Ready() {
+			res, err = transport.RoundTrip(outreq)
+			if err != nil {
+				breakerConf.CB.Fail()
+			} else {
+				breakerConf.CB.Success()
+			}
+		}
+	} else {
+		res, err = transport.RoundTrip(outreq)
+	}
+
 	if err != nil {
 		log.Error("http: proxy error: ", err)
 		if strings.Contains(err.Error(), "timeout awaiting response headers") {