Introduce DBCloud::CheckpointToCloud()

This is a useful utility to copy snapshot of the RocksDB-Cloud database
to a destination in the cloud. Another RocksDB-Cloud instance can then be
started from the specified location in the cloud.

Author: igorcanadi

cloud/aws/aws_env.cc

@@ -253,7 +253,7 @@ Aws::S3::Model::HeadObjectOutcome AwsS3ClientWrapper::HeadObject(
   t.SetSuccess(outcome.IsSuccess());
   return outcome;
 }
-  
+
 //
 // The AWS credentials are specified to the constructor via
 // access_key_id and secret_key.
@@ -1459,7 +1459,7 @@ Status AwsEnv::SaveIdentitytoS3(const std::string& localfile,
 
   // Save mapping from ID to cloud pathname
   if (st.ok() && !GetDestObjectPath().empty()) {
-    st = SaveDbid(dbid, GetDestObjectPath());
+    st = SaveDbid(GetDestBucketName(), dbid, GetDestObjectPath());
   }
   return st;
 }
@@ -1468,14 +1468,14 @@ Status AwsEnv::SaveIdentitytoS3(const std::string& localfile,
 // All db in a bucket are stored in path /.rockset/dbid/<dbid>
 // The value of the object is the pathname where the db resides.
 //
-Status AwsEnv::SaveDbid(const std::string& dbid, const std::string& dirname) {
+Status AwsEnv::SaveDbid(const std::string& bucket_name, const std::string& dbid,
+                        const std::string& dirname) {
   assert(status().ok());
   Log(InfoLogLevel::DEBUG_LEVEL, info_log_, "[s3] SaveDbid dbid %s dir '%s'",
       dbid.c_str(), dirname.c_str());
 
   std::string dbidkey = dbid_registry_ + dbid;
-  std::string bucket = GetDestBucketName();
-  Aws::String s3_bucket = ToAwsString(bucket);
+  Aws::String s3_bucket = ToAwsString(bucket_name);
   Aws::String key = ToAwsString(dbidkey);
 
   Aws::Map<Aws::String, Aws::String> metadata;
@@ -1497,11 +1497,11 @@ Status AwsEnv::SaveDbid(const std::string& dbid, const std::string& dirname) {
     std::string errmsg(error.GetMessage().c_str(), error.GetMessage().size());
     Log(InfoLogLevel::ERROR_LEVEL, info_log_,
         "[s3] Bucket %s SaveDbid error in saving dbid %s dirname %s %s",
-        bucket.c_str(), dbid.c_str(), dirname.c_str(), errmsg.c_str());
+        bucket_name.c_str(), dbid.c_str(), dirname.c_str(), errmsg.c_str());
     return Status::IOError(dirname, errmsg.c_str());
   }
   Log(InfoLogLevel::INFO_LEVEL, info_log_,
-      "[s3] Bucket %s SaveDbid dbid %s dirname %s %s", bucket.c_str(),
+      "[s3] Bucket %s SaveDbid dbid %s dirname %s %s", bucket_name.c_str(),
       dbid.c_str(), dirname.c_str(), "ok");
   return Status::OK();
 };
@@ -1880,7 +1880,7 @@ Status AwsEnv::NewLogger(const std::string& fname, shared_ptr<Logger>* result) {
 }
 
 // The factory method for creating an S3 Env
-Status AwsEnv::NewAwsEnv(Env* base_env, 
+Status AwsEnv::NewAwsEnv(Env* base_env,
                          const CloudEnvOptions& cloud_options,
                          const std::shared_ptr<Logger> & info_log, CloudEnv** cenv) {
   Status status;

cloud/aws/aws_env.h

@@ -97,7 +97,7 @@ struct JobHandle;
 class AwsEnv : public CloudEnvImpl {
  public:
   // A factory method for creating S3 envs
-  static Status NewAwsEnv(Env* env, 
+  static Status NewAwsEnv(Env* env,
                           const CloudEnvOptions& env_options,
                           const std::shared_ptr<Logger> & info_log, CloudEnv** cenv);
 
@@ -241,8 +241,8 @@ class AwsEnv : public CloudEnvImpl {
 
   bool IsRunning() const { return running_; }
 
-  
-  
+
+
   std::string GetWALCacheDir();
 
   std::shared_ptr<Logger> info_log_;  // informational messages
@@ -263,7 +263,8 @@ class AwsEnv : public CloudEnvImpl {
   Status StartTailingStream();
 
   // Saves and retrieves the dbid->dirname mapping in S3
-  Status SaveDbid(const std::string& dbid, const std::string& dirname) override;
+  Status SaveDbid(const std::string& bucket_name, const std::string& dbid,
+                  const std::string& dirname) override;
   Status GetPathForDbid(const std::string& bucket,
                         const std::string& dbid, std::string* dirname) override;
   Status GetDbidList(const std::string& bucket,

cloud/cloud_env.cc

@@ -19,7 +19,7 @@
 namespace rocksdb {
 
 bool CloudEnvOptions::GetNameFromEnvironment(const char *name, const char *alt, std::string * result) {
-  
+
   char *value = getenv(name);               // See if name is set in the environment
   if (value == nullptr && alt != nullptr) { // Not set.  Do we have an alt name?
     value = getenv(alt);                    // See if alt is in the environment
@@ -41,21 +41,21 @@ void CloudEnvOptions::TEST_Initialize(const std::string & bucket,
 BucketOptions::BucketOptions() {
     prefix_ = "rockset.";
 }
-  
+
 void BucketOptions::SetBucketName(const std::string & bucket,
 				  const std::string & prefix) {
   if (!prefix.empty()) {
     prefix_ = prefix;
   }
-  
+
   bucket_ = bucket;
   if (bucket_.empty()) {
     name_.clear();
   } else {
     name_ = prefix_ + bucket_;
   }
 }
-  
+
 
 // Initializes the bucket properties
 
@@ -77,13 +77,13 @@ void BucketOptions::TEST_Initialize(const std::string & bucket,
   if (! CloudEnvOptions::GetNameFromEnvironment("ROCKSDB_CLOUD_TEST_OBECT_PATH",
 						"ROCKSDB_CLOUD_OBJECT_PATH", &object_)) {
     object_ = object;
-  }  
+  }
   if (! CloudEnvOptions::GetNameFromEnvironment("ROCKSDB_CLOUD_TEST_REGION",
 						"ROCKSDB_CLOUD_REGION", &region_)) {
     region_ = region;
   }
 }
-  
+
 CloudEnv::~CloudEnv() {}
 
 CloudEnvWrapper::~CloudEnvWrapper() {}
@@ -228,8 +228,8 @@ Status CloudEnv::NewAwsEnv(Env* base_env,
   if (!dest_cloud_region.empty()) options.dest_bucket.SetRegion(dest_cloud_region);
   return NewAwsEnv(base_env, options, logger, cenv);
 }
-  
-Status CloudEnv::NewAwsEnv(Env* base_env, 
+
+Status CloudEnv::NewAwsEnv(Env* base_env,
                            const CloudEnvOptions& options,
                            const std::shared_ptr<Logger> & logger, CloudEnv** cenv) {
 #ifndef USE_AWS
@@ -238,13 +238,6 @@ Status CloudEnv::NewAwsEnv(Env* base_env,
   // Dump out cloud env options
   options.Dump(logger.get());
 
-  // If the src bucket is not specified, then this is a pass-through cloud env.
-  if (! options.dest_bucket.IsValid() &&
-      ! options.src_bucket.IsValid()) {
-    *cenv = new CloudEnvWrapper(options, base_env);
-    return Status::OK();
-  }
-
   Status st = AwsEnv::NewAwsEnv(base_env, options, logger, cenv);
   if (st.ok()) {
     // store a copy of the logger

cloud/cloud_env_options.cc

@@ -29,6 +29,8 @@ void CloudEnvOptions::Dump(Logger* log) const {
          ephemeral_resync_on_open ? "true" : "false");
   Header(log, "             COptions.skip_dbid_verification: %s",
          skip_dbid_verification ? "true" : "false");
+  Header(log, "           COptions.use_aws_transfer_manager: %s",
+         use_aws_transfer_manager ? "true" : "false");
 }
 
 }  // namespace rocksdb

cloud/cloud_env_wrapper.h

@@ -37,7 +37,8 @@ class CloudEnvWrapper : public CloudEnvImpl {
                                         const EnvOptions& options) override {
     return notsup_;
   }
-  virtual Status SaveDbid(const std::string& dbid,
+  virtual Status SaveDbid(const std::string& bucket_name,
+                          const std::string& dbid,
                           const std::string& dirname) override {
     return notsup_;
   }