[ https://issues.apache.org/jira/browse/FLINK-10423?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16656477#comment-16656477 ]
ASF GitHub Bot commented on FLINK-10423: ---------------------------------------- zentol closed pull request #6814: [FLINK-10423][rocksdb][metrics] rocksdb native metrics monitor URL: https://github.com/apache/flink/pull/6814 This is a PR merged from a forked repository. As GitHub hides the original diff on merge, it is displayed below for the sake of provenance: As this is a foreign pull request (from a fork), the diff is supplied below (as it won't show otherwise due to GitHub magic): 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 00000000000..78fc523b35b --- /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 85c60a67a22..04efdef26fc 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 d370e56d695..267b670d15a 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 4e09b345e10..b95d2b7e4db 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 176e55a7d8c..fff8f02285f 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.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 void testListSerialization() throws Exception { 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 void testMapSerialization() throws Exception { 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 adcf3aefb7c..bfadb75cfe6 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.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 void notifyKvStateUnregistered(JobID jobId, 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 d397a88c214..6c72acd888e 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 @@ 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 @@ 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 2775b71ac72..4235f762ff1 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 @@ 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 @@ 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 @@ 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 f5a86e1ac51..135030e88d6 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.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 CheckpointStorage createCheckpointStorage(JobID jobId) throws IOException 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 1c464d7fa97..4cc3a2c7376 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.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 OperatorStateBackend createOperatorStateBackend( 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 ffebc52f8a0..4b5ae06f217 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.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 CheckpointStorage createCheckpointStorage(JobID jobId) throws IOException 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 ccfafec806f..f88e6d7d31b 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.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 @@ 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 a8d49dde3ff..8ed84c031a1 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 @@ 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 CheckpointStreamFactory resolveCheckpointStorageLocation(long checkpointI 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 CheckpointStreamFactory resolveCheckpointStorageLocation(long checkpointI 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 84db465f12c..32045804aab 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 @@ */ 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 42a1e26b8b5..c56bd20b580 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.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 @@ /** 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 RocksDBKeyedStateBackend( 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 RocksDBKeyedStateBackend( 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 ColumnFamilyHandle getColumnFamilyHandle(String state) { 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 void dispose() { 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 @@ private RocksDB openDB( 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 @@ private void restoreKVStateMetaData() throws IOException, StateMigrationExceptio 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 @@ private ColumnFamilyHandle getOrRegisterColumnFamilyHandle( columnFamilyHandle != null ? columnFamilyHandle : stateBackend.db.createColumnFamily(columnFamilyDescriptor), stateMetaInfo); - stateBackend.kvStateInformation.put( + stateBackend.registerKvStateInformation( stateMetaInfoSnapshot.getName(), registeredStateMetaInfoEntry); } @@ -1180,7 +1216,7 @@ private void restoreLocalStateIntoFullInstance( RegisteredStateMetaInfoBase stateMetaInfo = RegisteredStateMetaInfoBase.fromMetaInfoSnapshot(stateMetaInfoSnapshot); - stateBackend.kvStateInformation.put( + stateBackend.registerKvStateInformation( stateMetaInfoSnapshot.getName(), new Tuple2<>(columnFamilyHandle, stateMetaInfo)); } @@ -1291,6 +1327,7 @@ private void transferAllDataFromStateHandles( StreamStateHandle remoteFileHandle = entry.getValue(); copyStateDataHandleData(new Path(restoreInstancePath, stateHandleID.toString()), remoteFileHandle); } + } /** @@ -1383,7 +1420,7 @@ private void copyStateDataHandleData( 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 static RocksIteratorWrapper getRocksIterator( 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 00000000000..4e3a526dc23 --- /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 00000000000..f647e4c153f --- /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 00000000000..ef680c90075 --- /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 b8bd73c08e1..794e22160a2 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.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 @@ /** 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 RocksDBStateBackend(StateBackend checkpointStreamBackend, TernaryBoolean 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 @@ private RocksDBStateBackend(RocksDBStateBackend original, Configuration config) } } + // configure metric options + this.defaultMetricOptions = RocksDBNativeMetricOptions.fromConfig(config); + // copy remaining settings this.predefinedOptions = original.predefinedOptions; this.optionsFactory = original.optionsFactory; @@ -412,7 +420,8 @@ public CheckpointStorage createCheckpointStorage(JobID jobId) throws IOException 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 CheckpointStorage createCheckpointStorage(JobID jobId) throws IOException isIncrementalCheckpointsEnabled(), localRecoveryConfig, priorityQueueStateType, - ttlTimeProvider); + ttlTimeProvider, + getMemoryWatcherOptions(), + metricGroup); } @Override @@ -666,6 +677,15 @@ public ColumnFamilyOptions getColumnOptions() { 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 00000000000..7a817ebb0e1 --- /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 00000000000..f5becef43b8 --- /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 00000000000..da8d07c9434 --- /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 4916251fc1f..d7d6bdea879 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.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 void testCorrectMergeOperatorSet() throws IOException { 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 f3c22080ab7..a63a7971679 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 final void initializeState() throws Exception { 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 4c9153506be..60e81828c54 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 @@ 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 @@ * @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 @@ StreamOperatorStateContext streamOperatorStateContext( @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 64af9939834..42abb5139bf 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.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 StreamOperatorStateContext streamOperatorStateContext( @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 StreamOperatorStateContext streamOperatorStateContext( keySerializer, operatorIdentifierText, prioritizedOperatorSubtaskStates, - streamTaskCloseableRegistry); + streamTaskCloseableRegistry, + metricGroup); // -------------- Operator State Backend -------------- operatorStateBackend = operatorStateBackend( @@ -247,7 +250,8 @@ protected OperatorStateBackend operatorStateBackend( 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 @@ protected OperatorStateBackend operatorStateBackend( 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 7d9dcd5ef11..bcfe195f340 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.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 void setUp() throws Exception { // 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 6233c4c12c4..3dafa024f8b 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.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 void testNoRestore() throws Exception { streamOperator.getClass().getSimpleName(), streamOperator, typeSerializer, - closeableRegistry); + closeableRegistry, + new UnregisteredMetricsGroup()); OperatorStateBackend operatorStateBackend = stateContext.operatorStateBackend(); AbstractKeyedStateBackend<?> keyedStateBackend = stateContext.keyedStateBackend(); @@ -140,7 +143,8 @@ public CheckpointStorage createCheckpointStorage(JobID jobId) throws IOException 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 OperatorStateBackend createOperatorStateBackend( 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 aebad543d06..09276986c58 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.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 CheckpointStorage createCheckpointStorage(JobID jobId) throws IOException 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 997e05c5bf4..bc145799fb3 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 @@ protected void cancelTask() throws Exception {} @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 214fab5b5a6..15f1ddd29f7 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 @@ 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 TestSpyWrapperStateBackend(AbstractStateBackend delegate) { 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 TestSpyWrapperStateBackend(AbstractStateBackend delegate) { 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 f3bae166a03..d1543b5765b 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.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 CheckpointStorage createCheckpointStorage(JobID jobId) throws IOException int numberOfKeyGroups, KeyGroupRange keyGroupRange, TaskKvStateRegistry kvStateRegistry, - TtlTimeProvider ttlTimeProvider) throws IOException { + TtlTimeProvider ttlTimeProvider, + MetricGroup metricGroup) throws IOException { throw new SuccessException(); } ---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org > Forward RocksDB native metrics to Flink metrics reporter > --------------------------------------------------------- > > Key: FLINK-10423 > URL: https://issues.apache.org/jira/browse/FLINK-10423 > Project: Flink > Issue Type: New Feature > Components: Metrics, State Backends, Checkpointing > Reporter: Seth Wiesman > Assignee: Seth Wiesman > Priority: Major > Labels: pull-request-available > Fix For: 1.7.0 > > > RocksDB contains a number of metrics at the column family level about current > memory usage, open memtables, etc that would be useful to users wishing > greater insight what rocksdb is doing. This work is inspired heavily by the > comments on this rocksdb issue thread > (https://github.com/facebook/rocksdb/issues/3216#issuecomment-348779233) -- This message was sent by Atlassian JIRA (v7.6.3#76005)