日志开源组件（六）Adaptive Sampling 自适应采样

0650

业务背景

有时候日志的信息比较多，怎么样才可以让系统做到自适应采样呢？

拓展阅读

日志开源组件（一）java 注解结合 spring aop 实现自动输出日志

日志开源组件（二）java 注解结合 spring aop 实现日志traceId唯一标识

日志开源组件（三）java 注解结合 spring aop 自动输出日志新增拦截器与过滤器

日志开源组件（四）如何动态修改 spring aop 切面信息？让自动日志输出框架更好用

日志开源组件（五）如何将 dubbo filter 拦截器原理运用到日志拦截器中？

自适应采样

是什么？

系统生成的日志可以包含大量信息，包括错误、警告、性能指标等，但在实际应用中，处理和分析所有的日志数据可能会对系统性能和资源产生负担。

自适应采样在这种情况下发挥作用，它能够根据当前系统状态和日志信息的重要性，智能地决定哪些日志需要被采样记录，从而有效地管理和分析日志数据。

采样的必要性

日志采样系统会给业务系统额外增加消耗，很多系统在接入的时候会比较排斥。

给他们一个百分比的选择，或许是一个不错的开始，然后根据实际需要选择合适的比例。

自适应采样是一个对用户透明，同时又非常优雅的方案。

自适应

如何通过 java 实现自适应采样?

接口定义

首先我们定义一个接口，返回 boolean。

根据是否为 true 来决定是否输出日志。

/**
 * 采样条件
 * @author binbin.hou
 * @since 0.5.0
 */
public interface IAutoLogSampleCondition {
    /**
     * 条件
     *
     * @param context 上下文
     * @return 结果
     * @since 0.5.0
     */
    boolean sampleCondition(IAutoLogContext context);
}
/**
 * 采样条件
 * @author binbin.hou
 * @since 0.5.0
 */
public interface IAutoLogSampleCondition {

    /**
     * 条件
     *
     * @param context 上下文
     * @return 结果
     * @since 0.5.0
     */
    boolean sampleCondition(IAutoLogContext context);

}
/**
 * 采样条件
 * @author binbin.hou
 * @since 0.5.0
 */
public interface IAutoLogSampleCondition {

    /**
     * 条件
     *
     * @param context 上下文
     * @return 结果
     * @since 0.5.0
     */
    boolean sampleCondition(IAutoLogContext context);

}

百分比概率采样

我们先实现一个简单的概率采样。

0-100 的值，让用户指定，按照百分比决定是否采样。

public class InnerRandomUtil {
    /**
     * 1. 计算一个 1-100 的随机数 randomVal
     * 2. targetRatePercent 值越大，则返回 true 的概率越高
     * @param targetRatePercent 目标百分比
     * @return 结果
     */
    public static boolean randomRateCondition(int targetRatePercent) {
        if(targetRatePercent <= 0) {
            return false;
        }
        if(targetRatePercent >= 100) {
            return true;
        }
        // 随机
        ThreadLocalRandom threadLocalRandom = ThreadLocalRandom.current();
        int value = threadLocalRandom.nextInt(1, 100);
        // 随机概率
        return targetRatePercent >= value;
    }
}
public class InnerRandomUtil {

    /**
     * 1. 计算一个 1-100 的随机数 randomVal
     * 2. targetRatePercent 值越大，则返回 true 的概率越高
     * @param targetRatePercent 目标百分比
     * @return 结果
     */
    public static boolean randomRateCondition(int targetRatePercent) {
        if(targetRatePercent <= 0) {
            return false;
        }
        if(targetRatePercent >= 100) {
            return true;
        }

        // 随机
        ThreadLocalRandom threadLocalRandom = ThreadLocalRandom.current();
        int value = threadLocalRandom.nextInt(1, 100);

        // 随机概率
        return targetRatePercent >= value;
    }

}
public class InnerRandomUtil {

    /**
     * 1. 计算一个 1-100 的随机数 randomVal
     * 2. targetRatePercent 值越大，则返回 true 的概率越高
     * @param targetRatePercent 目标百分比
     * @return 结果
     */
    public static boolean randomRateCondition(int targetRatePercent) {
        if(targetRatePercent <= 0) {
            return false;
        }
        if(targetRatePercent >= 100) {
            return true;
        }

        // 随机
        ThreadLocalRandom threadLocalRandom = ThreadLocalRandom.current();
        int value = threadLocalRandom.nextInt(1, 100);

        // 随机概率
        return targetRatePercent >= value;
    }

}

实现起来也非常简单，直接一个随机数，然后比较大小即可。

自适应采样

思路

我们计算一下当前日志的 QPS，让输出的概率和 QPS 称反比。

/**
 * 自适应采样
 *
 * 1. 初始化采样率为 100%，全部采样
 *
 * 2. QPS 如果越来越高，那么采样率应该越来越低。这样避免 cpu 等资源的损耗。最低为 1%
 * 如果 QPS 越来越低，采样率应该越来越高。增加样本，最高为 100%
 *
 * 3. QPS 如何计算问题
 *
 * 直接设置大小为 100 的队列，每一次在里面放入时间戳。
 * 当大小等于 100 的时候，计算首尾的时间差，currentQps = 100 / (endTime - startTime) * 1000
 *
 * 触发 rate 重新计算。
 *
 * 3.1 rate 计算逻辑
 *
 * 这里我们存储一下 preRate = 100, preQPS = ?
 *
 * newRate = (preQps / currentQps) * rate
 *
 * 范围限制：
 * newRate = Math.min(100, newRate);
 * newRate = Math.max(1, newRate);
 *
 * 3.2 时间队列的清空
 *
 * 更新完 rate 之后，对应的队列可以清空？
 *
 * 如果额外使用一个 count，好像也可以。
 * 可以调整为 atomicLong 的计算器，和 preTime。
 *
/**
 * 自适应采样
 *
 * 1. 初始化采样率为 100%，全部采样
 *
 * 2. QPS 如果越来越高，那么采样率应该越来越低。这样避免 cpu 等资源的损耗。最低为 1%
 * 如果 QPS 越来越低，采样率应该越来越高。增加样本，最高为 100%
 *
 * 3. QPS 如何计算问题
 *
 * 直接设置大小为 100 的队列，每一次在里面放入时间戳。
 * 当大小等于 100 的时候，计算首尾的时间差，currentQps = 100 / (endTime - startTime) * 1000
 *
 * 触发 rate 重新计算。
 *
 * 3.1 rate 计算逻辑
 *
 * 这里我们存储一下 preRate = 100, preQPS = ?
 *
 * newRate = (preQps / currentQps) * rate
 *
 * 范围限制：
 * newRate = Math.min(100, newRate);
 * newRate = Math.max(1, newRate);
 *
 * 3.2 时间队列的清空
 *
 * 更新完 rate 之后，对应的队列可以清空？
 *
 * 如果额外使用一个 count，好像也可以。
 * 可以调整为 atomicLong 的计算器，和 preTime。
 *
/**
 * 自适应采样
 *
 * 1. 初始化采样率为 100%，全部采样
 *
 * 2. QPS 如果越来越高，那么采样率应该越来越低。这样避免 cpu 等资源的损耗。最低为 1%
 * 如果 QPS 越来越低，采样率应该越来越高。增加样本，最高为 100%
 *
 * 3. QPS 如何计算问题
 *
 * 直接设置大小为 100 的队列，每一次在里面放入时间戳。
 * 当大小等于 100 的时候，计算首尾的时间差，currentQps = 100 / (endTime - startTime) * 1000
 *
 * 触发 rate 重新计算。
 *
 * 3.1 rate 计算逻辑
 *
 * 这里我们存储一下 preRate = 100, preQPS = ?
 *
 * newRate = (preQps / currentQps) * rate
 *
 * 范围限制：
 * newRate = Math.min(100, newRate);
 * newRate = Math.max(1, newRate);
 *
 * 3.2 时间队列的清空
 *
 * 更新完 rate 之后，对应的队列可以清空？
 *
 * 如果额外使用一个 count，好像也可以。
 * 可以调整为 atomicLong 的计算器，和 preTime。
 *

代码实现

public class AutoLogSampleConditionAdaptive implements IAutoLogSampleCondition {
    private static final AutoLogSampleConditionAdaptive INSTANCE = new AutoLogSampleConditionAdaptive();
    /**
     * 单例的方式获取实例
     * @return 结果
     */
    public static AutoLogSampleConditionAdaptive getInstance() {
        return INSTANCE;
    }
    /**
     * 次数大小限制，即接收到多少次请求更新一次 adaptive 计算
     *
     * TODO: 这个如何可以让用户可以自定义呢？后续考虑配置从默认的配置文件中读取。
     */
    private static final int COUNT_LIMIT = 1000;
    /**
     * 自适应比率，初始化为 100.全部采集
     */
    private volatile int adaptiveRate = 100;
    /**
     * 上一次的 QPS
     *
     * TODO: 这个如何可以让用户可以自定义呢？后续考虑配置从默认的配置文件中读取。
     */
    private volatile double preQps = 100.0;
    /**
     * 上一次的时间
     */
    private volatile long preTime;
    /**
     * 总数，请求计数器
     */
    private final AtomicInteger counter;
    public AutoLogSampleConditionAdaptive() {
        preTime = System.currentTimeMillis();
        counter = new AtomicInteger(0);
    }
    @Override
    public boolean sampleCondition(IAutoLogContext context) {
        int count = counter.incrementAndGet();
        // 触发一次重新计算
        if(count >= COUNT_LIMIT) {
            updateAdaptiveRate();
        }
        // 直接计算是否满足
        return InnerRandomUtil.randomRateCondition(adaptiveRate);
    }
}
public class AutoLogSampleConditionAdaptive implements IAutoLogSampleCondition {

    private static final AutoLogSampleConditionAdaptive INSTANCE = new AutoLogSampleConditionAdaptive();

    /**
     * 单例的方式获取实例
     * @return 结果
     */
    public static AutoLogSampleConditionAdaptive getInstance() {
        return INSTANCE;
    }

    /**
     * 次数大小限制，即接收到多少次请求更新一次 adaptive 计算
     *
     * TODO: 这个如何可以让用户可以自定义呢？后续考虑配置从默认的配置文件中读取。
     */
    private static final int COUNT_LIMIT = 1000;

    /**
     * 自适应比率，初始化为 100.全部采集
     */
    private volatile int adaptiveRate = 100;

    /**
     * 上一次的 QPS
     *
     * TODO: 这个如何可以让用户可以自定义呢？后续考虑配置从默认的配置文件中读取。
     */
    private volatile double preQps = 100.0;

    /**
     * 上一次的时间
     */
    private volatile long preTime;

    /**
     * 总数，请求计数器
     */
    private final AtomicInteger counter;

    public AutoLogSampleConditionAdaptive() {
        preTime = System.currentTimeMillis();
        counter = new AtomicInteger(0);
    }

    @Override
    public boolean sampleCondition(IAutoLogContext context) {
        int count = counter.incrementAndGet();

        // 触发一次重新计算
        if(count >= COUNT_LIMIT) {
            updateAdaptiveRate();
        }

        // 直接计算是否满足
        return InnerRandomUtil.randomRateCondition(adaptiveRate);
    }

}
public class AutoLogSampleConditionAdaptive implements IAutoLogSampleCondition {

    private static final AutoLogSampleConditionAdaptive INSTANCE = new AutoLogSampleConditionAdaptive();

    /**
     * 单例的方式获取实例
     * @return 结果
     */
    public static AutoLogSampleConditionAdaptive getInstance() {
        return INSTANCE;
    }

    /**
     * 次数大小限制，即接收到多少次请求更新一次 adaptive 计算
     *
     * TODO: 这个如何可以让用户可以自定义呢？后续考虑配置从默认的配置文件中读取。
     */
    private static final int COUNT_LIMIT = 1000;

    /**
     * 自适应比率，初始化为 100.全部采集
     */
    private volatile int adaptiveRate = 100;

    /**
     * 上一次的 QPS
     *
     * TODO: 这个如何可以让用户可以自定义呢？后续考虑配置从默认的配置文件中读取。
     */
    private volatile double preQps = 100.0;

    /**
     * 上一次的时间
     */
    private volatile long preTime;

    /**
     * 总数，请求计数器
     */
    private final AtomicInteger counter;

    public AutoLogSampleConditionAdaptive() {
        preTime = System.currentTimeMillis();
        counter = new AtomicInteger(0);
    }

    @Override
    public boolean sampleCondition(IAutoLogContext context) {
        int count = counter.incrementAndGet();

        // 触发一次重新计算
        if(count >= COUNT_LIMIT) {
            updateAdaptiveRate();
        }

        // 直接计算是否满足
        return InnerRandomUtil.randomRateCondition(adaptiveRate);
    }

}

每次累加次数超过限定次数之后，我们就更新一下对应的日志概率。

最后的概率计算和上面的百分比类似，不再赘述。

/**
 * 更新自适应的概率
 *
 * 100 计算一次，其实还好。实际应该可以适当调大这个阈值，本身不会经常变化的东西。
 */
private synchronized void updateAdaptiveRate() {
    //消耗的毫秒数
    long costTimeMs = System.currentTimeMillis() - preTime;
    //qps 的计算，时间差是毫秒。所以次数需要乘以 1000
    double currentQps = COUNT_LIMIT*1000.0 / costTimeMs;
    // preRate * preQps = currentRate * currentQps; 保障采样均衡，服务器压力均衡
    // currentRate = (preRate * preQps) / currentQps;
    // 更新比率
    int newRate = 100;
    if(currentQps > 0) {
        newRate = (int) ((adaptiveRate * preQps) / currentQps);
        newRate = Math.min(100, newRate);
        newRate = Math.max(1, newRate);
    }
    // 更新 rate
    adaptiveRate = newRate;
    // 更新 QPS
    preQps = currentQps;
    // 更新上一次的时间内戳
    preTime = System.currentTimeMillis();
    // 归零
    counter.set(0);
}
/**
 * 更新自适应的概率
 *
 * 100 计算一次，其实还好。实际应该可以适当调大这个阈值，本身不会经常变化的东西。
 */
private synchronized void updateAdaptiveRate() {
    //消耗的毫秒数
    long costTimeMs = System.currentTimeMillis() - preTime;
    //qps 的计算，时间差是毫秒。所以次数需要乘以 1000
    double currentQps = COUNT_LIMIT*1000.0 / costTimeMs;
    // preRate * preQps = currentRate * currentQps; 保障采样均衡，服务器压力均衡
    // currentRate = (preRate * preQps) / currentQps;
    // 更新比率
    int newRate = 100;
    if(currentQps > 0) {
        newRate = (int) ((adaptiveRate * preQps) / currentQps);
        newRate = Math.min(100, newRate);
        newRate = Math.max(1, newRate);
    }
    // 更新 rate
    adaptiveRate = newRate;
    // 更新 QPS
    preQps = currentQps;
    // 更新上一次的时间内戳
    preTime = System.currentTimeMillis();
    // 归零
    counter.set(0);
}
/**
 * 更新自适应的概率
 *
 * 100 计算一次，其实还好。实际应该可以适当调大这个阈值，本身不会经常变化的东西。
 */
private synchronized void updateAdaptiveRate() {
    //消耗的毫秒数
    long costTimeMs = System.currentTimeMillis() - preTime;
    //qps 的计算，时间差是毫秒。所以次数需要乘以 1000
    double currentQps = COUNT_LIMIT*1000.0 / costTimeMs;
    // preRate * preQps = currentRate * currentQps; 保障采样均衡，服务器压力均衡
    // currentRate = (preRate * preQps) / currentQps;
    // 更新比率
    int newRate = 100;
    if(currentQps > 0) {
        newRate = (int) ((adaptiveRate * preQps) / currentQps);
        newRate = Math.min(100, newRate);
        newRate = Math.max(1, newRate);
    }
    // 更新 rate
    adaptiveRate = newRate;
    // 更新 QPS
    preQps = currentQps;
    // 更新上一次的时间内戳
    preTime = System.currentTimeMillis();
    // 归零
    counter.set(0);
}

自适应代码-改良

问题

上面的自适应算法一般情况下都可以运行的很好。

但是有一种情况会不太好，那就是流量从高峰期到低峰期。

比如凌晨11点是请求高峰期，我们的输出日志概率很低。深夜之后请求数会很少，想达到累计值就会很慢，这个时间段就会导致日志输出很少。

如何解决这个问题呢？

思路

我们可以通过固定时间窗口的方式，来定时调整流量概率。

java 实现

我们初始化一个定时任务，1min 定时更新一次。

public class AutoLogSampleConditionAdaptiveSchedule implements IAutoLogSampleCondition {
    private static final ScheduledExecutorService EXECUTOR_SERVICE = Executors.newSingleThreadScheduledExecutor();
    /**
     * 时间分钟间隔
     */
    private static final int TIME_INTERVAL_MINUTES = 5;
    /**
     * 自适应比率，初始化为 100.全部采集
     */
    private volatile int adaptiveRate = 100;
    /**
     * 上一次的总数
     *
     * TODO: 这个如何可以让用户可以自定义呢？后续考虑配置从默认的配置文件中读取。
     */
    private volatile long preCount;
    /**
     * 总数，请求计数器
     */
    private final AtomicLong counter;
    public AutoLogSampleConditionAdaptiveSchedule() {
        counter = new AtomicLong(0);
        preCount = TIME_INTERVAL_MINUTES * 60 * 100;
        //1. 1min 后开始执行
        //2. 中间默认 5 分钟更新一次
        EXECUTOR_SERVICE.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                updateAdaptiveRate();
            }
        }, 60, TIME_INTERVAL_MINUTES * 60, TimeUnit.SECONDS);
    }
    @Override
    public boolean sampleCondition(IAutoLogContext context) {
        counter.incrementAndGet();
        // 直接计算是否满足
        return InnerRandomUtil.randomRateCondition(adaptiveRate);
    }
}
public class AutoLogSampleConditionAdaptiveSchedule implements IAutoLogSampleCondition {

    private static final ScheduledExecutorService EXECUTOR_SERVICE = Executors.newSingleThreadScheduledExecutor();

    /**
     * 时间分钟间隔
     */
    private static final int TIME_INTERVAL_MINUTES = 5;

    /**
     * 自适应比率，初始化为 100.全部采集
     */
    private volatile int adaptiveRate = 100;

    /**
     * 上一次的总数
     *
     * TODO: 这个如何可以让用户可以自定义呢？后续考虑配置从默认的配置文件中读取。
     */
    private volatile long preCount;

    /**
     * 总数，请求计数器
     */
    private final AtomicLong counter;

    public AutoLogSampleConditionAdaptiveSchedule() {
        counter = new AtomicLong(0);
        preCount = TIME_INTERVAL_MINUTES * 60 * 100;

        //1. 1min 后开始执行
        //2. 中间默认 5 分钟更新一次
        EXECUTOR_SERVICE.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                updateAdaptiveRate();
            }
        }, 60, TIME_INTERVAL_MINUTES * 60, TimeUnit.SECONDS);
    }

    @Override
    public boolean sampleCondition(IAutoLogContext context) {
        counter.incrementAndGet();

        // 直接计算是否满足
        return InnerRandomUtil.randomRateCondition(adaptiveRate);
    }

}
public class AutoLogSampleConditionAdaptiveSchedule implements IAutoLogSampleCondition {

    private static final ScheduledExecutorService EXECUTOR_SERVICE = Executors.newSingleThreadScheduledExecutor();

    /**
     * 时间分钟间隔
     */
    private static final int TIME_INTERVAL_MINUTES = 5;

    /**
     * 自适应比率，初始化为 100.全部采集
     */
    private volatile int adaptiveRate = 100;

    /**
     * 上一次的总数
     *
     * TODO: 这个如何可以让用户可以自定义呢？后续考虑配置从默认的配置文件中读取。
     */
    private volatile long preCount;

    /**
     * 总数，请求计数器
     */
    private final AtomicLong counter;

    public AutoLogSampleConditionAdaptiveSchedule() {
        counter = new AtomicLong(0);
        preCount = TIME_INTERVAL_MINUTES * 60 * 100;

        //1. 1min 后开始执行
        //2. 中间默认 5 分钟更新一次
        EXECUTOR_SERVICE.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                updateAdaptiveRate();
            }
        }, 60, TIME_INTERVAL_MINUTES * 60, TimeUnit.SECONDS);
    }

    @Override
    public boolean sampleCondition(IAutoLogContext context) {
        counter.incrementAndGet();

        // 直接计算是否满足
        return InnerRandomUtil.randomRateCondition(adaptiveRate);
    }

}

其中更新概率的逻辑和上面类似：

/**
 * 更新自适应的概率
 *
 * QPS = count / time_interval
 *
 * 其中时间维度是固定的，所以可以不用考虑时间。
 */
private synchronized void updateAdaptiveRate() {
    // preRate * preCount = currentRate * currentCount; 保障采样均衡，服务器压力均衡
    // currentRate = (preRate * preCount) / currentCount;
    // 更新比率
    long currentCount = counter.get();
    int newRate = 100;
    if(currentCount != 0) {
        newRate = (int) ((adaptiveRate * preCount) / currentCount);
        newRate = Math.min(100, newRate);
        newRate = Math.max(1, newRate);
    }
    // 更新自适应频率
    adaptiveRate = newRate;
    // 更新 QPS
    preCount = currentCount;
    // 归零
    counter.set(0);
}
/**
 * 更新自适应的概率
 *
 * QPS = count / time_interval
 *
 * 其中时间维度是固定的，所以可以不用考虑时间。
 */
private synchronized void updateAdaptiveRate() {
    // preRate * preCount = currentRate * currentCount; 保障采样均衡，服务器压力均衡
    // currentRate = (preRate * preCount) / currentCount;
    // 更新比率
    long currentCount = counter.get();
    int newRate = 100;
    if(currentCount != 0) {
        newRate = (int) ((adaptiveRate * preCount) / currentCount);
        newRate = Math.min(100, newRate);
        newRate = Math.max(1, newRate);
    }
    // 更新自适应频率
    adaptiveRate = newRate;
    // 更新 QPS
    preCount = currentCount;
    // 归零
    counter.set(0);
}
/**
 * 更新自适应的概率
 *
 * QPS = count / time_interval
 *
 * 其中时间维度是固定的，所以可以不用考虑时间。
 */
private synchronized void updateAdaptiveRate() {
    // preRate * preCount = currentRate * currentCount; 保障采样均衡，服务器压力均衡
    // currentRate = (preRate * preCount) / currentCount;
    // 更新比率
    long currentCount = counter.get();
    int newRate = 100;
    if(currentCount != 0) {
        newRate = (int) ((adaptiveRate * preCount) / currentCount);
        newRate = Math.min(100, newRate);
        newRate = Math.max(1, newRate);
    }
    // 更新自适应频率
    adaptiveRate = newRate;
    // 更新 QPS
    preCount = currentCount;
    // 归零
    counter.set(0);
}

小结

让系统自动化分配资源，是一种非常好的思路，可以让资源利用最大化。

实现起来也不是很困难，实际要根据我们的业务量进行观察和调整。

开源地址

auto-log https://github.com/houbb/auto-log

© 版权声明
文章版权归作者所有，未经允许请勿转载，侵权请联系 admin@trc20.tw 删除。
THE END
博客文章
喜欢就支持一下吧
点赞0

Warning: mysqli_query(): (HY000/3): Error writing file '/tmp/MYvImJBg' (Errcode: 28 - No space left on device) in /www/wwwroot/583.cn/wp-includes/class-wpdb.php on line 2345
admin的头像-五八三
相关推荐
评论 抢沙发
头像
欢迎您留下宝贵的见解！
提交
头像

昵称

图形验证码
取消
昵称

请填写用户信息：

  • 昵称(必填)
  • 邮箱(必填)
代码

请输入代码：

确认
图片
文章目录
文章目录
热门推荐