feat: watch db support sharding

--story=121164695

Author: wcy00000000000000

docs/apidoc/inner/admin-server/get_sharding_db_config.md

@@ -16,7 +16,7 @@ GET /migrate/v3/find/system/sharding_db_config
   "permission": null,
   "data": {
     "master_db": "masteruuid",
-    "for_new_tenant": "slave1uuid",
+    "for_new_data": "slave1uuid",
     "slave_db": {
       "slave1uuid": {
         "name": "slave1",
@@ -53,7 +53,7 @@ GET /migrate/v3/find/system/sharding_db_config
 | 参数名称           | 参数类型              | 描述                         |
 |----------------|-------------------|----------------------------|
 | master_db      | string            | 主库唯一标识                     |
-| for_new_tenant | string            | 指定新增租户数据写入哪个库，存储这个数据库的唯一标识 |
+| for_new_data | string            | 指定新增租户数据写入哪个库，存储这个数据库的唯一标识 |
 | slave_db       | map[string]object | 从库唯一标识->从库配置的映射            |
 
 #### data.slave_db[key]

docs/apidoc/inner/admin-server/update_sharding_db_config.md

@@ -10,7 +10,7 @@ PUT /migrate/v3/update/system/sharding_db_config
 
 | 参数名称            | 参数类型              | 必选 | 描述                                          |
 |-----------------|-------------------|----|---------------------------------------------|
-| for_new_tenant  | string            | 否  | 指定新增租户数据写入哪个库。对于存量数据库指定它的唯一标识。对于新增的从库指定它的名称 |
+| for_new_data  | string            | 否  | 指定新增租户数据写入哪个库。对于存量数据库指定它的唯一标识。对于新增的从库指定它的名称 |
 | create_slave_db | object array      | 否  | 新增的从库配置数组                                   |
 | update_slave_db | map[string]object | 否  | 更新的从库唯一标识->从库配置的映射                          |
 
@@ -41,7 +41,7 @@ PUT /migrate/v3/update/system/sharding_db_config
 
 ```json
 {
-  "for_new_tenant": "slave1uuid",
+  "for_new_data": "slave1uuid",
   "create_slave_db": [
     {
       "name": "slave2",

src/common/tablenames.go

@@ -117,6 +117,9 @@ const (
 
 	// BKTableNameTenantTemplate is the tenant template(public data that needs to be initialized for all tenants) table
 	BKTableNameTenantTemplate = "TenantTemplate"
+
+	// BKTableNameWatchDBRelation is the db and watch db relation table
+	BKTableNameWatchDBRelation = "WatchDBRelation"
 )
 
 // AllTables is all table names, not include the sharding tables which is created dynamically,
@@ -281,6 +284,7 @@ var platformTableMap = map[string]struct{}{
 	BKTableNameLastWatchEvent:     {},
 	BKTableNameAPITask:            {},
 	BKTableNameAPITaskSyncHistory: {},
+	BKTableNameWatchDBRelation:    {},
 }
 
 // IsPlatformTable returns if the target table is a platform table

src/scene_server/admin_server/app/server.go

@@ -16,7 +16,6 @@ package app
 import (
 	"context"
 	"fmt"
-	"time"
 
 	iamcli "configcenter/src/ac/iam"
 	"configcenter/src/common/auth"
@@ -31,7 +30,6 @@ import (
 	"configcenter/src/scene_server/admin_server/iam"
 	"configcenter/src/scene_server/admin_server/logics"
 	svc "configcenter/src/scene_server/admin_server/service"
-	"configcenter/src/storage/dal/mongo/sharding"
 	"configcenter/src/storage/dal/redis"
 	"configcenter/src/storage/driver/mongodb"
 	"configcenter/src/thirdparty/monitor"
@@ -64,11 +62,10 @@ func Run(ctx context.Context, cancel context.CancelFunc, op *options.ServerOptio
 	db := mongodb.Dal()
 	process.Service.SetDB(db)
 
-	watchDB, err := sharding.NewDisableDBShardingMongo(process.Config.WatchDB.GetMongoConf(), time.Minute)
-	if err != nil {
+	if err = mongodb.SetWatchCli("", &process.Config.WatchDB, process.Config.Crypto); err != nil {
 		return fmt.Errorf("connect watch mongo server failed, err: %v", err)
 	}
-	process.Service.SetWatchDB(watchDB)
+	process.Service.SetWatchDB(mongodb.Dal())
 
 	cache, err := redis.NewFromConfig(process.Config.Redis)
 	if err != nil {

src/scene_server/admin_server/service/sharding.go

@@ -46,9 +46,9 @@ func (s *Service) initShardingApi(api *restful.WebService) {
 
 // ShardingDBConfig is the sharding db config for api
 type ShardingDBConfig struct {
-	MasterDB     string                   `json:"master_db"`
-	ForNewTenant string                   `json:"for_new_tenant"`
-	SlaveDB      map[string]SlaveDBConfig `json:"slave_db"`
+	MasterDB   string                   `json:"master_db"`
+	ForNewData string                   `json:"for_new_data"`
+	SlaveDB    map[string]SlaveDBConfig `json:"slave_db"`
 }
 
 // SlaveDBConfig is the slave db config for api
@@ -69,9 +69,9 @@ func (s *Service) GetShardingDBConfig(req *restful.Request, resp *restful.Respon
 	}
 
 	result := &ShardingDBConfig{
-		MasterDB:     conf.MasterDB,
-		ForNewTenant: conf.ForNewTenant,
-		SlaveDB:      make(map[string]SlaveDBConfig),
+		MasterDB:   conf.MasterDB,
+		ForNewData: conf.ForNewData,
+		SlaveDB:    make(map[string]SlaveDBConfig),
 	}
 
 	for uuid, mongoConf := range conf.SlaveDB {
@@ -134,7 +134,7 @@ func (s *Service) getShardingDBConf(kit *rest.Kit) (*sharding.ShardingDBConf, er
 
 // UpdateShardingDBReq is the update sharding db config request
 type UpdateShardingDBReq struct {
-	ForNewTenant  string                       `json:"for_new_tenant,omitempty"`
+	ForNewData    string                       `json:"for_new_data,omitempty"`
 	CreateSlaveDB []SlaveDBConfig              `json:"create_slave_db,omitempty"`
 	UpdateSlaveDB map[string]UpdateSlaveDBInfo `json:"update_slave_db,omitempty"`
 }
@@ -177,8 +177,8 @@ func (s *Service) UpdateShardingDBConfig(req *restful.Request, resp *restful.Res
 
 	cond := map[string]any{common.MongoMetaID: common.ShardingDBConfID}
 	updateData := map[string]any{
-		"for_new_tenant": updateConf.ForNewTenant,
-		"slave_db":       updateConf.SlaveDB,
+		"for_new_data": updateConf.ForNewData,
+		"slave_db":     updateConf.SlaveDB,
 	}
 	err = s.db.Shard(kit.SysShardOpts()).Table(common.BKTableNameSystem).Update(s.ctx, cond, updateData)
 	if err != nil {
@@ -247,23 +247,23 @@ func (s *Service) genUpdatedShardingDBConf(kit *rest.Kit, dbConf *sharding.Shard
 	}
 
 	// update new tenant db config, check if the new tenant db config exists
-	if conf.ForNewTenant != "" {
+	if conf.ForNewData != "" {
 		// use uuid to specify the new tenant db config for db that already exists
-		_, uuidExists := dbConf.SlaveDB[conf.ForNewTenant]
-		if conf.ForNewTenant == dbConf.MasterDB || uuidExists {
-			dbConf.ForNewTenant = conf.ForNewTenant
+		_, uuidExists := dbConf.SlaveDB[conf.ForNewData]
+		if conf.ForNewData == dbConf.MasterDB || uuidExists {
+			dbConf.ForNewData = conf.ForNewData
 			return dbConf, nil
 		}
 
 		// use name to specify the new tenant db config for new db that doesn't have uuid before creation
-		uuid, nameExists := nameUUIDMap[conf.ForNewTenant]
+		uuid, nameExists := nameUUIDMap[conf.ForNewData]
 		if nameExists {
-			dbConf.ForNewTenant = uuid
+			dbConf.ForNewData = uuid
 			return dbConf, nil
 		}
 
-		blog.Errorf("add new tenant db %s is invalid, rid: %s", conf.ForNewTenant, kit.Rid)
-		return nil, kit.CCError.CCErrorf(common.CCErrCommParamsInvalid, "for_new_tenant")
+		blog.Errorf("add new tenant db %s is invalid, rid: %s", conf.ForNewData, kit.Rid)
+		return nil, kit.CCError.CCErrorf(common.CCErrCommParamsInvalid, "for_new_data")
 	}
 	return dbConf, nil
 }

src/source_controller/cacheservice/cache/cache.go

@@ -27,13 +27,12 @@ import (
 	"configcenter/src/storage/dal"
 	"configcenter/src/storage/driver/mongodb"
 	"configcenter/src/storage/driver/redis"
-	"configcenter/src/storage/reflector"
 	"configcenter/src/storage/stream"
 )
 
 // NewCache new cache service
-func NewCache(reflector reflector.Interface, loopW stream.LoopInterface, isMaster discovery.ServiceManageInterface,
-	watchDB dal.DB) (*ClientSet, error) {
+func NewCache(loopW stream.LoopInterface, isMaster discovery.ServiceManageInterface, watchDB dal.DB) (*ClientSet,
+	error) {
 
 	if err := mainline.NewMainlineCache(loopW); err != nil {
 		return nil, fmt.Errorf("new business cache failed, err: %v", err)

src/source_controller/cacheservice/service/service.go

@@ -37,7 +37,6 @@ import (
 	"configcenter/src/source_controller/cacheservice/event/identifier"
 	"configcenter/src/source_controller/coreservice/core"
 	"configcenter/src/storage/dal/mongo/local"
-	"configcenter/src/storage/reflector"
 	"configcenter/src/storage/stream"
 	"configcenter/src/thirdparty/logplatform/opentelemetry"
 
@@ -94,37 +93,25 @@ func (s *cacheService) SetConfig(cfg options.Config, engine *backbone.Engine, er
 	}
 	s.authManager = extensions.NewAuthManager(engine.CoreAPI, iamCli)
 
-	loopW, loopErr := stream.NewLoopStream(s.cfg.Mongo.GetMongoConf(), engine.ServiceManageInterface)
+	watcher, loopErr := stream.NewLoopStream(s.cfg.Mongo.GetMongoConf(), engine.ServiceManageInterface)
 	if loopErr != nil {
 		blog.Errorf("new loop stream failed, err: %v", loopErr)
 		return loopErr
 	}
 
-	event, eventErr := reflector.NewReflector(s.cfg.Mongo.GetMongoConf())
-	if eventErr != nil {
-		blog.Errorf("new reflector failed, err: %v", eventErr)
-		return eventErr
-	}
-
 	watchDB, dbErr := local.NewMgo(s.cfg.WatchMongo.GetMongoConf(), time.Minute)
 	if dbErr != nil {
 		blog.Errorf("new watch mongo client failed, err: %v", dbErr)
 		return dbErr
 	}
 
-	c, cacheErr := cacheop.NewCache(event, loopW, engine.ServiceManageInterface, watchDB)
+	c, cacheErr := cacheop.NewCache(watcher, engine.ServiceManageInterface, watchDB)
 	if cacheErr != nil {
 		blog.Errorf("new cache instance failed, err: %v", cacheErr)
 		return cacheErr
 	}
 	s.cacheSet = c
 
-	watcher, watchErr := stream.NewLoopStream(s.cfg.Mongo.GetMongoConf(), engine.ServiceManageInterface)
-	if watchErr != nil {
-		blog.Errorf("new loop watch stream failed, err: %v", watchErr)
-		return watchErr
-	}
-
 	ccDB, dbErr := local.NewMgo(s.cfg.Mongo.GetMongoConf(), time.Minute)
 	if dbErr != nil {
 		blog.Errorf("new cc mongo client failed, err: %v", dbErr)

src/storage/dal/mongo/local/db.go

@@ -21,7 +21,6 @@ import (
 	"context"
 
 	"configcenter/src/common/metadata"
-	"configcenter/src/storage/dal/redis"
 	"configcenter/src/storage/dal/types"
 )
 
@@ -60,8 +59,4 @@ type DB interface {
 	CommitTransaction(context.Context, *metadata.TxnCapable) error
 	// AbortTransaction 取消事务
 	AbortTransaction(context.Context, *metadata.TxnCapable) (bool, error)
-
-	// InitTxnManager TxnID management of initial transaction
-	// TODO 后续放到TenantDB里，只用初始化一次
-	InitTxnManager(r redis.Client) error
 }