Wrapping call in observe(...) and nulling sender

Due to the fact that in the tests we have both the producer and the consumer and they have the same parent, the zipkin graph looks bizarre.

With these changes we're changing the test sending code to simulate sending a message from a different service that will result in a creation of a new trace identifier. Due to this we will have 2 sets of trace ids created, one for sending and one for polling.

Sending: null_observation -> send -> receive message
Polling: test_span -> receive (very short)

Author: marcingrzejszczak

src/main/java/com/rabbitmq/client/observation/micrometer/DefaultDeliverObservationConvention.java

@@ -35,6 +35,7 @@ public DefaultDeliverObservationConvention(String name, String operation) {
     this.operation = operation;
   }
 
+  // TODO: If the name is not fixed we won't be able to parse it to automatically document the name
   @Override
   public String getName() {
     return name;

src/main/java/com/rabbitmq/client/observation/micrometer/MicrometerObservationCollector.java

@@ -102,8 +102,7 @@ public Consumer basicConsume(String queue, String consumerTag, Consumer consumer
 
   @Override
   public GetResponse basicGet(BasicGetCall call, String queue) {
-    Observation parentObservation = Observation.start("rabbitmq.receive", registry);
-    try {
+    return Observation.createNotStarted("rabbitmq.receive", registry).observe(() -> {
       GetResponse response = call.get();
       if (response != null) {
         Map<String, Object> headers;
@@ -113,26 +112,20 @@ public GetResponse basicGet(BasicGetCall call, String queue) {
           headers = response.getProps().getHeaders();
         }
         DeliverContext context =
-            new DeliverContext(
-                response.getEnvelope().getExchange(),
-                response.getEnvelope().getRoutingKey(),
-                queue,
-                headers,
-                response.getBody() == null ? 0 : response.getBody().length);
+                new DeliverContext(
+                        response.getEnvelope().getExchange(),
+                        response.getEnvelope().getRoutingKey(),
+                        queue,
+                        headers,
+                        response.getBody() == null ? 0 : response.getBody().length);
         Observation observation =
-            RabbitMqObservationDocumentation.RECEIVE_OBSERVATION.observation(
-                customReceiveConvention, defaultReceiveConvention, () -> context, registry);
-        observation.parentObservation(parentObservation);
+                RabbitMqObservationDocumentation.RECEIVE_OBSERVATION.observation(
+                        customReceiveConvention, defaultReceiveConvention, () -> context, registry);
         observation.start();
         observation.stop();
       }
       return response;
-    } catch (RuntimeException e) {
-      parentObservation.error(e);
-      throw e;
-    } finally {
-      parentObservation.stop();
-    }
+    });
   }
 
   private static class ObservationConsumer implements Consumer {

src/test/java/com/rabbitmq/client/test/functional/MicrometerObservationCollectorMetrics.java

@@ -22,16 +22,25 @@
 import com.rabbitmq.client.observation.micrometer.MicrometerObservationCollectorBuilder;
 import com.rabbitmq.client.test.BrokerTestCase;
 import com.rabbitmq.client.test.TestUtils;
+import io.micrometer.observation.Observation;
 import io.micrometer.observation.ObservationRegistry;
+import io.micrometer.tracing.Tracer;
+import io.micrometer.tracing.exporter.FinishedSpan;
 import io.micrometer.tracing.test.SampleTestRunner;
 import io.micrometer.tracing.test.simple.SpanAssert;
 import io.micrometer.tracing.test.simple.SpansAssert;
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
+import java.util.Collection;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Map;
 import java.util.Objects;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
+
+import org.assertj.core.api.BDDAssertions;
 import org.junit.jupiter.api.Nested;
 
 public class MicrometerObservationCollectorMetrics extends BrokerTestCase {
@@ -111,6 +120,17 @@ private abstract static class IntegrationTest extends SampleTestRunner {
     public TracingSetup[] getTracingSetup() {
       return new TracingSetup[] {TracingSetup.IN_MEMORY_BRAVE, TracingSetup.ZIPKIN_BRAVE};
     }
+
+    void runWithNullingObservation(ObservationRegistry registry, Tracer tracer, Observation.CheckedRunnable<?> runnable) {
+      Observation noParentObservation = Observation.createNotStarted("null_observation", registry);
+      noParentObservation.parentObservation(null);
+      try (Tracer.SpanInScope ws = tracer.withSpan(null)) {
+        noParentObservation.observeChecked(runnable);
+      }
+      catch (Throwable e) {
+        throw new RuntimeException(e);
+      }
+    }
   }
 
   @Nested
@@ -186,7 +206,7 @@ public SampleTestRunnerConsumer yourCode() {
           publishConnection = connectionFactory.newConnection();
           Channel channel = publishConnection.createChannel();
 
-          sendMessage(channel);
+          runWithNullingObservation(getObservationRegistry(), buildingBlocks.getTracer(), () -> sendMessage(channel));
 
           consumeConnection = connectionFactory.newConnection();
           Channel basicGetChannel = consumeConnection.createChannel();
@@ -196,8 +216,14 @@ public SampleTestRunnerConsumer yourCode() {
               buildingBlocks.getFinishedSpans().stream()
                   .map(Objects::toString)
                   .collect(Collectors.joining("\n")));
-          SpansAssert.assertThat(buildingBlocks.getFinishedSpans()).haveSameTraceId();
-          SpanAssert.assertThat(buildingBlocks.getFinishedSpans().get(0))
+          Map<String, List<FinishedSpan>> finishedSpans = buildingBlocks.getFinishedSpans().stream()
+                  .collect(Collectors.groupingBy(FinishedSpan::getTraceId));
+          BDDAssertions.then(finishedSpans).as("One trace id for sending, one for polling").hasSize(2);
+          Collection<List<FinishedSpan>> spans = finishedSpans.values();
+          List<FinishedSpan> sendAndReceiveSpans = spans.stream().filter(f -> f.size() == 3).findFirst().orElseThrow(() -> new AssertionError("null_observation (fake nulling observation) -> produce -> consume"));
+          sendAndReceiveSpans.sort(Comparator.comparing(FinishedSpan::getStartTimestamp));
+          SpanAssert.assertThat(sendAndReceiveSpans.get(0)).hasNameEqualTo("null_observation");
+          SpanAssert.assertThat(sendAndReceiveSpans.get(1))
               .hasNameEqualTo("metrics.queue publish")
               .hasTag("messaging.rabbitmq.destination.routing_key", "metrics.queue")
               .hasTag("messaging.destination.name", "amq.default")
@@ -206,12 +232,15 @@ public SampleTestRunnerConsumer yourCode() {
               .hasTag("net.sock.peer.port", "5672")
               .hasTag("net.protocol.name", "amqp")
               .hasTag("net.protocol.version", "0.9.1");
-          SpanAssert.assertThat(buildingBlocks.getFinishedSpans().get(1))
+          SpanAssert.assertThat(sendAndReceiveSpans.get(2))
               .hasNameEqualTo("metrics.queue receive")
               .hasTag("messaging.rabbitmq.destination.routing_key", "metrics.queue")
               .hasTag("messaging.destination.name", "amq.default")
               .hasTag("messaging.source.name", "metrics.queue")
               .hasTag("messaging.message.payload_size_bytes", String.valueOf(PAYLOAD.length));
+          List<FinishedSpan> pollingSpans = spans.stream().filter(f -> f.size() == 1).findFirst().orElseThrow(() -> new AssertionError("rabbitmq.receive (child of test span)"));
+          SpanAssert.assertThat(pollingSpans.get(0))
+                  .hasNameEqualTo("rabbitmq.receive");
           waitAtMost(
               () ->
                   getMeterRegistry().find("rabbitmq.publish").timer() != null