feat: support circuit breaker (#228)

Author: dk-lockdown

cmd/cmd.go

@@ -37,6 +37,7 @@ import (
 	"github.com/cectc/dbpack/pkg/executor"
 	"github.com/cectc/dbpack/pkg/filter"
 	_ "github.com/cectc/dbpack/pkg/filter/audit_log"
+	_ "github.com/cectc/dbpack/pkg/filter/breaker"
 	_ "github.com/cectc/dbpack/pkg/filter/crypto"
 	_ "github.com/cectc/dbpack/pkg/filter/dt"
 	_ "github.com/cectc/dbpack/pkg/filter/metrics"

go.mod

@@ -34,25 +34,25 @@ require (
 	go.etcd.io/etcd/api/v3 v3.5.0-alpha.0
 	go.etcd.io/etcd/client/v3 v3.5.0-alpha.0
 	go.uber.org/atomic v1.9.0
-	go.uber.org/goleak v1.1.11
+	go.uber.org/goleak v1.1.12
 	go.uber.org/ratelimit v0.2.1-0.20220713224938-b62b799bc9a5
 	go.uber.org/zap v1.21.0
 	golang.org/x/net v0.0.0-20220225172249-27dd8689420f
 	golang.org/x/text v0.3.7
-	google.golang.org/grpc v1.40.0
+	google.golang.org/grpc v1.43.0
 	gopkg.in/yaml.v3 v3.0.0
 	k8s.io/client-go v0.23.5
 	vimagination.zapto.org/byteio v1.0.1
 )
 
 require (
 	github.com/Azure/go-ansiterm v0.0.0-20210617225240-d185dfc1b5a1 // indirect
-	github.com/Microsoft/go-winio v0.4.17 // indirect
-	github.com/Microsoft/hcsshim v0.8.23 // indirect
+	github.com/Microsoft/go-winio v0.5.1 // indirect
+	github.com/Microsoft/hcsshim v0.9.4 // indirect
 	github.com/cenkalti/backoff/v4 v4.1.2 // indirect
-	github.com/containerd/cgroups v1.0.1 // indirect
-	github.com/containerd/containerd v1.5.9 // indirect
-	github.com/docker/distribution v2.7.1+incompatible // indirect
+	github.com/containerd/cgroups v1.0.4 // indirect
+	github.com/containerd/containerd v1.6.8 // indirect
+	github.com/docker/distribution v2.8.1+incompatible // indirect
 	github.com/docker/docker v20.10.11+incompatible // indirect
 	github.com/docker/go-connections v0.4.0 // indirect
 	github.com/docker/go-units v0.4.0 // indirect
@@ -71,10 +71,10 @@ require (
 	github.com/moby/term v0.0.0-20210619224110-3f7ff695adc6 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
-	github.com/morikuni/aec v0.0.0-20170113033406-39771216ff4c // indirect
+	github.com/morikuni/aec v1.0.0 // indirect
 	github.com/opencontainers/go-digest v1.0.0 // indirect
-	github.com/opencontainers/image-spec v1.0.2 // indirect
-	github.com/opencontainers/runc v1.0.2 // indirect
+	github.com/opencontainers/image-spec v1.0.3-0.20211202183452-c5a74bcca799 // indirect
+	github.com/opencontainers/runc v1.1.3 // indirect
 	github.com/sirupsen/logrus v1.8.1 // indirect
 	github.com/ugorji/go/codec v1.1.7 // indirect
 	go.opencensus.io v0.23.0 // indirect
@@ -105,7 +105,7 @@ require (
 	github.com/gobuffalo/packr v1.30.1 // indirect
 	github.com/golang/protobuf v1.5.2 // indirect
 	github.com/gorilla/mux v1.8.0
-	github.com/grpc-ecosystem/go-grpc-middleware v1.2.2 // indirect
+	github.com/grpc-ecosystem/go-grpc-middleware v1.3.0 // indirect
 	github.com/inconshreveable/mousetrap v1.0.0 // indirect
 	github.com/joho/godotenv v1.3.0 // indirect
 	github.com/klauspost/compress v1.15.0 // indirect
@@ -141,7 +141,7 @@ require (
 	golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a // indirect
 	golang.org/x/time v0.0.0-20210723032227-1f47c861a9ac // indirect
 	golang.org/x/tools v0.1.10 // indirect
-	google.golang.org/genproto v0.0.0-20210402141018-6c239bbf2bb1 // indirect
+	google.golang.org/genproto v0.0.0-20211208223120-3a66f561d7aa // indirect
 	google.golang.org/protobuf v1.27.1
 	gopkg.in/natefinch/lumberjack.v2 v2.0.0
 	k8s.io/apimachinery v0.23.5 // indirect

go.sum

@@ -163,12 +163,9 @@ func (executor *ReadWriteSplittingExecutor) ExecutorComQuery(
 	defer func() {
 		if err == nil {
 			result, err = decodeResult(result)
-			if err != nil {
-				span.RecordError(err)
-				return
-			}
-			err = executor.doPostFilter(spanCtx, result)
-		} else {
+		}
+		err = executor.doPostFilter(spanCtx, result, err)
+		if err != nil {
 			span.RecordError(err)
 		}
 	}()
@@ -307,12 +304,9 @@ func (executor *ReadWriteSplittingExecutor) ExecutorComStmtExecute(
 	defer func() {
 		if err == nil {
 			result, err = decodeResult(result)
-			if err != nil {
-				span.RecordError(err)
-				return
-			}
-			err = executor.doPostFilter(spanCtx, result)
-		} else {
+		}
+		err = executor.doPostFilter(spanCtx, result, err)
+		if err != nil {
 			span.RecordError(err)
 		}
 	}()
@@ -371,10 +365,10 @@ func (executor *ReadWriteSplittingExecutor) doPreFilter(ctx context.Context) err
 	return nil
 }
 
-func (executor *ReadWriteSplittingExecutor) doPostFilter(ctx context.Context, result proto.Result) error {
+func (executor *ReadWriteSplittingExecutor) doPostFilter(ctx context.Context, result proto.Result, err error) error {
 	for i := 0; i < len(executor.PostFilters); i++ {
 		f := executor.PostFilters[i]
-		err := f.PostHandle(ctx, result)
+		err := f.PostHandle(ctx, result, err)
 		if err != nil {
 			return err
 		}

pkg/executor/read_write_splitting.go

@@ -256,9 +256,7 @@ func (executor *ShardingExecutor) ExecutorComStmtExecute(
 		return nil, 0, err
 	}
 	defer func() {
-		if err == nil {
-			err = executor.doPostFilter(ctx, result)
-		}
+		err = executor.doPostFilter(ctx, result, err)
 	}()
 
 	var (
@@ -306,10 +304,10 @@ func (executor *ShardingExecutor) doPreFilter(ctx context.Context) error {
 	return nil
 }
 
-func (executor *ShardingExecutor) doPostFilter(ctx context.Context, result proto.Result) error {
+func (executor *ShardingExecutor) doPostFilter(ctx context.Context, result proto.Result, err error) error {
 	for i := 0; i < len(executor.PostFilters); i++ {
 		f := executor.PostFilters[i]
-		err := f.PostHandle(ctx, result)
+		err := f.PostHandle(ctx, result, err)
 		if err != nil {
 			return err
 		}

pkg/executor/sharding.go

@@ -136,12 +136,9 @@ func (executor *SingleDBExecutor) ExecutorComQuery(
 	defer func() {
 		if err == nil {
 			result, err = decodeResult(result)
-			if err != nil {
-				span.RecordError(err)
-				return
-			}
-			err = executor.doPostFilter(spanCtx, result)
-		} else {
+		}
+		err = executor.doPostFilter(spanCtx, result, err)
+		if err != nil {
 			span.RecordError(err)
 		}
 	}()
@@ -238,12 +235,9 @@ func (executor *SingleDBExecutor) ExecutorComStmtExecute(
 	defer func() {
 		if err == nil {
 			result, err = decodeResult(result)
-			if err != nil {
-				span.RecordError(err)
-				return
-			}
-			err = executor.doPostFilter(spanCtx, result)
-		} else {
+		}
+		err = executor.doPostFilter(spanCtx, result, err)
+		if err != nil {
 			span.RecordError(err)
 		}
 	}()
@@ -283,10 +277,10 @@ func (executor *SingleDBExecutor) doPreFilter(ctx context.Context) error {
 	return nil
 }
 
-func (executor *SingleDBExecutor) doPostFilter(ctx context.Context, result proto.Result) error {
+func (executor *SingleDBExecutor) doPostFilter(ctx context.Context, result proto.Result, err error) error {
 	for i := 0; i < len(executor.PostFilters); i++ {
 		f := executor.PostFilters[i]
-		err := f.PostHandle(ctx, result)
+		err := f.PostHandle(ctx, result, err)
 		if err != nil {
 			return err
 		}

pkg/executor/single_db.go

@@ -0,0 +1,165 @@
+/*
+ * Copyright 2022 CECTC, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package breaker
+
+import (
+	"context"
+	"encoding/json"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/pkg/errors"
+
+	"github.com/cectc/dbpack/pkg/filter"
+	"github.com/cectc/dbpack/pkg/log"
+	"github.com/cectc/dbpack/pkg/proto"
+)
+
+const circuitBreakFilter = "CircuitBreakerFilter"
+
+// ErrBreakerOpen is the error returned from PreHandle() when the function is not executed
+// because the breaker is currently open.
+var ErrBreakerOpen = errors.New("circuit breaker is open")
+
+const (
+	closed uint32 = iota
+	open
+	halfOpen
+)
+
+type _factory struct{}
+
+func (factory *_factory) NewFilter(_ string, config map[string]interface{}) (proto.Filter, error) {
+	var (
+		err     error
+		content []byte
+		conf    *CircuitBreakerConfig
+	)
+	if content, err = json.Marshal(config); err != nil {
+		return nil, errors.Wrap(err, "marshal circuit breaker filter config failed.")
+	}
+	if err = json.Unmarshal(content, &conf); err != nil {
+		log.Errorf("unmarshal circuit breaker filter failed, %v", err)
+		return nil, err
+	}
+
+	return &_filter{
+		errorThreshold:   conf.ErrorThreshold,
+		successThreshold: conf.SuccessThreshold,
+		timeout:          time.Duration(conf.Timeout) * time.Second,
+	}, nil
+}
+
+type _filter struct {
+	errorThreshold   int
+	successThreshold int
+	timeout          time.Duration
+
+	lock      sync.Mutex
+	state     uint32
+	errors    int
+	successes int
+	lastError time.Time
+}
+
+type CircuitBreakerConfig struct {
+	ErrorThreshold   int
+	SuccessThreshold int
+	Timeout          int
+}
+
+func (f *_filter) GetKind() string {
+	return circuitBreakFilter
+}
+
+func (f *_filter) PreHandle(ctx context.Context) error {
+	state := atomic.LoadUint32(&f.state)
+
+	if state == open {
+		return ErrBreakerOpen
+	}
+
+	return nil
+}
+
+func (f *_filter) PostHandle(ctx context.Context, result proto.Result, err error) error {
+	state := atomic.LoadUint32(&f.state)
+	if err == nil && state == closed {
+		return nil
+	}
+
+	f.lock.Lock()
+	defer f.lock.Unlock()
+
+	if err == nil {
+		if f.state == halfOpen {
+			f.successes++
+			if f.successes == f.successThreshold {
+				f.closeBreaker()
+			}
+		}
+	} else {
+		if f.errors > 0 {
+			expiry := f.lastError.Add(f.timeout)
+			if time.Now().After(expiry) {
+				f.errors = 0
+			}
+		}
+
+		switch f.state {
+		case closed:
+			f.errors++
+			if f.errors == f.errorThreshold {
+				f.openBreaker()
+			} else {
+				f.lastError = time.Now()
+			}
+		case halfOpen:
+			f.openBreaker()
+		}
+	}
+	return err
+}
+
+func (f *_filter) openBreaker() {
+	f.changeState(open)
+	go f.timer()
+}
+
+func (f *_filter) closeBreaker() {
+	f.changeState(closed)
+}
+
+func (f *_filter) timer() {
+	time.Sleep(f.timeout)
+
+	f.lock.Lock()
+	defer f.lock.Unlock()
+
+	f.changeState(halfOpen)
+}
+
+func (f *_filter) changeState(newState uint32) {
+	f.errors = 0
+	f.successes = 0
+	atomic.StoreUint32(&f.state, newState)
+}
+
+func init() {
+	filter.RegistryFilterFactory(circuitBreakFilter, &_factory{})
+}