Merge branch 'release-7.3' into 7.3_DD_init_logging

Author: saintstack

.gitignore

@@ -94,6 +94,7 @@ flow/coveragetool/obj
 /.ccls-cache
 /.clangd
 /.cache
+TAGS
 
 # Temporary and user configuration files
 *~

fdbclient/ServerKnobs.cpp

@@ -969,6 +969,7 @@ void ServerKnobs::initialize(Randomize randomize, ClientKnobs* clientKnobs, IsSi
 	init( DD_PREFER_LOW_READ_UTIL_TEAM,                          true );
 	init( DD_TRACE_MOVE_BYTES_AVERAGE_INTERVAL,                   120);
 	init( MOVING_WINDOW_SAMPLE_SIZE,                         10000000); // 10MB
+	init( DD_TEAMS_BY_SERVER_IDS_CONSISTENCY_CHECK_PROB_SIM,       0.0 ); if( isSimulated ) DD_TEAMS_BY_SERVER_IDS_CONSISTENCY_CHECK_PROB_SIM = (deterministicRandom()->random01() * 0.4) + 0.1;
 
 	//Storage Server
 	init( STORAGE_LOGGING_DELAY,                                 5.0 );
@@ -1132,6 +1133,7 @@ void ServerKnobs::initialize(Randomize randomize, ClientKnobs* clientKnobs, IsSi
 	init( CONFIGURATION_ROWS_TO_FETCH,                         20000 );
 	init( DISABLE_DUPLICATE_LOG_WARNING,                       false );
 	init( HISTOGRAM_REPORT_INTERVAL,                           300.0 );
+	init( GENERIC_METRICS_REPORT_INTERVAL,                     isSimulated ? 10.0 : 300.0 );
 
 	// Timekeeper
 	init( TIME_KEEPER_DELAY,                                      10 );

fdbclient/include/fdbclient/ServerKnobs.h

@@ -986,6 +986,9 @@ class ServerKnobs : public KnobsImpl<ServerKnobs> {
 	// Rolling window duration over which the average bytes moved by DD is calculated for the 'MovingData' trace event.
 	double DD_TRACE_MOVE_BYTES_AVERAGE_INTERVAL;
 	int64_t MOVING_WINDOW_SAMPLE_SIZE;
+	// Probability of running the consistency check between teams and teamsByServerIDs
+	// in getTeamByServers (simulation only). 0.0 in prod, random [0.1, 0.5) in simulation.
+	double DD_TEAMS_BY_SERVER_IDS_CONSISTENCY_CHECK_PROB_SIM;
 
 	// Storage Server
 	double STORAGE_LOGGING_DELAY;
@@ -1144,6 +1147,7 @@ class ServerKnobs : public KnobsImpl<ServerKnobs> {
 	int CONFIGURATION_ROWS_TO_FETCH;
 	bool DISABLE_DUPLICATE_LOG_WARNING;
 	double HISTOGRAM_REPORT_INTERVAL;
+	double GENERIC_METRICS_REPORT_INTERVAL;
 
 	// Timekeeper
 	int64_t TIME_KEEPER_DELAY;

fdbrpc/FlowTransport.actor.cpp

@@ -315,16 +315,6 @@ class TransportData {
 		countConnEstablished.init("Net2.CountConnEstablished"_sr);
 		countConnClosedWithError.init("Net2.CountConnClosedWithError"_sr);
 		countConnClosedWithoutError.init("Net2.CountConnClosedWithoutError"_sr);
-		countConnIncompatible.init("Net2.CountConnIncompatible"_sr);
-		countConnIncompatibleWithOldClient.init("Net2.CountConnIncompatibleWithOldClient"_sr);
-		countConnHandshakeAccepted.init("Net2.CountConnHandshakeAccepted"_sr);
-		countConnHandshakeRequested.init("Net2.CountConnHandshakeRequested"_sr);
-		countIncomingConnRequested.init("Net2.CountIncomingConnRequested"_sr);
-		countIncomingConnAccepted.init("Net2.CountIncomingConnAccepted"_sr);
-		countOutgoingConnHandshakeComplete.init("Net2.CountOutgoingConnHandshakeComplete"_sr);
-		countOutgoingConnHandshakeRequested.init("Net2.CountOutgoingConnHandshakeRequested"_sr);
-		countIncomingConnectionTimedout.init("Net2.CountIncomingConnectionTimedout"_sr);
-		countIncomingConnConnected.init("Net2.CountIncomingConnConnected"_sr);
 	}
 
 	Reference<struct Peer> getPeer(NetworkAddress const& address);
@@ -353,16 +343,6 @@ class TransportData {
 	Int64MetricHandle countConnEstablished;
 	Int64MetricHandle countConnClosedWithError;
 	Int64MetricHandle countConnClosedWithoutError;
-	Int64MetricHandle countConnIncompatible;
-	Int64MetricHandle countConnIncompatibleWithOldClient;
-	Int64MetricHandle countConnHandshakeAccepted;
-	Int64MetricHandle countConnHandshakeRequested;
-	Int64MetricHandle countIncomingConnRequested;
-	Int64MetricHandle countIncomingConnAccepted;
-	Int64MetricHandle countOutgoingConnHandshakeComplete;
-	Int64MetricHandle countOutgoingConnHandshakeRequested;
-	Int64MetricHandle countIncomingConnectionTimedout;
-	Int64MetricHandle countIncomingConnConnected;
 
 	std::map<NetworkAddress, std::pair<uint64_t, double>> incompatiblePeers;
 	AsyncTrigger incompatiblePeersChanged;
@@ -839,9 +819,14 @@ ACTOR Future<Void> connectionKeeper(Reference<Peer> self,
 						when(Reference<IConnection> _conn =
 						         wait(INetworkConnections::net()->connect(self->destination))) {
 							conn = _conn;
-							self->transport->countOutgoingConnHandshakeRequested++;
+							static SimpleCounter<int64_t>* countOutgoingConnectionCreated =
+							    SimpleCounter<int64_t>::makeCounter("/Transport/TLS/OutgoingConnectionCreated");
+							countOutgoingConnectionCreated->increment(1);
 							wait(conn->connectHandshake());
-							self->transport->countOutgoingConnHandshakeComplete++;
+							static SimpleCounter<int64_t>* countOutgoingConnectionHandshakeComplete =
+							    SimpleCounter<int64_t>::makeCounter(
+							        "/Transport/TLS/OutgoingConnectionHandshakeComplete");
+							countOutgoingConnectionHandshakeComplete->increment(1);
 							self->connectLatencies.addSample(now() - self->lastConnectTime);
 							if (FlowTransport::isClient()) {
 								IFailureMonitor::failureMonitor().setStatus(self->destination, FailureStatus(false));
@@ -1527,12 +1512,17 @@ ACTOR static Future<Void> connectionReader(TransportData* transport,
 								    now() + FLOW_KNOBS->CONNECTION_ID_TIMEOUT;
 							}
 							compatible = false;
-							transport->countConnIncompatible++;
+							static SimpleCounter<int64_t>* countConnectionIncompatible =
+							    SimpleCounter<int64_t>::makeCounter("/Transport/TLS/ConnectionIncompatible");
+							countConnectionIncompatible->increment(1);
 							if (!protocolVersion.hasInexpensiveMultiVersionClient()) {
 								if (peer) {
 									peer->protocolVersion->set(protocolVersion);
 								}
-								transport->countConnIncompatibleWithOldClient++;
+								static SimpleCounter<int64_t>* countConnectionIncompatibleWithVeryOldClient =
+								    SimpleCounter<int64_t>::makeCounter(
+								        "/Transport/TLS/ConnectionIncompatibleWithVeryOldClient");
+								countConnectionIncompatibleWithVeryOldClient->increment(1);
 								// Older versions expected us to hang up. It may work even if we don't hang up here, but
 								// it's safer to keep the old behavior.
 								throw incompatible_protocol_version();
@@ -1631,9 +1621,10 @@ ACTOR static Future<Void> connectionIncoming(TransportData* self, Reference<ICon
 	entry.time = now();
 	entry.addr = conn->getPeerAddress();
 	try {
-		self->countConnHandshakeRequested++;
 		wait(conn->acceptHandshake());
-		self->countConnHandshakeAccepted++;
+		static SimpleCounter<int64_t>* countIncomingConnectionHandshakeAccepted =
+		    SimpleCounter<int64_t>::makeCounter("/Transport/TLS/IncomingConnectionHandshakeAccepted");
+		countIncomingConnectionHandshakeAccepted->increment(1);
 		state Promise<Reference<Peer>> onConnected;
 		state Future<Void> reader = connectionReader(self, conn, Reference<Peer>(), onConnected);
 		if (FLOW_KNOBS->LOG_CONNECTION_ATTEMPTS_ENABLED) {
@@ -1650,17 +1641,24 @@ ACTOR static Future<Void> connectionIncoming(TransportData* self, Reference<ICon
 			}
 			when(wait(delayJittered(FLOW_KNOBS->CONNECTION_MONITOR_TIMEOUT))) {
 				CODE_PROBE(true, "Incoming connection timed out");
-				self->countIncomingConnectionTimedout++;
+				static SimpleCounter<int64_t>* countIncomingConnectionTimedout =
+				    SimpleCounter<int64_t>::makeCounter("/Transport/TLS/IncomingConnectionTimedout");
+				countIncomingConnectionTimedout->increment(1);
 				throw timed_out();
 			}
 		}
-		self->countIncomingConnConnected++;
+		static SimpleCounter<int64_t>* countIncomingConnectionConnected =
+		    SimpleCounter<int64_t>::makeCounter("/Transport/TLS/IncomingConnectionConnected");
+		countIncomingConnectionConnected->increment(1);
 	} catch (Error& e) {
 		if (e.code() != error_code_actor_cancelled) {
 			TraceEvent("IncomingConnectionError", conn->getDebugID())
 			    .errorUnsuppressed(e)
 			    .suppressFor(1.0)
 			    .detail("FromAddress", conn->getPeerAddress());
+			static SimpleCounter<int64_t>* countIncomingConnectionFailed =
+			    SimpleCounter<int64_t>::makeCounter("/Transport/TLS/IncomingConnectionFailed");
+			countIncomingConnectionFailed->increment(1);
 			if (FLOW_KNOBS->LOG_CONNECTION_ATTEMPTS_ENABLED) {
 				entry.failed = true;
 				self->connectionHistory.push_back(entry);
@@ -1685,9 +1683,10 @@ ACTOR static Future<Void> listen(TransportData* self, NetworkAddress listenAddr)
 	state uint64_t connectionCount = 0;
 	try {
 		loop {
-			self->countIncomingConnRequested++;
 			Reference<IConnection> conn = wait(listener->accept());
-			self->countIncomingConnAccepted++;
+			static SimpleCounter<int64_t>* countIncomingConnectionCreated =
+			    SimpleCounter<int64_t>::makeCounter("/Transport/TLS/IncomingConnectionCreated");
+			countIncomingConnectionCreated->increment(1);
 			if (conn) {
 				TraceEvent("ConnectionFrom", conn->getDebugID())
 				    .suppressFor(1.0)

fdbserver/DDTeamCollection.actor.cpp

@@ -185,17 +185,43 @@ class DDTeamCollectionImpl {
 
 	// Find the team with the exact storage servers as req.src.
 	static void getTeamByServers(DDTeamCollection* self, GetTeamRequest req) {
-		const std::string servers = TCTeamInfo::serversToString(req.src);
+		getTeamByServersConsistencyCheckInSim(self);
 		Optional<Reference<IDataDistributionTeam>> res;
-		for (const auto& team : self->teams) {
-			if (team->getServerIDsStr() == servers) {
-				res = team;
-				break;
-			}
+		auto it = self->teamsByServerIDs.find(TCTeamInfo::serversToString(req.src));
+		if (it != self->teamsByServerIDs.end()) {
+			res = it->second;
 		}
 		req.reply.send(std::make_pair(res, false));
 	}
 
+	// Probabilistic consistency check between teams and teamsByServerIDs
+	// Run only in simulation with a probability of DD_TEAMS_BY_SERVER_IDS_CONSISTENCY_CHECK_PROB_SIM
+	// We may need to tune this knob if simulation runs too slowly (in real-time) and results in
+	// ExternalTimeout in Joshua
+	static void getTeamByServersConsistencyCheckInSim(DDTeamCollection* self) {
+		// This check can be expensive in prod so only run it in simulation
+		if (!g_network->isSimulated()) {
+			return;
+		}
+
+		if (deterministicRandom()->random01() < SERVER_KNOBS->DD_TEAMS_BY_SERVER_IDS_CONSISTENCY_CHECK_PROB_SIM) {
+			std::unordered_map<std::string, Reference<TCTeamInfo>> expected;
+			for (const auto& team : self->teams) {
+				expected[team->getServerIDsStr()] = team;
+			}
+			ASSERT(expected.size() == self->teamsByServerIDs.size());
+			for (const auto& [key, value] : expected) {
+				auto it = self->teamsByServerIDs.find(key);
+				ASSERT(it != self->teamsByServerIDs.end());
+				ASSERT(it->second == value);
+			}
+			TraceEvent("TeamByServerIDsConsistencyCheckPassed")
+			    .suppressFor(5.0)
+			    .detail("TeamsSize", self->teams.size())
+			    .detail("MapSize", self->teamsByServerIDs.size());
+		}
+	}
+
 	// Return a threshold of team queue size which guarantees at least DD_LONG_STORAGE_QUEUE_TEAM_MAJORITY_PERCENTILE
 	// portion of teams that have longer storage queues
 	// A team storage queue size is defined as the longest storage queue size among all SSes of the team
@@ -4783,6 +4809,7 @@ void DDTeamCollection::addTeam(const std::vector<Reference<TCServerInfo>>& newTe
 
 	// For a good team, we add it to teams and create machine team for it when necessary
 	teams.push_back(teamInfo);
+	teamsByServerIDs[teamInfo->getServerIDsStr()] = teamInfo;
 	for (auto& server : newTeamServers) {
 		server->addTeam(teamInfo);
 	}
@@ -5708,6 +5735,10 @@ void DDTeamCollection::addServer(StorageServerInterface newServer,
 
 bool DDTeamCollection::removeTeam(Reference<TCTeamInfo> team) {
 	TraceEvent("RemovedServerTeam", distributorId).detail("Team", team->getDesc());
+	auto it = teamsByServerIDs.find(team->getServerIDsStr());
+	if (it != teamsByServerIDs.end()) {
+		teamsByServerIDs.erase(it);
+	}
 	bool found = false;
 	for (int t = 0; t < teams.size(); t++) {
 		if (teams[t] == team) {

fdbserver/VersionedBTree.actor.cpp

@@ -10008,8 +10008,8 @@ TEST_CASE("Lredwood/correctness/unit/deltaTree/IntIntPair") {
 	return Void();
 }
 
-struct SimpleCounter {
-	SimpleCounter() : x(0), t(timer()), start(t), xt(0) {}
+struct ReallySimpleCounter {
+	ReallySimpleCounter() : x(0), t(timer()), start(t), xt(0) {}
 	void operator+=(int n) { x += n; }
 	void operator++() { x++; }
 	int64_t get() { return x; }
@@ -10214,12 +10214,12 @@ TEST_CASE("Lredwood/correctness/btree") {
 
 	state Version version = lastVer + 1;
 
-	state SimpleCounter mutationBytes;
-	state SimpleCounter keyBytesInserted;
-	state SimpleCounter valueBytesInserted;
-	state SimpleCounter sets;
-	state SimpleCounter rangeClears;
-	state SimpleCounter keyBytesCleared;
+	state ReallySimpleCounter mutationBytes;
+	state ReallySimpleCounter keyBytesInserted;
+	state ReallySimpleCounter valueBytesInserted;
+	state ReallySimpleCounter sets;
+	state ReallySimpleCounter rangeClears;
+	state ReallySimpleCounter keyBytesCleared;
 	state int mutationBytesThisCommit = 0;
 	state int mutationBytesTargetThisCommit = randomSize(maxCommitSize);
 

fdbserver/fdbserver.actor.cpp

@@ -85,6 +85,7 @@
 #include "flow/FaultInjection.h"
 #include "flow/flow.h"
 #include "flow/network.h"
+#include "flow/SimpleCounter.h"
 
 #if defined(__linux__) || defined(__FreeBSD__)
 #include <execinfo.h>
@@ -381,6 +382,14 @@ ACTOR Future<Void> histogramReport() {
 	}
 }
 
+ACTOR Future<Void> metricsReport() {
+	loop {
+		wait(delay(SERVER_KNOBS->GENERIC_METRICS_REPORT_INTERVAL));
+
+		simpleCounterReport();
+	}
+}
+
 void testSerializationSpeed() {
 	double tstart;
 	double build = 0, serialize = 0, deserialize = 0, copy = 0, deallocate = 0;
@@ -1944,11 +1953,6 @@ int main(int argc, char* argv[]) {
 		// Enables profiling on this thread (but does not start it)
 		registerThreadForProfiling();
 
-#ifdef _WIN32
-		// Windows needs a gentle nudge to format floats correctly
-		//_set_output_format(_TWO_DIGIT_EXPONENT);
-#endif
-
 		auto opts = CLIOptions::parseArgs(argc, argv);
 		const auto role = opts.role;
 
@@ -2019,8 +2023,8 @@ int main(int argc, char* argv[]) {
 			flushAndExit(FDB_EXIT_SUCCESS);
 		}
 
-		// Initialize the thread pool
 		CoroThreadPool::init();
+
 		// Ordinarily, this is done when the network is run. However, network thread should be set before TraceEvents
 		// are logged. This thread will eventually run the network, so call it now.
 		TraceEvent::setNetworkThread();
@@ -2159,6 +2163,7 @@ int main(int argc, char* argv[]) {
 			TraceEvent("Simulation").detail("TestFile", opts.testFile);
 
 			auto histogramReportActor = histogramReport();
+			auto metricsReportActor = metricsReport();
 
 			CLIENT_KNOBS->trace();
 			FLOW_KNOBS->trace();
@@ -2345,7 +2350,14 @@ int main(int argc, char* argv[]) {
 				                      opts.configDBType,
 				                      opts.consistencyCheckUrgentMode));
 				actors.push_back(histogramReport());
-				// actors.push_back( recurring( []{}, .001 ) );  // for ASIO latency measurement
+				actors.push_back(metricsReport());
+
+#ifdef FLOW_GRPC_ENABLED
+				if (opts.grpcAddressStrs.size() > 0) {
+					FlowGrpc::init(&opts.tlsConfig, NetworkAddress::parse(opts.grpcAddressStrs[0]));
+					actors.push_back(GrpcServer::instance()->run());
+				}
+#endif
 
 				f = stopAfter(waitForAll(actors));
 				g_network->run();
@@ -2397,6 +2409,7 @@ int main(int argc, char* argv[]) {
 			setupRunLoopProfiler();
 			auto m =
 			    startSystemMonitor(opts.dataFolder, opts.dcId, opts.zoneId, opts.zoneId, opts.localities.dataHallId());
+			auto metricsReportActor = metricsReport();
 			f = stopAfter(runTests(opts.connectionFile,
 			                       TEST_TYPE_UNIT_TESTS,
 			                       TEST_HERE,

fdbserver/include/fdbserver/DDTeamCollection.h

@@ -681,7 +681,12 @@ class DDTeamCollection : public ReferenceCounted<DDTeamCollection> {
 	std::map<Standalone<StringRef>, Reference<TCMachineInfo>> machine_info;
 	std::vector<Reference<TCMachineTeamInfo>> machineTeams; // all machine teams
 
+	// IMPORTANT: teams and teamsByServerIDs MUST be consistent, so any time we
+	// mutate teams, we must also mutate teamsByServerIDs
 	std::vector<Reference<TCTeamInfo>> teams;
+	// O(1) hash map from server ID string to team information
+	// Currently used by getTeamByServers
+	std::unordered_map<std::string, Reference<TCTeamInfo>> teamsByServerIDs;
 
 	std::vector<DDTeamCollection*> teamCollections;
 	AsyncTrigger printDetailedTeamsInfo;

fdbserver/workloads/UnitTests.actor.cpp

@@ -52,6 +52,11 @@ void forceLinkCommitProxyTests();
 void forceLinkWipedStringTests();
 void forceLinkRandomKeyValueUtilsTests();
 void forceLinkSimKmsVaultTests();
+void forceLinkRESTSimKmsVaultTest();
+void forceLinkActorFuzzUnitTests();
+void forceLinkGrpcTests();
+void forceLinkGrpcTests2();
+void forceLinkSimpleCounterTests();
 
 struct UnitTestWorkload : TestWorkload {
 	static constexpr auto NAME = "UnitTests";
@@ -118,7 +123,12 @@ struct UnitTestWorkload : TestWorkload {
 		forceLinkDDSketchTests();
 		forceLinkWipedStringTests();
 		forceLinkRandomKeyValueUtilsTests();
-		forceLinkSimKmsVaultTests();
+		forceLinkSimpleCounterTests();
+
+#ifdef FLOW_GRPC_ENABLED
+		forceLinkGrpcTests();
+		forceLinkGrpcTests2();
+#endif
 	}
 
 	Future<Void> setup(Database const& cx) override {