SERVER-17969 Move shardCollection operation to the catalog manager

Used by both mapReduce and shardCollection commands.
Conflicts:
    src/mongo/s/catalog/legacy/catalog_manager_legacy.cpp

Author: alabid

src/mongo/s/catalog/catalog_manager.h

@@ -44,6 +44,7 @@ namespace mongo {
     class DatabaseType;
     class OperationContext;
     class Shard;
+    class ShardKeyPattern;
     class ShardType;
     class Status;
     template<typename T> class StatusWith;
@@ -82,6 +83,26 @@ namespace mongo {
          */
         virtual Status enableSharding(const std::string& dbName) = 0;
 
+        /**
+         * Shards a collection. Assumes that the database is enabled for sharding.
+         *
+         * @param ns: namespace of collection to shard
+         * @param fieldsAndOrder: shardKey pattern
+         * @param unique: if true, ensure underlying index enforces a unique constraint.
+         * @param initPoints: create chunks based on a set of specified split points.
+         * @param initShards: if NULL, use primary shard as lone shard for DB.
+         *
+         * WARNING: It's not completely safe to place initial chunks onto non-primary
+         *          shards using this method because a conflict may result if multiple map-reduce
+         *          operations are writing to the same output collection, for instance.
+         *
+         */
+        virtual Status shardCollection(const std::string& ns,
+                                       const ShardKeyPattern& fieldsAndOrder,
+                                       bool unique,
+                                       std::vector<BSONObj>* initPoints,
+                                       std::vector<Shard>* initShards = NULL) = 0;
+
         /**
          *
          * Adds a new shard. It expects a standalone mongod process or replica set to be running

src/mongo/s/catalog/legacy/catalog_manager_legacy.cpp

@@ -49,10 +49,13 @@
 #include "mongo/s/catalog/type_changelog.h"
 #include "mongo/s/catalog/type_chunk.h"
 #include "mongo/s/catalog/type_shard.h"
+#include "mongo/s/chunk_manager.h"
 #include "mongo/s/client/dbclient_multi_command.h"
 #include "mongo/s/client/shard_connection.h"
 #include "mongo/s/distlock.h"
 #include "mongo/s/client/shard.h"
+#include "mongo/s/config.h"
+#include "mongo/s/shard_key_pattern.h"
 #include "mongo/s/type_database.h"
 #include "mongo/s/write_ops/batched_command_request.h"
 #include "mongo/s/write_ops/batched_command_response.h"
@@ -403,6 +406,94 @@ namespace {
         return updateDatabase(dbName, db);
     }
 
+    Status CatalogManagerLegacy::shardCollection(const string& ns,
+                                                 const ShardKeyPattern& fieldsAndOrder,
+                                                 bool unique,
+                                                 vector<BSONObj>* initPoints,
+                                                 vector<Shard>* initShards) {
+
+        StatusWith<DatabaseType> status = getDatabase(nsToDatabase(ns));
+        if (!status.isOK()) {
+            return status.getStatus();
+        }
+        DatabaseType dbt = status.getValue();
+        Shard dbPrimary = Shard::make(dbt.getPrimary());
+
+        log() << "enable sharding on: " << ns << " with shard key: " << fieldsAndOrder;
+
+        // Record start in changelog
+        BSONObjBuilder collectionDetail;
+        collectionDetail.append("shardKey", fieldsAndOrder.toBSON());
+        collectionDetail.append("collection", ns);
+        collectionDetail.append("primary", dbPrimary.toString());
+
+        BSONArray initialShards;
+        if (initShards == NULL)
+            initialShards = BSONArray();
+        else {
+            BSONArrayBuilder b;
+            for (unsigned i = 0; i < initShards->size(); i++) {
+                b.append((*initShards)[i].getName());
+            }
+            initialShards = b.arr();
+        }
+
+        collectionDetail.append("initShards", initialShards);
+        collectionDetail.append("numChunks", static_cast<int>(initPoints->size() + 1));
+
+        logChange(NULL, "shardCollection.start", ns, collectionDetail.obj());
+
+        ChunkManagerPtr manager(new ChunkManager(ns, fieldsAndOrder, unique));
+        manager->createFirstChunks(_configServerConnectionString.toString(),
+                                   dbPrimary,
+                                   initPoints,
+                                   initShards);
+        manager->loadExistingRanges(_configServerConnectionString.toString(), NULL);
+
+        CollectionInfo collInfo;
+        collInfo.useChunkManager(manager);
+        collInfo.save(ns);
+        manager->reload(true);
+
+        // Tell the primary mongod to refresh its data
+        // TODO:  Think the real fix here is for mongos to just
+        //        assume that all collections are sharded, when we get there
+        for (int i = 0;i < 4;i++) {
+            if (i == 3) {
+                warning() << "too many tries updating initial version of " << ns
+                          << " on shard primary " << dbPrimary
+                          << ", other mongoses may not see the collection as sharded immediately";
+                break;
+            }
+            try {
+                ShardConnection conn(dbPrimary, ns);
+                bool isVersionSet = conn.setVersion();
+                conn.done();
+                if (!isVersionSet) {
+                    warning() << "could not update initial version of "
+                              << ns << " on shard primary " << dbPrimary;
+                } else {
+                    break;
+                }
+            }
+            catch (const DBException& e) {
+                warning() << "could not update initial version of " << ns
+                          << " on shard primary " << dbPrimary
+                          << causedBy(e);
+            }
+            sleepsecs(i);
+        }
+
+        // Record finish in changelog
+        BSONObjBuilder finishDetail;
+
+        finishDetail.append("version", manager->getVersion().toString());
+
+        logChange(NULL, "shardCollection", ns, finishDetail.obj());
+
+        return Status::OK();
+    }
+
     Status CatalogManagerLegacy::createDatabase(const std::string& dbName, const Shard* shard) {
         invariant(nsIsDbOnly(dbName));
 
@@ -559,7 +650,7 @@ namespace {
     }
 
     StatusWith<ShardDrainingStatus> CatalogManagerLegacy::removeShard(OperationContext* txn,
-                                                                 const std::string& name) {
+                                                                      const std::string& name) {
         ScopedDbConnection conn(_configServerConnectionString, 30);
 
         if (conn->count(ShardType::ConfigNS,

src/mongo/s/catalog/legacy/catalog_manager_legacy.h

@@ -53,6 +53,12 @@ namespace mongo {
 
         virtual Status enableSharding(const std::string& dbName);
 
+        virtual Status shardCollection(const std::string& ns,
+                                       const ShardKeyPattern& fieldsAndOrder,
+                                       bool unique,
+                                       std::vector<BSONObj>* initPoints,
+                                       std::vector<Shard>* initShards);
+
         virtual StatusWith<std::string> addShard(const std::string& name,
                                                  const ConnectionString& shardConnectionString,
                                                  const long long maxSize);

src/mongo/s/commands/cluster_shard_collection_cmd.cpp

@@ -46,6 +46,7 @@
 #include "mongo/db/hasher.h"
 #include "mongo/db/write_concern_options.h"
 #include "mongo/s/catalog/catalog_cache.h"
+#include "mongo/s/catalog/catalog_manager.h"
 #include "mongo/s/chunk_manager.h"
 #include "mongo/s/cluster_write.h"
 #include "mongo/s/grid.h"
@@ -406,7 +407,13 @@ namespace {
                                       proposedKey,
                                       careAboutUnique);
 
-            config->shardCollection(ns, proposedShardKey, careAboutUnique, &initSplits);
+            Status status = grid.catalogManager()->shardCollection(ns,
+                                                                   proposedShardKey,
+                                                                   careAboutUnique,
+                                                                   &initSplits);
+            if (!status.isOK()) {
+                return appendCommandStatus(result, status);
+            }
 
             result << "collectionsharded" << ns;
 

src/mongo/s/commands_public.cpp

@@ -1857,11 +1857,14 @@ namespace {
                         vector<Shard> outShards( shardSet.begin() , shardSet.end() );
 
                         ShardKeyPattern sortKeyPattern(sortKey);
-                        confOut->shardCollection(finalColLong,
-                                                 sortKeyPattern,
-                                                 true,
-                                                 &sortedSplitPts,
-                                                 &outShards);
+                        Status status = grid.catalogManager()->shardCollection(finalColLong,
+                                                                               sortKeyPattern,
+                                                                               true,
+                                                                               &sortedSplitPts,
+                                                                               &outShards);
+                        if (!status.isOK()) {
+                            return appendCommandStatus(result, status);
+                        }
 
                     }
 

src/mongo/s/config.cpp

@@ -76,7 +76,7 @@ namespace mongo {
     Shard Shard::EMPTY;
 
 
-    DBConfig::CollectionInfo::CollectionInfo( const BSONObj& in ) {
+    CollectionInfo::CollectionInfo(const BSONObj& in) {
         _dirty = false;
         _dropped = in[CollectionType::dropped()].trueValue();
 
@@ -87,18 +87,18 @@ namespace mongo {
         _dirty = false;
     }
 
-    DBConfig::CollectionInfo::~CollectionInfo() {
+    CollectionInfo::~CollectionInfo() {
 
     }
 
-    void DBConfig::CollectionInfo::resetCM(ChunkManager* cm) {
+    void CollectionInfo::resetCM(ChunkManager* cm) {
         invariant(cm);
         invariant(_cm);
 
         _cm.reset(cm);
     }
 
-    void DBConfig::CollectionInfo::shard(ChunkManager* manager){
+    void CollectionInfo::shard(ChunkManager* manager) {
         // Do this *first* so we're invisible to everyone else
         manager->loadExistingRanges(configServer.getPrimary().getConnString(), NULL);
 
@@ -107,12 +107,8 @@ namespace mongo {
         // This helps prevent errors when dropping in a different process
         //
 
-        if (manager->numChunks() != 0){
-            _cm = ChunkManagerPtr(manager);
-            _key = manager->getShardKeyPattern().toBSON().getOwned();
-            _unqiue = manager->isUnique();
-            _dirty = true;
-            _dropped = false;
+        if (manager->numChunks() != 0) {
+            useChunkManager(ChunkManagerPtr(manager));
         }
         else{
             warning() << "no chunks found for collection " << manager->getns()
@@ -121,14 +117,22 @@ namespace mongo {
         }
     }
 
-    void DBConfig::CollectionInfo::unshard() {
+    void CollectionInfo::unshard() {
         _cm.reset();
         _dropped = true;
         _dirty = true;
         _key = BSONObj();
     }
 
-    void DBConfig::CollectionInfo::save( const string& ns ) {
+    void CollectionInfo::useChunkManager(ChunkManagerPtr manager) {
+        _cm = manager;
+        _key = manager->getShardKeyPattern().toBSON().getOwned();
+        _unique = manager->isUnique();
+        _dirty = true;
+        _dropped = false;
+    }
+
+    void CollectionInfo::save(const string& ns) {
         BSONObj key = BSON( "_id" << ns );
 
         BSONObjBuilder val;
@@ -218,97 +222,6 @@ namespace mongo {
         if( save ) _save();
     }
 
-    boost::shared_ptr<ChunkManager> DBConfig::shardCollection(
-                                                const string& ns,
-                                                const ShardKeyPattern& fieldsAndOrder,
-                                                bool unique,
-                                                vector<BSONObj>* initPoints,
-                                                vector<Shard>* initShards) {
-
-        uassert(8042, "db doesn't have sharding enabled", _shardingEnabled);
-
-        ChunkManagerPtr manager;
-        
-        {
-            boost::lock_guard<boost::mutex> lk( _lock );
-
-            CollectionInfo& ci = _collections[ns];
-            uassert( 8043 , "collection already sharded" , ! ci.isSharded() );
-
-            log() << "enable sharding on: " << ns << " with shard key: " << fieldsAndOrder << endl;
-
-            // Record start in changelog
-            BSONObjBuilder collectionDetail;
-            collectionDetail.append("shardKey", fieldsAndOrder.toBSON());
-            collectionDetail.append("collection", ns);
-            collectionDetail.append("primary", getPrimary().toString());
-
-            BSONArray initialShards;
-            if (initShards == NULL)
-                initialShards = BSONArray();
-            else {
-                BSONArrayBuilder b;
-                for (unsigned i = 0; i < initShards->size(); i++) {
-                    b.append((*initShards)[i].getName());
-                }
-                initialShards = b.arr();
-            }
-
-            collectionDetail.append("initShards", initialShards);
-            collectionDetail.append("numChunks", (int)(initPoints->size() + 1));
-
-            grid.catalogManager()->logChange(NULL,
-                                             "shardCollection.start",
-                                             ns,
-                                             collectionDetail.obj());
-
-            ChunkManager* cm = new ChunkManager( ns, fieldsAndOrder, unique );
-            cm->createFirstChunks(configServer.getPrimary().getConnString(),
-                                  getPrimary(),
-                                  initPoints,
-                                  initShards);
-            ci.shard(cm);
-
-            _save();
-
-            // Save the initial chunk manager for later, no need to reload if we're in this lock
-            manager = ci.getCM();
-            verify( manager.get() );
-        }
-
-        // Tell the primary mongod to refresh it's data
-        // TODO:  Think the real fix here is for mongos to just assume all collections sharded, when we get there
-        for( int i = 0; i < 4; i++ ){
-            if( i == 3 ){
-                warning() << "too many tries updating initial version of " << ns << " on shard primary " << getPrimary() <<
-                             ", other mongoses may not see the collection as sharded immediately" << endl;
-                break;
-            }
-            try {
-                ShardConnection conn( getPrimary(), ns );
-                if (!conn.setVersion()) {
-                    warning() << "could not update initial version of "
-                              << ns << " on shard primary " << getPrimary();
-                }
-                conn.done();
-                break;
-            }
-            catch( DBException& e ){
-                warning() << "could not update initial version of " << ns << " on shard primary " << getPrimary() <<
-                             causedBy( e ) << endl;
-            }
-            sleepsecs( i );
-        }
-
-        // Record finish in changelog
-        BSONObjBuilder finishDetail;
-        finishDetail.append("version", manager->getVersion().toString());
-
-        grid.catalogManager()->logChange(NULL, "shardCollection", ns, finishDetail.obj());
-
-        return manager;
-    }
-
     bool DBConfig::removeSharding( const string& ns ) {
         if ( ! _shardingEnabled ) {