上一篇【第44篇】自定义OAL指标实战打造你的专属可观测仪表盘下一篇【第46篇】OAP集群通信协议节点间对话的暗语一、开篇从一个极端的假设开始假设你的系统每秒产生10,000条Span数据部署了10个OAP节点组成集群。如果每个OAP节点都把原始Span数据写入存储然后由存储层负责聚合会发生什么10个OAP节点 × 10,000条/秒 100,000次/秒写入ES 再加上 100,000次/秒 × (AVG SUM COUNT P99) 400,000次聚合查询/秒 结论在写入的同时还要做聚合查询ES会直接挂掉这就是SkyWalking采用两级聚合的根本原因。它的设计思路就像物流体系的分级配送各门店订单 区域集散中心 全国总仓 L1 本地聚合 L2 集群聚合 写入存储 (每个OAP节点) (集群协调) (ES/MySQL) 每收到一条数据 收集所有节点结果 持久化 立刻在内存聚合 做二次合并计算 供查询 图1两级聚合 分布式计算中的物流分拣中心二、两级聚合的架构全景----------------------- | Agent探针集群 | | (100 Service, 1000 | | Instances) | ---------------------- | gRPC (Segment/Log/Meter) | ------------------------------------------------------ | | | v v v -------------------- -------------------- -------------------- | OAP Node 1 | | OAP Node 2 | | OAP Node N | | | | | | | | [L1 本地聚合] | | [L1 本地聚合] | | [L1 本地聚合] | | | | | | | | 收到数据 - 内存聚合 | | 收到数据 - 内存聚合 | | 收到数据 - 内存聚合 | | (分钟级窗口) | | (分钟级窗口) | | (分钟级窗口) | --------------------- --------------------- --------------------- | | | | L1结果 | L1结果 | L1结果 ------------------------------------------------------ | -----------v----------- | 集群协调层 | | (Nacos/ZK/K8s) | ---------------------- | -----------v----------- | L2 全局聚合 | | | | 合并所有节点的L1结果 | | 生成最终的小时/天指标 | | 写入存储 | ---------------------- | -----------v----------- | 存储层 | | (ES/MySQL/H2) | ----------------------- 图2两级聚合的完整架构图三、L1聚合每个节点的本地会计L1聚合是分布式的第一层防线。它运行在每个OAP节点内部直接消费探针数据。3.1 L1聚合的工作流程/** * L1聚合的核心逻辑简化版 */publicclassL1AggregationWorker{// 分钟级聚合窗口// Key: MetricKey(指标名实体ID时间窗口)// Value: 聚合中Metric对象privatefinalMapMetricKey,Metricsl1BuffernewConcurrentHashMap();/** * 每收到一条探针数据的回调 */publicvoidonSegmentReceived(SegmentObjectsegment){// 1. 解析Segment提取各维度数据ListSourcesourcessegmentToSources(segment);// 2. 对每个Source查找对应的OAL指标定义for(Sourcesource:sources){for(AnalyzeDefinitiondef:getOALDefinitions(source)){// 3. 确定时间窗口longminuteWindowtruncateToMinute(segment.getStartTime());// 4. 构建聚合KeyMetricKeykeynewMetricKey(def.getMetricName(),source.getEntityId(),minuteWindow);// 5. 获取或创建聚合对象Metricsexistingl1Buffer.get(key);if(existingnull){existingdef.createMetrics();l1Buffer.put(key,existing);}// 6. 执行流式聚合调用生成的combine方法def.getProcessor().combine(source,existing,context);}}}/** * L1聚合窗口闭合时的回调每分钟 */Scheduled(fixedRate60_000)publicvoidflushMinuteWindow(){longcurrentMinutetruncateToMinute(System.currentTimeMillis());// 找出所有上一分钟窗口的聚合结果longtargetMinutecurrentMinute-60_000;ListMetricscompletedMetricsl1Buffer.values().stream().filter(m-m.getTimeBucket()targetMinute).collect(Collectors.toList());for(Metricsmetrics:completedMetrics){// 1. 写入内存中的小时聚合窗口本地mergeToHourWindow(metrics);// 2. 发送给L2聚合协调器集群l2AggregationCoordinator.send(metrics);// 3. 从L1缓存中移除已处理完毕l1Buffer.remove(MetricKey.from(metrics));}}}3.2 L1聚合的关键特征特征说明窗口粒度1分钟聚合位置OAP节点内存数据来源gRPC接收的原始Segment输出目标L2协调器 本地小时窗口丢失容忍如果OAP节点崩溃丢失最多60秒的L1数据四、L2聚合集群层面的总会计师L2聚合汇总所有OAP节点的L1结果计算集群级别的最终指标。4.1 L2聚合的完整流程/** * L2聚合协调器 * 运行在集群中的某个节点上或所有节点协商分工 */publicclassL2AggregationCoordinator{// L2聚合缓冲区// 收集所有节点发来的L1结果privatefinalMapMetricKey,ListMetricsl2CollectBuffer;// 分配策略通过consistent hash决定哪个节点处理哪个指标privatefinalConsistentHashStringpartitioner;/** * 接收L1聚合结果 */publicvoidreceive(Metricsl1Result,StringsenderNodeId){MetricKeykeyMetricKey.from(l1Result);// 确定该指标由哪个节点负责L2聚合StringtargetNodepartitioner.getNode(key.toString());if(isSelf(targetNode)){// 由本节点处理L2聚合aggregateL2(key,l1Result);}else{// 转发给目标节点remoteClient.forward(targetNode,l1Result);}}/** * 执行L2聚合 */privatevoidaggregateL2(MetricKeykey,Metricsl1Result){// 获取当前L2窗口的聚合结果Metricsexistingl2Buffer.get(key);if(existingnull){// 第一个到达的结果直接记录l2Buffer.put(key,l1Result);}else{// 后续到达的结果调用combine合并// L2版本的combine接受两个Metrics对象做合并MetricsDefinitiondefmetricsRegistry.getDefinition(key.getMetricName());def.getProcessor().combine(l1Result,existing);}// 检查是否收集了足够的结果90%以上节点已上报if(checkCollectionComplete(key)){MetricsfinalResultl2Buffer.remove(key);// 写入存储storageWriter.write(finalResult);}}}4.2 L2聚合的窗口设计时间轴 ------ |--- 分钟1 ---|--- 分钟2 ---|--- 分钟3 ---| | | | | L1: [聚合中] L1: [聚合中] L1: [聚合中] | | | v v v L1结果 L1结果 L1结果 | | | ------------------------- | | v v L2: [小时窗口聚合中...] | | v v L2下发 L2最终结果 (每分钟) (每小时) | v 写入存储 (小时粒度) 再经过24个小时窗口 | v 降采样为天粒度 写入存储 图3L1→L2→存储的时序关系五、为什么选择两级聚合5.1 一级聚合 vs 两级聚合------------------------------------------------------------------- | 一级聚合 vs 两级聚合 | ------------------------------------------------------------------- | | | 一级聚合所有OAP节点直接写入存储由存储层聚合 | | ---------------------------------------------------------------- | | | 优点架构简单无需集群协调 | | | | 缺点 | | | | 1. 写入量大每秒数万次写入 | | | | 2. 存储层压力巨大 | | | | 3. 无法实时查询分钟级指标数据还没落入存储 | | | | 4. 不同存储实现的聚合能力不一致 | | | ---------------------------------------------------------------- | | | | 两级聚合OAP内存聚合 集群协调 批量写入 | | ---------------------------------------------------------------- | | | 优点 | | | | 1. 写入量降低99%聚合后写入 | | | | 2. 存储层只负责持久化不做计算 | | | | 3. OAP内存中可以直接查询实时指标 | | | | 4. 存储实现无关H2/ES/MySQL一视同仁 | | | | 缺点 | | | | 1. OAP节点内存使用较大 | | | | 2. 集群协调复杂度增加 | | | | 3. 多节点崩溃时可能有数据丢失 | | | ---------------------------------------------------------------- | | | ------------------------------------------------------------------- 图4两级聚合的权衡分析5.2 量化的性能收益假设场景100个服务每秒10000个Segment请求 【一级聚合方案】 写入次数: 10000次/秒原始Segment 每日写入量: 864,000,000次 存储磁盘IO: ~86.4GB/天假设每个Segment 100B 【两级聚合方案】 L1聚合后写入: ~100次/秒100个服务 × 聚合结果 每日写入量: 8,640,000次 存储磁盘IO: ~2.5GB/天按小时指标计 性能提升 写入量降低: 99% 磁盘IO降低: 97% CPU降低: ~90%存储层不再需要聚合计算六、容错与数据一致性两级聚合在极端情况下的行为6.1 OAP节点崩溃正常情况 Node1 L1结果 - L2聚合 - 写入存储 ✓ Node2 L1结果 - L2聚合 - 写入存储 ✓ Node3 L1结果 - L2聚合 - 写入存储 ✓ Node2崩溃 Node1 L1结果 - L2聚合 - 写入存储 ✓ Node2 L1结果 - 丢失最多60秒数据 ✗ Node3 L1结果 - L2聚合 - 写入存储 ✓ 数据丢失范围 最多丢失 Node2 上当前分钟窗口的L1聚合结果 大约相当于总数据的 1/NN节点数6.2 存储写入失败的退避publicclassStorageWriteBuffer{privatefinalBlockingQueueMetricspendingWrites;publicvoidwrite(Metricsmetrics)throwsStorageException{intretries3;intbackoffMs1000;for(inti0;iretries;i){try{storageDAO.insert(metrics);return;// 成功}catch(StorageExceptione){logger.warn(写入存储失败(重试 {}/{}): {},i1,retries,e.getMessage());if(iretries-1){Thread.sleep(backoffMs*(i1));// 指数退避}}}// 三次重试都失败放入pending队列pendingWrites.offer(metrics);logger.error(存储写入持久失败数据已缓存到pending队列);}}七、指标数据在两级聚合中的生命周期Segment到达 | v [L1聚合 - 内存Map] 每收到一条数据就调用 combine(source, metrics) | | (每分钟窗口闭合) v ---------------------- | | v v L1结果 L1结果存入 发送给L2 本地小时缓存 | | | v | [本地小时聚合] | 合并多个L1分钟结果 | v [L2收集缓冲区] 收集所有节点的L1结果 | | (收集足够节点结果后) v [L2聚合] -- combine(L1_A, L1_B, L1_C...) | v [写入存储] |-- 分钟粒度结果 (可选) | v [小时聚合] 在L2层对分钟结果做二次聚合 | v [写入存储 - 小时粒度] | v [天级聚合] 在存储层通过小时数据降采样 | v [写入存储 - 天粒度] | v [TTL清理] 超过保留期的数据删除 图5指标数据从生到死的完整生命周期八、总结两级聚合是SkyWalking在分布式可观测场景下的最优解层级位置粒度职责L1OAP节点内存分钟级本地实时聚合降低数据量L2集群协调分钟→小时级全局合并保证数据完整性存储ES/MySQL小时→天级持久化供查询和历史分析这个架构的精妙之处在于用内存换IO用集群协调换数据量缩减用局部容忍换全局性能。下一篇我们将深入OAP集群间的通信协议看看L1结果是如何在节点间高效传输的。上一篇【第44篇】自定义OAL指标实战打造你的专属可观测仪表盘下一篇【第46篇】OAP集群通信协议节点间对话的暗语
