第 6 章 核心业务流程逐段拆解讲解

1 学习目标

学完本章你应该能够：

1. 完整走通资源同步主链路：CreateTask → 解密凭证 → 分页拉云 → 逐台入库 → 任务记账，并说出每一步对应的源码位置；
2. 走通资源检索链路：参数绑定 → 标签选择器解析（四种操作符）→ JOIN 查询，并解释为什么带标签筛选时 JOIN 方向要从 LEFT 切成 RIGHT；
3. 走通凭证录入与加密链路：校验 → CBC 加密 → 落库，说清 xid 作为凭证 ID 的用意；
4. 解释 pager.Pager 抽象与对账（先查后写）的职责边界——provider 只"翻译数据"，对账职责完全在业务层；
5. 指出同步链路的三个健壮性缺口：30 分钟硬超时、无法外部取消、先 go 后 insert 的竞态窗口。

前置知识

• 第 4 章的数据模型：hash 列、血缘链路、多租户三件套；
• 第 5 章的分层结构：Controller 三段式、Service 两种装配、DAO 两范式（本章会反复用到）；
• Go 基础：goroutine、context.WithTimeout、defer/recover 的语义。

本章你会动手做的事

1. 跟着 6.2.1 的全景图，在编辑器里从 CreateTask 一路点跳到 doSyncHost，把五个函数的文件行号标出来；
2. 用 tag=env=~prod&tag=team=bigdata 实际调一次检索接口，对照四种操作符看解析结果；
3. 录入一条测试凭证，去库里确认 api_secret 带上了 @ciphered@ 前缀。

先看本章知识地图——三条主动脉加两个核心抽象：

2 知识点通俗系统化讲解

2.1 资源同步：CMDB 的"主动脉"

同步是 CMDB 存在的意义。整个链路横跨 5 个模块（task 编排 → secret 解密 → provider 采集 → host 对账 → resource 落库），我们先把全景图挂出来，然后逐段精讲：

① 客户端 → gRPC/HTTP CreateTask（指定: 凭证ID + 资源类型 + 地域）
② task.CreatTask          apps/task/impl/task.go:26
   ├─ task.NewTaskFromReq          生成 Task（id 用 xid）
   ├─ s.secret.DescribeSecret      按凭证ID取凭证
   ├─ 校验 region 是否在凭证允许范围
   ├─ secret.DecryptAPISecret      解密出明文 SK
   └─ switch ResourceType: TYPE_HOST → go s.syncHost(...)
③ task.syncHost           apps/task/impl/host.go:23
   ├─ t.Run() 标记任务开始
   ├─ switch secret.Vendor: 按厂商 new 对应 Operator
   └─ for pager.Next(): pager.Scan(ctx, set) → 逐台 doSyncHost
④ task.doSyncHost         apps/task/impl/host.go:116
   ├─ ins.Resource.Meta.SyncAt = now
   ├─ s.host.SyncHost(ctx, ins)    调 host 模块 upsert
   └─ 成功/失败 → task_record 落一条明细
⑤ host.SyncHost           apps/host/impl/host.go:14
   ├─ DescribeHost 按 Meta.Id 查重
   ├─ 存在 → update；不存在 → save（对账）
   └─ 写 resource_host 表

2.2 分页器抽象：云端的"水龙头"

为什么 provider 层返回 pager.Pager 而不是一次拉全量？因为你不知道对方面前有多少数据——云上可能有几万台主机，一次拉回来内存直接爆掉。pager.Pager 接口（mcube 提供）把"逐页拉取"抽象成两步：

for pager.Next() {                  // 还有下一页吗？
	set := host.NewHostSet()
	if err := pager.Scan(ctx, set); err != nil {   // 拉当前页到 set
		...
	}
	for i := range set.Items {      // 逐台处理
		s.doSyncHost(ctx, set.Items[i], t)
	}
}

厂商侧的实现（以阿里云 ECS 为例，provider/aliyun/ecs/ecs_pager.go）：

func (p *ecsPager) Scan(ctx context.Context, set pager.Set) error {
	// 调阿里云 SDK 拉当前页
	resp, err := p.operator.queryInstance(p.nextReq())
	if err != nil {
		return err
	}
	p.CheckHasNext(resp)            // 根据返回条数判断是否还有下一页
	set.Add(resp.ToAny()...)        // 翻译后的数据塞进 set
	return nil
}

任务层完全不知道阿里云的 PageNumber/PageSize 怎么传——这些细节被封在 nextReq() 里。这就是"云端分页与本地处理解耦"：今天接 AWS（用 token 翻页）、明天接 VMware（根本没分页），任务层代码一行不改。

VMware 特例：vSphere 的 API 没有页的概念，所以走回调式——operator.QueryHost(callback) 每查到一台就回调一次，回调里直接 doSyncHost（apps/task/impl/host.go:81-85）。抽象层允许两种姿势并存。

2.3 对账（Reconcile）：同步的"灵魂步骤"

拉回来数据直接 INSERT 行不行？不行——第二次同步会把重复数据插成两行。对账 = 让库里数据和云端事实一致：

// apps/host/impl/host.go:14-41 —— SyncHost 就是"对账"的落地
func (s *service) SyncHost(ctx context.Context, ins *host.Host) (*host.Host, error) {
	// 按云厂商 ID（Meta.Id）查库里有没有这条
	exist, err := s.DescribeHost(ctx, host.NewDescribeHostRequestWithID(ins.Resource.Meta.Id))
	if err != nil {
		// 查库出错（且不是"不存在"错误）→ 直接失败返回
		if !exception.IsNotFoundError(err) {
			return nil, err
		}
	}

	if exist != nil {                    // 库里已有 → 更新
		exist.Put(host.NewUpdateHostDataByIns(ins))
		if err := s.update(ctx, exist); err != nil {
			return nil, err
		}
		return ins, nil
	}
	// 库里没有 → 新增
	if err := s.save(ctx, ins); err != nil {
		return nil, err
	}
	return ins, nil
}

大白话：先查后写（存在即更新，不存在即插入）——这就是最朴素的 upsert 对账。底稿特别强调一个事实：provider/ 层不含任何写库/对账逻辑（全目录 grep reconcile|Sync|SaveOrUpdate 无匹配），provider 只负责"翻译数据"，对账职责完全在业务层。单一职责不是口号，是真的按层切干净的。

2.4 数据流转总览：一条主机数据的"翻译之旅"

阿里云 DescribeInstances 响应（阿里方言：Status="Running", InstanceChargeType="PostPaid"）
  → provider/aliyun/ecs 的 transferInstance
       状态: "Running" --ALIYUN_STATUS_MAP--> host.STATUS_RUNNING(11)
       付费: "PostPaid" --mapping.PrasePAY_MODE--> resource.PAY_MODE_POST_PAY(1)
       填充 Meta(Id/Domain...)/Spec(Vendor=ALIYUN...)/Status/Cost/Describe
  → *resource.Resource（统一普通话模型，包在 host.Host 里）
  → task.doSyncHost 补 Meta.SyncAt + InjectBaseFromSecret（补凭证信息）
  → host.SyncHost 对账 upsert
  → resource_host 表（host 模块，原生 SQL）
  → 同时 resource 模块的 Put 可写 resource_meta/spec/cost/status（GORM）

2.5 检索链路：标签选择器的花样玩法

检索接口支持一种很实用的标签查询语法（apps/resource/interface.go:202-218）：

GET /cmdb/api/v1/resource/search?tag=env=~prod,stag&tag=team=bigdata

含义：标签 env 的值 LIKE 匹配 prod 或 stag，并且标签 team 精确等于 bigdata。解析器 ParExpr 支持 4 种操作符（interface.go:155-160）：

值还支持 * 通配（只按 key 匹配）与 .* 转百分号（dao.go:52,59 的 strings.ReplaceAll(selector.Key, ".*", "%")）。多值之间的连接关系由 RelationShip() 决定：等值类用 OR、非等值类用 AND（interface.go:220-229）——语义上很讲究：env!=a,b 应该是"既不是 a 也不是 b"（AND），而 env=a,b 是"a 或 b 都行"（OR）。

另外，带标签筛选时 JOIN 方向会从 LEFT 变 RIGHT（dao.go:43-44）：RIGHT JOIN resource_tag tag ON tag.resource_id = m.id——因为此时"有这个标签"是硬条件，必须以 tag 表为基准反查资源，LEFT JOIN 会把没标签的资源也带出来。

2.6 凭证录入链路：三行核心代码

POST /cmdb/api/v1/secrets/  {vendor: "aliyun", api_key: "LTAI...", api_secret: "明文", regions: ["cn-hangzhou"]}
  → CreateSecret handler（apps/secret/api/http.go:40）
  → secret.CreateSecret（apps/secret/impl/secret.go:11）
       ├─ secret.NewSecret(req)          校验（Regions 必填至少一个）
       ├─ ins.Spec.EncryptAPISecret(key) CBC加密+base64+前缀  ← 第5章片段2
       └─ s.db.WithContext(ctx).Create(ins)  GORM 落库

注意 NewSecret 里会生成 Meta.Id = xid.New().String()（apps/secret/interface.go:152-157）——凭证 ID 不是自增、不是 UUID v4，而是 xid（时间有序的短 ID），兼顾唯一性与排序友好。

3 核心代码片段逐行解析

3.1 片段：CreatTask 的"前置检查五连"（apps/task/impl/task.go:26-74 节选）

func (s *service) CreatTask(ctx context.Context, req *task.CreateTaskRequst) (
	*task.Task, error) {
	// ① 构造任务对象（生成 xid 任务 ID）
	t, err := task.NewTaskFromReq(req)
	if err != nil {
		return nil, err
	}

	// ② 取凭证：跨模块调用 secret 服务
	secretIns, err := s.secret.DescribeSecret(ctx, secret.NewDescribeSecretRequest(req.CredentialId))
	if err != nil {
		return nil, err
	}

	secret := secretIns.Spec
	t.UpdateSecretDesc(secret.ShortDesc())     // 任务上记录凭证描述（ShortDesc 带掩码）

	// ③ 地域白名单校验：vSphere（私有云无地域概念）和账单同步除外
	if !(secret.Vendor.Equal(resource.VENDOR_VSPHERE) || req.ResourceType.IsIn(resource.TYPE_BILL)) {
		if req.Region == "" {
			return nil, exception.NewBadRequest("region required")
		}
		if !secret.IsAllowRegion(req.Region) {
			// 凭证只授权了 cn-hangzhou，你却要同步 cn-beijing → 拒绝
			return nil, exception.NewBadRequest("this secret not allow sync region %s", req.Region)
		}
	}

	// ④ 解密凭证（解密结果只在本函数调用链的内存中存在）
	err = secret.DecryptAPISecret(conf.C().App.EncryptKey)
	if err != nil {
		s.log.Warnf("decrypt api secret error, %s", err)
	}

	// ⑤ 异步启动同步（30 分钟硬超时），任务先落库再返回
	syncCtx, _ := context.WithTimeout(context.Background(), time.Minute*30)
	switch req.ResourceType {
	case resource.TYPE_HOST:
		go s.syncHost(syncCtx, secretIns, t, s.SyncTaskCallback)
	case resource.TYPE_BILL:
		go s.syncBill(syncCtx, secretIns, t, s.SyncTaskCallback)
	case resource.TYPE_RDS:
		go s.syncRds(syncCtx, secretIns, t, s.SyncTaskCallback)
	}

	if err = s.insert(ctx, t); err != nil {    // 任务入库（此时状态=进行中）
		return nil, err
	}
	return t, nil
}

值得咀嚼的三个设计：

• 地域白名单在凭证上（regions 字段支持 * 通配，interface.go:86-98 的 IsAllowedRegion），一个 AK 只授权它该管的地域——最小权限原则的落地；
• syncCtx 丢弃了 cancel 函数（_），意味着这个 30 分钟超时只能等它自然到期，无法外部取消——健壮性缺口之一；
• 任务是先 go 后 insert：极端情况下 goroutine 可能先于 insert 执行完（同步很快时），此时 SyncTaskCallback 更新一条还没插入的任务——小概率竞态，真实项目里值得留意。

3.2 片段：syncHost 的"厂商分发 + panic 兜底"（apps/task/impl/host.go:23-113 节选）

func (s *service) syncHost(ctx context.Context, secretIns *secret.Secret,
	t *task.Task, cb SyncTaskCallback) {
	var (
		pager pager.Pager
	)

	t.Run()                        // 任务状态 → RUNNING
	defer func() {
		if err := recover(); err != nil {
			t.Failed(fmt.Sprintf("pannic, %v", err))   // panic 不会丢任务（注：源码拼写 pannic）
		} else {
			t.Completed()
		}
		cb(t)                       // 无论成败，回调更新任务状态
	}()

	secret := secretIns.Spec
	req := provider.NewQueryRequestWithRate(secret.RequestRate)  // 用凭证上的限速值

	switch secret.Vendor {          // 厂商分发：一个 case 一家云
	case resource.VENDOR_ALIYUN:
		op, err := aliyun.NewOperator(secret.ApiKey, secret.ApiSecret, t.Data.Region)
		if err != nil {
			t.Failed(err.Error())
			return
		}
		pager = op.HostOperator().PageQueryHost(req)   // 拿到分页器
	case resource.VENDOR_TENCENT:
		op, err := txyun.NewOperator(secret.ApiKey, secret.ApiSecret, t.Data.Region)
		// ……（同构代码，华为云/AWS 同理）
	case resource.VENDOR_VSPHERE:
		// VMware 特例：回调式采集，不走 pager
		client := vsConn.NewVsphereClient(secret.Address, secret.ApiKey, secret.ApiSecret)
		ec, err := client.VimClient()
		if err != nil {
			t.Failed(err.Error())
			return
		}
		operator := vmOp.NewVMOperator(ec)
		err = operator.QueryHost(func(h *host.Host) {
			h.Resource.Meta.CredentialId = secretIns.Meta.Id   // 回填凭证 ID
			s.doSyncHost(ctx, h, t)                            // 每台机器直接入库
		})
		if err != nil {
			t.Failed(err.Error())
			return
		}
	default:
		t.Failed(fmt.Sprintf("unsuport vendor %s", secret.Vendor))
		return
	}

	// 分页拉取（仅非 VMware 厂商）
	if pager != nil {
		for pager.Next() {
			set := host.NewHostSet()
			if err := pager.Scan(ctx, set); err != nil {
				t.Failed(fmt.Sprintf("sync error, %s", err))
				return
			}
			for i := range set.Items {
				target := set.Items[i]
				// 补充管理信息（domain/namespace/credential_id 等来自凭证）
				InjectBaseFromSecret(target.Resource.Meta, secretIns)
				s.doSyncHost(ctx, target, t)
			}
		}
	}
}

两个细节：

• InjectBaseFromSecret：云厂商只知道机器本身，不知道"这机器归哪个域哪个空间"——这些管理属性来自凭证（谁录入的凭证，同步来的资源就归属谁的 domain/namespace）。多云数据落库前完成"管理信息注入"；
• defer 里的 recover + cb：同步 goroutine 崩了也不会让任务永远停在"进行中"——panic 被捕获并转成任务失败状态。但底稿也指出：recover 只保状态不丢，不会重试，失败后需要人工重新发任务。

3.3 片段：doSyncHost —— 单台机器的"入库 + 记账"

// apps/task/impl/host.go:116-135 —— Host 主机数据入库
func (s *service) doSyncHost(ctx context.Context, ins *host.Host, t *task.Task) {
	// 刷新同步时间戳（doSyncHost 在第4步被调用）
	ins.Resource.Meta.SyncAt = time.Now().Unix()
	// 调 host 模块对账入库（存在即更新，不存在即插入）
	h, err := s.host.SyncHost(ctx, ins)

	// 无论成败，都往 task_record 记一条明细——审计的粒度是"单台机器"
	var detail *task.Record
	if err != nil {
		s.log.Warnf("save host error, %s", err)
		detail = task.NewSyncFailedRecord(t.Id, ins.Resource.Meta.Id,
			ins.Resource.Spec.Name, err.Error())
	} else {
		s.log.Debugf("save host %s to db", h.ShortDesc())
		detail = task.NewSyncSucceedRecord(t.Id, ins.Resource.Meta.Id,
			ins.Resource.Spec.Name)
	}

	t.AddDetail(detail)                        // 内存中的任务计数 +1
	if err := s.insertTaskDetail(ctx, detail); err != nil {
		s.log.Errorf("update detail error, %s", err)
	}
}

失败不中断：某台机器入库失败，只记一条失败明细，循环继续处理下一台——同步 500 台挂 1 台，任务仍算完成（total_failed=1），单点失败不会拖垮整批。这是批量任务处理的基本修养。

4 业务逻辑流程图与核心要点总结

4.1 同步全链路时序图

调用方          task模块         secret模块       provider(aliyun)    host模块      MySQL
  │ CreateTask    │                │                 │                │            │
  ├──────────────→│                │                 │                │            │
  │               │ DescribeSecret │                 │                │            │
  │               ├───────────────→│                 │                │            │
  │               │←─ Secret(密文) ┤                 │                │            │
  │               │ DecryptAPISecret(内存中解密)      │                │            │
  │               │ insert task(进行中) ──────────────────────────────────────────→ │
  │←─ Task(立即返回)│                │                │                │            │
  │               │ go syncHost ────────────────────→│                │            │
  │               │   (goroutine, 30min超时)          │                │            │
  │               │        for pager.Next()          │                │            │
  │               │           pager.Scan ───────────→│ DescribeInstances(第N页)     │
  │               │           ←── []*Host(已归一化) ──┤                │            │
  │               │        InjectBaseFromSecret       │                │            │
  │               │        doSyncHost ────────────────────────────────→│            │
  │               │                                   │    DescribeHost(按云ID查重)→│
  │               │                                   │    update or save ────────→│
  │               │        insert task_record(成败明细) ──────────────────────────→ │
  │               │   defer: t.Completed() + cb(t) 更新任务终态 ───────────────────→ │
  │ (轮询) DescribeTask ─→ 任务状态 + total_succeed/total_failed                    │

4.2 核心要点总结

1. 同步五步：建任务 → 取凭证+校验+解密 → 异步分页拉取 → 逐台对账入库 → 逐台记明细；
2. pager.Pager 抽象让任务层与"云厂商怎么翻页"彻底解耦；VMware 走回调特例；
3. 对账 = 先查后写（按云厂商 ID），provider 层零写库职责；
4. mapping 归一化是"海关"：方言进、普通话出；
5. 管理属性（domain 等）来自凭证注入，不是云厂商给的；
6. 单台失败只记账不中断整批；panic 有 recover 兜底但不重试。

5 自测题（对话式）

三道原书思考题，都是"真要在二开时动手改"的问题。先自己答，再看参考答案。

6 本章小结