Add Pulsar Consumer metrics #11891
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,114 @@ | ||
| /* | ||
| * 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.AttributeKey; | ||
| import io.opentelemetry.api.common.Attributes; | ||
| import io.opentelemetry.api.metrics.DoubleHistogram; | ||
| import io.opentelemetry.api.metrics.DoubleHistogramBuilder; | ||
| import io.opentelemetry.api.metrics.LongCounter; | ||
| import io.opentelemetry.api.metrics.LongCounterBuilder; | ||
| 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; | ||
|
|
||
| /** | ||
| * {@link OperationListener} which keeps track of <a | ||
| * href="https://github.com/open-telemetry/semantic-conventions/blob/v1.26.0/docs/messaging/messaging-metrics.md#consumer-metrics">Consumer | ||
| * metrics</a>. | ||
| */ | ||
| public final class MessagingConsumerMetrics implements OperationListener { | ||
| private static final double NANOS_PER_S = TimeUnit.SECONDS.toNanos(1); | ||
|
|
||
| // copied from MessagingIncubatingAttributes | ||
| private static final AttributeKey<Long> MESSAGING_BATCH_MESSAGE_COUNT = | ||
| AttributeKey.longKey("messaging.batch.message_count"); | ||
| private static final ContextKey<MessagingConsumerMetrics.State> MESSAGING_CONSUMER_METRICS_STATE = | ||
| ContextKey.named("messaging-consumer-metrics-state"); | ||
| private static final Logger logger = Logger.getLogger(MessagingConsumerMetrics.class.getName()); | ||
|
|
||
| private final DoubleHistogram receiveDurationHistogram; | ||
| private final LongCounter receiveMessageCount; | ||
|
|
||
| private MessagingConsumerMetrics(Meter meter) { | ||
| DoubleHistogramBuilder durationBuilder = | ||
| meter | ||
| .histogramBuilder("messaging.receive.duration") | ||
| .setDescription("Measures the duration of receive operation.") | ||
| .setExplicitBucketBoundariesAdvice(MessagingMetricsAdvice.DURATION_SECONDS_BUCKETS) | ||
| .setUnit("s"); | ||
| MessagingMetricsAdvice.applyReceiveDurationAdvice(durationBuilder); | ||
| receiveDurationHistogram = durationBuilder.build(); | ||
|
|
||
| LongCounterBuilder longCounterBuilder = | ||
| meter | ||
| .counterBuilder("messaging.receive.messages") | ||
| .setDescription("Measures the number of received messages.") | ||
| .setUnit("{message}"); | ||
| MessagingMetricsAdvice.applyReceiveMessagesAdvice(longCounterBuilder); | ||
| receiveMessageCount = longCounterBuilder.build(); | ||
| } | ||
|
|
||
| public static OperationMetrics get() { | ||
| return OperationMetricsUtil.create("messaging consumer", MessagingConsumerMetrics::new); | ||
| } | ||
|
|
||
| @Override | ||
| @CanIgnoreReturnValue | ||
| public Context onStart(Context context, Attributes startAttributes, long startNanos) { | ||
| return context.with( | ||
| MESSAGING_CONSUMER_METRICS_STATE, | ||
| new AutoValue_MessagingConsumerMetrics_State(startAttributes, startNanos)); | ||
| } | ||
|
|
||
| @Override | ||
| public void onEnd(Context context, Attributes endAttributes, long endNanos) { | ||
| MessagingConsumerMetrics.State state = context.get(MESSAGING_CONSUMER_METRICS_STATE); | ||
| if (state == null) { | ||
| logger.log( | ||
| FINE, | ||
| "No state present when ending context {0}. Cannot record consumer receive metrics.", | ||
| context); | ||
| return; | ||
| } | ||
|
|
||
| Attributes attributes = state.startAttributes().toBuilder().putAll(endAttributes).build(); | ||
| receiveDurationHistogram.record( | ||
| (endNanos - state.startTimeNanos()) / NANOS_PER_S, attributes, context); | ||
|
|
||
| Long receiveMessagesCount = getReceiveMessagesCount(state.startAttributes(), endAttributes); | ||
| if (receiveMessagesCount != null) { | ||
| receiveMessageCount.add(receiveMessagesCount, attributes, context); | ||
| } | ||
|
There was a problem hiding this comment. I think this isn't quite correct. We should record the count of all messages but currently we are recording only the count of batch messages. Perhaps There was a problem hiding this comment.
This change will cause the There was a problem hiding this comment. Imo you are misinterpreting the spec. It says that this metric is not required when messaging system does not support batch receives because the receive count can be obtained from a different metric. The metric description says There was a problem hiding this comment. Thank you for your addition. There was a problem hiding this comment. let's align with the latest semconv which I think answer this question? |
||
| } | ||
|
|
||
| private static Long getReceiveMessagesCount(Attributes... attributesList) { | ||
| for (Attributes attributes : attributesList) { | ||
| Long value = attributes.get(MESSAGING_BATCH_MESSAGE_COUNT); | ||
| if (value != null) { | ||
| return value; | ||
| } | ||
| } | ||
| return null; | ||
| } | ||
|
|
||
| @AutoValue | ||
| abstract static class State { | ||
|
|
||
| abstract Attributes startAttributes(); | ||
|
|
||
| abstract long startTimeNanos(); | ||
| } | ||
| } | ||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
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.
should we go ahead and update to the latest spec to avoid additional churn? https://github.com/open-telemetry/semantic-conventions/blob/main/docs/messaging/messaging-metrics.md
cc @open-telemetry/semconv-messaging-approvers
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.
The
semConvVersionhasn't been upgraded to the latest version yet. Is it okay to only upgrade some semantics?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.
yes it's ok, if we don't have the new constants available yet, you can create static final constants in this class and reference those