bug fix

hystrix/circuit.go

@@ -16,9 +16,9 @@ type CircuitBreaker struct {
 	forceOpen              bool
 	mutex                  *sync.RWMutex
 	openedOrLastTestedTime int64
-
-	executorPool *executorPool
-	metrics      *metricExchange
+	openedLastTime         int64
+	executorPool           *executorPool
+	metrics                *metricExchange
 }
 
 var (
@@ -53,6 +53,14 @@ func GetCircuit(name string) (*CircuitBreaker, bool, error) {
 	return circuitBreakers[name], !ok, nil
 }
 
+// ResetCircuit returns the circuit for the given command and whether this call created it.
+func ResetCircuit(name string) (*CircuitBreaker, bool, error) {
+	circuitBreakersMutex.Lock()
+	defer circuitBreakersMutex.Unlock()
+	circuitBreakers[name] = newCircuitBreaker(name)
+	return circuitBreakers[name], false, nil
+}
+
 // Flush purges all circuit and metric information from memory.
 func Flush() {
 	circuitBreakersMutex.Lock()
@@ -144,24 +152,24 @@ func (circuit *CircuitBreaker) setOpen() {
 		return
 	}
 
-	log.Printf("hystrix-go: opening circuit %v", circuit.Name)
-
+	log.Printf("hystrix-go: opening circuit %v;LastTestedTime:%d", circuit.Name, circuit.openedOrLastTestedTime)
+	circuit.openedLastTime = time.Now().UnixNano()
 	circuit.openedOrLastTestedTime = time.Now().UnixNano()
 	circuit.open = true
 }
 
 func (circuit *CircuitBreaker) setClose() {
 	circuit.mutex.Lock()
 	defer circuit.mutex.Unlock()
-
 	if !circuit.open {
 		return
 	}
-
-	log.Printf("hystrix-go: closing circuit %v", circuit.Name)
-
-	circuit.open = false
-	circuit.metrics.Reset()
+	now := time.Now().UnixNano()
+	if now > circuit.openedLastTime+getSettings(circuit.Name).SleepWindow.Nanoseconds() {
+		log.Printf("hystrix-go: closing circuit,%v;LastTestedTime:%d", circuit.Name, circuit.openedOrLastTestedTime)
+		circuit.open = false
+		circuit.metrics.Reset()
+	}
 }
 
 // ReportEvent records command metrics for tracking recent error rates and exposing data to the dashboard.

hystrix/eventstream.go

@@ -3,6 +3,8 @@ package hystrix
 import (
 	"bytes"
 	"encoding/json"
+	"fmt"
+	"io/ioutil"
 	"net/http"
 	"sync"
 	"time"
@@ -41,6 +43,55 @@ func (sh *StreamHandler) Stop() {
 var _ http.Handler = (*StreamHandler)(nil)
 
 func (sh *StreamHandler) ServeHTTP(rw http.ResponseWriter, req *http.Request) {
+	if req.URL.Path == "/config" {
+		data := struct {
+			Code    int         `json:"code"`
+			Message string      `json:"message"`
+			Data    interface{} `json:"data"`
+		}{
+			Code:    0,
+			Message: "Successful!",
+			Data:    nil,
+		}
+
+		if req.Method == "GET" {
+			configs := map[string]CommandConfig{}
+			for name, r := range GetCircuitSettings() {
+				configs[name] = CommandConfig{
+					Timeout:                int(r.Timeout.Nanoseconds() / 1e6),
+					MaxConcurrentRequests:  r.MaxConcurrentRequests,
+					ErrorPercentThreshold:  r.ErrorPercentThreshold,
+					RequestVolumeThreshold: int(r.RequestVolumeThreshold),
+					SleepWindow:            int(r.SleepWindow.Nanoseconds() / 1e6),
+				}
+			}
+			data.Data = configs
+		} else {
+			result, err := ioutil.ReadAll(req.Body)
+			if err != nil {
+				sh.error(rw, 1001, err)
+				return
+			}
+			var data struct {
+				CommandName string        `json:"command_name"`
+				Config      CommandConfig `json:"config"`
+			}
+			err = json.Unmarshal(result, &data)
+			if err != nil {
+				sh.error(rw, 1002, err)
+				return
+			}
+			if data.CommandName == "" || data.Config.Timeout == 0 || data.Config.ErrorPercentThreshold == 0 || data.Config.MaxConcurrentRequests == 0 || data.Config.RequestVolumeThreshold == 0 || data.Config.SleepWindow == 0 {
+				sh.error(rw, 1003, fmt.Errorf(`Params Valid error, "command_name,timeout,max_concurrent_requests,request_volume_threshold,sleep_window,error_percent_threshold"`))
+				return
+			}
+			ConfigureCommand(data.CommandName, data.Config)
+			ResetCircuit(data.CommandName)
+		}
+		rturnData, _ := json.Marshal(data)
+		rw.Write(rturnData)
+		return
+	}
 	// Make sure that the writer supports flushing.
 	f, ok := rw.(http.Flusher)
 	if !ok {
@@ -55,6 +106,7 @@ func (sh *StreamHandler) ServeHTTP(rw http.ResponseWriter, req *http.Request) {
 	rw.Header().Add("Content-Type", "text/event-stream")
 	rw.Header().Set("Cache-Control", "no-cache")
 	rw.Header().Set("Connection", "keep-alive")
+	rw.Header().Set("Access-Control-Allow-Origin", "*")
 	for {
 		select {
 		case <-notify:
@@ -70,6 +122,17 @@ func (sh *StreamHandler) ServeHTTP(rw http.ResponseWriter, req *http.Request) {
 	}
 }
 
+func (sh *StreamHandler) error(rw http.ResponseWriter, code int, err error) {
+	var data struct {
+		Code    int    `json:"code"`
+		Message string `json:"message"`
+	}
+	data.Code = code
+	data.Message = fmt.Sprintf("error :%s", err.Error())
+	d, _ := json.Marshal(data)
+	rw.Write(d)
+}
+
 func (sh *StreamHandler) loop() {
 	tick := time.Tick(1 * time.Second)
 	for {

hystrix/hystrix.go

@@ -24,16 +24,15 @@ func (e CircuitError) Error() string {
 // used to describe the pairing of your run/fallback functions with a circuit.
 type command struct {
 	sync.Mutex
-
-	ticket       *struct{}
-	start        time.Time
-	errChan      chan error
-	finished     chan bool
-	circuit      *CircuitBreaker
-	run          runFunc
-	fallback     fallbackFunc
-	runDuration  time.Duration
-	events       []string
+	ticket      *struct{}
+	start       time.Time
+	errChan     chan error
+	finished    chan bool
+	circuit     *CircuitBreaker
+	run         runFunc
+	fallback    fallbackFunc
+	runDuration time.Duration
+	events      []string
 }
 
 var (
@@ -52,11 +51,11 @@ var (
 // Define a fallback function if you want to define some code to execute during outages.
 func Go(name string, run runFunc, fallback fallbackFunc) chan error {
 	cmd := &command{
-		run:          run,
-		fallback:     fallback,
-		start:        time.Now(),
-		errChan:      make(chan error, 1),
-		finished:     make(chan bool, 1),
+		run:      run,
+		fallback: fallback,
+		start:    time.Now(),
+		errChan:  make(chan error, 1),
+		finished: make(chan bool, 1),
 	}
 
 	// dont have methods with explicit params and returns