0
# Native Metrics Configuration
1

2
Configuration for RocksDB native metrics monitoring, enabling detailed performance monitoring and observability of RocksDB internal operations.
3

4
## Capabilities
5

6
### RocksDBNativeMetricOptions
7

8
Configuration class for enabling and managing RocksDB native metrics collection.
9

10
```java { .api }
11
/**
12
 * Configuration for RocksDB native metrics collection.
13
 * Allows enabling specific RocksDB metrics for monitoring and performance analysis.
14
 */
15
class RocksDBNativeMetricOptions {
16
    
17
    /**
18
     * Creates RocksDBNativeMetricOptions from Flink configuration.
19
     * @param config configuration to read metric settings from
20
     * @return configured native metric options
21
     */
22
    static RocksDBNativeMetricOptions fromConfig(ReadableConfig config);
23
}
24
```
25

26
### Memory Table Metrics
27

28
Enable monitoring of RocksDB memory table (memtable) statistics.
29

30
```java { .api }
31
/**
32
 * Enables monitoring of the number of immutable memory tables.
33
 * Tracks memory tables that are waiting to be flushed to disk.
34
 */
35
void enableNumImmutableMemTable();
36

37
/**
38
 * Enables monitoring of memory table flush pending status.
39
 * Indicates whether a flush operation is currently pending.
40
 */
41
void enableMemTableFlushPending();
42

43
/**
44
 * Enables monitoring of current size of active memory table.
45
 * Tracks memory usage of the currently active memtable.
46
 */
47
void enableCurSizeActiveMemTable();
48

49
/**
50
 * Enables monitoring of current size of all memory tables.
51
 * Tracks total memory usage of all active and immutable memtables.
52
 */
53
void enableCurSizeAllMemTables();
54

55
/**
56
 * Enables monitoring of size of all memory tables.
57
 * Similar to current size but includes additional metadata.
58
 */
59
void enableSizeAllMemTables();
60

61
/**
62
 * Enables monitoring of number of entries in active memory table.
63
 * Tracks the count of key-value pairs in the active memtable.
64
 */
65
void enableNumEntriesActiveMemTable();
66

67
/**
68
 * Enables monitoring of number of entries in immutable memory tables.
69
 * Tracks total entries across all immutable memtables.
70
 */
71
void enableNumEntriesImmMemTables();
72

73
/**
74
 * Enables monitoring of number of deletes in active memory table.
75
 * Tracks delete operations in the currently active memtable.
76
 */
77
void enableNumDeletesActiveMemTable();
78

79
/**
80
 * Enables monitoring of number of deletes in immutable memory tables.
81
 * Tracks delete operations across all immutable memtables.
82
 */
83
void enableNumDeletesImmMemTables();
84
```
85

86
### Compaction and Database Metrics
87

88
Enable monitoring of RocksDB compaction operations and database-level statistics.
89

90
```java { .api }
91
/**
92
 * Enables monitoring of compaction pending status.
93
 * Indicates whether compaction operations are currently pending.
94
 */
95
void enableCompactionPending();
96

97
/**
98
 * Enables monitoring of background errors.
99
 * Tracks errors that occur during background operations like compaction.
100
 */
101
void enableBackgroundErrors();
102

103
/**
104
 * Enables monitoring of estimated number of keys in the database.
105
 * Provides an estimate of total key-value pairs across all levels.
106
 */
107
void enableEstimateNumKeys();
108

109
/**
110
 * Enables monitoring of number of running compactions.
111
 * Tracks currently active compaction operations.
112
 */
113
void enableNumRunningCompactions();
114

115
/**
116
 * Enables monitoring of number of running flushes.
117
 * Tracks currently active flush operations from memtables to disk.
118
 */
119
void enableNumRunningFlushes();
120

121
/**
122
 * Enables monitoring of actual delayed write rate.
123
 * Tracks write throttling when RocksDB slows down writes due to compaction lag.
124
 */
125
void enableActualDelayedWriteRate();
126

127
/**
128
 * Enables monitoring of write stopped status.
129
 * Indicates whether writes are currently stopped due to resource constraints.
130
 */
131
void enableIsWriteStopped();
132
```
133

134
### Storage and Memory Estimation Metrics
135

136
Enable monitoring of RocksDB storage usage and memory consumption estimates.
137

138
```java { .api }
139
/**
140
 * Enables monitoring of estimated table readers memory usage.
141
 * Tracks memory used by SST file readers (index and filter blocks).
142
 */
143
void enableEstimateTableReadersMem();
144

145
/**
146
 * Enables monitoring of number of snapshots.
147
 * Tracks currently active database snapshots.
148
 */
149
void enableNumSnapshots();
150

151
/**
152
 * Enables monitoring of number of live versions.
153
 * Tracks active database versions (used for MVCC).
154
 */
155
void enableNumLiveVersions();
156

157
/**
158
 * Enables monitoring of estimated live data size.
159
 * Estimates the size of data that's not deleted or overwritten.
160
 */
161
void enableEstimateLiveDataSize();
162

163
/**
164
 * Enables monitoring of total SST files size.
165
 * Tracks total disk space used by all SST files.
166
 */
167
void enableTotalSstFilesSize();
168

169
/**
170
 * Enables monitoring of live SST files size.
171
 * Tracks disk space used by currently active SST files.
172
 */
173
void enableLiveSstFilesSize();
174

175
/**
176
 * Enables monitoring of estimated pending compaction bytes.
177
 * Estimates the amount of data waiting to be compacted.
178
 */
179
void enableEstimatePendingCompactionBytes();
180
```
181

182
### Block Cache Metrics
183

184
Enable monitoring of RocksDB block cache performance and usage.
185

186
```java { .api }
187
/**
188
 * Enables monitoring of block cache capacity.
189
 * Tracks the maximum size configured for the block cache.
190
 */
191
void enableBlockCacheCapacity();
192

193
/**
194
 * Enables monitoring of block cache usage.
195
 * Tracks current memory usage of the block cache.
196
 */
197
void enableBlockCacheUsage();
198

199
/**
200
 * Enables monitoring of block cache pinned usage.
201
 * Tracks memory used by pinned blocks in the cache.
202
 */
203
void enableBlockCachePinnedUsage();
204
```
205

206
### Configuration and Query Methods
207

208
Configure metric collection behavior and query enabled metrics.
209

210
```java { .api }
211
/**
212
 * Configures whether column family name should be used as a variable in metrics.
213
 * When enabled, metrics are tagged with column family names for better granularity.
214
 * @param columnFamilyAsVariable whether to use column family as metric variable
215
 */
216
void setColumnFamilyAsVariable(boolean columnFamilyAsVariable);
217

218
/**
219
 * Checks whether column family is used as a variable in metrics.
220
 * @return true if column family names are included in metric tags
221
 */
222
boolean isColumnFamilyAsVariable();
223

224
/**
225
 * Gets the collection of enabled metric properties.
226
 * @return collection of RocksDB property names that are enabled for monitoring
227
 */
228
Collection<String> getProperties();
229

230
/**
231
 * Checks whether any metrics are enabled.
232
 * @return true if at least one metric is enabled for collection
233
 */
234
boolean isEnabled();
235
```
236

237
## Usage Examples
238

239
### Basic Metrics Configuration
240

241
```java
242
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
243
import org.apache.flink.contrib.streaming.state.RocksDBNativeMetricOptions;
244
import org.apache.flink.configuration.Configuration;
245

246
// Create state backend
247
EmbeddedRocksDBStateBackend stateBackend = new EmbeddedRocksDBStateBackend(true);
248

249
// Create custom options factory with metrics
250
RocksDBOptionsFactory optionsFactory = new RocksDBOptionsFactory() {
251
    @Override
252
    public DBOptions createDBOptions(DBOptions currentOptions, Collection<AutoCloseable> handlesToClose) {
253
        return currentOptions;
254
    }
255
    
256
    @Override
257
    public ColumnFamilyOptions createColumnOptions(ColumnFamilyOptions currentOptions, Collection<AutoCloseable> handlesToClose) {
258
        return currentOptions;
259
    }
260
    
261
    @Override
262
    public RocksDBNativeMetricOptions createNativeMetricsOptions(RocksDBNativeMetricOptions nativeMetricOptions) {
263
        // Enable basic performance metrics
264
        nativeMetricOptions.enableNumImmutableMemTable();
265
        nativeMetricOptions.enableCompactionPending();
266
        nativeMetricOptions.enableBlockCacheUsage();
267
        nativeMetricOptions.enableEstimateNumKeys();
268
        
269
        return nativeMetricOptions;
270
    }
271
};
272

273
stateBackend.setRocksDBOptions(optionsFactory);
274
```
275

276
### Comprehensive Metrics Configuration
277

278
```java
279
// Enable extensive monitoring for production environments
280
RocksDBOptionsFactory comprehensiveFactory = new RocksDBOptionsFactory() {
281
    @Override
282
    public DBOptions createDBOptions(DBOptions currentOptions, Collection<AutoCloseable> handlesToClose) {
283
        return currentOptions;
284
    }
285
    
286
    @Override
287
    public ColumnFamilyOptions createColumnOptions(ColumnFamilyOptions currentOptions, Collection<AutoCloseable> handlesToClose) {
288
        return currentOptions;
289
    }
290
    
291
    @Override
292
    public RocksDBNativeMetricOptions createNativeMetricsOptions(RocksDBNativeMetricOptions nativeMetricOptions) {
293
        // Memory table metrics
294
        nativeMetricOptions.enableNumImmutableMemTable();
295
        nativeMetricOptions.enableMemTableFlushPending();
296
        nativeMetricOptions.enableCurSizeActiveMemTable();
297
        nativeMetricOptions.enableCurSizeAllMemTables();
298
        
299
        // Compaction metrics
300
        nativeMetricOptions.enableCompactionPending();
301
        nativeMetricOptions.enableNumRunningCompactions();
302
        nativeMetricOptions.enableNumRunningFlushes();
303
        nativeMetricOptions.enableEstimatePendingCompactionBytes();
304
        
305
        // Storage metrics
306
        nativeMetricOptions.enableEstimateNumKeys();
307
        nativeMetricOptions.enableEstimateLiveDataSize();
308
        nativeMetricOptions.enableTotalSstFilesSize();
309
        
310
        // Cache metrics
311
        nativeMetricOptions.enableBlockCacheCapacity();
312
        nativeMetricOptions.enableBlockCacheUsage();
313
        nativeMetricOptions.enableBlockCachePinnedUsage();
314
        
315
        // Write throttling metrics
316
        nativeMetricOptions.enableActualDelayedWriteRate();
317
        nativeMetricOptions.enableIsWriteStopped();
318
        
319
        // Use column family names in metrics for better granularity
320
        nativeMetricOptions.setColumnFamilyAsVariable(true);
321
        
322
        return nativeMetricOptions;
323
    }
324
};
325

326
stateBackend.setRocksDBOptions(comprehensiveFactory);
327
```
328

329
### Configuration-Based Metrics Setup
330

331
```java
332
import org.apache.flink.configuration.Configuration;
333

334
// Configure metrics through Flink configuration
335
Configuration config = new Configuration();
336
config.setString("state.backend.rocksdb.metrics.num-immutable-mem-table", "true");
337
config.setString("state.backend.rocksdb.metrics.compaction-pending", "true");
338
config.setString("state.backend.rocksdb.metrics.block-cache-usage", "true");
339
config.setString("state.backend.rocksdb.metrics.estimate-num-keys", "true");
340
config.setString("state.backend.rocksdb.metrics.column-family-as-variable", "true");
341

342
// Create metrics options from configuration
343
RocksDBNativeMetricOptions metricsOptions = RocksDBNativeMetricOptions.fromConfig(config);
344

345
// Check what metrics are enabled
346
if (metricsOptions.isEnabled()) {
347
    Collection<String> enabledProperties = metricsOptions.getProperties();
348
    System.out.println("Enabled metrics: " + enabledProperties);
349
}
350
```
351

352
## Monitoring Best Practices
353

354
### Essential Metrics for Production
355

356
For production environments, consider enabling these key metrics:
357

358
```java
359
// Essential performance indicators
360
nativeMetricOptions.enableNumImmutableMemTable();    // Memory pressure
361
nativeMetricOptions.enableCompactionPending();       // Compaction lag  
362
nativeMetricOptions.enableBlockCacheUsage();         // Cache efficiency
363
nativeMetricOptions.enableEstimateNumKeys();         // Data growth
364
nativeMetricOptions.enableActualDelayedWriteRate();  // Write throttling
365
nativeMetricOptions.enableIsWriteStopped();          // Write blocking
366
```
367

368
### Memory Monitoring
369

370
For memory-constrained environments:
371

372
```java
373
// Memory usage tracking
374
nativeMetricOptions.enableCurSizeAllMemTables();     // Memory table usage
375
nativeMetricOptions.enableEstimateTableReadersMem(); // Reader memory
376
nativeMetricOptions.enableBlockCacheUsage();         // Cache memory
377
nativeMetricOptions.enableBlockCachePinnedUsage();   // Pinned cache memory
378
```
379

380
### I/O Performance Monitoring
381

382
For I/O-intensive workloads:
383

384
```java
385
// I/O and compaction monitoring
386
nativeMetricOptions.enableNumRunningCompactions();   // Active compactions
387
nativeMetricOptions.enableNumRunningFlushes();       // Active flushes
388
nativeMetricOptions.enableEstimatePendingCompactionBytes(); // Compaction backlog
389
nativeMetricOptions.enableTotalSstFilesSize();       // Disk usage
390
```
391

392
## Performance Impact
393

394
- **Low Impact**: Basic metrics like `enableEstimateNumKeys()`, `enableCompactionPending()`
395
- **Medium Impact**: Cache metrics, memory table metrics
396
- **Higher Impact**: Detailed compaction metrics, fine-grained storage metrics
397

398
Enable metrics incrementally and monitor the performance impact on your specific workload.