Copilot commented on code in PR #1106: URL: https://github.com/apache/incubator-seata-go/pull/1106#discussion_r3132211418
########## pkg/discovery/naming_server.go: ########## @@ -0,0 +1,673 @@ +/* + * 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" + "errors" + "fmt" + gostnet "github.com/dubbogo/gost/net" + "go.uber.org/zap" + "net/http" + "net/url" + "seata.apache.org/seata-go/v2/pkg/util/rand" + "strconv" + "strings" + "sync" + "sync/atomic" + "time" +) + +const ( + httpPrefix = "http://"; + healthCheckThreshold = 3 + longPollTimeoutPeriod = 28 * time.Second + authorizationHeader = "Authorization" + contentTypeJSON = "application/json" + namingAddrCacheTTL = 30 * time.Second + failureRecoveryThreshold = 3 + maxRetryAttempts = 3 + retryDelayMs = 1000 +) + +type MetaResponse struct { + Term int64 `json:"term"` + ClusterList []Cluster `json:"clusterList"` +} + +type Cluster struct { + ClusterName string `json:"clusterName"` + ClusterType string `json:"clusterType"` + UnitData []Unit `json:"unitData"` +} + +type Unit struct { + UnitName string `json:"unitName"` + NamingInstanceList []NamingServerNode `json:"namingInstanceList"` +} + +type NamingServerNode struct { + Role ClusterRole `json:"role"` + Term int64 `json:"term"` + Transaction Endpoint `json:"transaction"` + Control Endpoint `json:"control"` + Internal Endpoint `json:"internal"` + Group string `json:"group"` + Version string `json:"version"` + Metadata map[string]interface{} `json:"metadata"` + TimeStamp int64 `json:"timeStamp"` + Weight float64 `json:"weight"` + Healthy bool `json:"healthy"` + Unit string `json:"unit"` +} + +type Endpoint struct { + Host string `json:"host"` + Port int `json:"port"` + Protocol string `json:"protocol"` +} + +type ExternalEndpoint struct { + Host string `json:"host"` + ControlPort int `json:"controlPort"` + TransactionPort int `json:"transactionPort"` +} + +type ClusterRole string + +const ( + ClusterRoleLeader ClusterRole = "LEADER" + ClusterRoleMember ClusterRole = "MEMBER" +) + +type NamingServerClient struct { + config *NamingServerConfig + logger *zap.Logger + mu sync.Mutex + instance *NamingServerClient + term int64 + jwtToken string + tokenTimeStamp int64 + isSubscribed bool + namingAddrCache string + namingAddrCacheTimestamp int64 + + availableNamingMap sync.Map + vgroupAddressMap sync.Map + listenerServiceMap sync.Map + + healthCheckTicker *time.Ticker + closeChan chan struct{} + wg sync.WaitGroup + + httpClient *http.Client + longPollClient *http.Client +} + +type NamingListener interface { + OnEvent(vGroup string) error +} + +type NamingServerRegistryService struct { + client *NamingServerClient +} + +var _ NamingserverRegistry = (*NamingServerRegistryService)(nil) + +func (n *NamingServerRegistryService) Lookup(key string) ([]*ServiceInstance, error) { + return n.client.Lookup(key) +} + +func (n *NamingServerRegistryService) Close() { + n.client.Close() +} + +func newNamingServerRegistryService(_ *ServiceConfig, cfg *NamingServerConfig) RegistryService { + client := GetInstance(cfg) + return &NamingServerRegistryService{ + client: client, + } +} + +var ( + namingServerInstance *NamingServerClient + namingServerOnce sync.Once +) + +func GetInstance(config *NamingServerConfig) *NamingServerClient { + namingServerOnce.Do(func() { + namingServerInstance = &NamingServerClient{ + config: config, + logger: zap.L().Named("naming-server-client"), + closeChan: make(chan struct{}), + healthCheckTicker: time.NewTicker(time.Duration(config.HeartbeatPeriod) * time.Millisecond), + httpClient: &http.Client{Timeout: 3 * time.Second}, + longPollClient: &http.Client{Timeout: 30 * time.Second}, + } + // Initialize available naming server addresses from config + namingServerInstance.initNamingAddrs() + namingServerInstance.initHealthCheck() + }) + return namingServerInstance +} + +func resetInstance() { + if namingServerInstance != nil { + namingServerInstance.Close() + namingServerInstance.mu.Lock() + namingServerInstance.clearNamingAddrCache() + namingServerInstance.mu.Unlock() + } + namingServerInstance = nil + namingServerOnce = sync.Once{} +} + +func (c *NamingServerClient) initNamingAddrs() { + addrs := c.getNamingAddrs() + for _, addr := range addrs { + c.availableNamingMap.Store(addr, int32(0)) + } +} + +func (c *NamingServerClient) initHealthCheck() { + c.wg.Add(1) + go func() { + defer c.wg.Done() + for { + select { + case <-c.healthCheckTicker.C: + urlList := c.getNamingAddrs() + c.checkAvailableNamingAddr(urlList) + case <-c.closeChan: + return + } + } + }() +} + +func (c *NamingServerClient) checkAvailableNamingAddr(urlList []string) { + for _, addr := range urlList { + isHealthy := c.doHealthCheck(addr) + + val, _ := c.availableNamingMap.LoadOrStore(addr, int32(0)) + failCount := val.(int32) + + if !isHealthy { + newFailCount := atomic.AddInt32(&failCount, 1) + c.availableNamingMap.Store(addr, newFailCount) + if newFailCount == 1 { + c.logger.Warn("naming server check failed", zap.String("addr", addr), zap.Int32("failCount", newFailCount)) + } else if newFailCount >= healthCheckThreshold { + c.logger.Error("naming server offline", zap.String("addr", addr), zap.Int32("failCount", newFailCount)) + c.mu.Lock() + if c.namingAddrCache == addr { + c.clearNamingAddrCache() + } + c.mu.Unlock() + } + } else { + if failCount > 0 { + c.availableNamingMap.Store(addr, int32(0)) + c.logger.Info("naming server recovered", zap.String("addr", addr), zap.Int32("previousFailCount", failCount)) + } + } + } +} + +func (c *NamingServerClient) doHealthCheck(addr string) bool { + checkURL := fmt.Sprintf("%s%s/naming/v1/health", httpPrefix, addr) + + req, err := http.NewRequest(http.MethodGet, checkURL, nil) + if err != nil { + c.logger.Error("create health check request failed", zap.Error(err)) + return false + } + req.Header.Set("Content-Type", contentTypeJSON) + + resp, err := c.httpClient.Do(req) + if err != nil { + c.logger.Error("health check failed", zap.String("addr", addr), zap.Error(err)) + return false + } + defer resp.Body.Close() + return resp.StatusCode == http.StatusOK +} + +func (c *NamingServerClient) getNamingAddrs() []string { + return strings.Split(c.config.ServerAddr, ",") +} + +func (c *NamingServerClient) Lookup(vGroup string) ([]*ServiceInstance, error) { + if !c.isSubscribed { + if err := c.RefreshGroup(vGroup); err != nil { + return nil, fmt.Errorf("refresh group failed: %w", err) + } + listener := &RefreshListener{client: c} + if err := c.Subscribe(vGroup, listener); err != nil { + return nil, fmt.Errorf("subscribe failed: %w", err) + } + } + + val, ok := c.vgroupAddressMap.Load(vGroup) + if !ok { + if err := c.RefreshGroup(vGroup); err != nil { + return nil, fmt.Errorf("refresh group failed: %w", err) + } + val, ok = c.vgroupAddressMap.Load(vGroup) + if !ok { + return nil, errors.New("no nodes found for vgroup") + } + } + + nodes := val.([]NamingServerNode) + + var instances []*ServiceInstance + for _, node := range nodes { + if !node.Healthy { + continue + } + + if node.Transaction.Host == "" || node.Transaction.Port <= 0 || node.Transaction.Port > 65535 { + c.logger.Warn("invalid node address", zap.String("host", node.Transaction.Host), zap.Int("port", node.Transaction.Port)) + continue + } + + instances = append(instances, &ServiceInstance{ + Addr: node.Transaction.Host, + Port: node.Transaction.Port, + }) + } + return instances, nil +} + +func (c *NamingServerClient) RefreshGroup(vGroup string) error { + namingAddr, err := c.getNamingAddr() + if err != nil { + return err + } + + if c.isTokenExpired() { + if err := c.RefreshToken(namingAddr); err != nil { + return err + } + } + + params := url.Values{} + params.Add("vGroup", vGroup) + params.Add("namespace", c.config.Namespace) + + discoveryURL := fmt.Sprintf("%s%s/naming/v1/discovery?%s", httpPrefix, namingAddr, params.Encode()) + req, err := http.NewRequest(http.MethodGet, discoveryURL, nil) + if err != nil { + return err + } + if c.jwtToken != "" { + req.Header.Set(authorizationHeader, c.jwtToken) + } + req.Header.Set("Content-Type", contentTypeJSON) + + resp, err := c.httpClient.Do(req) + if err != nil { + return fmt.Errorf("discovery request failed: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + if resp.StatusCode == http.StatusUnauthorized { + return fmt.Errorf("discovery failed: unauthorized (401), authentication required. please configure username and password in the naming server config") + } + return fmt.Errorf("discovery failed, status: %d", resp.StatusCode) + } + + var metaResp MetaResponse + if err := json.NewDecoder(resp.Body).Decode(&metaResp); err != nil { + return fmt.Errorf("decode meta response failed: %w", err) + } + + return c.handleMetadata(&metaResp, vGroup) +} + +func (c *NamingServerClient) getNamingAddr() (string, error) { + c.mu.Lock() + defer c.mu.Unlock() + + now := time.Now().UnixMilli() + + if c.namingAddrCache != "" && + now-c.namingAddrCacheTimestamp < namingAddrCacheTTL.Milliseconds() { + + if val, ok := c.availableNamingMap.Load(c.namingAddrCache); ok { + failCount := val.(int32) + if failCount < healthCheckThreshold { + return c.namingAddrCache, nil + } + } + + c.clearNamingAddrCache() + } + + var availableAddrs []string + c.availableNamingMap.Range(func(key, value interface{}) bool { + addr := key.(string) + failCount := value.(int32) + if failCount < healthCheckThreshold { + availableAddrs = append(availableAddrs, addr) + } + return true + }) + + if len(availableAddrs) == 0 { + return "", errors.New("no available naming server") + } + + addr := availableAddrs[rand.RandIntn(len(availableAddrs))] + c.namingAddrCache = addr + c.namingAddrCacheTimestamp = now + return addr, nil +} + +func (c *NamingServerClient) clearNamingAddrCache() { + c.namingAddrCache = "" + c.namingAddrCacheTimestamp = 0 +} + +func (c *NamingServerClient) isTokenExpired() bool { + if c.config.Username == "" || c.config.Password == "" { + return false + } + + ts := atomic.LoadInt64(&c.tokenTimeStamp) + if ts == 0 { + return true + } + + return time.Now().UnixMilli() >= ts+c.config.TokenValidityInMilliseconds +} + +func (c *NamingServerClient) RefreshToken(addr string) error { + if c.config.Username == "" || c.config.Password == "" { + return nil + } + + var lastErr error + for attempt := 1; attempt <= maxRetryAttempts; attempt++ { + err := c.doRefreshToken(addr) + if err == nil { + return nil + } + + lastErr = err + if attempt < maxRetryAttempts { + c.logger.Warn("token refresh failed, retrying", + zap.String("addr", addr), + zap.Int("attempt", attempt), + zap.Error(err)) + time.Sleep(time.Duration(retryDelayMs*attempt) * time.Millisecond) + } + } + + c.logger.Error("token refresh failed after all retries", + zap.String("addr", addr), + zap.Int("maxAttempts", maxRetryAttempts), + zap.Error(lastErr)) + return fmt.Errorf("token refresh failed after %d attempts: %w", maxRetryAttempts, lastErr) +} + +func (c *NamingServerClient) doRefreshToken(addr string) error { + loginURL := fmt.Sprintf("%s%s/api/v1/auth/login", httpPrefix, addr) + + loginReq := struct { + Username string `json:"username"` + Password string `json:"password"` + }{ + Username: c.config.Username, + Password: c.config.Password, + } + + body, err := json.Marshal(loginReq) + if err != nil { + return fmt.Errorf("marshal login request failed: %w", err) + } + + req, err := http.NewRequest(http.MethodPost, loginURL, strings.NewReader(string(body))) + if err != nil { + return err + } + req.Header.Set("Content-Type", contentTypeJSON) + + resp, err := c.httpClient.Do(req) + if err != nil { + return fmt.Errorf("login request failed: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + c.logger.Error("login request failed", + zap.String("addr", addr), + zap.Int("statusCode", resp.StatusCode), + zap.String("username", c.config.Username)) + return fmt.Errorf("authentication failed: status %d", resp.StatusCode) + } + + var loginResp struct { + Success bool `json:"success"` + Data string `json:"data"` + Code interface{} `json:"code"` + Msg string `json:"msg"` + } + if err := json.NewDecoder(resp.Body).Decode(&loginResp); err != nil { + return fmt.Errorf("decode login response failed: %w", err) + } + + c.logger.Debug("login response received", + zap.Bool("success", loginResp.Success), + zap.String("code", fmt.Sprint(loginResp.Code)), + zap.String("msg", loginResp.Msg), + zap.String("data", loginResp.Data)) + + if !loginResp.Success { + errMsg := fmt.Sprintf("authentication failed: success=false, msg=%s", loginResp.Msg) + c.logger.Error(errMsg) + return fmt.Errorf(errMsg) + } + if loginResp.Data == "" { + return fmt.Errorf("authentication failed: token is empty") + } + + c.jwtToken = loginResp.Data + atomic.StoreInt64(&c.tokenTimeStamp, time.Now().UnixMilli()) + c.logger.Info("token refreshed successfully", zap.String("addr", addr)) + return nil +} + +func (c *NamingServerClient) handleMetadata(metaResp *MetaResponse, vGroup string) error { + if metaResp.Term > 0 { + atomic.StoreInt64(&c.term, metaResp.Term) + } + + var newNodes []NamingServerNode + for _, cluster := range metaResp.ClusterList { + for _, unit := range cluster.UnitData { + for _, node := range unit.NamingInstanceList { + if (node.Role == ClusterRoleLeader && node.Term >= atomic.LoadInt64(&c.term)) || + node.Role == ClusterRoleMember { + newNodes = append(newNodes, node) + } + } + } + } + + c.vgroupAddressMap.Store(vGroup, newNodes) + return nil +} + +type RefreshListener struct { + client *NamingServerClient +} + +func (l *RefreshListener) OnEvent(vGroup string) error { + return l.client.RefreshGroup(vGroup) +} + +func (c *NamingServerClient) Subscribe(vGroup string, listener NamingListener) error { + c.mu.Lock() + defer c.mu.Unlock() + + val, _ := c.listenerServiceMap.LoadOrStore(vGroup, []NamingListener{}) + listeners := append(val.([]NamingListener), listener) + c.listenerServiceMap.Store(vGroup, listeners) + + if !c.isSubscribed { + c.isSubscribed = true + c.wg.Add(1) + go c.watchLoop(vGroup) + } + return nil +} + +func (c *NamingServerClient) watchLoop(vGroup string) { + defer c.wg.Done() + var retryCount int + + for { + select { + case <-c.closeChan: + return + default: + } + + changed, err := c.Watch(vGroup) + if err != nil { + retryCount++ + if retryCount > failureRecoveryThreshold { + c.logger.Error("watch failed continuously, will retry", + zap.Error(err), zap.Int("retryCount", retryCount)) + select { + case <-time.After(time.Duration(retryDelayMs) * time.Millisecond): + case <-c.closeChan: + return + } + } else { + c.logger.Warn("watch error, will retry immediately", zap.Error(err)) + } + continue + } + + retryCount = 0 + + if changed { + if err := c.RefreshGroup(vGroup); err != nil { + c.logger.Error("refresh group failed in watch", zap.Error(err)) + continue + } + val, ok := c.listenerServiceMap.Load(vGroup) + if !ok { + continue + } + for _, listener := range val.([]NamingListener) { + if err := listener.OnEvent(vGroup); err != nil { + c.logger.Warn("listener callback failed", zap.Error(err)) + } + } Review Comment: watchLoop() refreshes the vGroup via RefreshGroup() and then invokes each listener's OnEvent(). The built-in RefreshListener.OnEvent() also calls RefreshGroup(), so a single change event triggers duplicate discovery calls. Consider making watchLoop either (a) only notify listeners and let listeners decide whether to refresh, or (b) keep the refresh in watchLoop and ensure listeners are pure notifications (and adjust RefreshListener accordingly). ```suggestion val, ok := c.listenerServiceMap.Load(vGroup) if ok { for _, listener := range val.([]NamingListener) { if err := listener.OnEvent(vGroup); err != nil { c.logger.Warn("listener callback failed", zap.Error(err)) } } continue } if err := c.RefreshGroup(vGroup); err != nil { c.logger.Error("refresh group failed in watch", zap.Error(err)) continue } ``` -- 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]
