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 ac60ee5f10..a4ab737a37 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 @@ -256,6 +256,20 @@ public class MetricNames { public static final String ROCKSDB_SHARED_WRITE_BUFFER_CAPACITY = "rocksdbSharedWriteBufferCapacity"; + // Server-level pre-write buffer metrics (aggregated from all KV tablets, Sum aggregation) + /** + * Estimated memory usage of the pre-write buffers across all KV tablets in this server (Sum + * aggregation). + */ + public static final String KV_PRE_WRITE_BUFFER_MEMORY_USAGE_BYTES = + "kvPreWriteBufferMemoryUsageBytes"; + + /** + * Number of entries buffered in the pre-write buffers across all KV tablets in this server (Sum + * aggregation). + */ + public static final String KV_PRE_WRITE_BUFFER_ENTRY_COUNT = "kvPreWriteBufferEntryCount"; + // 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/KvTablet.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/KvTablet.java index 2312b6fd4e..ffe33dcf26 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/KvTablet.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/KvTablet.java @@ -1345,6 +1345,9 @@ public void close(KvCloseMode closeMode) throws Exception { // Terminal transition: closing forces IDLE regardless of the current // state, see the FlushState state graph. flushState = FlushState.IDLE; + // Release the remaining pre-write buffer accounting to the shared + // ledger while the local accounting values are still exact. + kvPreWriteBuffer.close(); return true; }); if (shouldClose && closeFlushScheduler) { diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/prewrite/KvPreWriteBuffer.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/prewrite/KvPreWriteBuffer.java index 18e922cf3f..6a1fa4d796 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/prewrite/KvPreWriteBuffer.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/prewrite/KvPreWriteBuffer.java @@ -87,7 +87,20 @@ * head to tail, it will stop flush. */ @NotThreadSafe -public class KvPreWriteBuffer { +public class KvPreWriteBuffer implements AutoCloseable { + + /** + * Estimated JVM heap overhead of a single buffered entry besides its key/value payload bytes, + * covering the {@link KvEntry} object, its {@link Key} and {@link Value} wrappers, the byte + * array headers and the linked list node. + */ + private static final long PER_ENTRY_OVERHEAD_BYTES = 144; + + /** + * Estimated JVM heap overhead of a hash map node for an entry that is the latest version of its + * key in the buffer. + */ + private static final long PER_MAP_NODE_OVERHEAD_BYTES = 32; // a mapping from the key to the kv-entry private final Map kvEntryMap = new HashMap<>(); @@ -99,15 +112,29 @@ public class KvPreWriteBuffer { private final Counter truncateAsDuplicatedCount; private final Counter truncateAsErrorCount; + // The TabletServer-wide ledger shared by all pre-write buffers, updated atomically on the + // write path and serving as the single source of truth for metrics and backpressure. + private final KvPreWriteBufferMemoryLedger memoryLedger; + // the max LSN in the buffer private long maxLogSequenceNumber = -1; // Accumulated byte size of entries not yet completed by a flush. private long pendingFlushBytes = 0; + // Local accounting of this buffer, released to the shared ledger on close. Must be read + // under the kv write lock (or in single-threaded tests) to be exact. + private long memoryUsageBytes = 0; + + // Number of held entries, maintained together with the local memory accounting. + private int entryCount = 0; + + private boolean closed; + public KvPreWriteBuffer(TabletServerMetricGroup serverMetricGroup) { truncateAsDuplicatedCount = serverMetricGroup.kvTruncateAsDuplicatedCount(); truncateAsErrorCount = serverMetricGroup.kvTruncateAsErrorCount(); + memoryLedger = serverMetricGroup.kvPreWriteBufferMemoryLedger(); } /** @@ -181,8 +208,8 @@ private void doPut(ChangeType changeType, Key key, Value value, long lsn) { allKvEntries.addLast(kvEntry); // update the max lsn maxLogSequenceNumber = lsn; - // track accumulated bytes for flush budget gating - pendingFlushBytes += entryBytes(key, value); + // update the accounting for flush budget gating and metrics + addToAccounting(kvEntry); } /** @@ -226,8 +253,7 @@ public void truncateTo(long targetLogSequenceNumber, TruncateReason truncateReas + ", targetLogSequenceNumber=" + targetLogSequenceNumber); } - pendingFlushBytes -= entryBytes(entry.getKey(), entry.getValue()); - boolean removed = kvEntryMap.remove(entry.getKey(), entry); + boolean removed = removeFromMapAndAccounting(entry); // the removed entry is no longer the successor of its previous version; clear the // forward link so the truncated entry does not stay reachable through it if (entry.previousEntry != null) { @@ -251,6 +277,21 @@ public long pendingFlushBytes() { return pendingFlushBytes; } + /** Returns the number of entries currently held in this buffer. */ + public int entryCount() { + return entryCount; + } + + /** + * Returns an estimation of the total memory footprint of the entries currently held in this + * buffer, including the key/value payload bytes tracked by {@link #pendingFlushBytes()} and the + * per-entry JVM object overhead. This is an approximation for observability purposes, not an + * exact measurement. + */ + public long memoryUsageBytes() { + return memoryUsageBytes; + } + /** * Prepares a prefix of entries for asynchronous flush without removing them from the buffer. * @@ -286,8 +327,7 @@ public int completeFlush(PreparedFlush preparedFlush) { throw new IllegalStateException("Prepared flush entry is not in PREPARED state."); } entry.state = EntryState.FLUSHED; - pendingFlushBytes -= entryBytes(entry.getKey(), entry.getValue()); - kvEntryMap.remove(entry.getKey(), entry); + removeFromMapAndAccounting(entry); // the immediate successor is the only live referencer of a flushed entry; clearing // its reference makes the flushed entry (and, transitively, its older versions) // unreachable instead of being retained while no longer counted by pendingFlushBytes @@ -322,6 +362,59 @@ public void abortAllPrepared() { } } + /** + * Adds an entry to the incrementally maintained memory estimate and entry count. An entry + * without a previous version is the latest version of a new key and thus adds one map node. + */ + private void addToAccounting(KvEntry entry) { + long bytes = entryBytes(entry.getKey(), entry.getValue()); + pendingFlushBytes += bytes; + long accountedBytes = + bytes + + PER_ENTRY_OVERHEAD_BYTES + + (entry.previousEntry == null ? PER_MAP_NODE_OVERHEAD_BYTES : 0L); + memoryUsageBytes += accountedBytes; + entryCount++; + memoryLedger.add(accountedBytes, 1); + } + + /** + * Removes an entry from the key map and deducts it from the incrementally maintained memory + * estimate and entry count. Returns whether the entry was the latest version of its key and + * thus removed from the map. + */ + private boolean removeFromMapAndAccounting(KvEntry entry) { + long bytes = entryBytes(entry.getKey(), entry.getValue()); + pendingFlushBytes -= bytes; + boolean removedFromMap = kvEntryMap.remove(entry.getKey(), entry); + long accountedBytes = + bytes + + PER_ENTRY_OVERHEAD_BYTES + + (removedFromMap ? PER_MAP_NODE_OVERHEAD_BYTES : 0L); + memoryUsageBytes -= accountedBytes; + entryCount--; + memoryLedger.subtract(accountedBytes, 1); + return removedFromMap; + } + + /** + * Closes the buffer and releases its remaining accounting to the shared ledger. Must be called + * under the kv write lock so the local accounting values are exact. Idempotent. + */ + @Override + public void close() { + if (closed) { + return; + } + closed = true; + memoryLedger.subtract(memoryUsageBytes, entryCount); + memoryUsageBytes = 0; + entryCount = 0; + allKvEntries.clear(); + kvEntryMap.clear(); + maxLogSequenceNumber = -1; + } + private static long entryBytes(Key key, Value value) { return (long) key.key.length + (value.value != null ? value.value.length : 0L); } diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/prewrite/KvPreWriteBufferMemoryLedger.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/prewrite/KvPreWriteBufferMemoryLedger.java new file mode 100644 index 0000000000..48536969ab --- /dev/null +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/prewrite/KvPreWriteBufferMemoryLedger.java @@ -0,0 +1,83 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.fluss.server.kv.prewrite; + +import org.apache.fluss.annotation.Internal; + +import javax.annotation.concurrent.ThreadSafe; + +import java.util.concurrent.atomic.AtomicLong; + +import static org.apache.fluss.utils.Preconditions.checkArgument; + +/** + * TabletServer-wide memory ledger shared by all KV pre-write buffers. It atomically tracks the + * total estimated memory usage and the total number of entries held across all buffers, and serves + * as the single source of truth for both the exposed metrics and future backpressure decisions. + * + *

Each buffer reports its accounting deltas to this ledger on the write path through {@link + * #add} and {@link #subtract}, so a metric read is a single atomic snapshot instead of a sum of + * non-atomic samples collected from multiple buffers. + * + *

The memory usage covers the key/value payload bytes plus a per-entry object overhead + * approximation, reflecting the real retained heap of the buffered entries. + */ +@Internal +@ThreadSafe +public final class KvPreWriteBufferMemoryLedger { + + private final AtomicLong memoryUsageBytes = new AtomicLong(); + + private final AtomicLong entryCount = new AtomicLong(); + + /** + * Adds the given amount of memory usage and entries to the ledger. Called when entries are + * appended to a pre-write buffer. + */ + public void add(long memoryBytes, int entries) { + checkArgument(memoryBytes >= 0, "The added memory bytes must not be negative."); + checkArgument(entries >= 0, "The added entry count must not be negative."); + memoryUsageBytes.addAndGet(memoryBytes); + entryCount.addAndGet(entries); + } + + /** + * Subtracts the given amount of memory usage and entries from the ledger. Called when entries + * leave a pre-write buffer by flushing or truncation, or when a buffer is closed. + */ + public void subtract(long memoryBytes, int entries) { + checkArgument(memoryBytes >= 0, "The subtracted memory bytes must not be negative."); + checkArgument(entries >= 0, "The subtracted entry count must not be negative."); + memoryUsageBytes.addAndGet(-memoryBytes); + entryCount.addAndGet(-entries); + } + + /** + * Returns the total estimated memory usage across all pre-write buffers in bytes, including the + * key/value payload bytes and the per-entry object overhead. This is an approximation for + * observability purposes, not an exact measurement. + */ + public long memoryUsageBytes() { + return memoryUsageBytes.get(); + } + + /** Returns the total number of entries held across all pre-write buffers. */ + public long entryCount() { + return entryCount.get(); + } +} 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 dac5562c10..c62e864c3b 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 @@ -30,6 +30,7 @@ import org.apache.fluss.metrics.ThreadSafeSimpleCounter; import org.apache.fluss.metrics.groups.AbstractMetricGroup; import org.apache.fluss.metrics.registry.MetricRegistry; +import org.apache.fluss.server.kv.prewrite.KvPreWriteBufferMemoryLedger; import org.apache.fluss.server.kv.rocksdb.RocksDBStatistics; import java.util.Map; @@ -90,6 +91,10 @@ public class TabletServerMetricGroup extends AbstractMetricGroup { private volatile long sharedWriteBufferCapacity; + /** Ledger shared by all KV pre-write buffers, serving as the single accounting source. */ + private final KvPreWriteBufferMemoryLedger kvPreWriteBufferMemoryLedger = + new KvPreWriteBufferMemoryLedger(); + public TabletServerMetricGroup( MetricRegistry registry, String clusterId, String rack, String hostname, int serverId) { super(registry, new String[] {clusterId, hostname, NAME}, null); @@ -151,6 +156,9 @@ public TabletServerMetricGroup( // Register server-level RocksDB aggregated metrics registerServerRocksDBMetrics(); + + // Register server-level pre-write buffer aggregated metrics + registerServerKvPreWriteBufferMetrics(); } /** @@ -215,6 +223,24 @@ public void setSharedWriteBufferMetrics(LongSupplier usageSupplier, long capacit this.sharedWriteBufferCapacity = capacity; } + /** + * Register server-level pre-write buffer aggregated metrics. These metrics aggregate the memory + * usage of the pre-write buffers of all KV tablets in this server. + */ + private void registerServerKvPreWriteBufferMetrics() { + gauge( + MetricNames.KV_PRE_WRITE_BUFFER_MEMORY_USAGE_BYTES, + kvPreWriteBufferMemoryLedger::memoryUsageBytes); + gauge( + MetricNames.KV_PRE_WRITE_BUFFER_ENTRY_COUNT, + kvPreWriteBufferMemoryLedger::entryCount); + } + + /** Returns the memory ledger shared by all KV pre-write buffers of this server. */ + public KvPreWriteBufferMemoryLedger kvPreWriteBufferMemoryLedger() { + return kvPreWriteBufferMemoryLedger; + } + /** * Registers gauges for the server-wide WAL memory pool used by primary key tables. Called once * by KvManager when creating the server buffer pool. diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/KvManagerTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/KvManagerTest.java index e9c69e4fef..4333478dc7 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/kv/KvManagerTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/KvManagerTest.java @@ -282,6 +282,37 @@ void testSharedWriteBufferConfiguredThroughKvManagerCreateAndLoad() throws Excep .isEqualTo(capacity.getBytes()); } + @Test + void testPreWriteBufferServerLevelMetrics() throws Exception { + initTableBuckets(null); + TabletServerMetricGroup metricGroup = TestingMetricGroups.TABLET_SERVER_METRICS; + long memoryUsageBefore = + gaugeValue(metricGroup, MetricNames.KV_PRE_WRITE_BUFFER_MEMORY_USAGE_BYTES); + long entryCountBefore = + gaugeValue(metricGroup, MetricNames.KV_PRE_WRITE_BUFFER_ENTRY_COUNT); + + // write one kv record without flushing so that it stays in the pre-write buffer + KvTablet kvTablet = getOrCreateKv(tablePath1, null, tableBucket1); + KvRecordBatch kvRecordBatch = + kvRecordBatchFactory.ofRecords( + Collections.singletonList( + kvRecordFactory.ofRecord( + "key1".getBytes(), new Object[] {1, "a"}))); + kvTablet.putAsLeader(kvRecordBatch, null); + + assertThat(gaugeValue(metricGroup, MetricNames.KV_PRE_WRITE_BUFFER_ENTRY_COUNT)) + .isEqualTo(entryCountBefore + 1); + assertThat(gaugeValue(metricGroup, MetricNames.KV_PRE_WRITE_BUFFER_MEMORY_USAGE_BYTES)) + .isGreaterThan(memoryUsageBefore); + + // the accounting returns to its previous values once the buffered entry is flushed + flushAndWait(kvTablet, Long.MAX_VALUE); + assertThat(gaugeValue(metricGroup, MetricNames.KV_PRE_WRITE_BUFFER_ENTRY_COUNT)) + .isEqualTo(entryCountBefore); + assertThat(gaugeValue(metricGroup, MetricNames.KV_PRE_WRITE_BUFFER_MEMORY_USAGE_BYTES)) + .isEqualTo(memoryUsageBefore); + } + @ParameterizedTest @MethodSource("partitionProvider") void testCreateKv(String partitionName) throws Exception { diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/prewrite/KvPreWriteBufferTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/prewrite/KvPreWriteBufferTest.java index 45d983133a..a277f02e11 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/kv/prewrite/KvPreWriteBufferTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/prewrite/KvPreWriteBufferTest.java @@ -17,8 +17,10 @@ package org.apache.fluss.server.kv.prewrite; +import org.apache.fluss.metrics.registry.NOPMetricRegistry; import org.apache.fluss.server.kv.prewrite.KvPreWriteBuffer.PreparedFlush; import org.apache.fluss.server.kv.prewrite.KvPreWriteBuffer.TruncateReason; +import org.apache.fluss.server.metrics.group.TabletServerMetricGroup; import org.apache.fluss.server.metrics.group.TestingMetricGroups; import org.junit.jupiter.api.Test; @@ -468,6 +470,71 @@ void testPendingFlushBytesTracking() { assertThat(buffer.pendingFlushBytes()).isEqualTo(0); } + @Test + void testEstimatedMemoryUsage() { + TabletServerMetricGroup metricGroup = + new TabletServerMetricGroup(NOPMetricRegistry.INSTANCE, "fluss", "rack", "host", 0); + KvPreWriteBuffer buffer = new KvPreWriteBuffer(metricGroup); + + assertThat(buffer.memoryUsageBytes()).isEqualTo(0L); + assertThat(buffer.entryCount()).isEqualTo(0); + assertThat(metricGroup.kvPreWriteBufferMemoryLedger().memoryUsageBytes()).isEqualTo(0L); + assertThat(metricGroup.kvPreWriteBufferMemoryLedger().entryCount()).isEqualTo(0L); + + // +key1(10 bytes), +key2(11 bytes), -key3(4 bytes): 25 payload bytes in total + bufferInsert(buffer, "key1", "value1", 1); + bufferInsert(buffer, "key2", "value22", 2); + bufferDelete(buffer, "key3", 3); + long payloadBytes = 25; + + // the estimation covers the payload bytes plus the per-entry object overhead + assertThat(buffer.memoryUsageBytes()).isGreaterThan(payloadBytes); + assertThat(buffer.entryCount()).isEqualTo(3); + // the shared ledger mirrors the local accounting of the buffer + assertThat(metricGroup.kvPreWriteBufferMemoryLedger().memoryUsageBytes()) + .isEqualTo(buffer.memoryUsageBytes()); + assertThat(metricGroup.kvPreWriteBufferMemoryLedger().entryCount()).isEqualTo(3L); + + // flushing all entries releases the whole accounted usage + flushBuffer(buffer, Long.MAX_VALUE); + assertThat(buffer.memoryUsageBytes()).isEqualTo(0L); + assertThat(buffer.entryCount()).isEqualTo(0); + assertThat(metricGroup.kvPreWriteBufferMemoryLedger().memoryUsageBytes()).isEqualTo(0L); + assertThat(metricGroup.kvPreWriteBufferMemoryLedger().entryCount()).isEqualTo(0L); + + // truncating entries also releases their accounted usage + bufferInsert(buffer, "key1", "value1", 4); + assertThat(buffer.memoryUsageBytes()).isPositive(); + buffer.truncateTo(4, TruncateReason.ERROR); + assertThat(buffer.memoryUsageBytes()).isEqualTo(0L); + assertThat(buffer.entryCount()).isEqualTo(0); + assertThat(metricGroup.kvPreWriteBufferMemoryLedger().memoryUsageBytes()).isEqualTo(0L); + assertThat(metricGroup.kvPreWriteBufferMemoryLedger().entryCount()).isEqualTo(0L); + } + + @Test + void testCloseReleasesAccounting() { + TabletServerMetricGroup metricGroup = + new TabletServerMetricGroup(NOPMetricRegistry.INSTANCE, "fluss", "rack", "host", 0); + KvPreWriteBuffer buffer = new KvPreWriteBuffer(metricGroup); + bufferInsert(buffer, "key1", "value1", 0); + bufferInsert(buffer, "key2", "value2", 1); + assertThat(metricGroup.kvPreWriteBufferMemoryLedger().memoryUsageBytes()).isPositive(); + assertThat(metricGroup.kvPreWriteBufferMemoryLedger().entryCount()).isEqualTo(2L); + + // closing releases the remaining accounting to the shared ledger exactly once + buffer.close(); + assertThat(buffer.memoryUsageBytes()).isEqualTo(0L); + assertThat(buffer.entryCount()).isEqualTo(0); + assertThat(metricGroup.kvPreWriteBufferMemoryLedger().memoryUsageBytes()).isEqualTo(0L); + assertThat(metricGroup.kvPreWriteBufferMemoryLedger().entryCount()).isEqualTo(0L); + + // closing again is idempotent and must not over-release the ledger + buffer.close(); + assertThat(metricGroup.kvPreWriteBufferMemoryLedger().memoryUsageBytes()).isEqualTo(0L); + assertThat(metricGroup.kvPreWriteBufferMemoryLedger().entryCount()).isEqualTo(0L); + } + @Test void testCompleteFlushDetachesFlushedEntriesFromPreviousChain() { KvPreWriteBuffer buffer = new KvPreWriteBuffer(TestingMetricGroups.TABLET_SERVER_METRICS); diff --git a/website/docs/maintenance/observability/monitor-metrics.md b/website/docs/maintenance/observability/monitor-metrics.md index 0d31c09dd3..20f758ab20 100644 --- a/website/docs/maintenance/observability/monitor-metrics.md +++ b/website/docs/maintenance/observability/monitor-metrics.md @@ -588,6 +588,16 @@ Some metrics might not be exposed when using other JVM implementations (e.g. IBM preWriteBufferTruncateAsErrorPerSecond The number of kv pre-write buffer truncate due to the error happened when writing cdc to log per second. Meter + + + kvPreWriteBufferMemoryUsageBytes + Estimated total memory usage of the KV pre-write buffers across all KV tablets in this server (in bytes), including the key/value payload bytes and the per-entry object overhead. It is an approximation for observability, not an exact measurement. + Gauge + + + kvPreWriteBufferEntryCount + The number of entries buffered in the KV pre-write buffers across all KV tablets in this server. + Gauge kvWalMemoryPoolUsage