Change CreateNewAwsEnv to use the created base. The env_ that was created earlier is now used as the "default" env for the AwsEnv. This fixes an occasional timing-related failure of the DbTest.Snapshot test.

Move methods that are not AWS-specific to the CloudEnvImpl

This change moves many of the methods from the AWS Env into CloudEnvImpl.  With the addition of the CloudStorageProvider and CloudLogController classes, many methods in the AWS Env no longe
 methods that are moved in this change do not use any methods or code that was specific to the AWS Environment.

Note that DeleteFile was left for a later change, as it requires moving/changing much more code.

Run "clang-format -i cloud/*.{cc,h} cloud/aws/*.{cc,h} include/rocksdb/cloud/*,{cc,h}.

Author: mrambacher

cloud/aws/aws_env.cc

@@ -22,7 +22,6 @@ namespace rocksdb {
 
 class S3ReadableFile;
 
-
 namespace detail {
 struct JobHandle;
 }  // namespace detail
@@ -53,9 +52,9 @@ struct JobHandle;
 class AwsEnv : public CloudEnvImpl {
  public:
   // A factory method for creating S3 envs
-  static Status NewAwsEnv(Env* env,
-                          const CloudEnvOptions& env_options,
-                          const std::shared_ptr<Logger> & info_log, CloudEnv** cenv);
+  static Status NewAwsEnv(Env* env, const CloudEnvOptions& env_options,
+                          const std::shared_ptr<Logger>& info_log,
+                          CloudEnv** cenv);
 
   virtual ~AwsEnv();
 
@@ -72,136 +71,20 @@ class AwsEnv : public CloudEnvImpl {
   // explicitly make the default region be us-west-2.
   static constexpr const char* default_region = "us-west-2";
 
-  virtual Status NewSequentialFile(const std::string& fname,
-                                   std::unique_ptr<SequentialFile>* result,
-                                   const EnvOptions& options) override;
-
-  virtual Status NewSequentialFileCloud(const std::string& bucket,
-                                        const std::string& fname,
-                                        std::unique_ptr<SequentialFile>* result,
-                                        const EnvOptions& options) override;
-
-  virtual Status NewRandomAccessFile(const std::string& fname,
-                                     std::unique_ptr<RandomAccessFile>* result,
-                                     const EnvOptions& options) override;
-
-  virtual Status NewWritableFile(const std::string& fname,
-                                 std::unique_ptr<WritableFile>* result,
-                                 const EnvOptions& options) override;
-
-  virtual Status ReopenWritableFile(const std::string& /*fname*/,
-                                    std::unique_ptr<WritableFile>* /*result*/,
-                                    const EnvOptions& /*options*/) override;
-
-  virtual Status NewDirectory(const std::string& name,
-                              std::unique_ptr<Directory>* result) override;
-
-  virtual Status FileExists(const std::string& fname) override;
-
-  virtual Status GetChildren(const std::string& path,
-                             std::vector<std::string>* result) override;
-
   virtual Status DeleteFile(const std::string& fname) override;
 
-  virtual Status CreateDir(const std::string& name) override;
-
-  virtual Status CreateDirIfMissing(const std::string& name) override;
-
-  virtual Status DeleteDir(const std::string& name) override;
-
-  virtual Status GetFileSize(const std::string& fname, uint64_t* size) override;
-
-  virtual Status GetFileModificationTime(const std::string& fname,
-                                         uint64_t* file_mtime) override;
-
-  virtual Status RenameFile(const std::string& src,
-                            const std::string& target) override;
-
-  virtual Status LinkFile(const std::string& src,
-                          const std::string& target) override;
-
   virtual Status LockFile(const std::string& fname, FileLock** lock) override;
 
   virtual Status UnlockFile(FileLock* lock) override;
 
-  virtual Status NewLogger(const std::string& fname,
-                           std::shared_ptr<Logger>* result) override;
-
-  virtual void Schedule(void (*function)(void* arg), void* arg,
-                        Priority pri = LOW, void* tag = nullptr,
-                        void (*unschedFunction)(void* arg) = 0) override {
-    base_env_->Schedule(function, arg, pri, tag, unschedFunction);
-  }
-
-  virtual int UnSchedule(void* tag, Priority pri) override {
-    return base_env_->UnSchedule(tag, pri);
-  }
-
-  virtual void StartThread(void (*function)(void* arg), void* arg) override {
-    base_env_->StartThread(function, arg);
-  }
-
-  virtual void WaitForJoin() override { base_env_->WaitForJoin(); }
-
-  virtual unsigned int GetThreadPoolQueueLen(
-      Priority pri = LOW) const override {
-    return base_env_->GetThreadPoolQueueLen(pri);
-  }
-
-  virtual Status GetTestDirectory(std::string* path) override {
-    return base_env_->GetTestDirectory(path);
-  }
-
-  virtual uint64_t NowMicros() override { return base_env_->NowMicros(); }
-
-  virtual void SleepForMicroseconds(int micros) override {
-    base_env_->SleepForMicroseconds(micros);
-  }
-
-  virtual Status GetHostName(char* name, uint64_t len) override {
-    return base_env_->GetHostName(name, len);
-  }
-
-  virtual Status GetCurrentTime(int64_t* unix_time) override {
-    return base_env_->GetCurrentTime(unix_time);
-  }
-
-  virtual Status GetAbsolutePath(const std::string& db_path,
-                                 std::string* output_path) override {
-    return base_env_->GetAbsolutePath(db_path, output_path);
-  }
-
-  virtual void SetBackgroundThreads(int number, Priority pri = LOW) override {
-    base_env_->SetBackgroundThreads(number, pri);
-  }
-  int GetBackgroundThreads(Priority pri) override {
-    return base_env_->GetBackgroundThreads(pri);
-  }
-
-  virtual void IncBackgroundThreadsIfNeeded(int number, Priority pri) override {
-    base_env_->IncBackgroundThreadsIfNeeded(number, pri);
-  }
-
-  virtual std::string TimeToString(uint64_t number) override {
-    return base_env_->TimeToString(number);
-  }
-
-  static uint64_t gettid() {
-    assert(sizeof(pthread_t) <= sizeof(uint64_t));
-    return (uint64_t)pthread_self();
-  }
-
-  virtual uint64_t GetThreadID() const override { return AwsEnv::gettid(); }
-
   std::string GetWALCacheDir();
 
   // Saves and retrieves the dbid->dirname mapping in S3
   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,
-                     DbidList* dblist) override;
+  Status GetPathForDbid(const std::string& bucket, const std::string& dbid,
+                        std::string* dirname) override;
+  Status GetDbidList(const std::string& bucket, DbidList* dblist) override;
   Status DeleteDbid(const std::string& bucket,
                     const std::string& dbid) override;
   Status DeleteCloudFileFromDest(const std::string& fname) override;
@@ -220,41 +103,17 @@ class AwsEnv : public CloudEnvImpl {
   // The AWS credentials are specified to the constructor via
   // access_key_id and secret_key.
   //
-  explicit AwsEnv(Env* underlying_env,
-                  const CloudEnvOptions& cloud_options,
-                  const std::shared_ptr<Logger> & info_log = nullptr);
-
-
+  explicit AwsEnv(Env* underlying_env, const CloudEnvOptions& cloud_options,
+                  const std::shared_ptr<Logger>& info_log = nullptr);
 
   // The pathname that contains a list of all db's inside a bucket.
   static constexpr const char* dbid_registry_ = "/.rockset/dbid/";
 
-  Status create_bucket_status_;
-
   std::mutex files_to_delete_mutex_;
   std::chrono::seconds file_deletion_delay_ = std::chrono::hours(1);
   std::unordered_map<std::string, std::shared_ptr<detail::JobHandle>>
       files_to_delete_;
   Random64 rng_;
-
-  Status status();
-
-  // Validate options
-  Status CheckOption(const EnvOptions& options);
-
-
-  Status NewS3ReadableFile(const std::string& bucket, const std::string& fname,
-                           std::unique_ptr<S3ReadableFile>* result);
-
-  // Save IDENTITY file to S3. Update dbid registry.
-  Status SaveIdentitytoS3(const std::string& localfile,
-                          const std::string& target_idfile);
-
-  // Converts a local pathname to an object name in the src bucket
-  std::string srcname(const std::string& localname);
-
-  // Converts a local pathname to an object name in the dest bucket
-  std::string destname(const std::string& localname);
 };
 
 }  // namespace rocksdb

cloud/aws/aws_env.h

@@ -10,6 +10,6 @@ inline Aws::String ToAwsString(const std::string& s) {
   return Aws::String(s.data(), s.size());
 }
 
-}  // namepace rocksdb
+}  // namespace rocksdb
 
 #endif /* USE_AWS */

cloud/aws/aws_file.h

@@ -20,21 +20,22 @@
 namespace rocksdb {
 namespace cloud {
 namespace kafka {
-  
+
 /***************************************************/
 /*                KafkaWritableFile                */
 /***************************************************/
 class KafkaWritableFile : public CloudLogWritableFile {
  public:
   static const std::chrono::microseconds kFlushTimeout;
 
-  KafkaWritableFile(CloudEnv* env, const std::string& fname, const EnvOptions& options,
+  KafkaWritableFile(CloudEnv* env, const std::string& fname,
+                    const EnvOptions& options,
                     std::shared_ptr<RdKafka::Producer> producer,
                     std::shared_ptr<RdKafka::Topic> topic)
-    : CloudLogWritableFile(env, fname, options),
-      producer_(producer),
-      topic_(topic),
-      current_offset_(0) {
+      : CloudLogWritableFile(env, fname, options),
+        producer_(producer),
+        topic_(topic),
+        current_offset_(0) {
     Log(InfoLogLevel::DEBUG_LEVEL, env_->info_log_,
         "[kafka] WritableFile opened file %s", fname_.c_str());
   }
@@ -58,11 +59,10 @@ class KafkaWritableFile : public CloudLogWritableFile {
 const std::chrono::microseconds KafkaWritableFile::kFlushTimeout =
     std::chrono::seconds(10);
 
-
 Status KafkaWritableFile::ProduceRaw(const std::string& operation_name,
                                      const Slice& message) {
-  if (!status_.ok()){
-      return status_;
+  if (!status_.ok()) {
+    return status_;
   }
 
   RdKafka::ErrorCode resp;
@@ -116,13 +116,9 @@ Status KafkaWritableFile::Close() {
   return ProduceRaw("Close", serialized_data);
 }
 
-bool KafkaWritableFile::IsSyncThreadSafe() const {
-  return true;
-}
+bool KafkaWritableFile::IsSyncThreadSafe() const { return true; }
 
-Status KafkaWritableFile::Sync() {
-  return Flush();
-}
+Status KafkaWritableFile::Sync() { return Flush(); }
 
 Status KafkaWritableFile::Flush() {
   std::chrono::microseconds start(env_->NowMicros());
@@ -175,33 +171,33 @@ Status KafkaWritableFile::LogDelete() {
 //
 class KafkaController : public CloudLogControllerImpl {
  public:
-
   ~KafkaController() {
     for (size_t i = 0; i < partitions_.size(); i++) {
       consumer_->stop(consumer_topic_.get(), partitions_[i]->partition());
     }
-    
+
     Log(InfoLogLevel::DEBUG_LEVEL, env_->info_log_,
         "[%s] KafkaController closed.", Name());
   }
-  
-  const char *Name() const override { return "kafka"; }
+
+  const char* Name() const override { return "kafka"; }
 
   virtual Status CreateStream(const std::string& /* bucket_prefix */) override {
     // Kafka client cannot create a topic. Topics are either manually created
     // or implicitly created on first write if auto.create.topics.enable is
     // true.
     return status_;
   }
-  virtual Status WaitForStreamReady(const std::string& /* bucket_prefix */) override {
+  virtual Status WaitForStreamReady(
+      const std::string& /* bucket_prefix */) override {
     // Kafka topics don't need to be waited on.
     return status_;
   }
 
   virtual Status TailStream() override;
 
-  virtual CloudLogWritableFile* CreateWritableFile(const std::string& fname,
-                                                   const EnvOptions& options) override;
+  virtual CloudLogWritableFile* CreateWritableFile(
+      const std::string& fname, const EnvOptions& options) override;
 
  protected:
   Status Initialize(CloudEnv* env) override;
@@ -293,8 +289,7 @@ Status KafkaController::TailStream() {
   }
 
   Log(InfoLogLevel::DEBUG_LEVEL, env_->info_log_, "[%s] TailStream topic %s %s",
-      Name(), consumer_topic_->name().c_str(),
-      status_.ToString().c_str());
+      Name(), consumer_topic_->name().c_str(), status_.ToString().c_str());
 
   Status lastErrorStatus;
   int retryAttempt = 0;
@@ -319,14 +314,14 @@ Status KafkaController::TailStream() {
           Log(InfoLogLevel::ERROR_LEVEL, env_->info_log_,
               "[%s] error processing message size %ld "
               "extracted from stream %s %s",
-              Name(), message->len(),
-              consumer_topic_->name().c_str(), status_.ToString().c_str());
+              Name(), message->len(), consumer_topic_->name().c_str(),
+              status_.ToString().c_str());
         } else {
           Log(InfoLogLevel::DEBUG_LEVEL, env_->info_log_,
               "[%s] successfully processed message size %ld "
               "extracted from stream %s %s",
-              Name(), message->len(),
-              consumer_topic_->name().c_str(), status_.ToString().c_str());
+              Name(), message->len(), consumer_topic_->name().c_str(),
+              status_.ToString().c_str());
         }
 
         // Remember last read offset from topic (currently unused).
@@ -344,8 +339,7 @@ Status KafkaController::TailStream() {
                             RdKafka::err2str(message->err()).c_str());
 
         Log(InfoLogLevel::DEBUG_LEVEL, env_->info_log_,
-            "[%s] error reading %s %s", Name(),
-            consumer_topic_->name().c_str(),
+            "[%s] error reading %s %s", Name(), consumer_topic_->name().c_str(),
             RdKafka::err2str(message->err()).c_str());
 
         ++retryAttempt;
@@ -376,9 +370,8 @@ Status KafkaController::InitializePartitions() {
                               RdKafka::err2str(err).c_str());
 
     Log(InfoLogLevel::DEBUG_LEVEL, env_->info_log_,
-        "[%s] S3ReadableFile file %s Unable to find shards %s",
-        Name(), consumer_topic_->name().c_str(),
-        status_.ToString().c_str());
+        "[%s] S3ReadableFile file %s Unable to find shards %s", Name(),
+        consumer_topic_->name().c_str(), status_.ToString().c_str());
 
     return status_;
   }
@@ -432,11 +425,12 @@ Status CloudLogControllerImpl::CreateKafkaController(
     std::shared_ptr<CloudLogController>* output) {
 #ifndef USE_KAFKA
   output->reset();
-  return Status::NotSupported("In order to use Kafka, make sure you're compiling with "
-                              "USE_KAFKA=1");
+  return Status::NotSupported(
+      "In order to use Kafka, make sure you're compiling with "
+      "USE_KAFKA=1");
 #else
   output->reset(new rocksdb::cloud::kafka::KafkaController());
   return Status::OK();
-#endif // USE_KAFKA
+#endif  // USE_KAFKA
 }
 }  // namespace rocksdb