fix:修复资源清理逻辑以及配置中心的bug

admin/api.go

@@ -54,8 +54,10 @@ type AdminOperateServer interface {
 	ReleaseLeaderElection(ctx context.Context, electKey string) error
 	// GetCMDBInfo get cmdb info
 	GetCMDBInfo(ctx context.Context) ([]model.LocationView, error)
-	// InitMainUser
-	InitMainUser(ctx context.Context, user apisecurity.User) error
+	// HasMainUser .
+	HasMainUser(ctx context.Context) *apiservice.Response
+	// InitMainUser .
+	InitMainUser(ctx context.Context, user *apisecurity.User) *apiservice.Response
 	// GetServerFunctions Get server functions
 	GetServerFunctions(ctx context.Context) []authcommon.ServerFunctionGroup
 }

admin/default.go

@@ -23,6 +23,7 @@ import (
 	"fmt"
 
 	"github.com/polarismesh/polaris/admin/job"
+	"github.com/polarismesh/polaris/auth"
 	"github.com/polarismesh/polaris/cache"
 	"github.com/polarismesh/polaris/service"
 	"github.com/polarismesh/polaris/service/healthcheck"
@@ -48,13 +49,15 @@ func RegisterServerProxy(name string, factor ServerProxyFactory) error {
 
 // Initialize 初始化
 func Initialize(ctx context.Context, cfg *Config, namingService service.DiscoverServer,
-	healthCheckServer *healthcheck.Server, cacheMgn *cache.CacheManager, storage store.Store) error {
+	healthCheckServer *healthcheck.Server, cacheMgn *cache.CacheManager, storage store.Store,
+	userSvr auth.UserServer, policySvr auth.StrategyServer) error {
 
 	if finishInit {
 		return nil
 	}
 
-	proxySvr, actualSvr, err := InitServer(ctx, cfg, namingService, healthCheckServer, cacheMgn, storage)
+	proxySvr, actualSvr, err := InitServer(ctx, cfg, namingService, healthCheckServer, cacheMgn,
+		storage, userSvr, policySvr)
 	if err != nil {
 		return err
 	}
@@ -66,13 +69,16 @@ func Initialize(ctx context.Context, cfg *Config, namingService service.Discover
 }
 
 func InitServer(ctx context.Context, cfg *Config, namingService service.DiscoverServer,
-	healthCheckServer *healthcheck.Server, cacheMgn *cache.CacheManager, storage store.Store) (AdminOperateServer, *Server, error) {
+	healthCheckServer *healthcheck.Server, cacheMgn *cache.CacheManager, storage store.Store,
+	userSvr auth.UserServer, policySvr auth.StrategyServer) (AdminOperateServer, *Server, error) {
 
 	actualSvr := new(Server)
 
+	actualSvr.userSvr = userSvr
+	actualSvr.policySvr = policySvr
 	actualSvr.namingServer = namingService
 	actualSvr.healthCheckServer = healthCheckServer
-	actualSvr.cacheMgn = cacheMgn
+	actualSvr.cacheMgr = cacheMgn
 	actualSvr.storage = storage
 
 	maintainJobs := job.NewMaintainJobs(namingService, cacheMgn, storage)

admin/interceptor/auth/server.go

@@ -62,12 +62,12 @@ func (svr *Server) collectMaintainAuthContext(ctx context.Context, resourceOp au
 	)
 }
 
-func (s *Server) HasMainUser(ctx context.Context) (bool, error) {
-	return false, nil
+func (s *Server) HasMainUser(ctx context.Context) *apiservice.Response {
+	return s.nextSvr.HasMainUser(ctx)
 }
 
-func (s *Server) InitMainUser(ctx context.Context, user apisecurity.User) error {
-	return nil
+func (s *Server) InitMainUser(ctx context.Context, user *apisecurity.User) *apiservice.Response {
+	return s.nextSvr.InitMainUser(ctx, user)
 }
 
 func (svr *Server) GetServerConnections(ctx context.Context, req *admincommon.ConnReq) (*admincommon.ConnCountResp, error) {

admin/job/clean_deleted_resource.go

@@ -27,9 +27,10 @@ import (
 )
 
 var cleanFuncMapping = map[string]func(timeout time.Duration, job *cleanDeletedResourceJob){
-	"instance": cleanDeletedInstances,
-	"service":  cleanDeletedServices,
-	"clients":  cleanDeletedClients,
+	"instance":         cleanDeletedInstances,
+	"service":          cleanDeletedServices,
+	"clients":          cleanDeletedClients,
+	"service_contract": cleanDeletedServiceContracts,
 	"circuitbreaker_rule": func(timeout time.Duration, job *cleanDeletedResourceJob) {
 		cleanDeletedRules("circuitbreaker_rule", timeout, job)
 	},
@@ -48,6 +49,51 @@ var cleanFuncMapping = map[string]func(timeout time.Duration, job *cleanDeletedR
 	"config_file_release": cleanDeletedConfigFiles,
 }
 
+var defaultCleanDeletedResourceConfig = CleandeletedResourceConf{
+	Resources: []CleanDeletedResource{
+		{
+			Resource: "instance",
+			Enable:   true,
+		},
+		{
+			Resource: "service",
+			Enable:   true,
+		},
+		{
+			Resource: "service_contract",
+			Enable:   true,
+		},
+		{
+			Resource: "clients",
+			Enable:   true,
+		},
+		{
+			Resource: "circuitbreaker_rule",
+			Enable:   true,
+		},
+		{
+			Resource: "ratelimit_rule",
+			Enable:   true,
+		},
+		{
+			Resource: "router_rule",
+			Enable:   true,
+		},
+		{
+			Resource: "faultdetect_rule",
+			Enable:   true,
+		},
+		{
+			Resource: "lane_rule",
+			Enable:   true,
+		},
+		{
+			Resource: "config_file_release",
+			Enable:   true,
+		},
+	},
+}
+
 type CleanDeletedResource struct {
 	// Resource 记录需要清理的资源类型
 	Resource string `mapstructure:"resource"`
@@ -59,7 +105,7 @@ type CleanDeletedResource struct {
 
 type CleandeletedResourceConf struct {
 	// ResourceTimeout 记录资源的额外超时时间，用户可自定义
-	Resources []CleanDeletedResource `json:"resourceTimeout"`
+	Resources []CleanDeletedResource `json:"resources" mapstructure:"resources"`
 	// Timeout 记录清理资源的超时时间，默认20分钟
 	Timeout time.Duration `mapstructure:"timeout"`
 }
@@ -90,6 +136,16 @@ func (job *cleanDeletedResourceJob) init(raw map[string]interface{}) error {
 		cfg.Timeout = 2 * time.Minute
 	}
 	job.cfg = cfg
+
+	if len(cfg.Resources) == 0 {
+		job.cfg.Resources = defaultCleanDeletedResourceConfig.Resources
+		for i := range job.cfg.Resources {
+			if job.cfg.Resources[i].Timeout == nil {
+				job.cfg.Resources[i].Timeout = &job.cfg.Timeout
+			}
+		}
+	}
+
 	return nil
 }
 
@@ -182,6 +238,21 @@ func cleanDeletedInstances(timeout time.Duration, job *cleanDeletedResourceJob)
 	}
 }
 
+func cleanDeletedServiceContracts(timeout time.Duration, job *cleanDeletedResourceJob) {
+	batchSize := uint32(100)
+	for {
+		count, err := job.storage.BatchCleanDeletedClients(timeout, batchSize)
+		if err != nil {
+			log.Errorf("[Maintain][Job][CleanDeletedClients] batch clean deleted client, err: %v", err)
+			break
+		}
+		log.Infof("[Maintain][Job][CleanDeletedClients] clean deleted client count %d", count)
+		if count < batchSize {
+			break
+		}
+	}
+}
+
 func cleanDeletedRules(rule string, timeout time.Duration, job *cleanDeletedResourceJob) {
 	batchSize := uint32(100)
 	for {

admin/maintain.go

@@ -27,6 +27,7 @@ import (
 	apisecurity "github.com/polarismesh/specification/source/go/api/v1/security"
 	apiservice "github.com/polarismesh/specification/source/go/api/v1/service_manage"
 	"go.uber.org/zap"
+	"google.golang.org/protobuf/types/known/wrapperspb"
 
 	api "github.com/polarismesh/polaris/common/api/v1"
 	connlimit "github.com/polarismesh/polaris/common/conn/limit"
@@ -39,12 +40,31 @@ import (
 	"github.com/polarismesh/polaris/plugin"
 )
 
-func (s *Server) HasMainUser(ctx context.Context, user apisecurity.User) (bool, error) {
-	return false, nil
+// HasMainUser 判断是否存在主用户
+func (s *Server) HasMainUser(ctx context.Context) *apiservice.Response {
+	mainUser, err := s.storage.GetMainUser()
+	if err != nil {
+		log.Error("check hash main user", zap.Error(err), utils.RequestID(ctx))
+		return api.NewResponse(apimodel.Code_ExecuteException)
+	}
+	if mainUser == nil {
+		return api.NewResponse(apimodel.Code_NotFoundResource)
+	}
+	ret := mainUser.ToSpec()
+	ret.AuthToken = wrapperspb.String("")
+	return api.NewUserResponse(apimodel.Code_ExecuteSuccess, ret)
 }
 
-func (s *Server) InitMainUser(ctx context.Context, user apisecurity.User) error {
-	return nil
+// InitMainUser 初始化主用户
+func (s *Server) InitMainUser(_ context.Context, user *apisecurity.User) *apiservice.Response {
+	if user.GetSource().GetValue() == "" {
+		user.Source = utils.NewStringValue("Polaris")
+	}
+	ctx := context.WithValue(context.Background(), authcommon.ContextKeyInitMainUser{}, true)
+	rsp := s.userSvr.CreateUsers(ctx, []*apisecurity.User{
+		user,
+	})
+	return rsp.Responses[0]
 }
 
 func (s *Server) GetServerConnections(_ context.Context, req *admin.ConnReq) (*admin.ConnCountResp, error) {

admin/server.go

@@ -20,6 +20,7 @@ package admin
 import (
 	"sync"
 
+	"github.com/polarismesh/polaris/auth"
 	"github.com/polarismesh/polaris/cache"
 	"github.com/polarismesh/polaris/service"
 	"github.com/polarismesh/polaris/service/healthcheck"
@@ -32,8 +33,10 @@ type Server struct {
 	mu                sync.Mutex
 	namingServer      service.DiscoverServer
 	healthCheckServer *healthcheck.Server
-	cacheMgn          *cache.CacheManager
+	cacheMgr          *cache.CacheManager
 	storage           store.Store
+	userSvr           auth.UserServer
+	policySvr         auth.StrategyServer
 }
 
 func GetChainOrder() []string {

apiserver/grpcserver/config/client_access.go

@@ -91,7 +91,7 @@
 func (g *ConfigGRPCServer) UpsertAndPublishConfigFile(ctx context.Context,
 	req *apiconfig.ConfigFilePublishInfo) (*apiconfig.ConfigClientResponse, error) {
 	ctx = utils.ConvertGRPCContext(ctx)
-	response := g.configServer.CasUpsertAndReleaseConfigFileFromClient(ctx, req)
+	response := g.configServer.UpsertAndReleaseConfigFileFromClient(ctx, req)
 	return &apiconfig.ConfigClientResponse{
 		Code: response.Code,
 		Info: response.Info,
@@ -182,55 +182,67 @@
 			continue
 		}
 
-		var out *apiconfig.ConfigDiscoverResponse
-		var action string
-		startTime := commontime.CurrentMillisecond()
-		defer func() {
-			plugin.GetStatis().ReportDiscoverCall(metrics.ClientDiscoverMetric{
-				Action:    action,
-				ClientIP:  utils.ParseClientAddress(ctx),
-				Namespace: in.GetConfigFile().GetNamespace().GetValue(),
-				Resource:  metrics.ResourceOfConfigFile(in.GetConfigFile().GetGroup().GetValue(), in.GetConfigFile().GetFileName().GetValue()),
-				Timestamp: startTime,
-				CostTime:  commontime.CurrentMillisecond() - startTime,
-				Revision:  out.GetRevision(),
-				Success:   out.GetCode() > uint32(apimodel.Code_DataNoChange),
-			})
-		}()
-
-		switch in.Type {
-		case apiconfig.ConfigDiscoverRequest_CONFIG_FILE:
-			action = metrics.ActionGetConfigFile
-			ret := g.configServer.GetConfigFileWithCache(ctx, &apiconfig.ClientConfigFileInfo{})
-			out = api.NewConfigDiscoverResponse(apimodel.Code(ret.GetCode().GetValue()))
-			out.ConfigFile = ret.GetConfigFile()
-			out.Type = apiconfig.ConfigDiscoverResponse_CONFIG_FILE
-			out.Revision = strconv.Itoa(int(out.GetConfigFile().GetVersion().GetValue()))
-		case apiconfig.ConfigDiscoverRequest_CONFIG_FILE_Names:
-			action = metrics.ActionListConfigFiles
-			ret := g.configServer.GetConfigFileNamesWithCache(ctx, &apiconfig.ConfigFileGroupRequest{
-				Revision: wrapperspb.String(in.GetRevision()),
-				ConfigFileGroup: &apiconfig.ConfigFileGroup{
-					Namespace: in.GetConfigFile().GetNamespace(),
-					Name:      in.GetConfigFile().GetGroup(),
-				},
-			})
-			out = api.NewConfigDiscoverResponse(apimodel.Code(ret.GetCode().GetValue()))
-			out.ConfigFileNames = ret.GetConfigFileInfos()
-			out.Type = apiconfig.ConfigDiscoverResponse_CONFIG_FILE_Names
-			out.Revision = ret.GetRevision().GetValue()
-		case apiconfig.ConfigDiscoverRequest_CONFIG_FILE_GROUPS:
-			action = metrics.ActionListConfigGroups
-			req := in.GetConfigFile()
-			req.Md5 = wrapperspb.String(in.GetRevision())
-			out = g.configServer.GetConfigGroupsWithCache(ctx, req)
-			out.Type = apiconfig.ConfigDiscoverResponse_CONFIG_FILE_GROUPS
-		default:
-			out = api.NewConfigDiscoverResponse(apimodel.Code_InvalidDiscoverResource)
-		}
-
+		out := g.handleDiscoverRequest(ctx, in)
 		if err := svr.Send(out); err != nil {
 			return err
 		}
 	}
 }
+
+func (g *ConfigGRPCServer) handleDiscoverRequest(ctx context.Context, in *apiconfig.ConfigDiscoverRequest) *apiconfig.ConfigDiscoverResponse {
+	var out *apiconfig.ConfigDiscoverResponse
+	var action string
+	startTime := commontime.CurrentMillisecond()
+	defer func() {
+		plugin.GetStatis().ReportDiscoverCall(metrics.ClientDiscoverMetric{
+			Action:    action,
+			ClientIP:  utils.ParseClientAddress(ctx),
+			Namespace: in.GetConfigFile().GetNamespace().GetValue(),
+			Resource:  metrics.ResourceOfConfigFile(in.GetConfigFile().GetGroup().GetValue(), in.GetConfigFile().GetFileName().GetValue()),
+			Timestamp: startTime,
+			CostTime:  commontime.CurrentMillisecond() - startTime,
+			Revision:  out.GetRevision(),
+			Success:   out.GetCode() > uint32(apimodel.Code_DataNoChange),
+		})
+	}()
+
+	switch in.Type {
+	case apiconfig.ConfigDiscoverRequest_CONFIG_FILE:
+		action = metrics.ActionGetConfigFile
+		version, _ := strconv.ParseUint(in.GetRevision(), 10, 64)
+		ret := g.configServer.GetConfigFileWithCache(ctx, &apiconfig.ClientConfigFileInfo{
+			Namespace: in.GetConfigFile().GetNamespace(),
+			Group:     in.GetConfigFile().GetGroup(),
+			FileName:  in.GetConfigFile().GetFileName(),
+			Version:   wrapperspb.UInt64(version),
+			PublicKey: in.GetConfigFile().GetPublicKey(),
+		})
+		out = api.NewConfigDiscoverResponse(apimodel.Code(ret.GetCode().GetValue()))
+		out.ConfigFile = ret.GetConfigFile()
+		out.Type = apiconfig.ConfigDiscoverResponse_CONFIG_FILE
+		out.Revision = strconv.Itoa(int(out.GetConfigFile().GetVersion().GetValue()))
+	case apiconfig.ConfigDiscoverRequest_CONFIG_FILE_Names:
+		action = metrics.ActionListConfigFiles
+		ret := g.configServer.GetConfigFileNamesWithCache(ctx, &apiconfig.ConfigFileGroupRequest{
+			Revision: wrapperspb.String(in.GetRevision()),
+			ConfigFileGroup: &apiconfig.ConfigFileGroup{
+				Namespace: in.GetConfigFile().GetNamespace(),
+				Name:      in.GetConfigFile().GetGroup(),
+			},
+		})
+		out = api.NewConfigDiscoverResponse(apimodel.Code(ret.GetCode().GetValue()))
+		out.ConfigFileNames = ret.GetConfigFileInfos()
+		out.Type = apiconfig.ConfigDiscoverResponse_CONFIG_FILE_Names
+		out.Revision = ret.GetRevision().GetValue()
+	case apiconfig.ConfigDiscoverRequest_CONFIG_FILE_GROUPS:
+		action = metrics.ActionListConfigGroups
+		req := in.GetConfigFile()
+		req.Md5 = wrapperspb.String(in.GetRevision())
+		out = g.configServer.GetConfigGroupsWithCache(ctx, req)
+		out.Type = apiconfig.ConfigDiscoverResponse_CONFIG_FILE_GROUPS
+	default:
+		out = api.NewConfigDiscoverResponse(apimodel.Code_InvalidDiscoverResource)
+	}
+
+	return out
+}