Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Kaixuan-Duan wants to merge 1 commit into
base: main
Choose a base branch
from
Open

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions fluss-common/src/main/java/org/apache/fluss/memory/LazyMemorySegmentPool.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);

@platinumhamburg platinumhamburg Sep 8, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Could we make this accessor a lock-free read? The new gauge takes the same exclusive lock as page allocation and return, and allocation also performs heap allocation while holding that lock. This couples metric collection to the production allocation path in both directions: collection can contend with writers, and allocation can delay collection. An instantaneous usage value is sufficient here; we do not need a consistent snapshot across gauges. Please make pageUsage volatile and read it directly in usedPages(), while retaining the existing lock around accounting updates. The accessor should remain safe after pool closure so an in-flight collection can finish without throwing. Do not reset the accounting counter merely for monitoring during close, since pages can still be returned afterward.

}

@Override
public long availableMemory() {
return ((long) freePages()) * pageSize;
Expand All @@ -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);
}
Expand Down
10 changes: 10 additions & 0 deletions fluss-common/src/main/java/org/apache/fluss/metrics/MetricNames.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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";

@platinumhamburg platinumhamburg Sep 8, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Could we name these kvWalMemoryPoolUsage and kvWalMemoryPoolCapacity? This pool serves the KV write path, while the tabletserver metric group contains both Log and KV metrics. A kv prefix would make that distinction explicit and align with names such as kvFlushPerSecond and kvBackpressureMaxPressure. The Usage/Capacity suffixes already match the existing memory metrics.
Please also move these constants out of the RocksDB metrics section into a separate server-level KV WAL memory pool section. This is a Fluss WAL buffer, not a RocksDB resource. The registration method could follow the same naming: registerKvWalMemoryPoolMetrics.


/** 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 =
Expand Down
4 changes: 2 additions & 2 deletions fluss-server/src/main/java/org/apache/fluss/server/kv/KvManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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;

Expand Down Expand Up @@ -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);
Expand Down
14 changes: 14 additions & 0 deletions ...s-server/src/main/java/org/apache/fluss/server/metrics/group/TabletServerMetricGroup.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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");

@platinumhamburg platinumhamburg Sep 8, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Could we narrow this to registerWalMemoryPoolMetrics(LongSupplier usageSupplier, long capacity)? The metric group only needs a byte count and a fixed capacity, but currently depends on LazyMemorySegmentPool and knows how to convert pages into bytes. KvManager should bind the data source and perform the conversion; the metric group should only register the gauges. Capture a local pool variable in the supplier rather than implicitly capturing the entire manager. Register the supplier directly with the gauge without an additional supplier field. The existing metric-group shutdown unregisters and clears the gauges, and a safe numeric accessor tolerates any in-flight read. No separate cleanup callback or weak reference is needed. register also expresses the one-time operation accurately: duplicate metric registration retains the original gauge, so this method should not imply that it replaces an existing data source.

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);

@platinumhamburg platinumhamburg Sep 8, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please keep this PR focused on WAL pool usage and capacity and omit walMemoryPoolWaitingThreads. I do not see sufficient operational value in the waiting-thread gauge for this change to justify adding it to the exposed metric surface. Remove its constant, registration, assertions, and documentation entry, including the corresponding table row-span adjustment. Leave the existing queued() behavior unchanged and omit the Javadoc added to it by this PR. This is a scope recommendation, not a claim that queued() returns an incorrect value.

}

@Override
protected final void putVariables(Map<String, String> variables) {
variables.put("cluster_id", clusterId);
Expand Down
38 changes: 38 additions & 0 deletions ...rver/src/test/java/org/apache/fluss/server/metrics/group/TabletServerMetricGroupTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand Down Expand Up @@ -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<MemorySegment> 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();
}
Expand Down
19 changes: 17 additions & 2 deletions website/docs/maintenance/observability/monitor-metrics.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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="40"><strong>tabletserver</strong></th>
<td style={{textAlign: 'center', verticalAlign: 'middle' }} rowspan="28">-</td>
<td>messagesInPerSecond</td>
<td>The number of messages written per second to this server.</td>
<td>Meter</td>
Expand Down Expand Up @@ -589,6 +589,21 @@ 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>walMemoryPoolUsage</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>walMemoryPoolCapacity</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>walMemoryPoolWaitingThreads</td>
<td>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.</td>
<td>Gauge</td>
</tr>
<tr>
<td rowspan="4">historical</td>
<td>inflightRequests</td>
Expand Down
Loading