refactor:配置文件缓存逻辑重构 (polarismesh#1202)

Author: chuntaojun

.github/workflows/codecov.yaml

@@ -75,10 +75,22 @@ jobs:
       - name: Vert check
         run: bash vert.sh -install && bash vert.sh
 
-      - name: Go Test
+      - name: Standalone Test
         env:
           MYSQL_DB_USER: root
           MYSQL_DB_PWD: root
+          RUN_MODE: STANDALONE
+        run: |
+          # bash coverage.sh
+          export MYSQL_DB_USER=${{ env.MYSQL_DB_USER }}
+          export MYSQL_DB_PWD=${{ env.MYSQL_DB_PWD }}
+          bash test/codecov.sh
+
+      - name: Cluster Test
+        env:
+          MYSQL_DB_USER: root
+          MYSQL_DB_PWD: root
+          RUN_MODE: CLUSTER
         run: |
           # bash coverage.sh
           export MYSQL_DB_USER=${{ env.MYSQL_DB_USER }}

.golangci.yml

@@ -189,7 +189,7 @@ linters-settings:
     # Max line length, lines longer will be reported.
     # '\t' is counted as 1 character by default, and can be changed with the tab-width option.
     # Default: 120.
-    line-length: 120
+    line-length: 140
     # Tab width in spaces.
     # Default: 1
     tab-width: 1

README-zh.md

@@ -14,11 +14,12 @@
 
 README：
 
-- [介绍](#介绍)
-- [如何安装](#如何安装)
-- [如何开发服务](#如何开发服务)
-- [如何集成服务网关](#如何集成服务网关)
-- [交流群](#交流群)
+- [北极星：服务发现和治理](#北极星服务发现和治理)
+  - [介绍](#介绍)
+  - [如何安装](#如何安装)
+  - [如何开发服务](#如何开发服务)
+  - [如何集成服务网关](#如何集成服务网关)
+  - [交流群](#交流群)
 
 更多文档请查看[北极星官网](https://polarismesh.cn)
 
@@ -93,4 +94,4 @@ README：
 
 扫码二维码，加入北极星开源交流群。欢迎用户反馈使用问题和优化建议。
 
-<img src="https://main.qcloudimg.com/raw/bff4285d70498058caa212805b83a620.jpg" width="20%" height="20%" />
+<img src="./qrcode.png" width="20%" height="20%" />

README.md

@@ -14,11 +14,12 @@ English | [简体中文](./README-zh.md)
 
 README：
 
-- [Introduction](#introduction)
-- [How to install](#how-to-install)
-- [How to develop service](#how-to-develop-service)
-- [How to integrate service gateway](#how-to-integrate-service-gateway)
-- [Chat group](#chat-group)
+- [Polaris: Service Discovery and Governance](#polaris-service-discovery-and-governance)
+  - [Introduction](#introduction)
+  - [How to install](#how-to-install)
+  - [How to develop service](#how-to-develop-service)
+  - [How to integrate service gateway](#how-to-integrate-service-gateway)
+  - [Chat group](#chat-group)
 
 Visit [Website](https://polarismesh.cn/) to learn more
 
@@ -93,4 +94,4 @@ You can integrate service gateways with Polaris service discovery and governance
 
 Please scan the QR code to join the chat group.
 
-<img src="https://main.qcloudimg.com/raw/bff4285d70498058caa212805b83a620.jpg" width="20%" height="20%" />
+<img src="./qrcode.png" width="20%" height="20%" />

apiserver/apiserver.go

@@ -20,9 +20,6 @@ package apiserver
 import (
 	"context"
 	"fmt"
-	"strings"
-
-	"github.com/polarismesh/polaris/common/log"
 )
 
 const (
@@ -75,33 +72,3 @@ func Register(name string, server Apiserver) error {
 
 	return nil
 }
-
-// GetClientOpenMethod 获取客户端openMethod
-func GetClientOpenMethod(include []string, protocol string) (map[string]bool, error) {
-	clientAccess := make(map[string][]string)
-	clientAccess[DiscoverAccess] = []string{"Discover", "ReportClient"}
-	clientAccess[RegisterAccess] = []string{"RegisterInstance", "DeregisterInstance"}
-	clientAccess[HealthcheckAccess] = []string{"Heartbeat", "BatchHeartbeat", "BatchGetHeartbeat", "BatchDelHeartbeat"}
-
-	openMethod := make(map[string]bool)
-	// 如果为空，开启全部接口
-	if len(include) == 0 {
-		for key := range clientAccess {
-			include = append(include, key)
-		}
-	}
-
-	for _, item := range include {
-		if methods, ok := clientAccess[item]; ok {
-			for _, method := range methods {
-				method = "/v1.Polaris" + strings.ToUpper(protocol) + "/" + method
-				openMethod[method] = true
-			}
-		} else {
-			log.Errorf("method %s does not exist in %sserver client access", item, protocol)
-			return nil, fmt.Errorf("method %s does not exist in %sserver client access", item, protocol)
-		}
-	}
-
-	return openMethod, nil
-}

apiserver/apiserver_test.go

@@ -30,6 +30,7 @@ import (
 
 	"github.com/polarismesh/polaris/auth"
 	"github.com/polarismesh/polaris/cache"
+	cachetypes "github.com/polarismesh/polaris/cache/api"
 	commonlog "github.com/polarismesh/polaris/common/log"
 	"github.com/polarismesh/polaris/common/utils"
 	"github.com/polarismesh/polaris/namespace"
@@ -76,6 +77,7 @@ type EurekaTestSuit struct {
 	updateCacheInterval time.Duration
 	cancel              context.CancelFunc
 	storage             store.Store
+	cacheMgr            *cache.CacheManager
 }
 
 type options func(cfg *TestConfig)
@@ -117,6 +119,7 @@ func (d *EurekaTestSuit) initialize(t *testing.T, callback func(t *testing.T, s
 	if err != nil {
 		return err
 	}
+	d.cacheMgr = cacheMgn
 
 	// 批量控制器
 	namingBatchConfig, err := batch.ParseBatchConfig(d.cfg.Naming.Batch)
@@ -158,8 +161,8 @@ func (d *EurekaTestSuit) initialize(t *testing.T, callback func(t *testing.T, s
 	healthCheckServer.SetInstanceCache(cacheMgn.Instance())
 
 	// 为 instance 的 cache 添加 健康检查的 Listener
-	cacheMgn.AddListener(cache.CacheNameInstance, []cache.Listener{cacheProvider})
-	cacheMgn.AddListener(cache.CacheNameClient, []cache.Listener{cacheProvider})
+	cacheMgn.AddListener(cachetypes.CacheInstance, []cachetypes.Listener{cacheProvider})
+	cacheMgn.AddListener(cachetypes.CacheClient, []cachetypes.Listener{cacheProvider})
 
 	d.healthSvr = healthCheckServer
 	time.Sleep(5 * time.Second)
@@ -201,11 +204,6 @@ func replaceEnv(configContent string) string {
 
 func (d *EurekaTestSuit) Destroy() {
 	d.cancel()
-	time.Sleep(5 * time.Second)
-
+	d.cacheMgr.Close()
 	d.storage.Destroy()
-	time.Sleep(5 * time.Second)
-
-	healthcheck.TestDestroy()
-	time.Sleep(5 * time.Second)
 }

apiserver/eurekaserver/eureka_suit_test.go

@@ -286,7 +286,6 @@ type InstanceInfo struct {
 
 	LeaseInfo *LeaseInfo `json:"leaseInfo" xml:"leaseInfo,omitempty"`
 
-	//nolint:lll
 	IsCoordinatingDiscoveryServer interface{} `json:"isCoordinatingDiscoveryServer" xml:"isCoordinatingDiscoveryServer,omitempty"`
 
 	Metadata *Metadata `json:"metadata" xml:"metadata"`

apiserver/eurekaserver/model.go

@@ -151,6 +151,7 @@ type EurekaServer struct {
 
 	replicatePeers       map[string][]string
 	generateUniqueInstId bool
+	subCtxs              []*eventhub.SubscribtionContext
 }
 
 // GetPort 获取端口
@@ -178,6 +179,7 @@ func (h *EurekaServer) Initialize(ctx context.Context, option map[string]interfa
 	}
 	h.option = option
 	h.openAPI = api
+	h.subCtxs = make([]*eventhub.SubscribtionContext, 0, 4)
 
 	var namespace = DefaultNamespace
 	if namespaceValue, ok := option[optionNamespace]; ok {
@@ -341,11 +343,12 @@ func (h *EurekaServer) Run(errCh chan error) {
 	if len(h.replicatePeers) > 0 {
 		h.eventHandlerHandler = &EurekaInstanceEventHandler{
 			BaseInstanceEventHandler: service.NewBaseInstanceEventHandler(h.namingServer), svr: h}
-		if err = eventhub.Subscribe(
-			eventhub.InstanceEventTopic, "eureka-replication", h.eventHandlerHandler); nil != err {
+		subCtx, err := eventhub.Subscribe(eventhub.InstanceEventTopic, h.eventHandlerHandler)
+		if err != nil {
 			errCh <- err
 			return
 		}
+		h.subCtxs = append(h.subCtxs, subCtx)
 	}
 	h.workers = NewApplicationsWorkers(h.refreshInterval, h.deltaExpireInterval, h.enableSelfPreservation,
 		h.namingServer, h.healthCheckServer, h.namespace)

apiserver/eurekaserver/server.go

@@ -24,6 +24,8 @@ import (
 
 	"google.golang.org/grpc/metadata"
 
+	"github.com/polarismesh/polaris/apiserver"
+	grpchelp "github.com/polarismesh/polaris/apiserver/grpcserver/utils"
 	"github.com/polarismesh/polaris/common/utils"
 )
 
@@ -120,3 +122,57 @@ func TestConvertContext(t *testing.T) {
 		})
 	}
 }
+
+func TestGetClientOpenMethod(t *testing.T) {
+	type args struct {
+		include  []string
+		protocol string
+	}
+	tests := []struct {
+		name    string
+		args    args
+		want    map[string]bool
+		wantErr bool
+	}{
+		{
+			name: "case=1",
+			args: args{
+				include: []string{
+					apiserver.RegisterAccess,
+				},
+				protocol: "grpc",
+			},
+			want: map[string]bool{
+				"/v1.PolarisGRPC/RegisterInstance":   true,
+				"/v1.PolarisGRPC/DeregisterInstance": true,
+			},
+			wantErr: false,
+		},
+		{
+			name: "case=1",
+			args: args{
+				include: []string{
+					apiserver.DiscoverAccess,
+				},
+				protocol: "grpc",
+			},
+			want: map[string]bool{
+				"/v1.PolarisGRPC/Discover":     true,
+				"/v1.PolarisGRPC/ReportClient": true,
+			},
+			wantErr: false,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := grpchelp.GetClientOpenMethod(tt.args.include, tt.args.protocol)
+			if (err != nil) != tt.wantErr {
+				t.Errorf("GetClientOpenMethod() error = %v, wantErr %v", err, tt.wantErr)
+				return
+			}
+			if !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("GetClientOpenMethod() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}

apiserver/grpcserver/base_test.go

@@ -46,12 +46,13 @@ type Cache interface {
 
 // CacheObject
 type CacheObject struct {
+	// OriginVal
 	OriginVal proto.Message
-
+	// preparedVal
 	preparedVal *grpc.PreparedMsg
-
+	// CacheType
 	CacheType string
-
+	// Key
 	Key string
 }
 

apiserver/grpcserver/cache.go

@@ -72,7 +72,7 @@ func (g *ConfigGRPCServer) Run(errCh chan error) {
 			case "client":
 				if apiConfig.Enable {
 					apiconfig.RegisterPolarisConfigGRPCServer(server, g)
-					openMethod, getErr := getConfigClientOpenMethod(g.GetProtocol())
+					openMethod, getErr := GetClientOpenMethod(g.GetProtocol())
 					if getErr != nil {
 						return getErr
 					}
@@ -126,9 +126,10 @@ func (g *ConfigGRPCServer) allowAccess(method string) bool {
 	return g.BaseGrpcServer.AllowAccess(method)
 }
 
-func getConfigClientOpenMethod(protocol string) (map[string]bool, error) {
+// GetClientOpenMethod .
+func GetClientOpenMethod(protocol string) (map[string]bool, error) {
 	openMethods := []string{"GetConfigFile", "CreateConfigFile",
-		"UpdateConfigFile", "PublishConfigFile", "WatchConfigFiles"}
+		"UpdateConfigFile", "PublishConfigFile", "WatchConfigFiles", "GetConfigFileMetadataList"}
 
 	openMethod := make(map[string]bool)