Skip to content

Commit 3e64403

Browse files
trasktrask
committed
Add messaging v1.43 shared metrics
1 parent 653c0b1 commit 3e64403

12 files changed

Lines changed: 1037 additions & 114 deletions

‎instrumentation-api-incubator/src/main/java/io/opentelemetry/instrumentation/api/incubator/semconv/messaging/MessagingAttributesExtractorBuilder.java‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -38,6 +38,9 @@ public final class MessagingAttributesExtractorBuilder<REQUEST, RESPONSE> {
3838
@CanIgnoreReturnValue
3939
public MessagingAttributesExtractorBuilder<REQUEST, RESPONSE> setOperationName(
4040
String operationName) {
41+
if (!supportsStableSemconv) {
42+
throw new IllegalStateException("Operation name is not configurable for legacy builders");
43+
}
4144
this.operationName = requireNonNull(operationName, "operationName");
4245
return this;
4346
}

‎instrumentation-api-incubator/src/main/java/io/opentelemetry/instrumentation/api/incubator/semconv/messaging/MessagingConsumerMetrics.java‎

Lines changed: 125 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,9 @@
55

66
package io.opentelemetry.instrumentation.api.incubator.semconv.messaging;
77

8+
import static io.opentelemetry.instrumentation.api.internal.SemconvStability.emitOldMessagingSemconv;
9+
import static io.opentelemetry.instrumentation.api.internal.SemconvStability.emitStableMessagingSemconv;
10+
import static io.opentelemetry.semconv.ErrorAttributes.ERROR_TYPE;
811
import static java.util.concurrent.TimeUnit.SECONDS;
912
import static java.util.logging.Level.FINE;
1013

@@ -23,10 +26,11 @@
2326
import io.opentelemetry.instrumentation.api.instrumenter.OperationMetrics;
2427
import io.opentelemetry.instrumentation.api.internal.OperationMetricsUtil;
2528
import java.util.logging.Logger;
29+
import javax.annotation.Nullable;
2630

2731
/**
2832
* {@link OperationListener} which keeps track of <a
29-
* href="https://github.com/open-telemetry/semantic-conventions/blob/v1.26.0/docs/messaging/messaging-metrics.md#consumer-metrics">Consumer
33+
* href="https://github.com/open-telemetry/semantic-conventions/blob/v1.43.0/docs/messaging/messaging-metrics.md#consumer-metrics">consumer
3034
* metrics</a>.
3135
*/
3236
public final class MessagingConsumerMetrics implements OperationListener {
@@ -35,34 +39,52 @@ public final class MessagingConsumerMetrics implements OperationListener {
3539
// copied from MessagingIncubatingAttributes
3640
private static final AttributeKey<Long> MESSAGING_BATCH_MESSAGE_COUNT =
3741
AttributeKey.longKey("messaging.batch.message_count");
42+
private static final AttributeKey<String> MESSAGING_OPERATION_TYPE =
43+
AttributeKey.stringKey("messaging.operation.type");
3844
private static final ContextKey<MessagingConsumerMetrics.State> MESSAGING_CONSUMER_METRICS_STATE =
3945
ContextKey.named("messaging-consumer-metrics-state");
4046
private static final Logger logger = Logger.getLogger(MessagingConsumerMetrics.class.getName());
4147

42-
private final DoubleHistogram receiveDurationHistogram;
43-
private final LongCounter receiveMessageCount;
48+
private final boolean consumedMessagesOnly;
49+
@Nullable private final DoubleHistogram receiveDurationHistogram;
50+
@Nullable private final LongCounter receiveMessageCount;
51+
@Nullable private final DoubleHistogram clientOperationDurationHistogram;
52+
@Nullable private final LongCounter consumedMessagesCounter;
4453

45-
private MessagingConsumerMetrics(Meter meter) {
46-
DoubleHistogramBuilder durationBuilder =
47-
meter
48-
.histogramBuilder("messaging.receive.duration")
49-
.setDescription("Measures the duration of receive operation.")
50-
.setExplicitBucketBoundariesAdvice(MessagingMetricsAdvice.DURATION_SECONDS_BUCKETS)
51-
.setUnit("s");
52-
MessagingMetricsAdvice.applyReceiveDurationAdvice(durationBuilder);
53-
receiveDurationHistogram = durationBuilder.build();
54+
private MessagingConsumerMetrics(Meter meter, boolean supportsStableSemconv) {
55+
this(meter, supportsStableSemconv, false);
56+
}
5457

55-
LongCounterBuilder longCounterBuilder =
56-
meter
57-
.counterBuilder("messaging.receive.messages")
58-
.setDescription("Measures the number of received messages.")
59-
.setUnit("{message}");
60-
MessagingMetricsAdvice.applyReceiveMessagesAdvice(longCounterBuilder);
61-
receiveMessageCount = longCounterBuilder.build();
58+
private MessagingConsumerMetrics(
59+
Meter meter, boolean supportsStableSemconv, boolean consumedMessagesOnly) {
60+
this.consumedMessagesOnly = consumedMessagesOnly;
61+
boolean emitOldSemconv = !supportsStableSemconv || emitOldMessagingSemconv();
62+
boolean emitStableSemconv = supportsStableSemconv && emitStableMessagingSemconv();
63+
receiveDurationHistogram =
64+
!consumedMessagesOnly && emitOldSemconv ? buildReceiveDuration(meter) : null;
65+
receiveMessageCount =
66+
!consumedMessagesOnly && emitOldSemconv ? buildReceiveMessages(meter) : null;
67+
clientOperationDurationHistogram =
68+
!consumedMessagesOnly && emitStableSemconv ? buildClientOperationDuration(meter) : null;
69+
consumedMessagesCounter = emitStableSemconv ? buildConsumedMessages(meter) : null;
6270
}
6371

72+
/** Returns metrics for extractors configured with {@link MessageOperation}. */
6473
public static OperationMetrics get() {
65-
return OperationMetricsUtil.create("messaging consumer", MessagingConsumerMetrics::new);
74+
return OperationMetricsUtil.create(
75+
"messaging consumer", meter -> new MessagingConsumerMetrics(meter, false));
76+
}
77+
78+
/** Returns metrics for extractors configured with {@link MessagingOperationType}. */
79+
public static OperationMetrics getForOperationType() {
80+
return OperationMetricsUtil.create(
81+
"messaging consumer", meter -> new MessagingConsumerMetrics(meter, true));
82+
}
83+
84+
/** Returns only the stable consumed-messages metric for a delivered message. */
85+
public static OperationMetrics getConsumedMessages() {
86+
return OperationMetricsUtil.create(
87+
"messaging consumed messages", meter -> new MessagingConsumerMetrics(meter, true, true));
6688
}
6789

6890
@Override
@@ -85,21 +107,98 @@ public void onEnd(Context context, Attributes endAttributes, long endNanos) {
85107
}
86108

87109
Attributes attributes = state.startAttributes().toBuilder().putAll(endAttributes).build();
88-
receiveDurationHistogram.record(
89-
(endNanos - state.startTimeNanos()) / NANOS_PER_S, attributes, context);
110+
double duration = (endNanos - state.startTimeNanos()) / NANOS_PER_S;
111+
if (receiveDurationHistogram != null) {
112+
receiveDurationHistogram.record(duration, attributes, context);
113+
}
114+
// Metric view attribute advice can only select keys statically. The concrete destination name
115+
// must be omitted when a template is available or the destination is temporary or anonymous,
116+
// so this conditional requirement must be enforced before recording.
117+
Attributes filteredAttributes =
118+
clientOperationDurationHistogram != null || consumedMessagesCounter != null
119+
? MessagingMetricsAdvice.filterAttributes(attributes)
120+
: attributes;
121+
String operationType = attributes.get(MESSAGING_OPERATION_TYPE);
122+
if (clientOperationDurationHistogram != null
123+
&& !MessagingOperationType.PROCESS.value().equals(operationType)) {
124+
clientOperationDurationHistogram.record(duration, filteredAttributes, context);
125+
}
90126

91-
long receiveMessagesCount = getReceiveMessagesCount(state.startAttributes(), endAttributes);
92-
receiveMessageCount.add(receiveMessagesCount, attributes, context);
127+
Long batchMessageCount = getBatchMessageCount(state.startAttributes(), endAttributes);
128+
if (receiveMessageCount != null) {
129+
receiveMessageCount.add(
130+
batchMessageCount == null ? 1 : batchMessageCount, attributes, context);
131+
}
132+
if (consumedMessagesCounter != null
133+
&& (consumedMessagesOnly || MessagingOperationType.RECEIVE.value().equals(operationType))) {
134+
long consumedMessagesCount =
135+
getConsumedMessagesCount(attributes, batchMessageCount, consumedMessagesOnly);
136+
if (consumedMessagesCount > 0) {
137+
consumedMessagesCounter.add(consumedMessagesCount, filteredAttributes, context);
138+
}
139+
}
93140
}
94141

95-
private static long getReceiveMessagesCount(Attributes... attributesList) {
142+
@Nullable
143+
private static Long getBatchMessageCount(Attributes... attributesList) {
96144
for (Attributes attributes : attributesList) {
97145
Long value = attributes.get(MESSAGING_BATCH_MESSAGE_COUNT);
98146
if (value != null) {
99147
return value;
100148
}
101149
}
102-
return 1;
150+
return null;
151+
}
152+
153+
private static long getConsumedMessagesCount(
154+
Attributes attributes, @Nullable Long batchMessageCount, boolean consumedMessagesOnly) {
155+
if (batchMessageCount != null) {
156+
return batchMessageCount;
157+
}
158+
return consumedMessagesOnly || attributes.get(ERROR_TYPE) == null ? 1 : 0;
159+
}
160+
161+
private static DoubleHistogram buildReceiveDuration(Meter meter) {
162+
DoubleHistogramBuilder builder =
163+
meter
164+
.histogramBuilder("messaging.receive.duration")
165+
.setDescription("Measures the duration of receive operation.")
166+
.setExplicitBucketBoundariesAdvice(MessagingMetricsAdvice.DURATION_SECONDS_BUCKETS)
167+
.setUnit("s");
168+
MessagingMetricsAdvice.applyOldDurationAdvice(builder);
169+
return builder.build();
170+
}
171+
172+
private static LongCounter buildReceiveMessages(Meter meter) {
173+
LongCounterBuilder builder =
174+
meter
175+
.counterBuilder("messaging.receive.messages")
176+
.setDescription("Measures the number of received messages.")
177+
.setUnit("{message}");
178+
MessagingMetricsAdvice.applyOldMessagesAdvice(builder);
179+
return builder.build();
180+
}
181+
182+
private static DoubleHistogram buildClientOperationDuration(Meter meter) {
183+
DoubleHistogramBuilder builder =
184+
meter
185+
.histogramBuilder("messaging.client.operation.duration")
186+
.setDescription(
187+
"Duration of messaging operation initiated by a producer or consumer client.")
188+
.setExplicitBucketBoundariesAdvice(MessagingMetricsAdvice.DURATION_SECONDS_BUCKETS)
189+
.setUnit("s");
190+
MessagingMetricsAdvice.applyClientOperationDurationAdvice(builder);
191+
return builder.build();
192+
}
193+
194+
private static LongCounter buildConsumedMessages(Meter meter) {
195+
LongCounterBuilder builder =
196+
meter
197+
.counterBuilder("messaging.client.consumed.messages")
198+
.setDescription("Number of messages that were delivered to the application.")
199+
.setUnit("{message}");
200+
MessagingMetricsAdvice.applyConsumedMessagesAdvice(builder);
201+
return builder.build();
103202
}
104203

105204
@AutoValue

‎instrumentation-api-incubator/src/main/java/io/opentelemetry/instrumentation/api/incubator/semconv/messaging/MessagingMetricsAdvice.java‎

Lines changed: 86 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -12,10 +12,12 @@
1212
import static java.util.Collections.unmodifiableList;
1313

1414
import io.opentelemetry.api.common.AttributeKey;
15+
import io.opentelemetry.api.common.Attributes;
1516
import io.opentelemetry.api.incubator.metrics.ExtendedDoubleHistogramBuilder;
1617
import io.opentelemetry.api.incubator.metrics.ExtendedLongCounterBuilder;
1718
import io.opentelemetry.api.metrics.DoubleHistogramBuilder;
1819
import io.opentelemetry.api.metrics.LongCounterBuilder;
20+
import java.util.ArrayList;
1921
import java.util.List;
2022

2123
final class MessagingMetricsAdvice {
@@ -28,14 +30,26 @@ final class MessagingMetricsAdvice {
2830
AttributeKey.stringKey("messaging.system");
2931
private static final AttributeKey<String> MESSAGING_DESTINATION_NAME =
3032
AttributeKey.stringKey("messaging.destination.name");
33+
private static final AttributeKey<Boolean> MESSAGING_DESTINATION_ANONYMOUS =
34+
AttributeKey.booleanKey("messaging.destination.anonymous");
35+
private static final AttributeKey<Boolean> MESSAGING_DESTINATION_TEMPORARY =
36+
AttributeKey.booleanKey("messaging.destination.temporary");
3137
private static final AttributeKey<String> MESSAGING_OPERATION =
3238
AttributeKey.stringKey("messaging.operation");
39+
private static final AttributeKey<String> MESSAGING_OPERATION_NAME =
40+
AttributeKey.stringKey("messaging.operation.name");
41+
private static final AttributeKey<String> MESSAGING_OPERATION_TYPE =
42+
AttributeKey.stringKey("messaging.operation.type");
43+
private static final AttributeKey<String> MESSAGING_CONSUMER_GROUP_NAME =
44+
AttributeKey.stringKey("messaging.consumer.group.name");
45+
private static final AttributeKey<String> MESSAGING_DESTINATION_SUBSCRIPTION_NAME =
46+
AttributeKey.stringKey("messaging.destination.subscription.name");
3347
private static final AttributeKey<String> MESSAGING_DESTINATION_PARTITION_ID =
3448
AttributeKey.stringKey("messaging.destination.partition.id");
3549
private static final AttributeKey<String> MESSAGING_DESTINATION_TEMPLATE =
3650
AttributeKey.stringKey("messaging.destination.template");
3751

38-
private static final List<AttributeKey<?>> MESSAGING_ATTRIBUTES =
52+
private static final List<AttributeKey<?>> OLD_ATTRIBUTES =
3953
asList(
4054
MESSAGING_SYSTEM,
4155
MESSAGING_DESTINATION_NAME,
@@ -46,25 +60,90 @@ final class MessagingMetricsAdvice {
4660
SERVER_PORT,
4761
SERVER_ADDRESS);
4862

49-
static void applyPublishDurationAdvice(DoubleHistogramBuilder builder) {
63+
private static final List<AttributeKey<?>> CLIENT_OPERATION_DURATION_ATTRIBUTES =
64+
buildAttributes(true, true, true);
65+
private static final List<AttributeKey<?>> SENT_MESSAGES_ATTRIBUTES =
66+
buildAttributes(false, false, true);
67+
private static final List<AttributeKey<?>> CONSUMED_MESSAGES_ATTRIBUTES =
68+
buildAttributes(true, false, true);
69+
private static final List<AttributeKey<?>> PROCESS_DURATION_ATTRIBUTES =
70+
buildAttributes(true, false, true);
71+
72+
private static List<AttributeKey<?>> buildAttributes(
73+
boolean includeConsumerAttributes, boolean includeOperationType, boolean includeErrorType) {
74+
List<AttributeKey<?>> attributes = new ArrayList<>();
75+
attributes.add(MESSAGING_OPERATION_NAME);
76+
attributes.add(MESSAGING_SYSTEM);
77+
if (includeErrorType) {
78+
attributes.add(ERROR_TYPE);
79+
}
80+
if (includeConsumerAttributes) {
81+
attributes.add(MESSAGING_CONSUMER_GROUP_NAME);
82+
}
83+
attributes.add(MESSAGING_DESTINATION_NAME);
84+
if (includeConsumerAttributes) {
85+
attributes.add(MESSAGING_DESTINATION_SUBSCRIPTION_NAME);
86+
}
87+
attributes.add(MESSAGING_DESTINATION_TEMPLATE);
88+
if (includeOperationType) {
89+
attributes.add(MESSAGING_OPERATION_TYPE);
90+
}
91+
attributes.add(SERVER_ADDRESS);
92+
attributes.add(MESSAGING_DESTINATION_PARTITION_ID);
93+
attributes.add(SERVER_PORT);
94+
return unmodifiableList(attributes);
95+
}
96+
97+
static Attributes filterAttributes(Attributes attributes) {
98+
if (attributes.get(MESSAGING_DESTINATION_TEMPLATE) == null
99+
&& !Boolean.TRUE.equals(attributes.get(MESSAGING_DESTINATION_ANONYMOUS))
100+
&& !Boolean.TRUE.equals(attributes.get(MESSAGING_DESTINATION_TEMPORARY))) {
101+
return attributes;
102+
}
103+
return attributes.toBuilder().remove(MESSAGING_DESTINATION_NAME).build();
104+
}
105+
106+
static void applyOldDurationAdvice(DoubleHistogramBuilder builder) {
50107
if (!(builder instanceof ExtendedDoubleHistogramBuilder)) {
51108
return;
52109
}
53-
((ExtendedDoubleHistogramBuilder) builder).setAttributesAdvice(MESSAGING_ATTRIBUTES);
110+
((ExtendedDoubleHistogramBuilder) builder).setAttributesAdvice(OLD_ATTRIBUTES);
54111
}
55112

56-
static void applyReceiveDurationAdvice(DoubleHistogramBuilder builder) {
113+
static void applyClientOperationDurationAdvice(DoubleHistogramBuilder builder) {
57114
if (!(builder instanceof ExtendedDoubleHistogramBuilder)) {
58115
return;
59116
}
60-
((ExtendedDoubleHistogramBuilder) builder).setAttributesAdvice(MESSAGING_ATTRIBUTES);
117+
((ExtendedDoubleHistogramBuilder) builder)
118+
.setAttributesAdvice(CLIENT_OPERATION_DURATION_ATTRIBUTES);
119+
}
120+
121+
static void applyProcessDurationAdvice(DoubleHistogramBuilder builder) {
122+
if (!(builder instanceof ExtendedDoubleHistogramBuilder)) {
123+
return;
124+
}
125+
((ExtendedDoubleHistogramBuilder) builder).setAttributesAdvice(PROCESS_DURATION_ATTRIBUTES);
126+
}
127+
128+
static void applyOldMessagesAdvice(LongCounterBuilder builder) {
129+
if (!(builder instanceof ExtendedLongCounterBuilder)) {
130+
return;
131+
}
132+
((ExtendedLongCounterBuilder) builder).setAttributesAdvice(OLD_ATTRIBUTES);
133+
}
134+
135+
static void applySentMessagesAdvice(LongCounterBuilder builder) {
136+
if (!(builder instanceof ExtendedLongCounterBuilder)) {
137+
return;
138+
}
139+
((ExtendedLongCounterBuilder) builder).setAttributesAdvice(SENT_MESSAGES_ATTRIBUTES);
61140
}
62141

63-
static void applyReceiveMessagesAdvice(LongCounterBuilder builder) {
142+
static void applyConsumedMessagesAdvice(LongCounterBuilder builder) {
64143
if (!(builder instanceof ExtendedLongCounterBuilder)) {
65144
return;
66145
}
67-
((ExtendedLongCounterBuilder) builder).setAttributesAdvice(MESSAGING_ATTRIBUTES);
146+
((ExtendedLongCounterBuilder) builder).setAttributesAdvice(CONSUMED_MESSAGES_ATTRIBUTES);
68147
}
69148

70149
private MessagingMetricsAdvice() {}

0 commit comments

Comments
 (0)