org.quartz.utils.StringKeyDirtyFlagMap#brave.Tracer.SpanInScope源码实例Demo

@SuppressWarnings({"try", "unused"})
@Override
public void handle(Invocation invocation, AsyncResponse asyncResp) throws Exception {
  Span span = tracingDelegate.createSpan(invocation);
  try (SpanInScope scope = tracer.tracer().withSpanInScope(span)) {
    LOGGER.debug("{}: Generated tracing span for {}",
        tracingDelegate.name(),
        invocation.getOperationName());

    invocation.next(onResponse(invocation, asyncResp, span));
  } catch (Exception e) {
    LOGGER.debug("{}: Failed invocation on {}",
        tracingDelegate.name(),
        invocation.getOperationName(),
        e);

    tracingDelegate.onResponse(span, null, e);
    throw e;
  }
}
 

private void handleSendMessageBatchRequest(SendMessageBatchRequest request) {
  TraceContext maybeParent = currentTraceContext.get();

  Span span;
  if (maybeParent == null) {
    span = tracer.nextSpan();
  } else {
    // If we have a span in scope assume headers were cleared before
    span = tracer.newChild(maybeParent);
  }

  span.name("publish-batch").remoteServiceName("amazon-sqs").start();
  try (SpanInScope scope = tracer.withSpanInScope(span)) {
    for (SendMessageBatchRequestEntry entry : request.getEntries()) {
      injectPerMessage(request.getQueueUrl(), entry.getMessageAttributes());
    }
  } finally {
    span.finish();
  }
}
 

protected TraceScopeHolder<TraceScope> startTraceSpan(final Map<String, List<String>> requestHeaders,
        URI uri, String method) {

    final Request request = HttpAdapterFactory.request(requestHeaders, uri, method);
    final HttpClientAdapter<Request, ?> adapter = HttpClientAdapterFactory.create(request);
    
    final HttpClientHandler<Request, ?> handler = HttpClientHandler.create(brave, adapter);
    final Span span = handler.handleSend(
        brave
            .tracing()
            .propagation()
            .injector(inject(requestHeaders)), 
        request);

    // In case of asynchronous client invocation, the span should be detached as JAX-RS
    // client request / response filters are going to be executed in different threads.
    SpanInScope scope = null;
    if (!isAsyncInvocation() && span != null) {
        scope = brave.tracing().tracer().withSpanInScope(span);
    }

    return new TraceScopeHolder<TraceScope>(new TraceScope(span, scope), scope == null /* detached */);
}
 

@Override
public <T> T continueSpan(final Traceable<T> traceable) throws Exception {
    SpanInScope scope = null;
    
    if (tracer.currentSpan() == null && continuationSpan != null) {
        scope = tracer.withSpanInScope(continuationSpan);
    }

    try { //NOPMD
        return traceable.call(new BraveTracerContext(brave));
    } finally {
        if (continuationSpan != null && scope != null) {
            scope.close();
        }
    }
}
 

@Test
public void testThatProvidedSpanIsNotClosedWhenActive() throws MalformedURLException {
    final WebClient client = createWebClient("/bookstore/books", new BraveClientProvider(brave));
    final Span span = brave.tracer().nextSpan().name("test span").start();

    try (SpanInScope scope = brave.tracer().withSpanInScope(span)) {
        final Response r = client.get();
        assertEquals(Status.OK.getStatusCode(), r.getStatus());

        assertThat(TestSpanReporter.getAllSpans().size(), equalTo(3));
        assertThat(TestSpanReporter.getAllSpans().get(0).name(), equalTo("get books"));
        assertThat(TestSpanReporter.getAllSpans().get(0).parentId(), not(nullValue()));
        assertThat(TestSpanReporter.getAllSpans().get(1).name(), equalTo("get /bookstore/books"));
        assertThat(TestSpanReporter.getAllSpans().get(2).name(), equalTo("get " + client.getCurrentURI()));
    } finally {
        span.finish();
    }

    // Await till flush happens, usually a second is enough
    await().atMost(Duration.ofSeconds(1L)).until(()-> TestSpanReporter.getAllSpans().size() == 4);

    assertThat(TestSpanReporter.getAllSpans().size(), equalTo(4));
    assertThat(TestSpanReporter.getAllSpans().get(3).name(), equalTo("test span"));
}
 

@Test
public void testThatProvidedSpanIsNotDetachedWhenActiveUsingAsyncClient() throws Exception {
    final WebClient client = createWebClient("/bookstore/books", new BraveClientProvider(brave));
    final Span span = brave.tracer().nextSpan().name("test span").start();

    try (SpanInScope scope = brave.tracer().withSpanInScope(span)) {
        final Future<Response> f = client.async().get();

        final Response r = f.get(1, TimeUnit.SECONDS);
        assertEquals(Status.OK.getStatusCode(), r.getStatus());
        assertThat(brave.tracer().currentSpan().context().spanId(), equalTo(span.context().spanId()));

        assertThat(TestSpanReporter.getAllSpans().size(), equalTo(3));
        assertThat(TestSpanReporter.getAllSpans().get(0).name(), equalTo("get books"));
        assertThat(TestSpanReporter.getAllSpans().get(1).name(), equalTo("get /bookstore/books"));
        assertThat(TestSpanReporter.getAllSpans().get(2).name(), equalTo("get " + client.getCurrentURI()));
    } finally {
        span.finish();
    }

    // Await till flush happens, usually a second is enough
    await().atMost(Duration.ofSeconds(1L)).until(()-> TestSpanReporter.getAllSpans().size() == 4);

    assertThat(TestSpanReporter.getAllSpans().size(), equalTo(4));
    assertThat(TestSpanReporter.getAllSpans().get(3).name(), equalTo("test span"));
}
 

@Test public void withSpanInScope_nested() {
  Span parent = tracer.newTrace();

  try (SpanInScope wsParent = tracer.withSpanInScope(parent)) {

    Span child = tracer.newChild(parent.context());
    try (SpanInScope wsChild = tracer.withSpanInScope(child)) {
      assertThat(tracer.currentSpan())
        .isEqualTo(child);
    }

    // old parent reverted
    assertThat(tracer.currentSpan())
      .isEqualTo(parent);
  }
}
 

@Test public void withSpanInScope_clear() {
  Span parent = tracer.newTrace();

  try (SpanInScope wsParent = tracer.withSpanInScope(parent)) {
    try (SpanInScope clearScope = tracer.withSpanInScope(null)) {
      assertThat(tracer.currentSpan())
        .isNull();
      assertThat(tracer.currentSpanCustomizer())
        .isEqualTo(NoopSpanCustomizer.INSTANCE);
    }

    // old parent reverted
    assertThat(tracer.currentSpan())
      .isEqualTo(parent);
  }
}
 

@Override public CloseableHttpResponse execute(HttpRoute route,
  org.apache.http.client.methods.HttpRequestWrapper req,
  HttpClientContext context, HttpExecutionAware execAware)
  throws IOException, HttpException {
  HttpRequestWrapper request = new HttpRequestWrapper(req, context.getTargetHost());
  Span span = tracer.nextSpan(httpSampler, request);
  context.setAttribute(Span.class.getName(), span);

  CloseableHttpResponse response = null;
  Throwable error = null;
  try (SpanInScope ws = tracer.withSpanInScope(span)) {
    return response = protocolExec.execute(route, req, context, execAware);
  } catch (Throwable e) {
    error = e;
    throw e;
  } finally {
    handler.handleReceive(new HttpResponseWrapper(response, context, error), span);
  }
}
 

@Override public void send(Message message) throws JMSException {
  Span span = createAndStartProducerSpan(message, destination(message));
  SpanInScope ws = tracer.withSpanInScope(span);
  Throwable error = null;
  try {
    delegate.send(message);
  } catch (Throwable t) {
    propagateIfFatal(t);
    error = t;
    throw t;
  } finally {
    if (error != null) span.error(error);
    span.finish();
    ws.close();
  }
}
 

@Override public void send(Message message, int deliveryMode, int priority, long timeToLive)
  throws JMSException {
  Span span = createAndStartProducerSpan(message, destination(message));
  SpanInScope ws = tracer.withSpanInScope(span); // animal-sniffer mistakes this for AutoCloseable
  Throwable error = null;
  try {
    delegate.send(message, deliveryMode, priority, timeToLive);
  } catch (Throwable t) {
    propagateIfFatal(t);
    error = t;
    throw t;
  } finally {
    if (error != null) span.error(error);
    span.finish();
    ws.close();
  }
}
 

void send(SendDestination sendDestination, Destination destination, Message message)
  throws JMSException {
  Span span = createAndStartProducerSpan(message, destination);
  SpanInScope ws = tracer.withSpanInScope(span); // animal-sniffer mistakes this for AutoCloseable
  Throwable error = null;
  try {
    sendDestination.apply(delegate, destination, message);
  } catch (Throwable t) {
    propagateIfFatal(t);
    error = t;
    throw t;
  } finally {
    if (error != null) span.error(error);
    span.finish();
    ws.close();
  }
}
 

@Override
public void send(Destination destination, Message message, int deliveryMode, int priority,
  long timeToLive) throws JMSException {
  Span span = createAndStartProducerSpan(message, destination);
  SpanInScope ws = tracer.withSpanInScope(span);
  Throwable error = null;
  try {
    delegate.send(destination, message, deliveryMode, priority, timeToLive);
  } catch (Throwable t) {
    propagateIfFatal(t);
    error = t;
    throw t;
  } finally {
    if (error != null) span.error(error);
    span.finish();
    ws.close();
  }
}
 

@JMS2_0
public void send(Message message, CompletionListener completionListener) throws JMSException {
  Destination destination = destination(message);
  Span span = createAndStartProducerSpan(message, destination);
  SpanInScope ws = tracer.withSpanInScope(span); // animal-sniffer mistakes this for AutoCloseable
  Throwable error = null;
  try {
    delegate.send(message, TracingCompletionListener.create(completionListener, destination, span, current));
  } catch (Throwable t) {
    propagateIfFatal(t);
    error = t;
    throw t;
  } finally {
    if (error != null) span.error(error).finish();
    ws.close();
  }
}
 

@JMS2_0 public void send(Message message, int deliveryMode, int priority, long timeToLive,
  CompletionListener completionListener) throws JMSException {
  Destination destination = destination(message);
  Span span = createAndStartProducerSpan(message, destination);
  completionListener = TracingCompletionListener.create(completionListener, destination, span, current);
  SpanInScope ws = tracer.withSpanInScope(span); // animal-sniffer mistakes this for AutoCloseable
  Throwable error = null;
  try {
    delegate.send(message, deliveryMode, priority, timeToLive, completionListener);
  } catch (Throwable t) {
    propagateIfFatal(t);
    error = t;
    throw t;
  } finally {
    if (error != null) span.error(error).finish();
    ws.close();
  }
}
 

@JMS2_0 public void send(Destination destination, Message message,
  CompletionListener completionListener) throws JMSException {
  Span span = createAndStartProducerSpan(message, destination);
  completionListener = TracingCompletionListener.create(completionListener, destination, span, current);
  SpanInScope ws = tracer.withSpanInScope(span);
  Throwable error = null;
  try {
    delegate.send(destination, message, completionListener);
  } catch (Throwable t) {
    propagateIfFatal(t);
    error = t;
    throw t;
  } finally {
    if (error != null) span.error(error).finish();
    ws.close();
  }
}
 

@JMS2_0 public void send(Destination destination, Message message, int deliveryMode, int priority,
  long timeToLive, CompletionListener completionListener) throws JMSException {
  Span span = createAndStartProducerSpan(message, destination);
  completionListener = TracingCompletionListener.create(completionListener, destination, span, current);
  SpanInScope ws = tracer.withSpanInScope(span);
  Throwable error = null;
  try {
    delegate.send(destination, message, deliveryMode, priority, timeToLive, completionListener);
  } catch (Throwable t) {
    propagateIfFatal(t);
    error = t;
    throw t;
  } finally {
    if (error != null) span.error(error).finish();
    ws.close();
  }
}
 

@Override
public void send(Queue queue, Message message, int deliveryMode, int priority, long timeToLive)
  throws JMSException {
  checkQueueSender();
  QueueSender qs = (QueueSender) delegate;
  Span span = createAndStartProducerSpan(message, destination(message));
  SpanInScope ws = tracer.withSpanInScope(span);
  Throwable error = null;
  try {
    qs.send(queue, message, deliveryMode, priority, timeToLive);
  } catch (Throwable t) {
    propagateIfFatal(t);
    error = t;
    throw t;
  } finally {
    if (error != null) span.error(error);
    span.finish();
    ws.close();
  }
}
 

@Override public void publish(Message message) throws JMSException {
  checkTopicPublisher();
  TopicPublisher tp = (TopicPublisher) delegate;

  Span span = createAndStartProducerSpan(message, destination(message));
  SpanInScope ws = tracer.withSpanInScope(span);
  Throwable error = null;
  try {
    tp.publish(message);
  } catch (Throwable t) {
    propagateIfFatal(t);
    error = t;
    throw t;
  } finally {
    if (error != null) span.error(error);
    span.finish();
    ws.close();
  }
}
 

@Override public void publish(Message message, int deliveryMode, int priority, long timeToLive)
  throws JMSException {
  checkTopicPublisher();
  TopicPublisher tp = (TopicPublisher) delegate;

  Span span = createAndStartProducerSpan(message, destination(message));
  SpanInScope ws = tracer.withSpanInScope(span);
  Throwable error = null;
  try {
    tp.publish(message, deliveryMode, priority, timeToLive);
  } catch (Throwable t) {
    propagateIfFatal(t);
    error = t;
    throw t;
  } finally {
    if (error != null) span.error(error);
    span.finish();
    ws.close();
  }
}
 

@Override
public void publish(Topic topic, Message message, int deliveryMode, int priority, long timeToLive)
  throws JMSException {
  checkTopicPublisher();
  TopicPublisher tp = (TopicPublisher) delegate;

  Span span = createAndStartProducerSpan(message, destination(message));
  SpanInScope ws = tracer.withSpanInScope(span);
  Throwable error = null;
  try {
    tp.publish(topic, message, deliveryMode, priority, timeToLive);
  } catch (Throwable t) {
    propagateIfFatal(t);
    error = t;
    throw t;
  } finally {
    if (error != null) span.error(error);
    span.finish();
    ws.close();
  }
}
 

@Override public void onMessage(Message message) {
  Span listenerSpan = startMessageListenerSpan(message);
  SpanInScope ws = tracer.withSpanInScope(listenerSpan);
  Throwable error = null;
  try {
    delegate.onMessage(message);
  } catch (Throwable t) {
    propagateIfFatal(t);
    error = t;
    throw t;
  } finally {
    if (error != null) listenerSpan.error(error);
    listenerSpan.finish();
    ws.close();
  }
}
 

@SuppressWarnings({"try", "unused"})
@Override
public CompletableFuture<Response> onFilter(Invocation invocation, FilterNode nextNode) {
  ZipkinTracingDelegate tracing = collectTracing(invocation);

  Span span = tracing.createSpan(invocation);
  try (SpanInScope scope = tracing.tracer().tracer().withSpanInScope(span)) {
    return nextNode.onFilter(invocation)
        .whenComplete((response, exception) -> tracing.onResponse(span, response, Exceptions.unwrap(exception)));
  }
}
 

@SuppressWarnings({"unused", "try"})
<T> T invoke(String spanName, String path, ThrowableSupplier<T> supplier) throws Throwable {
  Span span = createSpan(spanName, path);
  try (SpanInScope spanInScope = tracer.withSpanInScope(span)) {
    return supplier.get();
  } catch (Throwable throwable) {
    span.tag("error", throwable.getClass().getSimpleName() + ": " + throwable.getMessage());
    throw throwable;
  } finally {
    span.finish();
  }
}
 

@SuppressWarnings({"unused", "try"})
@Test
public void startsNewChildSpan() {
  CyclicBarrier cyclicBarrier = new CyclicBarrier(nThreads);

  CompletableFuture<?>[] futures = (CompletableFuture<?>[]) Array.newInstance(CompletableFuture.class, nThreads);
  for (int i = 0; i < nThreads; i++) {
    futures[i] = CompletableFuture.runAsync(() -> {
      Span currentSpan = tracing.tracer().newTrace().start();

      waitTillAllAreReady(cyclicBarrier);

      try (SpanInScope spanInScope = tracing.tracer().withSpanInScope(currentSpan)) {
        assertThat(tracingAdviser.invoke(spanName, path, supplier), is(expected));
      } catch (Throwable throwable) {
        fail(throwable.getMessage());
      } finally {
        currentSpan.finish();
      }
    }, Executors.newFixedThreadPool(nThreads));
  }

  CompletableFuture.allOf(futures).join();

  assertThat(traces.size(), is(nThreads));

  for (Queue<zipkin2.Span> queue : traces.values()) {
    zipkin2.Span child = queue.poll();
    assertThat(child.name(), is(spanName));

    zipkin2.Span parent = queue.poll();
    assertThat(child.parentId(), is(parent.id()));
    assertThat(child.traceId(), is(parent.traceId()));
    assertThat(tracedValues(child), contains(this.getClass().getCanonicalName()));
  }
}
 

/**
 * @param delegate a SpanInScope to be closed upon deactivation of this ActiveSpan
 * @param source the BraveActiveSpanSource that created this BraveActiveSpan
 * @param wrapped the wrapped BraveSpan to which we will delegate all span operations
 */
v0_32_BraveScope(SpanInScope delegate, v0_32_BraveScopeManager source, BraveSpan wrapped,
    boolean finishSpanOnClose) {
  super(delegate);
  this.source = source;
  this.wrapped = wrapped;
  this.finishSpanOnClose = finishSpanOnClose;
}
 

private void closeTrace(JobExecutionContext context) {
	Object spanInScope = context.get(CONTEXT_SPAN_IN_SCOPE_KEY);
	Object span = context.get(CONTEXT_SPAN_KEY);
	if (spanInScope instanceof SpanInScope) {
		((SpanInScope) spanInScope).close();
	}
	if (span instanceof Span) {
		((Span) span).finish();
	}
}
 

void addSpanToJobData(JobDataMap data) {
	brave.Span span = tracing.tracer().nextSpan().start();
	try (SpanInScope spanInScope = tracing.tracer().withSpanInScope(span)) {
		tracing.propagation()
				.injector((Setter<JobDataMap, String>) StringKeyDirtyFlagMap::put)
				.inject(tracing.currentTraceContext().get(), data);
	}
	finally {
		span.finish();
	}
}
 

protected TraceScopeHolder<TraceScope> startTraceSpan(final Map<String, List<String>> requestHeaders,
        URI uri, String method) {

    final Request request = HttpAdapterFactory.request(requestHeaders, uri, method);
    final HttpServerAdapter<Request, ?> adapter = HttpServerAdapterFactory.create(request);
    
    final HttpServerHandler<Request, ?> handler = HttpServerHandler.create(brave, adapter);
    
    Span span = handler.handleReceive(
        brave
            .tracing()
            .propagation()
            .extractor(adapter::requestHeader), 
        request);
    
    // If the service resource is using asynchronous processing mode, the trace
    // scope will be closed in another thread and as such should be detached.
    SpanInScope scope = null;
    if (isAsyncResponse() && span != null) {
       // Do not modify the current context span
        propagateContinuationSpan(span);
    } else if (span != null) {
        scope = brave.tracing().tracer().withSpanInScope(span);
    }

    return new TraceScopeHolder<TraceScope>(new TraceScope(span, scope), scope == null /* detached */);
}
 

@Test
public void testThatProvidedSpanIsNotClosedWhenActive() throws Exception {
    try (Tracing brave = createTracer()) {
        final BookStoreService service = createJaxWsService(new BraveClientFeature(brave));

        final Span span = brave.tracer().nextSpan().name("test span").start();
        try {
            try (SpanInScope scope = brave.tracer().withSpanInScope(span)) {
                assertThat(service.getBooks().size(), equalTo(2));
                assertThat(brave.tracer().currentSpan(), not(nullValue()));

                assertThat(TestSpanReporter.getAllSpans().size(), equalTo(3));
                assertThat(TestSpanReporter.getAllSpans().get(0).name(), equalTo("get books"));
                assertThat(TestSpanReporter.getAllSpans().get(0).parentId(), not(nullValue()));
                assertThat(TestSpanReporter.getAllSpans().get(1).name(), equalTo("post /bookstore"));
                assertThat(TestSpanReporter.getAllSpans().get(2).name(),
                    equalTo("post http://localhost:" + PORT + "/bookstore"));
            }
        } finally {
            if (span != null) {
                span.finish();
            }
        }

        assertThat(TestSpanReporter.getAllSpans().size(), equalTo(4));
        assertThat(TestSpanReporter.getAllSpans().get(3).name(), equalTo("test span"));
    }
}