Skip to content

Commit 87e4508

Browse files
cirquarecirquare
authored
Merge pull request #182 from linkedin/yf186035/beam-2.75-phase4b-global-metrics
Publish process-wide metrics in Flink runner
2 parents 6acd56b + 843d972 commit 87e4508

3 files changed

Lines changed: 154 additions & 0 deletions

File tree

‎runners/flink/src/main/java/org/apache/beam/runners/flink/metrics/FlinkMetricContainerBase.java‎

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,8 @@
3232
import org.apache.beam.sdk.metrics.MetricQueryResults;
3333
import org.apache.beam.sdk.metrics.MetricResult;
3434
import org.apache.beam.sdk.metrics.MetricResults;
35+
import org.apache.beam.sdk.metrics.MetricsContainer;
36+
import org.apache.beam.sdk.metrics.MetricsEnvironment;
3537
import org.apache.beam.sdk.metrics.MetricsFilter;
3638
import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
3739
import org.apache.flink.configuration.GlobalConfiguration;
@@ -49,12 +51,15 @@
4951
*/
5052
abstract class FlinkMetricContainerBase {
5153

54+
private static final String PROCESS_WIDE_METRICS_STEP_NAME = "__process_wide_metrics";
55+
5256
private static final String METRIC_KEY_SEPARATOR =
5357
GlobalConfiguration.loadConfiguration()
5458
.getOptional(MetricOptions.SCOPE_DELIMITER)
5559
.orElseGet(MetricOptions.SCOPE_DELIMITER::defaultValue);
5660

5761
protected final MetricsContainerStepMap metricsContainers;
62+
private final MetricsContainerImpl processWideMetricsContainer;
5863
private final Map<String, Counter> flinkCounterCache;
5964
private final Map<String, FlinkDistributionGauge> flinkDistributionGaugeCache;
6065
private final Map<String, FlinkGauge> flinkGaugeCache;
@@ -65,6 +70,8 @@ public FlinkMetricContainerBase(MetricGroup metricGroup) {
6570
this.flinkDistributionGaugeCache = new HashMap<>();
6671
this.flinkGaugeCache = new HashMap<>();
6772
this.metricsContainers = new MetricsContainerStepMap();
73+
this.processWideMetricsContainer = MetricsContainerImpl.createProcessWideContainer();
74+
GlobalMetricsUtils.setProcessWideMetricsContainer(processWideMetricsContainer);
6875
this.metricGroup = metricGroup;
6976
}
7077

@@ -96,6 +103,23 @@ void updateMetrics(String stepName) {
96103
updateCounters(metricQueryResults.getCounters());
97104
updateDistributions(metricQueryResults.getDistributions());
98105
updateGauge(metricQueryResults.getGauges());
106+
updateProcessWideMetrics();
107+
}
108+
109+
private void updateProcessWideMetrics() {
110+
MetricsContainer processWideContainer = MetricsEnvironment.getProcessWideContainer();
111+
if (processWideContainer != processWideMetricsContainer) {
112+
return;
113+
}
114+
115+
MetricsContainerStepMap processWideMetrics = new MetricsContainerStepMap();
116+
processWideMetrics.update(PROCESS_WIDE_METRICS_STEP_NAME, processWideMetricsContainer);
117+
MetricQueryResults metricQueryResults =
118+
asAttemptedOnlyMetricResults(processWideMetrics)
119+
.queryMetrics(MetricsFilter.builder().addStep(PROCESS_WIDE_METRICS_STEP_NAME).build());
120+
updateCounters(metricQueryResults.getCounters());
121+
updateDistributions(metricQueryResults.getDistributions());
122+
updateGauge(metricQueryResults.getGauges());
99123
}
100124

101125
private void updateCounters(Iterable<MetricResult<Long>> counters) {

‎runners/flink/src/main/java/org/apache/beam/runners/flink/metrics/GlobalMetricsUtils.java‎

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,32 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing, software
13+
* distributed under the License is distributed on an "AS IS" BASIS,
14+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15+
* See the License for the specific language governing permissions and
16+
* limitations under the License.
17+
*/
18+
package org.apache.beam.runners.flink.metrics;
19+
20+
import org.apache.beam.runners.core.metrics.MetricsContainerImpl;
21+
import org.apache.beam.sdk.metrics.MetricsEnvironment;
22+
23+
/** Utility class for managing process-wide Beam metrics in a Flink environment. */
24+
final class GlobalMetricsUtils {
25+
private GlobalMetricsUtils() {}
26+
27+
static synchronized void setProcessWideMetricsContainer(MetricsContainerImpl container) {
28+
if (MetricsEnvironment.getProcessWideContainer() == null) {
29+
MetricsEnvironment.setProcessWideContainer(container);
30+
}
31+
}
32+
}

‎runners/flink/src/test/java/org/apache/beam/runners/flink/metrics/FlinkMetricContainerTest.java‎

Lines changed: 98 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,7 @@
2525
import static org.mockito.ArgumentMatchers.anyString;
2626
import static org.mockito.ArgumentMatchers.argThat;
2727
import static org.mockito.ArgumentMatchers.eq;
28+
import static org.mockito.Mockito.never;
2829
import static org.mockito.Mockito.verify;
2930
import static org.mockito.Mockito.when;
3031

@@ -39,19 +40,26 @@
3940
import org.apache.beam.runners.core.metrics.SimpleMonitoringInfoBuilder;
4041
import org.apache.beam.runners.flink.metrics.FlinkMetricContainerBase.FlinkDistributionGauge;
4142
import org.apache.beam.sdk.metrics.Counter;
43+
import org.apache.beam.sdk.metrics.DelegatingCounter;
44+
import org.apache.beam.sdk.metrics.DelegatingDistribution;
45+
import org.apache.beam.sdk.metrics.DelegatingGauge;
4246
import org.apache.beam.sdk.metrics.Distribution;
4347
import org.apache.beam.sdk.metrics.DistributionResult;
4448
import org.apache.beam.sdk.metrics.Gauge;
4549
import org.apache.beam.sdk.metrics.GaugeResult;
4650
import org.apache.beam.sdk.metrics.MetricKey;
4751
import org.apache.beam.sdk.metrics.MetricName;
52+
import org.apache.beam.sdk.metrics.Metrics;
4853
import org.apache.beam.sdk.metrics.MetricsContainer;
54+
import org.apache.beam.sdk.metrics.MetricsEnvironment;
4955
import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList;
5056
import org.apache.flink.api.common.functions.RuntimeContext;
5157
import org.apache.flink.metrics.SimpleCounter;
5258
import org.apache.flink.metrics.groups.OperatorMetricGroup;
59+
import org.junit.After;
5360
import org.junit.Before;
5461
import org.junit.Test;
62+
import org.mockito.ArgumentCaptor;
5563
import org.mockito.ArgumentMatcher;
5664
import org.mockito.Mock;
5765
import org.mockito.MockitoAnnotations;
@@ -74,6 +82,11 @@ public void beforeTest() {
7482
container = new FlinkMetricContainer(runtimeContext);
7583
}
7684

85+
@After
86+
public void afterTest() {
87+
MetricsEnvironment.setProcessWideContainer(null);
88+
}
89+
7790
@Test
7891
public void testMetricNameGeneration() {
7992
MetricKey key = MetricKey.create("step", MetricName.named("namespace", "name"));
@@ -97,6 +110,91 @@ public void testCounter() {
97110
assertThat(flinkCounter.getCount(), is(2L));
98111
}
99112

113+
@Test
114+
public void testProcessWideMetrics() {
115+
SimpleCounter flinkCounter = new SimpleCounter();
116+
when(metricGroup.counter("global.counter")).thenReturn(flinkCounter);
117+
when(metricGroup.gauge(eq("global.gauge"), any()))
118+
.thenAnswer(invocation -> invocation.getArgument(1));
119+
when(metricGroup.gauge(eq("global.distribution"), any()))
120+
.thenAnswer(invocation -> invocation.getArgument(1));
121+
122+
Counter counter = new DelegatingCounter(MetricName.named("global", "counter"), true);
123+
Gauge gauge = new DelegatingGauge(MetricName.named("global", "gauge"), true);
124+
Distribution distribution =
125+
new DelegatingDistribution(MetricName.named("global", "distribution"), true);
126+
counter.inc(5);
127+
gauge.set(3);
128+
distribution.update(11);
129+
distribution.update(13);
130+
MetricsContainerImpl processWideContainer =
131+
(MetricsContainerImpl) MetricsEnvironment.getProcessWideContainer();
132+
assertThat(
133+
processWideContainer
134+
.getGauge(MetricName.named("global", "gauge"))
135+
.getCumulative()
136+
.extractResult()
137+
.getValue(),
138+
is(3L));
139+
140+
container.updateMetrics("step");
141+
assertThat(flinkCounter.getCount(), is(5L));
142+
ArgumentCaptor<FlinkMetricContainerBase.FlinkGauge> gaugeCaptor =
143+
ArgumentCaptor.forClass(FlinkMetricContainerBase.FlinkGauge.class);
144+
ArgumentCaptor<FlinkDistributionGauge> distributionCaptor =
145+
ArgumentCaptor.forClass(FlinkDistributionGauge.class);
146+
verify(metricGroup).gauge(eq("global.gauge"), gaugeCaptor.capture());
147+
verify(metricGroup).gauge(eq("global.distribution"), distributionCaptor.capture());
148+
FlinkMetricContainerBase.FlinkGauge flinkGauge = gaugeCaptor.getValue();
149+
FlinkDistributionGauge flinkDistributionGauge = distributionCaptor.getValue();
150+
assertThat(flinkGauge.getValue(), is(3L));
151+
assertThat(flinkDistributionGauge.getValue(), is(DistributionResult.create(24, 2, 11, 13)));
152+
153+
container.updateMetrics("step");
154+
assertThat(flinkCounter.getCount(), is(5L));
155+
assertThat(flinkGauge.getValue(), is(3L));
156+
assertThat(flinkDistributionGauge.getValue(), is(DistributionResult.create(24, 2, 11, 13)));
157+
158+
counter.inc(2);
159+
gauge.set(4);
160+
distribution.update(17);
161+
container.updateMetrics("step");
162+
assertThat(flinkCounter.getCount(), is(7L));
163+
assertThat(flinkGauge.getValue(), is(4L));
164+
assertThat(flinkDistributionGauge.getValue(), is(DistributionResult.create(41, 3, 11, 17)));
165+
}
166+
167+
@Test
168+
public void testProcessWideMetricsArePublishedOnlyByOwningContainer() {
169+
RuntimeContext otherRuntimeContext = org.mockito.Mockito.mock(RuntimeContext.class);
170+
OperatorMetricGroup otherMetricGroup = org.mockito.Mockito.mock(OperatorMetricGroup.class);
171+
when(otherRuntimeContext.<MetricsContainerStepMap, MetricsContainerStepMap>getAccumulator(
172+
anyString()))
173+
.thenReturn(new MetricsAccumulator());
174+
when(otherRuntimeContext.getMetricGroup()).thenReturn(otherMetricGroup);
175+
FlinkMetricContainer otherContainer = new FlinkMetricContainer(otherRuntimeContext);
176+
177+
Counter counter = new DelegatingCounter(MetricName.named("global", "counter"), true);
178+
counter.inc(5);
179+
180+
otherContainer.updateMetrics("otherStep");
181+
verify(otherMetricGroup, never()).counter("global.counter");
182+
183+
SimpleCounter flinkCounter = new SimpleCounter();
184+
when(metricGroup.counter("global.counter")).thenReturn(flinkCounter);
185+
container.updateMetrics("step");
186+
assertThat(flinkCounter.getCount(), is(5L));
187+
}
188+
189+
@Test
190+
public void testUnscopedThreadMetricsAreNotPublishedAsProcessWideMetrics() {
191+
Metrics.counter("global", "unscoped").inc(5);
192+
193+
container.updateMetrics("step");
194+
195+
verify(metricGroup, never()).counter("global.unscoped");
196+
}
197+
100198
@Test
101199
public void testGauge() {
102200
FlinkMetricContainerBase.FlinkGauge flinkGauge =

0 commit comments

Comments
 (0)