/** * Copyright (C) 2018-present MongoDB, Inc. * * This program is free software: you can redistribute it and/or modify * it under the terms of the Server Side Public License, version 1, * as published by MongoDB, Inc. * * 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 * Server Side Public License for more details. * * You should have received a copy of the Server Side Public License * along with this program. If not, see * . * * 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 Server Side 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_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kCommand #include "mongo/platform/basic.h" #include "mongo/bson/simple_bsonelement_comparator.h" #include "mongo/db/catalog/collection_catalog.h" #include "mongo/db/catalog/database.h" #include "mongo/db/catalog/database_holder.h" #include "mongo/db/catalog/health_log_interface.h" #include "mongo/db/catalog/index_catalog.h" #include "mongo/db/db_raii.h" #include "mongo/db/operation_context.h" #include "mongo/db/query/internal_plans.h" #include "mongo/db/record_id_helpers.h" #include "mongo/db/repl/dbcheck.h" #include "mongo/db/repl/dbcheck_gen.h" #include "mongo/db/repl/oplog.h" #include "mongo/db/repl/optime.h" #include "mongo/db/repl/repl_server_parameters_gen.h" #include "mongo/logv2/log.h" namespace mongo { MONGO_FAIL_POINT_DEFINE(SleepDbCheckInBatch); MONGO_FAIL_POINT_DEFINE(hangAfterGeneratingHashForExtraIndexKeysCheck); namespace { /* * Some utilities for dealing with the expected/found documents in health log entries. */ bool operator==(const std::vector& lhs, const std::vector& rhs) { return std::equal(lhs.cbegin(), lhs.cend(), rhs.cbegin(), rhs.cend(), [](const auto& x, const auto& y) -> bool { return x.woCompare(y) == 0; }); } /** * Get whether the expected and found objects match, plus an expected/found object to report to the * health log. */ template std::pair expectedFound(const T& expected, const T& found) { auto obj = BSON("expected" << expected << "found" << found); return std::pair(expected == found, obj); } template <> std::pair expectedFound(const BSONObj& expected, const BSONObj& found) { auto obj = BSON("expected" << expected << "found" << found); return std::pair(expected.woCompare(found) == 0, obj); } /** * An overload for boost::optionals, which omits boost::none fields. */ template std::pair expectedFound(const boost::optional& expected, const boost::optional& found) { BSONObjBuilder builder; if (expected) { builder << "expected" << *expected; } if (found) { builder << "found" << *found; } auto obj = builder.obj(); if (expected && found) { return std::pair(*expected == *found, obj); } else if (expected || found) { return std::pair(false, obj); } return std::pair(true, obj); } std::string renderForHealthLog(OplogEntriesEnum op) { switch (op) { case OplogEntriesEnum::Batch: return "dbCheckBatch"; case OplogEntriesEnum::Collection: return "dbCheckCollection"; case OplogEntriesEnum::Start: return "dbCheckStart"; case OplogEntriesEnum::Stop: return "dbCheckStop"; } MONGO_UNREACHABLE; } } // namespace /** * Fills in the timestamp and scope, which are always the same for dbCheck's entries. */ std::unique_ptr dbCheckHealthLogEntry(const boost::optional& nss, SeverityEnum severity, const std::string& msg, OplogEntriesEnum operation, const boost::optional& data) { auto entry = std::make_unique(); if (nss) { entry->setNss(*nss); } entry->setTimestamp(Date_t::now()); entry->setSeverity(severity); entry->setScope(ScopeEnum::Cluster); entry->setMsg(msg); entry->setOperation(renderForHealthLog(operation)); if (data) { entry->setData(*data); } return entry; } /** * Get an error message if the check fails. */ std::unique_ptr dbCheckErrorHealthLogEntry( const boost::optional& nss, const std::string& msg, OplogEntriesEnum operation, const Status& err, const BSONObj& context) { return dbCheckHealthLogEntry( nss, SeverityEnum::Error, msg, operation, BSON("success" << false << "error" << err.toString() << "context" << context)); } std::unique_ptr dbCheckWarningHealthLogEntry(const NamespaceString& nss, const std::string& msg, OplogEntriesEnum operation, const Status& err) { return dbCheckHealthLogEntry(nss, SeverityEnum::Warning, msg, operation, BSON("success" << false << "error" << err.toString())); } /** * Get a HealthLogEntry for a dbCheck batch. */ std::unique_ptr dbCheckBatchEntry( const NamespaceString& nss, int64_t count, int64_t bytes, const std::string& expectedHash, const std::string& foundHash, const BSONKey& minKey, const BSONKey& maxKey, const boost::optional& readTimestamp, const repl::OpTime& optime, const boost::optional& options) { auto hashes = expectedFound(expectedHash, foundHash); BSONObjBuilder builder; builder.append("success", true); builder.append("count", count); builder.append("bytes", bytes); builder.append("md5", hashes.second); builder.appendAs(minKey.elem(), "minKey"); builder.appendAs(maxKey.elem(), "maxKey"); if (readTimestamp) { builder.append("readTimestamp", *readTimestamp); } builder.append("optime", optime.toBSON()); const auto hashesMatch = hashes.first; const auto severity = [&] { if (hashesMatch) { return SeverityEnum::Info; } // Implcitily replicated collections and capped collections not replicating truncation are // not designed to be consistent, so inconsistency is not necessarily pathological. if (nss.isChangeStreamPreImagesCollection() || nss.isConfigImagesCollection() || (options && options->capped)) { return SeverityEnum::Warning; } return SeverityEnum::Error; }(); std::string msg = "dbCheck batch " + (hashesMatch ? std::string("consistent") : std::string("inconsistent")); return dbCheckHealthLogEntry(nss, severity, msg, OplogEntriesEnum::Batch, builder.obj()); } DbCheckHasher::DbCheckHasher(OperationContext* opCtx, const CollectionPtr& collection, const BSONKey& start, const BSONKey& end, int64_t maxCount, int64_t maxBytes) : _opCtx(opCtx), _maxKey(end), _maxCount(maxCount), _maxBytes(maxBytes) { // Get the MD5 hasher set up. md5_init(&_state); if (!collection->isClustered()) { // Get the _id index. const IndexDescriptor* desc = collection->getIndexCatalog()->findIdIndex(opCtx); uassert(ErrorCodes::IndexNotFound, "dbCheck needs _id index", desc); // Set up a simple index scan on that. _exec = InternalPlanner::indexScan(opCtx, &collection, desc, start.obj(), end.obj(), BoundInclusion::kIncludeEndKeyOnly, PlanYieldPolicy::YieldPolicy::NO_YIELD, InternalPlanner::FORWARD, InternalPlanner::IXSCAN_FETCH); } else { CollectionScanParams params; params.minRecord = RecordIdBound(uassertStatusOK( record_id_helpers::keyForDoc(start.obj(), collection->getClusteredInfo()->getIndexSpec(), collection->getDefaultCollator()))); params.maxRecord = RecordIdBound(uassertStatusOK( record_id_helpers::keyForDoc(end.obj(), collection->getClusteredInfo()->getIndexSpec(), collection->getDefaultCollator()))); params.boundInclusion = CollectionScanParams::ScanBoundInclusion::kIncludeEndRecordOnly; _exec = InternalPlanner::collectionScan( opCtx, &collection, params, PlanYieldPolicy::YieldPolicy::NO_YIELD); } } template const md5_byte_t* md5Cast(const T* ptr) { return reinterpret_cast(ptr); } void maybeAppend(md5_state_t* state, const boost::optional& uuid) { if (uuid) { md5_append(state, md5Cast(uuid->toCDR().data()), uuid->toCDR().length()); } } Status DbCheckHasher::hashAll(OperationContext* opCtx, Date_t deadline) { BSONObj currentObj; PlanExecutor::ExecState lastState; while (PlanExecutor::ADVANCED == (lastState = _exec->getNext(¤tObj, nullptr))) { SleepDbCheckInBatch.execute([opCtx](const BSONObj& data) { int sleepMs = data["sleepMs"].safeNumberInt(); opCtx->sleepFor(Milliseconds(sleepMs)); }); if (!currentObj.hasField("_id")) { return Status(ErrorCodes::NoSuchKey, "Document missing _id"); } // If this would put us over a limit, stop here. if (!_canHash(currentObj)) { return Status::OK(); } // Update `last` every time. _last = BSONKey::parseFromBSON(currentObj["_id"]); _bytesSeen += currentObj.objsize(); _countSeen += 1; md5_append(&_state, md5Cast(currentObj.objdata()), currentObj.objsize()); if (Date_t::now() > deadline) { break; } } // If we got to the end of the collection, set the last key to MaxKey. if (lastState == PlanExecutor::IS_EOF) { _last = _maxKey; } return Status::OK(); } std::string DbCheckHasher::total(void) { md5digest digest; md5_finish(&_state, digest); return digestToString(digest); } BSONKey DbCheckHasher::lastKey(void) const { return _last; } int64_t DbCheckHasher::bytesSeen(void) const { return _bytesSeen; } int64_t DbCheckHasher::docsSeen(void) const { return _countSeen; } bool DbCheckHasher::_canHash(const BSONObj& obj) { // Make sure we hash at least one document. if (_countSeen == 0) { return true; } // Check that this won't push us over our byte limit if (_bytesSeen + obj.objsize() > _maxBytes) { return false; } // or our document limit. if (_countSeen + 1 > _maxCount) { return false; } return true; } namespace { // Cumulative number of batches processed. Can wrap around; it's not guaranteed to be in lockstep // with other replica set members. unsigned int batchesProcessed = 0; Status dbCheckBatchOnSecondary(OperationContext* opCtx, const repl::OpTime& optime, const DbCheckOplogBatch& entry) { const auto msg = "replication consistency check"; // Set up the hasher, boost::optional hasher; try { auto lockMode = MODE_S; if (entry.getReadTimestamp()) { lockMode = MODE_IS; opCtx->recoveryUnit()->setTimestampReadSource(RecoveryUnit::ReadSource::kProvided, entry.getReadTimestamp()); } AutoGetCollection coll(opCtx, entry.getNss(), lockMode); const auto& collection = coll.getCollection(); if (!collection) { const auto msg = "Collection under dbCheck no longer exists"; auto logEntry = dbCheckHealthLogEntry(entry.getNss(), SeverityEnum::Info, "dbCheck failed", OplogEntriesEnum::Batch, BSON("success" << false << "info" << msg)); HealthLogInterface::get(opCtx)->log(*logEntry); return Status::OK(); } hasher.emplace(opCtx, collection, entry.getMinKey(), entry.getMaxKey()); uassertStatusOK(hasher->hashAll(opCtx)); std::string expected = entry.getMd5().toString(); std::string found = hasher->total(); auto logEntry = dbCheckBatchEntry(entry.getNss(), hasher->docsSeen(), hasher->bytesSeen(), expected, found, entry.getMinKey(), hasher->lastKey(), entry.getReadTimestamp(), optime, collection->getCollectionOptions()); batchesProcessed++; if (kDebugBuild || logEntry->getSeverity() != SeverityEnum::Info || (batchesProcessed % gDbCheckHealthLogEveryNBatches.load() == 0)) { // On debug builds, health-log every batch result; on release builds, health-log // every N batches. HealthLogInterface::get(opCtx)->log(*logEntry); } if (MONGO_unlikely(hangAfterGeneratingHashForExtraIndexKeysCheck.shouldFail())) { LOGV2_DEBUG(3083200, 3, "Hanging due to hangAfterGeneratingHashForExtraIndexKeysCheck failpoint"); // hangAfterGeneratingHashForExtraIndexKeysCheck.pauseWhileSet(opCtx); opCtx->sleepFor(Milliseconds(1000)); } } catch (const DBException& exception) { // In case of an error, report it to the health log, auto logEntry = dbCheckErrorHealthLogEntry( entry.getNss(), msg, OplogEntriesEnum::Batch, exception.toStatus(), entry.toBSON()); HealthLogInterface::get(opCtx)->log(*logEntry); return Status::OK(); } return Status::OK(); } } // namespace namespace repl { /* * The corresponding command run during command application. */ Status dbCheckOplogCommand(OperationContext* opCtx, const repl::OplogEntry& entry, OplogApplication::Mode mode) { const auto& cmd = entry.getObject(); OpTime opTime; if (!opCtx->writesAreReplicated()) { opTime = entry.getOpTime(); } auto type = OplogEntries_parse(IDLParserErrorContext("type"), cmd.getStringField("type")); IDLParserErrorContext ctx("o"); auto skipDbCheck = mode != OplogApplication::Mode::kSecondary; std::string oplogApplicationMode; if (mode == OplogApplication::Mode::kInitialSync) { oplogApplicationMode = "initial sync"; } else if (mode == OplogApplication::Mode::kUnstableRecovering) { oplogApplicationMode = "unstable recovering"; } else if (mode == OplogApplication::Mode::kStableRecovering) { oplogApplicationMode = "stable recovering"; } else if (mode == OplogApplication::Mode::kApplyOpsCmd) { oplogApplicationMode = "applyOps"; } else { oplogApplicationMode = "secondary"; } switch (type) { case OplogEntriesEnum::Batch: { auto invocation = DbCheckOplogBatch::parse(ctx, cmd); /* // TODO SERVER-78399: Clean up handling minKey/maxKey once feature flag is removed. // If the dbcheck oplog entry doesn't contain batchStart, convert minKey to a BSONObj to // be used as batchStart. BSONObj batchStart, batchEnd, batchId; if (!invocation.getBatchStart()) { batchStart = BSON("_id" << invocation.getMinKey().elem()); } else { batchStart = invocation.getBatchStart().get(); } if (!invocation.getBatchEnd()) { batchEnd = BSON("_id" << invocation.getMaxKey().elem()); } else { batchEnd = invocation.getBatchEnd().get(); } */ if (!skipDbCheck && !repl::skipApplyingDbCheckBatchOnSecondary.load()) { return dbCheckBatchOnSecondary(opCtx, opTime, invocation); } // TODO SERVER-89921: Uncomment once the relevant tickets are backported. /* if (invocation.getBatchId()) { batchId = invocation.getBatchId().get().toBSON(); } BSONObjBuilder data; data.append("batchStart", batchStart); data.append("batchEnd", batchEnd); if (!batchId.isEmpty()) { data.append("batchId", batchId); } */ auto warningMsg = "cannot execute dbcheck due to ongoing " + oplogApplicationMode; if (repl::skipApplyingDbCheckBatchOnSecondary.load()) { warningMsg = "skipping applying dbcheck batch because the " "'skipApplyingDbCheckBatchOnSecondary' parameter is on"; } LOGV2_DEBUG(8888500, 3, "skipping applying dbcheck batch", "reason"_attr = warningMsg); // TODO SERVER-89921: Uncomment these logging attributes once the relevant tickets are // backported. //"batchStart"_attr = batchStart, //"batchEnd"_attr = batchEnd, //"batchId"_attr = batchId); auto healthLogEntry = mongo::dbCheckHealthLogEntry( invocation.getNss(), SeverityEnum::Warning, warningMsg, type, boost::none /*data*/); HealthLogInterface::get(Client::getCurrent()->getServiceContext()) ->log(*healthLogEntry); return Status::OK(); } case OplogEntriesEnum::Collection: { // TODO SERVER-61963. return Status::OK(); } case OplogEntriesEnum::Start: // fallthrough case OplogEntriesEnum::Stop: const auto healthLogEntry = mongo::dbCheckHealthLogEntry( boost::none /*nss*/, skipDbCheck ? SeverityEnum::Warning : SeverityEnum::Info, skipDbCheck ? "cannot execute dbcheck due to ongoing " + oplogApplicationMode : "", type, boost::none /*data*/ ); // TODO: need to change this HealthLogInterface::get(Client::getCurrent()->getServiceContext()) ->log(*healthLogEntry); return Status::OK(); } MONGO_UNREACHABLE; } } // namespace repl } // namespace mongo