stop circuit breaker Go routine before reloading api spec (TykTechnol…

…ogies#2256)

added changes for:
```
File: tyk
Type: goroutine
Time: May 1, 2019 at 4:32pm (EDT)
Entering interactive mode (type "help" for commands, "o" for options)
(pprof) top10
Showing nodes accounting for 10488, 100% of 10490 total
Dropped 74 nodes (cum <= 52)
Showing top 10 nodes out of 22
      flat  flat%   sum%        cum   cum%
     10488   100%   100%      10488   100%  runtime.gopark
         0     0%   100%         96  0.92%  bufio.(*Reader).Peek
         0     0%   100%        141  1.34%  bufio.(*Reader).fill
         0     0%   100%        999  9.52%  github.com/TykTechnologies/tyk/vendor/github.com/TykTechnologies/leakybucket/memorycache.(*Cache).startCleanupTimer.func1
         0     0%   100%       4557 43.44%  github.com/TykTechnologies/tyk/vendor/github.com/rubyist/circuitbreaker.(*Breaker).Subscribe.func1
         0     0%   100%        145  1.38%  internal/poll.(*FD).Read
         0     0%   100%        147  1.40%  internal/poll.(*pollDesc).wait
         0     0%   100%        147  1.40%  internal/poll.(*pollDesc).waitRead
         0     0%   100%        147  1.40%  internal/poll.runtime_pollWait
         0     0%   100%       4557 43.44%  main.APIDefinitionLoader.compileCircuitBreakerPathSpec.func1
(pprof)
```
this change addresses `4557 43.44%  main.APIDefinitionLoader.compileCircuitBreakerPathSpec.func1`

to fix it completely we need to fix `github.com/rubyist/circuitbreaker.(*Breaker).Subscribe`

api_definition.go

@@ -144,7 +144,7 @@ type TransformSpec struct {
 
 type ExtendedCircuitBreakerMeta struct {
 	apidef.CircuitBreakerMeta
-	CB *circuit.Breaker
+	CB *circuit.Breaker `json:"-"`
 }
 
 // APISpec represents a path specification for an API, to avoid enumerating multiple nested lists, a single
@@ -179,6 +179,29 @@ type APISpec struct {
 	OrgHasNoSession          bool
 
 	middlewareChain *ChainObject
+
+	shouldRelease bool
+}
+
+// Release re;leases all resources associated with API spec
+func (s *APISpec) Release() {
+	// mark spec as to be released
+	s.shouldRelease = true
+
+	// release circuit breaker resources
+	for _, path := range s.RxPaths {
+		for _, urlSpec := range path {
+			if urlSpec.CircuitBreaker.CB != nil {
+				// this will force CB-event reading routing to exit
+				urlSpec.CircuitBreaker.CB.Reset()
+
+				// TODO: close all circuit breaker's event receivers
+				// which should let event reading Go-routines to exit (need to change circuitbreaker package)
+			}
+		}
+	}
+
+	// release all other resources associated with spec
 }
 
 // APIDefinitionLoader will load an Api definition from a storage
@@ -714,6 +737,12 @@ func (a APIDefinitionLoader) compileCircuitBreakerPathSpec(paths []apidef.Circui
 					})
 
 				case circuit.BreakerReset:
+					// check if this spec is set to release resources
+					if spec.shouldRelease {
+						// time to stop this Go-routine
+						return
+					}
+
 					spec.FireEvent(EventBreakerTriggered, EventCurcuitBreakerMeta{
 						EventMetaDefault: EventMetaDefault{Message: "Breaker Reset"},
 						CircuitEvent:     e,

api_loader.go

@@ -668,6 +668,12 @@ func loadApps(specs []*APISpec, muxer *mux.Router) {
 
 	// Swap in the new register
 	apisMu.Lock()
+
+	// release current specs resources before overwriting map
+	for _, curSpec := range apisByID {
+		curSpec.Release()
+	}
+
 	apisByID = tmpSpecRegister
 	apisMu.Unlock()
 

gateway_test.go

@@ -918,21 +918,62 @@ func TestReloadGoroutineLeakWithAsyncWrites(t *testing.T) {
 
 	globalConf := config.Global()
 	globalConf.UseAsyncSessionWrite = true
+	globalConf.EnableJSVM = false
 	config.SetGlobal(globalConf)
 	defer resetTestConfig()
 
-	buildAndLoadAPI(func(spec *APISpec) {
+	specs := buildAndLoadAPI(func(spec *APISpec) {
 		spec.Proxy.ListenPath = "/"
 	})
 
 	before := runtime.NumGoroutine()
-	doReload()
+
+	loadAPI(specs...) // just doing doReload() doesn't load anything as buildAndLoadAPI cleans up folder with API specs
+
+	time.Sleep(100 * time.Millisecond)
+
+	after := runtime.NumGoroutine()
+
+	if before < after {
+		t.Errorf("Goroutine leak, was: %d, after reload: %d", before, after)
+	}
+}
+
+func TestReloadGoroutineLeakWithCircuitBreaker(t *testing.T) {
+	ts := newTykTestServer()
+	defer ts.Close()
+
+	globalConf := config.Global()
+	globalConf.EnableJSVM = false
+	config.SetGlobal(globalConf)
+	defer resetTestConfig()
+
+	specs := buildAndLoadAPI(func(spec *APISpec) {
+		spec.Proxy.ListenPath = "/"
+		updateAPIVersion(spec, "v1", func(version *apidef.VersionInfo) {
+			version.ExtendedPaths = apidef.ExtendedPathsSet{
+				CircuitBreaker: []apidef.CircuitBreakerMeta{
+					{
+						Path:                 "/",
+						Method:               http.MethodGet,
+						ThresholdPercent:     0.5,
+						Samples:              5,
+						ReturnToServiceAfter: 10,
+					},
+				},
+			}
+		})
+	})
+
+	before := runtime.NumGoroutine()
+
+	loadAPI(specs...) // just doing doReload() doesn't load anything as buildAndLoadAPI cleans up folder with API specs
 
 	time.Sleep(100 * time.Millisecond)
 
 	after := runtime.NumGoroutine()
 
-	if before != after {
+	if before < after-1 { // -1 because there is one will be running until we fix circuitbreaker Subscribe() method
 		t.Errorf("Goroutine leak, was: %d, after reload: %d", before, after)
 	}
 }

helpers_test.go

@@ -518,7 +518,10 @@ func loadAPI(specs ...*APISpec) (out []*APISpec) {
 	}()
 
 	for i, spec := range specs {
-		specBytes, _ := json.Marshal(spec)
+		specBytes, err := json.Marshal(spec)
+		if err != nil {
+			panic(err)
+		}
 		specFilePath := filepath.Join(config.Global().AppPath, spec.APIID+strconv.Itoa(i)+".json")
 		if err := ioutil.WriteFile(specFilePath, specBytes, 0644); err != nil {
 			panic(err)