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..cc19eaac02e 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( @@ -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; 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..8a070c5b33a 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 @@ -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 // -------------------------------------------------------------------------------------------- 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..14a14714bc5 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,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); 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..dac5562c10d 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 @@ -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 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..924e306ebc8 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 @@ -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(); } diff --git a/website/docs/maintenance/observability/monitor-metrics.md b/website/docs/maintenance/observability/monitor-metrics.md index 4d69d66484c..4bca3191d84 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,16 @@ 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 + + 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 + + + kvWalMemoryPoolCapacity + Total capacity of the server-wide WAL memory pool for primary key tables in this server (in bytes). + Gauge + historical inflightRequests