[Service Bus] Prepare tracing methods for message processor and scheduleMessage #16524
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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -4,7 +4,13 @@ | |
| package com.azure.messaging.servicebus.implementation; | ||
|
|
||
| import com.azure.core.amqp.implementation.AmqpConstants; | ||
| import com.azure.core.amqp.implementation.TracerProvider; | ||
| import com.azure.core.util.Context; | ||
| import com.azure.core.util.CoreUtils; | ||
| import com.azure.core.util.logging.ClientLogger; | ||
| import com.azure.core.util.tracing.ProcessKind; | ||
| import com.azure.messaging.servicebus.ServiceBusMessage; | ||
| import com.azure.messaging.servicebus.ServiceBusReceivedMessage; | ||
| import com.azure.messaging.servicebus.ServiceBusTransactionContext; | ||
| import org.apache.qpid.proton.amqp.Binary; | ||
| import org.apache.qpid.proton.amqp.Symbol; | ||
|
|
@@ -15,16 +21,32 @@ | |
| import org.apache.qpid.proton.amqp.transaction.TransactionalState; | ||
| import org.apache.qpid.proton.amqp.transport.DeliveryState; | ||
| import org.apache.qpid.proton.amqp.transport.ErrorCondition; | ||
| import reactor.core.publisher.Signal; | ||
|
|
||
| import java.io.Closeable; | ||
| import java.io.IOException; | ||
| import java.nio.ByteBuffer; | ||
| import java.time.Duration; | ||
| import java.time.Instant; | ||
| import java.time.OffsetDateTime; | ||
| import java.time.ZoneOffset; | ||
| import java.util.HashMap; | ||
| import java.util.Locale; | ||
| import java.util.Map; | ||
| import java.util.Optional; | ||
| import java.util.UUID; | ||
|
|
||
| import static com.azure.core.util.tracing.Tracer.AZ_TRACING_NAMESPACE_KEY; | ||
| import static com.azure.core.util.tracing.Tracer.DIAGNOSTIC_ID_KEY; | ||
| import static com.azure.core.util.tracing.Tracer.ENTITY_PATH_KEY; | ||
| import static com.azure.core.util.tracing.Tracer.HOST_NAME_KEY; | ||
| import static com.azure.core.util.tracing.Tracer.MESSAGE_ENQUEUED_TIME; | ||
| import static com.azure.core.util.tracing.Tracer.SCOPE_KEY; | ||
| import static com.azure.core.util.tracing.Tracer.SPAN_CONTEXT_KEY; | ||
| import static com.azure.messaging.servicebus.implementation.ServiceBusConstants.AZ_TRACING_SERVICE_NAME; | ||
| import static com.azure.messaging.servicebus.implementation.ServiceBusConstants.AZ_TRACING_NAMESPACE_VALUE; | ||
|
|
||
|
|
||
| /** | ||
| * Contains helper methods for message conversions, reading status codes, and getting delivery state. | ||
| */ | ||
|
|
@@ -253,4 +275,79 @@ private static TransactionalState getTransactionState(ByteBuffer transactionId, | |
| transactionalState.setOutcome(outcome); | ||
| return transactionalState; | ||
| } | ||
|
|
||
| /** | ||
| * Used in ServiceBusMessageBatch.tryAddMessage() to start tracing for to-be-sent out messages. | ||
| */ | ||
| public static ServiceBusMessage traceMessageSpan(ServiceBusMessage serviceBusMessage, | ||
| Context messageContext, String hostname, String entityPath, TracerProvider tracerProvider) { | ||
|
Contributor
There was a problem hiding this comment. How does Kind fits into this ?, should that be passed here ? https://github.com/Azure/azure-sdk-for-java/blob/master/sdk/core/azure-core/src/main/java/com/azure/core/util/tracing/ProcessKind.java
Contributor
Author
There was a problem hiding this comment.
Member
There was a problem hiding this comment. The ProcessKind helps the tracer apply specific attributes on the span. For example, in the case of MESSAGE, the tracer will attribute the spanType=Producer, here
Contributor
Author
There was a problem hiding this comment. @samvaity is this javadoc misleading? It looks like MESSAGE is used for receiving. /**
* Amqp message process call to receive data.
*/
MESSAGE,
Member
There was a problem hiding this comment. Yeah, I think we initially had this as RECEIVE and later updated to MESSAGE so the left over javadoc. But yes it should be updated to suit process kind "message". AMQP process kind for message spans. |
||
| Optional<Object> eventContextData = messageContext.getData(SPAN_CONTEXT_KEY); | ||
| if (eventContextData.isPresent()) { | ||
| // if message has context (in case of retries), don't start a message span or add a new context | ||
| return serviceBusMessage; | ||
| } else { | ||
| // Starting the span makes the sampling decision (nothing is logged at this time) | ||
| Context newMessageContext = messageContext | ||
| .addData(AZ_TRACING_NAMESPACE_KEY, AZ_TRACING_NAMESPACE_VALUE) | ||
| .addData(ENTITY_PATH_KEY, entityPath) | ||
| .addData(HOST_NAME_KEY, hostname); | ||
YijunXieMS marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| Context eventSpanContext = tracerProvider.startSpan(AZ_TRACING_SERVICE_NAME, newMessageContext, | ||
| ProcessKind.MESSAGE); | ||
| Optional<Object> eventDiagnosticIdOptional = eventSpanContext.getData(DIAGNOSTIC_ID_KEY); | ||
| if (eventDiagnosticIdOptional.isPresent()) { | ||
| serviceBusMessage.getApplicationProperties().put(DIAGNOSTIC_ID_KEY, eventDiagnosticIdOptional.get() | ||
| .toString()); | ||
| tracerProvider.endSpan(eventSpanContext, Signal.complete()); | ||
| serviceBusMessage.addContext(SPAN_CONTEXT_KEY, eventSpanContext); | ||
| } | ||
| } | ||
| return serviceBusMessage; | ||
| } | ||
|
|
||
| /* | ||
| * Starts a new process tracing span and attaches the returned context to the ServiceBusReceivedMessage object for | ||
| * users. | ||
| */ | ||
| public static Context startProcessTracingSpan(ServiceBusReceivedMessage receivedMessage, | ||
YijunXieMS marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| String hostname, String entityPath, TracerProvider tracerProvider, ProcessKind processKind) { | ||
| Object diagnosticId = receivedMessage.getApplicationProperties().get(DIAGNOSTIC_ID_KEY); | ||
| if (diagnosticId == null || !tracerProvider.isEnabled()) { | ||
| return Context.NONE; | ||
| } | ||
|
|
||
| Context spanContext = tracerProvider.extractContext(diagnosticId.toString(), Context.NONE) | ||
| .addData(ENTITY_PATH_KEY, entityPath) | ||
| .addData(HOST_NAME_KEY, hostname) | ||
| .addData(AZ_TRACING_NAMESPACE_KEY, AZ_TRACING_NAMESPACE_VALUE); | ||
| spanContext = receivedMessage.getEnqueuedTime() == null | ||
| ? spanContext | ||
| : spanContext.addData(MESSAGE_ENQUEUED_TIME, receivedMessage.getEnqueuedTime().toEpochSecond()); | ||
| return tracerProvider.startSpan(AZ_TRACING_SERVICE_NAME, spanContext, processKind); | ||
| } | ||
|
|
||
| /* | ||
| * Ends the process tracing span and the scope of that span. | ||
| */ | ||
| public static void endProcessTracingSpan(Context processSpanContext, Signal<Void> signal, | ||
YijunXieMS marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| TracerProvider tracerProvider, ClientLogger logger) { | ||
| if (processSpanContext != null) { | ||
| Optional<Object> spanScope = processSpanContext.getData(SCOPE_KEY); | ||
| // Disposes of the scope when the trace span closes. | ||
| if (tracerProvider.isEnabled() && spanScope.isPresent()) { | ||
| if (spanScope.get() instanceof Closeable) { | ||
| Closeable close = (Closeable) spanScope.get(); | ||
| try { | ||
| close.close(); | ||
| tracerProvider.endSpan(processSpanContext, signal); | ||
| } catch (IOException ioException) { | ||
| logger.error(Messages.MESSAGE_PROCESSOR_RUN_END, ioException); | ||
| } | ||
|
|
||
| } else { | ||
| logger.warning(String.format(Locale.US, | ||
| Messages.PROCESS_SPAN_SCOPE_TYPE_ERROR, spanScope.getClass())); | ||
| } | ||
| } | ||
| } | ||
| } | ||
| } | ||
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
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.
Uh oh!
There was an error while loading. Please reload this page.