From 7b95368ea6209cc0831782d28c1e508b9a6ab636 Mon Sep 17 00:00:00 2001 From: Sameeksha Vaity Date: Fri, 22 Nov 2019 11:01:11 -0800 Subject: [PATCH 1/9] Revert "update samples and package versions" This reverts commit 9e9951bc406cc0152e6b8ed0c46664b42b144c55. --- .../azure-core-tracing-opencensus/README.md | 15 +++----- .../src/samples/AsyncListKeyVaultSecrets.md | 2 +- .../AsyncQueueClientEnqueueMessages.md | 2 +- .../src/samples/ListeKeyVaultSecrets.md | 2 +- .../src/samples/PublishEvents.md | 38 ++++++++----------- .../src/samples/QueueClientEnqueueMessages.md | 2 +- .../CHANGELOG.md | 9 ----- .../README.md | 12 ++---- .../src/samples/PublishEventsTracingSample.md | 33 ++++++---------- 9 files changed, 39 insertions(+), 76 deletions(-) diff --git a/sdk/core/azure-core-tracing-opencensus/README.md b/sdk/core/azure-core-tracing-opencensus/README.md index 00a7c834c67e..8497caa53333 100644 --- a/sdk/core/azure-core-tracing-opencensus/README.md +++ b/sdk/core/azure-core-tracing-opencensus/README.md @@ -19,7 +19,7 @@ documentation][api_documentation] | [Samples][samples] com.azure azure-core-tracing-opencensus - 1.0.0-beta.5 + 1.0.0-beta.4 ``` [//]: # ({x-version-update-end}) @@ -90,15 +90,10 @@ private static final Tracer TRACER; .buildProducerClient(); try (Scope scope = TRACER.spanBuilder("tracing-user-span").startScopedSpan()) { - EventData event1 = new EventData("1".getBytes(UTF_8)); - event1.addContext(PARENT_SPAN_KEY, span); - - EventDataBatch eventDataBatch = producer.createBatch(); - - if (!eventDataBatch.tryAdd(eventData)) { - producer.send(eventDataBatch); - eventDataBatch = producer.createBatch(); - } + Context tracingContext = new Context(PARENT_SPAN_KEY, TRACER.getCurrentSpan()); + // Create an event to send + final EventData eventData = new EventData("Hello world!".getBytes(UTF_8), traceContext); + producer.send(eventData); } finally { Tracing.getExportComponent().shutdown(); } diff --git a/sdk/core/azure-core-tracing-opencensus/src/samples/AsyncListKeyVaultSecrets.md b/sdk/core/azure-core-tracing-opencensus/src/samples/AsyncListKeyVaultSecrets.md index 1e29612b229d..d1714b8b2b57 100644 --- a/sdk/core/azure-core-tracing-opencensus/src/samples/AsyncListKeyVaultSecrets.md +++ b/sdk/core/azure-core-tracing-opencensus/src/samples/AsyncListKeyVaultSecrets.md @@ -22,7 +22,7 @@ Sample uses **[opencensus-impl][opencensus_impl]** as implementation package and com.azure azure-core-tracing-opencensus - 1.0.0-beta.5 + 1.0.0-beta.4 ``` [//]: # ({x-version-update-end}) diff --git a/sdk/core/azure-core-tracing-opencensus/src/samples/AsyncQueueClientEnqueueMessages.md b/sdk/core/azure-core-tracing-opencensus/src/samples/AsyncQueueClientEnqueueMessages.md index 27e2066b6f39..eb2997a65e13 100644 --- a/sdk/core/azure-core-tracing-opencensus/src/samples/AsyncQueueClientEnqueueMessages.md +++ b/sdk/core/azure-core-tracing-opencensus/src/samples/AsyncQueueClientEnqueueMessages.md @@ -24,7 +24,7 @@ Sample uses **[opencensus-impl][opencensus_impl]** as implementation package and com.azure azure-core-tracing-opencensus - 1.0.0-beta.5 + 1.0.0-beta.4 ``` [//]: # ({x-version-update-end}) diff --git a/sdk/core/azure-core-tracing-opencensus/src/samples/ListeKeyVaultSecrets.md b/sdk/core/azure-core-tracing-opencensus/src/samples/ListeKeyVaultSecrets.md index 4ed8864983f8..43e63e44b96e 100644 --- a/sdk/core/azure-core-tracing-opencensus/src/samples/ListeKeyVaultSecrets.md +++ b/sdk/core/azure-core-tracing-opencensus/src/samples/ListeKeyVaultSecrets.md @@ -23,7 +23,7 @@ Sample uses **[opencensus-impl][opencensus_impl]** as implementation package and com.azure azure-core-tracing-opencensus - 1.0.0-beta.5 + 1.0.0-beta.4 ``` [//]: # ({x-version-update-end}) diff --git a/sdk/core/azure-core-tracing-opencensus/src/samples/PublishEvents.md b/sdk/core/azure-core-tracing-opencensus/src/samples/PublishEvents.md index 29674153f9b8..afb0878bc0d3 100644 --- a/sdk/core/azure-core-tracing-opencensus/src/samples/PublishEvents.md +++ b/sdk/core/azure-core-tracing-opencensus/src/samples/PublishEvents.md @@ -14,7 +14,7 @@ Sample uses **[opencensus-impl][opencensus_impl]** as implementation package and com.azure azure-messaging-eventhubs - 5.0.0-beta.6 + 5.0.0-beta.5 ``` [//]: # ({x-version-update-end}) @@ -68,31 +68,25 @@ public class PublishEvents { } public static void main(String[] args) { - EventHubProducerClient producer = new EventHubClientBuilder() - .connectionString(connectionString) - .buildProducerClient(); + final int count = 2; + final byte[] body = "Hello World!".getBytes(UTF_8); try(Scope scope = TRACER.spanBuilder("user-parent-span").startScopedSpan()) { - final EventData event1 = new EventData("1".getBytes(UTF_8)); - event1.addContext(PARENT_SPAN_KEY, TRACER.getCurrentSpan()); - - final EventData event2 = new EventData("2".getBytes(UTF_8)); - event2.addContext(PARENT_SPAN_KEY, TRACER.getCurrentSpan()); - final List telemetryEvents = Arrays.asList(event1, event2); - final CreateBatchOptions options = new CreateBatchOptions() - .setPartitionKey("telemetry") - .setMaximumSizeInBytes(256); - - EventDataBatch currentBatch = producer.createBatch(options); + Context traceContext = new Context(PARENT_SPAN_KEY, TRACER.getCurrentSpan()); + + EventHubProducerClient producer = new EventHubClientBuilder() + .connectionString(CONNECTION_STRING) + .buildProducer(); - // For each telemetry event, we try to add it to the current batch. - for (EventData event : telemetryEvents) { - if (!currentBatch.tryAdd(event)) { - producer.send(currentBatch); - currentBatch = producer.createBatch(options); - } - } + final Context traceContext = new Context(PARENT_SPAN_KEY, TRACER.getCurrentSpan()); + final Flux testData = Flux.range(0, count) + .flatMap(number -> { + final EventData data = new EventData(body, traceContext); + return Flux.just(data); + }); + + producer.send(testData.toIterable(1)); } finally { producer.close(); Tracing.getExportComponent().shutdown(); diff --git a/sdk/core/azure-core-tracing-opencensus/src/samples/QueueClientEnqueueMessages.md b/sdk/core/azure-core-tracing-opencensus/src/samples/QueueClientEnqueueMessages.md index 8ce90a2ff790..f910c9758d3c 100644 --- a/sdk/core/azure-core-tracing-opencensus/src/samples/QueueClientEnqueueMessages.md +++ b/sdk/core/azure-core-tracing-opencensus/src/samples/QueueClientEnqueueMessages.md @@ -24,7 +24,7 @@ Sample uses **[opencensus-impl][opencensus_impl]** as implementation package and com.azure azure-core-tracing-opencensus - 1.0.0-beta.5 + 1.0.0-beta.4 ``` [//]: # ({x-version-update-end}) diff --git a/sdk/core/azure-core-tracing-opentelemetry/CHANGELOG.md b/sdk/core/azure-core-tracing-opentelemetry/CHANGELOG.md index 780d77ac2ad6..7180efe5049e 100644 --- a/sdk/core/azure-core-tracing-opentelemetry/CHANGELOG.md +++ b/sdk/core/azure-core-tracing-opentelemetry/CHANGELOG.md @@ -11,12 +11,3 @@ and ## Version 1.0.0-beta.1 (2019-11-26) For details on the Azure SDK for Java (Decemeber 2019 Preview) release refer to the [release announcement](https://aka.ms/azure-sdk-preview5-java). -Version 1.0.0-beta.1 is the first preview of our efforts to provide low level interfaces and helper methods to support tracing for Java client libraries. -This library includes [OpenTelemetry](https://opentelemetry.io/) implementation of the interface. -This library added tracing instrumentation for AMQP and HTTP Java SDK client libraries across different languages and platforms. - -This package's -[documentation](https://github.com/Azure/azure-sdk-for-java/blob/azure-core-tracing-opentelemetry_1.0.0-beta.1/sdk/core/azure-core-tracing-opentelemetry/README.md) -and -[samples](https://github.com/Azure/azure-sdk-for-java/blob/azure-core-tracing-opentelemetry_1.0.0-beta.1/sdk/core/azure-core-tracing-opentelemetry/src/samples). - diff --git a/sdk/core/azure-core-tracing-opentelemetry/README.md b/sdk/core/azure-core-tracing-opentelemetry/README.md index 772a8b26ac5a..8e6c8854b24c 100644 --- a/sdk/core/azure-core-tracing-opentelemetry/README.md +++ b/sdk/core/azure-core-tracing-opentelemetry/README.md @@ -94,15 +94,9 @@ private static final TracerSdkFactory TRACER_SDK_FACTORY; Span span = TRACER.spanBuilder("user-parent-span").startSpan(); try (Scope scope = TRACER.withSpan(span)) { - EventData event1 = new EventData("1".getBytes(UTF_8)); - event1.addContext(PARENT_SPAN_KEY, span); - - EventDataBatch eventDataBatch = producer.createBatch(); - - if (!eventDataBatch.tryAdd(eventData)) { - producer.send(eventDataBatch); - eventDataBatch = producer.createBatch(); - } + Context traceContext = new Context(PARENT_SPAN_KEY, span); + EventData eventData = new EventData("Hello world!".getBytes(UTF_8), traceContext); + producer.send(eventData); } finally { span.end(); } diff --git a/sdk/core/azure-core-tracing-opentelemetry/src/samples/PublishEventsTracingSample.md b/sdk/core/azure-core-tracing-opentelemetry/src/samples/PublishEventsTracingSample.md index 2bd9d1bd6ab7..ec3cb269e6fe 100644 --- a/sdk/core/azure-core-tracing-opentelemetry/src/samples/PublishEventsTracingSample.md +++ b/sdk/core/azure-core-tracing-opentelemetry/src/samples/PublishEventsTracingSample.md @@ -81,31 +81,20 @@ public class Sample { private static void doClientWork() { EventHubProducerClient producer = new EventHubClientBuilder() - .connectionString(connectionString) - .buildProducerClient(); + .connectionString(CONNECTION_STRING) + .buildProducer(); + + final int count = 2; + final byte[] body = "Hello World!".getBytes(UTF_8); Span span = TRACER.spanBuilder("user-parent-span").startSpan(); try (final Scope scope = TRACER.withSpan(span)) { - final EventData event1 = new EventData("1".getBytes(UTF_8)); - event1.addContext(PARENT_SPAN_KEY, span); - - final EventData event2 = new EventData("2".getBytes(UTF_8)); - event2.addContext(PARENT_SPAN_KEY, span); - - final List telemetryEvents = Arrays.asList(event1, event2); - final CreateBatchOptions options = new CreateBatchOptions() - .setPartitionKey("telemetry") - .setMaximumSizeInBytes(256); - - EventDataBatch currentBatch = producer.createBatch(options); - - // For each telemetry event, we try to add it to the current batch. - for (EventData event : telemetryEvents) { - if (!currentBatch.tryAdd(event)) { - producer.send(currentBatch); - currentBatch = producer.createBatch(options); - } - } + final Context traceContext = new Context(PARENT_SPAN_KEY, span); + final Flux testData = Flux.range(0, count).flatMap(number -> { + final EventData data = new EventData(body, traceContext); + return Flux.just(data); + }); + producer.send(testData.toIterable(1)); } finally { span.end(); producer.close(); From 22d1c99ff4498d24d839c38a7b2874c12d6fc878 Mon Sep 17 00:00:00 2001 From: Sameeksha Vaity Date: Wed, 11 Dec 2019 16:10:25 -0800 Subject: [PATCH 2/9] add links on batch send --- .../amqp/implementation/TracerProvider.java | 1 + .../opentelemetry/OpenTelemetryTracer.java | 8 +++++++- .../opentelemetry/OpenTelemetryTracerTest.java | 17 +++++++++++++---- .../azure/core/util/tracing/ProcessKind.java | 6 +++++- .../eventhubs/EventHubProducerAsyncClient.java | 9 +++++++-- .../EventHubProducerAsyncClientTest.java | 14 ++++++++++++++ .../eventhubs/EventHubProducerClientTest.java | 15 +++++++++++++++ 7 files changed, 62 insertions(+), 8 deletions(-) diff --git a/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/TracerProvider.java b/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/TracerProvider.java index f39076bf10e7..143fa83d748c 100644 --- a/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/TracerProvider.java +++ b/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/TracerProvider.java @@ -118,6 +118,7 @@ private String getSpanName(ProcessKind processKind) { String spanName = "EventHubs."; switch (processKind) { case SEND: + case LINK: spanName += "send"; break; case MESSAGE: diff --git a/sdk/core/azure-core-tracing-opentelemetry/src/main/java/com/azure/core/tracing/opentelemetry/OpenTelemetryTracer.java b/sdk/core/azure-core-tracing-opentelemetry/src/main/java/com/azure/core/tracing/opentelemetry/OpenTelemetryTracer.java index d0040fcf23fc..57faba6e86ba 100644 --- a/sdk/core/azure-core-tracing-opentelemetry/src/main/java/com/azure/core/tracing/opentelemetry/OpenTelemetryTracer.java +++ b/sdk/core/azure-core-tracing-opentelemetry/src/main/java/com/azure/core/tracing/opentelemetry/OpenTelemetryTracer.java @@ -66,7 +66,11 @@ public Context start(String spanName, Context context, ProcessKind processKind) switch (processKind) { case SEND: - spanBuilder = getSpanBuilder(spanName, context); + // use previosuly created span builder from the LINK process. + spanBuilder = getOrDefault(context, SPAN_BUILDER_KEY, null, Builder.class); + if (spanBuilder == null) { + return Context.NONE; + } span = spanBuilder.setSpanKind(Span.Kind.PRODUCER).startSpan(); if (span.isRecording()) { // If span is sampled in, add additional request attributes @@ -81,6 +85,8 @@ public Context start(String spanName, Context context, ProcessKind processKind) return context.addData(PARENT_SPAN_KEY, span); case PROCESS: return startScopedSpan(spanName, context); + case LINK: + return context.addData(SPAN_BUILDER_KEY, getSpanBuilder(spanName, context)); default: return Context.NONE; } diff --git a/sdk/core/azure-core-tracing-opentelemetry/src/test/java/com/azure/core/tracing/opentelemetry/OpenTelemetryTracerTest.java b/sdk/core/azure-core-tracing-opentelemetry/src/test/java/com/azure/core/tracing/opentelemetry/OpenTelemetryTracerTest.java index dfb362de9756..d687bde4647d 100644 --- a/sdk/core/azure-core-tracing-opentelemetry/src/test/java/com/azure/core/tracing/opentelemetry/OpenTelemetryTracerTest.java +++ b/sdk/core/azure-core-tracing-opentelemetry/src/test/java/com/azure/core/tracing/opentelemetry/OpenTelemetryTracerTest.java @@ -92,7 +92,6 @@ public void startSpanParentContextFlowTest() { final ReadableSpan recordEventsSpan = (ReadableSpan) updatedContext.getData(PARENT_SPAN_KEY).get(); assertEquals(Span.Kind.INTERNAL, recordEventsSpan.toSpanData().getKind()); - assertEquals(true, true); } @Test @@ -117,9 +116,11 @@ public void startSpanTestNoUserParent() { public void startSpanProcessKindSend() { // Arrange final SpanId parentSpanId = parentSpan.getContext().getSpanId(); + // Start user parent span. + final Span.Builder spanBuilder = tracer.spanBuilder(METHOD_NAME); // Add additional metadata to spans for SEND final Context traceContext = tracingContext.addData(ENTITY_PATH_KEY, ENTITY_PATH_VALUE) - .addData(HOST_NAME_KEY, HOSTNAME_VALUE); + .addData(HOST_NAME_KEY, HOSTNAME_VALUE).addData(SPAN_BUILDER_KEY, spanBuilder); // Act final Context updatedContext = openTelemetryTracer.start(METHOD_NAME, traceContext, ProcessKind.SEND); @@ -181,6 +182,14 @@ public void startSpanProcessKindProcess() { verifySpanAttributes(attributeMap); } + @Test + public void startSpanProcessKindLink() { + // Act + final Context updatedContext = openTelemetryTracer.start(METHOD_NAME, Context.NONE, ProcessKind.LINK); + + assertTrue(updatedContext.getData(SPAN_BUILDER_KEY).isPresent()); + } + @Test public void startProcessSpanWithRemoteParent() { // Arrange @@ -277,7 +286,7 @@ public void endSpanTestThrowableResponseCode() { } private static void assertSpanWithExplicitParent(Context updatedContext, SpanId parentSpanId) { - assertNotNull(updatedContext.getData(PARENT_SPAN_KEY)); + assertNotNull(updatedContext.getData(PARENT_SPAN_KEY).get()); // verify instance created of opentelemetry-sdk (test impl), span implementation assertTrue(updatedContext.getData(PARENT_SPAN_KEY).get() instanceof ReadableSpan); @@ -292,7 +301,7 @@ private static void assertSpanWithExplicitParent(Context updatedContext, SpanId } private static void assertSpanWithRemoteParent(Context updatedContext, SpanId parentSpanId) { - assertNotNull(updatedContext.getData(PARENT_SPAN_KEY)); + assertNotNull(updatedContext.getData(PARENT_SPAN_KEY).get()); // verify instance created of opentelemetry-sdk (test impl), span implementation assertTrue(updatedContext.getData(PARENT_SPAN_KEY).get() instanceof ReadableSpan); diff --git a/sdk/core/azure-core/src/main/java/com/azure/core/util/tracing/ProcessKind.java b/sdk/core/azure-core/src/main/java/com/azure/core/util/tracing/ProcessKind.java index d0410d8e5007..96239adc2924 100644 --- a/sdk/core/azure-core/src/main/java/com/azure/core/util/tracing/ProcessKind.java +++ b/sdk/core/azure-core/src/main/java/com/azure/core/util/tracing/ProcessKind.java @@ -18,5 +18,9 @@ public enum ProcessKind { /** * Custom process call to process received messages. */ - PROCESS + PROCESS, + /** + * Custom process call to link messages to Send. + */ + LINK } diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClient.java b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClient.java index 1b68062096f2..48e23d755469 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClient.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClient.java @@ -47,6 +47,7 @@ import static com.azure.core.util.FluxUtil.monoError; 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.SPAN_CONTEXT_KEY; import static com.azure.messaging.eventhubs.implementation.ClientConstants.MAX_MESSAGE_LENGTH_BYTES; /** @@ -397,10 +398,13 @@ public Mono send(EventDataBatch batch) { ? new AtomicReference<>(Context.NONE) : null; + // initiate addLink on send span, get shared span builder for all links + Context sharedContext = isTracingEnabled ? tracerProvider.startSpan(parentContext.get(), ProcessKind.LINK) : + null; final List messages = batch.getEvents().stream().map(event -> { final Message message = messageSerializer.serialize(event); if (isTracingEnabled) { - parentContext.set(event.getContext()); + tracerProvider.addSpanLinks(sharedContext.addData(SPAN_CONTEXT_KEY, event.getContext())); } if (!CoreUtils.isNullOrEmpty(partitionKey)) { final MessageAnnotations messageAnnotations = message.getMessageAnnotations() == null @@ -408,6 +412,7 @@ public Mono send(EventDataBatch batch) { : message.getMessageAnnotations(); messageAnnotations.getValue().put(AmqpConstants.PARTITION_KEY, partitionKey); message.setMessageAnnotations(messageAnnotations); + } return message; @@ -416,7 +421,7 @@ public Mono send(EventDataBatch batch) { return getSendLink(batch.getPartitionId()) .flatMap(link -> { if (isTracingEnabled) { - Context userSpanContext = parentContext.get(); + Context userSpanContext = sharedContext; Context entityContext = userSpanContext.addData(ENTITY_PATH_KEY, link.getEntityPath()); // start send span and store updated context parentContext.set(tracerProvider.startSpan( diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClientTest.java b/sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClientTest.java index 0676d5f1145f..f21d8734f4bf 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClientTest.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClientTest.java @@ -44,6 +44,7 @@ import static com.azure.core.util.tracing.Tracer.DIAGNOSTIC_ID_KEY; import static com.azure.core.util.tracing.Tracer.PARENT_SPAN_KEY; +import static com.azure.core.util.tracing.Tracer.SPAN_BUILDER_KEY; import static com.azure.core.util.tracing.Tracer.SPAN_CONTEXT_KEY; import static java.nio.charset.StandardCharsets.UTF_8; import static org.mockito.ArgumentMatchers.any; @@ -244,6 +245,13 @@ public void sendStartSpanSingleMessage() { } ); + when(tracer1.start(eq("Azure.eventhubs.send"), any(), eq(ProcessKind.LINK))).thenAnswer( + invocation -> { + Context passed = invocation.getArgument(1, Context.class); + return passed.addData(SPAN_BUILDER_KEY, "value"); + } + ); + // Act StepVerifier.create(asyncProducer.send(testData, sendOptions)) .verifyComplete(); @@ -293,6 +301,12 @@ public void sendMessageRetrySpanTest() { return passed.addData(PARENT_SPAN_KEY, "value"); } ); + when(tracer1.start(eq("Azure.eventhubs.send"), any(), eq(ProcessKind.LINK))).thenAnswer( + invocation -> { + Context passed = invocation.getArgument(1, Context.class); + return passed.addData(SPAN_BUILDER_KEY, "value"); + } + ); //Act StepVerifier.create(asyncProducer.send(testData, sendOptions)).verifyComplete(); diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/EventHubProducerClientTest.java b/sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/EventHubProducerClientTest.java index b06267314e79..516bfb79b739 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/EventHubProducerClientTest.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/EventHubProducerClientTest.java @@ -44,6 +44,7 @@ import static com.azure.core.util.tracing.Tracer.DIAGNOSTIC_ID_KEY; import static com.azure.core.util.tracing.Tracer.PARENT_SPAN_KEY; +import static com.azure.core.util.tracing.Tracer.SPAN_BUILDER_KEY; import static com.azure.core.util.tracing.Tracer.SPAN_CONTEXT_KEY; import static java.nio.charset.StandardCharsets.UTF_8; import static org.mockito.ArgumentMatchers.any; @@ -168,6 +169,13 @@ public void sendStartSpanSingleMessage() { return passed.addData(PARENT_SPAN_KEY, "value").addData(DIAGNOSTIC_ID_KEY, "value2"); } ); + when(tracer1.start(eq("Azure.eventhubs.send"), any(), eq(ProcessKind.LINK))).thenAnswer( + invocation -> { + Context passed = invocation.getArgument(1, Context.class); + return passed.addData(SPAN_BUILDER_KEY, "value"); + } + ); + //Act producer.send(eventData); @@ -209,6 +217,13 @@ public void sendMessageRetrySpanTest() { } ); + when(tracer1.start(eq("Azure.eventhubs.send"), any(), eq(ProcessKind.LINK))).thenAnswer( + invocation -> { + Context passed = invocation.getArgument(1, Context.class); + return passed.addData(SPAN_BUILDER_KEY, "value"); + } + ); + //Act producer.send(eventData); From 2a2d41e33b38bb11c07fa1e284256112d354f321 Mon Sep 17 00:00:00 2001 From: Sameeksha Vaity Date: Wed, 11 Dec 2019 17:46:09 -0800 Subject: [PATCH 3/9] update method name to getSpanBuilder() --- .../amqp/implementation/TracerProvider.java | 15 +- .../opentelemetry/OpenTelemetryTracer.java | 17 +- .../OpenTelemetryTracerTest.java | 4 +- .../azure/core/util/tracing/ProcessKind.java | 6 +- .../com/azure/core/util/tracing/Tracer.java | 16 ++ .../tracing/TracerJavaDocCodeSnippets.java | 17 ++ .../EventHubProducerAsyncClient.java | 161 +++++++++--------- .../EventHubProducerAsyncClientTest.java | 5 +- .../eventhubs/EventHubProducerClientTest.java | 4 +- 9 files changed, 143 insertions(+), 102 deletions(-) diff --git a/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/TracerProvider.java b/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/TracerProvider.java index 143fa83d748c..7fb7ab659813 100644 --- a/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/TracerProvider.java +++ b/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/TracerProvider.java @@ -108,6 +108,20 @@ public Context extractContext(String diagnosticId, Context context) { return local; } + /** + * For each tracer plugged into the SDK a new context containing the span builder is returned. + * + * @param context Additional metadata containing the span name for creating the span builer. + */ + public Context getSpanBuilder(Context context) { + Context local = Objects.requireNonNull(context, "'context' cannot be null."); + String spanName = getSpanName(ProcessKind.SEND); + for (Tracer tracer : tracers) { + local = tracer.getSpanBuilder(spanName, local); + } + return local; + } + private void end(String statusMessage, Throwable throwable, Context context) { for (Tracer tracer : tracers) { tracer.end(statusMessage, throwable, context); @@ -118,7 +132,6 @@ private String getSpanName(ProcessKind processKind) { String spanName = "EventHubs."; switch (processKind) { case SEND: - case LINK: spanName += "send"; break; case MESSAGE: diff --git a/sdk/core/azure-core-tracing-opentelemetry/src/main/java/com/azure/core/tracing/opentelemetry/OpenTelemetryTracer.java b/sdk/core/azure-core-tracing-opentelemetry/src/main/java/com/azure/core/tracing/opentelemetry/OpenTelemetryTracer.java index 57faba6e86ba..b631d887bc24 100644 --- a/sdk/core/azure-core-tracing-opentelemetry/src/main/java/com/azure/core/tracing/opentelemetry/OpenTelemetryTracer.java +++ b/sdk/core/azure-core-tracing-opentelemetry/src/main/java/com/azure/core/tracing/opentelemetry/OpenTelemetryTracer.java @@ -46,7 +46,7 @@ public Context start(String spanName, Context context) { Objects.requireNonNull(spanName, "'spanName' cannot be null."); Objects.requireNonNull(context, "'context' cannot be null."); - Builder spanBuilder = getSpanBuilder(spanName, context); + Builder spanBuilder = getParentSpanBuilder(spanName, context); Span span = spanBuilder.startSpan(); return context.addData(PARENT_SPAN_KEY, span); @@ -66,7 +66,7 @@ public Context start(String spanName, Context context, ProcessKind processKind) switch (processKind) { case SEND: - // use previosuly created span builder from the LINK process. + // use previously created span builder from the LINK process. spanBuilder = getOrDefault(context, SPAN_BUILDER_KEY, null, Builder.class); if (spanBuilder == null) { return Context.NONE; @@ -78,15 +78,13 @@ public Context start(String spanName, Context context, ProcessKind processKind) } return context.addData(PARENT_SPAN_KEY, span); case MESSAGE: - spanBuilder = getSpanBuilder(spanName, context); + spanBuilder = getParentSpanBuilder(spanName, context); span = spanBuilder.startSpan(); // Add diagnostic Id and trace-headers to Context context = setContextData(span); return context.addData(PARENT_SPAN_KEY, span); case PROCESS: return startScopedSpan(spanName, context); - case LINK: - return context.addData(SPAN_BUILDER_KEY, getSpanBuilder(spanName, context)); default: return Context.NONE; } @@ -178,6 +176,11 @@ public Context extractContext(String diagnosticId, Context context) { return AmqpPropagationFormatUtil.extractContext(diagnosticId, context); } + @Override + public Context getSpanBuilder(String spanName, Context context) { + return context.addData(SPAN_BUILDER_KEY, getParentSpanBuilder(spanName, context)); + } + /** * Starts a new child {@link Span} with parent being the remote and uses the {@link Span} is in the current Context, * to return an object that represents that scope. @@ -194,7 +197,7 @@ private Context startScopedSpan(String spanName, Context context) { if (spanContext != null) { span = startSpanWithRemoteParent(spanName, spanContext); } else { - Builder spanBuilder = getSpanBuilder(spanName, context); + Builder spanBuilder = getParentSpanBuilder(spanName, context); span = spanBuilder.setSpanKind(Span.Kind.SERVER).startSpan(); } if (span.isRecording()) { @@ -274,7 +277,7 @@ private static String parseComponentValue(String spanName) { * @param context The context containing the span and the span name. * @return A {@code Span.Builder} to create and start a new {@code Span}. */ - private Builder getSpanBuilder(String spanName, Context context) { + private Builder getParentSpanBuilder(String spanName, Context context) { Span parentSpan = getOrDefault(context, PARENT_SPAN_KEY, null, Span.class); String spanNameKey = getOrDefault(context, USER_SPAN_NAME_KEY, null, String.class); diff --git a/sdk/core/azure-core-tracing-opentelemetry/src/test/java/com/azure/core/tracing/opentelemetry/OpenTelemetryTracerTest.java b/sdk/core/azure-core-tracing-opentelemetry/src/test/java/com/azure/core/tracing/opentelemetry/OpenTelemetryTracerTest.java index d687bde4647d..693b3d237642 100644 --- a/sdk/core/azure-core-tracing-opentelemetry/src/test/java/com/azure/core/tracing/opentelemetry/OpenTelemetryTracerTest.java +++ b/sdk/core/azure-core-tracing-opentelemetry/src/test/java/com/azure/core/tracing/opentelemetry/OpenTelemetryTracerTest.java @@ -183,9 +183,9 @@ public void startSpanProcessKindProcess() { } @Test - public void startSpanProcessKindLink() { + public void getSpanBuilderTest() { // Act - final Context updatedContext = openTelemetryTracer.start(METHOD_NAME, Context.NONE, ProcessKind.LINK); + final Context updatedContext = openTelemetryTracer.getSpanBuilder(METHOD_NAME, Context.NONE); assertTrue(updatedContext.getData(SPAN_BUILDER_KEY).isPresent()); } diff --git a/sdk/core/azure-core/src/main/java/com/azure/core/util/tracing/ProcessKind.java b/sdk/core/azure-core/src/main/java/com/azure/core/util/tracing/ProcessKind.java index 96239adc2924..d0410d8e5007 100644 --- a/sdk/core/azure-core/src/main/java/com/azure/core/util/tracing/ProcessKind.java +++ b/sdk/core/azure-core/src/main/java/com/azure/core/util/tracing/ProcessKind.java @@ -18,9 +18,5 @@ public enum ProcessKind { /** * Custom process call to process received messages. */ - PROCESS, - /** - * Custom process call to link messages to Send. - */ - LINK + PROCESS } diff --git a/sdk/core/azure-core/src/main/java/com/azure/core/util/tracing/Tracer.java b/sdk/core/azure-core/src/main/java/com/azure/core/util/tracing/Tracer.java index 463450338325..2090b8307cec 100644 --- a/sdk/core/azure-core/src/main/java/com/azure/core/util/tracing/Tracer.java +++ b/sdk/core/azure-core/src/main/java/com/azure/core/util/tracing/Tracer.java @@ -204,4 +204,20 @@ public interface Tracer { * @throws NullPointerException if {@code diagnosticId} or {@code context} is {@code null}. */ Context extractContext(String diagnosticId, Context context); + + /** + * Returns a span builder with the provided name in {@link Context}. + * + *

Code samples

+ * + *

Returns a builder with the provided span name.

+ * {@codesnippet com.azure.core.util.tracing.getSpanBuilder#string-context} + * + * @param spanName Name to give the span for the created builder. + * @param context Additional metadata that is passed through the call stack. + * + * @return The updated {@link Context} object containing the span builder. + * @throws NullPointerException if {@code context} or {@code spanName} is {@code null}. + */ + Context getSpanBuilder(String spanName, Context context); } diff --git a/sdk/core/azure-core/src/samples/java/com/azure/core/util/tracing/TracerJavaDocCodeSnippets.java b/sdk/core/azure-core/src/samples/java/com/azure/core/util/tracing/TracerJavaDocCodeSnippets.java index b84e833acbd8..bc99564e894a 100644 --- a/sdk/core/azure-core/src/samples/java/com/azure/core/util/tracing/TracerJavaDocCodeSnippets.java +++ b/sdk/core/azure-core/src/samples/java/com/azure/core/util/tracing/TracerJavaDocCodeSnippets.java @@ -7,6 +7,7 @@ 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.PARENT_SPAN_KEY; +import static com.azure.core.util.tracing.Tracer.SPAN_BUILDER_KEY; /** * Contains code snippets when generating javadocs through doclets for {@link Tracer}. @@ -120,6 +121,18 @@ public void extractContext() { // END: com.azure.core.util.tracing.extractContext#string-context } + /** + * Code snippet for {@link Tracer#getSpanBuilder(String, Context)} + */ + public void getSpanBuilder() { + // BEGIN: com.azure.core.util.tracing.getSpanBuilder#string-context + // Returns a span builder with the provided name + String methodName = "message-span"; + Context spanContext = tracer.getSpanBuilder(methodName, Context.NONE); + System.out.printf("Span context of the current tracing span: %s%n", spanContext.getData(SPAN_BUILDER_KEY).get()); + // END: com.azure.core.util.tracing.getSpanBuilder#string-context + } + //Noop Tracer private static final class TracerImplementation implements Tracer { @Override @@ -161,5 +174,9 @@ public void addLink(Context context) { public Context extractContext(String diagnosticId, Context context) { return null; } + + @Override + public Context getSpanBuilder(String spanName, Context context) { return null; + } } } diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClient.java b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClient.java index 48e23d755469..967209676adc 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClient.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClient.java @@ -116,8 +116,8 @@ public class EventHubProducerAsyncClient implements Closeable { * load balance the messages amongst available partitions. */ EventHubProducerAsyncClient(String fullyQualifiedNamespace, String eventHubName, EventHubConnection connection, - AmqpRetryOptions retryOptions, TracerProvider tracerProvider, MessageSerializer messageSerializer, - boolean isSharedConnection) { + AmqpRetryOptions retryOptions, TracerProvider tracerProvider, MessageSerializer messageSerializer, + boolean isSharedConnection) { this.fullyQualifiedNamespace = fullyQualifiedNamespace; this.eventHubName = eventHubName; this.connection = connection; @@ -170,9 +170,7 @@ public Flux getPartitionIds() { * events in the partition event stream. * * @param partitionId The unique identifier of a partition associated with the Event Hub. - * * @return The set of information for the requested partition under the Event Hub this client is associated with. - * * @throws NullPointerException if {@code partitionId} is null. */ @ServiceMethod(returns = ReturnType.SINGLE) @@ -193,9 +191,7 @@ public Mono createBatch() { * Creates an {@link EventDataBatch} configured with the options specified. * * @param options A set of options used to configure the {@link EventDataBatch}. - * * @return A new {@link EventDataBatch} that can fit as many events as the transport allows. - * * @throws NullPointerException if {@code options} is null. */ public Mono createBatch(CreateBatchOptions options) { @@ -208,39 +204,39 @@ public Mono createBatch(CreateBatchOptions options) { final int batchMaxSize = options.getMaximumSizeInBytes(); if (!CoreUtils.isNullOrEmpty(partitionKey) - && !CoreUtils.isNullOrEmpty(partitionId)) { + && !CoreUtils.isNullOrEmpty(partitionId)) { return monoError(logger, new IllegalArgumentException(String.format(Locale.US, "CreateBatchOptions.getPartitionKey() and CreateBatchOptions.getPartitionId() are both set. " + "Only one or the other can be used. partitionKey: '%s'. partitionId: '%s'", partitionKey, partitionId))); } else if (!CoreUtils.isNullOrEmpty(partitionKey) - && partitionKey.length() > MAX_PARTITION_KEY_LENGTH) { + && partitionKey.length() > MAX_PARTITION_KEY_LENGTH) { return monoError(logger, new IllegalArgumentException(String.format(Locale.US, "Partition key '%s' exceeds the maximum allowed length: '%s'.", partitionKey, MAX_PARTITION_KEY_LENGTH))); } return getSendLink(partitionId) - .flatMap(link -> link.getLinkSize() - .flatMap(size -> { - final int maximumLinkSize = size > 0 - ? size - : MAX_MESSAGE_LENGTH_BYTES; - - if (batchMaxSize > maximumLinkSize) { - return monoError(logger, - new IllegalArgumentException(String.format(Locale.US, - "BatchOptions.maximumSizeInBytes (%s bytes) is larger than the link size (%s bytes).", - batchMaxSize, maximumLinkSize))); - } - - final int batchSize = batchMaxSize > 0 - ? batchMaxSize - : maximumLinkSize; - - return Mono.just(new EventDataBatch(batchSize, partitionId, partitionKey, link::getErrorContext, - tracerProvider)); - })); + .flatMap(link -> link.getLinkSize() + .flatMap(size -> { + final int maximumLinkSize = size > 0 + ? size + : MAX_MESSAGE_LENGTH_BYTES; + + if (batchMaxSize > maximumLinkSize) { + return monoError(logger, + new IllegalArgumentException(String.format(Locale.US, + "BatchOptions.maximumSizeInBytes (%s bytes) is larger than the link size (%s bytes).", + batchMaxSize, maximumLinkSize))); + } + + final int batchSize = batchMaxSize > 0 + ? batchMaxSize + : maximumLinkSize; + + return Mono.just(new EventDataBatch(batchSize, partitionId, partitionKey, link::getErrorContext, + tracerProvider)); + })); } /** @@ -254,7 +250,6 @@ public Mono createBatch(CreateBatchOptions options) { *

* * @param event Event to send to the service. - * * @return A {@link Mono} that completes when the event is pushed to the service. */ Mono send(EventData event) { @@ -277,7 +272,6 @@ Mono send(EventData event) { * * @param event Event to send to the service. * @param options The set of options to consider when sending this event. - * * @return A {@link Mono} that completes when the event is pushed to the service. */ Mono send(EventData event, SendOptions options) { @@ -296,7 +290,6 @@ Mono send(EventData event, SendOptions options) { * size is the max amount allowed on the link. * * @param events Events to send to the service. - * * @return A {@link Mono} that completes when all events are pushed to the service. */ Mono send(Iterable events) { @@ -314,7 +307,6 @@ Mono send(Iterable events) { * * @param events Events to send to the service. * @param options The set of options to consider when sending this batch. - * * @return A {@link Mono} that completes when all events are pushed to the service. */ Mono send(Iterable events, SendOptions options) { @@ -333,7 +325,6 @@ Mono send(Iterable events, SendOptions options) { * size is the max amount allowed on the link. * * @param events Events to send to the service. - * * @return A {@link Mono} that completes when all events are pushed to the service. */ Mono send(Flux events) { @@ -351,7 +342,6 @@ Mono send(Flux events) { * * @param events Events to send to the service. * @param options The set of options to consider when sending this batch. - * * @return A {@link Mono} that completes when all events are pushed to the service. */ Mono send(Flux events, SendOptions options) { @@ -368,9 +358,7 @@ Mono send(Flux events, SendOptions options) { * Sends the batch to the associated Event Hub. * * @param batch The batch to send to the service. - * * @return A {@link Mono} that completes when the batch is pushed to the service. - * * @throws NullPointerException if {@code batch} is {@code null}. * @see EventHubProducerAsyncClient#createBatch() * @see EventHubProducerAsyncClient#createBatch(CreateBatchOptions) @@ -395,21 +383,28 @@ public Mono send(EventDataBatch batch) { final String partitionKey = batch.getPartitionKey(); final boolean isTracingEnabled = tracerProvider.isEnabled(); final AtomicReference parentContext = isTracingEnabled - ? new AtomicReference<>(Context.NONE) - : null; + ? new AtomicReference<>(Context.NONE) + : null; + final AtomicReference sharedContext = isTracingEnabled + ? new AtomicReference<>(Context.NONE) + : null; + final AtomicReference isFirst = new AtomicReference<>(true); - // initiate addLink on send span, get shared span builder for all links - Context sharedContext = isTracingEnabled ? tracerProvider.startSpan(parentContext.get(), ProcessKind.LINK) : - null; final List messages = batch.getEvents().stream().map(event -> { final Message message = messageSerializer.serialize(event); if (isTracingEnabled) { - tracerProvider.addSpanLinks(sharedContext.addData(SPAN_CONTEXT_KEY, event.getContext())); + if (isFirst.getAndSet(false)) { + parentContext.set(event.getContext()); + // get shared span builder for all links + sharedContext.set(tracerProvider.getSpanBuilder(parentContext.get())); + } + Context sharedSpanBuilderContext = sharedContext.get(); + tracerProvider.addSpanLinks(sharedSpanBuilderContext.addData(SPAN_CONTEXT_KEY, event.getContext())); } if (!CoreUtils.isNullOrEmpty(partitionKey)) { final MessageAnnotations messageAnnotations = message.getMessageAnnotations() == null - ? new MessageAnnotations(new HashMap<>()) - : message.getMessageAnnotations(); + ? new MessageAnnotations(new HashMap<>()) + : message.getMessageAnnotations(); messageAnnotations.getValue().put(AmqpConstants.PARTITION_KEY, partitionKey); message.setMessageAnnotations(messageAnnotations); @@ -419,19 +414,19 @@ public Mono send(EventDataBatch batch) { }).collect(Collectors.toList()); return getSendLink(batch.getPartitionId()) - .flatMap(link -> { - if (isTracingEnabled) { - Context userSpanContext = sharedContext; - Context entityContext = userSpanContext.addData(ENTITY_PATH_KEY, link.getEntityPath()); - // start send span and store updated context - parentContext.set(tracerProvider.startSpan( - entityContext.addData(HOST_NAME_KEY, link.getHostname()), ProcessKind.SEND)); - } - return messages.size() == 1 - ? link.send(messages.get(0)) - : link.send(messages); - - }).doOnEach(signal -> { + .flatMap(link -> { + if (isTracingEnabled) { + Context userSpanContext = sharedContext.get(); + Context entityContext = userSpanContext.addData(ENTITY_PATH_KEY, link.getEntityPath()); + // start send span and store updated context + parentContext.set(tracerProvider.startSpan( + entityContext.addData(HOST_NAME_KEY, link.getHostname()), ProcessKind.SEND)); + } + return messages.size() == 1 + ? link.send(messages.get(0)) + : link.send(messages); + + }).doOnEach(signal -> { if (isTracingEnabled) { tracerProvider.endSpan(parentContext.get(), signal); } @@ -443,7 +438,7 @@ private Mono sendInternal(Flux events, SendOptions options) { final String partitionId = options.getPartitionId(); if (!CoreUtils.isNullOrEmpty(partitionKey) - && !CoreUtils.isNullOrEmpty(partitionId)) { + && !CoreUtils.isNullOrEmpty(partitionId)) { return monoError(logger, new IllegalArgumentException(String.format(Locale.US, "SendOptions.getPartitionKey() and SendOptions.getPartitionId() are both set. Only one or the" + " other can be used. partitionKey: '%s'. partitionId: '%s'", @@ -451,38 +446,38 @@ private Mono sendInternal(Flux events, SendOptions options) { } return getSendLink(options.getPartitionId()) - .flatMap(link -> link.getLinkSize() - .flatMap(size -> { - final int batchSize = size > 0 ? size : MAX_MESSAGE_LENGTH_BYTES; - final CreateBatchOptions batchOptions = new CreateBatchOptions() - .setPartitionKey(options.getPartitionKey()) - .setPartitionId(options.getPartitionId()) - .setMaximumSizeInBytes(batchSize); - return events.collect(new EventDataCollector(batchOptions, 1, link::getErrorContext, - tracerProvider)); - }) - .flatMap(list -> sendInternal(Flux.fromIterable(list)))); + .flatMap(link -> link.getLinkSize() + .flatMap(size -> { + final int batchSize = size > 0 ? size : MAX_MESSAGE_LENGTH_BYTES; + final CreateBatchOptions batchOptions = new CreateBatchOptions() + .setPartitionKey(options.getPartitionKey()) + .setPartitionId(options.getPartitionId()) + .setMaximumSizeInBytes(batchSize); + return events.collect(new EventDataCollector(batchOptions, 1, link::getErrorContext, + tracerProvider)); + }) + .flatMap(list -> sendInternal(Flux.fromIterable(list)))); } private Mono sendInternal(Flux eventBatches) { return eventBatches - .flatMap(this::send) - .then() - .doOnError(error -> { - logger.error(Messages.ERROR_SENDING_BATCH, error); - }); + .flatMap(this::send) + .then() + .doOnError(error -> { + logger.error(Messages.ERROR_SENDING_BATCH, error); + }); } private String getEntityPath(String partitionId) { return CoreUtils.isNullOrEmpty(partitionId) - ? eventHubName - : String.format(Locale.US, SENDER_ENTITY_PATH_FORMAT, eventHubName, partitionId); + ? eventHubName + : String.format(Locale.US, SENDER_ENTITY_PATH_FORMAT, eventHubName, partitionId); } private String getLinkName(String partitionId) { return CoreUtils.isNullOrEmpty(partitionId) - ? StringUtil.getRandomString("EC") - : StringUtil.getRandomString("PS"); + ? StringUtil.getRandomString("EC") + : StringUtil.getRandomString("PS"); } private Mono getSendLink(String partitionId) { @@ -493,7 +488,7 @@ private Mono getSendLink(String partitionId) { return Mono.just(openLink); } else { return connection.createSendLink(getLinkName(partitionId), entityPath, retryOptions) - .map(link -> openLinks.computeIfAbsent(entityPath, unusedKey -> link)); + .map(link -> openLinks.computeIfAbsent(entityPath, unusedKey -> link)); } } @@ -522,7 +517,7 @@ public void close() { * AmqpErrorCondition#LINK_PAYLOAD_SIZE_EXCEEDED}. */ private static class EventDataCollector implements Collector, - List> { + List> { private final String partitionKey; private final String partitionId; private final int maxMessageSize; @@ -533,11 +528,11 @@ private static class EventDataCollector implements Collector 0 - ? options.getMaximumSizeInBytes() - : MAX_MESSAGE_LENGTH_BYTES; + ? options.getMaximumSizeInBytes() + : MAX_MESSAGE_LENGTH_BYTES; this.partitionKey = options.getPartitionKey(); this.partitionId = options.getPartitionId(); this.contextProvider = contextProvider; diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClientTest.java b/sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClientTest.java index f21d8734f4bf..0863275b439f 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClientTest.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClientTest.java @@ -245,7 +245,7 @@ public void sendStartSpanSingleMessage() { } ); - when(tracer1.start(eq("Azure.eventhubs.send"), any(), eq(ProcessKind.LINK))).thenAnswer( + when(tracer1.getSpanBuilder(eq("Azure.eventhubs.send"), any())).thenAnswer( invocation -> { Context passed = invocation.getArgument(1, Context.class); return passed.addData(SPAN_BUILDER_KEY, "value"); @@ -301,7 +301,8 @@ public void sendMessageRetrySpanTest() { return passed.addData(PARENT_SPAN_KEY, "value"); } ); - when(tracer1.start(eq("Azure.eventhubs.send"), any(), eq(ProcessKind.LINK))).thenAnswer( + + when(tracer1.getSpanBuilder(eq("Azure.eventhubs.send"), any())).thenAnswer( invocation -> { Context passed = invocation.getArgument(1, Context.class); return passed.addData(SPAN_BUILDER_KEY, "value"); diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/EventHubProducerClientTest.java b/sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/EventHubProducerClientTest.java index 516bfb79b739..fbcf352b3b0a 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/EventHubProducerClientTest.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/EventHubProducerClientTest.java @@ -169,7 +169,7 @@ public void sendStartSpanSingleMessage() { return passed.addData(PARENT_SPAN_KEY, "value").addData(DIAGNOSTIC_ID_KEY, "value2"); } ); - when(tracer1.start(eq("Azure.eventhubs.send"), any(), eq(ProcessKind.LINK))).thenAnswer( + when(tracer1.getSpanBuilder(eq("Azure.eventhubs.send"), any())).thenAnswer( invocation -> { Context passed = invocation.getArgument(1, Context.class); return passed.addData(SPAN_BUILDER_KEY, "value"); @@ -217,7 +217,7 @@ public void sendMessageRetrySpanTest() { } ); - when(tracer1.start(eq("Azure.eventhubs.send"), any(), eq(ProcessKind.LINK))).thenAnswer( + when(tracer1.getSpanBuilder(eq("Azure.eventhubs.send"), any())).thenAnswer( invocation -> { Context passed = invocation.getArgument(1, Context.class); return passed.addData(SPAN_BUILDER_KEY, "value"); From ba1bf10dd6a84fa65a33d19f547a228edf09f8f3 Mon Sep 17 00:00:00 2001 From: Sameeksha Vaity Date: Wed, 11 Dec 2019 18:05:34 -0800 Subject: [PATCH 4/9] add tracer provider test --- .../amqp/implementation/TracerProvider.java | 4 +- .../implementation/TracerProviderTest.java | 37 +++++++++++++++++++ .../tracing/opencensus/OpenCensusTracer.java | 6 +++ .../opentelemetry/OpenTelemetryTracer.java | 12 +++--- .../OpenTelemetryTracerTest.java | 2 +- .../com/azure/core/util/tracing/Tracer.java | 4 +- .../tracing/TracerJavaDocCodeSnippets.java | 8 ++-- .../EventHubProducerAsyncClient.java | 2 +- .../EventHubProducerAsyncClientTest.java | 4 +- .../eventhubs/EventHubProducerClientTest.java | 4 +- 10 files changed, 63 insertions(+), 20 deletions(-) diff --git a/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/TracerProvider.java b/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/TracerProvider.java index 7fb7ab659813..526d657154fe 100644 --- a/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/TracerProvider.java +++ b/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/TracerProvider.java @@ -113,11 +113,11 @@ public Context extractContext(String diagnosticId, Context context) { * * @param context Additional metadata containing the span name for creating the span builer. */ - public Context getSpanBuilder(Context context) { + public Context getSharedSpanBuilder(Context context) { Context local = Objects.requireNonNull(context, "'context' cannot be null."); String spanName = getSpanName(ProcessKind.SEND); for (Tracer tracer : tracers) { - local = tracer.getSpanBuilder(spanName, local); + local = tracer.getSharedSpanBuilder(spanName, local); } return local; } diff --git a/sdk/core/azure-core-amqp/src/test/java/com/azure/core/amqp/implementation/TracerProviderTest.java b/sdk/core/azure-core-amqp/src/test/java/com/azure/core/amqp/implementation/TracerProviderTest.java index 1c48c1c00181..e9d3f120e6d0 100644 --- a/sdk/core/azure-core-amqp/src/test/java/com/azure/core/amqp/implementation/TracerProviderTest.java +++ b/sdk/core/azure-core-amqp/src/test/java/com/azure/core/amqp/implementation/TracerProviderTest.java @@ -24,6 +24,7 @@ import static org.junit.jupiter.api.Assertions.assertThrows; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.ArgumentMatchers.isNull; import static org.mockito.Mockito.never; @@ -194,4 +195,40 @@ public void addSpanLinks() { // Act assertThrows(NullPointerException.class, () -> tracerProvider.addSpanLinks(null)); } + + @Test + public void getSpanBuilderReturnsUpdatedContext() { + // Arrange + final String spanBuilderKey = "spanBuilder-key"; + final String spanBuilderValue = "spanBuilder-value"; + + final String spanBuilderKey1 = "spanBuilder-key1"; + final String spanBuilderValue1 = "spanBuilder-value1"; + final Context startingContext = Context.NONE; + + when(tracer.getSharedSpanBuilder(anyString(), any())).thenAnswer( + invocation -> { + Context passed = invocation.getArgument(1, Context.class); + return passed.addData(spanBuilderKey, spanBuilderValue); + } + ); + when(tracer2.getSharedSpanBuilder(anyString(), any())).thenAnswer( + invocation -> { + Context passed = invocation.getArgument(1, Context.class); + return passed.addData(spanBuilderKey1, spanBuilderValue1); + } + ); + + // Act + final Context updatedContext = tracerProvider.getSharedSpanBuilder(startingContext); + + // Assert + final Optional spanBuilderData = updatedContext.getData(spanBuilderKey); + Assertions.assertTrue(spanBuilderData.isPresent()); + Assertions.assertEquals(spanBuilderValue, spanBuilderData.get()); + + final Optional spanBuilderData1 = updatedContext.getData(spanBuilderKey1); + Assertions.assertTrue(spanBuilderData1.isPresent()); + Assertions.assertEquals(spanBuilderValue1, spanBuilderData1.get()); + } } diff --git a/sdk/core/azure-core-tracing-opencensus/src/main/java/com/azure/core/tracing/opencensus/OpenCensusTracer.java b/sdk/core/azure-core-tracing-opencensus/src/main/java/com/azure/core/tracing/opencensus/OpenCensusTracer.java index c786fdfdbfe7..ee1042b871ab 100644 --- a/sdk/core/azure-core-tracing-opencensus/src/main/java/com/azure/core/tracing/opencensus/OpenCensusTracer.java +++ b/sdk/core/azure-core-tracing-opencensus/src/main/java/com/azure/core/tracing/opencensus/OpenCensusTracer.java @@ -181,6 +181,12 @@ public Context extractContext(String diagnosticId, Context context) { return AmqpPropagationFormatUtil.extractContext(diagnosticId, context); } + @Override + public Context getSharedSpanBuilder(String spanName, Context context) { + return Context.NONE; + // Remove OpenCensus support for tracing in https://github.com/Azure/azure-sdk-for-java/issues/6781 + } + /** * Starts a new child {@link Span} with parent being the remote and uses the {@link Span} is in the current Context, * to return an object that represents that scope. diff --git a/sdk/core/azure-core-tracing-opentelemetry/src/main/java/com/azure/core/tracing/opentelemetry/OpenTelemetryTracer.java b/sdk/core/azure-core-tracing-opentelemetry/src/main/java/com/azure/core/tracing/opentelemetry/OpenTelemetryTracer.java index b631d887bc24..51cfe65b84ee 100644 --- a/sdk/core/azure-core-tracing-opentelemetry/src/main/java/com/azure/core/tracing/opentelemetry/OpenTelemetryTracer.java +++ b/sdk/core/azure-core-tracing-opentelemetry/src/main/java/com/azure/core/tracing/opentelemetry/OpenTelemetryTracer.java @@ -46,7 +46,7 @@ public Context start(String spanName, Context context) { Objects.requireNonNull(spanName, "'spanName' cannot be null."); Objects.requireNonNull(context, "'context' cannot be null."); - Builder spanBuilder = getParentSpanBuilder(spanName, context); + Builder spanBuilder = getSpanBuilder(spanName, context); Span span = spanBuilder.startSpan(); return context.addData(PARENT_SPAN_KEY, span); @@ -78,7 +78,7 @@ public Context start(String spanName, Context context, ProcessKind processKind) } return context.addData(PARENT_SPAN_KEY, span); case MESSAGE: - spanBuilder = getParentSpanBuilder(spanName, context); + spanBuilder = getSpanBuilder(spanName, context); span = spanBuilder.startSpan(); // Add diagnostic Id and trace-headers to Context context = setContextData(span); @@ -177,8 +177,8 @@ public Context extractContext(String diagnosticId, Context context) { } @Override - public Context getSpanBuilder(String spanName, Context context) { - return context.addData(SPAN_BUILDER_KEY, getParentSpanBuilder(spanName, context)); + public Context getSharedSpanBuilder(String spanName, Context context) { + return context.addData(SPAN_BUILDER_KEY, getSpanBuilder(spanName, context)); } /** @@ -197,7 +197,7 @@ private Context startScopedSpan(String spanName, Context context) { if (spanContext != null) { span = startSpanWithRemoteParent(spanName, spanContext); } else { - Builder spanBuilder = getParentSpanBuilder(spanName, context); + Builder spanBuilder = getSpanBuilder(spanName, context); span = spanBuilder.setSpanKind(Span.Kind.SERVER).startSpan(); } if (span.isRecording()) { @@ -277,7 +277,7 @@ private static String parseComponentValue(String spanName) { * @param context The context containing the span and the span name. * @return A {@code Span.Builder} to create and start a new {@code Span}. */ - private Builder getParentSpanBuilder(String spanName, Context context) { + private Builder getSpanBuilder(String spanName, Context context) { Span parentSpan = getOrDefault(context, PARENT_SPAN_KEY, null, Span.class); String spanNameKey = getOrDefault(context, USER_SPAN_NAME_KEY, null, String.class); diff --git a/sdk/core/azure-core-tracing-opentelemetry/src/test/java/com/azure/core/tracing/opentelemetry/OpenTelemetryTracerTest.java b/sdk/core/azure-core-tracing-opentelemetry/src/test/java/com/azure/core/tracing/opentelemetry/OpenTelemetryTracerTest.java index 693b3d237642..668c5fb93505 100644 --- a/sdk/core/azure-core-tracing-opentelemetry/src/test/java/com/azure/core/tracing/opentelemetry/OpenTelemetryTracerTest.java +++ b/sdk/core/azure-core-tracing-opentelemetry/src/test/java/com/azure/core/tracing/opentelemetry/OpenTelemetryTracerTest.java @@ -185,7 +185,7 @@ public void startSpanProcessKindProcess() { @Test public void getSpanBuilderTest() { // Act - final Context updatedContext = openTelemetryTracer.getSpanBuilder(METHOD_NAME, Context.NONE); + final Context updatedContext = openTelemetryTracer.getSharedSpanBuilder(METHOD_NAME, Context.NONE); assertTrue(updatedContext.getData(SPAN_BUILDER_KEY).isPresent()); } diff --git a/sdk/core/azure-core/src/main/java/com/azure/core/util/tracing/Tracer.java b/sdk/core/azure-core/src/main/java/com/azure/core/util/tracing/Tracer.java index 2090b8307cec..c6d052b5dc32 100644 --- a/sdk/core/azure-core/src/main/java/com/azure/core/util/tracing/Tracer.java +++ b/sdk/core/azure-core/src/main/java/com/azure/core/util/tracing/Tracer.java @@ -211,7 +211,7 @@ public interface Tracer { *

Code samples

* *

Returns a builder with the provided span name.

- * {@codesnippet com.azure.core.util.tracing.getSpanBuilder#string-context} + * {@codesnippet com.azure.core.util.tracing.getSharedSpanBuilder#string-context} * * @param spanName Name to give the span for the created builder. * @param context Additional metadata that is passed through the call stack. @@ -219,5 +219,5 @@ public interface Tracer { * @return The updated {@link Context} object containing the span builder. * @throws NullPointerException if {@code context} or {@code spanName} is {@code null}. */ - Context getSpanBuilder(String spanName, Context context); + Context getSharedSpanBuilder(String spanName, Context context); } diff --git a/sdk/core/azure-core/src/samples/java/com/azure/core/util/tracing/TracerJavaDocCodeSnippets.java b/sdk/core/azure-core/src/samples/java/com/azure/core/util/tracing/TracerJavaDocCodeSnippets.java index bc99564e894a..6710fe08ab50 100644 --- a/sdk/core/azure-core/src/samples/java/com/azure/core/util/tracing/TracerJavaDocCodeSnippets.java +++ b/sdk/core/azure-core/src/samples/java/com/azure/core/util/tracing/TracerJavaDocCodeSnippets.java @@ -122,13 +122,13 @@ public void extractContext() { } /** - * Code snippet for {@link Tracer#getSpanBuilder(String, Context)} + * Code snippet for {@link Tracer#getSharedSpanBuilder(String, Context)} */ - public void getSpanBuilder() { + public void getSharedSpanBuilder() { // BEGIN: com.azure.core.util.tracing.getSpanBuilder#string-context // Returns a span builder with the provided name String methodName = "message-span"; - Context spanContext = tracer.getSpanBuilder(methodName, Context.NONE); + Context spanContext = tracer.getSharedSpanBuilder(methodName, Context.NONE); System.out.printf("Span context of the current tracing span: %s%n", spanContext.getData(SPAN_BUILDER_KEY).get()); // END: com.azure.core.util.tracing.getSpanBuilder#string-context } @@ -176,7 +176,7 @@ public Context extractContext(String diagnosticId, Context context) { } @Override - public Context getSpanBuilder(String spanName, Context context) { return null; + public Context getSharedSpanBuilder(String spanName, Context context) { return null; } } } diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClient.java b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClient.java index 967209676adc..e07c9f4a1007 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClient.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClient.java @@ -396,7 +396,7 @@ public Mono send(EventDataBatch batch) { if (isFirst.getAndSet(false)) { parentContext.set(event.getContext()); // get shared span builder for all links - sharedContext.set(tracerProvider.getSpanBuilder(parentContext.get())); + sharedContext.set(tracerProvider.getSharedSpanBuilder(parentContext.get())); } Context sharedSpanBuilderContext = sharedContext.get(); tracerProvider.addSpanLinks(sharedSpanBuilderContext.addData(SPAN_CONTEXT_KEY, event.getContext())); diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClientTest.java b/sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClientTest.java index 0863275b439f..5927054d0ab8 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClientTest.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClientTest.java @@ -245,7 +245,7 @@ public void sendStartSpanSingleMessage() { } ); - when(tracer1.getSpanBuilder(eq("Azure.eventhubs.send"), any())).thenAnswer( + when(tracer1.getSharedSpanBuilder(eq("Azure.eventhubs.send"), any())).thenAnswer( invocation -> { Context passed = invocation.getArgument(1, Context.class); return passed.addData(SPAN_BUILDER_KEY, "value"); @@ -302,7 +302,7 @@ public void sendMessageRetrySpanTest() { } ); - when(tracer1.getSpanBuilder(eq("Azure.eventhubs.send"), any())).thenAnswer( + when(tracer1.getSharedSpanBuilder(eq("Azure.eventhubs.send"), any())).thenAnswer( invocation -> { Context passed = invocation.getArgument(1, Context.class); return passed.addData(SPAN_BUILDER_KEY, "value"); diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/EventHubProducerClientTest.java b/sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/EventHubProducerClientTest.java index fbcf352b3b0a..5a3a018f7a54 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/EventHubProducerClientTest.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/EventHubProducerClientTest.java @@ -169,7 +169,7 @@ public void sendStartSpanSingleMessage() { return passed.addData(PARENT_SPAN_KEY, "value").addData(DIAGNOSTIC_ID_KEY, "value2"); } ); - when(tracer1.getSpanBuilder(eq("Azure.eventhubs.send"), any())).thenAnswer( + when(tracer1.getSharedSpanBuilder(eq("Azure.eventhubs.send"), any())).thenAnswer( invocation -> { Context passed = invocation.getArgument(1, Context.class); return passed.addData(SPAN_BUILDER_KEY, "value"); @@ -217,7 +217,7 @@ public void sendMessageRetrySpanTest() { } ); - when(tracer1.getSpanBuilder(eq("Azure.eventhubs.send"), any())).thenAnswer( + when(tracer1.getSharedSpanBuilder(eq("Azure.eventhubs.send"), any())).thenAnswer( invocation -> { Context passed = invocation.getArgument(1, Context.class); return passed.addData(SPAN_BUILDER_KEY, "value"); From 3f0387dc9cd86a292a2336c91335ea4fc7f9a9aa Mon Sep 17 00:00:00 2001 From: Sameeksha Vaity Date: Wed, 11 Dec 2019 18:37:46 -0800 Subject: [PATCH 5/9] update checkstyle --- .../EventHubProducerAsyncClient.java | 29 ++++++++++--------- 1 file changed, 15 insertions(+), 14 deletions(-) diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClient.java b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClient.java index e07c9f4a1007..13a0ba2522bc 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClient.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClient.java @@ -116,8 +116,8 @@ public class EventHubProducerAsyncClient implements Closeable { * load balance the messages amongst available partitions. */ EventHubProducerAsyncClient(String fullyQualifiedNamespace, String eventHubName, EventHubConnection connection, - AmqpRetryOptions retryOptions, TracerProvider tracerProvider, MessageSerializer messageSerializer, - boolean isSharedConnection) { + AmqpRetryOptions retryOptions, TracerProvider tracerProvider, + MessageSerializer messageSerializer, boolean isSharedConnection) { this.fullyQualifiedNamespace = fullyQualifiedNamespace; this.eventHubName = eventHubName; this.connection = connection; @@ -226,7 +226,8 @@ public Mono createBatch(CreateBatchOptions options) { if (batchMaxSize > maximumLinkSize) { return monoError(logger, new IllegalArgumentException(String.format(Locale.US, - "BatchOptions.maximumSizeInBytes (%s bytes) is larger than the link size (%s bytes).", + "BatchOptions.maximumSizeInBytes (%s bytes)" + + " is larger than the link size (%s bytes).", batchMaxSize, maximumLinkSize))); } @@ -234,8 +235,8 @@ public Mono createBatch(CreateBatchOptions options) { ? batchMaxSize : maximumLinkSize; - return Mono.just(new EventDataBatch(batchSize, partitionId, partitionKey, link::getErrorContext, - tracerProvider)); + return Mono.just(new EventDataBatch(batchSize, + partitionId, partitionKey, link::getErrorContext, tracerProvider)); })); } @@ -427,9 +428,9 @@ public Mono send(EventDataBatch batch) { : link.send(messages); }).doOnEach(signal -> { - if (isTracingEnabled) { - tracerProvider.endSpan(parentContext.get(), signal); - } + if (isTracingEnabled) { + tracerProvider.endSpan(parentContext.get(), signal); + } }); } @@ -449,12 +450,12 @@ private Mono sendInternal(Flux events, SendOptions options) { .flatMap(link -> link.getLinkSize() .flatMap(size -> { final int batchSize = size > 0 ? size : MAX_MESSAGE_LENGTH_BYTES; - final CreateBatchOptions batchOptions = new CreateBatchOptions() - .setPartitionKey(options.getPartitionKey()) - .setPartitionId(options.getPartitionId()) - .setMaximumSizeInBytes(batchSize); - return events.collect(new EventDataCollector(batchOptions, 1, link::getErrorContext, - tracerProvider)); + final CreateBatchOptions batchOptions = + new CreateBatchOptions().setPartitionKey(options.getPartitionKey()) + .setPartitionId(options.getPartitionId()) + .setMaximumSizeInBytes(batchSize); + return events.collect(new EventDataCollector(batchOptions, 1, + link::getErrorContext, tracerProvider)); }) .flatMap(list -> sendInternal(Flux.fromIterable(list)))); } From e543802a46a75b76b7ef4bcbdd17480bf1236d07 Mon Sep 17 00:00:00 2001 From: Sameeksha Vaity Date: Thu, 12 Dec 2019 12:18:32 -0800 Subject: [PATCH 6/9] update identation --- .../tracing/opencensus/OpenCensusTracer.java | 2 + .../OpenTelemetryTracerTest.java | 2 +- .../EventHubProducerAsyncClient.java | 160 ++++++++++-------- 3 files changed, 88 insertions(+), 76 deletions(-) diff --git a/sdk/core/azure-core-tracing-opencensus/src/main/java/com/azure/core/tracing/opencensus/OpenCensusTracer.java b/sdk/core/azure-core-tracing-opencensus/src/main/java/com/azure/core/tracing/opencensus/OpenCensusTracer.java index ee1042b871ab..6d142b90acda 100644 --- a/sdk/core/azure-core-tracing-opencensus/src/main/java/com/azure/core/tracing/opencensus/OpenCensusTracer.java +++ b/sdk/core/azure-core-tracing-opencensus/src/main/java/com/azure/core/tracing/opencensus/OpenCensusTracer.java @@ -183,6 +183,8 @@ public Context extractContext(String diagnosticId, Context context) { @Override public Context getSharedSpanBuilder(String spanName, Context context) { + logger.logExceptionAsError( + new UnsupportedOperationException("This operation is not supported in OpenCensus")); return Context.NONE; // Remove OpenCensus support for tracing in https://github.com/Azure/azure-sdk-for-java/issues/6781 } diff --git a/sdk/core/azure-core-tracing-opentelemetry/src/test/java/com/azure/core/tracing/opentelemetry/OpenTelemetryTracerTest.java b/sdk/core/azure-core-tracing-opentelemetry/src/test/java/com/azure/core/tracing/opentelemetry/OpenTelemetryTracerTest.java index 668c5fb93505..0103a549707a 100644 --- a/sdk/core/azure-core-tracing-opentelemetry/src/test/java/com/azure/core/tracing/opentelemetry/OpenTelemetryTracerTest.java +++ b/sdk/core/azure-core-tracing-opentelemetry/src/test/java/com/azure/core/tracing/opentelemetry/OpenTelemetryTracerTest.java @@ -120,7 +120,7 @@ public void startSpanProcessKindSend() { final Span.Builder spanBuilder = tracer.spanBuilder(METHOD_NAME); // Add additional metadata to spans for SEND final Context traceContext = tracingContext.addData(ENTITY_PATH_KEY, ENTITY_PATH_VALUE) - .addData(HOST_NAME_KEY, HOSTNAME_VALUE).addData(SPAN_BUILDER_KEY, spanBuilder); + .addData(HOST_NAME_KEY, HOSTNAME_VALUE).addData(SPAN_BUILDER_KEY, spanBuilder); // Act final Context updatedContext = openTelemetryTracer.start(METHOD_NAME, traceContext, ProcessKind.SEND); diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClient.java b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClient.java index 13a0ba2522bc..b58d82375964 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClient.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClient.java @@ -116,8 +116,8 @@ public class EventHubProducerAsyncClient implements Closeable { * load balance the messages amongst available partitions. */ EventHubProducerAsyncClient(String fullyQualifiedNamespace, String eventHubName, EventHubConnection connection, - AmqpRetryOptions retryOptions, TracerProvider tracerProvider, - MessageSerializer messageSerializer, boolean isSharedConnection) { + AmqpRetryOptions retryOptions, TracerProvider tracerProvider, MessageSerializer messageSerializer, + boolean isSharedConnection) { this.fullyQualifiedNamespace = fullyQualifiedNamespace; this.eventHubName = eventHubName; this.connection = connection; @@ -170,7 +170,9 @@ public Flux getPartitionIds() { * events in the partition event stream. * * @param partitionId The unique identifier of a partition associated with the Event Hub. + * * @return The set of information for the requested partition under the Event Hub this client is associated with. + * * @throws NullPointerException if {@code partitionId} is null. */ @ServiceMethod(returns = ReturnType.SINGLE) @@ -191,7 +193,9 @@ public Mono createBatch() { * Creates an {@link EventDataBatch} configured with the options specified. * * @param options A set of options used to configure the {@link EventDataBatch}. + * * @return A new {@link EventDataBatch} that can fit as many events as the transport allows. + * * @throws NullPointerException if {@code options} is null. */ public Mono createBatch(CreateBatchOptions options) { @@ -204,40 +208,39 @@ public Mono createBatch(CreateBatchOptions options) { final int batchMaxSize = options.getMaximumSizeInBytes(); if (!CoreUtils.isNullOrEmpty(partitionKey) - && !CoreUtils.isNullOrEmpty(partitionId)) { + && !CoreUtils.isNullOrEmpty(partitionId)) { return monoError(logger, new IllegalArgumentException(String.format(Locale.US, "CreateBatchOptions.getPartitionKey() and CreateBatchOptions.getPartitionId() are both set. " + "Only one or the other can be used. partitionKey: '%s'. partitionId: '%s'", partitionKey, partitionId))); } else if (!CoreUtils.isNullOrEmpty(partitionKey) - && partitionKey.length() > MAX_PARTITION_KEY_LENGTH) { + && partitionKey.length() > MAX_PARTITION_KEY_LENGTH) { return monoError(logger, new IllegalArgumentException(String.format(Locale.US, "Partition key '%s' exceeds the maximum allowed length: '%s'.", partitionKey, MAX_PARTITION_KEY_LENGTH))); } return getSendLink(partitionId) - .flatMap(link -> link.getLinkSize() - .flatMap(size -> { - final int maximumLinkSize = size > 0 - ? size - : MAX_MESSAGE_LENGTH_BYTES; - - if (batchMaxSize > maximumLinkSize) { - return monoError(logger, - new IllegalArgumentException(String.format(Locale.US, - "BatchOptions.maximumSizeInBytes (%s bytes)" - + " is larger than the link size (%s bytes).", - batchMaxSize, maximumLinkSize))); - } - - final int batchSize = batchMaxSize > 0 - ? batchMaxSize - : maximumLinkSize; - - return Mono.just(new EventDataBatch(batchSize, - partitionId, partitionKey, link::getErrorContext, tracerProvider)); - })); + .flatMap(link -> link.getLinkSize() + .flatMap(size -> { + final int maximumLinkSize = size > 0 + ? size + : MAX_MESSAGE_LENGTH_BYTES; + + if (batchMaxSize > maximumLinkSize) { + return monoError(logger, + new IllegalArgumentException(String.format(Locale.US, + "BatchOptions.maximumSizeInBytes (%s bytes) is larger than the link size (%s bytes).", + batchMaxSize, maximumLinkSize))); + } + + final int batchSize = batchMaxSize > 0 + ? batchMaxSize + : maximumLinkSize; + + return Mono.just(new EventDataBatch(batchSize, partitionId, partitionKey, link::getErrorContext, + tracerProvider)); + })); } /** @@ -251,6 +254,7 @@ public Mono createBatch(CreateBatchOptions options) { *

* * @param event Event to send to the service. + * * @return A {@link Mono} that completes when the event is pushed to the service. */ Mono send(EventData event) { @@ -273,6 +277,7 @@ Mono send(EventData event) { * * @param event Event to send to the service. * @param options The set of options to consider when sending this event. + * * @return A {@link Mono} that completes when the event is pushed to the service. */ Mono send(EventData event, SendOptions options) { @@ -291,6 +296,7 @@ Mono send(EventData event, SendOptions options) { * size is the max amount allowed on the link. * * @param events Events to send to the service. + * * @return A {@link Mono} that completes when all events are pushed to the service. */ Mono send(Iterable events) { @@ -308,6 +314,7 @@ Mono send(Iterable events) { * * @param events Events to send to the service. * @param options The set of options to consider when sending this batch. + * * @return A {@link Mono} that completes when all events are pushed to the service. */ Mono send(Iterable events, SendOptions options) { @@ -326,6 +333,7 @@ Mono send(Iterable events, SendOptions options) { * size is the max amount allowed on the link. * * @param events Events to send to the service. + * * @return A {@link Mono} that completes when all events are pushed to the service. */ Mono send(Flux events) { @@ -343,6 +351,7 @@ Mono send(Flux events) { * * @param events Events to send to the service. * @param options The set of options to consider when sending this batch. + * * @return A {@link Mono} that completes when all events are pushed to the service. */ Mono send(Flux events, SendOptions options) { @@ -359,7 +368,9 @@ Mono send(Flux events, SendOptions options) { * Sends the batch to the associated Event Hub. * * @param batch The batch to send to the service. + * * @return A {@link Mono} that completes when the batch is pushed to the service. + * * @throws NullPointerException if {@code batch} is {@code null}. * @see EventHubProducerAsyncClient#createBatch() * @see EventHubProducerAsyncClient#createBatch(CreateBatchOptions) @@ -384,18 +395,18 @@ public Mono send(EventDataBatch batch) { final String partitionKey = batch.getPartitionKey(); final boolean isTracingEnabled = tracerProvider.isEnabled(); final AtomicReference parentContext = isTracingEnabled - ? new AtomicReference<>(Context.NONE) - : null; + ? new AtomicReference<>(Context.NONE) + : null; final AtomicReference sharedContext = isTracingEnabled - ? new AtomicReference<>(Context.NONE) - : null; + ? new AtomicReference<>(Context.NONE) + : null; final AtomicReference isFirst = new AtomicReference<>(true); final List messages = batch.getEvents().stream().map(event -> { final Message message = messageSerializer.serialize(event); if (isTracingEnabled) { + parentContext.set(event.getContext()); if (isFirst.getAndSet(false)) { - parentContext.set(event.getContext()); // get shared span builder for all links sharedContext.set(tracerProvider.getSharedSpanBuilder(parentContext.get())); } @@ -404,33 +415,32 @@ public Mono send(EventDataBatch batch) { } if (!CoreUtils.isNullOrEmpty(partitionKey)) { final MessageAnnotations messageAnnotations = message.getMessageAnnotations() == null - ? new MessageAnnotations(new HashMap<>()) - : message.getMessageAnnotations(); + ? new MessageAnnotations(new HashMap<>()) + : message.getMessageAnnotations(); messageAnnotations.getValue().put(AmqpConstants.PARTITION_KEY, partitionKey); message.setMessageAnnotations(messageAnnotations); - } return message; }).collect(Collectors.toList()); return getSendLink(batch.getPartitionId()) - .flatMap(link -> { - if (isTracingEnabled) { - Context userSpanContext = sharedContext.get(); - Context entityContext = userSpanContext.addData(ENTITY_PATH_KEY, link.getEntityPath()); - // start send span and store updated context - parentContext.set(tracerProvider.startSpan( - entityContext.addData(HOST_NAME_KEY, link.getHostname()), ProcessKind.SEND)); - } - return messages.size() == 1 - ? link.send(messages.get(0)) - : link.send(messages); - - }).doOnEach(signal -> { - if (isTracingEnabled) { - tracerProvider.endSpan(parentContext.get(), signal); - } + .flatMap(link -> { + if (isTracingEnabled) { + Context userSpanContext = sharedContext.get(); + Context entityContext = userSpanContext.addData(ENTITY_PATH_KEY, link.getEntityPath()); + // start send span and store updated context + parentContext.set(tracerProvider.startSpan( + entityContext.addData(HOST_NAME_KEY, link.getHostname()), ProcessKind.SEND)); + } + return messages.size() == 1 + ? link.send(messages.get(0)) + : link.send(messages); + + }).doOnEach(signal -> { + if (isTracingEnabled) { + tracerProvider.endSpan(parentContext.get(), signal); + } }); } @@ -439,7 +449,7 @@ private Mono sendInternal(Flux events, SendOptions options) { final String partitionId = options.getPartitionId(); if (!CoreUtils.isNullOrEmpty(partitionKey) - && !CoreUtils.isNullOrEmpty(partitionId)) { + && !CoreUtils.isNullOrEmpty(partitionId)) { return monoError(logger, new IllegalArgumentException(String.format(Locale.US, "SendOptions.getPartitionKey() and SendOptions.getPartitionId() are both set. Only one or the" + " other can be used. partitionKey: '%s'. partitionId: '%s'", @@ -447,38 +457,38 @@ private Mono sendInternal(Flux events, SendOptions options) { } return getSendLink(options.getPartitionId()) - .flatMap(link -> link.getLinkSize() - .flatMap(size -> { - final int batchSize = size > 0 ? size : MAX_MESSAGE_LENGTH_BYTES; - final CreateBatchOptions batchOptions = - new CreateBatchOptions().setPartitionKey(options.getPartitionKey()) - .setPartitionId(options.getPartitionId()) - .setMaximumSizeInBytes(batchSize); - return events.collect(new EventDataCollector(batchOptions, 1, - link::getErrorContext, tracerProvider)); - }) - .flatMap(list -> sendInternal(Flux.fromIterable(list)))); + .flatMap(link -> link.getLinkSize() + .flatMap(size -> { + final int batchSize = size > 0 ? size : MAX_MESSAGE_LENGTH_BYTES; + final CreateBatchOptions batchOptions = new CreateBatchOptions() + .setPartitionKey(options.getPartitionKey()) + .setPartitionId(options.getPartitionId()) + .setMaximumSizeInBytes(batchSize); + return events.collect(new EventDataCollector(batchOptions, 1, link::getErrorContext, + tracerProvider)); + }) + .flatMap(list -> sendInternal(Flux.fromIterable(list)))); } private Mono sendInternal(Flux eventBatches) { return eventBatches - .flatMap(this::send) - .then() - .doOnError(error -> { - logger.error(Messages.ERROR_SENDING_BATCH, error); - }); + .flatMap(this::send) + .then() + .doOnError(error -> { + logger.error(Messages.ERROR_SENDING_BATCH, error); + }); } private String getEntityPath(String partitionId) { return CoreUtils.isNullOrEmpty(partitionId) - ? eventHubName - : String.format(Locale.US, SENDER_ENTITY_PATH_FORMAT, eventHubName, partitionId); + ? eventHubName + : String.format(Locale.US, SENDER_ENTITY_PATH_FORMAT, eventHubName, partitionId); } private String getLinkName(String partitionId) { return CoreUtils.isNullOrEmpty(partitionId) - ? StringUtil.getRandomString("EC") - : StringUtil.getRandomString("PS"); + ? StringUtil.getRandomString("EC") + : StringUtil.getRandomString("PS"); } private Mono getSendLink(String partitionId) { @@ -489,7 +499,7 @@ private Mono getSendLink(String partitionId) { return Mono.just(openLink); } else { return connection.createSendLink(getLinkName(partitionId), entityPath, retryOptions) - .map(link -> openLinks.computeIfAbsent(entityPath, unusedKey -> link)); + .map(link -> openLinks.computeIfAbsent(entityPath, unusedKey -> link)); } } @@ -518,7 +528,7 @@ public void close() { * AmqpErrorCondition#LINK_PAYLOAD_SIZE_EXCEEDED}. */ private static class EventDataCollector implements Collector, - List> { + List> { private final String partitionKey; private final String partitionId; private final int maxMessageSize; @@ -529,11 +539,11 @@ private static class EventDataCollector implements Collector 0 - ? options.getMaximumSizeInBytes() - : MAX_MESSAGE_LENGTH_BYTES; + ? options.getMaximumSizeInBytes() + : MAX_MESSAGE_LENGTH_BYTES; this.partitionKey = options.getPartitionKey(); this.partitionId = options.getPartitionId(); this.contextProvider = contextProvider; From 820d8d6b8af4aee8ccfc2ab1c4ca4a3b4ee77156 Mon Sep 17 00:00:00 2001 From: Sameeksha Vaity Date: Sat, 14 Dec 2019 23:13:25 -0800 Subject: [PATCH 7/9] add default to interface --- .../src/main/java/com/azure/core/util/tracing/Tracer.java | 5 ++++- .../azure/core/util/tracing/TracerJavaDocCodeSnippets.java | 3 ++- .../eventhubs/EventHubProducerAsyncClientTest.java | 6 +++--- .../messaging/eventhubs/EventHubProducerClientTest.java | 6 +++--- 4 files changed, 12 insertions(+), 8 deletions(-) diff --git a/sdk/core/azure-core/src/main/java/com/azure/core/util/tracing/Tracer.java b/sdk/core/azure-core/src/main/java/com/azure/core/util/tracing/Tracer.java index c6d052b5dc32..acf30ddcc7cb 100644 --- a/sdk/core/azure-core/src/main/java/com/azure/core/util/tracing/Tracer.java +++ b/sdk/core/azure-core/src/main/java/com/azure/core/util/tracing/Tracer.java @@ -219,5 +219,8 @@ public interface Tracer { * @return The updated {@link Context} object containing the span builder. * @throws NullPointerException if {@code context} or {@code spanName} is {@code null}. */ - Context getSharedSpanBuilder(String spanName, Context context); + default Context getSharedSpanBuilder(String spanName, Context context) { + // no-op + return Context.NONE; + } } diff --git a/sdk/core/azure-core/src/samples/java/com/azure/core/util/tracing/TracerJavaDocCodeSnippets.java b/sdk/core/azure-core/src/samples/java/com/azure/core/util/tracing/TracerJavaDocCodeSnippets.java index 6710fe08ab50..5ae137b2be54 100644 --- a/sdk/core/azure-core/src/samples/java/com/azure/core/util/tracing/TracerJavaDocCodeSnippets.java +++ b/sdk/core/azure-core/src/samples/java/com/azure/core/util/tracing/TracerJavaDocCodeSnippets.java @@ -176,7 +176,8 @@ public Context extractContext(String diagnosticId, Context context) { } @Override - public Context getSharedSpanBuilder(String spanName, Context context) { return null; + public Context getSharedSpanBuilder(String spanName, Context context) { + return null; } } } diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClientTest.java b/sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClientTest.java index 5927054d0ab8..198cb77c8407 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClientTest.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClientTest.java @@ -245,7 +245,7 @@ public void sendStartSpanSingleMessage() { } ); - when(tracer1.getSharedSpanBuilder(eq("Azure.eventhubs.send"), any())).thenAnswer( + when(tracer1.getSharedSpanBuilder(eq("EventHubs.send"), any())).thenAnswer( invocation -> { Context passed = invocation.getArgument(1, Context.class); return passed.addData(SPAN_BUILDER_KEY, "value"); @@ -302,7 +302,7 @@ public void sendMessageRetrySpanTest() { } ); - when(tracer1.getSharedSpanBuilder(eq("Azure.eventhubs.send"), any())).thenAnswer( + when(tracer1.getSharedSpanBuilder(eq("EventHubs.send"), any())).thenAnswer( invocation -> { Context passed = invocation.getArgument(1, Context.class); return passed.addData(SPAN_BUILDER_KEY, "value"); @@ -316,7 +316,7 @@ public void sendMessageRetrySpanTest() { verify(tracer1, times(1)) .start(eq("EventHubs.send"), any(), eq(ProcessKind.SEND)); verify(tracer1, never()).start(eq("EventHubs.message"), any(), eq(ProcessKind.MESSAGE)); - verify(tracer1, never()).addLink(any()); + verify(tracer1, times(2)).addLink(any()); verify(tracer1, times(1)).end(eq("success"), isNull(), any()); } diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/EventHubProducerClientTest.java b/sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/EventHubProducerClientTest.java index 5a3a018f7a54..246753b9273c 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/EventHubProducerClientTest.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/EventHubProducerClientTest.java @@ -169,7 +169,7 @@ public void sendStartSpanSingleMessage() { return passed.addData(PARENT_SPAN_KEY, "value").addData(DIAGNOSTIC_ID_KEY, "value2"); } ); - when(tracer1.getSharedSpanBuilder(eq("Azure.eventhubs.send"), any())).thenAnswer( + when(tracer1.getSharedSpanBuilder(eq("EventHubs.send"), any())).thenAnswer( invocation -> { Context passed = invocation.getArgument(1, Context.class); return passed.addData(SPAN_BUILDER_KEY, "value"); @@ -217,7 +217,7 @@ public void sendMessageRetrySpanTest() { } ); - when(tracer1.getSharedSpanBuilder(eq("Azure.eventhubs.send"), any())).thenAnswer( + when(tracer1.getSharedSpanBuilder(eq("EventHubs.send"), any())).thenAnswer( invocation -> { Context passed = invocation.getArgument(1, Context.class); return passed.addData(SPAN_BUILDER_KEY, "value"); @@ -230,7 +230,7 @@ public void sendMessageRetrySpanTest() { //Assert verify(tracer1, times(1)).start(eq("EventHubs.send"), any(), eq(ProcessKind.SEND)); verify(tracer1, never()).start(eq("EventHubs.message"), any(), eq(ProcessKind.MESSAGE)); - verify(tracer1, never()).addLink(any()); + verify(tracer1, times(1)).addLink(any()); verify(tracer1, times(1)).end(eq("success"), isNull(), any()); } From a2e40119d6596528af7874b52a32aab796791f34 Mon Sep 17 00:00:00 2001 From: Sameeksha Vaity Date: Sun, 15 Dec 2019 12:59:57 -0800 Subject: [PATCH 8/9] review feedback --- .../azure-core-tracing-opencensus/README.md | 15 +++-- .../tracing/opencensus/OpenCensusTracer.java | 3 +- .../src/samples/AsyncListKeyVaultSecrets.md | 2 +- .../AsyncQueueClientEnqueueMessages.md | 2 +- .../src/samples/ListeKeyVaultSecrets.md | 2 +- .../src/samples/PublishEvents.md | 38 ++++++----- .../src/samples/QueueClientEnqueueMessages.md | 2 +- .../CHANGELOG.md | 11 +++- .../README.md | 12 +++- .../src/samples/PublishEventsTracingSample.md | 65 +++++++++++++++---- .../azure/messaging/eventhubs/EventData.java | 2 +- .../EventHubProducerAsyncClient.java | 31 ++++----- 12 files changed, 123 insertions(+), 62 deletions(-) diff --git a/sdk/core/azure-core-tracing-opencensus/README.md b/sdk/core/azure-core-tracing-opencensus/README.md index 8497caa53333..00a7c834c67e 100644 --- a/sdk/core/azure-core-tracing-opencensus/README.md +++ b/sdk/core/azure-core-tracing-opencensus/README.md @@ -19,7 +19,7 @@ documentation][api_documentation] | [Samples][samples] com.azure azure-core-tracing-opencensus - 1.0.0-beta.4 + 1.0.0-beta.5 ``` [//]: # ({x-version-update-end}) @@ -90,10 +90,15 @@ private static final Tracer TRACER; .buildProducerClient(); try (Scope scope = TRACER.spanBuilder("tracing-user-span").startScopedSpan()) { - Context tracingContext = new Context(PARENT_SPAN_KEY, TRACER.getCurrentSpan()); - // Create an event to send - final EventData eventData = new EventData("Hello world!".getBytes(UTF_8), traceContext); - producer.send(eventData); + EventData event1 = new EventData("1".getBytes(UTF_8)); + event1.addContext(PARENT_SPAN_KEY, span); + + EventDataBatch eventDataBatch = producer.createBatch(); + + if (!eventDataBatch.tryAdd(eventData)) { + producer.send(eventDataBatch); + eventDataBatch = producer.createBatch(); + } } finally { Tracing.getExportComponent().shutdown(); } diff --git a/sdk/core/azure-core-tracing-opencensus/src/main/java/com/azure/core/tracing/opencensus/OpenCensusTracer.java b/sdk/core/azure-core-tracing-opencensus/src/main/java/com/azure/core/tracing/opencensus/OpenCensusTracer.java index 6d142b90acda..87e26d840623 100644 --- a/sdk/core/azure-core-tracing-opencensus/src/main/java/com/azure/core/tracing/opencensus/OpenCensusTracer.java +++ b/sdk/core/azure-core-tracing-opencensus/src/main/java/com/azure/core/tracing/opencensus/OpenCensusTracer.java @@ -183,9 +183,8 @@ public Context extractContext(String diagnosticId, Context context) { @Override public Context getSharedSpanBuilder(String spanName, Context context) { - logger.logExceptionAsError( + throw logger.logExceptionAsError( new UnsupportedOperationException("This operation is not supported in OpenCensus")); - return Context.NONE; // Remove OpenCensus support for tracing in https://github.com/Azure/azure-sdk-for-java/issues/6781 } diff --git a/sdk/core/azure-core-tracing-opencensus/src/samples/AsyncListKeyVaultSecrets.md b/sdk/core/azure-core-tracing-opencensus/src/samples/AsyncListKeyVaultSecrets.md index d1714b8b2b57..1e29612b229d 100644 --- a/sdk/core/azure-core-tracing-opencensus/src/samples/AsyncListKeyVaultSecrets.md +++ b/sdk/core/azure-core-tracing-opencensus/src/samples/AsyncListKeyVaultSecrets.md @@ -22,7 +22,7 @@ Sample uses **[opencensus-impl][opencensus_impl]** as implementation package and com.azure azure-core-tracing-opencensus - 1.0.0-beta.4 + 1.0.0-beta.5 ``` [//]: # ({x-version-update-end}) diff --git a/sdk/core/azure-core-tracing-opencensus/src/samples/AsyncQueueClientEnqueueMessages.md b/sdk/core/azure-core-tracing-opencensus/src/samples/AsyncQueueClientEnqueueMessages.md index eb2997a65e13..27e2066b6f39 100644 --- a/sdk/core/azure-core-tracing-opencensus/src/samples/AsyncQueueClientEnqueueMessages.md +++ b/sdk/core/azure-core-tracing-opencensus/src/samples/AsyncQueueClientEnqueueMessages.md @@ -24,7 +24,7 @@ Sample uses **[opencensus-impl][opencensus_impl]** as implementation package and com.azure azure-core-tracing-opencensus - 1.0.0-beta.4 + 1.0.0-beta.5 ``` [//]: # ({x-version-update-end}) diff --git a/sdk/core/azure-core-tracing-opencensus/src/samples/ListeKeyVaultSecrets.md b/sdk/core/azure-core-tracing-opencensus/src/samples/ListeKeyVaultSecrets.md index 43e63e44b96e..4ed8864983f8 100644 --- a/sdk/core/azure-core-tracing-opencensus/src/samples/ListeKeyVaultSecrets.md +++ b/sdk/core/azure-core-tracing-opencensus/src/samples/ListeKeyVaultSecrets.md @@ -23,7 +23,7 @@ Sample uses **[opencensus-impl][opencensus_impl]** as implementation package and com.azure azure-core-tracing-opencensus - 1.0.0-beta.4 + 1.0.0-beta.5 ``` [//]: # ({x-version-update-end}) diff --git a/sdk/core/azure-core-tracing-opencensus/src/samples/PublishEvents.md b/sdk/core/azure-core-tracing-opencensus/src/samples/PublishEvents.md index afb0878bc0d3..29674153f9b8 100644 --- a/sdk/core/azure-core-tracing-opencensus/src/samples/PublishEvents.md +++ b/sdk/core/azure-core-tracing-opencensus/src/samples/PublishEvents.md @@ -14,7 +14,7 @@ Sample uses **[opencensus-impl][opencensus_impl]** as implementation package and com.azure azure-messaging-eventhubs - 5.0.0-beta.5 + 5.0.0-beta.6 ``` [//]: # ({x-version-update-end}) @@ -68,25 +68,31 @@ public class PublishEvents { } public static void main(String[] args) { - final int count = 2; - final byte[] body = "Hello World!".getBytes(UTF_8); + EventHubProducerClient producer = new EventHubClientBuilder() + .connectionString(connectionString) + .buildProducerClient(); try(Scope scope = TRACER.spanBuilder("user-parent-span").startScopedSpan()) { + final EventData event1 = new EventData("1".getBytes(UTF_8)); + event1.addContext(PARENT_SPAN_KEY, TRACER.getCurrentSpan()); + + final EventData event2 = new EventData("2".getBytes(UTF_8)); + event2.addContext(PARENT_SPAN_KEY, TRACER.getCurrentSpan()); - Context traceContext = new Context(PARENT_SPAN_KEY, TRACER.getCurrentSpan()); - - EventHubProducerClient producer = new EventHubClientBuilder() - .connectionString(CONNECTION_STRING) - .buildProducer(); + final List telemetryEvents = Arrays.asList(event1, event2); + final CreateBatchOptions options = new CreateBatchOptions() + .setPartitionKey("telemetry") + .setMaximumSizeInBytes(256); + + EventDataBatch currentBatch = producer.createBatch(options); - final Context traceContext = new Context(PARENT_SPAN_KEY, TRACER.getCurrentSpan()); - final Flux testData = Flux.range(0, count) - .flatMap(number -> { - final EventData data = new EventData(body, traceContext); - return Flux.just(data); - }); - - producer.send(testData.toIterable(1)); + // For each telemetry event, we try to add it to the current batch. + for (EventData event : telemetryEvents) { + if (!currentBatch.tryAdd(event)) { + producer.send(currentBatch); + currentBatch = producer.createBatch(options); + } + } } finally { producer.close(); Tracing.getExportComponent().shutdown(); diff --git a/sdk/core/azure-core-tracing-opencensus/src/samples/QueueClientEnqueueMessages.md b/sdk/core/azure-core-tracing-opencensus/src/samples/QueueClientEnqueueMessages.md index f910c9758d3c..8ce90a2ff790 100644 --- a/sdk/core/azure-core-tracing-opencensus/src/samples/QueueClientEnqueueMessages.md +++ b/sdk/core/azure-core-tracing-opencensus/src/samples/QueueClientEnqueueMessages.md @@ -24,7 +24,7 @@ Sample uses **[opencensus-impl][opencensus_impl]** as implementation package and com.azure azure-core-tracing-opencensus - 1.0.0-beta.4 + 1.0.0-beta.5 ``` [//]: # ({x-version-update-end}) diff --git a/sdk/core/azure-core-tracing-opentelemetry/CHANGELOG.md b/sdk/core/azure-core-tracing-opentelemetry/CHANGELOG.md index 7180efe5049e..46e69e7326cb 100644 --- a/sdk/core/azure-core-tracing-opentelemetry/CHANGELOG.md +++ b/sdk/core/azure-core-tracing-opentelemetry/CHANGELOG.md @@ -1,7 +1,8 @@ # Release History ## Version 1.0.0-beta.2 (2020-01-03) -- Add eventhub properties to attributes of processing spans. +- Add eventHub properties to attributes of processing spans. - Remove `Azure` prefix from convenience layer span names. +- Add links for batch send operation in EventHubs client library. This package's [documentation](https://github.com/Azure/azure-sdk-for-java/blob/azure-core-tracing-opentelemetry_1.0.0-beta.2/sdk/core/azure-core-tracing-opentelemetry/README.md) @@ -11,3 +12,11 @@ and ## Version 1.0.0-beta.1 (2019-11-26) For details on the Azure SDK for Java (Decemeber 2019 Preview) release refer to the [release announcement](https://aka.ms/azure-sdk-preview5-java). +Version 1.0.0-beta.1 is the first preview of our efforts to provide low level interfaces and helper methods to support tracing for Java client libraries. +This library includes [OpenTelemetry](https://opentelemetry.io/) implementation of the interface. +This library added tracing instrumentation for AMQP and HTTP Java SDK client libraries across different languages and platforms. + +This package's +[documentation](https://github.com/Azure/azure-sdk-for-java/blob/azure-core-tracing-opentelemetry_1.0.0-beta.1/sdk/core/azure-core-tracing-opentelemetry/README.md) +and +[samples](https://github.com/Azure/azure-sdk-for-java/blob/azure-core-tracing-opentelemetry_1.0.0-beta.1/sdk/core/azure-core-tracing-opentelemetry/src/samples). diff --git a/sdk/core/azure-core-tracing-opentelemetry/README.md b/sdk/core/azure-core-tracing-opentelemetry/README.md index 8e6c8854b24c..772a8b26ac5a 100644 --- a/sdk/core/azure-core-tracing-opentelemetry/README.md +++ b/sdk/core/azure-core-tracing-opentelemetry/README.md @@ -94,9 +94,15 @@ private static final TracerSdkFactory TRACER_SDK_FACTORY; Span span = TRACER.spanBuilder("user-parent-span").startSpan(); try (Scope scope = TRACER.withSpan(span)) { - Context traceContext = new Context(PARENT_SPAN_KEY, span); - EventData eventData = new EventData("Hello world!".getBytes(UTF_8), traceContext); - producer.send(eventData); + EventData event1 = new EventData("1".getBytes(UTF_8)); + event1.addContext(PARENT_SPAN_KEY, span); + + EventDataBatch eventDataBatch = producer.createBatch(); + + if (!eventDataBatch.tryAdd(eventData)) { + producer.send(eventDataBatch); + eventDataBatch = producer.createBatch(); + } } finally { span.end(); } diff --git a/sdk/core/azure-core-tracing-opentelemetry/src/samples/PublishEventsTracingSample.md b/sdk/core/azure-core-tracing-opentelemetry/src/samples/PublishEventsTracingSample.md index ec3cb269e6fe..55fd97671381 100644 --- a/sdk/core/azure-core-tracing-opentelemetry/src/samples/PublishEventsTracingSample.md +++ b/sdk/core/azure-core-tracing-opentelemetry/src/samples/PublishEventsTracingSample.md @@ -80,24 +80,63 @@ public class Sample { } private static void doClientWork() { - EventHubProducerClient producer = new EventHubClientBuilder() + EventHubProducerAsyncClient producer = new EventHubClientBuilder() .connectionString(CONNECTION_STRING) - .buildProducer(); - - final int count = 2; - final byte[] body = "Hello World!".getBytes(UTF_8); + .buildAsyncProducerClient(); Span span = TRACER.spanBuilder("user-parent-span").startSpan(); try (final Scope scope = TRACER.withSpan(span)) { - final Context traceContext = new Context(PARENT_SPAN_KEY, span); - final Flux testData = Flux.range(0, count).flatMap(number -> { - final EventData data = new EventData(body, traceContext); - return Flux.just(data); - }); - producer.send(testData.toIterable(1)); - } finally { + String firstPartition = producer.getPartitionIds().blockFirst(OPERATION_TIMEOUT); + + final byte[] body = "EventData Sample 1".getBytes(UTF_8); + final byte[] body2 = "EventData Sample 2".getBytes(UTF_8); + + // We will publish three events based on simple sentences. + Flux data = Flux.just( + new EventData(body).addContext(PARENT_SPAN_KEY, TRACER.getCurrentSpan()), + new EventData(body2).addContext(PARENT_SPAN_KEY, TRACER.getCurrentSpan())); + + // Create a batch to send the events. + final CreateBatchOptions options = new CreateBatchOptions() + .setPartitionId(firstPartition) + .setMaximumSizeInBytes(256); + + final AtomicReference currentBatch = new AtomicReference<>( + producer.createBatch(options).block()); + + data.flatMap(event -> { + final EventDataBatch batch = currentBatch.get(); + if (batch.tryAdd(event)) { + return Mono.empty(); + } + + // The batch is full, so we create a new batch and send the batch. Mono.when completes when both operations + // have completed. + return Mono.when( + producer.send(batch), + producer.createBatch(options).map(newBatch -> { + currentBatch.set(newBatch); + + // Add that event that we couldn't before. + if (!newBatch.tryAdd(event)) { + throw Exceptions.propagate(new IllegalArgumentException(String.format( + "Event is too large for an empty batch. Max size: %s. Event: %s", + newBatch.getMaxSizeInBytes(), event.getBodyAsString()))); + } + + return newBatch; + })); + }).then() + .doFinally(signal -> { + final EventDataBatch batch = currentBatch.getAndSet(null); + if (batch != null) { + producer.send(batch).block(OPERATION_TIMEOUT); + } + }) + .subscribe(unused -> System.out.println("Complete"), + error -> System.out.println("Error sending events: " + error), + () -> System.out.println("Completed sending events.")); span.end(); - producer.close(); } } } diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventData.java b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventData.java index 3483c51ae712..6d394d2c414d 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventData.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventData.java @@ -254,7 +254,7 @@ Context getContext() { * @throws NullPointerException if {@code key} or {@code value} is null. * @return The updated {@link EventData}. */ - EventData addContext(String key, Object value) { + public EventData addContext(String key, Object value) { Objects.requireNonNull(key, "The 'key' parameter cannot be null."); Objects.requireNonNull(value, "The 'value' parameter cannot be null."); this.context = context.addData(key, value); diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClient.java b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClient.java index b58d82375964..2c8f37fb2a4a 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClient.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClient.java @@ -42,7 +42,6 @@ import java.util.function.Function; import java.util.function.Supplier; import java.util.stream.Collector; -import java.util.stream.Collectors; import static com.azure.core.util.FluxUtil.monoError; import static com.azure.core.util.tracing.Tracer.ENTITY_PATH_KEY; @@ -397,22 +396,21 @@ public Mono send(EventDataBatch batch) { final AtomicReference parentContext = isTracingEnabled ? new AtomicReference<>(Context.NONE) : null; - final AtomicReference sharedContext = isTracingEnabled - ? new AtomicReference<>(Context.NONE) - : null; - final AtomicReference isFirst = new AtomicReference<>(true); - final List messages = batch.getEvents().stream().map(event -> { - final Message message = messageSerializer.serialize(event); + Context sharedContext = null; + List messages = new ArrayList<>(); + + for (int i = 0; i < batch.getEvents().size(); i++) { + final EventData event = batch.getEvents().get(i); if (isTracingEnabled) { parentContext.set(event.getContext()); - if (isFirst.getAndSet(false)) { - // get shared span builder for all links - sharedContext.set(tracerProvider.getSharedSpanBuilder(parentContext.get())); + if (i == 0) { + sharedContext = tracerProvider.getSharedSpanBuilder(parentContext.get()); } - Context sharedSpanBuilderContext = sharedContext.get(); - tracerProvider.addSpanLinks(sharedSpanBuilderContext.addData(SPAN_CONTEXT_KEY, event.getContext())); + tracerProvider.addSpanLinks(sharedContext.addData(SPAN_CONTEXT_KEY, event.getContext())); } + final Message message = messageSerializer.serialize(event); + if (!CoreUtils.isNullOrEmpty(partitionKey)) { final MessageAnnotations messageAnnotations = message.getMessageAnnotations() == null ? new MessageAnnotations(new HashMap<>()) @@ -420,15 +418,14 @@ public Mono send(EventDataBatch batch) { messageAnnotations.getValue().put(AmqpConstants.PARTITION_KEY, partitionKey); message.setMessageAnnotations(messageAnnotations); } + messages.add(message); + } - return message; - }).collect(Collectors.toList()); - + Context finalSharedContext = sharedContext; return getSendLink(batch.getPartitionId()) .flatMap(link -> { if (isTracingEnabled) { - Context userSpanContext = sharedContext.get(); - Context entityContext = userSpanContext.addData(ENTITY_PATH_KEY, link.getEntityPath()); + Context entityContext = finalSharedContext.addData(ENTITY_PATH_KEY, link.getEntityPath()); // start send span and store updated context parentContext.set(tracerProvider.startSpan( entityContext.addData(HOST_NAME_KEY, link.getHostname()), ProcessKind.SEND)); From a07682cd7d5d6b3838e18bb6a2d4c137c89f5a07 Mon Sep 17 00:00:00 2001 From: Sameeksha Vaity Date: Sun, 15 Dec 2019 20:57:14 -0800 Subject: [PATCH 9/9] update changelog --- sdk/core/azure-core-tracing-opentelemetry/CHANGELOG.md | 4 ++-- .../src/samples/PublishEventsTracingSample.md | 10 +++++++--- 2 files changed, 9 insertions(+), 5 deletions(-) diff --git a/sdk/core/azure-core-tracing-opentelemetry/CHANGELOG.md b/sdk/core/azure-core-tracing-opentelemetry/CHANGELOG.md index 46e69e7326cb..3fd77148407c 100644 --- a/sdk/core/azure-core-tracing-opentelemetry/CHANGELOG.md +++ b/sdk/core/azure-core-tracing-opentelemetry/CHANGELOG.md @@ -1,8 +1,8 @@ # Release History ## Version 1.0.0-beta.2 (2020-01-03) -- Add eventHub properties to attributes of processing spans. +- Add `EventHubs.*` properties to attributes of processing spans. - Remove `Azure` prefix from convenience layer span names. -- Add links for batch send operation in EventHubs client library. +- Add links for batch send operation in Event Hubs client library. This package's [documentation](https://github.com/Azure/azure-sdk-for-java/blob/azure-core-tracing-opentelemetry_1.0.0-beta.2/sdk/core/azure-core-tracing-opentelemetry/README.md) diff --git a/sdk/core/azure-core-tracing-opentelemetry/src/samples/PublishEventsTracingSample.md b/sdk/core/azure-core-tracing-opentelemetry/src/samples/PublishEventsTracingSample.md index 55fd97671381..6256f61fba68 100644 --- a/sdk/core/azure-core-tracing-opentelemetry/src/samples/PublishEventsTracingSample.md +++ b/sdk/core/azure-core-tracing-opentelemetry/src/samples/PublishEventsTracingSample.md @@ -60,6 +60,7 @@ public class Sample { private static final Logger LOGGER = getLogger("Sample"); private static final Tracer TRACER; private static final TracerSdkFactory TRACER_SDK_FACTORY; + private static final Duration OPERATION_TIMEOUT = Duration.ofSeconds(30); static { TRACER_SDK_FACTORY = configureOpenTelemetryAndLoggingExporter(); @@ -102,7 +103,7 @@ public class Sample { .setMaximumSizeInBytes(256); final AtomicReference currentBatch = new AtomicReference<>( - producer.createBatch(options).block()); + producer.createBatch(options).block(OPERATION_TIMEOUT)); data.flatMap(event -> { final EventDataBatch batch = currentBatch.get(); @@ -135,8 +136,11 @@ public class Sample { }) .subscribe(unused -> System.out.println("Complete"), error -> System.out.println("Error sending events: " + error), - () -> System.out.println("Completed sending events.")); - span.end(); + () -> { + span.end(); + producer.close(); + System.out.println("Completed sending events."); + }); } } }