feat(backend): db-event-consumer 启用只用配置模式 TencentBlueKing#16453

Author: seanlook

dbm-services/common/db-config/internal/api/config_base.go

@@ -16,7 +16,7 @@ type BKBizIDDef struct {
 // RequestType TODO
 type RequestType struct {
 	// 配置文件修改动作的请求类型，`SaveOnly`: 仅保存, `SaveAndPublish`保存并发布
-	ReqType string `json:"req_type" form:"req_type" validate:"required,enums" enums:"SaveOnly,SaveAndPublish"`
+	ReqType string `json:"req_type" form:"req_type" validate:"enums" enums:"SaveOnly,SaveAndPublish"`
 }
 
 // OperationType TODO

dbm-services/common/db-config/internal/handler/simple/config_item.go

@@ -1,13 +1,14 @@
 package simple
 
 import (
+	"fmt"
+
 	"bk-dbconfig/internal/api"
 	"bk-dbconfig/internal/handler"
 	"bk-dbconfig/internal/service/simpleconfig"
 	"bk-dbconfig/pkg/constvar"
 	"bk-dbconfig/pkg/core/logger"
 	"bk-dbconfig/pkg/util"
-	"fmt"
 
 	"github.com/gin-gonic/gin"
 )
@@ -156,10 +157,20 @@ func (cf *Config) UpdateConfigFileItems(ctx *gin.Context) {
 		handler.SendResponse(ctx, err, nil)
 		return
 	}
-	if err = simpleconfig.CheckValidConfType(r.ConfFileInfo.Namespace, r.ConfFileInfo.ConfType,
-		r.ConfFileInfo.ConfFile, r.LevelName, 1); err != nil {
+	if versioned, err := simpleconfig.CheckValidConfFile(r.ConfFileInfo.Namespace, r.ConfFileInfo.ConfType,
+		r.ConfFileInfo.ConfFile, r.LevelName); err != nil {
 		handler.SendResponse(ctx, err, nil)
 		return
+	} else if versioned != "" {
+		r = api.UpsertConfItemsReq{
+			RequestType:      api.RequestType{ReqType: constvar.MethodGenAndPublish},
+			SaveConfItemsReq: r.SaveConfItemsReq,
+		}
+	} else {
+		r = api.UpsertConfItemsReq{
+			RequestType:      api.RequestType{ReqType: constvar.MethodSave},
+			SaveConfItemsReq: r.SaveConfItemsReq,
+		}
 	}
 	opUser := api.GetHeaderUsername(ctx.GetHeader(constvar.BKApiAuthorization))
 	if resp, err = simpleconfig.UpdateConfigFileItems(&r, opUser); err != nil {
@@ -184,33 +195,5 @@ func (cf *Config) UpdateConfigFileItems(ctx *gin.Context) {
 // @Failure      400  {object}  api.HTTPClientErrResp
 // @Router       /bkconfig/v1/confitem/save [post]
 func (cf *Config) SaveConfigFileItems(ctx *gin.Context) {
-	var r api.SaveConfItemsReq
-	var resp *api.UpsertConfItemsResp
-	var err error
-	defer util.LoggerErrorStack(logger.Error, err)
-
-	if err = ctx.BindJSON(&r); err != nil {
-		handler.SendResponse(ctx, err, nil)
-		return
-	}
-	if err = r.Validate(); err != nil {
-		handler.SendResponse(ctx, err, nil)
-		return
-	}
-	if err = simpleconfig.CheckValidConfType(r.ConfFileInfo.Namespace, r.ConfFileInfo.ConfType,
-		r.ConfFileInfo.ConfFile, r.LevelName, 0); err != nil {
-		handler.SendResponse(ctx, err, nil)
-		return
-	}
-	r2 := api.UpsertConfItemsReq{
-		RequestType:      api.RequestType{ReqType: constvar.MethodSave},
-		SaveConfItemsReq: r,
-	}
-	opUser := api.GetHeaderUsername(ctx.GetHeader(constvar.BKApiAuthorization))
-	if resp, err = simpleconfig.UpdateConfigFileItems(&r2, opUser); err != nil {
-		handler.SendResponse(ctx, err, nil)
-		return
-	} else {
-		handler.SendResponse(ctx, nil, resp)
-	}
+	cf.UpdateConfigFileItems(ctx)
 }

dbm-services/common/db-config/internal/service/simpleconfig/config_meta.go

@@ -149,6 +149,23 @@ func QueryConfigTypeInfo(r *api.QueryConfigTypeReq) (*api.QueryConfigTypeResp, e
 	return resp, nil
 }
 
+func CheckValidConfFile(namespace, confType, confFile, levelName string) (string, error) {
+	msg := fmt.Sprintf("namespace=%s, conf_type=%s, conf_file=%s", namespace, confType, confFile)
+	fd := api.BaseConfFileDef{Namespace: namespace, ConfType: confType, ConfFile: confFile}
+	if f, e := model.CacheGetConfigFile(fd); e != nil {
+		return "", errors.Wrapf(errno.ErrConfFile, "ErrFound:%s for %s", e.Error(), msg)
+	} else if f == nil {
+		return "", errors.Wrapf(errno.ErrNamespaceType, msg)
+	} else {
+		if levelName != "" {
+			if !util.StringsHas(f.LevelNameList, levelName) {
+				return "", errors.Wrapf(errno.ErrLevelName, "allowed [%s] but given %s", f.LevelNames, levelName)
+			}
+		}
+		return f.LevelVersioned, nil
+	}
+}
+
 // CheckValidConfType 检查 namespace, conf_type, conf_file, level_name 的合法性
 // 如果 level_name = "" 不检查 level_name
 // 如果 needVersioned >=2 不做版本化相关检查

dbm-services/common/db-event-consumer/README.md

@@ -41,18 +41,26 @@ db-event-consumer -c config.yaml
 
 ### 2. 非严格 schema
 自动根据 kafka 消息的内容，使用 `map[string]interface{}` 来反序列化，然后直接生成 insert 语句。
+这种方式不需要在 db-event-consumer 里定义 model 的代码，直接启动时拉起配置即可。
 
-这种方式需要提前在目标 datasource 创建好表结构，如果表结构字段缺少，会导致入库失败。
+这种方式需要提前在目标 datasource 创建好表结构，或者在其他地方有 migrate 来维护表结构和索引约束。
+如果表结构字段缺少，会导致入库失败。
 
 ```
 - topic: "mysql_binlog_result"
   model_table: "tb_mysql_binlog_result"
   strict_schema: false
-  datasource: "prod_bk_dbm_report"
+  skip_migrate_schema: true
+  write_mode: upsert
+  datasource: "prod_bk_dbm_report_raw"
+  omit_fields: []
 ```
-`strict_schema=false` 时，model_table 的值用于拼成 insert table name.
 
-非严格 schema 方式可以快速验证入库效果，不用代码定义 model 。但不方便在各个环境环境移植，不推荐。
+- `strict_schema=false` 时，model_table 的值用于拼成 insert table name.
+- `write_mode=upsert`, 当遇到冲突时，使用何种处理方式，`replace`,`upsert`,`insert_ignore`, `insert`
+- `omit_fields`, 忽略哪些字段，不拼在 insert values 里
+
+非严格 schema 方式可以快速验证入库效果，不用代码定义 model。如果你需要对上报的内容做重组，二次加工后再入库，则不适合这个方式。
 
 ## 自定义入库方式
 可以自定义 schema migrate 方式和数据入库方式，可以参考 `MysqlBackupResultModel`

dbm-services/common/db-event-consumer/pkg/config/sinker_config.go

@@ -33,7 +33,7 @@ type SinkerConfig struct {
 	FetchMinBytes int32 `yaml:"fetch_min_bytes"`
 	// SinkBatchSize 一次 fetch 可能有多条记录，sink_batch_size 控制多少次 fetch 合并成一次 sink. default 1
 	SinkBatchSize int `yaml:"sink_batch_size"`
-	// WriteMode default is insert_ignore, allowed: insert_ignore, insert, upsert
+	// WriteMode default is upsert, allowed: insert_ignore, insert, upsert
 	WriteMode  string `yaml:"write_mode"`
 	Datasource string `yaml:"datasource"`
 }

dbm-services/common/db-event-consumer/pkg/consumer/consumer.go

@@ -34,6 +34,7 @@ type AnySinker struct {
 	strictSchema bool
 }
 
+// Setup run default migrate or custom migrate
 func (s *AnySinker) Setup(sarama.ConsumerGroupSession) error {
 	var err error
 	if s.Sinker.RuntimeConfig.SkipMigrateSchema || !s.strictSchema {

dbm-services/common/db-event-consumer/pkg/consumer/new_consumer_group.go

@@ -37,7 +37,7 @@ func (s *Sinker) NewSinkHandler() (sarama.ConsumerGroupHandler, error) {
 	s.DSWriter.SetWriteMode(s.RuntimeConfig.WriteMode)
 	if ok {
 		if !*s.RuntimeConfig.StrictSchema {
-			return nil, fmt.Errorf("registerd table[%s] need strict_schema=true", s.RuntimeConfig.ModelTable)
+			return nil, fmt.Errorf("registered table[%s] need strict_schema=true", s.RuntimeConfig.ModelTable)
 		}
 		modelType := reflect.TypeOf(modelTable).Elem()
 		if modelType.Kind() == reflect.Ptr {

dbm-services/common/db-event-consumer/pkg/sinker/mysql_raw_writer.go

@@ -79,8 +79,10 @@ func (w *MysqlRawWriter) WriteBatch(table interface{}, models interface{}) error
 	}
 	if w.writeMode == cst.ModeUpsert || w.writeMode == cst.ModeReplace {
 		_, err = w.session.Replace()
-	} else {
+	} else if w.writeMode == cst.ModeInsertIgnore {
 		_, err = w.session.InsertIgnore()
+	} else {
+		_, err = w.session.Insert()
 	}
 	return err
 }

dbm-ui/backend/components/dbconfig/client.py

@@ -56,12 +56,12 @@ def __init__(self):
         self.save_conf_item = self.generate_data_api(
             method="POST",
             url="bkconfig/v1/confitem/save",
-            description=_("保存不可变配置（如字符集等）"),
+            description=_("保存无版本概念的配置（如部署，备份，监控），现在与 upsert 接口完全相同了"),
         )
         self.upsert_conf_item = self.generate_data_api(
             method="POST",
             url="bkconfig/v1/confitem/upsert",
-            description=_("编辑发布层级（业务、集群、模块）配置"),
+            description=_("保存有版本概念的配置（如参数）"),
         )
         self.batch_get_conf_item = self.generate_data_api(
             method="POST",

helm-charts/bk-dbm/templates/configmaps/db-event-consumer-configmap.yaml

@@ -45,7 +45,7 @@ data:
         password: "{{ $reportDB.password }}"
         database: "{{ $reportDB.name }}"
         charset: utf8
-    - name: prod_bk_dbm_report2
+    - name: prod_bk_dbm_report_raw
       type: mysql_raw
       dsn:
         address: "{{ $reportDB.host }}:{{ $reportDB.port }}"