Skip to content

Commit 60dbca4

Browse files
luke-sterkowiczluke-sterkowicz
committed
Fixes and options for gRPC instrumentation
1 parent f49bf3c commit 60dbca4

File tree

9 files changed

+173
-75
lines changed

9 files changed

+173
-75
lines changed

‎instrumentation/armeria/armeria-grpc-1.14/javaagent/src/test/java/io/opentelemetry/javaagent/instrumentation/armeria/grpc/v1_14/ArmeriaGrpcTest.java

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -98,7 +98,7 @@ void grpcInstrumentation() {
9898
.hasAttributesSatisfyingExactly(
9999
equalTo(
100100
MessageIncubatingAttributes.MESSAGE_TYPE, "RECEIVED"),
101-
equalTo(MessageIncubatingAttributes.MESSAGE_ID, 2L))),
101+
equalTo(MessageIncubatingAttributes.MESSAGE_ID, 1L))),
102102
span ->
103103
span.hasName("example.Greeter/SayHello")
104104
.hasKind(SpanKind.SERVER)
@@ -123,6 +123,6 @@ void grpcInstrumentation() {
123123
.hasName("message")
124124
.hasAttributesSatisfyingExactly(
125125
equalTo(MessageIncubatingAttributes.MESSAGE_TYPE, "SENT"),
126-
equalTo(MessageIncubatingAttributes.MESSAGE_ID, 2L)))));
126+
equalTo(MessageIncubatingAttributes.MESSAGE_ID, 1L)))));
127127
}
128128
}

‎instrumentation/grpc-1.6/README.md

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@
22

33
| System property | Type | Default | Description |
44
|-------------------------------------------------------------|---------|---------|----------------------------------------------------------------------------------------------------------------------------------------------------------------|
5+
| `otel.instrumentation.grpc.message-events` | Boolean | `true` | Determines whether to add span event for each individual message received and sent. Set this to false in case of streaming large volumes of messages. |
56
| `otel.instrumentation.grpc.experimental-span-attributes` | Boolean | `false` | Enable the capture of experimental span attributes. |
67
| `otel.instrumentation.grpc.capture-metadata.client.request` | String | | A comma-separated list of request metadata keys. gRPC client instrumentation will capture metadata values corresponding to configured keys as span attributes. |
78
| `otel.instrumentation.grpc.capture-metadata.server.request` | String | | A comma-separated list of request metadata keys. gRPC server instrumentation will capture metadata values corresponding to configured keys as span attributes. |

‎instrumentation/grpc-1.6/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/grpc/v1_6/GrpcSingletons.java

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,10 @@ public final class GrpcSingletons {
2727
private static final AtomicReference<Context.Storage> STORAGE_REFERENCE = new AtomicReference<>();
2828

2929
static {
30+
boolean addMessageEvents =
31+
AgentInstrumentationConfig.get()
32+
.getBoolean("otel.instrumentation.grpc.message-events", true);
33+
3034
boolean experimentalSpanAttributes =
3135
AgentInstrumentationConfig.get()
3236
.getBoolean("otel.instrumentation.grpc.experimental-span-attributes", false);
@@ -40,6 +44,7 @@ public final class GrpcSingletons {
4044

4145
GrpcTelemetry telemetry =
4246
GrpcTelemetry.builder(GlobalOpenTelemetry.get())
47+
.setAddMessageEvents(addMessageEvents)
4348
.setCaptureExperimentalSpanAttributes(experimentalSpanAttributes)
4449
.setCapturedClientRequestMetadata(clientRequestMetadata)
4550
.setCapturedServerRequestMetadata(serverRequestMetadata)

‎instrumentation/grpc-1.6/library/src/main/java/io/opentelemetry/instrumentation/grpc/v1_6/GrpcTelemetry.java

Lines changed: 8 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -29,31 +29,36 @@ public static GrpcTelemetryBuilder builder(OpenTelemetry openTelemetry) {
2929
private final Instrumenter<GrpcRequest, Status> clientInstrumenter;
3030
private final ContextPropagators propagators;
3131
private final boolean captureExperimentalSpanAttributes;
32+
private final boolean addMessageEvents;
3233

3334
GrpcTelemetry(
3435
Instrumenter<GrpcRequest, Status> serverInstrumenter,
3536
Instrumenter<GrpcRequest, Status> clientInstrumenter,
3637
ContextPropagators propagators,
37-
boolean captureExperimentalSpanAttributes) {
38+
boolean captureExperimentalSpanAttributes,
39+
boolean addMessageEvents) {
3840
this.serverInstrumenter = serverInstrumenter;
3941
this.clientInstrumenter = clientInstrumenter;
4042
this.propagators = propagators;
4143
this.captureExperimentalSpanAttributes = captureExperimentalSpanAttributes;
44+
this.addMessageEvents = addMessageEvents;
4245
}
4346

4447
/**
4548
* Returns a new {@link ClientInterceptor} for use with methods like {@link
4649
* io.grpc.ManagedChannelBuilder#intercept(ClientInterceptor...)}.
4750
*/
4851
public ClientInterceptor newClientInterceptor() {
49-
return new TracingClientInterceptor(clientInstrumenter, propagators);
52+
return new TracingClientInterceptor(
53+
clientInstrumenter, propagators, captureExperimentalSpanAttributes, addMessageEvents);
5054
}
5155

5256
/**
5357
* Returns a new {@link ServerInterceptor} for use with methods like {@link
5458
* io.grpc.ServerBuilder#intercept(ServerInterceptor)}.
5559
*/
5660
public ServerInterceptor newServerInterceptor() {
57-
return new TracingServerInterceptor(serverInstrumenter, captureExperimentalSpanAttributes);
61+
return new TracingServerInterceptor(
62+
serverInstrumenter, captureExperimentalSpanAttributes, addMessageEvents);
5863
}
5964
}

‎instrumentation/grpc-1.6/library/src/main/java/io/opentelemetry/instrumentation/grpc/v1_6/GrpcTelemetryBuilder.java

Lines changed: 13 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -49,6 +49,7 @@ public final class GrpcTelemetryBuilder {
4949
additionalServerExtractors = new ArrayList<>();
5050

5151
private boolean captureExperimentalSpanAttributes;
52+
private boolean addMessageEvents = true;
5253
private List<String> capturedClientRequestMetadata = Collections.emptyList();
5354
private List<String> capturedServerRequestMetadata = Collections.emptyList();
5455

@@ -130,6 +131,16 @@ public GrpcTelemetryBuilder setPeerService(String peerService) {
130131
return this;
131132
}
132133

134+
/**
135+
* Determines whether to add span event for each individual message received and sent. The default
136+
* is true. Set this to false in case of streaming large volumes of messages.
137+
*/
138+
@CanIgnoreReturnValue
139+
public GrpcTelemetryBuilder setAddMessageEvents(boolean addMessageEvents) {
140+
this.addMessageEvents = addMessageEvents;
141+
return this;
142+
}
143+
133144
/**
134145
* Sets whether experimental attributes should be set to spans. These attributes may be changed or
135146
* removed in the future, so only enable this if you know you do not require attributes filled by
@@ -211,6 +222,7 @@ public GrpcTelemetry build() {
211222
// So we go ahead and inject manually in this instrumentation.
212223
clientInstrumenterBuilder.buildInstrumenter(SpanKindExtractor.alwaysClient()),
213224
openTelemetry.getPropagators(),
214-
captureExperimentalSpanAttributes);
225+
captureExperimentalSpanAttributes,
226+
addMessageEvents);
215227
}
216228
}

‎instrumentation/grpc-1.6/library/src/main/java/io/opentelemetry/instrumentation/grpc/v1_6/TracingClientInterceptor.java

Lines changed: 37 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -34,16 +34,27 @@ final class TracingClientInterceptor implements ClientInterceptor {
3434
private static final String RECEIVED = "RECEIVED";
3535

3636
@SuppressWarnings("rawtypes")
37-
private static final AtomicLongFieldUpdater<TracingClientCall> MESSAGE_ID_UPDATER =
38-
AtomicLongFieldUpdater.newUpdater(TracingClientCall.class, "messageId");
37+
private static final AtomicLongFieldUpdater<TracingClientCall> SENT_MESSAGE_ID_UPDATER =
38+
AtomicLongFieldUpdater.newUpdater(TracingClientCall.class, "sentMessageId");
39+
40+
@SuppressWarnings("rawtypes")
41+
private static final AtomicLongFieldUpdater<TracingClientCall> RECEIVED_MESSAGE_ID_UPDATER =
42+
AtomicLongFieldUpdater.newUpdater(TracingClientCall.class, "receivedMessageId");
3943

4044
private final Instrumenter<GrpcRequest, Status> instrumenter;
4145
private final ContextPropagators propagators;
46+
private final boolean captureExperimentalSpanAttributes;
47+
private final boolean addMessageEvents;
4248

4349
TracingClientInterceptor(
44-
Instrumenter<GrpcRequest, Status> instrumenter, ContextPropagators propagators) {
50+
Instrumenter<GrpcRequest, Status> instrumenter,
51+
ContextPropagators propagators,
52+
boolean captureExperimentalSpanAttributes,
53+
boolean addMessageEvents) {
4554
this.instrumenter = instrumenter;
4655
this.propagators = propagators;
56+
this.captureExperimentalSpanAttributes = captureExperimentalSpanAttributes;
57+
this.addMessageEvents = addMessageEvents;
4758
}
4859

4960
@Override
@@ -75,9 +86,13 @@ final class TracingClientCall<REQUEST, RESPONSE>
7586
private final Context context;
7687
private final GrpcRequest request;
7788

78-
// Used by MESSAGE_ID_UPDATER
89+
// Used by SENT_MESSAGE_ID_UPDATER
90+
@SuppressWarnings("UnusedVariable")
91+
volatile long sentMessageId;
92+
93+
// Used by RECEIVED_MESSAGE_ID_UPDATER
7994
@SuppressWarnings("UnusedVariable")
80-
volatile long messageId;
95+
volatile long receivedMessageId;
8196

8297
TracingClientCall(
8398
ClientCall<REQUEST, RESPONSE> delegate,
@@ -113,10 +128,11 @@ public void sendMessage(REQUEST message) {
113128
instrumenter.end(context, request, Status.UNKNOWN, e);
114129
throw e;
115130
}
116-
Span span = Span.fromContext(context);
117-
Attributes attributes =
118-
Attributes.of(MESSAGE_TYPE, SENT, MESSAGE_ID, MESSAGE_ID_UPDATER.incrementAndGet(this));
119-
span.addEvent("message", attributes);
131+
long messageId = SENT_MESSAGE_ID_UPDATER.incrementAndGet(this);
132+
if (addMessageEvents) {
133+
Attributes attributes = Attributes.of(MESSAGE_TYPE, SENT, MESSAGE_ID, messageId);
134+
Span.fromContext(context).addEvent("message", attributes);
135+
}
120136
}
121137

122138
final class TracingClientCallListener
@@ -139,14 +155,11 @@ final class TracingClientCallListener
139155

140156
@Override
141157
public void onMessage(RESPONSE message) {
142-
Span span = Span.fromContext(context);
143-
Attributes attributes =
144-
Attributes.of(
145-
MESSAGE_TYPE,
146-
RECEIVED,
147-
MESSAGE_ID,
148-
MESSAGE_ID_UPDATER.incrementAndGet(TracingClientCall.this));
149-
span.addEvent("message", attributes);
158+
long messageId = RECEIVED_MESSAGE_ID_UPDATER.incrementAndGet(TracingClientCall.this);
159+
if (addMessageEvents) {
160+
Attributes attributes = Attributes.of(MESSAGE_TYPE, RECEIVED, MESSAGE_ID, messageId);
161+
Span.fromContext(context).addEvent("message", attributes);
162+
}
150163
try (Scope ignored = context.makeCurrent()) {
151164
delegate().onMessage(message);
152165
}
@@ -155,6 +168,13 @@ public void onMessage(RESPONSE message) {
155168
@Override
156169
public void onClose(Status status, Metadata trailers) {
157170
request.setPeerSocketAddress(getAttributes().get(Grpc.TRANSPORT_ATTR_REMOTE_ADDR));
171+
if (captureExperimentalSpanAttributes) {
172+
Span span = Span.fromContext(context);
173+
span.setAttribute(
174+
"grpc.messages.received", RECEIVED_MESSAGE_ID_UPDATER.get(TracingClientCall.this));
175+
span.setAttribute(
176+
"grpc.messages.sent", SENT_MESSAGE_ID_UPDATER.get(TracingClientCall.this));
177+
}
158178
instrumenter.end(context, request, status, status.getCause());
159179
try (Scope ignored = parentContext.makeCurrent()) {
160180
delegate().onClose(status, trailers);

‎instrumentation/grpc-1.6/library/src/main/java/io/opentelemetry/instrumentation/grpc/v1_6/TracingServerInterceptor.java

Lines changed: 46 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -33,19 +33,27 @@ final class TracingServerInterceptor implements ServerInterceptor {
3333
private static final String RECEIVED = "RECEIVED";
3434

3535
@SuppressWarnings("rawtypes")
36-
private static final AtomicLongFieldUpdater<TracingServerCall> MESSAGE_ID_UPDATER =
37-
AtomicLongFieldUpdater.newUpdater(TracingServerCall.class, "messageId");
36+
private static final AtomicLongFieldUpdater<TracingServerCall> SENT_MESSAGE_ID_UPDATER =
37+
AtomicLongFieldUpdater.newUpdater(TracingServerCall.class, "sentMessageId");
38+
39+
@SuppressWarnings("rawtypes")
40+
private static final AtomicLongFieldUpdater<TracingServerCall> RECEIVED_MESSAGE_ID_UPDATER =
41+
AtomicLongFieldUpdater.newUpdater(TracingServerCall.class, "receivedMessageId");
3842

3943
private static final Metadata.Key<String> AUTHORITY_KEY =
4044
InternalMetadata.keyOf(":authority", Metadata.ASCII_STRING_MARSHALLER);
4145

4246
private final Instrumenter<GrpcRequest, Status> instrumenter;
4347
private final boolean captureExperimentalSpanAttributes;
48+
private final boolean addMessageEvents;
4449

4550
TracingServerInterceptor(
46-
Instrumenter<GrpcRequest, Status> instrumenter, boolean captureExperimentalSpanAttributes) {
51+
Instrumenter<GrpcRequest, Status> instrumenter,
52+
boolean captureExperimentalSpanAttributes,
53+
boolean addMessageEvents) {
4754
this.instrumenter = instrumenter;
4855
this.captureExperimentalSpanAttributes = captureExperimentalSpanAttributes;
56+
this.addMessageEvents = addMessageEvents;
4957
}
5058

5159
@Override
@@ -85,9 +93,13 @@ final class TracingServerCall<REQUEST, RESPONSE>
8593
private final GrpcRequest request;
8694
private Status status;
8795

88-
// Used by MESSAGE_ID_UPDATER
96+
// Used by SENT_MESSAGE_ID_UPDATER
97+
@SuppressWarnings("UnusedVariable")
98+
volatile long sentMessageId;
99+
100+
// Used by RECEIVED_MESSAGE_ID_UPDATER
89101
@SuppressWarnings("UnusedVariable")
90-
volatile long messageId;
102+
volatile long receivedMessageId;
91103

92104
TracingServerCall(
93105
ServerCall<REQUEST, RESPONSE> delegate, Context context, GrpcRequest request) {
@@ -106,10 +118,11 @@ public void sendMessage(RESPONSE message) {
106118
try (Scope ignored = context.makeCurrent()) {
107119
super.sendMessage(message);
108120
}
109-
Span span = Span.fromContext(context);
110-
Attributes attributes =
111-
Attributes.of(MESSAGE_TYPE, SENT, MESSAGE_ID, MESSAGE_ID_UPDATER.incrementAndGet(this));
112-
span.addEvent("message", attributes);
121+
long messageId = SENT_MESSAGE_ID_UPDATER.incrementAndGet(this);
122+
if (addMessageEvents) {
123+
Attributes attributes = Attributes.of(MESSAGE_TYPE, SENT, MESSAGE_ID, messageId);
124+
Span.fromContext(context).addEvent("message", attributes);
125+
}
113126
}
114127

115128
@Override
@@ -134,15 +147,27 @@ final class TracingServerCallListener
134147
this.request = request;
135148
}
136149

150+
private void end(Context context, GrpcRequest request, Status response, Throwable error) {
151+
if (captureExperimentalSpanAttributes) {
152+
Span span = Span.fromContext(context);
153+
span.setAttribute(
154+
"grpc.messages.received", RECEIVED_MESSAGE_ID_UPDATER.get(TracingServerCall.this));
155+
span.setAttribute(
156+
"grpc.messages.sent", SENT_MESSAGE_ID_UPDATER.get(TracingServerCall.this));
157+
if (Status.CANCELLED.equals(status)) {
158+
span.setAttribute("grpc.canceled", true);
159+
}
160+
}
161+
instrumenter.end(context, request, response, error);
162+
}
163+
137164
@Override
138165
public void onMessage(REQUEST message) {
139-
Attributes attributes =
140-
Attributes.of(
141-
MESSAGE_TYPE,
142-
RECEIVED,
143-
MESSAGE_ID,
144-
MESSAGE_ID_UPDATER.incrementAndGet(TracingServerCall.this));
145-
Span.fromContext(context).addEvent("message", attributes);
166+
long messageId = RECEIVED_MESSAGE_ID_UPDATER.incrementAndGet(TracingServerCall.this);
167+
if (addMessageEvents) {
168+
Attributes attributes = Attributes.of(MESSAGE_TYPE, RECEIVED, MESSAGE_ID, messageId);
169+
Span.fromContext(context).addEvent("message", attributes);
170+
}
146171
delegate().onMessage(message);
147172
}
148173

@@ -160,36 +185,33 @@ public void onHalfClose() {
160185
public void onCancel() {
161186
try {
162187
delegate().onCancel();
163-
if (captureExperimentalSpanAttributes) {
164-
Span.fromContext(context).setAttribute("grpc.canceled", true);
165-
}
166188
} catch (Throwable e) {
167-
instrumenter.end(context, request, Status.UNKNOWN, e);
189+
end(context, request, Status.UNKNOWN, e);
168190
throw e;
169191
}
170-
instrumenter.end(context, request, Status.CANCELLED, null);
192+
end(context, request, Status.CANCELLED, null);
171193
}
172194

173195
@Override
174196
public void onComplete() {
175197
try {
176198
delegate().onComplete();
177199
} catch (Throwable e) {
178-
instrumenter.end(context, request, Status.UNKNOWN, e);
200+
end(context, request, Status.UNKNOWN, e);
179201
throw e;
180202
}
181203
if (status == null) {
182204
status = Status.UNKNOWN;
183205
}
184-
instrumenter.end(context, request, status, status.getCause());
206+
end(context, request, status, status.getCause());
185207
}
186208

187209
@Override
188210
public void onReady() {
189211
try {
190212
delegate().onReady();
191213
} catch (Throwable e) {
192-
instrumenter.end(context, request, Status.UNKNOWN, e);
214+
end(context, request, Status.UNKNOWN, e);
193215
throw e;
194216
}
195217
}

0 commit comments

Comments
 (0)