孤儿文档是怎样产生的(MongoDB orphaned document)

　　使用MongoDB的开发人员应该都听说过孤儿文档（orphaned document）这回事儿，可谓闻着沉默，遇者流泪。本文基于MongoDB3.0来看看怎么产生一个orphaned document，要求MongoDB的运行方式需要是sharded cluster，如果对这一部分还不是很了解，可以参考一下这篇文章。

　　在MongoDB的官方文档中，对orphaned document的描述非常简单：

　　In a sharded cluster, orphaned documents are those documents on a shard that also exist in chunks on other shards as a result of failed migrations or incomplete migration cleanup due to abnormal shutdown. Delete orphaned documents using cleanupOrphaned to reclaim disk space and reduce confusion

　　可以看到，orphaned document是指在sharded cluster环境下，一些同时存在于不同shard上的document。我们知道，在mongodb sharded cluster中，分布在不同shard的数据子集是正交的，即理论上一个document只能出现在一个shard上，document与shard的映射关系维护在config server中。官方文档指出了可能产生orphaned document的情况：在chunk迁移的过程中，mongod实例异常宕机，导致迁移过程失败或者部分完成。文档中还指出，可以使用 cleanupOrphaned 来删除orphaned document。

　　新闻报道灾难、事故的时候，一般都有这么一个潜规则：内容越短，事情也严重。不知道MongoDB对于orphaned document是不是也采用了这个套路，一来对orphaned document发生的可能原因描述不够详尽，二来也没有提供检测是否存在orphaned document的方法。对于cleanupOrphaned，要在生产环境使用也需要一定的勇气。

　　作为一个没有看过MongoDB源码的普通应用开发人员，拍脑袋想想，chuck的迁移应该有以下三个步骤：将数据从源shard拷贝到目标shard，更新config server中的metadata，从源shard删除数据。当然，这三个步骤的顺序不一定是上面的顺序。这三个步骤，如果能保证原子性，那么理论上是不会出问题的。但是，orphaned document具体怎么出现的一直不是很清楚。

　　前些天在浏览官方文档的时候，发现有对迁移过程描述（chunk-migration-procedure），大致过程翻译如下：

　　balancer向源shard发送moveChunk命令
　　源shard内部执行moveChunk命令，并保证在迁移的过程中，新插入的document还是写入源shard
　　如果需要的话，目标shard创建需要的索引
　　目标shard从源shard请求数据；注意，这里是一个copy操作，而不是move操作
　　在接收完chunk的最后一个文档后，目标shard启动一个同步拷贝进程，保证拷贝到在迁移过程中又写入源shard上的相关文档
　　完全同步之后，目标shard向config server报告新的metadata（chunk的新位置信息）
　　在上一步完成之后，源shard开始删除旧的document

　　如果能保证以上操作的原子性，在任何步骤出问题应该都没问题；如果不能保证，那么在第4，5，6，7步出现机器宕机，都有可能出问题。对于出问题的原因，官网（chunk-migration-queuing ）是这么解释的：

　　the balancer does not wait for the current migration’s delete phase to complete before starting the next chunk migration

　　This queuing behavior allows shards to unload chunks more quickly in cases of heavily imbalanced cluster, such as when performing initial data loads without pre-splitting and when adding new shards.

　　If multiple delete phases are queued but not yet complete, a crash of the replica set’s primary can orphan data from multiple migrations.

　　简而言之，为了加速chunk 迁移的速度（比如在新的shard加入的时候，有大量的chunk迁移），因此delete phase（第7步）不会立刻执行，而是放入一个队列，异步执行，此时如果crash，就可能产生孤儿文档

产生一个orphaned document

　　基于官方文档，如何产生一个orphaned document呢？我的想法很简单：监控MongoDB日志，在出现标志迁移过程的日志出现的时候，kill掉shard中的primary！

预备知识

　　在《通过一步步创建sharded cluster来认识mongodb》一文中，我详细介绍了如何搭建一个sharded cluster，在我的实例中，使用了两个shard，其中每个shard包括一个primary、一个secondary，一个arbiter。另外，创建了一个允许sharding的db -- test_db, 然后sharded_col这个集合使用_id分片，本文基于这个sharded cluster进行实验。但需要注意的是：在前文中，为了节省磁盘空间，我禁用了mongod实例的journal机制(启动选项中的 --nojourbal)，但在本文中，为了尽量符合真实情况，在启动mongod的时候使用了--journal来启用journal机制。

　　另外，再补充两点，第一个是chunk迁移的条件，只有当shards之间chunk的数目差异达到一定程度才会发生迁移：

Number of Chunks	Migration Threshold
Fewer than 20	2
20-79	4
80 and greater	8

　　第二个是，如果没有在document中包含_id，那么mongodb会自动添加这个字段，其value是一个ObjectId，ObjectId由一下部分组成：

a 4-byte value representing the seconds since the Unix epoch,
a 3-byte machine identifier,
a 2-byte process id, and
a 3-byte counter, starting with a random value.

　　因此，在没有使用hash sharding key（默认是ranged sharding key）的情况下，在短时间内插入大量没有携带_id字段的ducoment，会插入到同一个shard，这也有利于出现chunk分裂和迁移的情况。

准备

　　首先，得知道chunk迁移的时候日志是什么样子的，因此我用python脚本插入了一些记录，通过sh.status()发现有chunk分裂、迁移的时候去查看mongodb日志，在rs1（sharded_col这个集合的primary shard）的primary(rs1_1.log）里面发现了如下的输出：

 34 2017-07-06T21:43:21.629+0800 I NETWORK  [conn6] starting new replica set monitor for replica set rs2 with seeds 127.0.0.1:27021,127.0.0.1:27022
 48 2017-07-06T21:43:23.685+0800 I SHARDING [conn6] moveChunk data transfer progress: { active: true, ns: "test_db.sharded_col", from: "rs1/127.0.0.1:27018,127.0.0.1:27019"    , min: { _id: ObjectId('595e3e74d71ffd5c7be8c8b7') }, max: { _id: MaxKey }, shardKeyPattern: { _id: 1.0 }, state: "steady", counts: { cloned: 1, clonedBytes: 83944, cat    chup: 0, steady: 0 }, ok: 1.0, $gleStats: { lastOpTime: Timestamp 0|0, electionId: ObjectId('595e3b0ff70a0e5c3d75d684') } } my mem used: 0
 52 -017-07-06T21:43:23.977+0800 I SHARDING [conn6] moveChunk migrate commit accepted by TO-shard: { active: false, ns: "test_db.sharded_col", from: "rs1/127.0.0.1:27018,12    7.0.0.1:27019", min: { _id: ObjectId('595e3e74d71ffd5c7be8c8b7') }, max: { _id: MaxKey }, shardKeyPattern: { _id: 1.0 }, state: "done", counts: { cloned: 1, clonedBytes    : 83944, catchup: 0, steady: 0 }, ok: 1.0, $gleStats: { lastOpTime: Timestamp 0|0, electionId: ObjectId('595e3b0ff70a0e5c3d75d684') } }
 53 w017-07-06T21:43:23.977+0800 I SHARDING [conn6] moveChunk updating self version to: 3|1||590a8d4cd2575f23f5d0c9f3 through { _id: ObjectId('5937e11f48e2c04f793b1242') }     -> { _id: ObjectId('595b829fd71ffd546f9e5b05') } for collection 'test_db.sharded_col'
 54 2017-07-06T21:43:23.977+0800 I NETWORK  [conn6] SyncClusterConnection connecting to [127.0.0.1:40000]
 55 2017-07-06T21:43:23.978+0800 I NETWORK  [conn6] SyncClusterConnection connecting to [127.0.0.1:40001]
 56 2017-07-06T21:43:23.978+0800 I NETWORK  [conn6] SyncClusterConnection connecting to [127.0.0.1:40002]
 57 2017-07-06T21:43:24.413+0800 I SHARDING [conn6] about to log metadata event: { _id: "xxx-2017-07-06T13:43:24-595e3e7c0db0d72b7244e620", server: "xxx", clientAddr: "127.0.0.1:52312", time: new Date(1499348604413), what: "moveChunk.commit", ns: "test_db.sharded_col", details: { min: { _id: ObjectId(    '595e3e74d71ffd5c7be8c8b7') }, max: { _id: MaxKey }, from: "rs1", to: "rs2", cloned: 1, clonedBytes: 83944, catchup: 0, steady: 0 } }
 58 2017-07-06T21:43:24.417+0800 I SHARDING [conn6] MigrateFromStatus::done About to acquire global lock to exit critical section
 59 2017-07-06T21:43:24.417+0800 I SHARDING [conn6] forking for cleanup of chunk data
 60 2017-07-06T21:43:24.417+0800 I SHARDING [conn6] MigrateFromStatus::done About to acquire global lock to exit critical section
 61 2017-07-06T21:43:24.417+0800 I SHARDING [RangeDeleter] Deleter starting delete for: test_db.sharded_col from { _id: ObjectId('595e3e74d71ffd5c7be8c8b7') } -> { _id: MaxKey }, with opId: 6
 62 2017-07-06T21:43:24.417+0800 I SHARDING [RangeDeleter] rangeDeleter deleted 1 documents for test_db.sharded_col from { _id: ObjectId('595e3e74d71ffd5c7be8c8b7') } -> { _id: MaxKey }

　　上面第59行，“forking for cleanup of chunk data”，看起来是准备删除旧的数据了

　　于是我写了一个shell脚本: 在rs1_1.log日志中出现“forking for cleanup of chunk data”时kill掉rs1_1这个进程，脚本如下：

check_loop()
{
    echo 'checking'
    ret=`grep -c 'forking for cleanup of chunk data' /home/mongo_db/log/rs1_1.log`
    if [ $ret -gt 0 ]; then
         echo "will kill rs1 primary"
         kill -s 9 `ps aux | grep rs1_1 | awk '{print $2}'`
         exit 0
    fi

    ret=`grep -c 'forking for cleanup of chunk data' /home/mongo_db/log/rs2_1.log`
    if [ $ret -gt 0 ]; then
         echo "will kill rs2 primary"
         kill -s 9 `ps aux | grep rs2_1 | awk '{print $2}'`
         exit 0
    fi

    sleep 0.1
    check_loop
}
check_loop

第一次尝试

　第一次尝试就是使用的上面的脚本。

　　首先运行上面的shell脚本，然后另起一个终端开始插入数据，在shell脚本kill掉进程之后，立即登上rs1和rs2查看统计数据，发现并没有产生orphaned document（怎么检测看第二次尝试）

　　再回看前面的日志，几乎是出现“forking for cleanup of chunk data”的同一时刻就出现了“rangeDeleter deleted 1 documents for test_db.sharded_col from”，后者表明数据已经被删除。而shell脚本0.1s才检查一次，很可能在迁移过程已经完成之后才发出kill信号。于是将kill的时机提前，在shell脚本中检查“moveChunk migrate commit accepted”（上述文档中的第52行）

　　对shell脚本的修改也很简单，替换一下grep的内容：

check_loop()
{
    echo 'checking'
    ret=`grep -c 'moveChunk migrate commit accepted' /home/mongo_db/log/rs1_1.log`
    if [ $ret -gt 0 ]; then
         echo "will kill rs1 primary"
         kill -s 9 `ps aux | grep rs1_1 | awk '{print $2}'`
         exit 0
    fi

    ret=`grep -c 'moveChunk migrate commit accepted' /home/mongo_db/log/rs2_1.log`
    if [ $ret -gt 0 ]; then
         echo "will kill rs2 primary"
         kill -s 9 `ps aux | grep rs2_1 | awk '{print $2}'`
         exit 0
    fi

    sleep 0.1
    check_loop
}
check_loop

View Code