Copilot commented on code in PR #1107:
URL:
https://github.com/apache/incubator-seata-go/pull/1107#discussion_r3132204675
##########
testdata/conf/seatago.yml:
##########
@@ -136,6 +139,10 @@ seata:
etcd3:
cluster: "default"
server-addr: "http://localhost:2379";
+ raft:
+ metadata-max-age-ms: 30000
+ server-ddr:
Review Comment:
`server-ddr` appears to be a typo and has no value, which will prevent raft
server addresses from being configured via YAML. Rename this key to
`server-addr` (and set a value, e.g. `127.0.0.1:7091`), keeping it consistent
with the `raft.server-addr` flag/koanf key used in code.
```suggestion
server-addr: 127.0.0.1:7091
```
##########
pkg/discovery/config.go:
##########
@@ -90,3 +97,15 @@ 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 RaftConfig struct {
+ MetadataMaxAgeMs int64 `yaml:"metadata-max-age-ms"
json:"metadata-max-age-ms" koanf:"metadata-max-age-ms"`
+ ServerAddr string `yaml:"serverAddr"
json:"server-addr" koanf:"server-addr"`
+ TokenValidityInMilliseconds int64
`yaml:"token-validity-in-milliseconds" json:"token-validity-in-milliseconds"
koanf:"token-validity-in-milliseconds"`
+}
Review Comment:
`ServerAddr` uses `yaml:\"serverAddr\"` but the rest of the config and flags
use `server-addr`. As a result, YAML config with `server-addr` will not
populate `ServerAddr`. Change the YAML tag to `yaml:\"server-addr\"` to match
`json/koanf` and the example config.
##########
pkg/discovery/metadata/metadata.go:
##########
@@ -0,0 +1,161 @@
+/*
+ * 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) {
+ var nodes []*Node
+ for _, node := range response.Nodes {
+ if node.Role == LEADER {
+ groupMapAny, _ := m.leaders.LoadOrStore(clusterName,
&sync.Map{})
+ groupMap := groupMapAny.(*sync.Map)
+ groupMap.Store(node.Group, node)
+ }
+ nodes = append(nodes, node)
+ }
Review Comment:
`RefreshMetadata` stores *all* nodes under `nodes[0].Group`, which breaks
lookups if the response contains nodes from multiple groups (nodes will be
attached to the wrong group key). Consider grouping `response.Nodes` by
`node.Group` and calling `SetNodes` / `termMap.Store` per group (and
clearing/updating old group entries as appropriate).
##########
pkg/discovery/metadata/metadata.go:
##########
@@ -0,0 +1,161 @@
+/*
+ * 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) {
+ var nodes []*Node
+ for _, node := range response.Nodes {
+ if node.Role == LEADER {
+ groupMapAny, _ := m.leaders.LoadOrStore(clusterName,
&sync.Map{})
+ groupMap := groupMapAny.(*sync.Map)
+ groupMap.Store(node.Group, node)
+ }
+ nodes = append(nodes, node)
+ }
+
+ switch response.StoreMode {
+ case "RAFT":
+ m.storeMode = RAFT
+ default:
+ m.storeMode = FILE
+ }
+
+ if len(nodes) > 0 {
+ group := nodes[0].Group
+ m.SetNodes(clusterName, group, nodes)
+ termMapAny, _ := m.clusterTerm.LoadOrStore(clusterName,
&sync.Map{})
+ termMap := termMapAny.(*sync.Map)
+ termMap.Store(group, response.Term)
+ }
+}
Review Comment:
`RefreshMetadata` stores *all* nodes under `nodes[0].Group`, which breaks
lookups if the response contains nodes from multiple groups (nodes will be
attached to the wrong group key). Consider grouping `response.Nodes` by
`node.Group` and calling `SetNodes` / `termMap.Store` per group (and
clearing/updating old group entries as appropriate).
##########
pkg/discovery/raft.go:
##########
@@ -0,0 +1,520 @@
+/*
+ * 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.NamingserverAddr,
+ username: raftConfig.Username,
+ password: raftConfig.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 doesnt exist")
+ }
+ 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 {
+ r.mu.RLock()
+ clusterName :=
r.currentTransactionClusterName
+ r.mu.RUnlock()
+ 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) {
+ header := map[string]string{
+ "Content-Type": "application/x-www-form-urlencoded",
+ }
+ clusterNames := r.clusterNamesFromInit()
+ for _, clusterName := range clusterNames {
+ groupTerms := r.metadata.GetClusterTerm(clusterName)
+ if groupTerms == nil {
+ groupTerms = map[string]int64{"": 0}
+ }
+ for group := range groupTerms {
+ tcAddress, err := r.queryHttpAddress(clusterName, group)
+ if err != nil {
+ log.Infof("no tc address to watch for cluster
%s: %v", clusterName, err)
+ continue
+ }
+ if r.isTokenExpired() {
+ if err = r.refreshToken(); err != nil {
+ return false, err
+ }
+ }
+ if r.jwtToken != "" {
+ header["Authorization"] = r.jwtToken
+ }
+
+ form := url.Values{}
+ for k, v := range groupTerms {
+ form.Set(k, strconv.FormatInt(v, 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")
+ }
+ return resp.StatusCode == http.StatusOK, nil
+ }
+ }
+ return false, 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)
Review Comment:
`dataVal := node[\"data\"].(string)` can panic if `data` is missing or not a
string (e.g., auth service returns `{data:{...}}` or `null`). Use a checked
type assertion and return a descriptive error when the expected field/type is
not present.
```suggestion
data, ok := node["data"]
if !ok {
return errors.New("invalid auth response: missing data field")
}
dataVal, ok := data.(string)
if !ok {
return fmt.Errorf("invalid auth response: data field is %T,
want string", data)
}
```
##########
pkg/discovery/raft.go:
##########
@@ -0,0 +1,520 @@
+/*
+ * 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.NamingserverAddr,
+ username: raftConfig.Username,
+ password: raftConfig.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 doesnt exist")
+ }
+ 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 {
+ r.mu.RLock()
+ clusterName :=
r.currentTransactionClusterName
+ r.mu.RUnlock()
+ 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) {
+ header := map[string]string{
+ "Content-Type": "application/x-www-form-urlencoded",
+ }
+ clusterNames := r.clusterNamesFromInit()
+ for _, clusterName := range clusterNames {
+ groupTerms := r.metadata.GetClusterTerm(clusterName)
+ if groupTerms == nil {
+ groupTerms = map[string]int64{"": 0}
+ }
+ for group := range groupTerms {
+ tcAddress, err := r.queryHttpAddress(clusterName, group)
Review Comment:
`watch()` returns on the first request attempted, so it never watches
remaining groups within the cluster or additional clusters from
`initAddresses`. Additionally, the form payload sets *all* group terms while
iterating per-group, causing redundant/misaligned requests. Consider either (a)
sending a single watch request per cluster/node containing all group terms and
processing all clusters before returning, or (b) limiting the form to the
current `group` and aggregating results across the loops.
##########
pkg/discovery/raft.go:
##########
@@ -0,0 +1,520 @@
+/*
+ * 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.NamingserverAddr,
+ username: raftConfig.Username,
+ password: raftConfig.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 doesnt exist")
+ }
+ 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 {
+ r.mu.RLock()
+ clusterName :=
r.currentTransactionClusterName
+ r.mu.RUnlock()
+ 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) {
+ header := map[string]string{
+ "Content-Type": "application/x-www-form-urlencoded",
+ }
+ clusterNames := r.clusterNamesFromInit()
+ for _, clusterName := range clusterNames {
+ groupTerms := r.metadata.GetClusterTerm(clusterName)
+ if groupTerms == nil {
+ groupTerms = map[string]int64{"": 0}
+ }
+ for group := range groupTerms {
+ tcAddress, err := r.queryHttpAddress(clusterName, group)
+ if err != nil {
+ log.Infof("no tc address to watch for cluster
%s: %v", clusterName, err)
+ continue
+ }
+ if r.isTokenExpired() {
+ if err = r.refreshToken(); err != nil {
+ return false, err
+ }
+ }
+ if r.jwtToken != "" {
+ header["Authorization"] = r.jwtToken
+ }
+
+ form := url.Values{}
+ for k, v := range groupTerms {
+ form.Set(k, strconv.FormatInt(v, 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")
+ }
+ return resp.StatusCode == http.StatusOK, nil
+ }
+ }
Review Comment:
`watch()` returns on the first request attempted, so it never watches
remaining groups within the cluster or additional clusters from
`initAddresses`. Additionally, the form payload sets *all* group terms while
iterating per-group, causing redundant/misaligned requests. Consider either (a)
sending a single watch request per cluster/node containing all group terms and
processing all clusters before returning, or (b) limiting the form to the
current `group` and aggregating results across the loops.
##########
pkg/discovery/raft.go:
##########
@@ -0,0 +1,520 @@
+/*
+ * 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.NamingserverAddr,
+ username: raftConfig.Username,
+ password: raftConfig.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 doesnt exist")
+ }
+ 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 {
+ r.mu.RLock()
+ clusterName :=
r.currentTransactionClusterName
+ r.mu.RUnlock()
+ 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) {
+ header := map[string]string{
+ "Content-Type": "application/x-www-form-urlencoded",
+ }
+ clusterNames := r.clusterNamesFromInit()
+ for _, clusterName := range clusterNames {
+ groupTerms := r.metadata.GetClusterTerm(clusterName)
+ if groupTerms == nil {
+ groupTerms = map[string]int64{"": 0}
+ }
+ for group := range groupTerms {
+ tcAddress, err := r.queryHttpAddress(clusterName, group)
+ if err != nil {
+ log.Infof("no tc address to watch for cluster
%s: %v", clusterName, err)
+ continue
+ }
+ if r.isTokenExpired() {
+ if err = r.refreshToken(); err != nil {
+ return false, err
+ }
+ }
+ if r.jwtToken != "" {
+ header["Authorization"] = r.jwtToken
+ }
+
+ form := url.Values{}
+ for k, v := range groupTerms {
+ form.Set(k, strconv.FormatInt(v, 10))
+ }
Review Comment:
`watch()` returns on the first request attempted, so it never watches
remaining groups within the cluster or additional clusters from
`initAddresses`. Additionally, the form payload sets *all* group terms while
iterating per-group, causing redundant/misaligned requests. Consider either (a)
sending a single watch request per cluster/node containing all group terms and
processing all clusters before returning, or (b) limiting the form to the
current `group` and aggregating results across the loops.
##########
pkg/discovery/raft.go:
##########
@@ -0,0 +1,520 @@
+/*
+ * 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.NamingserverAddr,
+ username: raftConfig.Username,
+ password: raftConfig.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 doesnt exist")
Review Comment:
The error message has a typo and lacks context. Consider changing it to
something like `cluster doesn't exist for serviceGroup=%s` (including the
`key`) to make debugging misconfigured vgroup mappings easier.
```suggestion
return nil, fmt.Errorf("cluster doesn't exist for
serviceGroup=%s", key)
```
##########
pkg/discovery/raft.go:
##########
@@ -0,0 +1,520 @@
+/*
+ * 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.NamingserverAddr,
+ username: raftConfig.Username,
+ password: raftConfig.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 doesnt exist")
+ }
+ 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 {
+ r.mu.RLock()
+ clusterName :=
r.currentTransactionClusterName
+ r.mu.RUnlock()
+ 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) {
+ header := map[string]string{
+ "Content-Type": "application/x-www-form-urlencoded",
+ }
+ clusterNames := r.clusterNamesFromInit()
+ for _, clusterName := range clusterNames {
+ groupTerms := r.metadata.GetClusterTerm(clusterName)
+ if groupTerms == nil {
+ groupTerms = map[string]int64{"": 0}
+ }
+ for group := range groupTerms {
+ tcAddress, err := r.queryHttpAddress(clusterName, group)
+ if err != nil {
+ log.Infof("no tc address to watch for cluster
%s: %v", clusterName, err)
+ continue
+ }
+ if r.isTokenExpired() {
+ if err = r.refreshToken(); err != nil {
+ return false, err
+ }
+ }
+ if r.jwtToken != "" {
+ header["Authorization"] = r.jwtToken
+ }
+
+ form := url.Values{}
+ for k, v := range groupTerms {
+ form.Set(k, strconv.FormatInt(v, 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")
+ }
+ return resp.StatusCode == http.StatusOK, nil
+ }
+ }
+ return false, 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)
Review Comment:
On `401 Unauthorized`, the code refreshes the token but then immediately
returns an error without retrying the metadata request using the new token.
This prevents automatic recovery from token expiration. A common approach is to
refresh the token and retry the request once (guarded to avoid infinite loops)
before returning an error.
```suggestion
u := fmt.Sprintf("http://%s/metadata/v1/cluster";, tcAddress)
for attempt := 0; attempt < 2; attempt++ {
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)
}
if r.jwtToken != "" {
req.Header.Set("Authorization", r.jwtToken)
}
resp, err := r.doRequest(req, 1*time.Second)
if err != nil {
return fmt.Errorf("http get cluster failed: %w", err)
}
if resp.StatusCode == http.StatusOK {
defer resp.Body.Close()
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
}
if resp.StatusCode == http.StatusUnauthorized {
resp.Body.Close()
if attempt == 0 {
if err = r.refreshToken(); err != nil {
return err
}
continue
}
return fmt.Errorf("authentication failed! you should
configure the correct username and password")
}
resp.Body.Close()
return fmt.Errorf("unexpected status code: %d", resp.StatusCode)
}
return fmt.Errorf("authentication failed! you should configure the
correct username and password")
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
