wbtiger
/
foundationdb
mirror of https://github.com/apple/foundationdb.git
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity
foundationdb/fdbserver/workloads/ConsistencyCheck.cpp

1154 lines
48 KiB
C++
Raw Permalink Blame History

/*
* ConsistencyCheck.cpp
*
* This source file is part of the FoundationDB open source project
*
* Copyright 2013-2026 Apple Inc. and the FoundationDB project authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#include <math.h>
#include "boost/lexical_cast.hpp"
#include "flow/IRandom.h"
#include "flow/ProcessEvents.h"
#include "fdbclient/NativeAPI.actor.h"
#include "fdbclient/FDBTypes.h"
#include "fdbserver/core/TesterInterface.h"
#include "fdbserver/tester/workloads.h"
#include "flow/IRateControl.h"
#include "fdbrpc/simulator.h"
#include "fdbserver/core/FDBSimulatorProcessInfo.h"
#include "fdbserver/core/Knobs.h"
#include "fdbserver/core/ProcessClassRecruitment.h"
#include "fdbserver/core/FDBSimulationPolicy.h"
#include "fdbserver/consistencyscan/ConsistencyScan.h"
#include "fdbserver/core/StorageMetrics.h"
#include "fdbserver/core/QuietDatabase.h"
#include "fdbserver/core/TSSMappingUtil.h"
#include "flow/DeterministicRandom.h"
#include "fdbclient/ManagementAPI.h"
#include "fdbclient/StorageServerInterface.h"
#include "flow/network.h"
#include "fdbrpc/SimulatorProcessInfo.h"
#include "flow/CoroUtils.h"
// #define SevCCheckInfo SevVerbose
#define SevCCheckInfo SevInfo
struct ConsistencyCheckWorkload : TestWorkload {
struct OnTimeout {
ConsistencyCheckWorkload& self;
explicit OnTimeout(ConsistencyCheckWorkload& self) : self(self) {}
void operator()(StringRef name, std::any const& msg, Error const& e) {
TraceEvent(SevError, "ConsistencyCheckFailure")
.error(e)
.detail("EventName", name)
.detail("EventMessage", std::any_cast<StringRef>(msg))
.log();
}
};
static constexpr auto NAME = "ConsistencyCheck";
// Whether or not we should perform checks that will only pass if the database is in a quiescent state
bool performQuiescentChecks;
// Whether or not to perform consistency check between storage servers and pair TSS
bool performTSSCheck;
// Maximum time Data Distributor can run before being considered stuck (for quiescent checks)
double maxDDRunTime;
// If true, then perform all checks on this client. The first client is the only one to perform all of the fast
// checks All other clients will perform slow checks if this test is distributed
bool firstClient;
// If true, then the expensive checks will be distributed to multiple clients
bool distributed;
// Determines how many shards are checked for consistency: out of every <shardSampleFactor> shards, 1 will be
// checked
int shardSampleFactor;
// The previous data distribution mode
int oldDataDistributionMode;
// If true, then any failure of the consistency check will be logged as SevError. Otherwise, it will be logged as
// SevWarn
bool failureIsError;
// Max number of bytes per second to read from each storage server
int rateLimitMax;
// DataSet Size
int64_t bytesReadInPreviousRound;
// Randomize shard order with each iteration if true
bool shuffleShards;
bool success;
// Number of times this client has run its portion of the consistency check
int64_t repetitions;
// Whether to continuously perform the consistency check
bool indefinite;
// Whether to suspendConsistencyCheck
AsyncVar<bool> suspendConsistencyCheck;
Future<Void> monitorConsistencyCheckSettingsActor;
OnTimeout onTimeout;
ProcessEvents::Event onTimeoutEvent;
explicit ConsistencyCheckWorkload(WorkloadContext const& wcx)
: TestWorkload(wcx), onTimeout(*this), onTimeoutEvent({ "Timeout"_sr, "TracedTooManyLines"_sr }, onTimeout) {
performQuiescentChecks = getOption(options, "performQuiescentChecks"_sr, false);
performTSSCheck = getOption(options, "performTSSCheck"_sr, true);
maxDDRunTime = getOption(options, "maxDDRunTime"_sr, 600.0);
distributed = getOption(options, "distributed"_sr, true);
shardSampleFactor = std::max(getOption(options, "shardSampleFactor"_sr, 1), 1);
failureIsError = getOption(options, "failureIsError"_sr, false);
rateLimitMax = getOption(options, "rateLimitMax"_sr, 0);
shuffleShards = getOption(options, "shuffleShards"_sr, false);
indefinite = getOption(options, "indefinite"_sr, false);
suspendConsistencyCheck.set(true);
success = true;
firstClient = clientId == 0;
repetitions = 0;
bytesReadInPreviousRound = 0;
}
Future<Void> setup(Database const& cx) override { return _setup(cx, this); }
Future<Void> _setup(Database cx, ConsistencyCheckWorkload* self) {
// If performing quiescent checks, wait for the database to go quiet
if (self->firstClient && self->performQuiescentChecks) {
if (g_network->isSimulated()) {
co_await timeKeeperSetDisable(cx);
}
try {
co_await timeoutError(
quietDatabase(
cx, self->dbInfo, "ConsistencyCheckStart", 0, 1e5, 0, 0, 30e6, 1e6, self->maxDDRunTime),
self->maxDDRunTime); // FIXME: should be zero?
if (g_network->isSimulated()) {
fdbSimulationPolicyState().quiesced = true;
TraceEvent("ConsistencyCheckQuiesced").detail("Quiesced", fdbSimulationPolicyState().quiesced);
}
} catch (Error& e) {
TraceEvent("ConsistencyCheck_QuietDatabaseError").error(e);
self->testFailure("Unable to achieve a quiet database");
self->performQuiescentChecks = false;
}
}
self->monitorConsistencyCheckSettingsActor = self->monitorConsistencyCheckSettings(cx, self);
}
Future<Void> start(Database const& cx) override {
TraceEvent("ConsistencyCheck").log();
return _start(cx, this);
}
Future<bool> check(Database const& cx) override { return success; }
void getMetrics(std::vector<PerfMetric>& m) override {}
void testFailure(std::string message, bool isError = false) {
success = false;
TraceEvent failEvent((failureIsError || isError) ? SevError : SevWarn, "TestFailure");
if (performQuiescentChecks)
failEvent.detail("Workload", "QuiescentCheck");
else
failEvent.detail("Workload", "ConsistencyCheck");
failEvent.detail("Reason", "Consistency check: " + message);
}
Future<Void> monitorConsistencyCheckSettings(Database cx, ConsistencyCheckWorkload* self) {
while (true) {
ReadYourWritesTransaction tr(cx);
{
Error err;
try {
tr.setOption(FDBTransactionOptions::ACCESS_SYSTEM_KEYS);
tr.setOption(FDBTransactionOptions::PRIORITY_SYSTEM_IMMEDIATE);
tr.setOption(FDBTransactionOptions::LOCK_AWARE);
Optional<Value> ccSuspendVal = co_await tr.get(fdbShouldConsistencyCheckBeSuspended);
bool ccSuspend = ccSuspendVal.present()
? BinaryReader::fromStringRef<bool>(ccSuspendVal.get(), Unversioned())
: false;
self->suspendConsistencyCheck.set(ccSuspend);
Future<Void> watchCCSuspendFuture = tr.watch(fdbShouldConsistencyCheckBeSuspended);
co_await tr.commit();
co_await watchCCSuspendFuture;
} catch (Error& e) {
err = e;
}
co_await tr.onError(err);
}
}
}
Future<Void> _start(Database cx, ConsistencyCheckWorkload* self) {
while (true) {
while (self->suspendConsistencyCheck.get()) {
TraceEvent("ConsistencyCheck_Suspended").log();
co_await self->suspendConsistencyCheck.onChange();
}
TraceEvent("ConsistencyCheck_StartingOrResuming").log();
auto choice = co_await race(self->runCheck(cx, self), self->suspendConsistencyCheck.onChange());
if (choice.index() == 0) {
if (!self->indefinite)
break;
self->repetitions++;
co_await delay(5.0);
} else if (choice.index() == 1) {
} else {
UNREACHABLE();
}
}
if (self->firstClient && g_network->isSimulated() && self->performQuiescentChecks) {
fdbSimulationPolicyState().quiesced = false;
TraceEvent("ConsistencyCheckQuiescedEnd").detail("Quiesced", fdbSimulationPolicyState().quiesced);
}
}
Future<Void> runCheck(Database cx, ConsistencyCheckWorkload* self) {
CODE_PROBE(self->performQuiescentChecks, "Quiescent consistency check");
CODE_PROBE(!self->performQuiescentChecks, "Non-quiescent consistency check");
double consistenyCheckerBeginTime = now();
if (self->firstClient || self->distributed) {
try {
DatabaseConfiguration configuration;
std::map<UID, StorageServerInterface> tssMapping;
Transaction tr(cx);
tr.setOption(FDBTransactionOptions::LOCK_AWARE);
while (true) {
Error err;
try {
if (self->performTSSCheck) {
tssMapping.clear();
co_await readTSSMapping(&tr, &tssMapping);
}
RangeResult res = co_await tr.getRange(configKeys, 1000);
if (res.size() == 1000) {
TraceEvent("ConsistencyCheck_TooManyConfigOptions").log();
self->testFailure("Read too many configuration options");
}
for (int i = 0; i < res.size(); i++)
configuration.set(res[i].key, res[i].value);
break;
} catch (Error& e) {
err = e;
}
co_await tr.onError(err);
}
// Perform quiescence-only checks
if (self->firstClient && self->performQuiescentChecks) {
// Check for undesirable servers (storage servers with exact same network address or using the wrong
// key value store type)
bool hasUndesirableServers = co_await self->checkForUndesirableServers(cx, configuration, self);
// Check that nothing is in-flight or in queue in data distribution
int64_t inDataDistributionQueue = co_await getDataDistributionQueueSize(cx, self->dbInfo, true);
if (inDataDistributionQueue > 0) {
TraceEvent("ConsistencyCheck_NonZeroDataDistributionQueue")
.detail("QueueSize", inDataDistributionQueue);
self->testFailure("Non-zero data distribution queue/in-flight size");
}
// Check that the number of process (and machine) teams is no larger than
// the allowed maximum number of teams
bool teamCollectionValid = co_await getTeamCollectionValid(cx, self->dbInfo);
if (!teamCollectionValid) {
TraceEvent(SevError, "ConsistencyCheck_TooManyTeams").log();
self->testFailure("The number of process or machine teams is larger than the allowed maximum "
"number of teams");
}
// Check that nothing is in the TLog queues
std::pair<int64_t, int64_t> maxTLogQueueInfo = co_await getTLogQueueInfo(cx, self->dbInfo);
if (maxTLogQueueInfo.first > 1e5) // FIXME: Should be zero?
{
TraceEvent("ConsistencyCheck_NonZeroTLogQueue").detail("MaxQueueSize", maxTLogQueueInfo.first);
self->testFailure("Non-zero tlog queue size");
}
if (maxTLogQueueInfo.second > 30e6) {
TraceEvent("ConsistencyCheck_PoppedVersionLag")
.detail("PoppedVersionLag", maxTLogQueueInfo.second);
self->testFailure("large popped version lag");
}
// Check that nothing is in the storage server queues
try {
int64_t maxStorageServerQueueSize =
co_await getMaxStorageServerQueueSize(cx, self->dbInfo, invalidVersion);
if (maxStorageServerQueueSize > 0) {
TraceEvent("ConsistencyCheck_ExceedStorageServerQueueLimit")
.detail("MaxQueueSize", maxStorageServerQueueSize);
self->testFailure("Storage server queue size exceeds limit");
}
} catch (Error& e) {
if (e.code() == error_code_attribute_not_found) {
TraceEvent("ConsistencyCheck_StorageQueueSizeError")
.error(e)
.detail("Reason", "Could not read queue size");
// This error occurs if we have undesirable servers; in that case just report the
// undesirable servers error
if (!hasUndesirableServers)
self->testFailure("Could not read storage queue size");
} else {
throw;
}
}
co_await self->checkForStorage(cx, configuration, tssMapping, self);
co_await self->checkForExtraDataStores(cx, self);
co_await self->checkStorageMetadata(cx, self);
// Check that each machine is operating as its desired class
bool usingDesiredClasses = co_await self->checkUsingDesiredClasses(cx, self);
if (!usingDesiredClasses)
self->testFailure("Cluster has machine(s) not using requested classes");
bool workerListCorrect = co_await self->checkWorkerList(cx, self);
if (!workerListCorrect)
self->testFailure("Worker list incorrect");
bool coordinatorsCorrect = co_await self->checkCoordinators(cx);
if (!coordinatorsCorrect)
self->testFailure("Coordinators incorrect");
bool consistencyScanStopped = co_await self->checkConsistencyScan(cx);
if (!consistencyScanStopped)
self->testFailure("Consistency scan active");
// FIXME: re-enable this check!
// bool singleSingletons = self->checkSingleSingletons(self, configuration);
// if (!singleSingletons)
// self->testFailure("Cluster has multiple instances of a singleton!");
}
// Get a list of key servers; verify that the TLogs and master all agree about who the key servers are
Promise<std::vector<std::pair<KeyRange, std::vector<StorageServerInterface>>>> keyServerPromise;
bool keyServerResult = co_await getKeyServers(cx,
keyServerPromise,
keyServersKeys,
self->performQuiescentChecks,
self->failureIsError,
&self->success);
if (keyServerResult) {
std::vector<std::pair<KeyRange, std::vector<StorageServerInterface>>> keyServers =
keyServerPromise.getFuture().get();
// Get the locations of all the shards in the database
Promise<Standalone<VectorRef<KeyValueRef>>> keyLocationPromise;
bool keyLocationResult = co_await getKeyLocations(
cx, keyServers, keyLocationPromise, self->performQuiescentChecks, &self->success);
if (keyLocationResult) {
Standalone<VectorRef<KeyValueRef>> keyLocations = keyLocationPromise.getFuture().get();
// Check that each shard has the same data on all storage servers that it resides on
co_await checkDataConsistency(cx,
keyLocations,
configuration,
tssMapping,
self->performQuiescentChecks,
self->performTSSCheck,
self->firstClient,
self->failureIsError,
self->clientId,
self->clientCount,
self->distributed,
self->shuffleShards,
self->shardSampleFactor,
self->sharedRandomNumber,
self->repetitions,
&(self->bytesReadInPreviousRound),
true,
self->rateLimitMax,
CLIENT_KNOBS->CONSISTENCY_CHECK_ONE_ROUND_TARGET_COMPLETION_TIME,
&self->success);
}
}
} catch (Error& e) {
if (e.code() == error_code_transaction_too_old || e.code() == error_code_future_version ||
e.code() == error_code_wrong_shard_server || e.code() == error_code_all_alternatives_failed ||
e.code() == error_code_process_behind || e.code() == error_code_actor_cancelled) {
TraceEvent("ConsistencyCheck_Retry")
.error(e); // FIXME: consistency check does not retry in this case
} else {
self->testFailure(format("Error %d - %s", e.code(), e.name()));
}
}
}
TraceEvent("ConsistencyCheck_FinishedCheck")
.detail("Repetitions", self->repetitions)
.detail("TimeSpan", now() - consistenyCheckerBeginTime);
}
// Comparison function used to compare map elements by value
template <class K, class T>
static bool compareByValue(std::pair<K, T> a, std::pair<K, T> b) {
return a.second < b.second;
}
// Returns true if any storage servers have the exact same network address or are not using the correct key value
// store type
Future<bool> checkForUndesirableServers(Database cx,
DatabaseConfiguration configuration,
ConsistencyCheckWorkload* self) {
int i{ 0 };
int j{ 0 };
std::vector<StorageServerInterface> storageServers = co_await getStorageServers(cx);
std::string wiggleLocalityKeyValue = configuration.perpetualStorageWiggleLocality;
std::vector<std::pair<Optional<Value>, Optional<Value>>> wiggleLocalityKeyValues =
ParsePerpetualStorageWiggleLocality(configuration.perpetualStorageWiggleLocality);
// Check each pair of storage servers for an address match
for (i = 0; i < storageServers.size(); i++) {
// Check that each storage server has the correct key value store type
ReplyPromise<KeyValueStoreType> typeReply;
ErrorOr<KeyValueStoreType> keyValueStoreType =
co_await storageServers[i].getKeyValueStoreType.getReplyUnlessFailedFor(typeReply, 2, 0);
if (!keyValueStoreType.present()) {
TraceEvent("ConsistencyCheck_ServerUnavailable").detail("ServerID", storageServers[i].id());
self->testFailure("Storage server unavailable");
} else if (configuration.perpetualStoreType.isValid()) {
// Perpetual storage wiggle is used to migrate storage. Check that the matched storage servers are
// correctly migrated.
if (wiggleLocalityKeyValue == "0" ||
localityMatchInList(wiggleLocalityKeyValues, storageServers[i].locality)) {
if (keyValueStoreType.get() != configuration.perpetualStoreType) {
TraceEvent("ConsistencyCheck_WrongKeyValueStoreType")
.detail("ServerID", storageServers[i].id())
.detail("StoreType", keyValueStoreType.get().toString())
.detail("DesiredType", configuration.perpetualStoreType.toString())
.detail("IsPerpetualStoreType", true);
self->testFailure("Storage server has wrong key-value store type");
co_return true;
}
} else if ((!storageServers[i].isTss() &&
keyValueStoreType.get() != configuration.storageServerStoreType) ||
(storageServers[i].isTss() &&
keyValueStoreType.get() != configuration.testingStorageServerStoreType)) {
TraceEvent("ConsistencyCheck_WrongKeyValueStoreType")
.detail("ServerID", storageServers[i].id())
.detail("StoreType", keyValueStoreType.get().toString())
.detail("DesiredType", configuration.perpetualStoreType.toString())
.detail("IsPerpetualStoreType", false);
self->testFailure("Storage server has wrong key-value store type");
co_return true;
}
} else if (((!storageServers[i].isTss() &&
keyValueStoreType.get() != configuration.storageServerStoreType) ||
(storageServers[i].isTss() &&
keyValueStoreType.get() != configuration.testingStorageServerStoreType)) &&
(wiggleLocalityKeyValue == "0" ||
localityMatchInList(wiggleLocalityKeyValues, storageServers[i].locality))) {
TraceEvent("ConsistencyCheck_WrongKeyValueStoreType")
.detail("ServerID", storageServers[i].id())
.detail("StoreType", keyValueStoreType.get().toString())
.detail("DesiredType", configuration.storageServerStoreType.toString())
.detail("IsPerpetualStoreType", false);
self->testFailure("Storage server has wrong key-value store type");
co_return true;
}
// Check each pair of storage servers for an address match
for (j = i + 1; j < storageServers.size(); j++) {
if (storageServers[i].address() == storageServers[j].address()) {
TraceEvent("ConsistencyCheck_UndesirableServer")
.detail("StorageServer1", storageServers[i].id())
.detail("StorageServer2", storageServers[j].id())
.detail("Address", storageServers[i].address());
self->testFailure("Multiple storage servers have the same address");
co_return true;
}
}
}
co_return false;
}
// Every storage server should have it metadata populated and no metadata leak when the database reach the quiescent
// state
Future<bool> checkStorageMetadata(Database cx, ConsistencyCheckWorkload* self) {
KeyBackedObjectMap<UID, StorageMetadataType, decltype(IncludeVersion())> metadataMap(serverMetadataKeys.begin,
IncludeVersion());
std::vector<StorageServerInterface> servers;
std::unordered_map<UID, StorageMetadataType> id_ssi;
Transaction tr(cx);
while (true) {
servers.clear();
id_ssi.clear();
tr.setOption(FDBTransactionOptions::READ_SYSTEM_KEYS);
tr.setOption(FDBTransactionOptions::LOCK_AWARE);
{
Error err;
try {
KeyBackedRangeResult<std::pair<UID, StorageMetadataType>> metadata =
co_await metadataMap.getRange(&tr, {}, {}, CLIENT_KNOBS->TOO_MANY);
ASSERT(!metadata.more && metadata.results.size() < CLIENT_KNOBS->TOO_MANY);
RangeResult serverList = co_await tr.getRange(serverListKeys, CLIENT_KNOBS->TOO_MANY);
ASSERT(!serverList.more && serverList.size() < CLIENT_KNOBS->TOO_MANY);
ASSERT_EQ(metadata.results.size(), serverList.size());
id_ssi =
std::unordered_map<UID, StorageMetadataType>(metadata.results.begin(), metadata.results.end());
servers.reserve(serverList.size());
for (int i = 0; i < serverList.size(); i++)
servers.push_back(decodeServerListValue(serverList[i].value));
break;
} catch (Error& e) {
err = e;
}
co_await tr.onError(err);
}
}
for (auto& ssi : servers) {
ASSERT(id_ssi.contains(ssi.id()));
}
co_return true;
}
// Returns false if any worker that should have a storage server does not have one
Future<bool> checkForStorage(Database cx,
DatabaseConfiguration configuration,
std::map<UID, StorageServerInterface> tssMapping,
ConsistencyCheckWorkload* self) {
std::vector<WorkerDetails> workers = co_await getWorkers(self->dbInfo);
std::vector<StorageServerInterface> storageServers = co_await getStorageServers(cx);
std::vector<Optional<Key>> missingStorage; // vector instead of a set to get the count
for (int i = 0; i < workers.size(); i++) {
NetworkAddress addr = workers[i].interf.stableAddress();
if (!configuration.isExcludedServer(workers[i].interf.addresses(), workers[i].interf.locality) &&
(workers[i].processClass == ProcessClass::StorageClass ||
workers[i].processClass == ProcessClass::UnsetClass)) {
bool found = false;
for (int j = 0; j < storageServers.size(); j++) {
if (storageServers[j].stableAddress() == addr) {
found = true;
break;
}
}
if (!found) {
TraceEvent("ConsistencyCheck_NoStorage")
.detail("Address", addr)
.detail("ProcessId", workers[i].interf.locality.processId())
.detail("ProcessClassEqualToStorageClass",
(int)(workers[i].processClass == ProcessClass::StorageClass));
missingStorage.push_back(workers[i].interf.locality.dcId());
}
}
}
int missingDc0 = configuration.regions.empty()
? 0
: std::count(missingStorage.begin(), missingStorage.end(), configuration.regions[0].dcId);
int missingDc1 = configuration.regions.size() < 2
? 0
: std::count(missingStorage.begin(), missingStorage.end(), configuration.regions[1].dcId);
if ((configuration.regions.empty() && !missingStorage.empty()) ||
(configuration.regions.size() == 1 && missingDc0) ||
(configuration.regions.size() == 2 && configuration.usableRegions == 1 && missingDc0 && missingDc1) ||
(configuration.regions.size() == 2 && configuration.usableRegions > 1 && (missingDc0 || missingDc1))) {
// TODO could improve this check by also ensuring DD is currently recruiting a TSS by using quietdb?
bool couldExpectMissingTss = (configuration.desiredTSSCount - tssMapping.size()) > 0;
int countMissing = missingStorage.size();
int acceptableTssMissing = 1;
if (configuration.regions.size() == 1) {
countMissing = missingDc0;
} else if (configuration.regions.size() == 2) {
if (configuration.usableRegions == 1) {
// all processes should be missing from 1, so take the number missing from the other
countMissing = std::min(missingDc0, missingDc1);
} else if (configuration.usableRegions == 2) {
countMissing = missingDc0 + missingDc1;
acceptableTssMissing = 2;
} else {
ASSERT(false); // in case fdb ever adds 3+ region support?
}
}
if (!couldExpectMissingTss || countMissing > acceptableTssMissing) {
self->testFailure("No storage server on worker");
co_return false;
} else {
TraceEvent(SevWarn, "ConsistencyCheck_TSSMissing").log();
}
}
co_return true;
}
Future<bool> checkForExtraDataStores(Database cx, ConsistencyCheckWorkload* self) {
std::vector<WorkerDetails> workers = co_await getWorkers(self->dbInfo);
std::vector<StorageServerInterface> storageServers = co_await getStorageServers(cx);
std::vector<WorkerInterface> coordWorkers = co_await getCoordWorkers(cx, self->dbInfo);
auto& db = self->dbInfo->get();
std::vector<TLogInterface> logs = db.logSystemConfig.allPresentLogs();
std::vector<WorkerDetails>::iterator itr;
bool foundExtraDataStore = false;
std::vector<struct ProcessInfo*> protectedProcessesToKill;
std::map<NetworkAddress, std::set<UID>> statefulProcesses;
for (const auto& ss : storageServers) {
statefulProcesses[ss.address()].insert(ss.id());
// A process may have two addresses (same ip, different ports)
if (ss.secondaryAddress().present()) {
statefulProcesses[ss.secondaryAddress().get()].insert(ss.id());
}
TraceEvent(SevCCheckInfo, "StatefulProcess")
.detail("StorageServer", ss.id())
.detail("PrimaryAddress", ss.address().toString())
.detail("SecondaryAddress",
ss.secondaryAddress().present() ? ss.secondaryAddress().get().toString() : "Unset");
}
for (const auto& log : logs) {
statefulProcesses[log.address()].insert(log.id());
if (log.secondaryAddress().present()) {
statefulProcesses[log.secondaryAddress().get()].insert(log.id());
}
TraceEvent(SevCCheckInfo, "StatefulProcess")
.detail("Log", log.id())
.detail("PrimaryAddress", log.address().toString())
.detail("SecondaryAddress",
log.secondaryAddress().present() ? log.secondaryAddress().get().toString() : "Unset");
}
// Coordinators are also stateful processes
for (const auto& cWorker : coordWorkers) {
statefulProcesses[cWorker.address()].insert(cWorker.id());
if (cWorker.secondaryAddress().present()) {
statefulProcesses[cWorker.secondaryAddress().get()].insert(cWorker.id());
}
TraceEvent(SevCCheckInfo, "StatefulProcess")
.detail("Coordinator", cWorker.id())
.detail("PrimaryAddress", cWorker.address().toString())
.detail("SecondaryAddress",
cWorker.secondaryAddress().present() ? cWorker.secondaryAddress().get().toString() : "Unset");
}
for (itr = workers.begin(); itr != workers.end(); ++itr) {
ErrorOr<Standalone<VectorRef<UID>>> stores =
co_await itr->interf.diskStoreRequest.getReplyUnlessFailedFor(DiskStoreRequest(false), 2, 0);
if (stores.isError()) {
TraceEvent("ConsistencyCheck_GetDataStoreFailure")
.error(stores.getError())
.detail("Address", itr->interf.address());
self->testFailure("Failed to get data stores");
co_return false;
}
TraceEvent(SevCCheckInfo, "ConsistencyCheck_ExtraDataStore")
.detail("Worker", itr->interf.id().toString())
.detail("PrimaryAddress", itr->interf.address().toString())
.detail("SecondaryAddress",
itr->interf.secondaryAddress().present() ? itr->interf.secondaryAddress().get().toString()
: "Unset");
for (const auto& id : stores.get()) {
if (statefulProcesses[itr->interf.address()].contains(id)) {
continue;
}
// For extra data store
TraceEvent("ConsistencyCheck_ExtraDataStore")
.detail("Address", itr->interf.address())
.detail("DataStoreID", id);
if (g_network->isSimulated()) {
// FIXME: this is hiding the fact that we can recruit a new storage server on a location the has
// files left behind by a previous failure
// this means that the process is wasting disk space until the process is rebooting
ISimulator::ProcessInfo* p = g_simulator->getProcessByAddress(itr->interf.address());
// Note: itr->interf.address() may not equal to p->address() because role's endpoint's primary
// addr can be swapped by choosePrimaryAddress() based on its peer's tls config.
TraceEvent("ConsistencyCheck_RebootProcess")
.detail("Address",
itr->interf.address()) // worker's primary address (i.e., the first address)
.detail("ProcessPrimaryAddress", p->address)
.detail("ProcessAddresses", p->addresses.toString())
.detail("DataStoreID", id)
.detail("Protected", g_simulator->isProtectedAddress(itr->interf.address()))
.detail("Reliable", p->isReliable())
.detail("ReliableInfo", p->getReliableInfo())
.detail("KillOrRebootProcess", p->address);
if (p->isReliable()) {
g_simulator->rebootProcess(p, ISimulator::KillType::RebootProcess);
} else {
g_simulator->killProcess(p, ISimulator::KillType::KillInstantly);
}
}
foundExtraDataStore = true;
}
}
if (foundExtraDataStore) {
self->testFailure("Extra data stores present on workers");
co_return false;
}
co_return true;
}
Future<bool> checkWorkerList(Database cx, ConsistencyCheckWorkload* self) {
if (!fdbSimulationPolicyState().extraDatabases.empty()) {
co_return true;
}
std::vector<WorkerDetails> workers = co_await getWorkers(self->dbInfo);
std::set<NetworkAddress> workerAddresses;
for (const auto& it : workers) {
NetworkAddress addr = it.interf.tLog.getEndpoint().addresses.getTLSAddress();
ISimulator::ProcessInfo* info = g_simulator->getProcessByAddress(addr);
if (!info || info->failed) {
TraceEvent("ConsistencyCheck_FailedWorkerInList").detail("Addr", it.interf.address());
co_return false;
}
workerAddresses.insert(NetworkAddress(addr.ip, addr.port, true, addr.isTLS()));
}
std::vector<ISimulator::ProcessInfo*> all = g_simulator->getAllProcesses();
for (int i = 0; i < all.size(); i++) {
if (all[i]->isReliable() && all[i]->name == std::string("Server") &&
getSimulatorProcessClass(all[i]) != ProcessClass::TesterClass &&
getSimulatorProcessClass(all[i]) != ProcessClass::SimHTTPServerClass &&
all[i]->protocolVersion == g_network->protocolVersion()) {
if (!workerAddresses.contains(all[i]->address)) {
TraceEvent("ConsistencyCheck_WorkerMissingFromList").detail("Addr", all[i]->address);
co_return false;
}
}
}
co_return true;
}
static recruitment::Fitness getBestAvailableFitness(const std::vector<ProcessClass::ClassType>& availableClassTypes,
recruitment::ClusterRole role) {
recruitment::Fitness bestAvailableFitness = recruitment::NeverAssign;
for (auto classType : availableClassTypes) {
bestAvailableFitness =
std::min(bestAvailableFitness,
recruitment::machineClassFitness(ProcessClass(classType, ProcessClass::InvalidSource), role));
}
return bestAvailableFitness;
}
template <class T>
static std::string getOptionalString(Optional<T> opt) {
if (opt.present())
return opt.get().toString();
return "NotSet";
}
Future<bool> checkCoordinators(Database cx) {
Transaction tr(cx);
while (true) {
Error err;
try {
tr.setOption(FDBTransactionOptions::LOCK_AWARE);
Optional<Value> currentKey = co_await tr.get(coordinatorsKey);
if (!currentKey.present()) {
TraceEvent("ConsistencyCheck_NoCoordinatorKey").log();
co_return false;
}
ClusterConnectionString old(currentKey.get().toString());
std::vector<NetworkAddress> oldCoordinators = co_await old.tryResolveHostnames();
std::vector<ProcessData> workers = co_await ::getWorkers(&tr);
std::map<NetworkAddress, LocalityData> addr_locality;
for (const auto& w : workers) {
addr_locality[w.address] = w.locality;
}
std::set<Optional<Standalone<StringRef>>> checkDuplicates;
for (const auto& addr : oldCoordinators) {
auto findResult = addr_locality.find(addr);
if (findResult != addr_locality.end()) {
if (checkDuplicates.contains(findResult->second.zoneId())) {
TraceEvent("ConsistencyCheck_BadCoordinator")
.detail("Addr", addr)
.detail("NotFound", findResult == addr_locality.end());
co_return false;
}
checkDuplicates.insert(findResult->second.zoneId());
}
}
co_return true;
} catch (Error& e) {
err = e;
}
co_await tr.onError(err);
}
}
// Returns true if all machines in the cluster that specified a desired class are operating in that class
Future<bool> checkUsingDesiredClasses(Database cx, ConsistencyCheckWorkload* self) {
Optional<Key> expectedPrimaryDcId;
Optional<Key> expectedRemoteDcId;
DatabaseConfiguration config = co_await getDatabaseConfiguration(cx);
std::vector<WorkerDetails> allWorkers = co_await getWorkers(self->dbInfo);
std::vector<WorkerDetails> nonExcludedWorkers =
co_await getWorkers(self->dbInfo, GetWorkersRequest::NON_EXCLUDED_PROCESSES_ONLY);
auto& db = self->dbInfo->get();
std::map<NetworkAddress, WorkerDetails> allWorkerProcessMap;
std::map<Optional<Key>, std::vector<ProcessClass::ClassType>> dcToAllClassTypes;
for (const auto& worker : allWorkers) {
allWorkerProcessMap[worker.interf.address()] = worker;
Optional<Key> dc = worker.interf.locality.dcId();
if (!dcToAllClassTypes.contains(dc))
dcToAllClassTypes.insert({});
dcToAllClassTypes[dc].push_back(worker.processClass.classType());
}
std::map<NetworkAddress, WorkerDetails> nonExcludedWorkerProcessMap;
std::map<Optional<Key>, std::vector<ProcessClass::ClassType>> dcToNonExcludedClassTypes;
for (const auto& worker : nonExcludedWorkers) {
nonExcludedWorkerProcessMap[worker.interf.address()] = worker;
Optional<Key> dc = worker.interf.locality.dcId();
if (!dcToNonExcludedClassTypes.contains(dc))
dcToNonExcludedClassTypes.insert({});
dcToNonExcludedClassTypes[dc].push_back(worker.processClass.classType());
}
if (!allWorkerProcessMap.contains(db.clusterInterface.clientInterface.address())) {
TraceEvent("ConsistencyCheck_CCNotInWorkerList")
.detail("CCAddress", db.clusterInterface.clientInterface.address().toString());
co_return false;
}
if (!allWorkerProcessMap.contains(db.master.address())) {
TraceEvent("ConsistencyCheck_MasterNotInWorkerList")
.detail("MasterAddress", db.master.address().toString());
co_return false;
}
Optional<Key> ccDcId =
allWorkerProcessMap[db.clusterInterface.clientInterface.address()].interf.locality.dcId();
Optional<Key> masterDcId = allWorkerProcessMap[db.master.address()].interf.locality.dcId();
if (ccDcId != masterDcId) {
TraceEvent("ConsistencyCheck_CCAndMasterNotInSameDC")
.detail("ClusterControllerDcId", getOptionalString(ccDcId))
.detail("MasterDcId", getOptionalString(masterDcId));
co_return false;
}
// Check if master and cluster controller are in the desired DC for fearless cluster when running under
// simulation
// FIXME: g_simulator->datacenterDead could return false positives. Relaxing checks until it is fixed.
if (g_network->isSimulated() && config.usableRegions > 1 && fdbSimulationPolicyState().primaryDcId.present() &&
!g_simulator->datacenterDead(fdbSimulationPolicyState().primaryDcId) &&
!g_simulator->datacenterDead(fdbSimulationPolicyState().remoteDcId)) {
expectedPrimaryDcId = config.regions[0].dcId;
expectedRemoteDcId = config.regions[1].dcId;
// If the priorities are equal, either could be the primary
if (config.regions[0].priority == config.regions[1].priority) {
expectedPrimaryDcId = masterDcId;
expectedRemoteDcId = config.regions[0].dcId == expectedPrimaryDcId.get() ? config.regions[1].dcId
: config.regions[0].dcId;
}
if (ccDcId != expectedPrimaryDcId) {
TraceEvent("ConsistencyCheck_ClusterControllerDcNotBest")
.detail("PreferredDcId", getOptionalString(expectedPrimaryDcId))
.detail("ExistingDcId", getOptionalString(ccDcId));
co_return false;
}
if (masterDcId != expectedPrimaryDcId) {
TraceEvent("ConsistencyCheck_MasterDcNotBest")
.detail("PreferredDcId", getOptionalString(expectedPrimaryDcId))
.detail("ExistingDcId", getOptionalString(masterDcId));
co_return false;
}
}
// Check CC
recruitment::Fitness bestClusterControllerFitness =
getBestAvailableFitness(dcToNonExcludedClassTypes[ccDcId], recruitment::ClusterController);
if (!nonExcludedWorkerProcessMap.contains(db.clusterInterface.clientInterface.address()) ||
recruitment::machineClassFitness(
nonExcludedWorkerProcessMap[db.clusterInterface.clientInterface.address()].processClass,
recruitment::ClusterController) != bestClusterControllerFitness) {
TraceEvent("ConsistencyCheck_ClusterControllerNotBest")
.detail("BestClusterControllerFitness", bestClusterControllerFitness)
.detail(
"ExistingClusterControllerFit",
nonExcludedWorkerProcessMap.contains(db.clusterInterface.clientInterface.address())
? recruitment::machineClassFitness(
nonExcludedWorkerProcessMap[db.clusterInterface.clientInterface.address()].processClass,
recruitment::ClusterController)
: -1);
co_return false;
}
// Check Master
recruitment::Fitness bestMasterFitness =
getBestAvailableFitness(dcToNonExcludedClassTypes[masterDcId], recruitment::Master);
if (bestMasterFitness == recruitment::NeverAssign) {
bestMasterFitness = getBestAvailableFitness(dcToAllClassTypes[masterDcId], recruitment::Master);
if (bestMasterFitness != recruitment::NeverAssign) {
bestMasterFitness = recruitment::ExcludeFit;
}
}
if ((!nonExcludedWorkerProcessMap.contains(db.master.address()) &&
bestMasterFitness != recruitment::ExcludeFit) ||
recruitment::machineClassFitness(nonExcludedWorkerProcessMap[db.master.address()].processClass,
recruitment::Master) != bestMasterFitness) {
TraceEvent("ConsistencyCheck_MasterNotBest")
.detail("BestMasterFitness", bestMasterFitness)
.detail("ExistingMasterFit",
nonExcludedWorkerProcessMap.contains(db.master.address())
? recruitment::machineClassFitness(
nonExcludedWorkerProcessMap[db.master.address()].processClass, recruitment::Master)
: -1);
co_return false;
}
// Check commit proxy
recruitment::Fitness bestCommitProxyFitness =
getBestAvailableFitness(dcToNonExcludedClassTypes[masterDcId], recruitment::CommitProxy);
for (const auto& commitProxy : db.client.commitProxies) {
if (!nonExcludedWorkerProcessMap.contains(commitProxy.address()) ||
recruitment::machineClassFitness(nonExcludedWorkerProcessMap[commitProxy.address()].processClass,
recruitment::CommitProxy) != bestCommitProxyFitness) {
TraceEvent("ConsistencyCheck_CommitProxyNotBest")
.detail("BestCommitProxyFitness", bestCommitProxyFitness)
.detail("ExistingCommitProxyFitness",
nonExcludedWorkerProcessMap.contains(commitProxy.address())
? recruitment::machineClassFitness(
nonExcludedWorkerProcessMap[commitProxy.address()].processClass,
recruitment::CommitProxy)
: -1);
co_return false;
}
}
// Check grv proxy
recruitment::Fitness bestGrvProxyFitness =
getBestAvailableFitness(dcToNonExcludedClassTypes[masterDcId], recruitment::GrvProxy);
for (const auto& grvProxy : db.client.grvProxies) {
if (!nonExcludedWorkerProcessMap.contains(grvProxy.address()) ||
recruitment::machineClassFitness(nonExcludedWorkerProcessMap[grvProxy.address()].processClass,
recruitment::GrvProxy) != bestGrvProxyFitness) {
TraceEvent("ConsistencyCheck_GrvProxyNotBest")
.detail("BestGrvProxyFitness", bestGrvProxyFitness)
.detail(
"ExistingGrvProxyFitness",
nonExcludedWorkerProcessMap.contains(grvProxy.address())
? recruitment::machineClassFitness(
nonExcludedWorkerProcessMap[grvProxy.address()].processClass, recruitment::GrvProxy)
: -1);
co_return false;
}
}
// Check resolver
recruitment::Fitness bestResolverFitness =
getBestAvailableFitness(dcToNonExcludedClassTypes[masterDcId], recruitment::Resolver);
for (const auto& resolver : db.resolvers) {
if (!nonExcludedWorkerProcessMap.contains(resolver.address()) ||
recruitment::machineClassFitness(nonExcludedWorkerProcessMap[resolver.address()].processClass,
recruitment::Resolver) != bestResolverFitness) {
TraceEvent("ConsistencyCheck_ResolverNotBest")
.detail("BestResolverFitness", bestResolverFitness)
.detail(
"ExistingResolverFitness",
nonExcludedWorkerProcessMap.contains(resolver.address())
? recruitment::machineClassFitness(
nonExcludedWorkerProcessMap[resolver.address()].processClass, recruitment::Resolver)
: -1);
co_return false;
}
}
// Check LogRouter
if (g_network->isSimulated() && config.usableRegions > 1 && fdbSimulationPolicyState().primaryDcId.present() &&
!g_simulator->datacenterDead(fdbSimulationPolicyState().primaryDcId) &&
!g_simulator->datacenterDead(fdbSimulationPolicyState().remoteDcId)) {
for (auto& tlogSet : db.logSystemConfig.tLogs) {
if (!tlogSet.isLocal && !tlogSet.logRouters.empty()) {
for (auto& logRouter : tlogSet.logRouters) {
if (!nonExcludedWorkerProcessMap.contains(logRouter.interf().address())) {
TraceEvent("ConsistencyCheck_LogRouterNotInNonExcludedWorkers")
.detail("Id", logRouter.id());
co_return false;
}
if (logRouter.interf().filteredLocality.dcId() != expectedRemoteDcId) {
TraceEvent("ConsistencyCheck_LogRouterNotBestDC")
.detail("expectedDC", getOptionalString(expectedRemoteDcId))
.detail("ActualDC", getOptionalString(logRouter.interf().filteredLocality.dcId()));
co_return false;
}
}
}
}
}
// Check DataDistributor
recruitment::Fitness fitnessLowerBound = recruitment::machineClassFitness(
allWorkerProcessMap[db.master.address()].processClass, recruitment::DataDistributor);
if (db.distributor.present() &&
(!nonExcludedWorkerProcessMap.contains(db.distributor.get().address()) ||
recruitment::machineClassFitness(nonExcludedWorkerProcessMap[db.distributor.get().address()].processClass,
recruitment::DataDistributor) > fitnessLowerBound)) {
TraceEvent("ConsistencyCheck_DistributorNotBest")
.detail("DataDistributorFitnessLowerBound", fitnessLowerBound)
.detail("ExistingDistributorFitness",
nonExcludedWorkerProcessMap.contains(db.distributor.get().address())
? recruitment::machineClassFitness(
nonExcludedWorkerProcessMap[db.distributor.get().address()].processClass,
recruitment::DataDistributor)
: -1);
co_return false;
}
// Check Ratekeeper
if (db.ratekeeper.present() &&
(!nonExcludedWorkerProcessMap.contains(db.ratekeeper.get().address()) ||
recruitment::machineClassFitness(nonExcludedWorkerProcessMap[db.ratekeeper.get().address()].processClass,
recruitment::Ratekeeper) > fitnessLowerBound)) {
TraceEvent("ConsistencyCheck_RatekeeperNotBest")
.detail("BestRatekeeperFitness", fitnessLowerBound)
.detail("ExistingRatekeeperFitness",
nonExcludedWorkerProcessMap.contains(db.ratekeeper.get().address())
? recruitment::machineClassFitness(
nonExcludedWorkerProcessMap[db.ratekeeper.get().address()].processClass,
recruitment::Ratekeeper)
: -1);
co_return false;
}
// Check ConsistencyScan
if (db.consistencyScan.present() &&
(!nonExcludedWorkerProcessMap.contains(db.consistencyScan.get().address()) ||
recruitment::machineClassFitness(
nonExcludedWorkerProcessMap[db.consistencyScan.get().address()].processClass,
recruitment::ConsistencyScan) > fitnessLowerBound)) {
TraceEvent("ConsistencyCheck_ConsistencyScanNotBest")
.detail("BestConsistencyScanFitness", fitnessLowerBound)
.detail("ExistingConsistencyScanFitness",
nonExcludedWorkerProcessMap.contains(db.consistencyScan.get().address())
? recruitment::machineClassFitness(
nonExcludedWorkerProcessMap[db.consistencyScan.get().address()].processClass,
recruitment::ConsistencyScan)
: -1);
co_return false;
}
// TODO: Check Tlog
co_return true;
}
// returns true if stopped, false otherwise
Future<bool> checkConsistencyScan(Database cx) {
if (!g_network->isSimulated()) {
co_return true;
}
auto tr = makeReference<ReadYourWritesTransaction>(cx);
ConsistencyScanState cs;
while (true) {
Error err;
try {
SystemDBWriteLockedNow(cx.getReference())->setOptions(tr);
ConsistencyScanState::Config config = co_await cs.config().getD(tr);
co_return !config.enabled;
} catch (Error& e) {
err = e;
}
co_await tr->onError(err);
}
}
bool checkSingleSingleton(std::vector<ISimulator::ProcessInfo*> const& allProcesses,
TraceEvent& ev,
std::string const& role,
int expectedCount) {
// FIXME: this doesn't actually check that there aren't multiple of the same role running on the same process
// either
int count = 0;
for (int i = 0; i < allProcesses.size(); i++) {
if (g_simulator->hasRole(allProcesses[i]->address, role)) {
count++;
ev.detail(role + std::to_string(count), allProcesses[i]->address.toString());
}
}
ev.detail(role + "Count", count).detail(role + "ExpectedCount", expectedCount);
if (count != expectedCount) {
fmt::print("ConsistencyCheck failure: incorrect number {0} of singleton {1} running (expected {2})\n",
count,
role,
expectedCount);
}
return count == expectedCount;
}
// checks that there is only one instance of each singleton running in the cluster in simulation
bool checkSingleSingletons(ConsistencyCheckWorkload* self, DatabaseConfiguration config) {
if (!g_network->isSimulated()) {
return true;
}
std::vector<ISimulator::ProcessInfo*> allProcesses = g_simulator->getAllProcesses();
bool success = true;
TraceEvent ev("CheckSingletons");
success &= self->checkSingleSingleton(allProcesses, ev, "Ratekeeper", 1);
success &= self->checkSingleSingleton(allProcesses, ev, "DataDistributor", 1);
success &= self->checkSingleSingleton(allProcesses, ev, "ConsistencyScan", 1);
if (!success) {
// TODO REMOVE
fmt::print("ConsistencyCheck singletons: roles map:\n");
for (int i = 0; i < allProcesses.size(); i++) {
fmt::print(
"{0}: {1}\n", allProcesses[i]->address.toString(), g_simulator->getRoles(allProcesses[i]->address));
}
}
return success;
}
};
WorkloadFactory<ConsistencyCheckWorkload> ConsistencyCheckWorkloadFactory;
Reference in New Issue View Git Blame Copy Permalink