From 0bc25f04388e498cfb42f8a3c0b6ced499eba088 Mon Sep 17 00:00:00 2001 From: gxglass Date: Tue, 28 Apr 2026 17:37:09 -0700 Subject: [PATCH] Add 66 counters for 22 DD or related transactions to count begin, commit, and abort for each (#13062) (#13097) Problem: some of these transactions have been observed to involve cascades of many shards trying to execute them simultaneously owing to somewhat unpredictable DD dynamics. This then results in DD being pegged on CPU with many failing transactions. Solution: the counters here will make it clear if this is happening and if so, where, so that further mitigations/solutions can be devised. Some of these code paths have TraceEvents, but some don't and of the existing TraceEvents, many are sampled. This PR avoids all of that and counts every time through, making us less reliant on guesswork. An alternative approach (not taken here for many reasons) would be to instrument transaction code to let callers pass a tag. Then have the transaction client and/or server side emit a few metrics (started, committed, aborted) parameterized by tag. One may reasonably object that the boilerplate here is kind of fugly. I guess my view is that at this late date the time for more subtle approaches has come and gone. We need this code to tell us what it is doing and if this looks a little intrusive, so be it. Testing: 20260422-234326-gglass-2f514436068d99e6 compressed=True data_size=35519571 duration=5298275 ended=100000 fail=1 fail_fast=10 max_runs=100000 pass=99999 priority=100 remaining=0 runtime=0:55:44 sanity=False started=100000 stopped=20260423-003910 submitted=20260422-234326 timeout=5400 username=gglass --- fdbserver/DDTeamCollection.actor.cpp | 76 +++++++++- fdbserver/DDTxnProcessor.actor.cpp | 58 +++++++- fdbserver/DataDistribution.actor.cpp | 76 ++++++++++ fdbserver/MoveKeys.actor.cpp | 206 ++++++++++++++++++++++++++- 4 files changed, 413 insertions(+), 3 deletions(-) diff --git a/fdbserver/DDTeamCollection.actor.cpp b/fdbserver/DDTeamCollection.actor.cpp index 8ca916e4cd..9481e8bb5d 100644 --- a/fdbserver/DDTeamCollection.actor.cpp +++ b/fdbserver/DDTeamCollection.actor.cpp @@ -30,6 +30,7 @@ #include "flow/IRandom.h" #include "flow/Trace.h" #include "flow/network.h" +#include "flow/SimpleCounter.h" #include "flow/actorcompiler.h" // This must be the last #include. @@ -74,6 +75,55 @@ int EligibilityCounter::getCount(int combinedType) const { } // namespace data_distribution +static SimpleCounter* counterUpdateNextWigglingStorageIDStarted() { + static auto* c = SimpleCounter::makeCounter("/dd/updateNextWigglingStorageID/started"); + return c; +} +static SimpleCounter* counterUpdateNextWigglingStorageIDCommitted() { + static auto* c = SimpleCounter::makeCounter("/dd/updateNextWigglingStorageID/committed"); + return c; +} +static SimpleCounter* counterUpdateNextWigglingStorageIDAborted() { + static auto* c = SimpleCounter::makeCounter("/dd/updateNextWigglingStorageID/aborted"); + return c; +} +static SimpleCounter* counterPerpetualStorageWigglerStarted() { + static auto* c = SimpleCounter::makeCounter("/dd/perpetualStorageWiggler/started"); + return c; +} +static SimpleCounter* counterPerpetualStorageWigglerCommitted() { + static auto* c = SimpleCounter::makeCounter("/dd/perpetualStorageWiggler/committed"); + return c; +} +static SimpleCounter* counterPerpetualStorageWigglerAborted() { + static auto* c = SimpleCounter::makeCounter("/dd/perpetualStorageWiggler/aborted"); + return c; +} +static SimpleCounter* counterWaitHealthyZoneChangeStarted() { + static auto* c = SimpleCounter::makeCounter("/dd/waitHealthyZoneChange/started"); + return c; +} +static SimpleCounter* counterWaitHealthyZoneChangeCommitted() { + static auto* c = SimpleCounter::makeCounter("/dd/waitHealthyZoneChange/committed"); + return c; +} +static SimpleCounter* counterWaitHealthyZoneChangeAborted() { + static auto* c = SimpleCounter::makeCounter("/dd/waitHealthyZoneChange/aborted"); + return c; +} +static SimpleCounter* counterUpdateStorageMetadataStarted() { + static auto* c = SimpleCounter::makeCounter("/dd/updateStorageMetadata/started"); + return c; +} +static SimpleCounter* counterUpdateStorageMetadataCommitted() { + static auto* c = SimpleCounter::makeCounter("/dd/updateStorageMetadata/committed"); + return c; +} +static SimpleCounter* counterUpdateStorageMetadataAborted() { + static auto* c = SimpleCounter::makeCounter("/dd/updateStorageMetadata/aborted"); + return c; +} + class DDTeamCollectionImpl { ACTOR static Future checkAndRemoveInvalidLocalityAddr(DDTeamCollection* self) { state double start = now(); @@ -2276,6 +2326,9 @@ public: } ACTOR static Future updateNextWigglingStorageID(DDTeamCollection* self) { + state SimpleCounter* txnStarted = counterUpdateNextWigglingStorageIDStarted(); + state SimpleCounter* txnCommitted = counterUpdateNextWigglingStorageIDCommitted(); + state SimpleCounter* txnAborted = counterUpdateNextWigglingStorageIDAborted(); state StorageWiggleData wiggleState; state KeyBackedObjectMap metadataMap = wiggleState.wigglingStorageServer(PrimaryRegion(self->primary)); @@ -2284,13 +2337,16 @@ public: state StorageWiggleValue value(nextId); state Reference tr(new ReadYourWritesTransaction(self->dbContext())); loop { + txnStarted->increment(1); // write the next server id try { tr->setOption(FDBTransactionOptions::ACCESS_SYSTEM_KEYS); metadataMap.set(tr, nextId, value); wait(tr->commit()); + txnCommitted->increment(1); break; } catch (Error& e) { + txnAborted->increment(1); wait(tr->onError(e)); } } @@ -2560,6 +2616,9 @@ public: // command `configure perpetual_storage_wiggle=$value` if the value is 1, this actor start 2 actors, // `perpetualStorageWiggleIterator` and `perpetualStorageWiggler`. Otherwise, it sends stop signal to them. ACTOR static Future monitorPerpetualStorageWiggle(DDTeamCollection* self) { + state SimpleCounter* txnPSWStarted = counterPerpetualStorageWigglerStarted(); + state SimpleCounter* txnPSWCommitted = counterPerpetualStorageWigglerCommitted(); + state SimpleCounter* txnPSWAborted = counterPerpetualStorageWigglerAborted(); state int speed = 0; state PromiseStream finishStorageWiggleSignal; state SignalableActorCollection collection; @@ -2569,6 +2628,7 @@ public: loop { state ReadYourWritesTransaction tr(self->dbContext()); loop { + txnPSWStarted->increment(1); try { tr.setOption(FDBTransactionOptions::ACCESS_SYSTEM_KEYS); Optional> value = wait(tr.get(perpetualStorageWiggleKey)); @@ -2578,6 +2638,7 @@ public: } state Future watchFuture = tr.watch(perpetualStorageWiggleKey); wait(tr.commit()); + txnPSWCommitted->increment(1); ASSERT(speed == 1 || speed == 0); if (speed == 1 && self->storageWiggler->isStopped()) { // avoid duplicated start @@ -2600,6 +2661,7 @@ public: wait(watchFuture); break; } catch (Error& e) { + txnPSWAborted->increment(1); wait(tr.onError(e)); } } @@ -2607,8 +2669,12 @@ public: } ACTOR static Future waitHealthyZoneChange(DDTeamCollection* self) { + state SimpleCounter* txnStarted = counterWaitHealthyZoneChangeStarted(); + state SimpleCounter* txnCommitted = counterWaitHealthyZoneChangeCommitted(); + state SimpleCounter* txnAborted = counterWaitHealthyZoneChangeAborted(); state ReadYourWritesTransaction tr(self->dbContext()); loop { + txnStarted->increment(1); try { tr.setOption(FDBTransactionOptions::READ_SYSTEM_KEYS); tr.setOption(FDBTransactionOptions::LOCK_AWARE); @@ -2649,9 +2715,11 @@ public: state Future watchFuture = tr.watch(healthyZoneKey); wait(tr.commit()); + txnCommitted->increment(1); wait(watchFuture || healthyZoneTimeout); tr.reset(); } catch (Error& e) { + txnAborted->increment(1); wait(tr.onError(e)); } } @@ -3374,6 +3442,9 @@ public: } ACTOR static Future updateStorageMetadata(DDTeamCollection* self, TCServerInfo* server) { + state SimpleCounter* txnStarted = counterUpdateStorageMetadataStarted(); + state SimpleCounter* txnCommitted = counterUpdateStorageMetadataCommitted(); + state SimpleCounter* txnAborted = counterUpdateStorageMetadataAborted(); state KeyBackedObjectMap metadataMap( serverMetadataKeys.begin, IncludeVersion()); state Reference tr = makeReference(self->dbContext()); @@ -3397,6 +3468,7 @@ public: // read storage metadata loop { + txnStarted->increment(1); try { tr->setOption(FDBTransactionOptions::ACCESS_SYSTEM_KEYS); Optional serverInterfaceValue = wait(tr->get(serverListKeyFor(server->getId()))); @@ -3417,8 +3489,10 @@ public: metadataMap.set(tr, server->getId(), data); tr->set(serverMetadataChangeKey, deterministicRandom()->randomUniqueID().toString()); wait(tr->commit()); + txnCommitted->increment(1); break; } catch (Error& e) { + txnAborted->increment(1); wait(tr->onError(e)); } } @@ -7212,4 +7286,4 @@ TEST_CASE("/DataDistribution/GetTeam/PreferWithinShardRange") { } wait(DDTeamCollectionUnitTest::GetTeam_PreferShardsWithinLimit()); return Void(); -} \ No newline at end of file +} diff --git a/fdbserver/DDTxnProcessor.actor.cpp b/fdbserver/DDTxnProcessor.actor.cpp index f04e6e3cb2..5d4249fcce 100644 --- a/fdbserver/DDTxnProcessor.actor.cpp +++ b/fdbserver/DDTxnProcessor.actor.cpp @@ -23,6 +23,7 @@ #include "fdbclient/ManagementAPI.actor.h" #include "fdbserver/DataDistribution.actor.h" #include "fdbclient/DatabaseContext.h" +#include "flow/SimpleCounter.h" #include "flow/genericactors.actor.h" #include "flow/actorcompiler.h" // This must be the last #include. @@ -42,6 +43,43 @@ static void updateServersAndCompleteSources(std::set& servers, } } +static SimpleCounter* counterUpdateReplicaKeysStarted() { + static auto* c = SimpleCounter::makeCounter("/dd/updateReplicaKeys/started"); + return c; +} +static SimpleCounter* counterUpdateReplicaKeysCommitted() { + static auto* c = SimpleCounter::makeCounter("/dd/updateReplicaKeys/committed"); + return c; +} +static SimpleCounter* counterUpdateReplicaKeysAborted() { + static auto* c = SimpleCounter::makeCounter("/dd/updateReplicaKeys/aborted"); + return c; +} +static SimpleCounter* counterTryUpdateReplicasKeyForDcStarted() { + static auto* c = SimpleCounter::makeCounter("/dd/tryUpdateReplicasKeyForDc/started"); + return c; +} +static SimpleCounter* counterTryUpdateReplicasKeyForDcCommitted() { + static auto* c = SimpleCounter::makeCounter("/dd/tryUpdateReplicasKeyForDc/committed"); + return c; +} +static SimpleCounter* counterTryUpdateReplicasKeyForDcAborted() { + static auto* c = SimpleCounter::makeCounter("/dd/tryUpdateReplicasKeyForDc/aborted"); + return c; +} +static SimpleCounter* counterWaitDDTeamInfoPrintSignalStarted() { + static auto* c = SimpleCounter::makeCounter("/dd/waitDDTeamInfoPrintSignal/started"); + return c; +} +static SimpleCounter* counterWaitDDTeamInfoPrintSignalCommitted() { + static auto* c = SimpleCounter::makeCounter("/dd/waitDDTeamInfoPrintSignal/committed"); + return c; +} +static SimpleCounter* counterWaitDDTeamInfoPrintSignalAborted() { + static auto* c = SimpleCounter::makeCounter("/dd/waitDDTeamInfoPrintSignal/aborted"); + return c; +} + class DDTxnProcessorImpl { friend class DDTxnProcessor; @@ -178,8 +216,12 @@ class DDTxnProcessorImpl { std::vector> primaryDcId, std::vector> remoteDcIds, DatabaseConfiguration configuration) { + state SimpleCounter* txnStarted = counterUpdateReplicaKeysStarted(); + state SimpleCounter* txnCommitted = counterUpdateReplicaKeysCommitted(); + state SimpleCounter* txnAborted = counterUpdateReplicaKeysAborted(); state Transaction tr(cx); loop { + txnStarted->increment(1); try { tr.setOption(FDBTransactionOptions::ACCESS_SYSTEM_KEYS); tr.setOption(FDBTransactionOptions::PRIORITY_SYSTEM_IMMEDIATE); @@ -200,8 +242,10 @@ class DDTxnProcessorImpl { } wait(tr.commit()); + txnCommitted->increment(1); break; } catch (Error& e) { + txnAborted->increment(1); wait(tr.onError(e)); } } @@ -209,8 +253,12 @@ class DDTxnProcessorImpl { } ACTOR static Future tryUpdateReplicasKeyForDc(Database cx, Optional dcId, int storageTeamSize) { + state SimpleCounter* txnStarted = counterTryUpdateReplicasKeyForDcStarted(); + state SimpleCounter* txnCommitted = counterTryUpdateReplicasKeyForDcCommitted(); + state SimpleCounter* txnAborted = counterTryUpdateReplicasKeyForDcAborted(); state Transaction tr(cx); loop { + txnStarted->increment(1); tr.setOption(FDBTransactionOptions::ACCESS_SYSTEM_KEYS); tr.setOption(FDBTransactionOptions::PRIORITY_SYSTEM_IMMEDIATE); @@ -225,9 +273,11 @@ class DDTxnProcessorImpl { } tr.set(datacenterReplicasKeyFor(dcId), datacenterReplicasValue(storageTeamSize)); wait(tr.commit()); + txnCommitted->increment(1); return oldReplicas; } catch (Error& e) { + txnAborted->increment(1); wait(tr.onError(e)); } } @@ -654,15 +704,21 @@ class DDTxnProcessorImpl { } ACTOR static Future waitDDTeamInfoPrintSignal(Database cx) { + state SimpleCounter* txnStarted = counterWaitDDTeamInfoPrintSignalStarted(); + state SimpleCounter* txnCommitted = counterWaitDDTeamInfoPrintSignalCommitted(); + state SimpleCounter* txnAborted = counterWaitDDTeamInfoPrintSignalAborted(); state ReadYourWritesTransaction tr(cx); loop { + txnStarted->increment(1); try { tr.setOption(FDBTransactionOptions::ACCESS_SYSTEM_KEYS); state Future watchFuture = tr.watch(triggerDDTeamInfoPrintKey); wait(tr.commit()); + txnCommitted->increment(1); wait(watchFuture); return Void(); } catch (Error& e) { + txnAborted->increment(1); wait(tr.onError(e)); } } @@ -1194,4 +1250,4 @@ Future DDMockTxnProcessor::waitForAllDataRemoved( shardsAffectedByTeamFailure->getNumberOfShards(serverID) == 0; }, TaskPriority::DataDistribution); -} \ No newline at end of file +} diff --git a/fdbserver/DataDistribution.actor.cpp b/fdbserver/DataDistribution.actor.cpp index 3d2a365181..4c6802c663 100644 --- a/fdbserver/DataDistribution.actor.cpp +++ b/fdbserver/DataDistribution.actor.cpp @@ -48,8 +48,10 @@ #include "flow/Arena.h" #include "flow/Error.h" #include "flow/Platform.h" +#include "flow/SimpleCounter.h" #include "flow/Trace.h" #include "flow/UnitTest.h" + #include "flow/flow.h" #include "flow/genericactors.actor.h" #include "flow/serialize.h" @@ -401,6 +403,19 @@ struct DDBulkDumpJobManager { bool isValid() const { return jobState.isValid(); } }; +static SimpleCounter* counterRemoveDataMoveTombstoneStarted() { + static auto* c = SimpleCounter::makeCounter("/dd/removeDataMoveTombstone/started"); + return c; +} +static SimpleCounter* counterRemoveDataMoveTombstoneCommitted() { + static auto* c = SimpleCounter::makeCounter("/dd/removeDataMoveTombstone/committed"); + return c; +} +static SimpleCounter* counterRemoveDataMoveTombstoneAborted() { + static auto* c = SimpleCounter::makeCounter("/dd/removeDataMoveTombstone/aborted"); + return c; +} + struct DataDistributor : NonCopyable, ReferenceCounted { public: Reference const> dbInfo; @@ -738,11 +753,15 @@ public: } ACTOR static Future removeDataMoveTombstoneBackground(Reference self) { + state SimpleCounter* txnStarted = counterRemoveDataMoveTombstoneStarted(); + state SimpleCounter* txnCommitted = counterRemoveDataMoveTombstoneCommitted(); + state SimpleCounter* txnAborted = counterRemoveDataMoveTombstoneAborted(); state UID currentID; try { state Database cx = openDBOnServer(self->dbInfo, TaskPriority::DefaultEndpoint, LockAware::True); state Transaction tr(cx); loop { + txnStarted->increment(1); try { tr.setOption(FDBTransactionOptions::ACCESS_SYSTEM_KEYS); tr.setOption(FDBTransactionOptions::PRIORITY_SYSTEM_IMMEDIATE); @@ -752,8 +771,10 @@ public: TraceEvent(SevDebug, "RemoveDataMoveTombstone", self->ddId).detail("DataMoveID", currentID); } wait(tr.commit()); + txnCommitted->increment(1); break; } catch (Error& e) { + txnAborted->increment(1); wait(tr.onError(e)); } } @@ -3124,11 +3145,43 @@ ACTOR Future>> } } +static SimpleCounter* counterDdSnapSetRecoveryStarted() { + static auto* c = SimpleCounter::makeCounter("/dd/ddSnapSetRecovery/started"); + return c; +} +static SimpleCounter* counterDdSnapSetRecoveryCommitted() { + static auto* c = SimpleCounter::makeCounter("/dd/ddSnapSetRecovery/committed"); + return c; +} +static SimpleCounter* counterDdSnapSetRecoveryAborted() { + static auto* c = SimpleCounter::makeCounter("/dd/ddSnapSetRecovery/aborted"); + return c; +} +static SimpleCounter* counterDdSnapClearRecoveryStarted() { + static auto* c = SimpleCounter::makeCounter("/dd/ddSnapClearRecovery/started"); + return c; +} +static SimpleCounter* counterDdSnapClearRecoveryCommitted() { + static auto* c = SimpleCounter::makeCounter("/dd/ddSnapClearRecovery/committed"); + return c; +} +static SimpleCounter* counterDdSnapClearRecoveryAborted() { + static auto* c = SimpleCounter::makeCounter("/dd/ddSnapClearRecovery/aborted"); + return c; +} + ACTOR Future ddSnapCreateCore(DistributorSnapRequest snapReq, Reference const> db) { state Database cx = openDBOnServer(db, TaskPriority::DefaultDelay, LockAware::True); + state SimpleCounter* setRecoveryStarted = counterDdSnapSetRecoveryStarted(); + state SimpleCounter* setRecoveryCommitted = counterDdSnapSetRecoveryCommitted(); + state SimpleCounter* setRecoveryAborted = counterDdSnapSetRecoveryAborted(); + state SimpleCounter* clearRecoveryStarted = counterDdSnapClearRecoveryStarted(); + state SimpleCounter* clearRecoveryCommitted = counterDdSnapClearRecoveryCommitted(); + state SimpleCounter* clearRecoveryAborted = counterDdSnapClearRecoveryAborted(); state ReadYourWritesTransaction tr(cx); loop { + setRecoveryStarted->increment(1); try { tr.setOption(FDBTransactionOptions::ACCESS_SYSTEM_KEYS); tr.setOption(FDBTransactionOptions::LOCK_AWARE); @@ -3137,8 +3190,10 @@ ACTOR Future ddSnapCreateCore(DistributorSnapRequest snapReq, Referenceincrement(1); break; } catch (Error& e) { + setRecoveryAborted->increment(1); TraceEvent("SnapDataDistributor_WriteFlagError").error(e); wait(tr.onError(e)); } @@ -3230,6 +3285,7 @@ ACTOR Future ddSnapCreateCore(DistributorSnapRequest snapReq, Referenceincrement(1); try { tr.setOption(FDBTransactionOptions::ACCESS_SYSTEM_KEYS); tr.setOption(FDBTransactionOptions::LOCK_AWARE); @@ -3238,8 +3294,10 @@ ACTOR Future ddSnapCreateCore(DistributorSnapRequest snapReq, Referenceincrement(1); break; } catch (Error& e) { + clearRecoveryAborted->increment(1); TraceEvent("SnapDataDistributor_ClearFlagError").error(e); wait(tr.onError(e)); } @@ -3373,18 +3431,36 @@ ACTOR Future ddExclusionSafetyCheck(DistributorExclusionSafetyCheckRequest return Void(); } +static SimpleCounter* counterWaitFailCacheServerStarted() { + static auto* c = SimpleCounter::makeCounter("/dd/waitFailCacheServer/started"); + return c; +} +static SimpleCounter* counterWaitFailCacheServerCommitted() { + static auto* c = SimpleCounter::makeCounter("/dd/waitFailCacheServer/committed"); + return c; +} +static SimpleCounter* counterWaitFailCacheServerAborted() { + static auto* c = SimpleCounter::makeCounter("/dd/waitFailCacheServer/aborted"); + return c; +} ACTOR Future waitFailCacheServer(Database* db, StorageServerInterface ssi) { + state SimpleCounter* txnStarted = counterWaitFailCacheServerStarted(); + state SimpleCounter* txnCommitted = counterWaitFailCacheServerCommitted(); + state SimpleCounter* txnAborted = counterWaitFailCacheServerAborted(); state Transaction tr(*db); state Key key = storageCacheServerKey(ssi.id()); wait(waitFailureClient(ssi.waitFailure)); loop { + txnStarted->increment(1); tr.setOption(FDBTransactionOptions::ACCESS_SYSTEM_KEYS); try { tr.addReadConflictRange(storageCacheServerKeys); tr.clear(key); wait(tr.commit()); + txnCommitted->increment(1); break; } catch (Error& e) { + txnAborted->increment(1); wait(tr.onError(e)); } } diff --git a/fdbserver/MoveKeys.actor.cpp b/fdbserver/MoveKeys.actor.cpp index a16eeaacf5..59c5ab21fa 100644 --- a/fdbserver/MoveKeys.actor.cpp +++ b/fdbserver/MoveKeys.actor.cpp @@ -36,6 +36,7 @@ #include "fdbclient/ReadYourWrites.h" #include "fdbserver/BlobMigratorInterface.h" #include "fdbserver/TSSMappingUtil.actor.h" +#include "flow/SimpleCounter.h" #include "flow/actorcompiler.h" // This must be the last #include. @@ -283,9 +284,26 @@ ACTOR Future readMoveKeysLock(Database cx) { } } +static SimpleCounter* counterTakeMoveKeysLockStarted() { + static auto* c = SimpleCounter::makeCounter("/movekeys/takeMoveKeysLock/started"); + return c; +} +static SimpleCounter* counterTakeMoveKeysLockCommitted() { + static auto* c = SimpleCounter::makeCounter("/movekeys/takeMoveKeysLock/committed"); + return c; +} +static SimpleCounter* counterTakeMoveKeysLockAborted() { + static auto* c = SimpleCounter::makeCounter("/movekeys/takeMoveKeysLock/aborted"); + return c; +} + ACTOR Future takeMoveKeysLock(Database cx, UID ddId) { + state SimpleCounter* txnStarted = counterTakeMoveKeysLockStarted(); + state SimpleCounter* txnCommitted = counterTakeMoveKeysLockCommitted(); + state SimpleCounter* txnAborted = counterTakeMoveKeysLockAborted(); state Transaction tr(cx); loop { + txnStarted->increment(1); try { state MoveKeysLock lock; state UID txnId; @@ -300,6 +318,7 @@ ACTOR Future takeMoveKeysLock(Database cx, UID ddId) { lock.myOwner = deterministicRandom()->randomUniqueID(); tr.set(moveKeysLockOwnerKey, BinaryWriter::toValue(lock.myOwner, Unversioned())); wait(tr.commit()); + txnCommitted->increment(1); TraceEvent("TakeMoveKeysLockTransaction", ddId) .detail("TransactionUID", txnId) .detail("PrevOwner", lock.prevOwner.toString()) @@ -307,6 +326,7 @@ ACTOR Future takeMoveKeysLock(Database cx, UID ddId) { .detail("MyOwner", lock.myOwner.toString()); return lock; } catch (Error& e) { + txnAborted->increment(1); wait(tr.onError(e)); CODE_PROBE(true, "takeMoveKeysLock retry"); } @@ -667,6 +687,18 @@ ACTOR Future auditLocationMetadataPostCheck(Database occ, KeyRange range, } // Cleans up dest servers of a single shard, and unassigns the keyrange from the dest servers if necessary. +static SimpleCounter* counterCleanUpSingleShardDataMoveStarted() { + static auto* c = SimpleCounter::makeCounter("/movekeys/cleanUpSingleShardDataMove/started"); + return c; +} +static SimpleCounter* counterCleanUpSingleShardDataMoveCommitted() { + static auto* c = SimpleCounter::makeCounter("/movekeys/cleanUpSingleShardDataMove/committed"); + return c; +} +static SimpleCounter* counterCleanUpSingleShardDataMoveAborted() { + static auto* c = SimpleCounter::makeCounter("/movekeys/cleanUpSingleShardDataMove/aborted"); + return c; +} ACTOR Future cleanUpSingleShardDataMove(Database occ, KeyRange keys, MoveKeysLock lock, @@ -675,10 +707,14 @@ ACTOR Future cleanUpSingleShardDataMove(Database occ, const DDEnabledState* ddEnabledState) { ASSERT(SERVER_KNOBS->SHARD_ENCODE_LOCATION_METADATA); TraceEvent(SevInfo, "CleanUpSingleShardDataMoveBegin", dataMoveId).detail("Range", keys); + state SimpleCounter* txnStarted = counterCleanUpSingleShardDataMoveStarted(); + state SimpleCounter* txnCommitted = counterCleanUpSingleShardDataMoveCommitted(); + state SimpleCounter* txnAborted = counterCleanUpSingleShardDataMoveAborted(); state bool runPreCheck = true; loop { + txnStarted->increment(1); state Transaction tr(occ); try { @@ -740,6 +776,7 @@ ACTOR Future cleanUpSingleShardDataMove(Database occ, wait(waitForAll(actors)); wait(tr.commit()); + txnCommitted->increment(1); // Post validate consistency of update of keyServers and serverKeys if (SERVER_KNOBS->AUDIT_DATAMOVE_POST_CHECK) { @@ -747,6 +784,7 @@ ACTOR Future cleanUpSingleShardDataMove(Database occ, } break; } catch (Error& e) { + txnAborted->increment(1); state Error err = e; if (err.code() == error_code_location_metadata_corruption) { throw location_metadata_corruption(); @@ -961,6 +999,18 @@ ACTOR Future logWarningAfter(const char* context, double duration, std::ve // subrange of keys that the server did not already have, = complete for each subrange that it already has. Set // serverKeys[dest][keys] = "" for the dest servers of each existing shard in keys (unless that destination is a member // of servers OR if the source list is sufficiently degraded) +static SimpleCounter* counterStartMoveKeysStarted() { + static auto* c = SimpleCounter::makeCounter("/movekeys/startMoveKeys/started"); + return c; +} +static SimpleCounter* counterStartMoveKeysCommitted() { + static auto* c = SimpleCounter::makeCounter("/movekeys/startMoveKeys/committed"); + return c; +} +static SimpleCounter* counterStartMoveKeysAborted() { + static auto* c = SimpleCounter::makeCounter("/movekeys/startMoveKeys/aborted"); + return c; +} ACTOR static Future startMoveKeys(Database occ, KeyRange keys, std::vector servers, @@ -972,6 +1022,9 @@ ACTOR static Future startMoveKeys(Database occ, state TraceInterval interval("RelocateShard_StartMoveKeys"); state Future warningLogger = logWarningAfter("StartMoveKeysTooLong", 600, servers); // state TraceInterval waitInterval(""); + state SimpleCounter* txnStarted = counterStartMoveKeysStarted(); + state SimpleCounter* txnCommitted = counterStartMoveKeysCommitted(); + state SimpleCounter* txnAborted = counterStartMoveKeysAborted(); wait(startMoveKeysLock->take(TaskPriority::DataDistributionLaunch)); state FlowLock::Releaser releaser(*startMoveKeysLock); @@ -997,6 +1050,7 @@ ACTOR static Future startMoveKeys(Database occ, state int retries = 0; loop { + txnStarted->increment(1); try { retries++; @@ -1126,6 +1180,7 @@ ACTOR static Future startMoveKeys(Database occ, wait(waitForAll(actors)); wait(tr->commit()); + txnCommitted->increment(1); /*TraceEvent("StartMoveKeysCommitDone", relocationIntervalId) .detail("CommitVersion", tr.getCommittedVersion()) @@ -1134,6 +1189,7 @@ ACTOR static Future startMoveKeys(Database occ, shards += old.size() - 1; break; } catch (Error& e) { + txnAborted->increment(1); state Error err = e; if (err.code() == error_code_move_to_removed_server) throw; @@ -1266,6 +1322,18 @@ ACTOR Future checkFetchingState(Database cx, // keyServers[k].dest must be the same for all k in keys // Set serverKeys[dest][keys] = true; serverKeys[src][keys] = false for all src not in dest // Should be cancelled and restarted if keyServers[keys].dest changes (?so this is no longer true?) +static SimpleCounter* counterFinishMoveKeysStarted() { + static auto* c = SimpleCounter::makeCounter("/movekeys/finishMoveKeys/started"); + return c; +} +static SimpleCounter* counterFinishMoveKeysCommitted() { + static auto* c = SimpleCounter::makeCounter("/movekeys/finishMoveKeys/committed"); + return c; +} +static SimpleCounter* counterFinishMoveKeysAborted() { + static auto* c = SimpleCounter::makeCounter("/movekeys/finishMoveKeys/aborted"); + return c; +} ACTOR static Future finishMoveKeys(Database occ, KeyRange keys, std::vector destinationTeam, @@ -1278,6 +1346,9 @@ ACTOR static Future finishMoveKeys(Database occ, state TraceInterval interval("RelocateShard_FinishMoveKeys"); state TraceInterval waitInterval(""); state Future warningLogger = logWarningAfter("FinishMoveKeysTooLong", 600, destinationTeam); + state SimpleCounter* txnStarted = counterFinishMoveKeysStarted(); + state SimpleCounter* txnCommitted = counterFinishMoveKeysCommitted(); + state SimpleCounter* txnAborted = counterFinishMoveKeysAborted(); state Key begin = keys.begin; state Key endKey; state int retries = 0; @@ -1303,6 +1374,7 @@ ACTOR static Future finishMoveKeys(Database occ, // printf("finishMoveKeys( '%s'-'%s' )\n", begin.toString().c_str(), keys.end.toString().c_str()); loop { + txnStarted->increment(1); try { tr.trState->taskID = TaskPriority::MoveKeys; @@ -1577,12 +1649,16 @@ ACTOR static Future finishMoveKeys(Database occ, wait(waitForAll(actors)); wait(tr.commit()); + txnCommitted->increment(1); begin = endKey; break; } + // This leads to a count of transactions starting that exceeds the sum of + // committed or aborted, but this is intentional here. tr.reset(); } catch (Error& error) { + txnAborted->increment(1); if (error.code() == error_code_actor_cancelled) throw; state Error err = error; @@ -1616,6 +1692,18 @@ ACTOR static Future finishMoveKeys(Database occ, // Set keyServers[keys].dest = servers Set serverKeys[servers][keys] = dataMoveId for each // subrange of keys. // Set dataMoves[dataMoveId] = DataMoveMetaData. +static SimpleCounter* counterStartMoveShardsStarted() { + static auto* c = SimpleCounter::makeCounter("/movekeys/startMoveShards/started"); + return c; +} +static SimpleCounter* counterStartMoveShardsCommitted() { + static auto* c = SimpleCounter::makeCounter("/movekeys/startMoveShards/committed"); + return c; +} +static SimpleCounter* counterStartMoveShardsAborted() { + static auto* c = SimpleCounter::makeCounter("/movekeys/startMoveShards/aborted"); + return c; +} ACTOR static Future startMoveShards(Database occ, UID dataMoveId, std::vector ranges, @@ -1628,6 +1716,9 @@ ACTOR static Future startMoveShards(Database occ, CancelConflictingDataMoves cancelConflictingDataMoves, Optional bulkLoadTaskState) { state Future warningLogger = logWarningAfter("StartMoveShardsTooLong", 600, servers); + state SimpleCounter* txnStarted = counterStartMoveShardsStarted(); + state SimpleCounter* txnCommitted = counterStartMoveShardsCommitted(); + state SimpleCounter* txnAborted = counterStartMoveShardsAborted(); wait(startMoveKeysLock->take(TaskPriority::DataDistributionLaunch)); state FlowLock::Releaser releaser(*startMoveKeysLock); @@ -1647,6 +1738,7 @@ ACTOR static Future startMoveShards(Database occ, state bool runPreCheck = true; try { loop { + txnStarted->increment(1); state Key begin = keys.begin; state KeyRange currentKeys = keys; @@ -1680,6 +1772,7 @@ ACTOR static Future startMoveShards(Database occ, dataMove.setPhase(DataMoveMetaData::Deleting); tr.set(dataMoveKeyFor(dataMoveId), dataMoveValue(dataMove)); wait(tr.commit()); + // Don't increment committed as we prefer to increment aborted below. throw movekeys_conflict(); } if (dataMove.getPhase() == DataMoveMetaData::Running) { @@ -1922,6 +2015,7 @@ ACTOR static Future startMoveShards(Database occ, wait(waitForAll(actors)); wait(tr.commit()); + txnCommitted->increment(1); if (currentKeys.end == keys.end && bulkLoadTaskState.present()) { Version commitVersion = tr.getCommittedVersion(); @@ -1952,6 +2046,7 @@ ACTOR static Future startMoveShards(Database occ, break; } } catch (Error& e) { + txnAborted->increment(1); if (e.code() == error_code_location_metadata_corruption) { throw location_metadata_corruption(); } else if (e.code() == error_code_retry) { @@ -2048,6 +2143,18 @@ ACTOR static Future checkDataMoveComplete(Database occ, UID dataMoveId, Ke // keyServers[k].dest must be the same for all k in keys. // Set serverKeys[dest][keys] = dataMoveId; serverKeys[src][keys] = false for all src not in dest. // Clear dataMoves[dataMoveId]. +static SimpleCounter* counterFinishMoveShardsStarted() { + static auto* c = SimpleCounter::makeCounter("/movekeys/finishMoveShards/started"); + return c; +} +static SimpleCounter* counterFinishMoveShardsCommitted() { + static auto* c = SimpleCounter::makeCounter("/movekeys/finishMoveShards/committed"); + return c; +} +static SimpleCounter* counterFinishMoveShardsAborted() { + static auto* c = SimpleCounter::makeCounter("/movekeys/finishMoveShards/aborted"); + return c; +} ACTOR static Future finishMoveShards(Database occ, UID dataMoveId, std::vector targetRanges, @@ -2063,6 +2170,9 @@ ACTOR static Future finishMoveShards(Database occ, ASSERT(targetRanges.size() == 1); state KeyRange keys = targetRanges[0]; state Future warningLogger = logWarningAfter("FinishMoveShardsTooLong", 600, destinationTeam); + state SimpleCounter* txnStarted = counterFinishMoveShardsStarted(); + state SimpleCounter* txnCommitted = counterFinishMoveShardsCommitted(); + state SimpleCounter* txnAborted = counterFinishMoveShardsAborted(); state int retries = 0; state DataMoveMetaData dataMove; state bool cancelDataMove = false; @@ -2084,6 +2194,7 @@ ACTOR static Future finishMoveShards(Database occ, // This process can be split up into multiple transactions if getRange() doesn't return the entire // target range. loop { + txnStarted->increment(1); state std::vector completeSrc; state std::vector destServers; state std::unordered_set allServers; @@ -2110,6 +2221,7 @@ ACTOR static Future finishMoveShards(Database occ, dataMove.setPhase(DataMoveMetaData::Deleting); tr.set(dataMoveKeyFor(dataMoveId), dataMoveValue(dataMove)); wait(tr.commit()); + // Don't increment committed as we prefer to increment aborted below. throw movekeys_conflict(); } destServers.insert(destServers.end(), dataMove.dest.begin(), dataMove.dest.end()); @@ -2370,6 +2482,7 @@ ACTOR static Future finishMoveShards(Database occ, } wait(tr.commit()); + txnCommitted->increment(1); if (range.end == dataMove.ranges.front().end && bulkLoadTaskState.present()) { Version commitVersion = tr.getCommittedVersion(); @@ -2394,6 +2507,7 @@ ACTOR static Future finishMoveShards(Database occ, tr.reset(); } } catch (Error& error) { + txnAborted->increment(1); TraceEvent(SevWarn, "TryFinishMoveShardsError", relocationIntervalId) .errorUnsuppressed(error) .detail("DataMoveID", dataMoveId); @@ -2434,7 +2548,22 @@ ACTOR static Future finishMoveShards(Database occ, }; // anonymous namespace +static SimpleCounter* counterAddStorageServerStarted() { + static auto* c = SimpleCounter::makeCounter("/movekeys/addStorageServer/started"); + return c; +} +static SimpleCounter* counterAddStorageServerCommitted() { + static auto* c = SimpleCounter::makeCounter("/movekeys/addStorageServer/committed"); + return c; +} +static SimpleCounter* counterAddStorageServerAborted() { + static auto* c = SimpleCounter::makeCounter("/movekeys/addStorageServer/aborted"); + return c; +} ACTOR Future> addStorageServer(Database cx, StorageServerInterface server) { + state SimpleCounter* txnStarted = counterAddStorageServerStarted(); + state SimpleCounter* txnCommitted = counterAddStorageServerCommitted(); + state SimpleCounter* txnAborted = counterAddStorageServerAborted(); state Reference tr = makeReference(cx); state KeyBackedMap tssMapDB = KeyBackedMap(tssMappingKeys.begin); state KeyBackedObjectMap metadataMap(serverMetadataKeys.begin, @@ -2443,6 +2572,7 @@ ACTOR Future> addStorageServer(Database cx, StorageServe state int maxSkipTags = 1; loop { + txnStarted->increment(1); try { tr->setOption(FDBTransactionOptions::ACCESS_SYSTEM_KEYS); tr->setOption(FDBTransactionOptions::PRIORITY_SYSTEM_IMMEDIATE); @@ -2601,12 +2731,14 @@ ACTOR Future> addStorageServer(Database cx, StorageServe tr->set(serverListKeyFor(server.id()), serverListValue(server)); wait(tr->commit()); + txnCommitted->increment(1); TraceEvent("AddedStorageServerSystemKey") .detail("ServerID", server.id()) .detail("CommitVersion", tr->getCommittedVersion()); return std::make_pair(tr->getCommittedVersion(), tag); } catch (Error& e) { + txnAborted->increment(1); if (e.code() == error_code_commit_unknown_result) throw recruitment_failed(); // There is a remote possibility that we successfully added ourselves and // then someone removed us, so we have to fail @@ -2649,11 +2781,26 @@ ACTOR Future canRemoveStorageServer(Reference t return !assigned && keys[1].key == allKeys.end; } +static SimpleCounter* counterRemoveStorageServerStarted() { + static auto* c = SimpleCounter::makeCounter("/movekeys/removeStorageServer/started"); + return c; +} +static SimpleCounter* counterRemoveStorageServerCommitted() { + static auto* c = SimpleCounter::makeCounter("/movekeys/removeStorageServer/committed"); + return c; +} +static SimpleCounter* counterRemoveStorageServerAborted() { + static auto* c = SimpleCounter::makeCounter("/movekeys/removeStorageServer/aborted"); + return c; +} ACTOR Future removeStorageServer(Database cx, UID serverID, Optional tssPairID, MoveKeysLock lock, const DDEnabledState* ddEnabledState) { + state SimpleCounter* txnStarted = counterRemoveStorageServerStarted(); + state SimpleCounter* txnCommitted = counterRemoveStorageServerCommitted(); + state SimpleCounter* txnAborted = counterRemoveStorageServerAborted(); state KeyBackedMap tssMapDB = KeyBackedMap(tssMappingKeys.begin); state KeyBackedObjectMap metadataMap( serverMetadataKeys.begin, IncludeVersion()); @@ -2662,6 +2809,7 @@ ACTOR Future removeStorageServer(Database cx, state int noCanRemoveCount = 0; loop { + txnStarted->increment(1); try { tr->setOption(FDBTransactionOptions::PRIORITY_SYSTEM_IMMEDIATE); tr->setOption(FDBTransactionOptions::ACCESS_SYSTEM_KEYS); @@ -2759,6 +2907,7 @@ ACTOR Future removeStorageServer(Database cx, retry = true; wait(tr->commit()); + txnCommitted->increment(1); TraceEvent("RemoveStorageServer") .detail("State", "Success") .detail("ServerID", serverID) @@ -2766,6 +2915,7 @@ ACTOR Future removeStorageServer(Database cx, return Void(); } } catch (Error& e) { + txnAborted->increment(1); state Error err = e; wait(tr->onError(e)); TraceEvent("RemoveStorageServer").error(err).detail("State", "Retry").detail("ServerID", serverID); @@ -2776,11 +2926,26 @@ ACTOR Future removeStorageServer(Database cx, // Changes to keyServer and serverKey must happen symmetrically in a transaction. // If serverID is the last source server for a shard, the shard will be erased, and then be assigned // to teamForDroppedRange. +static SimpleCounter* counterRemoveKeysFromFailedServerStarted() { + static auto* c = SimpleCounter::makeCounter("/movekeys/removeKeysFromFailedServer/started"); + return c; +} +static SimpleCounter* counterRemoveKeysFromFailedServerCommitted() { + static auto* c = SimpleCounter::makeCounter("/movekeys/removeKeysFromFailedServer/committed"); + return c; +} +static SimpleCounter* counterRemoveKeysFromFailedServerAborted() { + static auto* c = SimpleCounter::makeCounter("/movekeys/removeKeysFromFailedServer/aborted"); + return c; +} ACTOR Future removeKeysFromFailedServer(Database cx, UID serverID, std::vector teamForDroppedRange, MoveKeysLock lock, const DDEnabledState* ddEnabledState) { + state SimpleCounter* txnStarted = counterRemoveKeysFromFailedServerStarted(); + state SimpleCounter* txnCommitted = counterRemoveKeysFromFailedServerCommitted(); + state SimpleCounter* txnAborted = counterRemoveKeysFromFailedServerAborted(); state Key begin = allKeys.begin; state std::vector src; @@ -2791,6 +2956,7 @@ ACTOR Future removeKeysFromFailedServer(Database cx, while (begin < allKeys.end) { state Transaction tr(cx); loop { + txnStarted->increment(1); try { tr.trState->taskID = TaskPriority::MoveKeys; tr.setOption(FDBTransactionOptions::PRIORITY_SYSTEM_IMMEDIATE); @@ -2938,6 +3104,7 @@ ACTOR Future removeKeysFromFailedServer(Database cx, .detail("End", currentKeys.end); wait(krmSetRangeCoalescing(&tr, serverKeysPrefixFor(serverID), currentKeys, allKeys, serverKeysFalse)); wait(tr.commit()); + txnCommitted->increment(1); TraceEvent(SevDebug, "FailedServerCommitSuccess", serverID) .detail("Begin", currentKeys.begin) .detail("End", currentKeys.end) @@ -2946,6 +3113,7 @@ ACTOR Future removeKeysFromFailedServer(Database cx, begin = currentKeys.end; break; } catch (Error& e) { + txnAborted->increment(1); TraceEvent("FailedServerError", serverID).error(e); wait(tr.onError(e)); } @@ -2973,6 +3141,18 @@ ACTOR Future removeKeysFromFailedServer(Database cx, // the place holder on the metadata put by cleanUpDataMoveCore. Then, startMoveShard gives up and exits. // No update to the metadata by the startMoveShard // For all three cases, the background cleanup only needs to cleanup the place holder +static SimpleCounter* counterCleanUpDataMoveBackgroundStarted() { + static auto* c = SimpleCounter::makeCounter("/movekeys/cleanUpDataMoveBackground/started"); + return c; +} +static SimpleCounter* counterCleanUpDataMoveBackgroundCommitted() { + static auto* c = SimpleCounter::makeCounter("/movekeys/cleanUpDataMoveBackground/committed"); + return c; +} +static SimpleCounter* counterCleanUpDataMoveBackgroundAborted() { + static auto* c = SimpleCounter::makeCounter("/movekeys/cleanUpDataMoveBackground/aborted"); + return c; +} ACTOR Future cleanUpDataMoveBackground(Database occ, UID dataMoveId, MoveKeysLock lock, @@ -2980,6 +3160,9 @@ ACTOR Future cleanUpDataMoveBackground(Database occ, KeyRange keys, const DDEnabledState* ddEnabledState, double delaySeconds) { + state SimpleCounter* txnStarted = counterCleanUpDataMoveBackgroundStarted(); + state SimpleCounter* txnCommitted = counterCleanUpDataMoveBackgroundCommitted(); + state SimpleCounter* txnAborted = counterCleanUpDataMoveBackgroundAborted(); wait(delay(std::max(10.0, delaySeconds))); TraceEvent(SevDebug, "CleanUpDataMoveBackgroundBegin", dataMoveId) .detail("DataMoveID", dataMoveId) @@ -2989,6 +3172,7 @@ ACTOR Future cleanUpDataMoveBackground(Database occ, state DataMoveMetaData dataMove; state Transaction tr(occ); loop { + txnStarted->increment(1); try { tr.trState->taskID = TaskPriority::MoveKeys; tr.setOption(FDBTransactionOptions::PRIORITY_SYSTEM_IMMEDIATE); @@ -3004,8 +3188,10 @@ ACTOR Future cleanUpDataMoveBackground(Database occ, ASSERT(dataMove.getPhase() == DataMoveMetaData::Deleting); tr.clear(dataMoveKeyFor(dataMoveId)); wait(tr.commit()); + txnCommitted->increment(1); break; } catch (Error& e) { + txnAborted->increment(1); TraceEvent(SevWarn, "CleanUpDataMoveBackgroundFail", dataMoveId).errorUnsuppressed(e); wait(tr.onError(e)); } @@ -3018,12 +3204,27 @@ ACTOR Future cleanUpDataMoveBackground(Database occ, return Void(); } +static SimpleCounter* counterCleanUpDataMoveCoreStarted() { + static auto* c = SimpleCounter::makeCounter("/movekeys/cleanUpDataMoveCore/started"); + return c; +} +static SimpleCounter* counterCleanUpDataMoveCoreCommitted() { + static auto* c = SimpleCounter::makeCounter("/movekeys/cleanUpDataMoveCore/committed"); + return c; +} +static SimpleCounter* counterCleanUpDataMoveCoreAborted() { + static auto* c = SimpleCounter::makeCounter("/movekeys/cleanUpDataMoveCore/aborted"); + return c; +} ACTOR Future cleanUpDataMoveCore(Database occ, UID dataMoveId, MoveKeysLock lock, FlowLock* cleanUpDataMoveParallelismLock, KeyRange keys, const DDEnabledState* ddEnabledState) { + state SimpleCounter* txnStarted = counterCleanUpDataMoveCoreStarted(); + state SimpleCounter* txnCommitted = counterCleanUpDataMoveCoreCommitted(); + state SimpleCounter* txnAborted = counterCleanUpDataMoveCoreAborted(); state KeyRange range; state Severity sevDm = static_cast(SERVER_KNOBS->PHYSICAL_SHARD_MOVE_LOG_SEVERITY); TraceEvent(SevInfo, "CleanUpDataMoveBegin", dataMoveId).detail("DataMoveID", dataMoveId).detail("Range", keys); @@ -3035,6 +3236,7 @@ ACTOR Future cleanUpDataMoveCore(Database occ, try { loop { + txnStarted->increment(1); state Transaction tr(occ); state std::unordered_map> physicalShardMap; state std::set oldDests; @@ -3172,6 +3374,7 @@ ACTOR Future cleanUpDataMoveCore(Database occ, wait(waitForAll(actors)); wait(tr.commit()); + txnCommitted->increment(1); TraceEvent(sevDm, "CleanUpDataMoveCommitted", dataMoveId) .detail("DataMoveID", dataMoveId) @@ -3186,6 +3389,7 @@ ACTOR Future cleanUpDataMoveCore(Database occ, break; } } catch (Error& e) { + txnAborted->increment(1); if (e.code() == error_code_location_metadata_corruption) { throw location_metadata_corruption(); } else { @@ -3504,4 +3708,4 @@ ACTOR Future prepareBlobRestore(Database occ, } } } -} \ No newline at end of file +}