Opentelemetry-Trace流程源码解析

opentelemetry Trace流程源码解析：

trace流程当然是从我们使用sdk开始。

SdkTracerProvider sdkTracerProvider = openTelemetrySdk.getSdkTracerProvider();
Span startSpan = sdkTracerProvider.get("orderService")
        .spanBuilder("pay")
        .startSpan();
startSpan.end();

从最简单的用法开始，从openTelemetrySdk获取traceProvider，构造span，startSpan再到end。

这就是我们trace中span最简单的流程。

在spanBuilder中只接受了一个spanName，其中sharedState指得是trace中所有span共享的一些状态和配置，比如id生成器，span处理器，span的shutdown状态等。

在最后的构造中其实只构造了一个名字，并没有做过多的事情，毕竟是builder。

下面直接看startSpan方法做了什么。

1.获取当前线程的context，从context中获取parentSpan，生成一个spanId。当startSpan为整个trace第一次调用时，parent是空。

ContextStorage是一个接口，实现也很有趣，一个是final类，一个是枚举。

我们看到StrictContextStorage实现了Storage，不过却用的delegate做模板方法。

例如current使用的delegate方法，在实际代码中,delegate其实就是ThreadLocalContextStorage。

ThreadLocalContextStorage是一个枚举类，有一个枚举叫做INSTANCE。

而current方法就是通过threadlocal存储并获取的。

而threadlocal在目前是没有值的。

所以最开始context就是个null。

我们继续往下，在context null时，会返回一个无效的span：PropagatedSpan.INVALID。

通过这个标记，我们可以得知是root span还是child span，从而生成或者获取traceId。

links是我们在spanbuiler使用addLink方法添加的,放入一个spancontext和一些attributes。

第二段：

从sharedState获取sampler采样器，默认是alwaysOn类型，当然也提供了其他策略，所以这里才使用shouldSample方法放入需要的参数，contxt，linkdatas等判断是否需要采样。

然后创建一个spanContext,spanContext其实也是存放span所需信息。

后续就是保存获取attribute，attribute具体放了什么，是否使用都是由shouldSample方法决定的。

startSpan方法中记录了一个span开始时间：

创建SdkSpan并调用spanProcessor中的onStart方法。

Processor有四种：

不过onStart方法基本空实现

此时我们已经拿到了span，总结一下startSpan做了什么：

从当前线程Context中获取parentContext，parentSpan和parentSpanContext，其中核心逻辑是是否采样的逻辑判断，其中一个SpanContext对应着一个span，会存放对应span的id和traceId。

接下来看看end方法，end方法直接调用internal。

可知span并不是线程安全，存在同时end的情况。

大家使用时注意，后end的线程将无法记录span信息到trace中。

记录时间，再次调用spanProcessor处理。

首先看batch

batch使用了queue存放span，存放失败时会使用counter类型的指标（如果不清楚，可以去看看prometheus四大指标类型）记录失败个数。

存放成功后，如果size大于等于spansNeeded时会发送一个信号。

这里的spansNeeded是一个AtomicInteger，我们知道这是BatchSpanprocessor批量发送的，那必然是收集到一个batch，一起发送的过程，所以spansNeeded其实就代表下一次批量发送还需要多少个span。那么这个信号其实就是通知发送线程开始发送。

下图是发送span的worker线程run方法：

着重讲下worker成员变量：

queue用于存放当前添加的还未export的span。

spansNeeded表示当前还需要多少个span才能export。

signal用于通知worker线程在poll阻塞中苏醒。

flushRequested用于通知外部强制flush是否完成。

batch存在下一次export的span。

worker线程逻辑：

public void run() {
  updateNextExportTime();

  while (continueWork) {
  // 先判断是否有flush请求，有就flush，以避免span积压太多，后续的span迟迟无法上报。
    if (flushRequested.get() != null) {
      flush();
    }
    // 从queue中出队maxExportBatchSize-batch.size还能export的span
    JcTools.drain(
        queue, maxExportBatchSize - batch.size(), span -> batch.add(span.toSpanData()));
	 // 当batch达到可导出数量时，导出
    if (batch.size() >= maxExportBatchSize || System.nanoTime() >= nextExportTime) {
      exportCurrentBatch();
      updateNextExportTime();
    }
    // 不管当前batch有没有导出，当queue为空，为避免cpu空转，让线程等待，等待queue中又有足够的span数量时能够一次性batch时，再唤醒线程
    if (queue.isEmpty()) {
      try {
        long pollWaitTime = nextExportTime - System.nanoTime();
        if (pollWaitTime > 0) {
          spansNeeded.set(maxExportBatchSize - batch.size());
          signal.poll(pollWaitTime, TimeUnit.NANOSECONDS);
          // 唤醒后设置为最大值,为什么这样？
          // 从之前添加span的代码可以得知,当queue超过spansNeeded时就会通知线程苏醒，实际环境是addSpan是一个非常频繁的动作，所以在苏醒后，queue并没有那么快就出队，就会导致下一个add时又去通知线程苏醒做不必要的操作。所以不如在此时把spansNeeded设置为最大值,减少没有必要的操作。
          spansNeeded.set(Integer.MAX_VALUE);
        }
      } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
        return;
      }
    }
  }
}

if (queue.size() >= spansNeeded.get()) {
  signal.offer(true);
}

剩下exportCurrentBatch的逻辑比较简单。

opentelemetry默认提供了三个exporter。

loggingSpanExporter就是从日志中把本次span的信息打印出来。

MultiSpanExporter就是多个spanExporter分别处理这些span。

NoopSpanExporter是空实现。

还剩几个问题大家可以自己探索源码：

1.trace中span层级怎么确定的？

2.queue没有上限吗？

3.如果span数量一致没有到need个数，是不是有些span永远无法上报呢？

• 本文作者： 宏
• 本文链接： http://sasuke.top/2024/07/15/Opentelemetry-Trace流程源码解析/
• 版权声明： 本博客所有文章除特别声明外，均采用 MIT 许可协议。转载请注明出处！