Merge pull request #748 from louyuting/feature/sentinel-integration

Ftr: add sentinel-golang  flow control/circuit breaker

filter/filter_impl/sentinel_filter.go

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 filter_impl
+
+import (
+	"context"
+	"fmt"
+	"strings"
+)
+
+import (
+	sentinel "github.com/alibaba/sentinel-golang/api"
+	"github.com/alibaba/sentinel-golang/core/base"
+	"github.com/alibaba/sentinel-golang/logging"
+)
+
+import (
+	"github.com/apache/dubbo-go/common"
+	"github.com/apache/dubbo-go/common/constant"
+	"github.com/apache/dubbo-go/common/extension"
+	"github.com/apache/dubbo-go/common/logger"
+	"github.com/apache/dubbo-go/filter"
+	"github.com/apache/dubbo-go/protocol"
+)
+
+func init() {
+	extension.SetFilter(SentinelProviderFilterName, GetSentinelProviderFilter)
+	extension.SetFilter(SentinelConsumerFilterName, GetSentinelConsumerFilter)
+
+	if err := sentinel.InitDefault(); err != nil {
+		logger.Errorf("[Sentinel Filter] fail to initialize Sentinel")
+	}
+	if err := logging.ResetGlobalLogger(DubboLoggerWrapper{Logger: logger.GetLogger()}); err != nil {
+		logger.Errorf("[Sentinel Filter] fail to ingest dubbo logger into sentinel")
+	}
+}
+
+type DubboLoggerWrapper struct {
+	logger.Logger
+}
+
+func (d DubboLoggerWrapper) Fatal(v ...interface{}) {
+	d.Logger.Error(v...)
+}
+
+func (d DubboLoggerWrapper) Fatalf(format string, v ...interface{}) {
+	d.Logger.Errorf(format, v...)
+}
+
+func (d DubboLoggerWrapper) Panic(v ...interface{}) {
+	d.Logger.Error(v...)
+}
+
+func (d DubboLoggerWrapper) Panicf(format string, v ...interface{}) {
+	d.Logger.Errorf(format, v...)
+}
+
+func GetSentinelConsumerFilter() filter.Filter {
+	return &SentinelConsumerFilter{}
+}
+
+func GetSentinelProviderFilter() filter.Filter {
+	return &SentinelProviderFilter{}
+}
+
+func sentinelExit(ctx context.Context, result protocol.Result) {
+	if methodEntry := ctx.Value(MethodEntryKey); methodEntry != nil {
+		e := methodEntry.(*base.SentinelEntry)
+		sentinel.TraceError(e, result.Error())
+		e.Exit()
+	}
+	if interfaceEntry := ctx.Value(InterfaceEntryKey); interfaceEntry != nil {
+		e := interfaceEntry.(*base.SentinelEntry)
+		sentinel.TraceError(e, result.Error())
+		e.Exit()
+	}
+}
+
+type SentinelProviderFilter struct{}
+
+func (d *SentinelProviderFilter) Invoke(ctx context.Context, invoker protocol.Invoker, invocation protocol.Invocation) protocol.Result {
+	interfaceResourceName, methodResourceName := getResourceName(invoker, invocation, getProviderPrefix())
+
+	var (
+		interfaceEntry *base.SentinelEntry
+		methodEntry    *base.SentinelEntry
+		b              *base.BlockError
+	)
+	interfaceEntry, b = sentinel.Entry(interfaceResourceName, sentinel.WithResourceType(base.ResTypeRPC), sentinel.WithTrafficType(base.Inbound))
+	if b != nil {
+		// interface blocked
+		return sentinelDubboProviderFallback(ctx, invoker, invocation, b)
+	}
+	ctx = context.WithValue(ctx, InterfaceEntryKey, interfaceEntry)
+
+	methodEntry, b = sentinel.Entry(methodResourceName,
+		sentinel.WithResourceType(base.ResTypeRPC),
+		sentinel.WithTrafficType(base.Inbound),
+		sentinel.WithArgs(invocation.Arguments()...))
+	if b != nil {
+		// method blocked
+		return sentinelDubboProviderFallback(ctx, invoker, invocation, b)
+	}
+	ctx = context.WithValue(ctx, MethodEntryKey, methodEntry)
+	return invoker.Invoke(ctx, invocation)
+}
+
+func (d *SentinelProviderFilter) OnResponse(ctx context.Context, result protocol.Result, _ protocol.Invoker, _ protocol.Invocation) protocol.Result {
+	sentinelExit(ctx, result)
+	return result
+}
+
+type SentinelConsumerFilter struct{}
+
+func (d *SentinelConsumerFilter) Invoke(ctx context.Context, invoker protocol.Invoker, invocation protocol.Invocation) protocol.Result {
+	interfaceResourceName, methodResourceName := getResourceName(invoker, invocation, getConsumerPrefix())
+	var (
+		interfaceEntry *base.SentinelEntry
+		methodEntry    *base.SentinelEntry
+		b              *base.BlockError
+	)
+
+	interfaceEntry, b = sentinel.Entry(interfaceResourceName, sentinel.WithResourceType(base.ResTypeRPC), sentinel.WithTrafficType(base.Outbound))
+	if b != nil {
+		// interface blocked
+		return sentinelDubboConsumerFallback(ctx, invoker, invocation, b)
+	}
+	ctx = context.WithValue(ctx, InterfaceEntryKey, interfaceEntry)
+
+	methodEntry, b = sentinel.Entry(methodResourceName, sentinel.WithResourceType(base.ResTypeRPC),
+		sentinel.WithTrafficType(base.Outbound), sentinel.WithArgs(invocation.Arguments()...))
+	if b != nil {
+		// method blocked
+		return sentinelDubboConsumerFallback(ctx, invoker, invocation, b)
+	}
+	ctx = context.WithValue(ctx, MethodEntryKey, methodEntry)
+
+	return invoker.Invoke(ctx, invocation)
+}
+
+func (d *SentinelConsumerFilter) OnResponse(ctx context.Context, result protocol.Result, _ protocol.Invoker, _ protocol.Invocation) protocol.Result {
+	sentinelExit(ctx, result)
+	return result
+}
+
+var (
+	sentinelDubboConsumerFallback = getDefaultDubboFallback()
+	sentinelDubboProviderFallback = getDefaultDubboFallback()
+)
+
+type DubboFallback func(context.Context, protocol.Invoker, protocol.Invocation, *base.BlockError) protocol.Result
+
+func SetDubboConsumerFallback(f DubboFallback) {
+	sentinelDubboConsumerFallback = f
+}
+func SetDubboProviderFallback(f DubboFallback) {
+	sentinelDubboProviderFallback = f
+}
+func getDefaultDubboFallback() DubboFallback {
+	return func(ctx context.Context, invoker protocol.Invoker, invocation protocol.Invocation, blockError *base.BlockError) protocol.Result {
+		result := &protocol.RPCResult{}
+		result.SetResult(nil)
+		result.SetError(blockError)
+		return result
+	}
+}
+
+const (
+	SentinelProviderFilterName = "sentinel-provider"
+	SentinelConsumerFilterName = "sentinel-consumer"
+
+	DefaultProviderPrefix = "dubbo:provider:"
+	DefaultConsumerPrefix = "dubbo:consumer:"
+
+	MethodEntryKey    = "$$sentinelMethodEntry"
+	InterfaceEntryKey = "$$sentinelInterfaceEntry"
+)
+
+func getResourceName(invoker protocol.Invoker, invocation protocol.Invocation, prefix string) (interfaceResourceName, methodResourceName string) {
+	var sb strings.Builder
+
+	sb.WriteString(prefix)
+	if getInterfaceGroupAndVersionEnabled() {
+		interfaceResourceName = getColonSeparatedKey(invoker.GetUrl())
+	} else {
+		interfaceResourceName = invoker.GetUrl().Service()
+	}
+	sb.WriteString(interfaceResourceName)
+	sb.WriteString(":")
+	sb.WriteString(invocation.MethodName())
+	sb.WriteString("(")
+	isFirst := true
+	for _, v := range invocation.ParameterTypes() {
+		if !isFirst {
+			sb.WriteString(",")
+		}
+		sb.WriteString(v.Name())
+		isFirst = false
+	}
+	sb.WriteString(")")
+	methodResourceName = sb.String()
+	return
+}
+
+func getConsumerPrefix() string {
+	return DefaultConsumerPrefix
+}
+
+func getProviderPrefix() string {
+	return DefaultProviderPrefix
+}
+
+func getInterfaceGroupAndVersionEnabled() bool {
+	return true
+}
+
+func getColonSeparatedKey(url common.URL) string {
+	return fmt.Sprintf("%s:%s:%s",
+		url.Service(),
+		url.GetParam(constant.GROUP_KEY, ""),
+		url.GetParam(constant.VERSION_KEY, ""))
+}

filter/filter_impl/sentinel_filter_test.go

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 filter_impl
+
+import (
+	"context"
+	"sync"
+	"sync/atomic"
+	"testing"
+)
+
+import (
+	"github.com/alibaba/sentinel-golang/core/flow"
+	"github.com/stretchr/testify/assert"
+)
+
+import (
+	"github.com/apache/dubbo-go/common"
+	"github.com/apache/dubbo-go/protocol"
+	"github.com/apache/dubbo-go/protocol/invocation"
+)
+
+func TestSentinelFilter_QPS(t *testing.T) {
+	url, err := common.NewURL("dubbo://127.0.0.1:20000/UserProvider?anyhost=true&" +
+		"version=1.0.0&group=myGroup&" +
+		"application=BDTService&category=providers&default.timeout=10000&dubbo=dubbo-provider-golang-1.0.0&" +
+		"environment=dev&interface=com.ikurento.user.UserProvider&ip=192.168.56.1&methods=GetUser%2C&" +
+		"module=dubbogo+user-info+server&org=ikurento.com&owner=ZX&pid=1447&revision=0.0.1&" +
+		"side=provider&timeout=3000&timestamp=1556509797245&bean.name=UserProvider")
+	assert.NoError(t, err)
+	mockInvoker := protocol.NewBaseInvoker(url)
+	interfaceResourceName, _ := getResourceName(mockInvoker,
+		invocation.NewRPCInvocation("hello", []interface{}{"OK"}, make(map[string]interface{})), "prefix_")
+	mockInvocation := invocation.NewRPCInvocation("hello", []interface{}{"OK"}, make(map[string]interface{}))
+
+	_, err = flow.LoadRules([]*flow.Rule{
+		{
+			Resource:               interfaceResourceName,
+			MetricType:             flow.QPS,
+			TokenCalculateStrategy: flow.Direct,
+			ControlBehavior:        flow.Reject,
+			Count:                  100,
+			RelationStrategy:       flow.CurrentResource,
+		},
+	})
+	assert.NoError(t, err)
+
+	wg := sync.WaitGroup{}
+	wg.Add(10)
+	f := GetSentinelProviderFilter()
+	pass := int64(0)
+	block := int64(0)
+	for i := 0; i < 10; i++ {
+		go func() {
+			for j := 0; j < 30; j++ {
+				result := f.Invoke(context.TODO(), mockInvoker, mockInvocation)
+				if result.Error() == nil {
+					atomic.AddInt64(&pass, 1)
+				} else {
+					atomic.AddInt64(&block, 1)
+				}
+			}
+			wg.Done()
+		}()
+	}
+	wg.Wait()
+	assert.True(t, atomic.LoadInt64(&pass) == 100)
+	assert.True(t, atomic.LoadInt64(&block) == 200)
+}
+
+func TestConsumerFilter_Invoke(t *testing.T) {
+	f := GetSentinelConsumerFilter()
+	url, err := common.NewURL("dubbo://127.0.0.1:20000/UserProvider?anyhost=true&" +
+		"application=BDTService&category=providers&default.timeout=10000&dubbo=dubbo-provider-golang-1.0.0&" +
+		"environment=dev&interface=com.ikurento.user.UserProvider&ip=192.168.56.1&methods=GetUser%2C&" +
+		"module=dubbogo+user-info+server&org=ikurento.com&owner=ZX&pid=1447&revision=0.0.1&" +
+		"side=provider&timeout=3000&timestamp=1556509797245&bean.name=UserProvider")
+	assert.NoError(t, err)
+	mockInvoker := protocol.NewBaseInvoker(url)
+	mockInvocation := invocation.NewRPCInvocation("hello", []interface{}{"OK"}, make(map[string]interface{}))
+	result := f.Invoke(context.TODO(), mockInvoker, mockInvocation)
+	assert.NoError(t, result.Error())
+}
+
+func TestProviderFilter_Invoke(t *testing.T) {
+	f := GetSentinelProviderFilter()
+	url, err := common.NewURL("dubbo://127.0.0.1:20000/UserProvider?anyhost=true&" +
+		"application=BDTService&category=providers&default.timeout=10000&dubbo=dubbo-provider-golang-1.0.0&" +
+		"environment=dev&interface=com.ikurento.user.UserProvider&ip=192.168.56.1&methods=GetUser%2C&" +
+		"module=dubbogo+user-info+server&org=ikurento.com&owner=ZX&pid=1447&revision=0.0.1&" +
+		"side=provider&timeout=3000&timestamp=1556509797245&bean.name=UserProvider")
+	assert.NoError(t, err)
+	mockInvoker := protocol.NewBaseInvoker(url)
+	mockInvocation := invocation.NewRPCInvocation("hello", []interface{}{"OK"}, make(map[string]interface{}))
+	result := f.Invoke(context.TODO(), mockInvoker, mockInvocation)
+	assert.NoError(t, result.Error())
+}
+
+func TestGetResourceName(t *testing.T) {
+	url, err := common.NewURL("dubbo://127.0.0.1:20000/UserProvider?anyhost=true&" +
+		"version=1.0.0&group=myGroup&" +
+		"application=BDTService&category=providers&default.timeout=10000&dubbo=dubbo-provider-golang-1.0.0&" +
+		"environment=dev&interface=com.ikurento.user.UserProvider&ip=192.168.56.1&methods=GetUser%2C&" +
+		"module=dubbogo+user-info+server&org=ikurento.com&owner=ZX&pid=1447&revision=0.0.1&" +
+		"side=provider&timeout=3000&timestamp=1556509797245&bean.name=UserProvider")
+	assert.NoError(t, err)
+	mockInvoker := protocol.NewBaseInvoker(url)
+	interfaceResourceName, methodResourceName := getResourceName(mockInvoker,
+		invocation.NewRPCInvocation("hello", []interface{}{"OK"}, make(map[string]interface{})), "prefix_")
+	assert.Equal(t, "com.ikurento.user.UserProvider:myGroup:1.0.0", interfaceResourceName)
+	assert.Equal(t, "prefix_com.ikurento.user.UserProvider:myGroup:1.0.0:hello()", methodResourceName)
+}

go.mod

 module github.com/apache/dubbo-go
 
 require (
-	github.com/Microsoft/go-winio v0.4.13 // indirect
 	github.com/NYTimes/gziphandler v1.1.1 // indirect
 	github.com/RoaringBitmap/roaring v0.4.23
 	github.com/Workiva/go-datastructures v1.0.50
 	github.com/afex/hystrix-go v0.0.0-20180502004556-fa1af6a1f4f5
+	github.com/alibaba/sentinel-golang v0.6.1
 	github.com/apache/dubbo-getty v1.3.10
 	github.com/apache/dubbo-go-hessian2 v1.6.0-rc1.0.20200906044240-6c1fb5c3bd44
-	github.com/coreos/bbolt v1.3.3 // indirect
-	github.com/coreos/etcd v3.3.13+incompatible
-	github.com/coreos/go-systemd v0.0.0-20190719114852-fd7a80b32e1f // indirect
+	github.com/coreos/etcd v3.3.25+incompatible
 	github.com/creasty/defaults v1.3.0
-	github.com/docker/go-connections v0.4.0 // indirect
 	github.com/dubbogo/go-zookeeper v1.0.1
 	github.com/dubbogo/gost v1.9.1
 	github.com/elazarl/go-bindata-assetfs v1.0.0 // indirect
@@ -22,22 +19,16 @@ require (
 	github.com/ghodss/yaml v1.0.1-0.20190212211648-25d852aebe32 // indirect
 	github.com/go-co-op/gocron v0.1.1
 	github.com/go-resty/resty/v2 v2.1.0
-	github.com/golang/groupcache v0.0.0-20190702054246-869f871628b6 // indirect
 	github.com/golang/mock v1.3.1
-	github.com/golang/protobuf v1.3.2
-	github.com/google/go-cmp v0.3.1 // indirect
-	github.com/grpc-ecosystem/go-grpc-middleware v1.0.1-0.20190118093823-f849b5445de4 // indirect
-	github.com/grpc-ecosystem/grpc-gateway v1.9.5 // indirect
+	github.com/golang/protobuf v1.4.0
 	github.com/grpc-ecosystem/grpc-opentracing v0.0.0-20180507213350-8e809c8a8645
 	github.com/hashicorp/consul v1.8.0
 	github.com/hashicorp/consul/api v1.5.0
 	github.com/hashicorp/go-raftchunking v0.6.3-0.20191002164813-7e9e8525653a // indirect
-	github.com/hashicorp/golang-lru v0.5.3 // indirect
 	github.com/hashicorp/vault/api v1.0.5-0.20191108163347-bdd38fca2cff // indirect
 	github.com/hashicorp/vault/sdk v0.1.14-0.20191112033314-390e96e22eb2
 	github.com/jinzhu/copier v0.0.0-20190625015134-976e0346caa8
 	github.com/magiconair/properties v1.8.1
-	github.com/mitchellh/hashstructure v1.0.0 // indirect
 	github.com/mitchellh/mapstructure v1.2.3
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd
 	github.com/nacos-group/nacos-sdk-go v1.0.0
@@ -46,14 +37,12 @@ require (
 	github.com/pkg/errors v0.9.1
 	github.com/prometheus/client_golang v1.1.0
 	github.com/satori/go.uuid v1.2.1-0.20181028125025-b2ce2384e17b
-	github.com/shirou/gopsutil v2.19.9+incompatible // indirect
 	github.com/stretchr/objx v0.2.0 // indirect
 	github.com/stretchr/testify v1.5.1
 	github.com/zouyx/agollo/v3 v3.4.4
-	go.etcd.io/bbolt v1.3.4 // indirect
 	go.uber.org/atomic v1.6.0
 	go.uber.org/zap v1.15.0
-	google.golang.org/grpc v1.23.0
+	google.golang.org/grpc v1.26.0
 	gopkg.in/yaml.v2 v2.2.8
 	k8s.io/api v0.16.9
 	k8s.io/apimachinery v0.16.9
@@ -65,3 +54,5 @@ require (
 go 1.13
 
 replace launchpad.net/gocheck => github.com/go-check/check v0.0.0-20140225173054-eb6ee6f84d0a
+
+replace github.com/envoyproxy/go-control-plane => github.com/envoyproxy/go-control-plane v0.8.0