Kamon 采样原理

冷宏茂

2023-12-01

采样方式

通过kamon.trace.sampler可以配置一个采样器，这可以决定哪些 span 应该被发送到 span 报告器。采样器类型有几个可能的值，分别是：

always：报告所有 trace，会忽略配置的其他规则。ConstantSampler类
never：不报告任何 trace，会忽略配置的其他规则。ConstantSampler类
random：随机采样器，由于概率设置中定义的概率决定。RandomSampler类，默认概率为 1%
adaptive：自适应采样器，为每个操作保留动态采样器，同时试图实现设定的吞吐量目标。AdaptiveSampler类
自己实现一个：填需要写一个继承了kamon.trace.Sampler接口的类的全类名，如果类加载失败则回退到概率为 10% 的 random 采样器

虽然上面有多种采样器，但其实 kamon 中的采样器只有三类：

恒定采样器 – always 和 never 都属于恒定
随机采样器 – 原理是使用ThreadLocalRandom.current().nextLong()产生随机数，如果随机数是落在一个区间内（下界_lowerBoundary，上界_upperBoundary），则需要采样，否则不采样。
自适应采样器 – 结合了用户设定的吞吐量目标和随机数，会每秒一次自动调整采样率。

判断操作是否应该被采样，下文均使用 判定采样 一词，判定成功，采样。判定失败，不采样。

不采样的决定仍然会产生可以收集指标并与上下文一起传播的 span，它们只是不会被发送到span 报告者那里。

配置采样规则

配置自适应采样器

通常，如果需要的是抽样采样而不是全量采样，此时我们需要使用 random 采样器或 adaptive 采样器，甚至自己定义一个。默认情况下，kamon 默认提供的是 adaptive，并且总的最大吞吐量kamon.trace.adaptive-sampler.throughput默认是 600。采样器将尽最大努力平衡采样，以产生不超过此数量的采样。

此外kamon还提供了更详细的配置以更细粒度决定采样，下面是如何配置一组操作的吞吐量范围：

kamon.trace.groups.groupName.minimum-throughput: [number] 定义了组内每个操作每分钟的最小采样次数。尽管采样器会尽力提供最小的采样次数，但实际的最小次数会因应用流量和总体吞吐量目标的不同而不同。
kamon.trace.groups.groupName.maximum-throughput: [number] 定义了组内每个操作每分钟预期的最大采样次数，而不管在达到全局吞吐量目标之前是否有剩余空间。

组是指描述了一组命名的操作和适用于它们的规则。如果 kamon 启用的是 always 或 never 采样器，则组配置会被忽略。

例如，如果想确保一个操作永远不会被采样，比如健康检查，可以这样配置：

groups {
    health-checks {
        operations = ["GET \/status"]
        rules {
          sample = never
        }
}

配置随机采样器

使用kamon.trace.random-sampler.probability配置采样的概率。必须是一个介于0和1之间的值，用途如下。

val _upperBoundary = Long.MaxValue * probability
val _lowerBoundary = -_upperBoundary

如果配置的是 0.01，那么最终区间就是：[-0.01 * Long.MaxValue, 0.01 * Long.MaxValue]

随机采样的实现

最后来看看随机采样的实现，核心代码如下：

def decide(): SamplingDecision = {
    _decisions.increment()
// 伪随机数
val random = ThreadLocalRandom.current().nextLong()
if (random >= _lowerBoundary && random <= _upperBoundary)
    SamplingDecision.Sample // 意味着会采样
else
    SamplingDecision.DoNotSample // 意味着不会采样
}

简而言之，随机采样器确实是仅随机的，它只根据伪随机数是否在区间来判定采样。

decide是采样器Sampler特质的唯一一个公开方法，在需要判断是否应该采样时被调用。SamplingDecision是一个采样判定结果，比如子类DoNotSample就表示某个 trace 的所有 span 都不应该被捕获或报告。

自适应采样器的实现稍微复杂一点，先了解了随机采样，更方便后面理解自适应采样器。

自适应采样的实现

因为采样器不会超出最大吞吐量配置，但是可能存在一种相反的特殊情况：吞吐量离预设值还有一段距离。

而自适应采样器便能实现对较低吞吐量的采样器进行补偿，补偿策略便是基于操作的历史平均吞吐量、当前吞吐量、rules 配置的该操作最大吞吐量，动态地将所有采样器的采样率调整到接近设定的目标吞吐量。

因此自适应采样器的实现稍微复杂一点，它还分为 rebalancing（再平衡） 和 adapting（适应） 两个阶段，

历史记录是指：计算在过去60个区间内做出了多少个采样决定。如果自采样器创建以来的时间少于60个间隔，则使用可用值的平均值来填补空白。这种特殊的逻辑用于在每个单独操作的启动过程中使之顺利进行。

rebalancing 阶段

rebalancing 阶段发生在有新操作时（判定采样时），判定操作是否有采样器，有则直接取，并获得判定，否则构造采样器，并获得判定。最终返回判定，调用方就知道是否应该采样了。

操作采样器使用内部状态来决定是否应该对一个特定的操作进行采样。这里说的采样器具体是指OperationSampler及其子类，而不是上面的Sampler。

rebalancing 阶段的主要逻辑：

获取采样器，没有就根据配置的 groups 来创建新的采样器。
- 创建判定，是否有自定义采样器，有的话还需平衡一下：为什么？因为 Delta 可能为负数
  1. 基于配置的 rules，获取所有非恒定采样器
  2. Delta = 基本吞吐量（配置的 throughputGoal / 随机采样器的数量）- 操作吞吐量（使用 rules 校正后的基本吞吐量），计算每个采样器分配 Delta 值
  3. 补偿采样器因满足规则产生的赤字或盈余（有 Delta 值），把 Delta 均分给每个非自定义采样器。
  4. 给所有采样器初始化 吞吐量
- 没有自定义采样器，不用平衡了，使用 rules 配置校正基本吞吐量即可。
保存采样器到集合中，同样的操作时再次判定时，可以直接获取
返回该采样器的采样判定
1. 恒定 – 根据配置，配置什么就是什么，如果 rules 的 sample配置的是 always ，这里就是 Sample，否则就是 DoNotSample
2. 随机 – 这里的随机与上面的随机采样器实现基本是一样的，只不过区间的上界和下界可以被动态调整。然而动态调整不属于 rebalancing 阶段的工作。主要是由 adapting 阶段完成。

adapting 阶段

adapting 阶段是一个后台定时任务（每秒执行一次）在执行，使用所有操作的随机采样器信息和历史采样信息来更新它们的采样概率，如果有剩余的吞吐量，还可以选择提升操作。

当前操作的吞吐量如果比历史记录平均值和该操作配置的最大吞吐量小，那么就可以尝试提升。

adapting 阶段的主要逻辑：

获得所有操作的随机采样器
根据历史记录计算平均吞吐量，并试图找到未使用的吞吐量，后面尝试分配给其他操作
有未使用的吞吐量且不是首次 adapting，计算出在不打破预期的最大吞吐量的情况下，我们可以将这个操作提高到什么程度，并需要考虑，即使一个操作从极限角度看可以达到更高的吞吐量，但它的历史吞吐量才是真正的预期吞吐量。
根据历史记录和提升操作的个数，得出平均每个操作可获得的提升吞吐量
- 如果这个提升小于该可接受的最大值（大于0），该操作直接提升这么多吞吐量
- 如果这个提升大于该可接受的最大值（大于0），该操作能提升吞吐量只能是自己可接受的最大值
- 如果该操作已经不能被提升（小于等于0），则不会提升。

提升就是修改吞吐量，并得到 新吞吐量与平均吞吐量的占比 ，即为新的概率，该概率被用于与上界相乘，如果最后上界增大了，那么随后随机数就会更大概率落在区间内。

与随机采样器一样，如果计算出的占比的是 0.01，那么最终区间就是：[-0.01 * Long.MaxValue, 0.01 * Long.MaxValue]

总结

由于 adapting 阶段是一种自适应算法，会自动根据历史数据计算出最合适的吞吐量，所以我们只能配置全局最大吞吐量，和每个操作的最大吞吐量。（有些接口可能不需要太详细精确的 trace，就可以降低吞吐量（采样率））

span生成算法

第一种：span 和 trace，大小 8 bytes 默认设置
第二种：trace 大小 16 bytes， span 大小 8 bytes
生成逻辑 kamon.trace.Identifier.Factory#generate
- long random = ThreadLocalRandom.current().nextLong()
- bytes 使用 random 填充的二进制数组
- String 将 random 转为16进制字符串 - 我们看到的就是这个