You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@fluss.apache.org by "wuchong (via GitHub)" <gi...@apache.org> on 2025/08/05 15:43:07 UTC
Re: [PR] [kv] support export key metrics for rocksdb [fluss]
wuchong commented on code in PR #1355:
URL: https://github.com/apache/fluss/pull/1355#discussion_r2254638304
##########
fluss-server/src/main/java/com/alibaba/fluss/server/kv/KvTablet.java:
##########
@@ -259,6 +263,36 @@ public void registerMetrics(BucketMetricGroup bucketMetricGroup) {
metricGroup.meter(
MetricNames.KV_PRE_WRITE_BUFFER_TRUNCATE_AS_ERROR_RATE,
new MeterView(kvPreWriteBuffer.getTruncateAsErrorCount()));
+
+ // Initialize RocksDB metrics reporter
+ inWriteLock(
+ kvLock,
Review Comment:
I think we don't need to initialize `rocksDBMetricsCollector` in lock, because there is no concurrency problem here. Besides, I think it's better to move the `rocksDBMetricsCollector` initialization to the constructor to simplify the logic and also make the `rocksDBMetricsCollector` to be final.
##########
fluss-server/src/main/java/com/alibaba/fluss/server/kv/rocksdb/RocksDBMetricsCollector.java:
##########
@@ -0,0 +1,527 @@
+/*
+ * 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 com.alibaba.fluss.server.kv.rocksdb;
+
+import com.alibaba.fluss.metadata.TableBucket;
+import com.alibaba.fluss.metadata.TablePath;
+import com.alibaba.fluss.metrics.Gauge;
+import com.alibaba.fluss.metrics.MetricNames;
+import com.alibaba.fluss.metrics.groups.MetricGroup;
+import com.alibaba.fluss.server.metrics.group.BucketMetricGroup;
+
+import org.rocksdb.Cache;
+import org.rocksdb.HistogramData;
+import org.rocksdb.HistogramType;
+import org.rocksdb.MemoryUsageType;
+import org.rocksdb.MemoryUtil;
+import org.rocksdb.RocksDB;
+import org.rocksdb.RocksDBException;
+import org.rocksdb.Statistics;
+import org.rocksdb.TickerType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.io.Closeable;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * RocksDB metrics collector that collects and reports RocksDB statistics including cache,
+ * compaction, memory, and I/O metrics with table bucket and server information tags.
+ */
+public class RocksDBMetricsCollector implements Closeable {
+
+ private static final Logger LOG = LoggerFactory.getLogger(RocksDBMetricsCollector.class);
+
+ private final RocksDB rocksDB;
+ private final Statistics statistics;
+ private final MetricGroup bucketMetricGroup;
+ private final TableBucket tableBucket;
+ private final TablePath tablePath;
+ private final String partitionName;
+ private final RocksDBResourceContainer resourceContainer;
+
+ private volatile boolean registered = false;
+
+ // Cached values for efficient metric access
+ private volatile long blockCacheHitCount = 0;
+ private volatile long blockCacheMissCount = 0;
+ private volatile long indexBlockCacheHitCount = 0;
+ private volatile long indexBlockCacheMissCount = 0;
+ private volatile long filterBlockCacheHitCount = 0;
+ private volatile long filterBlockCacheMissCount = 0;
+ private volatile long dataBlockCacheHitCount = 0;
+ private volatile long dataBlockCacheMissCount = 0;
+ private volatile long compactionCount = 0;
+ private volatile long compactionBytesRead = 0;
+ private volatile long compactionBytesWritten = 0;
+ private volatile long compactionCpuTimeMicros = 0;
+ private volatile long flushCount = 0;
+ private volatile long flushBytesWritten = 0;
+ private volatile long bytesRead = 0;
+ private volatile long bytesWritten = 0;
+ private volatile long numberDbNext = 0;
+ private volatile long numberDbPrev = 0;
+ private volatile long numberDbSeek = 0;
+ private volatile long numberDbSeekFound = 0;
+ private volatile long numberKeysRead = 0;
+ private volatile long numberKeysWritten = 0;
+ private volatile long numberKeysUpdated = 0;
+ private volatile long numLiveVersions = 0;
+ private volatile long numImmutableMemtables = 0;
+ private volatile long activeMemtableSize = 0;
+ private volatile long unflushedMemtableSize = 0;
+ private volatile long totalSstFilesSize = 0;
+ private volatile long liveSstFilesSize = 0;
+ private volatile long sizeAllMemtables = 0;
+ private volatile long blockCacheAddCount = 0;
+ private volatile long memtableMemoryUsage = 0;
+ private volatile long blockCacheMemoryUsage = 0;
+ private volatile long tableReadersMemoryUsage = 0;
+ private volatile long totalMemoryUsage = 0;
+ private volatile long blockCacheUsage = 0;
+ private volatile long blockCachePinnedUsage = 0;
+ private volatile long stallTimeMicros = 0;
+
+ // Level SST files metrics
+ private volatile long numFilesAtLevel0 = 0;
+ private volatile long numFilesAtLevel1 = 0;
+ private volatile long numFilesAtLevel2 = 0;
+ private volatile long numFilesAtLevel3 = 0;
+ private volatile long numFilesAtLevel4 = 0;
+ private volatile long numFilesAtLevel5 = 0;
+ private volatile long numFilesAtLevel6 = 0;
+
+ // Compression and compaction pressure metrics
+ private volatile long compactionPending = 0;
+ private volatile long flushPending = 0;
+ private volatile long numRunningCompactions = 0;
+ private volatile long numRunningFlushes = 0;
+
+ // Write amplification metrics
+ private volatile long writeAmplification = 0;
+ private volatile long readAmplification = 0;
+
+ private volatile long backgroundErrors = 0;
+
+ // Histogram metrics
+ private volatile long dbGetLatencyMicros = 0;
+ private volatile long dbWriteLatencyMicros = 0;
+ private volatile long compactionTimeMicros = 0;
+ private volatile long compressionTimeNanos = 0;
+ private volatile long decompressionTimeNanos = 0;
+
+ public RocksDBMetricsCollector(
+ RocksDB rocksDB,
+ Statistics statistics,
+ BucketMetricGroup bucketMetricGroup,
+ TableBucket tableBucket,
+ TablePath tablePath,
+ @Nullable String partitionName,
+ RocksDBResourceContainer resourceContainer) {
+ this.rocksDB = rocksDB;
+ this.statistics = statistics;
+ this.tableBucket = tableBucket;
+ this.tablePath = tablePath;
+ this.partitionName = partitionName;
+ this.resourceContainer = resourceContainer;
+
+ // Create RocksDB metrics group under the bucket metrics
+ this.bucketMetricGroup = bucketMetricGroup;
+
+ // Register all metrics
+ registerMetrics();
+
+ // Register this reporter with the global collector
+ RocksDBMetricsManager.getInstance().registerCollector(this);
+ this.registered = true;
+
+ LOG.info(
+ "RocksDB metrics collector started for table {} bucket {}",
+ tablePath,
+ tableBucket.getBucket());
+ }
+
+ private void registerMetrics() {
+ // Cache metrics - 统一使用缓存变量
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_BLOCK_CACHE_HIT_COUNT, (Gauge<Long>) () -> blockCacheHitCount);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_BLOCK_CACHE_MISS_COUNT,
+ (Gauge<Long>) () -> blockCacheMissCount);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_INDEX_BLOCK_CACHE_HIT_COUNT,
+ (Gauge<Long>) () -> indexBlockCacheHitCount);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_INDEX_BLOCK_CACHE_MISS_COUNT,
+ (Gauge<Long>) () -> indexBlockCacheMissCount);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_FILTER_BLOCK_CACHE_HIT_COUNT,
+ (Gauge<Long>) () -> filterBlockCacheHitCount);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_FILTER_BLOCK_CACHE_MISS_COUNT,
+ (Gauge<Long>) () -> filterBlockCacheMissCount);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_DATA_BLOCK_CACHE_HIT_COUNT,
+ (Gauge<Long>) () -> dataBlockCacheHitCount);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_DATA_BLOCK_CACHE_MISS_COUNT,
+ (Gauge<Long>) () -> dataBlockCacheMissCount);
+
+ // Compaction metrics
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_COMPACTION_COUNT, (Gauge<Long>) () -> compactionCount);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_COMPACTION_BYTES_READ, (Gauge<Long>) () -> compactionBytesRead);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_COMPACTION_BYTES_WRITTEN,
+ (Gauge<Long>) () -> compactionBytesWritten);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_COMPACTION_CPU_TIME_MICROS,
+ (Gauge<Long>) () -> compactionCpuTimeMicros);
+ bucketMetricGroup.gauge(MetricNames.ROCKSDB_FLUSH_COUNT, (Gauge<Long>) () -> flushCount);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_FLUSH_BYTES_WRITTEN, (Gauge<Long>) () -> flushBytesWritten);
+
+ // Memory metrics - 统一使用缓存变量
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_MEMTABLE_MEMORY_USAGE, (Gauge<Long>) () -> memtableMemoryUsage);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_BLOCK_CACHE_MEMORY_USAGE,
+ (Gauge<Long>) () -> blockCacheMemoryUsage);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_TABLE_READERS_MEMORY_USAGE,
+ (Gauge<Long>) () -> tableReadersMemoryUsage);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_TOTAL_MEMORY_USAGE, (Gauge<Long>) () -> totalMemoryUsage);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_ACTIVE_MEMTABLE_SIZE, (Gauge<Long>) () -> activeMemtableSize);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_UNFLUSHED_MEMTABLE_SIZE,
+ (Gauge<Long>) () -> unflushedMemtableSize);
+
+ // BlockCache usage指标 - 统一使用缓存变量
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_BLOCK_CACHE_USAGE, (Gauge<Long>) () -> blockCacheUsage);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_BLOCK_CACHE_PINNED_USAGE,
+ (Gauge<Long>) () -> blockCachePinnedUsage);
+
+ // I/O metrics
+ bucketMetricGroup.gauge(MetricNames.ROCKSDB_BYTES_READ, (Gauge<Long>) () -> bytesRead);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_BYTES_WRITTEN, (Gauge<Long>) () -> bytesWritten);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_NUMBER_DB_NEXT, (Gauge<Long>) () -> numberDbNext);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_NUMBER_DB_PREV, (Gauge<Long>) () -> numberDbPrev);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_NUMBER_DB_SEEK, (Gauge<Long>) () -> numberDbSeek);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_NUMBER_DB_SEEK_FOUND, (Gauge<Long>) () -> numberDbSeekFound);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_NUMBER_KEYS_READ, (Gauge<Long>) () -> numberKeysRead);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_NUMBER_KEYS_WRITTEN, (Gauge<Long>) () -> numberKeysWritten);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_NUMBER_KEYS_UPDATED, (Gauge<Long>) () -> numberKeysUpdated);
+
+ // SST files metrics
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_NUM_LIVE_VERSIONS, (Gauge<Long>) () -> numLiveVersions);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_NUM_IMMUTABLE_MEMTABLES,
+ (Gauge<Long>) () -> numImmutableMemtables);
+
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_TOTAL_SST_FILES_SIZE, (Gauge<Long>) () -> totalSstFilesSize);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_LIVE_SST_FILES_SIZE, (Gauge<Long>) () -> liveSstFilesSize);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_SIZE_ALL_MEMTABLES, (Gauge<Long>) () -> sizeAllMemtables);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_BLOCK_CACHE_ADD_COUNT, (Gauge<Long>) () -> blockCacheAddCount);
+
+ // Level SST files metrics
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_NUM_FILES_AT_LEVEL_0, (Gauge<Long>) () -> numFilesAtLevel0);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_NUM_FILES_AT_LEVEL_1, (Gauge<Long>) () -> numFilesAtLevel1);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_NUM_FILES_AT_LEVEL_2, (Gauge<Long>) () -> numFilesAtLevel2);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_NUM_FILES_AT_LEVEL_3, (Gauge<Long>) () -> numFilesAtLevel3);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_NUM_FILES_AT_LEVEL_4, (Gauge<Long>) () -> numFilesAtLevel4);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_NUM_FILES_AT_LEVEL_5, (Gauge<Long>) () -> numFilesAtLevel5);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_NUM_FILES_AT_LEVEL_6, (Gauge<Long>) () -> numFilesAtLevel6);
+
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_WRITE_STALL_MICROS, (Gauge<Long>) () -> stallTimeMicros);
+
+ // Compression and compaction pressure metrics
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_COMPACTION_PENDING, (Gauge<Long>) () -> compactionPending);
+
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_FLUSH_PENDING, (Gauge<Long>) () -> flushPending);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_NUM_RUNNING_COMPACTIONS,
+ (Gauge<Long>) () -> numRunningCompactions);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_NUM_RUNNING_FLUSHES, (Gauge<Long>) () -> numRunningFlushes);
+
+ // Background job metrics
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_BACKGROUND_ERRORS, (Gauge<Long>) () -> backgroundErrors);
+
+ // Histogram metrics
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_DB_GET_LATENCY_MICROS, (Gauge<Long>) () -> dbGetLatencyMicros);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_DB_WRITE_LATENCY_MICROS,
+ (Gauge<Long>) () -> dbWriteLatencyMicros);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_COMPACTION_TIME_MICROS,
+ (Gauge<Long>) () -> compactionTimeMicros);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_COMPRESSION_TIME_NANOS,
+ (Gauge<Long>) () -> compressionTimeNanos);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_DECOMPRESSION_TIME_NANOS,
+ (Gauge<Long>) () -> decompressionTimeNanos);
+ }
+
+ public void updateMetrics() {
+ try {
+ // Update cache metrics
+ blockCacheHitCount = getTickerValue(TickerType.BLOCK_CACHE_HIT);
+ blockCacheMissCount = getTickerValue(TickerType.BLOCK_CACHE_MISS);
+ indexBlockCacheHitCount = getTickerValue(TickerType.BLOCK_CACHE_INDEX_HIT);
+ indexBlockCacheMissCount = getTickerValue(TickerType.BLOCK_CACHE_INDEX_MISS);
+ filterBlockCacheHitCount = getTickerValue(TickerType.BLOCK_CACHE_FILTER_HIT);
+ filterBlockCacheMissCount = getTickerValue(TickerType.BLOCK_CACHE_FILTER_MISS);
+ dataBlockCacheHitCount = getTickerValue(TickerType.BLOCK_CACHE_DATA_HIT);
+ dataBlockCacheMissCount = getTickerValue(TickerType.BLOCK_CACHE_DATA_MISS);
+
+ // Update compaction metrics
+ compactionCount =
+ getTickerValue(TickerType.COMPACT_READ_BYTES) > 0
+ ? 1
+ : 0; // Simplified since exact count may not be available
Review Comment:
It's confusing to report a count that only has values 0 and 1. Is it necessary to report this metric? Can `compactionBytesRead` be satisfied?
##########
fluss-common/src/main/java/com/alibaba/fluss/config/ConfigOptions.java:
##########
@@ -1410,6 +1410,12 @@ public class ConfigOptions {
"The max size of the consumed memory for RocksDB batch write, "
+ "will flush just based on item count if this config set to 0.");
+ public static final ConfigOption<Boolean> KV_STATISTICS_ENABLED =
+ key("kv.rocksdb.statistics.enabled")
Review Comment:
Rename to `kv.rocksdb.metrics.enabled`? I think it also includes some metrics beyond statistics, like cache metrics.
##########
fluss-server/src/test/java/com/alibaba/fluss/server/kv/rocksdb/RocksDBMetricsManagerTest.java:
##########
@@ -0,0 +1,423 @@
+/*
+ * 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 com.alibaba.fluss.server.kv.rocksdb;
+
+import com.alibaba.fluss.config.Configuration;
+import com.alibaba.fluss.metadata.TableBucket;
+import com.alibaba.fluss.metadata.TablePath;
+import com.alibaba.fluss.server.metrics.group.TestingMetricGroups;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.jupiter.api.io.TempDir;
+import org.mockito.MockitoAnnotations;
+import org.rocksdb.RocksDB;
+import org.rocksdb.Statistics;
+
+import java.lang.reflect.Field;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+/** Tests for {@link RocksDBMetricsManager}. */
+class RocksDBMetricsManagerTest {
+
+ @TempDir Path tempDir;
+ private RocksDBResourceContainer resourceContainer;
+ private List<RocksDBMetricsCollector> collectors;
+ private AutoCloseable mocksCloseable;
+ @RegisterExtension public RocksDBExtension rocksDBExtension = new RocksDBExtension();
+
+ @BeforeEach
+ void setUp() {
+ RocksDB.loadLibrary();
+ mocksCloseable = MockitoAnnotations.openMocks(this);
+ // Setup resource container
+ resourceContainer = new RocksDBResourceContainer(new Configuration(), tempDir.toFile());
+
+ collectors = new ArrayList<>();
+
+ // Reset singleton instance for testing
+ resetSingleton();
+ }
+
+ @AfterEach
+ void tearDown() throws Exception {
+ // Close all collectors
+ for (RocksDBMetricsCollector collector : collectors) {
+ collector.close();
+ }
+ collectors.clear();
+
+ // Shutdown manager
+ RocksDBMetricsManager.getInstance().shutdown();
+
+ // Clean up resources
+ if (resourceContainer != null) {
+ resourceContainer.close();
+ }
+ if (mocksCloseable != null) {
+ mocksCloseable.close();
+ }
+
+ // Reset singleton for next test
+ resetSingleton();
+ }
+
+ @Test
+ void testRegisterCollector() throws Exception {
+
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ assertThat(manager.getCollectorCount()).isEqualTo(0);
+
+ RocksDBMetricsCollector collector = createMockCollector("test_db", "test_table", 0);
+ manager.registerCollector(collector);
+
+ assertThat(manager.getCollectorCount()).isEqualTo(1);
+ assertThat(manager.isRunning()).isTrue();
+
+ collectors.add(collector);
+ }
+
+ @Test
+ void testRegisterMultipleCollectors() throws Exception {
+
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ for (int i = 0; i < 5; i++) {
+ RocksDBMetricsCollector collector = createMockCollector("test_db", "test_table", i);
+ manager.registerCollector(collector);
+ collectors.add(collector);
+ }
+
+ assertThat(manager.getCollectorCount()).isEqualTo(5);
+ assertThat(manager.isRunning()).isTrue();
+ }
+
+ @Test
+ void testUnregisterCollector() throws Exception {
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ RocksDBMetricsCollector collector = createMockCollector("test_db", "test_table", 0);
+ manager.registerCollector(collector);
+ collectors.add(collector);
+
+ assertThat(manager.getCollectorCount()).isEqualTo(1);
+
+ manager.unregisterCollector(collector);
+
+ assertThat(manager.getCollectorCount()).isEqualTo(0);
+ }
+
+ @Test
+ void testRegisterAfterShutdown() throws Exception {
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+ manager.shutdown();
+
+ RocksDBMetricsCollector collector = createMockCollector("test_db", "test_table", 0);
+ manager.registerCollector(collector);
+
+ // Should not be registered after shutdown
+ assertThat(manager.getCollectorCount()).isEqualTo(0);
+
+ collectors.add(collector);
+ }
+
+ @Test
+ void testMetricsCollectionStartsWithFirstCollector() throws Exception {
+
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ assertThat(manager.isRunning()).isFalse();
+
+ RocksDBMetricsCollector collector = createMockCollector("test_db", "test_table", 0);
+ manager.registerCollector(collector);
+ collectors.add(collector);
+
+ assertThat(manager.isRunning()).isTrue();
+ }
+
+ @Test
+ void testBatchProcessing() throws Exception {
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ // Create many collectors to test batch processing
+ int collectorCount = 250; // More than batch size (100)
+ List<RocksDBMetricsCollector> spyCollectors = new ArrayList<>();
+
+ for (int i = 0; i < collectorCount; i++) {
+ RocksDBMetricsCollector collector = createMockCollector("test_db", "test_table", i);
+ RocksDBMetricsCollector spyCollector = spy(collector);
+ manager.registerCollector(spyCollector);
+ spyCollectors.add(spyCollector);
+ collectors.add(collector);
+ }
+
+ // Wait for metrics collection to occur
+ Thread.sleep(6000); // Wait for at least one collection cycle
+
+ // Verify that updateMetrics was called on all collectors
+ for (RocksDBMetricsCollector collector : spyCollectors) {
+ verify(collector, times(1)).updateMetrics();
+ }
+ }
+
+ @Test
+ void testExceptionHandlingInMetricsCollection() throws Exception {
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ // Create a collector that throws exception
+ RocksDBMetricsCollector faultyCollector = createMockCollector("test_db", "test_table", 0);
+ doThrow(new RuntimeException("Test exception")).when(faultyCollector).updateMetrics();
+
+ // Create a normal collector
+ RocksDBMetricsCollector normalCollector = createMockCollector("test_db", "test_table", 1);
+ RocksDBMetricsCollector spyNormalCollector = spy(normalCollector);
+
+ manager.registerCollector(faultyCollector);
+ manager.registerCollector(spyNormalCollector);
+
+ collectors.add(faultyCollector);
+ collectors.add(normalCollector);
+
+ // Wait for metrics collection
+ try {
+ Thread.sleep(6000);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+
+ // Normal collector should still be called despite exception in faulty collector
+ verify(spyNormalCollector, times(1)).updateMetrics();
+ }
+
+ @Test
+ void testShutdown() {
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ // Register some collectors
+ for (int i = 0; i < 3; i++) {
+ RocksDBMetricsCollector collector = createMockCollector("test_db", "test_table", i);
+ manager.registerCollector(collector);
+ collectors.add(collector);
+ }
+
+ assertThat(manager.isRunning()).isTrue();
+ assertThat(manager.getCollectorCount()).isEqualTo(3);
+
+ manager.shutdown();
+
+ assertThat(manager.isRunning()).isFalse();
+ assertThat(manager.isShutdown()).isTrue();
+ assertThat(manager.getCollectorCount()).isEqualTo(0);
+ }
+
+ @Test
+ void testDoubleShutdown() {
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ manager.shutdown();
+ assertThat(manager.isShutdown()).isTrue();
+
+ // Second shutdown should not cause issues
+ manager.shutdown();
+ assertThat(manager.isShutdown()).isTrue();
+ }
+
+ @Test
+ void testConcurrentRegistrationAndUnregistration() throws Exception {
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ int threadCount = 10;
+ CountDownLatch latch = new CountDownLatch(threadCount);
+ ExecutorService executor = Executors.newFixedThreadPool(threadCount);
+ AtomicInteger collectorId = new AtomicInteger(0);
+
+ for (int i = 0; i < threadCount; i++) {
+ executor.submit(
+ () -> {
+ try {
+ for (int j = 0; j < 10; j++) {
+ RocksDBMetricsCollector collector =
+ createMockCollector(
+ "test_db",
+ "test_table",
+ collectorId.getAndIncrement());
+ manager.registerCollector(collector);
+ collectors.add(collector);
+
+ // Immediately unregister some collectors
+ if (j % 2 == 0) {
+ manager.unregisterCollector(collector);
+ }
+ }
+ } finally {
+ latch.countDown();
+ }
+ });
+ }
+
+ latch.await(10, TimeUnit.SECONDS);
+ executor.shutdown();
+
+ // Should handle concurrent access gracefully
+ assertThat(manager.getCollectorCount()).isGreaterThanOrEqualTo(0);
+ }
+
+ @Test
+ void testEmptyCollectorsDoesNotCauseIssues() throws Exception {
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ // Start with no collectors
+ assertThat(manager.getCollectorCount()).isEqualTo(0);
+
+ // Wait a bit to ensure no exceptions are thrown
+ Thread.sleep(1000);
+
+ // Should still be not running
+ assertThat(manager.isRunning()).isFalse();
+ }
+
+ @Test
+ void testCreateReporterKey() {
+ RocksDBMetricsCollector collector1 = createMockCollector("db1", "table1", 0);
+ RocksDBMetricsCollector collector2 = createMockCollector("db1", "table1", 1);
+ RocksDBMetricsCollector collector3 = createMockCollector("db2", "table1", 0);
+
+ // Different buckets should have different keys
+ assertThat(collector1.getTableBucket().getBucket())
+ .isNotEqualTo(collector2.getTableBucket().getBucket());
+
+ // Different databases should have different keys
+ assertThat(collector1.getTablePath().getDatabaseName())
+ .isNotEqualTo(collector3.getTablePath().getDatabaseName());
+
+ collectors.add(collector1);
+ collectors.add(collector2);
+ collectors.add(collector3);
+ }
+
+ @Test
+ void testPartitionedTableCollector() {
+ RocksDBMetricsCollector collector =
+ createMockCollectorWithPartition("test_db", "test_table", 0, "partition1");
+
+ assertThat(collector.getPartitionName()).isEqualTo("partition1");
+
+ collectors.add(collector);
+ }
+
+ @Test
+ void testManagerStateAfterAllCollectorsUnregistered() {
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ // Register collectors
+ List<RocksDBMetricsCollector> testCollectors = new ArrayList<>();
+ for (int i = 0; i < 3; i++) {
+ RocksDBMetricsCollector collector = createMockCollector("test_db", "test_table", i);
+ manager.registerCollector(collector);
+ testCollectors.add(collector);
+ collectors.add(collector);
+ }
+
+ assertThat(manager.isRunning()).isTrue();
+ assertThat(manager.getCollectorCount()).isEqualTo(3);
+
+ // Unregister all collectors
+ for (RocksDBMetricsCollector collector : testCollectors) {
+ manager.unregisterCollector(collector);
+ }
+
+ assertThat(manager.getCollectorCount()).isEqualTo(0);
+ // Manager should still be running (collection thread continues)
+ assertThat(manager.isRunning()).isTrue();
+ }
+
+ @Test
+ void testResourceCleanupOnShutdown() throws Exception {
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ // Register a collector to start the manager
+ RocksDBMetricsCollector collector = createMockCollector("test_db", "test_table", 0);
+ manager.registerCollector(collector);
+ collectors.add(collector);
+
+ assertThat(manager.isRunning()).isTrue();
+
+ // Get the executor service via reflection to verify it's cleaned up
+ Field executorField =
+ RocksDBMetricsManager.class.getDeclaredField("metricsCollectionExecutor");
+ executorField.setAccessible(true);
+ ScheduledExecutorService executor = (ScheduledExecutorService) executorField.get(manager);
+
+ assertThat(executor.isShutdown()).isFalse();
+
+ manager.shutdown();
+
+ assertThat(manager.isShutdown()).isTrue();
+ assertThat(executor.isShutdown()).isTrue();
+ }
+
+ private RocksDBMetricsCollector createMockCollector(String database, String table, int bucket) {
+ return createMockCollectorWithPartition(database, table, bucket, null);
+ }
+
+ private RocksDBMetricsCollector createMockCollectorWithPartition(
+ String database, String table, int bucket, String partition) {
+ Statistics mockStatistics = mock(Statistics.class);
+
+ TablePath tablePath = TablePath.of(database, table);
+ TableBucket tableBucket = new TableBucket(0, bucket);
+
+ return spy(
+ new RocksDBMetricsCollector(
+ rocksDBExtension.getRocksDb(),
+ mockStatistics,
+ TestingMetricGroups.BUCKET_METRICS,
+ tableBucket,
+ tablePath,
+ partition,
+ resourceContainer));
+ }
+
+ private void resetSingleton() {
+ try {
+ Field instanceField = RocksDBMetricsManager.class.getDeclaredField("instance");
+ instanceField.setAccessible(true);
+ instanceField.set(null, null);
+ } catch (Exception e) {
Review Comment:
Usually, reflection is highly not recommended, unless it is absolutely necessary. Here we can declare the `instance` variable as `protected` to make it can be accessed in the `RocksDBMetricsManagerTest` but not exposed publicly.
##########
fluss-server/src/test/java/com/alibaba/fluss/server/kv/rocksdb/RocksDBResourceContainerTest.java:
##########
@@ -187,8 +193,237 @@ void testConfigurationOptionsFromConfig() throws Exception {
(BlockBasedTableConfig) columnOptions.tableFormatConfig();
assertThat(tableConfig.blockSize()).isEqualTo(4 * SizeUnit.KB);
assertThat(tableConfig.metadataBlockSize()).isEqualTo(8 * SizeUnit.KB);
- assertThat(tableConfig.blockCacheSize()).isEqualTo(512 * SizeUnit.MB);
assertThat(tableConfig.filterPolicy() instanceof BloomFilter).isTrue();
}
}
+
+ @Test
+ void testBlockCacheCreation() throws Exception {
+ Configuration config = new Configuration();
+ config.setString(ConfigOptions.KV_BLOCK_CACHE_SIZE.key(), "16mb");
+
+ try (RocksDBResourceContainer container = new RocksDBResourceContainer(config, null)) {
+ Cache blockCache = container.getBlockCache();
+ assertThat(blockCache).isNotNull();
+ }
+ }
+
+ @Test
+ void testBlockCacheNotCreatedWhenSizeIsZero() throws Exception {
+ Configuration config = new Configuration();
+ config.setString(ConfigOptions.KV_BLOCK_CACHE_SIZE.key(), "0mb");
+
+ try (RocksDBResourceContainer container = new RocksDBResourceContainer(config, null)) {
+ Cache blockCache = container.getBlockCache();
+ assertThat(blockCache).isNull();
+ }
+ }
+
+ @Test
+ void testStatisticsEnabledByDefault() throws Exception {
+ Configuration config = new Configuration();
+ // Statistics should be enabled by default
+ try (RocksDBResourceContainer container = new RocksDBResourceContainer(config, null)) {
+ DBOptions dbOptions = container.getDbOptions();
+ assertThat(dbOptions.statistics()).isNotNull();
+ assertThat(container.getStatistics()).isNotNull();
+ }
+ }
+
+ @Test
+ void testStatisticsDisabled() throws Exception {
+ Configuration config = new Configuration();
+ config.setBoolean(ConfigOptions.KV_STATISTICS_ENABLED.key(), false);
+
+ try (RocksDBResourceContainer container = new RocksDBResourceContainer(config, null)) {
+ DBOptions dbOptions = container.getDbOptions();
+ // Statistics should be null when disabled
+ assertThat(dbOptions.statistics()).isNull();
+ assertThat(container.getStatistics()).isNull();
+ }
+ }
+
+ @Test
+ void testStatisticsExplicitlyEnabled() throws Exception {
+ Configuration config = new Configuration();
+ config.setBoolean(ConfigOptions.KV_STATISTICS_ENABLED.key(), true);
+
+ try (RocksDBResourceContainer container = new RocksDBResourceContainer(config, null)) {
+ DBOptions dbOptions = container.getDbOptions();
+ assertThat(dbOptions.statistics()).isNotNull();
+ assertThat(container.getStatistics()).isNotNull();
+ }
+ }
+
+ @Test
+ void testStatisticsClosedWithContainer() throws Exception {
+ Configuration config = new Configuration();
+ config.setBoolean(ConfigOptions.KV_STATISTICS_ENABLED.key(), true);
+
+ Statistics statistics;
+ try (RocksDBResourceContainer container = new RocksDBResourceContainer(config, null)) {
+ statistics = container.getStatistics();
+ assertThat(statistics).isNotNull();
+ assertThat(statistics.isOwningHandle()).isTrue();
+ }
+
+ // Statistics should be closed when container is closed
+ assertThat(statistics.isOwningHandle()).isFalse();
+ }
+
+ @Test
+ void testMultipleStatisticsObjectsClosedProperly() throws Exception {
+ Configuration config = new Configuration();
+ config.setBoolean(ConfigOptions.KV_STATISTICS_ENABLED.key(), true);
+
+ List<Statistics> statisticsList = new ArrayList<>();
+
+ // Create multiple containers and collect statistics
+ for (int i = 0; i < 5; i++) {
+ RocksDBResourceContainer container = new RocksDBResourceContainer(config, null);
+ Statistics stats = container.getStatistics();
+ if (stats != null) {
+ statisticsList.add(stats);
+ }
+ container.close();
+ }
+
+ // All statistics should be closed
+ for (Statistics stats : statisticsList) {
+ assertThat(stats.isOwningHandle()).isFalse();
+ }
+ }
+
+ @Test
+ void testStatisticsWithNullInstancePath() throws Exception {
+ Configuration config = new Configuration();
+ config.setBoolean(ConfigOptions.KV_STATISTICS_ENABLED.key(), true);
+
+ try (RocksDBResourceContainer container = new RocksDBResourceContainer(config, null)) {
+ Statistics statistics = container.getStatistics();
+ assertThat(statistics).isNotNull();
+ assertThat(statistics.isOwningHandle()).isTrue();
+ }
+ }
+
+ @Test
+ void testStatisticsWithValidInstancePath(@TempDir Path tempFolder) throws Exception {
+ Configuration config = new Configuration();
+ config.setBoolean(ConfigOptions.KV_STATISTICS_ENABLED.key(), true);
+
+ File instancePath = tempFolder.resolve("rocksdb-instance").toFile();
+ try (RocksDBResourceContainer container =
+ new RocksDBResourceContainer(config, instancePath)) {
+ Statistics statistics = container.getStatistics();
+ assertThat(statistics).isNotNull();
+ assertThat(statistics.isOwningHandle()).isTrue();
+ }
+ }
+
+ @Test
+ void testStatisticsConfigurationConsistency() throws Exception {
+ Configuration config = new Configuration();
+
+ // Test with statistics disabled
+ config.setBoolean(ConfigOptions.KV_STATISTICS_ENABLED.key(), false);
+ try (RocksDBResourceContainer container = new RocksDBResourceContainer(config, null)) {
+ assertThat(container.getStatistics()).isNull();
+ }
+
+ // Test with statistics enabled
+ config.setBoolean(ConfigOptions.KV_STATISTICS_ENABLED.key(), true);
+ try (RocksDBResourceContainer container = new RocksDBResourceContainer(config, null)) {
+ assertThat(container.getStatistics()).isNotNull();
+ }
+ }
+
+ @Test
+ void testStatisticsMemoryManagement() throws Exception {
+ Configuration config = new Configuration();
+ config.setBoolean(ConfigOptions.KV_STATISTICS_ENABLED.key(), true);
+
+ // Create and close many containers to test memory management
+ List<RocksDBResourceContainer> containers = new ArrayList<>();
+ for (int i = 0; i < 10; i++) {
+ RocksDBResourceContainer container = new RocksDBResourceContainer(config, null);
+ containers.add(container);
+
+ // Verify statistics is created and valid
+ Statistics stats = container.getStatistics();
+ assertThat(stats).isNotNull();
+ assertThat(stats.isOwningHandle()).isTrue();
+ }
+
+ // Close all containers
+ for (RocksDBResourceContainer container : containers) {
+ container.close();
+ }
+
+ // This should not cause memory leaks or crashes
+ System.gc(); // Suggest garbage collection
+ }
+
+ @Test
+ void testStatisticsThreadSafety() throws Exception {
+ Configuration config = new Configuration();
+ config.setBoolean(ConfigOptions.KV_STATISTICS_ENABLED.key(), true);
+
+ try (RocksDBResourceContainer container = new RocksDBResourceContainer(config, null)) {
+ Statistics statistics = container.getStatistics();
+ assertThat(statistics).isNotNull();
+
+ // Test concurrent access to statistics
+ int threadCount = 10;
+ CountDownLatch latch = new CountDownLatch(threadCount);
+ List<Exception> exceptions = Collections.synchronizedList(new ArrayList<>());
+
+ for (int i = 0; i < threadCount; i++) {
+ new Thread(
+ () -> {
+ try {
+ // Access statistics concurrently
Review Comment:
I don't see any tests for the concurrency accessing. I think we don't need to add concurrency test just for `RocksDBResourceContainer#getStatistics`.
##########
fluss-server/src/test/java/com/alibaba/fluss/server/kv/rocksdb/RocksDBResourceContainerTest.java:
##########
@@ -187,8 +193,237 @@ void testConfigurationOptionsFromConfig() throws Exception {
(BlockBasedTableConfig) columnOptions.tableFormatConfig();
assertThat(tableConfig.blockSize()).isEqualTo(4 * SizeUnit.KB);
assertThat(tableConfig.metadataBlockSize()).isEqualTo(8 * SizeUnit.KB);
- assertThat(tableConfig.blockCacheSize()).isEqualTo(512 * SizeUnit.MB);
assertThat(tableConfig.filterPolicy() instanceof BloomFilter).isTrue();
}
}
+
+ @Test
+ void testBlockCacheCreation() throws Exception {
+ Configuration config = new Configuration();
+ config.setString(ConfigOptions.KV_BLOCK_CACHE_SIZE.key(), "16mb");
+
+ try (RocksDBResourceContainer container = new RocksDBResourceContainer(config, null)) {
+ Cache blockCache = container.getBlockCache();
+ assertThat(blockCache).isNotNull();
+ }
+ }
+
+ @Test
+ void testBlockCacheNotCreatedWhenSizeIsZero() throws Exception {
+ Configuration config = new Configuration();
+ config.setString(ConfigOptions.KV_BLOCK_CACHE_SIZE.key(), "0mb");
+
+ try (RocksDBResourceContainer container = new RocksDBResourceContainer(config, null)) {
+ Cache blockCache = container.getBlockCache();
+ assertThat(blockCache).isNull();
+ }
+ }
+
+ @Test
+ void testStatisticsEnabledByDefault() throws Exception {
+ Configuration config = new Configuration();
+ // Statistics should be enabled by default
+ try (RocksDBResourceContainer container = new RocksDBResourceContainer(config, null)) {
+ DBOptions dbOptions = container.getDbOptions();
+ assertThat(dbOptions.statistics()).isNotNull();
+ assertThat(container.getStatistics()).isNotNull();
+ }
+ }
+
+ @Test
+ void testStatisticsDisabled() throws Exception {
+ Configuration config = new Configuration();
+ config.setBoolean(ConfigOptions.KV_STATISTICS_ENABLED.key(), false);
+
+ try (RocksDBResourceContainer container = new RocksDBResourceContainer(config, null)) {
+ DBOptions dbOptions = container.getDbOptions();
+ // Statistics should be null when disabled
+ assertThat(dbOptions.statistics()).isNull();
+ assertThat(container.getStatistics()).isNull();
+ }
+ }
+
+ @Test
+ void testStatisticsExplicitlyEnabled() throws Exception {
+ Configuration config = new Configuration();
+ config.setBoolean(ConfigOptions.KV_STATISTICS_ENABLED.key(), true);
+
+ try (RocksDBResourceContainer container = new RocksDBResourceContainer(config, null)) {
+ DBOptions dbOptions = container.getDbOptions();
+ assertThat(dbOptions.statistics()).isNotNull();
+ assertThat(container.getStatistics()).isNotNull();
+ }
+ }
+
+ @Test
+ void testStatisticsClosedWithContainer() throws Exception {
+ Configuration config = new Configuration();
+ config.setBoolean(ConfigOptions.KV_STATISTICS_ENABLED.key(), true);
+
+ Statistics statistics;
+ try (RocksDBResourceContainer container = new RocksDBResourceContainer(config, null)) {
+ statistics = container.getStatistics();
+ assertThat(statistics).isNotNull();
+ assertThat(statistics.isOwningHandle()).isTrue();
+ }
+
+ // Statistics should be closed when container is closed
+ assertThat(statistics.isOwningHandle()).isFalse();
+ }
+
+ @Test
+ void testMultipleStatisticsObjectsClosedProperly() throws Exception {
+ Configuration config = new Configuration();
+ config.setBoolean(ConfigOptions.KV_STATISTICS_ENABLED.key(), true);
+
+ List<Statistics> statisticsList = new ArrayList<>();
+
+ // Create multiple containers and collect statistics
+ for (int i = 0; i < 5; i++) {
+ RocksDBResourceContainer container = new RocksDBResourceContainer(config, null);
+ Statistics stats = container.getStatistics();
+ if (stats != null) {
+ statisticsList.add(stats);
+ }
+ container.close();
+ }
+
+ // All statistics should be closed
+ for (Statistics stats : statisticsList) {
+ assertThat(stats.isOwningHandle()).isFalse();
+ }
+ }
+
+ @Test
+ void testStatisticsWithNullInstancePath() throws Exception {
+ Configuration config = new Configuration();
+ config.setBoolean(ConfigOptions.KV_STATISTICS_ENABLED.key(), true);
+
+ try (RocksDBResourceContainer container = new RocksDBResourceContainer(config, null)) {
+ Statistics statistics = container.getStatistics();
+ assertThat(statistics).isNotNull();
+ assertThat(statistics.isOwningHandle()).isTrue();
+ }
+ }
+
+ @Test
+ void testStatisticsWithValidInstancePath(@TempDir Path tempFolder) throws Exception {
Review Comment:
I don't see any values of this test, is it necessary?
##########
fluss-server/src/test/java/com/alibaba/fluss/server/kv/rocksdb/RocksDBMetricsCollectorTest.java:
##########
@@ -0,0 +1,248 @@
+/*
+ * 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 com.alibaba.fluss.server.kv.rocksdb;
+
+import com.alibaba.fluss.config.Configuration;
+import com.alibaba.fluss.metadata.TableBucket;
+import com.alibaba.fluss.metadata.TablePath;
+import com.alibaba.fluss.metrics.MetricNames;
+import com.alibaba.fluss.server.metrics.group.BucketMetricGroup;
+import com.alibaba.fluss.server.metrics.group.TestingMetricGroups;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.jupiter.api.io.TempDir;
+import org.mockito.MockitoAnnotations;
+import org.rocksdb.RocksDB;
+import org.rocksdb.Statistics;
+
+import java.nio.file.Path;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/** Tests for {@link RocksDBMetricsCollector}. */
+class RocksDBMetricsCollectorTest {
+
+ @TempDir Path tempDir;
+ private RocksDBResourceContainer resourceContainer;
+ private RocksDBMetricsCollector collector;
+ private Statistics mockStatistics;
+ private AutoCloseable mocksCloseable;
+ private BucketMetricGroup bucketMetricGroup;
+ private AutoCloseable mockMemUtilAutocloseable;
Review Comment:
never used, can remove
##########
fluss-server/src/main/java/com/alibaba/fluss/server/kv/rocksdb/RocksDBMetricsManager.java:
##########
@@ -0,0 +1,223 @@
+/*
+ * 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 com.alibaba.fluss.server.kv.rocksdb;
+
+import com.alibaba.fluss.utils.MapUtils;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * Global collector for RocksDB metrics that efficiently manages metric collection for thousands of
+ * RocksDB instances using a shared thread pool and batch processing.
+ */
+public class RocksDBMetricsManager {
+
+ private static final Logger LOG = LoggerFactory.getLogger(RocksDBMetricsManager.class);
+
+ /** Singleton instance. */
+ private static volatile RocksDBMetricsManager instance;
+
+ /** Update interval for metrics collection (in seconds). */
+ private static final long METRICS_UPDATE_INTERVAL_SECONDS = 5;
+
+ /** Number of threads for metrics collection. */
+ private static final int METRICS_COLLECTION_THREADS =
+ Math.max(1, Runtime.getRuntime().availableProcessors() / 4);
Review Comment:
IMO, single thread for rocksdb metrics updating is enough. A metric reporter shouldn't consume so many threads, otherwise, there must be too many metrics.
##########
fluss-server/src/test/java/com/alibaba/fluss/server/kv/rocksdb/RocksDBMetricsManagerTest.java:
##########
@@ -0,0 +1,423 @@
+/*
+ * 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 com.alibaba.fluss.server.kv.rocksdb;
+
+import com.alibaba.fluss.config.Configuration;
+import com.alibaba.fluss.metadata.TableBucket;
+import com.alibaba.fluss.metadata.TablePath;
+import com.alibaba.fluss.server.metrics.group.TestingMetricGroups;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.jupiter.api.io.TempDir;
+import org.mockito.MockitoAnnotations;
+import org.rocksdb.RocksDB;
+import org.rocksdb.Statistics;
+
+import java.lang.reflect.Field;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+/** Tests for {@link RocksDBMetricsManager}. */
+class RocksDBMetricsManagerTest {
+
+ @TempDir Path tempDir;
+ private RocksDBResourceContainer resourceContainer;
+ private List<RocksDBMetricsCollector> collectors;
+ private AutoCloseable mocksCloseable;
+ @RegisterExtension public RocksDBExtension rocksDBExtension = new RocksDBExtension();
+
+ @BeforeEach
+ void setUp() {
+ RocksDB.loadLibrary();
+ mocksCloseable = MockitoAnnotations.openMocks(this);
+ // Setup resource container
+ resourceContainer = new RocksDBResourceContainer(new Configuration(), tempDir.toFile());
+
+ collectors = new ArrayList<>();
+
+ // Reset singleton instance for testing
+ resetSingleton();
+ }
+
+ @AfterEach
+ void tearDown() throws Exception {
+ // Close all collectors
+ for (RocksDBMetricsCollector collector : collectors) {
+ collector.close();
+ }
+ collectors.clear();
+
+ // Shutdown manager
+ RocksDBMetricsManager.getInstance().shutdown();
+
+ // Clean up resources
+ if (resourceContainer != null) {
+ resourceContainer.close();
+ }
+ if (mocksCloseable != null) {
+ mocksCloseable.close();
+ }
+
+ // Reset singleton for next test
+ resetSingleton();
+ }
+
+ @Test
+ void testRegisterCollector() throws Exception {
+
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ assertThat(manager.getCollectorCount()).isEqualTo(0);
+
+ RocksDBMetricsCollector collector = createMockCollector("test_db", "test_table", 0);
+ manager.registerCollector(collector);
+
+ assertThat(manager.getCollectorCount()).isEqualTo(1);
+ assertThat(manager.isRunning()).isTrue();
+
+ collectors.add(collector);
+ }
+
+ @Test
+ void testRegisterMultipleCollectors() throws Exception {
+
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ for (int i = 0; i < 5; i++) {
+ RocksDBMetricsCollector collector = createMockCollector("test_db", "test_table", i);
+ manager.registerCollector(collector);
+ collectors.add(collector);
+ }
+
+ assertThat(manager.getCollectorCount()).isEqualTo(5);
+ assertThat(manager.isRunning()).isTrue();
+ }
+
+ @Test
+ void testUnregisterCollector() throws Exception {
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ RocksDBMetricsCollector collector = createMockCollector("test_db", "test_table", 0);
+ manager.registerCollector(collector);
+ collectors.add(collector);
+
+ assertThat(manager.getCollectorCount()).isEqualTo(1);
+
+ manager.unregisterCollector(collector);
+
+ assertThat(manager.getCollectorCount()).isEqualTo(0);
+ }
+
+ @Test
+ void testRegisterAfterShutdown() throws Exception {
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+ manager.shutdown();
+
+ RocksDBMetricsCollector collector = createMockCollector("test_db", "test_table", 0);
+ manager.registerCollector(collector);
+
+ // Should not be registered after shutdown
+ assertThat(manager.getCollectorCount()).isEqualTo(0);
+
+ collectors.add(collector);
+ }
+
+ @Test
+ void testMetricsCollectionStartsWithFirstCollector() throws Exception {
+
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ assertThat(manager.isRunning()).isFalse();
+
+ RocksDBMetricsCollector collector = createMockCollector("test_db", "test_table", 0);
+ manager.registerCollector(collector);
+ collectors.add(collector);
+
+ assertThat(manager.isRunning()).isTrue();
+ }
+
+ @Test
+ void testBatchProcessing() throws Exception {
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ // Create many collectors to test batch processing
+ int collectorCount = 250; // More than batch size (100)
+ List<RocksDBMetricsCollector> spyCollectors = new ArrayList<>();
+
+ for (int i = 0; i < collectorCount; i++) {
+ RocksDBMetricsCollector collector = createMockCollector("test_db", "test_table", i);
+ RocksDBMetricsCollector spyCollector = spy(collector);
+ manager.registerCollector(spyCollector);
+ spyCollectors.add(spyCollector);
+ collectors.add(collector);
+ }
+
+ // Wait for metrics collection to occur
+ Thread.sleep(6000); // Wait for at least one collection cycle
Review Comment:
Do not use `Thread.sleep` for testing. This kills time and is unstable when the CPU load is high on testing machines.
The suggested way for testing scheduler is using `ManuallyTriggeredScheduledExecutorService` in tests. See `MetricViewUpdater` and `MetricViewUpdaterTest` as an example. The key idea is adding a protected constructor for `RocksDBMetricsManager` with `ScheduledExecutorService metricsCollectionExecutor` as parameter, and create the `RocksDBMetricsManager` with the `ManuallyTriggeredScheduledExecutorService` in the test method. And manually trigger the scheduler instead of sleeping.
##########
fluss-server/src/main/java/com/alibaba/fluss/server/kv/rocksdb/RocksDBMetricsManager.java:
##########
@@ -0,0 +1,223 @@
+/*
+ * 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 com.alibaba.fluss.server.kv.rocksdb;
+
+import com.alibaba.fluss.utils.MapUtils;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * Global collector for RocksDB metrics that efficiently manages metric collection for thousands of
+ * RocksDB instances using a shared thread pool and batch processing.
+ */
+public class RocksDBMetricsManager {
+
+ private static final Logger LOG = LoggerFactory.getLogger(RocksDBMetricsManager.class);
+
+ /** Singleton instance. */
+ private static volatile RocksDBMetricsManager instance;
+
+ /** Update interval for metrics collection (in seconds). */
+ private static final long METRICS_UPDATE_INTERVAL_SECONDS = 5;
+
+ /** Number of threads for metrics collection. */
+ private static final int METRICS_COLLECTION_THREADS =
+ Math.max(1, Runtime.getRuntime().availableProcessors() / 4);
+
+ /** Batch size for processing metrics collectors. */
+ private static final int BATCH_SIZE = 100;
+
+ /** Shared thread pool for metrics collection. */
+ private final ScheduledExecutorService metricsCollectionExecutor;
+
+ /** Map of registered metrics collectors. */
+ private final ConcurrentMap<String, RocksDBMetricsCollector> collectors =
+ MapUtils.newConcurrentHashMap();
+
+ /** Flag to track if collector is running. */
+ private final AtomicBoolean running = new AtomicBoolean(false);
+
+ /** Flag to track if collector is shutdown. */
+ private final AtomicBoolean shutdown = new AtomicBoolean(false);
+
+ private RocksDBMetricsManager() {
+ this.metricsCollectionExecutor =
+ Executors.newScheduledThreadPool(
+ METRICS_COLLECTION_THREADS,
+ r -> {
+ Thread thread = new Thread(r, "rocksdb-metrics-collector");
+ thread.setDaemon(true);
+ return thread;
+ });
+ }
+
+ /** Get the singleton instance of RocksDBMetricsCollector. */
+ public static RocksDBMetricsManager getInstance() {
+ if (instance == null) {
+ synchronized (RocksDBMetricsManager.class) {
+ if (instance == null) {
+ instance = new RocksDBMetricsManager();
+ }
+ }
+ }
+ return instance;
+ }
+
+ /**
+ * Register a metrics collector for collection.
+ *
+ * @param collector the collector to register
+ */
+ public void registerCollector(RocksDBMetricsCollector collector) {
+ if (shutdown.get()) {
+ LOG.warn("Cannot register collector after collector is shutdown");
+ return;
+ }
+
+ String key = createCollectorKey(collector);
+ collectors.put(key, collector);
+
+ // Start the collection if this is the first collector
+ if (collectors.size() == 1 && !running.get()) {
+ startCollection();
+ }
+
+ LOG.debug(
+ "Registered RocksDB metrics collector for {}, total collectors: {}",
+ key,
+ collectors.size());
+ }
+
+ /**
+ * Unregister a metrics collector.
+ *
+ * @param collector the collector to unregister
+ */
+ public void unregisterCollector(RocksDBMetricsCollector collector) {
+ String key = createCollectorKey(collector);
+ collectors.remove(key);
+
+ LOG.debug(
+ "Unregistered RocksDB metrics collector for {}, remaining collectors: {}",
+ key,
+ collectors.size());
+ }
+
+ /** Start the metrics collection if not already running. */
+ private void startCollection() {
+ if (running.compareAndSet(false, true)) {
+ LOG.info(
+ "Starting RocksDB metrics collection with {} threads",
+ METRICS_COLLECTION_THREADS);
+
+ metricsCollectionExecutor.scheduleAtFixedRate(
+ this::gatherMetrics,
+ METRICS_UPDATE_INTERVAL_SECONDS,
+ METRICS_UPDATE_INTERVAL_SECONDS,
+ TimeUnit.SECONDS);
+ }
+ }
+
+ /** Collect metrics from all registered collectors in batches. */
+ private void gatherMetrics() {
+ if (collectors.isEmpty()) {
+ return;
+ }
+
+ List<RocksDBMetricsCollector> collectorList = new ArrayList<>(collectors.values());
+
+ // Process collectors in batches to avoid overwhelming the system
+ for (int i = 0; i < collectorList.size(); i += BATCH_SIZE) {
+ int endIndex = Math.min(i + BATCH_SIZE, collectorList.size());
+ List<RocksDBMetricsCollector> batch = collectorList.subList(i, endIndex);
+
+ // Submit batch for processing
+ metricsCollectionExecutor.submit(() -> processBatch(batch));
+ }
+ }
+
+ /** Process a batch of collectors. */
+ private void processBatch(List<RocksDBMetricsCollector> batch) {
+ for (RocksDBMetricsCollector collector : batch) {
+ try {
+ collector.updateMetrics();
+ } catch (Exception e) {
+ LOG.warn(
+ "Error updating metrics for collector {}",
+ createCollectorKey(collector),
+ e);
+ }
+ }
+ }
+
+ /** Create a unique key for a collector. */
+ private String createCollectorKey(RocksDBMetricsCollector collector) {
+ return String.format(
+ "%s-%s-%s-%s",
Review Comment:
`-` is a valid character in database name and table name, therefore, it's better to use `/` as separator, to avoid the key conflicts.
##########
fluss-server/src/main/java/com/alibaba/fluss/server/kv/rocksdb/RocksDBMetricsCollector.java:
##########
@@ -0,0 +1,527 @@
+/*
+ * 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 com.alibaba.fluss.server.kv.rocksdb;
+
+import com.alibaba.fluss.metadata.TableBucket;
+import com.alibaba.fluss.metadata.TablePath;
+import com.alibaba.fluss.metrics.Gauge;
+import com.alibaba.fluss.metrics.MetricNames;
+import com.alibaba.fluss.metrics.groups.MetricGroup;
+import com.alibaba.fluss.server.metrics.group.BucketMetricGroup;
+
+import org.rocksdb.Cache;
+import org.rocksdb.HistogramData;
+import org.rocksdb.HistogramType;
+import org.rocksdb.MemoryUsageType;
+import org.rocksdb.MemoryUtil;
+import org.rocksdb.RocksDB;
+import org.rocksdb.RocksDBException;
+import org.rocksdb.Statistics;
+import org.rocksdb.TickerType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.io.Closeable;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * RocksDB metrics collector that collects and reports RocksDB statistics including cache,
+ * compaction, memory, and I/O metrics with table bucket and server information tags.
+ */
+public class RocksDBMetricsCollector implements Closeable {
+
+ private static final Logger LOG = LoggerFactory.getLogger(RocksDBMetricsCollector.class);
+
+ private final RocksDB rocksDB;
+ private final Statistics statistics;
+ private final MetricGroup bucketMetricGroup;
+ private final TableBucket tableBucket;
+ private final TablePath tablePath;
+ private final String partitionName;
+ private final RocksDBResourceContainer resourceContainer;
+
+ private volatile boolean registered = false;
+
+ // Cached values for efficient metric access
+ private volatile long blockCacheHitCount = 0;
+ private volatile long blockCacheMissCount = 0;
+ private volatile long indexBlockCacheHitCount = 0;
+ private volatile long indexBlockCacheMissCount = 0;
+ private volatile long filterBlockCacheHitCount = 0;
+ private volatile long filterBlockCacheMissCount = 0;
+ private volatile long dataBlockCacheHitCount = 0;
+ private volatile long dataBlockCacheMissCount = 0;
+ private volatile long compactionCount = 0;
+ private volatile long compactionBytesRead = 0;
+ private volatile long compactionBytesWritten = 0;
+ private volatile long compactionCpuTimeMicros = 0;
+ private volatile long flushCount = 0;
+ private volatile long flushBytesWritten = 0;
+ private volatile long bytesRead = 0;
+ private volatile long bytesWritten = 0;
+ private volatile long numberDbNext = 0;
+ private volatile long numberDbPrev = 0;
+ private volatile long numberDbSeek = 0;
+ private volatile long numberDbSeekFound = 0;
+ private volatile long numberKeysRead = 0;
+ private volatile long numberKeysWritten = 0;
+ private volatile long numberKeysUpdated = 0;
+ private volatile long numLiveVersions = 0;
+ private volatile long numImmutableMemtables = 0;
+ private volatile long activeMemtableSize = 0;
+ private volatile long unflushedMemtableSize = 0;
+ private volatile long totalSstFilesSize = 0;
+ private volatile long liveSstFilesSize = 0;
+ private volatile long sizeAllMemtables = 0;
+ private volatile long blockCacheAddCount = 0;
+ private volatile long memtableMemoryUsage = 0;
+ private volatile long blockCacheMemoryUsage = 0;
+ private volatile long tableReadersMemoryUsage = 0;
+ private volatile long totalMemoryUsage = 0;
+ private volatile long blockCacheUsage = 0;
+ private volatile long blockCachePinnedUsage = 0;
+ private volatile long stallTimeMicros = 0;
+
+ // Level SST files metrics
+ private volatile long numFilesAtLevel0 = 0;
+ private volatile long numFilesAtLevel1 = 0;
+ private volatile long numFilesAtLevel2 = 0;
+ private volatile long numFilesAtLevel3 = 0;
+ private volatile long numFilesAtLevel4 = 0;
+ private volatile long numFilesAtLevel5 = 0;
+ private volatile long numFilesAtLevel6 = 0;
+
+ // Compression and compaction pressure metrics
+ private volatile long compactionPending = 0;
+ private volatile long flushPending = 0;
+ private volatile long numRunningCompactions = 0;
+ private volatile long numRunningFlushes = 0;
+
+ // Write amplification metrics
+ private volatile long writeAmplification = 0;
+ private volatile long readAmplification = 0;
+
+ private volatile long backgroundErrors = 0;
+
+ // Histogram metrics
+ private volatile long dbGetLatencyMicros = 0;
+ private volatile long dbWriteLatencyMicros = 0;
+ private volatile long compactionTimeMicros = 0;
+ private volatile long compressionTimeNanos = 0;
+ private volatile long decompressionTimeNanos = 0;
+
+ public RocksDBMetricsCollector(
+ RocksDB rocksDB,
+ Statistics statistics,
+ BucketMetricGroup bucketMetricGroup,
+ TableBucket tableBucket,
+ TablePath tablePath,
+ @Nullable String partitionName,
+ RocksDBResourceContainer resourceContainer) {
+ this.rocksDB = rocksDB;
+ this.statistics = statistics;
+ this.tableBucket = tableBucket;
+ this.tablePath = tablePath;
+ this.partitionName = partitionName;
+ this.resourceContainer = resourceContainer;
+
+ // Create RocksDB metrics group under the bucket metrics
+ this.bucketMetricGroup = bucketMetricGroup;
+
+ // Register all metrics
+ registerMetrics();
+
+ // Register this reporter with the global collector
+ RocksDBMetricsManager.getInstance().registerCollector(this);
+ this.registered = true;
+
+ LOG.info(
+ "RocksDB metrics collector started for table {} bucket {}",
+ tablePath,
+ tableBucket.getBucket());
+ }
+
+ private void registerMetrics() {
+ // Cache metrics - 统一使用缓存变量
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_BLOCK_CACHE_HIT_COUNT, (Gauge<Long>) () -> blockCacheHitCount);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_BLOCK_CACHE_MISS_COUNT,
+ (Gauge<Long>) () -> blockCacheMissCount);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_INDEX_BLOCK_CACHE_HIT_COUNT,
+ (Gauge<Long>) () -> indexBlockCacheHitCount);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_INDEX_BLOCK_CACHE_MISS_COUNT,
+ (Gauge<Long>) () -> indexBlockCacheMissCount);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_FILTER_BLOCK_CACHE_HIT_COUNT,
+ (Gauge<Long>) () -> filterBlockCacheHitCount);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_FILTER_BLOCK_CACHE_MISS_COUNT,
+ (Gauge<Long>) () -> filterBlockCacheMissCount);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_DATA_BLOCK_CACHE_HIT_COUNT,
+ (Gauge<Long>) () -> dataBlockCacheHitCount);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_DATA_BLOCK_CACHE_MISS_COUNT,
+ (Gauge<Long>) () -> dataBlockCacheMissCount);
+
+ // Compaction metrics
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_COMPACTION_COUNT, (Gauge<Long>) () -> compactionCount);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_COMPACTION_BYTES_READ, (Gauge<Long>) () -> compactionBytesRead);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_COMPACTION_BYTES_WRITTEN,
+ (Gauge<Long>) () -> compactionBytesWritten);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_COMPACTION_CPU_TIME_MICROS,
+ (Gauge<Long>) () -> compactionCpuTimeMicros);
+ bucketMetricGroup.gauge(MetricNames.ROCKSDB_FLUSH_COUNT, (Gauge<Long>) () -> flushCount);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_FLUSH_BYTES_WRITTEN, (Gauge<Long>) () -> flushBytesWritten);
+
+ // Memory metrics - 统一使用缓存变量
Review Comment:
Use English.
##########
fluss-server/src/test/java/com/alibaba/fluss/server/kv/rocksdb/RocksDBMetricsManagerTest.java:
##########
@@ -0,0 +1,423 @@
+/*
+ * 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 com.alibaba.fluss.server.kv.rocksdb;
+
+import com.alibaba.fluss.config.Configuration;
+import com.alibaba.fluss.metadata.TableBucket;
+import com.alibaba.fluss.metadata.TablePath;
+import com.alibaba.fluss.server.metrics.group.TestingMetricGroups;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.jupiter.api.io.TempDir;
+import org.mockito.MockitoAnnotations;
+import org.rocksdb.RocksDB;
+import org.rocksdb.Statistics;
+
+import java.lang.reflect.Field;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+/** Tests for {@link RocksDBMetricsManager}. */
+class RocksDBMetricsManagerTest {
+
+ @TempDir Path tempDir;
+ private RocksDBResourceContainer resourceContainer;
+ private List<RocksDBMetricsCollector> collectors;
+ private AutoCloseable mocksCloseable;
+ @RegisterExtension public RocksDBExtension rocksDBExtension = new RocksDBExtension();
+
+ @BeforeEach
+ void setUp() {
+ RocksDB.loadLibrary();
+ mocksCloseable = MockitoAnnotations.openMocks(this);
+ // Setup resource container
+ resourceContainer = new RocksDBResourceContainer(new Configuration(), tempDir.toFile());
+
+ collectors = new ArrayList<>();
+
+ // Reset singleton instance for testing
+ resetSingleton();
+ }
+
+ @AfterEach
+ void tearDown() throws Exception {
+ // Close all collectors
+ for (RocksDBMetricsCollector collector : collectors) {
+ collector.close();
+ }
+ collectors.clear();
+
+ // Shutdown manager
+ RocksDBMetricsManager.getInstance().shutdown();
+
+ // Clean up resources
+ if (resourceContainer != null) {
+ resourceContainer.close();
+ }
+ if (mocksCloseable != null) {
+ mocksCloseable.close();
+ }
+
+ // Reset singleton for next test
+ resetSingleton();
+ }
+
+ @Test
+ void testRegisterCollector() throws Exception {
+
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ assertThat(manager.getCollectorCount()).isEqualTo(0);
+
+ RocksDBMetricsCollector collector = createMockCollector("test_db", "test_table", 0);
+ manager.registerCollector(collector);
+
+ assertThat(manager.getCollectorCount()).isEqualTo(1);
+ assertThat(manager.isRunning()).isTrue();
+
+ collectors.add(collector);
+ }
+
+ @Test
+ void testRegisterMultipleCollectors() throws Exception {
+
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ for (int i = 0; i < 5; i++) {
+ RocksDBMetricsCollector collector = createMockCollector("test_db", "test_table", i);
+ manager.registerCollector(collector);
+ collectors.add(collector);
+ }
+
+ assertThat(manager.getCollectorCount()).isEqualTo(5);
+ assertThat(manager.isRunning()).isTrue();
+ }
+
+ @Test
+ void testUnregisterCollector() throws Exception {
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ RocksDBMetricsCollector collector = createMockCollector("test_db", "test_table", 0);
+ manager.registerCollector(collector);
+ collectors.add(collector);
+
+ assertThat(manager.getCollectorCount()).isEqualTo(1);
+
+ manager.unregisterCollector(collector);
+
+ assertThat(manager.getCollectorCount()).isEqualTo(0);
+ }
+
+ @Test
+ void testRegisterAfterShutdown() throws Exception {
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+ manager.shutdown();
+
+ RocksDBMetricsCollector collector = createMockCollector("test_db", "test_table", 0);
+ manager.registerCollector(collector);
+
+ // Should not be registered after shutdown
+ assertThat(manager.getCollectorCount()).isEqualTo(0);
+
+ collectors.add(collector);
+ }
+
+ @Test
+ void testMetricsCollectionStartsWithFirstCollector() throws Exception {
+
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ assertThat(manager.isRunning()).isFalse();
+
+ RocksDBMetricsCollector collector = createMockCollector("test_db", "test_table", 0);
+ manager.registerCollector(collector);
+ collectors.add(collector);
+
+ assertThat(manager.isRunning()).isTrue();
+ }
+
+ @Test
+ void testBatchProcessing() throws Exception {
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ // Create many collectors to test batch processing
+ int collectorCount = 250; // More than batch size (100)
+ List<RocksDBMetricsCollector> spyCollectors = new ArrayList<>();
+
+ for (int i = 0; i < collectorCount; i++) {
+ RocksDBMetricsCollector collector = createMockCollector("test_db", "test_table", i);
+ RocksDBMetricsCollector spyCollector = spy(collector);
+ manager.registerCollector(spyCollector);
+ spyCollectors.add(spyCollector);
+ collectors.add(collector);
+ }
+
+ // Wait for metrics collection to occur
+ Thread.sleep(6000); // Wait for at least one collection cycle
+
+ // Verify that updateMetrics was called on all collectors
+ for (RocksDBMetricsCollector collector : spyCollectors) {
+ verify(collector, times(1)).updateMetrics();
+ }
+ }
+
+ @Test
+ void testExceptionHandlingInMetricsCollection() throws Exception {
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ // Create a collector that throws exception
+ RocksDBMetricsCollector faultyCollector = createMockCollector("test_db", "test_table", 0);
+ doThrow(new RuntimeException("Test exception")).when(faultyCollector).updateMetrics();
+
+ // Create a normal collector
+ RocksDBMetricsCollector normalCollector = createMockCollector("test_db", "test_table", 1);
+ RocksDBMetricsCollector spyNormalCollector = spy(normalCollector);
+
+ manager.registerCollector(faultyCollector);
+ manager.registerCollector(spyNormalCollector);
+
+ collectors.add(faultyCollector);
+ collectors.add(normalCollector);
+
+ // Wait for metrics collection
+ try {
+ Thread.sleep(6000);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+
+ // Normal collector should still be called despite exception in faulty collector
+ verify(spyNormalCollector, times(1)).updateMetrics();
+ }
+
+ @Test
+ void testShutdown() {
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ // Register some collectors
+ for (int i = 0; i < 3; i++) {
+ RocksDBMetricsCollector collector = createMockCollector("test_db", "test_table", i);
+ manager.registerCollector(collector);
+ collectors.add(collector);
+ }
+
+ assertThat(manager.isRunning()).isTrue();
+ assertThat(manager.getCollectorCount()).isEqualTo(3);
+
+ manager.shutdown();
+
+ assertThat(manager.isRunning()).isFalse();
+ assertThat(manager.isShutdown()).isTrue();
+ assertThat(manager.getCollectorCount()).isEqualTo(0);
+ }
+
+ @Test
+ void testDoubleShutdown() {
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ manager.shutdown();
+ assertThat(manager.isShutdown()).isTrue();
+
+ // Second shutdown should not cause issues
+ manager.shutdown();
+ assertThat(manager.isShutdown()).isTrue();
+ }
+
+ @Test
+ void testConcurrentRegistrationAndUnregistration() throws Exception {
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ int threadCount = 10;
+ CountDownLatch latch = new CountDownLatch(threadCount);
+ ExecutorService executor = Executors.newFixedThreadPool(threadCount);
+ AtomicInteger collectorId = new AtomicInteger(0);
+
+ for (int i = 0; i < threadCount; i++) {
+ executor.submit(
+ () -> {
+ try {
+ for (int j = 0; j < 10; j++) {
+ RocksDBMetricsCollector collector =
+ createMockCollector(
+ "test_db",
+ "test_table",
+ collectorId.getAndIncrement());
+ manager.registerCollector(collector);
+ collectors.add(collector);
+
+ // Immediately unregister some collectors
+ if (j % 2 == 0) {
+ manager.unregisterCollector(collector);
+ }
+ }
+ } finally {
+ latch.countDown();
+ }
+ });
+ }
+
+ latch.await(10, TimeUnit.SECONDS);
+ executor.shutdown();
+
+ // Should handle concurrent access gracefully
+ assertThat(manager.getCollectorCount()).isGreaterThanOrEqualTo(0);
Review Comment:
I think we can assert the count value `50` here? It should be deterministic, right?
##########
fluss-server/src/main/java/com/alibaba/fluss/server/kv/rocksdb/RocksDBMetricsManager.java:
##########
@@ -0,0 +1,223 @@
+/*
+ * 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 com.alibaba.fluss.server.kv.rocksdb;
+
+import com.alibaba.fluss.utils.MapUtils;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * Global collector for RocksDB metrics that efficiently manages metric collection for thousands of
+ * RocksDB instances using a shared thread pool and batch processing.
+ */
+public class RocksDBMetricsManager {
+
+ private static final Logger LOG = LoggerFactory.getLogger(RocksDBMetricsManager.class);
+
+ /** Singleton instance. */
+ private static volatile RocksDBMetricsManager instance;
+
+ /** Update interval for metrics collection (in seconds). */
+ private static final long METRICS_UPDATE_INTERVAL_SECONDS = 5;
+
+ /** Number of threads for metrics collection. */
+ private static final int METRICS_COLLECTION_THREADS =
+ Math.max(1, Runtime.getRuntime().availableProcessors() / 4);
+
+ /** Batch size for processing metrics collectors. */
+ private static final int BATCH_SIZE = 100;
+
+ /** Shared thread pool for metrics collection. */
+ private final ScheduledExecutorService metricsCollectionExecutor;
+
+ /** Map of registered metrics collectors. */
+ private final ConcurrentMap<String, RocksDBMetricsCollector> collectors =
+ MapUtils.newConcurrentHashMap();
+
+ /** Flag to track if collector is running. */
+ private final AtomicBoolean running = new AtomicBoolean(false);
+
+ /** Flag to track if collector is shutdown. */
+ private final AtomicBoolean shutdown = new AtomicBoolean(false);
+
+ private RocksDBMetricsManager() {
+ this.metricsCollectionExecutor =
+ Executors.newScheduledThreadPool(
+ METRICS_COLLECTION_THREADS,
+ r -> {
+ Thread thread = new Thread(r, "rocksdb-metrics-collector");
+ thread.setDaemon(true);
+ return thread;
+ });
+ }
+
+ /** Get the singleton instance of RocksDBMetricsCollector. */
+ public static RocksDBMetricsManager getInstance() {
+ if (instance == null) {
+ synchronized (RocksDBMetricsManager.class) {
+ if (instance == null) {
+ instance = new RocksDBMetricsManager();
+ }
+ }
+ }
+ return instance;
+ }
+
+ /**
+ * Register a metrics collector for collection.
+ *
+ * @param collector the collector to register
+ */
+ public void registerCollector(RocksDBMetricsCollector collector) {
+ if (shutdown.get()) {
+ LOG.warn("Cannot register collector after collector is shutdown");
+ return;
+ }
+
+ String key = createCollectorKey(collector);
+ collectors.put(key, collector);
+
+ // Start the collection if this is the first collector
+ if (collectors.size() == 1 && !running.get()) {
+ startCollection();
+ }
+
+ LOG.debug(
+ "Registered RocksDB metrics collector for {}, total collectors: {}",
+ key,
+ collectors.size());
+ }
+
+ /**
+ * Unregister a metrics collector.
+ *
+ * @param collector the collector to unregister
+ */
+ public void unregisterCollector(RocksDBMetricsCollector collector) {
+ String key = createCollectorKey(collector);
+ collectors.remove(key);
+
+ LOG.debug(
+ "Unregistered RocksDB metrics collector for {}, remaining collectors: {}",
+ key,
+ collectors.size());
+ }
+
+ /** Start the metrics collection if not already running. */
+ private void startCollection() {
+ if (running.compareAndSet(false, true)) {
+ LOG.info(
+ "Starting RocksDB metrics collection with {} threads",
+ METRICS_COLLECTION_THREADS);
+
+ metricsCollectionExecutor.scheduleAtFixedRate(
+ this::gatherMetrics,
+ METRICS_UPDATE_INTERVAL_SECONDS,
+ METRICS_UPDATE_INTERVAL_SECONDS,
+ TimeUnit.SECONDS);
+ }
+ }
+
+ /** Collect metrics from all registered collectors in batches. */
+ private void gatherMetrics() {
+ if (collectors.isEmpty()) {
+ return;
+ }
+
+ List<RocksDBMetricsCollector> collectorList = new ArrayList<>(collectors.values());
+
+ // Process collectors in batches to avoid overwhelming the system
+ for (int i = 0; i < collectorList.size(); i += BATCH_SIZE) {
+ int endIndex = Math.min(i + BATCH_SIZE, collectorList.size());
Review Comment:
If the schduler is single-thread, then there is no need to submit batching tasks.
##########
fluss-server/src/test/java/com/alibaba/fluss/server/kv/rocksdb/RocksDBMetricsManagerTest.java:
##########
@@ -0,0 +1,423 @@
+/*
+ * 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 com.alibaba.fluss.server.kv.rocksdb;
+
+import com.alibaba.fluss.config.Configuration;
+import com.alibaba.fluss.metadata.TableBucket;
+import com.alibaba.fluss.metadata.TablePath;
+import com.alibaba.fluss.server.metrics.group.TestingMetricGroups;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.jupiter.api.io.TempDir;
+import org.mockito.MockitoAnnotations;
+import org.rocksdb.RocksDB;
+import org.rocksdb.Statistics;
+
+import java.lang.reflect.Field;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+/** Tests for {@link RocksDBMetricsManager}. */
+class RocksDBMetricsManagerTest {
+
+ @TempDir Path tempDir;
+ private RocksDBResourceContainer resourceContainer;
+ private List<RocksDBMetricsCollector> collectors;
+ private AutoCloseable mocksCloseable;
+ @RegisterExtension public RocksDBExtension rocksDBExtension = new RocksDBExtension();
+
+ @BeforeEach
+ void setUp() {
+ RocksDB.loadLibrary();
+ mocksCloseable = MockitoAnnotations.openMocks(this);
+ // Setup resource container
+ resourceContainer = new RocksDBResourceContainer(new Configuration(), tempDir.toFile());
+
+ collectors = new ArrayList<>();
+
+ // Reset singleton instance for testing
+ resetSingleton();
+ }
+
+ @AfterEach
+ void tearDown() throws Exception {
+ // Close all collectors
+ for (RocksDBMetricsCollector collector : collectors) {
+ collector.close();
+ }
+ collectors.clear();
+
+ // Shutdown manager
+ RocksDBMetricsManager.getInstance().shutdown();
+
+ // Clean up resources
+ if (resourceContainer != null) {
+ resourceContainer.close();
+ }
+ if (mocksCloseable != null) {
+ mocksCloseable.close();
+ }
+
+ // Reset singleton for next test
+ resetSingleton();
+ }
+
+ @Test
+ void testRegisterCollector() throws Exception {
+
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ assertThat(manager.getCollectorCount()).isEqualTo(0);
+
+ RocksDBMetricsCollector collector = createMockCollector("test_db", "test_table", 0);
+ manager.registerCollector(collector);
+
+ assertThat(manager.getCollectorCount()).isEqualTo(1);
+ assertThat(manager.isRunning()).isTrue();
+
+ collectors.add(collector);
+ }
+
+ @Test
+ void testRegisterMultipleCollectors() throws Exception {
+
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ for (int i = 0; i < 5; i++) {
+ RocksDBMetricsCollector collector = createMockCollector("test_db", "test_table", i);
+ manager.registerCollector(collector);
+ collectors.add(collector);
+ }
+
+ assertThat(manager.getCollectorCount()).isEqualTo(5);
+ assertThat(manager.isRunning()).isTrue();
+ }
+
+ @Test
+ void testUnregisterCollector() throws Exception {
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ RocksDBMetricsCollector collector = createMockCollector("test_db", "test_table", 0);
+ manager.registerCollector(collector);
+ collectors.add(collector);
+
+ assertThat(manager.getCollectorCount()).isEqualTo(1);
+
+ manager.unregisterCollector(collector);
+
+ assertThat(manager.getCollectorCount()).isEqualTo(0);
+ }
+
+ @Test
+ void testRegisterAfterShutdown() throws Exception {
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+ manager.shutdown();
+
+ RocksDBMetricsCollector collector = createMockCollector("test_db", "test_table", 0);
+ manager.registerCollector(collector);
+
+ // Should not be registered after shutdown
+ assertThat(manager.getCollectorCount()).isEqualTo(0);
+
+ collectors.add(collector);
+ }
+
+ @Test
+ void testMetricsCollectionStartsWithFirstCollector() throws Exception {
+
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ assertThat(manager.isRunning()).isFalse();
+
+ RocksDBMetricsCollector collector = createMockCollector("test_db", "test_table", 0);
+ manager.registerCollector(collector);
+ collectors.add(collector);
+
+ assertThat(manager.isRunning()).isTrue();
+ }
+
+ @Test
+ void testBatchProcessing() throws Exception {
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ // Create many collectors to test batch processing
+ int collectorCount = 250; // More than batch size (100)
+ List<RocksDBMetricsCollector> spyCollectors = new ArrayList<>();
+
+ for (int i = 0; i < collectorCount; i++) {
+ RocksDBMetricsCollector collector = createMockCollector("test_db", "test_table", i);
+ RocksDBMetricsCollector spyCollector = spy(collector);
+ manager.registerCollector(spyCollector);
+ spyCollectors.add(spyCollector);
+ collectors.add(collector);
+ }
+
+ // Wait for metrics collection to occur
+ Thread.sleep(6000); // Wait for at least one collection cycle
+
+ // Verify that updateMetrics was called on all collectors
+ for (RocksDBMetricsCollector collector : spyCollectors) {
+ verify(collector, times(1)).updateMetrics();
+ }
+ }
+
+ @Test
+ void testExceptionHandlingInMetricsCollection() throws Exception {
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ // Create a collector that throws exception
+ RocksDBMetricsCollector faultyCollector = createMockCollector("test_db", "test_table", 0);
+ doThrow(new RuntimeException("Test exception")).when(faultyCollector).updateMetrics();
+
+ // Create a normal collector
+ RocksDBMetricsCollector normalCollector = createMockCollector("test_db", "test_table", 1);
+ RocksDBMetricsCollector spyNormalCollector = spy(normalCollector);
+
+ manager.registerCollector(faultyCollector);
+ manager.registerCollector(spyNormalCollector);
+
+ collectors.add(faultyCollector);
+ collectors.add(normalCollector);
+
+ // Wait for metrics collection
+ try {
+ Thread.sleep(6000);
Review Comment:
ditto
##########
fluss-server/src/main/java/com/alibaba/fluss/server/kv/rocksdb/RocksDBMetricsCollector.java:
##########
@@ -0,0 +1,527 @@
+/*
+ * 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 com.alibaba.fluss.server.kv.rocksdb;
+
+import com.alibaba.fluss.metadata.TableBucket;
+import com.alibaba.fluss.metadata.TablePath;
+import com.alibaba.fluss.metrics.Gauge;
+import com.alibaba.fluss.metrics.MetricNames;
+import com.alibaba.fluss.metrics.groups.MetricGroup;
+import com.alibaba.fluss.server.metrics.group.BucketMetricGroup;
+
+import org.rocksdb.Cache;
+import org.rocksdb.HistogramData;
+import org.rocksdb.HistogramType;
+import org.rocksdb.MemoryUsageType;
+import org.rocksdb.MemoryUtil;
+import org.rocksdb.RocksDB;
+import org.rocksdb.RocksDBException;
+import org.rocksdb.Statistics;
+import org.rocksdb.TickerType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.io.Closeable;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * RocksDB metrics collector that collects and reports RocksDB statistics including cache,
+ * compaction, memory, and I/O metrics with table bucket and server information tags.
+ */
+public class RocksDBMetricsCollector implements Closeable {
+
+ private static final Logger LOG = LoggerFactory.getLogger(RocksDBMetricsCollector.class);
+
+ private final RocksDB rocksDB;
+ private final Statistics statistics;
+ private final MetricGroup bucketMetricGroup;
+ private final TableBucket tableBucket;
+ private final TablePath tablePath;
+ private final String partitionName;
+ private final RocksDBResourceContainer resourceContainer;
+
+ private volatile boolean registered = false;
+
+ // Cached values for efficient metric access
+ private volatile long blockCacheHitCount = 0;
+ private volatile long blockCacheMissCount = 0;
+ private volatile long indexBlockCacheHitCount = 0;
+ private volatile long indexBlockCacheMissCount = 0;
+ private volatile long filterBlockCacheHitCount = 0;
+ private volatile long filterBlockCacheMissCount = 0;
+ private volatile long dataBlockCacheHitCount = 0;
+ private volatile long dataBlockCacheMissCount = 0;
+ private volatile long compactionCount = 0;
+ private volatile long compactionBytesRead = 0;
+ private volatile long compactionBytesWritten = 0;
+ private volatile long compactionCpuTimeMicros = 0;
+ private volatile long flushCount = 0;
+ private volatile long flushBytesWritten = 0;
+ private volatile long bytesRead = 0;
+ private volatile long bytesWritten = 0;
+ private volatile long numberDbNext = 0;
+ private volatile long numberDbPrev = 0;
+ private volatile long numberDbSeek = 0;
+ private volatile long numberDbSeekFound = 0;
+ private volatile long numberKeysRead = 0;
+ private volatile long numberKeysWritten = 0;
+ private volatile long numberKeysUpdated = 0;
+ private volatile long numLiveVersions = 0;
+ private volatile long numImmutableMemtables = 0;
+ private volatile long activeMemtableSize = 0;
+ private volatile long unflushedMemtableSize = 0;
+ private volatile long totalSstFilesSize = 0;
+ private volatile long liveSstFilesSize = 0;
+ private volatile long sizeAllMemtables = 0;
+ private volatile long blockCacheAddCount = 0;
+ private volatile long memtableMemoryUsage = 0;
+ private volatile long blockCacheMemoryUsage = 0;
+ private volatile long tableReadersMemoryUsage = 0;
+ private volatile long totalMemoryUsage = 0;
+ private volatile long blockCacheUsage = 0;
+ private volatile long blockCachePinnedUsage = 0;
+ private volatile long stallTimeMicros = 0;
+
+ // Level SST files metrics
+ private volatile long numFilesAtLevel0 = 0;
+ private volatile long numFilesAtLevel1 = 0;
+ private volatile long numFilesAtLevel2 = 0;
+ private volatile long numFilesAtLevel3 = 0;
+ private volatile long numFilesAtLevel4 = 0;
+ private volatile long numFilesAtLevel5 = 0;
+ private volatile long numFilesAtLevel6 = 0;
+
+ // Compression and compaction pressure metrics
+ private volatile long compactionPending = 0;
+ private volatile long flushPending = 0;
+ private volatile long numRunningCompactions = 0;
+ private volatile long numRunningFlushes = 0;
+
+ // Write amplification metrics
+ private volatile long writeAmplification = 0;
+ private volatile long readAmplification = 0;
+
+ private volatile long backgroundErrors = 0;
+
+ // Histogram metrics
+ private volatile long dbGetLatencyMicros = 0;
+ private volatile long dbWriteLatencyMicros = 0;
+ private volatile long compactionTimeMicros = 0;
+ private volatile long compressionTimeNanos = 0;
+ private volatile long decompressionTimeNanos = 0;
+
+ public RocksDBMetricsCollector(
+ RocksDB rocksDB,
+ Statistics statistics,
+ BucketMetricGroup bucketMetricGroup,
+ TableBucket tableBucket,
+ TablePath tablePath,
+ @Nullable String partitionName,
+ RocksDBResourceContainer resourceContainer) {
+ this.rocksDB = rocksDB;
+ this.statistics = statistics;
+ this.tableBucket = tableBucket;
+ this.tablePath = tablePath;
+ this.partitionName = partitionName;
+ this.resourceContainer = resourceContainer;
+
+ // Create RocksDB metrics group under the bucket metrics
+ this.bucketMetricGroup = bucketMetricGroup;
+
+ // Register all metrics
+ registerMetrics();
+
+ // Register this reporter with the global collector
+ RocksDBMetricsManager.getInstance().registerCollector(this);
+ this.registered = true;
+
+ LOG.info(
+ "RocksDB metrics collector started for table {} bucket {}",
+ tablePath,
+ tableBucket.getBucket());
+ }
+
+ private void registerMetrics() {
+ // Cache metrics - 统一使用缓存变量
Review Comment:
Use English.
##########
fluss-server/src/main/java/com/alibaba/fluss/server/kv/rocksdb/RocksDBMetricsCollector.java:
##########
@@ -0,0 +1,527 @@
+/*
+ * 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 com.alibaba.fluss.server.kv.rocksdb;
+
+import com.alibaba.fluss.metadata.TableBucket;
+import com.alibaba.fluss.metadata.TablePath;
+import com.alibaba.fluss.metrics.Gauge;
+import com.alibaba.fluss.metrics.MetricNames;
+import com.alibaba.fluss.metrics.groups.MetricGroup;
+import com.alibaba.fluss.server.metrics.group.BucketMetricGroup;
+
+import org.rocksdb.Cache;
+import org.rocksdb.HistogramData;
+import org.rocksdb.HistogramType;
+import org.rocksdb.MemoryUsageType;
+import org.rocksdb.MemoryUtil;
+import org.rocksdb.RocksDB;
+import org.rocksdb.RocksDBException;
+import org.rocksdb.Statistics;
+import org.rocksdb.TickerType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.io.Closeable;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * RocksDB metrics collector that collects and reports RocksDB statistics including cache,
+ * compaction, memory, and I/O metrics with table bucket and server information tags.
+ */
+public class RocksDBMetricsCollector implements Closeable {
+
+ private static final Logger LOG = LoggerFactory.getLogger(RocksDBMetricsCollector.class);
+
+ private final RocksDB rocksDB;
+ private final Statistics statistics;
+ private final MetricGroup bucketMetricGroup;
+ private final TableBucket tableBucket;
+ private final TablePath tablePath;
+ private final String partitionName;
+ private final RocksDBResourceContainer resourceContainer;
+
+ private volatile boolean registered = false;
+
+ // Cached values for efficient metric access
+ private volatile long blockCacheHitCount = 0;
+ private volatile long blockCacheMissCount = 0;
+ private volatile long indexBlockCacheHitCount = 0;
+ private volatile long indexBlockCacheMissCount = 0;
+ private volatile long filterBlockCacheHitCount = 0;
+ private volatile long filterBlockCacheMissCount = 0;
+ private volatile long dataBlockCacheHitCount = 0;
+ private volatile long dataBlockCacheMissCount = 0;
+ private volatile long compactionCount = 0;
+ private volatile long compactionBytesRead = 0;
+ private volatile long compactionBytesWritten = 0;
+ private volatile long compactionCpuTimeMicros = 0;
+ private volatile long flushCount = 0;
+ private volatile long flushBytesWritten = 0;
+ private volatile long bytesRead = 0;
+ private volatile long bytesWritten = 0;
+ private volatile long numberDbNext = 0;
+ private volatile long numberDbPrev = 0;
+ private volatile long numberDbSeek = 0;
+ private volatile long numberDbSeekFound = 0;
+ private volatile long numberKeysRead = 0;
+ private volatile long numberKeysWritten = 0;
+ private volatile long numberKeysUpdated = 0;
+ private volatile long numLiveVersions = 0;
+ private volatile long numImmutableMemtables = 0;
+ private volatile long activeMemtableSize = 0;
+ private volatile long unflushedMemtableSize = 0;
+ private volatile long totalSstFilesSize = 0;
+ private volatile long liveSstFilesSize = 0;
+ private volatile long sizeAllMemtables = 0;
+ private volatile long blockCacheAddCount = 0;
+ private volatile long memtableMemoryUsage = 0;
+ private volatile long blockCacheMemoryUsage = 0;
+ private volatile long tableReadersMemoryUsage = 0;
+ private volatile long totalMemoryUsage = 0;
+ private volatile long blockCacheUsage = 0;
+ private volatile long blockCachePinnedUsage = 0;
+ private volatile long stallTimeMicros = 0;
+
+ // Level SST files metrics
+ private volatile long numFilesAtLevel0 = 0;
+ private volatile long numFilesAtLevel1 = 0;
+ private volatile long numFilesAtLevel2 = 0;
+ private volatile long numFilesAtLevel3 = 0;
+ private volatile long numFilesAtLevel4 = 0;
+ private volatile long numFilesAtLevel5 = 0;
+ private volatile long numFilesAtLevel6 = 0;
+
+ // Compression and compaction pressure metrics
+ private volatile long compactionPending = 0;
+ private volatile long flushPending = 0;
+ private volatile long numRunningCompactions = 0;
+ private volatile long numRunningFlushes = 0;
+
+ // Write amplification metrics
+ private volatile long writeAmplification = 0;
+ private volatile long readAmplification = 0;
+
+ private volatile long backgroundErrors = 0;
+
+ // Histogram metrics
+ private volatile long dbGetLatencyMicros = 0;
+ private volatile long dbWriteLatencyMicros = 0;
+ private volatile long compactionTimeMicros = 0;
+ private volatile long compressionTimeNanos = 0;
+ private volatile long decompressionTimeNanos = 0;
+
+ public RocksDBMetricsCollector(
+ RocksDB rocksDB,
+ Statistics statistics,
+ BucketMetricGroup bucketMetricGroup,
+ TableBucket tableBucket,
+ TablePath tablePath,
+ @Nullable String partitionName,
+ RocksDBResourceContainer resourceContainer) {
+ this.rocksDB = rocksDB;
+ this.statistics = statistics;
+ this.tableBucket = tableBucket;
+ this.tablePath = tablePath;
+ this.partitionName = partitionName;
+ this.resourceContainer = resourceContainer;
+
+ // Create RocksDB metrics group under the bucket metrics
+ this.bucketMetricGroup = bucketMetricGroup;
+
+ // Register all metrics
+ registerMetrics();
+
+ // Register this reporter with the global collector
+ RocksDBMetricsManager.getInstance().registerCollector(this);
+ this.registered = true;
+
+ LOG.info(
+ "RocksDB metrics collector started for table {} bucket {}",
+ tablePath,
+ tableBucket.getBucket());
+ }
+
+ private void registerMetrics() {
+ // Cache metrics - 统一使用缓存变量
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_BLOCK_CACHE_HIT_COUNT, (Gauge<Long>) () -> blockCacheHitCount);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_BLOCK_CACHE_MISS_COUNT,
+ (Gauge<Long>) () -> blockCacheMissCount);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_INDEX_BLOCK_CACHE_HIT_COUNT,
+ (Gauge<Long>) () -> indexBlockCacheHitCount);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_INDEX_BLOCK_CACHE_MISS_COUNT,
+ (Gauge<Long>) () -> indexBlockCacheMissCount);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_FILTER_BLOCK_CACHE_HIT_COUNT,
+ (Gauge<Long>) () -> filterBlockCacheHitCount);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_FILTER_BLOCK_CACHE_MISS_COUNT,
+ (Gauge<Long>) () -> filterBlockCacheMissCount);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_DATA_BLOCK_CACHE_HIT_COUNT,
+ (Gauge<Long>) () -> dataBlockCacheHitCount);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_DATA_BLOCK_CACHE_MISS_COUNT,
+ (Gauge<Long>) () -> dataBlockCacheMissCount);
+
+ // Compaction metrics
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_COMPACTION_COUNT, (Gauge<Long>) () -> compactionCount);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_COMPACTION_BYTES_READ, (Gauge<Long>) () -> compactionBytesRead);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_COMPACTION_BYTES_WRITTEN,
+ (Gauge<Long>) () -> compactionBytesWritten);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_COMPACTION_CPU_TIME_MICROS,
+ (Gauge<Long>) () -> compactionCpuTimeMicros);
+ bucketMetricGroup.gauge(MetricNames.ROCKSDB_FLUSH_COUNT, (Gauge<Long>) () -> flushCount);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_FLUSH_BYTES_WRITTEN, (Gauge<Long>) () -> flushBytesWritten);
+
+ // Memory metrics - 统一使用缓存变量
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_MEMTABLE_MEMORY_USAGE, (Gauge<Long>) () -> memtableMemoryUsage);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_BLOCK_CACHE_MEMORY_USAGE,
+ (Gauge<Long>) () -> blockCacheMemoryUsage);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_TABLE_READERS_MEMORY_USAGE,
+ (Gauge<Long>) () -> tableReadersMemoryUsage);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_TOTAL_MEMORY_USAGE, (Gauge<Long>) () -> totalMemoryUsage);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_ACTIVE_MEMTABLE_SIZE, (Gauge<Long>) () -> activeMemtableSize);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_UNFLUSHED_MEMTABLE_SIZE,
+ (Gauge<Long>) () -> unflushedMemtableSize);
+
+ // BlockCache usage指标 - 统一使用缓存变量
Review Comment:
Use English.
##########
fluss-server/src/test/java/com/alibaba/fluss/server/kv/rocksdb/RocksDBMetricsManagerTest.java:
##########
@@ -0,0 +1,423 @@
+/*
+ * 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 com.alibaba.fluss.server.kv.rocksdb;
+
+import com.alibaba.fluss.config.Configuration;
+import com.alibaba.fluss.metadata.TableBucket;
+import com.alibaba.fluss.metadata.TablePath;
+import com.alibaba.fluss.server.metrics.group.TestingMetricGroups;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.jupiter.api.io.TempDir;
+import org.mockito.MockitoAnnotations;
+import org.rocksdb.RocksDB;
+import org.rocksdb.Statistics;
+
+import java.lang.reflect.Field;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+/** Tests for {@link RocksDBMetricsManager}. */
+class RocksDBMetricsManagerTest {
+
+ @TempDir Path tempDir;
+ private RocksDBResourceContainer resourceContainer;
+ private List<RocksDBMetricsCollector> collectors;
+ private AutoCloseable mocksCloseable;
+ @RegisterExtension public RocksDBExtension rocksDBExtension = new RocksDBExtension();
+
+ @BeforeEach
+ void setUp() {
+ RocksDB.loadLibrary();
+ mocksCloseable = MockitoAnnotations.openMocks(this);
+ // Setup resource container
+ resourceContainer = new RocksDBResourceContainer(new Configuration(), tempDir.toFile());
+
+ collectors = new ArrayList<>();
+
+ // Reset singleton instance for testing
+ resetSingleton();
+ }
+
+ @AfterEach
+ void tearDown() throws Exception {
+ // Close all collectors
+ for (RocksDBMetricsCollector collector : collectors) {
+ collector.close();
+ }
+ collectors.clear();
+
+ // Shutdown manager
+ RocksDBMetricsManager.getInstance().shutdown();
+
+ // Clean up resources
+ if (resourceContainer != null) {
+ resourceContainer.close();
+ }
+ if (mocksCloseable != null) {
+ mocksCloseable.close();
+ }
+
+ // Reset singleton for next test
+ resetSingleton();
+ }
+
+ @Test
+ void testRegisterCollector() throws Exception {
+
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ assertThat(manager.getCollectorCount()).isEqualTo(0);
+
+ RocksDBMetricsCollector collector = createMockCollector("test_db", "test_table", 0);
+ manager.registerCollector(collector);
+
+ assertThat(manager.getCollectorCount()).isEqualTo(1);
+ assertThat(manager.isRunning()).isTrue();
+
+ collectors.add(collector);
+ }
+
+ @Test
+ void testRegisterMultipleCollectors() throws Exception {
+
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ for (int i = 0; i < 5; i++) {
+ RocksDBMetricsCollector collector = createMockCollector("test_db", "test_table", i);
+ manager.registerCollector(collector);
+ collectors.add(collector);
+ }
+
+ assertThat(manager.getCollectorCount()).isEqualTo(5);
+ assertThat(manager.isRunning()).isTrue();
+ }
+
+ @Test
+ void testUnregisterCollector() throws Exception {
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ RocksDBMetricsCollector collector = createMockCollector("test_db", "test_table", 0);
+ manager.registerCollector(collector);
+ collectors.add(collector);
+
+ assertThat(manager.getCollectorCount()).isEqualTo(1);
+
+ manager.unregisterCollector(collector);
+
+ assertThat(manager.getCollectorCount()).isEqualTo(0);
+ }
+
+ @Test
+ void testRegisterAfterShutdown() throws Exception {
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+ manager.shutdown();
+
+ RocksDBMetricsCollector collector = createMockCollector("test_db", "test_table", 0);
+ manager.registerCollector(collector);
+
+ // Should not be registered after shutdown
+ assertThat(manager.getCollectorCount()).isEqualTo(0);
+
+ collectors.add(collector);
+ }
+
+ @Test
+ void testMetricsCollectionStartsWithFirstCollector() throws Exception {
+
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ assertThat(manager.isRunning()).isFalse();
+
+ RocksDBMetricsCollector collector = createMockCollector("test_db", "test_table", 0);
+ manager.registerCollector(collector);
+ collectors.add(collector);
+
+ assertThat(manager.isRunning()).isTrue();
+ }
+
+ @Test
+ void testBatchProcessing() throws Exception {
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ // Create many collectors to test batch processing
+ int collectorCount = 250; // More than batch size (100)
+ List<RocksDBMetricsCollector> spyCollectors = new ArrayList<>();
+
+ for (int i = 0; i < collectorCount; i++) {
+ RocksDBMetricsCollector collector = createMockCollector("test_db", "test_table", i);
+ RocksDBMetricsCollector spyCollector = spy(collector);
+ manager.registerCollector(spyCollector);
+ spyCollectors.add(spyCollector);
+ collectors.add(collector);
+ }
+
+ // Wait for metrics collection to occur
+ Thread.sleep(6000); // Wait for at least one collection cycle
+
+ // Verify that updateMetrics was called on all collectors
+ for (RocksDBMetricsCollector collector : spyCollectors) {
+ verify(collector, times(1)).updateMetrics();
+ }
+ }
+
+ @Test
+ void testExceptionHandlingInMetricsCollection() throws Exception {
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ // Create a collector that throws exception
+ RocksDBMetricsCollector faultyCollector = createMockCollector("test_db", "test_table", 0);
+ doThrow(new RuntimeException("Test exception")).when(faultyCollector).updateMetrics();
+
+ // Create a normal collector
+ RocksDBMetricsCollector normalCollector = createMockCollector("test_db", "test_table", 1);
+ RocksDBMetricsCollector spyNormalCollector = spy(normalCollector);
+
+ manager.registerCollector(faultyCollector);
+ manager.registerCollector(spyNormalCollector);
+
+ collectors.add(faultyCollector);
+ collectors.add(normalCollector);
+
+ // Wait for metrics collection
+ try {
+ Thread.sleep(6000);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+
+ // Normal collector should still be called despite exception in faulty collector
+ verify(spyNormalCollector, times(1)).updateMetrics();
+ }
+
+ @Test
+ void testShutdown() {
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ // Register some collectors
+ for (int i = 0; i < 3; i++) {
+ RocksDBMetricsCollector collector = createMockCollector("test_db", "test_table", i);
+ manager.registerCollector(collector);
+ collectors.add(collector);
+ }
+
+ assertThat(manager.isRunning()).isTrue();
+ assertThat(manager.getCollectorCount()).isEqualTo(3);
+
+ manager.shutdown();
+
+ assertThat(manager.isRunning()).isFalse();
+ assertThat(manager.isShutdown()).isTrue();
+ assertThat(manager.getCollectorCount()).isEqualTo(0);
+ }
+
+ @Test
+ void testDoubleShutdown() {
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ manager.shutdown();
+ assertThat(manager.isShutdown()).isTrue();
+
+ // Second shutdown should not cause issues
+ manager.shutdown();
+ assertThat(manager.isShutdown()).isTrue();
+ }
+
+ @Test
+ void testConcurrentRegistrationAndUnregistration() throws Exception {
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ int threadCount = 10;
+ CountDownLatch latch = new CountDownLatch(threadCount);
+ ExecutorService executor = Executors.newFixedThreadPool(threadCount);
+ AtomicInteger collectorId = new AtomicInteger(0);
+
+ for (int i = 0; i < threadCount; i++) {
+ executor.submit(
+ () -> {
+ try {
+ for (int j = 0; j < 10; j++) {
+ RocksDBMetricsCollector collector =
+ createMockCollector(
+ "test_db",
+ "test_table",
+ collectorId.getAndIncrement());
+ manager.registerCollector(collector);
+ collectors.add(collector);
+
+ // Immediately unregister some collectors
+ if (j % 2 == 0) {
+ manager.unregisterCollector(collector);
+ }
+ }
+ } finally {
+ latch.countDown();
+ }
+ });
+ }
+
+ latch.await(10, TimeUnit.SECONDS);
+ executor.shutdown();
+
+ // Should handle concurrent access gracefully
+ assertThat(manager.getCollectorCount()).isGreaterThanOrEqualTo(0);
+ }
+
+ @Test
+ void testEmptyCollectorsDoesNotCauseIssues() throws Exception {
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ // Start with no collectors
+ assertThat(manager.getCollectorCount()).isEqualTo(0);
+
+ // Wait a bit to ensure no exceptions are thrown
+ Thread.sleep(1000);
+
+ // Should still be not running
+ assertThat(manager.isRunning()).isFalse();
+ }
+
+ @Test
+ void testCreateReporterKey() {
+ RocksDBMetricsCollector collector1 = createMockCollector("db1", "table1", 0);
+ RocksDBMetricsCollector collector2 = createMockCollector("db1", "table1", 1);
+ RocksDBMetricsCollector collector3 = createMockCollector("db2", "table1", 0);
+
+ // Different buckets should have different keys
+ assertThat(collector1.getTableBucket().getBucket())
+ .isNotEqualTo(collector2.getTableBucket().getBucket());
+
+ // Different databases should have different keys
+ assertThat(collector1.getTablePath().getDatabaseName())
+ .isNotEqualTo(collector3.getTablePath().getDatabaseName());
+
+ collectors.add(collector1);
+ collectors.add(collector2);
+ collectors.add(collector3);
+ }
+
+ @Test
+ void testPartitionedTableCollector() {
+ RocksDBMetricsCollector collector =
+ createMockCollectorWithPartition("test_db", "test_table", 0, "partition1");
+
+ assertThat(collector.getPartitionName()).isEqualTo("partition1");
Review Comment:
I don't see it asserts anything. If we want to assert the reporter key is different, I think we need to create another collector with different partition name and assert `manager.getCollectorCount() = 2` here?
##########
fluss-server/src/test/java/com/alibaba/fluss/server/kv/rocksdb/RocksDBResourceContainerTest.java:
##########
@@ -187,8 +193,237 @@ void testConfigurationOptionsFromConfig() throws Exception {
(BlockBasedTableConfig) columnOptions.tableFormatConfig();
assertThat(tableConfig.blockSize()).isEqualTo(4 * SizeUnit.KB);
assertThat(tableConfig.metadataBlockSize()).isEqualTo(8 * SizeUnit.KB);
- assertThat(tableConfig.blockCacheSize()).isEqualTo(512 * SizeUnit.MB);
assertThat(tableConfig.filterPolicy() instanceof BloomFilter).isTrue();
}
}
+
+ @Test
+ void testBlockCacheCreation() throws Exception {
+ Configuration config = new Configuration();
+ config.setString(ConfigOptions.KV_BLOCK_CACHE_SIZE.key(), "16mb");
+
+ try (RocksDBResourceContainer container = new RocksDBResourceContainer(config, null)) {
+ Cache blockCache = container.getBlockCache();
+ assertThat(blockCache).isNotNull();
+ }
+ }
+
+ @Test
+ void testBlockCacheNotCreatedWhenSizeIsZero() throws Exception {
+ Configuration config = new Configuration();
+ config.setString(ConfigOptions.KV_BLOCK_CACHE_SIZE.key(), "0mb");
+
+ try (RocksDBResourceContainer container = new RocksDBResourceContainer(config, null)) {
+ Cache blockCache = container.getBlockCache();
+ assertThat(blockCache).isNull();
+ }
+ }
+
+ @Test
+ void testStatisticsEnabledByDefault() throws Exception {
+ Configuration config = new Configuration();
+ // Statistics should be enabled by default
+ try (RocksDBResourceContainer container = new RocksDBResourceContainer(config, null)) {
+ DBOptions dbOptions = container.getDbOptions();
+ assertThat(dbOptions.statistics()).isNotNull();
+ assertThat(container.getStatistics()).isNotNull();
+ }
+ }
+
+ @Test
+ void testStatisticsDisabled() throws Exception {
+ Configuration config = new Configuration();
+ config.setBoolean(ConfigOptions.KV_STATISTICS_ENABLED.key(), false);
+
+ try (RocksDBResourceContainer container = new RocksDBResourceContainer(config, null)) {
+ DBOptions dbOptions = container.getDbOptions();
+ // Statistics should be null when disabled
+ assertThat(dbOptions.statistics()).isNull();
+ assertThat(container.getStatistics()).isNull();
+ }
+ }
+
+ @Test
+ void testStatisticsExplicitlyEnabled() throws Exception {
+ Configuration config = new Configuration();
+ config.setBoolean(ConfigOptions.KV_STATISTICS_ENABLED.key(), true);
+
+ try (RocksDBResourceContainer container = new RocksDBResourceContainer(config, null)) {
+ DBOptions dbOptions = container.getDbOptions();
+ assertThat(dbOptions.statistics()).isNotNull();
+ assertThat(container.getStatistics()).isNotNull();
+ }
+ }
+
+ @Test
+ void testStatisticsClosedWithContainer() throws Exception {
+ Configuration config = new Configuration();
+ config.setBoolean(ConfigOptions.KV_STATISTICS_ENABLED.key(), true);
+
+ Statistics statistics;
+ try (RocksDBResourceContainer container = new RocksDBResourceContainer(config, null)) {
+ statistics = container.getStatistics();
+ assertThat(statistics).isNotNull();
+ assertThat(statistics.isOwningHandle()).isTrue();
+ }
+
+ // Statistics should be closed when container is closed
+ assertThat(statistics.isOwningHandle()).isFalse();
+ }
+
+ @Test
+ void testMultipleStatisticsObjectsClosedProperly() throws Exception {
+ Configuration config = new Configuration();
+ config.setBoolean(ConfigOptions.KV_STATISTICS_ENABLED.key(), true);
+
+ List<Statistics> statisticsList = new ArrayList<>();
+
+ // Create multiple containers and collect statistics
+ for (int i = 0; i < 5; i++) {
+ RocksDBResourceContainer container = new RocksDBResourceContainer(config, null);
+ Statistics stats = container.getStatistics();
+ if (stats != null) {
+ statisticsList.add(stats);
+ }
+ container.close();
+ }
+
+ // All statistics should be closed
+ for (Statistics stats : statisticsList) {
+ assertThat(stats.isOwningHandle()).isFalse();
+ }
+ }
+
+ @Test
+ void testStatisticsWithNullInstancePath() throws Exception {
Review Comment:
I don't see any values of this test, what's the use case with a null instance path?
##########
fluss-server/src/main/java/com/alibaba/fluss/server/kv/rocksdb/RocksDBMetricsCollector.java:
##########
@@ -0,0 +1,527 @@
+/*
+ * 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 com.alibaba.fluss.server.kv.rocksdb;
+
+import com.alibaba.fluss.metadata.TableBucket;
+import com.alibaba.fluss.metadata.TablePath;
+import com.alibaba.fluss.metrics.Gauge;
+import com.alibaba.fluss.metrics.MetricNames;
+import com.alibaba.fluss.metrics.groups.MetricGroup;
+import com.alibaba.fluss.server.metrics.group.BucketMetricGroup;
+
+import org.rocksdb.Cache;
+import org.rocksdb.HistogramData;
+import org.rocksdb.HistogramType;
+import org.rocksdb.MemoryUsageType;
+import org.rocksdb.MemoryUtil;
+import org.rocksdb.RocksDB;
+import org.rocksdb.RocksDBException;
+import org.rocksdb.Statistics;
+import org.rocksdb.TickerType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.io.Closeable;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * RocksDB metrics collector that collects and reports RocksDB statistics including cache,
+ * compaction, memory, and I/O metrics with table bucket and server information tags.
+ */
+public class RocksDBMetricsCollector implements Closeable {
+
+ private static final Logger LOG = LoggerFactory.getLogger(RocksDBMetricsCollector.class);
+
+ private final RocksDB rocksDB;
+ private final Statistics statistics;
+ private final MetricGroup bucketMetricGroup;
+ private final TableBucket tableBucket;
+ private final TablePath tablePath;
+ private final String partitionName;
+ private final RocksDBResourceContainer resourceContainer;
+
+ private volatile boolean registered = false;
+
+ // Cached values for efficient metric access
+ private volatile long blockCacheHitCount = 0;
+ private volatile long blockCacheMissCount = 0;
+ private volatile long indexBlockCacheHitCount = 0;
+ private volatile long indexBlockCacheMissCount = 0;
+ private volatile long filterBlockCacheHitCount = 0;
+ private volatile long filterBlockCacheMissCount = 0;
+ private volatile long dataBlockCacheHitCount = 0;
+ private volatile long dataBlockCacheMissCount = 0;
+ private volatile long compactionCount = 0;
+ private volatile long compactionBytesRead = 0;
+ private volatile long compactionBytesWritten = 0;
+ private volatile long compactionCpuTimeMicros = 0;
+ private volatile long flushCount = 0;
+ private volatile long flushBytesWritten = 0;
+ private volatile long bytesRead = 0;
+ private volatile long bytesWritten = 0;
+ private volatile long numberDbNext = 0;
+ private volatile long numberDbPrev = 0;
+ private volatile long numberDbSeek = 0;
+ private volatile long numberDbSeekFound = 0;
+ private volatile long numberKeysRead = 0;
+ private volatile long numberKeysWritten = 0;
+ private volatile long numberKeysUpdated = 0;
+ private volatile long numLiveVersions = 0;
+ private volatile long numImmutableMemtables = 0;
+ private volatile long activeMemtableSize = 0;
+ private volatile long unflushedMemtableSize = 0;
+ private volatile long totalSstFilesSize = 0;
+ private volatile long liveSstFilesSize = 0;
+ private volatile long sizeAllMemtables = 0;
+ private volatile long blockCacheAddCount = 0;
+ private volatile long memtableMemoryUsage = 0;
+ private volatile long blockCacheMemoryUsage = 0;
+ private volatile long tableReadersMemoryUsage = 0;
+ private volatile long totalMemoryUsage = 0;
+ private volatile long blockCacheUsage = 0;
+ private volatile long blockCachePinnedUsage = 0;
+ private volatile long stallTimeMicros = 0;
+
+ // Level SST files metrics
+ private volatile long numFilesAtLevel0 = 0;
+ private volatile long numFilesAtLevel1 = 0;
+ private volatile long numFilesAtLevel2 = 0;
+ private volatile long numFilesAtLevel3 = 0;
+ private volatile long numFilesAtLevel4 = 0;
+ private volatile long numFilesAtLevel5 = 0;
+ private volatile long numFilesAtLevel6 = 0;
+
+ // Compression and compaction pressure metrics
+ private volatile long compactionPending = 0;
+ private volatile long flushPending = 0;
+ private volatile long numRunningCompactions = 0;
+ private volatile long numRunningFlushes = 0;
+
+ // Write amplification metrics
+ private volatile long writeAmplification = 0;
+ private volatile long readAmplification = 0;
+
+ private volatile long backgroundErrors = 0;
+
+ // Histogram metrics
+ private volatile long dbGetLatencyMicros = 0;
+ private volatile long dbWriteLatencyMicros = 0;
+ private volatile long compactionTimeMicros = 0;
+ private volatile long compressionTimeNanos = 0;
+ private volatile long decompressionTimeNanos = 0;
+
+ public RocksDBMetricsCollector(
+ RocksDB rocksDB,
+ Statistics statistics,
+ BucketMetricGroup bucketMetricGroup,
+ TableBucket tableBucket,
+ TablePath tablePath,
+ @Nullable String partitionName,
+ RocksDBResourceContainer resourceContainer) {
+ this.rocksDB = rocksDB;
+ this.statistics = statistics;
+ this.tableBucket = tableBucket;
+ this.tablePath = tablePath;
+ this.partitionName = partitionName;
+ this.resourceContainer = resourceContainer;
+
+ // Create RocksDB metrics group under the bucket metrics
+ this.bucketMetricGroup = bucketMetricGroup;
+
+ // Register all metrics
+ registerMetrics();
+
+ // Register this reporter with the global collector
+ RocksDBMetricsManager.getInstance().registerCollector(this);
+ this.registered = true;
+
+ LOG.info(
+ "RocksDB metrics collector started for table {} bucket {}",
+ tablePath,
+ tableBucket.getBucket());
+ }
+
+ private void registerMetrics() {
+ // Cache metrics - 统一使用缓存变量
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_BLOCK_CACHE_HIT_COUNT, (Gauge<Long>) () -> blockCacheHitCount);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_BLOCK_CACHE_MISS_COUNT,
+ (Gauge<Long>) () -> blockCacheMissCount);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_INDEX_BLOCK_CACHE_HIT_COUNT,
+ (Gauge<Long>) () -> indexBlockCacheHitCount);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_INDEX_BLOCK_CACHE_MISS_COUNT,
+ (Gauge<Long>) () -> indexBlockCacheMissCount);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_FILTER_BLOCK_CACHE_HIT_COUNT,
+ (Gauge<Long>) () -> filterBlockCacheHitCount);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_FILTER_BLOCK_CACHE_MISS_COUNT,
+ (Gauge<Long>) () -> filterBlockCacheMissCount);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_DATA_BLOCK_CACHE_HIT_COUNT,
+ (Gauge<Long>) () -> dataBlockCacheHitCount);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_DATA_BLOCK_CACHE_MISS_COUNT,
+ (Gauge<Long>) () -> dataBlockCacheMissCount);
+
+ // Compaction metrics
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_COMPACTION_COUNT, (Gauge<Long>) () -> compactionCount);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_COMPACTION_BYTES_READ, (Gauge<Long>) () -> compactionBytesRead);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_COMPACTION_BYTES_WRITTEN,
+ (Gauge<Long>) () -> compactionBytesWritten);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_COMPACTION_CPU_TIME_MICROS,
+ (Gauge<Long>) () -> compactionCpuTimeMicros);
+ bucketMetricGroup.gauge(MetricNames.ROCKSDB_FLUSH_COUNT, (Gauge<Long>) () -> flushCount);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_FLUSH_BYTES_WRITTEN, (Gauge<Long>) () -> flushBytesWritten);
+
+ // Memory metrics - 统一使用缓存变量
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_MEMTABLE_MEMORY_USAGE, (Gauge<Long>) () -> memtableMemoryUsage);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_BLOCK_CACHE_MEMORY_USAGE,
+ (Gauge<Long>) () -> blockCacheMemoryUsage);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_TABLE_READERS_MEMORY_USAGE,
+ (Gauge<Long>) () -> tableReadersMemoryUsage);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_TOTAL_MEMORY_USAGE, (Gauge<Long>) () -> totalMemoryUsage);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_ACTIVE_MEMTABLE_SIZE, (Gauge<Long>) () -> activeMemtableSize);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_UNFLUSHED_MEMTABLE_SIZE,
+ (Gauge<Long>) () -> unflushedMemtableSize);
+
+ // BlockCache usage指标 - 统一使用缓存变量
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_BLOCK_CACHE_USAGE, (Gauge<Long>) () -> blockCacheUsage);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_BLOCK_CACHE_PINNED_USAGE,
+ (Gauge<Long>) () -> blockCachePinnedUsage);
+
+ // I/O metrics
+ bucketMetricGroup.gauge(MetricNames.ROCKSDB_BYTES_READ, (Gauge<Long>) () -> bytesRead);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_BYTES_WRITTEN, (Gauge<Long>) () -> bytesWritten);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_NUMBER_DB_NEXT, (Gauge<Long>) () -> numberDbNext);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_NUMBER_DB_PREV, (Gauge<Long>) () -> numberDbPrev);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_NUMBER_DB_SEEK, (Gauge<Long>) () -> numberDbSeek);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_NUMBER_DB_SEEK_FOUND, (Gauge<Long>) () -> numberDbSeekFound);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_NUMBER_KEYS_READ, (Gauge<Long>) () -> numberKeysRead);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_NUMBER_KEYS_WRITTEN, (Gauge<Long>) () -> numberKeysWritten);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_NUMBER_KEYS_UPDATED, (Gauge<Long>) () -> numberKeysUpdated);
+
+ // SST files metrics
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_NUM_LIVE_VERSIONS, (Gauge<Long>) () -> numLiveVersions);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_NUM_IMMUTABLE_MEMTABLES,
+ (Gauge<Long>) () -> numImmutableMemtables);
+
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_TOTAL_SST_FILES_SIZE, (Gauge<Long>) () -> totalSstFilesSize);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_LIVE_SST_FILES_SIZE, (Gauge<Long>) () -> liveSstFilesSize);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_SIZE_ALL_MEMTABLES, (Gauge<Long>) () -> sizeAllMemtables);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_BLOCK_CACHE_ADD_COUNT, (Gauge<Long>) () -> blockCacheAddCount);
+
+ // Level SST files metrics
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_NUM_FILES_AT_LEVEL_0, (Gauge<Long>) () -> numFilesAtLevel0);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_NUM_FILES_AT_LEVEL_1, (Gauge<Long>) () -> numFilesAtLevel1);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_NUM_FILES_AT_LEVEL_2, (Gauge<Long>) () -> numFilesAtLevel2);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_NUM_FILES_AT_LEVEL_3, (Gauge<Long>) () -> numFilesAtLevel3);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_NUM_FILES_AT_LEVEL_4, (Gauge<Long>) () -> numFilesAtLevel4);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_NUM_FILES_AT_LEVEL_5, (Gauge<Long>) () -> numFilesAtLevel5);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_NUM_FILES_AT_LEVEL_6, (Gauge<Long>) () -> numFilesAtLevel6);
+
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_WRITE_STALL_MICROS, (Gauge<Long>) () -> stallTimeMicros);
+
+ // Compression and compaction pressure metrics
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_COMPACTION_PENDING, (Gauge<Long>) () -> compactionPending);
+
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_FLUSH_PENDING, (Gauge<Long>) () -> flushPending);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_NUM_RUNNING_COMPACTIONS,
+ (Gauge<Long>) () -> numRunningCompactions);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_NUM_RUNNING_FLUSHES, (Gauge<Long>) () -> numRunningFlushes);
+
+ // Background job metrics
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_BACKGROUND_ERRORS, (Gauge<Long>) () -> backgroundErrors);
+
+ // Histogram metrics
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_DB_GET_LATENCY_MICROS, (Gauge<Long>) () -> dbGetLatencyMicros);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_DB_WRITE_LATENCY_MICROS,
+ (Gauge<Long>) () -> dbWriteLatencyMicros);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_COMPACTION_TIME_MICROS,
+ (Gauge<Long>) () -> compactionTimeMicros);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_COMPRESSION_TIME_NANOS,
+ (Gauge<Long>) () -> compressionTimeNanos);
+ bucketMetricGroup.gauge(
+ MetricNames.ROCKSDB_DECOMPRESSION_TIME_NANOS,
+ (Gauge<Long>) () -> decompressionTimeNanos);
+ }
+
+ public void updateMetrics() {
+ try {
+ // Update cache metrics
+ blockCacheHitCount = getTickerValue(TickerType.BLOCK_CACHE_HIT);
+ blockCacheMissCount = getTickerValue(TickerType.BLOCK_CACHE_MISS);
+ indexBlockCacheHitCount = getTickerValue(TickerType.BLOCK_CACHE_INDEX_HIT);
+ indexBlockCacheMissCount = getTickerValue(TickerType.BLOCK_CACHE_INDEX_MISS);
+ filterBlockCacheHitCount = getTickerValue(TickerType.BLOCK_CACHE_FILTER_HIT);
+ filterBlockCacheMissCount = getTickerValue(TickerType.BLOCK_CACHE_FILTER_MISS);
+ dataBlockCacheHitCount = getTickerValue(TickerType.BLOCK_CACHE_DATA_HIT);
+ dataBlockCacheMissCount = getTickerValue(TickerType.BLOCK_CACHE_DATA_MISS);
+
+ // Update compaction metrics
+ compactionCount =
+ getTickerValue(TickerType.COMPACT_READ_BYTES) > 0
+ ? 1
+ : 0; // Simplified since exact count may not be available
+ compactionBytesRead = getTickerValue(TickerType.COMPACT_READ_BYTES);
+ compactionBytesWritten = getTickerValue(TickerType.COMPACT_WRITE_BYTES);
+ compactionCpuTimeMicros = getHistogramValue(HistogramType.COMPACTION_TIME);
+ flushCount = getTickerValue(TickerType.FLUSH_WRITE_BYTES) > 0 ? 1 : 0;
Review Comment:
ditto. Can `flushBytesWritten` be satisfied?
##########
fluss-server/src/test/java/com/alibaba/fluss/server/kv/rocksdb/RocksDBResourceContainerTest.java:
##########
@@ -187,8 +193,237 @@ void testConfigurationOptionsFromConfig() throws Exception {
(BlockBasedTableConfig) columnOptions.tableFormatConfig();
assertThat(tableConfig.blockSize()).isEqualTo(4 * SizeUnit.KB);
assertThat(tableConfig.metadataBlockSize()).isEqualTo(8 * SizeUnit.KB);
- assertThat(tableConfig.blockCacheSize()).isEqualTo(512 * SizeUnit.MB);
assertThat(tableConfig.filterPolicy() instanceof BloomFilter).isTrue();
}
}
+
+ @Test
+ void testBlockCacheCreation() throws Exception {
+ Configuration config = new Configuration();
+ config.setString(ConfigOptions.KV_BLOCK_CACHE_SIZE.key(), "16mb");
+
+ try (RocksDBResourceContainer container = new RocksDBResourceContainer(config, null)) {
+ Cache blockCache = container.getBlockCache();
+ assertThat(blockCache).isNotNull();
+ }
+ }
+
+ @Test
+ void testBlockCacheNotCreatedWhenSizeIsZero() throws Exception {
+ Configuration config = new Configuration();
+ config.setString(ConfigOptions.KV_BLOCK_CACHE_SIZE.key(), "0mb");
+
+ try (RocksDBResourceContainer container = new RocksDBResourceContainer(config, null)) {
+ Cache blockCache = container.getBlockCache();
+ assertThat(blockCache).isNull();
+ }
+ }
+
+ @Test
+ void testStatisticsEnabledByDefault() throws Exception {
+ Configuration config = new Configuration();
+ // Statistics should be enabled by default
+ try (RocksDBResourceContainer container = new RocksDBResourceContainer(config, null)) {
+ DBOptions dbOptions = container.getDbOptions();
+ assertThat(dbOptions.statistics()).isNotNull();
+ assertThat(container.getStatistics()).isNotNull();
+ }
+ }
+
+ @Test
+ void testStatisticsDisabled() throws Exception {
+ Configuration config = new Configuration();
+ config.setBoolean(ConfigOptions.KV_STATISTICS_ENABLED.key(), false);
+
+ try (RocksDBResourceContainer container = new RocksDBResourceContainer(config, null)) {
+ DBOptions dbOptions = container.getDbOptions();
+ // Statistics should be null when disabled
+ assertThat(dbOptions.statistics()).isNull();
+ assertThat(container.getStatistics()).isNull();
+ }
+ }
+
+ @Test
+ void testStatisticsExplicitlyEnabled() throws Exception {
+ Configuration config = new Configuration();
+ config.setBoolean(ConfigOptions.KV_STATISTICS_ENABLED.key(), true);
+
+ try (RocksDBResourceContainer container = new RocksDBResourceContainer(config, null)) {
+ DBOptions dbOptions = container.getDbOptions();
+ assertThat(dbOptions.statistics()).isNotNull();
+ assertThat(container.getStatistics()).isNotNull();
+ }
+ }
+
+ @Test
+ void testStatisticsClosedWithContainer() throws Exception {
+ Configuration config = new Configuration();
+ config.setBoolean(ConfigOptions.KV_STATISTICS_ENABLED.key(), true);
+
+ Statistics statistics;
+ try (RocksDBResourceContainer container = new RocksDBResourceContainer(config, null)) {
+ statistics = container.getStatistics();
+ assertThat(statistics).isNotNull();
+ assertThat(statistics.isOwningHandle()).isTrue();
+ }
+
+ // Statistics should be closed when container is closed
+ assertThat(statistics.isOwningHandle()).isFalse();
+ }
+
+ @Test
+ void testMultipleStatisticsObjectsClosedProperly() throws Exception {
+ Configuration config = new Configuration();
+ config.setBoolean(ConfigOptions.KV_STATISTICS_ENABLED.key(), true);
+
+ List<Statistics> statisticsList = new ArrayList<>();
+
+ // Create multiple containers and collect statistics
+ for (int i = 0; i < 5; i++) {
+ RocksDBResourceContainer container = new RocksDBResourceContainer(config, null);
+ Statistics stats = container.getStatistics();
+ if (stats != null) {
+ statisticsList.add(stats);
+ }
+ container.close();
+ }
+
+ // All statistics should be closed
+ for (Statistics stats : statisticsList) {
+ assertThat(stats.isOwningHandle()).isFalse();
+ }
+ }
+
+ @Test
+ void testStatisticsWithNullInstancePath() throws Exception {
+ Configuration config = new Configuration();
+ config.setBoolean(ConfigOptions.KV_STATISTICS_ENABLED.key(), true);
+
+ try (RocksDBResourceContainer container = new RocksDBResourceContainer(config, null)) {
+ Statistics statistics = container.getStatistics();
+ assertThat(statistics).isNotNull();
+ assertThat(statistics.isOwningHandle()).isTrue();
+ }
+ }
+
+ @Test
+ void testStatisticsWithValidInstancePath(@TempDir Path tempFolder) throws Exception {
+ Configuration config = new Configuration();
+ config.setBoolean(ConfigOptions.KV_STATISTICS_ENABLED.key(), true);
+
+ File instancePath = tempFolder.resolve("rocksdb-instance").toFile();
+ try (RocksDBResourceContainer container =
+ new RocksDBResourceContainer(config, instancePath)) {
+ Statistics statistics = container.getStatistics();
+ assertThat(statistics).isNotNull();
+ assertThat(statistics.isOwningHandle()).isTrue();
+ }
+ }
+
+ @Test
+ void testStatisticsConfigurationConsistency() throws Exception {
+ Configuration config = new Configuration();
+
+ // Test with statistics disabled
+ config.setBoolean(ConfigOptions.KV_STATISTICS_ENABLED.key(), false);
+ try (RocksDBResourceContainer container = new RocksDBResourceContainer(config, null)) {
+ assertThat(container.getStatistics()).isNull();
+ }
+
+ // Test with statistics enabled
+ config.setBoolean(ConfigOptions.KV_STATISTICS_ENABLED.key(), true);
+ try (RocksDBResourceContainer container = new RocksDBResourceContainer(config, null)) {
+ assertThat(container.getStatistics()).isNotNull();
+ }
+ }
+
+ @Test
+ void testStatisticsMemoryManagement() throws Exception {
+ Configuration config = new Configuration();
+ config.setBoolean(ConfigOptions.KV_STATISTICS_ENABLED.key(), true);
+
+ // Create and close many containers to test memory management
Review Comment:
I don't see any tests or assersions for memory memory.
##########
fluss-server/src/test/java/com/alibaba/fluss/server/kv/rocksdb/RocksDBMetricsManagerTest.java:
##########
@@ -0,0 +1,423 @@
+/*
+ * 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 com.alibaba.fluss.server.kv.rocksdb;
+
+import com.alibaba.fluss.config.Configuration;
+import com.alibaba.fluss.metadata.TableBucket;
+import com.alibaba.fluss.metadata.TablePath;
+import com.alibaba.fluss.server.metrics.group.TestingMetricGroups;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.jupiter.api.io.TempDir;
+import org.mockito.MockitoAnnotations;
+import org.rocksdb.RocksDB;
+import org.rocksdb.Statistics;
+
+import java.lang.reflect.Field;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+/** Tests for {@link RocksDBMetricsManager}. */
+class RocksDBMetricsManagerTest {
+
+ @TempDir Path tempDir;
+ private RocksDBResourceContainer resourceContainer;
+ private List<RocksDBMetricsCollector> collectors;
+ private AutoCloseable mocksCloseable;
+ @RegisterExtension public RocksDBExtension rocksDBExtension = new RocksDBExtension();
+
+ @BeforeEach
+ void setUp() {
+ RocksDB.loadLibrary();
+ mocksCloseable = MockitoAnnotations.openMocks(this);
+ // Setup resource container
+ resourceContainer = new RocksDBResourceContainer(new Configuration(), tempDir.toFile());
+
+ collectors = new ArrayList<>();
+
+ // Reset singleton instance for testing
+ resetSingleton();
+ }
+
+ @AfterEach
+ void tearDown() throws Exception {
+ // Close all collectors
+ for (RocksDBMetricsCollector collector : collectors) {
+ collector.close();
+ }
+ collectors.clear();
+
+ // Shutdown manager
+ RocksDBMetricsManager.getInstance().shutdown();
+
+ // Clean up resources
+ if (resourceContainer != null) {
+ resourceContainer.close();
+ }
+ if (mocksCloseable != null) {
+ mocksCloseable.close();
+ }
+
+ // Reset singleton for next test
+ resetSingleton();
+ }
+
+ @Test
+ void testRegisterCollector() throws Exception {
+
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ assertThat(manager.getCollectorCount()).isEqualTo(0);
+
+ RocksDBMetricsCollector collector = createMockCollector("test_db", "test_table", 0);
+ manager.registerCollector(collector);
+
+ assertThat(manager.getCollectorCount()).isEqualTo(1);
+ assertThat(manager.isRunning()).isTrue();
+
+ collectors.add(collector);
+ }
+
+ @Test
+ void testRegisterMultipleCollectors() throws Exception {
+
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ for (int i = 0; i < 5; i++) {
+ RocksDBMetricsCollector collector = createMockCollector("test_db", "test_table", i);
+ manager.registerCollector(collector);
+ collectors.add(collector);
+ }
+
+ assertThat(manager.getCollectorCount()).isEqualTo(5);
+ assertThat(manager.isRunning()).isTrue();
+ }
+
+ @Test
+ void testUnregisterCollector() throws Exception {
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ RocksDBMetricsCollector collector = createMockCollector("test_db", "test_table", 0);
+ manager.registerCollector(collector);
+ collectors.add(collector);
+
+ assertThat(manager.getCollectorCount()).isEqualTo(1);
+
+ manager.unregisterCollector(collector);
+
+ assertThat(manager.getCollectorCount()).isEqualTo(0);
+ }
+
+ @Test
+ void testRegisterAfterShutdown() throws Exception {
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+ manager.shutdown();
+
+ RocksDBMetricsCollector collector = createMockCollector("test_db", "test_table", 0);
+ manager.registerCollector(collector);
+
+ // Should not be registered after shutdown
+ assertThat(manager.getCollectorCount()).isEqualTo(0);
+
+ collectors.add(collector);
+ }
+
+ @Test
+ void testMetricsCollectionStartsWithFirstCollector() throws Exception {
+
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ assertThat(manager.isRunning()).isFalse();
+
+ RocksDBMetricsCollector collector = createMockCollector("test_db", "test_table", 0);
+ manager.registerCollector(collector);
+ collectors.add(collector);
+
+ assertThat(manager.isRunning()).isTrue();
+ }
+
+ @Test
+ void testBatchProcessing() throws Exception {
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ // Create many collectors to test batch processing
+ int collectorCount = 250; // More than batch size (100)
+ List<RocksDBMetricsCollector> spyCollectors = new ArrayList<>();
+
+ for (int i = 0; i < collectorCount; i++) {
+ RocksDBMetricsCollector collector = createMockCollector("test_db", "test_table", i);
+ RocksDBMetricsCollector spyCollector = spy(collector);
+ manager.registerCollector(spyCollector);
+ spyCollectors.add(spyCollector);
+ collectors.add(collector);
+ }
+
+ // Wait for metrics collection to occur
+ Thread.sleep(6000); // Wait for at least one collection cycle
+
+ // Verify that updateMetrics was called on all collectors
+ for (RocksDBMetricsCollector collector : spyCollectors) {
+ verify(collector, times(1)).updateMetrics();
+ }
+ }
+
+ @Test
+ void testExceptionHandlingInMetricsCollection() throws Exception {
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ // Create a collector that throws exception
+ RocksDBMetricsCollector faultyCollector = createMockCollector("test_db", "test_table", 0);
+ doThrow(new RuntimeException("Test exception")).when(faultyCollector).updateMetrics();
+
+ // Create a normal collector
+ RocksDBMetricsCollector normalCollector = createMockCollector("test_db", "test_table", 1);
+ RocksDBMetricsCollector spyNormalCollector = spy(normalCollector);
+
+ manager.registerCollector(faultyCollector);
+ manager.registerCollector(spyNormalCollector);
+
+ collectors.add(faultyCollector);
+ collectors.add(normalCollector);
+
+ // Wait for metrics collection
+ try {
+ Thread.sleep(6000);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+
+ // Normal collector should still be called despite exception in faulty collector
+ verify(spyNormalCollector, times(1)).updateMetrics();
+ }
+
+ @Test
+ void testShutdown() {
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ // Register some collectors
+ for (int i = 0; i < 3; i++) {
+ RocksDBMetricsCollector collector = createMockCollector("test_db", "test_table", i);
+ manager.registerCollector(collector);
+ collectors.add(collector);
+ }
+
+ assertThat(manager.isRunning()).isTrue();
+ assertThat(manager.getCollectorCount()).isEqualTo(3);
+
+ manager.shutdown();
+
+ assertThat(manager.isRunning()).isFalse();
+ assertThat(manager.isShutdown()).isTrue();
+ assertThat(manager.getCollectorCount()).isEqualTo(0);
+ }
+
+ @Test
+ void testDoubleShutdown() {
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ manager.shutdown();
+ assertThat(manager.isShutdown()).isTrue();
+
+ // Second shutdown should not cause issues
+ manager.shutdown();
+ assertThat(manager.isShutdown()).isTrue();
+ }
+
+ @Test
+ void testConcurrentRegistrationAndUnregistration() throws Exception {
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ int threadCount = 10;
+ CountDownLatch latch = new CountDownLatch(threadCount);
+ ExecutorService executor = Executors.newFixedThreadPool(threadCount);
+ AtomicInteger collectorId = new AtomicInteger(0);
+
+ for (int i = 0; i < threadCount; i++) {
+ executor.submit(
+ () -> {
+ try {
+ for (int j = 0; j < 10; j++) {
+ RocksDBMetricsCollector collector =
+ createMockCollector(
+ "test_db",
+ "test_table",
+ collectorId.getAndIncrement());
+ manager.registerCollector(collector);
+ collectors.add(collector);
+
+ // Immediately unregister some collectors
+ if (j % 2 == 0) {
+ manager.unregisterCollector(collector);
+ }
+ }
+ } finally {
+ latch.countDown();
+ }
+ });
+ }
+
+ latch.await(10, TimeUnit.SECONDS);
+ executor.shutdown();
+
+ // Should handle concurrent access gracefully
+ assertThat(manager.getCollectorCount()).isGreaterThanOrEqualTo(0);
+ }
+
+ @Test
+ void testEmptyCollectorsDoesNotCauseIssues() throws Exception {
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ // Start with no collectors
+ assertThat(manager.getCollectorCount()).isEqualTo(0);
+
+ // Wait a bit to ensure no exceptions are thrown
+ Thread.sleep(1000);
+
+ // Should still be not running
+ assertThat(manager.isRunning()).isFalse();
+ }
+
+ @Test
+ void testCreateReporterKey() {
+ RocksDBMetricsCollector collector1 = createMockCollector("db1", "table1", 0);
+ RocksDBMetricsCollector collector2 = createMockCollector("db1", "table1", 1);
+ RocksDBMetricsCollector collector3 = createMockCollector("db2", "table1", 0);
+
+ // Different buckets should have different keys
+ assertThat(collector1.getTableBucket().getBucket())
+ .isNotEqualTo(collector2.getTableBucket().getBucket());
+
+ // Different databases should have different keys
+ assertThat(collector1.getTablePath().getDatabaseName())
+ .isNotEqualTo(collector3.getTablePath().getDatabaseName());
Review Comment:
I don't see it asserts anything. If we want to assert the reporter key is different, I think we need to assert `manager.getCollectorCount() = 3` here?
##########
fluss-server/src/test/java/com/alibaba/fluss/server/kv/rocksdb/RocksDBMetricsManagerTest.java:
##########
@@ -0,0 +1,423 @@
+/*
+ * 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 com.alibaba.fluss.server.kv.rocksdb;
+
+import com.alibaba.fluss.config.Configuration;
+import com.alibaba.fluss.metadata.TableBucket;
+import com.alibaba.fluss.metadata.TablePath;
+import com.alibaba.fluss.server.metrics.group.TestingMetricGroups;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.jupiter.api.io.TempDir;
+import org.mockito.MockitoAnnotations;
+import org.rocksdb.RocksDB;
+import org.rocksdb.Statistics;
+
+import java.lang.reflect.Field;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+/** Tests for {@link RocksDBMetricsManager}. */
+class RocksDBMetricsManagerTest {
+
+ @TempDir Path tempDir;
+ private RocksDBResourceContainer resourceContainer;
+ private List<RocksDBMetricsCollector> collectors;
+ private AutoCloseable mocksCloseable;
+ @RegisterExtension public RocksDBExtension rocksDBExtension = new RocksDBExtension();
+
+ @BeforeEach
+ void setUp() {
+ RocksDB.loadLibrary();
+ mocksCloseable = MockitoAnnotations.openMocks(this);
+ // Setup resource container
+ resourceContainer = new RocksDBResourceContainer(new Configuration(), tempDir.toFile());
+
+ collectors = new ArrayList<>();
+
+ // Reset singleton instance for testing
+ resetSingleton();
+ }
+
+ @AfterEach
+ void tearDown() throws Exception {
+ // Close all collectors
+ for (RocksDBMetricsCollector collector : collectors) {
+ collector.close();
+ }
+ collectors.clear();
+
+ // Shutdown manager
+ RocksDBMetricsManager.getInstance().shutdown();
+
+ // Clean up resources
+ if (resourceContainer != null) {
+ resourceContainer.close();
+ }
+ if (mocksCloseable != null) {
+ mocksCloseable.close();
+ }
+
+ // Reset singleton for next test
+ resetSingleton();
+ }
+
+ @Test
+ void testRegisterCollector() throws Exception {
+
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ assertThat(manager.getCollectorCount()).isEqualTo(0);
+
+ RocksDBMetricsCollector collector = createMockCollector("test_db", "test_table", 0);
+ manager.registerCollector(collector);
+
+ assertThat(manager.getCollectorCount()).isEqualTo(1);
+ assertThat(manager.isRunning()).isTrue();
+
+ collectors.add(collector);
+ }
+
+ @Test
+ void testRegisterMultipleCollectors() throws Exception {
+
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ for (int i = 0; i < 5; i++) {
+ RocksDBMetricsCollector collector = createMockCollector("test_db", "test_table", i);
+ manager.registerCollector(collector);
+ collectors.add(collector);
+ }
+
+ assertThat(manager.getCollectorCount()).isEqualTo(5);
+ assertThat(manager.isRunning()).isTrue();
+ }
+
+ @Test
+ void testUnregisterCollector() throws Exception {
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ RocksDBMetricsCollector collector = createMockCollector("test_db", "test_table", 0);
+ manager.registerCollector(collector);
+ collectors.add(collector);
+
+ assertThat(manager.getCollectorCount()).isEqualTo(1);
+
+ manager.unregisterCollector(collector);
+
+ assertThat(manager.getCollectorCount()).isEqualTo(0);
+ }
+
+ @Test
+ void testRegisterAfterShutdown() throws Exception {
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+ manager.shutdown();
+
+ RocksDBMetricsCollector collector = createMockCollector("test_db", "test_table", 0);
+ manager.registerCollector(collector);
+
+ // Should not be registered after shutdown
+ assertThat(manager.getCollectorCount()).isEqualTo(0);
+
+ collectors.add(collector);
+ }
+
+ @Test
+ void testMetricsCollectionStartsWithFirstCollector() throws Exception {
+
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ assertThat(manager.isRunning()).isFalse();
+
+ RocksDBMetricsCollector collector = createMockCollector("test_db", "test_table", 0);
+ manager.registerCollector(collector);
+ collectors.add(collector);
+
+ assertThat(manager.isRunning()).isTrue();
+ }
+
+ @Test
+ void testBatchProcessing() throws Exception {
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ // Create many collectors to test batch processing
+ int collectorCount = 250; // More than batch size (100)
+ List<RocksDBMetricsCollector> spyCollectors = new ArrayList<>();
+
+ for (int i = 0; i < collectorCount; i++) {
+ RocksDBMetricsCollector collector = createMockCollector("test_db", "test_table", i);
+ RocksDBMetricsCollector spyCollector = spy(collector);
+ manager.registerCollector(spyCollector);
+ spyCollectors.add(spyCollector);
+ collectors.add(collector);
+ }
+
+ // Wait for metrics collection to occur
+ Thread.sleep(6000); // Wait for at least one collection cycle
+
+ // Verify that updateMetrics was called on all collectors
+ for (RocksDBMetricsCollector collector : spyCollectors) {
+ verify(collector, times(1)).updateMetrics();
+ }
+ }
+
+ @Test
+ void testExceptionHandlingInMetricsCollection() throws Exception {
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ // Create a collector that throws exception
+ RocksDBMetricsCollector faultyCollector = createMockCollector("test_db", "test_table", 0);
+ doThrow(new RuntimeException("Test exception")).when(faultyCollector).updateMetrics();
+
+ // Create a normal collector
+ RocksDBMetricsCollector normalCollector = createMockCollector("test_db", "test_table", 1);
+ RocksDBMetricsCollector spyNormalCollector = spy(normalCollector);
+
+ manager.registerCollector(faultyCollector);
+ manager.registerCollector(spyNormalCollector);
+
+ collectors.add(faultyCollector);
+ collectors.add(normalCollector);
+
+ // Wait for metrics collection
+ try {
+ Thread.sleep(6000);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+
+ // Normal collector should still be called despite exception in faulty collector
+ verify(spyNormalCollector, times(1)).updateMetrics();
+ }
+
+ @Test
+ void testShutdown() {
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ // Register some collectors
+ for (int i = 0; i < 3; i++) {
+ RocksDBMetricsCollector collector = createMockCollector("test_db", "test_table", i);
+ manager.registerCollector(collector);
+ collectors.add(collector);
+ }
+
+ assertThat(manager.isRunning()).isTrue();
+ assertThat(manager.getCollectorCount()).isEqualTo(3);
+
+ manager.shutdown();
+
+ assertThat(manager.isRunning()).isFalse();
+ assertThat(manager.isShutdown()).isTrue();
+ assertThat(manager.getCollectorCount()).isEqualTo(0);
+ }
+
+ @Test
+ void testDoubleShutdown() {
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ manager.shutdown();
+ assertThat(manager.isShutdown()).isTrue();
+
+ // Second shutdown should not cause issues
+ manager.shutdown();
+ assertThat(manager.isShutdown()).isTrue();
+ }
+
+ @Test
+ void testConcurrentRegistrationAndUnregistration() throws Exception {
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ int threadCount = 10;
+ CountDownLatch latch = new CountDownLatch(threadCount);
+ ExecutorService executor = Executors.newFixedThreadPool(threadCount);
+ AtomicInteger collectorId = new AtomicInteger(0);
+
+ for (int i = 0; i < threadCount; i++) {
+ executor.submit(
+ () -> {
+ try {
+ for (int j = 0; j < 10; j++) {
+ RocksDBMetricsCollector collector =
+ createMockCollector(
+ "test_db",
+ "test_table",
+ collectorId.getAndIncrement());
+ manager.registerCollector(collector);
+ collectors.add(collector);
+
+ // Immediately unregister some collectors
+ if (j % 2 == 0) {
+ manager.unregisterCollector(collector);
+ }
+ }
+ } finally {
+ latch.countDown();
+ }
+ });
+ }
+
+ latch.await(10, TimeUnit.SECONDS);
+ executor.shutdown();
+
+ // Should handle concurrent access gracefully
+ assertThat(manager.getCollectorCount()).isGreaterThanOrEqualTo(0);
+ }
+
+ @Test
+ void testEmptyCollectorsDoesNotCauseIssues() throws Exception {
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ // Start with no collectors
+ assertThat(manager.getCollectorCount()).isEqualTo(0);
+
+ // Wait a bit to ensure no exceptions are thrown
+ Thread.sleep(1000);
+
+ // Should still be not running
+ assertThat(manager.isRunning()).isFalse();
+ }
+
+ @Test
+ void testCreateReporterKey() {
+ RocksDBMetricsCollector collector1 = createMockCollector("db1", "table1", 0);
+ RocksDBMetricsCollector collector2 = createMockCollector("db1", "table1", 1);
+ RocksDBMetricsCollector collector3 = createMockCollector("db2", "table1", 0);
+
+ // Different buckets should have different keys
+ assertThat(collector1.getTableBucket().getBucket())
+ .isNotEqualTo(collector2.getTableBucket().getBucket());
+
+ // Different databases should have different keys
+ assertThat(collector1.getTablePath().getDatabaseName())
+ .isNotEqualTo(collector3.getTablePath().getDatabaseName());
+
+ collectors.add(collector1);
+ collectors.add(collector2);
+ collectors.add(collector3);
+ }
+
+ @Test
+ void testPartitionedTableCollector() {
+ RocksDBMetricsCollector collector =
+ createMockCollectorWithPartition("test_db", "test_table", 0, "partition1");
+
+ assertThat(collector.getPartitionName()).isEqualTo("partition1");
+
+ collectors.add(collector);
+ }
+
+ @Test
+ void testManagerStateAfterAllCollectorsUnregistered() {
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ // Register collectors
+ List<RocksDBMetricsCollector> testCollectors = new ArrayList<>();
+ for (int i = 0; i < 3; i++) {
+ RocksDBMetricsCollector collector = createMockCollector("test_db", "test_table", i);
+ manager.registerCollector(collector);
+ testCollectors.add(collector);
+ collectors.add(collector);
+ }
+
+ assertThat(manager.isRunning()).isTrue();
+ assertThat(manager.getCollectorCount()).isEqualTo(3);
+
+ // Unregister all collectors
+ for (RocksDBMetricsCollector collector : testCollectors) {
+ manager.unregisterCollector(collector);
+ }
+
+ assertThat(manager.getCollectorCount()).isEqualTo(0);
+ // Manager should still be running (collection thread continues)
+ assertThat(manager.isRunning()).isTrue();
+ }
+
+ @Test
+ void testResourceCleanupOnShutdown() throws Exception {
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ // Register a collector to start the manager
+ RocksDBMetricsCollector collector = createMockCollector("test_db", "test_table", 0);
+ manager.registerCollector(collector);
+ collectors.add(collector);
+
+ assertThat(manager.isRunning()).isTrue();
+
+ // Get the executor service via reflection to verify it's cleaned up
+ Field executorField =
+ RocksDBMetricsManager.class.getDeclaredField("metricsCollectionExecutor");
Review Comment:
ditto. use `protected` variables instead of Java reflection.
##########
fluss-server/src/test/java/com/alibaba/fluss/server/kv/rocksdb/RocksDBResourceContainerTest.java:
##########
@@ -187,8 +193,237 @@ void testConfigurationOptionsFromConfig() throws Exception {
(BlockBasedTableConfig) columnOptions.tableFormatConfig();
assertThat(tableConfig.blockSize()).isEqualTo(4 * SizeUnit.KB);
assertThat(tableConfig.metadataBlockSize()).isEqualTo(8 * SizeUnit.KB);
- assertThat(tableConfig.blockCacheSize()).isEqualTo(512 * SizeUnit.MB);
assertThat(tableConfig.filterPolicy() instanceof BloomFilter).isTrue();
}
}
+
+ @Test
+ void testBlockCacheCreation() throws Exception {
+ Configuration config = new Configuration();
+ config.setString(ConfigOptions.KV_BLOCK_CACHE_SIZE.key(), "16mb");
+
+ try (RocksDBResourceContainer container = new RocksDBResourceContainer(config, null)) {
+ Cache blockCache = container.getBlockCache();
+ assertThat(blockCache).isNotNull();
+ }
+ }
+
+ @Test
+ void testBlockCacheNotCreatedWhenSizeIsZero() throws Exception {
+ Configuration config = new Configuration();
+ config.setString(ConfigOptions.KV_BLOCK_CACHE_SIZE.key(), "0mb");
+
+ try (RocksDBResourceContainer container = new RocksDBResourceContainer(config, null)) {
+ Cache blockCache = container.getBlockCache();
+ assertThat(blockCache).isNull();
+ }
+ }
+
+ @Test
+ void testStatisticsEnabledByDefault() throws Exception {
+ Configuration config = new Configuration();
+ // Statistics should be enabled by default
+ try (RocksDBResourceContainer container = new RocksDBResourceContainer(config, null)) {
+ DBOptions dbOptions = container.getDbOptions();
+ assertThat(dbOptions.statistics()).isNotNull();
+ assertThat(container.getStatistics()).isNotNull();
+ }
+ }
+
+ @Test
+ void testStatisticsDisabled() throws Exception {
+ Configuration config = new Configuration();
+ config.setBoolean(ConfigOptions.KV_STATISTICS_ENABLED.key(), false);
+
+ try (RocksDBResourceContainer container = new RocksDBResourceContainer(config, null)) {
+ DBOptions dbOptions = container.getDbOptions();
+ // Statistics should be null when disabled
+ assertThat(dbOptions.statistics()).isNull();
+ assertThat(container.getStatistics()).isNull();
+ }
+ }
+
+ @Test
+ void testStatisticsExplicitlyEnabled() throws Exception {
+ Configuration config = new Configuration();
+ config.setBoolean(ConfigOptions.KV_STATISTICS_ENABLED.key(), true);
+
+ try (RocksDBResourceContainer container = new RocksDBResourceContainer(config, null)) {
+ DBOptions dbOptions = container.getDbOptions();
+ assertThat(dbOptions.statistics()).isNotNull();
+ assertThat(container.getStatistics()).isNotNull();
+ }
+ }
+
+ @Test
+ void testStatisticsClosedWithContainer() throws Exception {
+ Configuration config = new Configuration();
+ config.setBoolean(ConfigOptions.KV_STATISTICS_ENABLED.key(), true);
+
+ Statistics statistics;
+ try (RocksDBResourceContainer container = new RocksDBResourceContainer(config, null)) {
+ statistics = container.getStatistics();
+ assertThat(statistics).isNotNull();
+ assertThat(statistics.isOwningHandle()).isTrue();
+ }
+
+ // Statistics should be closed when container is closed
+ assertThat(statistics.isOwningHandle()).isFalse();
+ }
+
+ @Test
+ void testMultipleStatisticsObjectsClosedProperly() throws Exception {
+ Configuration config = new Configuration();
+ config.setBoolean(ConfigOptions.KV_STATISTICS_ENABLED.key(), true);
+
+ List<Statistics> statisticsList = new ArrayList<>();
+
+ // Create multiple containers and collect statistics
+ for (int i = 0; i < 5; i++) {
+ RocksDBResourceContainer container = new RocksDBResourceContainer(config, null);
+ Statistics stats = container.getStatistics();
+ if (stats != null) {
+ statisticsList.add(stats);
+ }
+ container.close();
+ }
+
+ // All statistics should be closed
+ for (Statistics stats : statisticsList) {
+ assertThat(stats.isOwningHandle()).isFalse();
+ }
+ }
+
+ @Test
+ void testStatisticsWithNullInstancePath() throws Exception {
+ Configuration config = new Configuration();
+ config.setBoolean(ConfigOptions.KV_STATISTICS_ENABLED.key(), true);
+
+ try (RocksDBResourceContainer container = new RocksDBResourceContainer(config, null)) {
+ Statistics statistics = container.getStatistics();
+ assertThat(statistics).isNotNull();
+ assertThat(statistics.isOwningHandle()).isTrue();
+ }
+ }
+
+ @Test
+ void testStatisticsWithValidInstancePath(@TempDir Path tempFolder) throws Exception {
+ Configuration config = new Configuration();
+ config.setBoolean(ConfigOptions.KV_STATISTICS_ENABLED.key(), true);
+
+ File instancePath = tempFolder.resolve("rocksdb-instance").toFile();
+ try (RocksDBResourceContainer container =
+ new RocksDBResourceContainer(config, instancePath)) {
+ Statistics statistics = container.getStatistics();
+ assertThat(statistics).isNotNull();
+ assertThat(statistics.isOwningHandle()).isTrue();
+ }
+ }
+
+ @Test
+ void testStatisticsConfigurationConsistency() throws Exception {
Review Comment:
I think this has been tested above.
##########
fluss-server/src/main/java/com/alibaba/fluss/server/kv/rocksdb/RocksDBResourceContainer.java:
##########
@@ -74,28 +77,25 @@ public class RocksDBResourceContainer implements AutoCloseable {
/** The handles to be closed when the container is closed. */
private final ArrayList<AutoCloseable> handlesToClose;
+ @Nullable private Statistics statistics;
+ @Nullable private Cache blockCache;
Review Comment:
They are initialized in the constructor, make them `final`, so the getters of them are thread-safe.
##########
fluss-server/src/test/java/com/alibaba/fluss/server/kv/rocksdb/RocksDBResourceContainerTest.java:
##########
@@ -187,8 +193,237 @@ void testConfigurationOptionsFromConfig() throws Exception {
(BlockBasedTableConfig) columnOptions.tableFormatConfig();
assertThat(tableConfig.blockSize()).isEqualTo(4 * SizeUnit.KB);
assertThat(tableConfig.metadataBlockSize()).isEqualTo(8 * SizeUnit.KB);
- assertThat(tableConfig.blockCacheSize()).isEqualTo(512 * SizeUnit.MB);
assertThat(tableConfig.filterPolicy() instanceof BloomFilter).isTrue();
}
}
+
+ @Test
+ void testBlockCacheCreation() throws Exception {
+ Configuration config = new Configuration();
+ config.setString(ConfigOptions.KV_BLOCK_CACHE_SIZE.key(), "16mb");
+
+ try (RocksDBResourceContainer container = new RocksDBResourceContainer(config, null)) {
+ Cache blockCache = container.getBlockCache();
+ assertThat(blockCache).isNotNull();
+ }
+ }
+
+ @Test
+ void testBlockCacheNotCreatedWhenSizeIsZero() throws Exception {
+ Configuration config = new Configuration();
+ config.setString(ConfigOptions.KV_BLOCK_CACHE_SIZE.key(), "0mb");
+
+ try (RocksDBResourceContainer container = new RocksDBResourceContainer(config, null)) {
+ Cache blockCache = container.getBlockCache();
+ assertThat(blockCache).isNull();
+ }
+ }
+
+ @Test
+ void testStatisticsEnabledByDefault() throws Exception {
+ Configuration config = new Configuration();
+ // Statistics should be enabled by default
+ try (RocksDBResourceContainer container = new RocksDBResourceContainer(config, null)) {
+ DBOptions dbOptions = container.getDbOptions();
+ assertThat(dbOptions.statistics()).isNotNull();
+ assertThat(container.getStatistics()).isNotNull();
+ }
+ }
+
+ @Test
+ void testStatisticsDisabled() throws Exception {
+ Configuration config = new Configuration();
+ config.setBoolean(ConfigOptions.KV_STATISTICS_ENABLED.key(), false);
+
+ try (RocksDBResourceContainer container = new RocksDBResourceContainer(config, null)) {
+ DBOptions dbOptions = container.getDbOptions();
+ // Statistics should be null when disabled
+ assertThat(dbOptions.statistics()).isNull();
+ assertThat(container.getStatistics()).isNull();
+ }
+ }
+
+ @Test
+ void testStatisticsExplicitlyEnabled() throws Exception {
+ Configuration config = new Configuration();
+ config.setBoolean(ConfigOptions.KV_STATISTICS_ENABLED.key(), true);
+
+ try (RocksDBResourceContainer container = new RocksDBResourceContainer(config, null)) {
+ DBOptions dbOptions = container.getDbOptions();
+ assertThat(dbOptions.statistics()).isNotNull();
+ assertThat(container.getStatistics()).isNotNull();
+ }
+ }
+
+ @Test
+ void testStatisticsClosedWithContainer() throws Exception {
+ Configuration config = new Configuration();
+ config.setBoolean(ConfigOptions.KV_STATISTICS_ENABLED.key(), true);
+
+ Statistics statistics;
+ try (RocksDBResourceContainer container = new RocksDBResourceContainer(config, null)) {
+ statistics = container.getStatistics();
+ assertThat(statistics).isNotNull();
+ assertThat(statistics.isOwningHandle()).isTrue();
+ }
+
+ // Statistics should be closed when container is closed
+ assertThat(statistics.isOwningHandle()).isFalse();
+ }
+
+ @Test
+ void testMultipleStatisticsObjectsClosedProperly() throws Exception {
+ Configuration config = new Configuration();
+ config.setBoolean(ConfigOptions.KV_STATISTICS_ENABLED.key(), true);
+
+ List<Statistics> statisticsList = new ArrayList<>();
+
+ // Create multiple containers and collect statistics
+ for (int i = 0; i < 5; i++) {
+ RocksDBResourceContainer container = new RocksDBResourceContainer(config, null);
+ Statistics stats = container.getStatistics();
+ if (stats != null) {
+ statisticsList.add(stats);
+ }
+ container.close();
+ }
+
+ // All statistics should be closed
+ for (Statistics stats : statisticsList) {
+ assertThat(stats.isOwningHandle()).isFalse();
+ }
+ }
+
+ @Test
+ void testStatisticsWithNullInstancePath() throws Exception {
+ Configuration config = new Configuration();
+ config.setBoolean(ConfigOptions.KV_STATISTICS_ENABLED.key(), true);
+
+ try (RocksDBResourceContainer container = new RocksDBResourceContainer(config, null)) {
+ Statistics statistics = container.getStatistics();
+ assertThat(statistics).isNotNull();
+ assertThat(statistics.isOwningHandle()).isTrue();
+ }
+ }
+
+ @Test
+ void testStatisticsWithValidInstancePath(@TempDir Path tempFolder) throws Exception {
+ Configuration config = new Configuration();
+ config.setBoolean(ConfigOptions.KV_STATISTICS_ENABLED.key(), true);
+
+ File instancePath = tempFolder.resolve("rocksdb-instance").toFile();
+ try (RocksDBResourceContainer container =
+ new RocksDBResourceContainer(config, instancePath)) {
+ Statistics statistics = container.getStatistics();
+ assertThat(statistics).isNotNull();
+ assertThat(statistics.isOwningHandle()).isTrue();
+ }
+ }
+
+ @Test
+ void testStatisticsConfigurationConsistency() throws Exception {
+ Configuration config = new Configuration();
+
+ // Test with statistics disabled
+ config.setBoolean(ConfigOptions.KV_STATISTICS_ENABLED.key(), false);
+ try (RocksDBResourceContainer container = new RocksDBResourceContainer(config, null)) {
+ assertThat(container.getStatistics()).isNull();
+ }
+
+ // Test with statistics enabled
+ config.setBoolean(ConfigOptions.KV_STATISTICS_ENABLED.key(), true);
+ try (RocksDBResourceContainer container = new RocksDBResourceContainer(config, null)) {
+ assertThat(container.getStatistics()).isNotNull();
+ }
+ }
+
+ @Test
+ void testStatisticsMemoryManagement() throws Exception {
+ Configuration config = new Configuration();
+ config.setBoolean(ConfigOptions.KV_STATISTICS_ENABLED.key(), true);
+
+ // Create and close many containers to test memory management
+ List<RocksDBResourceContainer> containers = new ArrayList<>();
+ for (int i = 0; i < 10; i++) {
+ RocksDBResourceContainer container = new RocksDBResourceContainer(config, null);
+ containers.add(container);
+
+ // Verify statistics is created and valid
+ Statistics stats = container.getStatistics();
+ assertThat(stats).isNotNull();
+ assertThat(stats.isOwningHandle()).isTrue();
+ }
+
+ // Close all containers
+ for (RocksDBResourceContainer container : containers) {
+ container.close();
+ }
+
+ // This should not cause memory leaks or crashes
+ System.gc(); // Suggest garbage collection
+ }
+
+ @Test
+ void testStatisticsThreadSafety() throws Exception {
+ Configuration config = new Configuration();
+ config.setBoolean(ConfigOptions.KV_STATISTICS_ENABLED.key(), true);
+
+ try (RocksDBResourceContainer container = new RocksDBResourceContainer(config, null)) {
+ Statistics statistics = container.getStatistics();
+ assertThat(statistics).isNotNull();
+
+ // Test concurrent access to statistics
+ int threadCount = 10;
+ CountDownLatch latch = new CountDownLatch(threadCount);
+ List<Exception> exceptions = Collections.synchronizedList(new ArrayList<>());
+
+ for (int i = 0; i < threadCount; i++) {
+ new Thread(
+ () -> {
+ try {
+ // Access statistics concurrently
+ Statistics stats = container.getStatistics();
+ assertThat(stats).isNotNull();
+ } catch (Exception e) {
+ exceptions.add(e);
+ } finally {
+ latch.countDown();
+ }
+ })
+ .start();
+ }
+
+ latch.await(5, TimeUnit.SECONDS);
+ assertThat(exceptions).isEmpty();
+ }
+ }
+
+ @Test
+ void testStatisticsWithDifferentConfigurations() throws Exception {
Review Comment:
this has been tested above.
##########
fluss-server/src/test/java/com/alibaba/fluss/server/kv/rocksdb/RocksDBMetricsManagerTest.java:
##########
@@ -0,0 +1,423 @@
+/*
+ * 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 com.alibaba.fluss.server.kv.rocksdb;
+
+import com.alibaba.fluss.config.Configuration;
+import com.alibaba.fluss.metadata.TableBucket;
+import com.alibaba.fluss.metadata.TablePath;
+import com.alibaba.fluss.server.metrics.group.TestingMetricGroups;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.jupiter.api.io.TempDir;
+import org.mockito.MockitoAnnotations;
+import org.rocksdb.RocksDB;
+import org.rocksdb.Statistics;
+
+import java.lang.reflect.Field;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+/** Tests for {@link RocksDBMetricsManager}. */
+class RocksDBMetricsManagerTest {
+
+ @TempDir Path tempDir;
+ private RocksDBResourceContainer resourceContainer;
+ private List<RocksDBMetricsCollector> collectors;
+ private AutoCloseable mocksCloseable;
+ @RegisterExtension public RocksDBExtension rocksDBExtension = new RocksDBExtension();
+
+ @BeforeEach
+ void setUp() {
+ RocksDB.loadLibrary();
+ mocksCloseable = MockitoAnnotations.openMocks(this);
+ // Setup resource container
+ resourceContainer = new RocksDBResourceContainer(new Configuration(), tempDir.toFile());
+
+ collectors = new ArrayList<>();
+
+ // Reset singleton instance for testing
+ resetSingleton();
+ }
+
+ @AfterEach
+ void tearDown() throws Exception {
+ // Close all collectors
+ for (RocksDBMetricsCollector collector : collectors) {
+ collector.close();
+ }
+ collectors.clear();
+
+ // Shutdown manager
+ RocksDBMetricsManager.getInstance().shutdown();
+
+ // Clean up resources
+ if (resourceContainer != null) {
+ resourceContainer.close();
+ }
+ if (mocksCloseable != null) {
+ mocksCloseable.close();
+ }
+
+ // Reset singleton for next test
+ resetSingleton();
+ }
+
+ @Test
+ void testRegisterCollector() throws Exception {
+
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ assertThat(manager.getCollectorCount()).isEqualTo(0);
+
+ RocksDBMetricsCollector collector = createMockCollector("test_db", "test_table", 0);
+ manager.registerCollector(collector);
+
+ assertThat(manager.getCollectorCount()).isEqualTo(1);
+ assertThat(manager.isRunning()).isTrue();
+
+ collectors.add(collector);
+ }
+
+ @Test
+ void testRegisterMultipleCollectors() throws Exception {
+
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ for (int i = 0; i < 5; i++) {
+ RocksDBMetricsCollector collector = createMockCollector("test_db", "test_table", i);
+ manager.registerCollector(collector);
+ collectors.add(collector);
+ }
+
+ assertThat(manager.getCollectorCount()).isEqualTo(5);
+ assertThat(manager.isRunning()).isTrue();
+ }
+
+ @Test
+ void testUnregisterCollector() throws Exception {
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ RocksDBMetricsCollector collector = createMockCollector("test_db", "test_table", 0);
+ manager.registerCollector(collector);
+ collectors.add(collector);
+
+ assertThat(manager.getCollectorCount()).isEqualTo(1);
+
+ manager.unregisterCollector(collector);
+
+ assertThat(manager.getCollectorCount()).isEqualTo(0);
+ }
+
+ @Test
+ void testRegisterAfterShutdown() throws Exception {
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+ manager.shutdown();
+
+ RocksDBMetricsCollector collector = createMockCollector("test_db", "test_table", 0);
+ manager.registerCollector(collector);
+
+ // Should not be registered after shutdown
+ assertThat(manager.getCollectorCount()).isEqualTo(0);
+
+ collectors.add(collector);
+ }
+
+ @Test
+ void testMetricsCollectionStartsWithFirstCollector() throws Exception {
+
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ assertThat(manager.isRunning()).isFalse();
+
+ RocksDBMetricsCollector collector = createMockCollector("test_db", "test_table", 0);
+ manager.registerCollector(collector);
+ collectors.add(collector);
+
+ assertThat(manager.isRunning()).isTrue();
+ }
+
+ @Test
+ void testBatchProcessing() throws Exception {
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ // Create many collectors to test batch processing
+ int collectorCount = 250; // More than batch size (100)
+ List<RocksDBMetricsCollector> spyCollectors = new ArrayList<>();
+
+ for (int i = 0; i < collectorCount; i++) {
+ RocksDBMetricsCollector collector = createMockCollector("test_db", "test_table", i);
+ RocksDBMetricsCollector spyCollector = spy(collector);
+ manager.registerCollector(spyCollector);
+ spyCollectors.add(spyCollector);
+ collectors.add(collector);
+ }
+
+ // Wait for metrics collection to occur
+ Thread.sleep(6000); // Wait for at least one collection cycle
+
+ // Verify that updateMetrics was called on all collectors
+ for (RocksDBMetricsCollector collector : spyCollectors) {
+ verify(collector, times(1)).updateMetrics();
+ }
+ }
+
+ @Test
+ void testExceptionHandlingInMetricsCollection() throws Exception {
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ // Create a collector that throws exception
+ RocksDBMetricsCollector faultyCollector = createMockCollector("test_db", "test_table", 0);
+ doThrow(new RuntimeException("Test exception")).when(faultyCollector).updateMetrics();
+
+ // Create a normal collector
+ RocksDBMetricsCollector normalCollector = createMockCollector("test_db", "test_table", 1);
+ RocksDBMetricsCollector spyNormalCollector = spy(normalCollector);
+
+ manager.registerCollector(faultyCollector);
+ manager.registerCollector(spyNormalCollector);
+
+ collectors.add(faultyCollector);
+ collectors.add(normalCollector);
+
+ // Wait for metrics collection
+ try {
+ Thread.sleep(6000);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+
+ // Normal collector should still be called despite exception in faulty collector
+ verify(spyNormalCollector, times(1)).updateMetrics();
+ }
+
+ @Test
+ void testShutdown() {
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ // Register some collectors
+ for (int i = 0; i < 3; i++) {
+ RocksDBMetricsCollector collector = createMockCollector("test_db", "test_table", i);
+ manager.registerCollector(collector);
+ collectors.add(collector);
+ }
+
+ assertThat(manager.isRunning()).isTrue();
+ assertThat(manager.getCollectorCount()).isEqualTo(3);
+
+ manager.shutdown();
+
+ assertThat(manager.isRunning()).isFalse();
+ assertThat(manager.isShutdown()).isTrue();
+ assertThat(manager.getCollectorCount()).isEqualTo(0);
+ }
+
+ @Test
+ void testDoubleShutdown() {
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ manager.shutdown();
+ assertThat(manager.isShutdown()).isTrue();
+
+ // Second shutdown should not cause issues
+ manager.shutdown();
+ assertThat(manager.isShutdown()).isTrue();
+ }
+
+ @Test
+ void testConcurrentRegistrationAndUnregistration() throws Exception {
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ int threadCount = 10;
+ CountDownLatch latch = new CountDownLatch(threadCount);
+ ExecutorService executor = Executors.newFixedThreadPool(threadCount);
+ AtomicInteger collectorId = new AtomicInteger(0);
+
+ for (int i = 0; i < threadCount; i++) {
+ executor.submit(
+ () -> {
+ try {
+ for (int j = 0; j < 10; j++) {
+ RocksDBMetricsCollector collector =
+ createMockCollector(
+ "test_db",
+ "test_table",
+ collectorId.getAndIncrement());
+ manager.registerCollector(collector);
+ collectors.add(collector);
+
+ // Immediately unregister some collectors
+ if (j % 2 == 0) {
+ manager.unregisterCollector(collector);
+ }
+ }
+ } finally {
+ latch.countDown();
+ }
+ });
+ }
+
+ latch.await(10, TimeUnit.SECONDS);
+ executor.shutdown();
+
+ // Should handle concurrent access gracefully
+ assertThat(manager.getCollectorCount()).isGreaterThanOrEqualTo(0);
+ }
+
+ @Test
+ void testEmptyCollectorsDoesNotCauseIssues() throws Exception {
+ RocksDBMetricsManager manager = RocksDBMetricsManager.getInstance();
+
+ // Start with no collectors
+ assertThat(manager.getCollectorCount()).isEqualTo(0);
+
+ // Wait a bit to ensure no exceptions are thrown
+ Thread.sleep(1000);
Review Comment:
ditto
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org