ceph/src/rocksdb/logging/auto_roll_logger.cc

   1 //  Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
   2 //  This source code is licensed under both the GPLv2 (found in the
   3 //  COPYING file in the root directory) and Apache 2.0 License
   4 //  (found in the LICENSE.Apache file in the root directory).
   5 //
   6 #include "logging/auto_roll_logger.h"
   7
   8 #include <algorithm>
   9 #include "file/filename.h"
  10 #include "logging/logging.h"
  11 #include "util/mutexlock.h"
  12
  13 namespace ROCKSDB_NAMESPACE {
  14
  15 #ifndef ROCKSDB_LITE
  16 // -- AutoRollLogger
  17
  18 AutoRollLogger::AutoRollLogger(Env* env, const std::string& dbname,
  19                                const std::string& db_log_dir,
  20                                size_t log_max_size,
  21                                size_t log_file_time_to_roll,
  22                                size_t keep_log_file_num,
  23                                const InfoLogLevel log_level)
  24     : Logger(log_level),
  25       dbname_(dbname),
  26       db_log_dir_(db_log_dir),
  27       env_(env),
  28       status_(Status::OK()),
  29       kMaxLogFileSize(log_max_size),
  30       kLogFileTimeToRoll(log_file_time_to_roll),
  31       kKeepLogFileNum(keep_log_file_num),
  32       cached_now(static_cast<uint64_t>(env_->NowMicros() * 1e-6)),
  33       ctime_(cached_now),
  34       cached_now_access_count(0),
  35       call_NowMicros_every_N_records_(100),
  36       mutex_() {
  37   Status s = env->GetAbsolutePath(dbname, &db_absolute_path_);
  38   if (s.IsNotSupported()) {
  39     db_absolute_path_ = dbname;
  40   } else {
  41     status_ = s;
  42   }
  43   log_fname_ = InfoLogFileName(dbname_, db_absolute_path_, db_log_dir_);
  44   if (env_->FileExists(log_fname_).ok()) {
  45     RollLogFile();
  46   }
  47   GetExistingFiles();
  48   s = ResetLogger();
  49   if (s.ok() && status_.ok()) {
  50     status_ = TrimOldLogFiles();
  51   }
  52 }
  53
  54 Status AutoRollLogger::ResetLogger() {
  55   TEST_SYNC_POINT("AutoRollLogger::ResetLogger:BeforeNewLogger");
  56   status_ = env_->NewLogger(log_fname_, &logger_);
  57   TEST_SYNC_POINT("AutoRollLogger::ResetLogger:AfterNewLogger");
  58
  59   if (!status_.ok()) {
  60     return status_;
  61   }
  62   assert(logger_);
  63   logger_->SetInfoLogLevel(Logger::GetInfoLogLevel());
  64
  65   if (logger_->GetLogFileSize() == Logger::kDoNotSupportGetLogFileSize) {
  66     status_ = Status::NotSupported(
  67         "The underlying logger doesn't support GetLogFileSize()");
  68   }
  69   if (status_.ok()) {
  70     cached_now = static_cast<uint64_t>(env_->NowMicros() * 1e-6);
  71     ctime_ = cached_now;
  72     cached_now_access_count = 0;
  73   }
  74
  75   return status_;
  76 }
  77
  78 void AutoRollLogger::RollLogFile() {
  79   // This function is called when log is rotating. Two rotations
  80   // can happen quickly (NowMicro returns same value). To not overwrite
  81   // previous log file we increment by one micro second and try again.
  82   uint64_t now = env_->NowMicros();
  83   std::string old_fname;
  84   do {
  85     old_fname = OldInfoLogFileName(
  86       dbname_, now, db_absolute_path_, db_log_dir_);
  87     now++;
  88   } while (env_->FileExists(old_fname).ok());
  89   Status s = env_->RenameFile(log_fname_, old_fname);
  90   if (!s.ok()) {
  91     // What should we do on error?
  92   }
  93   old_log_files_.push(old_fname);
  94 }
  95
  96 void AutoRollLogger::GetExistingFiles() {
  97   {
  98     // Empty the queue to avoid duplicated entries in the queue.
  99     std::queue<std::string> empty;
 100     std::swap(old_log_files_, empty);
 101   }
 102
 103   std::string parent_dir;
 104   std::vector<std::string> info_log_files;
 105   Status s =
 106       GetInfoLogFiles(env_, db_log_dir_, dbname_, &parent_dir, &info_log_files);
 107   if (status_.ok()) {
 108     status_ = s;
 109   }
 110   // We need to sort the file before enqueing it so that when we
 111   // delete file from the front, it is the oldest file.
 112   std::sort(info_log_files.begin(), info_log_files.end());
 113
 114   for (const std::string& f : info_log_files) {
 115     old_log_files_.push(parent_dir + "/" + f);
 116   }
 117 }
 118
 119 Status AutoRollLogger::TrimOldLogFiles() {
 120   // Here we directly list info files and delete them through Env.
 121   // The deletion isn't going through DB, so there are shortcomes:
 122   // 1. the deletion is not rate limited by SstFileManager
 123   // 2. there is a chance that an I/O will be issued here
 124   // Since it's going to be complicated to pass DB object down to
 125   // here, we take a simple approach to keep the code easier to
 126   // maintain.
 127
 128   // old_log_files_.empty() is helpful for the corner case that
 129   // kKeepLogFileNum == 0. We can instead check kKeepLogFileNum != 0 but
 130   // it's essentially the same thing, and checking empty before accessing
 131   // the queue feels safer.
 132   while (!old_log_files_.empty() && old_log_files_.size() >= kKeepLogFileNum) {
 133     Status s = env_->DeleteFile(old_log_files_.front());
 134     // Remove the file from the tracking anyway. It's possible that
 135     // DB cleaned up the old log file, or people cleaned it up manually.
 136     old_log_files_.pop();
 137     // To make the file really go away, we should sync parent directory.
 138     // Since there isn't any consistency issue involved here, skipping
 139     // this part to avoid one I/O here.
 140     if (!s.ok()) {
 141       return s;
 142     }
 143   }
 144   return Status::OK();
 145 }
 146
 147 std::string AutoRollLogger::ValistToString(const char* format,
 148                                            va_list args) const {
 149   // Any log messages longer than 1024 will get truncated.
 150   // The user is responsible for chopping longer messages into multi line log
 151   static const int MAXBUFFERSIZE = 1024;
 152   char buffer[MAXBUFFERSIZE];
 153
 154   int count = vsnprintf(buffer, MAXBUFFERSIZE, format, args);
 155   (void) count;
 156   assert(count >= 0);
 157
 158   return buffer;
 159 }
 160
 161 void AutoRollLogger::LogInternal(const char* format, ...) {
 162   mutex_.AssertHeld();
 163
 164   if (!logger_) {
 165     return;
 166   }
 167
 168   va_list args;
 169   va_start(args, format);
 170   logger_->Logv(format, args);
 171   va_end(args);
 172 }
 173
 174 void AutoRollLogger::Logv(const char* format, va_list ap) {
 175   assert(GetStatus().ok());
 176   if (!logger_) {
 177     return;
 178   }
 179
 180   std::shared_ptr<Logger> logger;
 181   {
 182     MutexLock l(&mutex_);
 183     if ((kLogFileTimeToRoll > 0 && LogExpired()) ||
 184         (kMaxLogFileSize > 0 && logger_->GetLogFileSize() >= kMaxLogFileSize)) {
 185       RollLogFile();
 186       Status s = ResetLogger();
 187       Status s2 = TrimOldLogFiles();
 188
 189       if (!s.ok()) {
 190         // can't really log the error if creating a new LOG file failed
 191         return;
 192       }
 193
 194       WriteHeaderInfo();
 195
 196       if (!s2.ok()) {
 197         ROCKS_LOG_WARN(logger.get(), "Fail to trim old info log file: %s",
 198                        s2.ToString().c_str());
 199       }
 200     }
 201
 202     // pin down the current logger_ instance before releasing the mutex.
 203     logger = logger_;
 204   }
 205
 206   // Another thread could have put a new Logger instance into logger_ by now.
 207   // However, since logger is still hanging on to the previous instance
 208   // (reference count is not zero), we don't have to worry about it being
 209   // deleted while we are accessing it.
 210   // Note that logv itself is not mutex protected to allow maximum concurrency,
 211   // as thread safety should have been handled by the underlying logger.
 212   logger->Logv(format, ap);
 213 }
 214
 215 void AutoRollLogger::WriteHeaderInfo() {
 216   mutex_.AssertHeld();
 217   for (auto& header : headers_) {
 218     LogInternal("%s", header.c_str());
 219   }
 220 }
 221
 222 void AutoRollLogger::LogHeader(const char* format, va_list args) {
 223   if (!logger_) {
 224     return;
 225   }
 226
 227   // header message are to be retained in memory. Since we cannot make any
 228   // assumptions about the data contained in va_list, we will retain them as
 229   // strings
 230   va_list tmp;
 231   va_copy(tmp, args);
 232   std::string data = ValistToString(format, tmp);
 233   va_end(tmp);
 234
 235   MutexLock l(&mutex_);
 236   headers_.push_back(data);
 237
 238   // Log the original message to the current log
 239   logger_->Logv(format, args);
 240 }
 241
 242 bool AutoRollLogger::LogExpired() {
 243   if (cached_now_access_count >= call_NowMicros_every_N_records_) {
 244     cached_now = static_cast<uint64_t>(env_->NowMicros() * 1e-6);
 245     cached_now_access_count = 0;
 246   }
 247
 248   ++cached_now_access_count;
 249   return cached_now >= ctime_ + kLogFileTimeToRoll;
 250 }
 251 #endif  // !ROCKSDB_LITE
 252
 253 Status CreateLoggerFromOptions(const std::string& dbname,
 254                                const DBOptions& options,
 255                                std::shared_ptr<Logger>* logger) {
 256   if (options.info_log) {
 257     *logger = options.info_log;
 258     return Status::OK();
 259   }
 260
 261   Env* env = options.env;
 262   std::string db_absolute_path;
 263   Status s = env->GetAbsolutePath(dbname, &db_absolute_path);
 264   if (!s.ok()) {
 265     return s;
 266   }
 267   std::string fname =
 268       InfoLogFileName(dbname, db_absolute_path, options.db_log_dir);
 269
 270   env->CreateDirIfMissing(dbname)
 271       .PermitUncheckedError();  // In case it does not exist
 272   // Currently we only support roll by time-to-roll and log size
 273 #ifndef ROCKSDB_LITE
 274   if (options.log_file_time_to_roll > 0 || options.max_log_file_size > 0) {
 275     AutoRollLogger* result = new AutoRollLogger(
 276         env, dbname, options.db_log_dir, options.max_log_file_size,
 277         options.log_file_time_to_roll, options.keep_log_file_num,
 278         options.info_log_level);
 279     s = result->GetStatus();
 280     if (!s.ok()) {
 281       delete result;
 282     } else {
 283       logger->reset(result);
 284     }
 285     return s;
 286   }
 287 #endif  // !ROCKSDB_LITE
 288   // Open a log file in the same directory as the db
 289   env->RenameFile(fname,
 290                   OldInfoLogFileName(dbname, env->NowMicros(), db_absolute_path,
 291                                      options.db_log_dir))
 292       .PermitUncheckedError();
 293   s = env->NewLogger(fname, logger);
 294   if (logger->get() != nullptr) {
 295     (*logger)->SetInfoLogLevel(options.info_log_level);
 296   }
 297   return s;
 298 }
 299
 300 }  // namespace ROCKSDB_NAMESPACE