-
Notifications
You must be signed in to change notification settings - Fork 850
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Add Pulsar MessagingProducerMetrics #11591
Changes from 2 commits
b954474
d855f97
b5d512b
3de71a8
6bb2c90
6f68ebd
9f42598
8b5e585
53b2dee
df187f4
e2d5e09
fc8c397
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,40 @@ | ||
/* | ||
* Copyright The OpenTelemetry Authors | ||
* SPDX-License-Identifier: Apache-2.0 | ||
*/ | ||
|
||
package io.opentelemetry.instrumentation.api.incubator.semconv.messaging; | ||
|
||
import static java.util.Arrays.asList; | ||
import static java.util.Collections.unmodifiableList; | ||
|
||
import io.opentelemetry.api.incubator.metrics.ExtendedDoubleHistogramBuilder; | ||
import io.opentelemetry.api.metrics.DoubleHistogramBuilder; | ||
import io.opentelemetry.semconv.ErrorAttributes; | ||
import io.opentelemetry.semconv.ServerAttributes; | ||
import io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes; | ||
import java.util.List; | ||
|
||
public class MessagingMetricsAdvice { | ||
crossoverJie marked this conversation as resolved.
Show resolved
Hide resolved
|
||
static final List<Double> DURATION_SECONDS_BUCKETS = | ||
unmodifiableList( | ||
asList(0.005, 0.01, 0.025, 0.05, 0.075, 0.1, 0.25, 0.5, 0.75, 1.0, 2.5, 5.0, 7.5, 10.0)); | ||
|
||
static void applyPublishDurationAdvice(DoubleHistogramBuilder builder) { | ||
if (!(builder instanceof ExtendedDoubleHistogramBuilder)) { | ||
return; | ||
} | ||
((ExtendedDoubleHistogramBuilder) builder) | ||
.setAttributesAdvice( | ||
asList( | ||
MessagingIncubatingAttributes.MESSAGING_SYSTEM, | ||
MessagingIncubatingAttributes.MESSAGING_DESTINATION_NAME, | ||
MessagingIncubatingAttributes.MESSAGING_OPERATION, | ||
MessagingIncubatingAttributes.MESSAGING_BATCH_MESSAGE_COUNT, | ||
ErrorAttributes.ERROR_TYPE, | ||
ServerAttributes.SERVER_PORT, | ||
ServerAttributes.SERVER_ADDRESS)); | ||
} | ||
|
||
private MessagingMetricsAdvice() {} | ||
} |
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,80 @@ | ||
/* | ||
* Copyright The OpenTelemetry Authors | ||
* SPDX-License-Identifier: Apache-2.0 | ||
*/ | ||
|
||
package io.opentelemetry.instrumentation.api.incubator.semconv.messaging; | ||
|
||
import static java.util.logging.Level.FINE; | ||
|
||
import com.google.auto.value.AutoValue; | ||
import com.google.errorprone.annotations.CanIgnoreReturnValue; | ||
import io.opentelemetry.api.common.Attributes; | ||
import io.opentelemetry.api.metrics.DoubleHistogram; | ||
import io.opentelemetry.api.metrics.DoubleHistogramBuilder; | ||
import io.opentelemetry.api.metrics.Meter; | ||
import io.opentelemetry.context.Context; | ||
import io.opentelemetry.context.ContextKey; | ||
import io.opentelemetry.instrumentation.api.instrumenter.OperationListener; | ||
import io.opentelemetry.instrumentation.api.instrumenter.OperationMetrics; | ||
import io.opentelemetry.instrumentation.api.internal.OperationMetricsUtil; | ||
import java.util.concurrent.TimeUnit; | ||
import java.util.logging.Logger; | ||
|
||
public class MessagingProducerMetrics implements OperationListener { | ||
crossoverJie marked this conversation as resolved.
Show resolved
Hide resolved
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Could you add javadoc similar to what the other There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Done; please take a look. Thanks. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. @trask should we call this There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. I think |
||
private static final double NANOS_PER_S = TimeUnit.SECONDS.toNanos(1); | ||
|
||
private static final ContextKey<MessagingProducerMetrics.State> PULSAR_PUBLISH_METRICS_STATE = | ||
ContextKey.named("pulsar-producer-metrics-state"); | ||
crossoverJie marked this conversation as resolved.
Show resolved
Hide resolved
|
||
private static final Logger logger = Logger.getLogger(MessagingProducerMetrics.class.getName()); | ||
|
||
private final DoubleHistogram publishDurationHistogram; | ||
|
||
private MessagingProducerMetrics(Meter meter) { | ||
DoubleHistogramBuilder durationBuilder = | ||
meter | ||
.histogramBuilder("messaging.publish.duration") | ||
.setDescription("Measures the duration of publish operation.") | ||
.setExplicitBucketBoundariesAdvice(MessagingMetricsAdvice.DURATION_SECONDS_BUCKETS) | ||
.setUnit("s"); | ||
MessagingMetricsAdvice.applyPublishDurationAdvice(durationBuilder); | ||
publishDurationHistogram = durationBuilder.build(); | ||
} | ||
|
||
public static OperationMetrics get() { | ||
return OperationMetricsUtil.create("messaging produce", MessagingProducerMetrics::new); | ||
} | ||
|
||
@Override | ||
@CanIgnoreReturnValue | ||
public Context onStart(Context context, Attributes startAttributes, long startNanos) { | ||
return context.with( | ||
PULSAR_PUBLISH_METRICS_STATE, | ||
new AutoValue_MessagingProducerMetrics_State(startAttributes, startNanos)); | ||
} | ||
|
||
@Override | ||
public void onEnd(Context context, Attributes endAttributes, long endNanos) { | ||
MessagingProducerMetrics.State state = context.get(PULSAR_PUBLISH_METRICS_STATE); | ||
if (state == null) { | ||
logger.log( | ||
FINE, | ||
"No state present when ending context {0}. Cannot record pulsar publish metrics.", | ||
context); | ||
return; | ||
} | ||
|
||
Attributes attributes = state.startAttributes().toBuilder().putAll(endAttributes).build(); | ||
|
||
publishDurationHistogram.record( | ||
(endNanos - state.startTimeNanos()) / NANOS_PER_S, attributes, context); | ||
} | ||
|
||
@AutoValue | ||
abstract static class State { | ||
|
||
abstract Attributes startAttributes(); | ||
|
||
abstract long startTimeNanos(); | ||
} | ||
} |
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,122 @@ | ||
/* | ||
* Copyright The OpenTelemetry Authors | ||
* SPDX-License-Identifier: Apache-2.0 | ||
*/ | ||
|
||
package io.opentelemetry.instrumentation.api.incubator.semconv.messaging; | ||
|
||
import static io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions.equalTo; | ||
import static org.assertj.core.api.Assertions.assertThat; | ||
|
||
import io.opentelemetry.api.common.Attributes; | ||
import io.opentelemetry.api.trace.Span; | ||
import io.opentelemetry.api.trace.SpanContext; | ||
import io.opentelemetry.api.trace.TraceFlags; | ||
import io.opentelemetry.api.trace.TraceState; | ||
import io.opentelemetry.context.Context; | ||
import io.opentelemetry.instrumentation.api.instrumenter.OperationListener; | ||
import io.opentelemetry.sdk.metrics.SdkMeterProvider; | ||
import io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions; | ||
import io.opentelemetry.sdk.testing.exporter.InMemoryMetricReader; | ||
import io.opentelemetry.semconv.ServerAttributes; | ||
import io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes; | ||
import java.util.concurrent.TimeUnit; | ||
import org.junit.jupiter.api.Test; | ||
|
||
public class MessagingProducerMetricsTest { | ||
crossoverJie marked this conversation as resolved.
Show resolved
Hide resolved
|
||
|
||
static final double[] DURATION_BUCKETS = | ||
crossoverJie marked this conversation as resolved.
Show resolved
Hide resolved
|
||
MessagingMetricsAdvice.DURATION_SECONDS_BUCKETS.stream().mapToDouble(d -> d).toArray(); | ||
|
||
@Test | ||
void collectsMetrics() { | ||
InMemoryMetricReader metricReader = InMemoryMetricReader.create(); | ||
SdkMeterProvider meterProvider = | ||
SdkMeterProvider.builder().registerMetricReader(metricReader).build(); | ||
|
||
OperationListener listener = MessagingProducerMetrics.get().create(meterProvider.get("test")); | ||
|
||
Attributes requestAttributes = | ||
Attributes.builder() | ||
.put(MessagingIncubatingAttributes.MESSAGING_SYSTEM, "pulsar") | ||
.put( | ||
MessagingIncubatingAttributes.MESSAGING_DESTINATION_NAME, | ||
"persistent://public/default/topic") | ||
.put(MessagingIncubatingAttributes.MESSAGING_OPERATION, "publish") | ||
.put(ServerAttributes.SERVER_PORT, 6650) | ||
.put(ServerAttributes.SERVER_ADDRESS, "localhost") | ||
.build(); | ||
|
||
Attributes responseAttributes = | ||
Attributes.builder() | ||
.put(MessagingIncubatingAttributes.MESSAGING_MESSAGE_ID, "1:1:0:0") | ||
.put(MessagingIncubatingAttributes.MESSAGING_BATCH_MESSAGE_COUNT, 2) | ||
.build(); | ||
|
||
Context parent = | ||
Context.root() | ||
.with( | ||
Span.wrap( | ||
SpanContext.create( | ||
"ff01020304050600ff0a0b0c0d0e0f00", | ||
"090a0b0c0d0e0f00", | ||
TraceFlags.getSampled(), | ||
TraceState.getDefault()))); | ||
|
||
Context context1 = listener.onStart(parent, requestAttributes, nanos(100)); | ||
|
||
assertThat(metricReader.collectAllMetrics()).isEmpty(); | ||
|
||
Context context2 = listener.onStart(Context.root(), requestAttributes, nanos(150)); | ||
|
||
assertThat(metricReader.collectAllMetrics()).isEmpty(); | ||
|
||
listener.onEnd(context1, responseAttributes, nanos(250)); | ||
|
||
assertThat(metricReader.collectAllMetrics()) | ||
.satisfiesExactlyInAnyOrder( | ||
metric -> | ||
OpenTelemetryAssertions.assertThat(metric) | ||
.hasName("messaging.publish.duration") | ||
.hasUnit("s") | ||
.hasDescription("Measures the duration of publish operation.") | ||
.hasHistogramSatisfying( | ||
histogram -> | ||
histogram.hasPointsSatisfying( | ||
point -> | ||
point | ||
.hasSum(0.15 /* seconds */) | ||
.hasAttributesSatisfying( | ||
equalTo( | ||
MessagingIncubatingAttributes.MESSAGING_SYSTEM, | ||
"pulsar"), | ||
equalTo( | ||
MessagingIncubatingAttributes | ||
.MESSAGING_DESTINATION_NAME, | ||
"persistent://public/default/topic"), | ||
equalTo(ServerAttributes.SERVER_PORT, 6650), | ||
equalTo(ServerAttributes.SERVER_ADDRESS, "localhost")) | ||
.hasExemplarsSatisfying( | ||
exemplar -> | ||
exemplar | ||
.hasTraceId("ff01020304050600ff0a0b0c0d0e0f00") | ||
.hasSpanId("090a0b0c0d0e0f00")) | ||
.hasBucketBoundaries(DURATION_BUCKETS)))); | ||
|
||
listener.onEnd(context2, responseAttributes, nanos(300)); | ||
|
||
assertThat(metricReader.collectAllMetrics()) | ||
.satisfiesExactlyInAnyOrder( | ||
metric -> | ||
OpenTelemetryAssertions.assertThat(metric) | ||
.hasName("messaging.publish.duration") | ||
.hasHistogramSatisfying( | ||
histogram -> | ||
histogram.hasPointsSatisfying( | ||
point -> point.hasSum(0.3 /* seconds */)))); | ||
} | ||
|
||
private static long nanos(int millis) { | ||
return TimeUnit.MILLISECONDS.toNanos(millis); | ||
} | ||
} |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
We don't add incubating semconv dependency to the libraries we publish, instead we copy the constants from the incubating semconv. The reason we do this is that incubating semconv can change in backwards incompatible way. Imagine if you have 2 libraries depending on different versions of incubating semconv, it is possible that a version of incubating semconv that works for both doesn't exist.
Using incubating semconv in javaagent and test code is fine and encouraged.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
thank you for the clarification.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Done; please take a look. Thanks.