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 bcf52bb314f..fca84fb97d8 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 644b11fd7e7..0139de4572b 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 d5b472e0123..2a72b25dbe0 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 d012e42ef22..115e3ab0bbb 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 b4531b7c15b..899cf0138cf 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 4d69d66484c..e4a5d95d821 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