[yugabyte#30307] DocDB: Convert spammy logs to DETAIL level

Summary:
Introduces a new `DETAIL_LEVEL` to the RocksDB `InfoLogLevel` enum, providing a way to tag high-frequency informational logs with a "DETAIL: " prefix while keeping them visible at the default INFO log level. This enables downstream log filtering/grep without changing log visibility semantics.
Summary of changes:
- Added DETAIL_LEVEL to InfoLogLevel enum and GetFilterLogLevel() helper so DETAIL and INFO share the same visibility threshold
- Added per-entry log level to LogBuffer so buffered DETAIL entries flush with the correct level
- Added log-level support to EventLogger/EventLoggerStream so individual event log call sites can opt into DETAIL
- Unified the "DETAIL: " string literal into the shared YB_DETAIL_LOG_PREFIX macro
- Using all of the above for log level migrations (INFO/DEBUG -> DETAIL)

Jira: DB-20188

Test Plan:
Jenkins: all tests

Added some unit tests to:
`yb_rocksdb_logger-test.cc`, `auto_roll_logger_test.cc` and `env_test.cc`

Verification through logs of local test runs:
`ybd release --cxx-test flush-test --gtest_filter '*.TestFlushPicksOldestInactiveTabletAfterCompaction'`
Some DETAIL log lines:
```
[ts-1] I0212 22:54:26.167357 86911 memtable_list.cc:410] DETAIL: T 08a2e6c27d134601876040adbd3c012a P 51dea67900734de7b314d7687b3ad0ea [R]: [default] Level-0 commit table #14 started
[ts-1] I0212 22:54:26.167373 86911 memtable_list.cc:426] DETAIL: T 08a2e6c27d134601876040adbd3c012a P 51dea67900734de7b314d7687b3ad0ea [R]: [default] Level-0 commit table #14: memtable #1 done
[ts-1] I0212 22:54:26.167378 86911 event_logger.cc:89] DETAIL: T 08a2e6c27d134601876040adbd3c012a P 51dea67900734de7b314d7687b3ad0ea [R]: EVENT_LOG_v1 {"time_micros": 1770936866166824, "job": 5, "event": "flush_finished", "lsm_state": [4]}
[ts-1] I0212 22:54:26.169718 87257 tablet_retention_policy.cc:203] DETAIL: T 08a2e6c27d134601876040adbd3c012a P 51dea67900734de7b314d7687b3ad0ea: SanitizeHistoryCutoff, cutoff from the provider { cotables_cutoff_ht: <invalid> primary_cutoff_ht: <max> }
```

Reviewers: hsunder, mlillibridge, sergei, arybochkin

Reviewed By: mlillibridge, arybochkin

Subscribers: sanketh, ybase

Differential Revision: https://phorge.dev.yugabyte.com/D50302

Author: rahulb-yb

src/yb/consensus/log.cc

@@ -838,7 +838,7 @@ Status Log::RollOver() {
 
     DCHECK_EQ(allocation_state(), SegmentAllocationState::kAllocationFinished);
 
-    LOG_WITH_PREFIX(INFO) << Format(
+    LOG_WITH_PREFIX_DETAIL << Format(
         "Last appended OpId in segment $0: $1", active_segment_->path(),
         last_appended_entry_op_id_.ToString());
 
@@ -847,7 +847,7 @@ Status Log::RollOver() {
 
     RETURN_NOT_OK(SwitchToAllocatedSegment());
 
-    LOG_WITH_PREFIX(INFO) << "Rolled over to a new segment: " << active_segment_->path();
+    LOG_WITH_PREFIX_DETAIL << "Rolled over to a new segment: " << active_segment_->path();
   }
   return Status::OK();
 }
@@ -1599,7 +1599,7 @@ OpId Log::WaitForSafeOpIdToApply(const OpId& min_allowed, MonoDelta duration) {
 Status Log::GC(int64_t min_op_idx, int32_t* num_gced) {
   CHECK_GE(min_op_idx, 0);
 
-  LOG_WITH_PREFIX(INFO) << "Running Log GC on " << wal_dir_ << ": retaining ops >= " << min_op_idx
+  LOG_WITH_PREFIX_DETAIL << "Running Log GC on " << wal_dir_ << ": retaining ops >= " << min_op_idx
                         << ", log segment size = " << options_.segment_size_bytes;
   VLOG_TIMING(1, "Log GC") {
     SegmentSequence segments_to_delete;
@@ -1625,9 +1625,10 @@ Status Log::GC(int64_t min_op_idx, int32_t* num_gced) {
     // Now that they are no longer referenced by the Log, delete the files.
     *num_gced = 0;
     for (const scoped_refptr<ReadableLogSegment>& segment : segments_to_delete) {
-      LOG_WITH_PREFIX(INFO) << "Deleting log segment in path: " << segment->path()
-                            << " (GCed ops < " << segment->footer().max_replicate_index() + 1
-                            << ")";
+      LOG_WITH_PREFIX_DETAIL
+          << "Deleting log segment in path: " << segment->path()
+          << " (GCed ops < " << segment->footer().max_replicate_index() + 1
+          << ")";
       RETURN_NOT_OK(get_env()->DeleteFile(segment->path()));
       (*num_gced)++;
 

src/yb/consensus/log_reader.cc

@@ -510,7 +510,7 @@ Status LogReader::TrimSegmentsUpToAndIncluding(const int64_t segment_sequence_nu
     RETURN_NOT_OK(segments_.pop_front());
     deleted_segments.push_back(current_seq_no);
   }
-  LOG_WITH_PREFIX(INFO) << "Removed log segment sequence numbers from log reader: "
+  LOG_WITH_PREFIX_DETAIL << "Removed log segment sequence numbers from log reader: "
                         << yb::ToString(deleted_segments);
   return Status::OK();
 }

src/yb/docdb/docdb_rocksdb_util.cc

@@ -933,7 +933,7 @@ class RocksDBPatcher::Impl {
       auto& consensus_frontier = down_cast<ConsensusFrontier&>(*file.largest.user_frontier);
       // If all the data in the file is already as of old time, no need to set any filter.
       if (consensus_frontier.hybrid_time() <= value) {
-        LOG(INFO) << "No need to set hybrid time filter since the largest frontier is already"
+        LOG_DETAIL << "No need to set hybrid time filter since the largest frontier is already"
                   << " older. Largest frontier HT " << consensus_frontier.hybrid_time()
                   << ", filter HT " << value;
         return;

src/yb/rocksdb/db/auto_roll_logger_test.cc

@@ -363,23 +363,37 @@ TEST_F(AutoRollLoggerTest, InfoLogLevel) {
       for (int log_type = InfoLogLevel::DEBUG_LEVEL;
            log_type <= InfoLogLevel::HEADER_LEVEL; log_type++) {
         // log messages with log level smaller than log_level will not be
-        // logged.
+        // logged, except that DETAIL_LEVEL messages are treated as equivalent
+        // to INFO_LEVEL for filtering (via ShouldLog): if log_level is
+        // INFO_LEVEL, DETAIL messages will also be logged alongside INFO.
         LogMessage((InfoLogLevel)log_type, &logger, kSampleMessage.c_str());
       }
       log_lines += InfoLogLevel::HEADER_LEVEL - log_level + 1;
+      // DETAIL_LEVEL is treated as equivalent to INFO_LEVEL for filtering
+      // (via ShouldLog), so at INFO_LEVEL the DETAIL message also passes.
+      if (log_level == InfoLogLevel::INFO_LEVEL) {
+        log_lines++;
+      }
     }
     for (int log_level = InfoLogLevel::HEADER_LEVEL;
          log_level >= InfoLogLevel::DEBUG_LEVEL; log_level--) {
       logger.SetInfoLogLevel((InfoLogLevel)log_level);
 
-      // again, messages with level smaller than log_level will not be logged.
+      // again, messages with level smaller than log_level will not be logged,
+      // with the exception of DETAIL_LEVEL messages and an INFO_LEVEL logger.
       RLOG(InfoLogLevel::HEADER_LEVEL, &logger, "%s", kSampleMessage.c_str());
       RDEBUG(&logger, "%s", kSampleMessage.c_str());
+      RLOG(InfoLogLevel::DETAIL_LEVEL, &logger, "%s", kSampleMessage.c_str());
       RINFO(&logger, "%s", kSampleMessage.c_str());
       RWARN(&logger, "%s", kSampleMessage.c_str());
       RERROR(&logger, "%s", kSampleMessage.c_str());
       RFATAL(&logger, "%s", kSampleMessage.c_str());
       log_lines += InfoLogLevel::HEADER_LEVEL - log_level + 1;
+      // DETAIL_LEVEL is treated as equivalent to INFO_LEVEL for filtering
+      // (via ShouldLog), so at INFO_LEVEL the DETAIL message also passes.
+      if (log_level == InfoLogLevel::INFO_LEVEL) {
+        log_lines++;
+      }
     }
   }
   std::ifstream inFile(AutoRollLoggerTest::kLogFile.c_str());
@@ -485,6 +499,23 @@ TEST_F(AutoRollLoggerTest, LogHeaderTest) {
   }
 }
 
+TEST_F(AutoRollLoggerTest, DetailLevelFiltering) {
+  InitTestDb();
+
+  size_t log_size = 8192;
+  {
+    AutoRollLogger logger(Env::Default(), kTestDir, "", log_size, 0);
+    logger.SetInfoLogLevel(InfoLogLevel::INFO_LEVEL);
+
+    RLOG(InfoLogLevel::DETAIL_LEVEL, &logger, "test_detail");
+    RLOG(InfoLogLevel::INFO_LEVEL, &logger, "test_info");
+  }
+
+  // DETAIL messages pass at INFO level (ShouldLog treats DETAIL as equivalent to INFO).
+  ASSERT_EQ(1u, GetLinesCount(kLogFile, "test_detail"));
+  ASSERT_EQ(1u, GetLinesCount(kLogFile, "test_info"));
+}
+
 TEST_F(AutoRollLoggerTest, LogFileExistence) {
   rocksdb::DB* db;
   rocksdb::Options options;

src/yb/rocksdb/db/compaction_job.cc

@@ -1053,7 +1053,7 @@ Status CompactionJob::InstallCompactionResults(
 
   {
     Compaction::InputLevelSummaryBuffer inputs_summary;
-    RLOG(InfoLogLevel::INFO_LEVEL, db_options_.info_log,
+    RLOG(InfoLogLevel::DETAIL_LEVEL, db_options_.info_log,
         "[%s] [JOB %d] Compacted %s => %" PRIu64 " bytes",
         compaction->column_family_data()->GetName().c_str(), job_id_,
         compaction->InputLevelSummary(&inputs_summary), compact_->total_bytes);
@@ -1068,7 +1068,7 @@ Status CompactionJob::InstallCompactionResults(
     }
   }
   if (largest_user_frontier_) {
-    LOG_WITH_PREFIX(INFO) << "Updating flushed frontier to " << largest_user_frontier_->ToString();
+    LOG_WITH_PREFIX_DETAIL << "Updating flushed frontier to " << largest_user_frontier_->ToString();
     compaction->edit()->UpdateFlushedFrontier(largest_user_frontier_);
   }
   return versions_->LogAndApply(compaction->column_family_data(),
@@ -1333,14 +1333,14 @@ void CompactionJob::LogCompaction() {
       }
     }
     Compaction::InputLevelSummaryBuffer inputs_summary;
-    RLOG(InfoLogLevel::INFO_LEVEL, db_options_.info_log,
+    RLOG(InfoLogLevel::DETAIL_LEVEL, db_options_.info_log,
         "[%s] [JOB %d] Compacting %s, score %.2f", cfd->GetName().c_str(),
         job_id_, compaction->InputLevelSummary(&inputs_summary),
         compaction->score());
     char scratch[2345];
     compaction->Summary(scratch, sizeof(scratch));
     RLOG(
-        InfoLogLevel::INFO_LEVEL, db_options_.info_log, "[%s] Compaction start summary: %s%s\n",
+        InfoLogLevel::DETAIL_LEVEL, db_options_.info_log, "[%s] Compaction start summary: %s%s\n",
         cfd->GetName().c_str(), scratch,
         compaction->skip_corrupt_data_blocks_unsafe()
             ? " WILL DELETE ANY CORRUPT DATA BLOCKS IF FOUND"

src/yb/rocksdb/db/compaction_picker.cc

@@ -1571,7 +1571,7 @@ std::unique_ptr<Compaction> UniversalCompactionPicker::DoPickCompaction(
         cf_name, mutable_cf_options, vstorage, score, sorted_runs, log_buffer);
   }
   if (c) {
-    LOG_TO_BUFFER(log_buffer, "[%s] Universal: compacting for direct deletion\n",
+    LOG_TO_BUFFER_DETAIL(log_buffer, "[%s] Universal: compacting for direct deletion\n",
                   cf_name.c_str());
   } else {
     // Check if the number of files to compact is greater than or equal to
@@ -1587,7 +1587,7 @@ std::unique_ptr<Compaction> UniversalCompactionPicker::DoPickCompaction(
     c = PickCompactionUniversalSizeAmp(cf_name, mutable_cf_options, vstorage,
                                             score, sorted_runs, log_buffer);
     if (c) {
-      LOG_TO_BUFFER(log_buffer, "[%s] Universal: compacting for size amp\n",
+      LOG_TO_BUFFER_DETAIL(log_buffer, "[%s] Universal: compacting for size amp\n",
                   cf_name.c_str());
     } else {
       // Size amplification is within limits. Try reducing read
@@ -1599,7 +1599,7 @@ std::unique_ptr<Compaction> UniversalCompactionPicker::DoPickCompaction(
           ioptions_.compaction_options_universal.always_include_size_threshold,
           sorted_runs, log_buffer);
       if (c) {
-        LOG_TO_BUFFER(log_buffer, "[%s] Universal: compacting for size ratio\n",
+        LOG_TO_BUFFER_DETAIL(log_buffer, "[%s] Universal: compacting for size ratio\n",
                     cf_name.c_str());
       } else {
         // ENG-1401: We trigger compaction logic when num files exceeds
@@ -1628,7 +1628,7 @@ std::unique_ptr<Compaction> UniversalCompactionPicker::DoPickCompaction(
                       cf_name, mutable_cf_options, vstorage, score, UINT_MAX, num_files,
                       ioptions_.compaction_options_universal.always_include_size_threshold,
                       sorted_runs, log_buffer)) != nullptr) {
-            LOG_TO_BUFFER(log_buffer,
+            LOG_TO_BUFFER_DETAIL(log_buffer,
                           "[%s] Universal: compacting for file num -- %u\n",
                           cf_name.c_str(), num_files);
           }
@@ -1896,8 +1896,8 @@ std::unique_ptr<Compaction> UniversalCompactionPicker::PickCompactionUniversalRe
     }
     char file_num_buf[kFormatFileSizeInfoBufSize];
     picking_sr.DumpSizeInfo(file_num_buf, sizeof(file_num_buf), i);
-    LOG_TO_BUFFER(log_buffer, "[%s] Universal: Picking %s", cf_name.c_str(),
-                file_num_buf);
+    LOG_TO_BUFFER_DETAIL(
+        log_buffer, "[%s] Universal: Picking %s", cf_name.c_str(), file_num_buf);
   }
 
   CompactionReason compaction_reason;

src/yb/rocksdb/db/db_filesnapshot.cc

@@ -48,8 +48,7 @@ Status DBImpl::DisableFileDeletions() {
   InstrumentedMutexLock l(&mutex_);
   ++disable_delete_obsolete_files_;
   if (disable_delete_obsolete_files_ == 1) {
-    RLOG(InfoLogLevel::INFO_LEVEL, db_options_.info_log,
-        "File Deletions Disabled");
+    RLOG(InfoLogLevel::DETAIL_LEVEL, db_options_.info_log, "File Deletions Disabled");
   } else {
     RLOG(InfoLogLevel::WARN_LEVEL, db_options_.info_log,
         "File Deletions Disabled, but already disabled. Counter: %d",
@@ -72,7 +71,7 @@ Status DBImpl::EnableFileDeletions(bool force) {
       --disable_delete_obsolete_files_;
     }
     if (disable_delete_obsolete_files_ == 0) {
-      RLOG(InfoLogLevel::INFO_LEVEL, db_options_.info_log,
+      RLOG(InfoLogLevel::DETAIL_LEVEL, db_options_.info_log,
           "File Deletions Enabled");
       should_purge_files = true;
       FindObsoleteFiles(&job_context, true);

src/yb/rocksdb/db/flush_job.cc

@@ -227,7 +227,7 @@ Result<FileNumbersHolder> FlushJob::Run(FileMetaData* file_meta) {
   // This includes both SST and MANIFEST files IO.
   RecordFlushIOStats();
 
-  auto stream = event_logger_->LogToBuffer(log_buffer_);
+  auto stream = event_logger_->LogToBuffer(log_buffer_, InfoLogLevel::DETAIL_LEVEL);
   stream << "job" << job_context_->job_id << "event"
          << "flush_finished";
   stream << "lsm_state";
@@ -263,7 +263,7 @@ Result<FileNumbersHolder> FlushJob::WriteLevel0Table(
     uint64_t total_num_entries = 0, total_num_deletes = 0;
     size_t total_memory_usage = 0;
     for (MemTable* m : mems) {
-      RLOG(InfoLogLevel::INFO_LEVEL, db_options_.info_log,
+      RLOG(InfoLogLevel::DETAIL_LEVEL, db_options_.info_log,
           "[%s] [JOB %d] Flushing memtable with next log file: %" PRIu64 "\n",
           cfd_->GetName().c_str(), job_context_->job_id, m->GetNextLogNumber());
       memtables.push_back(m->NewIterator(ro, &arena));
@@ -293,7 +293,7 @@ Result<FileNumbersHolder> FlushJob::WriteLevel0Table(
       ScopedArenaIterator iter(
           NewMergingIterator(cfd_->internal_comparator().get(), &memtables[0],
                              static_cast<int>(memtables.size()), &arena));
-      RLOG(InfoLogLevel::INFO_LEVEL, db_options_.info_log,
+      RLOG(InfoLogLevel::DETAIL_LEVEL, db_options_.info_log,
           "[%s] [JOB %d] Level-0 flush table #%" PRIu64 ": started",
           cfd_->GetName().c_str(), job_context_->job_id, meta->fd.GetNumber());
 
@@ -320,7 +320,7 @@ Result<FileNumbersHolder> FlushJob::WriteLevel0Table(
       info.table_properties = table_properties_;
       LogFlush(db_options_.info_log);
     }
-    RLOG(InfoLogLevel::INFO_LEVEL, db_options_.info_log,
+    RLOG(InfoLogLevel::DETAIL_LEVEL, db_options_.info_log,
         "[%s] [JOB %d] Level-0 flush table #%" PRIu64 ": %" PRIu64
         " bytes %s%s %s",
         cfd_->GetName().c_str(), job_context_->job_id, meta->fd.GetNumber(),

src/yb/rocksdb/db/memtable_list.cc

@@ -407,8 +407,8 @@ Status MemTableList::InstallMemtableFlushResults(
       break;
     }
 
-    LOG_TO_BUFFER(log_buffer, "[%s] Level-0 commit table #%" PRIu64 " started",
-                cfd->GetName().c_str(), m->file_number_);
+    LOG_TO_BUFFER_DETAIL(log_buffer, "[%s] Level-0 commit table #%" PRIu64 " started",
+                         cfd->GetName().c_str(), m->file_number_);
 
     // this can release and reacquire the mutex.
     s = vset->LogAndApply(cfd, mutable_cf_options, &m->edit_, mu, db_directory);
@@ -422,9 +422,9 @@ Status MemTableList::InstallMemtableFlushResults(
     uint64_t mem_id = 1;  // how many memtables have been flushed.
     do {
       if (s.ok()) { // commit new state
-        LOG_TO_BUFFER(log_buffer, "[%s] Level-0 commit table #%" PRIu64
-                                ": memtable #%" PRIu64 " done",
-                    cfd->GetName().c_str(), m->file_number_, mem_id);
+        LOG_TO_BUFFER_DETAIL(log_buffer, "[%s] Level-0 commit table #%" PRIu64
+                                         ": memtable #%" PRIu64 " done",
+                             cfd->GetName().c_str(), m->file_number_, mem_id);
         assert(m->file_number_ > 0);
         current_->Remove(m, to_delete);
       } else {

src/yb/rocksdb/env.h

@@ -454,6 +454,7 @@ class Directory {
 
 enum InfoLogLevel : unsigned char {
   DEBUG_LEVEL = 0,
+  DETAIL_LEVEL,
   INFO_LEVEL,
   WARN_LEVEL,
   ERROR_LEVEL,
@@ -462,6 +463,24 @@ enum InfoLogLevel : unsigned char {
   NUM_INFO_LOG_LEVELS,
 };
 
+// Returns true if a message at `message_level` should be logged given a logger
+// configured at `logger_level`. DETAIL_LEVEL is treated as equivalent to
+// INFO_LEVEL for filtering: a logger at INFO will show DETAIL messages and
+// vice versa.
+inline bool ShouldLog(InfoLogLevel message_level, InfoLogLevel logger_level) {
+  if (message_level >= logger_level) {
+    return true;
+  }
+
+  // Special handling for DETAIL_LEVEL.
+  if (message_level == InfoLogLevel::DETAIL_LEVEL && logger_level == InfoLogLevel::INFO_LEVEL) {
+    return true;
+  }
+
+  // Nothing to log in accordance with log levels.
+  return false;
+}
+
 // An interface for writing log messages.
 class Logger {
  public: