[#15274] docdb: Fix wal_size metric to reflect on-disk size, include active segment

Summary:
The `wal_size` metric reflects only the file size of readable log segments. `wal_size` is initialized to the size of all readable segments on `Log::Init()`, and only increased on the close of a current segment. This logic omits the size of the current active writable segment.

This code change fixes the metric by tracking the active segment size at these stages:
1. Allocation
1. Growth during append
1. Truncation at close before transition to readable segment

This change also introduces the APIs `WritableLogSegment::SizeOnDisk()` and `WritableFile::SizeOnDisk()` to get the true file size for the active segment, since the current `Size()` API returns the data size and not the file size.

The test `TestLogMetrics` grows the log to six segments and verifies the `wal_size` metric against actual file-sizes.
Jira: DB-4439

Test Plan:
```sh
./yb_build.sh --cxx-test='TEST_F(LogTest, TestLogMetrics) {' --test-args --vmodule=log=4
./yb_build.sh --cxx-test='TEST_F(TestEnv, TestHolePunch) {'
./yb_build.sh --cxx-test='TEST_F(TestEnv, TestRWFile) {'
```
Jenkins

Reviewers: yyan, rthallam, bkolagani, timur, sergei, mbautin, hector, qhu

Reviewed By: yyan, sergei

Subscribers: ybase, rthallam

Tags: #jenkins-ready

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

Author: srinskit

src/yb/consensus/log-test.cc

@@ -599,19 +599,26 @@ TEST_F(LogTest, TestLogMetrics) {
   ASSERT_OK(log_->GetLogReader()->GetSegmentsSnapshot(&segments));
   ASSERT_EQ(segments.size(), 1);
 
+  // Create at least 3 segments.
   while (segments.size() < 3) {
     ASSERT_OK(AppendNoOps(&op_id, kNumEntriesPerBatch));
     // Update the segments
     ASSERT_OK(log_->GetLogReader()->GetSegmentsSnapshot(&segments));
   }
 
-  ASSERT_OK(log_->Close());
+  uint64_t metrics_wal_size = log_->metrics_->wal_size->value();
+  uint64_t real_wal_size = 0;
 
-  int64_t wal_size_old = log_->metrics_->wal_size->value();
-  BuildLog();
-  int64_t wal_size_new = log_->metrics_->wal_size->value();
+  for (const scoped_refptr<ReadableLogSegment>& segment : segments) {
+    struct stat st;
+    ASSERT_EQ(stat(segment->path().c_str(), &st), 0);
+    uint64_t size_on_disk = st.st_size;
+    real_wal_size += size_on_disk;
+    LOG(INFO) << "Log size: " << size_on_disk << ", path: " << segment->path();
+  }
 
-  ASSERT_EQ(wal_size_old, wal_size_new);
+  ASSERT_OK(log_->Close());
+  ASSERT_EQ(real_wal_size, metrics_wal_size);
 }
 
 // Verify presence of min_start_time_running_txns in footer of closed segments.

src/yb/consensus/log.cc

@@ -752,11 +752,14 @@ Status Log::Init() {
     // In case where TServer process reboots, we need to reload the wal file size into the metric,
     // otherwise we do nothing
     if (metrics_ && metrics_->wal_size->value() == 0) {
-      std::for_each(segments.begin(), segments.end(),
-                    [this](const auto& segment) {
-                    this->metrics_->wal_size->IncrementBy(segment->file_size());});
+      for (const auto& segment : segments) {
+        int64_t size = segment->file_size();
+        this->metrics_->wal_size->IncrementBy(size);
+        VLOG_WITH_PREFIX(4)
+            << "Incremented wal size by " << size
+            << " to " << this->metrics_->wal_size->value();
+      }
     }
-
   }
 
   if (durable_wal_write_) {
@@ -815,9 +818,6 @@ Status Log::CloseCurrentSegment() {
                                                            &footer_builder_);
   }
 
-  if (status.ok() && metrics_) {
-      metrics_->wal_size->IncrementBy(active_segment_->Size());
-  }
   return status;
 }
 
@@ -986,17 +986,23 @@ Status Log::DoAppend(LogEntryBatch* entry_batch, SkipWalWrite skip_wal_write) {
     }
 
     // If the size of this entry overflows the current segment, get a new one.
+    bool segment_will_overflow =
+      active_segment_->Size() + entry_batch_bytes + kEntryHeaderSize > cur_max_segment_size_;
     if (allocation_state() == SegmentAllocationState::kAllocationNotStarted) {
-      if (active_segment_->Size() + entry_batch_bytes + kEntryHeaderSize > cur_max_segment_size_) {
+      if (segment_will_overflow) {
         LOG_WITH_PREFIX(INFO) << "Max segment size " << cur_max_segment_size_ << " reached. "
                               << "Starting new segment allocation.";
         RETURN_NOT_OK(AsyncAllocateSegment());
         if (!options_.async_preallocate_segments) {
           RETURN_NOT_OK(RollOver());
+          // Rollover prevents potential overflow.
+          segment_will_overflow = false;
         }
       }
     } else if (allocation_state() == SegmentAllocationState::kAllocationFinished) {
       RETURN_NOT_OK(RollOver());
+      // Rollover prevents potential overflow.
+      segment_will_overflow = false;
     } else {
       VLOG_WITH_PREFIX(1) << "Segment allocation already in progress...";
     }
@@ -1013,6 +1019,19 @@ Status Log::DoAppend(LogEntryBatch* entry_batch, SkipWalWrite skip_wal_write) {
 
     if (metrics_) {
       metrics_->bytes_logged->IncrementBy(active_segment_->written_offset() - start_offset);
+      VLOG_WITH_PREFIX(4) << "Total bytes logged: " << metrics_->bytes_logged->value();
+
+      // Check if append grew log file beyond pre-allocated size.
+      if (segment_will_overflow) {
+        int64_t size_on_disk = VERIFY_RESULT(active_segment_->SizeOnDisk());
+        if (size_on_disk != active_segment_ondisk_size_) {
+          metrics_->wal_size->IncrementBy(size_on_disk - active_segment_ondisk_size_);
+          VLOG_WITH_PREFIX(4)
+              << "Incremented wal size by " << (size_on_disk - active_segment_ondisk_size_)
+              << " to " << metrics_->wal_size->value();
+          active_segment_ondisk_size_ = size_on_disk;
+        }
+      }
     }
 
     // Populate the offset and sequence number for the entry batch if we did a WAL write.
@@ -1612,6 +1631,9 @@ Status Log::GC(int64_t min_op_idx, int32_t* num_gced) {
 
       if (metrics_) {
         metrics_->wal_size->IncrementBy(-1 * segment->file_size());
+        VLOG_WITH_PREFIX(4)
+            << "Decremented wal size by " << segment->file_size()
+            << " to " << metrics_->wal_size->value();
       }
     }
 
@@ -2054,6 +2076,17 @@ Status Log::SwitchToAllocatedSegment() {
       read_wal_mem_tracker_));
   RETURN_NOT_OK(readable_segment->Init(header, new_segment->first_entry_offset()));
   RETURN_NOT_OK(reader_->AppendEmptySegment(readable_segment));
+
+  if (metrics_) {
+    // Add the pre-allocated segment size to metric.
+    int64_t segment_size = VERIFY_RESULT(new_segment->SizeOnDisk());
+    metrics_->wal_size->IncrementBy(segment_size);
+    active_segment_ondisk_size_ = segment_size;
+    VLOG_WITH_PREFIX(4)
+        << "Incremented wal size by " << segment_size
+        << " to " << metrics_->wal_size->value();
+  }
+
   // Now set 'active_segment_' to the new segment.
   {
     std::lock_guard lock(active_segment_mutex_);
@@ -2083,6 +2116,17 @@ Status Log::ReplaceSegmentInReaderUnlocked() {
                                        active_segment_->footer(),
                                        active_segment_->first_entry_offset()));
 
+  if (metrics_) {
+    // The log may grow or truncate during close, hence update the log size metric before switching
+    // to the new segment.
+    int64_t size_on_disk = readable_segment->file_size();
+    metrics_->wal_size->IncrementBy(size_on_disk - active_segment_ondisk_size_);
+    VLOG_WITH_PREFIX(4)
+        << "Incremented wal size by " << (size_on_disk - active_segment_ondisk_size_)
+        << " to " << metrics_->wal_size->value();
+    active_segment_ondisk_size_ = 0;
+  }
+
   return reader_->ReplaceLastSegment(readable_segment);
 }
 

src/yb/consensus/log.h

@@ -706,6 +706,9 @@ class Log : public RefCountedThreadSafe<Log> {
   std::function<int64_t(const std::string&)> get_xcluster_min_index_to_retain_
       GUARDED_BY(get_xcluster_index_lock_);
 
+  // Tracks on-disk size of active log segment file for metric reporting.
+  int64_t active_segment_ondisk_size_ = 0;
+
   DISALLOW_COPY_AND_ASSIGN(Log);
 };
 

src/yb/consensus/log_util.h

@@ -32,6 +32,10 @@
 
 #pragma once
 
+#include <sys/stat.h>
+
+#include <iosfwd>
+#include <map>
 #include <memory>
 #include <string>
 #include <vector>
@@ -443,6 +447,10 @@ class WritableLogSegment {
     return writable_file_->Size();
   }
 
+  Result<uint64_t> SizeOnDisk() const {
+    return writable_file_->SizeOnDisk();
+  }
+
   // Appends the provided batch of data, including a header
   // and checksum.
   // Makes sure that the log segment has not been closed.

src/yb/util/env-test.cc

@@ -480,7 +480,7 @@ TEST_F(TestEnv, TestHolePunch) {
   ASSERT_OK(file->Write(0, slice));
   ASSERT_OK(file->Sync());
   uint64_t sz;
-  ASSERT_OK(file->Size(&sz));
+  sz = ASSERT_RESULT(file->Size());
   ASSERT_EQ(kOneMb, sz);
   uint64_t size_on_disk = ASSERT_RESULT(env_->GetFileSizeOnDisk(test_path));
   // Some kernels and filesystems (e.g. Centos 6.6 with XFS) aggressively
@@ -492,7 +492,7 @@ TEST_F(TestEnv, TestHolePunch) {
   uint64_t punch_amount = 4096 * 4;
   uint64_t new_size_on_disk;
   ASSERT_OK(file->PunchHole(4096, punch_amount));
-  ASSERT_OK(file->Size(&sz));
+  sz = ASSERT_RESULT(file->Size());
   ASSERT_EQ(kOneMb, sz);
   new_size_on_disk = ASSERT_RESULT(env_->GetFileSizeOnDisk(test_path));
   ASSERT_EQ(size_on_disk - punch_amount, new_size_on_disk);
@@ -850,7 +850,7 @@ TEST_F(TestEnv, TestRWFile) {
   ASSERT_OK(file->Read(0, kTestData.length(), &result, scratch.get()));
   ASSERT_EQ(result, kTestData);
   uint64_t sz;
-  ASSERT_OK(file->Size(&sz));
+  sz = ASSERT_RESULT(file->Size());
   ASSERT_EQ(kTestData.length(), sz);
 
   // Write past the end of the file and rewrite some of the interior.
@@ -863,7 +863,7 @@ TEST_F(TestEnv, TestRWFile) {
 
   // Retest.
   ASSERT_EQ(result, kNewTestData);
-  ASSERT_OK(file->Size(&sz));
+  sz = ASSERT_RESULT(file->Size());
   ASSERT_EQ(kNewTestData.length(), sz);
 
   // Make sure we can't overwrite it.

src/yb/util/env.h

@@ -295,7 +295,7 @@ class Env {
   // This should operate safely, not following any symlinks, etc.
   virtual Status DeleteRecursively(const std::string &dirname) = 0;
 
-  // Store the logical size of fname in *file_size.
+  // Returns the logical size of fname.
   virtual Result<uint64_t> GetFileSize(const std::string& fname) = 0;
 
   virtual Result<uint64_t> GetFileINode(const std::string& fname) = 0;
@@ -534,6 +534,9 @@ class WritableFile {
 
   virtual uint64_t Size() const = 0;
 
+  // Get the logical file size visible to the users.
+  virtual Result<uint64_t> SizeOnDisk() const = 0;
+
   // Returns the filename provided when the WritableFile was constructed.
   virtual const std::string& filename() const = 0;
 
@@ -563,6 +566,7 @@ class WritableFileWrapper : public WritableFile {
   Status Flush(FlushMode mode) override;
   Status Sync() override;
   uint64_t Size() const override { return target_->Size(); }
+  Result<uint64_t> SizeOnDisk() const override { return target_->SizeOnDisk(); }
   const std::string& filename() const override { return target_->filename(); }
 
  private:
@@ -649,7 +653,7 @@ class RWFile {
   virtual Status Close() = 0;
 
   // Retrieves the file's size.
-  virtual Status Size(uint64_t* size) const = 0;
+  virtual Result<uint64_t> Size() const = 0;
 
   // Returns the filename provided when the RWFile was constructed.
   virtual const std::string& filename() const = 0;

src/yb/util/env_posix.cc

@@ -255,6 +255,16 @@ static Status DoOpen(const string& filename, Env::CreateMode mode, int* fd, int
   return Status::OK();
 }
 
+// Get logical on-disk file size using file descriptor.
+Result<uint64_t> GetLogicalFileSize(int fd, const string& filename) {
+  ThreadRestrictions::AssertIOAllowed();
+  struct stat st;
+  if (fstat(fd, &st) == -1) {
+    return STATUS_IO_ERROR(filename, errno);
+  }
+  return st.st_size;
+}
+
 template <class Extractor>
 Result<uint64_t> GetFileStat(const std::string& fname, const char* event, Extractor extractor) {
   TRACE_EVENT1("io", event, "path", fname);
@@ -456,6 +466,11 @@ class PosixWritableFile : public WritableFile {
     return filesize_;
   }
 
+  Result<uint64_t> SizeOnDisk() const override {
+    TRACE_EVENT1("io", "PosixWritableFile::SizeOnDisk", "path", filename_);
+    return GetLogicalFileSize(fd_, filename_);
+  }
+
   const string& filename() const override { return filename_; }
 
  protected:
@@ -922,15 +937,9 @@ class PosixRWFile final : public RWFile {
     return s;
   }
 
-  Status Size(uint64_t* size) const override {
+  Result<uint64_t> Size() const override {
     TRACE_EVENT1("io", "PosixRWFile::Size", "path", filename_);
-    ThreadRestrictions::AssertIOAllowed();
-    struct stat st;
-    if (fstat(fd_, &st) == -1) {
-      return STATUS_IO_ERROR(filename_, errno);
-    }
-    *size = st.st_size;
-    return Status::OK();
+    return GetLogicalFileSize(fd_, filename_);
   }
 
   const string& filename() const override {

src/yb/util/memenv/memenv.cc

@@ -118,6 +118,8 @@ class WritableFileImpl : public WritableFile {
 
   uint64_t Size() const override { return file_->Size(); }
 
+  Result<uint64_t> SizeOnDisk() const override { return file_->Size(); }
+
   const string& filename() const override {
     return file_->filename();
   }
@@ -173,9 +175,8 @@ class RWFileImpl : public RWFile {
     return Status::OK();
   }
 
-  Status Size(uint64_t* size) const override {
-    *size = file_->Size();
-    return Status::OK();
+  Result<uint64_t> Size() const override {
+    return file_->Size();
   }
 
   const string& filename() const override {