Skip to content

Commit 7c21537

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

12 files changed

Lines changed: 977 additions & 116 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: 111 additions & 28 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,37 @@ 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;
44-
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-
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();
48+
@Nullable private final DoubleHistogram receiveDurationHistogram;
49+
@Nullable private final LongCounter receiveMessageCount;
50+
@Nullable private final DoubleHistogram clientOperationDurationHistogram;
51+
@Nullable private final LongCounter consumedMessagesCounter;
52+
53+
private MessagingConsumerMetrics(Meter meter, boolean supportsStableSemconv) {
54+
boolean emitOldSemconv = !supportsStableSemconv || emitOldMessagingSemconv();
55+
boolean emitStableSemconv = supportsStableSemconv && emitStableMessagingSemconv();
56+
receiveDurationHistogram = emitOldSemconv ? buildReceiveDuration(meter) : null;
57+
receiveMessageCount = emitOldSemconv ? buildReceiveMessages(meter) : null;
58+
clientOperationDurationHistogram =
59+
emitStableSemconv ? buildClientOperationDuration(meter) : null;
60+
consumedMessagesCounter = emitStableSemconv ? buildConsumedMessages(meter) : null;
6261
}
6362

63+
/** Returns metrics for extractors configured with {@link MessageOperation}. */
6464
public static OperationMetrics get() {
65-
return OperationMetricsUtil.create("messaging consumer", MessagingConsumerMetrics::new);
65+
return OperationMetricsUtil.create(
66+
"messaging consumer", meter -> new MessagingConsumerMetrics(meter, false));
67+
}
68+
69+
/** Returns metrics for extractors configured with {@link MessagingOperationType}. */
70+
public static OperationMetrics getForOperationType() {
71+
return OperationMetricsUtil.create(
72+
"messaging consumer", meter -> new MessagingConsumerMetrics(meter, true));
6673
}
6774

6875
@Override
@@ -85,21 +92,97 @@ public void onEnd(Context context, Attributes endAttributes, long endNanos) {
8592
}
8693

8794
Attributes attributes = state.startAttributes().toBuilder().putAll(endAttributes).build();
88-
receiveDurationHistogram.record(
89-
(endNanos - state.startTimeNanos()) / NANOS_PER_S, attributes, context);
95+
double duration = (endNanos - state.startTimeNanos()) / NANOS_PER_S;
96+
if (receiveDurationHistogram != null) {
97+
receiveDurationHistogram.record(duration, attributes, context);
98+
}
99+
// Metric view attribute advice can only select keys statically. The concrete destination name
100+
// must be omitted when a template is available or the destination is temporary or anonymous,
101+
// so this conditional requirement must be enforced before recording.
102+
Attributes filteredAttributes =
103+
clientOperationDurationHistogram != null || consumedMessagesCounter != null
104+
? MessagingMetricsAdvice.filterAttributes(attributes)
105+
: attributes;
106+
String operationType = attributes.get(MESSAGING_OPERATION_TYPE);
107+
if (clientOperationDurationHistogram != null
108+
&& !MessagingOperationType.PROCESS.value().equals(operationType)) {
109+
clientOperationDurationHistogram.record(duration, filteredAttributes, context);
110+
}
90111

91-
long receiveMessagesCount = getReceiveMessagesCount(state.startAttributes(), endAttributes);
92-
receiveMessageCount.add(receiveMessagesCount, attributes, context);
112+
Long batchMessageCount = getBatchMessageCount(state.startAttributes(), endAttributes);
113+
if (receiveMessageCount != null) {
114+
receiveMessageCount.add(
115+
batchMessageCount == null ? 1 : batchMessageCount, attributes, context);
116+
}
117+
if (consumedMessagesCounter != null
118+
&& MessagingOperationType.RECEIVE.value().equals(operationType)) {
119+
long consumedMessagesCount = getConsumedMessagesCount(attributes, batchMessageCount);
120+
if (consumedMessagesCount > 0) {
121+
consumedMessagesCounter.add(consumedMessagesCount, filteredAttributes, context);
122+
}
123+
}
93124
}
94125

95-
private static long getReceiveMessagesCount(Attributes... attributesList) {
126+
@Nullable
127+
private static Long getBatchMessageCount(Attributes... attributesList) {
96128
for (Attributes attributes : attributesList) {
97129
Long value = attributes.get(MESSAGING_BATCH_MESSAGE_COUNT);
98130
if (value != null) {
99131
return value;
100132
}
101133
}
102-
return 1;
134+
return null;
135+
}
136+
137+
private static long getConsumedMessagesCount(
138+
Attributes attributes, @Nullable Long batchMessageCount) {
139+
if (batchMessageCount != null) {
140+
return batchMessageCount;
141+
}
142+
return attributes.get(ERROR_TYPE) == null ? 1 : 0;
143+
}
144+
145+
private static DoubleHistogram buildReceiveDuration(Meter meter) {
146+
DoubleHistogramBuilder builder =
147+
meter
148+
.histogramBuilder("messaging.receive.duration")
149+
.setDescription("Measures the duration of receive operation.")
150+
.setExplicitBucketBoundariesAdvice(MessagingMetricsAdvice.DURATION_SECONDS_BUCKETS)
151+
.setUnit("s");
152+
MessagingMetricsAdvice.applyOldDurationAdvice(builder);
153+
return builder.build();
154+
}
155+
156+
private static LongCounter buildReceiveMessages(Meter meter) {
157+
LongCounterBuilder builder =
158+
meter
159+
.counterBuilder("messaging.receive.messages")
160+
.setDescription("Measures the number of received messages.")
161+
.setUnit("{message}");
162+
MessagingMetricsAdvice.applyOldMessagesAdvice(builder);
163+
return builder.build();
164+
}
165+
166+
private static DoubleHistogram buildClientOperationDuration(Meter meter) {
167+
DoubleHistogramBuilder builder =
168+
meter
169+
.histogramBuilder("messaging.client.operation.duration")
170+
.setDescription(
171+
"Duration of messaging operation initiated by a producer or consumer client.")
172+
.setExplicitBucketBoundariesAdvice(MessagingMetricsAdvice.DURATION_SECONDS_BUCKETS)
173+
.setUnit("s");
174+
MessagingMetricsAdvice.applyClientOperationDurationAdvice(builder);
175+
return builder.build();
176+
}
177+
178+
private static LongCounter buildConsumedMessages(Meter meter) {
179+
LongCounterBuilder builder =
180+
meter
181+
.counterBuilder("messaging.client.consumed.messages")
182+
.setDescription("Number of messages that were delivered to the application.")
183+
.setUnit("{message}");
184+
MessagingMetricsAdvice.applyConsumedMessagesAdvice(builder);
185+
return builder.build();
103186
}
104187

105188
@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)