From 644ad940a32bcb9f430f42562c08ea91683d1c26 Mon Sep 17 00:00:00 2001 From: duankaixuan <1417048384@qq.com> Date: Tue, 8 Sep 2026 14:57:34 +0800 Subject: [PATCH 1/2] [server] Expose WAL memory pool metrics for primary key tables --- .../fluss/memory/LazyMemorySegmentPool.java | 6 +++ .../org/apache/fluss/metrics/MetricNames.java | 10 +++++ .../org/apache/fluss/server/kv/KvManager.java | 4 +- .../group/TabletServerMetricGroup.java | 14 +++++++ .../group/TabletServerMetricGroupTest.java | 38 +++++++++++++++++++ .../observability/monitor-metrics.md | 19 +++++++++- 6 files changed, 87 insertions(+), 4 deletions(-) diff --git a/fluss-common/src/main/java/org/apache/fluss/memory/LazyMemorySegmentPool.java b/fluss-common/src/main/java/org/apache/fluss/memory/LazyMemorySegmentPool.java index bcf52bb314..fca84fb97d 100644 --- a/fluss-common/src/main/java/org/apache/fluss/memory/LazyMemorySegmentPool.java +++ b/fluss-common/src/main/java/org/apache/fluss/memory/LazyMemorySegmentPool.java @@ -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 inLock(lock, () -> pageUsage); + } + @Override public long availableMemory() { return ((long) freePages()) * pageSize; @@ -264,6 +269,7 @@ private void checkClosed() { } } + /** Returns the number of threads currently blocked waiting for pages. */ public int queued() { return inLock(lock, waiters::size); } diff --git a/fluss-common/src/main/java/org/apache/fluss/metrics/MetricNames.java b/fluss-common/src/main/java/org/apache/fluss/metrics/MetricNames.java index 644b11fd7e..0139de4572 100644 --- a/fluss-common/src/main/java/org/apache/fluss/metrics/MetricNames.java +++ b/fluss-common/src/main/java/org/apache/fluss/metrics/MetricNames.java @@ -250,6 +250,16 @@ public class MetricNames { public static final String ROCKSDB_SHARED_WRITE_BUFFER_CAPACITY = "rocksdbSharedWriteBufferCapacity"; + // Server-level WAL memory pool metrics for primary key tables + /** Memory used by the WAL memory pool for primary key tables in this server (bytes). */ + public static final String WAL_MEMORY_POOL_USAGE = "walMemoryPoolUsage"; + + /** Total capacity of the WAL memory pool for primary key tables in this server (bytes). */ + public static final String WAL_MEMORY_POOL_CAPACITY = "walMemoryPoolCapacity"; + + /** Number of threads currently waiting for pages from the WAL memory pool. */ + public static final String WAL_MEMORY_POOL_WAITING_THREADS = "walMemoryPoolWaitingThreads"; + // Table-level RocksDB memory metrics (Sum aggregation) /** Total memtable memory usage across all buckets of this table. */ public static final String ROCKSDB_MEMTABLE_MEMORY_USAGE_TOTAL = diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/KvManager.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/KvManager.java index d5b472e012..2a72b25dbe 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/KvManager.java +++ b/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,7 @@ private KvManager( this.sharedWriteBufferManager = createdWriteBufferManager; tabletServerMetricGroup.setSharedWriteBufferMetrics( this::getSharedWriteBufferUsage, sharedWriteBufferCapacity); + tabletServerMetricGroup.setWalMemoryPoolMetrics(memorySegmentPool); } catch (RuntimeException | Error e) { IOUtils.closeQuietly(createdWriteBufferManager); IOUtils.closeQuietly(createdWriteBufferAccountingCache); diff --git a/fluss-server/src/main/java/org/apache/fluss/server/metrics/group/TabletServerMetricGroup.java b/fluss-server/src/main/java/org/apache/fluss/server/metrics/group/TabletServerMetricGroup.java index d012e42ef2..115e3ab0bb 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/metrics/group/TabletServerMetricGroup.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/metrics/group/TabletServerMetricGroup.java @@ -17,6 +17,7 @@ package org.apache.fluss.server.metrics.group; +import org.apache.fluss.memory.LazyMemorySegmentPool; import org.apache.fluss.metadata.PhysicalTablePath; import org.apache.fluss.metadata.TableBucket; import org.apache.fluss.metadata.TablePath; @@ -215,6 +216,19 @@ 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 walMemoryPool the server-wide WAL memory segment pool + */ + public void setWalMemoryPoolMetrics(LazyMemorySegmentPool walMemoryPool) { + LazyMemorySegmentPool pool = checkNotNull(walMemoryPool, "walMemoryPool must not be null"); + gauge(MetricNames.WAL_MEMORY_POOL_USAGE, () -> (long) pool.usedPages() * pool.pageSize()); + gauge(MetricNames.WAL_MEMORY_POOL_CAPACITY, pool::totalSize); + gauge(MetricNames.WAL_MEMORY_POOL_WAITING_THREADS, pool::queued); + } + @Override protected final void putVariables(Map variables) { variables.put("cluster_id", clusterId); diff --git a/fluss-server/src/test/java/org/apache/fluss/server/metrics/group/TabletServerMetricGroupTest.java b/fluss-server/src/test/java/org/apache/fluss/server/metrics/group/TabletServerMetricGroupTest.java index b4531b7c15..899cf0138c 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/metrics/group/TabletServerMetricGroupTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/metrics/group/TabletServerMetricGroupTest.java @@ -17,6 +17,11 @@ package org.apache.fluss.server.metrics.group; +import org.apache.fluss.config.ConfigOptions; +import org.apache.fluss.config.Configuration; +import org.apache.fluss.config.MemorySize; +import org.apache.fluss.memory.LazyMemorySegmentPool; +import org.apache.fluss.memory.MemorySegment; import org.apache.fluss.metadata.PhysicalTablePath; import org.apache.fluss.metadata.TableBucket; import org.apache.fluss.metadata.TablePath; @@ -27,6 +32,9 @@ import org.junit.jupiter.api.Test; +import java.io.IOException; +import java.util.List; + import static org.assertj.core.api.Assertions.assertThat; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; @@ -77,6 +85,36 @@ void testSharedWriteBufferMetrics() { .isEqualTo(128L); } + @Test + void testWalMemoryPoolMetrics() throws IOException { + // 128kb total memory with 64kb pages gives a pool of 2 pages + Configuration conf = new Configuration(); + conf.set(ConfigOptions.SERVER_BUFFER_MEMORY_SIZE, MemorySize.parse("128kb")); + conf.set(ConfigOptions.SERVER_BUFFER_PAGE_SIZE, MemorySize.parse("64kb")); + LazyMemorySegmentPool pool = LazyMemorySegmentPool.createServerBufferPool(conf); + + TabletServerMetricGroup metricGroup = + new TabletServerMetricGroup( + NOPMetricRegistry.INSTANCE, "cluster", "rack", "host", 0); + metricGroup.setWalMemoryPoolMetrics(pool); + + assertThat(gaugeValue(metricGroup, MetricNames.WAL_MEMORY_POOL_USAGE)).isEqualTo(0L); + assertThat(gaugeValue(metricGroup, MetricNames.WAL_MEMORY_POOL_CAPACITY)) + .isEqualTo(128 * 1024L); + assertThat(gaugeValue(metricGroup, MetricNames.WAL_MEMORY_POOL_WAITING_THREADS)) + .isEqualTo(0); + + List pages = pool.allocatePages(2); + assertThat(gaugeValue(metricGroup, MetricNames.WAL_MEMORY_POOL_USAGE)) + .isEqualTo(2 * 64 * 1024L); + + pool.returnPage(pages.get(0)); + assertThat(gaugeValue(metricGroup, MetricNames.WAL_MEMORY_POOL_USAGE)) + .isEqualTo(64 * 1024L); + assertThat(gaugeValue(metricGroup, MetricNames.WAL_MEMORY_POOL_WAITING_THREADS)) + .isEqualTo(0); + } + private static Object gaugeValue(TabletServerMetricGroup metricGroup, String metricName) { return ((Gauge) metricGroup.getMetrics().get(metricName)).getValue(); } diff --git a/website/docs/maintenance/observability/monitor-metrics.md b/website/docs/maintenance/observability/monitor-metrics.md index 4d69d66484..e4a5d95d82 100644 --- a/website/docs/maintenance/observability/monitor-metrics.md +++ b/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 - tabletserver - - + tabletserver + - messagesInPerSecond The number of messages written per second to this server. Meter @@ -589,6 +589,21 @@ Some metrics might not be exposed when using other JVM implementations (e.g. IBM The number of kv pre-write buffer truncate due to the error happened when writing cdc to log per second. Meter + + walMemoryPoolUsage + 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 server.buffer.memory-size. + Gauge + + + walMemoryPoolCapacity + Total capacity of the server-wide WAL memory pool for primary key tables in this server (in bytes). + Gauge + + + walMemoryPoolWaitingThreads + The number of threads currently blocked waiting for pages from the server-wide WAL memory pool. A non-zero value indicates the pool is exhausted and writes to primary key tables are being throttled. + Gauge + historical inflightRequests From 2b32035a4ea4243c8c7629c8b463813c02175a7d Mon Sep 17 00:00:00 2001 From: duankaixuan <1417048384@qq.com> Date: Sat, 12 Sep 2026 01:20:15 +0800 Subject: [PATCH 2/2] address feedback --- .../fluss/memory/LazyMemorySegmentPool.java | 5 ++- .../org/apache/fluss/metrics/MetricNames.java | 17 ++++----- .../org/apache/fluss/server/kv/KvManager.java | 6 +++- .../group/TabletServerMetricGroup.java | 12 +++---- .../group/TabletServerMetricGroupTest.java | 36 +++---------------- .../observability/monitor-metrics.md | 13 +++---- 6 files changed, 28 insertions(+), 61 deletions(-) diff --git a/fluss-common/src/main/java/org/apache/fluss/memory/LazyMemorySegmentPool.java b/fluss-common/src/main/java/org/apache/fluss/memory/LazyMemorySegmentPool.java index fca84fb97d..cc19eaac02 100644 --- a/fluss-common/src/main/java/org/apache/fluss/memory/LazyMemorySegmentPool.java +++ b/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( @@ -244,7 +244,7 @@ public int freePages() { /** Returns the number of pages currently allocated to callers. */ public int usedPages() { - return inLock(lock, () -> pageUsage); + return pageUsage; } @Override @@ -269,7 +269,6 @@ private void checkClosed() { } } - /** Returns the number of threads currently blocked waiting for pages. */ public int queued() { return inLock(lock, waiters::size); } diff --git a/fluss-common/src/main/java/org/apache/fluss/metrics/MetricNames.java b/fluss-common/src/main/java/org/apache/fluss/metrics/MetricNames.java index 0139de4572..8a070c5b33 100644 --- a/fluss-common/src/main/java/org/apache/fluss/metrics/MetricNames.java +++ b/fluss-common/src/main/java/org/apache/fluss/metrics/MetricNames.java @@ -250,16 +250,6 @@ public class MetricNames { public static final String ROCKSDB_SHARED_WRITE_BUFFER_CAPACITY = "rocksdbSharedWriteBufferCapacity"; - // Server-level WAL memory pool metrics for primary key tables - /** Memory used by the WAL memory pool for primary key tables in this server (bytes). */ - public static final String WAL_MEMORY_POOL_USAGE = "walMemoryPoolUsage"; - - /** Total capacity of the WAL memory pool for primary key tables in this server (bytes). */ - public static final String WAL_MEMORY_POOL_CAPACITY = "walMemoryPoolCapacity"; - - /** Number of threads currently waiting for pages from the WAL memory pool. */ - public static final String WAL_MEMORY_POOL_WAITING_THREADS = "walMemoryPoolWaitingThreads"; - // Table-level RocksDB memory metrics (Sum aggregation) /** Total memtable memory usage across all buckets of this table. */ public static final String ROCKSDB_MEMTABLE_MEMORY_USAGE_TOTAL = @@ -281,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 // -------------------------------------------------------------------------------------------- diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/KvManager.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/KvManager.java index 2a72b25dbe..14a14714bc 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/KvManager.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/KvManager.java @@ -205,7 +205,11 @@ private KvManager( this.sharedWriteBufferManager = createdWriteBufferManager; tabletServerMetricGroup.setSharedWriteBufferMetrics( this::getSharedWriteBufferUsage, sharedWriteBufferCapacity); - tabletServerMetricGroup.setWalMemoryPoolMetrics(memorySegmentPool); + // 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); diff --git a/fluss-server/src/main/java/org/apache/fluss/server/metrics/group/TabletServerMetricGroup.java b/fluss-server/src/main/java/org/apache/fluss/server/metrics/group/TabletServerMetricGroup.java index 115e3ab0bb..dac5562c10 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/metrics/group/TabletServerMetricGroup.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/metrics/group/TabletServerMetricGroup.java @@ -17,7 +17,6 @@ package org.apache.fluss.server.metrics.group; -import org.apache.fluss.memory.LazyMemorySegmentPool; import org.apache.fluss.metadata.PhysicalTablePath; import org.apache.fluss.metadata.TableBucket; import org.apache.fluss.metadata.TablePath; @@ -220,13 +219,12 @@ public void setSharedWriteBufferMetrics(LongSupplier usageSupplier, long capacit * 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 walMemoryPool the server-wide WAL memory segment pool + * @param usageSupplier supplier for the bytes currently allocated from the pool + * @param capacity the total pool capacity in bytes */ - public void setWalMemoryPoolMetrics(LazyMemorySegmentPool walMemoryPool) { - LazyMemorySegmentPool pool = checkNotNull(walMemoryPool, "walMemoryPool must not be null"); - gauge(MetricNames.WAL_MEMORY_POOL_USAGE, () -> (long) pool.usedPages() * pool.pageSize()); - gauge(MetricNames.WAL_MEMORY_POOL_CAPACITY, pool::totalSize); - gauge(MetricNames.WAL_MEMORY_POOL_WAITING_THREADS, pool::queued); + 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 diff --git a/fluss-server/src/test/java/org/apache/fluss/server/metrics/group/TabletServerMetricGroupTest.java b/fluss-server/src/test/java/org/apache/fluss/server/metrics/group/TabletServerMetricGroupTest.java index 899cf0138c..924e306ebc 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/metrics/group/TabletServerMetricGroupTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/metrics/group/TabletServerMetricGroupTest.java @@ -17,11 +17,6 @@ package org.apache.fluss.server.metrics.group; -import org.apache.fluss.config.ConfigOptions; -import org.apache.fluss.config.Configuration; -import org.apache.fluss.config.MemorySize; -import org.apache.fluss.memory.LazyMemorySegmentPool; -import org.apache.fluss.memory.MemorySegment; import org.apache.fluss.metadata.PhysicalTablePath; import org.apache.fluss.metadata.TableBucket; import org.apache.fluss.metadata.TablePath; @@ -32,9 +27,6 @@ import org.junit.jupiter.api.Test; -import java.io.IOException; -import java.util.List; - import static org.assertj.core.api.Assertions.assertThat; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; @@ -86,33 +78,15 @@ void testSharedWriteBufferMetrics() { } @Test - void testWalMemoryPoolMetrics() throws IOException { - // 128kb total memory with 64kb pages gives a pool of 2 pages - Configuration conf = new Configuration(); - conf.set(ConfigOptions.SERVER_BUFFER_MEMORY_SIZE, MemorySize.parse("128kb")); - conf.set(ConfigOptions.SERVER_BUFFER_PAGE_SIZE, MemorySize.parse("64kb")); - LazyMemorySegmentPool pool = LazyMemorySegmentPool.createServerBufferPool(conf); - + void testWalMemoryPoolMetrics() { TabletServerMetricGroup metricGroup = new TabletServerMetricGroup( NOPMetricRegistry.INSTANCE, "cluster", "rack", "host", 0); - metricGroup.setWalMemoryPoolMetrics(pool); - - assertThat(gaugeValue(metricGroup, MetricNames.WAL_MEMORY_POOL_USAGE)).isEqualTo(0L); - assertThat(gaugeValue(metricGroup, MetricNames.WAL_MEMORY_POOL_CAPACITY)) - .isEqualTo(128 * 1024L); - assertThat(gaugeValue(metricGroup, MetricNames.WAL_MEMORY_POOL_WAITING_THREADS)) - .isEqualTo(0); + metricGroup.registerKvWalMemoryPoolMetrics(() -> 100L, 256L); - List pages = pool.allocatePages(2); - assertThat(gaugeValue(metricGroup, MetricNames.WAL_MEMORY_POOL_USAGE)) - .isEqualTo(2 * 64 * 1024L); - - pool.returnPage(pages.get(0)); - assertThat(gaugeValue(metricGroup, MetricNames.WAL_MEMORY_POOL_USAGE)) - .isEqualTo(64 * 1024L); - assertThat(gaugeValue(metricGroup, MetricNames.WAL_MEMORY_POOL_WAITING_THREADS)) - .isEqualTo(0); + 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) { diff --git a/website/docs/maintenance/observability/monitor-metrics.md b/website/docs/maintenance/observability/monitor-metrics.md index e4a5d95d82..4bca3191d8 100644 --- a/website/docs/maintenance/observability/monitor-metrics.md +++ b/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 - tabletserver - - + tabletserver + - messagesInPerSecond The number of messages written per second to this server. Meter @@ -590,20 +590,15 @@ Some metrics might not be exposed when using other JVM implementations (e.g. IBM Meter - walMemoryPoolUsage + kvWalMemoryPoolUsage 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 server.buffer.memory-size. Gauge - walMemoryPoolCapacity + kvWalMemoryPoolCapacity Total capacity of the server-wide WAL memory pool for primary key tables in this server (in bytes). Gauge - - walMemoryPoolWaitingThreads - The number of threads currently blocked waiting for pages from the server-wide WAL memory pool. A non-zero value indicates the pool is exhausted and writes to primary key tables are being throttled. - Gauge - historical inflightRequests