Skip to content

Commit a12f657

Browse files
0x006EA1E50x006EA1E5
committed
#887 otlp exporter exemplars
Signed-off-by: Greg Eales <0x006ea1e5@gmail.com>
1 parent ee99bb8 commit a12f657

File tree

4 files changed

+215
-9
lines changed

4 files changed

+215
-9
lines changed

‎prometheus-metrics-exporter-opentelemetry/pom.xml‎

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -79,6 +79,30 @@
7979
<version>4.13.2</version>
8080
<scope>test</scope>
8181
</dependency>
82+
<dependency>
83+
<groupId>org.wiremock</groupId>
84+
<artifactId>wiremock</artifactId>
85+
<version>3.2.0</version>
86+
<scope>test</scope>
87+
</dependency>
88+
<dependency>
89+
<groupId>org.awaitility</groupId>
90+
<artifactId>awaitility</artifactId>
91+
<version>4.2.0</version>
92+
<scope>test</scope>
93+
</dependency>
94+
<dependency>
95+
<groupId>io.opentelemetry</groupId>
96+
<artifactId>opentelemetry-proto</artifactId>
97+
<version>0.17.1</version>
98+
<scope>test</scope>
99+
</dependency>
100+
<dependency>
101+
<groupId>io.opentelemetry</groupId>
102+
<artifactId>opentelemetry-sdk-trace</artifactId>
103+
<version>${otel.version}</version>
104+
<scope>test</scope>
105+
</dependency>
82106
</dependencies>
83107

84108
<build>

‎prometheus-metrics-exporter-opentelemetry/src/main/java/io/prometheus/metrics/exporter/opentelemetry/OpenTelemetryExporter.java‎

Lines changed: 8 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,8 @@
1818
import java.util.Map;
1919
import java.util.concurrent.TimeUnit;
2020

21-
public class OpenTelemetryExporter {
21+
public class OpenTelemetryExporter implements AutoCloseable {
22+
private final PeriodicMetricReader reader;
2223

2324
private OpenTelemetryExporter(Builder builder, PrometheusProperties config, PrometheusRegistry registry) {
2425
InstrumentationScopeInfo instrumentationScopeInfo = PrometheusInstrumentationScope.loadInstrumentationScopeInfo();
@@ -42,13 +43,18 @@ private OpenTelemetryExporter(Builder builder, PrometheusProperties config, Prom
4243
}
4344
exporter = exporterBuilder.build();
4445
}
45-
PeriodicMetricReader reader = PeriodicMetricReader.builder(exporter)
46+
reader = PeriodicMetricReader.builder(exporter)
4647
.setInterval(Duration.ofSeconds(ConfigHelper.getIntervalSeconds(builder, properties)))
4748
.build();
49+
4850
PrometheusMetricProducer prometheusMetricProducer = new PrometheusMetricProducer(registry, instrumentationScopeInfo, resource);
4951
reader.register(prometheusMetricProducer);
5052
}
5153

54+
public void close() {
55+
reader.shutdown();
56+
}
57+
5258
private Resource initResourceAttributes(Builder builder, ExporterOpenTelemetryProperties properties, InstrumentationScopeInfo instrumentationScopeInfo) {
5359
String serviceName = ConfigHelper.getServiceName(builder, properties);
5460
String serviceNamespace = ConfigHelper.getServiceNamespace(builder, properties);

‎prometheus-metrics-exporter-opentelemetry/src/main/java/io/prometheus/metrics/exporter/opentelemetry/otelmodel/PrometheusData.java‎

Lines changed: 44 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -1,19 +1,22 @@
11
package io.prometheus.metrics.exporter.opentelemetry.otelmodel;
22

3-
import io.prometheus.metrics.model.snapshots.DataPointSnapshot;
4-
import io.prometheus.metrics.model.snapshots.Exemplars;
3+
import io.prometheus.metrics.model.snapshots.*;
54
import io.prometheus.metrics.shaded.io_opentelemetry_1_28_0.api.common.Attributes;
65
import io.prometheus.metrics.shaded.io_opentelemetry_1_28_0.api.common.AttributesBuilder;
6+
import io.prometheus.metrics.shaded.io_opentelemetry_1_28_0.api.trace.SpanContext;
7+
import io.prometheus.metrics.shaded.io_opentelemetry_1_28_0.api.trace.TraceFlags;
8+
import io.prometheus.metrics.shaded.io_opentelemetry_1_28_0.api.trace.TraceState;
79
import io.prometheus.metrics.shaded.io_opentelemetry_1_28_0.sdk.metrics.data.Data;
810
import io.prometheus.metrics.shaded.io_opentelemetry_1_28_0.sdk.metrics.data.DoubleExemplarData;
911
import io.prometheus.metrics.shaded.io_opentelemetry_1_28_0.sdk.metrics.data.MetricDataType;
1012
import io.prometheus.metrics.shaded.io_opentelemetry_1_28_0.sdk.metrics.data.PointData;
11-
import io.prometheus.metrics.model.snapshots.Exemplar;
12-
import io.prometheus.metrics.model.snapshots.Labels;
13+
import io.prometheus.metrics.shaded.io_opentelemetry_1_28_0.sdk.metrics.internal.data.ImmutableDoubleExemplarData;
1314

14-
import java.util.Collections;
1515
import java.util.List;
16+
import java.util.Objects;
1617
import java.util.concurrent.TimeUnit;
18+
import java.util.stream.Collectors;
19+
import java.util.stream.StreamSupport;
1720

1821
abstract class PrometheusData<T extends PointData> implements Data<T> {
1922

@@ -44,8 +47,41 @@ protected List<DoubleExemplarData> convertExemplar(Exemplar exemplar) {
4447
}
4548

4649
protected List<DoubleExemplarData> convertExemplars(Exemplars exemplars) {
47-
// TODO: Exemplars not implemented yet.
48-
return Collections.emptyList();
50+
return StreamSupport.stream(exemplars.spliterator(), false)
51+
.map(this::toDoubleExemplarData)
52+
.filter(Objects::nonNull)
53+
.collect(Collectors.toList());
54+
}
55+
56+
protected DoubleExemplarData toDoubleExemplarData(Exemplar exemplar) {
57+
if (exemplar == null) {
58+
return null;
59+
}
60+
61+
AttributesBuilder filteredAttributesBuilder = Attributes.builder();
62+
String traceId = null;
63+
String spanId = null;
64+
for (Label label : exemplar.getLabels()) {
65+
if (label.getName().equals(Exemplar.TRACE_ID)) {
66+
traceId = label.getValue();
67+
}
68+
else if (label.getName().equals(Exemplar.SPAN_ID)) {
69+
spanId = label.getValue();
70+
} else {
71+
filteredAttributesBuilder.put(label.getName(), label.getValue());
72+
}
73+
}
74+
Attributes filteredAttributes = filteredAttributesBuilder.build();
75+
76+
SpanContext spanContext = (traceId != null && spanId != null)
77+
? SpanContext.create(traceId, spanId, TraceFlags.getSampled(), TraceState.getDefault())
78+
: SpanContext.getInvalid();
79+
80+
return ImmutableDoubleExemplarData.create(
81+
filteredAttributes,
82+
TimeUnit.MILLISECONDS.toNanos(exemplar.getTimestampMillis()),
83+
spanContext,
84+
exemplar.getValue());
4985
}
5086

5187
protected long getStartEpochNanos(DataPointSnapshot dataPoint) {
@@ -55,4 +91,5 @@ protected long getStartEpochNanos(DataPointSnapshot dataPoint) {
5591
protected long getEpochNanos(DataPointSnapshot dataPoint, long currentTimeMillis) {
5692
return dataPoint.hasScrapeTimestamp() ? TimeUnit.MILLISECONDS.toNanos(dataPoint.getScrapeTimestampMillis()) : TimeUnit.MILLISECONDS.toNanos(currentTimeMillis);
5793
}
94+
5895
}

‎prometheus-metrics-exporter-opentelemetry/src/test/java/io/prometheus/metrics/exporter/opentelemetry/ExemplarTest.java‎

Lines changed: 139 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,139 @@
1+
package io.prometheus.metrics.exporter.opentelemetry;
2+
3+
import com.github.tomakehurst.wiremock.http.Request;
4+
import com.github.tomakehurst.wiremock.junit.WireMockRule;
5+
import com.github.tomakehurst.wiremock.matching.MatchResult;
6+
import com.github.tomakehurst.wiremock.matching.ValueMatcher;
7+
import com.google.protobuf.InvalidProtocolBufferException;
8+
import io.opentelemetry.api.trace.Span;
9+
import io.opentelemetry.api.trace.Tracer;
10+
import io.opentelemetry.context.Scope;
11+
import io.opentelemetry.proto.collector.metrics.v1.ExportMetricsServiceRequest;
12+
import io.opentelemetry.proto.metrics.v1.DoubleDataPoint;
13+
import io.opentelemetry.proto.metrics.v1.InstrumentationLibraryMetrics;
14+
import io.opentelemetry.proto.metrics.v1.Metric;
15+
import io.opentelemetry.proto.metrics.v1.ResourceMetrics;
16+
import io.opentelemetry.sdk.trace.SdkTracerProvider;
17+
import io.opentelemetry.sdk.trace.samplers.Sampler;
18+
import io.prometheus.metrics.core.metrics.Counter;
19+
import io.prometheus.metrics.model.registry.PrometheusRegistry;
20+
import org.awaitility.core.ConditionTimeoutException;
21+
import org.junit.After;
22+
import org.junit.Before;
23+
import org.junit.Rule;
24+
import org.junit.Test;
25+
26+
import static com.github.tomakehurst.wiremock.client.WireMock.*;
27+
import static java.util.concurrent.TimeUnit.SECONDS;
28+
import static org.awaitility.Awaitility.await;
29+
30+
public class ExemplarTest {
31+
private static final String ENDPOINT_PATH = "/v1/metrics";
32+
private static final int TIMEOUT = 3;
33+
private static final String INSTRUMENTATION_SCOPE_NAME = "testInstrumentationScope";
34+
private static final String SPAN_NAME = "test-span";
35+
public static final String TEST_COUNTER_NAME = "test_counter";
36+
private Counter testCounter;
37+
private OpenTelemetryExporter openTelemetryExporter;
38+
@Rule
39+
public WireMockRule wireMockRule = new WireMockRule(4317);
40+
41+
@Before
42+
public void setUp() {
43+
openTelemetryExporter = OpenTelemetryExporter.builder()
44+
.endpoint("http://localhost:4317")
45+
.protocol("http/protobuf")
46+
.intervalSeconds(1)
47+
.buildAndStart();
48+
49+
testCounter = Counter.builder()
50+
.name(TEST_COUNTER_NAME)
51+
.withExemplars()
52+
.register();
53+
54+
wireMockRule.stubFor(post(ENDPOINT_PATH)
55+
.withHeader("Content-Type", containing("application/x-protobuf"))
56+
.willReturn(ok()
57+
.withHeader("Content-Type", "application/json")
58+
.withBody("{\"partialSuccess\":{}}")));
59+
}
60+
61+
@After
62+
public void tearDown() {
63+
PrometheusRegistry.defaultRegistry.unregister(testCounter);
64+
openTelemetryExporter.close();
65+
}
66+
67+
@Test
68+
public void sampledExemplarIsForwarded() {
69+
try (SdkTracerProvider sdkTracerProvider = SdkTracerProvider.builder()
70+
.setSampler(Sampler.alwaysOn())
71+
.build()) {
72+
73+
Tracer test = sdkTracerProvider.get(INSTRUMENTATION_SCOPE_NAME);
74+
Span span = test.spanBuilder(SPAN_NAME)
75+
.startSpan();
76+
try (Scope scope = span.makeCurrent()) {
77+
testCounter.inc(2);
78+
}
79+
}
80+
81+
82+
await().atMost(TIMEOUT, SECONDS)
83+
.ignoreException(com.github.tomakehurst.wiremock.client.VerificationException.class)
84+
.until(() -> {
85+
verify(postRequestedFor(urlEqualTo(ENDPOINT_PATH))
86+
.withHeader("Content-Type", equalTo("application/x-protobuf"))
87+
.andMatching(getExemplarCountMatcher(1)));
88+
return true;
89+
});
90+
91+
}
92+
93+
@Test(expected = ConditionTimeoutException.class)
94+
public void notSampledExemplarIsNotForwarded() {
95+
try (SdkTracerProvider sdkTracerProvider = SdkTracerProvider.builder()
96+
.setSampler(Sampler.alwaysOff())
97+
.build()) {
98+
99+
Tracer test = sdkTracerProvider.get(INSTRUMENTATION_SCOPE_NAME);
100+
Span span = test.spanBuilder(SPAN_NAME)
101+
.startSpan();
102+
try (Scope scope = span.makeCurrent()) {
103+
testCounter.inc(2);
104+
}
105+
}
106+
107+
await().atMost(TIMEOUT, SECONDS)
108+
.ignoreException(com.github.tomakehurst.wiremock.client.VerificationException.class)
109+
.until(() -> {
110+
verify(postRequestedFor(urlEqualTo(ENDPOINT_PATH))
111+
.withHeader("Content-Type", equalTo("application/x-protobuf"))
112+
.andMatching(getExemplarCountMatcher(1)));
113+
return true;
114+
});
115+
116+
}
117+
118+
private static ValueMatcher<Request> getExemplarCountMatcher(int expectedCount) {
119+
return request -> {
120+
try {
121+
ExportMetricsServiceRequest exportMetricsServiceRequest = ExportMetricsServiceRequest.parseFrom(request.getBody());
122+
for (ResourceMetrics resourceMetrics : exportMetricsServiceRequest.getResourceMetricsList()) {
123+
for (InstrumentationLibraryMetrics instrumentationLibraryMetrics : resourceMetrics.getInstrumentationLibraryMetricsList()) {
124+
for (Metric metric : instrumentationLibraryMetrics.getMetricsList()) {
125+
for (DoubleDataPoint doubleDataPoint : metric.getDoubleSum().getDataPointsList()) {
126+
if (doubleDataPoint.getExemplarsCount() == expectedCount) {
127+
return MatchResult.exactMatch();
128+
}
129+
}
130+
}
131+
}
132+
}
133+
} catch (InvalidProtocolBufferException e) {
134+
throw new RuntimeException(e);
135+
}
136+
return MatchResult.noMatch();
137+
};
138+
}
139+
}

0 commit comments

Comments
 (0)