Feat/nacos registry (#659)

* feat(nacos): add nacos service registry&discovery

* feat(nacos): add nacos service client

* fix

* fix

cmd/admin/admin.go

@@ -85,7 +85,7 @@ func Run(bootstrapPath string, addr string) error {
 	}
 
 	registryConf := discovery.GetServiceRegistry(context.Background())
-	serviceDiscovery, err := registry.InitDiscovery(registryConf.Name, registryConf.RootPath, "service", registryConf.Options)
+	serviceDiscovery, err := registry.InitDiscovery(registryConf.Name, registryConf.Options)
 	if err != nil {
 		log.Fatal("init service discovert failed: %v", err)
 		return err

conf/bootstrap.yaml

@@ -34,9 +34,9 @@ registry:
 #    name: nacos
 #    options:
 #      endpoints: "127.0.0.1:8848"
-#      namespace: arana
+#      namespace_id: arana
 #      group: arana
-#      contextPath: /nacos
+#      context_path: /nacos
 #      scheme: http
 #      username: nacos
 #      password: nacos

example/service_discovery/etcd/main.go

@@ -30,11 +30,12 @@ import (
 
 func main() {
 	storeType := base.ETCD
-	basePath := "arana"
 	options := make(map[string]interface{})
 	options["endpoints"] = "http://127.0.0.1:2379"
+	options["root_path"] = "arana"
+	options["service_path"] = "service"
 
-	etcdDiscovery, err := registry.InitDiscovery(storeType, basePath, "service", options)
+	etcdDiscovery, err := registry.InitDiscovery(storeType, options)
 	if err != nil {
 		log.Fatalf("Init %s discovery err:%v", storeType, err)
 		return

example/service_discovery/nacos/main.go

@@ -0,0 +1,41 @@
+/*
+ * 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 main
+
+import (
+	"github.com/arana-db/arana/pkg/registry"
+	"github.com/arana-db/arana/pkg/registry/base"
+	"github.com/arana-db/arana/pkg/util/log"
+)
+
+func main() {
+	storeType := base.NACOS
+	options := make(map[string]interface{})
+	options["endpoints"] = "127.0.0.1:8848"
+	options["scheme"] = "http"
+	options["username"] = "nacos"
+	options["password"] = "nacos"
+
+	nacosDiscovery, err := registry.InitDiscovery(base.NACOS, options)
+	if err != nil {
+		log.Fatalf("Init %s discovery err:%v", storeType, err)
+		return
+	}
+
+	nacosDiscovery.GetServices()
+}

pkg/admin/config.go

@@ -76,7 +76,7 @@ type ServiceInstanceDTO struct {
 	// Version is the version of the compiled.
 	Version string `json:"version"`
 	// Endpoint addresses of the service instance.
-	Endpoints []*config.Listener
+	Endpoint *config.Listener
 }
 
 // XConfigWriter represents the mutations of configurations.

pkg/admin/service_discovery.go

@@ -18,7 +18,6 @@
 package admin
 
 import (
-	"github.com/arana-db/arana/pkg/config"
 	"github.com/arana-db/arana/pkg/registry/base"
 )
 
@@ -36,13 +35,11 @@ func (mysds *myServiceDiscovery) ListServices() []*ServiceInstanceDTO {
 		srvDTOs  = make([]*ServiceInstanceDTO, 0, len(services))
 	)
 	for _, srv := range services {
-		endpoints := make([]*config.Listener, len(srv.Endpoints))
-		copy(endpoints, srv.Endpoints)
 		srvDTOs = append(srvDTOs, &ServiceInstanceDTO{
-			ID:        srv.ID,
-			Name:      srv.Name,
-			Version:   srv.Version,
-			Endpoints: endpoints,
+			ID:       srv.ID,
+			Name:     srv.Name,
+			Version:  srv.Version,
+			Endpoint: srv.Endpoint,
 		})
 	}
 	return srvDTOs

pkg/config/nacos/nacos.go

@@ -37,20 +37,7 @@ import (
 import (
 	"github.com/arana-db/arana/pkg/config"
 	"github.com/arana-db/arana/pkg/util/bytesconv"
-)
-
-const (
-	_defaultGroupName string = "arana"
-
-	_namespaceKey string = "namespace"
-	_groupKey     string = "group"
-	_username     string = "username"
-	_password     string = "password"
-	_server       string = "endpoints"
-	_contextPath  string = "contextPath"
-	_scheme       string = "scheme"
-
-	_pathSplit string = "::"
+	u_conf "github.com/arana-db/arana/pkg/util/config"
 )
 
 var (
@@ -84,13 +71,13 @@ func (s *storeOperate) Init(options map[string]interface{}) error {
 }
 
 func (s *storeOperate) initNacosClient(options map[string]interface{}) error {
-	s.groupName = _defaultGroupName
-	if val, ok := options[_groupKey]; ok {
+	s.groupName = u_conf.DefaultGroupName
+	if val, ok := options[u_conf.GroupKey]; ok {
 		s.groupName = val.(string)
 	}
 
-	clientConfig := parseClientConfig(options)
-	serverConfigs := parseServerConfig(options)
+	clientConfig := u_conf.ParseNacosClientConfig(options)
+	serverConfigs := u_conf.ParseNacosServerConfig(options)
 
 	// a more graceful way to create config client
 	client, err := clients.NewConfigClient(
@@ -110,15 +97,15 @@ func parseServerConfig(options map[string]interface{}) []constant.ServerConfig {
 	cfgs := make([]constant.ServerConfig, 0)
 
 	scheme := "http"
-	if val, ok := options[_scheme]; ok {
+	if val, ok := options[u_conf.Scheme]; ok {
 		scheme = val.(string)
 	}
 	contextPath := "/nacos"
-	if val, ok := options[_contextPath]; ok {
+	if val, ok := options[u_conf.ContextPath]; ok {
 		contextPath = val.(string)
 	}
 
-	if servers, ok := options[_server]; ok {
+	if servers, ok := options[u_conf.Server]; ok {
 		addresses := strings.Split(servers.(string), ",")
 		for i := range addresses {
 			addr := strings.Split(strings.TrimSpace(addresses[i]), ":")
@@ -141,13 +128,13 @@ func parseServerConfig(options map[string]interface{}) []constant.ServerConfig {
 func parseClientConfig(options map[string]interface{}) constant.ClientConfig {
 	cc := constant.ClientConfig{}
 
-	if val, ok := options[_namespaceKey]; ok {
+	if val, ok := options[u_conf.NamespaceIdKey]; ok {
 		cc.NamespaceId = val.(string)
 	}
-	if val, ok := options[_username]; ok {
+	if val, ok := options[u_conf.Username]; ok {
 		cc.Username = val.(string)
 	}
-	if val, ok := options[_password]; ok {
+	if val, ok := options[u_conf.Password]; ok {
 		cc.Password = val.(string)
 	}
 	return cc
@@ -272,9 +259,9 @@ func (w *nacosWatcher) run(ctx context.Context) {
 }
 
 func buildNacosDataId(v string) string {
-	return strings.ReplaceAll(v, "/", _pathSplit)
+	return strings.ReplaceAll(v, "/", u_conf.PathSplit)
 }
 
 func revertNacosDataId(v string) string {
-	return strings.ReplaceAll(v, _pathSplit, "/")
+	return strings.ReplaceAll(v, u_conf.PathSplit, "/")
 }

pkg/config/nacos/nacos_test.go

@@ -239,32 +239,3 @@ func Test_storeOpertae(t *testing.T) {
 
 	assert.NoError(t, err, "blank string should be success")
 }
-
-func Test_parseServerConfig(t *testing.T) {
-	// _namespaceKey string = "namespace"
-	// _groupKey     string = "group"
-	// _username     string = "username"
-	// _password     string = "password"
-	// _server       string = "endpoints"
-	// _contextPath  string = "contextPath"
-	// _scheme       string = "scheme"
-
-	options := map[string]interface{}{
-		_namespaceKey: "arana_test",
-		_groupKey:     "arana_test",
-		_username:     "nacos_test",
-		_password:     "nacos_test",
-		_server:       "127.0.0.1:8848,127.0.0.2:8848",
-	}
-
-	clientConfig := parseClientConfig(options)
-	assert.Equal(t, options[_namespaceKey], clientConfig.NamespaceId)
-	assert.Equal(t, options[_username], clientConfig.Username)
-	assert.Equal(t, options[_password], clientConfig.Password)
-
-	serverConfigs := parseServerConfig(options)
-	assert.Equal(t, 2, len(serverConfigs))
-
-	assert.Equal(t, "127.0.0.1", serverConfigs[0].IpAddr)
-	assert.Equal(t, "127.0.0.2", serverConfigs[1].IpAddr)
-}

pkg/registry/base/base.go

@@ -41,15 +41,15 @@ type ServiceInstance struct {
 	// Version is the version of the compiled.
 	Version string `json:"version"`
 	// Endpoint addresses of the service instance.
-	Endpoints []*config.Listener
+	Endpoint *config.Listener
 }
 
 func (p ServiceInstance) String() string {
-	return fmt.Sprintf("Service instance: id:%s, name:%s, version:%s, endpoints:%s", p.ID, p.Name, p.Version, p.Endpoints)
+	return fmt.Sprintf("Service instance: id:%s, name:%s, version:%s, endpoints:%s", p.ID, p.Name, p.Version, p.Endpoint)
 }
 
 type Registry interface {
-	Register(ctx context.Context, name string, serviceInstance *ServiceInstance) error
+	Register(ctx context.Context, serviceInstance *ServiceInstance) error
 	Unregister(ctx context.Context, name string) error
 	UnregisterAllService(ctx context.Context) error
 }

pkg/registry/discovery.go

@@ -19,6 +19,7 @@ package registry
 
 import (
 	"fmt"
+	"strings"
 )
 
 import (
@@ -28,16 +29,27 @@ import (
 import (
 	"github.com/arana-db/arana/pkg/registry/base"
 	"github.com/arana-db/arana/pkg/registry/etcd"
+	"github.com/arana-db/arana/pkg/registry/nacos"
 	"github.com/arana-db/arana/pkg/util/log"
 )
 
-func InitDiscovery(storeType string, basePath string, servicePath string, options map[string]interface{}) (base.Discovery, error) {
+const (
+	_rootPath    = "root_path"
+	_servicePath = "service_path"
+	_endpoints   = "endpoints"
+
+	_defaultServicePath = "service"
+	_defaultRootPath    = "arana"
+)
+
+func InitDiscovery(storeType string, options map[string]interface{}) (base.Discovery, error) {
 	var serviceDiscovery base.Discovery
 	var err error
 	switch storeType {
 	case base.ETCD:
-		serviceDiscovery, err = initEtcdDiscovery(basePath, servicePath, []string{options["endpoints"].(string)})
+		serviceDiscovery, err = initEtcdDiscovery(options)
 	case base.NACOS:
+		initNacosV2Discovery(options)
 	default:
 		err = errors.Errorf("Service registry not support store:%s", storeType)
 	}
@@ -50,19 +62,37 @@ func InitDiscovery(storeType string, basePath string, servicePath string, option
 	return serviceDiscovery, nil
 }
 
-func initEtcdDiscovery(basePath string, servicePath string, storeAddrs []string) (base.Discovery, error) {
+func initEtcdDiscovery(options map[string]interface{}) (base.Discovery, error) {
+	var (
+		rootPath    = _defaultRootPath
+		servicePath = _defaultServicePath
+		storeAddrs  = make([]string, 0)
+	)
+
+	if r, ok := options[_rootPath]; ok {
+		rootPath = r.(string)
+	}
+
+	if s, ok := options[_servicePath]; ok {
+		servicePath = s.(string)
+	}
+
+	if e, ok := options[_endpoints]; ok {
+		storeAddrs = append(storeAddrs, strings.Split(e.(string), ",")...)
+	}
+
 	if len(storeAddrs) == 0 {
 		return nil, fmt.Errorf("service discovery init etcd error because get endpoints nil :%v", storeAddrs)
 	}
 
-	serviceDiscovery, err := etcd.NewEtcdV3Discovery(basePath, servicePath, storeAddrs, nil)
+	serviceDiscovery, err := etcd.NewEtcdV3Discovery(rootPath, servicePath, storeAddrs, nil)
 	if err != nil {
 		return nil, fmt.Errorf("service discovery init etcd error because err: :%v", err)
 	}
 
 	return serviceDiscovery, nil
 }
 
-func initNacosDiscovery(basePath string, servicePath string, storeAddrs []string) (base.Discovery, error) {
-	return nil, nil
+func initNacosV2Discovery(options map[string]interface{}) (base.Discovery, error) {
+	return nacos.NewNacosV2Discovery(options)
 }

pkg/registry/etcd/discovery.go

@@ -36,7 +36,7 @@ import (
 	"github.com/arana-db/arana/pkg/util/log"
 )
 
-// EtcdV3Discovery is a etcd service discovery.
+// EtcdV3Discovery is an etcd service discovery.
 // It always returns the registered servers in etcd.
 type EtcdV3Discovery struct {
 	BasePath string

pkg/registry/etcd/registery.go

@@ -109,7 +109,8 @@ func NewEtcdV3Registry(serviceAddr, path string, etcdAddrs []string, options *st
 	return etcdRegistry, nil
 }
 
-func (r *EtcdV3Registry) Register(ctx context.Context, name string, serviceInstance *base.ServiceInstance) error {
+func (r *EtcdV3Registry) Register(ctx context.Context, serviceInstance *base.ServiceInstance) error {
+	name := serviceInstance.Name
 	if strings.TrimSpace(name) == "" {
 		return errors.New("Register service `name` can't be empty")
 	}