open-telemetry
/
opentelemetry-java
mirror of https://github.com/open-telemetry/opentelemetry-java.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Simplify logic in AsynchronousInstrumentAccumulator (#2349) 

Signed-off-by: Bogdan Drutu <bogdandrutu@gmail.com>
This commit is contained in:
Bogdan Drutu 2020-12-18 09:08:19 -08:00 committed by GitHub
parent 94422d9dc0
commit 0412d47ffe
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 31 additions and 49 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

80
sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/AsynchronousInstrumentAccumulator.java
View File

@ -5,7 +5,6 @@
package io.opentelemetry.sdk.metrics; package io.opentelemetry.sdk.metrics;
import io.opentelemetry.api.common.Labels;
import io.opentelemetry.api.metrics.AsynchronousInstrument; import io.opentelemetry.api.metrics.AsynchronousInstrument;
import io.opentelemetry.sdk.metrics.aggregator.Aggregator; import io.opentelemetry.sdk.metrics.aggregator.Aggregator;
import io.opentelemetry.sdk.metrics.data.MetricData; import io.opentelemetry.sdk.metrics.data.MetricData;
@ -17,34 +16,24 @@ import javax.annotation.Nullable;
final class AsynchronousInstrumentAccumulator { final class AsynchronousInstrumentAccumulator {
private final ReentrantLock collectLock = new ReentrantLock(); private final ReentrantLock collectLock = new ReentrantLock();
private final InstrumentProcessor instrumentProcessor; private final InstrumentProcessor instrumentProcessor;
private final Consumer<InstrumentProcessor> metricUpdater; private final Runnable metricUpdater;
private AsynchronousInstrumentAccumulator(
InstrumentProcessor instrumentProcessor, Consumer<InstrumentProcessor> metricUpdater) {
this.metricUpdater = metricUpdater;
this.instrumentProcessor = instrumentProcessor;
}
public List<MetricData> collectAll() {
collectLock.lock();
try {
metricUpdater.accept(instrumentProcessor);
return instrumentProcessor.completeCollectionCycle();
} finally {
collectLock.unlock();
}
}
static AsynchronousInstrumentAccumulator doubleAsynchronousAccumulator( static AsynchronousInstrumentAccumulator doubleAsynchronousAccumulator(
InstrumentProcessor instrumentProcessor, InstrumentProcessor instrumentProcessor,
@Nullable Consumer<AsynchronousInstrument.DoubleResult> metricUpdater) { @Nullable Consumer<AsynchronousInstrument.DoubleResult> metricUpdater) {
// TODO: Decide what to do with null updater. // TODO: Decide what to do with null updater.
if (metricUpdater == null) { if (metricUpdater == null) {
return new AsynchronousInstrumentAccumulator(instrumentProcessor, instrumentProcessor1 -> {}); return new AsynchronousInstrumentAccumulator(instrumentProcessor, () -> {});
} }
DoubleResultSdk result = new DoubleResultSdk(instrumentProcessor); AsynchronousInstrument.DoubleResult result =
(value, labels) -> {
Aggregator aggregator = instrumentProcessor.getAggregator();
aggregator.recordDouble(value);
instrumentProcessor.batch(labels, aggregator, /* mappedAggregator= */ false);
};
return new AsynchronousInstrumentAccumulator( return new AsynchronousInstrumentAccumulator(
instrumentProcessor, instrumentProcessor1 -> metricUpdater.accept(result)); instrumentProcessor, () -> metricUpdater.accept(result));
} }
static AsynchronousInstrumentAccumulator longAsynchronousAccumulator( static AsynchronousInstrumentAccumulator longAsynchronousAccumulator(
@ -52,40 +41,33 @@ final class AsynchronousInstrumentAccumulator {
@Nullable Consumer<AsynchronousInstrument.LongResult> metricUpdater) { @Nullable Consumer<AsynchronousInstrument.LongResult> metricUpdater) {
// TODO: Decide what to do with null updater. // TODO: Decide what to do with null updater.
if (metricUpdater == null) { if (metricUpdater == null) {
return new AsynchronousInstrumentAccumulator(instrumentProcessor, instrumentProcessor1 -> {}); return new AsynchronousInstrumentAccumulator(instrumentProcessor, () -> {});
} }
LongResultSdk result = new LongResultSdk(instrumentProcessor);
AsynchronousInstrument.LongResult result =
(value, labels) -> {
Aggregator aggregator = instrumentProcessor.getAggregator();
aggregator.recordLong(value);
instrumentProcessor.batch(labels, aggregator, /* mappedAggregator= */ false);
};
return new AsynchronousInstrumentAccumulator( return new AsynchronousInstrumentAccumulator(
instrumentProcessor, instrumentProcessor1 -> metricUpdater.accept(result)); instrumentProcessor, () -> metricUpdater.accept(result));
} }
private static final class DoubleResultSdk implements AsynchronousInstrument.DoubleResult { private AsynchronousInstrumentAccumulator(
private final InstrumentProcessor instrumentProcessor; InstrumentProcessor instrumentProcessor, Runnable metricUpdater) {
this.metricUpdater = metricUpdater;
private DoubleResultSdk(InstrumentProcessor instrumentProcessor) { this.instrumentProcessor = instrumentProcessor;
this.instrumentProcessor = instrumentProcessor;
}
@Override
public void observe(double sum, Labels labels) {
Aggregator aggregator = instrumentProcessor.getAggregator();
aggregator.recordDouble(sum);
instrumentProcessor.batch(labels, aggregator, /* mappedAggregator= */ false);
}
} }
private static final class LongResultSdk implements AsynchronousInstrument.LongResult { public List<MetricData> collectAll() {
private final InstrumentProcessor instrumentProcessor; collectLock.lock();
try {
private LongResultSdk(InstrumentProcessor instrumentProcessor) { metricUpdater.run();
this.instrumentProcessor = instrumentProcessor; return instrumentProcessor.completeCollectionCycle();
} } finally {
collectLock.unlock();
@Override
public void observe(long sum, Labels labels) {
Aggregator aggregator = instrumentProcessor.getAggregator();
aggregator.recordLong(sum);
instrumentProcessor.batch(labels, aggregator, /* mappedAggregator= */ false);
} }
} }
} }