You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ta...@apache.org on 2021/09/30 16:34:58 UTC
[flink] branch release-1.14 updated: [FLINK-23519][metrics] Expose
state name as varibale if enable state latency tracking
This is an automated email from the ASF dual-hosted git repository.
tangyun pushed a commit to branch release-1.14
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.14 by this push:
new b5fab64 [FLINK-23519][metrics] Expose state name as varibale if enable state latency tracking
b5fab64 is described below
commit b5fab64432c9839f956c9c169f617edf1026669f
Author: Yun Tang <my...@live.com>
AuthorDate: Wed Aug 25 16:41:35 2021 +0800
[FLINK-23519][metrics] Expose state name as varibale if enable state latency tracking
---
.../generated/state_backend_configuration.html | 6 ++++++
.../state_backend_latency_tracking_section.html | 6 ++++++
.../flink/configuration/StateBackendOptions.java | 8 +++++++
.../metrics/LatencyTrackingAggregatingState.java | 11 +++++++---
.../state/metrics/LatencyTrackingListState.java | 11 +++++++---
.../state/metrics/LatencyTrackingMapState.java | 11 +++++++---
.../metrics/LatencyTrackingReducingState.java | 11 +++++++---
.../state/metrics/LatencyTrackingStateConfig.java | 25 +++++++++++++++++++---
.../state/metrics/LatencyTrackingValueState.java | 11 +++++++---
.../state/metrics/StateLatencyMetricBase.java | 12 +++++++++--
10 files changed, 92 insertions(+), 20 deletions(-)
diff --git a/docs/layouts/shortcodes/generated/state_backend_configuration.html b/docs/layouts/shortcodes/generated/state_backend_configuration.html
index ac1552f..54bff1d 100644
--- a/docs/layouts/shortcodes/generated/state_backend_configuration.html
+++ b/docs/layouts/shortcodes/generated/state_backend_configuration.html
@@ -32,5 +32,11 @@
<td>Integer</td>
<td>The sample interval of latency track once 'state.backend.latency-track.keyed-state-enabled' is enabled. The default value is 100, which means we would track the latency every 100 access requests.</td>
</tr>
+ <tr>
+ <td><h5>state.backend.latency-track.state-name-as-variable</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Whether to expose state name as a variable if tracking latency.</td>
+ </tr>
</tbody>
</table>
diff --git a/docs/layouts/shortcodes/generated/state_backend_latency_tracking_section.html b/docs/layouts/shortcodes/generated/state_backend_latency_tracking_section.html
index 3c5e877..d5cdf2f 100644
--- a/docs/layouts/shortcodes/generated/state_backend_latency_tracking_section.html
+++ b/docs/layouts/shortcodes/generated/state_backend_latency_tracking_section.html
@@ -26,5 +26,11 @@
<td>Integer</td>
<td>The sample interval of latency track once 'state.backend.latency-track.keyed-state-enabled' is enabled. The default value is 100, which means we would track the latency every 100 access requests.</td>
</tr>
+ <tr>
+ <td><h5>state.backend.latency-track.state-name-as-variable</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Whether to expose state name as a variable if tracking latency.</td>
+ </tr>
</tbody>
</table>
diff --git a/flink-core/src/main/java/org/apache/flink/configuration/StateBackendOptions.java b/flink-core/src/main/java/org/apache/flink/configuration/StateBackendOptions.java
index e371059..bb179f8 100644
--- a/flink-core/src/main/java/org/apache/flink/configuration/StateBackendOptions.java
+++ b/flink-core/src/main/java/org/apache/flink/configuration/StateBackendOptions.java
@@ -88,4 +88,12 @@ public class StateBackendOptions {
.defaultValue(128)
.withDescription(
"Defines the number of measured latencies to maintain at each state access operation.");
+
+ @Documentation.Section(Documentation.Sections.STATE_BACKEND_LATENCY_TRACKING)
+ public static final ConfigOption<Boolean> LATENCY_TRACK_STATE_NAME_AS_VARIABLE =
+ ConfigOptions.key("state.backend.latency-track.state-name-as-variable")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Whether to expose state name as a variable if tracking latency.");
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/metrics/LatencyTrackingAggregatingState.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/metrics/LatencyTrackingAggregatingState.java
index 25802d4..34d232f 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/metrics/LatencyTrackingAggregatingState.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/metrics/LatencyTrackingAggregatingState.java
@@ -51,7 +51,8 @@ class LatencyTrackingAggregatingState<K, N, IN, ACC, OUT>
stateName,
latencyTrackingStateConfig.getMetricGroup(),
latencyTrackingStateConfig.getSampleInterval(),
- latencyTrackingStateConfig.getHistorySize()));
+ latencyTrackingStateConfig.getHistorySize(),
+ latencyTrackingStateConfig.isStateNameAsVariable()));
}
@Override
@@ -108,8 +109,12 @@ class LatencyTrackingAggregatingState<K, N, IN, ACC, OUT>
private int mergeNamespaceCount = 0;
private AggregatingStateLatencyMetrics(
- String stateName, MetricGroup metricGroup, int sampleInterval, int historySize) {
- super(stateName, metricGroup, sampleInterval, historySize);
+ String stateName,
+ MetricGroup metricGroup,
+ int sampleInterval,
+ int historySize,
+ boolean stateNameAsVariable) {
+ super(stateName, metricGroup, sampleInterval, historySize, stateNameAsVariable);
}
int getGetCount() {
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/metrics/LatencyTrackingListState.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/metrics/LatencyTrackingListState.java
index ac24b53..21e4edd 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/metrics/LatencyTrackingListState.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/metrics/LatencyTrackingListState.java
@@ -50,7 +50,8 @@ class LatencyTrackingListState<K, N, T>
stateName,
latencyTrackingStateConfig.getMetricGroup(),
latencyTrackingStateConfig.getSampleInterval(),
- latencyTrackingStateConfig.getHistorySize()));
+ latencyTrackingStateConfig.getHistorySize(),
+ latencyTrackingStateConfig.isStateNameAsVariable()));
}
@Override
@@ -131,8 +132,12 @@ class LatencyTrackingListState<K, N, T>
private int mergeNamespaceCount = 0;
private ListStateLatencyMetrics(
- String stateName, MetricGroup metricGroup, int sampleInterval, int historySize) {
- super(stateName, metricGroup, sampleInterval, historySize);
+ String stateName,
+ MetricGroup metricGroup,
+ int sampleInterval,
+ int historySize,
+ boolean stateNameAsVariable) {
+ super(stateName, metricGroup, sampleInterval, historySize, stateNameAsVariable);
}
int getGetCount() {
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/metrics/LatencyTrackingMapState.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/metrics/LatencyTrackingMapState.java
index cffa790..6c1eb96 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/metrics/LatencyTrackingMapState.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/metrics/LatencyTrackingMapState.java
@@ -51,7 +51,8 @@ class LatencyTrackingMapState<K, N, UK, UV>
stateName,
latencyTrackingStateConfig.getMetricGroup(),
latencyTrackingStateConfig.getSampleInterval(),
- latencyTrackingStateConfig.getHistorySize()));
+ latencyTrackingStateConfig.getHistorySize(),
+ latencyTrackingStateConfig.isStateNameAsVariable()));
}
@Override
@@ -243,8 +244,12 @@ class LatencyTrackingMapState<K, N, UK, UV>
private int iteratorNextCount = 0;
private MapStateLatencyMetrics(
- String stateName, MetricGroup metricGroup, int sampleInterval, int historySize) {
- super(stateName, metricGroup, sampleInterval, historySize);
+ String stateName,
+ MetricGroup metricGroup,
+ int sampleInterval,
+ int historySize,
+ boolean stateNameAsVariable) {
+ super(stateName, metricGroup, sampleInterval, historySize, stateNameAsVariable);
}
int getGetCount() {
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/metrics/LatencyTrackingReducingState.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/metrics/LatencyTrackingReducingState.java
index 821ba68..2e80ebc 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/metrics/LatencyTrackingReducingState.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/metrics/LatencyTrackingReducingState.java
@@ -49,7 +49,8 @@ class LatencyTrackingReducingState<K, N, T>
stateName,
latencyTrackingStateConfig.getMetricGroup(),
latencyTrackingStateConfig.getSampleInterval(),
- latencyTrackingStateConfig.getHistorySize()));
+ latencyTrackingStateConfig.getHistorySize(),
+ latencyTrackingStateConfig.isStateNameAsVariable()));
}
@Override
@@ -105,8 +106,12 @@ class LatencyTrackingReducingState<K, N, T>
private int mergeNamespaceCount = 0;
ReducingStateLatencyMetrics(
- String stateName, MetricGroup metricGroup, int sampleInterval, int historySize) {
- super(stateName, metricGroup, sampleInterval, historySize);
+ String stateName,
+ MetricGroup metricGroup,
+ int sampleInterval,
+ int historySize,
+ boolean stateNameAsVariable) {
+ super(stateName, metricGroup, sampleInterval, historySize, stateNameAsVariable);
}
int getGetCount() {
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/metrics/LatencyTrackingStateConfig.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/metrics/LatencyTrackingStateConfig.java
index a2fb416..d4425e1 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/metrics/LatencyTrackingStateConfig.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/metrics/LatencyTrackingStateConfig.java
@@ -35,9 +35,14 @@ public class LatencyTrackingStateConfig {
private final boolean enabled;
private final int sampleInterval;
private final int historySize;
+ private final boolean stateNameAsVariable;
LatencyTrackingStateConfig(
- MetricGroup metricGroup, boolean enabled, int sampleInterval, int historySize) {
+ MetricGroup metricGroup,
+ boolean enabled,
+ int sampleInterval,
+ int historySize,
+ boolean stateNameAsVariable) {
if (enabled) {
Preconditions.checkNotNull(
metricGroup, "Metric group cannot be null if latency tracking is enabled.");
@@ -47,6 +52,7 @@ public class LatencyTrackingStateConfig {
this.enabled = enabled;
this.sampleInterval = sampleInterval;
this.historySize = historySize;
+ this.stateNameAsVariable = stateNameAsVariable;
}
public boolean isEnabled() {
@@ -65,6 +71,10 @@ public class LatencyTrackingStateConfig {
return sampleInterval;
}
+ public boolean isStateNameAsVariable() {
+ return stateNameAsVariable;
+ }
+
public static LatencyTrackingStateConfig disabled() {
return newBuilder().setEnabled(false).build();
}
@@ -80,6 +90,8 @@ public class LatencyTrackingStateConfig {
private int sampleInterval =
StateBackendOptions.LATENCY_TRACK_SAMPLE_INTERVAL.defaultValue();
private int historySize = StateBackendOptions.LATENCY_TRACK_HISTORY_SIZE.defaultValue();
+ private boolean stateNameAsVariable =
+ StateBackendOptions.LATENCY_TRACK_STATE_NAME_AS_VARIABLE.defaultValue();
private MetricGroup metricGroup;
public Builder setEnabled(boolean enabled) {
@@ -97,6 +109,11 @@ public class LatencyTrackingStateConfig {
return this;
}
+ public Builder setStateNameAsVariable(boolean stateNameAsVariable) {
+ this.stateNameAsVariable = stateNameAsVariable;
+ return this;
+ }
+
public Builder setMetricGroup(MetricGroup metricGroup) {
this.metricGroup = metricGroup;
return this;
@@ -106,13 +123,15 @@ public class LatencyTrackingStateConfig {
this.setEnabled(config.get(StateBackendOptions.LATENCY_TRACK_ENABLED))
.setSampleInterval(
config.get(StateBackendOptions.LATENCY_TRACK_SAMPLE_INTERVAL))
- .setHistorySize(config.get(StateBackendOptions.LATENCY_TRACK_HISTORY_SIZE));
+ .setHistorySize(config.get(StateBackendOptions.LATENCY_TRACK_HISTORY_SIZE))
+ .setStateNameAsVariable(
+ config.get(StateBackendOptions.LATENCY_TRACK_STATE_NAME_AS_VARIABLE));
return this;
}
public LatencyTrackingStateConfig build() {
return new LatencyTrackingStateConfig(
- metricGroup, enabled, sampleInterval, historySize);
+ metricGroup, enabled, sampleInterval, historySize, stateNameAsVariable);
}
}
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/metrics/LatencyTrackingValueState.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/metrics/LatencyTrackingValueState.java
index e80cfcc..fa008bf 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/metrics/LatencyTrackingValueState.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/metrics/LatencyTrackingValueState.java
@@ -49,7 +49,8 @@ class LatencyTrackingValueState<K, N, T>
stateName,
latencyTrackingStateConfig.getMetricGroup(),
latencyTrackingStateConfig.getSampleInterval(),
- latencyTrackingStateConfig.getHistorySize()));
+ latencyTrackingStateConfig.getHistorySize(),
+ latencyTrackingStateConfig.isStateNameAsVariable()));
}
@Override
@@ -81,8 +82,12 @@ class LatencyTrackingValueState<K, N, T>
private int updateCount = 0;
private ValueStateLatencyMetrics(
- String stateName, MetricGroup metricGroup, int sampleInterval, int historySize) {
- super(stateName, metricGroup, sampleInterval, historySize);
+ String stateName,
+ MetricGroup metricGroup,
+ int sampleInterval,
+ int historySize,
+ boolean stateNameAsVariable) {
+ super(stateName, metricGroup, sampleInterval, historySize, stateNameAsVariable);
}
int getGetCount() {
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/metrics/StateLatencyMetricBase.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/metrics/StateLatencyMetricBase.java
index 1450f6a..106032f 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/metrics/StateLatencyMetricBase.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/metrics/StateLatencyMetricBase.java
@@ -28,6 +28,7 @@ import java.util.function.Supplier;
/** Base class of state latency metric which counts and histogram the state metric. */
class StateLatencyMetricBase implements AutoCloseable {
+ protected static final String STATE_NAME_KEY = "state_name";
protected static final String STATE_CLEAR_LATENCY = "stateClearLatency";
private final MetricGroup metricGroup;
private final int sampleInterval;
@@ -36,8 +37,15 @@ class StateLatencyMetricBase implements AutoCloseable {
private int clearCount = 0;
StateLatencyMetricBase(
- String stateName, MetricGroup metricGroup, int sampleInterval, int historySize) {
- this.metricGroup = metricGroup.addGroup(stateName);
+ String stateName,
+ MetricGroup metricGroup,
+ int sampleInterval,
+ int historySize,
+ boolean stateNameAsVariable) {
+ this.metricGroup =
+ stateNameAsVariable
+ ? metricGroup.addGroup(STATE_NAME_KEY, stateName)
+ : metricGroup.addGroup(stateName);
this.sampleInterval = sampleInterval;
this.histogramMetrics = new HashMap<>();
this.histogramSupplier = () -> new DescriptiveStatisticsHistogram(historySize);