path: root/src/mongo/s/strategy.cpp

/*
 *    Copyright (C) 2010 10gen Inc.
 *
 *    This program is free software: you can redistribute it and/or  modify
 *    it under the terms of the GNU Affero General Public License, version 3,
 *    as published by the Free Software Foundation.
 *
 *    This program is distributed in the hope that it will be useful,
 *    but WITHOUT ANY WARRANTY; without even the implied warranty of
 *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *    GNU Affero General Public License for more details.
 *
 *    You should have received a copy of the GNU Affero General Public License
 *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
 *
 *    As a special exception, the copyright holders give permission to link the
 *    code of portions of this program with the OpenSSL library under certain
 *    conditions as described in each individual source file and distribute
 *    linked combinations including the program with the OpenSSL library. You
 *    must comply with the GNU Affero General Public License in all respects
 *    for all of the code used other than as permitted herein. If you modify
 *    file(s) with this exception, you may extend this exception to your
 *    version of the file(s), but you are not obligated to do so. If you do not
 *    wish to do so, delete this exception statement from your version. If you
 *    delete this exception statement from all source files in the program,
 *    then also delete it in the license file.
 */

#define MONGO_LOG_DEFAULT_COMPONENT ::mongo::logger::LogComponent::kSharding

#include "mongo/platform/basic.h"

#include "mongo/s/strategy.h"


#include "mongo/base/status.h"
#include "mongo/base/owned_pointer_vector.h"
#include "mongo/bson/util/builder.h"
#include "mongo/client/connpool.h"
#include "mongo/client/dbclientcursor.h"
#include "mongo/db/audit.h"
#include "mongo/db/auth/action_type.h"
#include "mongo/db/auth/authorization_session.h"
#include "mongo/db/commands.h"
#include "mongo/db/max_time.h"
#include "mongo/db/server_parameters.h"
#include "mongo/db/namespace_string.h"
#include "mongo/db/query/lite_parsed_query.h"
#include "mongo/db/stats/counters.h"
#include "mongo/s/bson_serializable.h"
#include "mongo/s/catalog/catalog_cache.h"
#include "mongo/s/chunk_manager_targeter.h"
#include "mongo/s/client/dbclient_multi_command.h"
#include "mongo/s/client/shard_registry.h"
#include "mongo/s/chunk_manager.h"
#include "mongo/s/chunk_version.h"
#include "mongo/s/config.h"
#include "mongo/s/cursors.h"
#include "mongo/s/dbclient_shard_resolver.h"
#include "mongo/s/grid.h"
#include "mongo/s/request.h"
#include "mongo/s/stale_exception.h"
#include "mongo/s/version_manager.h"
#include "mongo/s/write_ops/batched_command_request.h"
#include "mongo/s/write_ops/batch_upconvert.h"
#include "mongo/util/log.h"
#include "mongo/util/mongoutils/str.h"
#include "mongo/util/timer.h"

namespace mongo {

    using std::unique_ptr;
    using std::shared_ptr;
    using std::endl;
    using std::set;
    using std::string;
    using std::stringstream;
    using std::vector;

    static bool _isSystemIndexes( const char* ns ) {
        return nsToCollectionSubstring(ns) == "system.indexes";
    }

    /**
     * Returns true if request is a query for sharded indexes.
     */
    static bool doShardedIndexQuery(Request& r, const QuerySpec& qSpec) {
        // Extract the ns field from the query, which may be embedded within the "query" or
        // "$query" field.
        auto nsField = qSpec.filter()["ns"];
        if (nsField.eoo()) {
            return false;
        }
        const NamespaceString indexNSSQuery(nsField.str());

        auto status = grid.catalogCache()->getDatabase(indexNSSQuery.db().toString());
        if (!status.isOK()) {
            return false;
        }

        shared_ptr<DBConfig> config = status.getValue();
        if (!config->isSharded(indexNSSQuery.ns())) {
            return false;
        }

        // if you are querying on system.indexes, we need to make sure we go to a shard
        // that actually has chunks. This is not a perfect solution (what if you just
        // look at all indexes), but better than doing nothing.

        ShardPtr shard;
        ChunkManagerPtr cm;
        config->getChunkManagerOrPrimary(indexNSSQuery.ns(), cm, shard);
        if ( cm ) {
            set<ShardId> shardIds;
            cm->getAllShardIds(&shardIds);
            verify(shardIds.size() > 0);
            shard = grid.shardRegistry()->getShard(*shardIds.begin());
        }

        ShardConnection dbcon(shard->getConnString(), r.getns());
        DBClientBase &c = dbcon.conn();

        string actualServer;

        Message response;
        bool ok = c.call( r.m(), response, true , &actualServer );
        uassert( 10200 , "mongos: error calling db", ok );

        {
            QueryResult::View qr = response.singleData().view2ptr();
            if ( qr.getResultFlags() & ResultFlag_ShardConfigStale ) {
                dbcon.done();
                // Version is zero b/c this is deprecated codepath
                throw RecvStaleConfigException( r.getns(),
                                                "Strategy::doQuery",
                                                ChunkVersion( 0, 0, OID() ),
                                                ChunkVersion( 0, 0, OID() ));
            }
        }

        r.reply( response , actualServer.size() ? actualServer : c.getServerAddress() );
        dbcon.done();

        return true;
    }

    void Strategy::queryOp( Request& r ) {

        verify( !NamespaceString( r.getns() ).isCommand() );

        Timer queryTimer;

        QueryMessage q( r.d() );

        NamespaceString ns(q.ns);
        ClientBasic* client = ClientBasic::getCurrent();
        AuthorizationSession* authSession = AuthorizationSession::get(client);
        Status status = authSession->checkAuthForQuery(ns, q.query);
        audit::logQueryAuthzCheck(client, ns, q.query, status.code());
        uassertStatusOK(status);

        LOG(3) << "query: " << q.ns << " " << q.query << " ntoreturn: " << q.ntoreturn
               << " options: " << q.queryOptions << endl;

        if ( q.ntoreturn == 1 && strstr(q.ns, ".$cmd") )
            throw UserException( 8010 , "something is wrong, shouldn't see a command here" );

        if (q.queryOptions & QueryOption_Exhaust) {
            uasserted(18526,
                      string("the 'exhaust' query option is invalid for mongos queries: ") + q.ns
                      + " " + q.query.toString());
        }

        QuerySpec qSpec( (string)q.ns, q.query, q.fields, q.ntoskip, q.ntoreturn, q.queryOptions );

        // Parse "$maxTimeMS".
        StatusWith<int> maxTimeMS = LiteParsedQuery::parseMaxTimeMSQuery( q.query );
        uassert( 17233,
                 maxTimeMS.getStatus().reason(),
                 maxTimeMS.isOK() );

        if ( _isSystemIndexes( q.ns ) && doShardedIndexQuery( r, qSpec )) {
            return;
        }

        ParallelSortClusteredCursor * cursor = new ParallelSortClusteredCursor( qSpec, CommandInfo() );
        verify( cursor );

        // TODO:  Move out to Request itself, not strategy based
        try {
            cursor->init();

            if ( qSpec.isExplain() ) {
                BSONObjBuilder explain_builder;
                cursor->explain( explain_builder );
                explain_builder.appendNumber( "executionTimeMillis",
                                              static_cast<long long>(queryTimer.millis()) );
                BSONObj b = explain_builder.obj();

                replyToQuery( 0 , r.p() , r.m() , b );
                delete( cursor );
                return;
            }
        }
        catch(...) {
            delete cursor;
            throw;
        }

        // TODO: Revisit all of this when we revisit the sharded cursor cache

        if (cursor->getNumQueryShards() != 1) {

            // More than one shard (or zero), manage with a ShardedClientCursor
            // NOTE: We may also have *zero* shards here when the returnPartial flag is set.
            // Currently the code in ShardedClientCursor handles this.

            ShardedClientCursorPtr cc (new ShardedClientCursor( q , cursor ));

            BufBuilder buffer( ShardedClientCursor::INIT_REPLY_BUFFER_SIZE );
            int docCount = 0;
            const int startFrom = cc->getTotalSent();
            bool hasMore = cc->sendNextBatch(q.ntoreturn, buffer, docCount);

            if ( hasMore ) {
                LOG(5) << "storing cursor : " << cc->getId() << endl;

                int cursorLeftoverMillis = maxTimeMS.getValue() - queryTimer.millis();
                if ( maxTimeMS.getValue() == 0 ) { // 0 represents "no limit".
                    cursorLeftoverMillis = kMaxTimeCursorNoTimeLimit;
                }
                else if ( cursorLeftoverMillis <= 0 ) {
                    cursorLeftoverMillis = kMaxTimeCursorTimeLimitExpired;
                }

                cursorCache.store( cc, cursorLeftoverMillis );
            }

            replyToQuery( 0, r.p(), r.m(), buffer.buf(), buffer.len(), docCount,
                    startFrom, hasMore ? cc->getId() : 0 );
        }
        else{

            // Only one shard is used

            // Remote cursors are stored remotely, we shouldn't need this around.
            unique_ptr<ParallelSortClusteredCursor> cursorDeleter( cursor );

            ShardPtr shard = cursor->getQueryShard();
            verify( shard.get() );
            DBClientCursorPtr shardCursor = cursor->getShardCursor(shard->getId());

            // Implicitly stores the cursor in the cache
            r.reply( *(shardCursor->getMessage()) , shardCursor->originalHost() );

            // We don't want to kill the cursor remotely if there's still data left
            shardCursor->decouple();
        }
    }

    void Strategy::clientCommandOp( Request& r ) {
        QueryMessage q( r.d() );

        LOG(3) << "command: " << q.ns << " " << q.query << " ntoreturn: " << q.ntoreturn
               << " options: " << q.queryOptions << endl;

        if (q.queryOptions & QueryOption_Exhaust) {
            uasserted(18527,
                      string("the 'exhaust' query option is invalid for mongos commands: ") + q.ns
                      + " " + q.query.toString());
        }

        NamespaceString nss( r.getns() );
        // Regular queries are handled in strategy_shard.cpp
        verify( nss.isCommand() || nss.isSpecialCommand() );

        if ( handleSpecialNamespaces( r , q ) )
            return;

        int loops = 5;
        while ( true ) {
            BSONObjBuilder builder;
            try {
                BSONObj cmdObj = q.query;
                {
                    BSONElement e = cmdObj.firstElement();
                    if (e.type() == Object && (e.fieldName()[0] == '$'
                                                 ? str::equals("query", e.fieldName()+1)
                                                 : str::equals("query", e.fieldName()))) {
                        // Extract the embedded query object.

                        if (cmdObj.hasField(Query::ReadPrefField.name())) {
                            // The command has a read preference setting. We don't want
                            // to lose this information so we copy this to a new field
                            // called $queryOptions.$readPreference
                            BSONObjBuilder finalCmdObjBuilder;
                            finalCmdObjBuilder.appendElements(e.embeddedObject());

                            BSONObjBuilder queryOptionsBuilder(
                                    finalCmdObjBuilder.subobjStart("$queryOptions"));
                            queryOptionsBuilder.append(cmdObj[Query::ReadPrefField.name()]);
                            queryOptionsBuilder.done();

                            cmdObj = finalCmdObjBuilder.obj();
                        }
                        else {
                            cmdObj = e.embeddedObject();
                        }
                    }
                }

                Command::runAgainstRegistered(q.ns, cmdObj, builder, q.queryOptions);
                BSONObj x = builder.done();
                replyToQuery(0, r.p(), r.m(), x);
                return;
            }
            catch ( StaleConfigException& e ) {
                if ( loops <= 0 )
                    throw e;

                loops--;
                log() << "retrying command: " << q.query << endl;

                // For legacy reasons, ns may not actually be set in the exception :-(
                string staleNS = e.getns();
                if( staleNS.size() == 0 ) staleNS = q.ns;

                ShardConnection::checkMyConnectionVersions( staleNS );
                if( loops < 4 ) versionManager.forceRemoteCheckShardVersionCB( staleNS );
            }
            catch ( AssertionException& e ) {
                Command::appendCommandStatus(builder, e.toStatus());
                BSONObj x = builder.done();
                replyToQuery(0, r.p(), r.m(), x);
                return;
            }
        }
    }

    // TODO: remove after MongoDB 3.2
    bool Strategy::handleSpecialNamespaces( Request& r , QueryMessage& q ) {
        const char * ns = strstr( r.getns() , ".$cmd.sys." );
        if ( ! ns )
            return false;
        ns += 10;

        BSONObjBuilder reply;

        const auto upgradeToRealCommand = [&r, &q, &reply](StringData commandName) {
            BSONObjBuilder cmdBob;
            cmdBob.append(commandName, 1);
            cmdBob.appendElements(q.query); // fields are validated by Commands
            auto interposedCmd = cmdBob.done();
            NamespaceString nss(r.getns());
            NamespaceString interposedNss(nss.db(), "$cmd");
            Command::runAgainstRegistered(interposedNss.ns().c_str(),
                                          interposedCmd,
                                          reply,
                                          q.queryOptions);
        };

        if ( strcmp( ns , "inprog" ) == 0 ) {
            upgradeToRealCommand("currentOp");
        }
        else if ( strcmp( ns , "killop" ) == 0 ) {
            upgradeToRealCommand("killOp");
        }
        else if ( strcmp( ns , "unlock" ) == 0 ) {
            reply.append( "err" , "can't do unlock through mongos" );
        }
        else {
            warning() << "unknown sys command [" << ns << "]" << endl;
            return false;
        }

        BSONObj x = reply.done();
        replyToQuery(0, r.p(), r.m(), x);
        return true;
    }

    void Strategy::commandOp( const string& db,
                              const BSONObj& command,
                              int options,
                              const string& versionedNS,
                              const BSONObj& targetingQuery,
                              vector<CommandResult>* results )
    {

        QuerySpec qSpec(db + ".$cmd", command, BSONObj(), 0, 1, options);

        ParallelSortClusteredCursor cursor( qSpec, CommandInfo( versionedNS, targetingQuery ) );

        // Initialize the cursor
        cursor.init();

        set<ShardId> shardIds;
        cursor.getQueryShardIds(shardIds);

        for (const ShardId& shardId : shardIds) {
            CommandResult result;
            result.shardTargetId = shardId;

            string errMsg; // ignored, should never be invalid b/c an exception thrown earlier
            result.target =
                    ConnectionString::parse( cursor.getShardCursor(shardId)->originalHost(),
                                             errMsg );
            result.result = cursor.getShardCursor(shardId)->peekFirst().getOwned();
            results->push_back( result );
        }

    }

    Status Strategy::commandOpWrite(const std::string& dbName,
                                    const BSONObj& command,
                                    BatchItemRef targetingBatchItem,
                                    std::vector<CommandResult>* results) {

        // Note that this implementation will not handle targeting retries and does not completely
        // emulate write behavior

        ChunkManagerTargeter targeter(NamespaceString(
                                        targetingBatchItem.getRequest()->getTargetingNS()));
        Status status = targeter.init();
        if (!status.isOK())
            return status;

        OwnedPointerVector<ShardEndpoint> endpointsOwned;
        vector<ShardEndpoint*>& endpoints = endpointsOwned.mutableVector();

        if (targetingBatchItem.getOpType() == BatchedCommandRequest::BatchType_Insert) {
            ShardEndpoint* endpoint;
            Status status = targeter.targetInsert(targetingBatchItem.getDocument(), &endpoint);
            if (!status.isOK())
                return status;
            endpoints.push_back(endpoint);
        }
        else if (targetingBatchItem.getOpType() == BatchedCommandRequest::BatchType_Update) {
            Status status = targeter.targetUpdate(*targetingBatchItem.getUpdate(), &endpoints);
            if (!status.isOK())
                return status;
        }
        else {
            invariant(targetingBatchItem.getOpType() == BatchedCommandRequest::BatchType_Delete);
            Status status = targeter.targetDelete(*targetingBatchItem.getDelete(), &endpoints);
            if (!status.isOK())
                return status;
        }

        DBClientShardResolver resolver;
        DBClientMultiCommand dispatcher;

        // Assemble requests
        for (vector<ShardEndpoint*>::const_iterator it = endpoints.begin(); it != endpoints.end();
            ++it) {

            const ShardEndpoint* endpoint = *it;

            ConnectionString host;
            Status status = resolver.chooseWriteHost(endpoint->shardName, &host);
            if (!status.isOK())
                return status;

            RawBSONSerializable request(command);
            dispatcher.addCommand(host, dbName, request);
        }

        // Errors reported when recv'ing responses
        dispatcher.sendAll();
        Status dispatchStatus = Status::OK();

        // Recv responses
        while (dispatcher.numPending() > 0) {

            ConnectionString host;
            RawBSONSerializable response;

            Status status = dispatcher.recvAny(&host, &response);
            if (!status.isOK()) {
                // We always need to recv() all the sent operations
                dispatchStatus = status;
                continue;
            }

            CommandResult result;
            result.target = host;
            {
                const auto shard = grid.shardRegistry()->getShard(host.toString());
                result.shardTargetId = shard->getId();
            }
            result.result = response.toBSON();

            results->push_back(result);
        }

        return dispatchStatus;
    }

    Status Strategy::commandOpUnsharded(const std::string& db,
                                        const BSONObj& command,
                                        int options,
                                        const std::string& versionedNS,
                                        CommandResult* cmdResult) {

        // Note that this implementation will not handle targeting retries and fails when the
        // sharding metadata is too stale
        auto status = grid.catalogCache()->getDatabase(db);
        if (!status.isOK()) {
            mongoutils::str::stream ss;
            ss << "Passthrough command failed: " << command.toString()
               << " on ns " << versionedNS << ". Caused by " << causedBy(status.getStatus());
            return Status(ErrorCodes::IllegalOperation, ss);
        }

        shared_ptr<DBConfig> conf = status.getValue();
        if (conf->isSharded(versionedNS)) {
            mongoutils::str::stream ss;
            ss << "Passthrough command failed: " << command.toString()
               << " on ns " << versionedNS << ". Cannot run on sharded namespace.";
            return Status(ErrorCodes::IllegalOperation, ss);
        }

        const auto primaryShard = grid.shardRegistry()->getShard(conf->getPrimaryId());

        BSONObj shardResult;
        try {
            ShardConnection conn(primaryShard->getConnString(), "");

            // TODO: this can throw a stale config when mongos is not up-to-date -- fix.
            if (!conn->runCommand(db, command, shardResult, options)) {
                conn.done();
                return Status(ErrorCodes::OperationFailed,
                              str::stream() << "Passthrough command failed: " << command
                                            << " on ns " << versionedNS
                                            << "; result: " << shardResult);
            }
            conn.done();
        }
        catch (const DBException& ex) {
            return ex.toStatus();
        }

        // Fill out the command result.
        cmdResult->shardTargetId = conf->getPrimaryId();
        cmdResult->result = shardResult;
        cmdResult->target = primaryShard->getConnString();

        return Status::OK();
    }

    void Strategy::getMore( Request& r ) {
        Timer getMoreTimer;

        const char* ns = r.getns();
        const int ntoreturn = r.d().pullInt();
        const long long id = r.d().pullInt64();

        // TODO:  Handle stale config exceptions here from coll being dropped or sharded during op
        // for now has same semantics as legacy request
        const NamespaceString nss(ns);
        auto statusGetDb = grid.catalogCache()->getDatabase(nss.db().toString());
        if (statusGetDb == ErrorCodes::DatabaseNotFound) {
            cursorCache.remove(id);
            replyToQuery(ResultFlag_CursorNotFound, r.p(), r.m(), 0, 0, 0);
            return;
        }

        uassertStatusOK(statusGetDb);

        shared_ptr<DBConfig> config = statusGetDb.getValue();

        ShardPtr primary;
        ChunkManagerPtr info;
        config->getChunkManagerOrPrimary( ns, info, primary );

        //
        // TODO: Cleanup cursor cache, consolidate into single codepath
        //
        const string host = cursorCache.getRef(id);
        ShardedClientCursorPtr cursor = cursorCache.get( id );
        int cursorMaxTimeMS = cursorCache.getMaxTimeMS( id );

        // Cursor ids should not overlap between sharded and unsharded cursors
        massert( 17012, str::stream() << "duplicate sharded and unsharded cursor id "
                                      << id << " detected for " << ns
                                      << ", duplicated on host " << host,
                 NULL == cursorCache.get( id ).get() || host.empty() );

        ClientBasic* client = ClientBasic::getCurrent();
        NamespaceString nsString(ns);
        AuthorizationSession* authSession = AuthorizationSession::get(client);
        Status status = authSession->checkAuthForGetMore( nsString, id );
        audit::logGetMoreAuthzCheck( client, nsString, id, status.code() );
        uassertStatusOK(status);

        if( !host.empty() ){

            LOG(3) << "single getmore: " << ns << endl;

            // we used ScopedDbConnection because we don't get about config versions
            // not deleting data is handled elsewhere
            // and we don't want to call setShardVersion
            ScopedDbConnection conn(host);

            Message response;
            bool ok = conn->callRead( r.m() , response);
            uassert( 10204 , "dbgrid: getmore: error calling db", ok);

            bool hasMore = (response.singleData().getCursor() != 0);

            if ( !hasMore ) {
                cursorCache.removeRef( id );
            }

            r.reply( response , "" /*conn->getServerAddress() */ );
            conn.done();
            return;
        }
        else if ( cursor ) {

            if ( cursorMaxTimeMS == kMaxTimeCursorTimeLimitExpired ) {
                cursorCache.remove( id );
                uasserted( ErrorCodes::ExceededTimeLimit, "operation exceeded time limit" );
            }

            // TODO: Try to match logic of mongod, where on subsequent getMore() we pull lots more data?
            BufBuilder buffer( ShardedClientCursor::INIT_REPLY_BUFFER_SIZE );
            int docCount = 0;
            const int startFrom = cursor->getTotalSent();
            bool hasMore = cursor->sendNextBatch(ntoreturn, buffer, docCount);

            if ( hasMore ) {
                // still more data
                cursor->accessed();

                if ( cursorMaxTimeMS != kMaxTimeCursorNoTimeLimit ) {
                    // Update remaining amount of time in cursor cache.
                    int cursorLeftoverMillis = cursorMaxTimeMS - getMoreTimer.millis();
                    if ( cursorLeftoverMillis <= 0 ) {
                        cursorLeftoverMillis = kMaxTimeCursorTimeLimitExpired;
                    }
                    cursorCache.updateMaxTimeMS( id, cursorLeftoverMillis );
                }
            }
            else {
                // we've exhausted the cursor
                cursorCache.remove( id );
            }

            replyToQuery( 0, r.p(), r.m(), buffer.buf(), buffer.len(), docCount,
                    startFrom, hasMore ? cursor->getId() : 0 );
            return;
        }
        else {

            LOG( 3 ) << "could not find cursor " << id << " in cache for " << ns << endl;

            replyToQuery( ResultFlag_CursorNotFound , r.p() , r.m() , 0 , 0 , 0 );
            return;
        }
    }

    void Strategy::writeOp( int op , Request& r ) {

        // make sure we have a last error
        dassert(&LastError::get(cc()));

        OwnedPointerVector<BatchedCommandRequest> requestsOwned;
        vector<BatchedCommandRequest*>& requests = requestsOwned.mutableVector();

        msgToBatchRequests( r.m(), &requests );

        for ( vector<BatchedCommandRequest*>::iterator it = requests.begin();
            it != requests.end(); ++it ) {

            // Multiple commands registered to last error as multiple requests
            if ( it != requests.begin() )
                LastError::get(cc()).startRequest();

            BatchedCommandRequest* request = *it;

            // Adjust namespaces for command
            NamespaceString fullNS( request->getNS() );
            string cmdNS = fullNS.getCommandNS();
            // We only pass in collection name to command
            request->setNS( fullNS.coll() );

            BSONObjBuilder builder;
            BSONObj requestBSON = request->toBSON();

            {
                // Disable the last error object for the duration of the write cmd
                LastError::Disabled disableLastError(&LastError::get(cc()));
                Command::runAgainstRegistered( cmdNS.c_str(), requestBSON, builder, 0 );
            }

            BatchedCommandResponse response;
            bool parsed = response.parseBSON( builder.done(), NULL );
            (void) parsed; // for compile
            dassert( parsed && response.isValid( NULL ) );

            // Populate the lastError object based on the write response
            LastError::get(cc()).reset();
            bool hadError = batchErrorToLastError(*request, response, &LastError::get(cc()));

            // Check if this is an ordered batch and we had an error which should stop processing
            if ( request->getOrdered() && hadError )
                break;
        }
    }

}