feat: add transfer service (#8201)

--story=119241405

Author: wcy00000000000000

docs/apidoc/inner/transfer-service/sync_cmdb_data.md

@@ -0,0 +1,61 @@
+### 请求方式
+
+POST /transfer/v3/sync/cmdb/data
+
+### 描述
+
+触发同步cmdb数据任务
+
+注意：
+
+- 该接口仅用于在源环境触发同步指定的cmdb数据任务
+- 该接口是异步接口，触发后就会返回，可以通过rid追踪触发的同步任务
+- 同一时间只允许一个同步任务
+
+### 输入参数
+
+| 参数名称          | 参数类型             | 必选 | 描述                                                                                                 |
+|---------------|------------------|----|----------------------------------------------------------------------------------------------------|
+| resource_type | string           | 是  | 同步数据的资源类型，枚举值：biz,set,module,host,host_relation,object_instance,inst_asst,service_instance,process |
+| sub_resource  | string           | 否  | 下级数据类型。resource为object_instance和inst_asst时代表需要同步的模型的bk_obj_id                                      |
+| is_all        | bool             | 否  | 是否同步全量数据，is_all为false时必须传start或end                                                                 |
+| start         | map[string]int64 | 否  | 同步的起始区间ID信息，同步的数据不包含该ID代表的数据                                                                       |
+| end           | map[string]int64 | 否  | 同步的结束区间ID信息，同步的数据包含该ID代表的数据                                                                        |
+
+### 调用示例
+
+```json
+{
+  "resource_type": "object_instance",
+  "sub_resource": "bk_switch",
+  "is_all": false,
+  "start": {
+    "bk_inst_id": 10
+  },
+  "end": {
+    "bk_inst_id": 100
+  }
+}
+```
+
+### 响应示例
+
+```json
+{
+  "result": true,
+  "code": 0,
+  "message": "",
+  "permission": null,
+  "data": null
+}
+```
+
+### 响应参数说明
+
+| 参数名称       | 参数类型   | 描述                         |
+|------------|--------|----------------------------|
+| result     | bool   | 请求成功与否。true:请求成功；false请求失败 |
+| code       | int    | 错误编码。 0表示success，>0表示失败错误  |
+| message    | string | 请求失败返回的错误信息                |
+| permission | object | 权限信息                       |
+| data       | object | 请求返回的数据                    |

docs/resource/img/transfer-service/id_generator.png

@@ -0,0 +1,8 @@
+FROM hub.bktencent.com/blueking/centos7-cmdb:base
+ENV container docker
+COPY cmdb_transferservice /data/cmdb/cmdb_transferservice
+RUN mkdir /data/cmdb/cmdb_transferservice/logs
+RUN chmod +x /data/cmdb/cmdb_transferservice/cmdb_transferservice
+#time zone setting
+ENV TimeZone=Asia/Shanghai
+RUN ln -snf /usr/share/zoneinfo/$TimeZone /etc/localtime && echo $TimeZone > /etc/timezone

docs/resource/img/transfer-service/id_rule.png

@@ -0,0 +1,67 @@
+/*
+ * Tencent is pleased to support the open source community by making
+ * 蓝鲸智云 - 配置平台 (BlueKing - Configuration System) available.
+ * Copyright (C) 2017 THL A29 Limited,
+ * a Tencent company. All rights reserved.
+ * Licensed under the MIT License (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at http://opensource.org/licenses/MIT
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on
+ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
+ * either express or implied. See the License for the
+ * specific language governing permissions and limitations under the License.
+ * We undertake not to change the open source license (MIT license) applicable
+ * to the current version of the project delivered to anyone in the future.
+ */
+
+package types
+
+import (
+	"encoding/json"
+)
+
+// PushSyncDataOpt is the push full sync data option
+type PushSyncDataOpt struct {
+	ResType     ResType `json:"resource_type"`
+	SubRes      string  `json:"sub_resource"`
+	IsIncrement bool    `json:"is_increment"`
+	Data        any     `json:"data"`
+}
+
+// PullSyncDataOpt is the pull sync data option
+type PullSyncDataOpt struct {
+	ResType     ResType `json:"resource_type"`
+	SubRes      string  `json:"sub_resource"`
+	IsIncrement bool    `json:"is_increment"`
+	Ack         bool    `json:"ack"`
+}
+
+// TransferMediumResp is the transfer medium response
+type TransferMediumResp[T any] struct {
+	Result  bool   `json:"result"`
+	Code    int    `json:"code"`
+	Message string `json:"message"`
+	Data    T      `json:"data"`
+}
+
+// PullSyncDataRes is the pull sync data result data
+type PullSyncDataRes struct {
+	Total int64           `json:"total"`
+	Info  json.RawMessage `json:"info"`
+}
+
+// FullSyncTransData is the full sync transfer data
+type FullSyncTransData struct {
+	Name  string           `json:"name"`
+	Start map[string]int64 `json:"start"`
+	End   map[string]int64 `json:"end"`
+	Data  any              `json:"data"`
+}
+
+// IncrSyncTransData is the incremental sync transfer data
+type IncrSyncTransData struct {
+	Name       string                       `json:"name"`
+	UpsertInfo map[string][]json.RawMessage `json:"upsert_info"`
+	DeleteInfo map[string][]json.RawMessage `json:"delete_info"`
+}

docs/resource/img/transfer-service/sync.png

@@ -54,3 +54,41 @@ func (o *CreateSyncDataOption) Validate() errors.RawErrorInfo {
 
 	return errors.RawErrorInfo{}
 }
+
+// SyncCmdbDataOption defines sync cmdb data option
+type SyncCmdbDataOption struct {
+	ResType ResType          `json:"resource_type"`
+	SubRes  string           `json:"sub_resource"`
+	IsAll   bool             `json:"is_all"`
+	Start   map[string]int64 `json:"start"`
+	End     map[string]int64 `json:"end"`
+}
+
+// Validate sync cmdb data option
+func (o *SyncCmdbDataOption) Validate() errors.RawErrorInfo {
+	if rawErr := o.ResType.Validate(o.SubRes); rawErr.ErrCode != 0 {
+		return rawErr
+	}
+
+	if o.IsAll {
+		if len(o.Start) != 0 || len(o.End) != 0 {
+			return errors.RawErrorInfo{
+				ErrCode: common.CCErrCommParamsIsInvalid,
+				Args:    []interface{}{"is_all", "start", "end"},
+			}
+		}
+		return errors.RawErrorInfo{}
+	}
+
+	if len(o.Start) == 0 && len(o.End) == 0 {
+		return errors.RawErrorInfo{
+			ErrCode: common.CCErrCommParamsIsInvalid,
+			Args:    []interface{}{"start", "end"},
+		}
+	}
+
+	return errors.RawErrorInfo{}
+}
+
+// InfiniteEndID represent infinity for end id of id rule info
+const InfiniteEndID int64 = -1

docs/support-file/dockerfile/transferservice/dockerfile

@@ -15,6 +15,7 @@
  * to the current version of the project delivered to anyone in the future.
  */
 
+// Package types defines cmdb data syncer types
 package types
 
 import (
@@ -38,30 +39,58 @@ const (
 	HostRelation ResType = "host_relation"
 	// ObjectInstance is the object instance synchronize resource type
 	ObjectInstance ResType = "object_instance"
+	// QuotedInstance is the quoted instance synchronize resource type
+	QuotedInstance ResType = "quoted_instance"
 	// InstAsst is the instance association synchronize resource type
 	InstAsst ResType = "inst_asst"
+	// ServiceInstance is the service instance synchronize resource type
+	ServiceInstance ResType = "service_instance"
+	// Process is the process synchronize resource type
+	Process ResType = "process"
+	// ProcessRelation is the process relation synchronize resource type
+	ProcessRelation ResType = "process_relation"
 )
 
-// AllResTypeMap stores all synchronize resource type
-var AllResTypeMap = map[ResType]struct{}{
-	Biz:            {},
-	Set:            {},
-	Module:         {},
-	Host:           {},
-	HostRelation:   {},
-	ObjectInstance: {},
-	InstAsst:       {},
+var (
+	// allResType is all synchronize resource type in the order of dependency
+	allResType = []ResType{Biz, ObjectInstance, Set, Module, Host, HostRelation, InstAsst, ServiceInstance, Process,
+		ProcessRelation, QuotedInstance}
+	allResTypeMap = make(map[ResType]struct{})
+)
+
+func init() {
+	for _, resType := range allResType {
+		allResTypeMap[resType] = struct{}{}
+	}
+}
+
+// ListAllResType list all synchronize resource type
+func ListAllResType() []ResType {
+	return allResType
+}
+
+// ListAllResTypeForIncrSync list all synchronize resource type for incremental sync
+func ListAllResTypeForIncrSync() []ResType {
+	incrResTypes := make([]ResType, 0)
+	for _, resType := range allResType {
+		if resType == QuotedInstance {
+			continue
+		}
+		incrResTypes = append(incrResTypes, resType)
+	}
+	return incrResTypes
 }
 
 // ResTypeWithSubResMap stores all synchronize resource type with sub resource
 var ResTypeWithSubResMap = map[ResType]struct{}{
 	ObjectInstance: {},
 	InstAsst:       {},
+	QuotedInstance: {},
 }
 
 // Validate resource type
 func (r ResType) Validate(subRes string) ccErr.RawErrorInfo {
-	_, exists := AllResTypeMap[r]
+	_, exists := allResTypeMap[r]
 	if !exists {
 		return ccErr.RawErrorInfo{
 			ErrCode: common.CCErrCommParamsIsInvalid,

pkg/synchronize/types/medium.go

@@ -0,0 +1,58 @@
+/*
+ * Tencent is pleased to support the open source community by making
+ * 蓝鲸智云 - 配置平台 (BlueKing - Configuration System) available.
+ * Copyright (C) 2017 THL A29 Limited,
+ * a Tencent company. All rights reserved.
+ * Licensed under the MIT License (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at http://opensource.org/licenses/MIT
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on
+ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
+ * either express or implied. See the License for the
+ * specific language governing permissions and limitations under the License.
+ * We undertake not to change the open source license (MIT license) applicable
+ * to the current version of the project delivered to anyone in the future.
+ */
+
+package types
+
+import (
+	"encoding/json"
+
+	"configcenter/src/common/watch"
+)
+
+// ListDataOpt is the list data option
+type ListDataOpt struct {
+	SubRes string
+	Start  map[string]int64
+	End    map[string]int64
+}
+
+// ListDataRes is the list data result
+type ListDataRes struct {
+	IsAll     bool
+	Data      any
+	NextStart map[string]int64
+}
+
+// CompDataRes is the compare data result
+type CompDataRes struct {
+	Insert       any
+	Update       any
+	Delete       any
+	RemainingSrc any
+}
+
+// FullSyncLockKey is the full sync lock key
+const FullSyncLockKey = "cmdb_syncer:full_sync_lock"
+
+// EventInfo is the incremental sync event info
+type EventInfo struct {
+	EventType watch.EventType
+	ResType   ResType
+	Oid       string
+	SubRes    []string
+	Detail    json.RawMessage
+}

pkg/synchronize/types/protocol.go

@@ -535,6 +535,22 @@ def generate_config_file(
     caFile:
     # 用于解密根据RFC1423加密的证书密钥的PEM块
     password:
+
+# CMDB同步服务配置
+transferService:
+  # 是否开启同步, 默认为false
+  enableSync: false
+  # 是否开启增量同步, 默认为false
+  enableIncrSync: false
+  # 同步服务的名称
+  name:
+  # 同步角色，src表示源环境(将数据同步到目标环境)，dest表示目标环境(接收源环境的同步数据)
+  role:
+  # 全量同步周期，单位：小时，仅源环境需要配置
+  syncIntervalHours:
+  # 传输介质地址
+  transferMediumAddress:
+    - transfer.example.com
     '''
 
     template = FileTemplate(common_file_template_str)
@@ -725,7 +741,8 @@ def main(argv):
         "cmdb_taskserver": 60012,
         "cmdb_cloudserver": 60013,
         "cmdb_authserver": 60014,
-        "cmdb_cacheservice": 50010
+        "cmdb_cacheservice": 50010,
+        "cmdb_transferservice": 50011
     }
     arr = [
         "help", "discovery=", "database=", "redis_ip=", "redis_port=",

pkg/synchronize/types/resource.go

@@ -24,6 +24,8 @@ import (
 // Interface TODO
 type Interface interface {
 	WatchEvent(ctx context.Context, h http.Header, opts *watch.WatchEventOptions) (*string, errors.CCErrorCoder)
+	InnerWatchEvent(ctx context.Context, h http.Header, opts *watch.WatchEventOptions) (*watch.WatchResp,
+		errors.CCErrorCoder)
 }
 
 // NewCacheClient TODO

pkg/synchronize/types/syncer.go

@@ -42,3 +42,26 @@ func (e *eventCache) WatchEvent(ctx context.Context, h http.Header, opts *watch.
 	}
 	return &resp.Data, nil
 }
+
+// InnerWatchEvent watch event for inner api
+func (e *eventCache) InnerWatchEvent(ctx context.Context, h http.Header, opts *watch.WatchEventOptions) (
+	*watch.WatchResp, errors.CCErrorCoder) {
+
+	resp := new(watch.WatchEventResp)
+	err := e.client.Post().
+		WithContext(ctx).
+		Body(opts).
+		SubResourcef("/inner/watch/cache/event").
+		WithHeaders(h).
+		Do().
+		Into(resp)
+
+	if err != nil {
+		return nil, errors.CCHttpError
+	}
+
+	if err := resp.CCError(); err != nil {
+		return nil, err
+	}
+	return resp.Data, nil
+}

scripts/init.py

@@ -169,6 +169,9 @@ const (
 	// BKDBOR the db operator
 	BKDBOR = "$or"
 
+	// BKDBNOR the not or db operator
+	BKDBNOR = "$nor"
+
 	// BKDBAND the db operator
 	BKDBAND = "$and"