path: root/src/mongo/s/write_ops/batch_downconvert.cpp

/**
 *    Copyright (C) 2013 MongoDB 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/write_ops/batch_downconvert.h"

#include "mongo/bson/util/builder.h"
#include "mongo/db/write_concern_options.h"
#include "mongo/util/assert_util.h"
#include "mongo/util/log.h"

namespace mongo {

    Status extractGLEErrors( const BSONObj& gleResponse, GLEErrors* errors ) {

        // DRAGONS
        // Parsing GLE responses is incredibly finicky.
        // The order of testing here is extremely important.

        ///////////////////////////////////////////////////////////////////////
        // IMPORTANT!
        // Also update extractGLEErrors in batch_api.js for any changes made here.

        const bool isOK = gleResponse["ok"].trueValue();
        const string err = gleResponse["err"].str();
        const string errMsg = gleResponse["errmsg"].str();
        const string wNote = gleResponse["wnote"].str();
        const string jNote = gleResponse["jnote"].str();
        const int code = gleResponse["code"].numberInt();
        const bool timeout = gleResponse["wtimeout"].trueValue();

        if ( err == "norepl" || err == "noreplset" ) {
            // Know this is legacy gle and the repl not enforced - write concern error in 2.4
            errors->wcError.reset( new WCErrorDetail );
            errors->wcError->setErrCode( ErrorCodes::WriteConcernFailed );
            if ( !errMsg.empty() ) {
                errors->wcError->setErrMessage( errMsg );
            }
            else if ( !wNote.empty() ) {
                errors->wcError->setErrMessage( wNote );
            }
            else {
                errors->wcError->setErrMessage( err );
            }
        }
        else if ( timeout ) {
            // Know there was no write error
            errors->wcError.reset( new WCErrorDetail );
            errors->wcError->setErrCode( ErrorCodes::WriteConcernFailed );
            if ( !errMsg.empty() ) {
                errors->wcError->setErrMessage( errMsg );
            }
            else {
                errors->wcError->setErrMessage( err );
            }
            errors->wcError->setErrInfo( BSON( "wtimeout" << true ) );
        }
        else if ( code == 10990 /* no longer primary */
                  || code == 16805 /* replicatedToNum no longer primary */
                  || code == 14830 /* gle wmode changed / invalid */
                  // 2.6 Error codes
                  || code == ErrorCodes::NotMaster
                  || code == ErrorCodes::UnknownReplWriteConcern
                  || code == ErrorCodes::WriteConcernFailed ) {
            // Write concern errors that get returned as regular errors (result may not be ok: 1.0)
            errors->wcError.reset( new WCErrorDetail );
            errors->wcError->setErrCode( code );
            errors->wcError->setErrMessage( errMsg );
        }
        else if ( !isOK ) {

            //
            // !!! SOME GLE ERROR OCCURRED, UNKNOWN WRITE RESULT !!!
            //

            return Status( DBException::convertExceptionCode(
                               code ? code : ErrorCodes::UnknownError ),
                           errMsg );
        }
        else if ( !err.empty() ) {
            // Write error
            errors->writeError.reset( new WriteErrorDetail );
            int writeErrorCode = code == 0 ? ErrorCodes::UnknownError : code;

            // COMPATIBILITY
            // Certain clients expect write commands to always report 11000 for duplicate key
            // errors, while legacy GLE can return additional codes.
            if ( writeErrorCode == 11001 /* dup key in update */
                 || writeErrorCode == 12582 /* dup key capped */) {
                writeErrorCode = ErrorCodes::DuplicateKey;
            }

            errors->writeError->setErrCode( writeErrorCode );
            errors->writeError->setErrMessage( err );
        }
        else if ( !jNote.empty() ) {
            // Know this is legacy gle and the journaling not enforced - write concern error in 2.4
            errors->wcError.reset( new WCErrorDetail );
            errors->wcError->setErrCode( ErrorCodes::WriteConcernFailed );
            errors->wcError->setErrMessage( jNote );
        }

        // See if we had a version error reported as a writeback id - this is the only kind of
        // write error where the write concern may still be enforced.
        // The actual version that was stale is lost in the writeback itself.
        const int opsSinceWriteback = gleResponse["writebackSince"].numberInt();
        const bool hadWriteback = !gleResponse["writeback"].eoo();

        if ( hadWriteback && opsSinceWriteback == 0 ) {

            // We shouldn't have a previous write error
            dassert( !errors->writeError.get() );
            if ( errors->writeError.get() ) {
                // Somehow there was a write error *and* a writeback from the last write
                warning() << "both a write error and a writeback were reported "
                          << "when processing a legacy write: " << errors->writeError->toBSON()
                          << endl;
            }

            errors->writeError.reset( new WriteErrorDetail );
            errors->writeError->setErrCode( ErrorCodes::StaleShardVersion );
            errors->writeError->setErrInfo( BSON( "downconvert" << true ) ); // For debugging
            errors->writeError->setErrMessage( "shard version was stale" );
        }

        return Status::OK();
    }

    /**
     * Suppress the "err" and "code" field if they are coming from a previous write error and
     * are not related to write concern.  Also removes any write stats information (e.g. "n")
     *
     * Also, In some cases, 2.4 GLE w/ wOpTime can give us duplicate "err" and "code" fields b/c of
     * reporting a previous error.  The later field is what we want - dedup and use later field.
     *
     * Returns the stripped GLE response.
     */
    BSONObj stripNonWCInfo( const BSONObj& gleResponse ) {

        BSONObjIterator it( gleResponse );
        BSONObjBuilder builder;

        BSONElement codeField; // eoo
        BSONElement errField; // eoo

        while ( it.more() ) {
            BSONElement el = it.next();
            StringData fieldName( el.fieldName() );
            if ( fieldName.compare( "err" ) == 0 ) {
                errField = el;
            }
            else if ( fieldName.compare( "code" ) == 0 ) {
                codeField = el;
            }
            else if ( fieldName.compare( "n" ) == 0 || fieldName.compare( "nModified" ) == 0
                      || fieldName.compare( "upserted" ) == 0
                      || fieldName.compare( "updatedExisting" ) == 0 ) {
                // Suppress field
            }
            else {
                builder.append( el );
            }
        }

        if ( !codeField.eoo() ) {
            if ( !gleResponse["ok"].trueValue() ) {
                // The last code will be from the write concern
                builder.append( codeField );
            }
            else {
                // The code is from a non-wc error on this connection - suppress it
            }
        }

        if ( !errField.eoo() ) {
            string err = errField.str();
            if ( err == "norepl" || err == "noreplset" || err == "timeout" ) {
                // Append err if it's from a write concern issue
                builder.append( errField );
            }
            else {
                // Suppress non-write concern err as null, but we need to report null err if ok
                if ( gleResponse["ok"].trueValue() )
                    builder.appendNull( errField.fieldName() );
            }
        }

        return builder.obj();
    }

    // Adds a wOpTime and a wElectionId field to a set of gle options
    static BSONObj buildGLECmdWithOpTime( const BSONObj& gleOptions,
                                          const OpTime& opTime,
                                          const OID& electionId ) {
        BSONObjBuilder builder;
        BSONObjIterator it( gleOptions );

        for ( int i = 0; it.more(); ++i ) {
            BSONElement el = it.next();

            // Make sure first element is getLastError : 1
            if ( i == 0 ) {
                StringData elName( el.fieldName() );
                if ( !elName.equalCaseInsensitive( "getLastError" ) ) {
                    builder.append( "getLastError", 1 );
                }
            }

            builder.append( el );
        }
        builder.appendTimestamp( "wOpTime", opTime.asDate() );
        builder.appendOID( "wElectionId", const_cast<OID*>(&electionId) );
        return builder.obj();
    }

    Status enforceLegacyWriteConcern( MultiCommandDispatch* dispatcher,
                                      const StringData& dbName,
                                      const BSONObj& options,
                                      const HostOpTimeMap& hostOpTimes,
                                      vector<LegacyWCResponse>* legacyWCResponses ) {

        if ( hostOpTimes.empty() ) {
            return Status::OK();
        }

        for ( HostOpTimeMap::const_iterator it = hostOpTimes.begin(); it != hostOpTimes.end();
            ++it ) {

            const ConnectionString& shardEndpoint = it->first;
            const HostOpTime hot = it->second;
            const OpTime& opTime = hot.opTime;
            const OID& electionId = hot.electionId;

            LOG( 3 ) << "enforcing write concern " << options << " on " << shardEndpoint.toString()
                     << " at opTime " << opTime.toStringPretty() << " with electionID "
                     << electionId;

            BSONObj gleCmd = buildGLECmdWithOpTime( options, opTime, electionId );

            RawBSONSerializable gleCmdSerial( gleCmd );
            dispatcher->addCommand( shardEndpoint, dbName, gleCmdSerial );
        }

        dispatcher->sendAll();

        vector<Status> failedStatuses;

        while ( dispatcher->numPending() > 0 ) {

            ConnectionString shardEndpoint;
            RawBSONSerializable gleResponseSerial;

            Status dispatchStatus = dispatcher->recvAny( &shardEndpoint, &gleResponseSerial );
            if ( !dispatchStatus.isOK() ) {
                // We need to get all responses before returning
                failedStatuses.push_back( dispatchStatus );
                continue;
            }

            BSONObj gleResponse = stripNonWCInfo( gleResponseSerial.toBSON() );

            // Use the downconversion tools to determine if this GLE response is ok, a
            // write concern error, or an unknown error we should immediately abort for.
            GLEErrors errors;
            Status extractStatus = extractGLEErrors( gleResponse, &errors );
            if ( !extractStatus.isOK() ) {
                failedStatuses.push_back( extractStatus );
                continue;
            }

            LegacyWCResponse wcResponse;
            wcResponse.shardHost = shardEndpoint.toString();
            wcResponse.gleResponse = gleResponse;
            if ( errors.wcError.get() ) {
                wcResponse.errToReport = errors.wcError->getErrMessage();
            }

            legacyWCResponses->push_back( wcResponse );
        }

        if ( failedStatuses.empty() ) {
            return Status::OK();
        }

        StringBuilder builder;
        builder << "could not enforce write concern";

        for ( vector<Status>::const_iterator it = failedStatuses.begin();
            it != failedStatuses.end(); ++it ) {
            const Status& failedStatus = *it;
            if ( it == failedStatuses.begin() ) {
                builder << causedBy( failedStatus.toString() );
            }
            else {
                builder << ":: and ::" << failedStatus.toString();
            }
        }

        return Status( failedStatuses.size() == 1u ? failedStatuses.front().code() : 
                                                     ErrorCodes::MultipleErrorsOccurred, 
                       builder.str() );
    }
}