24. MongoDB 数据迁移与同步实战

# 逻辑备份：全库导出为 BSON（gzip 压缩）
mongodump \
  --uri="mongodb://backup:pass@mongodb-primary:27017/admin?replicaSet=rs0" \
  --out=/backup/dump-20260927 \
  --gzip \
  --numParallelCollections=4

# 仅导出指定数据库/集合
mongodump --uri="$MONGO_URI" --db=shop --collection=orders --gzip --out=/backup/orders-only

# 归档模式：输出单文件，便于管道传输
mongodump --uri="$MONGO_URI" --gzip --archive=/backup/mongo.archive.gz

# 带 oplog 的备份：可以恢复到最后一条 oplog 条目的时间点
mongodump \
  --uri="$MONGO_URI" \
  --oplog \
  --gzip \
  --out=/backup/dump-with-oplog

# 恢复整个 dump 目录
mongorestore --uri="mongodb://target:27017" --gzip --drop /backup/dump-20260927

# 恢复单集合
mongorestore --uri="$TARGET_URI" --gzip --db=shop --collection=orders /backup/dump-20260927/shop/orders.bson.gz

# 归档 + oplog 回放到指定时间点
mongorestore --uri="$TARGET_URI" --gzip --archive=/backup/mongo.archive.gz
mongorestore --uri="$TARGET_URI" --oplogReplay --gzip /backup/dump-with-oplog

// 触发一致性快照前，锁定写入
db.fsyncLock()

// 此时复制 /data/db 目录（示例：tar 打包）
// tar -czf /backup/physical.tgz /data/db

// 解锁
db.fsyncUnlock()

# 启动 mongosync：source 为源端，cluster0 为同步主节点
mongosync \
  --cluster0 mongodb://admin:pass@source-rs:27017/admin?replicaSet=src \
  --cluster1 mongodb://admin:pass@target-rs:27017/admin?replicaSet=dst \
  --logPath /var/log/mongosync

# 查询同步进度
mongosync --cluster0 "$SRC_URI" --cluster1 "$DST_URI" --progress

# 查询当前状态（开始/运行/提交）
curl -s http://localhost:27182/api/v1/status

# 以 Docker 运行 mongosync（示例）
docker run -d \
  --name mongosync \
  -p 27182:27182 \
  -v /data/mongosync:/var/log/mongosync \
  registry.mongodb.com/mongosync/mongosync \
  mongosync --cluster0 "$SRC_URI" --cluster1 "$DST_URI"

# 查看 API 端点与版本
curl -s http://localhost:27182/api/v1/status | python3 -m json.tool
curl -s http://localhost:27182/api/v1/version

// 查看一条 oplog 条目
db.getSiblingDB("local").oplog.rs.find().sort({ $natural: -1 }).limit(1).toArray()
// [
//   {
//     op: "i",                    // i=insert, u=update, d=delete, c=command, n=noop
//     ns: "shop.orders",          // 操作的命名空间
//     ui: UUID("..."),            // 集合 UUID
//     ts: Timestamp(1785400000, 1),  // 逻辑时间戳（秒 + 序号）
//     o: { _id: ObjectId("..."), amount: 120 },   // 操作内容
//     o2: { _id: ObjectId("...") }               // 更新条件的过滤键
//   }
// ]

// 手工模拟增量回放（概念演示）：读取并重放最新操作
const cursor = db.getSiblingDB("local").oplog.rs.find({
  ts: { $gt: Timestamp(1785390000, 1) }
}).sort({ ts: 1 }).addOption(2 /* tailable */)

cursor.forEach(entry => {
  // 依据 entry.op 分发到目标集合
  // "i" -> insertOne, "u" -> updateOne, "d" -> deleteOne
})

# 停机窗口内的完整操作序列（示意）
# 1) 停应用 -> 2) mongodump --oplog -> 3) mongorestore -> 4) 校验 -> 5) 切流量

# 手动迁移到 Atlas：mongodump + mongorestore（一次性的简单方案）
mongodump --uri="$SELF_HOSTED_URI" --gzip --archive | \
mongorestore --uri="$ATLAS_SRV" --gzip --archive

# 在线迁移：在 Atlas 控制台创建 Live Migration 任务后，
# 源端开启如下权限（示例）：
# - 具有 root 或 readAnyDatabase + 相应角色的备份用户

// 第 1 层：文档计数
const src = Mongo("mongodb://src:27017").getDB("shop")
const dst = Mongo("mongodb://dst:27017").getDB("shop")
print("src.orders =", src.orders.countDocuments({}))
print("dst.orders =", dst.orders.countDocuments({}))

// 第 2 层：按 _id 分桶计算文档哈希指纹
// 对每个分桶：聚合 $bsonSize 或字段拼接后哈希，对比两侧结果
src.orders.aggregate([
  { $group: { _id: { $bucket: { groupBy: "$_id", boundaries: [0, 100000, 200000, 300000] } },
              hash: { $accumulator: { init: () => "", accumulate: (acc, doc) => acc + JSON.stringify(doc), accumulateArgs: ["$$CURRENT"], merge: (a, b) => a + b, finalize: acc => md5(acc) } } } }
])
// 实际项目更推荐：导出样本 + 逐条对比，或使用三方校验工具

// 校验脚本：两侧按 _id 分桶，输出每个桶的文档数与内容摘要
const srcDB = Mongo("mongodb://src:27017").getDB("shop")
const dstDB = Mongo("mongodb://dst:27017").getDB("shop")

const MIN = ObjectId("000000000000000000000000")
const MAX = ObjectId("ffffffffffffffffffffffff")

for (let i = 0; i < 16; i++) {
  const lo = i * 0x1000000000000000
  const hi = (i + 1) * 0x1000000000000000
  const lower = { _id: { $gte: MIN } }
  const upper = { _id: { $lt: MAX } }
  // 简化示意：实际使用 _id 范围过滤（$lt/$gt）
  const srcCount = srcDB.orders.countDocuments({ _id: { $gte: new ObjectId(lo.toString(16).padStart(24, "0")), $lt: new ObjectId(hi.toString(16).padStart(24, "0")) } })
  const dstCount = dstDB.orders.countDocuments({ _id: { $gte: new ObjectId(lo.toString(16).padStart(24, "0")), $lt: new ObjectId(hi.toString(16).padStart(24, "0")) } })
  print(`bucket-${i}: src=${srcCount} dst=${dstCount} ${srcCount === dstCount ? "OK" : "MISMATCH"}`)
}

// 第 3 层：索引与集合选项对比
// 源端导出索引
src.orders.getIndexes().forEach(i => printjson(i))
// 目标端应完全一致（名称、key、options）
db.orders.getIndexes().forEach(i => printjson(i))

// 校验集合物理完整性
db.orders.validate({ full: true })
// { valid: true, nInvalidDocuments: 0, ... }

// 回滚场景 1：切换前发现数据不一致
// 动作：中止 mongosync commit，停止向目标端写入，恢复应用连接旧集群
// mongosync --cluster0 "$SRC" --cluster1 "$DST" --abort

// 回滚场景 2：切换后应用异常
// 动作：应用连接串切回旧集群，旧集群作为唯一真相源

# mongosync 常用控制命令汇总
mongosync --cluster0 "$SRC_URI" --cluster1 "$DST_URI" --progress    # 查询进度
mongosync --cluster0 "$SRC_URI" --cluster1 "$DST_URI" --commit      # 提交/切换
mongosync --cluster0 "$SRC_URI" --cluster1 "$DST_URI" --abort       # 中止并清理