Add EventGrid distributed tracing #15850
There are no files selected for viewing
This file contains hidden or 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 hidden or 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 hidden or 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 hidden or 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 hidden or 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,76 @@ | ||
| package com.azure.messaging.eventgrid.implementation; | ||
|
|
||
| import com.azure.core.http.HttpHeader; | ||
| import com.azure.core.http.HttpPipelineCallContext; | ||
| import com.azure.core.http.HttpPipelineNextPolicy; | ||
| import com.azure.core.http.HttpRequest; | ||
| import com.azure.core.http.HttpResponse; | ||
| import com.azure.core.http.policy.HttpPipelinePolicy; | ||
| import com.azure.core.util.tracing.TracerProxy; | ||
| import reactor.core.publisher.Mono; | ||
|
|
||
| import java.nio.charset.StandardCharsets; | ||
|
|
||
| import com.azure.messaging.eventgrid.CloudEvent; | ||
| /** | ||
| * This pipeline policy should be added after OpenTelemetryPolicy in the http pipeline. | ||
| * | ||
| * It checks whether the {@link HttpRequest} headers have "traceparent" or "tracestate" and whether the serialized | ||
| * http body json string for a list of {@link CloudEvent} instances has place holders | ||
| * {@link Constants#TRACE_PARENT_PLACEHOLDER} or {@link Constants#TRACE_STATE_PLACEHOLDER}. | ||
| * The place holders will be replaced by the value from headers if the headers have "traceparent" or "tracestate", | ||
| * or be removed if the headers don't have. | ||
| * | ||
| * The place holders won't exist in the json string if the {@link TracerProxy#isTracingEnabled()} returns false. | ||
| */ | ||
| public class CloudEventTracingPipelinePolicy implements HttpPipelinePolicy { | ||
|
|
||
| @Override | ||
| public Mono<HttpResponse> process(HttpPipelineCallContext context, HttpPipelineNextPolicy next) { | ||
| final HttpRequest request = context.getHttpRequest(); | ||
| final HttpHeader contentType = request.getHeaders().get(Constants.CONTENT_TYPE); | ||
| StringBuilder bodyStringBuilder = new StringBuilder(); | ||
| if (TracerProxy.isTracingEnabled() && contentType != null && | ||
| Constants.CLOUD_EVENT_CONTENT_TYPE.equals(contentType.getValue())) { | ||
| return request.getBody().map(byteBuffer -> bodyStringBuilder.append(new String(byteBuffer.array(), | ||
| StandardCharsets.UTF_8))) | ||
| .then(Mono.fromCallable(() -> replaceTracingPlaceHolder(request, bodyStringBuilder))) | ||
| .then(next.process()); | ||
| } | ||
| else { | ||
| return next.process(); | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * | ||
| * @param request The {@link HttpRequest}, whose body will be mutated by replacing traceparent and tracestate | ||
| * placeholders. | ||
| * @param bodyStringBuilder The {@link StringBuilder} that contains the full HttpRequest body string. | ||
| * @return The new body string with the place holders replaced (if header has tracing) | ||
| * or removed (if header no tracing). | ||
| */ | ||
| static String replaceTracingPlaceHolder(HttpRequest request, StringBuilder bodyStringBuilder) { | ||
| final int traceParentPlaceHolderIndex = bodyStringBuilder.indexOf(Constants.TRACE_PARENT_PLACEHOLDER); | ||
| if (traceParentPlaceHolderIndex >= 0) { // There is "traceparent" placeholder in body, replace it. | ||
| final HttpHeader traceparentHeader = request.getHeaders().get(Constants.TRACE_PARENT); | ||
| bodyStringBuilder.replace(traceParentPlaceHolderIndex, | ||
| Constants.TRACE_PARENT_PLACEHOLDER.length() + traceParentPlaceHolderIndex, | ||
| traceparentHeader != null | ||
| ? String.format(",\"%s\":\"%s\"", Constants.TRACE_PARENT, traceparentHeader.getValue()) | ||
| : ""); | ||
|
samvaity marked this conversation as resolved.
|
||
| } | ||
| final int traceStatePlaceHolderIndex = bodyStringBuilder.indexOf(Constants.TRACE_STATE_PLACEHOLDER); | ||
| if (traceStatePlaceHolderIndex >= 0) { // There is "tracestate" placeholder in body, replace it. | ||
| final HttpHeader tracestateHeader = request.getHeaders().get(Constants.TRACE_STATE); | ||
| bodyStringBuilder.replace(traceStatePlaceHolderIndex, | ||
| Constants.TRACE_STATE_PLACEHOLDER.length() + traceStatePlaceHolderIndex, | ||
| tracestateHeader != null | ||
| ? String.format(",\"%s\":\"%s\"", Constants.TRACE_STATE, tracestateHeader.getValue()) | ||
| : ""); | ||
| } | ||
| String newBodyString = bodyStringBuilder.toString(); | ||
| request.setHeader(Constants.CONTENT_LENGTH, String.valueOf(newBodyString.length())); | ||
| request.setBody(newBodyString); | ||
| return newBodyString; | ||
| } | ||
| } | ||
This file contains hidden or 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,17 @@ | ||
| package com.azure.messaging.eventgrid.implementation; | ||
|
|
||
| public class Constants { | ||
| public static final String CONTENT_TYPE = "Content-Type"; | ||
| public static final String CONTENT_LENGTH = "Content-Length"; | ||
| public static final String CLOUD_EVENT_CONTENT_TYPE = "application/cloudevents-batch+json; charset=utf-8"; | ||
| public static final String TRACE_PARENT = "traceparent"; | ||
| public static final String TRACE_STATE = "tracestate"; | ||
| public static final String TRACE_PARENT_PLACEHOLDER_UUID = "TP-14b6b15b-74b6-4178-847e-d142aa2727b2"; | ||
| public static final String TRACE_STATE_PLACEHOLDER_UUID = "TS-14b6b15b-74b6-4178-847e-d142aa2727b2"; | ||
| public static final String TRACE_PARENT_PLACEHOLDER = ",\"" + TRACE_PARENT + "\":\"TP-14b6b15b-74b6-4178-847e-d142aa2727b2\""; | ||
| public static final String TRACE_STATE_PLACEHOLDER = ",\"" + TRACE_STATE + "\":\"TS-14b6b15b-74b6-4178-847e-d142aa2727b2\""; | ||
|
|
||
| // Please see <a href=https://docs.microsoft.com/en-us/azure/azure-resource-manager/management/azure-services-resource-providers>here</a> | ||
| // for more information on Azure resource provider namespaces. | ||
| public static final String EVENT_GRID_TRACING_NAMESPACE_VALUE = "Microsoft.EventGrid"; | ||
| } |
This file contains hidden or 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 hidden or 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.
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 have a null check for context before using it.
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.
This internal api is called by a public API, which calls
FluxUtil.withContextto create a Context. So I assume the context won't be null.In debugging, I see it's an empty Context instance instead of null.
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.
@YijunXieMS - this method is also called from sync client and the user can pass a null context.
User can call
sendEventsWithResponse(events, null)