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

2895 lines
125 KiB
C++
Raw Permalink Blame History

/*
* BackupContainerFileSystem.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 "fdbclient/BackupAgent.h"
#include "fdbclient/BackupFileFormat.h"
#include "fdbclient/BackupContainer.h"
#include "flow/BooleanParam.h"
#ifdef BUILD_AZURE_BACKUP
#include "fdbclient/BackupContainerAzureBlobStore.h"
#endif
#include "fdbclient/BackupContainerFileSystem.h"
#include "BackupContainerLocalDirectory.h"
#include "BackupContainerBlobStore.h"
#include "fdbclient/JsonBuilder.h"
#include "fdbrpc/AsyncFileEncrypted.h"
#include "flow/StreamCipher.h"
#include "flow/UnitTest.h"
#include <algorithm>
#include <cinttypes>
class BackupContainerFileSystemImpl {
public:
// A snapshot manifest is normally a few hundred MB. Warn as it grows and error before it gets dangerously
// large, so we see the problem in the logs with time to act before a manifest actually becomes too large
// to handle.
static void traceManifestSize(const std::string& fileName, int64_t bytes) {
constexpr int64_t MB = 1048576; // 1024 * 1024
if (bytes >= 750 * MB) {
TraceEvent(SevError, "BackupSnapshotManifestTooLarge").detail("FileName", fileName).detail("Bytes", bytes);
} else if (bytes >= 500 * MB) {
TraceEvent(SevWarnAlways, "BackupSnapshotManifestLarge")
.detail("FileName", fileName)
.detail("Bytes", bytes);
}
}
// TODO: Do this more efficiently, as the range file list for a snapshot could potentially be hundreds of
// megabytes.
static Future<std::pair<std::vector<RangeFile>, std::map<std::string, KeyRange>>> readKeyspaceSnapshot(
Reference<BackupContainerFileSystem> bc,
KeyspaceSnapshotFile snapshot) {
// Read the range file list for the specified version range, and then index them by fileName.
// This is so we can verify that each of the files listed in the manifest file are also in the container at this
// time.
std::vector<RangeFile> files = co_await bc->listRangeFiles(snapshot.beginVersion, snapshot.endVersion);
std::map<std::string, RangeFile> rangeIndex;
for (auto& f : files)
rangeIndex[f.fileName] = std::move(f);
// Read the snapshot file, verify the version range, then find each of the range files by name in the index and
// return them.
Reference<IAsyncFile> f = co_await bc->readFile(snapshot.fileName);
int64_t size = co_await f->size();
traceManifestSize(snapshot.fileName, size);
// A manifest is normally a few hundred MB. Read it into a std::string in chunks; std::string and the
// chunked reads guard against an unexpectedly large manifest overflowing the int length that read() takes.
// TODO (optimization): the whole manifest is loaded into memory before parsing. Explore if a streaming JSON
// parser would avoid this.
std::string buf;
buf.resize(size);
for (int64_t offset = 0; offset < size;) {
int toRead = static_cast<int>(std::min<int64_t>(CLIENT_KNOBS->BACKUP_MANIFEST_CHUNK_SIZE, size - offset));
int r = co_await f->read((uint8_t*)buf.data() + offset, toRead, offset);
if (r != toRead)
throw restore_corrupted_data();
offset += r;
}
json_spirit::mValue json;
if (!json_spirit::read_string(buf, json)) {
fprintf(stderr,
"ERROR: Failed to read data. Verify that backup and restore encryption keys match (if provided) or "
"the data is corrupted.\n");
throw restore_error();
}
JSONDoc doc(json);
Version v;
if (!doc.tryGet("beginVersion", v) || v != snapshot.beginVersion)
throw restore_corrupted_data();
if (!doc.tryGet("endVersion", v) || v != snapshot.endVersion)
throw restore_corrupted_data();
json_spirit::mValue& filesArray = doc.create("files");
if (filesArray.type() != json_spirit::array_type)
throw restore_corrupted_data();
std::vector<RangeFile> results;
int missing = 0;
for (auto const& fileValue : filesArray.get_array()) {
if (fileValue.type() != json_spirit::str_type)
throw restore_corrupted_data();
// If the file is not in the index then log the error but don't throw yet, keep checking the whole list.
auto i = rangeIndex.find(fileValue.get_str());
if (i == rangeIndex.end()) {
TraceEvent(SevError, "FileRestoreMissingRangeFile")
.detail("URL", bc->getURL())
.detail("File", fileValue.get_str());
++missing;
}
// No point in using more memory once data is missing since an error will be thrown instead.
if (missing == 0) {
results.push_back(i->second);
}
}
if (missing > 0) {
TraceEvent(SevError, "FileRestoreMissingRangeFileSummary")
.detail("URL", bc->getURL())
.detail("Count", missing);
throw restore_missing_data();
}
// Check key ranges for files
std::map<std::string, KeyRange> fileKeyRanges;
JSONDoc ranges = doc.subDoc("keyRanges"); // Create an empty doc if not existed
for (auto i : ranges.obj()) {
const std::string& filename = i.first;
JSONDoc fields(i.second);
std::string begin, end;
if (fields.tryGet("beginKey", begin) && fields.tryGet("endKey", end)) {
TraceEvent("ManifestFields")
.detail("File", filename)
.detail("Begin", printable(StringRef(begin)))
.detail("End", printable(StringRef(end)));
fileKeyRanges.emplace(filename, KeyRange(KeyRangeRef(StringRef(begin), StringRef(end))));
} else {
TraceEvent("MalFormattedManifest").detail("Key", filename);
throw restore_corrupted_data();
}
}
co_return std::make_pair(results, fileKeyRanges);
}
// Find what should be the filename of a path by finding whatever is after the last forward or backward slash, or
// failing to find those, the whole string.
static std::string fileNameOnly(const std::string& path) {
// Find the last forward slash position, defaulting to 0 if not found
int pos = path.find_last_of('/');
if (pos == std::string::npos) {
pos = 0;
}
// Find the last backward slash position after pos, and update pos if found
int b = path.find_last_of('\\', pos);
if (b != std::string::npos) {
pos = b;
}
return path.substr(pos + 1);
}
static bool pathToRangeFile(RangeFile& out, const std::string& path, int64_t size) {
std::string name = fileNameOnly(path);
RangeFile f;
f.fileName = path;
f.fileSize = size;
int len;
if (sscanf(name.c_str(), "range,%" SCNd64 ",%*[^,],%u%n", &f.version, &f.blockSize, &len) == 2 &&
len == name.size()) {
out = f;
return true;
}
return false;
}
static Future<Void> writeKeyspaceSnapshotFile(Reference<BackupContainerFileSystem> bc,
std::vector<std::string> fileNames,
std::vector<std::pair<Key, Key>> beginEndKeys,
int64_t totalBytes,
IncludeKeyRangeMap includeKeyRangeMap,
Optional<SnapshotMetadata> metadata) {
Version minVer = std::numeric_limits<Version>::max();
Version maxVer = 0;
RangeFile rf;
json_spirit::mArray fileArray;
bool isBulkDump = metadata.present() && metadata.get().isBulkDump();
if (isBulkDump) {
// For BulkDump snapshots, version comes from metadata
ASSERT(metadata.get().snapshotVersion != invalidVersion);
minVer = metadata.get().snapshotVersion;
maxVer = metadata.get().snapshotVersion;
// fileNames is empty for BulkDump (files are referenced by job ID)
} else {
ASSERT(!fileNames.empty() && fileNames.size() == beginEndKeys.size());
// Validate each filename, update version range
for (const auto& f : fileNames) {
if (pathToRangeFile(rf, f, 0)) {
fileArray.push_back(f);
if (rf.version < minVer)
minVer = rf.version;
if (rf.version > maxVer)
maxVer = rf.version;
} else {
throw restore_unknown_file_type();
}
co_await yield();
}
}
json_spirit::mValue json;
JSONDoc doc(json);
doc.create("files") = fileArray;
doc.create("totalBytes") = totalBytes;
doc.create("beginVersion") = minVer;
doc.create("endVersion") = maxVer;
if (isBulkDump) {
// Add BulkDump-specific fields
doc.create("snapshotType") = metadata.get().snapshotType;
doc.create("bulkDumpJobId") = metadata.get().bulkDumpJobId;
doc.create("totalKeys") = metadata.get().totalKeys;
// Include the backup ranges
json_spirit::mArray rangesArray;
for (const auto& range : beginEndKeys) {
json_spirit::mObject rangeObj;
rangeObj["begin"] = range.first.toString();
rangeObj["end"] = range.second.toString();
rangesArray.push_back(rangeObj);
}
doc.create("ranges") = rangesArray;
} else if (includeKeyRangeMap) {
auto ranges = doc.subDoc("keyRanges");
for (int i = 0; i < beginEndKeys.size(); i++) {
auto fileDoc = ranges.subDoc(fileNames[i], /*split=*/false);
fileDoc.create("beginKey") = beginEndKeys[i].first.toString();
fileDoc.create("endKey") = beginEndKeys[i].second.toString();
}
}
co_await yield();
// TODO (optimization): the whole manifest is built and serialized in memory before writing. Explore if a
// streaming approach would avoid this.
std::string docString = json_spirit::write_string(json);
// Generate filename - add suffixes only when 'both' mode is active to prevent collision
// Single modes use original format for backward compatibility
std::string fileName;
std::string baseFileName = format("snapshots/snapshot,%lld,%lld,%lld", minVer, maxVer, totalBytes);
if (isBulkDump) {
// For BulkDump: check if we're in 'both' mode by looking for potential rangefile collision
std::vector<KeyspaceSnapshotFile> existingSnapshots = co_await bc->listKeyspaceSnapshots();
bool hasRangefileSnapshot = false;
for (const auto& snapshot : existingSnapshots) {
if (snapshot.fileName == baseFileName || snapshot.fileName == baseFileName + ",range") {
hasRangefileSnapshot = true;
break;
}
}
if (hasRangefileSnapshot) {
// Both mode detected - use suffix to distinguish
fileName = baseFileName + ",bulk";
} else {
// Single bulkdump mode - use original format
fileName = baseFileName;
}
} else {
// For Rangefile: check if we're in 'both' mode by looking for BulkDump collision
std::vector<KeyspaceSnapshotFile> existingSnapshots = co_await bc->listKeyspaceSnapshots();
bool hasBulkDumpSnapshot = false;
for (const auto& snapshot : existingSnapshots) {
if (snapshot.fileName == baseFileName + ",bulk") {
hasBulkDumpSnapshot = true;
break;
}
}
if (hasBulkDumpSnapshot) {
// Both mode detected - BulkDump already exists, use suffix to distinguish
fileName = baseFileName + ",range";
} else {
// Single rangefile mode - use original format for backward compatibility
fileName = baseFileName;
}
}
Reference<IBackupFile> f = co_await bc->writeFile(fileName);
traceManifestSize(fileName, docString.size());
co_await f->append(docString.data(), docString.size());
co_await f->finish();
}
static Future<BackupFileList> dumpFileList(Reference<BackupContainerFileSystem> bc, Version begin, Version end) {
Future<std::vector<RangeFile>> fRanges = bc->listRangeFiles(begin, end);
Future<std::vector<KeyspaceSnapshotFile>> fSnapshots = bc->listKeyspaceSnapshots(begin, end);
std::vector<LogFile> logs;
std::vector<LogFile> pLogs;
co_await (success(fRanges) && success(fSnapshots) &&
store(logs, bc->listLogFiles(begin, end, MutationLogType::DEFAULT)) &&
store(pLogs, bc->listLogFiles(begin, end, MutationLogType::PARTITIONED_LOG)));
logs.insert(logs.end(), std::make_move_iterator(pLogs.begin()), std::make_move_iterator(pLogs.end()));
co_return BackupFileList({ fRanges.get(), std::move(logs), fSnapshots.get() });
}
static Version resolveRelativeVersion(Optional<Version> max, Version v, const char* name, Error e) {
if (v == invalidVersion) {
TraceEvent(SevError, "BackupExpireInvalidVersion").detail(name, v);
throw e;
}
if (v < 0) {
if (!max.present()) {
TraceEvent(SevError, "BackupExpireCannotResolveRelativeVersion").detail(name, v);
throw e;
}
v += max.get();
}
return v;
}
// For a list of log files specified by their indices (of the same tag),
// returns if they are continuous in the range [begin, end]. If "tags" is not
// nullptr, then it will be populated with [begin, end] -> tags, where next
// pair's begin <= previous pair's end + 1. On return, the last pair's end
// version (inclusive) gives the continuous range from begin.
static bool isContinuous(const std::vector<LogFile>& files,
const std::vector<int>& indices,
Version begin,
Version end,
std::map<std::pair<Version, Version>, int>* tags) {
Version lastBegin = invalidVersion;
Version lastEnd = invalidVersion;
int lastTags = -1;
ASSERT(tags == nullptr || tags->empty());
for (int idx : indices) {
const LogFile& file = files[idx];
if (lastEnd == invalidVersion) {
if (file.beginVersion > begin) {
// the first version of the first file must be smaller or equal to the desired beginVersion
return false;
}
if (file.endVersion > begin) {
lastBegin = begin;
lastTags = file.totalTags;
} else {
// if endVerison of file is smaller than desired beginVersion, then do not include this file
continue;
}
} else if (lastEnd < file.beginVersion) {
if (tags != nullptr) {
tags->emplace(std::make_pair(lastBegin, lastEnd - 1), lastTags);
}
return false;
}
if (lastTags != file.totalTags) {
if (tags != nullptr) {
tags->emplace(std::make_pair(lastBegin, file.beginVersion - 1), lastTags);
}
lastBegin = file.beginVersion;
lastTags = file.totalTags;
}
lastEnd = file.endVersion;
if (lastEnd > end)
break;
}
if (tags != nullptr && lastBegin != invalidVersion) {
tags->emplace(std::make_pair(lastBegin, std::min(end, lastEnd - 1)), lastTags);
}
return lastBegin != invalidVersion && lastEnd > end;
}
// Returns the end version such that [begin, end] is continuous.
// "logs" should be already sorted.
static Version getPartitionedLogsContinuousEndVersion(const std::vector<LogFile>& logs, Version begin) {
Version end = 0;
std::map<int, std::vector<int>> tagIndices; // tagId -> indices in files
for (int i = 0; i < logs.size(); i++) {
ASSERT_GE(logs[i].tagId, 0);
ASSERT_LT(logs[i].tagId, logs[i].totalTags);
auto& indices = tagIndices[logs[i].tagId];
// filter out if indices.back() is subset of files[i] or vice versa
if (!indices.empty()) {
if (logs[indices.back()].isSubset(logs[i])) {
ASSERT_LE(logs[indices.back()].fileSize, logs[i].fileSize);
indices.back() = i;
} else if (!logs[i].isSubset(logs[indices.back()])) {
indices.push_back(i);
}
} else {
indices.push_back(i);
}
end = std::max(end, logs[i].endVersion - 1);
}
TraceEvent("ContinuousLogEnd").detail("Begin", begin).detail("InitVersion", end);
// check partition 0 is continuous in [begin, end] and create a map of ranges to partitions
std::map<std::pair<Version, Version>, int> tags; // range [start, end] -> partitions
isContinuous(logs, tagIndices[0], begin, end, &tags);
if (tags.empty() || end <= begin)
return 0;
end = std::min(end, tags.rbegin()->first.second);
TraceEvent("ContinuousLogEnd").detail("Partition", 0).detail("EndVersion", end).detail("Begin", begin);
// for each range in tags, check all partitions from 1 are continuous
Version lastEnd = begin;
for (const auto& [beginEnd, count] : tags) {
Version tagEnd = beginEnd.second; // This range's minimum continuous partition version
for (int i = 1; i < count; i++) {
std::map<std::pair<Version, Version>, int> rangeTags;
isContinuous(logs, tagIndices[i], beginEnd.first, beginEnd.second, &rangeTags);
tagEnd = rangeTags.empty() ? 0 : std::min(tagEnd, rangeTags.rbegin()->first.second);
TraceEvent("ContinuousLogEnd")
.detail("Partition", i)
.detail("EndVersion", tagEnd)
.detail("RangeBegin", beginEnd.first)
.detail("RangeEnd", beginEnd.second);
if (tagEnd == 0)
return lastEnd == begin ? 0 : lastEnd;
}
if (tagEnd < beginEnd.second) {
return tagEnd;
}
lastEnd = beginEnd.second;
}
return end;
}
// Analyze partitioned logs and set contiguousLogEnd for "desc" if larger
// than the "scanBegin" version.
static void updatePartitionedLogsContinuousEnd(BackupDescription* desc,
const std::vector<LogFile>& logs,
const Version scanBegin,
const Version scanEnd) {
if (logs.empty())
return;
Version snapshotBeginVersion = !desc->snapshots.empty() ? desc->snapshots[0].beginVersion : invalidVersion;
Version begin = std::max(scanBegin, desc->minLogBegin.get());
TraceEvent("ContinuousLogEnd")
.detail("ScanBegin", scanBegin)
.detail("ScanEnd", scanEnd)
.detail("Begin", begin)
.detail("ContiguousLogEnd", desc->contiguousLogEnd.get());
for (const auto& file : logs) {
if (file.beginVersion > begin) {
if (scanBegin > 0)
return;
// scanBegin is 0
desc->minLogBegin = file.beginVersion;
begin = file.beginVersion;
}
Version ver = getPartitionedLogsContinuousEndVersion(logs, begin);
if (ver >= desc->contiguousLogEnd.get()) {
// contiguousLogEnd is not inclusive, so +1 here.
desc->contiguousLogEnd.get() = ver + 1;
TraceEvent("UpdateContinuousLogEnd").detail("Version", ver + 1);
if (ver > snapshotBeginVersion)
return;
}
}
}
// Computes the continuous end version for non-partitioned mutation logs up to
// the "targetVersion". If "outLogs" is not nullptr, it will be updated with
// continuous log files. "*end" is updated with the continuous end version.
static void computeRestoreEndVersion(const std::vector<LogFile>& logs,
std::vector<LogFile>* outLogs,
Version* end,
Version targetVersion) {
auto i = logs.begin();
// Add logs to restorable logs set until continuity is broken OR we reach targetVersion
while (i != logs.end()) {
if (i->beginVersion > *end || i->beginVersion > targetVersion)
break;
// If the next link in the log chain is found, update the end
if (i->beginVersion == *end) {
if (outLogs != nullptr)
outLogs->push_back(*i);
*end = i->endVersion;
}
++i;
}
}
// Checks if list of sorted logfiles have the logs from snapshotBeginVersion to snapshotEndversion.
// Which means the sorted log files(have beginVersion and endVersion) should cover
// all the versions between snapshotBegingVersion and snapshotEndversion.
// Note: logs should be pre-sorted according to version order.
static bool hasContinuousLogsForSnapshot(const std::vector<LogFile>& logs,
Version snapshotBeginVersion,
Version snapshotEndVersion) {
auto it = logs.begin();
// find the first mutation log file that covers snapshotBeginVersion
while (it != logs.end()) {
if (it->beginVersion <= snapshotBeginVersion && it->endVersion > snapshotBeginVersion)
break;
++it;
}
// no log find found covering snaphostBeginVersion, return false
if (it == logs.end())
return false;
// If current log entry(it), covers the entire snapshot, return true
if (it->endVersion > snapshotEndVersion)
return true;
// Iterate over the next logs, check if they are continuous and if
// the log file is covering the snapshot.
Version prevEnd = it->endVersion;
++it;
while (it != logs.end()) {
if (it->beginVersion == prevEnd &&
it->endVersion > snapshotEndVersion) // continuous logs until snapshot is covered
return true;
else if (it->beginVersion != prevEnd) // not continuous logs
return false;
prevEnd = it->endVersion;
++it;
} // comes out if the logs are not found until snapshotEndVersion.
return prevEnd > snapshotEndVersion;
}
// Find the continuous log end version starting from beginVersion in the
// given list of sorted logfiles.
// Note: logs should be pre-sorted according to version order.
static Version findContinuousLogEnd(const std::vector<LogFile>& logs, Version beginVersion) {
auto it = logs.begin();
// find the first mutation log file that covers beginVersion
while (it != logs.end()) {
if (it->beginVersion <= beginVersion && it->endVersion > beginVersion)
break;
++it;
}
// no log find found covering beginVersion, return invalidVersion
if (it == logs.end())
return invalidVersion;
// Iterate over the next logs, check if they are continuous
Version prevEnd = it->endVersion;
++it;
while (it != logs.end()) {
if (it->beginVersion != prevEnd) // not continuous logs
return prevEnd;
prevEnd = it->endVersion;
++it;
} // out of logs.
return prevEnd;
}
// Read encryption metadata from JSON file
static Future<std::pair<bool, int>> readEncryptionMetadata(Reference<BackupContainerFileSystem> bc) {
try {
Reference<IAsyncFile> f = co_await bc->readFile(BackupContainerFileSystem::encryptionMetadataFileName());
int64_t size = co_await f->size();
std::string content;
content.resize(size);
co_await f->read((uint8_t*)content.data(), size, 0);
json_spirit::mValue json;
if (json_spirit::read_string(content, json) && json.type() == json_spirit::obj_type) {
auto& obj = json.get_obj();
// Both fields must be present with correct types.
if (!obj.contains("is_encryption_enabled") ||
obj.at("is_encryption_enabled").type() != json_spirit::bool_type ||
!obj.contains("encryption_block_size") ||
obj.at("encryption_block_size").type() != json_spirit::int_type) {
fprintf(stderr, "ERROR: Encryption metadata file is missing required fields or has wrong types.\n");
TraceEvent(SevError, "BackupContainerReadEncryptionMetadataMalformed")
.detail("URL", bc->getURL())
.detail("File", BackupContainerFileSystem::encryptionMetadataFileName());
throw file_corrupt();
}
bool enabled = obj.at("is_encryption_enabled").get_bool();
int blockSize = obj.at("encryption_block_size").get_int();
if ((enabled && blockSize <= 0) || (!enabled && blockSize != 0)) {
fprintf(stderr,
"ERROR: Encryption metadata is inconsistent (enabled=%d, blockSize=%d).\n",
enabled,
blockSize);
TraceEvent(SevError, "BackupContainerReadEncryptionMetadataInconsistent")
.detail("URL", bc->getURL())
.detail("IsEncryptionEnabled", enabled)
.detail("EncryptionBlockSize", blockSize);
throw file_corrupt();
}
TraceEvent("BackupContainerReadEncryptionMetadata")
.detail("URL", bc->getURL())
.detail("IsEncryptionEnabled", enabled)
.detail("EncryptionBlockSize", blockSize);
co_return std::make_pair(enabled, blockSize);
} else {
// If the file is malformed, throw an error.
fprintf(stderr, "ERROR: Failed to read encryption_metadata file due to incorrect format\n");
TraceEvent(SevError, "BackupContainerReadEncryptionMetadataMalformed")
.detail("URL", bc->getURL())
.detail("File", BackupContainerFileSystem::encryptionMetadataFileName());
throw file_corrupt();
}
} catch (Error& e) {
if (e.code() == error_code_file_not_found) {
// File may not be present for older backups, return encryption disabled.
TraceEvent(SevWarn, "BackupContainerEncryptionMetadataNotFound")
.detail("URL", bc->getURL())
.detail("File", BackupContainerFileSystem::encryptionMetadataFileName());
co_return std::make_pair(false, 0);
}
fprintf(stderr, "ERROR: Failed to read encryption_metadata file due to an error.\n");
TraceEvent(SevError, "BackupContainerReadEncryptionMetadataError")
.error(e)
.detail("URL", bc->getURL())
.detail("File", BackupContainerFileSystem::encryptionMetadataFileName());
throw file_not_readable();
}
}
static Future<BackupDescription> describeBackup(Reference<BackupContainerFileSystem> bc,
bool deepScan,
Version logStartVersionOverride) {
BackupDescription desc;
desc.url = bc->getURL();
desc.proxy = bc->getProxy();
TraceEvent("BackupContainerDescribe1")
.detail("URL", bc->getURL())
.detail("LogStartVersionOverride", logStartVersionOverride)
.detail("DeepScan", deepScan);
bool e = co_await bc->exists();
if (!e) {
TraceEvent(SevWarnAlways, "BackupContainerDoesNotExist").detail("URL", bc->getURL());
throw backup_does_not_exist();
}
// If logStartVersion is relative, then first do a recursive call without it to find the max log version
// from which to resolve the relative version.
// This could be handled more efficiently without recursion but it's tricky, this will do for now.
if (logStartVersionOverride != invalidVersion && logStartVersionOverride < 0) {
BackupDescription tmp = co_await bc->describeBackup(false, invalidVersion);
logStartVersionOverride = resolveRelativeVersion(
tmp.maxLogEnd, logStartVersionOverride, "LogStartVersionOverride", invalid_option_value());
}
// Get metadata versions
Optional<Version> metaLogBegin;
Optional<Version> metaLogEnd;
Optional<Version> metaExpiredEnd;
Optional<Version> metaUnreliableEnd;
Optional<Version> metaLogType;
bool fileLevelEncryptionEnabled = false;
int encryptionBlockSize = 0;
std::vector<Future<Void>> metaReads;
metaReads.push_back(store(metaExpiredEnd, bc->expiredEndVersion().get()));
metaReads.push_back(store(metaUnreliableEnd, bc->unreliableEndVersion().get()));
metaReads.push_back(store(metaLogType, bc->logType().get()));
// Only read log begin/end versions if not doing a deep scan, otherwise scan files and recalculate them.
if (!deepScan) {
metaReads.push_back(store(metaLogBegin, bc->logBeginVersion().get()));
metaReads.push_back(store(metaLogEnd, bc->logEndVersion().get()));
}
co_await waitForAll(metaReads);
std::pair<bool, int> encryptionMeta = co_await readEncryptionMetadata(bc);
fileLevelEncryptionEnabled = encryptionMeta.first;
encryptionBlockSize = encryptionMeta.second;
if (fileLevelEncryptionEnabled) {
bc->setEncryptionBlockSize(encryptionBlockSize);
}
TraceEvent("BackupContainerDescribe2")
.detail("URL", bc->getURL())
.detail("LogStartVersionOverride", logStartVersionOverride)
.detail("ExpiredEndVersion", metaExpiredEnd.orDefault(invalidVersion))
.detail("UnreliableEndVersion", metaUnreliableEnd.orDefault(invalidVersion))
.detail("LogBeginVersion", metaLogBegin.orDefault(invalidVersion))
.detail("LogEndVersion", metaLogEnd.orDefault(invalidVersion))
.detail("LogType", metaLogType.orDefault(-1))
.detail("FileLevelEncryption", fileLevelEncryptionEnabled)
.detail("EncryptionBlockSize", encryptionBlockSize);
// If the logStartVersionOverride is positive (not relative) then ensure that unreliableEndVersion is equal or
// greater
if (logStartVersionOverride != invalidVersion &&
metaUnreliableEnd.orDefault(invalidVersion) < logStartVersionOverride) {
metaUnreliableEnd = logStartVersionOverride;
}
// Don't use metaLogBegin or metaLogEnd if any of the following are true, the safest
// thing to do is rescan to verify log continuity and get exact begin/end versions
// - either are missing
// - metaLogEnd <= metaLogBegin (invalid range)
// - metaLogEnd < metaExpiredEnd (log continuity exists in missing data range)
// - metaLogEnd < metaUnreliableEnd (log continuity exists in incomplete data range)
if (!metaLogBegin.present() || !metaLogEnd.present() || metaLogEnd.get() <= metaLogBegin.get() ||
metaLogEnd.get() < metaExpiredEnd.orDefault(invalidVersion) ||
metaLogEnd.get() < metaUnreliableEnd.orDefault(invalidVersion)) {
TraceEvent(SevWarnAlways, "BackupContainerMetadataInvalid")
.detail("URL", bc->getURL())
.detail("ExpiredEndVersion", metaExpiredEnd.orDefault(invalidVersion))
.detail("UnreliableEndVersion", metaUnreliableEnd.orDefault(invalidVersion))
.detail("LogBeginVersion", metaLogBegin.orDefault(invalidVersion))
.detail("LogEndVersion", metaLogEnd.orDefault(invalidVersion));
metaLogBegin = Optional<Version>();
metaLogEnd = Optional<Version>();
}
// If the unreliable end version is not set or is < expiredEndVersion then increase it to expiredEndVersion.
// Describe does not update unreliableEnd in the backup metadata for safety reasons as there is no
// compare-and-set operation to atomically change it and an expire process could be advancing it simultaneously.
if (!metaUnreliableEnd.present() || metaUnreliableEnd.get() < metaExpiredEnd.orDefault(0))
metaUnreliableEnd = metaExpiredEnd;
desc.unreliableEndVersion = metaUnreliableEnd;
desc.expiredEndVersion = metaExpiredEnd;
// Start scanning at the end of the unreliable version range, which is the version before which data is likely
// missing because an expire process has operated on that range.
Version scanBegin = desc.unreliableEndVersion.orDefault(0);
Version scanEnd = std::numeric_limits<Version>::max();
// Use the known log range if present
// Logs are assumed to be contiguous between metaLogBegin and metaLogEnd, so initialize desc accordingly
if (metaLogBegin.present() && metaLogEnd.present()) {
// minLogBegin is the greater of the log begin metadata OR the unreliable end version since we can't count
// on log file presence before that version.
desc.minLogBegin = std::max(metaLogBegin.get(), desc.unreliableEndVersion.orDefault(0));
// Set the maximum known end version of a log file, so far, which is also the assumed contiguous log file
// end version
desc.maxLogEnd = metaLogEnd.get();
desc.contiguousLogEnd = desc.maxLogEnd;
// Advance scanBegin to the contiguous log end version
scanBegin = desc.contiguousLogEnd.get();
}
std::vector<LogFile> logs;
std::vector<LogFile> plogs;
TraceEvent("BackupContainerListFiles").detail("URL", bc->getURL());
co_await (store(logs, bc->listLogFiles(scanBegin, scanEnd, MutationLogType::DEFAULT)) &&
store(plogs, bc->listLogFiles(scanBegin, scanEnd, MutationLogType::PARTITIONED_LOG)) &&
store(desc.snapshots, bc->listKeyspaceSnapshots()));
TraceEvent("BackupContainerListFiles")
.detail("URL", bc->getURL())
.detail("LogFiles", logs.size())
.detail("PLogsFiles", plogs.size())
.detail("Snapshots", desc.snapshots.size());
if (!plogs.empty()) {
desc.mutationLogType = MutationLogType::PARTITIONED_LOG;
logs.swap(plogs);
} else {
desc.mutationLogType =
metaLogType.present() ? static_cast<MutationLogType>(metaLogType.get()) : MutationLogType::DEFAULT;
}
desc.fileLevelEncryption = fileLevelEncryptionEnabled;
desc.encryptionBlockSize = encryptionBlockSize;
// List logs in version order so log continuity can be analyzed
std::sort(logs.begin(), logs.end());
// Find out contiguous log end version
if (!logs.empty()) {
desc.maxLogEnd = logs.rbegin()->endVersion;
// If we didn't get log versions above then seed them using the first log file
if (!desc.contiguousLogEnd.present()) {
desc.minLogBegin = logs.begin()->beginVersion;
if (desc.mutationLogType == MutationLogType::PARTITIONED_LOG) {
// Cannot use the first file's end version, which may not be contiguous
// for other partitions. Set to its beginVersion to be safe.
desc.contiguousLogEnd = logs.begin()->beginVersion;
} else {
desc.contiguousLogEnd = logs.begin()->beginVersion;
}
}
if (desc.mutationLogType == MutationLogType::PARTITIONED_LOG) {
updatePartitionedLogsContinuousEnd(&desc, logs, scanBegin, scanEnd);
} else {
Version& end = desc.contiguousLogEnd.get();
computeRestoreEndVersion(logs, nullptr, &end, std::numeric_limits<Version>::max());
}
}
// Only update stored contiguous log begin and end versions if we did NOT use a log start override.
// Otherwise, a series of describe operations can result in a version range which is actually missing data.
if (logStartVersionOverride == invalidVersion) {
// If the log metadata begin/end versions are missing (or treated as missing due to invalidity) or
// differ from the newly calculated values for minLogBegin and contiguousLogEnd, respectively,
// then attempt to update the metadata in the backup container but ignore errors in case the
// container is not writeable.
try {
Future<Void> updates = Void();
if (desc.minLogBegin.present() && metaLogBegin != desc.minLogBegin) {
updates = updates && bc->logBeginVersion().set(desc.minLogBegin.get());
}
if (desc.contiguousLogEnd.present() && metaLogEnd != desc.contiguousLogEnd) {
updates = updates && bc->logEndVersion().set(desc.contiguousLogEnd.get());
}
if (!metaLogType.present()) {
updates = updates && bc->logType().set(static_cast<int>(desc.mutationLogType));
}
co_await updates;
} catch (Error& e) {
if (e.code() == error_code_actor_cancelled)
throw;
TraceEvent(SevWarn, "BackupContainerMetadataUpdateFailure").error(e).detail("URL", bc->getURL());
}
}
for (auto& s : desc.snapshots) {
// Calculate restorability of each snapshot. Assume true, then try to prove false
s.restorable = true;
// If this is not a single-version snapshot then see if the available contiguous logs cover its range
if (s.beginVersion != s.endVersion) {
if (!desc.minLogBegin.present() || desc.minLogBegin.get() > s.beginVersion)
s.restorable = false;
if (!desc.contiguousLogEnd.present() || desc.contiguousLogEnd.get() <= s.endVersion)
s.restorable = false;
// If there is logs gap after contiguousLogEnd, then check whether the current snapshot
// can be restored from the logs available after contiguousLogEnd.
if (desc.contiguousLogEnd.present() && desc.contiguousLogEnd.get() <= s.beginVersion) {
if (desc.mutationLogType == MutationLogType::PARTITIONED_LOG)
s.restorable = isPartitionedLogsContinuous(logs, s.beginVersion, s.endVersion);
else
s.restorable = hasContinuousLogsForSnapshot(logs, s.beginVersion, s.endVersion);
}
}
desc.snapshotBytes += s.totalSize;
// If the snapshot is at a single version and then it requires no logs. Update min and max restorable.
// Update only if minRestorableVersion and maxRestorableVersion are not set. If they are set, we should
// check for log continuity between current minRestorableVersion to s.endVersion which happens in the
// next if block.
if (s.beginVersion == s.endVersion && !desc.minRestorableVersion.present() &&
!desc.maxRestorableVersion.present()) {
desc.minRestorableVersion = s.endVersion;
desc.maxRestorableVersion = s.endVersion;
}
// If the snapshot is covered by the contiguous log chain then update min/max restorable.
if (desc.minLogBegin.present() && s.beginVersion >= desc.minLogBegin.get() &&
s.endVersion < desc.contiguousLogEnd.get()) {
// If minRestorableVersion not present, update minRestorableVersion to snapshot endVersion.
// If minRestorableVersion present and if it has continuous logs from minRestorableVersion
// to snapshot endVersion, don't update the minRestorableVersion.
// Else, means it has no continous logs, so update minRestorableVersion to s.endVersion.
if (!desc.minRestorableVersion.present() ||
!(desc.minRestorableVersion.get() >= desc.minLogBegin.get() &&
desc.minRestorableVersion.get() < desc.contiguousLogEnd.get()))
desc.minRestorableVersion = s.endVersion;
if (!desc.maxRestorableVersion.present() ||
(desc.contiguousLogEnd.get() - 1) > desc.maxRestorableVersion.get())
desc.maxRestorableVersion = desc.contiguousLogEnd.get() - 1;
}
// If there is logs gap after contiguousLogEnd and if current snapshot is restorable(have continuous logs)
if (desc.contiguousLogEnd.present() &&
((desc.contiguousLogEnd.get() < s.beginVersion) ||
// if contiguousLogEnd==s.beginVersion==s.endVersion, there is no need to check for continuous logs in
// single version snapshot. And this case is covered in above if condition.
(desc.contiguousLogEnd.get() == s.beginVersion && s.beginVersion != s.endVersion)) &&
s.restorable.get()) {
if (desc.minRestorableVersion.present() && desc.maxRestorableVersion.present()) {
// check if we have contiguous logs from minRestorableVersion to current snapshot endVersion
bool contiguousLogs = false;
if (desc.mutationLogType == MutationLogType::PARTITIONED_LOG) {
contiguousLogs =
isPartitionedLogsContinuous(logs, desc.minRestorableVersion.get(), s.endVersion);
} else {
contiguousLogs =
hasContinuousLogsForSnapshot(logs, desc.minRestorableVersion.get(), s.endVersion);
}
if (contiguousLogs) {
// The previous restorable version can be extended to current snapshot version,
// so minRestorableVersion remain same
desc.maxRestorableVersion = s.endVersion;
} else {
// Previous restorable version cannot be extended to current snapshot version,
// means some logs are missing inbetween.
// So set the snapshot beginversion as minRestorableVersion.
desc.minRestorableVersion = s.endVersion;
}
} else {
// There is no previous snapshot that is restorable.
// Since the current snapshot is restorable, set the snapshot beginversion as minRestorableVersion.
desc.minRestorableVersion = s.endVersion;
}
// Find the continuousLogEnd after snapshotEndVersion and set it as
// maxRestorableVersion.
if (desc.mutationLogType == MutationLogType::PARTITIONED_LOG) {
// TO DO: Yet to implement similar function findContinuousLogEnd for partitioned logs.
desc.maxRestorableVersion = s.endVersion;
} else {
Version maxContinuousLogEnd = findContinuousLogEnd(logs, s.endVersion);
desc.maxRestorableVersion =
(maxContinuousLogEnd == invalidVersion) ? s.endVersion : maxContinuousLogEnd - 1;
}
}
}
co_return desc;
}
static Future<Void> expireData(Reference<BackupContainerFileSystem> bc,
Version expireEndVersion,
bool force,
IBackupContainer::ExpireProgress* progress,
Version restorableBeginVersion) {
if (progress != nullptr) {
progress->step = "Describing backup";
progress->total = 0;
}
TraceEvent("BackupContainerFileSystemExpire1")
.detail("URL", bc->getURL())
.detail("ExpireEndVersion", expireEndVersion)
.detail("RestorableBeginVersion", restorableBeginVersion);
// Get the backup description.
BackupDescription desc = co_await bc->describeBackup(false, expireEndVersion);
// Resolve relative versions using max log version
expireEndVersion =
resolveRelativeVersion(desc.maxLogEnd, expireEndVersion, "ExpireEndVersion", invalid_option_value());
restorableBeginVersion = resolveRelativeVersion(
desc.maxLogEnd, restorableBeginVersion, "RestorableBeginVersion", invalid_option_value());
if (progress != nullptr) {
progress->requestedEndVersion = expireEndVersion;
}
// It would be impossible to have restorability to any version < expireEndVersion after expiring to that version
if (restorableBeginVersion < expireEndVersion)
throw backup_cannot_expire();
// If the expire request is to a version at or before the previous version to which data was already deleted
// then do nothing and just return
if (expireEndVersion <= desc.expiredEndVersion.orDefault(invalidVersion)) {
if (progress != nullptr) {
progress->actualEndVersion = desc.expiredEndVersion.orDefault(invalidVersion);
}
co_return;
}
// Assume force is needed, then try to prove otherwise.
// Force is required if there is not a restorable snapshot which both
// - begins at or after expireEndVersion
// - ends at or before restorableBeginVersion
bool forceNeeded = true;
for (KeyspaceSnapshotFile& s : desc.snapshots) {
if (s.restorable.orDefault(false) && s.beginVersion >= expireEndVersion &&
s.endVersion <= restorableBeginVersion) {
forceNeeded = false;
break;
}
}
// If force is needed but not passed then refuse to expire anything.
// Note that it is possible for there to be no actual files in the backup prior to expireEndVersion,
// if they were externally deleted or an expire operation deleted them but was terminated before
// updating expireEndVersion
if (forceNeeded && !force)
throw backup_cannot_expire();
// Start scan for files to delete at the last completed expire operation's end or 0.
Version scanBegin = desc.expiredEndVersion.orDefault(0);
TraceEvent("BackupContainerFileSystemExpire2")
.detail("URL", bc->getURL())
.detail("ExpireEndVersion", expireEndVersion)
.detail("RestorableBeginVersion", restorableBeginVersion)
.detail("ScanBeginVersion", scanBegin);
std::vector<LogFile> logs;
std::vector<LogFile> pLogs; // partitioned mutation logs
std::vector<RangeFile> ranges;
if (progress != nullptr) {
progress->step = "Listing files";
}
// Get log files or range files that contain any data at or before expireEndVersion
co_await (store(logs, bc->listLogFiles(scanBegin, expireEndVersion - 1, MutationLogType::DEFAULT)) &&
store(pLogs, bc->listLogFiles(scanBegin, expireEndVersion - 1, MutationLogType::PARTITIONED_LOG)) &&
store(ranges, bc->listRangeFiles(scanBegin, expireEndVersion - 1)));
logs.insert(logs.end(), std::make_move_iterator(pLogs.begin()), std::make_move_iterator(pLogs.end()));
// The new logBeginVersion will be taken from the last log file, if there is one
Optional<Version> newLogBeginVersion;
if (!logs.empty()) {
// Linear scan the unsorted logs to find the latest one in sorted order
LogFile& last = *std::max_element(logs.begin(), logs.end());
// If the last log ends at expireEndVersion then that will be the next log begin
if (last.endVersion == expireEndVersion) {
newLogBeginVersion = expireEndVersion;
} else {
// If the last log overlaps the expiredEnd then use the log's begin version and move the expiredEnd
// back to match it and keep the last log file
if (last.endVersion > expireEndVersion) {
newLogBeginVersion = last.beginVersion;
// Instead of modifying this potentially very large vector, just clear LogFile
last = LogFile();
expireEndVersion = newLogBeginVersion.get();
}
}
}
if (progress != nullptr) {
progress->actualEndVersion = expireEndVersion;
}
// Make a list of files to delete
std::vector<std::string> toDelete;
// Move filenames out of vector then destroy it to save memory
for (auto& f : logs) {
// We may have cleared the last log file earlier so skip any empty filenames
if (!f.fileName.empty()) {
toDelete.push_back(std::move(f.fileName));
}
}
logs.clear();
// Move filenames out of vector then destroy it to save memory
for (auto& f : ranges) {
// The file version must be checked here again because it is likely that expireEndVersion is in the middle
// of a log file, in which case after the log and range file listings are done (using the original
// expireEndVersion) the expireEndVersion will be moved back slightly to the begin version of the last log
// file found (which is also the first log to not be deleted)
if (f.version < expireEndVersion) {
toDelete.push_back(std::move(f.fileName));
}
}
ranges.clear();
for (auto& f : desc.snapshots) {
if (f.endVersion < expireEndVersion)
toDelete.push_back(std::move(f.fileName));
}
desc = BackupDescription();
// We are about to start deleting files, at which point all data prior to expireEndVersion is considered
// 'unreliable' as some or all of it will be missing. So before deleting anything, read unreliableEndVersion
// (don't use cached value in desc) and update its value if it is missing or < expireEndVersion
if (progress != nullptr) {
progress->step = "Initial metadata update";
}
Optional<Version> metaUnreliableEnd = co_await bc->unreliableEndVersion().get();
if (metaUnreliableEnd.orDefault(0) < expireEndVersion) {
co_await bc->unreliableEndVersion().set(expireEndVersion);
}
if (progress != nullptr) {
progress->step = "Deleting files";
progress->total = toDelete.size();
progress->done = 0;
}
// Delete files, but limit parallelism because the file list could use a lot of memory and the corresponding
// delete actor states would use even more if they all existed at the same time.
std::list<Future<Void>> deleteFutures;
while (!toDelete.empty() || !deleteFutures.empty()) {
// While there are files to delete and budget in the deleteFutures list, start a delete
while (!toDelete.empty() && deleteFutures.size() < CLIENT_KNOBS->BACKUP_CONCURRENT_DELETES) {
deleteFutures.push_back(bc->deleteFile(toDelete.back()));
toDelete.pop_back();
}
// Wait for deletes to finish until there are only targetDeletesInFlight remaining.
// If there are no files left to start then this value is 0, otherwise it is one less
// than the delete concurrency limit.
int targetFuturesSize = toDelete.empty() ? 0 : (CLIENT_KNOBS->BACKUP_CONCURRENT_DELETES - 1);
while (deleteFutures.size() > targetFuturesSize) {
co_await deleteFutures.front();
if (progress != nullptr) {
++progress->done;
}
deleteFutures.pop_front();
}
}
if (progress != nullptr) {
progress->step = "Final metadata update";
progress->total = 0;
}
// Update the expiredEndVersion metadata to indicate that everything prior to that version has been
// successfully deleted if the current version is lower or missing
Optional<Version> metaExpiredEnd = co_await bc->expiredEndVersion().get();
if (metaExpiredEnd.orDefault(0) < expireEndVersion) {
co_await bc->expiredEndVersion().set(expireEndVersion);
}
}
// Returns true if logs are continuous in the range [begin, end].
// "files" should be pre-sorted according to version order.
static bool isPartitionedLogsContinuous(const std::vector<LogFile>& files, Version begin, Version end) {
std::map<int, std::vector<int>> tagIndices; // tagId -> indices in files
for (int i = 0; i < files.size(); i++) {
ASSERT(files[i].tagId >= 0 && files[i].tagId < files[i].totalTags);
auto& indices = tagIndices[files[i].tagId];
indices.push_back(i);
}
// check partition 0 is continuous and create a map of ranges to tags
std::map<std::pair<Version, Version>, int> tags; // range [begin, end] -> tags
if (!isContinuous(files, tagIndices[0], begin, end, &tags)) {
TraceEvent(SevWarn, "BackupFileNotContinuous")
.detail("Partition", 0)
.detail("RangeBegin", begin)
.detail("RangeEnd", end);
return false;
}
// for each range in tags, check all tags from 1 are continouous
for (const auto& [beginEnd, count] : tags) {
for (int i = 1; i < count; i++) {
if (!isContinuous(files, tagIndices[i], beginEnd.first, std::min(beginEnd.second - 1, end), nullptr)) {
TraceEvent(SevWarn, "BackupFileNotContinuous")
.detail("Partition", i)
.detail("RangeBegin", beginEnd.first)
.detail("RangeEnd", beginEnd.second);
return false;
}
}
}
return true;
}
// Returns log files that are not duplicated, or subset of another log.
// If a log file's progress is not saved, a new log file will be generated
// with the same begin version. So we can have a file that contains a subset
// of contents in another log file.
// PRE-CONDITION: logs are already sorted by (tagId, beginVersion, endVersion).
static std::vector<LogFile> filterDuplicates(const std::vector<LogFile>& logs) {
std::vector<LogFile> filtered;
int i = 0;
for (int j = 1; j < logs.size(); j++) {
if (logs[j].isSubset(logs[i])) {
ASSERT_LE(logs[j].fileSize, logs[i].fileSize);
continue;
}
if (!logs[i].isSubset(logs[j])) {
filtered.push_back(logs[i]);
}
i = j;
}
if (i < logs.size())
filtered.push_back(logs[i]);
return filtered;
}
static Optional<RestorableFileSet> getRestoreSetFromLogs(const std::vector<LogFile>& logs,
Version targetVersion,
RestorableFileSet restorable) {
Version end = logs.begin()->beginVersion;
computeRestoreEndVersion(logs, &restorable.logs, &end, targetVersion);
if (end >= targetVersion) {
restorable.continuousBeginVersion = logs.begin()->beginVersion;
restorable.continuousEndVersion = end;
return Optional<RestorableFileSet>(restorable);
}
return Optional<RestorableFileSet>();
}
// Get a set of files that can restore the given "keyRangesFilter" to the "targetVersion".
// If "keyRangesFilter" is empty, the file set will cover all key ranges present in the backup.
// It's generally a good idea to specify "keyRangesFilter" to reduce the number of files for
// restore times.
// If "logsOnly" is true, then only log files are returned and "keyRangesFilter" is ignored,
// because the log can contain mutations of the whole key space, unlike range files that each
// is limited to a smaller key range.
static Future<Optional<RestorableFileSet>> getRestoreSet(Reference<BackupContainerFileSystem> bc,
Version targetVersion,
VectorRef<KeyRangeRef> keyRangesFilter,
bool logsOnly = false,
Version beginVersion = invalidVersion) {
for (const auto& range : keyRangesFilter) {
TraceEvent("BackupContainerGetRestoreSet").detail("RangeFilter", printable(range));
}
if (logsOnly) {
RestorableFileSet restorableSet;
restorableSet.targetVersion = targetVersion;
std::vector<LogFile> logFiles;
Version begin = beginVersion == invalidVersion ? 0 : beginVersion;
logFiles = co_await bc->listLogFiles(begin, targetVersion, MutationLogType::DEFAULT);
// List logs in version order so log continuity can be analyzed
std::sort(logFiles.begin(), logFiles.end());
if (!logFiles.empty()) {
co_return getRestoreSetFromLogs(logFiles, targetVersion, restorableSet);
}
}
// Find the most recent keyrange snapshot through which we can restore filtered key ranges into targetVersion.
std::vector<KeyspaceSnapshotFile> snapshots = co_await bc->listKeyspaceSnapshots();
for (int i = snapshots.size() - 1; i >= 0; i--) {
// The smallest version of filtered range files >= snapshot beginVersion > targetVersion
if (targetVersion >= 0 && snapshots[i].beginVersion > targetVersion) {
continue;
}
RestorableFileSet restorable;
Version minKeyRangeVersion = MAX_VERSION;
Version maxKeyRangeVersion = -1;
std::pair<std::vector<RangeFile>, std::map<std::string, KeyRange>> results =
co_await bc->readKeyspaceSnapshot(snapshots[i]);
// If there is no key ranges filter for the restore OR if the snapshot contains no per-file key range info
// then return all of the range files
if (keyRangesFilter.empty() || results.second.empty()) {
restorable.ranges = std::move(results.first);
restorable.keyRanges = std::move(results.second);
minKeyRangeVersion = snapshots[i].beginVersion;
maxKeyRangeVersion = snapshots[i].endVersion;
} else {
for (const auto& rangeFile : results.first) {
const auto& keyRange = results.second.at(rangeFile.fileName);
if (keyRange.intersects(keyRangesFilter)) {
restorable.ranges.push_back(rangeFile);
restorable.keyRanges[rangeFile.fileName] = keyRange;
minKeyRangeVersion = std::min(minKeyRangeVersion, rangeFile.version);
maxKeyRangeVersion = std::max(maxKeyRangeVersion, rangeFile.version);
}
}
// No range file matches 'keyRangesFilter'.
if (restorable.ranges.empty()) {
throw backup_not_overlapped_with_keys_filter();
}
}
// 'latestVersion' represents using the minimum restorable version in a snapshot.
restorable.targetVersion = targetVersion == latestVersion ? maxKeyRangeVersion : targetVersion;
// Any version < maxKeyRangeVersion is not restorable.
if (restorable.targetVersion < maxKeyRangeVersion)
continue;
// restorable.snapshot.beginVersion is set to the smallest(oldest) snapshot's beginVersion
restorable.snapshot = snapshots[i];
// No logs needed if there is a complete filtered key space snapshot at the target version.
if (minKeyRangeVersion == maxKeyRangeVersion && maxKeyRangeVersion == restorable.targetVersion) {
restorable.continuousBeginVersion = restorable.continuousEndVersion = invalidVersion;
TraceEvent("BackupContainerGetRestorableFilesWithoutLogs")
.detail("KeyRangeVersion", restorable.targetVersion)
.detail("NumberOfRangeFiles", restorable.ranges.size())
.detail("KeyRangesFilter", printable(keyRangesFilter));
co_return Optional<RestorableFileSet>(restorable);
}
// FIXME: check if there are tagged logs. for each tag, there is no version gap.
std::vector<LogFile> logs;
std::vector<LogFile> plogs;
co_await (
store(logs, bc->listLogFiles(minKeyRangeVersion, restorable.targetVersion, MutationLogType::DEFAULT)) &&
store(
plogs,
bc->listLogFiles(minKeyRangeVersion, restorable.targetVersion, MutationLogType::PARTITIONED_LOG)));
if (!plogs.empty()) {
logs.swap(plogs);
// sort by tag ID so that filterDuplicates works.
std::sort(logs.begin(), logs.end(), [](const LogFile& a, const LogFile& b) {
return std::tie(a.tagId, a.beginVersion, a.endVersion) <
std::tie(b.tagId, b.beginVersion, b.endVersion);
});
// Remove duplicated log files that can happen for old epochs.
std::vector<LogFile> filtered = filterDuplicates(logs);
restorable.logs.swap(filtered);
// sort by version order again for continuous analysis
std::sort(restorable.logs.begin(), restorable.logs.end());
if (isPartitionedLogsContinuous(restorable.logs, minKeyRangeVersion, restorable.targetVersion)) {
restorable.continuousBeginVersion = minKeyRangeVersion;
restorable.continuousEndVersion = restorable.targetVersion + 1; // not inclusive
co_return Optional<RestorableFileSet>(restorable);
}
co_return Optional<RestorableFileSet>();
}
// List logs in version order so log continuity can be analyzed
std::sort(logs.begin(), logs.end());
// If there are logs and the first one starts at or before the keyrange's snapshot begin version, then
// it is valid restore set and proceed
if (!logs.empty() && logs.front().beginVersion <= minKeyRangeVersion) {
co_return getRestoreSetFromLogs(logs, targetVersion, restorable);
}
}
co_return Optional<RestorableFileSet>();
}
static std::string versionFolderString(Version v, int smallestBucket) {
ASSERT_LT(smallestBucket, 14);
// Get a 0-padded fixed size representation of v
std::string vFixedPrecision = format("%019lld", v);
ASSERT_EQ(vFixedPrecision.size(), 19);
// Truncate smallestBucket from the fixed length representation
vFixedPrecision.resize(vFixedPrecision.size() - smallestBucket);
// Split the remaining digits with a '/' 4 places from the right
vFixedPrecision.insert(vFixedPrecision.size() - 4, 1, '/');
return vFixedPrecision;
}
// This useful for comparing version folder strings regardless of where their "/" dividers are, as it is possible
// that division points would change in the future.
static std::string cleanFolderString(std::string f) {
f.erase(std::remove(f.begin(), f.end(), '/'), f.end());
return f;
}
// The innermost folder covers 100 seconds (1e8 versions) During a full speed backup it is possible though very
// unlikely write about 10,000 snapshot range files during that time.
static std::string old_rangeVersionFolderString(Version v) {
return format("ranges/%s/", versionFolderString(v, 8).c_str());
}
// Get the root folder for a snapshot's data based on its begin version
static std::string snapshotFolderString(Version snapshotBeginVersion) {
return format("kvranges/snapshot.%018" PRId64, snapshotBeginVersion);
}
// Extract the snapshot begin version from a path
static Version extractSnapshotBeginVersion(const std::string& path) {
Version snapshotBeginVersion;
if (sscanf(path.c_str(), "kvranges/snapshot.%018" SCNd64, &snapshotBeginVersion) == 1) {
return snapshotBeginVersion;
}
return invalidVersion;
}
// The innermost folder covers 100,000 seconds (1e11 versions) which is 5,000 mutation log files at current
// settings.
static std::string logVersionFolderString(Version v, MutationLogType mutationLogType) {
return format("%s/%s/",
(mutationLogType == MutationLogType::PARTITIONED_LOG ? "plogs" : "logs"),
versionFolderString(v, 11).c_str());
}
static std::string logVersionFolderStringForRangePartitioned(Version v, Version baseVersion) {
Version directoryVersion =
baseVersion + ((v - baseVersion) / CLIENT_KNOBS->RANGE_PARTITIONED_BACKUP_VDIR_INTERVAL) *
CLIENT_KNOBS->RANGE_PARTITIONED_BACKUP_VDIR_INTERVAL;
std::string vFixed = format("%019lld", directoryVersion);
return format("rlogs/%s/", vFixed.c_str());
}
static bool pathToLogFile(LogFile& out, const std::string& path, int64_t size) {
std::string name = fileNameOnly(path);
LogFile f;
f.fileName = path;
f.fileSize = size;
int len;
if (sscanf(name.c_str(),
"log,%" SCNd64 ",%" SCNd64 ",%*[^,],%u%n",
&f.beginVersion,
&f.endVersion,
&f.blockSize,
&len) == 3 &&
len == name.size()) {
out = f;
return true;
} else if (sscanf(name.c_str(),
"log,%" SCNd64 ",%" SCNd64 ",%*[^,],%d-of-%d,%u%n",
&f.beginVersion,
&f.endVersion,
&f.tagId,
&f.totalTags,
&f.blockSize,
&len) == 5 &&
len == name.size() && f.tagId >= 0) {
out = f;
return true;
}
return false;
}
static bool pathToKeyspaceSnapshotFile(KeyspaceSnapshotFile& out, const std::string& path) {
std::string name = fileNameOnly(path);
KeyspaceSnapshotFile f;
f.fileName = path;
int len;
char typeBuf[64] = {};
// Try new format with type suffix: snapshot,beginVersion,endVersion,totalSize,type
if (sscanf(name.c_str(),
"snapshot,%" SCNd64 ",%" SCNd64 ",%" SCNd64 ",%63[^,]%n",
&f.beginVersion,
&f.endVersion,
&f.totalSize,
typeBuf,
&len) == 4 &&
len == name.size()) {
f.snapshotType = typeBuf;
out = f;
return true;
}
// Try original format: snapshot,beginVersion,endVersion,totalSize
if (sscanf(name.c_str(),
"snapshot,%" SCNd64 ",%" SCNd64 ",%" SCNd64 "%n",
&f.beginVersion,
&f.endVersion,
&f.totalSize,
&len) == 3 &&
len == name.size()) {
out = f;
return true;
}
return false;
}
// fallback for using existing write api if the underlying blob store doesn't support efficient writeEntireFile
static Future<Void> writeEntireFileFallback(Reference<BackupContainerFileSystem> bc,
std::string fileName,
std::string fileContents) {
Reference<IBackupFile> objectFile = co_await bc->writeFile(fileName);
co_await objectFile->append(&fileContents[0], fileContents.size());
co_await objectFile->finish();
}
static Future<Void> createTestEncryptionKeyFile(std::string filename) {
if (fileExists(filename)) {
// Key file already exists, don't overwrite it -> only for testing between backup and restore workloads to
// share the key.
TraceEvent("EncryptionKeyFileExists").detail("FileName", filename);
co_return;
}
Reference<IAsyncFile> keyFile = co_await IAsyncFileSystem::filesystem()->open(
filename,
IAsyncFile::OPEN_ATOMIC_WRITE_AND_CREATE | IAsyncFile::OPEN_READWRITE | IAsyncFile::OPEN_CREATE,
0600);
StreamCipherKey testKey(AES_256_KEY_LENGTH);
testKey.initializeRandomTestKey();
keyFile->write(testKey.data(), testKey.size(), 0);
co_await keyFile->sync();
}
static Future<Void> readEncryptionKey(std::string encryptionKeyFileName) {
Reference<IAsyncFile> keyFile;
StreamCipherKey const* cipherKey = StreamCipherKey::getGlobalCipherKey();
try {
Reference<IAsyncFile> _keyFile = co_await IAsyncFileSystem::filesystem()->open(
encryptionKeyFileName,
IAsyncFile::OPEN_NO_AIO | IAsyncFile::OPEN_READONLY | IAsyncFile::OPEN_UNCACHED,
0400);
keyFile = _keyFile;
} catch (Error& e) {
TraceEvent(SevError, "FailedToOpenEncryptionKeyFile").error(e).detail("FileName", encryptionKeyFileName);
throw e;
}
int bytesRead = co_await uncancellable(keyFile->read(cipherKey->data(), cipherKey->size(), 0));
if (bytesRead != cipherKey->size()) {
TraceEvent(SevError, "InvalidEncryptionKeyFileSize")
.detail("ExpectedSize", cipherKey->size())
.detail("ActualSize", bytesRead)
.detail("FileName", encryptionKeyFileName);
throw invalid_encryption_key_file();
}
ASSERT_EQ(bytesRead, cipherKey->size());
}
static Future<Void> writeEncryptionMetadataIfNotExists(Reference<BackupContainerFileSystem> bc,
int encryptionBlockSize) {
try {
Reference<IAsyncFile> f = co_await bc->readFile(BackupContainerFileSystem::encryptionMetadataFileName());
int64_t size = co_await f->size();
TraceEvent("WriteEncryptionMetadataAlreadyExists").detail("URL", bc->getURL()).detail("FileSize", size);
co_return;
} catch (Error& e) {
if (e.code() != error_code_file_not_found) {
TraceEvent(SevWarn, "WriteEncryptionMetadataReadError").error(e).detail("URL", bc->getURL());
throw e;
}
}
bool exists = co_await bc->exists();
if (!exists) {
TraceEvent("WriteEncryptionMetadataCreatingContainer").detail("URL", bc->getURL());
co_await bc->create();
}
// Write JSON with encryption metadata
bool enabled = bc->encryptionKeyFileName.present();
JsonBuilderObject doc;
doc.setKey("is_encryption_enabled", enabled);
doc.setKey("encryption_block_size", encryptionBlockSize);
std::string jsonStr = doc.getJson();
TraceEvent("WriteEncryptionMetadataCompleted").detail("URL", bc->getURL()).detail("JSON", jsonStr);
co_await bc->writeEntireFile(BackupContainerFileSystem::encryptionMetadataFileName(), jsonStr);
}
}; // class BackupContainerFileSystemImpl
Future<Reference<IBackupFile>> BackupContainerFileSystem::writeLogFile(Version beginVersion,
Version endVersion,
int blockSize) {
return writeFile(BackupContainerFileSystemImpl::logVersionFolderString(beginVersion, MutationLogType::DEFAULT) +
format("log,%lld,%lld,%s,%d",
beginVersion,
endVersion,
deterministicRandom()->randomUniqueID().toString().c_str(),
blockSize));
}
Future<Reference<IBackupFile>> BackupContainerFileSystem::writeTaggedLogFile(Version beginVersion,
Version endVersion,
int blockSize,
uint16_t tagId,
int totalTags) {
return writeFile(
BackupContainerFileSystemImpl::logVersionFolderString(beginVersion, MutationLogType::PARTITIONED_LOG) +
format("log,%lld,%lld,%s,%d-of-%d,%d",
beginVersion,
endVersion,
deterministicRandom()->randomUniqueID().toString().c_str(),
tagId,
totalTags,
blockSize));
}
Future<Reference<IBackupFile>> BackupContainerFileSystem::writeRangePartitionedLogFile(Version beginVersion,
Version endVersion,
Version baseVersion,
int32_t partitionId,
int blockSize) {
return writeFile(
BackupContainerFileSystemImpl::logVersionFolderStringForRangePartitioned(beginVersion, baseVersion) +
format("log,%lld,%lld,%d,%d", beginVersion, endVersion, partitionId, blockSize));
}
Future<Reference<IBackupFile>> BackupContainerFileSystem::writeRangeFile(Version snapshotBeginVersion,
int snapshotFileCount,
Version fileVersion,
int blockSize) {
std::string fileName = format(
"range,%" PRId64 ",%s,%d", fileVersion, deterministicRandom()->randomUniqueID().toString().c_str(), blockSize);
// In order to test backward compatibility in simulation, sometimes write to the old path format
if (g_network->isSimulated() && deterministicRandom()->coinflip()) {
return writeFile(BackupContainerFileSystemImpl::old_rangeVersionFolderString(fileVersion) + fileName);
}
return writeFile(BackupContainerFileSystemImpl::snapshotFolderString(snapshotBeginVersion) +
format("/%d/", snapshotFileCount / (buggify() ? 1 : 5000)) + fileName);
}
Future<Void> BackupContainerFileSystem::writePartitionListFile(Version v, std::string contents) {
return writeEntireFile(BackupContainerFileSystemImpl::logVersionFolderStringForRangePartitioned(v, v) +
"partitionId_keyRange_map",
contents);
}
Future<std::pair<std::vector<RangeFile>, std::map<std::string, KeyRange>>>
BackupContainerFileSystem::readKeyspaceSnapshot(KeyspaceSnapshotFile snapshot) {
return BackupContainerFileSystemImpl::readKeyspaceSnapshot(Reference<BackupContainerFileSystem>::addRef(this),
snapshot);
}
Future<Void> BackupContainerFileSystem::writeKeyspaceSnapshotFile(const std::vector<std::string>& fileNames,
const std::vector<std::pair<Key, Key>>& beginEndKeys,
int64_t totalBytes,
IncludeKeyRangeMap includeKeyRangeMap,
Optional<SnapshotMetadata> metadata) {
return BackupContainerFileSystemImpl::writeKeyspaceSnapshotFile(Reference<BackupContainerFileSystem>::addRef(this),
fileNames,
beginEndKeys,
totalBytes,
includeKeyRangeMap,
metadata);
};
Future<std::vector<LogFile>> BackupContainerFileSystem::listLogFiles(Version beginVersion,
Version targetVersion,
MutationLogType mutationLogType) {
// The first relevant log file could have a begin version less than beginVersion based on the knobs which
// determine log file range size, so start at an earlier version adjusted by how many versions a file could
// contain.
//
// Get the cleaned (without slashes) first and last folders that could contain relevant results.
std::string firstPath =
BackupContainerFileSystemImpl::cleanFolderString(BackupContainerFileSystemImpl::logVersionFolderString(
std::max<Version>(0,
beginVersion - static_cast<Version>(CLIENT_KNOBS->BACKUP_MAX_LOG_RANGES) *
CLIENT_KNOBS->LOG_RANGE_BLOCK_SIZE),
mutationLogType));
std::string lastPath = BackupContainerFileSystemImpl::cleanFolderString(
BackupContainerFileSystemImpl::logVersionFolderString(targetVersion, mutationLogType));
std::function<bool(std::string const&)> pathFilter = [=](const std::string& folderPath) {
// Remove slashes in the given folder path so that the '/' positions in the version folder string do not
// matter
std::string cleaned = BackupContainerFileSystemImpl::cleanFolderString(folderPath);
return StringRef(firstPath).startsWith(cleaned) || StringRef(lastPath).startsWith(cleaned) ||
(cleaned > firstPath && cleaned < lastPath);
};
return map(listFiles((mutationLogType == MutationLogType::PARTITIONED_LOG ? "plogs/" : "logs/"), pathFilter),
[=, self = Reference<BackupContainerFileSystem>::addRef(this)](const FilesAndSizesT& files) {
std::vector<LogFile> results;
LogFile lf;
for (auto& f : files) {
if (BackupContainerFileSystemImpl::pathToLogFile(lf, f.first, f.second) &&
lf.endVersion > beginVersion && lf.beginVersion <= targetVersion) {
if (self->usesEncryption()) {
lf.fileSize =
AsyncFileEncrypted::rawToLogicalSize(lf.fileSize, self->encryptionBlockSize);
}
results.push_back(lf);
}
}
return results;
});
}
Future<std::vector<RangeFile>> BackupContainerFileSystem::old_listRangeFiles(Version beginVersion, Version endVersion) {
// Get the cleaned (without slashes) first and last folders that could contain relevant results.
std::string firstPath = BackupContainerFileSystemImpl::cleanFolderString(
BackupContainerFileSystemImpl::old_rangeVersionFolderString(beginVersion));
std::string lastPath = BackupContainerFileSystemImpl::cleanFolderString(
BackupContainerFileSystemImpl::old_rangeVersionFolderString(endVersion));
std::function<bool(std::string const&)> pathFilter = [=](const std::string& folderPath) {
// Remove slashes in the given folder path so that the '/' positions in the version folder string do not
// matter
std::string cleaned = BackupContainerFileSystemImpl::cleanFolderString(folderPath);
return StringRef(firstPath).startsWith(cleaned) || StringRef(lastPath).startsWith(cleaned) ||
(cleaned > firstPath && cleaned < lastPath);
};
return map(listFiles("ranges/", pathFilter),
[=, self = Reference<BackupContainerFileSystem>::addRef(this)](const FilesAndSizesT& files) {
std::vector<RangeFile> results;
RangeFile rf;
for (auto& f : files) {
if (BackupContainerFileSystemImpl::pathToRangeFile(rf, f.first, f.second) &&
rf.version >= beginVersion && rf.version <= endVersion) {
if (self->usesEncryption()) {
rf.fileSize =
AsyncFileEncrypted::rawToLogicalSize(rf.fileSize, self->encryptionBlockSize);
}
results.push_back(rf);
}
}
return results;
});
}
Future<std::vector<RangeFile>> BackupContainerFileSystem::listRangeFiles(Version beginVersion, Version endVersion) {
// Until the old folder scheme is no longer supported, read files stored using old folder scheme
Future<std::vector<RangeFile>> oldFiles = old_listRangeFiles(beginVersion, endVersion);
// Define filter function (for listFiles() implementations that use it) to reject any folder
// starting after endVersion
std::function<bool(std::string const&)> pathFilter = [=](std::string const& path) {
return BackupContainerFileSystemImpl::extractSnapshotBeginVersion(path) <= endVersion;
};
Future<std::vector<RangeFile>> newFiles =
map(listFiles("kvranges/", pathFilter),
[=, self = Reference<BackupContainerFileSystem>::addRef(this)](const FilesAndSizesT& files) {
std::vector<RangeFile> results;
RangeFile rf;
for (auto& f : files) {
if (BackupContainerFileSystemImpl::pathToRangeFile(rf, f.first, f.second) &&
rf.version >= beginVersion && rf.version <= endVersion) {
if (self->usesEncryption()) {
rf.fileSize = AsyncFileEncrypted::rawToLogicalSize(rf.fileSize, self->encryptionBlockSize);
}
results.push_back(rf);
}
}
return results;
});
return map(success(oldFiles) && success(newFiles), [=](Void _) {
std::vector<RangeFile> results = newFiles.get();
std::vector<RangeFile> oldResults = oldFiles.get();
results.insert(
results.end(), std::make_move_iterator(oldResults.begin()), std::make_move_iterator(oldResults.end()));
return results;
});
}
Future<std::vector<KeyspaceSnapshotFile>> BackupContainerFileSystem::listKeyspaceSnapshots(Version begin, Version end) {
return map(listFiles("snapshots/"), [=](const FilesAndSizesT& files) {
std::vector<KeyspaceSnapshotFile> results;
KeyspaceSnapshotFile sf;
for (auto& f : files) {
if (BackupContainerFileSystemImpl::pathToKeyspaceSnapshotFile(sf, f.first) && sf.beginVersion < end &&
sf.endVersion >= begin)
results.push_back(sf);
}
std::sort(results.begin(), results.end());
return results;
});
}
Future<BackupFileList> BackupContainerFileSystem::dumpFileList(Version begin, Version end) {
return BackupContainerFileSystemImpl::dumpFileList(Reference<BackupContainerFileSystem>::addRef(this), begin, end);
}
Future<BackupDescription> BackupContainerFileSystem::describeBackup(bool deepScan, Version logStartVersionOverride) {
return BackupContainerFileSystemImpl::describeBackup(
Reference<BackupContainerFileSystem>::addRef(this), deepScan, logStartVersionOverride);
}
Future<Void> BackupContainerFileSystem::expireData(Version expireEndVersion,
bool force,
ExpireProgress* progress,
Version restorableBeginVersion) {
return BackupContainerFileSystemImpl::expireData(
Reference<BackupContainerFileSystem>::addRef(this), expireEndVersion, force, progress, restorableBeginVersion);
}
Future<Void> BackupContainerFileSystem::writeEncryptionMetadata(int encryptionBlockSize) {
return BackupContainerFileSystemImpl::writeEncryptionMetadataIfNotExists(
Reference<BackupContainerFileSystem>::addRef(this), encryptionBlockSize);
}
static Future<KeyRange> getSnapshotFileKeyRange_impl(Reference<BackupContainerFileSystem> bc,
RangeFile file,
Database cx) {
int readFileRetries = 0;
Key beginKey;
Key endKey;
while (true) {
Error err;
try {
Reference<IAsyncFile> inFile = co_await bc->readFile(file.fileName);
bool beginKeySet = false;
for (int64_t j = 0; j < file.fileSize; j += file.blockSize) {
int64_t len = std::min<int64_t>(file.blockSize, file.fileSize - j);
Standalone<VectorRef<KeyValueRef>> blockData =
co_await fileBackup::decodeRangeFileBlock(inFile, j, len, cx);
if (!beginKeySet) {
beginKey = blockData.front().key;
beginKeySet = true;
}
endKey = blockData.back().key;
}
break;
} catch (Error& e) {
err = e;
}
if (err.code() == error_code_restore_bad_read || err.code() == error_code_restore_unsupported_file_version ||
err.code() == error_code_restore_corrupted_data_padding) { // no retriable error
TraceEvent(SevError, "BackupContainerGetSnapshotFileKeyRange").error(err);
throw err;
} else if (err.code() == error_code_http_request_failed || err.code() == error_code_connection_failed ||
err.code() == error_code_timed_out || err.code() == error_code_lookup_failed) {
// blob http request failure, retry
TraceEvent(SevWarnAlways, "BackupContainerGetSnapshotFileKeyRangeConnectionFailure")
.error(err)
.detail("Retries", ++readFileRetries);
co_await delayJittered(0.1);
} else {
TraceEvent(SevError, "BackupContainerGetSnapshotFileKeyRangeUnexpectedError").error(err);
throw err;
}
}
co_return KeyRange(KeyRangeRef(beginKey, endKey));
}
static Future<Void> writeVersionProperty(Reference<BackupContainerFileSystem> bc, std::string path, Version v) {
try {
Reference<IBackupFile> f = co_await bc->writeFile(path);
std::string s = format("%lld", v);
co_await f->append(s.data(), s.size());
co_await f->finish();
} catch (Error& e) {
TraceEvent(SevWarn, "BackupContainerWritePropertyFailed")
.error(e)
.detail("URL", bc->getURL())
.detail("Path", path);
throw;
}
}
static Future<Optional<Version>> readVersionProperty(Reference<BackupContainerFileSystem> bc, std::string path) {
try {
Reference<IAsyncFile> f = co_await bc->readFile(path);
int64_t size = co_await f->size();
std::string s;
s.resize(size);
int rs = co_await f->read((uint8_t*)s.data(), size, 0);
Version v;
int len;
if (rs == size && sscanf(s.c_str(), "%" SCNd64 "%n", &v, &len) == 1 && len == size)
co_return v;
TraceEvent(SevWarn, "BackupContainerInvalidProperty").detail("URL", bc->getURL()).detail("Path", path);
throw backup_invalid_info();
} catch (Error& e) {
if (e.code() == error_code_file_not_found)
co_return Optional<Version>();
TraceEvent(SevWarn, "BackupContainerReadPropertyFailed")
.error(e)
.detail("URL", bc->getURL())
.detail("Path", path);
throw;
}
}
Future<KeyRange> BackupContainerFileSystem::getSnapshotFileKeyRange(const RangeFile& file, Database cx) {
ASSERT(g_network->isSimulated());
return getSnapshotFileKeyRange_impl(Reference<BackupContainerFileSystem>::addRef(this), file, cx);
}
Future<Optional<RestorableFileSet>> BackupContainerFileSystem::getRestoreSet(Version targetVersion,
VectorRef<KeyRangeRef> keyRangesFilter,
bool logsOnly,
Version beginVersion) {
return BackupContainerFileSystemImpl::getRestoreSet(
Reference<BackupContainerFileSystem>::addRef(this), targetVersion, keyRangesFilter, logsOnly, beginVersion);
}
Future<Optional<Version>> BackupContainerFileSystem::VersionProperty::get() {
return readVersionProperty(bc, path);
}
Future<Void> BackupContainerFileSystem::VersionProperty::set(Version v) {
return writeVersionProperty(bc, path, v);
}
Future<Void> BackupContainerFileSystem::VersionProperty::clear() {
return bc->deleteFile(path);
}
BackupContainerFileSystem::VersionProperty BackupContainerFileSystem::logBeginVersion() {
return { Reference<BackupContainerFileSystem>::addRef(this), "log_begin_version" };
}
BackupContainerFileSystem::VersionProperty BackupContainerFileSystem::logEndVersion() {
return { Reference<BackupContainerFileSystem>::addRef(this), "log_end_version" };
}
BackupContainerFileSystem::VersionProperty BackupContainerFileSystem::expiredEndVersion() {
return { Reference<BackupContainerFileSystem>::addRef(this), "expired_end_version" };
}
BackupContainerFileSystem::VersionProperty BackupContainerFileSystem::unreliableEndVersion() {
return { Reference<BackupContainerFileSystem>::addRef(this), "unreliable_end_version" };
}
BackupContainerFileSystem::VersionProperty BackupContainerFileSystem::logType() {
return { Reference<BackupContainerFileSystem>::addRef(this), "mutation_log_type" };
}
std::string BackupContainerFileSystem::encryptionMetadataFileName() {
return "properties/encryption_metadata";
}
bool BackupContainerFileSystem::usesEncryption() const {
return encryptionSetupFuture.isValid();
}
Future<Void> BackupContainerFileSystem::encryptionSetupComplete() const {
return encryptionSetupFuture;
}
Future<Void> BackupContainerFileSystem::writeEntireFileFallback(const std::string& fileName,
const std::string& fileContents) {
return BackupContainerFileSystemImpl::writeEntireFileFallback(
Reference<BackupContainerFileSystem>::addRef(this), fileName, fileContents);
}
void BackupContainerFileSystem::setEncryptionKey(Optional<std::string> const& encryptionKeyFileName) {
if (encryptionKeyFileName.present()) {
encryptionSetupFuture = BackupContainerFileSystemImpl::readEncryptionKey(encryptionKeyFileName.get());
}
}
Future<Void> BackupContainerFileSystem::createTestEncryptionKeyFile(std::string const& filename) {
return BackupContainerFileSystemImpl::createTestEncryptionKeyFile(filename);
}
namespace backup_test {
int chooseFileSize(std::vector<int>& sizes) {
if (!sizes.empty()) {
int size = sizes.back();
sizes.pop_back();
return size;
}
return deterministicRandom()->randomInt(0, 2e6);
}
Future<Void> writeAndVerifyFile(Reference<IBackupContainer> c, Reference<IBackupFile> f, int size, FlowLock* lock) {
Standalone<VectorRef<uint8_t>> content;
co_await lock->take(TaskPriority::DefaultYield, size);
FlowLock::Releaser releaser(*lock, size);
printf("writeAndVerify size=%d file=%s\n", size, f->getFileName().c_str());
content.resize(content.arena(), size);
for (auto& contentByte : content) {
contentByte = (uint8_t)deterministicRandom()->randomInt(0, 256);
}
VectorRef<uint8_t> sendBuf = content;
while (!sendBuf.empty()) {
int n = std::min(sendBuf.size(), deterministicRandom()->randomInt(1, 16384));
co_await f->append(sendBuf.begin(), n);
sendBuf.pop_front(n);
}
co_await f->finish();
Reference<IAsyncFile> inputFile = co_await c->readFile(f->getFileName());
int64_t fileSize = co_await inputFile->size();
ASSERT_EQ(size, fileSize);
if (size > 0) {
Standalone<VectorRef<uint8_t>> buf;
buf.resize(buf.arena(), fileSize);
int b = co_await inputFile->read(buf.begin(), buf.size(), 0);
ASSERT_EQ(b, buf.size());
ASSERT(buf == content);
}
}
// Randomly advance version by up to 1 second of versions
Version nextVersion(Version v) {
int64_t increment = deterministicRandom()->randomInt64(1, CLIENT_KNOBS->CORE_VERSIONSPERSECOND);
return v + increment;
}
// Write a snapshot file with only begin & end key
static Future<Void> testWriteSnapshotFile(Reference<IBackupFile> file, Key begin, Key end, uint32_t blockSize) {
ASSERT_GT(blockSize, 3 * sizeof(uint32_t) + begin.size() + end.size());
uint32_t fileVersion = BACKUP_AGENT_SNAPSHOT_FILE_VERSION;
// write Header
co_await file->append((uint8_t*)&fileVersion, sizeof(fileVersion));
// write begin key length and key
co_await file->appendStringRefWithLen(begin);
// write end key length and key
co_await file->appendStringRefWithLen(end);
int bytesLeft = blockSize - file->size();
if (bytesLeft > 0) {
Value paddings = fileBackup::makePadding(bytesLeft);
co_await file->append(paddings.begin(), bytesLeft);
}
co_await file->finish();
}
Future<Void> testBackupContainer(std::string url,
Optional<std::string> proxy,
Optional<std::string> encryptionKeyFileName) {
FlowLock lock(100e6);
if (encryptionKeyFileName.present()) {
co_await BackupContainerFileSystem::createTestEncryptionKeyFile(encryptionKeyFileName.get());
}
printf("BackupContainerTest URL %s\n", url.c_str());
int encryptionBlockSize = encryptionKeyFileName.present() ? 4096 : 0;
Reference<IBackupContainer> c =
IBackupContainer::openContainer(url, proxy, encryptionKeyFileName, encryptionBlockSize);
// Make sure container doesn't exist, then create it.
try {
co_await c->deleteContainer();
} catch (Error& e) {
if (e.code() != error_code_backup_invalid_url && e.code() != error_code_backup_does_not_exist)
throw;
}
co_await c->create();
std::vector<Future<Void>> writes;
std::map<Version, std::vector<std::string>> snapshots;
std::map<Version, int64_t> snapshotSizes;
std::map<Version, std::vector<std::pair<Key, Key>>> snapshotBeginEndKeys;
int nRangeFiles = 0;
std::map<Version, std::string> logs;
Version v = deterministicRandom()->randomInt64(0, std::numeric_limits<Version>::max() / 2);
// List of sizes to use to test edge cases on underlying file implementations
std::vector<int> fileSizes = { 0 };
if (StringRef(url).startsWith("blob"_sr)) {
fileSizes.push_back(CLIENT_KNOBS->BLOBSTORE_MULTIPART_MIN_PART_SIZE);
fileSizes.push_back(CLIENT_KNOBS->BLOBSTORE_MULTIPART_MIN_PART_SIZE + 10);
}
while (true) {
Version logStart = v;
int kvfiles = deterministicRandom()->randomInt(0, 3);
Key begin = ""_sr;
Key end = ""_sr;
int blockSize = 3 * sizeof(uint32_t) + begin.size() + end.size() + 8;
while (kvfiles > 0) {
if (snapshots.empty()) {
snapshots[v] = {};
snapshotBeginEndKeys[v] = {};
snapshotSizes[v] = 0;
if (deterministicRandom()->coinflip()) {
v = nextVersion(v);
}
}
Reference<IBackupFile> range = co_await c->writeRangeFile(snapshots.rbegin()->first, 0, v, blockSize);
++nRangeFiles;
v = nextVersion(v);
snapshots.rbegin()->second.push_back(range->getFileName());
snapshotBeginEndKeys.rbegin()->second.emplace_back(begin, end);
int size = chooseFileSize(fileSizes);
snapshotSizes.rbegin()->second += size;
// Write in actual range file format, instead of random data.
// writes.push_back(writeAndVerifyFile(c, range, size, &lock));
co_await testWriteSnapshotFile(range, begin, end, blockSize);
if (deterministicRandom()->random01() < .2) {
writes.push_back(c->writeKeyspaceSnapshotFile(snapshots.rbegin()->second,
snapshotBeginEndKeys.rbegin()->second,
snapshotSizes.rbegin()->second,
IncludeKeyRangeMap(buggify())));
snapshots[v] = {};
snapshotBeginEndKeys[v] = {};
snapshotSizes[v] = 0;
break;
}
--kvfiles;
}
if (logStart == v || deterministicRandom()->coinflip()) {
v = nextVersion(v);
}
Reference<IBackupFile> log = co_await c->writeLogFile(logStart, v, 10);
logs[logStart] = log->getFileName();
int size = chooseFileSize(fileSizes);
writes.push_back(writeAndVerifyFile(c, log, size, &lock));
// Randomly stop after a snapshot has finished and all manually seeded file sizes have been used.
if (fileSizes.empty() && !snapshots.empty() && snapshots.rbegin()->second.empty() &&
deterministicRandom()->random01() < .2) {
snapshots.erase(snapshots.rbegin()->first);
break;
}
}
co_await waitForAll(writes);
BackupFileList listing = co_await c->dumpFileList();
ASSERT_EQ(listing.ranges.size(), nRangeFiles);
ASSERT_EQ(listing.logs.size(), logs.size());
ASSERT_EQ(listing.snapshots.size(), snapshots.size());
BackupDescription desc = co_await c->describeBackup();
printf("\n%s\n", desc.toString().c_str());
// Do a series of expirations and verify resulting state
int i = 0;
for (; i < listing.snapshots.size(); ++i) {
{
// Ensure we can still restore to the latest version
Optional<RestorableFileSet> rest = co_await c->getRestoreSet(desc.maxRestorableVersion.get());
ASSERT(rest.present());
}
{
// Ensure we can restore to the end version of snapshot i
Optional<RestorableFileSet> rest = co_await c->getRestoreSet(listing.snapshots[i].endVersion);
ASSERT(rest.present());
}
// Test expiring to the end of this snapshot
Version expireVersion = listing.snapshots[i].endVersion;
// Expire everything up to but not including the snapshot end version
fmt::print("EXPIRE TO {}\n", expireVersion);
Future<Void> f = c->expireData(expireVersion);
co_await ready(f);
// If there is an error, it must be backup_cannot_expire and we have to be on the last snapshot
if (f.isError()) {
ASSERT_EQ(f.getError().code(), error_code_backup_cannot_expire);
ASSERT_EQ(i, listing.snapshots.size() - 1);
co_await c->expireData(expireVersion, true);
}
BackupDescription d = co_await c->describeBackup();
printf("\n%s\n", d.toString().c_str());
}
printf("DELETING\n");
co_await c->deleteContainer();
Future<BackupDescription> d = c->describeBackup();
co_await ready(d);
ASSERT(d.isError() && d.getError().code() == error_code_backup_does_not_exist);
BackupFileList empty = co_await c->dumpFileList();
ASSERT_EQ(empty.ranges.size(), 0);
ASSERT_EQ(empty.logs.size(), 0);
ASSERT_EQ(empty.snapshots.size(), 0);
printf("BackupContainerTest URL=%s PASSED.\n", url.c_str());
}
TEST_CASE("/backup/containers/localdir/unencrypted") {
co_await testBackupContainer(format("file://%s/fdb_backups/%llx", params.getDataDir().c_str(), timer_int()),
Optional<std::string>(),
Optional<std::string>());
}
TEST_CASE("/backup/containers/localdir/encrypted") {
co_await testBackupContainer(format("file://%s/fdb_backups/%llx", params.getDataDir().c_str(), timer_int()),
Optional<std::string>(),
format("%s/test_encryption_key", params.getDataDir().c_str()));
}
TEST_CASE("/backup/containers/localdir/encryptedDescribeWithoutBlockSize") {
std::string url = fmt::format("file://{}/fdb_backups/{:x}", params.getDataDir(), timer_int());
std::string keyFile = fmt::format("{}/test_encryption_key_describe", params.getDataDir());
co_await BackupContainerFileSystem::createTestEncryptionKeyFile(keyFile);
Reference<IBackupContainer> c = IBackupContainer::openContainer(url, {}, keyFile, 4096);
co_await c->create();
co_await c->writeEncryptionMetadata(4096);
Reference<IBackupFile> log = co_await c->writeLogFile(1, 2, 1);
uint8_t value = 1;
co_await log->append(&value, 1);
co_await log->finish();
c->setEncryptionBlockSize(0);
BackupDescription desc = co_await c->describeBackup(true);
ASSERT(desc.fileLevelEncryption);
ASSERT_EQ(desc.encryptionBlockSize, 4096);
ASSERT_EQ(c->getEncryptionBlockSize(), 4096);
co_await c->deleteContainer();
}
TEST_CASE("/backup/containers/url") {
if (!g_network->isSimulated()) {
const char* url = getenv("FDB_TEST_BACKUP_URL");
ASSERT(url != nullptr);
co_await testBackupContainer(url, Optional<std::string>(), Optional<std::string>());
}
}
TEST_CASE("/backup/containers_list") {
if (!g_network->isSimulated()) {
const char* url = getenv("FDB_TEST_BACKUP_URL");
ASSERT(url != nullptr);
printf("Listing %s\n", url);
std::vector<std::string> urls = co_await IBackupContainer::listContainers(url, {});
for (auto& u : urls) {
printf("%s\n", u.c_str());
}
}
}
TEST_CASE("/backup/time") {
// test formatTime()
for (int i = 0; i < 1000; ++i) {
int64_t ts = deterministicRandom()->randomInt64(0, std::numeric_limits<int32_t>::max());
ASSERT(BackupAgentBase::parseTime(BackupAgentBase::formatTime(ts)) == ts);
}
ASSERT(BackupAgentBase::parseTime("2019/03/18.17:51:11-0600") ==
BackupAgentBase::parseTime("2019/03/18.16:51:11-0700"));
ASSERT(BackupAgentBase::parseTime("2019/03/31.22:45:07-0700") ==
BackupAgentBase::parseTime("2019/04/01.03:45:07-0200"));
ASSERT(BackupAgentBase::parseTime("2019/03/31.22:45:07+0000") ==
BackupAgentBase::parseTime("2019/04/01.03:45:07+0500"));
ASSERT(BackupAgentBase::parseTime("2019/03/31.22:45:07+0030") ==
BackupAgentBase::parseTime("2019/04/01.03:45:07+0530"));
ASSERT(BackupAgentBase::parseTime("2019/03/31.22:45:07+0030") ==
BackupAgentBase::parseTime("2019/04/01.04:00:07+0545"));
return Void();
}
TEST_CASE("/backup/continuous") {
std::vector<LogFile> files;
// [0, 100) 2 tags
files.push_back({ 0, 100, 10, "file1", 100, 0, 2 }); // Tag 0: 0-100
ASSERT(!BackupContainerFileSystemImpl::isPartitionedLogsContinuous(files, 0, 99));
ASSERT(BackupContainerFileSystemImpl::getPartitionedLogsContinuousEndVersion(files, 0) == 0);
files.push_back({ 0, 100, 10, "file2", 200, 1, 2 }); // Tag 1: 0-100
std::sort(files.begin(), files.end());
ASSERT(BackupContainerFileSystemImpl::isPartitionedLogsContinuous(files, 0, 99));
ASSERT(!BackupContainerFileSystemImpl::isPartitionedLogsContinuous(files, 0, 100));
ASSERT(BackupContainerFileSystemImpl::getPartitionedLogsContinuousEndVersion(files, 0) == 99);
// [100, 300) 3 tags
files.push_back({ 100, 200, 10, "file3", 200, 0, 3 }); // Tag 0: 100-200
files.push_back({ 100, 250, 10, "file4", 200, 1, 3 }); // Tag 1: 100-250
std::sort(files.begin(), files.end());
ASSERT(BackupContainerFileSystemImpl::isPartitionedLogsContinuous(files, 0, 99));
ASSERT(!BackupContainerFileSystemImpl::isPartitionedLogsContinuous(files, 0, 100));
ASSERT(!BackupContainerFileSystemImpl::isPartitionedLogsContinuous(files, 50, 150));
ASSERT(BackupContainerFileSystemImpl::getPartitionedLogsContinuousEndVersion(files, 0) == 99);
files.push_back({ 100, 300, 10, "file5", 200, 2, 3 }); // Tag 2: 100-300
std::sort(files.begin(), files.end());
ASSERT(BackupContainerFileSystemImpl::isPartitionedLogsContinuous(files, 50, 150));
ASSERT(!BackupContainerFileSystemImpl::isPartitionedLogsContinuous(files, 50, 200));
ASSERT(BackupContainerFileSystemImpl::isPartitionedLogsContinuous(files, 10, 199));
ASSERT(BackupContainerFileSystemImpl::getPartitionedLogsContinuousEndVersion(files, 0) == 199);
ASSERT(BackupContainerFileSystemImpl::getPartitionedLogsContinuousEndVersion(files, 100) == 199);
files.push_back({ 250, 300, 10, "file6", 200, 0, 3 }); // Tag 0: 250-300, missing 200-250
std::sort(files.begin(), files.end());
ASSERT(!BackupContainerFileSystemImpl::isPartitionedLogsContinuous(files, 50, 240));
ASSERT(!BackupContainerFileSystemImpl::isPartitionedLogsContinuous(files, 100, 280));
ASSERT(BackupContainerFileSystemImpl::getPartitionedLogsContinuousEndVersion(files, 99) == 199);
files.push_back({ 250, 300, 10, "file7", 200, 1, 3 }); // Tag 1: 250-300
std::sort(files.begin(), files.end());
ASSERT(!BackupContainerFileSystemImpl::isPartitionedLogsContinuous(files, 100, 280));
files.push_back({ 200, 250, 10, "file8", 200, 0, 3 }); // Tag 0: 200-250
std::sort(files.begin(), files.end());
ASSERT(BackupContainerFileSystemImpl::isPartitionedLogsContinuous(files, 0, 299));
ASSERT(BackupContainerFileSystemImpl::isPartitionedLogsContinuous(files, 100, 280));
ASSERT(BackupContainerFileSystemImpl::getPartitionedLogsContinuousEndVersion(files, 150) == 299);
// [300, 400) 1 tag
// files.push_back({200, 250, 10, "file9", 200, 0, 3}); // Tag 0: 200-250, duplicate file
files.push_back({ 300, 400, 10, "file10", 200, 0, 1 }); // Tag 1: 300-400
std::sort(files.begin(), files.end());
ASSERT(BackupContainerFileSystemImpl::isPartitionedLogsContinuous(files, 0, 399));
ASSERT(BackupContainerFileSystemImpl::isPartitionedLogsContinuous(files, 100, 399));
ASSERT(BackupContainerFileSystemImpl::isPartitionedLogsContinuous(files, 150, 399));
ASSERT(BackupContainerFileSystemImpl::isPartitionedLogsContinuous(files, 250, 399));
ASSERT(BackupContainerFileSystemImpl::getPartitionedLogsContinuousEndVersion(files, 0) == 399);
ASSERT(BackupContainerFileSystemImpl::getPartitionedLogsContinuousEndVersion(files, 99) == 399);
ASSERT(BackupContainerFileSystemImpl::getPartitionedLogsContinuousEndVersion(files, 250) == 399);
return Void();
}
TEST_CASE("/backup/logs_continuous") {
std::vector<LogFile> files;
// [10, 100)
files.push_back({ 10, 100, 10, "file1", 100 });
ASSERT(!BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 0, 5));
ASSERT(!BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 5, 50));
ASSERT(!BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 5, 105));
ASSERT(!BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 100, 101));
ASSERT(!BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 101, 150));
ASSERT(BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 10, 99));
ASSERT(!BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 10, 100));
ASSERT(!BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 10, 101));
ASSERT(!BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 10, 150));
ASSERT(BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 50, 70));
ASSERT(BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 50, 99));
ASSERT(!BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 50, 100));
ASSERT(BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 10, 11));
ASSERT(BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 98, 99));
ASSERT(!BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 99, 100));
ASSERT(BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 99, 99));
ASSERT(!BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 100, 100));
ASSERT(BackupContainerFileSystemImpl::findContinuousLogEnd(files, 0) == invalidVersion);
ASSERT(BackupContainerFileSystemImpl::findContinuousLogEnd(files, 5) == invalidVersion);
ASSERT(BackupContainerFileSystemImpl::findContinuousLogEnd(files, 10) == 100);
ASSERT(BackupContainerFileSystemImpl::findContinuousLogEnd(files, 50) == 100);
ASSERT(BackupContainerFileSystemImpl::findContinuousLogEnd(files, 99) == 100);
ASSERT(BackupContainerFileSystemImpl::findContinuousLogEnd(files, 100) == invalidVersion);
ASSERT(BackupContainerFileSystemImpl::findContinuousLogEnd(files, 101) == invalidVersion);
// [10, 100), [100, 200)
files.push_back({ 100, 200, 10, "file2", 100 });
std::sort(files.begin(), files.end());
ASSERT(!BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 0, 5));
ASSERT(!BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 5, 50));
ASSERT(!BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 5, 105));
ASSERT(BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 100, 101));
ASSERT(BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 101, 150));
ASSERT(BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 10, 99));
ASSERT(BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 10, 100));
ASSERT(BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 10, 101));
ASSERT(BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 10, 150));
ASSERT(BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 50, 70));
ASSERT(BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 50, 99));
ASSERT(BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 50, 100));
ASSERT(BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 10, 11));
ASSERT(BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 98, 99));
ASSERT(BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 99, 100));
ASSERT(BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 99, 99));
ASSERT(BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 100, 100));
ASSERT(!BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 5, 150));
ASSERT(!BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 5, 205));
ASSERT(!BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 200, 201));
ASSERT(!BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 201, 250));
ASSERT(BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 10, 199));
ASSERT(!BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 10, 200));
ASSERT(!BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 10, 201));
ASSERT(BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 70, 170));
ASSERT(!BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 70, 200));
ASSERT(!BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 199, 200));
ASSERT(BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 199, 199));
ASSERT(!BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 200, 200));
ASSERT(BackupContainerFileSystemImpl::findContinuousLogEnd(files, 0) == invalidVersion);
ASSERT(BackupContainerFileSystemImpl::findContinuousLogEnd(files, 5) == invalidVersion);
ASSERT(BackupContainerFileSystemImpl::findContinuousLogEnd(files, 10) == 200);
ASSERT(BackupContainerFileSystemImpl::findContinuousLogEnd(files, 50) == 200);
ASSERT(BackupContainerFileSystemImpl::findContinuousLogEnd(files, 99) == 200);
ASSERT(BackupContainerFileSystemImpl::findContinuousLogEnd(files, 100) == 200);
ASSERT(BackupContainerFileSystemImpl::findContinuousLogEnd(files, 101) == 200);
ASSERT(BackupContainerFileSystemImpl::findContinuousLogEnd(files, 199) == 200);
ASSERT(BackupContainerFileSystemImpl::findContinuousLogEnd(files, 200) == invalidVersion);
ASSERT(BackupContainerFileSystemImpl::findContinuousLogEnd(files, 201) == invalidVersion);
// [10, 100), [100, 200), [300, 400)
files.push_back({ 300, 400, 10, "file3", 100 });
std::sort(files.begin(), files.end());
ASSERT(!BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 0, 5));
ASSERT(!BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 5, 50));
ASSERT(!BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 5, 105));
ASSERT(BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 100, 101));
ASSERT(BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 101, 150));
ASSERT(BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 10, 99));
ASSERT(BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 10, 100));
ASSERT(BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 10, 101));
ASSERT(BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 10, 150));
ASSERT(BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 50, 70));
ASSERT(BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 50, 99));
ASSERT(BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 50, 100));
ASSERT(BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 10, 11));
ASSERT(BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 98, 99));
ASSERT(BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 99, 100));
ASSERT(BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 99, 99));
ASSERT(BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 100, 100));
ASSERT(!BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 5, 150));
ASSERT(!BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 5, 205));
ASSERT(!BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 200, 201));
ASSERT(!BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 201, 250));
ASSERT(BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 10, 199));
ASSERT(!BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 10, 200));
ASSERT(!BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 10, 201));
ASSERT(BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 70, 170));
ASSERT(!BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 70, 200));
ASSERT(!BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 199, 200));
ASSERT(BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 199, 199));
ASSERT(!BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 200, 200));
ASSERT(!BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 250, 260));
ASSERT(!BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 250, 310));
ASSERT(!BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 250, 405));
ASSERT(!BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 400, 401));
ASSERT(!BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 401, 450));
ASSERT(BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 300, 399));
ASSERT(!BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 300, 400));
ASSERT(!BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 300, 401));
ASSERT(BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 300, 350));
ASSERT(BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 350, 370));
ASSERT(!BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 350, 400));
ASSERT(!BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 10, 400));
ASSERT(!BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 100, 400));
ASSERT(!BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 200, 400));
ASSERT(!BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 299, 400));
ASSERT(!BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 399, 400));
ASSERT(BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 399, 399));
ASSERT(!BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(files, 400, 400));
ASSERT(BackupContainerFileSystemImpl::findContinuousLogEnd(files, 0) == invalidVersion);
ASSERT(BackupContainerFileSystemImpl::findContinuousLogEnd(files, 5) == invalidVersion);
ASSERT(BackupContainerFileSystemImpl::findContinuousLogEnd(files, 10) == 200);
ASSERT(BackupContainerFileSystemImpl::findContinuousLogEnd(files, 50) == 200);
ASSERT(BackupContainerFileSystemImpl::findContinuousLogEnd(files, 99) == 200);
ASSERT(BackupContainerFileSystemImpl::findContinuousLogEnd(files, 100) == 200);
ASSERT(BackupContainerFileSystemImpl::findContinuousLogEnd(files, 101) == 200);
ASSERT(BackupContainerFileSystemImpl::findContinuousLogEnd(files, 199) == 200);
ASSERT(BackupContainerFileSystemImpl::findContinuousLogEnd(files, 200) == invalidVersion);
ASSERT(BackupContainerFileSystemImpl::findContinuousLogEnd(files, 201) == invalidVersion);
ASSERT(BackupContainerFileSystemImpl::findContinuousLogEnd(files, 250) == invalidVersion);
ASSERT(BackupContainerFileSystemImpl::findContinuousLogEnd(files, 299) == invalidVersion);
ASSERT(BackupContainerFileSystemImpl::findContinuousLogEnd(files, 300) == 400);
ASSERT(BackupContainerFileSystemImpl::findContinuousLogEnd(files, 301) == 400);
ASSERT(BackupContainerFileSystemImpl::findContinuousLogEnd(files, 399) == 400);
ASSERT(BackupContainerFileSystemImpl::findContinuousLogEnd(files, 400) == invalidVersion);
ASSERT(BackupContainerFileSystemImpl::findContinuousLogEnd(files, 401) == invalidVersion);
ASSERT(BackupContainerFileSystemImpl::findContinuousLogEnd(files, 450) == invalidVersion);
return Void();
}
void printFileList(BackupFileList& backupFileList) {
printf("\nRangeFiles count:%lu", backupFileList.ranges.size());
for (const auto& r : backupFileList.ranges)
printf("\n%s", r.toString().c_str());
printf("\nLogFiles count:%lu", backupFileList.logs.size());
for (const auto& l : backupFileList.logs)
printf("\n%s", l.toString().c_str());
printf("\nSnapshotFiles count:%lu", backupFileList.snapshots.size());
for (const auto& s : backupFileList.snapshots) {
printf("\n%" PRId64 ", %" PRId64 ", %s, %" PRId64 "\n",
s.beginVersion,
s.endVersion,
s.fileName.c_str(),
s.totalSize);
}
}
// Intentionally missing some log range files and checking if the snapshot can be restored.
Future<Void> testBackupContainerWithMissingLogRanges(std::string url, Optional<std::string> proxy) {
FlowLock lock(100e6);
printf("BackupContainerTest URL %s\n", url.c_str());
Reference<IBackupContainer> c =
IBackupContainer::openContainer(url, proxy, /*encryptionKeyFileName=*/{}, /*encryptionBlockSize=*/0);
// Make sure container doesn't exist, then create it.
try {
co_await c->deleteContainer();
} catch (Error& e) {
if (e.code() != error_code_backup_invalid_url && e.code() != error_code_backup_does_not_exist)
throw;
}
co_await c->create();
Key begin = randomKeyBetween(normalKeys);
Key end = randomKeyBetween(KeyRangeRef(begin, normalKeys.end));
int blockSize = 3 * sizeof(uint32_t) + begin.size() + end.size() + 8;
std::vector<Future<Void>> writes;
std::pair<Key, Key> beginEndKeys = std::make_pair(begin, end);
std::vector<bool> snapshotsMissingLogs;
Version v = deterministicRandom()->randomInt64(0, std::numeric_limits<Version>::max() / 2);
Version tempLogEnd = 0;
Version logStart = v;
Version logEnd = v;
Version snapshotBeginVersion = v;
Version snapshotEndVersion = v;
Version lastMissedLogFileEnd = 0;
// create a random number of snapshots
int numSnapshots = deterministicRandom()->randomInt(1, 10);
while (numSnapshots) {
std::vector<std::string> rangeFileNames;
std::vector<std::pair<Key, Key>> snapshotBeginEndKeys;
// create a random number of range files per snapshot
int numRangeFiles = deterministicRandom()->randomInt(2, 5);
snapshotBeginVersion = v;
while (numRangeFiles) {
Reference<IBackupFile> range = co_await c->writeRangeFile(v, 0, v, blockSize);
writes.push_back(writeAndVerifyFile(c, range, deterministicRandom()->randomInt(0, 2e6), &lock));
rangeFileNames.push_back(range->getFileName());
snapshotBeginEndKeys.push_back(beginEndKeys);
logEnd = v;
v = nextVersion(v);
--numRangeFiles;
}
snapshotEndVersion = logEnd;
// writing the snapshot file
writes.push_back(c->writeKeyspaceSnapshotFile(rangeFileNames,
snapshotBeginEndKeys,
deterministicRandom()->randomInt(0, 2e6),
IncludeKeyRangeMap(buggify())));
// if the last missing log file overlaps with the current snapshot,
// mark snapshotsMissingLogs for current snapshot as true.
snapshotsMissingLogs.push_back(lastMissedLogFileEnd > snapshotBeginVersion);
// creating log files for the snapshot range.
while (logStart < logEnd) {
tempLogEnd = nextVersion(logStart);
if (deterministicRandom()->random01() < 0.5) {
Reference<IBackupFile> log = co_await c->writeLogFile(logStart, tempLogEnd, blockSize);
writes.push_back(writeAndVerifyFile(c, log, deterministicRandom()->randomInt(0, 2e6), &lock));
} else { // intentionally missing writing of some log files.
// If the missing log range falls in the current snapshot range, mark it.
if (!(tempLogEnd < snapshotBeginVersion || snapshotEndVersion < logStart))
snapshotsMissingLogs.back() = true;
lastMissedLogFileEnd = tempLogEnd;
}
logStart = tempLogEnd;
}
--numSnapshots;
}
co_await waitForAll(writes);
BackupFileList listing = co_await c->dumpFileList();
printFileList(listing);
printf("\n\nSnapshots missing logs:");
int i = 0;
for (; i < snapshotsMissingLogs.size(); ++i)
printf("\nSnapshot%d: %s", i, snapshotsMissingLogs[i] ? "true" : "false");
BackupDescription desc = co_await c->describeBackup();
printf("\n\n%s\n", desc.toString().c_str());
for (i = 0; i < listing.snapshots.size(); ++i) {
// Ensure we can restore to the end version of snapshot i
Optional<RestorableFileSet> rest = co_await c->getRestoreSet(listing.snapshots[i].endVersion);
if (snapshotsMissingLogs[i])
ASSERT(!rest.present());
else
ASSERT(rest.present());
}
for (i = snapshotsMissingLogs.size() - 1; i >= 0; --i) {
if (!snapshotsMissingLogs[i]) {
int j = i - 1;
for (; j >= 0; --j) {
if (snapshotsMissingLogs[j] ||
!BackupContainerFileSystemImpl::hasContinuousLogsForSnapshot(
listing.logs, listing.snapshots[j].endVersion, listing.snapshots[j + 1].beginVersion))
break;
}
ASSERT(desc.minRestorableVersion.get() == listing.snapshots[j + 1].endVersion);
ASSERT(desc.maxRestorableVersion.get() >= listing.snapshots[i].endVersion);
if (i + 1 < snapshotsMissingLogs.size())
ASSERT(desc.maxRestorableVersion.get() < listing.snapshots[i + 1].endVersion);
break;
}
}
printf("DELETING\n");
co_await c->deleteContainer();
Future<BackupDescription> d = c->describeBackup();
co_await ready(d);
ASSERT(d.isError() && d.getError().code() == error_code_backup_does_not_exist);
BackupFileList empty = co_await c->dumpFileList();
ASSERT_EQ(empty.ranges.size(), 0);
ASSERT_EQ(empty.logs.size(), 0);
ASSERT_EQ(empty.snapshots.size(), 0);
printf("BackupContainerTest URL=%s PASSED.\n", url.c_str());
}
TEST_CASE("/backup/containers/localdir/missingLogRangesRestorability") {
co_await testBackupContainerWithMissingLogRanges(
format("file://%s/fdb_backups/%llx", params.getDataDir().c_str(), timer_int()), Optional<std::string>());
}
Future<Void> testBackupContinuousLogEndVer(std::string url, Optional<std::string> proxy) {
FlowLock lock(100e6);
printf("BackupContainerTest URL %s\n", url.c_str());
Reference<IBackupContainer> c =
IBackupContainer::openContainer(url, proxy, /*encryptionKeyFileName=*/{}, /*encryptionBlockSize=*/0);
// Make sure container doesn't exist, then create it.
try {
co_await c->deleteContainer();
} catch (Error& e) {
if (e.code() != error_code_backup_invalid_url && e.code() != error_code_backup_does_not_exist)
throw;
}
co_await c->create();
int blockSize = 1024;
std::vector<std::string> rangeFileNames;
Key begin = randomKeyBetween(normalKeys);
Key end = randomKeyBetween(KeyRangeRef(begin, normalKeys.end));
std::pair<Key, Key> beginEndKeys = std::make_pair(begin, end);
std::vector<std::pair<Key, Key>> snapshotBeginEndKeys;
// writing random number of range files with rangeSize 100
std::vector<Future<Void>> writes;
Version snapshotBeginVersion = 10;
Version snapshotEndVersion = deterministicRandom()->randomInt(500, 1000);
Version rangeSize = 100;
Version v = snapshotBeginVersion;
int numRangeFiles = 0;
while (v <= snapshotEndVersion) {
Reference<IBackupFile> range = co_await c->writeRangeFile(v, 0, v, blockSize);
writes.push_back(writeAndVerifyFile(c, range, 100, &lock));
rangeFileNames.push_back(range->getFileName());
snapshotBeginEndKeys.push_back(beginEndKeys);
v += rangeSize;
++numRangeFiles;
}
snapshotEndVersion = v - rangeSize;
// writing random number of log files with logSize 70, covering the entire snapshot
Version logSize = 70;
v = snapshotBeginVersion;
int numLogFiles = 0;
while (v <= snapshotEndVersion) {
Reference<IBackupFile> log = co_await c->writeLogFile(v, v + logSize, blockSize);
writes.push_back(writeAndVerifyFile(c, log, 100, &lock));
++numLogFiles;
v += logSize;
}
// writing snapshot file
writes.push_back(c->writeKeyspaceSnapshotFile(
rangeFileNames, snapshotBeginEndKeys, deterministicRandom()->randomInt(0, 2e6), IncludeKeyRangeMap(buggify())));
co_await waitForAll(writes);
BackupFileList fileList = co_await c->dumpFileList();
printFileList(fileList);
ASSERT_EQ(fileList.ranges.size(), numRangeFiles);
ASSERT_EQ(fileList.logs.size(), numLogFiles);
ASSERT_EQ(fileList.snapshots.size(), 1);
BackupDescription desc = co_await c->describeBackup();
printf("\n%s\n", desc.toString().c_str());
ASSERT_EQ(desc.minLogBegin, snapshotBeginVersion);
ASSERT_EQ(desc.maxLogEnd, v);
ASSERT_EQ(desc.minRestorableVersion, snapshotEndVersion);
ASSERT_EQ(desc.maxRestorableVersion, v - 1);
ASSERT_EQ(desc.snapshots[0].restorable, true);
ASSERT_EQ(desc.contiguousLogEnd, v);
// writing random number of more continuous log files
int newNumLogFiles = deterministicRandom()->randomInt(2, 8);
numLogFiles += newNumLogFiles;
writes.clear();
while (newNumLogFiles) {
Reference<IBackupFile> log = co_await c->writeLogFile(v, v + logSize, blockSize);
writes.push_back(writeAndVerifyFile(c, log, 100, &lock));
--newNumLogFiles;
v += logSize;
}
co_await waitForAll(writes);
BackupFileList fileList1 = co_await c->dumpFileList();
printFileList(fileList1);
ASSERT_EQ(fileList1.ranges.size(), numRangeFiles);
ASSERT_EQ(fileList1.logs.size(), numLogFiles);
ASSERT_EQ(fileList1.snapshots.size(), 1);
BackupDescription desc1 = co_await c->describeBackup();
printf("\n%s\n", desc1.toString().c_str());
ASSERT_EQ(desc1.minLogBegin, snapshotBeginVersion);
ASSERT_EQ(desc1.maxLogEnd, v);
ASSERT_EQ(desc1.minRestorableVersion, snapshotEndVersion);
ASSERT_EQ(desc1.maxRestorableVersion, v - 1);
ASSERT_EQ(desc1.snapshots[0].restorable, true);
ASSERT_EQ(desc1.contiguousLogEnd, v);
}
TEST_CASE("/backup/containers/localdir/continuousLogEndVersion") {
co_await testBackupContinuousLogEndVer(
format("file://%s/fdb_backups/%llx", params.getDataDir().c_str(), timer_int()), Optional<std::string>());
}
// Verifies that IBackupContainer::ExpireProgress reports both the requested expire version and
// the version expiration was actually performed to, in the two cases where they can differ:
// 1. The requested version falls inside a log file, so the actual version is rolled back to
// the log file's begin version to avoid splitting it.
// 2. The requested version is at or before a version that was already expired by a prior call,
// so nothing new is deleted and the actual version reflects the prior expiration.
Future<Void> testExpireProgressVersions(std::string url, Optional<std::string> proxy) {
FlowLock lock(100e6);
printf("BackupContainerTest URL %s\n", url.c_str());
Reference<IBackupContainer> c = IBackupContainer::openContainer(url, proxy, {}, 0);
// Make sure container doesn't exist, then create it.
try {
co_await c->deleteContainer();
} catch (Error& e) {
if (e.code() != error_code_backup_invalid_url && e.code() != error_code_backup_does_not_exist)
throw;
}
co_await c->create();
int blockSize = 1024;
Key begin = randomKeyBetween(normalKeys);
Key end = randomKeyBetween(KeyRangeRef(begin, normalKeys.end));
std::vector<Future<Void>> writes;
// A single-file snapshot ending well before the log file below.
Version snapshotVersion = 100;
Reference<IBackupFile> range = co_await c->writeRangeFile(snapshotVersion, 0, snapshotVersion, blockSize);
writes.push_back(writeAndVerifyFile(c, range, 100, &lock));
writes.push_back(c->writeKeyspaceSnapshotFile(
{ range->getFileName() }, { std::make_pair(begin, end) }, 100, IncludeKeyRangeMap(buggify())));
// A single log file spanning a wide version range that straddles the version we're about to
// request expiring to. A log file can't be partially deleted, so expireData() must roll the
// actual expiration point back to this file's begin version.
Version logBegin = 10;
Version logEnd = 1000;
Reference<IBackupFile> log = co_await c->writeLogFile(logBegin, logEnd, blockSize);
writes.push_back(writeAndVerifyFile(c, log, 100, &lock));
co_await waitForAll(writes);
BackupDescription desc = co_await c->describeBackup();
printf("\n%s\n", desc.toString().c_str());
ASSERT_EQ(desc.snapshots.size(), 1);
ASSERT_EQ(desc.snapshots[0].endVersion, snapshotVersion);
ASSERT_EQ(desc.maxLogEnd, logEnd);
// Case 1: request an expire version strictly inside the log file's range. The actual
// expiration point must be rolled back to the log file's begin version, which is before
// the snapshot's end version, so the snapshot must survive.
Version requestedVersion = logBegin + (logEnd - logBegin) / 2;
IBackupContainer::ExpireProgress progress1;
co_await c->expireData(requestedVersion, true, &progress1);
fmt::print("Case 1: requested={} actual={}\n", progress1.requestedEndVersion, progress1.actualEndVersion);
ASSERT_EQ(progress1.requestedEndVersion, requestedVersion);
ASSERT_EQ(progress1.actualEndVersion, logBegin);
ASSERT_LT(progress1.actualEndVersion, progress1.requestedEndVersion);
BackupDescription desc1 = co_await c->describeBackup();
printf("\n%s\n", desc1.toString().c_str());
ASSERT_EQ(desc1.snapshots.size(), 1); // Snapshot must not have been deleted.
// Case 2: request an expire version at or before what has already been expired. No new data
// can be deleted, so the actual version must reflect the version already achieved by the
// prior expiration, not the newly (smaller) requested version.
Version smallerVersion = 1;
IBackupContainer::ExpireProgress progress2;
co_await c->expireData(smallerVersion, true, &progress2);
fmt::print("Case 2: requested={} actual={}\n", progress2.requestedEndVersion, progress2.actualEndVersion);
ASSERT_EQ(progress2.requestedEndVersion, smallerVersion);
ASSERT_EQ(progress2.actualEndVersion, progress1.actualEndVersion);
ASSERT_GE(progress2.actualEndVersion, progress2.requestedEndVersion);
}
TEST_CASE("/backup/containers/localdir/expireProgressVersions") {
co_await testExpireProgressVersions(format("file://%s/fdb_backups/%llx", params.getDataDir().c_str(), timer_int()),
Optional<std::string>());
}
// Verify that writeKeyspaceSnapshotFile correctly writes and reads back a snapshot manifest even when the
// JSON document is larger than BACKUP_MANIFEST_CHUNK_SIZE, exercising the chunked-append path.
TEST_CASE("/backup/containers/localdir/writeKeyspaceSnapshotFile/chunked") {
// Force a tiny chunk size so a normal-sized manifest triggers multiple append() calls.
int savedChunkSize = CLIENT_KNOBS->BACKUP_MANIFEST_CHUNK_SIZE;
const_cast<ClientKnobs*>(CLIENT_KNOBS)->BACKUP_MANIFEST_CHUNK_SIZE = 64;
ASSERT_EQ(CLIENT_KNOBS->BACKUP_MANIFEST_CHUNK_SIZE, 64);
std::string url = format("file://%s/fdb_backups/%llx", params.getDataDir().c_str(), timer_int());
Reference<IBackupContainer> c = IBackupContainer::openContainer(url, {}, {}, 0);
co_await c->create();
// Use a fixed version and block size for deterministic assertions.
Version v = 1000;
int blockSize = 3 * sizeof(uint32_t) + 8;
// Write several range files so the resulting JSON manifest exceeds 64 bytes.
std::vector<std::string> rangeFileNames;
std::vector<std::pair<Key, Key>> beginEndKeys;
for (int i = 0; i < 5; ++i) {
Reference<IBackupFile> range = co_await c->writeRangeFile(v, 0, v, blockSize);
co_await testWriteSnapshotFile(range, ""_sr, ""_sr, blockSize);
rangeFileNames.push_back(range->getFileName());
beginEndKeys.push_back({ ""_sr, ""_sr });
++v;
}
int64_t totalSize = 99999;
co_await c->writeKeyspaceSnapshotFile(rangeFileNames, beginEndKeys, totalSize, IncludeKeyRangeMap(false));
BackupFileList listing = co_await c->dumpFileList();
ASSERT_EQ(listing.snapshots.size(), 1);
ASSERT_EQ(listing.snapshots[0].totalSize, totalSize);
ASSERT_EQ(listing.snapshots[0].beginVersion, 1000);
ASSERT_EQ(listing.snapshots[0].endVersion, 1004);
const_cast<ClientKnobs*>(CLIENT_KNOBS)->BACKUP_MANIFEST_CHUNK_SIZE = savedChunkSize;
co_await c->deleteContainer();
}
// Verify that readKeyspaceSnapshot correctly reassembles and parses a snapshot manifest when it is read
// back in many small pieces, exercising the chunked-read path. A tiny chunk size (that does not divide the
// manifest evenly) forces the read loop to run many iterations with a partial final chunk, which catches
// off-by-one / wrong-offset / short-read bugs in the loop. Note: a unit test cannot allocate a >2 GB
// manifest to reproduce the original int overflow, so this validates the chunking logic instead.
TEST_CASE("/backup/containers/localdir/readKeyspaceSnapshot/chunked") {
int savedChunkSize = CLIENT_KNOBS->BACKUP_MANIFEST_CHUNK_SIZE;
const_cast<ClientKnobs*>(CLIENT_KNOBS)->BACKUP_MANIFEST_CHUNK_SIZE = 7;
std::string url = format("file://%s/fdb_backups/%llx", params.getDataDir().c_str(), timer_int());
Reference<IBackupContainer> c = IBackupContainer::openContainer(url, {}, {}, 0);
co_await c->create();
Version v = 1000;
int blockSize = 64;
// Write several range files with distinct, non-empty key ranges so the manifest also contains a
// populated keyRanges section (exercising that part of the read path too).
std::vector<std::string> rangeFileNames;
std::vector<std::pair<Key, Key>> beginEndKeys;
std::map<std::string, std::pair<std::string, std::string>> expected;
for (int i = 0; i < 5; ++i) {
Key begin = StringRef(format("begin-%d", i));
Key end = StringRef(format("end-%d", i));
Reference<IBackupFile> range = co_await c->writeRangeFile(v, 0, v, blockSize);
co_await testWriteSnapshotFile(range, begin, end, blockSize);
rangeFileNames.push_back(range->getFileName());
beginEndKeys.push_back({ begin, end });
expected[range->getFileName()] = { begin.toString(), end.toString() };
++v;
}
int64_t totalSize = 99999;
co_await c->writeKeyspaceSnapshotFile(rangeFileNames, beginEndKeys, totalSize, IncludeKeyRangeMap::True);
// Read the manifest back through the chunked-read path and verify every range file and key range.
Reference<BackupContainerFileSystem> bcfs = c.castTo<BackupContainerFileSystem>();
std::vector<KeyspaceSnapshotFile> snapshots = co_await bcfs->listKeyspaceSnapshots();
ASSERT_EQ(snapshots.size(), 1);
auto [files, keyRanges] = co_await bcfs->readKeyspaceSnapshot(snapshots[0]);
ASSERT_EQ(files.size(), rangeFileNames.size());
ASSERT_EQ(keyRanges.size(), expected.size());
for (const auto& [fileName, range] : expected) {
auto it = keyRanges.find(fileName);
ASSERT(it != keyRanges.end());
ASSERT(it->second.begin == StringRef(range.first));
ASSERT(it->second.end == StringRef(range.second));
}
const_cast<ClientKnobs*>(CLIENT_KNOBS)->BACKUP_MANIFEST_CHUNK_SIZE = savedChunkSize;
co_await c->deleteContainer();
}
} // namespace backup_test
Reference in New Issue View Git Blame Copy Permalink