chore: ci & review bugfix

--story=120702860

Author: wcy00000000000000

pkg/tenant/event.go

@@ -26,13 +26,13 @@ import (
 var (
 	prevTenantInfo      = make(map[string]types.Tenant)
 	tenantEventChannels = make(map[string]chan TenantEvent)
-	tenantEventChLock   sync.RWMutex
+	tenantEventChLock   sync.Mutex
 )
 
 // TenantEvent is the tenant event info
 type TenantEvent struct {
 	EventType EventType
-	TenantID  string
+	Tenant    types.Tenant
 }
 
 // EventType is the tenant event type
@@ -54,14 +54,14 @@ func NewTenantEventChan(name string) <-chan TenantEvent {
 		return ch
 	}
 
-	eventChan := make(chan TenantEvent, 1)
+	eventChan := make(chan TenantEvent)
 	tenantEventChannels[name] = eventChan
 	go func() {
 		for _, tenant := range allTenants {
 			if tenant.Status == types.EnabledStatus {
 				eventChan <- TenantEvent{
 					EventType: Create,
-					TenantID:  tenant.TenantID,
+					Tenant:    tenant,
 				}
 			}
 		}
@@ -85,8 +85,8 @@ func RemoveTenantEventChan(name string) {
 
 // generateAndPushTenantEvent compare the tenant with the previous tenant info to generate and push event
 func generateAndPushTenantEvent(tenants []types.Tenant) {
-	tenantEventChLock.RLock()
-	defer tenantEventChLock.RUnlock()
+	tenantEventChLock.Lock()
+	defer tenantEventChLock.Unlock()
 
 	prevTenantMap := make(map[string]types.Tenant)
 
@@ -95,11 +95,15 @@ func generateAndPushTenantEvent(tenants []types.Tenant) {
 		prevTenantMap[tenantID] = tenant
 
 		prevTenant, exists := prevTenantInfo[tenantID]
-		if !exists && tenant.Status == types.EnabledStatus {
+		if !exists {
+			if tenant.Status != types.EnabledStatus {
+				continue
+			}
+
 			for _, eventChan := range tenantEventChannels {
 				eventChan <- TenantEvent{
 					EventType: Create,
-					TenantID:  tenantID,
+					Tenant:    tenant,
 				}
 			}
 			continue
@@ -113,19 +117,19 @@ func generateAndPushTenantEvent(tenants []types.Tenant) {
 			for _, eventChan := range tenantEventChannels {
 				eventChan <- TenantEvent{
 					EventType: eventType,
-					TenantID:  tenantID,
+					Tenant:    tenant,
 				}
 			}
 		}
 
 		delete(prevTenantInfo, tenantID)
 	}
 
-	for tenantID := range prevTenantInfo {
+	for _, tenant := range prevTenantInfo {
 		for _, eventChan := range tenantEventChannels {
 			eventChan <- TenantEvent{
 				EventType: Delete,
-				TenantID:  tenantID,
+				Tenant:    tenant,
 			}
 		}
 	}

pkg/tenant/tenant.go

@@ -99,8 +99,8 @@ func SetTenant(tenant []types.Tenant) {
 	for _, t := range allTenants {
 		tenantMap[t.TenantID] = &t
 	}
-	generateAndPushTenantEvent(allTenants)
 	lock.Unlock()
+	generateAndPushTenantEvent(allTenants)
 }
 
 func refreshTenantInfo() error {

src/apimachinery/refresh/api.go

@@ -34,10 +34,10 @@ func (r *refresh) RefreshTenant(moduleName string) ([]types.Tenant, error) {
 
 	case commontypes.CC_MODULE_APISERVER:
 		r.capability.Discover = r.disc.ApiServer()
-
 	case commontypes.CC_MODULE_TASK:
 		r.capability.Discover = r.disc.TaskServer()
-
+	case commontypes.CC_MODULE_CACHESERVICE:
+		r.capability.Discover = r.disc.CacheService()
 	default:
 		return nil, fmt.Errorf("unsupported refresh module: %s", moduleName)
 	}

src/common/tablenames.go

@@ -63,7 +63,6 @@ const (
 	BKTableNameHistory          = "History"
 	BKTableNameHostFavorite     = "HostFavourite"
 	BKTableNameAuditLog         = "AuditLog"
-	BKTableNamePlatformAuditLog = "PlatformAuditLog"
 	BKTableNameUserAPI          = "UserAPI"
 	BKTableNameDynamicGroup     = "DynamicGroup"
 	BKTableNameUserCustom       = "UserCustom"
@@ -288,12 +287,14 @@ var platformTableMap = map[string]struct{}{
 	BKTableNameIDgenerator:        {},
 	BKTableNameTenant:             {},
 	BKTableNameTenantTemplate:     {},
-	BKTableNamePlatformAuditLog:   {},
 	BKTableNameWatchToken:         {},
 	BKTableNameAPITask:            {},
 	BKTableNameAPITaskSyncHistory: {},
 	BKTableNameWatchDBRelation:    {},
 	BKTableNameFullSyncCond:       {},
+	BKTableNameCacheWatchToken:    {},
+	"SrcSyncDataToken":            {},
+	"SrcSyncDataCursor":           {},
 }
 
 // IsPlatformTable returns if the target table is a platform table

src/scene_server/admin_server/logics/tenant.go

@@ -18,15 +18,17 @@
 package logics
 
 import (
+	"context"
 	"fmt"
 
 	"configcenter/pkg/tenant"
-	"configcenter/pkg/tenant/types"
 	"configcenter/src/apimachinery"
 	"configcenter/src/common/blog"
 	"configcenter/src/common/http/rest"
 	commontypes "configcenter/src/common/types"
+	"configcenter/src/storage/dal"
 	"configcenter/src/storage/dal/mongo/local"
+	"configcenter/src/storage/dal/mongo/sharding"
 )
 
 // NewTenantInterface get new tenant cli interface
@@ -52,19 +54,24 @@ func GetNewTenantCli(kit *rest.Kit, cli interface{}) (local.DB, string, error) {
 }
 
 // RefreshTenants refresh tenant info, skip tenant verify for apiserver
-func RefreshTenants(coreAPI apimachinery.ClientSetInterface) error {
+func RefreshTenants(coreAPI apimachinery.ClientSetInterface, db dal.Dal) error {
+	tenants, err := tenant.GetAllTenantsFromDB(context.Background(),
+		db.Shard(sharding.NewShardOpts().WithIgnoreTenant()))
+	if err != nil {
+		blog.Errorf("get all tenants failed, err: %v", err)
+		return err
+	}
+	tenant.SetTenant(tenants)
 
-	var tenants []types.Tenant
-	var err error
-	needRefreshServer := []string{commontypes.CC_MODULE_APISERVER, commontypes.CC_MODULE_TASK}
+	needRefreshServer := []string{commontypes.CC_MODULE_APISERVER, commontypes.CC_MODULE_TASK,
+		commontypes.CC_MODULE_CACHESERVICE}
 	for _, module := range needRefreshServer {
-		tenants, err = coreAPI.Refresh().RefreshTenant(module)
+		_, err = coreAPI.Refresh().RefreshTenant(module)
 		if err != nil {
 			blog.Errorf("refresh tenant info failed, module: %s, err: %v", module, err)
 			return err
 		}
 	}
 
-	tenant.SetTenant(tenants)
 	return nil
 }

src/scene_server/admin_server/service/migrate.go

@@ -86,6 +86,11 @@ func (s *Service) migrateDatabase(req *restful.Request, resp *restful.Response)
 		return
 	}
 
+	// refresh tenants, ignore refresh tenants error
+	if err = logics.RefreshTenants(s.CoreAPI, s.db); err != nil {
+		blog.Errorf("refresh tenant failed, err: %v", err)
+	}
+
 	if err = s.createWatchDBChainCollections(kit); err != nil {
 		blog.Errorf("create watch db chain collections failed, err: %v", err)
 		resp.WriteError(http.StatusInternalServerError, &metadata.RespError{
@@ -94,11 +99,6 @@ func (s *Service) migrateDatabase(req *restful.Request, resp *restful.Response)
 		return
 	}
 
-	// refresh tenants, ignore refresh tenants error
-	if err = logics.RefreshTenants(s.CoreAPI); err != nil {
-		blog.Errorf("refresh tenant failed, err: %v", err)
-	}
-
 	resp.WriteEntity(metadata.NewSuccessResp(result))
 }
 
@@ -126,14 +126,12 @@ func (s *Service) createWatchDBChainCollections(kit *rest.Kit) error {
 		}
 
 		err = tenant.ExecForAllTenants(func(tenantID string) error {
-			// TODO 在新增租户初始化时同时增加watch相关表，并刷新cache的tenant
 			return s.addTenantWatchToken(kit.NewKit().WithTenant(tenantID), cursorType, key)
 		})
 		if err != nil {
 			return err
 		}
 
-		// TODO 在新增DB时同时增加db relation和token数据
 		err = s.createWatchTokenForEventKey(kit, key, watchDBToDBRelation)
 		if err != nil {
 			return err
@@ -321,16 +319,20 @@ func (s *Service) createWatchIndexes(kit *rest.Kit, cursorType watch.CursorType,
 		existIdxMap[index.Name] = true
 	}
 
+	createIndexes := make([]daltypes.Index, 0)
 	for _, index := range indexes {
-		if _, exist := existIdxMap[index.Name]; exist {
-			continue
+		if _, exist := existIdxMap[index.Name]; !exist {
+			createIndexes = append(createIndexes, index)
 		}
+	}
+	if len(createIndexes) == 0 {
+		return nil
+	}
 
-		err = s.watchDB.Shard(kit.ShardOpts()).Table(key.ChainCollection()).CreateIndex(s.ctx, index)
-		if err != nil && !mongodb.IsDuplicatedError(err) {
-			blog.Errorf("create indexes for table %s failed, err: %v, rid: %s", key.ChainCollection(), err, kit.Rid)
-			return err
-		}
+	err = s.watchDB.Shard(kit.ShardOpts()).Table(key.ChainCollection()).BatchCreateIndexes(s.ctx, createIndexes)
+	if err != nil && !mongodb.IsDuplicatedError(err) {
+		blog.Errorf("create indexes for table %s failed, err: %v, rid: %s", key.ChainCollection(), err, kit.Rid)
+		return err
 	}
 	return nil
 }

src/scene_server/admin_server/service/sharding.go

@@ -262,7 +262,7 @@ func (s *Service) genUpdatedShardingDBConf(kit *rest.Kit, dbConf *sharding.Shard
 			return dbConf, nil
 		}
 
-		blog.Errorf("add new tenant db %s is invalid, rid: %s", conf.ForNewData, kit.Rid)
+		blog.Errorf("for new data db %s is invalid, rid: %s", conf.ForNewData, kit.Rid)
 		return nil, kit.CCError.CCErrorf(common.CCErrCommParamsInvalid, "for_new_data")
 	}
 	return dbConf, nil
@@ -292,7 +292,7 @@ func (s *Service) genDBSlaveConf(kit *rest.Kit, name string, disabled bool, conf
 }
 
 func (s *Service) saveUpdateShardingDBAudit(kit *rest.Kit, preConf, curConf *sharding.ShardingDBConf) error {
-	id, err := s.db.Shard(kit.SysShardOpts()).NextSequence(kit.Ctx, common.BKTableNamePlatformAuditLog)
+	id, err := s.db.Shard(kit.SysShardOpts()).NextSequence(kit.Ctx, common.BKTableNameAuditLog)
 	if err != nil {
 		blog.Errorf("generate next audit log id failed, err: %v, rid: %s", err, kit.Rid)
 		return err