Add DD init visibility, metrics retries, shard tracking, scan progress, and team collection logging (apple#12913)

Add logging throughout DD startup and runtime to diagnose slow startups,
stuck data moves, and undesired server classification.

DDTxnProcessor: Log elapsed time for server list + data move read
transaction and keyServer scan. Warn when getRange(dataMoveKeys) takes
over 5 seconds.

DataDistribution: Add NumShards and NumServers to DDInitGotInitialDD.
Add DDInitResumedDataMoves summary with ValidMoves, CancelledMoves,
EmptyMoves counts and elapsed time. Log DD exit reason as DDExiting
at SevWarn. Add DDInit-prefixed trace events throughout startup sequence.

DDTeamCollection: Add Reason and Address details to UndesiredStorageServer
trace events to distinguish version lag, same-address, wrong-class, and
exclusion causes.

DDShardTracker: Log TrackInitialShardsComplete with shard count and
TrackInitialShardsMetricsComplete with elapsed time.

NativeAPI: Add retry counting and logging for getStorageMetrics timeouts.

Author: michael stack

fdbclient/NativeAPI.actor.cpp

@@ -8144,6 +8144,9 @@ ACTOR Future<std::pair<Optional<StorageMetrics>, int>> waitStorageMetrics(
     int expectedShardCount,
     Optional<Reference<TransactionState>> trState) {
 	state Span span("NAPI:WaitStorageMetrics"_loc, generateSpanID(cx->transactionTracingSample));
+	state double startTime = now();
+	state double lastLogTime = 0;
+	state int retryCount = 0;
 	loop {
 		if (trState.present()) {
 			wait(trState.get()->startTransaction());
@@ -8191,7 +8194,14 @@ ACTOR Future<std::pair<Optional<StorageMetrics>, int>> waitStorageMetrics(
 				return std::make_pair(res, -1);
 			}
 		} catch (Error& e) {
-			TraceEvent(SevDebug, "WaitStorageMetricsHandleError").error(e);
+			retryCount++;
+			// Upgrade from SevDebug to SevWarn after 60 seconds of retrying
+			Severity sev = (now() - startTime > 60.0) ? SevWarn : SevDebug;
+			TraceEvent(sev, "WaitStorageMetricsHandleError")
+			    .error(e)
+			    .detail("Keys", keys)
+			    .detail("Elapsed", now() - startTime)
+			    .detail("Retries", retryCount);
 			if (e.code() == error_code_wrong_shard_server || e.code() == error_code_all_alternatives_failed) {
 				cx->invalidateCache(tenantInfo.prefix, keys);
 				wait(delay(CLIENT_KNOBS->WRONG_SHARD_SERVER_DELAY, TaskPriority::DataDistribution));

fdbserver/DDShardTracker.actor.cpp

@@ -1426,9 +1426,23 @@ ACTOR Future<Void> trackInitialShards(DataDistributionTracker* self, Reference<I
 		wait(yield(TaskPriority::DataDistribution));
 	}
 
+	TraceEvent("TrackInitialShardsComplete", self->distributorId).detail("ShardsTracked", s);
+
+	state double changeSizesStart = now();
 	Future<Void> initialSize = changeSizes(self, KeyRangeRef(allKeys.begin, allKeys.end), 0, "ShardInit");
 	self->readyToStart.send(Void());
 	wait(initialSize);
+
+	TraceEvent("TrackInitialShardsMetricsComplete", self->distributorId)
+	    .detail("ElapsedSeconds", now() - changeSizesStart);
+
+	// DDInitDone bookends DDInitRunning — marks DD fully operational. Uses DD* prefix so
+	// the full startup sequence can be queried with Type="DD*" in trace logs.
+	// Ideally this would be in DataDistribution.actor.cpp but dataDistribution() has no
+	// callback after trackInitialShards completes — it launches all actors and waits forever.
+	// Adding a new promise to signal completion would work but risks disturbing current operation.
+	TraceEvent("DDInitDone", self->distributorId);
+
 	self->maxShardSizeUpdater = updateMaxShardSize(self->dbSizeEstimate, self->maxShardSize);
 
 	return Void();

fdbserver/DDTeamCollection.actor.cpp

@@ -1376,9 +1376,17 @@ class DDTeamCollectionImpl {
 		try {
 			loop {
 				state bool isBm = BlobMigratorInterface::isBlobMigrator(server->getLastKnownInterface().id());
-				status.isUndesired =
-				    (!self->disableFailingLaggingServers.get() && server->ssVersionTooFarBehind.get()) || isBm;
-
+				{
+					bool versionLagUndesired =
+					    !self->disableFailingLaggingServers.get() && server->ssVersionTooFarBehind.get();
+					if ((versionLagUndesired || isBm) && !status.isUndesired) {
+						TraceEvent(SevWarn, "UndesiredStorageServer", self->distributorId)
+						    .detail("Server", server->getId())
+						    .detail("Address", server->getLastKnownInterface().address())
+						    .detail("Reason", isBm ? "BlobMigrator" : "VersionLag");
+					}
+					status.isUndesired = versionLagUndesired || isBm;
+				}
 				status.isWrongConfiguration = isBm;
 				status.isWiggling = false;
 				hasWrongDC = !self->isCorrectDC(*server);
@@ -1414,6 +1422,7 @@ class DDTeamCollectionImpl {
 								TraceEvent(SevWarn, "UndesiredStorageServer", self->distributorId)
 								    .detail("Server", server->getId())
 								    .detail("Address", server->getLastKnownInterface().address())
+								    .detail("Reason", "SameAddress")
 								    .detail("OtherServer", i.second->getId())
 								    .detail("NumShards",
 								            self->shardsAffectedByTeamFailure->getNumberOfShards(server->getId()))
@@ -1439,6 +1448,8 @@ class DDTeamCollectionImpl {
 					if (self->optimalTeamCount > 0) {
 						TraceEvent(SevWarn, "UndesiredStorageServer", self->distributorId)
 						    .detail("Server", server->getId())
+						    .detail("Address", server->getLastKnownInterface().address())
+						    .detail("Reason", "WrongMachineClass")
 						    .detail("OptimalTeamCount", self->optimalTeamCount)
 						    .detail("Fitness", server->getLastKnownClass().machineClassFitness(ProcessClass::Storage));
 						status.isUndesired = true;
@@ -1515,9 +1526,16 @@ class DDTeamCollectionImpl {
 				}
 
 				if (worstStatus != DDTeamCollection::Status::NONE) {
+					const char* exclusionType = worstStatus == DDTeamCollection::Status::WIGGLING   ? "Wiggling"
+					                            : worstStatus == DDTeamCollection::Status::FAILED   ? "Failed"
+					                            : worstStatus == DDTeamCollection::Status::EXCLUDED ? "Excluded"
+					                                                                                : "Unknown";
 					TraceEvent(SevWarn, "UndesiredStorageServer", self->distributorId)
 					    .detail("Server", server->getId())
-					    .detail("Excluded", worstAddr.toString());
+					    .detail("Address", server->getLastKnownInterface().address())
+					    .detail("Reason", "Excluded")
+					    .detail("ExclusionType", exclusionType)
+					    .detail("ExcludedAddress", worstAddr.toString());
 					status.isUndesired = true;
 					status.isWrongConfiguration = true;
 

fdbserver/DDTxnProcessor.actor.cpp

@@ -261,6 +261,8 @@ class DDTxnProcessorImpl {
 		CODE_PROBE((bool)skipDDModeCheck, "DD Mode won't prevent read initial data distribution.");
 		// Get the server list in its own try/catch block since it modifies result.  We don't want a subsequent failure
 		// causing entries to be duplicated
+		// Phase 1: Single transaction to read server list and all persisted data moves
+		state double serverListAndDataMoveReadStart = now();
 		loop {
 			numDataMoves = 0;
 			server_dc.clear();
@@ -334,7 +336,12 @@ class DDTxnProcessorImpl {
 					}
 				}
 
+				state double dataMoveReadStart = now();
 				RangeResult dms = wait(tr.getRange(dataMoveKeys, CLIENT_KNOBS->TOO_MANY));
+				if (now() - dataMoveReadStart > 5.0) {
+					TraceEvent(SevWarn, "DDInitSlowDataMoveRead", distributorId)
+					    .detail("ElapsedSeconds", now() - dataMoveReadStart);
+				}
 				ASSERT(!dms.more && dms.size() < CLIENT_KNOBS->TOO_MANY);
 				// For each data move, find out the src or dst servers are in primary or remote DC.
 				for (int i = 0; i < dms.size(); ++i) {
@@ -382,6 +389,11 @@ class DDTxnProcessorImpl {
 
 				succeeded = true;
 
+				TraceEvent("DDInitServerListAndDataMoveReadComplete", distributorId)
+				    .detail("NumDataMoves", numDataMoves)
+				    .detail("NumServers", result->allServers.size())
+				    .detail("ElapsedSeconds", now() - serverListAndDataMoveReadStart);
+
 				break;
 			} catch (Error& e) {
 				TraceEvent("GetInitialTeamsRetry", distributorId).error(e);
@@ -393,6 +405,10 @@ class DDTxnProcessorImpl {
 
 		// If keyServers is too large to read in a single transaction, then we will have to break this process up into
 		// multiple transactions. In that case, each iteration should begin where the previous left off
+		// Scan keyServers in batches to build the shard map
+		state double keyServerScanStart = now();
+		state double lastScanLogTime = now();
+		state int scanBatchCount = 0;
 		while (beginKey < allKeys.end) {
 			CODE_PROBE(beginKey > allKeys.begin, "Multi-transactional getInitialDataDistribution");
 			loop {
@@ -479,6 +495,15 @@ class DDTxnProcessorImpl {
 
 					ASSERT_GT(keyServers.size(), 0);
 					beginKey = keyServers.end()[-1].key;
+					scanBatchCount++;
+					if (now() - lastScanLogTime >= 30.0) {
+						lastScanLogTime = now();
+						TraceEvent("DDInitKeyServerScanProgress", distributorId)
+						    .detail("BeginKey", beginKey)
+						    .detail("Batches", scanBatchCount)
+						    .detail("ShardsScanned", result->shards.size())
+						    .detail("ElapsedSeconds", now() - keyServerScanStart);
+					}
 					break;
 				} catch (Error& e) {
 					TraceEvent("GetInitialTeamsKeyServersRetry", distributorId).error(e);
@@ -495,6 +520,10 @@ class DDTxnProcessorImpl {
 		// a dummy shard at the end with no keys or servers makes life easier for trackInitialShards()
 		result->shards.push_back(DDShardInfo(allKeys.end));
 
+		TraceEvent("DDInitKeyServerScanComplete", distributorId)
+		    .detail("NumShards", result->shards.size())
+		    .detail("ElapsedSeconds", now() - keyServerScanStart);
+
 		if (SERVER_KNOBS->SHARD_ENCODE_LOCATION_METADATA && numDataMoves > 0) {
 			for (int shard = 0; shard < result->shards.size() - 1; ++shard) {
 				const DDShardInfo& iShard = result->shards[shard];

fdbserver/DataDistribution.actor.cpp

@@ -592,6 +592,31 @@ struct DataDistributor : NonCopyable, ReferenceCounted<DataDistributor> {
 	// Initialize the required internal states of DataDistributor from system metadata. It's necessary before
 	// DataDistributor start working. Doesn't include initialization of optional components, like TenantCache, DDQueue,
 	// Tracker, TeamCollection. The components should call its own ::init methods.
+	//
+	// DD Startup Progress (trace events in order):
+	//   DDInitRunning                     - DD process recruited and starting init
+	//   DDInitTakingMoveKeysLock          - Acquiring move keys lock
+	//   DDInitTookMoveKeysLock            - Lock acquired
+	//   DDInitGotConfiguration            - Database configuration loaded
+	//   DDInitUpdatedReplicaKeys          - Replica keys updated
+	//   DDInitSlowDataMoveRead            - (SevWarn) dataMoveKeys read taking >5s
+	//   DDInitServerListAndDataMoveReadComplete - Server list + data moves read: NumDataMoves, NumServers,
+	//   ElapsedSeconds
+	//   DDInitKeyServerScanProgress       - (every 30s) keyServer scan: BeginKey, Batches, ShardsScanned
+	//   DDInitKeyServerScanComplete       - keyServer scan done: NumShards, ElapsedSeconds
+	//   DDInitGotInitialDD                - Init data loaded: NumShards, NumServers
+	//   DDInitDataLoaded                  - Init data loaded, ElapsedSeconds (does NOT mean DD is fully operational)
+	//
+	// After init(), the following startup events fire from other components:
+	//   DDInitResumeDataMovesProgress     - (every 30s) data move resume: ValidMoves, CancelledMoves, EmptyMoves
+	//   DDInitResumedDataMoves            - Data move resume complete with counts
+	//   TrackInitialShards                - Shard tracker setup started with InitialShardCount
+	//   TrackInitialShardsComplete        - Shard trackers created: ShardsTracked
+	//   DDTrackerStarting                 - Teams ready (fires from DDTeamCollection after readyToStart + delay)
+	//   TrackInitialShardsMetricsComplete - All shard metrics received: ElapsedSeconds
+	//                                       WaitStorageMetricsHandleError may fire (SevWarn after 60s) if a
+	//                                       shard's metrics read is stuck retrying: Keys, Retries
+	//   DDInitDone                        - DD is fully operational with all shard sizes loaded
 	ACTOR static Future<Void> init(Reference<DataDistributor> self) {
 		loop {
 			wait(self->waitDataDistributorEnabled());
@@ -650,13 +675,17 @@ struct DataDistributor : NonCopyable, ReferenceCounted<DataDistributor> {
 				    .detail("E", self->initData->shards.end()[-1].key)
 				    .detail("Src", describe(self->initData->shards.end()[-2].primarySrc))
 				    .detail("Dest", describe(self->initData->shards.end()[-2].primaryDest))
+				    .detail("NumShards", self->initData->shards.size())
+				    .detail("NumServers", self->initData->allServers.size())
 				    .trackLatest(self->initialDDEventHolder->trackingKey);
 			} else {
 				TraceEvent("DDInitGotInitialDD", self->ddId)
 				    .detail("B", "")
 				    .detail("E", "")
 				    .detail("Src", "[no items]")
 				    .detail("Dest", "[no items]")
+				    .detail("NumShards", self->initData->shards.size())
+				    .detail("NumServers", self->initData->allServers.size())
 				    .trackLatest(self->initialDDEventHolder->trackingKey);
 			}
 
@@ -853,6 +882,11 @@ struct DataDistributor : NonCopyable, ReferenceCounted<DataDistributor> {
 	// TODO: unit test needed
 	ACTOR static Future<Void> resumeFromDataMoves(Reference<DataDistributor> self, Future<Void> readyToStart) {
 		state KeyRangeMap<std::shared_ptr<DataMove>>::iterator it = self->initData->dataMoveMap.ranges().begin();
+		state int validMoves = 0;
+		state int cancelledMoves = 0;
+		state int emptyMoves = 0;
+		state double resumeStart = now();
+		state double lastLogTime = now();
 
 		wait(readyToStart);
 
@@ -861,6 +895,7 @@ struct DataDistributor : NonCopyable, ReferenceCounted<DataDistributor> {
 			DataMoveType dataMoveType = getDataMoveTypeFromDataMoveId(meta.id);
 			if (meta.ranges.empty()) {
 				TraceEvent(SevInfo, "EmptyDataMoveRange", self->ddId).detail("DataMoveMetaData", meta.toString());
+				emptyMoves++;
 				continue;
 			}
 			if (meta.bulkLoadTaskState.present()) {
@@ -891,6 +926,7 @@ struct DataDistributor : NonCopyable, ReferenceCounted<DataDistributor> {
 				rs.cancelled = true;
 				self->relocationProducer.send(rs);
 				TraceEvent("DDInitScheduledCancelDataMove", self->ddId).detail("DataMove", meta.toString());
+				cancelledMoves++;
 			} else if (it.value()->valid) {
 				TraceEvent(SevDebug, "DDInitFoundDataMove", self->ddId).detail("DataMove", meta.toString());
 				ASSERT(meta.ranges.front() == it.range());
@@ -914,9 +950,24 @@ struct DataDistributor : NonCopyable, ReferenceCounted<DataDistributor> {
 				self->shardsAffectedByTeamFailure->moveShard(rs.keys, teams);
 				self->relocationProducer.send(rs);
 				wait(yield(TaskPriority::DataDistribution));
+				validMoves++;
+			}
+			if (now() - lastLogTime >= 30.0) {
+				lastLogTime = now();
+				TraceEvent("DDInitResumeDataMovesProgress", self->ddId)
+				    .detail("ValidMoves", validMoves)
+				    .detail("CancelledMoves", cancelledMoves)
+				    .detail("EmptyMoves", emptyMoves)
+				    .detail("ElapsedSeconds", now() - resumeStart);
 			}
 		}
 
+		TraceEvent("DDInitResumedDataMoves", self->ddId)
+		    .detail("ValidMoves", validMoves)
+		    .detail("CancelledMoves", cancelledMoves)
+		    .detail("EmptyMoves", emptyMoves)
+		    .detail("ElapsedSeconds", now() - resumeStart);
+
 		// Trigger background cleanup for datamove tombstones
 		if (!self->txnProcessor->isMocked()) {
 			self->addActor.send(self->removeDataMoveTombstoneBackground(self));
@@ -2588,6 +2639,7 @@ ACTOR Future<Void> dataDistribution(Reference<DataDistributor> self,
 
 	loop {
 		self->context->trackerCancelled = false;
+		state double ddStartTime = now();
 		// whether all initial shard are tracked
 		self->initialized = Promise<Void>();
 
@@ -2599,7 +2651,7 @@ ACTOR Future<Void> dataDistribution(Reference<DataDistributor> self,
 
 			wait(DataDistributor::init(self));
 
-			TraceEvent(SevInfo, "DataDistributionInitProgress", self->ddId).detail("Phase", "Metadata Initialized");
+			TraceEvent("DDInitDataLoaded", self->ddId).detail("ElapsedSeconds", now() - ddStartTime);
 
 			// When/If this assertion fails, Evan owes Ben a pat on the back for his foresight
 			ASSERT(self->configuration.storageTeamSize > 0);
@@ -2831,6 +2883,10 @@ ACTOR Future<Void> dataDistribution(Reference<DataDistributor> self,
 			self->context->markTrackerCancelled();
 			state Error err = e;
 			TraceEvent("DataDistributorDestroyTeamCollections", self->ddId).error(e);
+			// Log every DD exit with the reason. movekeys_conflict is the most common
+			// non-kill cause — it's a normal internal restart, but was previously
+			// invisible because reportErrorsExcept suppresses logging for "normal" DD errors.
+			TraceEvent(SevWarn, "DDExiting", self->ddId).error(e).detail("ErrorCode", e.code());
 			state std::vector<UID> teamForDroppedRange;
 			if (removeFailedServer.getFuture().isReady() && !removeFailedServer.getFuture().isError()) {
 				// Choose a random healthy team to host the to-be-dropped range.
@@ -5127,6 +5183,10 @@ ACTOR Future<Void> dataDistributor_impl(DataDistributorInterface di,
 	                                                    &normalDataDistributorErrors());
 
 	try {
+		TraceEvent("DataDistributorRunning", di.id());
+		// DDInitRunning duplicates the above with DDInit* prefix so the full startup sequence
+		// can be queried with Type="DDInit*" in trace logs
+		TraceEvent("DDInitRunning", di.id());
 		loop choose {
 			when(wait(distributor || collection)) {
 				ASSERT(false);

flow/include/flow/IndexedSet.h

@@ -344,11 +344,11 @@ struct IndexedSet {
 	// direction 0 = left, 1 = right
 	template <int direction>
 	static void moveIterator(Node const*& node) {
-		moveIteratorImpl<direction, true>(node);
+		IndexedSet::template moveIteratorImpl<direction, true>(node);
 	}
 	template <int direction>
 	static void moveIterator(Node*& node) {
-		moveIteratorImpl<direction, false>(node);
+		IndexedSet::template moveIteratorImpl<direction, false>(node);
 	}
 
 public: // but testonly