Merge pull request #67 from rockset/InfoLog

Unify Info_log and options

Author: igorcanadi

cloud/aws/aws_env.cc

@@ -391,8 +391,7 @@ Aws::S3::Model::HeadObjectOutcome AwsS3ClientWrapper::HeadObject(
 //
 AwsEnv::AwsEnv(Env* underlying_env, const CloudEnvOptions& _cloud_env_options,
                const std::shared_ptr<Logger>& info_log)
-    : CloudEnvImpl(_cloud_env_options, underlying_env),
-      info_log_(info_log),
+  : CloudEnvImpl(_cloud_env_options, underlying_env, info_log),
       running_(true) {
   Aws::InitAPI(Aws::SDKOptions());
   if (cloud_env_options.src_bucket.GetRegion().empty() ||
@@ -887,7 +886,7 @@ class S3Directory : public Directory {
  public:
   explicit S3Directory(AwsEnv* env, const std::string name)
       : env_(env), name_(name) {
-    status_ = env_->GetPosixEnv()->NewDirectory(name, &posixDir);
+    status_ = env_->GetBaseEnv()->NewDirectory(name, &posixDir);
   }
 
   ~S3Directory() {}

cloud/aws/aws_env.h

@@ -234,15 +234,10 @@ class AwsEnv : public CloudEnvImpl {
   virtual Status EmptyBucket(const std::string& bucket,
                              const std::string& path) override;
 
-  // get the posix env
-  Env* GetPosixEnv() const { return base_env_; }
-
   bool IsRunning() const { return running_; }
 
   std::string GetWALCacheDir();
 
-  std::shared_ptr<Logger> info_log_;  // informational messages
-
   // The S3 client
   std::shared_ptr<AwsS3ClientWrapper> s3client_;
 

cloud/aws/aws_log.cc

@@ -32,19 +32,19 @@ CloudLogController::CloudLogController(
   : env_(env), info_log_(info_log) {
 
   // Create a random number for the cache directory.
-  const std::string uid = trim(env_->GetPosixEnv()->GenerateUniqueId());
+  const std::string uid = trim(env_->GetBaseEnv()->GenerateUniqueId());
 
   // Temporary directory for cache.
   const std::string bucket_dir = kCacheDir + pathsep + env_->GetSrcBucketName();
   cache_dir_ = bucket_dir + pathsep + uid;
 
   // Create temporary directories.
-  status_ = env_->GetPosixEnv()->CreateDirIfMissing(kCacheDir);
+  status_ = env_->GetBaseEnv()->CreateDirIfMissing(kCacheDir);
   if (status_.ok()) {
-    status_ = env_->GetPosixEnv()->CreateDirIfMissing(bucket_dir);
+    status_ = env_->GetBaseEnv()->CreateDirIfMissing(bucket_dir);
   }
   if (status_.ok()) {
-    status_ = env_->GetPosixEnv()->CreateDirIfMissing(cache_dir_);
+    status_ = env_->GetBaseEnv()->CreateDirIfMissing(cache_dir_);
   }
 }
 
@@ -86,17 +86,17 @@ Status CloudLogController::Apply(const Slice& in) {
     // If this file is not yet open, open it and store it in cache.
     if (iter == cache_fds_.end()) {
       unique_ptr<RandomRWFile> result;
-      st = env_->GetPosixEnv()->NewRandomRWFile(
+      st = env_->GetBaseEnv()->NewRandomRWFile(
           pathname, &result, EnvOptions());
 
       if (!st.ok()) {
           // create the file
           unique_ptr<WritableFile> tmp_writable_file;
-          env_->GetPosixEnv()->NewWritableFile(pathname, &tmp_writable_file,
+          env_->GetBaseEnv()->NewWritableFile(pathname, &tmp_writable_file,
                                                EnvOptions());
           tmp_writable_file.reset();
           // Try again.
-          st = env_->GetPosixEnv()->NewRandomRWFile(
+          st = env_->GetBaseEnv()->NewRandomRWFile(
                   pathname, &result, EnvOptions());
       }
 
@@ -129,7 +129,7 @@ Status CloudLogController::Apply(const Slice& in) {
       cache_fds_.erase(iter);
     }
 
-    st = env_->GetPosixEnv()->DeleteFile(pathname);
+    st = env_->GetBaseEnv()->DeleteFile(pathname);
     Log(InfoLogLevel::DEBUG_LEVEL, env_->info_log_,
         "[%s] Tailer: Deleted file: %s %s",
         GetTypeName().c_str(), pathname.c_str(), st.ToString().c_str());

cloud/aws/aws_s3.cc

@@ -303,7 +303,7 @@ Status S3WritableFile::Close() {
 
   // delete local file
   if (!env_->GetCloudEnvOptions().keep_local_sst_files) {
-    status_ = env_->GetPosixEnv()->DeleteFile(fname_);
+    status_ = env_->GetBaseEnv()->DeleteFile(fname_);
     if (!status_.ok()) {
       Log(InfoLogLevel::ERROR_LEVEL, env_->info_log_,
           "[s3] S3WritableFile closing delete failed on local file %s",

cloud/cloud_env_impl.cc

@@ -19,8 +19,8 @@
 
 namespace rocksdb {
 
-CloudEnvImpl::CloudEnvImpl(const CloudEnvOptions& opts, Env* base_env)
-    : CloudEnv(opts), base_env_(base_env), purger_is_running_(true) {}
+  CloudEnvImpl::CloudEnvImpl(const CloudEnvOptions& opts, Env* base, const std::shared_ptr<Logger>& l)
+    : CloudEnv(opts, base, l), purger_is_running_(true) {}
 
 CloudEnvImpl::~CloudEnvImpl() { StopPurger(); }
 

cloud/cloud_env_impl.h

@@ -20,15 +20,12 @@ class CloudEnvImpl : public CloudEnv {
 
  public:
   // Constructor
-  CloudEnvImpl(const CloudEnvOptions & options, Env* base_env);
+  CloudEnvImpl(const CloudEnvOptions & options, Env* base_env, const std::shared_ptr<Logger>& logger);
 
   virtual ~CloudEnvImpl();
 
   const CloudType& GetCloudType() const { return cloud_env_options.cloud_type; }
 
-  // Returns the underlying env
-  Env* GetBaseEnv() override { return base_env_; }
-
   Status SanitizeDirectory(const DBOptions& options,
                            const std::string& clone_name, bool read_only);
   Status LoadCloudManifest(const std::string& local_dbname, bool read_only);
@@ -122,11 +119,6 @@ class CloudEnvImpl : public CloudEnv {
   // The pathname of the source database that is cloned
   std::string src_dbdir_;
 
-  // The underlying env
-  Env* base_env_;
-
-  std::shared_ptr<Logger> info_log_;  // informational messages
-
   // Protects purger_cv_
   std::mutex purger_lock_;
   std::condition_variable purger_cv_;

cloud/cloud_env_wrapper.h

@@ -19,7 +19,8 @@ namespace rocksdb {
 class CloudEnvWrapper : public CloudEnvImpl {
  public:
   // Initialize an EnvWrapper that delegates all calls to *t
-  explicit CloudEnvWrapper(const CloudEnvOptions& options, Env* t) : CloudEnvImpl(options, t) {
+  explicit CloudEnvWrapper(const CloudEnvOptions& options, Env* t, const std::shared_ptr<Logger>& l) :
+    CloudEnvImpl(options, t, l) {
     cloud_env_options.log_type = LogType::kLogNone;
     cloud_env_options.cloud_type = CloudType::kCloudNone;
     notsup_ = Status::NotSupported();

cloud/db_cloud_impl.cc

@@ -71,6 +71,9 @@ Status DBCloud::Open(const Options& opt, const std::string& local_dbname,
   }
 
   CloudEnvImpl* cenv = static_cast<CloudEnvImpl*>(options.env);
+  if (!cenv->info_log_) {
+    cenv->info_log_ = options.info_log;
+  }
   Env* local_env = cenv->GetBaseEnv();
   if (!read_only) {
     local_env->CreateDirIfMissing(

include/rocksdb/cloud/cloud_env_options.h

@@ -307,11 +307,19 @@ typedef std::map<std::string, std::string> DbidList;
 class CloudEnv : public Env {
  protected:
   CloudEnvOptions cloud_env_options;
-  CloudEnv(const CloudEnvOptions& options) : cloud_env_options(options) { }
- public:
-  // Returns the underlying env
-  virtual Env* GetBaseEnv() = 0;
+  Env* base_env_; // The underlying env
+  CloudEnv(const CloudEnvOptions& options, Env *base, const std::shared_ptr<Logger>& logger)
+    : cloud_env_options(options),
+      base_env_(base),
+      info_log_(logger) {
+  }
+public:
+  std::shared_ptr<Logger> info_log_;  // informational messages
   virtual ~CloudEnv();
+  // Returns the underlying env
+  Env* GetBaseEnv() {
+    return base_env_;
+  }
   virtual Status PreloadCloudManifest(const std::string& local_dbname) = 0;
 
   // Empties all contents of the associated cloud storage bucket.