HTTP Client circuit breaker integration (fixes #190) (#255)

Signed-off-by: Sotirios Mantziaris <[email protected]>

README.md

@@ -36,13 +36,12 @@ The latest version can be installed with
 go get github.com/mantzas/patron/cmd/patron
 ```
 
-The below is an example of a service created with the cli that has a module name `github.com/mantzas/test` and will be created in the test folder in the currend directory.
+The below is an example of a service created with the cli that has a module name `github.com/mantzas/test` and will be created in the test folder in the current directory.
 
 ```go
 patron -m "github.com/mantzas/test" -p "test"
 ```
 
-
 ## Service
 
 The `Service` has the role of glueing all of the above together, which are:
@@ -156,7 +155,7 @@ Everything else is exactly the same.
 
 ## Metrics and Tracing
 
-Tracing and metrics are provided by jaeger's implementation of the OpenTracing project.
+Tracing and metrics are provided by Jaeger's implementation of the OpenTracing project.
 Every component has been integrated with the above library and produces traces and metrics.
 Metrics are provided with the default HTTP component at the `/metrics` route for Prometheus to scrape.
 Tracing will be send to a jaeger agent which can be setup though environment variables mentioned in the config section. Sane defaults are applied for making the use easy.
@@ -166,6 +165,26 @@ downstream systems. The tracing information is added to each implementations hea
 - HTTP
 - AMQP
 - Kafka
+- SQL
+
+## Reliability
+
+The reliability package contains the following implementations:
+
+- Circuit Breaker
+
+### Circuit Breaker
+
+The circuit breaker supports a half-open state which allows to probe for successful responses in order to close the circuit again. Every aspect of the circuit breaker is configurable via it's settings.
+
+## Clients
+
+The following clients have been implemented:
+
+- http, with distributed tracing and optional circuit breaker
+- sql, with distributed tracing
+- kafka, with distributed tracing
+- amqp, with distributed tracing
 
 ## Logging
 

trace/http/http.go

@@ -5,6 +5,7 @@ import (
 	"net/http"
 	"time"
 
+	"github.com/mantzas/patron/reliability/circuitbreaker"
 	"github.com/mantzas/patron/trace"
 	"github.com/opentracing-contrib/go-stdlib/nethttp"
 	opentracing "github.com/opentracing/opentracing-go"
@@ -19,6 +20,7 @@ type Client interface {
 // TracedClient defines a HTTP client with tracing integrated.
 type TracedClient struct {
 	cl *http.Client
+	cb *circuitbreaker.CircuitBreaker
 }
 
 // New creates a new HTTP client.
@@ -28,6 +30,7 @@ func New(oo ...OptionFunc) (*TracedClient, error) {
 			Timeout:   60 * time.Second,
 			Transport: &nethttp.Transport{},
 		},
+		cb: nil,
 	}
 
 	for _, o := range oo {
@@ -48,14 +51,32 @@ func (tc *TracedClient) Do(ctx context.Context, req *http.Request) (*http.Respon
 		req,
 		nethttp.OperationName(trace.HTTPOpName("Client", req.Method, req.URL.String())),
 		nethttp.ComponentName(trace.HTTPClientComponent))
+
 	defer ht.Finish()
-	rsp, err := tc.cl.Do(req)
+
+	rsp, err := tc.do(req)
 	if err != nil {
 		ext.Error.Set(ht.Span(), true)
 	} else {
 		ext.HTTPStatusCode.Set(ht.Span(), uint16(rsp.StatusCode))
 	}
+
 	ext.HTTPMethod.Set(ht.Span(), req.Method)
 	ext.HTTPUrl.Set(ht.Span(), req.URL.String())
 	return rsp, err
 }
+
+func (tc *TracedClient) do(req *http.Request) (*http.Response, error) {
+	if tc.cb == nil {
+		return tc.cl.Do(req)
+	}
+
+	r, err := tc.cb.Execute(func() (interface{}, error) {
+		return tc.cl.Do(req)
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	return r.(*http.Response), nil
+}

trace/http/http_test.go

@@ -8,6 +8,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/mantzas/patron/reliability/circuitbreaker"
 	"github.com/mantzas/patron/trace"
 	opentracing "github.com/opentracing/opentracing-go"
 	"github.com/opentracing/opentracing-go/mocktracer"
@@ -26,53 +27,64 @@ func TestTracedClient_Do(t *testing.T) {
 	opentracing.SetGlobalTracer(mtr)
 	c, err := New()
 	assert.NoError(t, err)
+	cb, err := New(CircuitBreaker("test", circuitbreaker.Setting{}))
+	assert.NoError(t, err)
 	req, err := http.NewRequest("GET", ts.URL, nil)
 	assert.NoError(t, err)
-	rsp, err := c.Do(context.Background(), req)
+	reqErr, err := http.NewRequest("GET", "", nil)
 	assert.NoError(t, err)
-	assert.NotNil(t, rsp)
-	sp := mtr.FinishedSpans()[0]
-	assert.NotNil(t, sp)
-	assert.Equal(t, trace.HTTPOpName("Client", "GET", ts.URL), sp.OperationName)
-}
+	opName := trace.HTTPOpName("Client", "GET", ts.URL)
+	opNameError := "HTTP GET"
 
-func TestTracedClient_Do_Error(t *testing.T) {
-	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		assert.Equal(t, "true", r.Header.Get("Mockpfx-Ids-Sampled"))
-		assert.Equal(t, "46", r.Header.Get("Mockpfx-Ids-Spanid"))
-		assert.Equal(t, "43", r.Header.Get("Mockpfx-Ids-Traceid"))
-		fmt.Fprintln(w, "Hello, client")
-	}))
-	defer ts.Close()
-	mtr := mocktracer.New()
-	opentracing.SetGlobalTracer(mtr)
-	c, err := New()
-	assert.NoError(t, err)
-	req, err := http.NewRequest("GET", "", nil)
-	assert.NoError(t, err)
-	rsp, err := c.Do(context.Background(), req)
-	assert.Error(t, err)
-	assert.Nil(t, rsp)
-	sp := mtr.FinishedSpans()[0]
-	assert.NotNil(t, sp)
-	assert.Equal(t, "HTTP GET", sp.OperationName)
+	type args struct {
+		c   Client
+		req *http.Request
+	}
+	tests := []struct {
+		name       string
+		args       args
+		wantErr    bool
+		wantOpName string
+	}{
+		{name: "respose", args: args{c: c, req: req}, wantErr: false, wantOpName: opName},
+		{name: "response with circuit breaker", args: args{c: cb, req: req}, wantErr: false, wantOpName: opName},
+		{name: "error", args: args{c: cb, req: reqErr}, wantErr: true, wantOpName: opNameError},
+		{name: "error with circuit breaker", args: args{c: cb, req: reqErr}, wantErr: true, wantOpName: opNameError},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			rsp, err := tt.args.c.Do(context.Background(), tt.args.req)
+			if tt.wantErr {
+				assert.Error(t, err)
+				assert.Nil(t, rsp)
+			} else {
+				assert.NoError(t, err)
+				assert.NotNil(t, rsp)
+			}
+			sp := mtr.FinishedSpans()[0]
+			assert.NotNil(t, sp)
+			assert.Equal(t, tt.wantOpName, sp.OperationName)
+			mtr.Reset()
+		})
+	}
 }
 
 func TestNew(t *testing.T) {
 	type args struct {
-		opt OptionFunc
+		oo []OptionFunc
 	}
 	tests := []struct {
 		name    string
 		args    args
 		wantErr bool
 	}{
-		{name: "success", args: args{opt: Timeout(time.Second)}, wantErr: false},
-		{name: "failure, invalid timeout", args: args{opt: Timeout(0 * time.Second)}, wantErr: true},
+		{name: "success", args: args{oo: []OptionFunc{Timeout(time.Second), CircuitBreaker("test", circuitbreaker.Setting{})}}, wantErr: false},
+		{name: "failure, invalid timeout", args: args{oo: []OptionFunc{Timeout(0 * time.Second)}}, wantErr: true},
+		{name: "failure, invalid circuit breaker", args: args{[]OptionFunc{CircuitBreaker("", circuitbreaker.Setting{})}}, wantErr: true},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			got, err := New(tt.args.opt)
+			got, err := New(tt.args.oo...)
 			if tt.wantErr {
 				assert.Error(t, err)
 				assert.Nil(t, got)

trace/http/option.go

@@ -4,6 +4,7 @@ import (
 	"time"
 
 	"github.com/mantzas/patron/errors"
+	"github.com/mantzas/patron/reliability/circuitbreaker"
 )
 
 // OptionFunc definition for configuring the client in a functional way.
@@ -19,3 +20,15 @@ func Timeout(timeout time.Duration) OptionFunc {
 		return nil
 	}
 }
+
+// CircuitBreaker option for setting up a circuit breaker.
+func CircuitBreaker(name string, set circuitbreaker.Setting) OptionFunc {
+	return func(tc *TracedClient) error {
+		cb, err := circuitbreaker.New(name, set)
+		if err != nil {
+			return errors.Wrap(err, "failed to set circuit breaker")
+		}
+		tc.cb = cb
+		return nil
+	}
+}