[server] Expose KV WAL memory pool metrics for primary key tables

fluss-common/src/main/java/org/apache/fluss/memory/LazyMemorySegmentPool.java

@@ -64,7 +64,7 @@ public class LazyMemorySegmentPool implements MemorySegmentPool, Closeable {
     @GuardedBy("lock")
     private boolean closed;
 
-    private int pageUsage;
+    private volatile int pageUsage;
 
     @VisibleForTesting
     LazyMemorySegmentPool(
@@ -242,6 +242,11 @@ public int freePages() {
         return inLock(lock, () -> this.maxPages - this.pageUsage);
     }
 
+    /** Returns the number of pages currently allocated to callers. */
+    public int usedPages() {
+        return pageUsage;
+    }
+
     @Override
     public long availableMemory() {
         return ((long) freePages()) * pageSize;

fluss-common/src/main/java/org/apache/fluss/metrics/MetricNames.java

@@ -271,6 +271,13 @@ public class MetricNames {
     public static final String ROCKSDB_BLOCK_CACHE_PINNED_USAGE_TOTAL =
             "rocksdbBlockCachePinnedUsageTotal";
 
+    // Server-level KV WAL memory pool metrics for primary key tables
+    /** Memory used by the KV WAL memory pool for primary key tables in this server (bytes). */
+    public static final String KV_WAL_MEMORY_POOL_USAGE = "kvWalMemoryPoolUsage";
+
+    /** Total capacity of the KV WAL memory pool for primary key tables in this server (bytes). */
+    public static final String KV_WAL_MEMORY_POOL_CAPACITY = "kvWalMemoryPoolCapacity";
+
     // --------------------------------------------------------------------------------------------
     // metrics for table bucket
     // --------------------------------------------------------------------------------------------

fluss-server/src/main/java/org/apache/fluss/server/kv/KvManager.java

@@ -29,7 +29,6 @@
 import org.apache.fluss.fs.FileSystem;
 import org.apache.fluss.fs.FsPath;
 import org.apache.fluss.memory.LazyMemorySegmentPool;
-import org.apache.fluss.memory.MemorySegmentPool;
 import org.apache.fluss.metadata.KvFormat;
 import org.apache.fluss.metadata.PhysicalTablePath;
 import org.apache.fluss.metadata.SchemaGetter;
@@ -137,7 +136,7 @@ public static RateLimiter getDefaultRateLimiter() {
     private final BufferAllocator arrowBufferAllocator;
 
     /** The memory segment pool to allocate memorySegment. */
-    private final MemorySegmentPool memorySegmentPool;
+    private final LazyMemorySegmentPool memorySegmentPool;
 
     private final FsPath remoteKvDir;
 
@@ -206,6 +205,11 @@ private KvManager(
             this.sharedWriteBufferManager = createdWriteBufferManager;
             tabletServerMetricGroup.setSharedWriteBufferMetrics(
                     this::getSharedWriteBufferUsage, sharedWriteBufferCapacity);
+            // bind the pool as the data source locally so the supplier does not capture the
+            // whole KvManager, and convert pages to bytes here rather than in the metric group
+            LazyMemorySegmentPool walPool = memorySegmentPool;
+            tabletServerMetricGroup.registerKvWalMemoryPoolMetrics(
+                    () -> (long) walPool.usedPages() * walPool.pageSize(), walPool.totalSize());
         } catch (RuntimeException | Error e) {
             IOUtils.closeQuietly(createdWriteBufferManager);
             IOUtils.closeQuietly(createdWriteBufferAccountingCache);

fluss-server/src/main/java/org/apache/fluss/server/metrics/group/TabletServerMetricGroup.java

@@ -215,6 +215,18 @@ public void setSharedWriteBufferMetrics(LongSupplier usageSupplier, long capacit
         this.sharedWriteBufferCapacity = capacity;
     }
 
+    /**
+     * Registers gauges for the server-wide WAL memory pool used by primary key tables. Called once
+     * by KvManager when creating the server buffer pool.
+     *
+     * @param usageSupplier supplier for the bytes currently allocated from the pool
+     * @param capacity the total pool capacity in bytes
+     */
+    public void registerKvWalMemoryPoolMetrics(LongSupplier usageSupplier, long capacity) {
+        gauge(MetricNames.KV_WAL_MEMORY_POOL_USAGE, usageSupplier::getAsLong);
+        gauge(MetricNames.KV_WAL_MEMORY_POOL_CAPACITY, () -> capacity);
+    }
+
     @Override
     protected final void putVariables(Map<String, String> variables) {
         variables.put("cluster_id", clusterId);

fluss-server/src/test/java/org/apache/fluss/server/metrics/group/TabletServerMetricGroupTest.java

@@ -77,6 +77,18 @@ void testSharedWriteBufferMetrics() {
                 .isEqualTo(128L);
     }
 
+    @Test
+    void testWalMemoryPoolMetrics() {
+        TabletServerMetricGroup metricGroup =
+                new TabletServerMetricGroup(
+                        NOPMetricRegistry.INSTANCE, "cluster", "rack", "host", 0);
+        metricGroup.registerKvWalMemoryPoolMetrics(() -> 100L, 256L);
+
+        assertThat(gaugeValue(metricGroup, MetricNames.KV_WAL_MEMORY_POOL_USAGE)).isEqualTo(100L);
+        assertThat(gaugeValue(metricGroup, MetricNames.KV_WAL_MEMORY_POOL_CAPACITY))
+                .isEqualTo(256L);
+    }
+
     private static Object gaugeValue(TabletServerMetricGroup metricGroup, String metricName) {
         return ((Gauge<?>) metricGroup.getMetrics().get(metricName)).getValue();
     }

website/docs/maintenance/observability/monitor-metrics.md

@@ -463,8 +463,8 @@ Some metrics might not be exposed when using other JVM implementations (e.g. IBM
   </thead>
   <tbody>
     <tr>
-      <th rowspan="37"><strong>tabletserver</strong></th>
-      <td style={{textAlign: 'center', verticalAlign: 'middle' }} rowspan="25">-</td>
+      <th rowspan="39"><strong>tabletserver</strong></th>
+      <td style={{textAlign: 'center', verticalAlign: 'middle' }} rowspan="27">-</td>
       <td>messagesInPerSecond</td>
       <td>The number of messages written per second to this server.</td>
       <td>Meter</td>
@@ -589,6 +589,16 @@ Some metrics might not be exposed when using other JVM implementations (e.g. IBM
       <td>The number of kv pre-write buffer truncate due to the error happened when writing cdc to log per second.</td>
       <td>Meter</td>
     </tr>
+    <tr>
+      <td>kvWalMemoryPoolUsage</td>
+      <td>Memory currently allocated from the server-wide WAL memory pool for primary key tables in this server (in bytes). The pool capacity is configured by <code>server.buffer.memory-size</code>.</td>
+      <td>Gauge</td>
+    </tr>
+    <tr>
+      <td>kvWalMemoryPoolCapacity</td>
+      <td>Total capacity of the server-wide WAL memory pool for primary key tables in this server (in bytes).</td>
+      <td>Gauge</td>
+    </tr>
     <tr>
       <td rowspan="4">historical</td>
       <td>inflightRequests</td>