Packages

RocksDB Cloud for Erlang

Current section

Files

Jump to
rocksdb_cloud deps rocksdb cloud db_cloud_test.cc
Raw

deps/rocksdb/cloud/db_cloud_test.cc

// Copyright (c) 2017 Rockset
#ifndef ROCKSDB_LITE
#ifdef USE_AWS
#include "rocksdb/cloud/db_cloud.h"
#include <algorithm>
#include <chrono>
#include "cloud/aws/aws_env.h"
#include "cloud/aws/aws_file.h"
#include "cloud/db_cloud_impl.h"
#include "cloud/filename.h"
#include "cloud/manifest_reader.h"
#include "rocksdb/options.h"
#include "rocksdb/status.h"
#include "util/filename.h"
#include "util/logging.h"
#include "util/string_util.h"
#include "util/testharness.h"
#ifndef OS_WIN
#include <unistd.h>
#endif
namespace rocksdb {
class CloudTest : public testing::Test {
public:
CloudTest() {
base_env_ = Env::Default();
dbname_ = test::TmpDir() + "/db_cloud";
clone_dir_ = test::TmpDir() + "/ctest";
src_bucket_prefix_ = "dbcloud." + AwsEnv::GetTestBucketSuffix();
src_object_prefix_ = dbname_;
// Set dest = src by default
dest_bucket_prefix_ = src_bucket_prefix_;
dest_object_prefix_ = src_object_prefix_;
options_.create_if_missing = true;
persistent_cache_path_ = "";
persistent_cache_size_gb_ = 0;
db_ = nullptr;
DestroyDir(dbname_);
base_env_->CreateDirIfMissing(dbname_);
base_env_->NewLogger(test::TmpDir(base_env_) + "/rocksdb-cloud.log",
&options_.info_log);
options_.info_log->SetInfoLogLevel(InfoLogLevel::DEBUG_LEVEL);
// Get cloud credentials
AwsEnv::GetTestCredentials(&cloud_env_options_.credentials.access_key_id,
&cloud_env_options_.credentials.secret_key,
&region_);
Cleanup();
}
void Cleanup() {
ASSERT_TRUE(!aenv_);
CloudEnv* aenv;
// create a dummy aws env
ASSERT_OK(CloudEnv::NewAwsEnv(
base_env_, src_bucket_prefix_, src_object_prefix_, region_,
dest_bucket_prefix_, dest_object_prefix_, region_, cloud_env_options_,
options_.info_log, &aenv));
aenv_.reset(aenv);
// delete all pre-existing contents from the bucket
Status st = aenv_->EmptyBucket(src_bucket_prefix_, "");
ASSERT_TRUE(st.ok() || st.IsNotFound());
aenv_.reset();
// delete and create directory where clones reside
DestroyDir(clone_dir_);
ASSERT_OK(base_env_->CreateDir(clone_dir_));
}
std::set<std::string> GetSSTFiles(std::string name) {
std::vector<std::string> files;
aenv_->GetBaseEnv()->GetChildren(name, &files);
std::set<std::string> sst_files;
for (auto& f : files) {
if (IsSstFile(RemoveEpoch(f))) {
sst_files.insert(f);
}
}
return sst_files;
}
std::set<std::string> GetSSTFilesClone(std::string name) {
std::string cname = clone_dir_ + "/" + name;
return GetSSTFiles(cname);
}
void DestroyDir(const std::string& dir) {
std::string cmd = "rm -rf " + dir;
int rc = system(cmd.c_str());
ASSERT_EQ(rc, 0);
}
virtual ~CloudTest() { CloseDB(); }
void CreateAwsEnv() {
CloudEnv* aenv;
ASSERT_OK(CloudEnv::NewAwsEnv(
base_env_, src_bucket_prefix_, src_object_prefix_, region_,
dest_bucket_prefix_, dest_object_prefix_, region_, cloud_env_options_,
options_.info_log, &aenv));
// To catch any possible file deletion bugs, we set file deletion delay to
// smallest possible
if (!src_bucket_prefix_.empty() || !dest_bucket_prefix_.empty()) {
((AwsEnv*)aenv)->TEST_SetFileDeletionDelay(std::chrono::seconds(0));
}
aenv_.reset(aenv);
}
// Open database via the cloud interface
void OpenDB() {
ASSERT_NE(cloud_env_options_.credentials.access_key_id.size(), 0);
ASSERT_NE(cloud_env_options_.credentials.secret_key.size(), 0);
// Create new AWS env
CreateAwsEnv();
options_.env = aenv_.get();
// default column family
ColumnFamilyOptions cfopt = options_;
std::vector<ColumnFamilyDescriptor> column_families;
column_families.emplace_back(
ColumnFamilyDescriptor(kDefaultColumnFamilyName, cfopt));
std::vector<ColumnFamilyHandle*> handles;
ASSERT_TRUE(db_ == nullptr);
ASSERT_OK(DBCloud::Open(options_, dbname_, column_families,
persistent_cache_path_, persistent_cache_size_gb_,
&handles, &db_));
ASSERT_OK(db_->GetDbIdentity(dbid_));
// Delete the handle for the default column family because the DBImpl
// always holds a reference to it.
ASSERT_TRUE(handles.size() > 0);
delete handles[0];
}
// Creates and Opens a clone
void CloneDB(const std::string& clone_name, const std::string& src_bucket,
const std::string& src_object_path,
const std::string& dest_bucket,
const std::string& dest_object_path,
std::unique_ptr<DBCloud>* cloud_db,
std::unique_ptr<CloudEnv>* cloud_env) {
// The local directory where the clone resides
std::string cname = clone_dir_ + "/" + clone_name;
CloudEnv* cenv;
DBCloud* clone_db;
// If there is no destination bucket, then the clone needs to copy
// all sst fies from source bucket to local dir
CloudEnvOptions copt = cloud_env_options_;
if (dest_bucket.empty()) {
copt.keep_local_sst_files = true;
}
// Create new AWS env
ASSERT_OK(CloudEnv::NewAwsEnv(base_env_, src_bucket, src_object_path,
region_, dest_bucket, dest_object_path,
region_, copt, options_.info_log, &cenv));
// To catch any possible file deletion bugs, we set file deletion delay to
// smallest possible
((AwsEnv*)cenv)->TEST_SetFileDeletionDelay(std::chrono::seconds(0));
// sets the cloud env to be used by the env wrapper
options_.env = cenv;
// Returns the cloud env that was created
cloud_env->reset(cenv);
// default column family
ColumnFamilyOptions cfopt = options_;
std::vector<ColumnFamilyDescriptor> column_families;
column_families.emplace_back(
ColumnFamilyDescriptor(kDefaultColumnFamilyName, cfopt));
std::vector<ColumnFamilyHandle*> handles;
ASSERT_OK(DBCloud::Open(options_, cname, column_families,
persistent_cache_path_, persistent_cache_size_gb_,
&handles, &clone_db));
cloud_db->reset(clone_db);
// Delete the handle for the default column family because the DBImpl
// always holds a reference to it.
ASSERT_TRUE(handles.size() > 0);
delete handles[0];
}
void CloseDB() {
if (db_) {
db_->Flush(FlushOptions()); // convert pending writes to sst files
delete db_;
db_ = nullptr;
}
}
void SetPersistentCache(const std::string& path, uint64_t size_gb) {
persistent_cache_path_ = path;
persistent_cache_size_gb_ = size_gb;
}
Status GetCloudLiveFilesSrc(std::set<uint64_t>* list) {
std::unique_ptr<ManifestReader> manifest(
new ManifestReader(options_.info_log, aenv_.get(), src_bucket_prefix_));
return manifest->GetLiveFiles(src_object_prefix_, list);
}
// Verify that local files are the same as cloud files in src bucket path
void ValidateCloudLiveFilesSrcSize() {
// Loop though all the files in the cloud manifest
std::set<uint64_t> cloud_files;
ASSERT_OK(GetCloudLiveFilesSrc(&cloud_files));
for (uint64_t num: cloud_files) {
std::string pathname = MakeTableFileName(dbname_, num);
Log(options_.info_log, "cloud file list %s\n", pathname.c_str());
}
std::set<std::string> localFiles = GetSSTFiles(dbname_);
uint64_t cloudSize = 0;
uint64_t localSize = 0;
// loop through all the local files and validate
for (std::string path: localFiles) {
std::string cpath = src_object_prefix_ + "/" + path;
ASSERT_OK(aenv_->GetObjectSize(src_bucket_prefix_, cpath, &cloudSize));
// find the size of the file on local storage
std::string lpath = dbname_ + "/" + path;
ASSERT_OK(aenv_->GetBaseEnv()->GetFileSize(lpath, &localSize));
ASSERT_TRUE(localSize == cloudSize);
Log(options_.info_log, "local file %s size %ld\n", lpath.c_str(), localSize);
Log(options_.info_log, "cloud file %s size %ld\n", cpath.c_str(), cloudSize);
printf("local file %s size %ld\n", lpath.c_str(), localSize);
printf("cloud file %s size %ld\n", cpath.c_str(), cloudSize);
}
}
protected:
Env* base_env_;
Options options_;
std::string dbname_;
std::string clone_dir_;
std::string src_bucket_prefix_;
std::string src_object_prefix_;
std::string dest_bucket_prefix_;
std::string dest_object_prefix_;
CloudEnvOptions cloud_env_options_;
std::string region_;
std::string dbid_;
std::string persistent_cache_path_;
uint64_t persistent_cache_size_gb_;
DBCloud* db_;
unique_ptr<CloudEnv> aenv_;
};
//
// Most basic test. Create DB, write one key, close it and then check to see
// that the key exists.
//
TEST_F(CloudTest, BasicTest) {
// Put one key-value
OpenDB();
std::string value;
ASSERT_OK(db_->Put(WriteOptions(), "Hello", "World"));
ASSERT_OK(db_->Get(ReadOptions(), "Hello", &value));
ASSERT_TRUE(value.compare("World") == 0);
CloseDB();
value.clear();
// Reopen and validate
OpenDB();
ASSERT_OK(db_->Get(ReadOptions(), "Hello", &value));
ASSERT_EQ(value, "World");
std::set<uint64_t> live_files;
ASSERT_OK(GetCloudLiveFilesSrc(&live_files));
ASSERT_GT(live_files.size(), 0);
CloseDB();
}
TEST_F(CloudTest, GetChildrenTest) {
// Create some objects in S3
OpenDB();
ASSERT_OK(db_->Put(WriteOptions(), "Hello", "World"));
ASSERT_OK(db_->Flush(FlushOptions()));
CloseDB();
DestroyDir(dbname_);
OpenDB();
std::vector<std::string> children;
ASSERT_OK(aenv_->GetChildren(dbname_, &children));
int sst_files = 0;
for (auto c : children) {
if (IsSstFile(c)) {
sst_files++;
}
}
// This verifies that GetChildren() works on S3. We deleted the S3 file
// locally, so the only way to actually get it through GetChildren() if
// listing S3 buckets works.
EXPECT_EQ(sst_files, 1);
}
//
// Create and read from a clone.
//
TEST_F(CloudTest, Newdb) {
std::string master_dbid;
std::string newdb1_dbid;
std::string newdb2_dbid;
// Put one key-value
OpenDB();
std::string value;
ASSERT_OK(db_->Put(WriteOptions(), "Hello", "World"));
ASSERT_OK(db_->Get(ReadOptions(), "Hello", &value));
ASSERT_TRUE(value.compare("World") == 0);
ASSERT_OK(db_->GetDbIdentity(master_dbid));
CloseDB();
value.clear();
{
// Create and Open a new ephemeral instance
std::unique_ptr<CloudEnv> cloud_env;
std::unique_ptr<DBCloud> cloud_db;
CloneDB("newdb1", src_bucket_prefix_, src_object_prefix_, "", "", &cloud_db,
&cloud_env);
// Retrieve the id of the first reopen
ASSERT_OK(cloud_db->GetDbIdentity(newdb1_dbid));
// This is an ephemeral clone. Its dbid is a prefix of the master's.
ASSERT_NE(newdb1_dbid, master_dbid);
auto res = std::mismatch(master_dbid.begin(), master_dbid.end(),
newdb1_dbid.begin());
ASSERT_TRUE(res.first == master_dbid.end());
ASSERT_OK(cloud_db->Get(ReadOptions(), "Hello", &value));
ASSERT_TRUE(value.compare("World") == 0);
// Open master and write one more kv to it. This is written to
// src bucket as well.
OpenDB();
ASSERT_OK(db_->Put(WriteOptions(), "Dhruba", "Borthakur"));
// check that the newly written kv exists
value.clear();
ASSERT_OK(db_->Get(ReadOptions(), "Dhruba", &value));
ASSERT_TRUE(value.compare("Borthakur") == 0);
// check that the earlier kv exists too
value.clear();
ASSERT_OK(db_->Get(ReadOptions(), "Hello", &value));
ASSERT_TRUE(value.compare("World") == 0);
CloseDB();
// Assert that newdb1 cannot see the second kv because the second kv
// was written to local dir only of the ephemeral clone.
ASSERT_TRUE(cloud_db->Get(ReadOptions(), "Dhruba", &value).IsNotFound());
}
{
// Create another ephemeral instance using a different local dir but the same two
// buckets as newdb1. This should be identical in contents with newdb1.
std::unique_ptr<CloudEnv> cloud_env;
std::unique_ptr<DBCloud> cloud_db;
CloneDB("newdb2", src_bucket_prefix_, src_object_prefix_, "", "", &cloud_db,
&cloud_env);
// Retrieve the id of the second clone db
ASSERT_OK(cloud_db->GetDbIdentity(newdb2_dbid));
// Since we use two different local directories for the two ephemeral
// clones, their dbids should be different from one another
ASSERT_NE(newdb1_dbid, newdb2_dbid);
// check that both the kvs appear in the new ephemeral clone
value.clear();
ASSERT_OK(cloud_db->Get(ReadOptions(), "Hello", &value));
ASSERT_TRUE(value.compare("World") == 0);
value.clear();
ASSERT_OK(cloud_db->Get(ReadOptions(), "Dhruba", &value));
ASSERT_TRUE(value.compare("Borthakur") == 0);
}
CloseDB();
}
//
// Create and read from a clone.
//
TEST_F(CloudTest, TrueClone) {
std::string master_dbid;
std::string newdb1_dbid;
std::string newdb2_dbid;
std::string newdb3_dbid;
// Put one key-value
OpenDB();
std::string value;
ASSERT_OK(db_->Put(WriteOptions(), "Hello", "World"));
ASSERT_OK(db_->Get(ReadOptions(), "Hello", &value));
ASSERT_TRUE(value.compare("World") == 0);
ASSERT_OK(db_->GetDbIdentity(master_dbid));
ASSERT_OK(db_->Flush(FlushOptions()));
CloseDB();
value.clear();
{
// Create a new instance with different src and destination paths.
// This is true clone and should have all the contents of the masterdb
std::unique_ptr<CloudEnv> cloud_env;
std::unique_ptr<DBCloud> cloud_db;
CloneDB("localpath1", src_bucket_prefix_, src_object_prefix_,
src_bucket_prefix_, "clone1_path", &cloud_db, &cloud_env);
// Retrieve the id of the clone db
ASSERT_OK(cloud_db->GetDbIdentity(newdb1_dbid));
// Since we used the different src and destination paths for both
// the master and clone1, the clone should have its own identity.
ASSERT_NE(master_dbid, newdb1_dbid);
// check that the original kv appears in the clone
value.clear();
ASSERT_OK(cloud_db->Get(ReadOptions(), "Hello", &value));
ASSERT_TRUE(value.compare("World") == 0);
// write a new value to the clone
ASSERT_OK(cloud_db->Put(WriteOptions(), "Hello", "Clone1"));
value.clear();
ASSERT_OK(cloud_db->Get(ReadOptions(), "Hello", &value));
ASSERT_TRUE(value.compare("Clone1") == 0);
ASSERT_OK(cloud_db->Flush(FlushOptions()));
}
{
// Reopen clone1 with a different local path
std::unique_ptr<CloudEnv> cloud_env;
std::unique_ptr<DBCloud> cloud_db;
CloneDB("localpath2", src_bucket_prefix_, src_object_prefix_,
src_bucket_prefix_, "clone1_path", &cloud_db, &cloud_env);
// Retrieve the id of the clone db
ASSERT_OK(cloud_db->GetDbIdentity(newdb2_dbid));
ASSERT_EQ(newdb1_dbid, newdb2_dbid);
value.clear();
ASSERT_OK(cloud_db->Get(ReadOptions(), "Hello", &value));
ASSERT_TRUE(value.compare("Clone1") == 0);
ASSERT_OK(cloud_db->Flush(FlushOptions()));
}
{
// Reopen clone1 with the same local path as above.
std::unique_ptr<CloudEnv> cloud_env;
std::unique_ptr<DBCloud> cloud_db;
CloneDB("localpath2", src_bucket_prefix_, src_object_prefix_,
src_bucket_prefix_, "clone1_path", &cloud_db, &cloud_env);
// Retrieve the id of the clone db
ASSERT_OK(cloud_db->GetDbIdentity(newdb2_dbid));
ASSERT_EQ(newdb1_dbid, newdb2_dbid);
value.clear();
ASSERT_OK(cloud_db->Get(ReadOptions(), "Hello", &value));
ASSERT_TRUE(value.compare("Clone1") == 0);
ASSERT_OK(cloud_db->Flush(FlushOptions()));
}
{
// Create clone2
std::unique_ptr<CloudEnv> cloud_env;
std::unique_ptr<DBCloud> cloud_db;
CloneDB("localpath3", // xxx try with localpath2
src_bucket_prefix_, src_object_prefix_, src_bucket_prefix_,
"clone2_path", &cloud_db, &cloud_env);
// Retrieve the id of the clone db
ASSERT_OK(cloud_db->GetDbIdentity(newdb3_dbid));
ASSERT_NE(newdb2_dbid, newdb3_dbid);
// verify that data is still as it was in the original db.
value.clear();
ASSERT_OK(cloud_db->Get(ReadOptions(), "Hello", &value));
ASSERT_TRUE(value.compare("World") == 0);
// Assert that there are no redundant sst files
CloudEnvImpl* env = static_cast<CloudEnvImpl*>(cloud_env.get());
std::vector<std::string> to_be_deleted;
ASSERT_OK(env->FindObsoleteFiles(src_bucket_prefix_, &to_be_deleted));
// TODO(igor): Re-enable once purger code is fixed
// ASSERT_EQ(to_be_deleted.size(), 0);
// Assert that there are no redundant dbid
ASSERT_OK(env->FindObsoleteDbid(src_bucket_prefix_, &to_be_deleted));
// TODO(igor): Re-enable once purger code is fixed
// ASSERT_EQ(to_be_deleted.size(), 0);
}
}
//
// verify that dbid registry is appropriately handled
//
TEST_F(CloudTest, DbidRegistry) {
// Put one key-value
OpenDB();
std::string value;
ASSERT_OK(db_->Put(WriteOptions(), "Hello", "World"));
ASSERT_OK(db_->Get(ReadOptions(), "Hello", &value));
ASSERT_TRUE(value.compare("World") == 0);
// Assert that there is one db in the registry
DbidList dbs;
ASSERT_OK(aenv_->GetDbidList(src_bucket_prefix_, &dbs));
ASSERT_EQ(dbs.size(), 1);
CloseDB();
}
TEST_F(CloudTest, KeepLocalFiles) {
cloud_env_options_.keep_local_sst_files = true;
// Create two files
OpenDB();
std::string value;
ASSERT_OK(db_->Put(WriteOptions(), "Hello", "World"));
ASSERT_OK(db_->Flush(FlushOptions()));
ASSERT_OK(db_->Put(WriteOptions(), "Hello2", "World2"));
ASSERT_OK(db_->Flush(FlushOptions()));
CloseDB();
DestroyDir(dbname_);
OpenDB();
std::vector<std::string> files;
ASSERT_OK(Env::Default()->GetChildren(dbname_, &files));
long sst_files =
std::count_if(files.begin(), files.end(), [](const std::string& file) {
return file.find("sst") != std::string::npos;
});
ASSERT_EQ(sst_files, 2);
ASSERT_OK(db_->Get(ReadOptions(), "Hello", &value));
ASSERT_EQ(value, "World");
ASSERT_OK(db_->Get(ReadOptions(), "Hello2", &value));
ASSERT_EQ(value, "World2");
CloseDB();
ValidateCloudLiveFilesSrcSize();
}
TEST_F(CloudTest, CopyToFromS3) {
std::string fname = dbname_ + "/100000.sst";
// Create aws env
cloud_env_options_.keep_local_sst_files = true;
CreateAwsEnv();
((CloudEnvImpl*)aenv_.get())->TEST_InitEmptyCloudManifest();
char buffer[1 * 1024 * 1024];
// create a 10 MB file and upload it to cloud
{
unique_ptr<WritableFile> writer;
ASSERT_OK(aenv_->NewWritableFile(fname, &writer, EnvOptions()));
for (int i = 0; i < 10; i++) {
ASSERT_OK(writer->Append(Slice(buffer, sizeof(buffer))));
}
// sync and close file
}
// delete the file manually.
ASSERT_OK(base_env_->DeleteFile(fname));
// reopen file for reading. It should be refetched from cloud storage.
{
unique_ptr<RandomAccessFile> reader;
ASSERT_OK(aenv_->NewRandomAccessFile(fname, &reader, EnvOptions()));
uint64_t offset = 0;
for (int i = 0; i < 10; i++) {
Slice result;
char* scratch = &buffer[0];
ASSERT_OK(reader->Read(offset, sizeof(buffer), &result, scratch));
ASSERT_EQ(result.size(), sizeof(buffer));
offset += sizeof(buffer);
}
}
}
TEST_F(CloudTest, DelayFileDeletion) {
std::string fname = dbname_ + "/000010.sst";
// Create aws env
cloud_env_options_.keep_local_sst_files = true;
CreateAwsEnv();
((CloudEnvImpl*)aenv_.get())->TEST_InitEmptyCloudManifest();
((AwsEnv*)aenv_.get())->TEST_SetFileDeletionDelay(std::chrono::seconds(2));
auto createFile = [&]() {
unique_ptr<WritableFile> writer;
ASSERT_OK(aenv_->NewWritableFile(fname, &writer, EnvOptions()));
for (int i = 0; i < 10; i++) {
ASSERT_OK(writer->Append("igor"));
}
// sync and close file
};
for (int iter = 0; iter <= 1; ++iter) {
createFile();
// delete the file
ASSERT_OK(aenv_->DeleteFile(fname));
// file should still be there
ASSERT_OK(aenv_->FileExists(fname));
if (iter == 1) {
// should prevent the deletion
createFile();
}
std::this_thread::sleep_for(std::chrono::seconds(3));
auto st = aenv_->FileExists(fname);
if (iter == 0) {
// in iter==0 file should be deleted after 2 seconds
ASSERT_TRUE(st.IsNotFound());
} else {
// in iter==1 file should not be deleted because we wrote the new file
ASSERT_OK(st);
}
}
}
// Verify that a savepoint copies all src files to destination
TEST_F(CloudTest, Savepoint) {
// Put one key-value
OpenDB();
std::string value;
ASSERT_OK(db_->Put(WriteOptions(), "Hello", "World"));
ASSERT_OK(db_->Get(ReadOptions(), "Hello", &value));
ASSERT_TRUE(value.compare("World") == 0);
CloseDB();
value.clear();
std::string dest_path = "/clone2_path";
{
// Create a new instance with different src and destination paths.
// This is true clone and should have all the contents of the masterdb
std::unique_ptr<CloudEnv> cloud_env;
std::unique_ptr<DBCloud> cloud_db;
CloneDB("localpath1", src_bucket_prefix_, src_object_prefix_,
src_bucket_prefix_, dest_path, &cloud_db, &cloud_env);
// check that the original kv appears in the clone
value.clear();
ASSERT_OK(cloud_db->Get(ReadOptions(), "Hello", &value));
ASSERT_TRUE(value.compare("World") == 0);
// there should be only one sst file
std::vector<LiveFileMetaData> flist;
cloud_db->GetLiveFilesMetaData(&flist);
ASSERT_TRUE(flist.size() == 1);
auto remapped_fname =
((CloudEnvImpl*)cloud_env.get())->RemapFilename(flist[0].name);
// source path
std::string spath = src_object_prefix_ + "/" + remapped_fname;
ASSERT_OK(cloud_env->ExistsObject(src_bucket_prefix_, spath));
// Verify that the destination path does not have any sst files
std::string dpath = dest_path + "/" + remapped_fname;
ASSERT_TRUE(
cloud_env->ExistsObject(src_bucket_prefix_, dpath).IsNotFound());
// write a new value to the clone
ASSERT_OK(cloud_db->Put(WriteOptions(), "Hell", "Done"));
value.clear();
ASSERT_OK(cloud_db->Get(ReadOptions(), "Hell", &value));
ASSERT_TRUE(value.compare("Done") == 0);
// Invoke savepoint to populate destination path from source path
ASSERT_OK(cloud_db->Savepoint());
// check that the sst file is copied to dest path
ASSERT_OK(cloud_env->ExistsObject(src_bucket_prefix_, dpath));
ASSERT_OK(cloud_db->Flush(FlushOptions()));
}
{
// Reopen the clone
std::unique_ptr<CloudEnv> cloud_env;
std::unique_ptr<DBCloud> cloud_db;
CloneDB("localpath2", src_bucket_prefix_, src_object_prefix_,
src_bucket_prefix_, dest_path, &cloud_db, &cloud_env);
// check that the both kvs appears in the clone
value.clear();
ASSERT_OK(cloud_db->Get(ReadOptions(), "Hello", &value));
ASSERT_TRUE(value.compare("World") == 0);
value.clear();
ASSERT_OK(cloud_db->Get(ReadOptions(), "Hell", &value));
ASSERT_TRUE(value.compare("Done") == 0);
}
}
TEST_F(CloudTest, Encryption) {
// Create aws env
cloud_env_options_.server_side_encryption = true;
char* key_id = getenv("AWS_KMS_KEY_ID");
if (key_id != nullptr) {
cloud_env_options_.encryption_key_id = std::string(key_id);
Log(options_.info_log, "Found encryption key id in env variable %s",
key_id);
}
OpenDB();
ASSERT_OK(db_->Put(WriteOptions(), "Hello", "World"));
// create a file
ASSERT_OK(db_->Flush(FlushOptions()));
CloseDB();
OpenDB();
std::string value;
ASSERT_OK(db_->Get(ReadOptions(), "Hello", &value));
ASSERT_EQ(value, "World");
CloseDB();
}
#ifdef USE_KAFKA
TEST_F(CloudTest, KeepLocalLogKafka) {
cloud_env_options_.keep_local_log_files = false;
cloud_env_options_.log_type = LogType::kLogKafka;
cloud_env_options_.kafka_log_options.client_config_params["metadata.broker.list"]
= "localhost:9092";
OpenDB();
ASSERT_OK(db_->Put(WriteOptions(), "Franz", "Kafka"));
// Destroy DB in memory and on local file system.
delete db_;
db_ = nullptr;
aenv_.reset();
DestroyDir(dbname_);
DestroyDir("/tmp/ROCKSET");
// Create new env.
CreateAwsEnv();
// Give env enough time to consume WALs
std::this_thread::sleep_for(std::chrono::seconds(3));
// Open DB.
cloud_env_options_.keep_local_log_files = true;
options_.wal_dir = static_cast<AwsEnv*>(aenv_.get())->GetWALCacheDir();
OpenDB();
// Test read.
std::string value;
ASSERT_OK(db_->Get(ReadOptions(), "Franz", &value));
ASSERT_EQ(value, "Kafka");
CloseDB();
}
#endif /* USE_KAFKA */
// TODO(igor): determine why this fails,
// https://github.com/rockset/rocksdb-cloud/issues/35
TEST_F(CloudTest, DISABLED_KeepLocalLogKinesis) {
cloud_env_options_.keep_local_log_files = false;
cloud_env_options_.log_type = LogType::kLogKinesis;
OpenDB();
// Test write.
ASSERT_OK(db_->Put(WriteOptions(), "Tele", "Kinesis"));
// Destroy DB in memory and on local file system.
delete db_;
db_ = nullptr;
aenv_.reset();
DestroyDir(dbname_);
DestroyDir("/tmp/ROCKSET");
// Create new env.
CreateAwsEnv();
// Give env enough time to consume WALs
std::this_thread::sleep_for(std::chrono::seconds(3));
// Open DB.
cloud_env_options_.keep_local_log_files = true;
options_.wal_dir = static_cast<AwsEnv*>(aenv_.get())->GetWALCacheDir();
OpenDB();
// Test read.
std::string value;
ASSERT_OK(db_->Get(ReadOptions(), "Tele", &value));
ASSERT_EQ(value, "Kinesis");
CloseDB();
}
// Test whether we are able to recover nicely from two different writers to the
// same S3 bucket. (The feature that was enabled by CLOUDMANIFEST)
TEST_F(CloudTest, TwoDBsOneBucket) {
auto firstDB = dbname_;
auto secondDB = dbname_ + "-1";
cloud_env_options_.keep_local_sst_files = true;
std::string value;
OpenDB();
auto firstManifestFile =
dest_object_prefix_ + "/" +
((CloudEnvImpl*)aenv_.get())->RemapFilename("MANIFEST-1");
EXPECT_OK(aenv_->ExistsObject(dest_bucket_prefix_, firstManifestFile));
// Create two files
ASSERT_OK(db_->Put(WriteOptions(), "First", "File"));
ASSERT_OK(db_->Flush(FlushOptions()));
ASSERT_OK(db_->Put(WriteOptions(), "Second", "File"));
ASSERT_OK(db_->Flush(FlushOptions()));
auto files = GetSSTFiles(dbname_);
EXPECT_EQ(files.size(), 2);
CloseDB();
// Open again, with no destination bucket
dest_bucket_prefix_.clear();
dest_object_prefix_.clear();
OpenDB();
ASSERT_OK(db_->Put(WriteOptions(), "Third", "File"));
ASSERT_OK(db_->Flush(FlushOptions()));
auto newFiles = GetSSTFiles(dbname_);
EXPECT_EQ(newFiles.size(), 3);
// Remember the third file we created
std::vector<std::string> diff;
std::set_difference(newFiles.begin(), newFiles.end(), files.begin(),
files.end(), std::inserter(diff, diff.begin()));
ASSERT_EQ(diff.size(), 1);
auto thirdFile = diff[0];
CloseDB();
// Open in a different directory with destination bucket set
dbname_ = secondDB;
dest_bucket_prefix_ = src_bucket_prefix_;
dest_object_prefix_ = src_object_prefix_;
OpenDB();
ASSERT_OK(db_->Put(WriteOptions(), "Third", "DifferentFile"));
ASSERT_OK(db_->Flush(FlushOptions()));
CloseDB();
// Open back in the first directory with no destination
dbname_ = firstDB;
dest_bucket_prefix_.clear();
dest_object_prefix_.clear();
OpenDB();
// Changes to the cloud database should make no difference for us. This is an
// important check because we should not reinitialize from the cloud if we
// have a valid local directory!
ASSERT_OK(db_->Get(ReadOptions(), "Third", &value));
EXPECT_EQ(value, "File");
CloseDB();
// Reopen in the first directory, this time with destination path
dbname_ = firstDB;
dest_bucket_prefix_ = src_bucket_prefix_;
dest_object_prefix_ = src_object_prefix_;
OpenDB();
// Changes to the cloud database should be pulled down now.
ASSERT_OK(db_->Get(ReadOptions(), "Third", &value));
EXPECT_EQ(value, "DifferentFile");
files = GetSSTFiles(dbname_);
// Should no longer be in my directory because it's not part of the new
// MANIFEST.
EXPECT_TRUE(files.find(thirdFile) == files.end());
// We need to sleep a bit because file deletion happens in a different thread,
// so it might not be immediately deleted.
std::this_thread::sleep_for(std::chrono::milliseconds(10));
EXPECT_TRUE(
aenv_->ExistsObject(dest_bucket_prefix_, firstManifestFile).IsNotFound());
CloseDB();
}
// This test is similar to TwoDBsOneBucket, but is much more chaotic and illegal
// -- it runs two databases on exact same S3 bucket. The work on CLOUDMANIFEST
// enables us to run in that configuration for extended amount of time (1 hour
// by default) without any issues -- the last CLOUDMANIFEST writer wins.
TEST_F(CloudTest, TwoConcurrentWriters) {
auto firstDB = dbname_;
auto secondDB = dbname_ + "-1";
DBCloud *db1, *db2;
CloudEnv *aenv1, *aenv2;
auto openDB1 = [&] {
dbname_ = firstDB;
OpenDB();
db1 = db_;
db_ = nullptr;
aenv1 = aenv_.release();
};
auto openDB2 = [&] {
dbname_ = secondDB;
OpenDB();
db2 = db_;
db_ = nullptr;
aenv2 = aenv_.release();
};
auto closeDB1 = [&] {
db_ = db1;
aenv_.reset(aenv1);
CloseDB();
};
auto closeDB2 = [&] {
db_ = db2;
aenv_.reset(aenv2);
CloseDB();
};
openDB1();
openDB2();
// Create bunch of files, reopening the databases during
for (int i = 0; i < 5; ++i) {
closeDB1();
if (i == 2) {
DestroyDir(firstDB);
}
// opening the database makes me a master (i.e. CLOUDMANIFEST points to my
// manifest), my writes are applied to the shared space!
openDB1();
for (int j = 0; j < 5; ++j) {
auto key = ToString(i) + ToString(j) + "1";
ASSERT_OK(db1->Put(WriteOptions(), key, "FirstDB"));
ASSERT_OK(db1->Flush(FlushOptions()));
}
closeDB2();
if (i == 2) {
DestroyDir(secondDB);
}
// opening the database makes me a master (i.e. CLOUDMANIFEST points to my
// manifest), my writes are applied to the shared space!
openDB2();
for (int j = 0; j < 5; ++j) {
auto key = ToString(i) + ToString(j) + "2";
ASSERT_OK(db2->Put(WriteOptions(), key, "SecondDB"));
ASSERT_OK(db2->Flush(FlushOptions()));
}
}
dbname_ = firstDB;
// This write should not be applied, because DB2 is currently the owner of the
// S3 bucket
ASSERT_OK(db1->Put(WriteOptions(), "ShouldNotBeApplied", ""));
ASSERT_OK(db1->Flush(FlushOptions()));
closeDB1();
closeDB2();
openDB1();
for (int i = 0; i < 5; ++i) {
for (int j = 0; j < 5; ++j) {
std::string val;
auto key = ToString(i) + ToString(j);
ASSERT_OK(db1->Get(ReadOptions(), key + "1", &val));
EXPECT_EQ(val, "FirstDB");
ASSERT_OK(db1->Get(ReadOptions(), key + "2", &val));
EXPECT_EQ(val, "SecondDB");
}
}
std::string v;
ASSERT_TRUE(db1->Get(ReadOptions(), "ShouldNotBeApplied", &v).IsNotFound());
}
// Creates a pure RocksDB database and makes sure we can migrate to RocksDB
// Cloud
TEST_F(CloudTest, MigrateFromPureRocksDB) {
{ // Create local RocksDB
Options options;
options.create_if_missing = true;
DB* dbptr;
unique_ptr<DB> db;
ASSERT_OK(DB::Open(options, dbname_, &dbptr));
db.reset(dbptr);
// create 5 files
for (int i = 0; i < 5; ++i) {
auto key = "key" + ToString(i);
ASSERT_OK(db->Put(WriteOptions(), key, key));
ASSERT_OK(db->Flush(FlushOptions()));
}
}
// Now open RocksDB cloud
// TODO(dhruba) Figure out how to make this work without skipping dbid
// verification
cloud_env_options_.skip_dbid_verification = true;
OpenDB();
for (int i = 5; i < 10; ++i) {
auto key = "key" + ToString(i);
ASSERT_OK(db_->Put(WriteOptions(), key, key));
ASSERT_OK(db_->Flush(FlushOptions()));
}
for (int i = 0; i < 10; ++i) {
auto key = "key" + ToString(i);
std::string value;
ASSERT_OK(db_->Get(ReadOptions(), key, &value));
ASSERT_EQ(value, key);
}
CloseDB();
}
// Tests that we can open cloud DB without destination and source bucket set.
// This is useful for tests.
TEST_F(CloudTest, NoDestOrSrc) {
DestroyDir(dbname_);
// cloud_env_options_.keep_local_sst_files = true;
dest_bucket_prefix_ = dest_object_prefix_ = src_bucket_prefix_ =
src_object_prefix_ = "";
OpenDB();
ASSERT_OK(db_->Put(WriteOptions(), "key", "value"));
ASSERT_OK(db_->Flush(FlushOptions()));
std::string value;
ASSERT_OK(db_->Get(ReadOptions(), "key", &value));
ASSERT_EQ(value, "value");
CloseDB();
OpenDB();
ASSERT_OK(db_->Get(ReadOptions(), "key", &value));
ASSERT_EQ(value, "value");
CloseDB();
}
TEST_F(CloudTest, PreloadCloudManifest) {
DestroyDir(dbname_);
// Put one key-value
OpenDB();
std::string value;
ASSERT_OK(db_->Put(WriteOptions(), "Hello", "World"));
ASSERT_OK(db_->Get(ReadOptions(), "Hello", &value));
ASSERT_TRUE(value.compare("World") == 0);
CloseDB();
value.clear();
// Reopen and validate, preload cloud manifest
DBCloud::PreloadCloudManifest(aenv_.get(), options_, dbname_);
OpenDB();
ASSERT_OK(db_->Get(ReadOptions(), "Hello", &value));
ASSERT_EQ(value, "World");
}
//
// Test Ephemeral mode. In this mode, the database is cloned
// from a cloud bucket but new writes are not propagated
// back to any cloud bucket. Once cloned, all updates are local.
//
TEST_F(CloudTest, Ephemeral) {
cloud_env_options_.keep_local_sst_files = true;
options_.level0_file_num_compaction_trigger = 100; // never compact
// Create a primary DB with two files
OpenDB();
std::string value;
std::string newdb1_dbid;
std::set<uint64_t> cloud_files;
ASSERT_OK(db_->Put(WriteOptions(), "Name", "dhruba"));
ASSERT_OK(db_->Flush(FlushOptions()));
ASSERT_OK(db_->Put(WriteOptions(), "Hello2", "borthakur"));
ASSERT_OK(db_->Flush(FlushOptions()));
CloseDB();
ASSERT_EQ(2, GetSSTFiles(dbname_).size());
// Reopen the same database in ephemeral mode by cloning the original.
// Do not destroy the local dir. Writes to this db does not make it back
// to any cloud storage.
{
std::unique_ptr<CloudEnv> cloud_env;
std::unique_ptr<DBCloud> cloud_db;
CloneDB("db_ephemeral", src_bucket_prefix_, src_object_prefix_, "", "", &cloud_db,
&cloud_env);
// Retrieve the id of the first reopen
ASSERT_OK(cloud_db->GetDbIdentity(newdb1_dbid));
// verify that we still have two sst files
ASSERT_EQ(2, GetSSTFilesClone("db_ephemeral").size());
ASSERT_OK(cloud_db->Get(ReadOptions(), "Name", &value));
ASSERT_EQ(value, "dhruba");
ASSERT_OK(cloud_db->Get(ReadOptions(), "Hello2", &value));
ASSERT_EQ(value, "borthakur");
// Write one more record.
// There should be 3 local sst files in the ephemeral db.
ASSERT_OK(cloud_db->Put(WriteOptions(), "zip", "94087"));
ASSERT_OK(cloud_db->Flush(FlushOptions()));
ASSERT_EQ(3, GetSSTFilesClone("db_ephemeral").size());
// check that cloud files did not get updated
ASSERT_OK(GetCloudLiveFilesSrc(&cloud_files));
ASSERT_EQ(2, cloud_files.size());
cloud_files.clear();
}
// reopen main db and write two more records to it
OpenDB();
ASSERT_EQ(2, GetSSTFiles(dbname_).size());
// write two more records to it.
ASSERT_OK(db_->Put(WriteOptions(), "Key1", "onlyInMainDB"));
ASSERT_OK(db_->Flush(FlushOptions()));
ASSERT_OK(db_->Put(WriteOptions(), "Key2", "onlyInMainDB"));
ASSERT_OK(db_->Flush(FlushOptions()));
ASSERT_EQ(4, GetSSTFiles(dbname_).size());
CloseDB();
ASSERT_OK(GetCloudLiveFilesSrc(&cloud_files));
ASSERT_EQ(4, cloud_files.size());
cloud_files.clear();
// At this point, the main db has 4 files while the ephemeral
// database has diverged earlier with 3 local files. If we try
// to reopen the ephemeral clone, it should not download new
// files from the cloud
{
std::unique_ptr<CloudEnv> cloud_env;
std::unique_ptr<DBCloud> cloud_db;
std::string dbid;
options_.info_log = nullptr;
CreateLoggerFromOptions(clone_dir_ + "/db_ephemeral", options_, &options_.info_log);
CloneDB("db_ephemeral", src_bucket_prefix_, src_object_prefix_, "", "", &cloud_db,
&cloud_env);
// Retrieve the id of this clone. It should be same as before
ASSERT_OK(cloud_db->GetDbIdentity(dbid));
ASSERT_EQ(newdb1_dbid, dbid);
ASSERT_EQ(3, GetSSTFilesClone("db_ephemeral").size());
// verify that a key written to the ephemeral db still exists
ASSERT_OK(cloud_db->Get(ReadOptions(), "zip", &value));
ASSERT_EQ(value, "94087");
// verify that keys written to the main db after the ephemeral
// was clones do not appear in the ephemeral db.
ASSERT_NOK(cloud_db->Get(ReadOptions(), "Key1", &value));
ASSERT_NOK(cloud_db->Get(ReadOptions(), "Key2", &value));
}
}
//
// Test Ephemeral clones with resyncOnOpen mode.
// In this mode, every open of the ephemeral clone db causes its
// data to be resynced with the master db.
//
TEST_F(CloudTest, EphemeralResync) {
cloud_env_options_.keep_local_sst_files = true;
cloud_env_options_.ephemeral_resync_on_open = true;
options_.level0_file_num_compaction_trigger = 100; // never compact
// Create a primary DB with two files
OpenDB();
std::string value;
std::string newdb1_dbid;
std::set<uint64_t> cloud_files;
ASSERT_OK(db_->Put(WriteOptions(), "Name", "dhruba"));
ASSERT_OK(db_->Flush(FlushOptions()));
ASSERT_OK(db_->Put(WriteOptions(), "Hello2", "borthakur"));
ASSERT_OK(db_->Flush(FlushOptions()));
CloseDB();
ASSERT_EQ(2, GetSSTFiles(dbname_).size());
// Reopen the same database in ephemeral mode by cloning the original.
// Do not destroy the local dir. Writes to this db does not make it back
// to any cloud storage.
{
std::unique_ptr<CloudEnv> cloud_env;
std::unique_ptr<DBCloud> cloud_db;
CloneDB("db_ephemeral", src_bucket_prefix_, src_object_prefix_, "", "", &cloud_db,
&cloud_env);
// Retrieve the id of the first reopen
ASSERT_OK(cloud_db->GetDbIdentity(newdb1_dbid));
// verify that we still have two sst files
ASSERT_EQ(2, GetSSTFilesClone("db_ephemeral").size());
ASSERT_OK(cloud_db->Get(ReadOptions(), "Name", &value));
ASSERT_EQ(value, "dhruba");
ASSERT_OK(cloud_db->Get(ReadOptions(), "Hello2", &value));
ASSERT_EQ(value, "borthakur");
// Write one more record.
// There should be 3 local sst files in the ephemeral db.
ASSERT_OK(cloud_db->Put(WriteOptions(), "zip", "94087"));
ASSERT_OK(cloud_db->Flush(FlushOptions()));
ASSERT_EQ(3, GetSSTFilesClone("db_ephemeral").size());
// check that cloud files did not get updated
ASSERT_OK(GetCloudLiveFilesSrc(&cloud_files));
ASSERT_EQ(2, cloud_files.size());
cloud_files.clear();
}
// reopen main db and write two more records to it
OpenDB();
ASSERT_EQ(2, GetSSTFiles(dbname_).size());
// write two more records to it.
ASSERT_OK(db_->Put(WriteOptions(), "Key1", "onlyInMainDB"));
ASSERT_OK(db_->Flush(FlushOptions()));
ASSERT_OK(db_->Put(WriteOptions(), "Key2", "onlyInMainDB"));
ASSERT_OK(db_->Flush(FlushOptions()));
ASSERT_EQ(4, GetSSTFiles(dbname_).size());
CloseDB();
ASSERT_OK(GetCloudLiveFilesSrc(&cloud_files));
ASSERT_EQ(4, cloud_files.size());
cloud_files.clear();
// At this point, the main db has 4 files while the ephemeral
// database has diverged earlier with 3 local files.
// Reopen the ephemeral db with resync_on_open flag.
// This means that earlier updates to the ephemeral db are lost.
// It also means that the most latest updates in the master db
// are reflected in the newly opened ephemeral database.
{
std::unique_ptr<CloudEnv> cloud_env;
std::unique_ptr<DBCloud> cloud_db;
std::string dbid;
options_.info_log = nullptr;
CreateLoggerFromOptions(clone_dir_ + "/db_ephemeral", options_, &options_.info_log);
CloneDB("db_ephemeral", src_bucket_prefix_, src_object_prefix_, "", "", &cloud_db,
&cloud_env);
// Retrieve the id of this clone. It should be same as before
ASSERT_OK(cloud_db->GetDbIdentity(dbid));
ASSERT_EQ(newdb1_dbid, dbid);
// verify that a key written to the ephemeral db does not exist
ASSERT_NOK(cloud_db->Get(ReadOptions(), "zip", &value));
// verify that keys written to the main db after the ephemeral
// was clones appear in the ephemeral db.
ASSERT_OK(cloud_db->Get(ReadOptions(), "Key1", &value));
ASSERT_EQ(value, "onlyInMainDB");
ASSERT_OK(cloud_db->Get(ReadOptions(), "Key2", &value));
ASSERT_EQ(value, "onlyInMainDB");
}
}
#ifdef AWS_DO_NOT_RUN
//
// Verify that we can cache data from S3 in persistent cache.
//
TEST_F(CloudTest, PersistentCache) {
std::string pcache = test::TmpDir() + "/persistent_cache";
SetPersistentCache(pcache, 1);
// Put one key-value
OpenDB();
std::string value;
ASSERT_OK(db_->Put(WriteOptions(), "Hello", "World"));
ASSERT_OK(db_->Get(ReadOptions(), "Hello", &value));
ASSERT_TRUE(value.compare("World") == 0);
CloseDB();
value.clear();
// Reopen and validate
OpenDB();
ASSERT_OK(db_->Get(ReadOptions(), "Hello", &value));
ASSERT_EQ(value, "World");
CloseDB();
}
#endif /* AWS_DO_NOT_RUN */
} // namespace rocksdb
// A black-box test for the cloud wrapper around rocksdb
int main(int argc, char** argv) {
::testing::InitGoogleTest(&argc, argv);
return RUN_ALL_TESTS();
}
#else // USE_AWS
#include <stdio.h>
int main(int argc, char** argv) {
fprintf(stderr,
"SKIPPED as DBCloud is supported only when USE_AWS is defined.\n");
return 0;
}
#endif
#else // ROCKSDB_LITE
#include <stdio.h>
int main(int argc, char** argv) {
fprintf(stderr, "SKIPPED as DBCloud is not supported in ROCKSDB_LITE\n");
return 0;
}
#endif // !ROCKSDB_LITE