Enable log info with different levels.

Summary:
* Now each Log related function has a variant that takes an additional
  argument indicating its log level, which is one of the following:
 - DEBUG, INFO, WARN, ERROR, FATAL.

* To ensure backward-compatibility, old version Log functions are kept
  unchanged.

* Logger now has a member variable indicating its log level.  Any incoming
  Log request which log level is lower than Logger's log level will not
  be output.

* The output of the newer version Log will be prefixed by its log level.

Test Plan:
Add a LogType test in auto_roll_logger_test.cc

 = Sample log output =
    2014/02/11-00:03:07.683895 7feded179840 [DEBUG] this is the message to be written to the log file!!
    2014/02/11-00:03:07.683898 7feded179840 [INFO] this is the message to be written to the log file!!
    2014/02/11-00:03:07.683900 7feded179840 [WARN] this is the message to be written to the log file!!
    2014/02/11-00:03:07.683903 7feded179840 [ERROR] this is the message to be written to the log file!!
    2014/02/11-00:03:07.683906 7feded179840 [FATAL] this is the message to be written to the log file!!

Reviewers: dhruba, xjin, kailiu

Reviewed By: kailiu

CC: leveldb

Differential Revision: https://reviews.facebook.net/D16071

include/rocksdb/env.h

@@ -511,25 +511,56 @@ class Directory {
   virtual Status Fsync() = 0;
 };
 
+enum InfoLogLevel {
+  DEBUG = 0,
+  INFO,
+  WARN,
+  ERROR,
+  FATAL,
+  NUM_INFO_LOG_LEVELS,
+};
+
 // An interface for writing log messages.
 class Logger {
  public:
   enum { DO_NOT_SUPPORT_GET_LOG_FILE_SIZE = -1 };
-  Logger() { }
+  explicit Logger(const InfoLogLevel log_level = InfoLogLevel::ERROR)
+      : log_level_(log_level) {}
   virtual ~Logger();
 
   // Write an entry to the log file with the specified format.
   virtual void Logv(const char* format, va_list ap) = 0;
+
+  // Write an entry to the log file with the specified log level
+  // and format.  Any log with level under the internal log level
+  // of *this (see @SetInfoLogLevel and @GetInfoLogLevel) will not be
+  // printed.
+  void Logv(const InfoLogLevel log_level, const char* format, va_list ap) {
+    static const char* kInfoLogLevelNames[5] = {"DEBUG", "INFO", "WARN",
+                                                "ERROR", "FATAL"};
+    if (log_level < log_level_) {
+      return;
+    }
+    char new_format[500];
+    snprintf(new_format, sizeof(new_format) - 1, "[%s] %s",
+             kInfoLogLevelNames[log_level], format);
+    Logv(new_format, ap);
+  }
   virtual size_t GetLogFileSize() const {
     return DO_NOT_SUPPORT_GET_LOG_FILE_SIZE;
   }
   // Flush to the OS buffers
   virtual void Flush() {}
+  virtual InfoLogLevel GetInfoLogLevel() const { return log_level_; }
+  virtual void SetInfoLogLevel(const InfoLogLevel log_level) {
+    log_level_ = log_level;
+  }
 
  private:
   // No copying allowed
   Logger(const Logger&);
   void operator=(const Logger&);
+  InfoLogLevel log_level_;
 };
 
 
@@ -547,7 +578,18 @@ class FileLock {
 
 extern void LogFlush(const shared_ptr<Logger>& info_log);
 
+extern void Log(const InfoLogLevel log_level,
+                const shared_ptr<Logger>& info_log, const char* format, ...);
+
+// a set of log functions with different log levels.
+extern void Debug(const shared_ptr<Logger>& info_log, const char* format, ...);
+extern void Info(const shared_ptr<Logger>& info_log, const char* format, ...);
+extern void Warn(const shared_ptr<Logger>& info_log, const char* format, ...);
+extern void Error(const shared_ptr<Logger>& info_log, const char* format, ...);
+extern void Fatal(const shared_ptr<Logger>& info_log, const char* format, ...);
+
 // Log the specified data to *info_log if info_log is non-nullptr.
+// The default info log level is InfoLogLevel::ERROR.
 extern void Log(const shared_ptr<Logger>& info_log, const char* format, ...)
 #   if defined(__GNUC__) || defined(__clang__)
     __attribute__((__format__ (__printf__, 2, 3)))
@@ -556,12 +598,23 @@ extern void Log(const shared_ptr<Logger>& info_log, const char* format, ...)
 
 extern void LogFlush(Logger *info_log);
 
+extern void Log(const InfoLogLevel log_level, Logger* info_log,
+                const char* format, ...);
+
+// The default info log level is InfoLogLevel::ERROR.
 extern void Log(Logger* info_log, const char* format, ...)
 #   if defined(__GNUC__) || defined(__clang__)
     __attribute__((__format__ (__printf__, 2, 3)))
 #   endif
     ;
 
+// a set of log functions with different log levels.
+extern void Debug(Logger* info_log, const char* format, ...);
+extern void Info(Logger* info_log, const char* format, ...);
+extern void Warn(Logger* info_log, const char* format, ...);
+extern void Error(Logger* info_log, const char* format, ...);
+extern void Fatal(Logger* info_log, const char* format, ...);
+
 // A utility routine: write "data" to the named file.
 extern Status WriteStringToFile(Env* env, const Slice& data,
                                 const std::string& fname);

util/auto_roll_logger.h

@@ -17,20 +17,21 @@ namespace rocksdb {
 class AutoRollLogger : public Logger {
  public:
   AutoRollLogger(Env* env, const std::string& dbname,
-                 const std::string& db_log_dir,
-                 size_t log_max_size,
-                 size_t log_file_time_to_roll):
-     dbname_(dbname),
-     db_log_dir_(db_log_dir),
-     env_(env),
-     status_(Status::OK()),
-     kMaxLogFileSize(log_max_size),
-     kLogFileTimeToRoll(log_file_time_to_roll),
-     cached_now(static_cast<uint64_t>(env_->NowMicros() * 1e-6)),
-     ctime_(cached_now),
-     cached_now_access_count(0),
-     call_NowMicros_every_N_records_(100),
-     mutex_() {
+                 const std::string& db_log_dir, size_t log_max_size,
+                 size_t log_file_time_to_roll,
+                 const InfoLogLevel log_level = InfoLogLevel::ERROR)
+      : Logger(log_level),
+        dbname_(dbname),
+        db_log_dir_(db_log_dir),
+        env_(env),
+        status_(Status::OK()),
+        kMaxLogFileSize(log_max_size),
+        kLogFileTimeToRoll(log_file_time_to_roll),
+        cached_now(static_cast<uint64_t>(env_->NowMicros() * 1e-6)),
+        ctime_(cached_now),
+        cached_now_access_count(0),
+        call_NowMicros_every_N_records_(100),
+        mutex_() {
     env->GetAbsolutePath(dbname, &db_absolute_path_);
     log_fname_ = InfoLogFileName(dbname_, db_absolute_path_, db_log_dir_);
     RollLogFile();

util/auto_roll_logger_test.cc

@@ -39,10 +39,8 @@ class AutoRollLoggerTest {
 
 const string AutoRollLoggerTest::kSampleMessage(
     "this is the message to be written to the log file!!");
-const string AutoRollLoggerTest::kTestDir(
-    test::TmpDir() + "/db_log_test");
-const string AutoRollLoggerTest::kLogFile(
-    test::TmpDir() + "/db_log_test/LOG");
+const string AutoRollLoggerTest::kTestDir(test::TmpDir() + "/db_log_test");
+const string AutoRollLoggerTest::kLogFile(test::TmpDir() + "/db_log_test/LOG");
 Env* AutoRollLoggerTest::env = Env::Default();
 
 // In this test we only want to Log some simple log message with
@@ -53,6 +51,11 @@ void LogMessage(Logger* logger, const char* message) {
   Log(logger, "%s", message);
 }
 
+void LogMessage(const InfoLogLevel log_level, Logger* logger,
+                const char* message) {
+  Log(log_level, logger, "%s", message);
+}
+
 void GetFileCreateTime(const std::string& fname, uint64_t* file_ctime) {
   struct stat s;
   if (stat(fname.c_str(), &s) != 0) {
@@ -64,6 +67,7 @@ void GetFileCreateTime(const std::string& fname, uint64_t* file_ctime) {
 void AutoRollLoggerTest::RollLogFileBySizeTest(AutoRollLogger* logger,
                                                size_t log_max_size,
                                                const string& log_message) {
+  logger->SetInfoLogLevel(InfoLogLevel::INFO);
   // measure the size of each message, which is supposed
   // to be equal or greater than log_message.size()
   LogMessage(logger, log_message.c_str());
@@ -131,7 +135,6 @@ TEST(AutoRollLoggerTest, RollLogFileBySize) {
 
     RollLogFileBySizeTest(&logger, log_max_size,
                           kSampleMessage + ":RollLogFileBySize");
-
 }
 
 TEST(AutoRollLoggerTest, RollLogFileByTime) {
@@ -235,6 +238,55 @@ TEST(AutoRollLoggerTest, CreateLoggerFromOptions) {
       kSampleMessage + ":CreateLoggerFromOptions - both");
 }
 
+TEST(AutoRollLoggerTest, InfoLogLevel) {
+  InitTestDb();
+  // the lengths of DEBUG, INFO, WARN, ERROR, FATAL respectively
+  const int kInfoLogLevelNameLens[5] = {5, 4, 4, 5, 5};
+
+  size_t log_size = 8192;
+  AutoRollLogger* logger =
+      new AutoRollLogger(Env::Default(), kTestDir, "", log_size, 0);
+
+  int message_length = kSampleMessage.length();
+  int log_length = 0;
+  int total_logname_length = 0;
+  for (int log_level = InfoLogLevel::FATAL; log_level >= InfoLogLevel::DEBUG;
+       log_level--) {
+    logger->SetInfoLogLevel((InfoLogLevel)log_level);
+    total_logname_length += kInfoLogLevelNameLens[log_level];
+    for (int log_type = InfoLogLevel::DEBUG; log_type <= InfoLogLevel::FATAL;
+         log_type++) {
+      // log messages with log level smaller than log_level will not be logged.
+      LogMessage((InfoLogLevel)log_type, logger, kSampleMessage.c_str());
+    }
+    // 44 is the length of the message excluding the actual
+    // message and log name.
+    log_length += (message_length + 44) * (InfoLogLevel::FATAL - log_level + 1);
+    log_length += total_logname_length;
+    ASSERT_EQ(logger->GetLogFileSize(), log_length);
+  }
+
+  // rerun the test but using different log functions.
+  total_logname_length = 0;
+  for (int log_level = InfoLogLevel::FATAL; log_level >= InfoLogLevel::DEBUG;
+       log_level--) {
+    logger->SetInfoLogLevel((InfoLogLevel)log_level);
+    total_logname_length += kInfoLogLevelNameLens[log_level];
+
+    // again, messages with level smaller than log_level will not be logged.
+    Debug(logger, "%s", kSampleMessage.c_str());
+    Info(logger, "%s", kSampleMessage.c_str());
+    Warn(logger, "%s", kSampleMessage.c_str());
+    Error(logger, "%s", kSampleMessage.c_str());
+    Fatal(logger, "%s", kSampleMessage.c_str());
+    // 44 is the length of the message excluding the actual
+    // message and log name.
+    log_length += (message_length + 44) * (InfoLogLevel::FATAL - log_level + 1);
+    log_length += total_logname_length;
+    ASSERT_EQ(logger->GetLogFileSize(), log_length);
+  }
+}
+
 int OldLogFileCount(const string& dir) {
   std::vector<std::string> files;
   Env::Default()->GetChildren(dir, &files);

util/env.cc

@@ -45,12 +45,120 @@ void Log(Logger* info_log, const char* format, ...) {
   }
 }
 
+void Log(const InfoLogLevel log_level, Logger* info_log, const char* format,
+         ...) {
+  if (info_log) {
+    va_list ap;
+    va_start(ap, format);
+    info_log->Logv(log_level, format, ap);
+    va_end(ap);
+  }
+}
+
+void Debug(Logger* info_log, const char* format, ...) {
+  if (info_log) {
+    va_list ap;
+    va_start(ap, format);
+    info_log->Logv(InfoLogLevel::DEBUG, format, ap);
+    va_end(ap);
+  }
+}
+
+void Info(Logger* info_log, const char* format, ...) {
+  if (info_log) {
+    va_list ap;
+    va_start(ap, format);
+    info_log->Logv(InfoLogLevel::INFO, format, ap);
+    va_end(ap);
+  }
+}
+
+void Warn(Logger* info_log, const char* format, ...) {
+  if (info_log) {
+    va_list ap;
+    va_start(ap, format);
+    info_log->Logv(InfoLogLevel::WARN, format, ap);
+    va_end(ap);
+  }
+}
+void Error(Logger* info_log, const char* format, ...) {
+  if (info_log) {
+    va_list ap;
+    va_start(ap, format);
+    info_log->Logv(InfoLogLevel::ERROR, format, ap);
+    va_end(ap);
+  }
+}
+void Fatal(Logger* info_log, const char* format, ...) {
+  if (info_log) {
+    va_list ap;
+    va_start(ap, format);
+    info_log->Logv(InfoLogLevel::FATAL, format, ap);
+    va_end(ap);
+  }
+}
+
 void LogFlush(const shared_ptr<Logger>& info_log) {
   if (info_log) {
     info_log->Flush();
   }
 }
 
+void Log(const InfoLogLevel log_level, const shared_ptr<Logger>& info_log,
+         const char* format, ...) {
+  if (info_log) {
+    va_list ap;
+    va_start(ap, format);
+    info_log->Logv(log_level, format, ap);
+    va_end(ap);
+  }
+}
+
+void Debug(const shared_ptr<Logger>& info_log, const char* format, ...) {
+  if (info_log) {
+    va_list ap;
+    va_start(ap, format);
+    info_log->Logv(InfoLogLevel::DEBUG, format, ap);
+    va_end(ap);
+  }
+}
+
+void Info(const shared_ptr<Logger>& info_log, const char* format, ...) {
+  if (info_log) {
+    va_list ap;
+    va_start(ap, format);
+    info_log->Logv(InfoLogLevel::INFO, format, ap);
+    va_end(ap);
+  }
+}
+
+void Warn(const shared_ptr<Logger>& info_log, const char* format, ...) {
+  if (info_log) {
+    va_list ap;
+    va_start(ap, format);
+    info_log->Logv(InfoLogLevel::WARN, format, ap);
+    va_end(ap);
+  }
+}
+
+void Error(const shared_ptr<Logger>& info_log, const char* format, ...) {
+  if (info_log) {
+    va_list ap;
+    va_start(ap, format);
+    info_log->Logv(InfoLogLevel::ERROR, format, ap);
+    va_end(ap);
+  }
+}
+
+void Fatal(const shared_ptr<Logger>& info_log, const char* format, ...) {
+  if (info_log) {
+    va_list ap;
+    va_start(ap, format);
+    info_log->Logv(InfoLogLevel::FATAL, format, ap);
+    va_end(ap);
+  }
+}
+
 void Log(const shared_ptr<Logger>& info_log, const char* format, ...) {
   if (info_log) {
     va_list ap;

util/env_hdfs.cc

@@ -236,8 +236,9 @@ class HdfsLogger : public Logger {
   uint64_t (*gettid_)();  // Return the thread id for the current thread
 
  public:
-  HdfsLogger(HdfsWritableFile* f, uint64_t (*gettid)())
-    : file_(f), gettid_(gettid) {
+  HdfsLogger(HdfsWritableFile* f, uint64_t (*gettid)(),
+             const InfoLogLevel log_level = InfoLogLevel::ERROR)
+      : Logger(log_level), file_(f), gettid_(gettid) {
     Log(mylog, "[hdfs] HdfsLogger opened %s\n",
             file_->getName().c_str());
   }

util/posix_logger.h

@@ -38,9 +38,16 @@ class PosixLogger : public Logger {
   Env* env_;
   bool flush_pending_;
  public:
-  PosixLogger(FILE* f, uint64_t (*gettid)(), Env* env) :
-    file_(f), gettid_(gettid), log_size_(0), fd_(fileno(f)),
-    last_flush_micros_(0), env_(env), flush_pending_(false) { }
+  PosixLogger(FILE* f, uint64_t (*gettid)(), Env* env,
+              const InfoLogLevel log_level = InfoLogLevel::ERROR)
+      : Logger(log_level),
+        file_(f),
+        gettid_(gettid),
+        log_size_(0),
+        fd_(fileno(f)),
+        last_flush_micros_(0),
+        env_(env),
+        flush_pending_(false) {}
   virtual ~PosixLogger() {
     fclose(file_);
   }