Spring Boot中OpenTelemetry Span向Executor线程自动传播可行性问询

问题背景 #

我正在配置OpenTelemetry Tracer，希望将Span传播到ThreadPoolTaskExecutor创建的线程中，当前Executor的配置代码如下：

private DgsDataLoaderAsyncExecutor() {
  ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
  executor.setCorePoolSize(1);
  executor.setMaxPoolSize(100);
  executor.setQueueCapacity(500);
  executor.setThreadNamePrefix("threadAsync");
  executor.setWaitForTasksToCompleteOnShutdown(true);
  executor.setTaskDecorator(new DataLoaderTaskDecorator());
  executor.initialize();
  executorInstance = executor;
  // spring tracer is obtained here to extract the current span later on 
  tracer = GlobalOpenTelemetry.getTracer("org.springframework.boot");
}

Executor调用方式 #

后续通过以下方式获取并调用该Executor：

/**
 * Only one (or no) Executor will live in the service at all times. It will be retrieved in a thread safe manner
 * @return executor to be used in the DataLoader
 */
public static synchronized Executor get() {
  if (instance == null) {
    instance = new DgsDataLoaderAsyncExecutor();
  }
  return instance.getExecutorInstance();
}

static class DataLoaderTaskDecorator implements TaskDecorator {
  /**
   * Currently, tenant, security, and locale contexts are passed
   */
  @Override
  public Runnable decorate(Runnable runnable) {
    // Current Web Thread context to be passed to async
    var currentSpanContext = Context.current();
    var currentSpan = Span.current();
    Locale locale = LocaleContextHolder.getLocale();

    log.debug("Saving information for async thread...");
    return () -> {
      // this span shows up in the parent trace as expected 
      Span asyncSpan = tracer.spanBuilder("DGS Executor async op")
        .setParent(currentSpanContext.with(currentSpan))
        .startSpan();
      try {
        // Async thread context
        LocaleContextHolder.setLocale(locale);
        log.debug("Restoring information for async thread...");
        // however the spans in the thread fail to obtain the parent trace and are not linked to it
        runnable.run();
      } catch (Exception e) {
        log.error("Error in async task", e);
      } finally {
        asyncSpan.end();
        log.debug("DgsDataLoader has finished async execution");
      }
    };
  }
}

当前问题 #

目前的问题是：Span无法传播到目标线程，线程内生成的Span会创建新的父追踪，无法关联到原有的链路中。

已尝试的解决方案 #

我曾通过自定义上下文封装当前Span来临时解决该问题，修改后的代码如下：

Span asyncSpan = tracer.spanBuilder("DGS Executor async op")
  .setParent(currentSpanContext.with(currentSpan))
  .startSpan();
try {
  SpanThreadContextHolder.setTracingSpan(asyncSpan);
  // Async thread context
  LocaleContextHolder.setLocale(locale);
  log.debug("Restoring information for async thread...");
  runnable.run();
} catch (Exception e) {
  log.error("Error in async task", e);
} finally {
  SpanThreadContextHolder.removeTracingSpan();

但这种方式存在副作用，下游代码必须手动激活Span才能关联链路：

// inside runnable.run()
var span = SpanThreadContextHolder.getOpenTracingSpan()
if (span!= null) {
    Context.current().with(span).makeCurrent()
}

我的问题 #

能否将Span自动传播至主线程生成的其他线程，让Span以THREAD_LOCAL模式存在，无需下游代码手动激活？

内容的提问来源于stack exchange，提问作者Simas Joneliunas