Merge remote-tracking branch 'upstream/develop' into agent-skills

Author: Oxidaner

cluster/cluster/adaptivesvc/cluster_invoker.go

@@ -86,21 +86,21 @@ func (ivk *adaptiveServiceClusterInvoker) Invoke(ctx context.Context, invocation
 		}
 	}
 	if remainingStr == "" {
-		logger.Errorf("[adasvc cluster] The %s field type of value %v should be string.",
-			constant.AdaptiveServiceRemainingKey, remainingIface)
+		logger.Errorf("[Cluster][AdaptiveSvc] %s field type invalid, key=%s value=%v, expected string",
+			constant.AdaptiveServiceRemainingKey, constant.AdaptiveServiceRemainingKey, remainingIface)
 		return res
 	}
 	remaining, err := strconv.Atoi(remainingStr)
 	if err != nil {
-		logger.Warnf("the remaining is unexpected, we need a int type, but we got %s, err: %v.", remainingStr, err)
+		logger.Warnf("[Cluster][AdaptiveSvc] parse remaining failed, remaining=%s err=%v, expected int", remainingStr, err)
 		return res
 	}
-	logger.Debugf("[adasvc cluster] The server status was received successfully, %s: %#v",
+	logger.Debugf("[Cluster][AdaptiveSvc] received server status, %s=%s",
 		constant.AdaptiveServiceRemainingKey, remainingStr)
 	err = metrics.LocalMetrics.SetMethodMetrics(invoker.GetURL(),
 		invocation.MethodName(), metrics.HillClimbing, uint64(remaining))
 	if err != nil {
-		logger.Warnf("adaptive service metrics update is failed, err: %v", err)
+		logger.Warnf("[Cluster][AdaptiveSvc] update metrics failed, err=%v", err)
 		return &result.RPCResult{Err: err}
 	}
 

cluster/cluster/base/cluster_invoker.go

@@ -127,7 +127,7 @@ func (invoker *BaseClusterInvoker) doSelectInvoker(lb loadbalance.LoadBalance, i
 			return invokers[0]
 		}
 		base.SetInvokerUnhealthyStatus(invokers[0])
-		logger.Errorf("the invokers of %s is nil. ", invokers[0].GetURL().ServiceKey())
+		logger.Errorf("[Cluster] invoker unavailable, serviceKey=%s, reason=nil", invokers[0].GetURL().ServiceKey())
 		return nil
 	}
 
@@ -149,8 +149,7 @@ func (invoker *BaseClusterInvoker) doSelectInvoker(lb loadbalance.LoadBalance, i
 				continue
 			}
 			if !reselectedInvoker.IsAvailable() {
-				logger.Infof("the invoker of %s is not available, maybe some network error happened or the server is shutdown.",
-					invoker.GetURL().Ip)
+				logger.Infof("[Cluster] invoker unavailable, ip=%s", invoker.GetURL().Ip)
 				base.SetInvokerUnhealthyStatus(reselectedInvoker)
 				otherInvokers = getOtherInvokers(otherInvokers, reselectedInvoker)
 				continue

cluster/cluster/broadcast/cluster_invoker.go

@@ -58,7 +58,7 @@ func (invoker *broadcastClusterInvoker) Invoke(ctx context.Context, invocation p
 	for _, ivk := range invokers {
 		res = ivk.Invoke(ctx, invocation)
 		if res.Error() != nil {
-			logger.Warnf("broadcast invoker invoke err: %v when use invoker: %v\n", res.Error(), ivk)
+			logger.Warnf("[Cluster][Broadcast] invoke failed, err=%v invoker=%v", res.Error(), ivk)
 			err = res.Error()
 		}
 	}

cluster/cluster/failback/cluster_invoker.go

@@ -65,7 +65,7 @@ func newFailbackClusterInvoker(directory directory.Directory) protocolbase.Invok
 	retriesConfig := invoker.GetURL().GetParam(constant.RetriesKey, constant.DefaultFailbackTimes)
 	retries, err := strconv.Atoi(retriesConfig)
 	if err != nil || retries < 0 {
-		logger.Error("Your retries config is invalid,pls do a check. And will use the default fail back times configuration instead.")
+		logger.Error("[Cluster][Failback] retries config invalid, using default")
 		retries = constant.DefaultFailbackTimesInt
 	}
 
@@ -112,7 +112,7 @@ func (invoker *failbackClusterInvoker) process(ctx context.Context) {
 
 			// ignore return. the get must success.
 			if _, err = invoker.taskList.Get(1); err != nil {
-				logger.Warnf("get task found err: %v\n", err)
+				logger.Warnf("[Cluster][Failback] get task failed, err=%v", err)
 				break
 			}
 			go invoker.tryTimerTaskProc(ctx, retryTask)
@@ -124,7 +124,7 @@ func (invoker *failbackClusterInvoker) process(ctx context.Context) {
 func (invoker *failbackClusterInvoker) Invoke(ctx context.Context, invocation protocolbase.Invocation) result.Result {
 	invokers := invoker.Directory.List(invocation)
 	if err := invoker.CheckInvokers(invokers, invocation); err != nil {
-		logger.Errorf("Failed to invoke the method %v in the service %v, wait for retry in background. Ignored exception: %v.\n",
+		logger.Errorf("[Cluster][Failback] check invokers failed, method=%s service=%s err=%v",
 			invocation.MethodName(), invoker.GetURL().Service(), err)
 		return &result.RPCResult{}
 	}
@@ -151,14 +151,14 @@ func (invoker *failbackClusterInvoker) Invoke(ctx context.Context, invocation pr
 
 		taskLen := invoker.taskList.Len()
 		if taskLen >= invoker.failbackTasks {
-			logger.Warnf("tasklist is too full > %d.\n", taskLen)
+			logger.Warnf("[Cluster][Failback] task list full, len=%d", taskLen)
 			return &result.RPCResult{}
 		}
 
 		timerTask := newRetryTimerTask(loadBalance, invocation, invokers, ivk, invoker)
 		invoker.taskList.Put(timerTask)
 
-		logger.Errorf("Failback to invoke the method %v in the service %v, wait for retry in background. Ignored exception: %v.\n",
+		logger.Errorf("[Cluster][Failback] invoke failed, method=%s service=%s err=%v",
 			methodName, url.Service(), res.Error().Error())
 		// ignore
 		return &result.RPCResult{}
@@ -192,21 +192,21 @@ type retryTimerTask struct {
 }
 
 func (t *retryTimerTask) checkRetry() {
-	logger.Errorf("Failed retry to invoke the method %v in the service %v, wait again. The exception: %v",
+	logger.Errorf("[Cluster][Failback] retry failed, method=%s service=%s err=%v",
 		t.invocation.MethodName(), t.clusterInvoker.GetURL().Service(), t.lastErr)
 	t.retries++
 	t.nextBackoff = t.backoff.NextBackOff() // calculate next exponential backoff wait time
 
 	if t.retries > t.maxRetries || t.nextBackoff == backoff.Stop {
-		logger.Errorf("Retry times exceed threshold (%v), invocation-> %v",
+		logger.Errorf("[Cluster][Failback] retry exceeded, retries=%d invocation=%v",
 			t.retries, t.invocation)
 		return
 	}
 
-	logger.Infof("Failback retry scheduled after %v for method %v", t.nextBackoff, t.invocation.MethodName())
+	logger.Infof("[Cluster][Failback] retry scheduled, backoff=%v method=%s", t.nextBackoff, t.invocation.MethodName())
 
 	if err := t.clusterInvoker.taskList.Put(t); err != nil {
-		logger.Errorf("invoker.taskList.Put(retryTask:%#v) = error:%v", t, err)
+		logger.Errorf("[Cluster][Failback] put task failed, task=%v err=%v", t, err)
 		return
 	}
 	t.lastT = time.Now() // update lastT after successful Put

cluster/cluster/failover/cluster_invoker.go

@@ -19,7 +19,6 @@ package failover
 
 import (
 	"context"
-	"fmt"
 	"strconv"
 )
 
@@ -97,17 +96,15 @@ func (invoker *failoverClusterInvoker) Invoke(ctx context.Context, invocation pr
 	invokerSvc := invoker.GetURL().Service()
 	invokerUrl := invoker.Directory.GetURL()
 	if ivk == nil {
-		logger.Errorf("Failed to invoke the method %s of the service %s .No provider is available.", methodName, invokerSvc)
+		logger.Errorf("[Cluster][Failover] no provider available, method=%s service=%s", methodName, invokerSvc)
 		return &result.RPCResult{
 			Err: perrors.Errorf("Failed to invoke the method %s of the service %s .No provider is available because can't connect server.",
 				methodName, invokerSvc),
 		}
 	}
 
-	logger.Errorf(fmt.Sprintf("Failed to invoke the method %v in the service %v. "+
-		"Tried %v times of the providers %v (%v/%v) from the registry %v on the consumer %v using the dubbo version %v. "+
-		"Last error is %+v.", methodName, invokerSvc, retries, providers, len(providers), len(invokers),
-		invokerUrl, ip, constant.Version, res.Error().Error()))
+	logger.Errorf("[Cluster][Failover] invoke failed after retries, method=%s service=%s retries=%d providers=%d/%d registry=%s consumer=%s version=%s lastErr=%v",
+		methodName, invokerSvc, retries, len(providers), len(invokers), invokerUrl, ip, constant.Version, res.Error().Error())
 
 	return res
 }

cluster/cluster/failsafe/cluster_invoker.go

@@ -77,7 +77,7 @@ func (invoker *failsafeClusterInvoker) Invoke(ctx context.Context, invocation pr
 	res = ivk.Invoke(ctx, invocation)
 	if res.Error() != nil {
 		// ignore
-		logger.Errorf("Failsafe ignore exception: %v.\n", res.Error().Error())
+		logger.Errorf("[Cluster][Failsafe] ignore exception, err=%v", res.Error().Error())
 		return &result.RPCResult{}
 	}
 	return res

cluster/cluster/forking/cluster_invoker.go

@@ -76,7 +76,7 @@ func (invoker *forkingClusterInvoker) Invoke(ctx context.Context, invocation pro
 		go func(k protocolbase.Invoker) {
 			result := k.Invoke(ctx, invocation)
 			if err := resultQ.Put(result); err != nil {
-				logger.Errorf("resultQ put failed with exception: %v.\n", err)
+				logger.Errorf("[Cluster][Forking] resultQ put failed with exception err=%v", err)
 			}
 		}(ivk)
 	}

cluster/cluster/mock.go

@@ -102,7 +102,7 @@ func (bi *MockInvoker) Invoke(c context.Context, invocation base.Invocation) res
 }
 
 func (bi *MockInvoker) Destroy() {
-	logger.Infof("Destroy invoker: %v", bi.GetURL().String())
+	logger.Infof("[Mock] Destroy invoker url=%s", bi.GetURL().String())
 	bi.destroyed = true
 	bi.available = false
 }

cluster/directory/static/directory.go

@@ -49,7 +49,7 @@ func NewDirectory(invokers []protocolbase.Invoker) *directory {
 
 	err := dir.BuildRouterChain(invokers, url)
 	if err != nil {
-		logger.Error(err)
+		logger.Errorf("[Cluster][Directory] build router chain failed, err=%v", err)
 		dir.RouterChain().SetInvokers(invokers)
 	}
 

cluster/loadbalance/consistenthashing/loadbalance.go

@@ -21,6 +21,7 @@ import (
 	"encoding/json"
 	"hash/crc32"
 	"regexp"
+	"sync"
 )
 
 import (
@@ -38,8 +39,9 @@ const (
 )
 
 var (
-	selectors = make(map[string]*selector)
-	re        = regexp.MustCompile(constant.CommaSplitPattern)
+	selectors   = make(map[string]*selector)
+	selectorsMu sync.RWMutex
+	re          = regexp.MustCompile(constant.CommaSplitPattern)
 )
 
 func init() {
@@ -71,10 +73,19 @@ func (lb *conshashLoadBalance) Select(invokers []base.Invoker, invocation base.I
 		bs = append(bs, b...)
 	}
 	hashCode := crc32.ChecksumIEEE(bs)
+
+	selectorsMu.RLock()
 	selector, ok := selectors[key]
+	selectorsMu.RUnlock()
+
 	if !ok || selector.hashCode != hashCode {
-		selectors[key] = newSelector(invokers, methodName, hashCode)
+		selectorsMu.Lock()
+		selector, ok = selectors[key]
+		if !ok || selector.hashCode != hashCode {
+			selectors[key] = newSelector(invokers, methodName, hashCode)
+		}
 		selector = selectors[key]
+		selectorsMu.Unlock()
 	}
 	return selector.Select(invocation)
 }