feat: new feature for service discovery (#574)

* feat: new feature for service discovery

1.Completed the code for organizing the workflow.
2.Implemented FileRegistryService based on local configuration files.

* style: go imports

* refactor: avoid the uncertainty of SQL field order in unit test.

refactor select_for_update_executor_test.go to avoid the uncertainty of SQL field order causing test code to pass and fail sometimes.

pkg/client/client.go

@@ -23,6 +23,7 @@ import (
 	"github.com/seata/seata-go/pkg/datasource"
 	at "github.com/seata/seata-go/pkg/datasource/sql"
 	"github.com/seata/seata-go/pkg/datasource/sql/exec/config"
+	"github.com/seata/seata-go/pkg/discovery"
 	"github.com/seata/seata-go/pkg/integration"
 	remoteConfig "github.com/seata/seata-go/pkg/remoting/config"
 	"github.com/seata/seata-go/pkg/remoting/getty"
@@ -41,7 +42,7 @@ func Init() {
 // InitPath init client with config path
 func InitPath(configFilePath string) {
 	cfg := LoadPath(configFilePath)
-
+	initRegistry(cfg)
 	initRmClient(cfg)
 	initTmClient(cfg)
 	initDatasource()
@@ -51,6 +52,7 @@ var (
 	onceInitTmClient   sync.Once
 	onceInitRmClient   sync.Once
 	onceInitDatasource sync.Once
+	onceInitRegistry   sync.Once
 )
 
 // InitTmClient init client tm client
@@ -95,3 +97,9 @@ func initDatasource() {
 		datasource.Init()
 	})
 }
+
+func initRegistry(cfg *Config) {
+	onceInitRegistry.Do(func() {
+		discovery.InitRegistry(&cfg.ServiceConfig, &cfg.RegistryConfig)
+	})
+}

pkg/client/config.go

@@ -31,6 +31,7 @@ import (
 	"github.com/knadh/koanf/parsers/toml"
 	"github.com/knadh/koanf/parsers/yaml"
 	"github.com/knadh/koanf/providers/rawbytes"
+	"github.com/seata/seata-go/pkg/discovery"
 
 	"github.com/seata/seata-go/pkg/datasource/sql"
 	"github.com/seata/seata-go/pkg/datasource/sql/undo"
@@ -81,7 +82,8 @@ type Config struct {
 	ClientConfig      ClientConfig                 `yaml:"client" json:"client" koanf:"client"`
 	GettyConfig       remoteConfig.Config          `yaml:"getty" json:"getty" koanf:"getty"`
 	TransportConfig   remoteConfig.TransportConfig `yaml:"transport" json:"transport" koanf:"transport"`
-	ServiceConfig     tm.ServiceConfig             `yaml:"service" json:"service" koanf:"service"`
+	ServiceConfig     discovery.ServiceConfig      `yaml:"service" json:"service" koanf:"service"`
+	RegistryConfig    discovery.RegistryConfig     `yaml:"registry" json:"registry" koanf:"registry"`
 }
 
 func (c *Config) RegisterFlags(f *flag.FlagSet) {
@@ -98,6 +100,7 @@ func (c *Config) RegisterFlags(f *flag.FlagSet) {
 	c.ClientConfig.RegisterFlagsWithPrefix("client", f)
 	c.GettyConfig.RegisterFlagsWithPrefix("getty", f)
 	c.TransportConfig.RegisterFlagsWithPrefix("transport", f)
+	c.RegistryConfig.RegisterFlagsWithPrefix("registry", f)
 	c.ServiceConfig.RegisterFlagsWithPrefix("service", f)
 }
 

pkg/client/config_test.go

@@ -112,6 +112,20 @@ func TestLoadPath(t *testing.T) {
 	assert.Equal(t, "default", cfg.ServiceConfig.VgroupMapping["default_tx_group"])
 	assert.Equal(t, "127.0.0.1:8091", cfg.ServiceConfig.Grouplist["default"])
 
+	assert.NotNil(t, cfg.RegistryConfig)
+	assert.Equal(t, "file", cfg.RegistryConfig.Type)
+	assert.Equal(t, "seatago.yml", cfg.RegistryConfig.File.Name)
+	assert.Equal(t, "seata-server", cfg.RegistryConfig.Nacos.Application)
+	assert.Equal(t, "127.0.0.1:8848", cfg.RegistryConfig.Nacos.ServerAddr)
+	assert.Equal(t, "SEATA_GROUP", cfg.RegistryConfig.Nacos.Group)
+	assert.Equal(t, "test-namespace", cfg.RegistryConfig.Nacos.Namespace)
+	assert.Equal(t, "test-username", cfg.RegistryConfig.Nacos.Username)
+	assert.Equal(t, "test-password", cfg.RegistryConfig.Nacos.Password)
+	assert.Equal(t, "test-access-key", cfg.RegistryConfig.Nacos.AccessKey)
+	assert.Equal(t, "test-secret-key", cfg.RegistryConfig.Nacos.SecretKey)
+	assert.Equal(t, "default", cfg.RegistryConfig.Etcd3.Cluster)
+	assert.Equal(t, "http://localhost:2379", cfg.RegistryConfig.Etcd3.ServerAddr)
+
 	// reset flag.CommandLine
 	flag.CommandLine = flag.NewFlagSet(os.Args[0], flag.ExitOnError)
 }

pkg/datasource/sql/exec/at/select_for_update_executor_test.go

@@ -44,7 +44,8 @@ func TestBuildSelectPKSQL(t *testing.T) {
 	ctx, err := parser.DoParser(sql)
 
 	metaData := types.TableMeta{
-		TableName: "t_user",
+		TableName:   "t_user",
+		ColumnNames: []string{"id", "order_id", "age"},
 		Indexs: map[string]types.IndexMeta{
 			"id": {
 				IType:      types.IndexTypePrimaryKey,

pkg/discovery/base.go

@@ -0,0 +1,39 @@
+/*
+ * 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 discovery
+
+const (
+	FILE   string = "file"
+	NACOS  string = "nacos"
+	ETCD   string = "etcd"
+	EUREKA string = "eureka"
+	REDIS  string = "redis"
+	ZK     string = "zk"
+	CONSUL string = "consul"
+	SOFA   string = "sofa"
+)
+
+type ServiceInstance struct {
+	Addr string
+	Port int
+}
+
+type RegistryService interface {
+	Lookup(key string) ([]*ServiceInstance, error)
+	Close()
+}

pkg/discovery/config.go

@@ -0,0 +1,92 @@
+/*
+ * 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 discovery
+
+import (
+	"flag"
+
+	"github.com/seata/seata-go/pkg/util/flagext"
+)
+
+type ServiceConfig struct {
+	VgroupMapping            flagext.StringMap `yaml:"vgroup-mapping" json:"vgroup-mapping" koanf:"vgroup-mapping"`
+	Grouplist                flagext.StringMap `yaml:"grouplist" json:"grouplist" koanf:"grouplist"`
+	EnableDegrade            bool              `yaml:"enable-degrade" json:"enable-degrade" koanf:"enable-degrade"`
+	DisableGlobalTransaction bool              `yaml:"disable-global-transaction" json:"disable-global-transaction" koanf:"disable-global-transaction"`
+}
+
+func (cfg *ServiceConfig) RegisterFlagsWithPrefix(prefix string, f *flag.FlagSet) {
+	f.BoolVar(&cfg.EnableDegrade, prefix+".enable-degrade", false, "degrade current not support.")
+	f.BoolVar(&cfg.DisableGlobalTransaction, prefix+".disable-global-transaction", false, "disable globalTransaction.")
+	f.Var(&cfg.VgroupMapping, prefix+".vgroup-mapping", "The vgroup mapping.")
+	f.Var(&cfg.Grouplist, prefix+".grouplist", "The group list.")
+}
+
+type RegistryConfig struct {
+	Type  string      `yaml:"type" json:"type" koanf:"type"`
+	File  FileConfig  `yaml:"file" json:"file" koanf:"file"`
+	Nacos NacosConfig `yaml:"nacos" json:"nacos" koanf:"nacos"`
+	Etcd3 Etcd3Config `yaml:"etcd3" json:"etcd3" koanf:"etcd3"`
+}
+
+func (cfg *RegistryConfig) RegisterFlagsWithPrefix(prefix string, f *flag.FlagSet) {
+	f.StringVar(&cfg.Type, prefix+".type", "file", "The registry type.")
+	cfg.File.RegisterFlagsWithPrefix(prefix+".file", f)
+	cfg.Nacos.RegisterFlagsWithPrefix(prefix+".nacos", f)
+	cfg.Etcd3.RegisterFlagsWithPrefix(prefix+".etcd3", f)
+}
+
+type FileConfig struct {
+	Name string `yaml:"name" json:"name" koanf:"name"`
+}
+
+func (cfg *FileConfig) RegisterFlagsWithPrefix(prefix string, f *flag.FlagSet) {
+	f.StringVar(&cfg.Name, prefix+".name", "registry.conf", "The file name of registry.")
+}
+
+type NacosConfig struct {
+	Application string `yaml:"application" json:"application" koanf:"application"`
+	ServerAddr  string `yaml:"server-addr" json:"server-addr" koanf:"server-addr"`
+	Group       string `yaml:"group" json:"group" koanf:"group"`
+	Namespace   string `yaml:"namespace" json:"namespace" koanf:"namespace"`
+	Username    string `yaml:"username" json:"username" koanf:"username"`
+	Password    string `yaml:"password" json:"password" koanf:"password"`
+	AccessKey   string `yaml:"access-key" json:"access-key" koanf:"access-key"`
+	SecretKey   string `yaml:"secret-key" json:"secret-key" koanf:"secret-key"`
+}
+
+func (cfg *NacosConfig) RegisterFlagsWithPrefix(prefix string, f *flag.FlagSet) {
+	f.StringVar(&cfg.Application, prefix+".application", "seata", "The application name of registry.")
+	f.StringVar(&cfg.ServerAddr, prefix+".server-addr", "", "The server address of registry.")
+	f.StringVar(&cfg.Group, prefix+".group", "SEATA_GROUP", "The group of registry.")
+	f.StringVar(&cfg.Namespace, prefix+".namespace", "", "The namespace of registry.")
+	f.StringVar(&cfg.Username, prefix+".username", "", "The username of registry.")
+	f.StringVar(&cfg.Password, prefix+".password", "", "The password of registry.")
+	f.StringVar(&cfg.AccessKey, prefix+".access-key", "", "The access key of registry.")
+	f.StringVar(&cfg.SecretKey, prefix+".secret-key", "", "The secret key of registry.")
+}
+
+type Etcd3Config struct {
+	Cluster    string `yaml:"cluster" json:"cluster" koanf:"cluster"`
+	ServerAddr string `yaml:"server-addr" json:"server-addr" koanf:"server-addr"`
+}
+
+func (cfg *Etcd3Config) RegisterFlagsWithPrefix(prefix string, f *flag.FlagSet) {
+	f.StringVar(&cfg.Cluster, prefix+".cluster", "default", "The server address of registry.")
+	f.StringVar(&cfg.ServerAddr, prefix+".server-addr", "http://localhost:2379", "The server address of registry.")
+}

pkg/discovery/consul.go

@@ -0,0 +1,30 @@
+/*
+ * 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 discovery
+
+type ConsulRegistryService struct{}
+
+func (s *ConsulRegistryService) Lookup(key string) ([]*ServiceInstance, error) {
+	//TODO implement me
+	panic("implement me")
+}
+
+func (s *ConsulRegistryService) Close() {
+	//TODO implement me
+	panic("implement me")
+}

pkg/discovery/etcd3.go

@@ -0,0 +1,30 @@
+/*
+ * 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 discovery
+
+type EtcdRegistryService struct{}
+
+func (s *EtcdRegistryService) Lookup(key string) ([]*ServiceInstance, error) {
+	//TODO implement me
+	panic("implement me")
+}
+
+func (s *EtcdRegistryService) Close() {
+	//TODO implement me
+	panic("implement me")
+}

pkg/discovery/eureka.go

@@ -0,0 +1,30 @@
+/*
+ * 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 discovery
+
+type EurekaRegistryService struct{}
+
+func (s *EurekaRegistryService) Lookup(key string) ([]*ServiceInstance, error) {
+	//TODO implement me
+	panic("implement me")
+}
+
+func (s *EurekaRegistryService) Close() {
+	//TODO implement me
+	panic("implement me")
+}