feat: impletment discovery service by zookeeper

go.mod

@@ -61,6 +61,7 @@ require (
 	github.com/go-playground/locales v0.14.1 // indirect
 	github.com/go-playground/universal-translator v0.18.1 // indirect
 	github.com/go-viper/mapstructure/v2 v2.2.1 // indirect
+	github.com/go-zookeeper/zk v1.0.4 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/golang/snappy v0.0.4 // indirect
 	github.com/gorilla/websocket v1.4.2 // indirect

go.sum

@@ -261,6 +261,8 @@ github.com/go-test/deep v1.0.2-0.20181118220953-042da051cf31/go.mod h1:wGDj63lr6
 github.com/go-test/deep v1.0.2/go.mod h1:wGDj63lr65AM2AQyKZd/NYHGb0R+1RLqB8NKt3aSFNA=
 github.com/go-viper/mapstructure/v2 v2.2.1 h1:ZAaOCxANMuZx5RCeg0mBdEZk7DZasvvZIxtHqx8aGss=
 github.com/go-viper/mapstructure/v2 v2.2.1/go.mod h1:oJDH3BJKyqBA2TXFhDsKDGDTlndYOZ6rGS0BRZIxGhM=
+github.com/go-zookeeper/zk v1.0.4 h1:DPzxraQx7OrPyXq2phlGlNSIyWEsAox0RJmjTseMV6I=
+github.com/go-zookeeper/zk v1.0.4/go.mod h1:nOB03cncLtlp4t+UAkGSV+9beXP/akpekBwL+UX1Qcw=
 github.com/goccy/go-json v0.10.2 h1:CrxCmQqYDkv1z7lO7Wbh2HN93uovUHgrECaO5ZrCXAU=
 github.com/goccy/go-json v0.10.2/go.mod h1:6MelG93GURQebXPDq3khkgXZkazVtN9CRI+MGFi0w8I=
 github.com/godbus/dbus/v5 v5.0.3/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5xrFpKfA=

pkg/discovery/config.go

@@ -19,8 +19,8 @@ package discovery
 
 import (
 	"flag"
-
 	"seata.apache.org/seata-go/pkg/util/flagext"
+	"time"
 )
 
 type ServiceConfig struct {
@@ -38,17 +38,19 @@ func (cfg *ServiceConfig) RegisterFlagsWithPrefix(prefix string, f *flag.FlagSet
 }
 
 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"`
+	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"`
+	Zookeeper ZookeeperConfig `yaml:"zk" json:"zk" koanf:"zk"`
 }
 
 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)
+	cfg.Zookeeper.RegisterFlagsWithPrefix(prefix+".zk", f)
 }
 
 type FileConfig struct {
@@ -90,3 +92,21 @@ 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.")
 }
+
+type ZookeeperConfig struct {
+	ServerAddr        string        `yaml:"server-addr" json:"server-addr" koanf:"server-addr"`
+	SessionTimeout    time.Duration `yaml:"session-timeout" json:"session-timeout" koanf:"session-timeout"`
+	ConnectionTimeout time.Duration `yaml:"connection-timeout" json:"connection-timeout" koanf:"connection-timeout"`
+	NodePath          string        `yaml:"node-path" json:"node-path" koanf:"node-path"`
+	Username          string        `yaml:"username" json:"username" koanf:"username"`
+	Password          string        `yaml:"password" json:"password" koanf:"password"`
+}
+
+func (cfg *ZookeeperConfig) RegisterFlagsWithPrefix(prefix string, f *flag.FlagSet) {
+	f.StringVar(&cfg.ServerAddr, prefix+".server-addr", "localhost:2181", "The server address of the Zookeeper registry.")
+	f.DurationVar(&cfg.SessionTimeout, prefix+".session-timeout", 10*time.Second, "The session timeout for the Zookeeper connection.")
+	f.DurationVar(&cfg.ConnectionTimeout, prefix+".connection-timeout", 10*time.Second, "The session timeout for the Zookeeper connection.")
+	f.StringVar(&cfg.NodePath, prefix+".node-path", "default", "The cluster name of the Zookeeper registry.")
+	f.StringVar(&cfg.Username, prefix+".username", "", "The username of registry.")
+	f.StringVar(&cfg.Password, prefix+".password", "", "The password of registry.")
+}

pkg/discovery/init.go

@@ -42,7 +42,8 @@ func InitRegistry(serviceConfig *ServiceConfig, registryConfig *RegistryConfig)
 	case REDIS:
 		//TODO: init redis registry
 	case ZK:
-		//TODO: init zk registry
+		//init zk registry
+		registryService = newZookeeperRegistryService(serviceConfig, &registryConfig.Zookeeper)
 	case CONSUL:
 		//TODO: init consul registry
 	case SOFA:

pkg/discovery/mock/test_zookeeper_client.go

@@ -0,0 +1,32 @@
+/*
+ * 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 mock
+
+import (
+	"github.com/go-zookeeper/zk"
+)
+
+// ZkConnInterface defines the methods required by ZookeeperRegistryService.
+type ZkConnInterface interface {
+	// GetW retrieves data and sets a watch on the given path.
+	GetW(path string) ([]byte, *zk.Stat, <-chan zk.Event, error)
+	// Get retrieves data without setting a watch.
+	Get(path string) ([]byte, *zk.Stat, error)
+	// Close closes the connection.
+	Close() error
+}