This is an automated email from the ASF dual-hosted git repository.
Ethan-Xingyue pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-seata-go.git
The following commit(s) were added to refs/heads/master by this push:
new b61d1dd5 Feat: support raft (#1107)
b61d1dd5 is described below
commit b61d1dd502c5c226c05d4e964ffc5ab40a088660
Author: ssshr-66 <[email protected]>
AuthorDate: Wed Jun 3 11:32:00 2026 +0800
Feat: support raft (#1107)
* OSPP 2025 Go&Java Version alignment / support-raft (#923)
upd-raft
* fix ci
* Update seatago.yml
* Update raft.go
* upd
* fix ci & reply cr
* upd
* style: fix gofmt formatting in base.go
---------
Co-authored-by: flypiggy <[email protected]>
---
pkg/discovery/base.go | 1 +
pkg/discovery/config.go | 14 +
pkg/discovery/config_test.go | 62 +++
pkg/discovery/init.go | 2 +
pkg/discovery/{base.go => metadata/constant.go} | 32 +-
pkg/discovery/metadata/metadata.go | 162 +++++++
pkg/discovery/metadata/metadata_test.go | 62 +++
pkg/discovery/{base.go => metadata/node.go} | 35 +-
pkg/discovery/raft.go | 539 ++++++++++++++++++++++
pkg/discovery/raft_test.go | 568 ++++++++++++++++++++++++
pkg/util/log/logging.go | 95 ++--
testdata/conf/seatago.yml | 4 +
12 files changed, 1510 insertions(+), 66 deletions(-)
diff --git a/pkg/discovery/base.go b/pkg/discovery/base.go
index 8438a808..4fc0f3c6 100644
--- a/pkg/discovery/base.go
+++ b/pkg/discovery/base.go
@@ -27,6 +27,7 @@ const (
CONSUL string = "consul"
SOFA string = "sofa"
NAMINGSERVER string = "namingserver"
+ RAFT string = "raft"
)
type ServiceInstance struct {
diff --git a/pkg/discovery/config.go b/pkg/discovery/config.go
index 74becf2e..8f6af80f 100644
--- a/pkg/discovery/config.go
+++ b/pkg/discovery/config.go
@@ -43,6 +43,7 @@ type RegistryConfig struct {
Nacos NacosConfig `yaml:"nacos" json:"nacos"
koanf:"nacos"`
Etcd3 Etcd3Config `yaml:"etcd3" json:"etcd3"
koanf:"etcd3"`
NamingServer NamingServerConfig `yaml:"naming-server"
json:"naming-server" koanf:"naming-server"`
+ Raft RaftConfig `yaml:"raft" json:"raft" koanf:"raft"`
}
func (cfg *RegistryConfig) RegisterFlagsWithPrefix(prefix string, f
*flag.FlagSet) {
@@ -51,6 +52,7 @@ func (cfg *RegistryConfig) RegisterFlagsWithPrefix(prefix
string, f *flag.FlagSe
cfg.Nacos.RegisterFlagsWithPrefix(prefix+".nacos", f)
cfg.Etcd3.RegisterFlagsWithPrefix(prefix+".etcd3", f)
cfg.NamingServer.RegisterFlagsWithPrefix(prefix+".naming-server", f)
+ cfg.Raft.RegisterFlagsWithPrefix(prefix+".raft", f)
}
type FileConfig struct {
@@ -93,6 +95,18 @@ func (cfg *Etcd3Config) RegisterFlagsWithPrefix(prefix
string, f *flag.FlagSet)
f.StringVar(&cfg.ServerAddr, prefix+".server-addr",
"http://localhost:2379";, "The server address of registry.")
}
+type RaftConfig struct {
+ MetadataMaxAgeMs int64 `yaml:"metadata-max-age-ms"
json:"metadata-max-age-ms" koanf:"metadata-max-age-ms"`
+ ServerAddr string `yaml:"server-addr"
json:"server-addr" koanf:"server-addr"`
+ TokenValidityInMilliseconds int64
`yaml:"token-validity-in-milliseconds" json:"token-validity-in-milliseconds"
koanf:"token-validity-in-milliseconds"`
+}
+
+func (cfg *RaftConfig) RegisterFlagsWithPrefix(prefix string, f *flag.FlagSet)
{
+ f.Int64Var(&cfg.MetadataMaxAgeMs, prefix+".metadata-max-age-ms", 30000,
"Maximum age of metadata in milliseconds before refresh")
+ f.StringVar(&cfg.ServerAddr, prefix+".server-addr", "127.0.0.1:7091",
"The server address of raft registry")
+ f.Int64Var(&cfg.TokenValidityInMilliseconds,
prefix+".token-validity-in-milliseconds", 1740000, "Token validity duration in
milliseconds")
+}
+
type NamingServerConfig struct {
Cluster string `yaml:"cluster" json:"cluster"
koanf:"cluster"`
ServerAddr string `yaml:"server-addr"
json:"server-addr" koanf:"server-addr"`
diff --git a/pkg/discovery/config_test.go b/pkg/discovery/config_test.go
new file mode 100644
index 00000000..ed7c7c5d
--- /dev/null
+++ b/pkg/discovery/config_test.go
@@ -0,0 +1,62 @@
+/*
+ * 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"
+ "testing"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+ "gopkg.in/yaml.v2"
+)
+
+func TestRaftConfigYAMLServerAddr(t *testing.T) {
+ type testConfig struct {
+ Raft RaftConfig `yaml:"raft"`
+ }
+
+ input := []byte(`raft:
+ metadata-max-age-ms: 30000
+ server-addr: 127.0.0.1:7091
+ token-validity-in-milliseconds: 1740000
+`)
+
+ var cfg testConfig
+ err := yaml.Unmarshal(input, &cfg)
+ require.NoError(t, err)
+
+ assert.Equal(t, "127.0.0.1:7091", cfg.Raft.ServerAddr)
+ assert.Equal(t, int64(30000), cfg.Raft.MetadataMaxAgeMs)
+ assert.Equal(t, int64(1740000), cfg.Raft.TokenValidityInMilliseconds)
+}
+
+func TestRegistryConfigRegisterFlagsWithPrefix(t *testing.T) {
+ cfg := &RegistryConfig{}
+ fs := flag.NewFlagSet("test", flag.ContinueOnError)
+ cfg.RegisterFlagsWithPrefix("registry", fs)
+
+ err := fs.Parse([]string{
+ "-registry.naming-server.server-addr=127.0.0.1:8081",
+ "-registry.raft.server-addr=127.0.0.1:7091",
+ })
+ require.NoError(t, err)
+
+ assert.Equal(t, "127.0.0.1:8081", cfg.NamingServer.ServerAddr)
+ assert.Equal(t, "127.0.0.1:7091", cfg.Raft.ServerAddr)
+}
diff --git a/pkg/discovery/init.go b/pkg/discovery/init.go
index fe6127f3..98855b2f 100644
--- a/pkg/discovery/init.go
+++ b/pkg/discovery/init.go
@@ -35,6 +35,8 @@ func InitRegistry(serviceConfig *ServiceConfig,
registryConfig *RegistryConfig)
case ETCD:
//init etcd registry
registryService = newEtcdRegistryService(serviceConfig,
®istryConfig.Etcd3)
+ case RAFT:
+ registryService = NewRaftRegistryService(serviceConfig,
registryConfig)
case NACOS:
//TODO: init nacos registry
case EUREKA:
diff --git a/pkg/discovery/base.go b/pkg/discovery/metadata/constant.go
similarity index 64%
copy from pkg/discovery/base.go
copy to pkg/discovery/metadata/constant.go
index 8438a808..a86ba903 100644
--- a/pkg/discovery/base.go
+++ b/pkg/discovery/metadata/constant.go
@@ -15,26 +15,22 @@
* limitations under the License.
*/
-package discovery
+package metadata
+
+type ClusterRole string
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"
- NAMINGSERVER string = "namingserver"
+ LEADER ClusterRole = "LEADER"
+ FOLLOWER ClusterRole = "FOLLOWER"
+ LEARNER ClusterRole = "LEARNER"
+ MEMBER ClusterRole = "MEMBER"
)
-type ServiceInstance struct {
- Addr string
- Port int
-}
+type StoreMode string
-type RegistryService interface {
- Lookup(key string) ([]*ServiceInstance, error)
- Close()
-}
+const (
+ FILE StoreMode = "file"
+ DB StoreMode = "db"
+ REDIS StoreMode = "redis"
+ RAFT StoreMode = "raft"
+)
diff --git a/pkg/discovery/metadata/metadata.go
b/pkg/discovery/metadata/metadata.go
new file mode 100644
index 00000000..3e9b4ed4
--- /dev/null
+++ b/pkg/discovery/metadata/metadata.go
@@ -0,0 +1,162 @@
+/*
+ * 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 metadata
+
+import (
+ "math/rand"
+ "sync"
+)
+
+type MetadataResponse struct {
+ Nodes []*Node
+ StoreMode string
+ Term int64
+}
+
+type Metadata struct {
+ leaders sync.Map // clusterName -> sync.Map(group -> *Node)
+ clusterTerm sync.Map // clusterName -> sync.Map(group -> int64)
+ clusterNodes sync.Map // clusterName -> sync.Map(group -> []*Node)
+ storeMode StoreMode
+}
+
+func NewMetadata() *Metadata {
+ return &Metadata{
+ storeMode: FILE,
+ }
+}
+
+func (m *Metadata) GetLeader(clusterName string) *Node {
+ if groupMapAny, ok := m.leaders.Load(clusterName); ok {
+ if groupMap, ok := groupMapAny.(*sync.Map); ok {
+ var nodes []*Node
+ groupMap.Range(func(_, value any) bool {
+ if node, ok := value.(*Node); ok {
+ nodes = append(nodes, node)
+ }
+ return true
+ })
+ if len(nodes) > 0 {
+ return nodes[rand.Intn(len(nodes))]
+ }
+ }
+ }
+ return nil
+}
+
+func (m *Metadata) GetNodes(clusterName, group string) []*Node {
+ clusterNodesAny, ok := m.clusterNodes.Load(clusterName)
+ if !ok {
+ return nil
+ }
+ clusterMap, ok := clusterNodesAny.(*sync.Map)
+ if !ok {
+ return nil
+ }
+
+ if group == "" {
+ var result []*Node
+ clusterMap.Range(func(_, value any) bool {
+ if nodes, ok := value.([]*Node); ok {
+ result = append(result, nodes...)
+ }
+ return true
+ })
+ return result
+ }
+
+ if nodesAny, ok := clusterMap.Load(group); ok {
+ if nodes, ok := nodesAny.([]*Node); ok {
+ return nodes
+ }
+ }
+ return nil
+}
+
+func (m *Metadata) SetNodes(clusterName, group string, nodes []*Node) {
+ clusterMapAny, _ := m.clusterNodes.LoadOrStore(clusterName, &sync.Map{})
+ clusterMap := clusterMapAny.(*sync.Map)
+ clusterMap.Store(group, nodes)
+}
+
+func (m *Metadata) ContainsGroup(clusterName string) bool {
+ _, ok := m.clusterNodes.Load(clusterName)
+ return ok
+}
+
+func (m *Metadata) Groups(clusterName string) []string {
+ if clusterAny, ok := m.clusterNodes.Load(clusterName); ok {
+ if cluster, ok := clusterAny.(*sync.Map); ok {
+ var groups []string
+ cluster.Range(func(key, _ any) bool {
+ if group, ok := key.(string); ok {
+ groups = append(groups, group)
+ }
+ return true
+ })
+ return groups
+ }
+ }
+ return nil
+}
+
+func (m *Metadata) GetClusterTerm(clusterName string) map[string]int64 {
+ if termMapAny, ok := m.clusterTerm.Load(clusterName); ok {
+ if termMap, ok := termMapAny.(*sync.Map); ok {
+ result := make(map[string]int64)
+ termMap.Range(func(key, value any) bool {
+ k, ok1 := key.(string)
+ v, ok2 := value.(int64)
+ if ok1 && ok2 {
+ result[k] = v
+ }
+ return true
+ })
+ return result
+ }
+ }
+ return nil
+}
+
+func (m *Metadata) RefreshMetadata(clusterName string, response
MetadataResponse) {
+ nodesByGroup := make(map[string][]*Node)
+ for _, node := range response.Nodes {
+ nodesByGroup[node.Group] = append(nodesByGroup[node.Group],
node)
+ if node.Role == LEADER {
+ groupMapAny, _ := m.leaders.LoadOrStore(clusterName,
&sync.Map{})
+ groupMap := groupMapAny.(*sync.Map)
+ groupMap.Store(node.Group, node)
+ }
+ }
+
+ switch response.StoreMode {
+ case "RAFT":
+ m.storeMode = RAFT
+ default:
+ m.storeMode = FILE
+ }
+
+ if len(nodesByGroup) > 0 {
+ termMapAny, _ := m.clusterTerm.LoadOrStore(clusterName,
&sync.Map{})
+ termMap := termMapAny.(*sync.Map)
+ for group, nodes := range nodesByGroup {
+ m.SetNodes(clusterName, group, nodes)
+ termMap.Store(group, response.Term)
+ }
+ }
+}
diff --git a/pkg/discovery/metadata/metadata_test.go
b/pkg/discovery/metadata/metadata_test.go
new file mode 100644
index 00000000..01914634
--- /dev/null
+++ b/pkg/discovery/metadata/metadata_test.go
@@ -0,0 +1,62 @@
+/*
+ * 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 metadata
+
+import (
+ "testing"
+
+ "github.com/stretchr/testify/assert"
+)
+
+func TestRefreshMetadataGroupsNodesAndTermsByGroup(t *testing.T) {
+ m := NewMetadata()
+
+ m.RefreshMetadata("test-cluster", MetadataResponse{
+ Term: 7,
+ StoreMode: "RAFT",
+ Nodes: []*Node{
+ {
+ Control: &Endpoint{Host: "127.0.0.1", Port:
7001},
+ Transaction: &Endpoint{Host: "127.0.0.1", Port:
8001},
+ Internal: &Endpoint{Host: "127.0.0.1", Port:
9001},
+ Group: "group-a",
+ Role: LEADER,
+ },
+ {
+ Control: &Endpoint{Host: "127.0.0.1", Port:
7002},
+ Transaction: &Endpoint{Host: "127.0.0.1", Port:
8002},
+ Internal: &Endpoint{Host: "127.0.0.1", Port:
9002},
+ Group: "group-a",
+ Role: FOLLOWER,
+ },
+ {
+ Control: &Endpoint{Host: "127.0.0.1", Port:
7003},
+ Transaction: &Endpoint{Host: "127.0.0.1", Port:
8003},
+ Internal: &Endpoint{Host: "127.0.0.1", Port:
9003},
+ Group: "group-b",
+ Role: LEADER,
+ },
+ },
+ })
+
+ assert.Len(t, m.GetNodes("test-cluster", "group-a"), 2)
+ assert.Len(t, m.GetNodes("test-cluster", "group-b"), 1)
+ assert.Equal(t, int64(7), m.GetClusterTerm("test-cluster")["group-a"])
+ assert.Equal(t, int64(7), m.GetClusterTerm("test-cluster")["group-b"])
+ assert.Equal(t, RAFT, m.storeMode)
+}
diff --git a/pkg/discovery/base.go b/pkg/discovery/metadata/node.go
similarity index 64%
copy from pkg/discovery/base.go
copy to pkg/discovery/metadata/node.go
index 8438a808..ab931073 100644
--- a/pkg/discovery/base.go
+++ b/pkg/discovery/metadata/node.go
@@ -15,26 +15,25 @@
* limitations under the License.
*/
-package discovery
+package metadata
-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"
- NAMINGSERVER string = "namingserver"
-)
+type Node struct {
+ Control *Endpoint
+ Transaction *Endpoint
+ Internal *Endpoint
-type ServiceInstance struct {
- Addr string
- Port int
+ Weight float64
+ Healthy bool
+ Timestamp int64
+
+ Group string
+ Role ClusterRole
+ Version string
+ Metadata map[string]interface{}
}
-type RegistryService interface {
- Lookup(key string) ([]*ServiceInstance, error)
- Close()
+type Endpoint struct {
+ Host string
+ Protocol string
+ Port int
}
diff --git a/pkg/discovery/raft.go b/pkg/discovery/raft.go
new file mode 100644
index 00000000..f724ccbe
--- /dev/null
+++ b/pkg/discovery/raft.go
@@ -0,0 +1,539 @@
+/*
+ * 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 (
+ "bytes"
+ "context"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "io"
+ "math/rand"
+ "net"
+ "net/http"
+ "net/url"
+ "strconv"
+ "strings"
+ "sync"
+ "time"
+
+ "seata.apache.org/seata-go/v2/pkg/discovery/metadata"
+ "seata.apache.org/seata-go/v2/pkg/util/log"
+)
+
+const (
+ controlEndpoint = "control"
+ transactionEndpoint = "transaction"
+)
+
+type RaftRegistryService struct {
+ cfg *RaftConfig
+ metadata *metadata.Metadata
+ initAddresses sync.Map // clusterName ->
[]*ServiceInstance
+ aliveNodes sync.Map // transactionServiceGroup ->
[]*ServiceInstance
+ vgroupMapping map[string]string
+ namingserverAddress string
+ username string
+ password string
+ jwtToken string
+ tokenTimestamp int64
+ currentTransactionServiceGroup string
+ currentTransactionClusterName string
+ mu sync.RWMutex
+ stopCh chan struct{}
+ refreshOnce sync.Once
+ httpClient *http.Client
+ random *rand.Rand
+}
+
+func NewRaftRegistryService(config *ServiceConfig, raftConfig *RegistryConfig)
*RaftRegistryService {
+ vgroupMapping := config.VgroupMapping
+
+ r := &RaftRegistryService{
+ cfg: &raftConfig.Raft,
+ metadata: metadata.NewMetadata(),
+ initAddresses: sync.Map{},
+ aliveNodes: sync.Map{},
+ vgroupMapping: vgroupMapping,
+ namingserverAddress: raftConfig.NamingServer.ServerAddr,
+ username: raftConfig.NamingServer.Username,
+ password: raftConfig.NamingServer.Password,
+ stopCh: make(chan struct{}),
+ httpClient: &http.Client{},
+ tokenTimestamp: -1,
+ random:
rand.New(rand.NewSource(time.Now().UnixNano())),
+ }
+ return r
+}
+
+func (r *RaftRegistryService) Lookup(key string) ([]*ServiceInstance, error) {
+ clusterName := r.vgroupMapping[key]
+ if clusterName == "" {
+ return nil, fmt.Errorf("cluster doesn't exist for
serviceGroup=%s", key)
+ }
+ r.mu.Lock()
+ r.currentTransactionServiceGroup = key
+ r.currentTransactionClusterName = clusterName
+ r.mu.Unlock()
+
+ if !r.metadata.ContainsGroup(clusterName) {
+ if _, ok := r.loadInitAddresses(clusterName); !ok &&
r.cfg.ServerAddr != "" {
+ addrs := strings.Split(r.cfg.ServerAddr, ",")
+ list := make([]*ServiceInstance, 0, len(addrs))
+ for _, addr := range addrs {
+ h, p, err :=
net.SplitHostPort(strings.TrimSpace(addr))
+ if err != nil {
+ log.Infof("invalid init server addr:
%s, err: %v", addr, err)
+ continue
+ }
+ port, err := strconv.Atoi(p)
+ if err != nil {
+ log.Errorf("invalid port: %s", p)
+ continue
+ }
+ list = append(list, &ServiceInstance{Addr: h,
Port: port})
+ }
+ if len(list) == 0 {
+ return nil, fmt.Errorf("invalid service
group/key: %s", key)
+ }
+ r.initAddresses.Store(clusterName, list)
+
+ if err := r.refreshToken(); err != nil {
+ return nil, err
+ }
+
+ err := r.acquireClusterMetaData(clusterName, "")
+ if err != nil {
+ return nil, err
+ }
+ r.startQueryMetadata()
+ }
+ }
+ leader := r.metadata.GetLeader(clusterName)
+ if leader != nil {
+ endpoint, err := r.selectEndpoint(transactionEndpoint, leader)
+ if err != nil {
+ return nil, err
+ }
+ return []*ServiceInstance{endpoint}, nil
+ }
+ return r.getServiceInstances(clusterName, "")
+}
+
+func (r *RaftRegistryService) getServiceInstances(clusterName, group string)
([]*ServiceInstance, error) {
+ nodes := r.metadata.GetNodes(clusterName, group)
+ if len(nodes) > 0 {
+ instances := make([]*ServiceInstance, 0, len(nodes))
+ for _, n := range nodes {
+ inst, _ := r.selectEndpoint(transactionEndpoint, n)
+ if inst != nil {
+ instances = append(instances, inst)
+ }
+ }
+ return instances, nil
+ }
+ return nil, nil
+}
+
+func (r *RaftRegistryService) RefreshAliveLookup(transactionServiceGroup
string, aliveAddress []*ServiceInstance) ([]*ServiceInstance, error) {
+ clusterName := r.vgroupMapping[transactionServiceGroup]
+ if clusterName == "" {
+ return nil, fmt.Errorf("cluster not found for serviceGroup=%s",
transactionServiceGroup)
+ }
+
+ leader := r.metadata.GetLeader(clusterName)
+ if leader == nil {
+ return nil, fmt.Errorf("leader not found for cluster=%s",
clusterName)
+ }
+
+ leaderEndpoint, err := r.selectEndpoint(transactionEndpoint, leader)
+ if err != nil {
+ return nil, err
+ }
+
+ var result []*ServiceInstance
+ for _, addr := range aliveAddress {
+ if addr.Port != leaderEndpoint.Port || addr.Addr !=
leaderEndpoint.Addr {
+ result = append(result, addr)
+ }
+ }
+
+ r.aliveNodes.Store(transactionServiceGroup, result)
+ return result, nil
+}
+
+func (r *RaftRegistryService) Close() {
+ select {
+ case <-r.stopCh:
+ default:
+ close(r.stopCh)
+ }
+}
+
+func (r *RaftRegistryService) selectEndpoint(t string, n *metadata.Node)
(*ServiceInstance, error) {
+ switch t {
+ case controlEndpoint:
+ return &ServiceInstance{
+ Addr: n.Control.Host,
+ Port: n.Control.Port,
+ }, nil
+ case transactionEndpoint:
+ return &ServiceInstance{
+ Addr: n.Transaction.Host,
+ Port: n.Transaction.Port,
+ }, nil
+ default:
+ return nil, fmt.Errorf("SelectEndpoint is not support type:
%s", t)
+ }
+}
+
+func (r *RaftRegistryService) startQueryMetadata() {
+ r.refreshOnce.Do(func() {
+ go func() {
+ metadataMaxAge := int64(30000)
+ if r.cfg.MetadataMaxAgeMs > 0 {
+ metadataMaxAge = r.cfg.MetadataMaxAgeMs
+ }
+ currentTime := time.Now().UnixMilli()
+ ticker := time.NewTicker(5 * time.Second)
+ defer ticker.Stop()
+
+ for {
+ select {
+ case <-r.stopCh:
+ log.Info("raft registry service
stopped")
+ return
+ case <-ticker.C:
+ func() {
+ shouldFetch :=
time.Now().UnixMilli()-currentTime > metadataMaxAge
+ if !shouldFetch {
+ ok, err := r.watch()
+ if err != nil {
+
log.Errorf("watch error: %v", err)
+ shouldFetch =
true
+ } else {
+ shouldFetch = ok
+ }
+ }
+
+ if shouldFetch {
+ clusterNames :=
r.clusterNamesFromInit()
+ if len(clusterNames) ==
0 {
+ r.mu.RLock()
+ if
r.currentTransactionClusterName != "" {
+
clusterNames = append(clusterNames, r.currentTransactionClusterName)
+ }
+ r.mu.RUnlock()
+ }
+ for _, clusterName :=
range clusterNames {
+ groups :=
r.metadata.Groups(clusterName)
+ if len(groups)
== 0 {
+ groups
= append(groups, "")
+ }
+ for _, g :=
range groups {
+ err :=
r.acquireClusterMetaData(clusterName, g)
+ if err
!= nil {
+
log.Errorf("acquire cluster metadata failed: cluster=%s group=%s err=%v",
clusterName, g, err)
+ }
+ }
+ }
+
+ currentTime =
time.Now().UnixMilli()
+ }
+ }()
+ }
+ }
+ }()
+ })
+}
+
+func (r *RaftRegistryService) watch() (bool, error) {
+ clusterNames := r.clusterNamesFromInit()
+ needsRefresh := false
+ for _, clusterName := range clusterNames {
+ groupTerms := r.metadata.GetClusterTerm(clusterName)
+ if groupTerms == nil {
+ groupTerms = map[string]int64{"": 0}
+ }
+
+ var (
+ tcAddress string
+ err error
+ )
+ for group := range groupTerms {
+ tcAddress, err = r.queryHttpAddress(clusterName, group)
+ if err == nil {
+ break
+ }
+ log.Infof("no tc address to watch for cluster %s group
%s: %v", clusterName, group, err)
+ }
+ if tcAddress == "" {
+ continue
+ }
+
+ header := map[string]string{
+ "Content-Type": "application/x-www-form-urlencoded",
+ }
+ if r.isTokenExpired() {
+ if err = r.refreshToken(); err != nil {
+ return false, err
+ }
+ }
+ if r.jwtToken != "" {
+ header["Authorization"] = r.jwtToken
+ }
+
+ form := url.Values{}
+ for group, term := range groupTerms {
+ form.Set(group, strconv.FormatInt(term, 10))
+ }
+
+ endpoint := fmt.Sprintf("http://%s/metadata/v1/watch";,
tcAddress)
+ req, err := http.NewRequest("POST", endpoint,
strings.NewReader(form.Encode()))
+ if err != nil {
+ return false, err
+ }
+ for hk, hv := range header {
+ req.Header.Set(hk, hv)
+ }
+ resp, err := r.doRequest(req, 30*time.Second)
+ if err != nil {
+ log.Errorf("watch cluster node: %s, fail: %v",
tcAddress, err)
+ return false, err
+ }
+ defer resp.Body.Close()
+ if resp.StatusCode == http.StatusUnauthorized {
+ return false, errors.New("authentication failed:
missing username/password")
+ }
+ if resp.StatusCode == http.StatusOK {
+ needsRefresh = true
+ }
+ }
+ return needsRefresh, nil
+}
+
+func (r *RaftRegistryService) acquireClusterMetaData(clusterName, group
string) error {
+ tcAddress, err := r.queryHttpAddress(clusterName, group)
+ if err != nil {
+ return err
+ }
+ headers := map[string]string{
+ "Content-Type": "application/x-www-form-urlencoded",
+ }
+ if r.isTokenExpired() {
+ if err = r.refreshToken(); err != nil {
+ return err
+ }
+ }
+ if r.jwtToken != "" {
+ headers["Authorization"] = r.jwtToken
+ }
+ u := fmt.Sprintf("http://%s/metadata/v1/cluster";, tcAddress)
+ req, err := http.NewRequest("GET", u, nil)
+ if err != nil {
+ return err
+ }
+ q := req.URL.Query()
+ q.Add("group", group)
+ req.URL.RawQuery = q.Encode()
+ for hk, hv := range headers {
+ req.Header.Set(hk, hv)
+ }
+
+ resp, err := r.doRequest(req, 1*time.Second)
+ if err != nil {
+ return fmt.Errorf("http get cluster failed: %w", err)
+ }
+ defer resp.Body.Close()
+
+ if resp.StatusCode == http.StatusOK {
+ body, _ := io.ReadAll(resp.Body)
+ var mr metadata.MetadataResponse
+ if err = json.Unmarshal(body, &mr); err != nil {
+ return fmt.Errorf("unmarshal metadataResponse failed:
%w", err)
+ }
+ r.metadata.RefreshMetadata(clusterName, mr)
+ return nil
+ } else if resp.StatusCode == http.StatusUnauthorized {
+ if err = r.refreshToken(); err != nil {
+ return err
+ }
+ return fmt.Errorf("authentication failed! you should configure
the correct username and password")
+ }
+ return fmt.Errorf("unexpected status code: %d", resp.StatusCode)
+}
+
+/* -------------------- Token management -------------------- */
+
+func (r *RaftRegistryService) isTokenExpired() bool {
+ r.mu.RLock()
+ ts := r.tokenTimestamp
+ r.mu.RUnlock()
+ if ts == -1 {
+ return true
+ }
+ valid := int64(29 * 60 * 1000)
+ if r.cfg.TokenValidityInMilliseconds > 0 {
+ valid = r.cfg.TokenValidityInMilliseconds
+ }
+ expireTime := ts + valid
+ return time.Now().UnixMilli() >= expireTime
+}
+
+func (r *RaftRegistryService) refreshToken() error {
+ address := r.namingserverAddress
+ body, _ := json.Marshal(map[string]string{
+ "username": r.username,
+ "password": r.password,
+ })
+ req, err := http.NewRequest("POST",
fmt.Sprintf("http://%s/api/v1/auth/login";, address), bytes.NewReader(body))
+ if err != nil {
+ return err
+ }
+ req.Header.Set("Content-Type", "application/json")
+ resp, err := r.doRequest(req, 1*time.Second)
+ if err != nil {
+ return fmt.Errorf("refresh token failed: %w", err)
+ }
+ defer resp.Body.Close()
+ if resp.StatusCode != http.StatusOK {
+ return errors.New("authentication failed when refresh token")
+ }
+ respBody, _ := io.ReadAll(resp.Body)
+ var node map[string]interface{}
+ if err = json.Unmarshal(respBody, &node); err != nil {
+ return fmt.Errorf("invalid auth response: %w", err)
+ }
+ code, _ := node["code"].(string)
+ if code != "" && code != "200" {
+ return errors.New("authentication failed! you should configure
the correct username and password")
+ }
+ dataVal := node["data"].(string)
+ r.mu.Lock()
+ r.jwtToken = dataVal
+ r.tokenTimestamp = time.Now().UnixMilli()
+ r.mu.Unlock()
+ return nil
+}
+
+func (r *RaftRegistryService) queryHttpAddress(clusterName, group string)
(string, error) {
+ var addressList []string
+
+ nodes := r.metadata.GetNodes(clusterName, group)
+
+ formatAddr := func(addr string, port int) string {
+ return fmt.Sprintf("%s:%d", addr, port)
+ }
+
+ appendControlAddr := func(node *metadata.Node) error {
+ endpoint, err := r.selectEndpoint(controlEndpoint, node)
+ if err != nil {
+ return fmt.Errorf("failed to select control endpoint:
%v", err)
+ }
+ addressList = append(addressList, formatAddr(endpoint.Addr,
endpoint.Port))
+ return nil
+ }
+
+ if len(nodes) > 0 {
+ r.mu.RLock()
+ currentServiceGroup := r.currentTransactionServiceGroup
+ r.mu.RUnlock()
+ if aliveAny, ok := r.aliveNodes.Load(currentServiceGroup); ok {
+ aliveNodes, ok := aliveAny.([]*ServiceInstance)
+ if !ok {
+ log.Errorf("invalid type for alive nodes,
expected []*ServiceInstance")
+ return "", fmt.Errorf("invalid alive nodes type
for service group %s", currentServiceGroup)
+ }
+
+ if len(aliveNodes) == 0 {
+ for _, node := range nodes {
+ if err := appendControlAddr(node); err
!= nil {
+ return "", err
+ }
+ }
+ } else {
+ m := make(map[string]*metadata.Node)
+ for _, node := range nodes {
+ inetSocketAddress, _ :=
r.selectEndpoint(transactionEndpoint, node)
+ m[formatAddr(inetSocketAddress.Addr,
inetSocketAddress.Port)] = node
+ }
+
+ for _, alive := range aliveNodes {
+ addrKey := formatAddr(alive.Addr,
alive.Port)
+ node, exists := m[addrKey]
+ if !exists {
+ continue
+ }
+
+ contrEndpoint, err :=
r.selectEndpoint(controlEndpoint, node)
+ if err != nil {
+ log.Errorf("failed to select
control endpoint for node: %v", err)
+ continue
+ }
+ addressList = append(addressList,
formatAddr(contrEndpoint.Addr, contrEndpoint.Port))
+ }
+ }
+ } else {
+ for _, node := range nodes {
+ if err := appendControlAddr(node); err != nil {
+ return "", err
+ }
+ }
+ }
+ } else {
+ if initAny, ok := r.initAddresses.Load(clusterName); ok {
+ for _, inetSocketAddress := range
initAny.([]*ServiceInstance) {
+ addressList = append(addressList,
formatAddr(inetSocketAddress.Addr, inetSocketAddress.Port))
+ }
+ }
+ }
+
+ if len(addressList) == 0 {
+ return "", fmt.Errorf("no available address for cluster=%s
group=%s", clusterName, group)
+ }
+ return addressList[r.random.Intn(len(addressList))], nil
+}
+
+func (r *RaftRegistryService) clusterNamesFromInit() []string {
+ var names []string
+ r.initAddresses.Range(func(key, _ any) bool {
+ if k, ok := key.(string); ok {
+ names = append(names, k)
+ }
+ return true
+ })
+ return names
+}
+
+func (r *RaftRegistryService) doRequest(req *http.Request, timeout
time.Duration) (*http.Response, error) {
+ ctx, cancel := context.WithTimeout(req.Context(), timeout)
+ defer cancel()
+
+ req = req.WithContext(ctx)
+ return r.httpClient.Do(req)
+}
+
+func (r *RaftRegistryService) loadInitAddresses(clusterName string)
([]*ServiceInstance, bool) {
+ if val, ok := r.initAddresses.Load(clusterName); ok {
+ if list, ok := val.([]*ServiceInstance); ok {
+ return list, true
+ }
+ }
+ return nil, false
+}
diff --git a/pkg/discovery/raft_test.go b/pkg/discovery/raft_test.go
new file mode 100644
index 00000000..3ce7bd26
--- /dev/null
+++ b/pkg/discovery/raft_test.go
@@ -0,0 +1,568 @@
+/*
+ * 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 (
+ "encoding/json"
+ "math/rand"
+ "net"
+ "net/http"
+ "net/http/httptest"
+ "strconv"
+ "sync"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+
+ "seata.apache.org/seata-go/v2/pkg/discovery/metadata"
+)
+
+func newTestServiceConfig() *ServiceConfig {
+ return &ServiceConfig{VgroupMapping:
map[string]string{"default_tx_group": "default"}}
+}
+
+func newTestRegistryConfig(serverAddr string) *RegistryConfig {
+ return &RegistryConfig{
+ Type: "raft",
+ NamingServer: NamingServerConfig{
+ ServerAddr: serverAddr,
+ Username: "seata",
+ Password: "seata",
+ },
+ Raft: RaftConfig{
+ MetadataMaxAgeMs: int64(30000),
+ ServerAddr: serverAddr,
+ TokenValidityInMilliseconds: int64(1740000),
+ },
+ }
+}
+
+func newTestServiceInstance(t *testing.T, addr string) *ServiceInstance {
+ t.Helper()
+
+ host, portStr, err := net.SplitHostPort(addr)
+ require.NoError(t, err)
+ port, err := strconv.Atoi(portStr)
+ require.NoError(t, err)
+
+ return &ServiceInstance{Addr: host, Port: port}
+}
+
+func newTestMetadataNode(t *testing.T, addr, group string, role
metadata.ClusterRole) *metadata.Node {
+ t.Helper()
+
+ serviceInstance := newTestServiceInstance(t, addr)
+
+ return &metadata.Node{
+ Control: &metadata.Endpoint{Host: serviceInstance.Addr,
Port: serviceInstance.Port},
+ Transaction: &metadata.Endpoint{Host: serviceInstance.Addr,
Port: serviceInstance.Port},
+ Internal: &metadata.Endpoint{Host: serviceInstance.Addr,
Port: serviceInstance.Port},
+ Group: group,
+ Role: role,
+ Version: "2.5.0",
+ Metadata: map[string]interface{}{},
+ }
+}
+
+func clonePostForm(r *http.Request) map[string][]string {
+ result := make(map[string][]string, len(r.PostForm))
+ for key, values := range r.PostForm {
+ result[key] = append([]string(nil), values...)
+ }
+ return result
+}
+
+func TestMetadataHandler(t *testing.T) {
+ serviceConfig := newTestServiceConfig()
+ registryConfig := newTestRegistryConfig("")
+
+ mockServer := httptest.NewServer(http.HandlerFunc(func(w
http.ResponseWriter, r *http.Request) {
+ t.Logf("Received request: %s %s", r.Method, r.URL.Path)
+
+ switch r.URL.Path {
+ case "/api/v1/auth/login":
+ w.Header().Set("Content-Type", "application/json")
+ w.WriteHeader(http.StatusOK)
+ _ = json.NewEncoder(w).Encode(map[string]string{
+ "code": "200",
+ "data": "mock-jwt-token",
+ })
+
+ case "/metadata/v1/cluster":
+ if r.Header.Get("Authorization") != "mock-jwt-token" {
+ t.Errorf("Missing valid token: %s",
r.Header.Get("Authorization"))
+ w.WriteHeader(http.StatusUnauthorized)
+ return
+ }
+
+ resp := metadata.MetadataResponse{
+ Term: 1,
+ Nodes: []*metadata.Node{
+ {
+ Control:
&metadata.Endpoint{Host: "127.0.0.1", Port: 7001},
+ Transaction:
&metadata.Endpoint{Host: "127.0.0.1", Port: 7001},
+ Internal:
&metadata.Endpoint{Host: "127.0.0.1", Port: 9001},
+ Group: "default",
+ Role: metadata.LEADER,
+ Version: "2.5.0",
+ Metadata:
map[string]interface{}{},
+ },
+ {
+ Control:
&metadata.Endpoint{Host: "127.0.0.1", Port: 7002},
+ Transaction:
&metadata.Endpoint{Host: "127.0.0.1", Port: 7002},
+ Internal:
&metadata.Endpoint{Host: "127.0.0.1", Port: 9002},
+ Group: "default",
+ Role: metadata.FOLLOWER,
+ Version: "2.5.0",
+ Metadata:
map[string]interface{}{},
+ },
+ {
+ Control:
&metadata.Endpoint{Host: "127.0.0.1", Port: 7003},
+ Transaction:
&metadata.Endpoint{Host: "127.0.0.1", Port: 7003},
+ Internal:
&metadata.Endpoint{Host: "127.0.0.1", Port: 9003},
+ Group: "default",
+ Role: metadata.FOLLOWER,
+ Version: "2.5.0",
+ Metadata:
map[string]interface{}{},
+ },
+ },
+ }
+
+ w.Header().Set("Content-Type", "application/json")
+ w.WriteHeader(http.StatusOK)
+ if err := json.NewEncoder(w).Encode(resp); err != nil {
+ t.Errorf("Failed to encode response: %v", err)
+ }
+
+ default:
+ t.Errorf("Unknown path: %s", r.URL.Path)
+ w.WriteHeader(http.StatusNotFound)
+ }
+ }))
+ defer mockServer.Close()
+ mockAddr := mockServer.Listener.Addr().String()
+ registryConfig.Raft.ServerAddr = mockAddr
+ registryConfig.NamingServer.ServerAddr = mockAddr
+ service := NewRaftRegistryService(serviceConfig, registryConfig)
+ instances, err := service.Lookup("default_tx_group")
+ if err != nil {
+ t.Errorf("Failed to lookup service: %v", err)
+ }
+ serviceInstances := []*ServiceInstance{
+ {
+ Addr: "127.0.0.1",
+ Port: 7001,
+ },
+ }
+ assert.Equal(t, serviceInstances, instances)
+}
+
+func TestLookupIncludesServiceGroupInMissingClusterError(t *testing.T) {
+ service := NewRaftRegistryService(newTestServiceConfig(),
newTestRegistryConfig(""))
+
+ _, err := service.Lookup("missing_tx_group")
+
+ require.EqualError(t, err, "cluster doesn't exist for
serviceGroup=missing_tx_group")
+}
+
+func TestMultiClusterWatch(t *testing.T) {
+ clusterRequests := make(map[string]int)
+ var requestMu sync.Mutex
+
+ mockServer := httptest.NewServer(http.HandlerFunc(func(w
http.ResponseWriter, r *http.Request) {
+ requestMu.Lock()
+ clusterRequests[r.URL.Path]++
+ requestMu.Unlock()
+
+ switch r.URL.Path {
+ case "/api/v1/auth/login":
+ w.Header().Set("Content-Type", "application/json")
+ w.WriteHeader(http.StatusOK)
+ _ = json.NewEncoder(w).Encode(map[string]string{
+ "code": "200",
+ "data": "mock-jwt-token",
+ })
+
+ case "/metadata/v1/cluster":
+ resp := metadata.MetadataResponse{
+ Term: 1,
+ Nodes: []*metadata.Node{
+ {
+ Control:
&metadata.Endpoint{Host: "127.0.0.1", Port: 7001},
+ Transaction:
&metadata.Endpoint{Host: "127.0.0.1", Port: 7001},
+ Internal:
&metadata.Endpoint{Host: "127.0.0.1", Port: 9001},
+ Group: "",
+ Role: metadata.LEADER,
+ Version: "2.5.0",
+ Metadata:
map[string]interface{}{},
+ },
+ },
+ }
+ w.Header().Set("Content-Type", "application/json")
+ w.WriteHeader(http.StatusOK)
+ _ = json.NewEncoder(w).Encode(resp)
+
+ default:
+ w.WriteHeader(http.StatusNotFound)
+ }
+ }))
+ defer mockServer.Close()
+
+ multiServiceConfig := &ServiceConfig{
+ VgroupMapping: map[string]string{
+ "tx_group_1": "cluster1",
+ "tx_group_2": "cluster2",
+ },
+ }
+
+ mockAddr := mockServer.Listener.Addr().String()
+ multiRegistryConfig := &RegistryConfig{
+ Type: "raft",
+ NamingServer: NamingServerConfig{
+ ServerAddr: mockAddr,
+ Username: "seata",
+ Password: "seata",
+ },
+ Raft: RaftConfig{
+ MetadataMaxAgeMs: int64(30000),
+ ServerAddr: mockAddr + "," + mockAddr,
+ TokenValidityInMilliseconds: int64(1740000),
+ },
+ }
+
+ service := NewRaftRegistryService(multiServiceConfig,
multiRegistryConfig)
+ defer service.Close()
+
+ _, err1 := service.Lookup("tx_group_1")
+ assert.NoError(t, err1)
+
+ _, err2 := service.Lookup("tx_group_2")
+ assert.NoError(t, err2)
+
+ requestMu.Lock()
+ clusterCount := clusterRequests["/metadata/v1/cluster"]
+ requestMu.Unlock()
+
+ assert.True(t, clusterCount >= 2, "Multiple clusters should trigger
multiple cluster metadata requests")
+}
+
+func TestConcurrentAccess(t *testing.T) {
+ serviceConfig := newTestServiceConfig()
+
+ mockServer := httptest.NewServer(http.HandlerFunc(func(w
http.ResponseWriter, r *http.Request) {
+ switch r.URL.Path {
+ case "/api/v1/auth/login":
+ w.Header().Set("Content-Type", "application/json")
+ w.WriteHeader(http.StatusOK)
+ _ = json.NewEncoder(w).Encode(map[string]string{
+ "code": "200",
+ "data": "mock-jwt-token",
+ })
+
+ case "/metadata/v1/cluster":
+ resp := metadata.MetadataResponse{
+ Term: 1,
+ Nodes: []*metadata.Node{
+ {
+ Control:
&metadata.Endpoint{Host: "127.0.0.1", Port: 7001},
+ Transaction:
&metadata.Endpoint{Host: "127.0.0.1", Port: 7001},
+ Internal:
&metadata.Endpoint{Host: "127.0.0.1", Port: 9001},
+ Group: "default",
+ Role: metadata.LEADER,
+ Version: "2.5.0",
+ Metadata:
map[string]interface{}{},
+ },
+ },
+ }
+ w.Header().Set("Content-Type", "application/json")
+ w.WriteHeader(http.StatusOK)
+ _ = json.NewEncoder(w).Encode(resp)
+
+ case "/metadata/v1/watch":
+ w.WriteHeader(http.StatusOK)
+
+ default:
+ w.WriteHeader(http.StatusNotFound)
+ }
+ }))
+ defer mockServer.Close()
+
+ mockAddr := mockServer.Listener.Addr().String()
+ testRegistryConfig := newTestRegistryConfig(mockAddr)
+
+ service := NewRaftRegistryService(serviceConfig, testRegistryConfig)
+ defer service.Close()
+
+ _, err := service.Lookup("default_tx_group")
+ assert.NoError(t, err)
+
+ var wg sync.WaitGroup
+ successCount := int32(0)
+
+ for i := 0; i < 20; i++ {
+ wg.Add(1)
+ go func(id int) {
+ defer wg.Done()
+
+ _, err := service.Lookup("default_tx_group")
+ if err == nil {
+ atomic.AddInt32(&successCount, 1)
+ }
+ }(i)
+ }
+
+ wg.Wait()
+
+ assert.True(t, atomic.LoadInt32(&successCount) > 0, "Some concurrent
lookups should succeed")
+}
+
+func TestWatchErrorRecovery(t *testing.T) {
+ service := &RaftRegistryService{
+ metadata: metadata.NewMetadata(),
+ vgroupMapping: map[string]string{"test_group":
"test_cluster"},
+ currentTransactionClusterName: "test_cluster",
+ random:
rand.New(rand.NewSource(time.Now().UnixNano())),
+ initAddresses: sync.Map{},
+ cfg:
&RaftConfig{TokenValidityInMilliseconds: 1740000},
+ tokenTimestamp: time.Now().UnixMilli(),
+ httpClient: &http.Client{},
+ }
+
+ service.metadata.RefreshMetadata("test_cluster",
metadata.MetadataResponse{
+ Term: 1,
+ Nodes: []*metadata.Node{
+ {
+ Control: &metadata.Endpoint{Host:
"127.0.0.1", Port: 7001},
+ Transaction: &metadata.Endpoint{Host:
"127.0.0.1", Port: 8001},
+ Internal: &metadata.Endpoint{Host:
"127.0.0.1", Port: 9001},
+ Group: "",
+ Role: metadata.LEADER,
+ Version: "2.5.0",
+ Metadata: map[string]interface{}{},
+ },
+ },
+ })
+
+ service.initAddresses.Store("test_cluster", []*ServiceInstance{
+ {Addr: "127.0.0.1", Port: 7001},
+ })
+
+ result, err := service.watch()
+ assert.Error(t, err)
+ assert.False(t, result)
+
+ address, err := service.queryHttpAddress("test_cluster", "")
+ assert.NoError(t, err)
+ assert.NotEmpty(t, address)
+}
+
+func TestWatchProcessesAllClustersAndAllGroupTerms(t *testing.T) {
+ var (
+ cluster1WatchCount int32
+ cluster2WatchCount int32
+ formMu sync.Mutex
+ cluster1Form map[string][]string
+ )
+
+ cluster1Server := httptest.NewServer(http.HandlerFunc(func(w
http.ResponseWriter, r *http.Request) {
+ if r.URL.Path != "/metadata/v1/watch" {
+ w.WriteHeader(http.StatusNotFound)
+ return
+ }
+
+ if err := r.ParseForm(); err != nil {
+ t.Errorf("parse watch form failed: %v", err)
+ w.WriteHeader(http.StatusBadRequest)
+ return
+ }
+
+ formMu.Lock()
+ cluster1Form = clonePostForm(r)
+ formMu.Unlock()
+
+ atomic.AddInt32(&cluster1WatchCount, 1)
+ w.WriteHeader(http.StatusNotModified)
+ }))
+ defer cluster1Server.Close()
+
+ cluster2Server := httptest.NewServer(http.HandlerFunc(func(w
http.ResponseWriter, r *http.Request) {
+ if r.URL.Path != "/metadata/v1/watch" {
+ w.WriteHeader(http.StatusNotFound)
+ return
+ }
+
+ atomic.AddInt32(&cluster2WatchCount, 1)
+ w.WriteHeader(http.StatusOK)
+ }))
+ defer cluster2Server.Close()
+
+ cluster1Addr := cluster1Server.Listener.Addr().String()
+ cluster2Addr := cluster2Server.Listener.Addr().String()
+
+ service := &RaftRegistryService{
+ metadata: metadata.NewMetadata(),
+ initAddresses: sync.Map{},
+ cfg: &RaftConfig{TokenValidityInMilliseconds:
1740000},
+ tokenTimestamp: time.Now().UnixMilli(),
+ httpClient: &http.Client{},
+ random: rand.New(rand.NewSource(time.Now().UnixNano())),
+ }
+
+ service.metadata.RefreshMetadata("cluster1", metadata.MetadataResponse{
+ Term: 3,
+ Nodes: []*metadata.Node{
+ newTestMetadataNode(t, cluster1Addr, "group-a",
metadata.LEADER),
+ newTestMetadataNode(t, cluster1Addr, "group-b",
metadata.FOLLOWER),
+ },
+ })
+ service.metadata.RefreshMetadata("cluster2", metadata.MetadataResponse{
+ Term: 5,
+ Nodes: []*metadata.Node{
+ newTestMetadataNode(t, cluster2Addr, "group-c",
metadata.LEADER),
+ },
+ })
+
+ service.initAddresses.Store("cluster1",
[]*ServiceInstance{newTestServiceInstance(t, cluster1Addr)})
+ service.initAddresses.Store("cluster2",
[]*ServiceInstance{newTestServiceInstance(t, cluster2Addr)})
+
+ result, err := service.watch()
+
+ require.NoError(t, err)
+ assert.True(t, result)
+ assert.Equal(t, int32(1), atomic.LoadInt32(&cluster1WatchCount))
+ assert.Equal(t, int32(1), atomic.LoadInt32(&cluster2WatchCount))
+
+ formMu.Lock()
+ defer formMu.Unlock()
+ require.NotNil(t, cluster1Form)
+ assert.Equal(t, []string{"3"}, cluster1Form["group-a"])
+ assert.Equal(t, []string{"3"}, cluster1Form["group-b"])
+}
+
+func TestRefreshAliveLookup(t *testing.T) {
+ serviceConfig := newTestServiceConfig()
+
+ mockServer := httptest.NewServer(http.HandlerFunc(func(w
http.ResponseWriter, r *http.Request) {
+ switch r.URL.Path {
+ case "/api/v1/auth/login":
+ w.Header().Set("Content-Type", "application/json")
+ w.WriteHeader(http.StatusOK)
+ _ = json.NewEncoder(w).Encode(map[string]string{
+ "code": "200",
+ "data": "mock-jwt-token",
+ })
+
+ case "/metadata/v1/cluster":
+ resp := metadata.MetadataResponse{
+ Term: 1,
+ Nodes: []*metadata.Node{
+ {
+ Control:
&metadata.Endpoint{Host: "127.0.0.1", Port: 7001},
+ Transaction:
&metadata.Endpoint{Host: "127.0.0.1", Port: 8001},
+ Internal:
&metadata.Endpoint{Host: "127.0.0.1", Port: 9001},
+ Group: "default",
+ Role: metadata.LEADER,
+ Version: "2.5.0",
+ Metadata:
map[string]interface{}{},
+ },
+ },
+ }
+ w.Header().Set("Content-Type", "application/json")
+ w.WriteHeader(http.StatusOK)
+ _ = json.NewEncoder(w).Encode(resp)
+
+ default:
+ w.WriteHeader(http.StatusNotFound)
+ }
+ }))
+ defer mockServer.Close()
+
+ mockAddr := mockServer.Listener.Addr().String()
+ testRegistryConfig := newTestRegistryConfig(mockAddr)
+
+ service := NewRaftRegistryService(serviceConfig, testRegistryConfig)
+ defer service.Close()
+
+ _, err := service.Lookup("default_tx_group")
+ assert.NoError(t, err)
+
+ aliveInstances := []*ServiceInstance{
+ {Addr: "127.0.0.1", Port: 8001},
+ {Addr: "127.0.0.1", Port: 8002},
+ {Addr: "127.0.0.1", Port: 8003},
+ }
+
+ result, err := service.RefreshAliveLookup("default_tx_group",
aliveInstances)
+ assert.NoError(t, err)
+
+ expectedResult := []*ServiceInstance{
+ {Addr: "127.0.0.1", Port: 8002},
+ {Addr: "127.0.0.1", Port: 8003},
+ }
+ assert.Equal(t, expectedResult, result)
+
+ emptyResult, err := service.RefreshAliveLookup("default_tx_group",
[]*ServiceInstance{})
+ assert.NoError(t, err)
+ assert.Empty(t, emptyResult)
+
+ _, err = service.RefreshAliveLookup("non_existent_group",
aliveInstances)
+ assert.Error(t, err)
+ assert.Contains(t, err.Error(), "cluster not found")
+}
+
+func TestTypeAssertionSafety(t *testing.T) {
+ service := &RaftRegistryService{
+ aliveNodes: sync.Map{},
+ vgroupMapping: map[string]string{"test_group": "test_cluster"},
+ metadata: metadata.NewMetadata(),
+ random: rand.New(rand.NewSource(time.Now().UnixNano())),
+ }
+
+ service.metadata.RefreshMetadata("test_cluster",
metadata.MetadataResponse{
+ Term: 1,
+ Nodes: []*metadata.Node{
+ {
+ Control: &metadata.Endpoint{Host:
"127.0.0.1", Port: 7001},
+ Transaction: &metadata.Endpoint{Host:
"127.0.0.1", Port: 8001},
+ Internal: &metadata.Endpoint{Host:
"127.0.0.1", Port: 9001},
+ Group: "",
+ Role: metadata.LEADER,
+ Version: "2.5.0",
+ Metadata: map[string]interface{}{},
+ },
+ },
+ })
+
+ service.aliveNodes.Store("test_group", "invalid_type_string")
+ service.currentTransactionServiceGroup = "test_group"
+
+ _, err := service.queryHttpAddress("test_cluster", "")
+ assert.Error(t, err)
+ assert.Contains(t, err.Error(), "invalid alive nodes type")
+
+ service.aliveNodes.Store("test_group", []*ServiceInstance{
+ {Addr: "127.0.0.1", Port: 8001},
+ })
+
+ address, err := service.queryHttpAddress("test_cluster", "")
+ assert.NoError(t, err)
+ assert.NotEmpty(t, address)
+}
diff --git a/pkg/util/log/logging.go b/pkg/util/log/logging.go
index ae333e70..c91923ec 100644
--- a/pkg/util/log/logging.go
+++ b/pkg/util/log/logging.go
@@ -21,6 +21,7 @@ import (
"bytes"
"fmt"
"os"
+ "sync"
"time"
getty "github.com/apache/dubbo-getty"
@@ -101,6 +102,7 @@ type Logger interface {
}
var (
+ loggerMu sync.RWMutex
log Logger
zapLogger *zap.Logger
@@ -142,9 +144,15 @@ func encodeCaller(caller zapcore.EntryCaller, enc
zapcore.PrimitiveArrayEncoder)
func Init() {
zapLoggerConfig.EncoderConfig = zapLoggerEncoderConfig
- zapLogger, _ = zapLoggerConfig.Build(zap.AddCallerSkip(1))
- log = zapLogger.Sugar()
- getty.SetLogger(log)
+ logger, _ := zapLoggerConfig.Build(zap.AddCallerSkip(1))
+ sugared := logger.Sugar()
+
+ loggerMu.Lock()
+ zapLogger = logger
+ log = sugared
+ loggerMu.Unlock()
+
+ getty.SetLogger(sugared)
}
func InitWithOption(logPath string, level LogLevel) {
@@ -164,68 +172,89 @@ func InitWithOption(logPath string, level LogLevel) {
encoder := zapcore.NewConsoleEncoder(encoderConfig)
core := zapcore.NewCore(encoder, syncer,
zap.NewAtomicLevelAt(zapcore.Level(level)))
- zapLogger = zap.New(core, zap.AddCaller())
+ logger := zap.New(core, zap.AddCaller())
+ sugared := logger.Sugar()
+
+ loggerMu.Lock()
+ zapLogger = logger
+ log = sugared
+ loggerMu.Unlock()
- log = zapLogger.Sugar()
- getty.SetLogger(log)
+ getty.SetLogger(sugared)
}
// SetLogger: customize yourself logger.
func SetLogger(logger Logger) {
+ loggerMu.Lock()
log = logger
+ loggerMu.Unlock()
}
// GetLogger get logger
func GetLogger() Logger {
+ loggerMu.RLock()
+ defer loggerMu.RUnlock()
+ return log
+}
+
+func currentLogger() Logger {
+ loggerMu.RLock()
+ defer loggerMu.RUnlock()
return log
}
// Debug ...
func Debug(v ...interface{}) {
- if log == nil {
+ logger := currentLogger()
+ if logger == nil {
return
}
- log.Debug(v...)
+ logger.Debug(v...)
}
// Debugf ...
func Debugf(format string, v ...interface{}) {
- if log == nil {
+ logger := currentLogger()
+ if logger == nil {
return
}
- log.Debugf(format, v...)
+ logger.Debugf(format, v...)
}
// Info ...
func Info(v ...interface{}) {
- if log == nil {
+ logger := currentLogger()
+ if logger == nil {
return
}
- log.Info(v...)
+ logger.Info(v...)
}
// Infof ...
func Infof(format string, v ...interface{}) {
- if log == nil {
+ logger := currentLogger()
+ if logger == nil {
return
}
- log.Infof(format, v...)
+ logger.Infof(format, v...)
}
// Warn ...
func Warn(v ...interface{}) {
- if log == nil {
+ logger := currentLogger()
+ if logger == nil {
return
}
- log.Warn(v...)
+ logger.Warn(v...)
}
// Warnf ...
func Warnf(format string, v ...interface{}) {
- if log == nil {
+ logger := currentLogger()
+ if logger == nil {
return
}
- log.Warnf(format, v...)
+ logger.Warnf(format, v...)
}
// Error ...
@@ -235,11 +264,12 @@ func Error(v ...interface{}) {
fmt.Fprintf(os.Stderr, "ERROR: %v\n", v)
}
}()
- if log == nil {
+ logger := currentLogger()
+ if logger == nil {
fmt.Fprintf(os.Stderr, "ERROR: %v\n", v)
return
}
- log.Error(v...)
+ logger.Error(v...)
}
// Errorf ...
@@ -251,49 +281,54 @@ func Errorf(format string, v ...interface{}) {
fmt.Fprintf(os.Stderr, "\n")
}
}()
- if log == nil {
+ logger := currentLogger()
+ if logger == nil {
fmt.Fprintf(os.Stderr, "ERROR: ")
fmt.Fprintf(os.Stderr, format, v...)
fmt.Fprintf(os.Stderr, "\n")
return
}
- log.Errorf(format, v...)
+ logger.Errorf(format, v...)
}
// Panic ...
func Panic(v ...interface{}) {
- if log == nil {
+ logger := currentLogger()
+ if logger == nil {
panic(v)
}
- log.Panic(v...)
+ logger.Panic(v...)
}
// Panicf ...
func Panicf(format string, v ...interface{}) {
- if log == nil {
+ logger := currentLogger()
+ if logger == nil {
panic(fmt.Sprintf(format, v...))
}
- log.Panicf(format, v...)
+ logger.Panicf(format, v...)
}
// Fatal ...
func Fatal(v ...interface{}) {
- if log == nil {
+ logger := currentLogger()
+ if logger == nil {
fmt.Fprintf(os.Stderr, "FATAL: %v\n", v)
os.Exit(1)
return
}
- log.Fatal(v...)
+ logger.Fatal(v...)
}
// Fatalf ...
func Fatalf(format string, v ...interface{}) {
- if log == nil {
+ logger := currentLogger()
+ if logger == nil {
fmt.Fprintf(os.Stderr, "FATAL: ")
fmt.Fprintf(os.Stderr, format, v...)
fmt.Fprintf(os.Stderr, "\n")
os.Exit(1)
return
}
- log.Fatalf(format, v...)
+ logger.Fatalf(format, v...)
}
diff --git a/testdata/conf/seatago.yml b/testdata/conf/seatago.yml
index 12dbe25e..e82625f9 100644
--- a/testdata/conf/seatago.yml
+++ b/testdata/conf/seatago.yml
@@ -136,6 +136,10 @@ seata:
etcd3:
cluster: "default"
server-addr: "http://localhost:2379";
+ raft:
+ metadata-max-age-ms: 30000
+ server-addr: 127.0.0.1:7091
+ token-validity-in-milliseconds: 1740000
naming-server:
cluster: "default"
server-addr: "127.0.0.1:8081"
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
