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 +}