MongoDB分片迁移原理与源码(4)

张学伟 2020-04-04

架构 MongoDB 源码 集群 metadata

MongoDB分片迁移原理与源码
异步删除数据

在from shard将迁移结果提交到config服务器成功后,from shard就会执行删除原数据的操作;如果迁移的参数"waitForDelete"为false,则触发异步删除。"waitForDelete"的默认参数即是false,即异步删除是默认设计。

将此次迁移的数据范围调用cleanUpRange()函数进行后续处理。

默认情况下,900s 以后开始清理 chunks 的数据,每次清理 128 个文档,每隔 20ms 删除一次。具体通过以下参数设置:

rangeDeleterBatchDelayMS: 删除每个 chunk 数据的时候分批次删除,每批之间间隔的时间,单位 ms,默认 20ms;

internalQueryExecYieldIterations: 默认为 128;

rangeDeleterBatchSize:每次删除数据的数量,默认即为0;为0时 ,则每次删除的数量为max(internalQueryExecYieldIterations,1),

orphanCleanupDelaySecs: moveChunk 以后延迟删除数据的时间,单位 s ,默认 900 s

const ChunkRange range(_args.getMinKey(), _args.getMaxKey());

auto notification = [&] {
auto const whenToClean = _args.getWaitForDelete() ? CollectionShardingRuntime::kNow

                                                : CollectionShardingRuntime::kDelayed;

UninterruptibleLockGuard noInterrupt(opCtx->lockState());
AutoGetCollection autoColl(opCtx, getNss(), MODE_IS);
return CollectionShardingRuntime::get(opCtx, getNss())->cleanUpRange(range, whenToClean);
}();

// 默认的异步删除时间
//MONGO_EXPORT_SERVER_PARAMETER(orphanCleanupDelaySecs, int, 900); // 900s = 15m
auto CollectionShardingRuntime::cleanUpRange(ChunkRange const& range, CleanWhen when)
-> CleanupNotification {
Date_t time = (when == kNow) ? Date_t{} : Date_t::now() +

       stdx::chrono::seconds{orphanCleanupDelaySecs.load()};

return _metadataManager->cleanUpRange(range, time);
}

再删除之前,还要判断是否满足没有任何基于该chunk的查询了:如果没有则放到删除队列中,等删除时间到了;如果还有查询,则放到另外一个孤儿文档队列,后续再删除;

auto MetadataManager::cleanUpRange(ChunkRange const& range, Date_t whenToDelete)
-> CleanupNotification {
stdx::lock_guard lg(_managerLock);
invariant(!_metadata.empty());

auto* const activeMetadata = _metadata.back().get();
auto* const overlapMetadata = _findNewestOverlappingMetadata(lg, range);

if (overlapMetadata == activeMetadata) {

   return Status{ErrorCodes::RangeOverlapConflict,
                 str::stream() << "Requested deletion range overlaps a live shard chunk"};

}

if (rangeMapOverlaps(_receivingChunks, range.getMin(), range.getMax())) {

   return Status{ErrorCodes::RangeOverlapConflict,
                 str::stream() << "Requested deletion range overlaps a chunk being"
                                  " migrated in"};

}

if (!overlapMetadata) {

   //如果没有基于该chunk的查询了,则把该数据块放到删除队列中.
   const auto whenStr = (whenToDelete == Date_t{}) ? "immediate"_sd : "deferred"_sd;
   log() << "Scheduling " << whenStr << " deletion of " << _nss.ns() << " range "
         << redact(range.toString());
   return _pushRangeToClean(lg, range, whenToDelete);

}

log() << "Deletion of " << _nss.ns() << " range " << redact(range.toString())

     << " will be scheduled after all possibly dependent queries finish";

//如果还有查询,则放到孤儿文档的队列中,后续再删除.
auto& orphans = overlapMetadata->orphans;
orphans.emplace_back(ChunkRange(range.getMin().getOwned(), range.getMax().getOwned()),

                    whenToDelete);

return orphans.back().notification;
}

根据删除时间,则定是否放到最终的异步删除的任务线程中scheduleCleanup()

auto MetadataManager::_pushRangeToClean(WithLock lock, ChunkRange const& range, Date_t when)
-> CleanupNotification {
std::list ranges;
ranges.emplace_back(ChunkRange(range.getMin().getOwned(), range.getMax().getOwned()), when);
auto& notifn = ranges.back().notification;
_pushListToClean(lock, std::move(ranges));
return notifn;
}

void MetadataManager::_pushListToClean(WithLock, std::list ranges) {
auto when = _rangesToClean.add(std::move(ranges));
if (when) {

   scheduleCleanup(
       _executor, _nss, _metadata.back()->metadata.getCollVersion().epoch(), *when);

}
invariant(ranges.empty());
}

void scheduleCleanup(executor::TaskExecutor* executor,

                NamespaceString nss,
                OID epoch,
                Date_t when) {

LOG(1) << "Scheduling cleanup on " << nss.ns() << " at " << when;
auto swCallbackHandle = executor->scheduleWorkAt(

   when, [ executor, nss = std::move(nss), epoch = std::move(epoch) ](auto&) {
       Client::initThreadIfNotAlready("Collection Range Deleter");
       auto uniqueOpCtx = Client::getCurrent()->makeOperationContext();
       auto opCtx = uniqueOpCtx.get();

       const int maxToDelete = std::max(int(internalQueryExecYieldIterations.load()), 1);

       MONGO_FAIL_POINT_PAUSE_WHILE_SET(suspendRangeDeletion);

       //执行真正的删除,但是每批只删除maxToDelete(默认128)个文档;每批间隔时间默认为rangeDeleterBatchDelayMS(20)毫秒。
       //最终删除调用的是collection->deleteDocument()删除集合文档的接口,完成文档删除
       auto next = CollectionRangeDeleter::cleanUpNextRange(opCtx, nss, epoch, maxToDelete);
       if (next) {
           scheduleCleanup(executor, std::move(nss), std::move(epoch), *next);
      }
  });

if (!swCallbackHandle.isOK()) {

   log() << "Failed to schedule the orphan data cleanup task"
         << causedBy(redact(swCallbackHandle.getStatus()));

}
}

问题
孤儿文档(orphaned document)

在第一章已经说过,由于数据块的迁移不是原子操作,导致从拷贝数据到异步删除数据,中间任何地方出错,都会导致产生孤儿文档。

孤儿文档会造成数据的不一致,甚至一个数据块迁移了一部分然后被打断,后续相同的数据块重新迁移的时候,有可能造成迁移始终不成功的问题。
4.0 版本中迁移触发的阈值太低,导致迁移产生的性能问题太高

该问题主要从参考文献中得出来的结论。详情可参考《MongoDB疑难解析:为什么升级之后负载升高了》

除此之外,由于整个迁移不是原子的,且存在异步过程,导致中间失败,产生其他问题的可能。
总结

MongoDB基于分片集群架构,实现了存储能力和服务能力的水平扩展,实现了管理海量数据的能力;并且基于自身架构的特点和优势,解决了如下问题:

可靠性。各个shard和config server基于副本集架构,实现了数据冗余和容错能力;

可用性。除了副本集架构的可用性的提高,一个shard出问题也不影响其他分片,以及整个分片集群继续服务的能力;

一致性。用户通过哪个mongos访问分片集群,都可以获得正确的数据;

伸缩性。非常方便的实现了增加和删除分片的功能,极为方便的实现了水平扩容;

性能。整个集群的服务分摊到了各个shard上,而且基于动态均衡,实现了性能的最大化。

综上,MongoDB的分片集群,还挺好。
参考文档

MongoDB官方文档 

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

MongoDB疑难解析:为什么升级之后负载升高了? 

由数据迁移至MongoDB导致的数据不一致问题及解决方案
登录 后评论
下一篇
云栖号资讯小编
5828人浏览
2020-05-11
相关推荐
MongoDB Sharded cluster架构原理
35812人浏览
2016-04-21 19:24:32
mongodb分片扩展架构
1615人浏览
2018-03-13 21:58:00
MongoDB 分片集群技术
2079人浏览
2018-01-18 14:51:58
MongoShake全量迁移功能
1476人浏览
2019-06-01 20:33:33
MongoDB分片集群
1384人浏览
2019-09-22 19:26:09
0
0
0
298