Skip to content

Commit fea5ecc

Browse files
jkwatsonjkwatson
authored
Don't auto-start the IntervalMetricReader in the constructor. (open-telemetry#3026)
* Don't auto-start the IntervalMetricReader in the constructor. * a little more protection around the possibility of double-starting the IMR * add a test that verifies we only start once * tweak the values to see if CI likes it better * make the test simpler for CI * try a little bit longer time? * change the test to make an assertion on a passed-in scheduler.
1 parent aac9284 commit fea5ecc

5 files changed

Lines changed: 83 additions & 14 deletions

File tree

‎perf-harness/src/test/java/io/opentelemetry/perf/OtlpPipelineStressTest.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -252,7 +252,7 @@ private void setupSdk() {
252252
.setMetricExporter(metricExporter)
253253
.setMetricProducers(Collections.singleton(meterProvider))
254254
.setExportIntervalMillis(1000)
255-
.build();
255+
.buildAndStart();
256256

257257
// set up the span exporter and wire it into the SDK
258258
OtlpGrpcSpanExporter spanExporter =

‎sdk-extensions/autoconfigure/src/main/java/io/opentelemetry/sdk/autoconfigure/MetricExporterConfiguration.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -96,7 +96,7 @@ private static void configureIntervalMetricReader(
9696
if (exportIntervalMillis != null) {
9797
readerBuilder.setExportIntervalMillis(exportIntervalMillis);
9898
}
99-
IntervalMetricReader reader = readerBuilder.build();
99+
IntervalMetricReader reader = readerBuilder.buildAndStart();
100100
Runtime.getRuntime().addShutdownHook(new Thread(reader::shutdown));
101101
}
102102

‎sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/export/IntervalMetricReader.java‎

Lines changed: 35 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,7 @@
1515
import java.util.List;
1616
import java.util.concurrent.Executors;
1717
import java.util.concurrent.ScheduledExecutorService;
18+
import java.util.concurrent.ScheduledFuture;
1819
import java.util.concurrent.TimeUnit;
1920
import java.util.concurrent.atomic.AtomicBoolean;
2021
import java.util.logging.Level;
@@ -32,9 +33,15 @@ public final class IntervalMetricReader {
3233
private final Exporter exporter;
3334
private final ScheduledExecutorService scheduler;
3435

36+
private volatile ScheduledFuture<?> scheduledFuture;
37+
private final Object lock = new Object();
38+
3539
/** Stops the scheduled task and calls export one more time. */
3640
public CompletableResultCode shutdown() {
3741
final CompletableResultCode result = new CompletableResultCode();
42+
if (scheduledFuture != null) {
43+
scheduledFuture.cancel(false);
44+
}
3845
scheduler.shutdown();
3946
try {
4047
scheduler.awaitTermination(5, TimeUnit.SECONDS);
@@ -68,16 +75,36 @@ public static IntervalMetricReaderBuilder builder() {
6875
return new IntervalMetricReaderBuilder(InternalState.builder());
6976
}
7077

71-
@SuppressWarnings("FutureReturnValueIgnored")
7278
IntervalMetricReader(InternalState internalState) {
79+
this(
80+
internalState,
81+
Executors.newScheduledThreadPool(1, new DaemonThreadFactory("IntervalMetricReader")));
82+
}
83+
84+
// visible for testing
85+
IntervalMetricReader(InternalState internalState, ScheduledExecutorService intervalMetricReader) {
7386
this.exporter = new Exporter(internalState);
74-
this.scheduler =
75-
Executors.newScheduledThreadPool(1, new DaemonThreadFactory("IntervalMetricReader"));
76-
this.scheduler.scheduleAtFixedRate(
77-
exporter,
78-
internalState.getExportIntervalMillis(),
79-
internalState.getExportIntervalMillis(),
80-
TimeUnit.MILLISECONDS);
87+
this.scheduler = intervalMetricReader;
88+
}
89+
90+
/**
91+
* Starts this {@link IntervalMetricReader} to report to the configured exporter.
92+
*
93+
* @return this for fluent usage along with the builder.
94+
*/
95+
public IntervalMetricReader start() {
96+
synchronized (lock) {
97+
if (scheduledFuture != null) {
98+
return this;
99+
}
100+
scheduledFuture =
101+
scheduler.scheduleAtFixedRate(
102+
exporter,
103+
exporter.internalState.getExportIntervalMillis(),
104+
exporter.internalState.getExportIntervalMillis(),
105+
TimeUnit.MILLISECONDS);
106+
return this;
107+
}
81108
}
82109

83110
private static final class Exporter implements Runnable {

‎sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/export/IntervalMetricReaderBuilder.java‎

Lines changed: 12 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -52,7 +52,8 @@ public IntervalMetricReaderBuilder setMetricProducers(
5252
}
5353

5454
/**
55-
* Builds a new {@link IntervalMetricReader} with current settings.
55+
* Builds a new {@link IntervalMetricReader} with current settings. Does not start the background
56+
* thread. Please call {@link IntervalMetricReader#start()} to do that.
5657
*
5758
* @return a {@code IntervalMetricReader}.
5859
*/
@@ -63,4 +64,14 @@ public IntervalMetricReader build() {
6364

6465
return new IntervalMetricReader(internalState);
6566
}
67+
68+
/**
69+
* Builds a new {@link IntervalMetricReader} with current settings and starts the background
70+
* thread running.
71+
*
72+
* @return a {@code IntervalMetricReader}.
73+
*/
74+
public IntervalMetricReader buildAndStart() {
75+
return build().start();
76+
}
6677
}

‎sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/export/IntervalMetricReaderTest.java‎

Lines changed: 34 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,11 @@
66
package io.opentelemetry.sdk.metrics.export;
77

88
import static org.assertj.core.api.Assertions.assertThat;
9+
import static org.mockito.ArgumentMatchers.any;
10+
import static org.mockito.ArgumentMatchers.anyLong;
11+
import static org.mockito.Mockito.mock;
12+
import static org.mockito.Mockito.times;
13+
import static org.mockito.Mockito.verify;
914
import static org.mockito.Mockito.when;
1015

1116
import io.opentelemetry.api.metrics.common.Labels;
@@ -22,6 +27,8 @@
2227
import java.util.List;
2328
import java.util.concurrent.BlockingQueue;
2429
import java.util.concurrent.LinkedBlockingQueue;
30+
import java.util.concurrent.ScheduledExecutorService;
31+
import java.util.concurrent.ScheduledFuture;
2532
import java.util.concurrent.TimeUnit;
2633
import java.util.concurrent.atomic.AtomicBoolean;
2734
import javax.annotation.Nullable;
@@ -57,6 +64,28 @@ void setup() {
5764
when(metricProducer.collectAllMetrics()).thenReturn(Collections.singletonList(METRIC_DATA));
5865
}
5966

67+
@Test
68+
@SuppressWarnings({"rawtypes", "unchecked"})
69+
void startOnlyOnce() {
70+
ScheduledExecutorService scheduler = mock(ScheduledExecutorService.class);
71+
72+
ScheduledFuture mock = mock(ScheduledFuture.class);
73+
when(scheduler.scheduleAtFixedRate(any(), anyLong(), anyLong(), any())).thenReturn(mock);
74+
75+
IntervalMetricReader intervalMetricReader =
76+
new IntervalMetricReader(
77+
IntervalMetricReader.InternalState.builder()
78+
.setMetricProducers(Collections.emptyList())
79+
.setMetricExporter(mock(MetricExporter.class))
80+
.build(),
81+
scheduler);
82+
83+
intervalMetricReader.start();
84+
intervalMetricReader.start();
85+
86+
verify(scheduler, times(1)).scheduleAtFixedRate(any(), anyLong(), anyLong(), any());
87+
}
88+
6089
@Test
6190
void intervalExport() throws Exception {
6291
WaitingMetricExporter waitingMetricExporter = new WaitingMetricExporter();
@@ -65,7 +94,7 @@ void intervalExport() throws Exception {
6594
.setExportIntervalMillis(100)
6695
.setMetricExporter(waitingMetricExporter)
6796
.setMetricProducers(Collections.singletonList(metricProducer))
68-
.build();
97+
.buildAndStart();
6998

7099
try {
71100
assertThat(waitingMetricExporter.waitForNumberOfExports(1))
@@ -88,7 +117,7 @@ public void intervalExport_exporterThrowsException() throws Exception {
88117
.setExportIntervalMillis(100)
89118
.setMetricExporter(waitingMetricExporter)
90119
.setMetricProducers(Collections.singletonList(metricProducer))
91-
.build();
120+
.buildAndStart();
92121

93122
try {
94123
assertThat(waitingMetricExporter.waitForNumberOfExports(1))
@@ -106,7 +135,7 @@ void oneLastExportAfterShutdown() throws Exception {
106135
.setExportIntervalMillis(100_000)
107136
.setMetricExporter(waitingMetricExporter)
108137
.setMetricProducers(Collections.singletonList(metricProducer))
109-
.build();
138+
.buildAndStart();
110139

111140
// Assume that this will be called in less than 100 seconds.
112141
intervalMetricReader.shutdown();
@@ -123,6 +152,7 @@ private static class WaitingMetricExporter implements MetricExporter {
123152
private final AtomicBoolean hasShutdown = new AtomicBoolean(false);
124153
private final boolean shouldThrow;
125154
private final BlockingQueue<List<MetricData>> queue = new LinkedBlockingQueue<>();
155+
private final List<Long> exportTimes = Collections.synchronizedList(new ArrayList<>());
126156

127157
private WaitingMetricExporter() {
128158
this(false);
@@ -134,6 +164,7 @@ private WaitingMetricExporter(boolean shouldThrow) {
134164

135165
@Override
136166
public CompletableResultCode export(Collection<MetricData> metricList) {
167+
exportTimes.add(System.currentTimeMillis());
137168
queue.offer(new ArrayList<>(metricList));
138169

139170
if (shouldThrow) {

0 commit comments

Comments
 (0)