diff --git a/cmd/tenant-migration/cmd/root.go b/cmd/tenant-migration/cmd/root.go index d3a3a3559..e48966605 100644 --- a/cmd/tenant-migration/cmd/root.go +++ b/cmd/tenant-migration/cmd/root.go @@ -124,6 +124,8 @@ func applyBizIDsFlag() error { return nil } +var autoConfirmMigrate bool + // migrateCmd represents the migrate command var migrateCmd = &cobra.Command{ Use: "migrate", @@ -132,7 +134,9 @@ var migrateCmd = &cobra.Command{ This command migrates: - MySQL data with tenant_id population -- Vault KV data via API (if configured)`, +- Vault KV data via API (if configured) + +The --biz-ids flag is required to explicitly specify which businesses to migrate.`, Run: func(cmd *cobra.Command, args []string) { if cfg == nil { fmt.Println("Error: configuration not loaded") @@ -144,6 +148,12 @@ This command migrates: os.Exit(1) } + if !cfg.Migration.HasBizFilter() { + fmt.Println("Error: --biz-ids is required for migrate command.") + fmt.Println("Usage: migrate --biz-ids '1001,1002' -c config.yaml") + os.Exit(1) + } + m, err := migrator.NewMigrator(cfg) if err != nil { fmt.Printf("Error creating migrator: %v\n", err) @@ -151,6 +161,27 @@ This command migrates: } defer m.Close() + existingData := m.CheckTargetData() + if len(existingData) > 0 { + fmt.Printf("\nError: Target database already has data for biz_ids %v:\n", cfg.Migration.BizIDs) + for _, item := range existingData { + fmt.Printf(" - %s: %d records\n", item.Table, item.Count) + } + fmt.Println("\nPlease run 'cleanup' first to avoid duplicate data:") + fmt.Printf(" cleanup --biz-ids %s -c %s -f\n", bizIDs, cfgFile) + os.Exit(1) + } + + if !autoConfirmMigrate { + fmt.Printf("About to migrate data for biz_ids %v\n", cfg.Migration.BizIDs) + fmt.Print("Continue? [y/N]: ") + var confirm string + if _, scanErr := fmt.Scanln(&confirm); scanErr != nil || (confirm != "y" && confirm != "Y") { + fmt.Println("Migration canceled.") + return + } + } + report, err := m.Run() if err != nil { fmt.Printf("Error during migration: %v\n", err) @@ -223,6 +254,11 @@ Use --all to scan all tables in the databases.`, os.Exit(1) } + if err := applyBizIDsFlag(); err != nil { + fmt.Printf("Error parsing biz-ids: %v\n", err) + os.Exit(1) + } + m, err := migrator.NewMigrator(cfg) if err != nil { fmt.Printf("Error creating migrator: %v\n", err) @@ -316,7 +352,8 @@ var versionCmd = &cobra.Command{ func init() { // Add migrate flags migrateCmd.Flags().StringVar(&bizIDs, "biz-ids", "", - "Comma-separated list of business IDs to migrate (e.g., '1001,1002,1003'). Overrides config file setting.") + "Comma-separated list of business IDs to migrate (required, e.g., '1001,1002,1003').") + migrateCmd.Flags().BoolVarP(&autoConfirmMigrate, "yes", "y", false, "Skip confirmation prompt") // Add cleanup flags cleanupCmd.Flags().BoolVarP(&forceCleanup, "force", "f", false, "Skip confirmation prompt") @@ -325,4 +362,6 @@ func init() { // Add scan flags scanCmd.Flags().BoolVarP(&scanAllTables, "all", "a", false, "Scan all tables in databases (not just configured tables)") + scanCmd.Flags().StringVar(&bizIDs, "biz-ids", "", + "Comma-separated list of business IDs to scan (e.g., '1001,1002,1003'). Overrides config file setting.") } diff --git "a/cmd/tenant-migration/docs/\345\244\232\347\247\237\346\210\267\350\277\201\347\247\273plan.md" "b/cmd/tenant-migration/docs/\345\244\232\347\247\237\346\210\267\350\277\201\347\247\273plan.md" index 5da78a594..69654aab1 100644 --- "a/cmd/tenant-migration/docs/\345\244\232\347\247\237\346\210\267\350\277\201\347\247\273plan.md" +++ "b/cmd/tenant-migration/docs/\345\244\232\347\247\237\346\210\267\350\277\201\347\247\273plan.md" @@ -4,463 +4,461 @@ ```mermaid flowchart LR - subgraph source [源环境 v2.3.13] + subgraph source [源环境 - 单租户] MySQL_S[(MySQL)] Vault_S[(Vault)] end - subgraph target [目标环境 v2.3.8-multi-tenant.alpha.2] + subgraph tool [迁移工具] + ID_Mapper[ID 映射器] + FK_Rewriter[外键重写] + Special[特殊处理] + end + + subgraph target [目标环境 - 多租户] MySQL_T[(MySQL)] Vault_T[(Vault)] end - MySQL_S -->|"导出 + 填充tenant_id"| MySQL_T - Vault_S -->|"API读取明文 -> API写入"| Vault_T + MySQL_S -->|"分批读取"| tool + tool -->|"新ID + tenant_id + FK重写"| MySQL_T + Vault_S -->|"API读取明文"| tool + tool -->|"映射新路径 + API写入"| Vault_T ``` **迁移范围**： -- MySQL: 核心业务数据，填充 tenant_id 值（目标表结构已包含此字段） -- Vault: KV 配置值数据（必须通过 API 迁移，因为两套环境密钥不同） -- ITSM: 跳过，新环境重新配置 -- 文件存储: 跳过（bkrepo/cos 单独处理） +- MySQL：29 张核心业务表，自动分配新 ID、重写外键引用、填充 `tenant_id` +- Vault：KV 配置值数据，通过 API 迁移（两套环境密钥不同），路径中 `app_id` / `release_id` 自动映射为新 ID +- ITSM：迁移时清除审批工单字段，待审批的策略重置为 `pending_approval`，用户在新环境重新发起审批 +- 文件存储：跳过（bkrepo/cos 单独处理） --- -## 二、MySQL 数据迁移 - -### 1. 表分类说明 - -#### 必须迁移的核心业务表（共29张） - -| 分类 | 表名 | 说明 | - -|------|------|------| - -| 应用 | `applications` | 应用定义，核心数据 | - -| 配置项 | `config_items`, `commits`, `contents` | 配置文件元数据 | - -| 发布 | `releases`, `released_config_items` | 发布版本数据 | - -| 策略 | `strategies`, `strategy_sets`, `current_published_strategies` | 当前生效的发布策略 | - -| KV配置 | `kvs`, `released_kvs` | KV 配置元数据（值在 Vault） | - -| 模板 | `templates`, `template_spaces`, `template_sets`, `template_revisions` | 配置模板 | - -| 模板变量 | `template_variables`, `app_template_bindings`, `app_template_variables`, `released_app_templates`, `released_app_template_variables` | 模板变量绑定 | - -| 分组 | `groups`, `group_app_binds`, `released_groups` | 分组配置 | - -| 钩子 | `hooks`, `hook_revisions`, `released_hooks` | 前置/后置脚本 | - -| 凭证 | `credentials`, `credential_scopes` | 服务密钥凭证 | - -| 业务分片 | `sharding_bizs` | 业务数据库分片配置 | - -#### 可跳过的运行时/历史表（共13张） - -| 表名 | 说明 | 跳过原因 | +## 二、CLI 命令一览 +| 命令 | 功能 | 关键参数 | |------|------|----------| +| `migrate` | 全量迁移（MySQL + Vault） | `--biz-ids`（必填）、`-y` 跳过确认 | +| `cleanup` | 清理目标数据（Vault → MySQL） | `--biz-ids`（可选）、`-f` 跳过确认 | +| `validate` | 验证迁移数据完整性 | — | +| `scan` | 扫描源/目标数据库资产对比 | `--biz-ids`（可选）、`-a` 扫描所有表 | +| `version` | 打印版本信息 | — | -| `events` | 事件通知表 | 用于触发 feed-server 变更通知，系统运行时自动生成 | - -| `current_released_instances` | 客户端当前版本记录 | 客户端重新连接后自动更新 | - -| `resource_locks` | 资源锁 | 运行时临时数据 | +**全局参数**：`-c / --config` 配置文件路径（必填） -| `clients` | 客户端连接信息 | 运行时数据，客户端重连后自动注册 | +```bash +# 编译 +go build -o bk-bscp-tenant-migration ./cmd/tenant-migration/ -| `client_events` | 客户端事件 | 运行时监控数据 | +# 迁移指定业务 +./bk-bscp-tenant-migration -c migration.yaml migrate --biz-ids=100,200,300 -| `client_querys` | 客户端查询记录 | 用户自定义查询，非核心数据 | +# 清理目标数据（迁移失败后重来） +./bk-bscp-tenant-migration -c migration.yaml cleanup --biz-ids=100,200,300 -f -| `audits` | 审计日志 | 历史操作记录，可选迁移 | +# 验证迁移结果 +./bk-bscp-tenant-migration -c migration.yaml validate -| `published_strategy_histories` | 发布策略历史 | 历史记录，可选迁移 | - -| `archived_apps` | 归档应用 | 已删除的应用，通常无需迁移 | - -### 2. 程序化迁移实现 +# 扫描资产对比 +./bk-bscp-tenant-migration -c migration.yaml scan --biz-ids=100,200,300 +./bk-bscp-tenant-migration -c migration.yaml scan --all +``` -使用 Go 程序直接连接源和目标数据库，逐表迁移数据。 +> **注意**：`migrate` 命令**必须**通过 `--biz-ids` 指定业务 ID，不支持全量迁移模式。 -**核心迁移逻辑**： +--- -```go -// 迁移配置 -type MigrateConfig struct { - TargetTenantID string - BatchSize int - SkipTables []string -} +## 三、MySQL 数据迁移 -// 需要迁移的核心表（按依赖顺序） -var coreTables = []string{ - // 基础表（无外键依赖） - "sharding_bizs", - "applications", - "template_spaces", - "groups", - "hooks", - "credentials", - - // 二级表 - "config_items", - "releases", - "strategy_sets", - "template_sets", - "templates", - "template_variables", - "hook_revisions", - "credential_scopes", - "group_app_binds", - - // 三级表 - "commits", - "contents", - "strategies", - "current_published_strategies", - "kvs", - "released_config_items", - "released_groups", - "released_hooks", - "released_kvs", - "template_revisions", - "app_template_bindings", - "app_template_variables", - "released_app_templates", - "released_app_template_variables", -} - -// 跳过的表（运行时/历史数据） -var skipTables = []string{ - "events", - "current_released_instances", - "resource_locks", - "clients", - "client_events", - "client_querys", - "audits", - "published_strategy_histories", - "archived_apps", - "configs", // ITSM配置，新环境重配 -} -``` +### 1. 表分类说明 -**单表迁移示例**： +#### 必须迁移的核心业务表（29 张，按依赖层级） + +| 层级 | 表名 | 说明 | 外键依赖 | +|------|------|------|----------| +| L1（基础表） | `sharding_bizs` | 业务分片配置 | 无 | +| | `applications` | 应用定义 | 无 | +| | `template_spaces` | 模板空间 | 无 | +| | `groups` | 分组 | 无 | +| | `hooks` | 前置/后置脚本 | 无 | +| | `credentials` | 服务密钥凭证 | 无 | +| | `template_variables` | 模板变量 | 无 | +| L2（二级表） | `config_items` | 配置项元数据 | → `applications` | +| | `releases` | 发布版本 | → `applications` | +| | `strategy_sets` | 策略集合 | → `applications` | +| | `templates` | 配置模板 | → `template_spaces` | +| | `template_sets` | 模板套餐 | → `template_spaces`；JSON数组: `template_ids` → `templates`, `bound_apps` → `applications` | +| | `hook_revisions` | 脚本版本 | → `hooks` | +| | `credential_scopes` | 凭证作用域 | → `credentials` | +| | `group_app_binds` | 分组应用绑定 | → `groups`, `applications` | +| L3（三级表） | `contents` | 配置内容 | → `applications`, `config_items`(optional) | +| | `commits` | 配置提交 | → `applications`, `config_items`(optional), `contents` | +| | `strategies` | 发布策略 | → `applications`, `releases`, `strategy_sets` | +| | `current_published_strategies` | 当前生效策略 | → `applications`, `strategies`, `releases`, `strategy_sets` | +| | `kvs` | KV 配置元数据 | → `applications` | +| | `released_config_items` | 已发布配置项 | → `applications`, `releases`, `commits`, `config_items`(optional), `contents` | +| | `released_groups` | 已发布分组 | → `applications`, `groups`, `releases`, `strategies` | +| | `released_hooks` | 已发布脚本 | → `applications`, `releases`, `hooks`, `hook_revisions` | +| | `released_kvs` | 已发布 KV | → `applications`, `releases` | +| | `template_revisions` | 模板版本 | → `template_spaces`, `templates` | +| | `app_template_bindings` | 应用模板绑定 | → `applications`；JSON数组: 多个模板相关 ID | +| | `app_template_variables` | 应用模板变量 | → `applications` | +| | `released_app_templates` | 已发布应用模板 | → `applications`, `releases`, `template_spaces`, `template_sets`, `templates`, `template_revisions` | +| | `released_app_template_variables` | 已发布模板变量 | → `applications`, `releases` | + +#### 跳过的运行时/历史表（13 张） + +| 表名 | 跳过原因 | +|------|----------| +| `events` | 事件通知，系统运行时自动生成 | +| `current_released_instances` | 客户端当前版本记录，重连后自动更新 | +| `resource_locks` | 资源锁，运行时临时数据 | +| `clients` | 客户端连接信息，重连后自动注册 | +| `client_events` | 客户端事件，运行时监控数据 | +| `client_querys` | 客户端查询记录，非核心数据 | +| `audits` | 审计日志，历史记录（可选迁移） | +| `published_strategy_histories` | 发布策略历史（可选迁移） | +| `archived_apps` | 归档应用，已删除的应用 | +| `configs` | ITSM 配置，新环境需重新注册 | +| `schema_migrations` | 数据库迁移记录表 | +| `biz_hosts` | 业务主机关系表，定时同步 | +| `sharding_dbs` | 数据库分片配置 | + +### 2. 核心迁移机制 + +#### 2.1 ID 映射与外键重写 + +迁移的核心难点在于：目标环境的 `id_generators` 自增序列与源环境不同，因此每条记录在目标库中会获得**新的主键 ID**。所有外键引用必须同步更新为新 ID。 -```go -func migrateTable(srcDB, tgtDB *gorm.DB, tableName, tenantID string, batchSize int) error { - // 1. 关闭目标库外键检查 - tgtDB.Exec("SET FOREIGN_KEY_CHECKS = 0") - defer tgtDB.Exec("SET FOREIGN_KEY_CHECKS = 1") +```mermaid +flowchart LR + subgraph 源库 + A1["applications id=100"] + C1["config_items id=50, app_id=100"] + end - // 2. 获取源表总记录数 - var total int64 - srcDB.Table(tableName).Count(&total) + subgraph ID映射器 + M["100 → 2001 (applications)"] + end - // 3. 分批读取并写入 - for offset := 0; offset < int(total); offset += batchSize { - var rows []map[string]interface{} - srcDB.Table(tableName).Offset(offset).Limit(batchSize).Find(&rows) - - for _, row := range rows { - // 填充 tenant_id（目标表已有此字段） - row["tenant_id"] = tenantID - - // strategies 表特殊处理：itsm_ticket_state_id int -> string - if tableName == "strategies" { - if stateID, ok := row["itsm_ticket_state_id"]; ok && stateID != nil { - row["itsm_ticket_state_id"] = fmt.Sprintf("%v", stateID) - } - } - - // 写入目标库 - if err := tgtDB.Table(tableName).Create(row).Error; err != nil { - return fmt.Errorf("insert %s failed: %w", tableName, err) - } - } - - log.Printf("Table %s: %d/%d migrated", tableName, offset+len(rows), total) - } + subgraph 目标库 + A2["applications id=2001"] + C2["config_items id=3001, app_id=2001"] + end - return nil -} + A1 --> M + M --> A2 + C1 -.->|"app_id: 100→2001"| C2 ``` -**主迁移流程**： +**实现细节**： -```go -func RunMigration(cfg *MigrateConfig) error { - // 1. 连接源和目标数据库 - srcDB := connectDB(cfg.Source) - tgtDB := connectDB(cfg.Target) - - // 2. 按顺序迁移核心表 - for _, table := range coreTables { - if contains(cfg.SkipTables, table) { - log.Printf("Skip table: %s", table) - continue - } - - log.Printf("Migrating table: %s", table) - if err := migrateTable(srcDB, tgtDB, table, cfg.TargetTenantID, cfg.BatchSize); err != nil { - return err - } - } - - // 3. 更新 id_generators - if err := updateIDGenerators(srcDB, tgtDB); err != nil { - return err - } - - return nil -} - -// 更新目标库的 id_generators -func updateIDGenerators(srcDB, tgtDB *gorm.DB) error { - var generators []struct { - Resource string - MaxID uint64 - } - srcDB.Table("id_generators").Find(&generators) - - for _, g := range generators { - tgtDB.Table("id_generators"). - Where("resource = ?", g.Resource). - Update("max_id", g.MaxID) - } - return nil -} -``` +- 每行记录通过 `UPDATE id_generators SET max_id = max_id + 1 WHERE resource = ?` 获取新 ID +- `IDMapper` 记录所有 `(表名, 源ID) → 新ID` 的映射关系 +- 外键转换时，从 `IDMapper` 查找引用表的新 ID +- 支持三种外键类型： + - **普通外键**（`ForeignKeys`）：直接列值映射，源 ID 为 0 时跳过 + - **可选外键**（`OptionalFKs`）：映射不到时自动分配虚拟 ID（不插入父表记录），适用于可能已删除的引用（如 `config_item_id`） + - **JSON 数组外键**（`JSONArrayFKs`）：解析 JSON 数组中的每个 ID 并替换（如 `template_sets.template_ids`） +- 特殊处理 `app_template_bindings.bindings` 嵌套 JSON 结构中的多层 ID 映射 -**特殊处理说明**： +#### 2.2 Biz_id 过滤 -| 表名 | 特殊处理 | +- 有 `biz_id` 列的表：`WHERE biz_id IN (?)` 只读取/清理指定业务的数据 +- 无 `biz_id` 列的表（如 `sharding_bizs`）：迁移时**全量复制** -| --------------- | --------------------------------------------- | +#### 2.3 特殊处理 -| `strategies` | `itsm_ticket_state_id` 字段从 int 转为 string | +| 表名 | 特殊处理 | 说明 | +|------|----------|------| +| `strategies` | `itsm_ticket_state_id` int → string | 目标库字段类型变更 | +| `strategies` | 清除 ITSM 审批字段 | 跨环境审批单不互通，详见下文 | +| `kvs` / `released_kvs` | `version` 重置为 1 | Vault KV v2 的 Put 操作会重置版本号 | +| 所有表 | 填充 `tenant_id` | 目标表已有此字段，填充配置的 `target_tenant_id` | +| `app_template_bindings` | `bindings` JSON 嵌套 ID 重写 | 包含 template_set_id, template_revision_id 等多层引用 | -| 所有表 | 填充 `tenant_id` 字段值（目标表已有此字段） | +#### 2.4 ITSM 审批字段处理 -| `id_generators` | 迁移后更新各资源的 max_id | +跨环境迁移时，源环境的 ITSM 审批工单无法在目标环境使用： -### 3. 迁移顺序说明 +- 工单 SN 属于源 ITSM 实例，在目标 ITSM 中不存在 +- 审批回调 URL 指向源环境的 BSCP 网关 +- 目标环境的 ITSM 工作流/服务/状态 ID 配置不同 -由于表之间存在外键依赖关系，需要按依赖顺序迁移： +**迁移策略**： -``` -第一批（基础表）: sharding_bizs -> applications -> template_spaces -> groups -> hooks -> credentials - ↓ -第二批（二级表）: config_items, releases, strategy_sets, template_sets, templates, ... - ↓ -第三批（依赖表）: commits, contents, strategies, kvs, released_*, ... -``` +| 原始状态 | 迁移后处理 | +|----------|-----------| +| `pending_approval` / `pending_publish` | 重置为 `revoked_publish`（已撤销），清除 ITSM 字段和 `approver_progress` | +| `already_publish` / `rejected_approval` 等 | **不做任何处理**，保留全部字段原值 | -### 4. 不迁移的表 +> **为什么只处理待审批/待上线的策略？** +> - 已完成的策略（`already_publish` 等）：页面不会展示 ITSM 审批信息（`approveStatus = -1`，组件不渲染），保留原始记录无害且保持审计可追溯性。 +> +> **为什么待审批的改为 `revoked_publish` 而非保留 `pending_approval`？** +> 1. 保持 `pending_approval` 会**阻塞该应用的新上线操作**（代码检查到有待审批策略时拒绝新提交） +> 2. 页面会常驻显示"待审批"旋转图标，但没有审批单链接，用户无法操作 +> 3. `revoked_publish` 不会阻塞新发布，且在页面刷新后自动消失，用户可以正常重新提交上线 -| 表名 | 原因 | +#### 2.5 错误处理 -| ----------- | ------------------------- | +- **`continue_on_error: false`（默认）**：遇到第一个失败记录即停止当前表的迁移 +- **`continue_on_error: true`**：跳过失败记录继续迁移，累计错误计数 +- 失败记录详情写入日志文件 `logs/biz_/migrate_.json`，按 `biz_id` 分组 +- 迁移报告中展示每张表的源数据量、已迁移量、失败量和状态 -| `biz_hosts` | 多租户版本已删除此表 | +### 3. 迁移顺序 -| `configs` | ITSM 配置，新环境重新注册 | +迁移工具通过 `TablesInInsertOrder()` 严格按依赖顺序迁移，确保外键引用的目标记录已存在： -| 运行时表 | 见上方 skipTables 列表 | +``` +第一批（基础表）: sharding_bizs → applications → template_spaces → groups → hooks → credentials → template_variables + ↓ +第二批（二级表）: config_items → releases → strategy_sets → templates → template_sets → hook_revisions → credential_scopes → group_app_binds + ↓ +第三批（依赖表）: contents → commits → strategies → current_published_strategies → kvs → released_* → template_revisions → app_template_* → released_app_* +``` + +清理时使用 `TablesInCleanupOrder()` 按**反向依赖顺序**删除，先删三级表再删基础表。 --- -## 三、Vault KV 数据迁移 +## 四、Vault KV 数据迁移 ### 1. 关键约束 -**重要**：两套环境的 Vault 使用不同的加密密钥（unseal key），因此： +两套环境的 Vault 使用不同的加密密钥（unseal key），因此： -- 不能直接复制 Vault 底层存储数据 +- **不能**直接复制 Vault 底层存储数据 - 必须通过 Vault API 读取解密后的明文数据 - 再通过 API 写入目标 Vault（目标 Vault 用自己的密钥重新加密） -### 2. 存储结构 +### 2. 存储路径与 ID 映射 -Vault 中 KV 数据路径（两个版本完全相同）： +Vault KV 路径中包含 `app_id` 和 `release_id`，迁移时需使用 MySQL 阶段生成的 ID 映射进行路径重写： -- 未发布 KV: `bk_bscp/biz/{biz_id}/apps/{app_id}/kvs/{key}` -- 已发布 KV: `bk_bscp/biz/{biz_id}/apps/{app_id}/releases/{release_id}/kvs/{key}` +``` +源路径: bk_bscp/biz/{biz_id}/apps/{源app_id}/kvs/{key} +目标路径: bk_bscp/biz/{biz_id}/apps/{新app_id}/kvs/{key} -路径中不包含 tenant_id，两套环境使用各自的 Vault 实例，路径不会冲突。 +源路径: bk_bscp/biz/{biz_id}/apps/{源app_id}/releases/{源release_id}/kvs/{key} +目标路径: bk_bscp/biz/{biz_id}/apps/{新app_id}/releases/{新release_id}/kvs/{key} +``` -### 3. 迁移方式（必须通过 API） +### 3. 迁移流程 ```mermaid sequenceDiagram participant Tool as 迁移工具 + participant IDMap as ID映射器 + participant SrcDB as 源MySQL participant SrcVault as 源Vault participant TgtVault as 目标Vault - participant MySQL as MySQL kvs表 - Tool->>MySQL: 1. 查询所有 KV 记录 + Note over Tool: MySQL 迁移完成后 + Tool->>IDMap: 获取 app_id / release_id 映射表 + Tool->>SrcDB: 1. 分批查询 kvs 记录（按 biz_id 过滤） loop 遍历每条 KV - Tool->>SrcVault: 2. API 读取 (自动解密) - SrcVault-->>Tool: 返回明文 value - Tool->>TgtVault: 3. API 写入 (自动加密) + Tool->>SrcVault: 2. GetVersion(源路径, version) + SrcVault-->>Tool: 返回明文 {kv_type, value} + Tool->>IDMap: 3. 映射 app_id → 新 app_id + Tool->>TgtVault: 4. Put(新路径, {kv_type, value}) + end + Tool->>SrcDB: 5. 分批查询 released_kvs 记录 + loop 遍历每条 Released KV + Tool->>SrcVault: 6. GetVersion(源路径, version) + SrcVault-->>Tool: 返回明文 + Tool->>IDMap: 7. 映射 app_id, release_id + Tool->>TgtVault: 8. Put(新路径, {kv_type, value}) end ``` -**迁移脚本示例**： +### 4. 注意事项 -```go -// 基于 MySQL kvs/released_kvs 表中的记录遍历 -func migrateVaultKV(srcVault, tgtVault *vault.Client, mysqlDB *gorm.DB) error { - // 1. 迁移未发布的 KV (kvs 表) - var kvRecords []table.Kv - mysqlDB.Find(&kvRecords) - - for _, kv := range kvRecords { - // 从源 Vault 读取（API 自动解密） - path := fmt.Sprintf("biz/%d/apps/%d/kvs/%s", kv.BizID, kv.AppID, kv.Key) - secret, _ := srcVault.KVv2("bk_bscp").GetVersion(ctx, path, kv.Version) - - value := secret.Data["value"].(string) - kvType := secret.Data["kv_type"].(string) - - // 写入目标 Vault（API 自动加密） - tgtVault.KVv2("bk_bscp").Put(ctx, path, map[string]interface{}{ - "kv_type": kvType, - "value": value, - }) - } - - // 2. 迁移已发布的 KV (released_kvs 表) - var releasedKvs []table.ReleasedKv - mysqlDB.Find(&releasedKvs) - - for _, rkv := range releasedKvs { - path := fmt.Sprintf("biz/%d/apps/%d/releases/%d/kvs/%s", - rkv.BizID, rkv.AppID, rkv.ReleaseID, rkv.Key) - // ... 同上逻辑 - } - - return nil -} -``` +- **版本重置**：Vault KV v2 的 `Put` 操作会重置版本为 1，因此 MySQL 中 `kvs.version` 和 `released_kvs.version` 也同步重置为 1 +- **超时控制**：Vault 迁移设置 30 分钟超时（context） +- **错误处理**：`continue_on_error` 配置同时适用于 Vault 迁移 +- **签名校验**：`kvs` 表中的 `signature` 字段可用于验证迁移后数据完整性 -### 4. Vault 迁移注意事项 +--- -1. **版本处理**：Vault KVv2 支持版本，但迁移时只需迁移 MySQL 中记录的版本 -2. **批量处理**：建议分批处理，避免一次性加载过多数据 -3. **错误处理**：记录失败的 KV，支持断点续传 -4. **签名校验**：MySQL `kvs` 表中的 `signature` 字段可用于验证迁移后数据完整性 +## 五、清理机制 -### 5. 数据一致性验证 +清理（cleanup）用于迁移失败后重新执行前清除目标数据。 -```sql --- 检查 MySQL 中 KV 记录数 -SELECT COUNT(*) FROM kvs; -SELECT COUNT(*) FROM released_kvs; +**执行顺序**：Vault 先于 MySQL，因为 Vault 清理需要读取目标 MySQL 中的记录来构建正确的 Vault 路径。 --- 迁移后验证：遍历 MySQL 记录，确认每条在目标 Vault 中存在 +```mermaid +sequenceDiagram + participant Tool as 迁移工具 + participant TgtDB as 目标MySQL + participant TgtVault as 目标Vault + + rect rgb(255, 230, 230) + Note over Tool,TgtVault: Step 1: Vault 清理（先执行） + Tool->>TgtDB: 查询 kvs/released_kvs 记录（获取 app_id, release_id） + Tool->>TgtVault: DeleteMetadata 逐条删除 + end + + rect rgb(230, 230, 255) + Note over Tool,TgtDB: Step 2: MySQL 清理（后执行） + Note over Tool: 按反向依赖顺序删除 29 张表 + Tool->>TgtDB: 有 biz_id 过滤: DELETE WHERE biz_id IN (...) + Tool->>TgtDB: 无 biz_id 过滤: TRUNCATE / DELETE + end ``` --- -## 四、迁移工具架构 +## 六、验证与扫描 + +### 1. 验证（validate） + +对 29 张核心表逐一检查： + +| 检查项 | 说明 | +|--------|------| +| 记录数对比 | 源库 vs 目标库记录数（按 biz_id 过滤） | +| tenant_id 完整性 | 检查目标库中 tenant_id 为 NULL 或空的记录 | +| tenant_id 正确性 | 检查 tenant_id 不等于配置值的记录 | + +### 2. 扫描（scan） -### 核心模块 +资产扫描用于了解迁移前后的数据分布： + +- **默认模式**：只扫描配置的迁移表（`tables` 列表） +- **`--all` 模式**：扫描源/目标数据库中的所有表 +- 输出源/目标库各表记录数及差异对比 +- 如配置了 Vault，额外扫描 Vault KV 数据（与 MySQL 记录交叉验证存在性） + +--- + +## 七、迁移工具架构 ```mermaid flowchart TB - subgraph migrator [迁移工具] - config[配置管理器] - mysql_migrator[MySQL迁移器] - vault_migrator[Vault迁移器] - id_updater[ID生成器更新] - validator[数据验证器] - reporter[迁移报告] + subgraph CLI [命令行入口 cmd/root.go] + migrate_cmd[migrate] + cleanup_cmd[cleanup] + validate_cmd[validate] + scan_cmd[scan] end - config --> mysql_migrator - config --> vault_migrator - mysql_migrator --> id_updater - mysql_migrator --> validator - vault_migrator --> validator - validator --> reporter -``` - -### 程序入口 - -```go -func main() { - // 1. 加载配置 - cfg := loadConfig("migrate.yaml") - - // 2. 创建迁移器 - migrator := NewMigrator(cfg) - defer migrator.Close() + subgraph config [配置管理 config/] + cfg[Config] + table_list[CoreTables / SkipTables] + end - // 3. 执行迁移（MySQL + Vault，Vault 未配置时自动跳过） - report, err := migrator.Run() - if err != nil { - log.Fatalf("Migration failed: %v", err) - } + subgraph migrator_pkg [迁移器 migrator/] + orchestrator[Migrator 编排器] + mysql_m[MySQLMigrator] + vault_m[VaultMigrator] + id_mapper[IDMapper] + table_meta[TableMeta 外键元数据] + validator[Validator 验证器] + scanner[Scanner 扫描器] + end - // 4. 输出迁移报告 - migrator.PrintReport(report) -} + CLI --> cfg + cfg --> orchestrator + orchestrator --> mysql_m + orchestrator --> vault_m + orchestrator --> validator + orchestrator --> scanner + mysql_m --> id_mapper + mysql_m --> table_meta + id_mapper -.->|"SetIDMapper"| vault_m ``` -### 配置文件示例 +### 核心模块说明 + +| 模块 | 文件 | 职责 | +|------|------|------| +| 编排器 | `migrator.go` | 协调 MySQL → Vault 迁移流程，汇总报告 | +| MySQL 迁移器 | `mysql.go` | 分批读取、ID 分配、FK 重写、特殊处理、写入目标库 | +| Vault 迁移器 | `vault.go` | API 读源写目标，路径中 ID 映射 | +| 表元数据 | `table_meta.go` | 29 张表的 FK 关系、插入/清理顺序定义 | +| ID 映射器 | `id_mapper.go` | 线程安全的 `(表名, 源ID) → 新ID` 映射 | +| 验证器 | `validator.go` | 迁移后记录数、tenant_id 完整性检查 | +| 扫描器 | `scanner.go` | 源/目标资产对比，Vault 交叉验证 | + +--- + +## 八、配置文件说明 ```yaml migration: - target_tenant_id: "your_tenant_id" # 目标租户ID + # 目标租户 ID，将填充到所有迁移记录的 tenant_id 字段 + target_tenant_id: "10001" + + # 要迁移的业务 ID（配置文件设置，可被 --biz-ids 命令行参数覆盖） + # biz_ids: + # - 100 + # - 200 + + # 每批处理的记录数（默认 1000） + batch_size: 1000 + # 遇错继续（默认 false，遇到失败记录即停止） + continue_on_error: false + +# 源环境配置 source: mysql: - host: "source-mysql-host" - port: 3306 + endpoints: + - "source-mysql:3306" database: "bk_bscp" - user: "xxx" + user: "root" password: "xxx" vault: address: "http://source-vault:8200" token: "xxx" +# 目标环境配置 target: mysql: - host: "target-mysql-host" - port: 3306 + endpoints: + - "target-mysql:3306" database: "bk_bscp" - user: "xxx" + user: "root" password: "xxx" vault: address: "http://target-vault:8200" token: "xxx" -options: - batch_size: 1000 # 批量处理大小 - -# 跳过的表（运行时/历史数据） +# 跳过的表（运行时/历史数据，使用默认列表即可） skip_tables: - - events # 事件通知，系统自动生成 - - current_released_instances # 客户端当前版本，自动更新 - - resource_locks # 资源锁，运行时数据 - - clients # 客户端连接信息 - - client_events # 客户端事件 - - client_querys # 客户端查询记录 - - audits # 审计日志（可选迁移） - - published_strategy_histories # 发布历史（可选迁移） - - archived_apps # 归档应用 - - configs # ITSM配置，新环境重新配置 + - events + - current_released_instances + - resource_locks + - clients + - client_events + - client_querys + - audits + - published_strategy_histories + - archived_apps + - configs # ITSM 配置，新环境重新注册 + - schema_migrations + - biz_hosts + - sharding_dbs ``` +| 配置项 | 必填 | 默认值 | 说明 | +|--------|------|--------|------| +| `migration.target_tenant_id` | 是 | — | 目标租户 ID | +| `migration.biz_ids` | 否 | 空（需 `--biz-ids`） | 要迁移的业务 ID 列表 | +| `migration.batch_size` | 否 | 1000 | 每批处理记录数 | +| `migration.continue_on_error` | 否 | false | 遇错是否继续 | +| `source.mysql.*` | 是 | — | 源 MySQL 连接配置 | +| `source.vault.*` | 否 | — | 源 Vault 配置（迁移 KV 时必填） | +| `target.mysql.*` | 是 | — | 目标 MySQL 连接配置 | +| `target.vault.*` | 否 | — | 目标 Vault 配置（迁移 KV 时必填） | +| `skip_tables` | 否 | 13 张默认表 | 跳过的表列表 | + --- -## 五、迁移执行流程 +## 九、迁移执行流程 ```mermaid sequenceDiagram @@ -471,151 +469,133 @@ sequenceDiagram participant SrcVault as 源Vault participant TgtVault as 目标Vault - Op->>Tool: 1. 配置迁移参数(tenant_id等) - + Op->>Tool: 1. 配置 migration.yaml + Op->>Tool: 2. scan 扫描源/目标数据现状 + rect rgb(200, 220, 250) - Note over Tool,TgtMySQL: MySQL 迁移 - loop 遍历核心表 - Tool->>SrcMySQL: 2. 分批读取数据 - Tool->>Tool: 3. 填充tenant_id + 类型转换 - Tool->>TgtMySQL: 4. 批量写入 + Note over Tool,TgtMySQL: MySQL 迁移（migrate 命令） + Tool->>TgtMySQL: 检查目标库是否已有数据（CheckTargetData） + Tool->>TgtMySQL: SET FOREIGN_KEY_CHECKS = 0 + loop 按依赖顺序遍历 29 张表 + Tool->>SrcMySQL: 3. 分批读取（ORDER BY id, biz_id 过滤） + Tool->>Tool: 4. 分配新 ID → IDMapper 记录映射 + Tool->>Tool: 5. 重写外键（普通FK / JSON数组FK / 嵌套JSON） + Tool->>Tool: 6. 填充 tenant_id + 特殊处理 + Tool->>TgtMySQL: 7. 逐行写入 end - Tool->>TgtMySQL: 5. 更新id_generators + Tool->>TgtMySQL: SET FOREIGN_KEY_CHECKS = 1 end - + rect rgb(220, 250, 200) Note over Tool,TgtVault: Vault 迁移 - Tool->>SrcMySQL: 6. 查询kvs/released_kvs记录 - loop 遍历KV记录 - Tool->>SrcVault: 7. API读取(解密) - Tool->>TgtVault: 8. API写入(加密) + Tool->>Tool: 8. 从 MySQL 迁移获取 IDMapper + Tool->>SrcMySQL: 9. 分批查询 kvs/released_kvs 记录 + loop 遍历 KV 记录 + Tool->>SrcVault: 10. API 读取（GetVersion，自动解密） + Tool->>Tool: 11. 映射 app_id/release_id 为新 ID + Tool->>TgtVault: 12. API 写入（Put，自动加密） end end - - Tool->>Tool: 9. 数据验证 - Tool->>Op: 10. 输出迁移报告 -``` - -### 执行步骤 - -1. **准备阶段** - - - 停止源环境写入（或选择合适的迁移时间窗口） - - 备份目标环境数据库 - - 配置迁移参数（目标租户ID、数据库连接、Vault地址等） - -2. **MySQL 迁移**（程序自动执行） - - - 按依赖顺序遍历核心表 - - 分批读取源数据 - - 填充 tenant_id 值 + 类型转换 - - 批量写入目标数据库 - - 更新 id_generators 表 - -3. **Vault 迁移**（程序自动执行） - - - 从 MySQL 查询 kvs/released_kvs 记录 - - 根据记录从源 Vault API 读取（自动解密） - - 写入目标 Vault API（自动加密） - -4. **验证阶段** - - - 核对各表记录数 - - 抽样验证关键数据 - - 验证 Vault KV 数据完整性 -5. **ITSM 配置**（迁移完成后手动执行） + Tool->>Op: 13. 输出迁移报告 + 失败记录日志 - - 在目标环境执行 ITSM v4 服务注册 - - 配置审批流程 + Op->>Tool: 14. validate 验证迁移数据完整性 + Op->>Tool: 15. scan 扫描确认数据一致 +``` ---- +### 执行步骤 -## 六、BkRepo 文件存储分析 +#### 1. 准备阶段 -### 1. 存储路径结构 +- 停止源环境写入（或选择低峰期迁移窗口） +- 备份目标环境数据库 +- 确认迁移机器可访问源/目标的 MySQL 和 Vault +- 确认 MySQL 用户有读写权限，Vault Token 有 KV 读写权限 +- 准备配置文件 `migration.yaml` -BkRepo 中文件存储路径格式： +#### 2. 扫描（可选） -``` -/generic/{project}/bscp-{version}-{biz_id}/file/{sha256} +```bash +./bk-bscp-tenant-migration -c migration.yaml scan --biz-ids=100,200 ``` -**关键点**： +了解源/目标数据分布，确认迁移范围。 -- 文件按 **sha256 签名** 存储（内容寻址） -- 路径中 **不包含 tenant_id** -- 同一文件（相同sha256）只存储一份 +#### 3. 执行迁移 -### 2. 是否需要迁移文件？ +```bash +./bk-bscp-tenant-migration -c migration.yaml migrate --biz-ids=100,200 +``` -| 场景 | 是否需要迁移 | +工具自动执行： +- 检查目标库是否已有该业务数据（有则要求先 cleanup） +- 交互式确认（`-y` 跳过） +- MySQL 分表迁移（ID 映射 + FK 重写 + 特殊处理） +- Vault KV 迁移（API 方式，路径中 ID 自动映射） +- 输出迁移报告（各表数量、成功/失败状态） +- 失败记录写入日志文件 -| ------------------------------------------ | ---------------------------------- | +#### 4. 验证 -| 两套环境共用同一个 BkRepo，且 project 相同 | **无需迁移**，路径相同可直接访问 | +```bash +./bk-bscp-tenant-migration -c migration.yaml validate +``` -| 两套环境使用不同的 BkRepo 实例 | **需要迁移**文件内容 | +自动对比 29 张核心表的记录数和 tenant_id 完整性。 -| 同一 BkRepo 但 project 不同 | **需要迁移**或复制文件到新 project | +#### 5. ITSM 配置（迁移完成后手动执行） -**推荐方案**：如果可以，让两套环境共用同一个 BkRepo（相同 project），这样文件无需迁移。 +- 在目标环境执行 ITSM v4 服务注册 +- 配置审批流程（`configs` 表中的 ITSM workflow/service/state ID） +- 用户对待审批的策略重新发起审批 --- -## 七、不迁移表的影响分析 - -### 各表影响说明 - -| 表名 | 作用 | 不迁移影响 | 风险等级 | - -| ------------------------------ | ------------------------- | -------------------------------- | -------------- | - -| `events` | 触发 feed-server 变更通知 | 新环境无历史事件，但不影响新发布 | 低 | - -| `current_released_instances` | 记录客户端当前拉取的版本 | 客户端重连后自动更新，无影响 | 无 | - -| `clients` | 客户端连接信息 | 客户端重连后自动注册 | 无 | - -| `client_events` | 客户端事件监控 | 丢失历史监控数据 | 低 | - -| `client_querys` | 用户自定义查询 | 需要在新环境重新创建 | 低 | - -| `resource_locks` | 资源锁 | 运行时临时数据，无影响 | 无 | - -| `audits` | 审计日志 | 丢失历史操作记录 | 中（可选迁移） | +## 十、不迁移表的影响分析 -| `published_strategy_histories` | 发布策略历史 | 丢失历史发布记录 | 低（可选迁移） | +| 表名 | 不迁移影响 | 风险等级 | +|------|-----------|---------| +| `events` | 无历史事件，不影响新发布 | 低 | +| `current_released_instances` | 客户端重连后自动更新 | 无 | +| `clients` | 客户端重连后自动注册 | 无 | +| `client_events` | 丢失历史监控数据 | 低 | +| `client_querys` | 需要在新环境重新创建 | 低 | +| `resource_locks` | 运行时临时数据 | 无 | +| `audits` | 丢失历史操作记录 | 中（可选迁移） | +| `published_strategy_histories` | 丢失历史发布记录 | 低（可选迁移） | +| `configs` | ITSM 配置需重新注册 | 中（必须手动配置） | -### 结论 - -上述表不迁移 **不会影响核心业务功能**，客户端可以正常拉取配置。 +**结论**：上述表不迁移**不会影响核心业务功能**，客户端可以正常拉取配置。 --- -## 八、双环境并行运行风险分析 - -### 场景说明 - -迁移后两套环境会同时运行，客户端需要切换 feed-server 连接地址。 +## 十一、BkRepo 文件存储分析 -### 风险点及应对 +### 存储路径结构 -| 风险 | 说明 | 应对措施 | +``` +/generic/{project}/bscp-{version}-{biz_id}/file/{sha256} +``` -| ----------------------------------- | ------------------------------------------- | -------------------------------------------- | +- 文件按 **sha256 签名** 存储（内容寻址） +- 路径中**不包含 tenant_id** +- 同一文件（相同 sha256）只存储一份 -| **配置版本不一致** | 切换期间源环境有新发布 | 迁移时选择停服窗口，或迁移后在新环境同步发布 | +### 是否需要迁移 -| **客户端版本兼容** | v2.3.13 与 v2.3.8-multi-tenant SDK 是否兼容 | 测试验证客户端 SDK 版本兼容性 | +| 场景 | 是否需要迁移 | +|------|-------------| +| 两套环境共用同一个 BkRepo，且 project 相同 | **无需迁移** | +| 两套环境使用不同的 BkRepo 实例 | **需要迁移**文件内容 | +| 同一 BkRepo 但 project 不同 | **需要迁移**或复制文件到新 project | -| **切换过程配置拉取** | 客户端切换瞬间可能拉取失败 | 客户端有重试机制，短暂失败可自动恢复 | +**推荐**：让两套环境共用同一个 BkRepo（相同 project），这样文件无需迁移。 -| **回滚困难** | 切换后发现问题需要回滚 | 保留源环境一段时间，确认稳定后再下线 | +--- -| **current_released_instances 丢失** | 新环境无客户端版本记录 | 客户端重连后自动更新，仅影响首次查询统计 | +## 十二、双环境并行运行 -### 切换流程建议 +### 切换流程 ```mermaid sequenceDiagram @@ -625,67 +605,44 @@ sequenceDiagram participant Admin as 运维 Admin->>Admin: 1. 选择低峰期执行迁移 - Admin->>OldFeed: 2. 停止源环境写入(可选) - Admin->>Admin: 3. 执行数据迁移 - Admin->>Admin: 4. 验证迁移数据 - Admin->>Admin: 5. 更新客户端配置(feed地址) + Admin->>OldFeed: 2. 停止源环境写入（可选） + Admin->>Admin: 3. 执行数据迁移 + 验证 + Admin->>Admin: 4. 配置 ITSM / 审批流程 + Admin->>Admin: 5. 更新客户端配置（feed 地址） Client->>NewFeed: 6. 客户端重连新环境 - NewFeed-->>Client: 7. 返回配置(基于released_groups匹配) + NewFeed-->>Client: 7. 返回配置 Admin->>Admin: 8. 观察一段时间 Admin->>OldFeed: 9. 确认稳定后下线源环境 ``` -### 关键验证项 - -1. **迁移前**： - - - 确认源环境数据备份完成 - - 验证目标环境数据库连通性 - - 确认 BkRepo/Vault 访问正常 - -2. **迁移后/切换前**： - - - 对比核心表记录数 - - 抽样验证应用/配置项数据 - - 在新环境测试客户端拉取配置 - -3. **切换后**： - - - 监控客户端连接数恢复情况 - - 检查配置拉取成功率 - - 验证新发布流程正常 - ---- - -## 九、注意事项 +### 风险与应对 -1. **Vault 密钥问题**：两套环境 Vault 密钥不同，必须通过 API 迁移（读取解密 -> 写入加密），不能直接复制底层存储 -2. **BkRepo 共用**：建议两套环境共用同一个 BkRepo，避免文件迁移 -3. **ID 冲突**：如果目标环境已有数据，需要处理 ID 冲突或清空目标表 -4. **外键约束**：导入时需临时关闭外键检查 -5. **signature 字段**：`kvs` 和 `released_kvs` 表中的 signature 字段用于校验，迁移后应保持一致 -6. **id_generators 表**：迁移后需要更新 `id_generators` 表中各资源的 `max_id` 值 -7. **运行时表**：`events`, `current_released_instances`, `resource_locks`, `clients` 等表无需迁移，系统运行时自动生成 -8. **ITSM 配置**：迁移完成后需要在新环境重新注册 ITSM v4 服务 -9. **并行运行**：保留源环境一段时间作为回滚备份，确认新环境稳定后再下线 +| 风险 | 应对措施 | +|------|---------| +| 切换期间源环境有新发布 | 迁移时选择停服窗口，或迁移后在新环境同步发布 | +| 客户端 SDK 版本兼容性 | 提前测试验证客户端 SDK 版本兼容性 | +| 客户端切换瞬间拉取失败 | 客户端有重试机制，短暂失败可自动恢复 | +| 回滚困难 | 保留源环境一段时间，确认稳定后再下线 | --- -## 十、迁移后验证清单 +## 十三、迁移后验证清单 ### 数据验证 -- [ ] MySQL 核心表记录数一致 -- [ ] Vault KV 数据完整（对比 MySQL kvs/released_kvs 记录） -- [ ] id_generators 表 max_id 已更新 +- [ ] `validate` 命令通过（29 张核心表记录数一致、tenant_id 完整） +- [ ] `scan` 命令对比源/目标数据无异常差异 +- [ ] Vault KV 数据完整（kvs / released_kvs 与 Vault 交叉验证） +- [ ] `id_generators` 表 max_id 已正确更新 ### 功能验证 - [ ] 应用列表正常显示 -- [ ] 配置项/KV 配置可正常查看 +- [ ] 配置项 / KV 配置可正常查看 - [ ] 发布版本数据完整 - [ ] 模板、分组、钩子、凭证数据正常 - [ ] ITSM v4 服务已注册（如需审批功能） +- [ ] 待审批策略可正常重新发起审批 ### 客户端切换验证 @@ -693,4 +650,4 @@ sequenceDiagram - [ ] 测试客户端拉取配置正常 - [ ] 验证配置内容与源环境一致 - [ ] 监控客户端连接数恢复正常 -- [ ] 验证新发布流程正常工作 \ No newline at end of file +- [ ] 验证新发布流程正常工作 diff --git "a/cmd/tenant-migration/docs/\350\277\201\347\247\273\345\267\245\345\205\267\344\275\277\347\224\250\350\257\264\346\230\216.md" "b/cmd/tenant-migration/docs/\350\277\201\347\247\273\345\267\245\345\205\267\344\275\277\347\224\250\350\257\264\346\230\216.md" index 4108c998c..cb2c1d88f 100644 --- "a/cmd/tenant-migration/docs/\350\277\201\347\247\273\345\267\245\345\205\267\344\275\277\347\224\250\350\257\264\346\230\216.md" +++ "b/cmd/tenant-migration/docs/\350\277\201\347\247\273\345\267\245\345\205\267\344\275\277\347\224\250\350\257\264\346\230\216.md" @@ -8,7 +8,10 @@ BSCP 租户迁移工具 | 命令 | 功能 | |------|------| | `migrate` | 全量迁移（MySQL + Vault） | +| `validate` | 校验源/目标数据一致性 | +| `scan` | 扫描源/目标数据库资产 | | `cleanup` | 清理目标数据库中的迁移数据 | +| `version` | 打印版本信息 | ## 前置准备 1. 停止源环境的写入操作（避免迁移过程中数据变更） @@ -19,66 +22,135 @@ BSCP 租户迁移工具 ## 数据迁移 -按顺序迁移 MySQL 数据（30张核心表）和 Vault KV 数据（如配置）： +按依赖顺序迁移 MySQL 数据（29 张核心表）和 Vault KV 数据（如配置）： ```bash -# 迁移所有业务 -./bk-bscp-tenant-migration -c migration.yaml migrate +# 迁移指定业务（--biz-ids 必填） +./bk-bscp-tenant-migration -c migration.yaml migrate --biz-ids '100,200,300' -# 只迁移指定业务 -./bk-bscp-tenant-migration -c migration.yaml migrate --biz-ids=100,200,300 +# 跳过确认提示 +./bk-bscp-tenant-migration -c migration.yaml migrate --biz-ids '100,200,300' -y ``` -> 注意：如果配置文件中未配置 Vault，则只迁移 MySQL 数据。 +> 注意：`--biz-ids` 是 migrate 命令的必填参数，必须明确指定要迁移的业务 ID。 +> 如果配置文件中未配置 Vault，则只迁移 MySQL 数据。 + +迁移前会自动检查目标库是否已存在对应业务数据，如果存在则会提示先执行 `cleanup`。 ### 迁移命令参数 ``` --c, --config string 配置文件路径（必填） - --biz-ids uint32Slice 要迁移的业务ID列表（逗号分隔，覆盖配置文件） +-c, --config string 配置文件路径（必填，全局参数） + --biz-ids string 要迁移的业务 ID 列表（逗号分隔，必填） +-y, --yes 跳过迁移前确认提示 ``` -## 数据清理 +## 数据校验 -清理目标数据库中的迁移数据，用于迁移失败后重新执行： +校验迁移后源库和目标库的数据一致性： ```bash -# 清理所有数据（交互式确认） -./bk-bscp-tenant-migration -c migration.yaml cleanup - -# 只清理指定业务的数据（跳过确认） -./bk-bscp-tenant-migration -c migration.yaml cleanup --biz-ids=100,200 -f +./bk-bscp-tenant-migration -c migration.yaml validate ``` -### 清理命令参数 +校验范围由配置文件中的 `migration.biz_ids` 控制。 + +### 校验命令参数 ``` --c, --config string 配置文件路径（必填） --f, --force 跳过确认提示 - --biz-ids uint32Slice 要清理的业务ID列表（逗号分隔，覆盖配置文件） +-c, --config string 配置文件路径（必填，全局参数） ``` -## 按业务维度迁移 +## 资产扫描 -支持只迁移指定业务（biz_id）的数据，有两种方式指定业务ID： - -**方式一：通过命令行参数（推荐）** +扫描并对比源/目标数据库的表和记录数： ```bash -./bk-bscp-tenant-migration -c migration.yaml migrate --biz-ids=100,200,300 +# 只扫描配置的迁移表 +./bk-bscp-tenant-migration -c migration.yaml scan + +# 扫描数据库中所有表 +./bk-bscp-tenant-migration -c migration.yaml scan --all + +# 扫描指定业务 +./bk-bscp-tenant-migration -c migration.yaml scan --biz-ids '100,200' ``` -**方式二：通过配置文件** -```yaml -migration: - target_tenant_id: "your_tenant_id" - biz_ids: - - 100 - - 200 - - 300 +### 扫描命令参数 +``` +-c, --config string 配置文件路径（必填，全局参数） +-a, --all 扫描数据库中所有表（默认只扫描配置的迁移表） + --biz-ids string 要扫描的业务 ID 列表（逗号分隔，覆盖配置文件） ``` -> 注意：命令行参数 `--biz-ids` 会覆盖配置文件中的 `biz_ids` 设置。 +## 数据清理 -**适用场景**： -- 分批迁移大量业务数据 -- 只迁移部分测试业务进行验证 -- 针对特定业务进行数据回滚和重新迁移 +清理目标数据库中的迁移数据，用于迁移失败后重新执行： +```bash +# 清理所有数据（交互式确认） +./bk-bscp-tenant-migration -c migration.yaml cleanup + +# 只清理指定业务的数据（跳过确认） +./bk-bscp-tenant-migration -c migration.yaml cleanup --biz-ids '100,200' -f +``` + +### 清理命令参数 +``` +-c, --config string 配置文件路径（必填，全局参数） +-f, --force 跳过确认提示 + --biz-ids string 要清理的业务 ID 列表（逗号分隔，覆盖配置文件） +``` + +## 迁移表列表 + +按依赖顺序迁移以下 29 张核心表： + +| 层级 | 表名 | 说明 | +|------|------|------| +| Level 1 - 基础表 | `sharding_bizs` | 业务分片 | +| | `applications` | 应用 | +| | `template_spaces` | 模板空间 | +| | `groups` | 分组 | +| | `hooks` | 脚本 | +| | `credentials` | 凭据 | +| | `template_variables` | 模板变量 | +| Level 2 - 依赖基础表 | `config_items` | 配置项 | +| | `releases` | 版本 | +| | `strategy_sets` | 策略集 | +| | `templates` | 模板 | +| | `template_sets` | 模板套餐 | +| | `hook_revisions` | 脚本版本 | +| | `credential_scopes` | 凭据范围 | +| | `group_app_binds` | 分组应用绑定 | +| Level 3 - 依赖前两层 | `contents` | 内容 | +| | `commits` | 提交 | +| | `strategies` | 策略 | +| | `current_published_strategies` | 当前发布策略 | +| | `kvs` | KV 配置 | +| | `released_config_items` | 已发布配置项 | +| | `released_groups` | 已发布分组 | +| | `released_hooks` | 已发布脚本 | +| | `released_kvs` | 已发布 KV | +| | `template_revisions` | 模板版本 | +| | `app_template_bindings` | 应用模板绑定 | +| | `app_template_variables` | 应用模板变量 | +| | `released_app_templates` | 已发布应用模板 | +| | `released_app_template_variables` | 已发布应用模板变量 | + +### 默认跳过的表 + +以下表为运行时/历史数据，默认不迁移： + +| 表名 | 说明 | +|------|------| +| `events` | 事件通知，系统自动生成 | +| `current_released_instances` | 客户端当前版本，自动更新 | +| `resource_locks` | 资源锁，运行时数据 | +| `clients` | 客户端连接信息 | +| `client_events` | 客户端事件 | +| `client_querys` | 客户端查询记录 | +| `audits` | 审计日志 | +| `published_strategy_histories` | 发布历史 | +| `archived_apps` | 归档应用 | +| `configs` | ITSM 配置，新环境重新配置 | +| `schema_migrations` | 迁移记录表 | +| `biz_hosts` | 业务主机关系表，定时同步 | +| `sharding_dbs` | 业务数据库分片配置 | ## 配置文件说明 配置文件使用 YAML 格式，主要配置项： @@ -93,5 +165,9 @@ migration: | `source.vault.*` | 否 | 源 Vault 配置（迁移 Vault 时必填） | | `target.mysql.*` | 是 | 目标 MySQL 连接配置 | | `target.vault.*` | 否 | 目标 Vault 配置（迁移 Vault 时必填） | +| `tables` | 否 | 需要迁移的表列表，默认使用核心 29 张表 | +| `skip_tables` | 否 | 跳过的表列表，默认跳过运行时/历史数据表 | +| `log.level` | 否 | 日志级别：debug / info / warn / error，默认 info | +| `log.toStdErr` | 否 | 是否输出到 stderr | 完整配置示例见 `etc/migration.yaml`。 diff --git "a/cmd/tenant-migration/docs/\350\277\201\347\247\273\346\223\215\344\275\234\346\211\213\345\206\214.md" "b/cmd/tenant-migration/docs/\350\277\201\347\247\273\346\223\215\344\275\234\346\211\213\345\206\214.md" new file mode 100644 index 000000000..b26a5f494 --- /dev/null +++ "b/cmd/tenant-migration/docs/\350\277\201\347\247\273\346\223\215\344\275\234\346\211\213\345\206\214.md" @@ -0,0 +1,89 @@ +BSCP 租户迁移 - 运维操作手册 +============================ + +## 前置准备 + +1. 停止源环境写入操作 +2. 备份目标数据库 +3. 准备配置文件 `migration.yaml`（参考 `etc/migration.yaml`） + +## 迁移 + +```bash +# 迁移指定业务（--biz-ids 必填） +./bk-bscp-tenant-migration -c migration.yaml migrate --biz-ids '100,200,300' + +# 跳过确认提示 +./bk-bscp-tenant-migration -c migration.yaml migrate --biz-ids '100,200,300' -y +``` + +参数说明： + +| 参数 | 说明 | +|------|------| +| `-c` | 配置文件路径（必填） | +| `--biz-ids` | 业务 ID 列表，逗号分隔（必填） | +| `-y` | 跳过确认提示 | + +> 如果目标库已有该业务数据，工具会报错并提示先执行 cleanup。 + +## 清理 + +迁移失败后清理目标库数据，以便重新迁移： + +```bash +# 清理指定业务数据（跳过确认） +./bk-bscp-tenant-migration -c migration.yaml cleanup --biz-ids '100,200,300' -f + +# 清理所有数据（交互式确认） +./bk-bscp-tenant-migration -c migration.yaml cleanup +``` + +参数说明： + +| 参数 | 说明 | +|------|------| +| `-c` | 配置文件路径（必填） | +| `--biz-ids` | 业务 ID 列表，逗号分隔（可选） | +| `-f` | 跳过确认提示 | + +## 典型操作流程 + +```bash +# 1. 迁移 +./bk-bscp-tenant-migration -c migration.yaml migrate --biz-ids '100,200' -y + +# 2. 如果失败，先清理再重试 +./bk-bscp-tenant-migration -c migration.yaml cleanup --biz-ids '100,200' -f +./bk-bscp-tenant-migration -c migration.yaml migrate --biz-ids '100,200' -y +``` + +## 配置文件最小示例 + +```yaml +migration: + target_tenant_id: "your_tenant_id" + batch_size: 1000 + +source: + mysql: + endpoints: ["source-host:3306"] + database: "bk_bscp" + user: "root" + password: "xxx" + vault: + address: "http://source-vault:8200" + token: "xxx" + +target: + mysql: + endpoints: ["target-host:3306"] + database: "bk_bscp" + user: "root" + password: "xxx" + vault: + address: "http://target-vault:8200" + token: "xxx" +``` + +> 如果不需要迁移 Vault 数据，可省略 `source.vault` 和 `target.vault` 配置。 diff --git a/cmd/tenant-migration/migrator/migrator.go b/cmd/tenant-migration/migrator/migrator.go index c2ef8f6a1..f5b0acbd2 100644 --- a/cmd/tenant-migration/migrator/migrator.go +++ b/cmd/tenant-migration/migrator/migrator.go @@ -82,7 +82,7 @@ func NewMigrator(cfg *config.Config) (*Migrator, error) { } validator := NewValidator(cfg, mysqlMigrator.sourceDB, mysqlMigrator.targetDB) - scanner := NewScanner(cfg, mysqlMigrator.sourceDB, mysqlMigrator.targetDB) + scanner := NewScanner(cfg, mysqlMigrator.sourceDB, mysqlMigrator.targetDB, vaultMigrator) return &Migrator{ cfg: cfg, @@ -186,31 +186,21 @@ type FullCleanupResult struct { Success bool } -// Cleanup clears migrated data from target database and Vault -// If biz_id filter is configured (--biz-ids), only clears data for those businesses -// This should be run before migration to ensure clean state for the target business +// Cleanup clears migrated data from target database and Vault. +// If biz_id filter is configured (--biz-ids), only clears data for those businesses. +// Vault cleanup runs FIRST because it reads target DB records to construct correct Vault paths; +// MySQL cleanup runs SECOND so that target DB records are still available for Vault path lookup. func (m *Migrator) Cleanup() (*FullCleanupResult, error) { startTime := time.Now() result := &FullCleanupResult{ Success: true, } - log.Println("Starting full cleanup (MySQL + Vault)...") + log.Println("Starting full cleanup (Vault + MySQL)...") - // Step 1: Cleanup MySQL - log.Println("Step 1/2: MySQL cleanup...") - mysqlResult, err := m.mysqlMigrator.CleanupTarget() - result.MySQLResult = mysqlResult - if err != nil { - result.Success = false - log.Printf("MySQL cleanup failed: %v", err) - } else if !mysqlResult.Success { - result.Success = false - } - - // Step 2: Cleanup Vault (if configured) + // Step 1: Cleanup Vault FIRST (reads target DB to get mapped IDs for Vault paths) if m.vaultMigrator != nil { - log.Println("Step 2/2: Vault cleanup...") + log.Println("Step 1/2: Vault cleanup...") vaultResult, err := m.vaultMigrator.CleanupTarget() result.VaultResult = vaultResult if err != nil { @@ -220,7 +210,18 @@ func (m *Migrator) Cleanup() (*FullCleanupResult, error) { result.Success = false } } else { - log.Println("Step 2/2: Vault cleanup skipped (not configured)") + log.Println("Step 1/2: Vault cleanup skipped (not configured)") + } + + // Step 2: Cleanup MySQL SECOND (after Vault has read the records it needs) + log.Println("Step 2/2: MySQL cleanup...") + mysqlResult, err := m.mysqlMigrator.CleanupTarget() + result.MySQLResult = mysqlResult + if err != nil { + result.Success = false + log.Printf("MySQL cleanup failed: %v", err) + } else if !mysqlResult.Success { + result.Success = false } result.Duration = time.Since(startTime) @@ -300,11 +301,12 @@ func (m *Migrator) PrintCleanupReport(result *FullCleanupResult) { fmt.Println(strings.Repeat("=", 60)) } -// PrintReport prints the migration report to stdout +// PrintReport prints a concise migration summary to stdout. +// If there are failed records, full details are written to a JSON log file. func (m *Migrator) PrintReport(report *MigrationReport) { - fmt.Println("\n" + strings.Repeat("=", 60)) + fmt.Println("\n" + strings.Repeat("=", 70)) fmt.Println("MIGRATION REPORT") - fmt.Println(strings.Repeat("=", 60)) + fmt.Println(strings.Repeat("=", 70)) fmt.Printf("Start Time: %s\n", report.StartTime.Format(time.RFC3339)) fmt.Printf("End Time: %s\n", report.EndTime.Format(time.RFC3339)) fmt.Printf("Duration: %v\n", report.Duration) @@ -313,20 +315,21 @@ func (m *Migrator) PrintReport(report *MigrationReport) { if len(report.TableResults) > 0 { fmt.Println("MySQL Migration Results:") - fmt.Println(strings.Repeat("-", 60)) - fmt.Printf("%-35s %10s %10s %8s\n", "Table", "Source", "Migrated", "Status") - fmt.Println(strings.Repeat("-", 60)) + fmt.Println(strings.Repeat("-", 70)) + fmt.Printf("%-35s %10s %10s %8s %8s\n", "Table", "Source", "Migrated", "Failed", "Status") + fmt.Println(strings.Repeat("-", 70)) for _, tr := range report.TableResults { - fmt.Printf("%-35s %10d %10d %8s\n", - tr.TableName, tr.SourceCount, tr.MigratedCount, boolToStatus(tr.Success)) + fmt.Printf("%-35s %10d %10d %8d %8s\n", + tr.TableName, tr.SourceCount, tr.MigratedCount, + tr.ErrorCount, boolToStatus(tr.Success)) } fmt.Println() } if report.VaultResults != nil { fmt.Println("Vault Migration Results:") - fmt.Println(strings.Repeat("-", 60)) + fmt.Println(strings.Repeat("-", 70)) fmt.Printf("KV Records: %d migrated of %d\n", report.VaultResults.MigratedKvs, report.VaultResults.KvCount) fmt.Printf("Released KV Records: %d migrated of %d\n", @@ -337,14 +340,73 @@ func (m *Migrator) PrintReport(report *MigrationReport) { if len(report.Errors) > 0 { fmt.Println("Errors:") - fmt.Println(strings.Repeat("-", 60)) + fmt.Println(strings.Repeat("-", 70)) for _, err := range report.Errors { fmt.Printf(" - %s\n", err) } fmt.Println() } - fmt.Println(strings.Repeat("=", 60)) + // Write failed rows to log file if any + if m.mysqlMigrator.HasFailedRows() { + logDir := m.buildLogDir() + logFile, err := m.mysqlMigrator.WriteFailedRowsLog(logDir) + if err != nil { + log.Printf("Warning: failed to write error log: %v", err) + } else if logFile != "" { + fmt.Printf(" >> Failed records detail saved to: %s\n\n", logFile) + } + } + + fmt.Println(strings.Repeat("=", 70)) +} + +// ExistingDataItem describes a table with existing data in the target database +type ExistingDataItem struct { + Table string + Count int64 +} + +// CheckTargetData checks if the target database already has data for the configured biz_ids. +// Returns a list of tables with existing record counts (empty if no data found). +func (m *Migrator) CheckTargetData() []ExistingDataItem { + if !m.cfg.Migration.HasBizFilter() { + return nil + } + + var result []ExistingDataItem + coreLevel1Tables := []string{"applications", "template_spaces", "groups", "hooks", "credentials"} + + for _, table := range coreLevel1Tables { + var count int64 + err := m.mysqlMigrator.targetDB.Table(table). + Where("biz_id IN ?", m.cfg.Migration.BizIDs). + Count(&count).Error + if err != nil { + log.Printf("Warning: failed to check target table %s: %v", table, err) + continue + } + if count > 0 { + result = append(result, ExistingDataItem{Table: table, Count: count}) + } + } + return result +} + +// buildLogDir returns the log directory path based on biz_ids filter. +// e.g. logs/biz_100_200/ or logs/all/ +func (m *Migrator) buildLogDir() string { + var bizDir string + if m.cfg.Migration.HasBizFilter() { + parts := make([]string, 0, len(m.cfg.Migration.BizIDs)) + for _, id := range m.cfg.Migration.BizIDs { + parts = append(parts, fmt.Sprintf("%d", id)) + } + bizDir = "biz_" + strings.Join(parts, "_") + } else { + bizDir = "all" + } + return fmt.Sprintf("logs/%s", bizDir) } func boolToStatus(success bool) string { diff --git a/cmd/tenant-migration/migrator/mysql.go b/cmd/tenant-migration/migrator/mysql.go index 33d09dee8..93244cdb1 100644 --- a/cmd/tenant-migration/migrator/mysql.go +++ b/cmd/tenant-migration/migrator/mysql.go @@ -13,8 +13,10 @@ package migrator import ( + "encoding/json" "fmt" "log" + "os" "time" "gorm.io/driver/mysql" @@ -24,12 +26,23 @@ import ( "github.com/TencentBlueKing/bk-bscp/cmd/tenant-migration/config" ) +// FailedRow records a single row that failed during migration +type FailedRow struct { + Table string `json:"table"` + SourceID uint32 `json:"source_id"` + NewID uint32 `json:"new_id,omitempty"` + BizID uint32 `json:"biz_id"` + Error string `json:"error"` + Data map[string]interface{} `json:"data"` +} + // MySQLMigrator handles MySQL data migration type MySQLMigrator struct { - cfg *config.Config - sourceDB *gorm.DB - targetDB *gorm.DB - idMapper *IDMapper + cfg *config.Config + sourceDB *gorm.DB + targetDB *gorm.DB + idMapper *IDMapper + failedRows []FailedRow } // NewMySQLMigrator creates a new MySQLMigrator instance @@ -48,6 +61,12 @@ func NewMySQLMigrator(cfg *config.Config) (*MySQLMigrator, error) { sqlDB.SetMaxOpenConns(int(cfg.Source.MySQL.MaxOpenConn)) sqlDB.SetMaxIdleConns(int(cfg.Source.MySQL.MaxIdleConn)) + err = sqlDB.Ping() + if err != nil { + return nil, fmt.Errorf("source database is unreachable: %w", err) + } + log.Println("Source MySQL connection OK") + // Connect to target database targetDB, err := gorm.Open(mysql.Open(cfg.Target.MySQL.DSN()), &gorm.Config{Logger: logger.Default.LogMode(logger.Warn)}) @@ -62,6 +81,12 @@ func NewMySQLMigrator(cfg *config.Config) (*MySQLMigrator, error) { sqlDB.SetMaxOpenConns(int(cfg.Target.MySQL.MaxOpenConn)) sqlDB.SetMaxIdleConns(int(cfg.Target.MySQL.MaxIdleConn)) + err = sqlDB.Ping() + if err != nil { + return nil, fmt.Errorf("target database is unreachable: %w", err) + } + log.Println("Target MySQL connection OK") + return &MySQLMigrator{ cfg: cfg, sourceDB: sourceDB, @@ -196,7 +221,7 @@ func (m *MySQLMigrator) migrateTable(tableName string) TableMigrationResult { if hasBizID && hasBizFilter { batchQuery = batchQuery.Where("biz_id IN ?", m.cfg.Migration.BizIDs) } - if err := batchQuery.Offset(offset).Limit(batchSize).Find(&rows).Error; err != nil { + if err := batchQuery.Order("id").Offset(offset).Limit(batchSize).Find(&rows).Error; err != nil { result.Errors = append(result.Errors, fmt.Sprintf("failed to read batch at offset %d: %v", offset, err)) result.Success = false break @@ -215,7 +240,9 @@ func (m *MySQLMigrator) migrateTable(tableName string) TableMigrationResult { newID, err := m.getNextID(tableName) if err != nil { result.ErrorCount++ - result.Errors = append(result.Errors, fmt.Sprintf("failed to get next ID: %v", err)) + errMsg := fmt.Sprintf("failed to get next ID: %v", err) + result.Errors = append(result.Errors, errMsg) + m.logFailedRow(tableName, sourceID, 0, row, errMsg) if !m.cfg.Migration.ContinueOnError { result.Success = false break @@ -232,7 +259,22 @@ func (m *MySQLMigrator) migrateTable(tableName string) TableMigrationResult { // Convert foreign keys using ID mapper if err := m.convertForeignKeys(row, meta); err != nil { result.ErrorCount++ - result.Errors = append(result.Errors, fmt.Sprintf("failed to convert foreign keys (sourceID=%d): %v", sourceID, err)) + errMsg := fmt.Sprintf("failed to convert foreign keys (sourceID=%d): %v", sourceID, err) + result.Errors = append(result.Errors, errMsg) + m.logFailedRow(tableName, sourceID, newID, row, errMsg) + if !m.cfg.Migration.ContinueOnError { + result.Success = false + break + } + continue + } + + // Convert JSON array foreign keys (e.g. template_ids in template_sets) + if err := m.convertJSONArrayFKs(row, meta); err != nil { + result.ErrorCount++ + errMsg := fmt.Sprintf("failed to convert JSON array FKs (sourceID=%d): %v", sourceID, err) + result.Errors = append(result.Errors, errMsg) + m.logFailedRow(tableName, sourceID, newID, row, errMsg) if !m.cfg.Migration.ContinueOnError { result.Success = false break @@ -240,6 +282,21 @@ func (m *MySQLMigrator) migrateTable(tableName string) TableMigrationResult { continue } + // Convert complex bindings JSON in app_template_bindings + if tableName == "app_template_bindings" { + if err := m.convertBindingsJSON(row); err != nil { + result.ErrorCount++ + errMsg := fmt.Sprintf("failed to convert bindings JSON (sourceID=%d): %v", sourceID, err) + result.Errors = append(result.Errors, errMsg) + m.logFailedRow(tableName, sourceID, newID, row, errMsg) + if !m.cfg.Migration.ContinueOnError { + result.Success = false + break + } + continue + } + } + // Fill tenant_id if the column exists if hasTenantID { row["tenant_id"] = m.cfg.Migration.TargetTenantID @@ -251,7 +308,9 @@ func (m *MySQLMigrator) migrateTable(tableName string) TableMigrationResult { // Insert into target database if err := m.targetDB.Table(tableName).Create(row).Error; err != nil { result.ErrorCount++ - result.Errors = append(result.Errors, fmt.Sprintf("failed to insert row (sourceID=%d, newID=%d): %v", sourceID, newID, err)) + errMsg := fmt.Sprintf("failed to insert row (sourceID=%d, newID=%d): %v", sourceID, newID, err) + result.Errors = append(result.Errors, errMsg) + m.logFailedRow(tableName, sourceID, newID, row, errMsg) if !m.cfg.Migration.ContinueOnError { result.Success = false break @@ -354,6 +413,15 @@ func (m *MySQLMigrator) convertForeignKeys(row map[string]interface{}, meta Tabl // Look up the target ID from the mapper targetFK := m.idMapper.Get(refTable, sourceFK) if targetFK == 0 { + if meta.OptionalFKs[fkColumn] { + virtualID, err := m.getOrCreateVirtualID(refTable, sourceFK) + if err != nil { + return fmt.Errorf("failed to generate virtual ID for optional FK %s=%d: %w", + fkColumn, sourceFK, err) + } + row[fkColumn] = virtualID + continue + } return fmt.Errorf("foreign key %s=%d references %s, but no mapping found (referenced table may not have been migrated yet)", fkColumn, sourceFK, refTable) } @@ -362,6 +430,213 @@ func (m *MySQLMigrator) convertForeignKeys(row map[string]interface{}, meta Tabl return nil } +// getOrCreateVirtualID generates a virtual ID for a deleted record. +// If a virtual ID was already generated for this source ID, it returns the same one. +func (m *MySQLMigrator) getOrCreateVirtualID(refTable string, sourceID uint32) (uint32, error) { + if existingID := m.idMapper.Get(refTable, sourceID); existingID != 0 { + return existingID, nil + } + + newID, err := m.getNextID(refTable) + if err != nil { + return 0, err + } + + m.idMapper.Set(refTable, sourceID, newID) + return newID, nil +} + +// convertJSONArrayFKs converts ID values inside JSON array columns using ID mapper +func (m *MySQLMigrator) convertJSONArrayFKs(row map[string]interface{}, meta TableMeta) error { + if len(meta.JSONArrayFKs) == 0 { + return nil + } + + for jsonCol, refTable := range meta.JSONArrayFKs { + rawVal, ok := row[jsonCol] + if !ok || rawVal == nil { + continue + } + + sourceIDs, err := parseJSONUint32Array(rawVal) + if err != nil { + return fmt.Errorf("failed to parse JSON array column %s: %w", jsonCol, err) + } + + if len(sourceIDs) == 0 { + continue + } + + targetIDs := make([]uint32, 0, len(sourceIDs)) + for _, sid := range sourceIDs { + if sid == 0 { + targetIDs = append(targetIDs, 0) + continue + } + tid := m.idMapper.Get(refTable, sid) + if tid == 0 { + return fmt.Errorf("JSON array column %s: source ID %d references %s, but no mapping found", + jsonCol, sid, refTable) + } + targetIDs = append(targetIDs, tid) + } + + data, err := json.Marshal(targetIDs) + if err != nil { + return fmt.Errorf("failed to marshal converted IDs for column %s: %w", jsonCol, err) + } + row[jsonCol] = string(data) + } + return nil +} + +// convertBindingsJSON converts IDs inside the complex bindings JSON column of app_template_bindings. +// The structure is: [{"template_set_id": N, "template_revisions": [{"template_id": N, "template_revision_id": N, "is_latest": bool}]}] +func (m *MySQLMigrator) convertBindingsJSON(row map[string]interface{}) error { + rawVal, ok := row["bindings"] + if !ok || rawVal == nil { + return nil + } + + rawBytes, err := toJSONBytes(rawVal) + if err != nil { + return fmt.Errorf("failed to read bindings column: %w", err) + } + + var bindings []map[string]interface{} + if unmarshalErr := json.Unmarshal(rawBytes, &bindings); unmarshalErr != nil { + return fmt.Errorf("failed to parse bindings JSON: %w", unmarshalErr) + } + + for i, binding := range bindings { + if tsID, ok := binding["template_set_id"]; ok && tsID != nil { + sourceID := jsonNumberToUint32(tsID) + if sourceID != 0 { + targetID := m.idMapper.Get("template_sets", sourceID) + if targetID == 0 { + return fmt.Errorf("bindings[%d].template_set_id=%d: no mapping found in template_sets", i, sourceID) + } + binding["template_set_id"] = targetID + } + } + + rawRevisions, ok := binding["template_revisions"] + if !ok || rawRevisions == nil { + continue + } + + revisions, ok := rawRevisions.([]interface{}) + if !ok { + continue + } + + for j, rawRev := range revisions { + rev, ok := rawRev.(map[string]interface{}) + if !ok { + continue + } + + if tmplID, ok := rev["template_id"]; ok && tmplID != nil { + sourceID := jsonNumberToUint32(tmplID) + if sourceID != 0 { + targetID := m.idMapper.Get("templates", sourceID) + if targetID == 0 { + return fmt.Errorf("bindings[%d].template_revisions[%d].template_id=%d: no mapping found in templates", + i, j, sourceID) + } + rev["template_id"] = targetID + } + } + + if trID, ok := rev["template_revision_id"]; ok && trID != nil { + sourceID := jsonNumberToUint32(trID) + if sourceID != 0 { + targetID := m.idMapper.Get("template_revisions", sourceID) + if targetID == 0 { + return fmt.Errorf( + "bindings[%d].template_revisions[%d].template_revision_id=%d: no mapping found in template_revisions", + i, j, sourceID) + } + rev["template_revision_id"] = targetID + } + } + } + } + + data, err := json.Marshal(bindings) + if err != nil { + return fmt.Errorf("failed to marshal converted bindings: %w", err) + } + row["bindings"] = string(data) + return nil +} + +// parseJSONUint32Array parses a raw DB value ([]byte or string) as a JSON array of uint32 +func parseJSONUint32Array(val interface{}) ([]uint32, error) { + rawBytes, err := toJSONBytes(val) + if err != nil { + return nil, err + } + + var numbers []json.Number + if err := json.Unmarshal(rawBytes, &numbers); err != nil { + var floats []float64 + if err2 := json.Unmarshal(rawBytes, &floats); err2 != nil { + return nil, fmt.Errorf("cannot parse as number array: %w", err) + } + result := make([]uint32, len(floats)) + for i, f := range floats { + result[i] = uint32(f) + } + return result, nil + } + + result := make([]uint32, len(numbers)) + for i, n := range numbers { + v, err := n.Int64() + if err != nil { + return nil, fmt.Errorf("invalid number at index %d: %w", i, err) + } + result[i] = uint32(v) + } + return result, nil +} + +// toJSONBytes converts a raw DB value to JSON bytes +func toJSONBytes(val interface{}) ([]byte, error) { + switch v := val.(type) { + case []byte: + return v, nil + case string: + return []byte(v), nil + default: + data, err := json.Marshal(val) + if err != nil { + return nil, fmt.Errorf("unsupported type %T for JSON conversion", val) + } + return data, nil + } +} + +// jsonNumberToUint32 converts a JSON-deserialized number value to uint32 +func jsonNumberToUint32(val interface{}) uint32 { + switch v := val.(type) { + case float64: + return uint32(v) + case json.Number: + n, _ := v.Int64() + return uint32(n) + case int64: + return uint32(v) + case uint32: + return v + case uint64: + return uint32(v) + default: + return 0 + } +} + // hasTenantIDColumn checks if a table has tenant_id column func (m *MySQLMigrator) hasTenantIDColumn(tableName string) bool { return m.hasColumn(tableName, "tenant_id", m.cfg.Target.MySQL.Database) @@ -388,6 +663,36 @@ func (m *MySQLMigrator) hasColumn(tableName, columnName, database string) bool { return count > 0 } +// detectBizIDColumn returns the biz ID column name for a table in the target database. +// Most tables use "biz_id", but some (e.g. biz_hosts) use "bk_biz_id". +// Returns empty string if the table doesn't exist or has no biz ID column. +func (m *MySQLMigrator) detectBizIDColumn(tableName string) string { + if !m.tableExists(tableName, m.cfg.Target.MySQL.Database) { + return "" + } + if m.hasColumn(tableName, "biz_id", m.cfg.Target.MySQL.Database) { + return "biz_id" + } + if m.hasColumn(tableName, "bk_biz_id", m.cfg.Target.MySQL.Database) { + return "bk_biz_id" + } + return "" +} + +// tableExists checks if a table exists in the specified database +func (m *MySQLMigrator) tableExists(tableName, database string) bool { + var count int64 + query := `SELECT COUNT(*) FROM information_schema.tables WHERE table_schema = ? AND table_name = ?` + db := m.sourceDB + if database == m.cfg.Target.MySQL.Database { + db = m.targetDB + } + if err := db.Raw(query, database, tableName).Scan(&count).Error; err != nil { + return false + } + return count > 0 +} + // handleSpecialCases handles special type conversions and transformations func (m *MySQLMigrator) handleSpecialCases(tableName string, row map[string]interface{}) { if tableName == "strategies" { @@ -400,6 +705,8 @@ func (m *MySQLMigrator) handleSpecialCases(tableName string, row map[string]inte row["itsm_ticket_state_id"] = fmt.Sprintf("%.0f", v) } } + + m.clearItsmFields(row) } // KV tables version reset: @@ -410,6 +717,35 @@ func (m *MySQLMigrator) handleSpecialCases(tableName string, row map[string]inte } } +// clearItsmFields handles ITSM ticket fields for strategies during cross-environment migration. +// +// Only pending strategies need processing. Completed strategies (already_publish, rejected_approval, +// etc.) are left untouched because: +// - The UI does not display ITSM fields for these statuses (approveStatus = -1). +// - Preserving the original approval record is harmless and keeps audit traceability. +// +// Pending strategies (pending_approval / pending_publish) are set to "revoked_publish" because: +// 1. The source ITSM ticket SN, callback URL, and workflow config do not work in the target env. +// 2. Keeping "pending_approval" without a real ITSM ticket would block new publishes +// (SubmitPublishApprove rejects when an existing strategy is pending). +// 3. The UI would show a confusing "pending" spinner with no approval link. +// 4. "revoked_publish" is transient in the UI (disappears on page refresh) and does not +// block new publish operations, allowing users to re-submit in the new environment. +func (m *MySQLMigrator) clearItsmFields(row map[string]interface{}) { + publishStatus, _ := row["publish_status"].(string) + if publishStatus != "pending_approval" && publishStatus != "pending_publish" { + return + } + + row["publish_status"] = "revoked_publish" + row["approver_progress"] = "" + row["itsm_ticket_type"] = "" + row["itsm_ticket_url"] = "" + row["itsm_ticket_sn"] = "" + row["itsm_ticket_status"] = "" + row["itsm_ticket_state_id"] = "" +} + // GetSourceDB returns the source database connection func (m *MySQLMigrator) GetSourceDB() *gorm.DB { return m.sourceDB @@ -443,16 +779,12 @@ func (m *MySQLMigrator) CleanupTarget() (*CleanupResult, error) { } }() - // Use proper cleanup order - cleanupTables := TablesInCleanupOrder() - log.Println("Cleaning up target database...") - for _, tableName := range cleanupTables { - if m.cfg.ShouldSkipTable(tableName) { - continue - } - + // Step 1: Clean runtime tables (audits, events, clients, etc.) + // These are not migrated but accumulate data in the target env and must be cleaned per biz_id. + log.Println(" Cleaning runtime tables...") + for _, tableName := range RuntimeCleanupTables() { tableResult := m.cleanupTable(tableName) result.TableResults = append(result.TableResults, tableResult) @@ -464,6 +796,26 @@ func (m *MySQLMigrator) CleanupTarget() (*CleanupResult, error) { } } + // Step 2: Clean core migration tables in reverse dependency order + if result.Success || m.cfg.Migration.ContinueOnError { + log.Println(" Cleaning core tables...") + for _, tableName := range TablesInCleanupOrder() { + if m.cfg.ShouldSkipTable(tableName) { + continue + } + + tableResult := m.cleanupTable(tableName) + result.TableResults = append(result.TableResults, tableResult) + + if !tableResult.Success { + result.Success = false + if !m.cfg.Migration.ContinueOnError { + break + } + } + } + } + result.Duration = time.Since(startTime) log.Printf("Cleanup completed in %v", result.Duration) @@ -478,14 +830,19 @@ func (m *MySQLMigrator) cleanupTable(tableName string) TableCleanupResult { Success: true, } - // Check if table has biz_id column for filtering - hasBizID := m.hasColumn(tableName, "biz_id", m.cfg.Target.MySQL.Database) + if !m.tableExists(tableName, m.cfg.Target.MySQL.Database) { + log.Printf(" Table %s does not exist, skipping", tableName) + return result + } + + // Detect the biz ID column name: most tables use "biz_id", biz_hosts uses "bk_biz_id" + bizIDColumn := m.detectBizIDColumn(tableName) hasBizFilter := m.cfg.Migration.HasBizFilter() // Build base query with biz_id filter if applicable baseQuery := m.targetDB.Table(tableName) - if hasBizID && hasBizFilter { - baseQuery = baseQuery.Where("biz_id IN ?", m.cfg.Migration.BizIDs) + if bizIDColumn != "" && hasBizFilter { + baseQuery = baseQuery.Where(bizIDColumn+" IN ?", m.cfg.Migration.BizIDs) } // Count records before deletion @@ -498,19 +855,20 @@ func (m *MySQLMigrator) cleanupTable(tableName string) TableCleanupResult { result.DeletedCount = count if count == 0 { - log.Printf(" Table %s is empty (or no matching biz_id), skipping", tableName) + log.Printf(" Table %s is empty (or no matching records), skipping", tableName) return result } // Delete records - if hasBizID && hasBizFilter { - // Delete only records matching the biz_id filter - if err := m.targetDB.Table(tableName).Where("biz_id IN ?", m.cfg.Migration.BizIDs).Delete(nil).Error; err != nil { + if bizIDColumn != "" && hasBizFilter { + if err := m.targetDB.Table(tableName).Where(bizIDColumn+" IN ?", m.cfg.Migration.BizIDs). + Delete(nil).Error; err != nil { result.Error = fmt.Sprintf("failed to delete records: %v", err) result.Success = false return result } - log.Printf(" Deleted %d records from table %s (biz_id filter: %v)", count, tableName, m.cfg.Migration.BizIDs) + log.Printf(" Deleted %d records from table %s (%s filter: %v)", + count, tableName, bizIDColumn, m.cfg.Migration.BizIDs) } else { // Delete all records using TRUNCATE for better performance // Use backticks to handle reserved keywords like 'groups' @@ -535,6 +893,83 @@ type CleanupResult struct { Success bool } +// logFailedRow collects a failed row for later file output +func (m *MySQLMigrator) logFailedRow(tableName string, sourceID, newID uint32, row map[string]interface{}, errMsg string) { + bizID := m.getUint32FromRow(row, "biz_id") + + rowCopy := make(map[string]interface{}, len(row)) + for k, v := range row { + rowCopy[k] = v + } + + m.failedRows = append(m.failedRows, FailedRow{ + Table: tableName, + SourceID: sourceID, + NewID: newID, + BizID: bizID, + Error: errMsg, + Data: rowCopy, + }) +} + +// HasFailedRows returns true if there are any collected failed rows +func (m *MySQLMigrator) HasFailedRows() bool { + return len(m.failedRows) > 0 +} + +// WriteFailedRowsLog writes all collected failed rows to a JSON log file, grouped by biz_id. +// logDir specifies the directory (e.g. "logs/biz_100_200"), file is named migrate_.json. +func (m *MySQLMigrator) WriteFailedRowsLog(logDir string) (string, error) { + if len(m.failedRows) == 0 { + return "", nil + } + + grouped := make(map[uint32][]FailedRow) + for _, r := range m.failedRows { + grouped[r.BizID] = append(grouped[r.BizID], r) + } + + type bizFailedRows struct { + BizID uint32 `json:"biz_id"` + Count int `json:"count"` + FailedRows []FailedRow `json:"failed_rows"` + } + + output := struct { + Timestamp string `json:"timestamp"` + TotalFailed int `json:"total_failed"` + Businesses []bizFailedRows `json:"businesses"` + }{ + Timestamp: time.Now().Format(time.RFC3339), + TotalFailed: len(m.failedRows), + } + + for bizID, rows := range grouped { + output.Businesses = append(output.Businesses, bizFailedRows{ + BizID: bizID, + Count: len(rows), + FailedRows: rows, + }) + } + + data, err := json.MarshalIndent(output, "", " ") + if err != nil { + return "", fmt.Errorf("failed to marshal failed rows: %w", err) + } + + if err := os.MkdirAll(logDir, 0755); err != nil { + return "", fmt.Errorf("failed to create log directory %s: %w", logDir, err) + } + + filename := fmt.Sprintf("migrate_%s.json", time.Now().Format("20060102_150405")) + fullPath := logDir + "/" + filename + if err := os.WriteFile(fullPath, data, 0644); err != nil { + return "", fmt.Errorf("failed to write failed rows log: %w", err) + } + + return fullPath, nil +} + // TableCleanupResult contains the result of cleaning up a single table type TableCleanupResult struct { TableName string diff --git a/cmd/tenant-migration/migrator/scanner.go b/cmd/tenant-migration/migrator/scanner.go index 72e180964..76fd93471 100644 --- a/cmd/tenant-migration/migrator/scanner.go +++ b/cmd/tenant-migration/migrator/scanner.go @@ -25,9 +25,10 @@ import ( // Scanner handles asset scanning for source and target databases type Scanner struct { - cfg *config.Config - sourceDB *gorm.DB - targetDB *gorm.DB + cfg *config.Config + sourceDB *gorm.DB + targetDB *gorm.DB + vaultMigrator *VaultMigrator } // ScanReport contains the complete scan results @@ -70,6 +71,7 @@ type VaultScanSummary struct { SourceReleasedKvCount int64 TargetKvCount int64 TargetReleasedKvCount int64 + VaultDirect *VaultDirectScanResult } // ScanComparison contains comparison statistics @@ -83,11 +85,12 @@ type ScanComparison struct { } // NewScanner creates a new Scanner instance -func NewScanner(cfg *config.Config, sourceDB, targetDB *gorm.DB) *Scanner { +func NewScanner(cfg *config.Config, sourceDB, targetDB *gorm.DB, vaultMigrator *VaultMigrator) *Scanner { return &Scanner{ - cfg: cfg, - sourceDB: sourceDB, - targetDB: targetDB, + cfg: cfg, + sourceDB: sourceDB, + targetDB: targetDB, + vaultMigrator: vaultMigrator, } } @@ -177,6 +180,16 @@ func (s *Scanner) scanTables(tables []string, mode string) (*ScanReport, error) // Scan Vault if configured if s.cfg.Source.Vault.Address != "" || s.cfg.Target.Vault.Address != "" { report.VaultSummary = s.scanVaultFromDB() + + if s.vaultMigrator != nil { + log.Println("Scanning Vault directly...") + vaultDirect, err := s.vaultMigrator.ScanVault() + if err != nil { + log.Printf("Warning: Vault direct scan failed: %v", err) + } else { + report.VaultSummary.VaultDirect = vaultDirect + } + } } report.EndTime = time.Now() @@ -413,33 +426,86 @@ func (s *Scanner) PrintReport(report *ScanReport) { // Vault summary if report.VaultSummary != nil { - fmt.Println("VAULT KV SUMMARY") + s.printVaultSummary(report.VaultSummary) + } + + // Comparison summary and legend + s.printComparisonSummary(&report.Comparison) + + fmt.Println(strings.Repeat("=", 80)) +} + +// printVaultSummary prints the Vault section of the scan report +func (s *Scanner) printVaultSummary(vs *VaultScanSummary) { + fmt.Println("VAULT KV SUMMARY (Database)") + fmt.Println(strings.Repeat("-", 80)) + fmt.Printf("%-35s %12s %12s\n", "", "Source", "Target") + fmt.Printf("%-35s %12d %12d\n", "Unreleased KVs (kvs table)", + vs.SourceKvCount, vs.TargetKvCount) + fmt.Printf("%-35s %12d %12d\n", "Released KVs (released_kvs table)", + vs.SourceReleasedKvCount, vs.TargetReleasedKvCount) + fmt.Println() + + if vs.VaultDirect != nil { + vd := vs.VaultDirect + fmt.Println("VAULT DIRECT VERIFICATION (DB records vs Vault)") fmt.Println(strings.Repeat("-", 80)) - fmt.Printf("%-35s %12s %12s\n", "", "Source", "Target") - fmt.Printf("%-35s %12d %12d\n", "Unreleased KVs (kvs table)", - report.VaultSummary.SourceKvCount, report.VaultSummary.TargetKvCount) - fmt.Printf("%-35s %12d %12d\n", "Released KVs (released_kvs table)", - report.VaultSummary.SourceReleasedKvCount, report.VaultSummary.TargetReleasedKvCount) + + printVaultScanTable("Source Vault", vd.SourceBizScans) + fmt.Println() + printVaultScanTable("Target Vault", vd.TargetBizScans) + + fmt.Printf("\n Scan Duration: %v\n", vd.Duration) + + if len(vd.Errors) > 0 { + fmt.Println(" Errors:") + for _, e := range vd.Errors { + fmt.Printf(" - %s\n", e) + } + } fmt.Println() } +} - // Comparison summary +func printVaultScanTable(label string, scans []BizVaultScan) { + fmt.Printf(" %s:\n", label) + if len(scans) == 0 { + fmt.Println(" No DB records to probe") + return + } + fmt.Printf(" %-12s %8s %8s %8s %8s %8s\n", + "Biz", "KV(DB)", "KV(V)", "RKV(DB)", "RKV(V)", "Status") + fmt.Printf(" %s\n", strings.Repeat("-", 60)) + for _, scan := range scans { + status := "OK" + if scan.KvDB != scan.KvVault || scan.RKvDB != scan.RKvVault { + status = "MISMATCH" + } + fmt.Printf(" %-12d %8d %8d %8d %8d %8s\n", + scan.BizID, scan.KvDB, scan.KvVault, scan.RKvDB, scan.RKvVault, status) + } + fmt.Printf(" %s\n", strings.Repeat("-", 60)) + fmt.Printf(" %-12s %8d %8d %8d %8d\n", "Total", + totalKvDB(scans), totalKvVault(scans), + totalRKvDB(scans), totalRKvVault(scans)) +} + +// printComparisonSummary prints the comparison and legend sections +func (s *Scanner) printComparisonSummary(comp *ScanComparison) { fmt.Println("COMPARISON SUMMARY") fmt.Println(strings.Repeat("-", 80)) - fmt.Printf("Tables with matching records: %d\n", len(report.Comparison.MatchingTables)) - fmt.Printf("Tables with more data in source: %d\n", len(report.Comparison.TablesWithMoreData)) - fmt.Printf("Tables with more data in target: %d\n", len(report.Comparison.TablesWithLessData)) + fmt.Printf("Tables with matching records: %d\n", len(comp.MatchingTables)) + fmt.Printf("Tables with more data in source: %d\n", len(comp.TablesWithMoreData)) + fmt.Printf("Tables with more data in target: %d\n", len(comp.TablesWithLessData)) - if len(report.Comparison.TablesOnlyInSource) > 0 { - fmt.Printf("Tables only in source: %s\n", strings.Join(report.Comparison.TablesOnlyInSource, ", ")) + if len(comp.TablesOnlyInSource) > 0 { + fmt.Printf("Tables only in source: %s\n", strings.Join(comp.TablesOnlyInSource, ", ")) } - if len(report.Comparison.TablesOnlyInTarget) > 0 { - fmt.Printf("Tables only in target: %s\n", strings.Join(report.Comparison.TablesOnlyInTarget, ", ")) + if len(comp.TablesOnlyInTarget) > 0 { + fmt.Printf("Tables only in target: %s\n", strings.Join(comp.TablesOnlyInTarget, ", ")) } fmt.Println() - - // Status legend fmt.Println("STATUS LEGEND") fmt.Println(strings.Repeat("-", 80)) fmt.Println(" ✓ = Record counts match") @@ -447,6 +513,4 @@ func (s *Scanner) PrintReport(report *ScanReport) { fmt.Println(" S- = Table missing in source") fmt.Println(" T- = Table missing in target") fmt.Println() - - fmt.Println(strings.Repeat("=", 80)) } diff --git a/cmd/tenant-migration/migrator/table_meta.go b/cmd/tenant-migration/migrator/table_meta.go index 29172fbfc..afdc3c087 100644 --- a/cmd/tenant-migration/migrator/table_meta.go +++ b/cmd/tenant-migration/migrator/table_meta.go @@ -14,10 +14,12 @@ package migrator // TableMeta defines the migration metadata for a table type TableMeta struct { - Name string // Table name - IDColumn string // ID column name (default "id") - ForeignKeys map[string]string // Foreign key column -> referenced table name - HasBizID bool // Whether the table has biz_id column for filtering + Name string // Table name + IDColumn string // ID column name (default "id") + ForeignKeys map[string]string // Foreign key column -> referenced table name + JSONArrayFKs map[string]string // JSON array column -> referenced table name (e.g. template_ids -> templates) + OptionalFKs map[string]bool // FK columns that allow soft failure: generate virtual ID when mapping not found + HasBizID bool // Whether the table has biz_id column for filtering } // TableMetas defines metadata for all 29 core tables @@ -86,6 +88,10 @@ var TableMetas = map[string]TableMeta{ ForeignKeys: map[string]string{ "template_space_id": "template_spaces", }, + JSONArrayFKs: map[string]string{ + "template_ids": "templates", + "bound_apps": "applications", + }, }, "templates": { Name: "templates", @@ -134,6 +140,10 @@ var TableMetas = map[string]TableMeta{ ForeignKeys: map[string]string{ "app_id": "applications", "config_item_id": "config_items", + "content_id": "contents", + }, + OptionalFKs: map[string]bool{ + "config_item_id": true, }, }, "contents": { @@ -144,6 +154,9 @@ var TableMetas = map[string]TableMeta{ "app_id": "applications", "config_item_id": "config_items", }, + OptionalFKs: map[string]bool{ + "config_item_id": true, + }, }, "strategies": { Name: "strategies", @@ -183,6 +196,10 @@ var TableMetas = map[string]TableMeta{ "release_id": "releases", "commit_id": "commits", "config_item_id": "config_items", + "content_id": "contents", + }, + OptionalFKs: map[string]bool{ + "config_item_id": true, }, }, "released_groups": { @@ -232,6 +249,13 @@ var TableMetas = map[string]TableMeta{ ForeignKeys: map[string]string{ "app_id": "applications", }, + JSONArrayFKs: map[string]string{ + "template_space_ids": "template_spaces", + "template_set_ids": "template_sets", + "template_ids": "templates", + "template_revision_ids": "template_revisions", + "latest_template_ids": "templates", + }, }, "app_template_variables": { Name: "app_template_variables", @@ -246,8 +270,12 @@ var TableMetas = map[string]TableMeta{ IDColumn: "id", HasBizID: true, ForeignKeys: map[string]string{ - "app_id": "applications", - "release_id": "releases", + "app_id": "applications", + "release_id": "releases", + "template_space_id": "template_spaces", + "template_set_id": "template_sets", + "template_id": "templates", + "template_revision_id": "template_revisions", }, }, "released_app_template_variables": { @@ -261,6 +289,23 @@ var TableMetas = map[string]TableMeta{ }, } +// RuntimeCleanupTables returns tables that are skipped during migration but should +// still be cleaned during cleanup. These tables accumulate runtime data (audits, +// events, client records, etc.) in the target environment after migration. +func RuntimeCleanupTables() []string { + return []string{ + "audits", + "events", + "published_strategy_histories", + "current_released_instances", + "clients", + "client_events", + "client_querys", + "archived_apps", + "biz_hosts", + } +} + // TablesInCleanupOrder returns tables in reverse dependency order (for deletion) // Delete dependent tables first, then base tables func TablesInCleanupOrder() []string { @@ -278,8 +323,8 @@ func TablesInCleanupOrder() []string { "kvs", "current_published_strategies", "strategies", - "contents", "commits", + "contents", // Level 2 "group_app_binds", @@ -319,15 +364,15 @@ func TablesInInsertOrder() []string { "config_items", "releases", "strategy_sets", - "template_sets", "templates", + "template_sets", "hook_revisions", "credential_scopes", "group_app_binds", // Level 3 (depends on Level 1 and Level 2) - "commits", "contents", + "commits", "strategies", "current_published_strategies", "kvs", diff --git a/cmd/tenant-migration/migrator/vault.go b/cmd/tenant-migration/migrator/vault.go index 2f4c5ee9d..1f99825ca 100644 --- a/cmd/tenant-migration/migrator/vault.go +++ b/cmd/tenant-migration/migrator/vault.go @@ -92,6 +92,45 @@ func NewVaultMigrator(cfg *config.Config, sourceDB, targetDB *gorm.DB) (*VaultMi }, nil } +// checkConnectivity verifies that both source and target Vault instances are reachable and the tokens are valid +func (m *VaultMigrator) checkConnectivity() error { + log.Println("Checking Vault connectivity...") + + // Check source Vault + sourceHealth, err := m.sourceVault.Sys().Health() + if err != nil { + return fmt.Errorf("source Vault is unreachable at %s: %w", m.cfg.Source.Vault.Address, err) + } + if sourceHealth.Sealed { + return fmt.Errorf("source Vault at %s is sealed", m.cfg.Source.Vault.Address) + } + + // Verify source token by looking up self + _, err = m.sourceVault.Auth().Token().LookupSelf() + if err != nil { + return fmt.Errorf("source Vault token is invalid: %w", err) + } + log.Printf(" Source Vault OK (%s)", m.cfg.Source.Vault.Address) + + // Check target Vault + targetHealth, err := m.targetVault.Sys().Health() + if err != nil { + return fmt.Errorf("target Vault is unreachable at %s: %w", m.cfg.Target.Vault.Address, err) + } + if targetHealth.Sealed { + return fmt.Errorf("target Vault at %s is sealed", m.cfg.Target.Vault.Address) + } + + // Verify target token by looking up self + _, err = m.targetVault.Auth().Token().LookupSelf() + if err != nil { + return fmt.Errorf("target Vault token is invalid: %w", err) + } + log.Printf(" Target Vault OK (%s)", m.cfg.Target.Vault.Address) + + return nil +} + // SetIDMapper sets the ID mapper from MySQL migration // This must be called before Migrate() when using incremental migration func (m *VaultMigrator) SetIDMapper(mapper *IDMapper) { @@ -100,6 +139,10 @@ func (m *VaultMigrator) SetIDMapper(mapper *IDMapper) { // Migrate performs the Vault KV data migration with ID mapping support func (m *VaultMigrator) Migrate() (*VaultMigrationResult, error) { + if err := m.checkConnectivity(); err != nil { + return nil, fmt.Errorf("vault connectivity check failed: %w", err) + } + startTime := time.Now() result := &VaultMigrationResult{ Success: true, @@ -213,7 +256,7 @@ func (m *VaultMigrator) getKvRecordsBatch(offset, limit int) ([]KvRecord, error) if m.cfg.Migration.HasBizFilter() { query = query.Where("biz_id IN ?", m.cfg.Migration.BizIDs) } - if err := query.Offset(offset).Limit(limit).Find(&records).Error; err != nil { + if err := query.Order("id").Offset(offset).Limit(limit).Find(&records).Error; err != nil { return nil, err } return records, nil @@ -241,7 +284,7 @@ func (m *VaultMigrator) getReleasedKvRecordsBatch(offset, limit int) ([]Released if m.cfg.Migration.HasBizFilter() { query = query.Where("biz_id IN ?", m.cfg.Migration.BizIDs) } - if err := query.Offset(offset).Limit(limit).Find(&records).Error; err != nil { + if err := query.Order("id").Offset(offset).Limit(limit).Find(&records).Error; err != nil { return nil, err } return records, nil @@ -367,10 +410,15 @@ type VaultCleanupResult struct { Success bool } -// CleanupTarget deletes all migrated KV data from target Vault -// Uses source database to get KV records (since target DB may not have data yet) -// If biz_id filter is configured, only deletes KVs for those businesses +// CleanupTarget deletes all migrated KV data from target Vault. +// Reads from TARGET database to get the correct (already-mapped) IDs for Vault path construction. +// If biz_id filter is configured, only deletes KVs for those businesses. +// IMPORTANT: This must be called BEFORE MySQL cleanup, because it relies on target DB records. func (m *VaultMigrator) CleanupTarget() (*VaultCleanupResult, error) { + if err := m.checkConnectivity(); err != nil { + return nil, fmt.Errorf("vault connectivity check failed: %w", err) + } + startTime := time.Now() result := &VaultCleanupResult{ Success: true, @@ -379,18 +427,18 @@ func (m *VaultMigrator) CleanupTarget() (*VaultCleanupResult, error) { ctx, cancel := context.WithTimeout(context.Background(), 30*time.Minute) defer cancel() - // Log biz_id filter info if m.cfg.Migration.HasBizFilter() { log.Printf("Vault cleanup with biz_id filter: %v", m.cfg.Migration.BizIDs) } - // Step 1: Delete unreleased KV data (use source DB to get records) - log.Println("Cleaning up unreleased KV data from target Vault...") batchSize := m.cfg.Migration.BatchSize + + // Step 1: Delete unreleased KV data using target DB records (already have mapped IDs) + log.Println("Cleaning up unreleased KV data from target Vault...") for offset := 0; ; offset += batchSize { - kvRecords, err := m.getKvRecordsBatch(offset, batchSize) + kvRecords, err := m.getTargetKvRecordsBatch(offset, batchSize) if err != nil { - result.Errors = append(result.Errors, fmt.Sprintf("failed to get KV records batch at offset %d: %v", offset, err)) + result.Errors = append(result.Errors, fmt.Sprintf("failed to get target KV records batch at offset %d: %v", offset, err)) result.Success = false result.Duration = time.Since(startTime) return result, err @@ -401,8 +449,9 @@ func (m *VaultMigrator) CleanupTarget() (*VaultCleanupResult, error) { } for _, kv := range kvRecords { - if deleteErr := m.deleteKv(ctx, kv); deleteErr != nil { - result.Errors = append(result.Errors, fmt.Sprintf("failed to delete KV %d: %v", kv.ID, deleteErr)) + targetPath := fmt.Sprintf(kvPath, kv.BizID, kv.AppID, kv.Key) + if deleteErr := m.targetVault.KVv2(MountPath).DeleteMetadata(ctx, targetPath); deleteErr != nil { + result.Errors = append(result.Errors, fmt.Sprintf("failed to delete KV %d at %s: %v", kv.ID, targetPath, deleteErr)) if !m.cfg.Migration.ContinueOnError { result.Success = false result.Duration = time.Since(startTime) @@ -415,12 +464,12 @@ func (m *VaultMigrator) CleanupTarget() (*VaultCleanupResult, error) { } log.Printf(" Deleted %d unreleased KVs", result.DeletedKvs) - // Step 2: Delete released KV data (use source DB to get records) + // Step 2: Delete released KV data using target DB records (already have mapped IDs) log.Println("Cleaning up released KV data from target Vault...") for offset := 0; ; offset += batchSize { - releasedKvRecords, err := m.getReleasedKvRecordsBatch(offset, batchSize) + releasedKvRecords, err := m.getTargetReleasedKvRecordsBatch(offset, batchSize) if err != nil { - result.Errors = append(result.Errors, fmt.Sprintf("failed to get released KV records batch at offset %d: %v", offset, err)) + result.Errors = append(result.Errors, fmt.Sprintf("failed to get target released KV records batch at offset %d: %v", offset, err)) result.Success = false result.Duration = time.Since(startTime) return result, err @@ -431,8 +480,9 @@ func (m *VaultMigrator) CleanupTarget() (*VaultCleanupResult, error) { } for _, rkv := range releasedKvRecords { - if err := m.deleteReleasedKv(ctx, rkv); err != nil { - result.Errors = append(result.Errors, fmt.Sprintf("failed to delete released KV %d: %v", rkv.ID, err)) + targetPath := fmt.Sprintf(releasedKvPath, rkv.BizID, rkv.AppID, rkv.ReleaseID, rkv.Key) + if err := m.targetVault.KVv2(MountPath).DeleteMetadata(ctx, targetPath); err != nil { + result.Errors = append(result.Errors, fmt.Sprintf("failed to delete released KV %d at %s: %v", rkv.ID, targetPath, err)) if !m.cfg.Migration.ContinueOnError { result.Success = false result.Duration = time.Since(startTime) @@ -452,30 +502,215 @@ func (m *VaultMigrator) CleanupTarget() (*VaultCleanupResult, error) { return result, nil } -// deleteKv deletes a single unreleased KV from target Vault -// Uses ID mapping for the target path -func (m *VaultMigrator) deleteKv(ctx context.Context, kv KvRecord) error { - targetPath := m.getTargetKvPath(kv) +// getTargetKvRecordsBatch retrieves a batch of KV records from TARGET MySQL (for cleanup) +func (m *VaultMigrator) getTargetKvRecordsBatch(offset, limit int) ([]KvRecord, error) { + var records []KvRecord + query := m.targetDB.Table("kvs") + if m.cfg.Migration.HasBizFilter() { + query = query.Where("biz_id IN ?", m.cfg.Migration.BizIDs) + } + if err := query.Order("id").Offset(offset).Limit(limit).Find(&records).Error; err != nil { + return nil, err + } + return records, nil +} - // Delete all versions and metadata - err := m.targetVault.KVv2(MountPath).DeleteMetadata(ctx, targetPath) - if err != nil { - return fmt.Errorf("failed to delete from target Vault path %s: %w", targetPath, err) +// getTargetReleasedKvRecordsBatch retrieves a batch of released KV records from TARGET MySQL (for cleanup) +func (m *VaultMigrator) getTargetReleasedKvRecordsBatch(offset, limit int) ([]ReleasedKvRecord, error) { + var records []ReleasedKvRecord + query := m.targetDB.Table("released_kvs") + if m.cfg.Migration.HasBizFilter() { + query = query.Where("biz_id IN ?", m.cfg.Migration.BizIDs) + } + if err := query.Order("id").Offset(offset).Limit(limit).Find(&records).Error; err != nil { + return nil, err } + return records, nil +} - return nil +// VaultDirectScanResult contains results from probing Vault based on DB records +type VaultDirectScanResult struct { + SourceBizScans []BizVaultScan + TargetBizScans []BizVaultScan + Duration time.Duration + Errors []string } -// deleteReleasedKv deletes a single released KV from target Vault -// Uses ID mapping for the target path -func (m *VaultMigrator) deleteReleasedKv(ctx context.Context, rkv ReleasedKvRecord) error { - targetPath := m.getTargetReleasedKvPath(rkv) +// BizVaultScan contains per-biz DB record count vs Vault existence count +type BizVaultScan struct { + BizID uint32 + KvDB int64 + KvVault int64 + RKvDB int64 + RKvVault int64 +} - // Delete all versions and metadata - err := m.targetVault.KVv2(MountPath).DeleteMetadata(ctx, targetPath) - if err != nil { - return fmt.Errorf("failed to delete from target Vault path %s: %w", targetPath, err) +// TotalKvDB returns total unreleased KV DB count across all biz scans +func totalKvDB(scans []BizVaultScan) int64 { + var n int64 + for _, s := range scans { + n += s.KvDB + } + return n +} + +// TotalKvVault returns total unreleased KV Vault found count across all biz scans +func totalKvVault(scans []BizVaultScan) int64 { + var n int64 + for _, s := range scans { + n += s.KvVault } + return n +} - return nil +// TotalRKvDB returns total released KV DB count across all biz scans +func totalRKvDB(scans []BizVaultScan) int64 { + var n int64 + for _, s := range scans { + n += s.RKvDB + } + return n +} + +// TotalRKvVault returns total released KV Vault found count across all biz scans +func totalRKvVault(scans []BizVaultScan) int64 { + var n int64 + for _, s := range scans { + n += s.RKvVault + } + return n +} + +// ScanVault reads KV/released KV records from both source and target DB, +// then probes the corresponding Vault to check if each secret actually exists. +func (m *VaultMigrator) ScanVault() (*VaultDirectScanResult, error) { + if err := m.checkConnectivity(); err != nil { + return nil, fmt.Errorf("vault connectivity check failed: %w", err) + } + + startTime := time.Now() + result := &VaultDirectScanResult{} + ctx := context.Background() + batchSize := m.cfg.Migration.BatchSize + + // Source: read source DB records, probe source Vault + log.Println(" Scanning source Vault (based on source DB records)...") + sourceBizMap := make(map[uint32]*BizVaultScan) + + for offset := 0; ; offset += batchSize { + records, err := m.getKvRecordsBatch(offset, batchSize) + if err != nil { + result.Errors = append(result.Errors, fmt.Sprintf("failed to get source KV batch: %v", err)) + break + } + if len(records) == 0 { + break + } + for _, kv := range records { + biz := getOrCreateBizScan(sourceBizMap, kv.BizID) + biz.KvDB++ + path := m.getSourceKvPath(kv) + if _, err := m.sourceVault.KVv2(MountPath).Get(ctx, path); err == nil { + biz.KvVault++ + } + } + } + + for offset := 0; ; offset += batchSize { + records, err := m.getReleasedKvRecordsBatch(offset, batchSize) + if err != nil { + result.Errors = append(result.Errors, fmt.Sprintf("failed to get source released KV batch: %v", err)) + break + } + if len(records) == 0 { + break + } + for _, rkv := range records { + biz := getOrCreateBizScan(sourceBizMap, rkv.BizID) + biz.RKvDB++ + path := m.getSourceReleasedKvPath(rkv) + if _, err := m.sourceVault.KVv2(MountPath).Get(ctx, path); err == nil { + biz.RKvVault++ + } + } + } + + result.SourceBizScans = bizMapToSortedSlice(sourceBizMap) + log.Printf(" Source: KVs %d/%d, Released KVs %d/%d found in Vault", + totalKvVault(result.SourceBizScans), totalKvDB(result.SourceBizScans), + totalRKvVault(result.SourceBizScans), totalRKvDB(result.SourceBizScans)) + + // Target: read target DB records, probe target Vault + log.Println(" Scanning target Vault (based on target DB records)...") + targetBizMap := make(map[uint32]*BizVaultScan) + + for offset := 0; ; offset += batchSize { + records, err := m.getTargetKvRecordsBatch(offset, batchSize) + if err != nil { + result.Errors = append(result.Errors, fmt.Sprintf("failed to get target KV batch: %v", err)) + break + } + if len(records) == 0 { + break + } + for _, kv := range records { + biz := getOrCreateBizScan(targetBizMap, kv.BizID) + biz.KvDB++ + path := fmt.Sprintf(kvPath, kv.BizID, kv.AppID, kv.Key) + if _, err := m.targetVault.KVv2(MountPath).Get(ctx, path); err == nil { + biz.KvVault++ + } + } + } + + for offset := 0; ; offset += batchSize { + records, err := m.getTargetReleasedKvRecordsBatch(offset, batchSize) + if err != nil { + result.Errors = append(result.Errors, fmt.Sprintf("failed to get target released KV batch: %v", err)) + break + } + if len(records) == 0 { + break + } + for _, rkv := range records { + biz := getOrCreateBizScan(targetBizMap, rkv.BizID) + biz.RKvDB++ + path := fmt.Sprintf(releasedKvPath, rkv.BizID, rkv.AppID, rkv.ReleaseID, rkv.Key) + if _, err := m.targetVault.KVv2(MountPath).Get(ctx, path); err == nil { + biz.RKvVault++ + } + } + } + + result.TargetBizScans = bizMapToSortedSlice(targetBizMap) + log.Printf(" Target: KVs %d/%d, Released KVs %d/%d found in Vault", + totalKvVault(result.TargetBizScans), totalKvDB(result.TargetBizScans), + totalRKvVault(result.TargetBizScans), totalRKvDB(result.TargetBizScans)) + + result.Duration = time.Since(startTime) + return result, nil +} + +func getOrCreateBizScan(m map[uint32]*BizVaultScan, bizID uint32) *BizVaultScan { + if s, ok := m[bizID]; ok { + return s + } + s := &BizVaultScan{BizID: bizID} + m[bizID] = s + return s +} + +func bizMapToSortedSlice(m map[uint32]*BizVaultScan) []BizVaultScan { + scans := make([]BizVaultScan, 0, len(m)) + for _, s := range m { + scans = append(scans, *s) + } + for i := 0; i < len(scans)-1; i++ { + for j := i + 1; j < len(scans); j++ { + if scans[i].BizID > scans[j].BizID { + scans[i], scans[j] = scans[j], scans[i] + } + } + } + return scans }