shardCollection源码解析

// 设置请求参数，主要设置分片键、是否唯一索引和初始chunk数量
ShardsvrShardCollection shardsvrShardCollectionRequest;
...
// 发送给主分片执行shardsvrShardCollection命令
auto cmdResponse = uassertStatusOK(primaryShard>runCommandWithFixedRetryAttempts(opCtx,ReadPreferenceSetting(ReadPreference::PrimaryOnly),"admin", CommandHelpers::appendMajorityWriteConcern(CommandHelpers::appendPassthroughFields(cmdObj, shardsvrShardCollectionRequest.toBSON())), Shard::RetryPolicy::kIdempotent));

// 4.0版本的主分片都支持ShardsvrShardCollection命令，始终都走第一个分支的逻辑，进行返回参数判断并更新路由版本信息
if (cmdResponse.commandStatus != ErrorCodes::CommandNotFound) {
    uassertStatusOK(cmdResponse.commandStatus);
    ...
    // 更新并获取路由版本信息，如果返回的chunk manager不为空，说明创建分片表成功
    auto routingInfo = uassertStatusOK(catalogCache->getCollectionRoutingInfoWithRefresh(opCtx, nss));
    uassert(ErrorCodes::ConflictingOperationInProgress, "Collection was successfully written as sharded but got dropped before it ", "could be evenly distributed", routingInfo.cm());

    return true;
} else {
  ...
}

       // Take the collection critical section so that no writes can happen.
        CollectionCriticalSection critSec(opCtx, nss);
    ...
    
        // 禁止该集合任何读写操作
        critSec.enterCommitPhase();
    ...

        // The initial chunks are distributed evenly across shards if the initial split points were
        // specified in the request by mapReduce or if we are using a hashed shard key. Otherwise,
        // all the initial chunks are placed on the primary shard.
        const bool fromMapReduce = bool(request.getInitialSplitPoints());
        const int numContiguousChunksPerShard = initialSplitPoints.empty()
            ? 1
            : (finalSplitPoints.size() + 1) / (initialSplitPoints.size() + 1);

        // Step 6. Actually shard the collection.
        shardCollection(opCtx,
                        nss,
                        uuid,
                        shardKeyPattern,
                        *request.getCollation(),
                        request.getUnique(),
                        finalSplitPoints,
                        tags,
                        fromMapReduce,
                        ShardingState::get(opCtx)->shardId(),
                        numContiguousChunksPerShard);

void createCollectionOrValidateExisting(OperationContext* opCtx,
                                        const NamespaceString& nss,
                                        const BSONObj& proposedKey,
                                        const ShardKeyPattern& shardKeyPattern,
                                        const ShardsvrShardCollection& request) {
    DBDirectClient localClient(opCtx);
    std::list<BSONObj> indexes = localClient.getIndexSpecs(nss.ns());
    
    // 1. 除_id之外的唯一索引必须以分片键索引作为前缀
    for (const auto& idx : indexes) {
        BSONObj currentKey = idx["key"].embeddedObject();
        bool isUnique = idx["unique"].trueValue();
        uassert(ErrorCodes::InvalidOptions,
                str::stream() << "can't shard collection '" << nss.ns() << "' with unique index on "
                              << currentKey
                              << " and proposed shard key "
                              << proposedKey
                              << ". Uniqueness can't be maintained unless shard key is a prefix",
                !isUnique || shardKeyPattern.isUniqueIndexCompatible(currentKey));
    }

    // 2. 检查是否有可用索引
    bool hasUsefulIndexForKey = false;
    for (const auto& idx : indexes) {
        BSONObj currentKey = idx["key"].embeddedObject();
        // Check 2.i. and 2.ii.
        if (!idx["sparse"].trueValue() && idx["filter"].eoo() && idx["collation"].eoo() &&
            proposedKey.isPrefixOf(currentKey, SimpleBSONElementComparator::kInstance)) {
        // 哈希索引必须指定seed
            uassert(ErrorCodes::InvalidOptions,
                    str::stream() << "can't shard collection " << nss.ns()
                                  << " with hashed shard key "
                                  << proposedKey
                                  << " because the hashed index uses a non-default seed of "
                                  << idx["seed"].numberInt(),
                    !shardKeyPattern.isHashedPattern() || idx["seed"].eoo() ||
                        idx["seed"].numberInt() == BSONElementHasher::DEFAULT_HASH_SEED);
            hasUsefulIndexForKey = true;
        }
    }

    // 3. 指定唯一索引的分片键必须完全匹配
    if (hasUsefulIndexForKey && request.getUnique()) {
        ...
        BSONObj eqQueryResult;
        for (const auto& idx : indexes) {
            if (SimpleBSONObjComparator::kInstance.evaluate(idx["key"].embeddedObject() == proposedKey)) {
                eqQueryResult = idx;
                break;
            }
        }

        if (eqQueryResult.isEmpty()) {
            // If no exact match, index not useful, but still possible to create one later
            hasUsefulIndexForKey = false;
        } else {
            bool isExplicitlyUnique = eqQueryResult["unique"].trueValue();
            BSONObj currKey = eqQueryResult["key"].embeddedObject();
            bool isCurrentID = str::equals(currKey.firstElementFieldName(), "_id");
            uassert(ErrorCodes::InvalidOptions,
                    str::stream() << "can't shard collection " << nss.ns() << ", " << proposedKey
                                  << " index not unique, and unique index explicitly specified",
                    isExplicitlyUnique || isCurrentID);
        }
    }

    if (hasUsefulIndexForKey) {
        // 检查null和多键
        ...
        BSONObj res;
        auto success = localClient.runCommand("admin", checkShardingIndexCmd.obj(), res);
        uassert(ErrorCodes::OperationFailed, res["errmsg"].str(), success);
    } else if (localClient.count(nss.ns()) != 0) {
        // 4. 非空集合没有可用索引，失败返回
        uasserted(ErrorCodes::InvalidOptions,
                  "Please create an index that starts with the proposed shard key before "
                  "sharding the collection");
    } else {
        // 5. 空集合创建分片键索引
        ...
        auto createIndexesCmd =
            makeCreateIndexesCmd(nss, proposedKey, collation, request.getUnique());

        BSONObj res;
        localClient.runCommand(nss.db().toString(), createIndexesCmd, res);
        uassertStatusOK(getStatusFromCommandResult(res));
    ...
    }

void InitialSplitPolicy::calculateHashedSplitPointsForEmptyCollection(
    const ShardKeyPattern& shardKeyPattern,
    bool isEmpty,
    int numShards,
    int numInitialChunks,
    std::vector<BSONObj>* initialSplitPoints,
    std::vector<BSONObj>* finalSplitPoints) {
    ...
    const long long intervalSize = (std::numeric_limits<long long>::max() / numInitialChunks) * 2;
    long long current = 0;

    const auto proposedKey(shardKeyPattern.getKeyPattern().toBSON());

    if (numInitialChunks % 2 == 0) {
        finalSplitPoints->push_back(BSON(proposedKey.firstElementFieldName() << current));
        current += intervalSize;
    } else {
        current += intervalSize / 2;
    }

    for (int i = 0; i < (numInitialChunks - 1) / 2; i++) {
        finalSplitPoints->push_back(BSON(proposedKey.firstElementFieldName() << current));
        finalSplitPoints->push_back(BSON(proposedKey.firstElementFieldName() << -current));
        current += intervalSize;
    }

    // 排序
    sort(finalSplitPoints->begin(),
         finalSplitPoints->end(),
         SimpleBSONObjComparator::kInstance.makeLessThan());

    // The initial splits define the "big chunks" that we will subdivide later.
    int lastIndex = -1;
    for (int i = 1; i < numShards; i++) {
        if (lastIndex < (i * numInitialChunks) / numShards - 1) {
            lastIndex = (i * numInitialChunks) / numShards - 1;
            initialSplitPoints->push_back(finalSplitPoints->at(lastIndex));
        }
    }
}

   ChunkVersion version(1, 0, OID::gen());
    const auto& keyPattern(shardKeyPattern.getKeyPattern());

    std::vector<ChunkType> chunks;

    for (size_t i = 0; i <= splitPoints.size(); i++) {
        const BSONObj min = (i == 0) ? keyPattern.globalMin() : splitPoints[i - 1];
        const BSONObj max = (i < splitPoints.size()) ? splitPoints[i] : keyPattern.globalMax();

        // It's possible there are no split points or fewer split points than total number of
        // shards, and we need to be sure that at least one chunk is placed on the primary shard
        const ShardId shardId = (i == 0 && splitPoints.size() + 1 < allShardIds.size())
            ? databasePrimaryShardId
            : allShardIds[(i / numContiguousChunksPerShard) % allShardIds.size()];

        appendChunk(nss, min, max, &version, validAfter, shardId, &chunks);
    }

//  ChunkType
Expected config server config.chunks collection format:
{
    _id : "test.foo-a_MinKey",
    ns : "test.foo",
    min : {
        "a" : { "$minKey" : 1 }
    },
    max : {
        "a" : { "$maxKey" : 1 }
    },
    shard : "test-rs1",
    lastmod : Timestamp(1, 0),
    lastmodEpoch : ObjectId("587fc60cef168288439ad6ed"),
    jumbo : false              // optional field
}

void InitialSplitPolicy::writeFirstChunksToConfig(
    OperationContext* opCtx, const InitialSplitPolicy::ShardCollectionConfig& initialChunks) {
    for (const auto& chunk : initialChunks.chunks) {
        uassertStatusOK(Grid::get(opCtx)->catalogClient()->insertConfigDocument(
            opCtx,
            ChunkType::ConfigNS,
            chunk.toConfigBSON(),
            ShardingCatalogClient::kMajorityWriteConcern));
    }
}

{
        CollectionType coll;
        coll.setNs(nss);
        if (uuid)
            coll.setUUID(*uuid);
        coll.setEpoch(initialChunks.collVersion().epoch());
        coll.setUpdatedAt(Date_t::fromMillisSinceEpoch(initialChunks.collVersion().toLong()));
        coll.setKeyPattern(fieldsAndOrder.toBSON());
        coll.setDefaultCollation(defaultCollator ? defaultCollator->getSpec().toBSON() : BSONObj());
        coll.setUnique(unique);

        uassertStatusOK(ShardingCatalogClientImpl::updateShardingCatalogEntryForCollection(
            opCtx, nss, coll, true /*upsert*/));
}

std::vector<ShardId> shardsRefreshed;
    for (const auto& chunk : initialChunks.chunks) {
        if ((chunk.getShard() == dbPrimaryShardId) ||
            std::find(shardsRefreshed.begin(), shardsRefreshed.end(), chunk.getShard()) !=
                shardsRefreshed.end()) {
            continue;
        }

        auto shard = uassertStatusOK(shardRegistry->getShard(opCtx, chunk.getShard()));
        auto refreshCmdResponse = uassertStatusOK(shard->runCommandWithFixedRetryAttempts(
            opCtx,
            ReadPreferenceSetting{ReadPreference::PrimaryOnly},
            "admin",
            BSON("_flushRoutingTableCacheUpdates" << nss.ns()),
            Seconds{30},
            Shard::RetryPolicy::kIdempotent));

        uassertStatusOK(refreshCmdResponse.commandStatus);
        shardsRefreshed.emplace_back(chunk.getShard());
    }

   oss.waitForMigrationCriticalSectionSignal(opCtx);

    // 如果需要从config server获取路由来更新
    if (request().getSyncFromConfig()) {
                LOG(1) << "Forcing remote routing table refresh for " << ns();
                forceShardFilteringMetadataRefresh(opCtx, ns());
    }

    CatalogCacheLoader::get(opCtx).waitForCollectionFlush(opCtx, ns());

    repl::ReplClientInfo::forClient(opCtx->getClient()).setLastOpToSystemLastOpTime(opCtx);

std::vector<BSONObj> donorIndexSpecs;
    BSONObj donorIdIndexSpec;
    BSONObj donorOptions;
    {
        // 0. Get the collection indexes and options from the donor shard.
        ...
        // Get indexes by calling listIndexes against the donor.
        auto indexes = uassertStatusOK(fromShard->runExhaustiveCursorCommand(
            opCtx,
            ReadPreferenceSetting(ReadPreference::PrimaryOnly),
            nss.db().toString(),
            BSON("listIndexes" << nss.coll().toString()),
            Milliseconds(-1)));

        for (auto&& spec : indexes.docs) {
            donorIndexSpecs.push_back(spec);
            if (auto indexNameElem = spec[IndexDescriptor::kIndexNameFieldName]) {
                if (indexNameElem.type() == BSONType::String &&
                    indexNameElem.valueStringData() == "_id_"_sd) {
                    donorIdIndexSpec = spec;
                }
            }
        }

        // Get collection options by calling listCollections against the donor.
        auto infosRes = uassertStatusOK(fromShard->runExhaustiveCursorCommand(
            opCtx,
            ReadPreferenceSetting(ReadPreference::PrimaryOnly),
            nss.db().toString(),
            BSON("listCollections" << 1 << "filter" << BSON("name" << nss.coll())),
            Milliseconds(-1)));

        auto infos = infosRes.docs;
        ...

        BSONObj entry = infos.front();
        BSONObjBuilder donorOptionsBob;

        if (entry["options"].isABSONObj()) {
            donorOptionsBob.appendElements(entry["options"].Obj());
        }

        BSONObj info;
        if (entry["info"].isABSONObj()) {
            info = entry["info"].Obj();
        }

        uassert(ErrorCodes::InvalidUUID,
                str::stream() << "The donor shard did not return a UUID for collection " << nss.ns()
                              << " as part of its listCollections response: "
                              << entry
                              << ", but this node expects to see a UUID.",
                !info["uuid"].eoo());

        donorOptionsBob.append(info["uuid"]);
        donorOptions = donorOptionsBob.obj();
    }

{
        AutoGetOrCreateDb autoCreateDb(opCtx, nss.db(), MODE_X);
        uassert(ErrorCodes::NotMaster,
                str::stream() << "Unable to create collection " << nss.ns()
                              << " because the node is not primary",
                repl::ReplicationCoordinator::get(opCtx)->canAcceptWritesFor(opCtx, nss));

        Database* const db = autoCreateDb.getDb();

    // 尝试获取collection
        Collection* collection = db->getCollection(opCtx, nss);
        if (collection) { // 存在collection
            // 集合存在，检查uuid
            ...

            uassert(ErrorCodes::InvalidUUID,
                    str::stream()
                        << "Cannot create collection "
                        << nss.ns()
                        << " because we already have an identically named collection with UUID "
                        << (collection->uuid() ? collection->uuid()->toString() : "(none)")
                        << ", which differs from the donor's UUID "
                        << (donorUUID ? donorUUID->toString() : "(none)")
                        << ". Manually drop the collection on this shard if it contains data from "
                           "a previous incarnation of "
                        << nss.ns(),
                    collection->uuid() == donorUUID);
        } else { // 不存在collection
            // We do not have a collection by this name. Create the collection with the donor's options.
            WriteUnitOfWork wuow(opCtx);
            CollectionOptions collectionOptions;
            uassertStatusOK(collectionOptions.parse(donorOptions,
                                                    CollectionOptions::ParseKind::parseForStorage));
            const bool createDefaultIndexes = true;

            uassertStatusOK(Database::userCreateNS(
                opCtx, db, nss.ns(), collectionOptions, createDefaultIndexes, donorIdIndexSpec));
            wuow.commit();

            collection = db->getCollection(opCtx, nss);
        }

        MultiIndexBlock indexer(opCtx, collection);
        indexer.removeExistingIndexes(&donorIndexSpecs);

        if (!donorIndexSpecs.empty()) {
            ...

            auto indexInfoObjs = indexer.init(donorIndexSpecs);
            uassert(ErrorCodes::CannotCreateIndex,
                    str::stream() << "failed to create index before migrating data. "
                                  << " error: "
                                  << redact(indexInfoObjs.getStatus()),
                    indexInfoObjs.isOK());

            WriteUnitOfWork wunit(opCtx);
            indexer.commit();

            for (auto&& infoObj : indexInfoObjs.getValue()) {
                // make sure to create index on secondaries as well
                serviceContext->getOpObserver()->onCreateIndex(
                    opCtx, collection->ns(), collection->uuid(), infoObj, true /* fromMigrate */);
            }

            wunit.commit();
        }
}

// locks文档存储的是最后一次拉取某ns的分布式锁的记录
{
        "_id" : "records", // 标记加锁的ns，ns可能是db，也可能是collection
        "state" : 0, // 2表示加锁，1已经被废弃
        "process" : "ConfigServer",
        "ts" : ObjectId("5f82cf26410ff005943046cc"),
        "when" : ISODate("2020-10-11T09:23:50.898Z"),
        "who" : "ConfigServer:conn3", // 哪一个session获取该锁
        "why" : "shardCollection" // 操作名
  }