Merge 77ed42b into 3b038b4

clients/config_client/config_client.go

@@ -212,6 +212,10 @@ func (client *ConfigClient) getConfigInner(param vo.ConfigParam) (content string
 		clientConfig.TimeoutMs, false, client)
 	if err != nil {
 		logger.Infof("get config from server error:%+v ", err)
+		if clientConfig, err := client.GetClientConfig(); err == nil && clientConfig.DisableUseSnapShot {
+			logger.Errorf("get config from cache  error:%+v ", err)
+			return "", errors.New("get config from remote nacos server fail, and is not allowed to read local file")
+		}
 		content, err = cache.ReadConfigFromFile(cacheKey, client.configCacheDir)
 		if err != nil {
 			logger.Errorf("get config from cache  error:%+v ", err)
@@ -423,7 +427,12 @@ func (client *ConfigClient) executeConfigListen() {
 				logger.Warnf("ConfigBatchListenRequest failure,err:%+v", err)
 				continue
 			}
-			if iResponse == nil && !iResponse.IsSuccess() {
+			if iResponse == nil {
+				logger.Warnf("ConfigBatchListenRequest failure, response is nil")
+				continue
+			}
+			if !iResponse.IsSuccess() {
+				logger.Warnf("ConfigBatchListenRequest failure, error code:%+v", iResponse.GetErrorCode())
 				continue
 			}
 			changeKeys := make(map[string]struct{})

clients/config_client/config_proxy.go

@@ -152,11 +152,20 @@ func (cp *ConfigProxy) queryConfig(dataId, group, tenant string, timeout uint64,
 	return response, nil
 }
 
+func appName(client *ConfigClient) string {
+	if clientConfig, err := client.GetClientConfig(); err == nil {
+		appName := clientConfig.AppName
+		return appName
+	}
+	return "unknown"
+}
+
 func (cp *ConfigProxy) createRpcClient(ctx context.Context, taskId string, client *ConfigClient) *rpc.RpcClient {
 	labels := map[string]string{
-		constant.LABEL_SOURCE: constant.LABEL_SOURCE_SDK,
-		constant.LABEL_MODULE: constant.LABEL_MODULE_CONFIG,
-		"taskId":              taskId,
+		constant.LABEL_SOURCE:   constant.LABEL_SOURCE_SDK,
+		constant.LABEL_MODULE:   constant.LABEL_MODULE_CONFIG,
+		constant.APPNAME_HEADER: appName(client),
+		"taskId":                taskId,
 	}
 
 	iRpcClient, _ := rpc.CreateClient(ctx, "config-"+taskId+"-"+client.uid, rpc.GRPC, labels, cp.nacosServer)

common/constant/client_config_options.go

@@ -116,6 +116,13 @@ func WithCacheDir(cacheDir string) ClientOption {
 	}
 }
 
+// WithDisableUseSnapShot ...
+func WithDisableUseSnapShot(disableUseSnapShot bool) ClientOption {
+	return func(config *ClientConfig) {
+		config.DisableUseSnapShot = disableUseSnapShot
+	}
+}
+
 // WithUpdateThreadNum ...
 func WithUpdateThreadNum(updateThreadNum int) ClientOption {
 	return func(config *ClientConfig) {

common/constant/config.go

@@ -39,6 +39,7 @@ type ClientConfig struct {
 	SecretKey            string                   // the SecretKey for kms
 	OpenKMS              bool                     // it's to open kms,default is false. https://help.aliyun.com/product/28933.html
 	CacheDir             string                   // the directory for persist nacos service info,default value is current path
+	DisableUseSnapShot   bool                     // It's a switch, default is false, means that when get remote config fail, use local cache file instead
 	UpdateThreadNum      int                      // the number of goroutine for update nacos service info,default value is 20
 	NotLoadCacheAtStart  bool                     // not to load persistent nacos service info in CacheDir at start time
 	UpdateCacheWhenEmpty bool                     // update cache when get empty service instance from server

common/constant/const.go

@@ -89,6 +89,7 @@ const (
 	DEFAULT_TIMEOUT_MILLS       = 3000
 	ALL_SYNC_INTERNAL           = 5 * time.Minute
 	CLIENT_APPNAME_HEADER       = "Client-AppName"
+	APPNAME_HEADER              = "AppName"
 	CLIENT_REQUEST_TS_HEADER    = "Client-RequestTS"
 	CLIENT_REQUEST_TOKEN_HEADER = "Client-RequestToken"
 	EX_CONFIG_INFO              = "exConfigInfo"

common/nacos_server/nacos_server.go

@@ -87,7 +87,7 @@ func NewNacosServer(ctx context.Context, serverList []constant.ServerConfig, cli
 	_, err := securityLogin.Login()
 
 	if err != nil {
-		return &ns, err
+		logger.Errorf("login in err:%v", err)
 	}
 
 	securityLogin.AutoRefresh(ctx)

common/security/security_proxy.go

@@ -75,16 +75,24 @@ func (ac *AuthClient) AutoRefresh(ctx context.Context) {
 	}
 
 	go func() {
-		timer := time.NewTimer(time.Second * time.Duration(ac.tokenTtl-ac.tokenRefreshWindow))
+		var timer *time.Timer
+		if lastLoginSuccess := ac.lastRefreshTime > 0 && ac.tokenTtl > 0 && ac.tokenRefreshWindow > 0; lastLoginSuccess {
+			timer = time.NewTimer(time.Second * time.Duration(ac.tokenTtl-ac.tokenRefreshWindow))
+		} else {
+			timer = time.NewTimer(time.Second * time.Duration(5))
+		}
 		defer timer.Stop()
 		for {
 			select {
 			case <-timer.C:
 				_, err := ac.Login()
 				if err != nil {
 					logger.Errorf("login has error %+v", err)
+					timer.Reset(time.Second * time.Duration(5))
+				} else {
+					logger.Infof("login success, tokenTtl: %+v seconds, tokenRefreshWindow: %+v seconds", ac.tokenTtl, ac.tokenRefreshWindow)
+					timer.Reset(time.Second * time.Duration(ac.tokenTtl-ac.tokenRefreshWindow))
 				}
-				timer.Reset(time.Second * time.Duration(ac.tokenTtl-ac.tokenRefreshWindow))
 			case <-ctx.Done():
 				return
 			}
@@ -156,6 +164,7 @@ func (ac *AuthClient) login(server constant.ServerConfig) (bool, error) {
 
 		if val, ok := result[constant.KEY_ACCESS_TOKEN]; ok {
 			ac.accessToken.Store(val)
+			ac.lastRefreshTime = time.Now().Unix()
 			ac.tokenTtl = int64(result[constant.KEY_TOKEN_TTL].(float64))
 			ac.tokenRefreshWindow = ac.tokenTtl / 10
 		}