path: root/src/mongo/db/commands/cleanup_orphaned_cmd.cpp

/**
 *    Copyright (C) 2013 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::kCommand

#include "mongo/platform/basic.h"

#include <string>
#include <vector>

#include "mongo/base/init.h"
#include "mongo/db/auth/action_type.h"
#include "mongo/db/auth/authorization_manager.h"
#include "mongo/db/auth/authorization_session.h"
#include "mongo/db/auth/privilege.h"
#include "mongo/db/commands.h"
#include "mongo/db/field_parser.h"
#include "mongo/db/jsobj.h"
#include "mongo/db/namespace_string.h"
#include "mongo/db/range_arithmetic.h"
#include "mongo/db/range_deleter_service.h"
#include "mongo/db/repl/replication_coordinator_global.h"
#include "mongo/s/collection_metadata.h"
#include "mongo/s/d_state.h"
#include "mongo/util/log.h"

namespace {
using mongo::WriteConcernOptions;

const int kDefaultWTimeoutMs = 60 * 1000;
const WriteConcernOptions DefaultWriteConcern(WriteConcernOptions::kMajority,
                                              WriteConcernOptions::NONE,
                                              kDefaultWTimeoutMs);
}

namespace mongo {

using std::endl;
using std::string;

using mongoutils::str::stream;

enum CleanupResult { CleanupResult_Done, CleanupResult_Continue, CleanupResult_Error };

/**
 * Cleans up one range of orphaned data starting from a range that overlaps or starts at
 * 'startingFromKey'.  If empty, startingFromKey is the minimum key of the sharded range.
 *
 * @return CleanupResult_Continue and 'stoppedAtKey' if orphaned range was found and cleaned
 * @return CleanupResult_Done if no orphaned ranges remain
 * @return CleanupResult_Error and 'errMsg' if an error occurred
 *
 * If the collection is not sharded, returns CleanupResult_Done.
 */
CleanupResult cleanupOrphanedData(OperationContext* txn,
                                  const NamespaceString& ns,
                                  const BSONObj& startingFromKeyConst,
                                  const WriteConcernOptions& secondaryThrottle,
                                  BSONObj* stoppedAtKey,
                                  string* errMsg) {
    BSONObj startingFromKey = startingFromKeyConst;

    CollectionMetadataPtr metadata = shardingState.getCollectionMetadata(ns.toString());
    if (!metadata || metadata->getKeyPattern().isEmpty()) {
        warning() << "skipping orphaned data cleanup for " << ns.toString()
                  << ", collection is not sharded" << endl;

        return CleanupResult_Done;
    }

    BSONObj keyPattern = metadata->getKeyPattern();
    if (!startingFromKey.isEmpty()) {
        if (!metadata->isValidKey(startingFromKey)) {
            *errMsg = stream() << "could not cleanup orphaned data, start key " << startingFromKey
                               << " does not match shard key pattern " << keyPattern;

            warning() << *errMsg << endl;
            return CleanupResult_Error;
        }
    } else {
        startingFromKey = metadata->getMinKey();
    }

    KeyRange orphanRange;
    if (!metadata->getNextOrphanRange(startingFromKey, &orphanRange)) {
        LOG(1) << "orphaned data cleanup requested for " << ns.toString() << " starting from "
               << startingFromKey << ", no orphan ranges remain" << endl;

        return CleanupResult_Done;
    }
    orphanRange.ns = ns;
    *stoppedAtKey = orphanRange.maxKey;

    // We're done with this metadata now, no matter what happens
    metadata.reset();

    LOG(1) << "orphaned data cleanup requested for " << ns.toString() << " starting from "
           << startingFromKey << ", removing next orphan range"
           << " [" << orphanRange.minKey << "," << orphanRange.maxKey << ")" << endl;

    // Metadata snapshot may be stale now, but deleter checks metadata again in write lock
    // before delete.
    RangeDeleterOptions deleterOptions(orphanRange);
    deleterOptions.writeConcern = secondaryThrottle;
    deleterOptions.onlyRemoveOrphanedDocs = true;
    deleterOptions.fromMigrate = true;
    // Must wait for cursors since there can be existing cursors with an older
    // CollectionMetadata.
    deleterOptions.waitForOpenCursors = true;
    deleterOptions.removeSaverReason = "cleanup-cmd";

    if (!getDeleter()->deleteNow(txn, deleterOptions, errMsg)) {
        warning() << *errMsg << endl;
        return CleanupResult_Error;
    }

    return CleanupResult_Continue;
}

/**
 * Cleanup orphaned data command.  Called on a particular namespace, and if the collection
 * is sharded will clean up a single orphaned data range which overlaps or starts after a
 * passed-in 'startingFromKey'.  Returns true and a 'stoppedAtKey' (which will start a
 * search for the next orphaned range if the command is called again) or no key if there
 * are no more orphaned ranges in the collection.
 *
 * If the collection is not sharded, returns true but no 'stoppedAtKey'.
 * On failure, returns false and an error message.
 *
 * Calling this command repeatedly until no 'stoppedAtKey' is returned ensures that the
 * full collection range is searched for orphaned documents, but since sharding state may
 * change between calls there is no guarantee that all orphaned documents were found unless
 * the balancer is off.
 *
 * Safe to call with the balancer on.
 *
 * Format:
 *
 * {
 *      cleanupOrphaned: <ns>,
 *      // optional parameters:
 *      startingAtKey: { <shardKeyValue> }, // defaults to lowest value
 *      secondaryThrottle: <bool>, // defaults to true
 *      // defaults to { w: "majority", wtimeout: 60000 }. Applies to individual writes.
 *      writeConcern: { <writeConcern options> }
 * }
 */
class CleanupOrphanedCommand : public Command {
public:
    CleanupOrphanedCommand() : Command("cleanupOrphaned") {}

    virtual bool slaveOk() const {
        return false;
    }
    virtual bool adminOnly() const {
        return true;
    }
    virtual bool localHostOnlyIfNoAuth(const BSONObj& cmdObj) {
        return false;
    }

    virtual Status checkAuthForCommand(ClientBasic* client,
                                       const std::string& dbname,
                                       const BSONObj& cmdObj) {
        if (!AuthorizationSession::get(client)->isAuthorizedForActionsOnResource(
                ResourcePattern::forClusterResource(), ActionType::cleanupOrphaned)) {
            return Status(ErrorCodes::Unauthorized, "Not authorized for cleanupOrphaned command.");
        }
        return Status::OK();
    }

    virtual bool isWriteCommandForConfigServer() const {
        return false;
    }

    // Input
    static BSONField<string> nsField;
    static BSONField<BSONObj> startingFromKeyField;

    // Output
    static BSONField<BSONObj> stoppedAtKeyField;

    bool run(OperationContext* txn,
             string const& db,
             BSONObj& cmdObj,
             int,
             string& errmsg,
             BSONObjBuilder& result) {
        string ns;
        if (!FieldParser::extract(cmdObj, nsField, &ns, &errmsg)) {
            return false;
        }

        if (ns == "") {
            errmsg = "no collection name specified";
            return false;
        }

        BSONObj startingFromKey;
        if (!FieldParser::extract(cmdObj, startingFromKeyField, &startingFromKey, &errmsg)) {
            return false;
        }

        WriteConcernOptions writeConcern;
        Status status = writeConcern.parseSecondaryThrottle(cmdObj, NULL);

        if (!status.isOK()) {
            if (status.code() != ErrorCodes::WriteConcernNotDefined) {
                return appendCommandStatus(result, status);
            }

            writeConcern = DefaultWriteConcern;
        } else {
            repl::ReplicationCoordinator* replCoordinator = repl::getGlobalReplicationCoordinator();
            Status status = replCoordinator->checkIfWriteConcernCanBeSatisfied(writeConcern);

            if (replCoordinator->getReplicationMode() ==
                    repl::ReplicationCoordinator::modeMasterSlave &&
                writeConcern.shouldWaitForOtherNodes()) {
                warning() << "cleanupOrphaned cannot check if write concern setting "
                          << writeConcern.toBSON()
                          << " can be enforced in a master slave configuration";
            }

            if (!status.isOK() && status != ErrorCodes::NoReplicationEnabled) {
                return appendCommandStatus(result, status);
            }
        }

        if (writeConcern.shouldWaitForOtherNodes() &&
            writeConcern.wTimeout == WriteConcernOptions::kNoTimeout) {
            // Don't allow no timeout.
            writeConcern.wTimeout = kDefaultWTimeoutMs;
        }

        if (!shardingState.enabled()) {
            errmsg = str::stream() << "server is not part of a sharded cluster or "
                                   << "the sharding metadata is not yet initialized.";
            return false;
        }

        ChunkVersion shardVersion;
        status = shardingState.refreshMetadataNow(txn, ns, &shardVersion);
        if (!status.isOK()) {
            if (status.code() == ErrorCodes::RemoteChangeDetected) {
                warning() << "Shard version in transition detected while refreshing "
                          << "metadata for " << ns << " at version " << shardVersion << endl;
            } else {
                errmsg = str::stream() << "failed to refresh shard metadata: " << status.reason();
                return false;
            }
        }

        BSONObj stoppedAtKey;
        CleanupResult cleanupResult = cleanupOrphanedData(
            txn, NamespaceString(ns), startingFromKey, writeConcern, &stoppedAtKey, &errmsg);

        if (cleanupResult == CleanupResult_Error) {
            return false;
        }

        if (cleanupResult == CleanupResult_Continue) {
            result.append(stoppedAtKeyField(), stoppedAtKey);
        } else {
            dassert(cleanupResult == CleanupResult_Done);
        }

        return true;
    }
};

BSONField<string> CleanupOrphanedCommand::nsField("cleanupOrphaned");
BSONField<BSONObj> CleanupOrphanedCommand::startingFromKeyField("startingFromKey");
BSONField<BSONObj> CleanupOrphanedCommand::stoppedAtKeyField("stoppedAtKey");

MONGO_INITIALIZER(RegisterCleanupOrphanedCommand)(InitializerContext* context) {
    // Leaked intentionally: a Command registers itself when constructed.
    new CleanupOrphanedCommand();
    return Status::OK();
}

}  // namespace mongo