diff options
Diffstat (limited to 'src/mongo/db/storage/wiredtiger')
21 files changed, 1107 insertions, 483 deletions
diff --git a/src/mongo/db/storage/wiredtiger/SConscript b/src/mongo/db/storage/wiredtiger/SConscript index 9d9f5ca18f4..9916a7f27c7 100644 --- a/src/mongo/db/storage/wiredtiger/SConscript +++ b/src/mongo/db/storage/wiredtiger/SConscript @@ -35,6 +35,7 @@ wtEnv.Library( 'wiredtiger_global_options.cpp', 'wiredtiger_index.cpp', 'wiredtiger_kv_engine.cpp', + 'wiredtiger_stats.cpp', 'wiredtiger_oplog_manager.cpp', 'wiredtiger_parameters.cpp', 'wiredtiger_prepare_conflict.cpp', @@ -52,10 +53,9 @@ wtEnv.Library( '$BUILD_DIR/mongo/db/catalog/collection', '$BUILD_DIR/mongo/db/catalog/collection_options', '$BUILD_DIR/mongo/db/concurrency/lock_manager', - '$BUILD_DIR/mongo/db/concurrency/write_conflict_exception', '$BUILD_DIR/mongo/db/curop', '$BUILD_DIR/mongo/db/global_settings', - '$BUILD_DIR/mongo/db/index/index_descriptor', + '$BUILD_DIR/mongo/db/index/index_access_method', '$BUILD_DIR/mongo/db/namespace_string', '$BUILD_DIR/mongo/db/prepare_conflict_tracker', '$BUILD_DIR/mongo/db/record_id_helpers', @@ -81,6 +81,7 @@ wtEnv.Library( LIBDEPS_PRIVATE=[ '$BUILD_DIR/mongo/db/catalog/database_holder', '$BUILD_DIR/mongo/db/commands/server_status', + '$BUILD_DIR/mongo/db/concurrency/exception_util', '$BUILD_DIR/mongo/db/db_raii', '$BUILD_DIR/mongo/db/mongod_options', '$BUILD_DIR/mongo/db/multitenancy', @@ -135,13 +136,14 @@ wtEnv.CppUnitTest( source=[ 'wiredtiger_init_test.cpp', 'wiredtiger_kv_engine_test.cpp', + 'wiredtiger_stats_test.cpp', 'wiredtiger_recovery_unit_test.cpp', 'wiredtiger_session_cache_test.cpp', 'wiredtiger_util_test.cpp', ], LIBDEPS=[ '$BUILD_DIR/mongo/db/auth/authmocks', - '$BUILD_DIR/mongo/db/index/index_access_methods', + '$BUILD_DIR/mongo/db/index/index_access_method', '$BUILD_DIR/mongo/db/repl/repl_coordinator_interface', '$BUILD_DIR/mongo/db/repl/replmocks', '$BUILD_DIR/mongo/db/service_context', diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_index.cpp b/src/mongo/db/storage/wiredtiger/wiredtiger_index.cpp index cd9a25c2930..09b1603361e 100644 --- a/src/mongo/db/storage/wiredtiger/wiredtiger_index.cpp +++ b/src/mongo/db/storage/wiredtiger/wiredtiger_index.cpp @@ -83,6 +83,7 @@ namespace { MONGO_FAIL_POINT_DEFINE(WTCompactIndexEBUSY); MONGO_FAIL_POINT_DEFINE(WTIndexPauseAfterSearchNear); +MONGO_FAIL_POINT_DEFINE(WTValidateIndexStructuralDamage); static const WiredTigerItem emptyItem(nullptr, 0); } // namespace @@ -102,7 +103,7 @@ void WiredTigerIndex::getKey(OperationContext* opCtx, WT_CURSOR* cursor, WT_ITEM StatusWith<std::string> WiredTigerIndex::parseIndexOptions(const BSONObj& options) { StringBuilder ss; BSONForEach(elem, options) { - if (elem.fieldNameStringData() == "configString") { + if (elem.fieldNameStringData() == WiredTigerUtil::kConfigStringField) { Status status = WiredTigerUtil::checkTableCreationOptions(elem); if (!status.isOK()) { return status; @@ -329,6 +330,15 @@ void WiredTigerIndex::fullValidate(OperationContext* opCtx, IndexValidateResults* fullResults) const { dassert(opCtx->lockState()->isReadLocked()); if (fullResults && !WiredTigerRecoveryUnit::get(opCtx)->getSessionCache()->isEphemeral()) { + if (WTValidateIndexStructuralDamage.shouldFail()) { + std::string msg = str::stream() << "verify() returned an error. " + << "This indicates structural damage. " + << "Not examining individual index entries."; + fullResults->errors.push_back(msg); + fullResults->valid = false; + return; + } + int err = WiredTigerUtil::verifyTable(opCtx, _uri, &(fullResults->errors)); if (err == EBUSY) { std::string msg = str::stream() @@ -356,25 +366,25 @@ void WiredTigerIndex::fullValidate(OperationContext* opCtx, } } - auto cursor = newCursor(opCtx); - long long count = 0; - LOGV2_TRACE_INDEX(20094, "fullValidate"); - - const auto requestedInfo = TRACING_ENABLED ? Cursor::kKeyAndLoc : Cursor::kJustExistance; - - KeyString::Value keyStringForSeek = - IndexEntryComparison::makeKeyStringFromBSONKeyForSeek(BSONObj(), - getKeyStringVersion(), - getOrdering(), - true, /* forward */ - true /* inclusive */ - ); - - for (auto kv = cursor->seek(keyStringForSeek, requestedInfo); kv; kv = cursor->next()) { - LOGV2_TRACE_INDEX(20095, "fullValidate {kv}", "kv"_attr = kv); - count++; - } if (numKeysOut) { + auto cursor = newCursor(opCtx); + long long count = 0; + LOGV2_TRACE_INDEX(20094, "fullValidate"); + + const auto requestedInfo = TRACING_ENABLED ? Cursor::kKeyAndLoc : Cursor::kJustExistance; + + KeyString::Value keyStringForSeek = + IndexEntryComparison::makeKeyStringFromBSONKeyForSeek(BSONObj(), + getKeyStringVersion(), + getOrdering(), + true, /* forward */ + true /* inclusive */ + ); + + for (auto kv = cursor->seek(keyStringForSeek, requestedInfo); kv; kv = cursor->next()) { + LOGV2_TRACE_INDEX(20095, "fullValidate {kv}", "kv"_attr = kv); + count++; + } *numKeysOut = count; } } @@ -445,6 +455,74 @@ bool WiredTigerIndex::isEmpty(OperationContext* opCtx) { return false; } +void WiredTigerIndex::printIndexEntryMetadata(OperationContext* opCtx, + const KeyString::Value& keyString) const { + // Printing the index entry metadata requires a new session. We cannot open other cursors when + // there are open history store cursors in the session. We also need to make sure that the + // existing session has not written data to avoid potential deadlocks. + invariant(!opCtx->lockState()->inAWriteUnitOfWork()); + WiredTigerSession session(WiredTigerRecoveryUnit::get(opCtx)->getSessionCache()->conn()); + + // Per the version cursor API: + // - A version cursor can only be called with the read timestamp as the oldest timestamp. + // - If there is no oldest timestamp, the version cursor can only be called with a read + // timestamp of 1. + // - If there is an oldest timestamp, reading at timestamp 1 will get rounded up. + const std::string config = "read_timestamp=1,roundup_timestamps=(read=true)"; + WiredTigerBeginTxnBlock beginTxn(session.getSession(), config.c_str()); + + // Open a version cursor. This is a debug cursor that enables iteration through the history of + // values for a given index entry. + WT_CURSOR* cursor = session.getNewCursor(_uri, "debug=(dump_version=true)"); + + const WiredTigerItem searchKey(keyString.getBuffer(), keyString.getSize()); + cursor->set_key(cursor, searchKey.Get()); + + int ret = cursor->search(cursor); + while (ret != WT_NOTFOUND) { + invariantWTOK(ret, cursor->session); + + uint64_t startTs = 0, startDurableTs = 0, stopTs = 0, stopDurableTs = 0; + uint64_t startTxnId = 0, stopTxnId = 0; + uint8_t flags = 0, location = 0, prepare = 0, type = 0; + WT_ITEM value; + + invariantWTOK(cursor->get_value(cursor, + &startTxnId, + &startTs, + &startDurableTs, + &stopTxnId, + &stopTs, + &stopDurableTs, + &type, + &prepare, + &flags, + &location, + &value), + cursor->session); + + auto indexKey = KeyString::toBson( + keyString.getBuffer(), keyString.getSize(), _ordering, keyString.getTypeBits()); + + LOGV2(6601200, + "WiredTiger index entry metadata", + "keyString"_attr = keyString, + "indexKey"_attr = indexKey, + "startTxnId"_attr = startTxnId, + "startTs"_attr = Timestamp(startTs), + "startDurableTs"_attr = Timestamp(startDurableTs), + "stopTxnId"_attr = stopTxnId, + "stopTs"_attr = Timestamp(stopTs), + "stopDurableTs"_attr = Timestamp(stopDurableTs), + "type"_attr = type, + "prepare"_attr = prepare, + "flags"_attr = flags, + "location"_attr = location); + + ret = cursor->next(cursor); + } +} + long long WiredTigerIndex::getSpaceUsedBytes(OperationContext* opCtx) const { dassert(opCtx->lockState()->isReadLocked()); auto ru = WiredTigerRecoveryUnit::get(opCtx); @@ -1192,16 +1270,12 @@ protected: LOGV2_TRACE_CURSOR(5683900, "cmp after advance: {cmp}", "cmp"_attr = cmp); - // We do not expect any exact matches or matches of prefixes by comparing keys of - // different lengths. Callers either seek using keys with discriminators that always - // compare unequally, or in the case of restoring a cursor, perform exact searches. In - // the case of an exact search, we will have returned earlier. - dassert(cmp); - if (enforcingPrepareConflicts) { // If we are enforcing prepare conflicts, calling next() or prev() must always give // us a key that compares, respectively, greater than or less than our search key. - dassert(_forward ? cmp > 0 : cmp < 0); + // An exact match is also possible in the case of _id indexes, because the recordid + // is not a part of the key. + dassert(_forward ? cmp >= 0 : cmp <= 0); } } @@ -1786,16 +1860,74 @@ void WiredTigerIndexUnique::_unindex(OperationContext* opCtx, return; } - // After a rolling upgrade an index can have keys from both timestamp unsafe (old) and - // timestamp safe (new) unique indexes. Old format keys just had the index key while new - // format key has index key + Record id. WT_NOTFOUND is possible if index key is in old format. - // Retry removal of key using old format. + // WT_NOTFOUND is possible if index key is in old (v4.0) format. Retry removal of key using old + // format. + _unindexTimestampUnsafe(opCtx, c, keyString, dupsAllowed); +} + +void WiredTigerIndexUnique::_unindexTimestampUnsafe(OperationContext* opCtx, + WT_CURSOR* c, + const KeyString::Value& keyString, + bool dupsAllowed) { + // The old unique index format had a key-value of indexKey-RecordId. This means that the + // RecordId in an index entry might not match the indexKey+RecordId keyString passed into this + // function: an index on a field where multiple collection documents have the same field value + // but only one passes the partial index filter. + // + // The dupsAllowed flag is no longer relevant for the old unique index format. No new index + // entries are written in the old format, let alone during temporary phases of the server when + // duplicates are allowed. + + const RecordId id = + KeyString::decodeRecordIdLongAtEnd(keyString.getBuffer(), keyString.getSize()); + invariant(id.isValid()); + auto sizeWithoutRecordId = KeyString::sizeWithoutRecordIdLongAtEnd(keyString.getBuffer(), keyString.getSize()); WiredTigerItem keyItem(keyString.getBuffer(), sizeWithoutRecordId); setKey(c, keyItem.Get()); - ret = WT_OP_CHECK(wiredTigerCursorRemove(opCtx, c)); + if (_partial) { + int ret = wiredTigerPrepareConflictRetry(opCtx, [&] { return c->search(c); }); + if (ret == WT_NOTFOUND) { + return; + } + invariantWTOK(ret, c->session); + + WT_ITEM value; + invariantWTOK(c->get_value(c, &value), c->session); + BufReader br(value.data, value.size); + fassert(40416, br.remaining()); + + // Check that the record id matches. We may be called to unindex records that are not + // present in the index due to the partial filter expression. + bool foundRecord = [&]() { + if (KeyString::decodeRecordIdLong(&br) != id) { + return false; + } + return true; + }(); + + // Ensure the index entry value is not a list of RecordIds, which should only be possible + // temporarily in v4.0 when dupsAllowed is true, not ever across upgrades or in upgraded + // versions. + KeyString::TypeBits::fromBuffer(getKeyStringVersion(), &br); + if (br.remaining()) { + LOGV2_FATAL_NOTRACE( + 7592201, + "An index entry was found that contains an unexpected old format that should no " + "longer exist. The index should be dropped and rebuilt.", + "indexName"_attr = _indexName, + "uri"_attr = uri(), + "collection"_attr = getCollectionNamespace(opCtx)); + } + + if (!foundRecord) { + return; + } + } + + int ret = WT_OP_CHECK(wiredTigerCursorRemove(opCtx, c)); if (ret == WT_NOTFOUND) { return; } diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_index.h b/src/mongo/db/storage/wiredtiger/wiredtiger_index.h index 5a94980c22c..8fccc7d8c24 100644 --- a/src/mongo/db/storage/wiredtiger/wiredtiger_index.h +++ b/src/mongo/db/storage/wiredtiger/wiredtiger_index.h @@ -156,6 +156,9 @@ public: virtual Status initAsEmpty(OperationContext* opCtx); + virtual void printIndexEntryMetadata(OperationContext* opCtx, + const KeyString::Value& keyString) const; + Status compact(OperationContext* opCtx) override; const std::string& uri() const { @@ -295,6 +298,16 @@ protected: const KeyString::Value& keyString, bool dupsAllowed) override; + /** + * This function continues to exist in order to support v4.0 unique partial index format: the + * format changed in v4.2 and onward. _unindex will call this if an index entry in the new + * format cannot be found, and this function will check for the old format. + */ + void _unindexTimestampUnsafe(OperationContext* opCtx, + WT_CURSOR* c, + const KeyString::Value& keyString, + bool dupsAllowed); + private: bool _partial; }; diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_init.cpp b/src/mongo/db/storage/wiredtiger/wiredtiger_init.cpp index da82969596e..a093fb70661 100644 --- a/src/mongo/db/storage/wiredtiger/wiredtiger_init.cpp +++ b/src/mongo/db/storage/wiredtiger/wiredtiger_init.cpp @@ -116,7 +116,6 @@ public: "RAM. See http://dochub.mongodb.org/core/faq-memory-diagnostics-wt"); } } - const bool ephemeral = false; auto kv = std::make_unique<WiredTigerKVEngine>(getCanonicalName().toString(), params.dbpath, @@ -125,7 +124,7 @@ public: cacheMB, wiredTigerGlobalOptions.getMaxHistoryFileSizeMB(), params.dur, - ephemeral, + params.ephemeral, params.repair, params.readOnly); kv->setRecordStoreExtraOptions(wiredTigerGlobalOptions.collectionConfig); diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.cpp b/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.cpp index 2c5a6ed5559..1dcac669b84 100644 --- a/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.cpp +++ b/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.cpp @@ -472,6 +472,12 @@ WiredTigerKVEngine::WiredTigerKVEngine(const std::string& canonicalName, ss << WiredTigerUtil::generateRestoreConfig() << ","; } + // If we've requested an ephemeral instance we store everything into memory instead of backing + // it onto disk. Logging is not supported in this instance, thus we also have to disable it. + if (_ephemeral) { + ss << "in_memory=true,log=(enabled=false),"; + } + string config = ss.str(); LOGV2(22315, "Opening WiredTiger", "config"_attr = config); auto startTime = Date_t::now(); @@ -719,7 +725,6 @@ void WiredTigerKVEngine::_openWiredTiger(const std::string& path, const std::str void WiredTigerKVEngine::cleanShutdown() { LOGV2(22317, "WiredTigerKVEngine shutting down"); - WiredTigerUtil::resetTableLoggingInfo(); if (!_conn) { return; @@ -2004,6 +2009,11 @@ bool WiredTigerKVEngine::supportsDirectoryPerDB() const { } void WiredTigerKVEngine::_checkpoint(WT_SESSION* session) { + // Ephemeral WiredTiger instances cannot do a checkpoint to disk as there is no disk backing + // the data. + if (_ephemeral) { + return; + } // TODO: SERVER-64507: Investigate whether we can smartly rely on one checkpointer if two or // more threads checkpoint at the same time. stdx::lock_guard lk(_checkpointMutex); @@ -2746,4 +2756,38 @@ Status WiredTigerKVEngine::reconfigureLogging() { return wtRCToStatus(_conn->reconfigure(_conn, verboseConfig.c_str()), nullptr); } +KeyFormat WiredTigerKVEngine::getKeyFormat(OperationContext* opCtx, StringData ident) const { + + const std::string wtTableConfig = + uassertStatusOK(WiredTigerUtil::getMetadataCreate(opCtx, "table:{}"_format(ident))); + return wtTableConfig.find("key_format=u") != string::npos ? KeyFormat::String : KeyFormat::Long; +} + +StatusWith<BSONObj> WiredTigerKVEngine::getSanitizedStorageOptionsForSecondaryReplication( + const BSONObj& options) const { + + // Skip inMemory storage engine, encryption at rest only applies to storage backed engine. + if (_ephemeral || options.isEmpty()) { + return options; + } + + auto firstElem = options.firstElement(); + if (firstElem.fieldName() != kWiredTigerEngineName) { + return Status(ErrorCodes::InvalidOptions, + str::stream() << "Expected \"" << kWiredTigerEngineName + << "\" field, but got: " << firstElem.fieldName()); + } + + BSONObj wtObj = firstElem.Obj(); + if (auto configStringElem = wtObj.getField(WiredTigerUtil::kConfigStringField)) { + auto configString = configStringElem.String(); + WiredTigerUtil::removeEncryptionFromConfigString(&configString); + // Return a new BSONObj with the configString field sanitized. + return options.addFields(BSON(kWiredTigerEngineName << wtObj.addFields(BSON( + WiredTigerUtil::kConfigStringField << configString)))); + } + + return options; +} + } // namespace mongo diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.h b/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.h index 688db855b74..52e1731b6ae 100644 --- a/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.h +++ b/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.h @@ -383,6 +383,11 @@ public: Status reconfigureLogging() override; + KeyFormat getKeyFormat(OperationContext* opCtx, StringData ident) const override; + + StatusWith<BSONObj> getSanitizedStorageOptionsForSecondaryReplication( + const BSONObj& options) const override; + private: class WiredTigerSessionSweeper; diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_oplog_manager.cpp b/src/mongo/db/storage/wiredtiger/wiredtiger_oplog_manager.cpp index fb3bc211faa..1be18b4c8ff 100644 --- a/src/mongo/db/storage/wiredtiger/wiredtiger_oplog_manager.cpp +++ b/src/mongo/db/storage/wiredtiger/wiredtiger_oplog_manager.cpp @@ -237,8 +237,6 @@ void WiredTigerOplogManager::_updateOplogVisibilityLoop(WiredTigerSessionCache* invariant(_triggerOplogVisibilityUpdate); _triggerOplogVisibilityUpdate = false; - lk.unlock(); - // Fetch the all_durable timestamp from the storage engine, which is guaranteed not to have // any holes behind it in-memory. const uint64_t newTimestamp = sessionCache->getKVEngine()->getAllDurableTimestamp().asULL(); @@ -254,7 +252,6 @@ void WiredTigerOplogManager::_updateOplogVisibilityLoop(WiredTigerSessionCache* continue; } - lk.lock(); // Publish the new timestamp value. Avoid going backward. auto currentVisibleTimestamp = getOplogReadTimestamp(); if (newTimestamp > currentVisibleTimestamp) { diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_record_store.cpp b/src/mongo/db/storage/wiredtiger/wiredtiger_record_store.cpp index ace8dd90cf8..7b00609ee77 100644 --- a/src/mongo/db/storage/wiredtiger/wiredtiger_record_store.cpp +++ b/src/mongo/db/storage/wiredtiger/wiredtiger_record_store.cpp @@ -44,8 +44,8 @@ #include "mongo/base/static_assert.h" #include "mongo/bson/util/builder.h" #include "mongo/db/catalog/validate_results.h" +#include "mongo/db/concurrency/exception_util.h" #include "mongo/db/concurrency/locker.h" -#include "mongo/db/concurrency/write_conflict_exception.h" #include "mongo/db/global_settings.h" #include "mongo/db/namespace_string.h" #include "mongo/db/operation_context.h" @@ -134,6 +134,7 @@ std::size_t computeRecordIdSize(const RecordId& id) { } // namespace MONGO_FAIL_POINT_DEFINE(WTCompactRecordStoreEBUSY); +MONGO_FAIL_POINT_DEFINE(WTRecordStoreUassertOutOfOrder); MONGO_FAIL_POINT_DEFINE(WTWriteConflictException); MONGO_FAIL_POINT_DEFINE(WTWriteConflictExceptionForReads); MONGO_FAIL_POINT_DEFINE(slowOplogSamplingReads); @@ -648,7 +649,7 @@ void WiredTigerRecordStore::OplogStones::adjust(int64_t maxSize) { StatusWith<std::string> WiredTigerRecordStore::parseOptionsField(const BSONObj options) { StringBuilder ss; BSONForEach(elem, options) { - if (elem.fieldNameStringData() == "configString") { + if (elem.fieldNameStringData() == WiredTigerUtil::kConfigStringField) { Status status = WiredTigerUtil::checkTableCreationOptions(elem); if (!status.isOK()) { return status; @@ -801,7 +802,11 @@ StatusWith<std::string> WiredTigerRecordStore::generateCreateString( ident.startsWith("internal-") || // TODO (SERVER-60753): Remove special handling for index build during recovery. This // includes the following _mdb_catalog ident. - nss == NamespaceString::kIndexBuildEntryNamespace || ident.startsWith("_mdb_catalog")) { + nss == NamespaceString::kIndexBuildEntryNamespace || + // SERVER-68330: Reconstructing config.transactions after a rollback does a mixed-mode + // write. + nss == NamespaceString::kSessionTransactionsTableNamespace || + ident.startsWith("_mdb_catalog")) { ss << "write_timestamp_usage=mixed_mode,"; } else { ss << "write_timestamp_usage=ordered,"; @@ -1046,7 +1051,8 @@ bool WiredTigerRecordStore::inShutdown() const { } long long WiredTigerRecordStore::dataSize(OperationContext* opCtx) const { - return _sizeInfo->dataSize.load(); + auto dataSize = _sizeInfo->dataSize.load(); + return dataSize > 0 ? dataSize : 0; } long long WiredTigerRecordStore::numRecords(OperationContext* opCtx) const { @@ -1157,8 +1163,7 @@ void WiredTigerRecordStore::doDeleteRecord(OperationContext* opCtx, const Record auto keyLength = computeRecordIdSize(id); metricsCollector.incrementOneDocWritten(old_length + keyLength); - _changeNumRecords(opCtx, -1); - _increaseDataSize(opCtx, -old_length); + _changeNumRecordsAndDataSize(opCtx, -1, -old_length); } Timestamp WiredTigerRecordStore::getPinnedOplog() const { @@ -1283,8 +1288,7 @@ void WiredTigerRecordStore::reclaimOplog(OperationContext* opCtx, Timestamp mayT invariantWTOK(cursor->reset(cursor), cursor->session); setKey(cursor, &truncateUpToKey); invariantWTOK(session->truncate(session, nullptr, nullptr, cursor, nullptr), session); - _changeNumRecords(opCtx, -stone->records); - _increaseDataSize(opCtx, -stone->bytes); + _changeNumRecordsAndDataSize(opCtx, -stone->records, -stone->bytes); wuow.commit(); @@ -1424,9 +1428,7 @@ Status WiredTigerRecordStore::_insertRecords(OperationContext* opCtx, metricsCollector.incrementOneDocWritten(value.size + keyLength); } } - - _changeNumRecords(opCtx, nRecords); - _increaseDataSize(opCtx, totalLength); + _changeNumRecordsAndDataSize(opCtx, nRecords, totalLength); if (_oplogStones) { _oplogStones->updateCurrentStoneAfterInsertOnCommit( @@ -1601,7 +1603,7 @@ Status WiredTigerRecordStore::doUpdateRecord(OperationContext* opCtx, } invariantWTOK(ret, c->session); - _increaseDataSize(opCtx, len - old_length); + _changeNumRecordsAndDataSize(opCtx, 0, len - old_length); return Status::OK(); } @@ -1658,9 +1660,8 @@ StatusWith<RecordData> WiredTigerRecordStore::doUpdateWithDamages( } void WiredTigerRecordStore::printRecordMetadata(OperationContext* opCtx, - const RecordId& recordId) const { - LOGV2(6120300, "Printing record metadata", "recordId"_attr = recordId); - + const RecordId& recordId, + std::set<Timestamp>* recordTimestamps) const { // Printing the record metadata requires a new session. We cannot open other cursors when there // are open history store cursors in the session. WiredTigerSession session(_kvEngine->getConnection()); @@ -1705,7 +1706,7 @@ void WiredTigerRecordStore::printRecordMetadata(OperationContext* opCtx, cursor->session); RecordData recordData(static_cast<const char*>(value.data), value.size); - LOGV2(6120301, + LOGV2(6120300, "WiredTiger record metadata", "recordId"_attr = recordId, "startTxnId"_attr = startTxnId, @@ -1720,6 +1721,20 @@ void WiredTigerRecordStore::printRecordMetadata(OperationContext* opCtx, "location"_attr = location, "value"_attr = redact(recordData.toBson())); + // Save all relevant timestamps that we just printed. + if (recordTimestamps) { + auto saveRecordTimestampIfValid = [recordTimestamps](Timestamp ts) { + if (ts.isNull() || ts == Timestamp::max() || ts == Timestamp::min()) { + return; + } + (void)recordTimestamps->emplace(ts); + }; + saveRecordTimestampIfValid(Timestamp(startTs)); + saveRecordTimestampIfValid(Timestamp(startDurableTs)); + saveRecordTimestampIfValid(Timestamp(stopTs)); + saveRecordTimestampIfValid(Timestamp(stopDurableTs)); + } + ret = cursor->next(cursor); } } @@ -1742,8 +1757,7 @@ Status WiredTigerRecordStore::doTruncate(OperationContext* opCtx) { WT_SESSION* session = WiredTigerRecoveryUnit::get(opCtx)->getSession()->getSession(); invariantWTOK(WT_OP_CHECK(session->truncate(session, nullptr, start, nullptr, nullptr)), session); - _changeNumRecords(opCtx, -numRecords(opCtx)); - _increaseDataSize(opCtx, -dataSize(opCtx)); + _changeNumRecordsAndDataSize(opCtx, -numRecords(opCtx), -dataSize(opCtx)); if (_oplogStones) { _oplogStones->clearStonesOnCommit(opCtx); @@ -1974,24 +1988,9 @@ RecordId WiredTigerRecordStore::_nextId(OperationContext* opCtx) { return out; } -void WiredTigerRecordStore::_changeNumRecords(OperationContext* opCtx, int64_t diff) { - if (!_tracksSizeAdjustments) { - return; - } - - if (!sizeRecoveryState(getGlobalServiceContext()).collectionNeedsSizeAdjustment(getIdent())) { - return; - } - - opCtx->recoveryUnit()->onRollback([this, diff]() { - LOGV2_DEBUG( - 22404, 3, "WiredTigerRecordStore: rolling back NumRecordsChange", "diff"_attr = -diff); - _sizeInfo->numRecords.addAndFetch(-diff); - }); - _sizeInfo->numRecords.addAndFetch(diff); -} - -void WiredTigerRecordStore::_increaseDataSize(OperationContext* opCtx, int64_t amount) { +void WiredTigerRecordStore::_changeNumRecordsAndDataSize(OperationContext* opCtx, + int64_t numRecordDiff, + int64_t dataSizeDiff) { if (!_tracksSizeAdjustments) { return; } @@ -2000,15 +1999,23 @@ void WiredTigerRecordStore::_increaseDataSize(OperationContext* opCtx, int64_t a return; } - if (opCtx) - opCtx->recoveryUnit()->onRollback( - [this, amount]() { _increaseDataSize(nullptr, -amount); }); + const auto updateAndStoreSizeInfo = [this](int64_t numRecordDiff, int64_t dataSizeDiff) { + _sizeInfo->numRecords.addAndFetch(numRecordDiff); + _sizeInfo->dataSize.addAndFetch(dataSizeDiff); - if (_sizeInfo->dataSize.fetchAndAdd(amount) < 0) - _sizeInfo->dataSize.store(std::max(amount, int64_t(0))); + if (_sizeStorer) + _sizeStorer->store(_uri, _sizeInfo); + }; - if (_sizeStorer) - _sizeStorer->store(_uri, _sizeInfo); + opCtx->recoveryUnit()->onRollback([updateAndStoreSizeInfo, numRecordDiff, dataSizeDiff]() { + LOGV2_DEBUG(7105300, + 3, + "WiredTigerRecordStore: rolling back change to numRecords and dataSize", + "numRecordDiff"_attr = -numRecordDiff, + "dataSizeDiff"_attr = -dataSizeDiff); + updateAndStoreSizeInfo(-numRecordDiff, -dataSizeDiff); + }); + updateAndStoreSizeInfo(numRecordDiff, dataSizeDiff); } void WiredTigerRecordStore::setNumRecords(long long numRecords) { @@ -2092,8 +2099,7 @@ void WiredTigerRecordStore::doCappedTruncateAfter(OperationContext* opCtx, WT_SESSION* session = WiredTigerRecoveryUnit::get(opCtx)->getSession()->getSession(); invariantWTOK(session->truncate(session, nullptr, start, nullptr, nullptr), session); - _changeNumRecords(opCtx, -recordsRemoved); - _increaseDataSize(opCtx, -bytesRemoved); + _changeNumRecordsAndDataSize(opCtx, -recordsRemoved, -bytesRemoved); wuow.commit(); @@ -2219,20 +2225,23 @@ boost::optional<Record> WiredTigerRecordStoreCursorBase::next() { return {}; } - if (_forward && _lastReturnedId >= id) { - LOGV2_ERROR(22406, - "WTCursor::next -- c->next_key ( {next}) was not greater than _lastReturnedId " - "({last}) which is a bug.", - "WTCursor::next -- next was not greater than last which is a bug", - "next"_attr = id, - "last"_attr = _lastReturnedId); - - // Crash when testing diagnostics are enabled. - invariant(!TestingProctor::instance().isEnabled(), "next was not greater than last"); + const bool failWithOutOfOrderForTest = WTRecordStoreUassertOutOfOrder.shouldFail(); + if ((_forward && _lastReturnedId >= id) || MONGO_unlikely(failWithOutOfOrderForTest)) { + if (!failWithOutOfOrderForTest) { + // Crash when testing diagnostics are enabled and not explicitly uasserting on + // out-of-order keys. + invariant(!TestingProctor::instance().isEnabled(), "cursor returned out-of-order keys"); + } - // Force a retry of the operation from our last known position by acting as-if - // we received a WT_ROLLBACK error. - throw WriteConflictException(); + // uassert with 'DataCorruptionDetected' after logging. + LOGV2_ERROR_OPTIONS(22406, + {logv2::UserAssertAfterLog(ErrorCodes::DataCorruptionDetected)}, + "WT_Cursor::next -- returned out-of-order keys", + "forward"_attr = _forward, + "next"_attr = id, + "last"_attr = _lastReturnedId, + "ident"_attr = _rs._ident, + "ns"_attr = _rs.ns()); } WT_ITEM value; diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_record_store.h b/src/mongo/db/storage/wiredtiger/wiredtiger_record_store.h index 7630cc0900b..119b907f7a7 100644 --- a/src/mongo/db/storage/wiredtiger/wiredtiger_record_store.h +++ b/src/mongo/db/storage/wiredtiger/wiredtiger_record_store.h @@ -169,7 +169,9 @@ public: const char* damageSource, const mutablebson::DamageVector& damages) final; - virtual void printRecordMetadata(OperationContext* opCtx, const RecordId& recordId) const; + virtual void printRecordMetadata(OperationContext* opCtx, + const RecordId& recordId, + std::set<Timestamp>* recordTimestamps) const; virtual std::unique_ptr<SeekableRecordCursor> getCursor(OperationContext* opCtx, bool forward) const = 0; @@ -306,9 +308,9 @@ private: void _initNextIdIfNeeded(OperationContext* opCtx); /** - * Adjusts the record count and data size metadata for this record store, respectively. These - * functions consult the SizeRecoveryState to determine whether or not to actually change the - * size metadata if the server is undergoing recovery. + * Adjusts the record count and data size metadata for this record store. The function consults + * the SizeRecoveryState to determine whether or not to actually change the size metadata if the + * server is undergoing recovery. * * For most record stores, we will not update the size metadata during recovery, as we trust * that the values in the SizeStorer are accurate with respect to the end state of recovery. @@ -322,8 +324,9 @@ private: * are pending writes to this ident as part of the recovery process, and so we must * always adjust size metadata for these idents. */ - void _changeNumRecords(OperationContext* opCtx, int64_t diff); - void _increaseDataSize(OperationContext* opCtx, int64_t amount); + void _changeNumRecordsAndDataSize(OperationContext* opCtx, + int64_t numRecordDiff, + int64_t dataSizeDiff); const std::string _uri; const uint64_t _tableId; // not persisted diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_record_store_test.cpp b/src/mongo/db/storage/wiredtiger/wiredtiger_record_store_test.cpp index a2d9c5f7e99..f4909583ce3 100644 --- a/src/mongo/db/storage/wiredtiger/wiredtiger_record_store_test.cpp +++ b/src/mongo/db/storage/wiredtiger/wiredtiger_record_store_test.cpp @@ -1150,9 +1150,9 @@ TEST(WiredTigerRecordStoreTest, ClusteredRecordStore) { ASSERT_EQ(0, memcmp(dataUpdated, rd.data(), strlen(dataUpdated))); } -// Make sure numRecords is accurate after a delete rolls back and some other transaction deletes the -// same rows before we have a chance of patching up the metadata. -TEST(WiredTigerRecordStoreTest, NumRecordsAccurateAfterRollbackWithDelete) { +// Make sure numRecords and dataSize are accurate after a delete rolls back and some other +// transaction deletes the same rows before we have a chance of patching up the metadata. +TEST(WiredTigerRecordStoreTest, SizeInfoAccurateAfterRollbackWithDelete) { const auto harnessHelper(newRecordStoreHarnessHelper()); unique_ptr<RecordStore> rs(harnessHelper->newRecordStore()); @@ -1166,6 +1166,7 @@ TEST(WiredTigerRecordStoreTest, NumRecordsAccurateAfterRollbackWithDelete) { } ASSERT_EQ(1, rs->numRecords(ctx.get())); + ASSERT_EQ(2, rs->dataSize(ctx.get())); WriteUnitOfWork uow(ctx.get()); @@ -1197,6 +1198,7 @@ TEST(WiredTigerRecordStoreTest, NumRecordsAccurateAfterRollbackWithDelete) { abortedThread.join(); ASSERT_EQ(0, rs->numRecords(ctx.get())); + ASSERT_EQ(0, rs->dataSize(ctx.get())); } } // namespace diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_recovery_unit.cpp b/src/mongo/db/storage/wiredtiger/wiredtiger_recovery_unit.cpp index ca2e5d8c1b4..3da6bb97782 100644 --- a/src/mongo/db/storage/wiredtiger/wiredtiger_recovery_unit.cpp +++ b/src/mongo/db/storage/wiredtiger/wiredtiger_recovery_unit.cpp @@ -39,6 +39,7 @@ #include "mongo/db/storage/wiredtiger/wiredtiger_kv_engine.h" #include "mongo/db/storage/wiredtiger/wiredtiger_prepare_conflict.h" #include "mongo/db/storage/wiredtiger/wiredtiger_session_cache.h" +#include "mongo/db/storage/wiredtiger/wiredtiger_stats.h" #include "mongo/db/storage/wiredtiger/wiredtiger_util.h" #include "mongo/logv2/log.h" #include "mongo/util/hex.h" @@ -47,6 +48,7 @@ #include <fmt/compile.h> #include <fmt/format.h> +#include <memory> namespace mongo { namespace { @@ -79,104 +81,6 @@ void handleWriteContextForDebugging(WiredTigerRecoveryUnit& ru, Timestamp& ts) { AtomicWord<std::int64_t> snapshotTooOldErrorCount{0}; -using Section = WiredTigerOperationStats::Section; - -std::map<int, std::pair<StringData, Section>> WiredTigerOperationStats::_statNameMap = { - {WT_STAT_SESSION_BYTES_READ, std::make_pair("bytesRead"_sd, Section::DATA)}, - {WT_STAT_SESSION_BYTES_WRITE, std::make_pair("bytesWritten"_sd, Section::DATA)}, - {WT_STAT_SESSION_LOCK_DHANDLE_WAIT, std::make_pair("handleLock"_sd, Section::WAIT)}, - {WT_STAT_SESSION_READ_TIME, std::make_pair("timeReadingMicros"_sd, Section::DATA)}, - {WT_STAT_SESSION_WRITE_TIME, std::make_pair("timeWritingMicros"_sd, Section::DATA)}, - {WT_STAT_SESSION_LOCK_SCHEMA_WAIT, std::make_pair("schemaLock"_sd, Section::WAIT)}, - {WT_STAT_SESSION_CACHE_TIME, std::make_pair("cache"_sd, Section::WAIT)}}; - -std::shared_ptr<StorageStats> WiredTigerOperationStats::getCopy() { - std::shared_ptr<WiredTigerOperationStats> copy = std::make_shared<WiredTigerOperationStats>(); - *copy += *this; - return copy; -} - -void WiredTigerOperationStats::fetchStats(WT_SESSION* session, - const std::string& uri, - const std::string& config) { - invariant(session); - - WT_CURSOR* c = nullptr; - const char* cursorConfig = config.empty() ? nullptr : config.c_str(); - int ret = session->open_cursor(session, uri.c_str(), nullptr, cursorConfig, &c); - uassert(ErrorCodes::CursorNotFound, "Unable to open statistics cursor", ret == 0); - - invariant(c); - ON_BLOCK_EXIT([&] { c->close(c); }); - - const char* desc; - uint64_t value; - int32_t key; - while (c->next(c) == 0 && c->get_key(c, &key) == 0) { - fassert(51035, c->get_value(c, &desc, nullptr, &value) == 0); - _stats[key] = WiredTigerUtil::castStatisticsValue<long long>(value); - } - - // Reset the statistics so that the next fetch gives the recent values. - invariantWTOK(c->reset(c), c->session); -} - -BSONObj WiredTigerOperationStats::toBSON() { - BSONObjBuilder bob; - std::unique_ptr<BSONObjBuilder> dataSection; - std::unique_ptr<BSONObjBuilder> waitSection; - - for (auto const& stat : _stats) { - // Find the user consumable name for this statistic. - auto statIt = _statNameMap.find(stat.first); - invariant(statIt != _statNameMap.end()); - - auto statName = statIt->second.first; - Section subs = statIt->second.second; - long long val = stat.second; - // Add this statistic only if higher than zero. - if (val > 0) { - // Gather the statistic into its own subsection in the BSONObj. - switch (subs) { - case Section::DATA: - if (!dataSection) - dataSection = std::make_unique<BSONObjBuilder>(); - - dataSection->append(statName, val); - break; - case Section::WAIT: - if (!waitSection) - waitSection = std::make_unique<BSONObjBuilder>(); - - waitSection->append(statName, val); - break; - default: - MONGO_UNREACHABLE; - } - } - } - - if (dataSection) - bob.append("data", dataSection->obj()); - if (waitSection) - bob.append("timeWaitingMicros", waitSection->obj()); - - return bob.obj(); -} - -WiredTigerOperationStats& WiredTigerOperationStats::operator+=( - const WiredTigerOperationStats& other) { - for (auto const& otherStat : other._stats) { - _stats[otherStat.first] += otherStat.second; - } - return (*this); -} - -StorageStats& WiredTigerOperationStats::operator+=(const StorageStats& other) { - *this += checked_cast<const WiredTigerOperationStats&>(other); - return (*this); -} - WiredTigerRecoveryUnit::WiredTigerRecoveryUnit(WiredTigerSessionCache* sc) : WiredTigerRecoveryUnit(sc, sc->getKVEngine()->getOplogManager()) {} @@ -450,36 +354,22 @@ void WiredTigerRecoveryUnit::_txnClose(bool commit) { int wtRet; if (commit) { - // Avoid heap allocation in favour of a stack allocation for the commit string. - static constexpr auto commitTimestampFmtString = "commit_timestamp={:X},"; - static constexpr auto durableTimestampFmtString = "durable_timestamp={:X}"; - static constexpr auto bytesRequired = - std::char_traits<char>::length(commitTimestampFmtString) + - (sizeof(decltype(_commitTimestamp.asULL())) * 2) + - std::char_traits<char>::length(durableTimestampFmtString) + - (sizeof(decltype(_durableTimestamp.asULL())) * 2) + 1; - std::array<char, bytesRequired> conf; - auto end = conf.begin(); if (!_commitTimestamp.isNull()) { // There is currently no scenario where it is intentional to commit before the current // read timestamp. invariant(_readAtTimestamp.isNull() || _commitTimestamp >= _readAtTimestamp); if (MONGO_likely(!doUntimestampedWritesForIdempotencyTests.shouldFail())) { - end = fmt::format_to( - end, FMT_STRING(commitTimestampFmtString), _commitTimestamp.asULL()); + s->timestamp_transaction_uint(s, WT_TS_TXN_TYPE_COMMIT, _commitTimestamp.asULL()); } _isTimestamped = true; } if (!_durableTimestamp.isNull()) { - end = fmt::format_to( - end, FMT_STRING(durableTimestampFmtString), _durableTimestamp.asULL()); + s->timestamp_transaction_uint(s, WT_TS_TXN_TYPE_DURABLE, _durableTimestamp.asULL()); } - *end = '\0'; - - wtRet = s->commit_transaction(s, conf.data()); + wtRet = s->commit_transaction(s, nullptr); LOGV2_DEBUG( 22412, 3, "WT commit_transaction", "snapshotId"_attr = getSnapshotId().toNumber()); @@ -557,11 +447,16 @@ boost::optional<Timestamp> WiredTigerRecoveryUnit::getPointInTimeReadTimestamp( // The read timestamp is set by the user and does not require a transaction to be open. invariant(!_readAtTimestamp.isNull()); return _readAtTimestamp; - + case ReadSource::kLastApplied: + // The lastApplied timestamp is not always available if the system has not accepted + // writes, so it is not possible to invariant that it exists. + if (_readAtTimestamp.isNull()) { + return boost::none; + } + return _readAtTimestamp; // The following ReadSources can only establish a read timestamp when a transaction is // opened. case ReadSource::kNoOverlap: - case ReadSource::kLastApplied: case ReadSource::kAllDurableSnapshot: case ReadSource::kMajorityCommitted: break; @@ -622,7 +517,7 @@ void WiredTigerRecoveryUnit::_txnOpen() { break; } case ReadSource::kLastApplied: { - _readAtTimestamp = _beginTransactionAtLastAppliedTimestamp(session); + _beginTransactionAtLastAppliedTimestamp(session); break; } case ReadSource::kNoOverlap: { @@ -679,9 +574,8 @@ Timestamp WiredTigerRecoveryUnit::_beginTransactionAtAllDurableTimestamp(WT_SESS return readTimestamp; } -Timestamp WiredTigerRecoveryUnit::_beginTransactionAtLastAppliedTimestamp(WT_SESSION* session) { - auto lastApplied = _sessionCache->snapshotManager().getLastApplied(); - if (!lastApplied) { +void WiredTigerRecoveryUnit::_beginTransactionAtLastAppliedTimestamp(WT_SESSION* session) { + if (_readAtTimestamp.isNull()) { // When there is not a lastApplied timestamp available, read without a timestamp. Do not // round up the read timestamp to the oldest timestamp. @@ -695,20 +589,20 @@ Timestamp WiredTigerRecoveryUnit::_beginTransactionAtLastAppliedTimestamp(WT_SES session, _prepareConflictBehavior, _roundUpPreparedTimestamps); LOGV2_DEBUG(4847500, 2, "no read timestamp available for kLastApplied"); txnOpen.done(); - return Timestamp(); + return; } WiredTigerBeginTxnBlock txnOpen(session, _prepareConflictBehavior, _roundUpPreparedTimestamps, RoundUpReadTimestamp::kRound); - auto status = txnOpen.setReadSnapshot(*lastApplied); + auto status = txnOpen.setReadSnapshot(_readAtTimestamp); fassert(4847501, status); txnOpen.done(); // We might have rounded to oldest between calling getLastApplied and setReadSnapshot. We // need to get the actual read timestamp we used. - return _getTransactionReadTimestamp(session); + _readAtTimestamp = _getTransactionReadTimestamp(session); } Timestamp WiredTigerRecoveryUnit::_beginTransactionAtNoOverlapTimestamp(WT_SESSION* session) { @@ -950,7 +844,16 @@ void WiredTigerRecoveryUnit::setTimestampReadSource(ReadSource readSource, invariant(!(provided && provided->isNull())); _timestampReadSource = readSource; - _readAtTimestamp = (provided) ? *provided : Timestamp(); + if (readSource == kLastApplied) { + // The lastApplied timestamp is not always available if the system has not accepted writes. + if (auto lastApplied = _sessionCache->snapshotManager().getLastApplied()) { + _readAtTimestamp = *lastApplied; + } else { + _readAtTimestamp = Timestamp(); + } + } else { + _readAtTimestamp = (provided) ? *provided : Timestamp(); + } } RecoveryUnit::ReadSource WiredTigerRecoveryUnit::getTimestampReadSource() const { @@ -978,19 +881,21 @@ void WiredTigerRecoveryUnit::beginIdle() { } } -std::shared_ptr<StorageStats> WiredTigerRecoveryUnit::getOperationStatistics() const { - std::shared_ptr<WiredTigerOperationStats> statsPtr(nullptr); - +std::unique_ptr<StorageStats> WiredTigerRecoveryUnit::computeOperationStatisticsSinceLastCall() { if (!_session) - return statsPtr; + return nullptr; - WT_SESSION* s = _session->getSession(); - invariant(s); + // We compute operation statistics as the difference between the current session statistics and + // the session statistics of the last time the method was called, which should correspond to the + // end of one operation. + WiredTigerStats currentSessionStats{_session->getSession()}; + + auto operationStats = + std::make_unique<WiredTigerStats>(currentSessionStats - _sessionStatsAfterLastOperation); - statsPtr = std::make_shared<WiredTigerOperationStats>(); - statsPtr->fetchStats(s, "statistics:session", "statistics=(fast)"); + _sessionStatsAfterLastOperation = std::move(currentSessionStats); - return statsPtr; + return operationStats; } void WiredTigerRecoveryUnit::setCatalogConflictingTimestamp(Timestamp timestamp) { diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_recovery_unit.h b/src/mongo/db/storage/wiredtiger/wiredtiger_recovery_unit.h index 2d75b4e54f1..e0b26282779 100644 --- a/src/mongo/db/storage/wiredtiger/wiredtiger_recovery_unit.h +++ b/src/mongo/db/storage/wiredtiger/wiredtiger_recovery_unit.h @@ -45,8 +45,8 @@ #include "mongo/db/storage/recovery_unit.h" #include "mongo/db/storage/wiredtiger/wiredtiger_begin_transaction_block.h" #include "mongo/db/storage/wiredtiger/wiredtiger_session_cache.h" +#include "mongo/db/storage/wiredtiger/wiredtiger_stats.h" #include "mongo/util/timer.h" - namespace mongo { using RoundUpPreparedTimestamps = WiredTigerBeginTxnBlock::RoundUpPreparedTimestamps; @@ -56,41 +56,6 @@ extern AtomicWord<std::int64_t> snapshotTooOldErrorCount; class BSONObjBuilder; -class WiredTigerOperationStats final : public StorageStats { -public: - /** - * There are two types of statistics provided by WiredTiger engine - data and wait. - */ - enum class Section { DATA, WAIT }; - - BSONObj toBSON() final; - - StorageStats& operator+=(const StorageStats&) final; - - WiredTigerOperationStats& operator+=(const WiredTigerOperationStats&); - - /** - * Fetches an operation's storage statistics from WiredTiger engine. - */ - void fetchStats(WT_SESSION*, const std::string&, const std::string&); - - std::shared_ptr<StorageStats> getCopy() final; - -private: - /** - * Each statistic in WiredTiger has an integer key, which this map associates with a section - * (either DATA or WAIT) and user-readable name. - */ - static std::map<int, std::pair<StringData, Section>> _statNameMap; - - /** - * Stores the value for each statistic returned by a WiredTiger cursor. Each statistic is - * associated with an integer key, which can be mapped to a name and section using the - * '_statNameMap'. - */ - std::map<int, long long> _stats; -}; - class WiredTigerRecoveryUnit final : public RecoveryUnit { public: WiredTigerRecoveryUnit(WiredTigerSessionCache* sc); @@ -174,7 +139,7 @@ public: return _readOnce; }; - std::shared_ptr<StorageStats> getOperationStatistics() const override; + std::unique_ptr<StorageStats> computeOperationStatisticsSinceLastCall() override; void refreshSnapshot() override; @@ -257,10 +222,11 @@ private: Timestamp _beginTransactionAtNoOverlapTimestamp(WT_SESSION* session); /** - * Starts a transaction at the lastApplied timestamp. Returns the timestamp at which the - * transaction was started. + * Starts a transaction at the lastApplied timestamp stored in '_readAtTimestamp'. Sets + * '_readAtTimestamp' to the actual timestamp used by the storage engine in case rounding + * occured. */ - Timestamp _beginTransactionAtLastAppliedTimestamp(WT_SESSION* session); + void _beginTransactionAtLastAppliedTimestamp(WT_SESSION* session); /** * Returns the timestamp at which the current transaction is reading. @@ -315,6 +281,8 @@ private: boost::optional<int64_t> _oplogVisibleTs = boost::none; bool _gatherWriteContextForDebugging = false; std::vector<BSONObj> _writeContextForDebugging; + + WiredTigerStats _sessionStatsAfterLastOperation; }; } // namespace mongo diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_session_cache.cpp b/src/mongo/db/storage/wiredtiger/wiredtiger_session_cache.cpp index 3ed1d4e985b..5db8982d9d0 100644 --- a/src/mongo/db/storage/wiredtiger/wiredtiger_session_cache.cpp +++ b/src/mongo/db/storage/wiredtiger/wiredtiger_session_cache.cpp @@ -36,7 +36,6 @@ #include <memory> #include "mongo/base/error_codes.h" -#include "mongo/db/concurrency/write_conflict_exception.h" #include "mongo/db/global_settings.h" #include "mongo/db/repl/repl_settings.h" #include "mongo/db/storage/journal_listener.h" diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_standard_index_test.cpp b/src/mongo/db/storage/wiredtiger/wiredtiger_standard_index_test.cpp index f4cdc403149..0101172b81b 100644 --- a/src/mongo/db/storage/wiredtiger/wiredtiger_standard_index_test.cpp +++ b/src/mongo/db/storage/wiredtiger/wiredtiger_standard_index_test.cpp @@ -66,15 +66,11 @@ public: _fastClockSource = std::make_unique<SystemClockSource>(); _sessionCache = new WiredTigerSessionCache(_conn, _fastClockSource.get()); - - WiredTigerUtil::notifyStartupComplete(); } ~WiredTigerIndexHarnessHelper() final { delete _sessionCache; _conn->close(_conn, nullptr); - - WiredTigerUtil::resetTableLoggingInfo(); } std::unique_ptr<SortedDataInterface> newIdIndexSortedDataInterface() final { diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_standard_record_store_test.cpp b/src/mongo/db/storage/wiredtiger/wiredtiger_standard_record_store_test.cpp index 1b74a9ead4e..9d2568450cf 100644 --- a/src/mongo/db/storage/wiredtiger/wiredtiger_standard_record_store_test.cpp +++ b/src/mongo/db/storage/wiredtiger/wiredtiger_standard_record_store_test.cpp @@ -38,7 +38,6 @@ #include "mongo/base/init.h" #include "mongo/base/string_data.h" #include "mongo/bson/bsonobjbuilder.h" -#include "mongo/db/concurrency/write_conflict_exception.h" #include "mongo/db/json.h" #include "mongo/db/service_context.h" #include "mongo/db/storage/wiredtiger/wiredtiger_kv_engine.h" @@ -202,5 +201,45 @@ TEST_F(SizeStorerUpdateTest, Basic) { ASSERT_EQUALS(getDataSize(opCtx.get()), val); } +// Verify that the size storer contains accurate data after a transaction rollback just before a +// flush (simulating a shutdown). That is, that the rollback marks the size info as dirty, and is +// properly flushed to disk. +TEST_F(SizeStorerUpdateTest, ReloadAfterRollbackAndFlush) { + ServiceContext::UniqueOperationContext opCtx(harnessHelper->newOperationContext()); + // Do an op for which the sizeInfo is persisted, for safety so we don't check against 0. + { + WriteUnitOfWork uow(opCtx.get()); + auto rId = rs->insertRecord(opCtx.get(), "12345", 5, Timestamp{1}); + ASSERT_TRUE(rId.isOK()); + + uow.commit(); + } + + // An operation to rollback, with a flush between the original modification and the rollback. + { + WriteUnitOfWork uow(opCtx.get()); + auto rId = rs->insertRecord(opCtx.get(), "12345", 5, Timestamp{2}); + ASSERT_TRUE(rId.isOK()); + + ASSERT_EQ(getNumRecords(opCtx.get()), 2); + ASSERT_EQ(getDataSize(opCtx.get()), 10); + // Mark size info as clean, before rollback is done. + sizeStorer->flush(false); + } + + // Simulate a shutdown and restart, which loads the size storer from disk. + sizeStorer->flush(true); + sizeStorer.reset(new WiredTigerSizeStorer(harnessHelper->conn(), + WiredTigerKVEngine::kTableUriPrefix + "sizeStorer")); + WiredTigerRecordStore* wtrs = checked_cast<WiredTigerRecordStore*>(rs.get()); + wtrs->setSizeStorer(sizeStorer.get()); + + // As the operation was rolled back, numRecords and dataSize should be for the first op only. If + // rollback does not properly mark the sizeInfo as dirty, on load sizeInfo will account for the + // two operations, as the rollback sizeInfo update has not been flushed. + ASSERT_EQ(getNumRecords(opCtx.get()), 1); + ASSERT_EQ(getDataSize(opCtx.get()), 5); +}; + } // namespace } // namespace mongo diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_stats.cpp b/src/mongo/db/storage/wiredtiger/wiredtiger_stats.cpp new file mode 100644 index 00000000000..da3d2f461e5 --- /dev/null +++ b/src/mongo/db/storage/wiredtiger/wiredtiger_stats.cpp @@ -0,0 +1,150 @@ +/** + * Copyright (C) 2022-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 + * <http://www.mongodb.com/licensing/server-side-public-license>. + * + * 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. + */ + +#include "mongo/db/storage/wiredtiger/wiredtiger_stats.h" + +#include "mongo/bson/bsonobjbuilder.h" +#include "mongo/db/storage/wiredtiger/wiredtiger_util.h" + +namespace mongo { +namespace { + +enum class StatType { kData, kWait }; + +struct StatInfo { + StringData name; + StatType type; +}; + +const stdx::unordered_map<int, StatInfo> kWiredTigerStatCodeToStatInfo = { + {WT_STAT_SESSION_BYTES_READ, {"bytesRead"_sd, StatType::kData}}, + {WT_STAT_SESSION_BYTES_WRITE, {"bytesWritten"_sd, StatType::kData}}, + {WT_STAT_SESSION_LOCK_DHANDLE_WAIT, {"handleLock"_sd, StatType::kWait}}, + {WT_STAT_SESSION_READ_TIME, {"timeReadingMicros"_sd, StatType::kData}}, + {WT_STAT_SESSION_WRITE_TIME, {"timeWritingMicros"_sd, StatType::kData}}, + {WT_STAT_SESSION_LOCK_SCHEMA_WAIT, {"schemaLock"_sd, StatType::kWait}}, + {WT_STAT_SESSION_CACHE_TIME, {"cache"_sd, StatType::kWait}}}; + +} // namespace + +WiredTigerStats::WiredTigerStats(WT_SESSION* session) { + invariant(session); + + WT_CURSOR* c; + uassert(ErrorCodes::CursorNotFound, + "Unable to open statistics cursor", + !session->open_cursor(session, "statistics:session", nullptr, "statistics=(fast)", &c)); + + ScopeGuard guard{[c] { c->close(c); }}; + + int32_t key; + uint64_t value; + while (c->next(c) == 0 && c->get_key(c, &key) == 0) { + fassert(51035, c->get_value(c, nullptr, nullptr, &value) == 0); + _stats[key] = WiredTigerUtil::castStatisticsValue<long long>(value); + } +} + +BSONObj WiredTigerStats::toBSON() const { + boost::optional<BSONObjBuilder> dataSection; + boost::optional<BSONObjBuilder> waitSection; + + for (auto&& [stat, value] : _stats) { + if (value == 0) { + continue; + } + + auto it = kWiredTigerStatCodeToStatInfo.find(stat); + if (it == kWiredTigerStatCodeToStatInfo.end()) { + continue; + } + auto&& [name, type] = it->second; + + auto appendToSection = [name = name, + value = value](boost::optional<BSONObjBuilder>& section) { + if (!section) { + section.emplace(); + } + section->append(name, value); + }; + + switch (type) { + case StatType::kData: + appendToSection(dataSection); + break; + case StatType::kWait: + appendToSection(waitSection); + break; + } + } + + BSONObjBuilder builder; + if (dataSection) { + builder.append("data", dataSection->obj()); + } + if (waitSection) { + builder.append("timeWaitingMicros", waitSection->obj()); + } + + return builder.obj(); +} + +std::unique_ptr<StorageStats> WiredTigerStats::clone() const { + return std::make_unique<WiredTigerStats>(*this); +} + +WiredTigerStats& WiredTigerStats::operator=(WiredTigerStats&& other) { + _stats = std::move(other._stats); + return *this; +} + +WiredTigerStats& WiredTigerStats::operator+=(const WiredTigerStats& other) { + for (auto&& [stat, value] : other._stats) { + _stats[stat] += value; + } + return *this; +} + +StorageStats& WiredTigerStats::operator+=(const StorageStats& other) { + return *this += checked_cast<const WiredTigerStats&>(other); +} + +WiredTigerStats& WiredTigerStats::operator-=(const WiredTigerStats& other) { + for (auto const& otherStat : other._stats) { + _stats[otherStat.first] -= otherStat.second; + } + return (*this); +} + +StorageStats& WiredTigerStats::operator-=(const StorageStats& other) { + *this -= checked_cast<const WiredTigerStats&>(other); + return (*this); +} + +} // namespace mongo diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_stats.h b/src/mongo/db/storage/wiredtiger/wiredtiger_stats.h new file mode 100644 index 00000000000..d35a582cd34 --- /dev/null +++ b/src/mongo/db/storage/wiredtiger/wiredtiger_stats.h @@ -0,0 +1,72 @@ +/** + * Copyright (C) 2022-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 + * <http://www.mongodb.com/licensing/server-side-public-license>. + * + * 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. + */ + +#pragma once + +#include <wiredtiger.h> + +#include "mongo/db/storage/storage_stats.h" + +namespace mongo { + +class WiredTigerStats final : public StorageStats { +public: + /** + * Construct a new WiredTigerStats object with the statistics of the specified session. + */ + WiredTigerStats(WT_SESSION*); + + WiredTigerStats() = default; + WiredTigerStats(const WiredTigerStats&) = default; + WiredTigerStats(WiredTigerStats&&) = default; + + BSONObj toBSON() const final; + + std::unique_ptr<StorageStats> clone() const final; + + WiredTigerStats& operator=(WiredTigerStats&&); + + StorageStats& operator+=(const StorageStats&) final; + + WiredTigerStats& operator+=(const WiredTigerStats&); + + StorageStats& operator-=(const StorageStats&) final; + + WiredTigerStats& operator-=(const WiredTigerStats&); + +protected: + std::map<int, long long> _stats; +}; + +inline WiredTigerStats operator-(WiredTigerStats lhs, const WiredTigerStats& rhs) { + lhs -= rhs; + return lhs; +} + +} // namespace mongo diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_stats_test.cpp b/src/mongo/db/storage/wiredtiger/wiredtiger_stats_test.cpp new file mode 100644 index 00000000000..942d6ce8592 --- /dev/null +++ b/src/mongo/db/storage/wiredtiger/wiredtiger_stats_test.cpp @@ -0,0 +1,327 @@ +/** + * Copyright (C) 2022-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 + * <http://www.mongodb.com/licensing/server-side-public-license>. + * + * 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. + */ + +#include "mongo/db/storage/wiredtiger/wiredtiger_stats.h" +#include "mongo/unittest/temp_dir.h" +#include "mongo/unittest/unittest.h" +#include <memory> + +namespace mongo { +namespace { + +#define ASSERT_WT_OK(result) ASSERT_EQ(result, 0) << wiredtiger_strerror(result) + +class WiredTigerStatsTest : public unittest::Test { +protected: + void setUp() override { + openConnectionAndCreateSession(); + // Prepare data to be read by tests. Reading data written within the same transaction does + // not count towards bytes read into cache. + + // Tests in the fixture do up to _maxReads reads. + for (int64_t i = 0; i < _kMaxReads; ++i) { + // Make the write big enough to span different pages. + writeAtKey(std::string(20000, 'a'), i); + } + + // Closing the connection will ensure that tests actually have to read into cache. + closeConnection(); + openConnectionAndCreateSession(); + } + + void tearDown() override { + closeConnection(); + } + + void openConnectionAndCreateSession() { + ASSERT_WT_OK( + wiredtiger_open(_path.path().c_str(), nullptr, "create,statistics=(fast),", &_conn)); + ASSERT_WT_OK(_conn->open_session(_conn, nullptr, "isolation=snapshot", &_session)); + ASSERT_WT_OK(_session->create( + _session, _uri.c_str(), "type=file,key_format=q,value_format=u,log=(enabled=false)")); + } + + void closeConnection() { + ASSERT_EQ(_conn->close(_conn, nullptr), 0); + } + + /** + * Writes some data using WT. Causes the bytesWritten stat to be incremented, and may also + * increment timeWritingMicros. + */ + void write() { + writeAtKey(std::string(_writeKey + 1, 'a'), _writeKey); + ++_writeKey; + } + + /** + * Writes the specified data using WT. Causes the bytesWritten stat to be incremented, and may + * also increment timeWritingMicros. + */ + void write(const std::string& data) { + writeAtKey(data, _writeKey); + ++_writeKey; + } + + /** + * Writes at the specified key to WT. + */ + void writeAtKey(const std::string& data, int64_t key) { + ASSERT_WT_OK(_session->begin_transaction(_session, nullptr)); + + WT_CURSOR* cursor; + ASSERT_WT_OK(_session->open_cursor(_session, _uri.c_str(), nullptr, nullptr, &cursor)); + + cursor->set_key(cursor, key); + + WT_ITEM item{data.data(), data.size()}; + cursor->set_value(cursor, &item); + + ASSERT_WT_OK(cursor->insert(cursor)); + ASSERT_WT_OK(cursor->close(cursor)); + ASSERT_WT_OK(_session->commit_transaction(_session, nullptr)); + + // Without a checkpoint, an operation is not guaranteed to write to disk. + ASSERT_WT_OK(_session->checkpoint(_session, nullptr)); + } + + /** + * Reads at the specified key from WT. + */ + void readAtKey(int64_t key) { + ASSERT_WT_OK(_session->begin_transaction(_session, nullptr)); + + WT_CURSOR* cursor; + ASSERT_WT_OK(_session->open_cursor(_session, _uri.c_str(), nullptr, nullptr, &cursor)); + + cursor->set_key(cursor, key); + ASSERT_WT_OK(cursor->search(cursor)); + + WT_ITEM value; + ASSERT_WT_OK(cursor->get_value(cursor, &value)); + + ASSERT_WT_OK(cursor->close(cursor)); + ASSERT_WT_OK(_session->commit_transaction(_session, nullptr)); + } + + /** + * Reads fixture data from WT. Causes the bytesRead stat to be incremented. May also cause + * timeReadingMicros to be incremented, but not always. This function can only be called up to + * _kMaxReads times within a test. + */ + void read() { + ASSERT_LT(_readKey, _kMaxReads); + readAtKey(_readKey++); + } + + /** + * Reads data written by the test from WT. Causes the bytesRead stat to be incremented. May + * also cause timeReadingMicros to be incremented, but not always. + */ + void readTestWrites() { + for (int64_t i = _kMaxReads; i < _writeKey; i++) { + readAtKey(i); + } + } + + unittest::TempDir _path{"wiredtiger_operation_stats_test"}; + std::string _uri{"table:wiredtiger_operation_stats_test"}; + WT_CONNECTION* _conn; + WT_SESSION* _session; + /* Number of reads the fixture will prepare in setUp(), consequently max amount of times read() + * can be called in a test. */ + static constexpr int64_t _kMaxReads = 2; + /* Next key to be used by read(), must be initialized at 0. */ + int64_t _readKey = 0; + /* Next key to be used by write(), must be initialized >= _kMaxReads. */ + int64_t _writeKey = _kMaxReads; +}; + +TEST_F(WiredTigerStatsTest, EmptySession) { + // Read and write statistics should be empty. Check "data" field does not exist. "wait" fields + // such as the schemaLock might have some value. + auto statsBson = WiredTigerStats{_session}.toBSON(); + ASSERT_FALSE(statsBson.hasField("data")); +} + +TEST_F(WiredTigerStatsTest, SessionWithWrite) { + write(); + + auto statsObj = WiredTigerStats{_session}.toBSON(); + auto dataSection = statsObj["data"]; + ASSERT_EQ(dataSection.type(), BSONType::Object) << statsObj; + + ASSERT(dataSection["bytesWritten"]) << statsObj; + for (auto&& [name, value] : dataSection.Obj()) { + ASSERT_EQ(value.type(), BSONType::NumberLong) << statsObj; + ASSERT_GT(value.numberLong(), 0) << statsObj; + } +} + +TEST_F(WiredTigerStatsTest, SessionWithRead) { + read(); + + auto statsObj = WiredTigerStats{_session}.toBSON(); + + auto dataSection = statsObj["data"]; + ASSERT_EQ(dataSection.type(), BSONType::Object) << statsObj; + + ASSERT(dataSection["bytesRead"]) << statsObj; + for (auto&& [name, value] : dataSection.Obj()) { + ASSERT_EQ(value.type(), BSONType::NumberLong) << statsObj; + ASSERT_GT(value.numberLong(), 0) << statsObj; + } +} + +TEST_F(WiredTigerStatsTest, SessionWithLargeWriteAndLargeRead) { + auto remaining = static_cast<int64_t>(std::numeric_limits<uint32_t>::max()) + 1; + while (remaining > 0) { + std::string data(1024 * 1024, 'a'); + remaining -= data.size(); + write(data); + } + + auto statsObj = WiredTigerStats{_session}.toBSON(); + ASSERT_GT(statsObj["data"]["bytesWritten"].numberLong(), std::numeric_limits<uint32_t>::max()) + << statsObj; + + // Closing the connection will ensure that tests actually have to read into cache. + closeConnection(); + openConnectionAndCreateSession(); + + readTestWrites(); + + statsObj = WiredTigerStats{_session}.toBSON(); + ASSERT_GT(statsObj["data"]["bytesRead"].numberLong(), std::numeric_limits<uint32_t>::max()) + << statsObj; +} + +TEST_F(WiredTigerStatsTest, OperationsAddToSessionStats) { + std::vector<std::unique_ptr<WiredTigerStats>> operationStats; + + write(); + WiredTigerStats firstWrite(_session); + operationStats.push_back(std::make_unique<WiredTigerStats>(firstWrite - WiredTigerStats{})); + read(); + WiredTigerStats firstRead(_session); + operationStats.push_back(std::make_unique<WiredTigerStats>(firstRead - firstWrite)); + write(); + WiredTigerStats secondWrite(_session); + operationStats.push_back(std::make_unique<WiredTigerStats>(secondWrite - firstRead)); + read(); + WiredTigerStats secondRead(_session); + operationStats.push_back(std::make_unique<WiredTigerStats>(secondRead - secondWrite)); + + const WiredTigerStats& fetchedSessionStats = secondRead; + + long long bytesWritten = 0; + long long timeWritingMicros = 0; + long long bytesRead = 0; + long long timeReadingMicros = 0; + + WiredTigerStats addedSessionStats; + + for (auto&& op : operationStats) { + auto statsObj = op->toBSON(); + + bytesWritten += statsObj["data"]["bytesWritten"].numberLong(); + timeWritingMicros += statsObj["data"]["timeWritingMicros"].numberLong(); + bytesRead += statsObj["data"]["bytesRead"].numberLong(); + timeReadingMicros += statsObj["data"]["timeReadingMicros"].numberLong(); + + addedSessionStats += *op; + } + + auto addedObj = addedSessionStats.toBSON(); + auto dataSection = addedObj["data"]; + ASSERT_EQ(dataSection.type(), BSONType::Object) << addedObj; + ASSERT_EQ(dataSection["bytesWritten"].numberLong(), bytesWritten) << addedObj; + ASSERT_EQ(dataSection["timeWritingMicros"].numberLong(), timeWritingMicros) << addedObj; + ASSERT_EQ(dataSection["bytesRead"].numberLong(), bytesRead) << addedObj; + ASSERT_EQ(dataSection["timeReadingMicros"].numberLong(), timeReadingMicros) << addedObj; + + auto fetchedObj = fetchedSessionStats.toBSON(); + auto fetchedDataSection = fetchedObj["data"]; + ASSERT_EQ(fetchedDataSection.type(), BSONType::Object) << fetchedObj; + ASSERT_EQ(fetchedDataSection["bytesWritten"].numberLong(), bytesWritten) << fetchedObj; + ASSERT_EQ(fetchedDataSection["timeWritingMicros"].numberLong(), timeWritingMicros) + << fetchedObj; + ASSERT_EQ(fetchedDataSection["bytesRead"].numberLong(), bytesRead) << fetchedObj; + ASSERT_EQ(fetchedDataSection["timeReadingMicros"].numberLong(), timeReadingMicros) + << fetchedObj; +} + +TEST_F(WiredTigerStatsTest, OperationsSubtractToZero) { + std::vector<std::unique_ptr<WiredTigerStats>> operationStats; + + write(); + WiredTigerStats firstWrite(_session); + operationStats.push_back(std::make_unique<WiredTigerStats>(firstWrite - WiredTigerStats{})); + read(); + WiredTigerStats firstRead(_session); + operationStats.push_back(std::make_unique<WiredTigerStats>(firstRead - firstWrite)); + write(); + WiredTigerStats secondWrite(_session); + operationStats.push_back(std::make_unique<WiredTigerStats>(secondWrite - firstRead)); + read(); + WiredTigerStats secondRead(_session); + operationStats.push_back(std::make_unique<WiredTigerStats>(secondRead - secondWrite)); + + WiredTigerStats& fetchedSessionStats = secondRead; + + // Assert fetchedSessionStats was not zero before checking subtract results in it being zero. + // We ignore the time statistics as those might still be 0 from time to time. + auto preSubtractObj = fetchedSessionStats.toBSON(); + auto preSubtract = preSubtractObj["data"]; + ASSERT_EQ(preSubtract.type(), BSONType::Object) << preSubtractObj; + ASSERT_GT(preSubtract["bytesWritten"].numberLong(), 0) << preSubtractObj; + ASSERT_GT(preSubtract["bytesRead"].numberLong(), 0) << preSubtractObj; + + for (auto&& op : operationStats) { + fetchedSessionStats -= *op; + } + + auto subtractedObj = fetchedSessionStats.toBSON(); + ASSERT_BSONOBJ_EQ(subtractedObj, BSONObj{}); +} + +TEST_F(WiredTigerStatsTest, Clone) { + write(); + + WiredTigerStats stats{_session}; + auto clone = stats.clone(); + + ASSERT_BSONOBJ_EQ(stats.toBSON(), clone->toBSON()); + + stats += *clone; + ASSERT_BSONOBJ_NE(stats.toBSON(), clone->toBSON()); +} + +} // namespace +} // namespace mongo diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_util.cpp b/src/mongo/db/storage/wiredtiger/wiredtiger_util.cpp index de31ec10751..d374b285d8d 100644 --- a/src/mongo/db/storage/wiredtiger/wiredtiger_util.cpp +++ b/src/mongo/db/storage/wiredtiger/wiredtiger_util.cpp @@ -37,12 +37,15 @@ #include <boost/filesystem.hpp> #include <boost/filesystem/path.hpp> +#include <pcrecpp.h> #include "mongo/base/simple_string_data_comparator.h" #include "mongo/bson/bsonobjbuilder.h" +#include "mongo/bson/json.h" +#include "mongo/db/concurrency/exception_util.h" +#include "mongo/db/concurrency/exception_util_gen.h" #include "mongo/db/concurrency/temporarily_unavailable_exception.h" #include "mongo/db/concurrency/write_conflict_exception.h" -#include "mongo/db/concurrency/write_conflict_exception_gen.h" #include "mongo/db/global_settings.h" #include "mongo/db/server_options_general_gen.h" #include "mongo/db/snapshot_window_options_gen.h" @@ -54,7 +57,6 @@ #include "mongo/db/storage/wiredtiger/wiredtiger_session_cache.h" #include "mongo/logv2/log.h" #include "mongo/util/assert_util.h" -#include "mongo/util/fail_point.h" #include "mongo/util/processinfo.h" #include "mongo/util/scopeguard.h" #include "mongo/util/static_immortal.h" @@ -62,12 +64,13 @@ #include "mongo/util/testing_proctor.h" // From src/third_party/wiredtiger/src/include/txn.h -#define WT_TXN_ROLLBACK_REASON_CACHE "oldest pinned transaction ID rolled back for eviction" +#define WT_TXN_ROLLBACK_REASON_OLDEST_FOR_EVICTION \ + "oldest pinned transaction ID rolled back for eviction" +#define WT_TXN_ROLLBACK_REASON_TOO_LARGE_FOR_CACHE \ + "transaction is too large and will not fit in the storage engine cache" namespace mongo { -MONGO_FAIL_POINT_DEFINE(crashAfterUpdatingFirstTableLoggingSettings); - namespace { const std::string kTableChecksFileName = "_wt_table_checks"; @@ -75,41 +78,6 @@ const std::string kTableExtension = ".wt"; const std::string kWiredTigerBackupFile = "WiredTiger.backup"; /** - * Returns true if the 'kTableChecksFileName' file exists in the dbpath. - * - * Must be called before createTableChecksFile() or removeTableChecksFile() to get accurate results. - */ -bool hasPreviouslyIncompleteTableChecks() { - auto path = boost::filesystem::path(storageGlobalParams.dbpath) / - boost::filesystem::path(kTableChecksFileName); - - return boost::filesystem::exists(path); -} - -/** - * Creates the 'kTableChecksFileName' file in the dbpath. - */ -void createTableChecksFile() { - auto path = boost::filesystem::path(storageGlobalParams.dbpath) / - boost::filesystem::path(kTableChecksFileName); - - boost::filesystem::ofstream fileStream(path); - fileStream << "This file indicates that a WiredTiger table check operation is in progress or " - "incomplete." - << std::endl; - if (fileStream.fail()) { - LOGV2_FATAL_NOTRACE(4366400, - "Failed to write to file", - "file"_attr = path.generic_string(), - "error"_attr = errnoWithDescription()); - } - fileStream.close(); - - fassertNoTrace(4366401, fsyncFile(path)); - fassertNoTrace(4366402, fsyncParentDirectory(path)); -} - -/** * Removes the 'kTableChecksFileName' file in the dbpath, if it exists. */ void removeTableChecksFile() { @@ -159,34 +127,87 @@ void setTableWriteTimestampAssertion(WiredTigerSessionCache* sessionCache, using std::string; -Mutex WiredTigerUtil::_tableLoggingInfoMutex = - MONGO_MAKE_LATCH("WiredTigerUtil::_tableLoggingInfoMutex"); -WiredTigerUtil::TableLoggingInfo WiredTigerUtil::_tableLoggingInfo; - bool wasRollbackReasonCachePressure(WT_SESSION* session) { if (session) { const auto reason = session->get_rollback_reason(session); if (reason) { - return strncmp(WT_TXN_ROLLBACK_REASON_CACHE, + return strncmp(WT_TXN_ROLLBACK_REASON_OLDEST_FOR_EVICTION, reason, - sizeof(WT_TXN_ROLLBACK_REASON_CACHE)) == 0; + sizeof(WT_TXN_ROLLBACK_REASON_OLDEST_FOR_EVICTION)) == 0; } } return false; } +/** + * Configured WT cache is deemed insufficient for a transaction when its dirty bytes in cache + * exceed a certain threshold on the proportion of total cache which is used by transaction. + * + * For instance, if the transaction uses 80% of WT cache and the threshold is set to 75%, the + * transaction is considered too large. + */ +bool isCacheInsufficientForTransaction(WT_SESSION* session, double threshold) { + StatusWith<int64_t> txnDirtyBytes = WiredTigerUtil::getStatisticsValue( + session, "statistics:session", "", WT_STAT_SESSION_TXN_BYTES_DIRTY); + if (!txnDirtyBytes.isOK()) { + tasserted(6190900, + str::stream() << "unable to gather the WT session's txn dirty bytes: " + << txnDirtyBytes.getStatus()); + } + + StatusWith<int64_t> cacheDirtyBytes = WiredTigerUtil::getStatisticsValue( + session, "statistics:", "", WT_STAT_CONN_CACHE_BYTES_DIRTY); + if (!cacheDirtyBytes.isOK()) { + tasserted(6190901, + str::stream() << "unable to gather the WT connection's cache dirty bytes: " + << txnDirtyBytes.getStatus()); + } + + + double txnBytesDirtyOverCacheBytesDirty = + static_cast<double>(txnDirtyBytes.getValue()) / cacheDirtyBytes.getValue(); + + LOGV2_DEBUG(6190902, + 2, + "Checking if transaction can eventually succeed", + "txnDirtyBytes"_attr = txnDirtyBytes.getValue(), + "cacheDirtyBytes"_attr = cacheDirtyBytes.getValue(), + "txnBytesDirtyOverCacheBytesDirty"_attr = txnBytesDirtyOverCacheBytesDirty, + "threshold"_attr = threshold); + + return txnBytesDirtyOverCacheBytesDirty > threshold; +} + Status wtRCToStatus_slow(int retCode, WT_SESSION* session, StringData prefix) { if (retCode == 0) return Status::OK(); + const auto generateContextStrStream = [&](StringData reason) { + str::stream contextStrStream; + if (!prefix.empty()) + contextStrStream << prefix << " "; + contextStrStream << retCode << ": " << reason; + + return contextStrStream; + }; + if (retCode == WT_ROLLBACK) { - if (gEnableTemporarilyUnavailableExceptions.load() && - wasRollbackReasonCachePressure(session)) { - str::stream s; - if (!prefix.empty()) - s << prefix << " "; - s << retCode << ": " << WT_TXN_ROLLBACK_REASON_CACHE; - throw TemporarilyUnavailableException(s); + double cacheThreshold = gTransactionTooLargeForCacheThreshold.load(); + bool txnTooLargeEnabled = cacheThreshold < 1.0; + bool temporarilyUnavailableEnabled = gEnableTemporarilyUnavailableExceptions.load(); + bool reasonWasCachePressure = (txnTooLargeEnabled || temporarilyUnavailableEnabled) && + wasRollbackReasonCachePressure(session); + + if (reasonWasCachePressure) { + if (txnTooLargeEnabled && isCacheInsufficientForTransaction(session, cacheThreshold)) { + auto s = generateContextStrStream(WT_TXN_ROLLBACK_REASON_TOO_LARGE_FOR_CACHE); + throwTransactionTooLargeForCache(s); + } + + if (temporarilyUnavailableEnabled) { + auto s = generateContextStrStream(WT_TXN_ROLLBACK_REASON_OLDEST_FOR_EVICTION); + throw TemporarilyUnavailableException(s); + } } throw WriteConflictException(prefix); @@ -195,10 +216,7 @@ Status wtRCToStatus_slow(int retCode, WT_SESSION* session, StringData prefix) { // Don't abort on WT_PANIC when repairing, as the error will be handled at a higher layer. fassert(28559, retCode != WT_PANIC || storageGlobalParams.repair); - str::stream s; - if (!prefix.empty()) - s << prefix << " "; - s << retCode << ": " << wiredtiger_strerror(retCode); + auto s = generateContextStrStream(wiredtiger_strerror(retCode)); if (retCode == EINVAL) { return Status(ErrorCodes::BadValue, s); @@ -447,7 +465,7 @@ StatusWith<int64_t> WiredTigerUtil::checkApplicationMetadataFormatVersion(Operat // static Status WiredTigerUtil::checkTableCreationOptions(const BSONElement& configElem) { - invariant(configElem.fieldNameStringData() == "configString"); + invariant(configElem.fieldNameStringData() == WiredTigerUtil::kConfigStringField); if (configElem.type() != String) { return {ErrorCodes::TypeMismatch, "'configString' must be a string."}; @@ -826,20 +844,7 @@ int WiredTigerUtil::verifyTable(OperationContext* opCtx, } void WiredTigerUtil::notifyStartupComplete() { - { - stdx::lock_guard<Latch> lk(_tableLoggingInfoMutex); - invariant(_tableLoggingInfo.isInitializing); - _tableLoggingInfo.isInitializing = false; - } - - if (!storageGlobalParams.readOnly) { - removeTableChecksFile(); - } -} - -void WiredTigerUtil::resetTableLoggingInfo() { - stdx::lock_guard<Latch> lk(_tableLoggingInfoMutex); - _tableLoggingInfo = TableLoggingInfo(); + removeTableChecksFile(); } bool WiredTigerUtil::useTableLogging(const NamespaceString& nss) { @@ -873,120 +878,16 @@ bool WiredTigerUtil::useTableLogging(const NamespaceString& nss) { } Status WiredTigerUtil::setTableLogging(OperationContext* opCtx, const std::string& uri, bool on) { - // Try to close as much as possible to avoid EBUSY errors. - WiredTigerRecoveryUnit::get(opCtx)->getSession()->closeAllCursors(uri); - WiredTigerSessionCache* sessionCache = WiredTigerRecoveryUnit::get(opCtx)->getSessionCache(); - sessionCache->closeAllCursors(uri); - - invariant(!storageGlobalParams.readOnly); - stdx::lock_guard<Latch> lk(_tableLoggingInfoMutex); - - // Update the table logging settings regardless if we're no longer starting up the process. - if (!_tableLoggingInfo.isInitializing) { - return _setTableLogging(sessionCache, uri, on); - } - - // During the start up process, the table logging settings are checked for each table to verify - // that they are set appropriately. We can speed this process up by assuming that the logging - // setting is identical for each table. - // We cross reference the logging settings for the first table and if it isn't correctly set, we - // change the logging settings for all tables during start up. - // In the event that the server wasn't shutdown cleanly, the logging settings will be modified - // for all tables as a safety precaution, or if repair mode is running. - if (_tableLoggingInfo.isFirstTable && hasPreviouslyIncompleteTableChecks()) { - _tableLoggingInfo.hasPreviouslyIncompleteTableChecks = true; - } - if (gWiredTigerSkipTableLoggingChecksOnStartup) { - if (_tableLoggingInfo.hasPreviouslyIncompleteTableChecks) { - LOGV2_FATAL_NOTRACE( - 5548300, - "Cannot use the 'wiredTigerSkipTableLoggingChecksOnStartup' startup parameter when " - "there are previously incomplete table checks"); - } - - // Only log this warning once. - if (_tableLoggingInfo.isFirstTable) { - _tableLoggingInfo.isFirstTable = false; - LOGV2_WARNING_OPTIONS( - 5548301, - {logv2::LogTag::kStartupWarnings}, - "Skipping table logging checks for all existing WiredTiger tables on startup", - "wiredTigerSkipTableLoggingChecksOnStartup"_attr = - gWiredTigerSkipTableLoggingChecksOnStartup); - } - LOGV2_DEBUG(5548302, 1, "Skipping table logging check", "uri"_attr = uri); return Status::OK(); } - if (storageGlobalParams.repair || _tableLoggingInfo.hasPreviouslyIncompleteTableChecks) { - if (_tableLoggingInfo.isFirstTable) { - _tableLoggingInfo.isFirstTable = false; - if (!_tableLoggingInfo.hasPreviouslyIncompleteTableChecks) { - createTableChecksFile(); - } - - LOGV2(4366405, - "Modifying the table logging settings for all existing WiredTiger tables", - "loggingEnabled"_attr = on, - "repair"_attr = storageGlobalParams.repair, - "hasPreviouslyIncompleteTableChecks"_attr = - _tableLoggingInfo.hasPreviouslyIncompleteTableChecks); - } - - return _setTableLogging(sessionCache, uri, on); - } - - if (!_tableLoggingInfo.isFirstTable) { - if (_tableLoggingInfo.changeTableLogging) { - return _setTableLogging(sessionCache, uri, on); - } - - // The table logging settings do not need to be modified. - return Status::OK(); - } - - invariant(_tableLoggingInfo.isFirstTable); - invariant(!_tableLoggingInfo.hasPreviouslyIncompleteTableChecks); - - // When repair or a forced modification to the table logging settings isn't running, check that - // the first table is the catalog. - invariant(uri == "table:_mdb_catalog", str::stream() << "First table checked was: " << uri); - _tableLoggingInfo.isFirstTable = false; - - // Check if the first tables logging settings need to be modified. - const std::string setting = on ? "log=(enabled=true)" : "log=(enabled=false)"; - const std::string existingMetadata = getMetadataCreate(opCtx, uri).getValue(); - if (existingMetadata.find(setting) != std::string::npos) { - // The table is running with the expected logging settings. - LOGV2(4366408, - "No table logging settings modifications are required for existing WiredTiger tables", - "loggingEnabled"_attr = on); - return Status::OK(); - } - - // The first table is running with the incorrect logging settings. All tables will need to have - // their logging settings modified. - _tableLoggingInfo.changeTableLogging = true; - createTableChecksFile(); - - LOGV2(4366406, - "Modifying the table logging settings for all existing WiredTiger tables", - "loggingEnabled"_attr = on); - - Status status = _setTableLogging(sessionCache, uri, on); - - if (MONGO_unlikely(crashAfterUpdatingFirstTableLoggingSettings.shouldFail())) { - LOGV2_FATAL_NOTRACE( - 4366407, "Crashing due to 'crashAfterUpdatingFirstTableLoggingSettings' fail point"); - } - return status; -} + // Try to close as much as possible to avoid EBUSY errors. + WiredTigerRecoveryUnit::get(opCtx)->getSession()->closeAllCursors(uri); + WiredTigerSessionCache* sessionCache = WiredTigerRecoveryUnit::get(opCtx)->getSessionCache(); + sessionCache->closeAllCursors(uri); -Status WiredTigerUtil::_setTableLogging(WiredTigerSessionCache* sessionCache, - const std::string& uri, - bool on) { const std::string setting = on ? "log=(enabled=true)" : "log=(enabled=false)"; // This method does some "weak" parsing to see if the table is in the expected logging @@ -1290,5 +1191,9 @@ std::string WiredTigerUtil::generateWTVerboseConfiguration() { return cfg; } +void WiredTigerUtil::removeEncryptionFromConfigString(std::string* configString) { + static const StaticImmortal<pcrecpp::RE> encryptionOptsRegex(R"re(encryption=\([^\)]*\),?)re"); + encryptionOptsRegex->GlobalReplace("", configString); +} } // namespace mongo diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_util.h b/src/mongo/db/storage/wiredtiger/wiredtiger_util.h index 83bcd4e0293..618add61bf8 100644 --- a/src/mongo/db/storage/wiredtiger/wiredtiger_util.h +++ b/src/mongo/db/storage/wiredtiger/wiredtiger_util.h @@ -158,6 +158,8 @@ private: WiredTigerUtil(); public: + static constexpr StringData kConfigStringField = "configString"_sd; + /** * Fetch the type and source fields out of the colgroup metadata. 'tableUri' must be a * valid table: uri. @@ -315,8 +317,6 @@ public: static void notifyStartupComplete(); - static void resetTableLoggingInfo(); - static bool useTableLogging(const NamespaceString& nss); static Status setTableLogging(OperationContext* opCtx, const std::string& uri, bool on); @@ -334,6 +334,14 @@ public: template <typename T> static T castStatisticsValue(uint64_t statisticsValue); + /** + * Removes encryption configuration from a config string. Should only be applied on custom + * config strings on secondaries. Fixes an issue where encryption configuration might be + * replicated to non-encrypted nodes, or nodes with different encryption options, causing + * initial sync or replication to fail. See SERVER-68122. + */ + static void removeEncryptionFromConfigString(std::string* configString); + private: /** * Casts unsigned 64-bit statistics value to T. @@ -341,20 +349,6 @@ private: */ template <typename T> static T _castStatisticsValue(uint64_t statisticsValue, T maximumResultType); - - static Status _setTableLogging(WiredTigerSessionCache* sessionCache, - const std::string& uri, - bool on); - - // Used to keep track of the table logging setting modifications during start up. The mutex must - // be held prior to accessing any of the member variables in the struct. - static Mutex _tableLoggingInfoMutex; - static struct TableLoggingInfo { - bool isInitializing = true; - bool isFirstTable = true; - bool changeTableLogging = false; - bool hasPreviouslyIncompleteTableChecks = false; - } _tableLoggingInfo; }; class WiredTigerConfigParser { diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_util_test.cpp b/src/mongo/db/storage/wiredtiger/wiredtiger_util_test.cpp index 088fb820474..099dc5ce453 100644 --- a/src/mongo/db/storage/wiredtiger/wiredtiger_util_test.cpp +++ b/src/mongo/db/storage/wiredtiger/wiredtiger_util_test.cpp @@ -452,4 +452,67 @@ TEST(WiredTigerUtilTest, GenerateVerboseConfiguration) { } } +TEST(WiredTigerUtilTest, RemoveEncryptionFromConfigString) { + { // Found at the middle. + std::string input{ + "debug_mode=(table_logging=true,checkpoint_retention=4),encryption=(name=AES256-CBC," + "keyid=" + "\".system\"),extensions=[local={entry=mongo_addWiredTigerEncryptors,early_load=true},," + "],"}; + const std::string expectedOutput{ + "debug_mode=(table_logging=true,checkpoint_retention=4),extensions=[local={entry=mongo_" + "addWiredTigerEncryptors,early_load=true},,],"}; + WiredTigerUtil::removeEncryptionFromConfigString(&input); + ASSERT_EQUALS(input, expectedOutput); + } + { // Found at start. + std::string input{ + "encryption=(name=AES256-CBC,keyid=\".system\"),extensions=[local={entry=mongo_" + "addWiredTigerEncryptors,early_load=true},,],"}; + const std::string expectedOutput{ + "extensions=[local={entry=mongo_addWiredTigerEncryptors,early_load=true},,],"}; + WiredTigerUtil::removeEncryptionFromConfigString(&input); + ASSERT_EQUALS(input, expectedOutput); + } + { // Found at the end. + std::string input{ + "debug_mode=(table_logging=true,checkpoint_retention=4),encryption=(name=AES256-CBC," + "keyid=\".system\")"}; + const std::string expectedOutput{"debug_mode=(table_logging=true,checkpoint_retention=4),"}; + WiredTigerUtil::removeEncryptionFromConfigString(&input); + ASSERT_EQUALS(input, expectedOutput); + } + { // Matches full configString. + std::string input{"encryption=(name=AES256-CBC,keyid=\".system\")"}; + const std::string expectedOutput{""}; + WiredTigerUtil::removeEncryptionFromConfigString(&input); + ASSERT_EQUALS(input, expectedOutput); + } + { // Matches full configString, trailing comma. + std::string input{"encryption=(name=AES256-CBC,keyid=\".system\"),"}; + const std::string expectedOutput{""}; + WiredTigerUtil::removeEncryptionFromConfigString(&input); + ASSERT_EQUALS(input, expectedOutput); + } + { // No match. + std::string input{"debug_mode=(table_logging=true,checkpoint_retention=4)"}; + const std::string expectedOutput{"debug_mode=(table_logging=true,checkpoint_retention=4)"}; + WiredTigerUtil::removeEncryptionFromConfigString(&input); + ASSERT_EQUALS(input, expectedOutput); + } + { // No match, empty. + std::string input{""}; + const std::string expectedOutput{""}; + WiredTigerUtil::removeEncryptionFromConfigString(&input); + ASSERT_EQUALS(input, expectedOutput); + } + { // Removes multiple instances. + std::string input{ + "encryption=(name=AES256-CBC,keyid=\".system\"),debug_mode=(table_logging=true," + "checkpoint_retention=4),encryption=(name=AES256-CBC,keyid=\".system\")"}; + const std::string expectedOutput{"debug_mode=(table_logging=true,checkpoint_retention=4),"}; + WiredTigerUtil::removeEncryptionFromConfigString(&input); + ASSERT_EQUALS(input, expectedOutput); + } +} } // namespace mongo |