Add breaker to grpcx client middleware. (#162)

Author: Skandalik

.travis.yml

@@ -4,7 +4,6 @@ language: go
 go:
   - "1.14.x"
   - "1.13.x"
-  - "1.11.x"
   - "1.x"
 env:
 - GO111MODULE=on

grpcx/middleware/breaker.go

@@ -0,0 +1,34 @@
+package middleware
+
+import (
+	"context"
+	"errors"
+
+	"github.com/msales/pkg/v4/breaker"
+	"github.com/msales/pkg/v4/stats"
+	"google.golang.org/grpc"
+)
+
+const (
+	breakerErrorKey = "breaker.error"
+	stateTag        = "state"
+)
+
+// WithClientBreaker adds breaker to client request.
+func WithClientBreaker(br *breaker.Breaker) grpc.UnaryClientInterceptor {
+	return func(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
+		err := br.Run(func() error {
+			return invoker(ctx, method, req, reply, cc, opts...)
+		})
+
+		if errors.Is(err, breaker.ErrOpenState) {
+			_ = stats.Inc(ctx, breakerErrorKey, 1, 1.0, stateTag, "open")
+		}
+
+		if errors.Is(err, breaker.ErrTooManyRequests) {
+			_ = stats.Inc(ctx, breakerErrorKey, 1, 1.0, stateTag, "too_many_requests")
+		}
+
+		return err
+	}
+}

grpcx/middleware/breaker_test.go

@@ -0,0 +1,56 @@
+package middleware_test
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/msales/pkg/v4/breaker"
+	"github.com/stretchr/testify/assert"
+	"google.golang.org/grpc"
+
+	. "github.com/msales/pkg/v4/grpcx/middleware"
+)
+
+var breakerErr = errors.New("breaker: circuit breaker is open")
+
+func TestWithBreaker(t *testing.T) {
+	ctx := context.Background()
+
+	br := breaker.NewBreaker(
+		breaker.RateFuse(1),
+		breaker.WithSleep(1*time.Second),
+		breaker.WithTestRequests(1),
+	)
+
+	interceptor := WithClientBreaker(br)
+	err := interceptor(ctx, "method", nil, nil, nil, func(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, opts ...grpc.CallOption) error {
+		return nil
+	})
+
+	assert.Nil(t, err)
+}
+
+func TestWithBreaker_Errored(t *testing.T) {
+	ctx := context.Background()
+
+	br := breaker.NewBreaker(
+		breaker.RateFuse(10),
+		breaker.WithSleep(1*time.Second),
+		breaker.WithTestRequests(1),
+	)
+
+	interceptor := WithClientBreaker(br)
+	err := interceptor(ctx, "method", nil, nil, nil, func(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, opts ...grpc.CallOption) error {
+		return testErr
+	})
+
+	assert.Equal(t, testErr, err)
+
+	err = interceptor(ctx, "method", nil, nil, nil, func(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, opts ...grpc.CallOption) error {
+		return testErr
+	})
+
+	assert.Equal(t, breakerErr, err)
+}