闲鱼目前的实际生产部署环境越来越复杂，横向依赖各种服务交织在一起，纵向依赖运行环境也越来越复杂。图片来自Pexels当服务出现问题时，能否在海量数据中及时定位问题根源，成为考验闲鱼服务能力的严峻挑战。当线上出现问题时，往往需要十几分钟，甚至更长的时间才能找到问题的原因。因此，快速自动诊断的系统需求应运而生，而快速诊断的基础是高性能的实时数据处理系统。这个实时数据处理系统需要具备以下能力：实时数据采集、实时分析、复杂计算、分析结果持久化。可以处理各种数据。包含应用日志、主机性能监控指标、调用链路图。高可靠性。系统没有问题，数据不会丢失。高性能，低延迟。数据处理时延不超过3秒，支持每秒千万级数据处理。本文不涉及自动问题诊断的具体分析模型，仅讨论整体实时数据处理环节的设计。输入和输出定义为了理解系统的运行，我们定义了系统的整体输入和输出。输入服务请求日志（包括traceid、时间戳、客户端IP、服务器IP、耗时、返回码、服务名、方法名）。环境监测数据（指标名称、IP、时间戳、指标值）。比如CPU、JVMGC次数、JVMGC耗时、数据库指标等。输出一段时间内某个服务错误的根本原因，将每个服务的错误分析结果用有向无环图表示。（根节点是被分析的错误节点，叶子节点是错误的根本原因节点，叶子节点可能是外部依赖的服务错误，也可能是JVM异常等）。架构设计在实际的系统运行过程中，日志数据和监控数据是随着时间的推移不断产生的。每条生成的数据都有自己的时间戳。实时流式传输这些带有时间戳的数据就像流经不同管道的水一样。如果把连续的实时数据比作流水，那么数据处理过程就类似于自来水的生产过程：自然地，我们也将实时数据的处理过程分解为采集、传输、预处理、计算、存储、计算和持久化的几个阶段。整体系统架构设计如下：采用阿里自研的SLS日志服务产品（包括Logtail+LogHub组件）进行采集，Logtail为采集客户端。之所以选择Logtail，是因为其卓越的性能、高可靠性，以及灵活的插件扩展机制。闲鱼可以定制自己的采集插件，实现各种数据的实时采集。TransmissionLoghub可以理解为数据发布订阅组件，类似Kafka的功能，作为数据传输通道，更加稳定安全。详细对比文章参考：https://yq.aliyun.com/articles/35979?spm=5176.10695662.1996646101.searchclickresult.6f2c7fbe6g3xgPPreprocessing实时数据预处理部分使用了Blink流计算处理组件（开源版叫Flink，Blink是阿里内部基于Flink的一个增强版）。目前常用的实时流计算开源产品有Jstorm、SparkStream、Flink：由于Jstorm没有中间计算态，计算过程中需要的中间结果必须依赖外部存储，这会导致频繁IO并影响其性能。SparkStream本质上是使用微小的批处理来模拟实时计算，但实际上还是有一定延迟的。Flink由于其优秀的状态管理机制，保证了其计算的性能和实时性，同时提供了完整的SQL表达式，让流式计算变得更加简单。经过计算和持久化数据的预处理，最终生成调用链路聚合日志和主机监控数据。主机监控数据将独立存储在TSDB时序数据库中，供后续统计分析。TSDB对时间索引数据的特殊存储结构设计，非常适合时序数据的存储和查询。调用链路日志聚合数据，提供给Cep/GraphService进行诊断模型分析。Cep/GraphService是闲鱼自研的一款应用，实现模型分析、复杂数据处理、与外部服务交互，并借助RDB实现图数据的实时聚合。最后将Cep/GraphService分析的结果作为图数据，实时dump到Lindorm中，提供在线查询。Lindorm可以看作是Hbase的增强版，在系统中起到持久化存储的作用。详细设计和性能优化日志和索引数据采集使用Logtail。整个数据采集流程如图所示：它提供了非常灵活的插件机制，插件有四种类型：Inputs：输入插件获取数据。Processors：处理插件，对获取到的数据进行处理。聚合器：用于聚合数据的聚合插件。Flushers：输出插件，将数据输出到指定的Sink。由于指标数据（如CPU、内存、JVM指标）的获取需要在本机调用服务接口获取，因此尽量减少请求次数。在Logtail中，一个Input占用一个Goroutine。通过自定义Input插件和Processors插件，闲鱼通过一个服务请求在一个Input插件中获取多个指标数据（如CPU、内存、JVM指标）（获取指标的接口由基础接口提供）监督组）。并格式化成Json数组对象，然后在Processors插件中拆分成多条数据，以减少系统的IO次数，提高性能。数据传输采用LogHub。Logtail写入数据后，Blink直接消费数据。您只需要设置合理的分区数即可。分区数必须大于或等于Blink读任务的并发数，避免Blink出现空闲任务。预处理预处理主要由Blink实现。主要设计和优化点：①编写高效的计算流程Blink是一个有状态的流计算框架，非常适合实时聚合、Join等操作。在我们的应用中，只需要关注在错误请求上调用相关服务链接即可。因此，整个日志处理流程分为两个流程：服务的请求入口日志作为一个单独的流程进行处理，过滤掉请求错误的数据。其他中间环节的调用日志作为另一个独立的流程进行处理，错误服务所依赖的请求数据过滤是通过与上述流程JoinOnTraceid连接实现的。如上图双流join后，输出的是请求错误相关的所有链路的完整数据。②设置合理的State生命周期。Blink在做Join的时候，本质上是通过State缓存中间数据状态，然后进行数据匹配。但是如果State的生命周期过长，数据膨胀会影响性能。如果State的生命周期太短，将无法关联一些延迟数据。因此，需要合理配置State生命周期。对于此应用程序，允许的最大数据延迟为1分钟。使用niagara作为statebackend，并设置state数据生命周期，单位毫秒state.backend.type=niagarastate.backend.niagara.ttl.ms=60000③EnableMicroBatch/MiniBatchMicroBatch和MiniBatch都是微批处理，但是triggermicro-batch的机制略有不同。原则上缓存一定量的数据，然后触发处理，减少对State的访问，从而显着提高吞吐量，减少输出数据量。启用joinblink.miniBatch.join.enabled=true使用microbatch时需要保持以下两个minibatch配置blink.miniBatch.allowLatencyMs=5000防止OOM，每个batch最多缓存多少条数据blink.miniBatch.size=20000④Dynamic-Rebalance替换RebalanceBlink任务运行时最忌讳的就是计算热点的存在。为了保证数据被平均使用，DynamicRebalance可以根据每个Subpartition中累积的Buffer数量，选择一个负载较轻的Subpartition进行写入，从而达到动态负载均衡。与静态Rebalance策略相比，当下游任务的计算能力不平衡时，可以更加平衡各个任务的相对负载，从而提升整个作业的性能。启用动态加载task.dynamic.rebalance.enabled=true⑤自定义输出插件数据关联后，统一请求链路上的数据需要作为数据包通知下游图分析节点。传统的方式是通过消息服务传递数据。但是，通过消息服务有两个缺点：它的吞吐量仍然远远落后于内存数据库，如RDB（可能差一个数量级）。在接收端，需要根据traceid进行数据关联。我们通过自定义插件将数据异步写入RDB，同时设置数据过期时间。在RDB中写入时，只使用traceid作为消息内容通过MetaQ通知下游计算服务，大大降低了MetaQ的数据传输压力。GraphAggregationComputingCep/Graph计算服务节点收到MetaQ的通知后，会根据请求的链路数据和依赖的环境监测数据，实时生成诊断结果。诊断结果简化为如下形式：说明请求是由于下游JVM线程池满导致的，但一次调用无法说明服务不可用的根本原因。需要分析整体的错误情况，那么需要对图数据进行分析，做实时聚合。聚合设计如下（简化说明基本思想）：首先，利用Redis的Zrank能力，根据服务名称或IP信息，为每个节点分配一个全局唯一的排序编号。为图中每个节点生成对应的图节点代码，以及代码格式。对于头节点：头节点序列号|圆形时间戳|节点代码。对于普通节点：|编号时间戳|节点ID。由于每个节点在一个时间段内都有唯一的Key，因此可以将节点编码作为Key，使用Redis对每个节点进行统计。同时消除并发读写问题。使用Redis中的Set集合可以很方便地叠加图的边缘。记录根节点，可以通过遍历还原聚合图结构。聚合结果大致如下：这样，最终生成了服务不可用的整体原因，可以通过统计叶子节点来对根本原因进行排序。营收系统上线后，整个实时处理数据链路时延不超过3秒。在闲鱼服务器上定位问题的时间从十几分钟甚至更长的时间缩短到5秒以内。问题定位效率大大提高。预计当前系统可支撑闲鱼每秒千万级的数据处理能力。后续自动定位问题的服务可能会扩展到阿里内部更多的业务场景，随之而来的是数据量的指数级增长，对效率和成本提出了更好的要求。我们未来可能做的改进：可以自动减少或压缩处理后的数据。复杂的模型分析计算也可以在Blink中完成，减少IO，提高性能。支持多租户数据隔离。
