You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ch...@apache.org on 2018/10/19 09:01:12 UTC
[flink] branch master updated: [FLINK-10423][rocksdb][metrics]
Expose Rocksdb native metrics
This is an automated email from the ASF dual-hosted git repository.
chesnay pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new e62a7ea [FLINK-10423][rocksdb][metrics] Expose Rocksdb native metrics
e62a7ea is described below
commit e62a7eabe34da42e4cb56d1b14905a3b9e9c9bd4
Author: Seth Wiesman <sj...@gmail.com>
AuthorDate: Fri Oct 19 04:01:03 2018 -0500
[FLINK-10423][rocksdb][metrics] Expose Rocksdb native metrics
---
.../rocks_db_native_metric_configuration.html | 116 ++++++
docs/monitoring/metrics.md | 3 +
docs/ops/config.md | 10 +
docs/ops/state/state_backends.md | 2 +
.../KVStateRequestSerializerRocksDBTest.java | 11 +-
.../network/KvStateServerHandlerTest.java | 4 +-
.../flink/runtime/state/AbstractStateBackend.java | 4 +-
.../apache/flink/runtime/state/StateBackend.java | 42 ++-
.../runtime/state/filesystem/FsStateBackend.java | 4 +-
.../runtime/state/memory/MemoryStateBackend.java | 4 +-
.../CheckpointSettingsSerializableTest.java | 4 +-
.../state/ttl/mock/MockKeyedStateBackend.java | 4 +-
.../runtime/state/ttl/mock/MockStateBackend.java | 7 +-
.../contrib/streaming/state/OptionsFactory.java | 12 +
.../streaming/state/RocksDBKeyedStateBackend.java | 49 ++-
.../state/RocksDBNativeMetricMonitor.java | 170 +++++++++
.../state/RocksDBNativeMetricOptions.java | 419 +++++++++++++++++++++
.../contrib/streaming/state/RocksDBProperty.java | 80 ++++
.../streaming/state/RocksDBStateBackend.java | 24 +-
.../state/RocksDBNativeMetricMonitorTest.java | 207 ++++++++++
.../state/RocksDBNativeMetricOptionsTest.java | 48 +++
.../streaming/state/RocksDBPropertyTest.java | 49 +++
.../streaming/state/RocksDBStateBackendTest.java | 5 +-
.../api/operators/AbstractStreamOperator.java | 3 +-
.../api/operators/StreamTaskStateInitializer.java | 5 +-
.../operators/StreamTaskStateInitializerImpl.java | 13 +-
.../StateInitializationContextImplTest.java | 4 +-
.../StreamTaskStateInitializerImplTest.java | 11 +-
.../runtime/tasks/StreamTaskTerminationTest.java | 4 +-
.../streaming/runtime/tasks/StreamTaskTest.java | 5 +-
.../runtime/tasks/TestSpyWrapperStateBackend.java | 7 +-
.../test/streaming/runtime/StateBackendITCase.java | 4 +-
32 files changed, 1297 insertions(+), 37 deletions(-)
diff --git a/docs/_includes/generated/rocks_db_native_metric_configuration.html b/docs/_includes/generated/rocks_db_native_metric_configuration.html
new file mode 100644
index 0000000..78fc523
--- /dev/null
+++ b/docs/_includes/generated/rocks_db_native_metric_configuration.html
@@ -0,0 +1,116 @@
+<table class="table table-bordered">
+ <thead>
+ <tr>
+ <th class="text-left" style="width: 20%">Key</th>
+ <th class="text-left" style="width: 15%">Default</th>
+ <th class="text-left" style="width: 65%">Description</th>
+ </tr>
+ </thead>
+ <tbody>
+ <tr>
+ <td><h5>state.backend.rocksdb.metrics.actual-delayed-write-rate</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Monitor the current actual delayed write rate. 0 means no delay.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.rocksdb.metrics.background-errors</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Monitor the number of background errors in RocksDB.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.rocksdb.metrics.compaction-pending</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Track pending compactions in RocksDB. Returns 1 if a compaction is pending, 0 otherwise.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.rocksdb.metrics.cur-size-active-mem-table</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Monitor the approximate size of the active memtable in bytes.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.rocksdb.metrics.cur-size-all-mem-tables</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Monitor the approximate size of the active and unflushed immutable memtables in bytes.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.rocksdb.metrics.estimate-live-data-size</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Estimate of the amount of live data in bytes.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.rocksdb.metrics.estimate-num-keys</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Estimate the number of keys in RocksDB.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.rocksdb.metrics.estimate-pending-compaction-bytes</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Estimated total number of bytes compaction needs to rewrite to get all levels down to under target size. Not valid for other compactions than level-based.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.rocksdb.metrics.estimate-table-readers-mem</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Estimate the memory used for reading SST tables, excluding memory used in block cache (e.g.,filter and index blocks) in bytes.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.rocksdb.metrics.mem-table-flush-pending</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Monitor the number of pending memtable flushes in RocksDB.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.rocksdb.metrics.num-deletes-active-mem-table</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Monitor the total number of delete entries in the active memtable.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.rocksdb.metrics.num-deletes-imm-mem-tables</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Monitor the total number of delete entries in the unflushed immutable memtables.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.rocksdb.metrics.num-entries-active-mem-table</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Monitor the total number of entries in the active memtable.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.rocksdb.metrics.num-entries-imm-mem-tables</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Monitor the total number of entries in the unflushed immutable memtables.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.rocksdb.metrics.num-immutable-mem-table</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Monitor the number of immutable memtables in RocksDB.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.rocksdb.metrics.num-live-versions</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Monitor number of live versions. Version is an internal data structure. See RocksDB file version_set.h for details. More live versions often mean more SST files are held from being deleted, by iterators or unfinished compactions.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.rocksdb.metrics.num-running-compactions</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Monitor the number of currently running compactions.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.rocksdb.metrics.num-running-flushes</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Monitor the number of currently running flushes.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.rocksdb.metrics.num-snapshots</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Monitor the number of unreleased snapshots of the database.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.rocksdb.metrics.size-all-mem-tables</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Monitor the approximate size of the active, unflushed immutable, and pinned immutable memtables in bytes.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.rocksdb.metrics.total-sst-files-size</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Monitor the total size (bytes) of all SST files.WARNING: may slow down online queries if there are too many files.</td>
+ </tr>
+ </tbody>
+</table>
diff --git a/docs/monitoring/metrics.md b/docs/monitoring/metrics.md
index 85c60a6..04efdef 100644
--- a/docs/monitoring/metrics.md
+++ b/docs/monitoring/metrics.md
@@ -1205,6 +1205,9 @@ Thus, in order to infer the metric identifier:
</tbody>
</table>
+### RocksDB
+Certain RocksDB native metrics are available but disabled by default, you can find full documentation [here]({{ site.baseurl }}/ops/config.html#rocksdb-native-metrics)
+
### IO
<table class="table table-bordered">
<thead>
diff --git a/docs/ops/config.md b/docs/ops/config.md
index d370e56..267b670 100644
--- a/docs/ops/config.md
+++ b/docs/ops/config.md
@@ -162,6 +162,16 @@ The configuration keys in this section are independent of the used resource mana
{% include generated/metric_configuration.html %}
+### RocksDB Native Metrics
+Certain RocksDB native metrics may be forwarded to Flink's metrics reporter.
+All native metrics are scoped to operators and then further broken down by column family; values are reported as unsigned longs.
+
+<div class="alert alert-warning">
+ <strong>Note:</strong> Enabling native metrics may cause degraded performance and should be set carefully.
+</div>
+
+{% include generated/rocks_db_native_metric_configuration.html %}
+
### History Server
You have to configure `jobmanager.archive.fs.dir` in order to archive terminated jobs and add it to the list of monitored directories via `historyserver.archive.fs.dir` if you want to display them via the HistoryServer's web frontend.
diff --git a/docs/ops/state/state_backends.md b/docs/ops/state/state_backends.md
index 4e09b34..b95d2b7 100644
--- a/docs/ops/state/state_backends.md
+++ b/docs/ops/state/state_backends.md
@@ -121,6 +121,8 @@ on-heap representation as the heap-based backends are doing.
RocksDBStateBackend is currently the only backend that offers incremental checkpoints (see [here](large_state_tuning.html)).
+Certain RocksDB native metrics are available but disabled by default, you can find full documentation [here]({{ site.baseurl }}/ops/config.html#rocksdb-native-metrics)
+
## Configuring a State Backend
The default state backend, if you specify nothing, is the jobmanager. If you wish to establish a different default for all jobs on your cluster, you can do so by defining a new default state backend in **flink-conf.yaml**. The default state backend can be overridden on a per-job basis, as shown below.
diff --git a/flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/network/KVStateRequestSerializerRocksDBTest.java b/flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/network/KVStateRequestSerializerRocksDBTest.java
index 176e55a..fff8f02 100644
--- a/flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/network/KVStateRequestSerializerRocksDBTest.java
+++ b/flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/network/KVStateRequestSerializerRocksDBTest.java
@@ -25,7 +25,9 @@ import org.apache.flink.api.common.typeutils.base.LongSerializer;
import org.apache.flink.api.common.typeutils.base.StringSerializer;
import org.apache.flink.contrib.streaming.state.PredefinedOptions;
import org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackend;
+import org.apache.flink.contrib.streaming.state.RocksDBNativeMetricOptions;
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;
+import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
import org.apache.flink.queryablestate.client.VoidNamespace;
import org.apache.flink.queryablestate.client.VoidNamespaceSerializer;
import org.apache.flink.runtime.query.TaskKvStateRegistry;
@@ -82,7 +84,9 @@ public final class KVStateRequestSerializerRocksDBTest {
false,
TestLocalRecoveryConfig.disabled(),
RocksDBStateBackend.PriorityQueueStateType.HEAP,
- TtlTimeProvider.DEFAULT
+ TtlTimeProvider.DEFAULT,
+ new RocksDBNativeMetricOptions(),
+ new UnregisteredMetricsGroup()
);
longHeapKeyedStateBackend.restore(null);
longHeapKeyedStateBackend.setCurrentKey(key);
@@ -124,7 +128,10 @@ public final class KVStateRequestSerializerRocksDBTest {
false,
TestLocalRecoveryConfig.disabled(),
RocksDBStateBackend.PriorityQueueStateType.HEAP,
- TtlTimeProvider.DEFAULT);
+ TtlTimeProvider.DEFAULT,
+ new RocksDBNativeMetricOptions(),
+ new UnregisteredMetricsGroup()
+ );
longHeapKeyedStateBackend.restore(null);
longHeapKeyedStateBackend.setCurrentKey(key);
diff --git a/flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/network/KvStateServerHandlerTest.java b/flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/network/KvStateServerHandlerTest.java
index adcf3ae..bfadb75 100644
--- a/flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/network/KvStateServerHandlerTest.java
+++ b/flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/network/KvStateServerHandlerTest.java
@@ -26,6 +26,7 @@ import org.apache.flink.api.common.typeutils.base.IntSerializer;
import org.apache.flink.api.common.typeutils.base.LongSerializer;
import org.apache.flink.api.common.typeutils.base.StringSerializer;
import org.apache.flink.api.common.typeutils.base.array.BytePrimitiveArraySerializer;
+import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
import org.apache.flink.queryablestate.KvStateID;
import org.apache.flink.queryablestate.client.VoidNamespace;
import org.apache.flink.queryablestate.client.VoidNamespaceSerializer;
@@ -763,6 +764,7 @@ public class KvStateServerHandlerTest extends TestLogger {
numKeyGroups,
new KeyGroupRange(0, 0),
registry.createTaskRegistry(dummyEnv.getJobID(), dummyEnv.getJobVertexId()),
- TtlTimeProvider.DEFAULT);
+ TtlTimeProvider.DEFAULT,
+ new UnregisteredMetricsGroup());
}
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/AbstractStateBackend.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/AbstractStateBackend.java
index d397a88..6c72acd 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/AbstractStateBackend.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/AbstractStateBackend.java
@@ -21,6 +21,7 @@ package org.apache.flink.runtime.state;
import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.metrics.MetricGroup;
import org.apache.flink.runtime.execution.Environment;
import org.apache.flink.runtime.query.TaskKvStateRegistry;
import org.apache.flink.runtime.state.ttl.TtlTimeProvider;
@@ -50,7 +51,8 @@ public abstract class AbstractStateBackend implements StateBackend, java.io.Seri
int numberOfKeyGroups,
KeyGroupRange keyGroupRange,
TaskKvStateRegistry kvStateRegistry,
- TtlTimeProvider ttlTimeProvider) throws IOException;
+ TtlTimeProvider ttlTimeProvider,
+ MetricGroup metricGroup) throws IOException;
@Override
public abstract OperatorStateBackend createOperatorStateBackend(
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/StateBackend.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/StateBackend.java
index 2775b71..4235f76 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/StateBackend.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/StateBackend.java
@@ -21,6 +21,8 @@ package org.apache.flink.runtime.state;
import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
import org.apache.flink.runtime.execution.Environment;
import org.apache.flink.runtime.query.TaskKvStateRegistry;
import org.apache.flink.runtime.state.ttl.TtlTimeProvider;
@@ -144,7 +146,42 @@ public interface StateBackend extends java.io.Serializable {
numberOfKeyGroups,
keyGroupRange,
kvStateRegistry,
- TtlTimeProvider.DEFAULT);
+ TtlTimeProvider.DEFAULT
+ );
+ }
+
+ /**
+ * Creates a new {@link AbstractKeyedStateBackend} that is responsible for holding <b>keyed state</b>
+ * and checkpointing it.
+ *
+ * <p><i>Keyed State</i> is state where each value is bound to a key.
+ *
+ * @param <K> The type of the keys by which the state is organized.
+ *
+ * @return The Keyed State Backend for the given job, operator, and key group range.
+ *
+ * @throws Exception This method may forward all exceptions that occur while instantiating the backend.
+ */
+ default <K> AbstractKeyedStateBackend<K> createKeyedStateBackend(
+ Environment env,
+ JobID jobID,
+ String operatorIdentifier,
+ TypeSerializer<K> keySerializer,
+ int numberOfKeyGroups,
+ KeyGroupRange keyGroupRange,
+ TaskKvStateRegistry kvStateRegistry,
+ TtlTimeProvider ttlTimeProvider
+ ) throws Exception {
+ return createKeyedStateBackend(
+ env,
+ jobID,
+ operatorIdentifier,
+ keySerializer,
+ numberOfKeyGroups,
+ keyGroupRange,
+ kvStateRegistry,
+ ttlTimeProvider,
+ new UnregisteredMetricsGroup());
}
/**
@@ -167,7 +204,8 @@ public interface StateBackend extends java.io.Serializable {
int numberOfKeyGroups,
KeyGroupRange keyGroupRange,
TaskKvStateRegistry kvStateRegistry,
- TtlTimeProvider ttlTimeProvider) throws Exception;
+ TtlTimeProvider ttlTimeProvider,
+ MetricGroup metricGroup) throws Exception;
/**
* Creates a new {@link OperatorStateBackend} that can be used for storing operator state.
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsStateBackend.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsStateBackend.java
index f5a86e1..135030e 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsStateBackend.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsStateBackend.java
@@ -25,6 +25,7 @@ import org.apache.flink.configuration.CheckpointingOptions;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.core.fs.Path;
+import org.apache.flink.metrics.MetricGroup;
import org.apache.flink.runtime.execution.Environment;
import org.apache.flink.runtime.query.TaskKvStateRegistry;
import org.apache.flink.runtime.state.AbstractKeyedStateBackend;
@@ -456,7 +457,8 @@ public class FsStateBackend extends AbstractFileStateBackend implements Configur
int numberOfKeyGroups,
KeyGroupRange keyGroupRange,
TaskKvStateRegistry kvStateRegistry,
- TtlTimeProvider ttlTimeProvider) {
+ TtlTimeProvider ttlTimeProvider,
+ MetricGroup metricGroup) {
TaskStateManager taskStateManager = env.getTaskStateManager();
LocalRecoveryConfig localRecoveryConfig = taskStateManager.createLocalRecoveryConfig();
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/memory/MemoryStateBackend.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/memory/MemoryStateBackend.java
index 1c464d7..4cc3a2c 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/memory/MemoryStateBackend.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/memory/MemoryStateBackend.java
@@ -24,6 +24,7 @@ import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.configuration.CheckpointingOptions;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.fs.Path;
+import org.apache.flink.metrics.MetricGroup;
import org.apache.flink.runtime.execution.Environment;
import org.apache.flink.runtime.query.TaskKvStateRegistry;
import org.apache.flink.runtime.state.AbstractKeyedStateBackend;
@@ -309,7 +310,8 @@ public class MemoryStateBackend extends AbstractFileStateBackend implements Conf
int numberOfKeyGroups,
KeyGroupRange keyGroupRange,
TaskKvStateRegistry kvStateRegistry,
- TtlTimeProvider ttlTimeProvider) {
+ TtlTimeProvider ttlTimeProvider,
+ MetricGroup metricGroup) {
TaskStateManager taskStateManager = env.getTaskStateManager();
HeapPriorityQueueSetFactory priorityQueueSetFactory =
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointSettingsSerializableTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointSettingsSerializableTest.java
index ffebc52..4b5ae06 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointSettingsSerializableTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointSettingsSerializableTest.java
@@ -23,6 +23,7 @@ import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.testutils.CommonTestUtils;
+import org.apache.flink.metrics.MetricGroup;
import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
import org.apache.flink.runtime.blob.VoidBlobWriter;
import org.apache.flink.runtime.execution.Environment;
@@ -170,7 +171,8 @@ public class CheckpointSettingsSerializableTest extends TestLogger {
int numberOfKeyGroups,
KeyGroupRange keyGroupRange,
TaskKvStateRegistry kvStateRegistry,
- TtlTimeProvider ttlTimeProvider) throws Exception {
+ TtlTimeProvider ttlTimeProvider,
+ MetricGroup metricGroup) throws Exception {
throw new UnsupportedOperationException();
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/mock/MockKeyedStateBackend.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/mock/MockKeyedStateBackend.java
index ccfafec..f88e6d7 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/mock/MockKeyedStateBackend.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/mock/MockKeyedStateBackend.java
@@ -29,6 +29,7 @@ import org.apache.flink.api.common.state.StateDescriptor;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.metrics.MetricGroup;
import org.apache.flink.runtime.checkpoint.CheckpointOptions;
import org.apache.flink.runtime.query.TaskKvStateRegistry;
import org.apache.flink.runtime.state.AbstractKeyedStateBackend;
@@ -95,7 +96,8 @@ public class MockKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
int numberOfKeyGroups,
KeyGroupRange keyGroupRange,
ExecutionConfig executionConfig,
- TtlTimeProvider ttlTimeProvider) {
+ TtlTimeProvider ttlTimeProvider,
+ MetricGroup operatorMetricGroup) {
super(kvStateRegistry, keySerializer, userCodeClassLoader,
numberOfKeyGroups, keyGroupRange, executionConfig, ttlTimeProvider);
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/mock/MockStateBackend.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/mock/MockStateBackend.java
index a8d49dd..8ed84c0 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/mock/MockStateBackend.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/mock/MockStateBackend.java
@@ -20,6 +20,7 @@ package org.apache.flink.runtime.state.ttl.mock;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.metrics.MetricGroup;
import org.apache.flink.runtime.execution.Environment;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.query.KvStateRegistry;
@@ -93,7 +94,8 @@ public class MockStateBackend extends AbstractStateBackend {
int numberOfKeyGroups,
KeyGroupRange keyGroupRange,
TaskKvStateRegistry kvStateRegistry,
- TtlTimeProvider ttlTimeProvider) {
+ TtlTimeProvider ttlTimeProvider,
+ MetricGroup metricGroup) {
return new MockKeyedStateBackend<>(
new KvStateRegistry().createTaskRegistry(jobID, new JobVertexID()),
keySerializer,
@@ -101,7 +103,8 @@ public class MockStateBackend extends AbstractStateBackend {
numberOfKeyGroups,
keyGroupRange,
env.getExecutionConfig(),
- ttlTimeProvider);
+ ttlTimeProvider,
+ metricGroup);
}
@Override
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/OptionsFactory.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/OptionsFactory.java
index 84db465..3204580 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/OptionsFactory.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/OptionsFactory.java
@@ -69,4 +69,16 @@ public interface OptionsFactory extends java.io.Serializable {
*/
ColumnFamilyOptions createColumnOptions(ColumnFamilyOptions currentOptions);
+ /**
+ * This method should enable certain RocksDB metrics to be forwarded to
+ * Flink's metrics reporter.
+ *
+ * <p>Enabling these monitoring options may degrade RockDB performance
+ * and should be set with care.
+ * @param nativeMetricOptions The options object with the pre-defined options.
+ * @return The options object on which the additional options are set.
+ */
+ default RocksDBNativeMetricOptions createNativeMetricsOptions(RocksDBNativeMetricOptions nativeMetricOptions) {
+ return nativeMetricOptions;
+ }
}
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java
index 3d66db6..a4b4ce8 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java
@@ -47,6 +47,7 @@ import org.apache.flink.core.memory.DataInputDeserializer;
import org.apache.flink.core.memory.DataInputView;
import org.apache.flink.core.memory.DataInputViewStreamWrapper;
import org.apache.flink.core.memory.DataOutputSerializer;
+import org.apache.flink.metrics.MetricGroup;
import org.apache.flink.runtime.checkpoint.CheckpointOptions;
import org.apache.flink.runtime.checkpoint.CheckpointType;
import org.apache.flink.runtime.query.TaskKvStateRegistry;
@@ -241,6 +242,13 @@ public class RocksDBKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
/** Shared wrapper for batch writes to the RocksDB instance. */
private RocksDBWriteBatchWrapper writeBatchWrapper;
+ private final RocksDBNativeMetricOptions metricOptions;
+
+ private final MetricGroup metricGroup;
+
+ /** The native metrics monitor. */
+ private RocksDBNativeMetricMonitor nativeMetricMonitor;
+
public RocksDBKeyedStateBackend(
String operatorIdentifier,
ClassLoader userCodeClassLoader,
@@ -255,7 +263,9 @@ public class RocksDBKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
boolean enableIncrementalCheckpointing,
LocalRecoveryConfig localRecoveryConfig,
RocksDBStateBackend.PriorityQueueStateType priorityQueueStateType,
- TtlTimeProvider ttlTimeProvider
+ TtlTimeProvider ttlTimeProvider,
+ RocksDBNativeMetricOptions metricOptions,
+ MetricGroup metricGroup
) throws IOException {
super(kvStateRegistry, keySerializer, userCodeClassLoader,
@@ -291,6 +301,9 @@ public class RocksDBKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
this.writeOptions = new WriteOptions().setDisableWAL(true);
+ this.metricOptions = metricOptions;
+ this.metricGroup = metricGroup;
+
switch (priorityQueueStateType) {
case HEAP:
this.priorityQueueFactory = new HeapPriorityQueueSetFactory(keyGroupRange, numberOfKeyGroups, 128);
@@ -358,6 +371,14 @@ public class RocksDBKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
return columnInfo != null ? columnInfo.f0 : null;
}
+ private void registerKvStateInformation(String columnFamilyName, Tuple2<ColumnFamilyHandle, RegisteredStateMetaInfoBase> registeredColumn) {
+ kvStateInformation.put(columnFamilyName, registeredColumn);
+
+ if (nativeMetricMonitor != null) {
+ nativeMetricMonitor.registerColumnFamily(columnFamilyName, registeredColumn.f0);
+ }
+ }
+
/**
* Should only be called by one thread, and only after all accesses to the DB happened.
*/
@@ -375,6 +396,13 @@ public class RocksDBKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
IOUtils.closeQuietly(writeBatchWrapper);
+ // Metric collection occurs on a background thread. When this method returns
+ // it is guaranteed that thr RocksDB reference has been invalidated
+ // and no more metric collection will be attempted against the database.
+ if (nativeMetricMonitor != null) {
+ nativeMetricMonitor.close();
+ }
+
// RocksDB's native memory management requires that *all* CFs (including default) are closed before the
// DB is closed. See:
// https://github.com/facebook/rocksdb/wiki/RocksJava-Basics#opening-a-database-with-column-families
@@ -606,6 +634,14 @@ public class RocksDBKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
Preconditions.checkState(1 + stateColumnFamilyDescriptors.size() == stateColumnFamilyHandles.size(),
"Not all requested column family handles have been created");
+ if (this.metricOptions.isEnabled()) {
+ this.nativeMetricMonitor = new RocksDBNativeMetricMonitor(
+ dbRef,
+ metricOptions,
+ metricGroup
+ );
+ }
+
return dbRef;
}
@@ -733,7 +769,7 @@ public class RocksDBKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
ColumnFamilyHandle columnFamily = rocksDBKeyedStateBackend.db.createColumnFamily(columnFamilyDescriptor);
registeredColumn = new Tuple2<>(columnFamily, stateMetaInfo);
- rocksDBKeyedStateBackend.kvStateInformation.put(stateMetaInfo.getName(), registeredColumn);
+ rocksDBKeyedStateBackend.registerKvStateInformation(stateMetaInfo.getName(), registeredColumn);
} else {
// TODO with eager state registration in place, check here for serializer migration strategies
@@ -1051,7 +1087,7 @@ public class RocksDBKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
columnFamilyHandle != null ? columnFamilyHandle : stateBackend.db.createColumnFamily(columnFamilyDescriptor),
stateMetaInfo);
- stateBackend.kvStateInformation.put(
+ stateBackend.registerKvStateInformation(
stateMetaInfoSnapshot.getName(),
registeredStateMetaInfoEntry);
}
@@ -1180,7 +1216,7 @@ public class RocksDBKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
RegisteredStateMetaInfoBase stateMetaInfo =
RegisteredStateMetaInfoBase.fromMetaInfoSnapshot(stateMetaInfoSnapshot);
- stateBackend.kvStateInformation.put(
+ stateBackend.registerKvStateInformation(
stateMetaInfoSnapshot.getName(),
new Tuple2<>(columnFamilyHandle, stateMetaInfo));
}
@@ -1291,6 +1327,7 @@ public class RocksDBKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
StreamStateHandle remoteFileHandle = entry.getValue();
copyStateDataHandleData(new Path(restoreInstancePath, stateHandleID.toString()), remoteFileHandle);
}
+
}
/**
@@ -1383,7 +1420,7 @@ public class RocksDBKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
ColumnFamilyHandle columnFamily = createColumnFamily(stateName);
stateInfo = Tuple2.of(columnFamily, newMetaInfo);
- kvStateInformation.put(stateDesc.getName(), stateInfo);
+ registerKvStateInformation(stateDesc.getName(), stateInfo);
}
return Tuple2.of(stateInfo.f0, newMetaInfo);
@@ -1567,7 +1604,7 @@ public class RocksDBKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
new RegisteredPriorityQueueStateBackendMetaInfo<>(stateName, byteOrderedElementSerializer);
metaInfoTuple = new Tuple2<>(columnFamilyHandle, metaInfo);
- kvStateInformation.put(stateName, metaInfoTuple);
+ registerKvStateInformation(stateName, metaInfoTuple);
} else {
// TODO we implement the simple way of supporting the current functionality, mimicking keyed state
// because this should be reworked in FLINK-9376 and then we should have a common algorithm over
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBNativeMetricMonitor.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBNativeMetricMonitor.java
new file mode 100644
index 0000000..4e3a526
--- /dev/null
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBNativeMetricMonitor.java
@@ -0,0 +1,170 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.contrib.streaming.state;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.metrics.Gauge;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.metrics.View;
+
+import org.rocksdb.ColumnFamilyHandle;
+import org.rocksdb.RocksDB;
+import org.rocksdb.RocksDBException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nonnull;
+import javax.annotation.concurrent.GuardedBy;
+
+import java.io.Closeable;
+import java.math.BigInteger;
+
+/**
+ * A monitor which pulls {{@link RocksDB}} native metrics
+ * and forwards them to Flink's metric group. All metrics are
+ * unsigned longs and are reported at the column family level.
+ */
+@Internal
+public class RocksDBNativeMetricMonitor implements Closeable {
+ private static final Logger LOG = LoggerFactory.getLogger(RocksDBNativeMetricMonitor.class);
+
+ private final RocksDBNativeMetricOptions options;
+
+ private final MetricGroup metricGroup;
+
+ private final Object lock;
+
+ @GuardedBy("lock")
+ private RocksDB rocksDB;
+
+ RocksDBNativeMetricMonitor(
+ @Nonnull RocksDB rocksDB,
+ @Nonnull RocksDBNativeMetricOptions options,
+ @Nonnull MetricGroup metricGroup
+ ) {
+ this.options = options;
+ this.metricGroup = metricGroup;
+ this.rocksDB = rocksDB;
+
+ this.lock = new Object();
+ }
+
+ /**
+ * Register gauges to pull native metrics for the column family.
+ * @param columnFamilyName group name for the new gauges
+ * @param handle native handle to the column family
+ */
+ void registerColumnFamily(String columnFamilyName, ColumnFamilyHandle handle) {
+ MetricGroup group = metricGroup.addGroup(columnFamilyName);
+
+ for (String property : options.getProperties()) {
+ RocksDBNativeMetricView gauge = new RocksDBNativeMetricView(handle, property);
+ group.gauge(property, gauge);
+ }
+ }
+
+ /**
+ * Updates the value of metricView if the reference is still valid.
+ */
+ private void setProperty(ColumnFamilyHandle handle, String property, RocksDBNativeMetricView metricView) {
+ if (metricView.isClosed()) {
+ return;
+ }
+ try {
+ synchronized (lock) {
+ if (rocksDB != null) {
+ long value = rocksDB.getLongProperty(handle, property);
+ metricView.setValue(value);
+ }
+ }
+ } catch (RocksDBException e) {
+ metricView.close();
+ LOG.warn("Failed to read native metric %s from RocksDB", property, e);
+ }
+ }
+
+ @Override
+ public void close() {
+ synchronized (lock) {
+ rocksDB = null;
+ }
+ }
+
+ /**
+ * A gauge which periodically pulls a RocksDB native metric
+ * for the specified column family / metric pair.
+ *
+ *<p><strong>Note</strong>: As the returned property is of type
+ * {@code uint64_t} on C++ side the returning value can be negative.
+ * Because java does not support unsigned long types, this gauge
+ * wraps the result in a {@link BigInteger}.
+ */
+ class RocksDBNativeMetricView implements Gauge<BigInteger>, View {
+ private final String property;
+
+ private final ColumnFamilyHandle handle;
+
+ private BigInteger bigInteger;
+
+ private boolean closed;
+
+ private RocksDBNativeMetricView(
+ @Nonnull ColumnFamilyHandle handle,
+ @Nonnull String property
+ ) {
+ this.handle = handle;
+ this.property = property;
+ this.bigInteger = BigInteger.ZERO;
+ this.closed = false;
+ }
+
+ public void setValue(long value) {
+ if (value >= 0L) {
+ bigInteger = BigInteger.valueOf(value);
+ } else {
+ int upper = (int) (value >>> 32);
+ int lower = (int) value;
+
+ bigInteger = BigInteger
+ .valueOf(Integer.toUnsignedLong(upper))
+ .shiftLeft(32)
+ .add(BigInteger.valueOf(Integer.toUnsignedLong(lower)));
+ }
+ }
+
+ public void close() {
+ closed = true;
+ }
+
+ public boolean isClosed() {
+ return closed;
+ }
+
+ @Override
+ public BigInteger getValue() {
+ return bigInteger;
+ }
+
+ @Override
+ public void update() {
+ setProperty(handle, property, this);
+ }
+ }
+}
+
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBNativeMetricOptions.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBNativeMetricOptions.java
new file mode 100644
index 0000000..f647e4c
--- /dev/null
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBNativeMetricOptions.java
@@ -0,0 +1,419 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.contrib.streaming.state;
+
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.Configuration;
+
+import java.io.Serializable;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Set;
+
+/**
+ * Enable which RocksDB metrics to forward to Flink's metrics reporter.
+ * All metrics report at the column family level and return unsigned long values.
+ *
+ * <p>Properties and doc comments are taken from RocksDB documentation. See
+ * <a href="https://github.com/facebook/rocksdb/blob/64324e329eb0a9b4e77241a425a1615ff524c7f1/include/rocksdb/db.h#L429">
+ * db.h</a> for more information.
+ */
+public class RocksDBNativeMetricOptions implements Serializable {
+
+ public static final ConfigOption<Boolean> MONITOR_NUM_IMMUTABLE_MEM_TABLES = ConfigOptions
+ .key(RocksDBProperty.NumImmutableMemTable.getConfigKey())
+ .defaultValue(false)
+ .withDescription("Monitor the number of immutable memtables in RocksDB.");
+
+ public static final ConfigOption<Boolean> MONITOR_MEM_TABLE_FLUSH_PENDING = ConfigOptions
+ .key(RocksDBProperty.MemTableFlushPending.getConfigKey())
+ .defaultValue(false)
+ .withDescription("Monitor the number of pending memtable flushes in RocksDB.");
+
+ public static final ConfigOption<Boolean> TRACK_COMPACTION_PENDING = ConfigOptions
+ .key(RocksDBProperty.CompactionPending.getConfigKey())
+ .defaultValue(false)
+ .withDescription("Track pending compactions in RocksDB. Returns 1 if a compaction is pending, 0 otherwise.");
+
+ public static final ConfigOption<Boolean> MONITOR_BACKGROUND_ERRORS = ConfigOptions
+ .key(RocksDBProperty.BackgroundErrors.getConfigKey())
+ .defaultValue(false)
+ .withDescription("Monitor the number of background errors in RocksDB.");
+
+ public static final ConfigOption<Boolean> MONITOR_CUR_SIZE_ACTIVE_MEM_TABLE = ConfigOptions
+ .key(RocksDBProperty.CurSizeActiveMemTable.getConfigKey())
+ .defaultValue(false)
+ .withDescription("Monitor the approximate size of the active memtable in bytes.");
+
+ public static final ConfigOption<Boolean> MONITOR_CUR_SIZE_ALL_MEM_TABLE = ConfigOptions
+ .key(RocksDBProperty.CurSizeAllMemTables.getConfigKey())
+ .defaultValue(false)
+ .withDescription("Monitor the approximate size of the active and unflushed immutable memtables" +
+ " in bytes.");
+
+ public static final ConfigOption<Boolean> MONITOR_SIZE_ALL_MEM_TABLES = ConfigOptions
+ .key(RocksDBProperty.SizeAllMemTables.getConfigKey())
+ .defaultValue(false)
+ .withDescription("Monitor the approximate size of the active, unflushed immutable, " +
+ "and pinned immutable memtables in bytes.");
+
+ public static final ConfigOption<Boolean> MONITOR_NUM_ENTRIES_ACTIVE_MEM_TABLE = ConfigOptions
+ .key(RocksDBProperty.NumEntriesActiveMemTable.getConfigKey())
+ .defaultValue(false)
+ .withDescription("Monitor the total number of entries in the active memtable.");
+
+ public static final ConfigOption<Boolean> MONITOR_NUM_ENTRIES_IMM_MEM_TABLES = ConfigOptions
+ .key(RocksDBProperty.NumEntriesImmMemTables.getConfigKey())
+ .defaultValue(false)
+ .withDescription("Monitor the total number of entries in the unflushed immutable memtables.");
+
+ public static final ConfigOption<Boolean> MONITOR_NUM_DELETES_ACTIVE_MEM_TABLE = ConfigOptions
+ .key(RocksDBProperty.NumDeletesActiveMemTable.getConfigKey())
+ .defaultValue(false)
+ .withDescription("Monitor the total number of delete entries in the active memtable.");
+
+ public static final ConfigOption<Boolean> MONITOR_NUM_DELETES_IMM_MEM_TABLE = ConfigOptions
+ .key(RocksDBProperty.NumDeletesImmMemTables.getConfigKey())
+ .defaultValue(false)
+ .withDescription("Monitor the total number of delete entries in the unflushed immutable memtables.");
+
+ public static final ConfigOption<Boolean> ESTIMATE_NUM_KEYS = ConfigOptions
+ .key(RocksDBProperty.EstimateNumKeys.getConfigKey())
+ .defaultValue(false)
+ .withDescription("Estimate the number of keys in RocksDB.");
+
+ public static final ConfigOption<Boolean> ESTIMATE_TABLE_READERS_MEM = ConfigOptions
+ .key(RocksDBProperty.EstimateTableReadersMem.getConfigKey())
+ .defaultValue(false)
+ .withDescription("Estimate the memory used for reading SST tables, excluding memory" +
+ " used in block cache (e.g.,filter and index blocks) in bytes.");
+
+ public static final ConfigOption<Boolean> MONITOR_NUM_SNAPSHOTS = ConfigOptions
+ .key(RocksDBProperty.NumSnapshots.getConfigKey())
+ .defaultValue(false)
+ .withDescription("Monitor the number of unreleased snapshots of the database.");
+
+ public static final ConfigOption<Boolean> MONITOR_NUM_LIVE_VERSIONS = ConfigOptions
+ .key(RocksDBProperty.NumLiveVersions.getConfigKey())
+ .defaultValue(false)
+ .withDescription("Monitor number of live versions. Version is an internal data structure. " +
+ "See RocksDB file version_set.h for details. More live versions often mean more SST files are held " +
+ "from being deleted, by iterators or unfinished compactions.");
+
+ public static final ConfigOption<Boolean> ESTIMATE_LIVE_DATA_SIZE = ConfigOptions
+ .key(RocksDBProperty.EstimateLiveDataSize.getConfigKey())
+ .defaultValue(false)
+ .withDescription("Estimate of the amount of live data in bytes.");
+
+ public static final ConfigOption<Boolean> MONITOR_TOTAL_SST_FILES_SIZE = ConfigOptions
+ .key(RocksDBProperty.TotalSstFilesSize.getConfigKey())
+ .defaultValue(false)
+ .withDescription("Monitor the total size (bytes) of all SST files." +
+ "WARNING: may slow down online queries if there are too many files.");
+
+ public static final ConfigOption<Boolean> ESTIMATE_PENDING_COMPACTION_BYTES = ConfigOptions
+ .key(RocksDBProperty.EstimatePendingCompactionBytes.getConfigKey())
+ .defaultValue(false)
+ .withDescription("Estimated total number of bytes compaction needs to rewrite to get all levels " +
+ "down to under target size. Not valid for other compactions than level-based.");
+
+ public static final ConfigOption<Boolean> MONITOR_NUM_RUNNING_COMPACTIONS = ConfigOptions
+ .key(RocksDBProperty.NumRunningCompactions.getConfigKey())
+ .defaultValue(false)
+ .withDescription("Monitor the number of currently running compactions.");
+
+ public static final ConfigOption<Boolean> MONITOR_NUM_RUNNING_FLUSHES = ConfigOptions
+ .key(RocksDBProperty.NumRunningFlushes.getConfigKey())
+ .defaultValue(false)
+ .withDescription("Monitor the number of currently running flushes.");
+
+ public static final ConfigOption<Boolean> MONITOR_ACTUAL_DELAYED_WRITE_RATE = ConfigOptions
+ .key(RocksDBProperty.ActualDelayedWriteRate.getConfigKey())
+ .defaultValue(false)
+ .withDescription("Monitor the current actual delayed write rate. 0 means no delay.");
+ /**
+ * Creates a {@link RocksDBNativeMetricOptions} based on an
+ * external configuration.
+ */
+ public static RocksDBNativeMetricOptions fromConfig(Configuration config) {
+ RocksDBNativeMetricOptions options = new RocksDBNativeMetricOptions();
+ if (config.getBoolean(MONITOR_NUM_IMMUTABLE_MEM_TABLES)) {
+ options.enableNumImmutableMemTable();
+ }
+
+ if (config.getBoolean(MONITOR_MEM_TABLE_FLUSH_PENDING)) {
+ options.enableMemTableFlushPending();
+ }
+
+ if (config.getBoolean(TRACK_COMPACTION_PENDING)) {
+ options.enableCompactionPending();
+ }
+
+ if (config.getBoolean(MONITOR_BACKGROUND_ERRORS)) {
+ options.enableBackgroundErrors();
+ }
+
+ if (config.getBoolean(MONITOR_CUR_SIZE_ACTIVE_MEM_TABLE)) {
+ options.enableCurSizeActiveMemTable();
+ }
+
+ if (config.getBoolean(MONITOR_CUR_SIZE_ALL_MEM_TABLE)) {
+ options.enableCurSizeAllMemTables();
+ }
+
+ if (config.getBoolean(MONITOR_SIZE_ALL_MEM_TABLES)) {
+ options.enableSizeAllMemTables();
+ }
+
+ if (config.getBoolean(MONITOR_NUM_ENTRIES_ACTIVE_MEM_TABLE)) {
+ options.enableNumEntriesActiveMemTable();
+ }
+
+ if (config.getBoolean(MONITOR_NUM_ENTRIES_IMM_MEM_TABLES)) {
+ options.enableNumEntriesImmMemTables();
+ }
+
+ if (config.getBoolean(MONITOR_NUM_DELETES_ACTIVE_MEM_TABLE)) {
+ options.enableNumDeletesActiveMemTable();
+ }
+
+ if (config.getBoolean(MONITOR_NUM_DELETES_IMM_MEM_TABLE)) {
+ options.enableNumDeletesImmMemTables();
+ }
+
+ if (config.getBoolean(ESTIMATE_NUM_KEYS)) {
+ options.enableEstimateNumKeys();
+ }
+
+ if (config.getBoolean(ESTIMATE_TABLE_READERS_MEM)) {
+ options.enableEstimateTableReadersMem();
+ }
+
+ if (config.getBoolean(MONITOR_NUM_SNAPSHOTS)) {
+ options.enableNumSnapshots();
+ }
+
+ if (config.getBoolean(MONITOR_NUM_LIVE_VERSIONS)) {
+ options.enableNumLiveVersions();
+ }
+
+ if (config.getBoolean(ESTIMATE_LIVE_DATA_SIZE)) {
+ options.enableEstimateLiveDataSize();
+ }
+
+ if (config.getBoolean(MONITOR_TOTAL_SST_FILES_SIZE)) {
+ options.enableTotalSstFilesSize();
+ }
+
+ if (config.getBoolean(ESTIMATE_PENDING_COMPACTION_BYTES)) {
+ options.enableEstimatePendingCompactionBytes();
+ }
+
+ if (config.getBoolean(MONITOR_NUM_RUNNING_COMPACTIONS)) {
+ options.enableNumRunningCompactions();
+ }
+
+ if (config.getBoolean(MONITOR_NUM_RUNNING_FLUSHES)) {
+ options.enableNumRunningFlushes();
+ }
+
+ if (config.getBoolean(MONITOR_ACTUAL_DELAYED_WRITE_RATE)) {
+ options.enableActualDelayedWriteRate();
+ }
+
+ return options;
+ }
+
+ private Set<String> properties;
+
+ public RocksDBNativeMetricOptions() {
+ this.properties = new HashSet<>();
+ }
+
+ /**
+ * Returns number of immutable memtables that have not yet been flushed.
+ */
+ public void enableNumImmutableMemTable() {
+ this.properties.add(RocksDBProperty.NumImmutableMemTable.getRocksDBProperty());
+ }
+
+ /**
+ * Returns 1 if a memtable flush is pending; otherwise, returns 0.
+ */
+ public void enableMemTableFlushPending() {
+ this.properties.add(RocksDBProperty.MemTableFlushPending.getRocksDBProperty());
+ }
+
+ /**
+ * Returns 1 if at least one compaction is pending; otherwise, returns 0.
+ */
+ public void enableCompactionPending() {
+ this.properties.add(RocksDBProperty.CompactionPending.getRocksDBProperty());
+ }
+
+ /**
+ * Returns accumulated number of background errors.
+ */
+ public void enableBackgroundErrors() {
+ this.properties.add(RocksDBProperty.BackgroundErrors.getRocksDBProperty());
+ }
+
+ /**
+ * Returns approximate size of active memtable (bytes).
+ */
+ public void enableCurSizeActiveMemTable() {
+ this.properties.add(RocksDBProperty.CurSizeActiveMemTable.getRocksDBProperty());
+ }
+
+ /**
+ * Returns approximate size of active and unflushed immutable memtables (bytes).
+ */
+ public void enableCurSizeAllMemTables() {
+ this.properties.add(RocksDBProperty.CurSizeAllMemTables.getRocksDBProperty());
+ }
+
+ /**
+ * Returns approximate size of active, unflushed immutable, and pinned immutable memtables (bytes).
+ */
+ public void enableSizeAllMemTables() {
+ this.properties.add(RocksDBProperty.SizeAllMemTables.getRocksDBProperty());
+ }
+
+ /**
+ * Returns total number of entries in the active memtable.
+ */
+ public void enableNumEntriesActiveMemTable() {
+ this.properties.add(RocksDBProperty.NumEntriesActiveMemTable.getRocksDBProperty());
+ }
+
+ /**
+ * Returns total number of entries in the unflushed immutable memtables.
+ */
+ public void enableNumEntriesImmMemTables() {
+ this.properties.add(RocksDBProperty.NumEntriesImmMemTables.getRocksDBProperty());
+ }
+
+ /**
+ * Returns total number of delete entries in the active memtable.
+ */
+ public void enableNumDeletesActiveMemTable() {
+ this.properties.add(RocksDBProperty.NumDeletesActiveMemTable.getRocksDBProperty());
+ }
+
+ /**
+ * Returns total number of delete entries in the unflushed immutable memtables.
+ */
+ public void enableNumDeletesImmMemTables() {
+ this.properties.add(RocksDBProperty.NumDeletesImmMemTables.getRocksDBProperty());
+ }
+
+ /**
+ * Returns estimated number of total keys in the active and unflushed immutable memtables and storage.
+ */
+ public void enableEstimateNumKeys() {
+ this.properties.add(RocksDBProperty.EstimateNumKeys.getRocksDBProperty());
+ }
+
+ /**
+ * Returns estimated memory used for reading SST tables, excluding memory
+ * used in block cache (e.g.,filter and index blocks).
+ */
+ public void enableEstimateTableReadersMem() {
+ this.properties.add(RocksDBProperty.EstimateTableReadersMem.getRocksDBProperty());
+ }
+
+ /**
+ * Returns number of unreleased snapshots of the database.
+ */
+ public void enableNumSnapshots() {
+ this.properties.add(RocksDBProperty.NumSnapshots.getRocksDBProperty());
+ }
+
+ /**
+ * Returns number of live versions. `Version`
+ * is an internal data structure. See version_set.h for details. More
+ * live versions often mean more SST files are held from being deleted,
+ * by iterators or unfinished compactions.
+ */
+ public void enableNumLiveVersions() {
+ this.properties.add(RocksDBProperty.NumLiveVersions.getRocksDBProperty());
+ }
+
+ /**
+ * Returns an estimate of the amount of live data in bytes.
+ */
+ public void enableEstimateLiveDataSize() {
+ this.properties.add(RocksDBProperty.EstimateLiveDataSize.getRocksDBProperty());
+ }
+
+ /**
+ * Returns total size (bytes) of all SST files.
+ * <strong>WARNING</strong>: may slow down online queries if there are too many files.
+ */
+ public void enableTotalSstFilesSize() {
+ this.properties.add(RocksDBProperty.TotalSstFilesSize.getRocksDBProperty());
+ }
+
+ /**
+ * Returns estimated total number of bytes compaction needs to rewrite to get all levels down
+ * to under target size. Not valid for other compactions than level-based.
+ */
+ public void enableEstimatePendingCompactionBytes() {
+ this.properties.add(RocksDBProperty.EstimatePendingCompactionBytes.getRocksDBProperty());
+ }
+
+ /**
+ * Returns the number of currently running compactions.
+ */
+ public void enableNumRunningCompactions() {
+ this.properties.add(RocksDBProperty.NumRunningCompactions.getRocksDBProperty());
+ }
+
+ /**
+ * Returns the number of currently running flushes.
+ */
+ public void enableNumRunningFlushes() {
+ this.properties.add(RocksDBProperty.NumRunningFlushes.getRocksDBProperty());
+ }
+
+ /**
+ * Returns the current actual delayed write rate. 0 means no delay.
+ */
+ public void enableActualDelayedWriteRate() {
+ this.properties.add(RocksDBProperty.ActualDelayedWriteRate.getRocksDBProperty());
+ }
+
+ /**
+ * @return the enabled RocksDB metrics
+ */
+ public Collection<String> getProperties() {
+ return Collections.unmodifiableCollection(properties);
+ }
+
+ /**
+ * {{@link RocksDBNativeMetricMonitor}} is enabled is any property is set.
+ *
+ * @return true if {{RocksDBNativeMetricMonitor}} should be enabled, false otherwise.
+ */
+ public boolean isEnabled() {
+ return !properties.isEmpty();
+ }
+}
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBProperty.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBProperty.java
new file mode 100644
index 0000000..ef680c9
--- /dev/null
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBProperty.java
@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.contrib.streaming.state;
+
+import org.apache.flink.annotation.Internal;
+
+import org.rocksdb.ColumnFamilyHandle;
+import org.rocksdb.RocksDB;
+
+/**
+ * {@link RocksDB} properties that can be queried by Flink's metrics reporter.
+ *
+ * <p>Note: Metrics properties are added in each new version of {@link RocksDB},
+ * when upgrading to a latter version consider updating this class with newly added
+ * properties.
+ */
+@Internal
+public enum RocksDBProperty {
+ NumImmutableMemTable("num-immutable-mem-table"),
+ MemTableFlushPending("mem-table-flush-pending"),
+ CompactionPending("compaction-pending"),
+ BackgroundErrors("background-errors"),
+ CurSizeActiveMemTable("cur-size-active-mem-table"),
+ CurSizeAllMemTables("cur-size-all-mem-tables"),
+ SizeAllMemTables("size-all-mem-tables"),
+ NumEntriesActiveMemTable("num-entries-active-mem-table"),
+ NumEntriesImmMemTables("num-entries-imm-mem-tables"),
+ NumDeletesActiveMemTable("num-deletes-active-mem-table"),
+ NumDeletesImmMemTables("num-deletes-imm-mem-tables"),
+ EstimateNumKeys("estimate-num-keys"),
+ EstimateTableReadersMem("estimate-table-readers-mem"),
+ NumSnapshots("num-snapshots"),
+ NumLiveVersions("num-live-versions"),
+ EstimateLiveDataSize("estimate-live-data-size"),
+ TotalSstFilesSize("total-sst-files-size"),
+ EstimatePendingCompactionBytes("estimate-pending-compaction-bytes"),
+ NumRunningCompactions("num-running-compactions"),
+ NumRunningFlushes("num-running-flushes"),
+ ActualDelayedWriteRate("actual-delayed-write-rate");
+
+ private static final String ROCKS_DB_PROPERTY_FORMAT = "rocksdb.%s";
+
+ private static final String CONFIG_KEY_FORMAT = "state.backend.rocksdb.metrics.%s";
+
+ private final String property;
+
+ RocksDBProperty(String property) {
+ this.property = property;
+ }
+
+ /**
+ * @return property string that can be used to query {@link RocksDB#getLongProperty(ColumnFamilyHandle, String)}.
+ */
+ public String getRocksDBProperty() {
+ return String.format(ROCKS_DB_PROPERTY_FORMAT, property);
+ }
+
+ /**
+ * @return key for enabling metric using {@link org.apache.flink.configuration.Configuration}.
+ */
+ public String getConfigKey() {
+ return String.format(CONFIG_KEY_FORMAT, property);
+ }
+}
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackend.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackend.java
index b8bd73c..794e221 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackend.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackend.java
@@ -24,6 +24,7 @@ import org.apache.flink.configuration.CheckpointingOptions;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.IllegalConfigurationException;
import org.apache.flink.core.fs.Path;
+import org.apache.flink.metrics.MetricGroup;
import org.apache.flink.runtime.execution.Environment;
import org.apache.flink.runtime.query.TaskKvStateRegistry;
import org.apache.flink.runtime.state.AbstractKeyedStateBackend;
@@ -122,6 +123,9 @@ public class RocksDBStateBackend extends AbstractStateBackend implements Configu
/** This determines the type of priority queue state. */
private final PriorityQueueStateType priorityQueueStateType;
+ /** The default rocksdb metrics options. */
+ private final RocksDBNativeMetricOptions defaultMetricOptions;
+
// -- runtime values, set on TaskManager when initializing / using the backend
/** Base paths for RocksDB directory, as initialized. */
@@ -236,6 +240,7 @@ public class RocksDBStateBackend extends AbstractStateBackend implements Configu
this.enableIncrementalCheckpointing = enableIncrementalCheckpointing;
// for now, we use still the heap-based implementation as default
this.priorityQueueStateType = PriorityQueueStateType.HEAP;
+ this.defaultMetricOptions = new RocksDBNativeMetricOptions();
}
/**
@@ -295,6 +300,9 @@ public class RocksDBStateBackend extends AbstractStateBackend implements Configu
}
}
+ // configure metric options
+ this.defaultMetricOptions = RocksDBNativeMetricOptions.fromConfig(config);
+
// copy remaining settings
this.predefinedOptions = original.predefinedOptions;
this.optionsFactory = original.optionsFactory;
@@ -412,7 +420,8 @@ public class RocksDBStateBackend extends AbstractStateBackend implements Configu
int numberOfKeyGroups,
KeyGroupRange keyGroupRange,
TaskKvStateRegistry kvStateRegistry,
- TtlTimeProvider ttlTimeProvider) throws IOException {
+ TtlTimeProvider ttlTimeProvider,
+ MetricGroup metricGroup) throws IOException {
// first, make sure that the RocksDB JNI library is loaded
// we do this explicitly here to have better error handling
@@ -445,7 +454,9 @@ public class RocksDBStateBackend extends AbstractStateBackend implements Configu
isIncrementalCheckpointsEnabled(),
localRecoveryConfig,
priorityQueueStateType,
- ttlTimeProvider);
+ ttlTimeProvider,
+ getMemoryWatcherOptions(),
+ metricGroup);
}
@Override
@@ -666,6 +677,15 @@ public class RocksDBStateBackend extends AbstractStateBackend implements Configu
return opt;
}
+ public RocksDBNativeMetricOptions getMemoryWatcherOptions() {
+ RocksDBNativeMetricOptions options = this.defaultMetricOptions;
+ if (optionsFactory != null) {
+ options = optionsFactory.createNativeMetricsOptions(options);
+ }
+
+ return options;
+ }
+
// ------------------------------------------------------------------------
// utilities
// ------------------------------------------------------------------------
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBNativeMetricMonitorTest.java b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBNativeMetricMonitorTest.java
new file mode 100644
index 0000000..7a817eb
--- /dev/null
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBNativeMetricMonitorTest.java
@@ -0,0 +1,207 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.contrib.streaming.state;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.MetricOptions;
+import org.apache.flink.metrics.Metric;
+import org.apache.flink.runtime.metrics.MetricRegistry;
+import org.apache.flink.runtime.metrics.groups.AbstractMetricGroup;
+import org.apache.flink.runtime.metrics.groups.GenericMetricGroup;
+import org.apache.flink.runtime.metrics.groups.UnregisteredMetricGroups;
+import org.apache.flink.runtime.metrics.scope.ScopeFormats;
+
+import org.junit.Assert;
+import org.junit.Rule;
+import org.junit.Test;
+import org.rocksdb.ColumnFamilyHandle;
+
+import javax.annotation.Nullable;
+
+import java.math.BigInteger;
+import java.util.ArrayList;
+
+/**
+ * validate native metric monitor.
+ */
+public class RocksDBNativeMetricMonitorTest {
+
+ private static final String OPERATOR_NAME = "dummy";
+
+ private static final String COLUMN_FAMILY_NAME = "column-family";
+
+ @Rule
+ public RocksDBResource rocksDBResource = new RocksDBResource();
+
+ @Test
+ public void testMetricMonitorLifecycle() throws Throwable {
+ //We use a local variable here to manually control the life-cycle.
+ // This allows us to verify that metrics do not try to access
+ // RocksDB after the monitor was closed.
+ RocksDBResource localRocksDBResource = new RocksDBResource();
+ localRocksDBResource.before();
+
+ SimpleMetricRegistry registry = new SimpleMetricRegistry();
+ GenericMetricGroup group = new GenericMetricGroup(
+ registry,
+ UnregisteredMetricGroups.createUnregisteredTaskMetricGroup(),
+ OPERATOR_NAME
+ );
+
+ RocksDBNativeMetricOptions options = new RocksDBNativeMetricOptions();
+ // always returns a non-zero
+ // value since empty memtables
+ // have overhead.
+ options.enableSizeAllMemTables();
+
+ RocksDBNativeMetricMonitor monitor = new RocksDBNativeMetricMonitor(
+ localRocksDBResource.getRocksDB(),
+ options,
+ group
+ );
+
+ ColumnFamilyHandle handle = localRocksDBResource.createNewColumnFamily(COLUMN_FAMILY_NAME);
+ monitor.registerColumnFamily(COLUMN_FAMILY_NAME, handle);
+
+ Assert.assertEquals("Failed to register metrics for column family", 1, registry.metrics.size());
+
+ RocksDBNativeMetricMonitor.RocksDBNativeMetricView view = registry.metrics.get(0);
+
+ view.update();
+
+ Assert.assertNotEquals("Failed to pull metric from RocksDB", BigInteger.ZERO, view.getValue());
+
+ view.setValue(0L);
+
+ //After the monitor is closed no metric should be accessing RocksDB anymore.
+ //If they do, then this test will likely fail with a segmentation fault.
+ monitor.close();
+
+ localRocksDBResource.after();
+
+ view.update();
+
+ Assert.assertEquals("Failed to release RocksDB reference", BigInteger.ZERO, view.getValue());
+ }
+
+ @Test
+ public void testReturnsUnsigned() {
+ SimpleMetricRegistry registry = new SimpleMetricRegistry();
+ GenericMetricGroup group = new GenericMetricGroup(
+ registry,
+ UnregisteredMetricGroups.createUnregisteredTaskMetricGroup(),
+ OPERATOR_NAME
+ );
+
+ RocksDBNativeMetricOptions options = new RocksDBNativeMetricOptions();
+ options.enableSizeAllMemTables();
+
+ RocksDBNativeMetricMonitor monitor = new RocksDBNativeMetricMonitor(
+ null,
+ options,
+ group
+ );
+
+ monitor.registerColumnFamily(COLUMN_FAMILY_NAME, null);
+ RocksDBNativeMetricMonitor.RocksDBNativeMetricView view = registry.metrics.get(0);
+
+ view.setValue(-1);
+ BigInteger result = view.getValue();
+
+ Assert.assertEquals("Failed to interpret RocksDB result as an unsigned long", 1, result.signum());
+ }
+
+ @Test
+ public void testClosedGaugesDontRead() {
+ SimpleMetricRegistry registry = new SimpleMetricRegistry();
+ GenericMetricGroup group = new GenericMetricGroup(
+ registry,
+ UnregisteredMetricGroups.createUnregisteredTaskMetricGroup(),
+ OPERATOR_NAME
+ );
+
+ RocksDBNativeMetricOptions options = new RocksDBNativeMetricOptions();
+ options.enableSizeAllMemTables();
+
+ RocksDBNativeMetricMonitor monitor = new RocksDBNativeMetricMonitor(
+ rocksDBResource.getRocksDB(),
+ options,
+ group
+ );
+
+ ColumnFamilyHandle handle = rocksDBResource.createNewColumnFamily(COLUMN_FAMILY_NAME);
+ monitor.registerColumnFamily(COLUMN_FAMILY_NAME, handle);
+
+ RocksDBNativeMetricMonitor.RocksDBNativeMetricView view = registry.metrics.get(0);
+
+ view.close();
+ view.update();
+
+ Assert.assertEquals("Closed gauge still queried RocksDB", BigInteger.ZERO, view.getValue());
+ }
+
+ static class SimpleMetricRegistry implements MetricRegistry {
+ ArrayList<RocksDBNativeMetricMonitor.RocksDBNativeMetricView> metrics = new ArrayList<>();
+
+ @Override
+ public char getDelimiter() {
+ return 0;
+ }
+
+ @Override
+ public char getDelimiter(int index) {
+ return 0;
+ }
+
+ @Override
+ public int getNumberReporters() {
+ return 0;
+ }
+
+ @Override
+ public void register(Metric metric, String metricName, AbstractMetricGroup group) {
+ if (metric instanceof RocksDBNativeMetricMonitor.RocksDBNativeMetricView) {
+ metrics.add((RocksDBNativeMetricMonitor.RocksDBNativeMetricView) metric);
+ }
+ }
+
+ @Override
+ public void unregister(Metric metric, String metricName, AbstractMetricGroup group) {
+
+ }
+
+ @Override
+ public ScopeFormats getScopeFormats() {
+ Configuration config = new Configuration();
+
+ config.setString(MetricOptions.SCOPE_NAMING_TM, "A");
+ config.setString(MetricOptions.SCOPE_NAMING_TM_JOB, "B");
+ config.setString(MetricOptions.SCOPE_NAMING_TASK, "C");
+ config.setString(MetricOptions.SCOPE_NAMING_OPERATOR, "D");
+
+ return ScopeFormats.fromConfig(config);
+ }
+
+ @Nullable
+ @Override
+ public String getMetricQueryServicePath() {
+ return null;
+ }
+ }
+}
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBNativeMetricOptionsTest.java b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBNativeMetricOptionsTest.java
new file mode 100644
index 0000000..f5becef
--- /dev/null
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBNativeMetricOptionsTest.java
@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.contrib.streaming.state;
+
+import org.apache.flink.configuration.Configuration;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+/**
+ * Test all native metrics can be set using configuration.
+ */
+public class RocksDBNativeMetricOptionsTest {
+ @Test
+ public void testNativeMetricsConfigurable() {
+ for (RocksDBProperty property : RocksDBProperty.values()) {
+ Configuration config = new Configuration();
+ config.setBoolean(property.getConfigKey(), true);
+
+ RocksDBNativeMetricOptions options = RocksDBNativeMetricOptions.fromConfig(config);
+
+ Assert.assertTrue(
+ String.format("Failed to enable native metrics with property %s", property.getConfigKey()),
+ options.isEnabled());
+
+ Assert.assertTrue(
+ String.format("Failed to enable native metric %s using config", property.getConfigKey()),
+ options.getProperties().contains(property.getRocksDBProperty())
+ );
+ }
+ }
+}
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBPropertyTest.java b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBPropertyTest.java
new file mode 100644
index 0000000..da8d07c
--- /dev/null
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBPropertyTest.java
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.contrib.streaming.state;
+
+import org.junit.Rule;
+import org.junit.Test;
+import org.rocksdb.ColumnFamilyHandle;
+import org.rocksdb.RocksDB;
+import org.rocksdb.RocksDBException;
+
+
+/**
+ * Validate RocksDB properties.
+ */
+public class RocksDBPropertyTest {
+
+ @Rule
+ public RocksDBResource rocksDBResource = new RocksDBResource();
+
+ @Test
+ public void testRocksDBPropertiesValid() throws RocksDBException {
+ RocksDB db = rocksDBResource.getRocksDB();
+ ColumnFamilyHandle handle = rocksDBResource.getDefaultColumnFamily();
+
+ for (RocksDBProperty property : RocksDBProperty.values()) {
+ try {
+ db.getLongProperty(handle, property.getRocksDBProperty());
+ } catch (RocksDBException e) {
+ throw new AssertionError(String.format("Invalid RocksDB property %s", property.getRocksDBProperty()), e);
+ }
+ }
+ }
+}
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackendTest.java b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackendTest.java
index 4916251..d7d6bde 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackendTest.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackendTest.java
@@ -25,6 +25,7 @@ import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeutils.base.IntSerializer;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.testutils.OneShotLatch;
+import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
import org.apache.flink.runtime.checkpoint.CheckpointOptions;
import org.apache.flink.runtime.execution.Environment;
import org.apache.flink.runtime.operators.testutils.DummyEnvironment;
@@ -250,7 +251,9 @@ public class RocksDBStateBackendTest extends StateBackendTestBase<RocksDBStateBa
enableIncrementalCheckpointing,
TestLocalRecoveryConfig.disabled(),
RocksDBStateBackend.PriorityQueueStateType.HEAP,
- TtlTimeProvider.DEFAULT);
+ TtlTimeProvider.DEFAULT,
+ new RocksDBNativeMetricOptions(),
+ new UnregisteredMetricsGroup());
verify(columnFamilyOptions, Mockito.times(1))
.setMergeOperatorName(RocksDBKeyedStateBackend.MERGE_OPERATOR_NAME);
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperator.java
index f3c2208..a63a797 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperator.java
@@ -252,7 +252,8 @@ public abstract class AbstractStreamOperator<OUT>
getClass().getSimpleName(),
this,
keySerializer,
- streamTaskCloseableRegistry);
+ streamTaskCloseableRegistry,
+ metrics);
this.operatorStateBackend = context.operatorStateBackend();
this.keyedStateBackend = context.keyedStateBackend();
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamTaskStateInitializer.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamTaskStateInitializer.java
index 4c91535..60e8182 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamTaskStateInitializer.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamTaskStateInitializer.java
@@ -20,6 +20,7 @@ package org.apache.flink.streaming.api.operators;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.core.fs.CloseableRegistry;
+import org.apache.flink.metrics.MetricGroup;
import org.apache.flink.runtime.jobgraph.OperatorID;
import javax.annotation.Nonnull;
@@ -41,6 +42,7 @@ public interface StreamTaskStateInitializer {
* @param keyContext the key context of the operator instance for which the context is created Cannot be null.
* @param keySerializer the key-serializer for the operator. Can be null.
* @param streamTaskCloseableRegistry the closeable registry to which created closeable objects will be registered.
+ * @param metricGroup the parent metric group for all statebackend metrics
* @return a context from which the given operator can initialize everything related to state.
* @throws Exception when something went wrong while creating the context.
*/
@@ -49,5 +51,6 @@ public interface StreamTaskStateInitializer {
@Nonnull String operatorClassName,
@Nonnull KeyContext keyContext,
@Nullable TypeSerializer<?> keySerializer,
- @Nonnull CloseableRegistry streamTaskCloseableRegistry) throws Exception;
+ @Nonnull CloseableRegistry streamTaskCloseableRegistry,
+ @Nonnull MetricGroup metricGroup) throws Exception;
}
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamTaskStateInitializerImpl.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamTaskStateInitializerImpl.java
index 64af993..42abb51 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamTaskStateInitializerImpl.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamTaskStateInitializerImpl.java
@@ -23,6 +23,7 @@ import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.core.fs.CloseableRegistry;
import org.apache.flink.core.fs.FSDataInputStream;
+import org.apache.flink.metrics.MetricGroup;
import org.apache.flink.runtime.checkpoint.PrioritizedOperatorSubtaskState;
import org.apache.flink.runtime.checkpoint.StateObjectCollection;
import org.apache.flink.runtime.execution.Environment;
@@ -106,7 +107,8 @@ public class StreamTaskStateInitializerImpl implements StreamTaskStateInitialize
@Nonnull String operatorClassName,
@Nonnull KeyContext keyContext,
@Nullable TypeSerializer<?> keySerializer,
- @Nonnull CloseableRegistry streamTaskCloseableRegistry) throws Exception {
+ @Nonnull CloseableRegistry streamTaskCloseableRegistry,
+ @Nonnull MetricGroup metricGroup) throws Exception {
TaskInfo taskInfo = environment.getTaskInfo();
OperatorSubtaskDescriptionText operatorSubtaskDescription =
@@ -134,7 +136,8 @@ public class StreamTaskStateInitializerImpl implements StreamTaskStateInitialize
keySerializer,
operatorIdentifierText,
prioritizedOperatorSubtaskStates,
- streamTaskCloseableRegistry);
+ streamTaskCloseableRegistry,
+ metricGroup);
// -------------- Operator State Backend --------------
operatorStateBackend = operatorStateBackend(
@@ -247,7 +250,8 @@ public class StreamTaskStateInitializerImpl implements StreamTaskStateInitialize
TypeSerializer<K> keySerializer,
String operatorIdentifierText,
PrioritizedOperatorSubtaskState prioritizedOperatorSubtaskStates,
- CloseableRegistry backendCloseableRegistry) throws Exception {
+ CloseableRegistry backendCloseableRegistry,
+ MetricGroup metricGroup) throws Exception {
if (keySerializer == null) {
return null;
@@ -272,7 +276,8 @@ public class StreamTaskStateInitializerImpl implements StreamTaskStateInitialize
taskInfo.getMaxNumberOfParallelSubtasks(),
keyGroupRange,
environment.getTaskKvStateRegistry(),
- TtlTimeProvider.DEFAULT),
+ TtlTimeProvider.DEFAULT,
+ metricGroup),
backendCloseableRegistry,
logDescription);
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/StateInitializationContextImplTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/StateInitializationContextImplTest.java
index 7d9dcd5..bcfe195 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/StateInitializationContextImplTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/StateInitializationContextImplTest.java
@@ -28,6 +28,7 @@ import org.apache.flink.core.memory.DataInputView;
import org.apache.flink.core.memory.DataInputViewStreamWrapper;
import org.apache.flink.core.memory.DataOutputView;
import org.apache.flink.core.memory.DataOutputViewStreamWrapper;
+import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
import org.apache.flink.runtime.checkpoint.JobManagerTaskRestore;
import org.apache.flink.runtime.checkpoint.OperatorSubtaskState;
import org.apache.flink.runtime.checkpoint.StateObjectCollection;
@@ -196,7 +197,8 @@ public class StateInitializationContextImplTest {
// notice that this essentially disables the previous test of the keyed stream because it was and is always
// consumed by the timer service.
IntSerializer.INSTANCE,
- closableRegistry);
+ closableRegistry,
+ new UnregisteredMetricsGroup());
this.initializationContext =
new StateInitializationContextImpl(
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/StreamTaskStateInitializerImplTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/StreamTaskStateInitializerImplTest.java
index 6233c4c..3dafa02 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/StreamTaskStateInitializerImplTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/StreamTaskStateInitializerImplTest.java
@@ -22,6 +22,8 @@ import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.common.typeutils.base.IntSerializer;
import org.apache.flink.core.fs.CloseableRegistry;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
import org.apache.flink.runtime.checkpoint.JobManagerTaskRestore;
import org.apache.flink.runtime.checkpoint.OperatorSubtaskState;
import org.apache.flink.runtime.checkpoint.TaskStateSnapshot;
@@ -91,7 +93,8 @@ public class StreamTaskStateInitializerImplTest {
streamOperator.getClass().getSimpleName(),
streamOperator,
typeSerializer,
- closeableRegistry);
+ closeableRegistry,
+ new UnregisteredMetricsGroup());
OperatorStateBackend operatorStateBackend = stateContext.operatorStateBackend();
AbstractKeyedStateBackend<?> keyedStateBackend = stateContext.keyedStateBackend();
@@ -140,7 +143,8 @@ public class StreamTaskStateInitializerImplTest {
TypeSerializer<K> keySerializer,
int numberOfKeyGroups, KeyGroupRange keyGroupRange,
TaskKvStateRegistry kvStateRegistry,
- TtlTimeProvider ttlTimeProvider) throws Exception {
+ TtlTimeProvider ttlTimeProvider,
+ MetricGroup metricGroup) throws Exception {
return mock(AbstractKeyedStateBackend.class);
}
@@ -192,7 +196,8 @@ public class StreamTaskStateInitializerImplTest {
streamOperator.getClass().getSimpleName(),
streamOperator,
typeSerializer,
- closeableRegistry);
+ closeableRegistry,
+ new UnregisteredMetricsGroup());
OperatorStateBackend operatorStateBackend = stateContext.operatorStateBackend();
AbstractKeyedStateBackend<?> keyedStateBackend = stateContext.keyedStateBackend();
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTerminationTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTerminationTest.java
index aebad54..0927698 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTerminationTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTerminationTest.java
@@ -23,6 +23,7 @@ import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.testutils.OneShotLatch;
+import org.apache.flink.metrics.MetricGroup;
import org.apache.flink.runtime.blob.BlobCacheService;
import org.apache.flink.runtime.blob.PermanentBlobCache;
import org.apache.flink.runtime.blob.TransientBlobCache;
@@ -263,7 +264,8 @@ public class StreamTaskTerminationTest extends TestLogger {
int numberOfKeyGroups,
KeyGroupRange keyGroupRange,
TaskKvStateRegistry kvStateRegistry,
- TtlTimeProvider ttlTimeProvider) {
+ TtlTimeProvider ttlTimeProvider,
+ MetricGroup metricGroup) {
return null;
}
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java
index 997e05c..bc14579 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java
@@ -1073,14 +1073,15 @@ public class StreamTaskTest extends TestLogger {
@Override
public StreamTaskStateInitializer createStreamTaskStateInitializer() {
final StreamTaskStateInitializer streamTaskStateManager = super.createStreamTaskStateInitializer();
- return (operatorID, operatorClassName, keyContext, keySerializer, closeableRegistry) -> {
+ return (operatorID, operatorClassName, keyContext, keySerializer, closeableRegistry, metricGroup) -> {
final StreamOperatorStateContext context = streamTaskStateManager.streamOperatorStateContext(
operatorID,
operatorClassName,
keyContext,
keySerializer,
- closeableRegistry);
+ closeableRegistry,
+ metricGroup);
return new StreamOperatorStateContext() {
@Override
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/TestSpyWrapperStateBackend.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/TestSpyWrapperStateBackend.java
index 214fab5..15f1ddd 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/TestSpyWrapperStateBackend.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/TestSpyWrapperStateBackend.java
@@ -20,6 +20,7 @@ package org.apache.flink.streaming.runtime.tasks;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.metrics.MetricGroup;
import org.apache.flink.runtime.execution.Environment;
import org.apache.flink.runtime.query.TaskKvStateRegistry;
import org.apache.flink.runtime.state.AbstractKeyedStateBackend;
@@ -55,7 +56,8 @@ public class TestSpyWrapperStateBackend extends AbstractStateBackend {
int numberOfKeyGroups,
KeyGroupRange keyGroupRange,
TaskKvStateRegistry kvStateRegistry,
- TtlTimeProvider ttlTimeProvider) throws IOException {
+ TtlTimeProvider ttlTimeProvider,
+ MetricGroup metricGroup) throws IOException {
return spy(delegate.createKeyedStateBackend(
env,
jobID,
@@ -64,7 +66,8 @@ public class TestSpyWrapperStateBackend extends AbstractStateBackend {
numberOfKeyGroups,
keyGroupRange,
kvStateRegistry,
- ttlTimeProvider));
+ ttlTimeProvider,
+ metricGroup));
}
@Override
diff --git a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/StateBackendITCase.java b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/StateBackendITCase.java
index f3bae16..d1543b5 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/StateBackendITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/StateBackendITCase.java
@@ -25,6 +25,7 @@ import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.metrics.MetricGroup;
import org.apache.flink.runtime.client.JobExecutionException;
import org.apache.flink.runtime.execution.Environment;
import org.apache.flink.runtime.query.TaskKvStateRegistry;
@@ -113,7 +114,8 @@ public class StateBackendITCase extends AbstractTestBase {
int numberOfKeyGroups,
KeyGroupRange keyGroupRange,
TaskKvStateRegistry kvStateRegistry,
- TtlTimeProvider ttlTimeProvider) throws IOException {
+ TtlTimeProvider ttlTimeProvider,
+ MetricGroup metricGroup) throws IOException {
throw new SuccessException();
}