Nacos datasource

ext/datasource/nacos/nacos.go

@@ -0,0 +1,115 @@
+package nacos
+
+import (
+	"github.com/alibaba/sentinel-golang/ext/datasource"
+	"github.com/alibaba/sentinel-golang/logging"
+	"github.com/alibaba/sentinel-golang/util"
+	"github.com/nacos-group/nacos-sdk-go/clients/config_client"
+	"github.com/nacos-group/nacos-sdk-go/clients/nacos_client"
+	"github.com/nacos-group/nacos-sdk-go/common/constant"
+	"github.com/nacos-group/nacos-sdk-go/common/http_agent"
+	"github.com/nacos-group/nacos-sdk-go/vo"
+	"github.com/pkg/errors"
+)
+
+var (
+	logger = logging.GetDefaultLogger()
+)
+
+type ConfigParam struct {
+	DataId string
+	Group  string
+}
+
+var clientConfig = constant.ClientConfig{
+	TimeoutMs:      10 * 1000,
+	ListenInterval: 30 * 1000,
+	BeatInterval:   5 * 1000,
+	LogDir:         "/nacos/logs",
+	CacheDir:       "/nacos/cache",
+}
+
+type NacosDataSource struct {
+	datasource.Base
+	serverConfig  constant.ServerConfig
+	clientConfig  constant.ClientConfig
+	client        *config_client.ConfigClient
+	isInitialized util.AtomicBool
+	configParam   ConfigParam
+	listener      *vo.ConfigParam
+}
+
+func NewNacosDataSource(clientConfig constant.ClientConfig, serverConfig constant.ServerConfig, configParam ConfigParam, handlers ...datasource.PropertyHandler) (*NacosDataSource, error) {
+	if len(serverConfig.IpAddr) <= 0 || serverConfig.Port <= 0 {
+		return nil, errors.New("The nacos serverConfig is incorrect.")
+	}
+	var ds = &NacosDataSource{
+		clientConfig: clientConfig,
+		serverConfig: serverConfig,
+		configParam:  configParam,
+	}
+	for _, h := range handlers {
+		ds.AddPropertyHandler(h)
+	}
+	return ds, nil
+}
+
+func (s *NacosDataSource) Initialize() error {
+	if !s.isInitialized.CompareAndSet(false, true) {
+		return nil
+	}
+	nc, err := buildNacosClient(s)
+	client, err := config_client.NewConfigClient(nc)
+	if err != nil {
+		return errors.Errorf("Nacosclient failed to build, err: %+v", err)
+	}
+	s.client = &client
+	err = s.listen(s.client)
+	return err
+}
+
+func buildNacosClient(s *NacosDataSource) (nacos_client.INacosClient, error) {
+	nc := nacos_client.NacosClient{}
+	nc.SetServerConfig([]constant.ServerConfig{s.serverConfig})
+	err := nc.SetClientConfig(s.clientConfig)
+	err = nc.SetHttpAgent(&http_agent.HttpAgent{})
+	return &nc, err
+}
+func (s *NacosDataSource) ReadSource() ([]byte, error) {
+	content, err := s.client.GetConfig(vo.ConfigParam{
+		DataId: s.configParam.DataId,
+		Group:  s.configParam.Group,
+	})
+	if err != nil {
+		return nil, errors.Errorf("Failed to read the nacos data source, err: %+v", err)
+	}
+	return []byte(content), err
+}
+
+func (s *NacosDataSource) listen(client *config_client.ConfigClient) (err error) {
+	s.listener = &vo.ConfigParam{
+		DataId: s.configParam.DataId,
+		Group:  s.configParam.Group,
+		OnChange: func(namespace, group, dataId, data string) {
+			if err := s.Handle([]byte(data)); err != nil {
+				logger.Errorf("Fail to update data for dataId:[%s] group:[%s] namespaceId:[%s]  when execute "+
+					"listen function, err: %+v", data, group, namespace, err)
+			}
+
+		},
+		ListenCloseChan: make(chan struct{}, 1),
+	}
+	err = client.ListenConfig(*s.listener)
+	if err != nil {
+		return errors.Errorf("Failed to listen to the nacos data source, err: %+v", err)
+	}
+	return
+}
+
+func (s *NacosDataSource) Close() error {
+	s.listener.ListenCloseChan <- struct{}{}
+
+	logger.Infof("The RefreshableFileDataSource   had been closed. DataId:[%s],Group:[%s]",
+		s.configParam.DataId, s.configParam.Group)
+	return nil
+}

ext/datasource/nacos/nacos_test.go

@@ -0,0 +1,166 @@
+package nacos
+
+import (
+	"strings"
+	"testing"
+
+	"github.com/alibaba/sentinel-golang/ext/datasource"
+	"github.com/nacos-group/nacos-sdk-go/clients/config_client"
+	"github.com/nacos-group/nacos-sdk-go/clients/nacos_client"
+	"github.com/nacos-group/nacos-sdk-go/common/constant"
+	"github.com/nacos-group/nacos-sdk-go/common/http_agent"
+	"github.com/nacos-group/nacos-sdk-go/vo"
+	"github.com/stretchr/testify/assert"
+	tmock "github.com/stretchr/testify/mock"
+)
+
+const (
+	TestSystemRules = `[
+    {
+        "id": 0,
+        "metricType": 0,
+        "adaptiveStrategy": 0
+    },
+    {
+        "id": 1,
+        "metricType": 0,
+        "adaptiveStrategy": 0
+    },
+    {
+        "id": 2,
+        "metricType": 0,
+        "adaptiveStrategy": 0
+    }
+]`
+)
+
+var serverConfig = constant.ServerConfig{
+	ContextPath: "/nacos",
+	Port:        80,
+	IpAddr:      "console.nacos.io",
+}
+var serverConfigErr = constant.ServerConfig{
+	ContextPath: "/nacos",
+	IpAddr:      "console.nacos.io1",
+}
+var clientConfigTest = constant.ClientConfig{
+	TimeoutMs:      10000,
+	ListenInterval: 20000,
+	BeatInterval:   10000,
+}
+
+var configParam = ConfigParam{
+	DataId: "system-rules",
+	Group:  "sentinel-go",
+}
+
+func cretateConfigClientTest() (*config_client.ConfigClient, error) {
+	nc := nacos_client.NacosClient{}
+	err := nc.SetServerConfig([]constant.ServerConfig{serverConfig})
+	err = nc.SetClientConfig(constant.ClientConfig{
+		TimeoutMs:      10000,
+		ListenInterval: 20000,
+		BeatInterval:   10000,
+	})
+	err = nc.SetHttpAgent(&http_agent.HttpAgent{})
+	client, err := config_client.NewConfigClient(&nc)
+
+	return &client, err
+}
+
+func prePushSystemRules(client *config_client.ConfigClient, content string) (bool, error) {
+	success, err := client.PublishConfig(vo.ConfigParam{
+		DataId:  configParam.DataId,
+		Group:   configParam.Group,
+		Content: content})
+
+	return success, err
+}
+func getNacosDataSource(clientConfigTest constant.ClientConfig, serverConfig constant.ServerConfig, configParam ConfigParam) (*NacosDataSource, error) {
+	mh1 := &datasource.MockPropertyHandler{}
+	mh1.On("Handle", tmock.Anything).Return(nil)
+	mh1.On("isPropertyConsistent", tmock.Anything).Return(false)
+	nds, err := NewNacosDataSource(clientConfigTest, serverConfig, configParam, mh1)
+
+	return nds, err
+}
+
+func TestNewNacosDataSource(t *testing.T) {
+	t.Run("NewNacosDataSource", func(t *testing.T) {
+		nds, err := getNacosDataSource(clientConfigTest, serverConfig, configParam)
+		assert.True(t, nds != nil && err == nil, "New NacosDataSource success.")
+	})
+	t.Run("NewNacosDataSourceErr", func(t *testing.T) {
+		mh1 := &datasource.MockPropertyHandler{}
+		nds, err := NewNacosDataSource(clientConfigTest, serverConfigErr, configParam, mh1)
+		assert.True(t, nds == nil && err != nil && strings.Contains(err.Error(), "The nacos serverConfig is incorrect."), "New NacosDataSource failed.")
+	})
+}
+
+func TestNacosDataSource_Initialize(t *testing.T) {
+
+	t.Run("NacosDataSource_Initialize_BuildNacosClient", func(t *testing.T) {
+		client, err := cretateConfigClientTest()
+		assert.Nil(t, err)
+		published, err := prePushSystemRules(client, TestSystemRules)
+		assert.True(t, err == nil && published, "Push systemRules configuration is successful.")
+
+		nds, err := getNacosDataSource(clientConfigTest, serverConfig, configParam)
+		assert.True(t, err == nil, "New NacosDataSource success.")
+		err = nds.Initialize()
+		assert.True(t, err == nil, "NacosDataSource initialize.")
+	})
+
+	t.Run("NacosDataSource_Initialize_BuildNacosClientErr", func(t *testing.T) {
+		client, err := cretateConfigClientTest()
+		assert.Nil(t, err)
+		published, err := prePushSystemRules(client, TestSystemRules)
+		assert.True(t, err == nil && published, "Push systemRules configuration is successful.")
+
+		clientConfigTest.TimeoutMs = 0
+		nds, err := getNacosDataSource(clientConfigTest, serverConfig, configParam)
+		assert.True(t, err == nil, "New NacosDataSource success.")
+		err = nds.Initialize()
+		assert.True(t, err != nil && strings.Contains(err.Error(), "Nacosclient failed to build"), "NacosDataSource failed.")
+	})
+}
+
+func TestNacosDataSource_ReadSource(t *testing.T) {
+	t.Run("NacosDataSource_ReadSource", func(t *testing.T) {
+		client, err := cretateConfigClientTest()
+		assert.Nil(t, err)
+		published, err := prePushSystemRules(client, TestSystemRules)
+		assert.True(t, err == nil && published, "Push systemRules configuration is successful.")
+
+		nds, err := getNacosDataSource(constant.ClientConfig{
+			TimeoutMs:      10000,
+			ListenInterval: 20000,
+			BeatInterval:   10000,
+		}, serverConfig, configParam)
+		assert.True(t, err == nil, "New NacosDataSource success.")
+		err = nds.Initialize()
+		assert.True(t, err == nil, "NacosDataSource initialize.")
+
+		data, err := nds.ReadSource()
+		assert.True(t, data != nil && err == nil, "NacosDataSource read source success.")
+	})
+}
+
+func TestNacosDataSource_Close(t *testing.T) {
+	client, err := cretateConfigClientTest()
+	assert.Nil(t, err)
+	published, err := prePushSystemRules(client, TestSystemRules)
+	assert.True(t, err == nil && published, "Push systemRules configuration is successful.")
+
+	nds, err := getNacosDataSource(constant.ClientConfig{
+		TimeoutMs:      10000,
+		ListenInterval: 20000,
+		BeatInterval:   10000,
+	}, serverConfig, configParam)
+	assert.True(t, err == nil, "New NacosDataSource success.")
+	err = nds.Initialize()
+	assert.True(t, err == nil, "NacosDataSource initialize.")
+
+	err = nds.Close()
+	assert.Nil(t, err)
+}

go.mod

@@ -4,20 +4,21 @@ go 1.13
 
 require (
 	github.com/StackExchange/wmi v0.0.0-20190523213315-cbe66965904d // indirect
-	github.com/apache/dubbo-go v0.1.2-0.20200224151332-dd1a3c24d656
+	github.com/apache/dubbo-go v1.4.2
 	github.com/coreos/etcd v3.3.13+incompatible
 	github.com/fsnotify/fsnotify v1.4.7
 	github.com/gin-gonic/gin v1.5.0
 	github.com/go-ole/go-ole v1.2.4 // indirect
 	github.com/google/uuid v1.1.1
 	github.com/hashicorp/consul/api v1.4.0
 	github.com/labstack/echo/v4 v4.1.15
-	github.com/pkg/errors v0.8.1
+	github.com/nacos-group/nacos-sdk-go v0.4.0
+	github.com/pkg/errors v0.9.1
 	github.com/shirou/gopsutil v2.19.12+incompatible
-	github.com/stretchr/testify v1.4.0
+	github.com/stretchr/testify v1.5.1
+	github.com/tidwall/gjson v1.6.0
 	go.uber.org/multierr v1.5.0
 	golang.org/x/tools v0.0.0-20200426102838-f3a5411a4c3b // indirect
 	google.golang.org/grpc v1.22.1
-	github.com/tidwall/gjson v1.6.0
 	gopkg.in/yaml.v2 v2.2.8
 )