diff --git a/fdbclient/SystemData.cpp b/fdbclient/SystemData.cpp index f6e1ef6d1e..16441f0d27 100644 --- a/fdbclient/SystemData.cpp +++ b/fdbclient/SystemData.cpp @@ -690,8 +690,21 @@ UID decodeTssQuarantineKey(KeyRef const& key) { const KeyRangeRef tssMismatchKeys("\xff/tssMismatch/"_sr, "\xff/tssMismatch0"_sr); +const KeyRef serverMetadataChangeKey = "\xff\x02/serverMetadataChanges"_sr; const KeyRangeRef serverMetadataKeys("\xff/serverMetadata/"_sr, "\xff/serverMetadata0"_sr); +UID decodeServerMetadataKey(const KeyRef& key) { + // Key is packed by KeyBackedObjectMap::packKey + return TupleCodec::unpack(key.removePrefix(serverMetadataKeys.begin)); +} + +StorageMetadataType decodeServerMetadataValue(const KeyRef& value) { + StorageMetadataType type; + ObjectReader rd(value.begin(), IncludeVersion()); + rd.deserialize(type); + return type; +} + const KeyRangeRef serverTagKeys("\xff/serverTag/"_sr, "\xff/serverTag0"_sr); const KeyRef serverTagPrefix = serverTagKeys.begin; diff --git a/fdbclient/include/fdbclient/StorageServerInterface.h b/fdbclient/include/fdbclient/StorageServerInterface.h index 4057112d17..c82d4c0845 100644 --- a/fdbclient/include/fdbclient/StorageServerInterface.h +++ b/fdbclient/include/fdbclient/StorageServerInterface.h @@ -260,6 +260,14 @@ struct StorageInfo : NonCopyable, public ReferenceCounted { StorageInfo() : tag(invalidTag) {} }; +struct StorageServerMetaInfo : public StorageServerInterface { + Optional metadata; + + StorageServerMetaInfo(const StorageServerInterface& interface, + Optional metadata = Optional()) + : StorageServerInterface(interface), metadata(metadata) {} +}; + struct ServerCacheInfo { std::vector tags; // all tags in both primary and remote DC for the key-range std::vector> src_info; diff --git a/fdbclient/include/fdbclient/SystemData.h b/fdbclient/include/fdbclient/SystemData.h index 694e78c733..42ba2405c7 100644 --- a/fdbclient/include/fdbclient/SystemData.h +++ b/fdbclient/include/fdbclient/SystemData.h @@ -215,6 +215,12 @@ extern const KeyRangeRef tssMismatchKeys; // Note: storageInterfaceUID is the one stated in the file name extern const KeyRangeRef serverMetadataKeys; +// Any update to serverMetadataKeys will update this key to a random UID. +extern const KeyRef serverMetadataChangeKey; + +UID decodeServerMetadataKey(const KeyRef&); +StorageMetadataType decodeServerMetadataValue(const KeyRef&); + // "\xff/serverTag/[[serverID]]" = "[[Tag]]" // Provides the Tag for the given serverID. Used to access a // storage server's corresponding TLog in order to apply mutations. diff --git a/fdbserver/ClusterController.actor.cpp b/fdbserver/ClusterController.actor.cpp index 0b76ebf7bb..f53347db3a 100644 --- a/fdbserver/ClusterController.actor.cpp +++ b/fdbserver/ClusterController.actor.cpp @@ -1514,6 +1514,7 @@ ACTOR Future statusServer(FutureStream requests, self->cx, workers, workerIssues, + self->storageStatusInfos, &self->db.clientStatus, coordinators, incompatibleConnections, @@ -1669,6 +1670,53 @@ ACTOR Future monitorServerInfoConfig(ClusterControllerData::DBInfo* db) { } } +// Monitors storage metadata changes and updates to storage servers. +ACTOR Future monitorStorageMetadata(ClusterControllerData* self) { + state KeyBackedObjectMap metadataMap(serverMetadataKeys.begin, + IncludeVersion()); + state Reference tr = makeReference(self->cx); + state std::vector servers; + loop { + try { + servers.clear(); + tr->setOption(FDBTransactionOptions::READ_SYSTEM_KEYS); + tr->setOption(FDBTransactionOptions::LOCK_AWARE); + tr->setOption(FDBTransactionOptions::PRIORITY_SYSTEM_IMMEDIATE); + state RangeResult serverList = wait(tr->getRange(serverListKeys, CLIENT_KNOBS->TOO_MANY)); + ASSERT(!serverList.more && serverList.size() < CLIENT_KNOBS->TOO_MANY); + + servers.reserve(serverList.size()); + for (const auto ss : serverList) { + servers.push_back(StorageServerMetaInfo(decodeServerListValue(ss.value))); + } + + state RangeResult serverMetadata = wait(tr->getRange(serverMetadataKeys, CLIENT_KNOBS->TOO_MANY)); + ASSERT(!serverMetadata.more && serverMetadata.size() < CLIENT_KNOBS->TOO_MANY); + std::map idMetadata; + for (const auto& sm : serverMetadata) { + const UID id = decodeServerMetadataKey(sm.key); + idMetadata[id] = decodeServerMetadataValue(sm.value); + } + for (auto& s : servers) { + if (idMetadata.count(s.id())) { + s.metadata = idMetadata[s.id()]; + } else { + TraceEvent(SevWarn, "StorageServerMetadataMissing", self->id).detail("ServerID", s.id()); + } + } + + state Future watchFuture = tr->watch(serverMetadataChangeKey); + wait(tr->commit()); + + self->storageStatusInfos = std::move(servers); + wait(watchFuture); + tr->reset(); + } catch (Error& e) { + wait(tr->onError(e)); + } + } +} + // Monitors the global configuration version key for changes. When changes are // made, the global configuration history is read and any updates are sent to // all processes in the system by updating the ClientDBInfo object. The @@ -3074,6 +3122,7 @@ ACTOR Future clusterControllerCore(ClusterControllerFullInterface interf, self.addActor.send(timeKeeper(&self)); self.addActor.send(monitorProcessClasses(&self)); self.addActor.send(monitorServerInfoConfig(&self.db)); + self.addActor.send(monitorStorageMetadata(&self)); self.addActor.send(monitorGlobalConfig(&self.db)); self.addActor.send(updatedChangingDatacenters(&self)); self.addActor.send(updatedChangedDatacenters(&self)); diff --git a/fdbserver/DDTeamCollection.actor.cpp b/fdbserver/DDTeamCollection.actor.cpp index a4f7b447e3..77c7ee445a 100644 --- a/fdbserver/DDTeamCollection.actor.cpp +++ b/fdbserver/DDTeamCollection.actor.cpp @@ -18,15 +18,19 @@ * limitations under the License. */ -#include "fdbserver/DDTeamCollection.h" +#include + +#include "fdbclient/SystemData.h" #include "fdbrpc/simulator.h" +#include "fdbserver/BlobMigratorInterface.h" +#include "fdbserver/DDTeamCollection.h" #include "fdbserver/ExclusionTracker.actor.h" #include "fdbserver/DataDistributionTeam.h" -#include "fdbserver/BlobMigratorInterface.h" +#include "flow/IRandom.h" #include "flow/Trace.h" -#include "flow/actorcompiler.h" // This must be the last #include. #include "flow/network.h" -#include + +#include "flow/actorcompiler.h" // This must be the last #include. namespace { @@ -3078,6 +3082,7 @@ public: data.createdTime = metadata.get().createdTime; } metadataMap.set(tr, server->getId(), data); + tr->set(serverMetadataChangeKey, deterministicRandom()->randomUniqueID().toString()); wait(tr->commit()); break; } catch (Error& e) { diff --git a/fdbserver/MoveKeys.actor.cpp b/fdbserver/MoveKeys.actor.cpp index ffcace1a73..6b4750ea2f 100644 --- a/fdbserver/MoveKeys.actor.cpp +++ b/fdbserver/MoveKeys.actor.cpp @@ -1159,10 +1159,6 @@ ACTOR Future waitForShardReady(StorageServerInterface server, try { GetShardStateReply rep = wait(server.getShardState.getReply(GetShardStateRequest(keys, mode), TaskPriority::MoveKeys)); - TraceEvent("GetShardStateReadyDD", server.id()) - .detail("RepVersion", rep.first) - .detail("MinVersion", rep.second) - .log(); if (rep.first >= minVersion) { return Void(); } @@ -2483,6 +2479,7 @@ ACTOR Future> addStorageServer(Database cx, StorageServe StorageMetadataType metadata(StorageMetadataType::currentTime()); metadataMap.set(tr, server.id(), metadata); + tr->set(serverMetadataChangeKey, deterministicRandom()->randomUniqueID().toString()); tr->set(serverListKeyFor(server.id()), serverListValue(server)); wait(tr->commit()); @@ -2639,6 +2636,7 @@ ACTOR Future removeStorageServer(Database cx, } metadataMap.erase(tr, serverID); + tr->set(serverMetadataChangeKey, deterministicRandom()->randomUniqueID().toString()); retry = true; wait(tr->commit()); @@ -3262,6 +3260,7 @@ void seedShardServers(Arena& arena, CommitTransactionRef& tr, std::vectorrandomUniqueID().toString()); std::vector serverTags; std::vector serverSrcUID; diff --git a/fdbserver/Status.actor.cpp b/fdbserver/Status.actor.cpp index 4d98ad2378..e7b92bc980 100644 --- a/fdbserver/Status.actor.cpp +++ b/fdbserver/Status.actor.cpp @@ -114,12 +114,9 @@ extern const char* limitReasonDesc[]; typedef std::map EventMap; -struct StorageServerStatusInfo : public StorageServerInterface { - Optional metadata; +struct StorageServerStatusInfo : public StorageServerMetaInfo { EventMap eventMap; - StorageServerStatusInfo(const StorageServerInterface& interface, - Optional metadata = Optional()) - : StorageServerInterface(interface), metadata(metadata) {} + StorageServerStatusInfo(const StorageServerMetaInfo& info) : StorageServerMetaInfo(info, info.metadata) {} }; ACTOR static Future> latestEventOnWorker(WorkerInterface worker, std::string eventName) { @@ -1979,43 +1976,6 @@ static Future>> getServerMetrics( return results; } -ACTOR -static Future> readStorageInterfaceAndMetadata(Database cx, - bool use_system_priority) { - state KeyBackedObjectMap metadataMap(serverMetadataKeys.begin, - IncludeVersion()); - state Reference tr = makeReference(cx); - state std::vector servers; - loop { - try { - servers.clear(); - tr->setOption(FDBTransactionOptions::READ_SYSTEM_KEYS); - tr->setOption(FDBTransactionOptions::LOCK_AWARE); - if (use_system_priority) { - tr->setOption(FDBTransactionOptions::PRIORITY_SYSTEM_IMMEDIATE); - } - state RangeResult serverList = wait(tr->getRange(serverListKeys, CLIENT_KNOBS->TOO_MANY)); - ASSERT(!serverList.more && serverList.size() < CLIENT_KNOBS->TOO_MANY); - - servers.reserve(serverList.size()); - for (int i = 0; i < serverList.size(); i++) { - servers.push_back(StorageServerStatusInfo(decodeServerListValue(serverList[i].value))); - } - state std::vector> futures(servers.size()); - for (int i = 0; i < servers.size(); ++i) { - futures[i] = store(servers[i].metadata, metadataMap.get(tr, servers[i].id())); - // TraceEvent(SevDebug, "MetadataAppear", servers[i].id()).detail("Present", metadata.present()); - } - wait(waitForAll(futures)); - wait(tr->commit()); - break; - } catch (Error& e) { - wait(tr->onError(e)); - } - } - return servers; -} - namespace { const std::vector STORAGE_SERVER_METRICS_LIST{ "StorageMetrics", @@ -2027,11 +1987,14 @@ const std::vector STORAGE_SERVER_METRICS_LIST{ "StorageMetrics", } // namespace ACTOR static Future> getStorageServerStatusInfos( - Database cx, + std::vector storageMetadatas, std::unordered_map address_workers, WorkerDetails rkWorker) { - state std::vector servers = - wait(timeoutError(readStorageInterfaceAndMetadata(cx, true), 5.0)); + state std::vector servers; + servers.reserve(storageMetadatas.size()); + for (const auto& meta : storageMetadatas) { + servers.push_back(StorageServerStatusInfo(meta)); + } state std::vector> results; wait(store(results, getServerMetrics(servers, address_workers, STORAGE_SERVER_METRICS_LIST))); for (int i = 0; i < results.size(); ++i) { @@ -3062,6 +3025,7 @@ ACTOR Future clusterGetStatus( Database cx, std::vector workers, std::vector workerIssues, + std::vector storageMetadatas, std::map>* clientStatus, ServerCoordinators coordinators, std::vector incompatibleConnections, @@ -3326,7 +3290,7 @@ ACTOR Future clusterGetStatus( } state Future>> storageServerFuture = - errorOr(getStorageServerStatusInfos(cx, address_workers, rkWorker)); + errorOr(getStorageServerStatusInfos(storageMetadatas, address_workers, rkWorker)); state Future>>> tLogFuture = errorOr(getTLogsAndMetrics(db, address_workers)); state Future>>> commitProxyFuture = diff --git a/fdbserver/include/fdbserver/ClusterController.actor.h b/fdbserver/include/fdbserver/ClusterController.actor.h index 727e9d1892..4d4ba18429 100644 --- a/fdbserver/include/fdbserver/ClusterController.actor.h +++ b/fdbserver/include/fdbserver/ClusterController.actor.h @@ -18,24 +18,23 @@ * limitations under the License. */ -#pragma once - // When actually compiled (NO_INTELLISENSE), include the generated version of this file. In intellisense use the source // version. -#include "fdbclient/StorageServerInterface.h" -#include "fdbserver/BlobMigratorInterface.h" -#include - #if defined(NO_INTELLISENSE) && !defined(FDBSERVER_CLUSTERCONTROLLER_ACTOR_G_H) #define FDBSERVER_CLUSTERCONTROLLER_ACTOR_G_H #include "fdbserver/ClusterController.actor.g.h" #elif !defined(FDBSERVER_CLUSTERCONTROLLER_ACTOR_H) #define FDBSERVER_CLUSTERCONTROLLER_ACTOR_H +#pragma once + +#include #include "fdbclient/DatabaseContext.h" #include "fdbclient/MetaclusterRegistration.h" +#include "fdbclient/StorageServerInterface.h" #include "fdbrpc/Replication.h" #include "fdbrpc/ReplicationUtils.h" +#include "fdbserver/BlobMigratorInterface.h" #include "fdbserver/Knobs.h" #include "fdbserver/WorkerInterface.actor.h" #include "flow/SystemMonitor.h" @@ -3339,6 +3338,7 @@ public: AsyncTrigger updateDBInfo; std::set updateDBInfoEndpoints; std::set removedDBInfoEndpoints; + std::vector storageStatusInfos; DBInfo db; Database cx; diff --git a/fdbserver/include/fdbserver/Status.actor.h b/fdbserver/include/fdbserver/Status.actor.h index 1d3b037937..4685c559c8 100644 --- a/fdbserver/include/fdbserver/Status.actor.h +++ b/fdbserver/include/fdbserver/Status.actor.h @@ -48,6 +48,7 @@ Future clusterGetStatus( Database const& cx, std::vector const& workers, std::vector const& workerIssues, + std::vector const& storageMetadatas, std::map>* const& clientStatus, ServerCoordinators const& coordinators, std::vector const& incompatibleConnections, diff --git a/fdbserver/worker.actor.cpp b/fdbserver/worker.actor.cpp index 77b4dfeb04..964d8bc85f 100644 --- a/fdbserver/worker.actor.cpp +++ b/fdbserver/worker.actor.cpp @@ -3801,7 +3801,7 @@ TEST_CASE("/fdbserver/worker/swversion/runNewer") { namespace { KeyValueStoreType randomStoreType() { int type = deterministicRandom()->randomInt(0, (int)KeyValueStoreType::END); - if (type == KeyValueStoreType::NONE) { + if (type == KeyValueStoreType::NONE || type == KeyValueStoreType::SSD_REDWOOD_V1) { type = KeyValueStoreType::SSD_BTREE_V2; } #ifndef WITH_ROCKSDB