Java服务_ELK日志采集与可视化链路建设实战经验

一、分布式链路追踪的前世今生

1.起源

1.1 服务化框架 –> 微服务架构的演进

产生新事物的原因一定是新事物的优越性和旧事物的缺陷两方面导致。传统的服务架构已经无法处理「服务的用户量逐渐增加、大规模高并发请求」的问题。

微服务架构（分布式系统）：

微服务架构并不是为了拆分而拆分，拆分微服务的目的是通过对微服务进行水平扩展解决传统的单体应用在业务急剧增长时遇到的问题，

而且由于拆分的微服务系统中专业的人做专业的事，人员和项目的职责单一、低耦合、高内聚，所以产生问题的概率就会降到最小。

1.2 微服务架构与传统单体框架的对比示意

微服务的架构图示例：

从图中可以看出 :

• 微服务把每一个职责单一的功能放在一个独立的服务中。
• 每个服务运行在一个单独的进程中。
• 每个服务有多个实例在运行，每个实例可以运行在容器化的平台内，达到平滑伸缩的效果。
• 每个服务都有自己的储存。
• 每个服务都有自己的运营平台，每个服务都高度自治，内部的变化对外透明。
• 每个服务都可根据性能需求独立地进行水平伸缩。

传统单体架构的伸缩架构：

对比微服务架构与传统单体架构，传统单体架构特点：

• 传统单体架构将所有模块化组件混合后运行在同一个JVM进程中。
• 可对包含多个模块化组件的整体JVM进程进行水平扩展，而无法对某个模块化组件进行水平化扩展。
• 某个模块化组件发生变化时，需要所有的模块化组件进行编译、打包和上线。
• 久而久之，模块间的依赖将会不清晰，相互耦合、相互依赖成为家常便饭。

1.3 分布式系统与微服务的优势

• 扩展性： 分布式系统天然具备“按需扩展”的能力，比如双11大促前通过添加机器实现快速水平扩容，大促结束后释放机器，充分利用云计算的分时复用能力，节约成本。利用微服务，还可以实现按需精准扩容，比如登录服务扩容10倍，下单服务扩容3倍，最大化的节省资源。
• 可靠性： 分布式系统可以有效抵抗“单点风险”，不会因为某一个节点的故障，影响整体的服务可用性。结合流量调度、离群实例摘除和弹性扩容等技术，甚至可以实现故障自愈。
• 可维护性： 分布式系统的可维护性更强，一方面我们将一个复杂服务拆分成多个简单的微服务，每一个微服务的逻辑都更加清晰、更易理解。就好比我们写代码，将一个几百行的复杂函数重构成若干个简单函数，代码可读性就会直线上升。另一方面，一些通用的微服务可以被高度复用，无需重复开发和维护，比如你在开发一个电商 APP，可以直接调用第三方提供的支付、物流等服务接口，整体开发和维护效率将大幅提升。

1.4 分布式系统与微服务引入的问题

• 模糊性： 随着系统的分布式程度越来越高，异常表象与根因之间的逻辑联系变得愈加模糊，问题诊断的难度急剧上升。比如A、B两个服务共享同一个数据库实例，当A服务在压测期间，大量占用数据库的服务端连接和计算资源，会导致B服务出现连接超时或响应变慢等问题。如果B服务是通过C服务间接依赖该数据库实例，问题的定位就会变得更加困难。
• 不一致： 虽然分布式应用从总体上变的更加可靠，但是每一个独立节点的状态却难以保证。导致这种不一致的原因有很多，比如前文提到的单机故障这种预期外的不一致，或者应用Owner执行分批发布或流量灰度时导致的预期内行为不一致。这种不一致性导致我们难以确定一个用户请求在系统内的准确执行路径与行为逻辑，可能引发不可预知的逻辑灾难。
• 去中心化： 当你的系统拥有上千个微服务镜像运行在数百台机器实例上，你该如何梳理它们之间的依赖关系，又该如何找到核心业务的关键执行路径？特别是在分布式的场景下，你没有一个中心化的节点（Master）来保存每个服务之间的依赖与调度状态，每个独立节点都在自行其是，无法分辨自己在整个系统中的位置，只能“盲人摸象、管中窥豹”，缺乏全局视图。

2.诞生

为了应对分布式环境下的不一致、模糊性等问题，人们试图通过请求粒度的轨迹追踪与数据透传，实现节点间的确定性关联，分布式链路追踪技术也由此诞生。

2.1 里程碑事件：Google Dapper

分布式链路追踪诞生的标志性事件就是 Google Dapper 论文的发表。2010年4月，Benjamin H. Sigelman 等人在 Google Technical Report 上发表了《Dapper, a Large-Scale Distributed Systems Tracing Infrastructure》，揭开了分布式链路追踪的技术大幕，开启了一段全新的技术浪潮。

Dapper 首先明确了分布式链路追踪的两个目标：任意部署和持续监测。进而给出了三个具体的设计准则：

• 低开销： 确保核心系统不会因为额外的性能开销拒绝使用。
• 应用级透明： 对应用开发透明，无需开发人员的协助，降低接入门槛，提高迭代效率。
• 可扩展： 在未来相当长一段时间内，随着业务的高速发展，仍然可以有效运转。

下面几张图展示了 Dapper 对链路透传、调用链结构和数据采集流程的描述，对 Dapper 感兴趣的同学建议直接阅读原作。

Dapper 论文有两个重要的意义，

1. 详细阐述了分布式链路追踪的设计理念，为后来的实现者提供了重要的理论指导；
2. 通过 Dapper 在 Google 生产环境的大规模落地实践，证明了分布式链路追踪技术的企业级价值，为分布式链路追踪的推广作出了不可磨灭的贡献。

2.2 基本原理

分布式链路追踪并不是无中生有、凭空诞生的新概念，而是轨迹追踪在 IT 领域的又一次成功运用。轨迹追踪的理念早已被广泛应用于社会生活的方方面面，比如物流订单追踪。

一个快递包裹会在发件站点被赋予一个快递单号，沿途的中转节点会记录该快递到达的时间等信息，而用户通过快递单号就可以查询自己的包裹途径了哪些站点，耗时多久，是否存在滞留或丢件的情况。

下表对比了物流追踪与链路追踪的关联与差异性。

分布式链路追踪的基本原理就是在分布式应用的接口方法上设置一些观察点（类似快递中转站记录点），然后在入口节点给每个请求分配一个全局唯一的标识 TraceId（类似快递单号），当请求流经这些观察点时就会记录一行对应的链路日志（包含链路唯一标识，接口名称，时间戳，主机信息等）。最后通过 TraceId 将一次请求的所有链路日志进行组装，就可以还原出该次请求的链路轨迹，如下图所示。

分布式链路追踪实现请求回溯的关键点有两个：

1. 一是低成本、高质量的观察点设置，也就是链路插桩，确保我们追踪的信息足够丰富，能够快速定位异常根因；
2. 二是保证链路上下文在不同环境下都能够完整透传，避免出现上下文丢失导致的断链现象。

下面，我们来看一个高速公路的例子，进一步加深对链路追踪实现原理的认识。

一辆汽车飞驰在高速公路上

小明、小红、小玉计划在“五一”期间去自驾游，他们的旅游路线各不相同。如果我们想追踪他们的行程轨迹与时间该如何实现？

可能你会建议在每辆车上安装一个追踪器，确实，这是一种行之有效的方法。但是，当出行车辆扩展到全国数以十亿计的规模，安装追踪器的成本就会很高。此时，让我们换个角度思考一下，高速公路的路线是固定的，每隔一段距离就会有一个收费站，如果我们在每个收费站上安装监控，记录车辆在每个收费站的轨迹与时间，就可以很经济的实现车辆轨迹与行驶时间的追踪。最终，我们得到了如下所示的行程记录：

游客	行程路线	行驶距离	行驶时间
小明	北京 -> 石家庄 -> 郑州 -> 西安	1140 公里	13 小时 34 分钟
小红	北京 -> 天津 -> 济南 -> 南京 -> 杭州	1280 公里	14 小时 33 分钟
小玉	北京 -> 天津 -> 济南 -> 南京 -> 上海	1234 公里	13 小时 53 分钟

如果我们将每个游客替换为服务请求，收费站替换为服务接口，那我们就可以得到每次请求在分布式系统中的调用轨迹与状态，这就是分布式链路追踪的含义。

3.基础术语

虽然分布式链路追踪的实现方式多种多样，不同开源或商业化产品都有自己的数据模型和定义。但是仍然有一些基础术语在业界具备广泛的共识，以 OpenTracing 为例。

3.1 Trace

一条 Trace 代表一次入口请求在 IT 系统内的完整调用轨迹及其关联数据集合。其中，全局唯一的链路标识TraceId，是最具代表的一个属性。

通过 TraceId 我们才能将同一个请求分散在不同节点的链路数据准确的关联起来，实现请求粒度的“确定性关联”价值。这也是 Trace 区别于 Metrics、Log 其他两类可观测技术的关键属性。

3.2 Span

光有 TraceId 还不够，请求在每一跳的接口方法上执行了什么动作，耗时多久，执行状态是成功还是失败？承载这些信息的基础对象就是 Span。通常一个完整的 Span 具有如下属性：

• Operation Name：描述了当前接口的行为语义，比如 /api/createOrder 代表执行了一次创建订单的动作。
• SpanId/ParentSpanId：接口调用的层级标识，用于还原 Trace 内部的层次调用关系。
• Start/FinishTime：接口调用的开始和结束时间，二者相减就是该次调用的耗时。
• StatusCode：响应状态，标识当次调用是成功或失败。
• Tags & Events：调用附加信息，详见下面的描述。

3.3 Tags

SpanName 的描述通常是高度抽象的，仅仅回答这个接口是做什么的。如果需要进一步记录请求的行为特征，可以使用 Tags 来扩展语义。Tags 是一组由 {Key:Value} 组成的键值对集合，描述这一次接口调用的具体属性，比如将 UserType 添加到 Tags 中，就可以观察某一类用户（比如 VIP 用户）的链路行为状态。如果将设备类型加到 Tags 中，可以对比不同设备的性能差异。

由于 Tags 只支持结构化的 KV 键值对，因此，它可以作为标签添加到聚合后的链路指标中，有效提升监控告警的数据精度。更准确的回答异常或性能问题发生的原因，比如集中在某个地域、设备或版本。

3.4 Logs

Tags 会随着链路上下文自动向下游透传，如果希望记录一些不需要透传的事件信息，可以使用 Logs 字段。每个 Span 都可以进行多次 Logs 操作，但每个 Logs 对象都需要带有一个时间戳，Logs 的内容可以是非结构化的复杂对象。

为了节省成本，一般不会对 Logs 字段建立索引，也不支持 Logs 的查询或统计，仅仅作为附加信息关联在调用链上，用于单请求诊断。

下图展示了一个 OpenTracing 的 Span 示例，不同开源实现的链路模型我们将在后续章节再展开介绍。

4.分布式链路追踪的应用

狭义上的分布式链路追踪（Tracing），是指跟踪请求在分布式系统中的流转路径与状态，主要用途是协助开发运维人员进行故障诊断、容量预估、性能瓶颈分析与调用链路梳理等工作。技术实现上包含了数据埋点、采集、存储、分析、可视化等环节，形成了一套完整的技术体系。

而更广义的分布式链路追踪，则涵盖了由数据透传能力衍生的生态系统，比如全链路压测、微服务流量路由、业务场景链路拆分等。我们可以为调用链路赋予业务语义，也可以将一次调用生命周期内的所有数据进行关联整合，不再局限于链路数据本身。

由此可见，分布式链路追踪的应用场景广阔，潜力巨大，它的核心属性就是“关联”。然而，分布式链路追踪（Tracing）相对于统计指标（Metrics）和应用日志（Logging）来说更加难以理解，不容易运用，更难用好。接下来，我们通过一个生动形象的例子，了解下分布式链路追踪的经典用法，加深对它的技术本质的掌握。

游客、收费站和交通局

分布式链路追踪的用法有很多，但是最经典、最常用的有三种，还是以上一节的高速公路为例，不同角色对应着不同的用法。

• 游客，只关心自身的行程路线，需要途经哪些收费站点？行驶时间有多长？沿途是否有拥堵或危险路段等。
• 收费站，只关心自身站点的状态，比如站点吞吐量、平均过闸时间等，以便于提前安排检票口值班人数。
• 交通局，会将所有的出行记录汇总，提前估算整个高速公路网的出行流量、易拥堵路段、事故多发路段等，以便于提前疏通或加固问题路段，并给出合理的建议出行路线，有时还需要提前制定车辆限流策略等。

分布式链路追踪的应用和行程轨迹追踪类似，游客关心的是单次请求的轨迹回溯，收费站关注的是服务接口维度的汇总统计，旅游局则类似全局链路拓扑梳理。

4.1 单请求轨迹回溯

单请求轨迹回溯是分布式链路追踪最基础的功能，它记录了一次请求经过的所有服务节点以及对应的节点状态信息（接口名称、耗时、状态码等），这就好比记录了游客自驾游时经过的所有收费站，以及沿途的路况与行驶时间等信息。单请求轨迹回溯是诊断特定请求异常/超时原因的有效手段，可以快速定位异常节点（拥堵的收费站）。

比较成熟的 Tracing 产品（比如阿里云的 ARMS）除了基础的链路数据外，还会记录请求出入参、本地方法栈、关联 SQL 与异常堆栈等信息。这些细节信息就好比车辆的型号大小、驾驶员驾龄、是否醉酒、沿途每一路段的详细路况等，当调用不符合预期（行程异常）时，就可以精准的定位根因，如下图所示:

4.2 服务监控

假如你是收费站的站长，你会关注哪些信息？收费站的车辆吞吐量？平均的过闸时间？车辆的来源与去向？同理，每一个服务节点，将途经的所有调用信息汇总后，就可以得到当前服务接口的吞吐量、耗时、来源与去向等统计指标。这些指标可以帮助我们快速识别当前服务的健康状态。在实际生产系统中，还可以与告警系统结合，实现风险的快速识别与处理，降低业务损失。

4.3 链路拓扑

假如你是交通局的局长，你可能会关注全国高速公路网的整体运行状态，有哪些易拥堵或事故多发路段与站点，如何确保春运高峰期核心路段运行通畅，不会出现重大交通瘫痪事件等等。此时，你需要对所有的车辆行程轨迹进行汇总分析。

同理，链路拓扑就是将全局或某一入口服务的所有调用链路进行汇总，聚合为链路拓扑大图，进而分析当前链路的性能瓶颈点、易故障点等，提前进行性能优化或风险防控，还可以根据历史流量来指导未来（比如双11大促）的容量评估。

5.分布式链路追踪的发展现状

截止到 2021年，分布式链路追踪（Tracing）已经成为主流软件架构不可或缺的基础技术之一，它与指标（Metrics）、日志（Logging）并称为可观测领域的“三驾马车”，它们三者之间的关系如下图所示：

随着 Kubenetes 容器技术与云计算的普及，未来的软件架构会更加趋向分布式云、微服务化的方向，软件开发、部署成本将大幅下降，但是系统维护和问题诊断的难度却急剧上升。因此，分布式链路追踪以及由它提供的“确定性关联”价值将愈加凸显，如下图所示：

Tracing 在开源社区也颇受喜爱，拥有着旺盛的生命力，主流的开源标准包括 OpenTelemetry、OpenTracing、OpenCensus 和国内使用较多的 SkyWalking。其他影响力较强的实现还包括 Jaeger、Zipkin、Pinpoint等，如下图所示。

在商业化领域，Tracing 与 APM（Application Performance Mornitoring） 密切绑定，绝大部分厂商会面向应用视角提供更加全面、易用的 APM 服务，而不仅仅是 Tracing 服务。参考 2021 年 Gartner 评测机构给出的 APM 魔力象限，可以大致评估各大厂商的 APM 与 Tracing 产品能力，如下图所示。

截止 2021年，阿里巴巴 98% 的 Java 应用（上万级别）均已接入内部自研的分布式链路追踪系统 EagleEye；阿里云上有近万家企业在持续使用 ARMS 提供的分布式链路追踪服务。而从整个业界来看，无论是谷歌、亚马逊这样的国际大厂，还是新兴的互联网公司，或是传统企业，都在大规模接入和应用分布式链路追踪技术，Tracing 生态正在蓬勃发展。

6.分布式链路追踪的挑战与限制

作为一门新兴技术，分布式链路追踪的技术演进史并不算长，仅有十数年。目前，它仍处于不断被探索、快速迭代的周期。为了更好的了解与应用分布式链路追踪技术，我们来看下它目前面临的几项关键挑战与限制。

6.1 关键挑战与应对

分布式链路追踪技术从诞生到大规模应用，中间经历了一段较长的蛰伏期，直到近几年才逐渐被大家广泛接受和认可。影响其快速推广的关键挑战包括如下几点：

• 前期建设成本高： 无论是在不同组件接口上进行插桩埋点，还是保证链路上下文能够正确传播，亦或是搭建一套稳定可靠的链路数据后端处理系统，都不是一件易事，需要投入大量的研发人力。
• 数据处理成本高： 由于链路数据与请求流量成正比，每一次请求都会记录相应的链路日志，当系统流量爆炸式增长，相应的链路数据生成、采集、处理、存储、查询的成本也会急剧上升，带来巨大的 IT 资源开销。
• 价值没有得到普遍认可： 基础的链路数据仅仅表达了接口间的调用依赖，没有释放足够的业务价值，难以得到领导和同事们的全力支持。
• 链路标准不统一： 分布式链路追踪发展前期没有统一的业界标准，各家厂商百花齐放，虽然一定程度上促进 Tracing 技术的多元化探索，但也为链路融合、迁移和推广带来了巨大的挑战。

当然，挑战同样也是机遇，为了应对上述问题，分布式链路追踪近几年实现了如下技术突破：

• 无侵入探针 + 一体化解决方案： 类似 JavaAgent 的探针插桩技术，实现了0代码入侵，0改造成本的链路自动埋点，而类似 SkyWalking 的开源实现还提供了端到端的一体化解决方案，从链路数据生成到最后的可视化，中小企业可以快速搭建并享受到分布式链路追踪技术的价值，大幅降低了 Tracing 的前期建设成本和接入门槛。
• 链路采样 + 边缘计算： 链路采样策略，例如固定比例采样、限流采样、错慢全采、自定义标签采样等，可以大幅降低链路数据的传输、处理、存储成本；结合用户网络内的指标聚合，长文本编码/压缩等边缘计算技术，可以合理控制分布式链路追踪的数据成本，保障链路系统持续健康运转。
• 关联分析 + 立体化可观测： 单条链路的价值难以凸显，但是基于成千上万条链路的聚合/关联分析却能快速定位，导致系统异常的关键因素，比如版本、地域、用户类型等。同时，结合业务、容器、基础设施等其他层面的可观测数据，建立一套端到端、立体化的可观测体系，能够更加有效地释放分布式链路追踪的技术价值。
• 开源标准趋向统一： 自从 2019 年 OpenTelemetry 开源立项，得到了两大主流开源实现 OpenTracing 和 OpenCensus 的大力支持，开启了可观测性的新时代。虽然，目前 OpenTelemetry 仅在 Tracing 领域拥有比较完善的技术标准，Metrics 和 Logging 仍在探索阶段，但是可观测性“三驾马车”融合一统的趋势已经势不可挡。未来基于统一完善的可观测数据标准，分布式链路追踪的“确定性关联”将得到更加广泛的应用。

6.2 现阶段能力限制

分布式链路追踪现有的模型设计与实现，可以有效满足许多经典场景的分布式诊断诉求。但是，仍然有大量场景超出了现阶段分布式链路追踪的能力范畴，需要我们去探索更好的方案。

树形 YES！图形 NO！

分布式链路追踪是通过 ParentSpanId 和 SpanId 来标识依赖关系，从而准确还原链路层级与顺序。但是，每个 Span 有且仅有一个 ParentSpanId，这就限制了所有链路形态只能是单个父节点的树形结构，而不能是多个父节点的图形结构。

某些系统为了提供重复调用的效率，会将多次 RPC 调用打包成一次 RPC 调用合并发送，这种入度大于1的图形结构，就无法通过调用链真实还原调用状态，而是会被拆成多条调用链，如下图所示：

人工插桩 YES！智能插桩 NO！

无论是 SDK 或是 Agent 模式，目前工业界的链路插桩主要是依赖人工发现插桩点并实现插桩过程，很难通过算法自适应的实现插桩点的智能发现。然而，学术界在这方面已经进行了一些有意思的探索，虽然在性能开销、安全等方面还不够成熟，但是值得关注。

2019 年波士顿大学发表了一篇研究智能插桩的文章，他们实现的 Pythia 原型系统针对性能退化问题，可以自动发现更有价值的内部插桩点。例如，我们在请求一个存储系统时，可能会直接命中缓存快速返回结果，也可能未命中缓存导致加载磁盘花费了较多时间。我们仅在 RPC 层面进行插桩，只能看到请求耗时高低起伏，呈现一种双峰式的分布，但无法确认根因是什么。Pythia 通过比对分析不同的链路数据，会自动发现影响性能的潜在插桩点，比如慢请求可能会额外调用一次 fetchFromDisk 方法，从而更清晰的解释影响请求耗时的根因，如下图所示。

分布式链路追踪的能力限制远不止以上两种场景，在离线分析、机器学习等多个领域也等待我们去探索攻克。我们既要充分发挥现有的分布式链路追踪技术价值，解决当下的企业运维困难；同时也要把视野放宽，在未来更多的领域中去拓展分布式链路追踪的边界。

二、数据服务日志统一建设接入组件

日志采集流程：日志->Filebeat->Logstash→Elasticsearch→Kibana or Grafana

ELK架构示意图

ELK部署架构

1.应用接入filebeat（需接入应用改造）

1.1 概述

应用接入filebeat，需要对JDOS工程做以下一次性的操作。

1.2 带filebeat的镜像

接入filebeat的前提是部署用的镜像中有filebeat。

可选镜像

在进行容器镜像的编译时，可选以下两种类型基础镜像。

镜像名称	作用	其他
datamill9n_java_filebeat	java application模式使用
datamill9n_tom_filebeat	tomcat模式使用

镜像关联步骤

步骤一：路径：镜像管理 - 镜像市场

步骤二：搜索 “datamill9n”

步骤三：关联需要使用的景象

基础镜像使用

在构建镜像时-在运行镜像时需要选择上述关联的基础镜像

1.3 日志采集配置

容器变量

变量名称 ： LOG_PROFILE

变量值格式：应用英文标识_应用环境信息，如”dataasset_prod”

配置文件

文件路径：/etc/filebeat/filebeat.yml

文件内容：

filebeat.prospectors:
  - type: log
    paths:
      - /export/Logs/data-asset-service*.log
    #handle with Exception
    multiline.pattern: '^20\d{2}.*$'
    multiline.negate: true
    multiline.match: after
    #handle profile
    fields:
      profile: LOG_PROFILE
      #profile:
    fields_under_root: true
filebeat.registry_file: /export/Data/filebeat/configs/registry
output.logstash:
  hosts: ["11.104.70.56:5044"]
  loadbalance: true

1.4 filebeat的启停

start.sh 启动脚本

在对应的java进程启动之后，额外新增启动filebeat的步骤

#################################### filebeat start
#export container args to env args
count=`ps -ef |grep filebeat |grep -v "grep" |wc -l`
if [ 0 == $count ];then
    echo START filebeat
    export $(cat /proc/${cpid}/environ |tr '\0' '\n' | grep LOG_PROFILE | xargs)
    echo $LOG_PROFILE
    mkdir -p /export/filebeat
    cp /etc/filebeat/filebeat.yml /export/filebeat/filebeat.yml
    sed -i "s/LOG_PROFILE/${LOG_PROFILE}/g" /export/filebeat/filebeat.yml
    #run filebeat
    nohup /usr/share/filebeat/bin/filebeat -e -c /export/filebeat/filebeat.yml > /dev/null 2>&1 &
    echo "filebeat start success."
fi
#################################### filebeat end

全部脚本举例

#!/usr/bin/env bash
#userdir
SHDIR=$(cd "$(dirname "$0")"; pwd)
echo current path:$SHDIR
 
#
PIDFILE="./start.pid"
if [ -f $PIDFILE ]; then
    if kill -0 `cat $PIDFILE` > /dev/null 2>&1; then
        echo server already running as process `cat $PIDFILE`.
        exit 0
    fi
fi
 
# exec
BASEDIR=`dirname $0`/..
LOGFILE=/export/Logs/xxx.log
 
curl -s "http://pfinder-master.jd.com/access/script" -o /tmp/pfinder.sh ; source /tmp/pfinder.sh || :
java -version
nohup java ${PFINDER_AGENT:-} -XX:MaxRAMFraction=1 -XX:MaxRAM=2g -XX:+HeapDumpOnOutOfMemoryError -XX:ErrorFile=/export/Logs/hs_err.log -XX:HeapDumpPath=/export/Logs/heap_dump.hprof $JAVA_OPTS -jar $BASEDIR/xxx.jar --spring.config.location=file:/home/export/App/bin/application.yml  > /dev/null 2>&1 &
 
 
# wirte pid to file
if [ $? -eq 0 ]
then
    cpid=$!
    echo "pid is ${cpid}"
    #################################### filebeat start
    #export container args to env args
    count=`ps -ef |grep filebeat |grep -v "grep" |wc -l`
    if [ 0 == $count ];then
        echo START filebeat
        export $(cat /proc/${cpid}/environ |tr '\0' '\n' | grep LOG_PROFILE | xargs)
        echo $LOG_PROFILE
        mkdir -p /export/filebeat
        cp /etc/filebeat/filebeat.yml /export/filebeat/filebeat.yml
        sed -i "s/LOG_PROFILE/${LOG_PROFILE}/g" /export/filebeat/filebeat.yml
        #run filebeat
        nohup /usr/share/filebeat/bin/filebeat -e -c /export/filebeat/filebeat.yml > /dev/null 2>&1 &
        echo "filebeat start success."
    fi
    #################################### filebeat end
    if /bin/echo -n ${cpid} > "$PIDFILE"
    then
        sleep 5
        echo STARTED SUCCESS
    else
        echo FAILED TO WRITE PID
        exit 1
    fi
    (timeout 60 tail -f -n0 $LOGFILE &) | grep -q "Started Application"
else
    echo SERVER DID NOT START
    exit 1
fi
exit 0

stop.sh 停止脚本

停止脚本额外新增

############################################# filebeat stop
# kill filebeat
pkill -9 filebeat
echo STOP FILEBEAT
############################################# filebeat stop

1.5 日志格式-统一日志格式与打印形式

日志格式配置

application.yml

# 日志配置相关
logging:
  pattern:
    level: ' %X{uuid:-uuid} %X{erp:-erp} %X{PFTID:-PFTID} %5p'
  file:
    name: /export/Logs/data-asset-service.log

日志打印encoder配置

需要定义日志组件内部的encoder

<encoder class="com.jd.datamill9n.common.logger.PatternLayoutEncoder">
    <pattern>${FILE_LOG_PATTERN}</pattern>
    <charset>utf-8</charset>
</encoder>

完整配置举例：

<?xml version="1.0" encoding="UTF-8"?>
<configuration>
    <include resource="org/springframework/boot/logging/logback/defaults.xml"/>
    <include resource="org/springframework/boot/logging/logback/console-appender.xml"/>
 
    <springProperty scope="context" name="LOG_FILE_NAME" source="logging.file.name"/>
    <property name="LOG_FILE_PATH" value="${LOG_FILE_NAME}"/>
 
    <!--    文件输出日志-->
    <appender name="ROLLING-FILE" class="ch.qos.logback.core.rolling.RollingFileAppender">
        <file>${LOG_FILE_PATH}</file>
        <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
            <fileNamePattern>${LOG_FILE_PATH}.%d{yyyy-MM-dd}.%i.log</fileNamePattern>
            <timeBasedFileNamingAndTriggeringPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP">
                <maxFileSize>200MB</maxFileSize>
            </timeBasedFileNamingAndTriggeringPolicy>
        </rollingPolicy>
        <append>true</append>
        <encoder class="com.jd.datamill9n.common.logger.PatternLayoutEncoder">
            <pattern>${FILE_LOG_PATTERN}</pattern>
            <charset>utf-8</charset>
        </encoder>
    </appender>
 
    <!-- console输出日志 -->
    <appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
        <encoder class="com.jd.datamill9n.common.logger.PatternLayoutEncoder">
            <pattern>${CONSOLE_LOG_PATTERN}</pattern>
            <charset>utf-8</charset>
        </encoder>
    </appender>
 
    <root level="INFO">
        <appender-ref ref="CONSOLE"/>
        <appender-ref ref="ROLLING-FILE"/>
    </root>
</configuration>

日志打印新方式

经过上述“日志打印encoder配置”之后，我们打印日志在原有的基础上，可以解析下面这种日志的打印方式。

LOGGER.info("消息内容，不包括参数值。 keys={k1, k2,k3}", v1, v2, v3)；
LOGGER.error("出错消息内容，不包括参数值。 keys={k1, k2,k3}", v1, v2, v3, e)；

其中keys={k1, k2,k3}为固定pattern。k1,k2,k3为参数的名称，v1,v2,v3为k1,k2,k3对应的值 。注意顺序一一对应。

举例：

logger.info("Test with param. keys={k1,k2, k3}", "v1", 2, 3L);

输出的结果为：

2019-09-17 09:51:01.728 INFO 52531 --- [ main] c.j.datamill9n.common.logger.TestLogger : Test with param. param={"k1":"v1","k2":2,"k3":3}

日志为什么会输出这个样子，具体原因稍后再表。

2.logstash集群部署及解析规则配置（接入应用无需关注）

2.1 集群部署

Docker镜像部署集群。 推荐书籍 ： 《深入浅出Docker》→ 《Kubernetes快速入门》 作者：Nigel，Poulton（奈吉尔·波尔顿）

Dockfile内容

FROM centos
RUN yum install -y wget && yum clean all
RUN mkdir /home/logstash
copy . /home/logstash
RUN ls /home/logstash
#RUN wget 'https://artifacts.elastic.co/downloads/logstash/logstash-6.8.0.tar.gz'
RUN wget -O logstash-6.8.0.tar.gz 'http://storage.jd.local/mirror/logstash-6.8.0.tar.gz?Expires=3772414887&AccessKey=CYVc16EdCiyJnawA&Signature=L3hs%2B5dvaDSYpGuZ3HJT3STfEco%3D'
RUN tar -xvf  logstash-6.8.0.tar.gz -C /home/logstash
RUN rm -rf logstash-6.8.0.tar.gz
RUN ls /home/logstash
RUN mv /home/logstash/logstash-6.8.0/config/logstash.yml /home/logstash/logstash-6.8.0/config/logstash-copy.yml
RUN mv /home/logstash/logstash/logstash.yml /home/logstash/logstash-6.8.0/config
RUN ls /home/logstash/logstash-6.8.0/config
WORKDIR /home/logstash
RUN chmod 777 logstash-6.8.0/bin
RUN ls /home/logstash
RUN echo $PATH
ENTRYPOINT ./logstash-6.8.0/bin/logstash -f ./logstash/logstash.conf
#ENTRYPOINT ./logstash-6.8.0/bin/logstash -f ./logstash/logstash.conf >> /export/Logs/logstash.log

GIT地址：https://coding.jd.com/datamill9n/jdos-images/

其他12种可用镜像：

2.2 logstash配置

logstash配置解析内容

以下为logstash配置内容

input {
    beats {
        port => "5044"
        client_inactivity_timeout => 3600
    }
 
    #stdin{}
}
 
filter {
 
    dissect {
        mapping => {"profile" => "%{doc_type}_%{profileName}"}
    }
     
    grok {
            match => {
                "message" => "%{TS:ts}\s+%{UUID:uuid}\s+%{ERP:erp}\s+%{PFTID:traceId}\s+%{LOGLEVEL:loglevel}\s+%{PID:pid}\s+---\s+\[%{PNAME:thread}\]\s+%{CLASSNAME:classname}\s+\:\s+%{CONTEXT:context}"
            }
 
            pattern_definitions => {
                "TS" => "\d{4}-\d{2}-\d{2} \d{2}\:\d{2}\:[0-9\.]+"
            }
 
            pattern_definitions => {
                "UUID" => "[a-zA-Z0-9\-_]+"
            }
            pattern_definitions => {
                "PFTID" => "[a-zA-Z0-9\.]+"
            }
            pattern_definitions => {
                "ERP" => "[a-zA-Z0-9]+"
            }
            pattern_definitions => {
                "LOGLEVEL" => "(INFO|DEBUG|ERROR|WARN)"
            }
            pattern_definitions => {
                "PID" => "\d+"
            }
            pattern_definitions => {
                "PNAME" => "[_a-zA-Z\-\d&\.\s\/]+"
            }
            pattern_definitions => {
                "CLASSNAME" => "[a-zA-Z\d_$\.\/\[\]]+"
            }
            pattern_definitions => {
                "CONTEXT" => ".+"
            }
        }
 
        if "_grokparsefailure" in [tags] {
            mutate {
                replace => {
                    "doc_type" => "logs_grok_failure"
                }
            }
        } else {
            mutate {remove_field => ["message"]}
 
            grok {
                #(.*?)param=(.*)[\n]*([\s\S]*)
                match => {"context" => "%{DATA:prefix}param=%{JSON_DETAIL_PATTERN:details}[\n]*%{ANY_STRING_INCLUDE_WARP:ex}"}
                pattern_definitions => {"ANY_STRING_INCLUDE_WARP" => "[\s\S]*"}
                pattern_definitions => {"JSON_DETAIL_PATTERN" => "{.*}"}
            }
 
            if [prefix] or [details] {
             
                mutate {
                    replace => { "json_details" => '{"%{doc_type}": %{details}}' }
                }
              
                json {
                    source => "json_details"
                    target => "params"
                }
              
                mutate {rename => ["prefix", "message"]}
                mutate {remove_field => ["details", "json_details", "context"]}
            } else {
                mutate {rename => ["context", "message"]}
            }
        }
         
        if "beats_input_codec_plain_applied" in [tags] {
            mutate {
                remove_tag => ["beats_input_codec_plain_applied"]
            }
        }
 
        mutate {
            add_field => {
                "ip_address" => "%{[@metadata][ip_address]}"
            }
        }
 
        # 2019-05-31 10:21:23.873
        date {
            match => ["ts", "yyyy-MM-dd HH:mm:ss.SSS"]
            target => "timestamp"
            timezone => "Asia/Shanghai"
        }
 
        mutate {remove_field => ["ts", "offset", "@metadata", "debug", "version", "type", "beat", "prospector", "host", "profile","input","log"]}
 
}
 
output {
    if [doc_type] and [profileName] {
        stdout {codec => dots}
        #stdout{codec => rubydebug{metadata => true}}
        #ouput to es
        if [profileName] =~ /.*?prod.*?/ {
            elasticsearch {
                action => "index"
                hosts => ["es-nlb-es-87wgaecdzm.jvessel-open-hb.jdcloud.com:9200"]
                user => "elastic"
                password => "Es-logstash"
                index => "bdaa_service_log_prod_v1_%{+YYYY.MM}"
                document_type => "log"
            }
        } else {
            elasticsearch {
                action => "index"
                hosts => ["es-nlb-es-87wgaecdzm.jvessel-open-hb.jdcloud.com:9200"]
                user => "elastic"
                password => "Es-logstash"
                index => "bdaa_service_log_pre_v1_%{+YYYY.MM}"
                document_type => "log"
            }
        }
    } else {
        stdout {codec => rubydebug {metadata => true}}
    }
}

2.3 实际举例

数据来源

日志举例

2021-04-23 14:36:24.321  5be059e26ce558058a32-178fd7195ed njr21397101 erp 45242.36153.16191597842474811 datamill  INFO 159 --- [JSF-BZ-22000-82-T-11] c.j.d.c.s.i.JsfLogMessageInterceptor     : JSF invoked.  param={"receiveTime":1619159784319,"args":["njr21397101"],"headers":{"1":60000,"7":1660},"costTime":1,"response":{"status":0,"result":{"id":747748,"lastUpdateTime":1614248349424,"status":"STARTED","pin":"njr21397101","name":"njr21397101","role":{"id":622070,"name":"基础角色","desc":"支持基础功能","order":-1,"startTime":"2000-02-01 00:00:00","expirationTime":"2100-02-01 00:00:00","available":true},"company":"柏丽德珠宝（广州）有限公司","mainSubRoleType":"MAIN","roleType":"EXTERNAL_BASIC","channels":[],"channelNames":[],"isvAccount":false,"virtualAccount":false,"healthAccount":false,"brandCount":1,"basicAccount":true,"advancedAccount":false,"thirdPartyAccount":false,"brandMember":[],"available":true,"startDate":"2021-02-25","endDate":"2022-01-01","createdTime":"2021-02-25"}},"methodName":"findPin","alias":"AccountAuthorizeService-4.prod","className":"com.jd.datamill.auth.service.open.AccountAuthorizeService","receiveTimeStr":"2021-04-23 14:36:24.320","invokedTime":"2021-04-23 14:36:24.321"}

es中的存储

{
  "_index" : "ads_polaris_datamill9n_log_prod_v3_2021.04",
  "_type" : "log",
  "_id" : "T4xx_XgBIDfoGaLsobt-",
  "_score" : 0.0,
  "_source" : {
    "message" : "JSF invoked.  ",
    "appId" : "datamill",
    "@version" : "1",
    "profileName" : "prod3",
    "thread" : "JSF-BZ-22000-82-T-11",
    "ip_address" : "10.175.121.217",
    "doc_type" : "auth",
    "loglevel" : "INFO",
    "uuid" : "5be059e26ce558058a32-178fd7195ed",
    "@timestamp" : "2021-04-23T06:36:24.604Z",
    "source" : "/export/Logs/datamill.log",
    "erp" : "erp",
    "timestamp" : "2021-04-23T06:36:24.321Z",
    "pid" : "159",
    "traceId" : "45242.36153.16191597842474811",
    "params" : {
      "auth" : {
        "className" : "com.jd.datamill.auth.service.open.AccountAuthorizeService",
        "receiveTime" : 1619159784319,
        "args" : [
          "njr21397101"
        ],
        "headers" : {
          "7" : 1660,
          "1" : 60000
        },
        "costTime" : 1,
        "receiveTimeStr" : "2021-04-23 14:36:24.320",
        "invokedTime" : "2021-04-23 14:36:24.321",
        "response" : {
          "result" : {
            "channelNames" : [ ],
            "startDate" : "2021-02-25",
            "name" : "njr21397101",
            "healthAccount" : false,
            "createdTime" : "2021-02-25",
            "available" : true,
            "company" : "柏丽德珠宝（广州）有限公司",
            "roleType" : "EXTERNAL_BASIC",
            "advancedAccount" : false,
            "status" : "STARTED",
            "mainSubRoleType" : "MAIN",
            "thirdPartyAccount" : false,
            "lastUpdateTime" : 1614248349424,
            "role" : {
              "startTime" : "2000-02-01 00:00:00",
              "order" : -1,
              "expirationTime" : "2100-02-01 00:00:00",
              "id" : 622070,
              "name" : "基础角色",
              "available" : true,
              "desc" : "支持基础功能"
            },
            "brandMember" : [ ],
            "virtualAccount" : false,
            "endDate" : "2022-01-01",
            "channels" : [ ],
            "id" : 747748,
            "pin" : "njr21397101",
            "isvAccount" : false,
            "brandCount" : 1,
            "basicAccount" : true
          },
          "status" : 0
        },
        "alias" : "AccountAuthorizeService-4.prod",
        "methodName" : "findPin"
      }
    },
    "pin" : "njr21397101",
    "classname" : "c.j.d.c.s.i.JsfLogMessageInterceptor",
    "tags" : [ ]
  }
}

3.grafana看板的配置

3.1 设计自己想看的面板以及打印对应日志

LOGGER.info("auth validate invoke process success. keys={authValidateCostTime,invokeCostTime}",
           authValidateEndTime - authValidateStartTime, authInvokeEndTime - authValidateEndTime);

3.2 配置grafana看板

3.3 ES中的日志存储

POST _sql
{
  "sql" : "select * from bdaa_uds_log_prod_v1_202207* where doc_type = 'uds' and message = 'auth validate invoke process success.' limit 3"
}
 
ES返回结构
{
  "took" : 167,
  "timed_out" : false,
  "_shards" : {
    "total" : 145,
    "successful" : 145,
    "skipped" : 0,
    "failed" : 0
  },
  "hits" : {
    "total" : 1676,
    "max_score" : 0.0,
    "hits" : [
      {
        "_index" : "bdaa_uds_log_prod_v1_20220705",
        "_type" : "log",
        "_id" : "N531zYEBklnJeavdQh5M",
        "_score" : 0.0,
        "_source" : {
          "profileName" : "prodbi",
          "ip_address" : "10.184.12.54",
          "message" : "auth validate invoke process success.  ",
          "@timestamp" : "2022-07-05T10:43:25.428Z",
          "method" : "c.j.g.core.aop.UnionAuthValidateAspect.around",
          "doc_type" : "uds",
          "erp" : "erp",
          "lineNum" : "258",
          "timestamp" : "2022-07-05T10:43:22.000Z",
          "loglevel" : "INFO",
          "thread" : "JSF-BZ-22000-404-T-17",
          "loginErp" : "loginErp",
          "params" : {
            "uds" : {
              "authValidateCostTime" : 255,
              "invokeCostTime" : 1008
            }
          },
          "@version" : "1",
          "traceId" : "1341457.36523.16570178010864966",
          "source" : "/export/Logs/uds/uds.log",
          "tags" : [ ]
        }
      },
      {
        "_index" : "bdaa_uds_log_prod_v1_20220705",
        "_type" : "log",
        "_id" : "ApvPzYEBklnJeavd3H6y",
        "_score" : 0.0,
        "_source" : {
          "erp" : "erp",
          "tags" : [ ],
          "@version" : "1",
          "doc_type" : "uds",
          "params" : {
            "uds" : {
              "invokeCostTime" : 1021,
              "authValidateCostTime" : 239
            }
          },
          "loginErp" : "loginErp",
          "message" : "auth validate invoke process success.  ",
          "@timestamp" : "2022-07-05T10:02:34.586Z",
          "timestamp" : "2022-07-05T10:02:31.000Z",
          "profileName" : "prodbi",
          "source" : "/export/Logs/uds/uds.log",
          "method" : "c.j.g.core.aop.UnionAuthValidateAspect.around",
          "thread" : "JSF-BZ-22000-404-T-16",
          "ip_address" : "11.16.99.144",
          "lineNum" : "258",
          "loglevel" : "INFO",
          "traceId" : "724602.36523.16570153504936099"
        }
      },
      {
        "_index" : "bdaa_uds_log_prod_v1_20220705",
        "_type" : "log",
        "_id" : "P5vPzYEBklnJeavdyX1K",
        "_score" : 0.0,
        "_source" : {
          "profileName" : "prodbi",
          "ip_address" : "10.188.19.225",
          "message" : "auth validate invoke process success.  ",
          "@timestamp" : "2022-07-05T10:02:34.616Z",
          "method" : "c.j.g.core.aop.UnionAuthValidateAspect.around",
          "doc_type" : "uds",
          "erp" : "erp",
          "lineNum" : "258",
          "timestamp" : "2022-07-05T10:02:31.000Z",
          "loglevel" : "INFO",
          "thread" : "JSF-BZ-22000-404-T-18",
          "loginErp" : "loginErp",
          "params" : {
            "uds" : {
              "authValidateCostTime" : 259,
              "invokeCostTime" : 1005
            }
          },
          "@version" : "1",
          "traceId" : "724611.36523.16570153504917852",
          "source" : "/export/Logs/uds/uds.log",
          "tags" : [ ]
        }
      }
    ]
  }
}

4.调用链路追踪实现

4.1 现阶段组件化

http与java服务的传输

@Override
public Object run() throws ZuulException {
    // 因为下面会调用MDC.clear()整体清理MDC中的内容,会导致pfinder在这个filter之前往MDC中设置的PFTID也会被清理,所以在这里暂时性的将PFTID取出之后再set进MDC,后续有更好的办法会更新这部分逻辑
    String traceId = null;
    if (MDC.getCopyOfContextMap() != null && MDC.getCopyOfContextMap().containsKey("PFTID")) {
        traceId = MDC.get("PFTID");
    }
    //1.清空MDC等threadLocal信息
    MDC.clear();
    RpcContext.getContext().clearAttachments();
    LoginContext.remove();
 
    //2.往mdc中set uuid/PFTID
    MDC.put("PFTID", traceId);
    RequestContext ctx = RequestContext.getCurrentContext();
    String uuid = ctx.getRequest().getHeader("uuid");
    MDC.put("uuid", uuid);
 
    //3.记录请求的startTime以及复位记录标志位
    long startTime = System.currentTimeMillis();
    ctx.set("startTime", String.valueOf(startTime));
    return null;
}

java服务之间的传输

JsfMdcInterceptor依赖与使用

<!--JSF配置 jsfFilter-->
<!--    MDC相关，用来获取来自于client的uuid、pin、erp等信息-->
<bean id="jsfMdcInterceptor" class="com.jd.datamill9n.common.spring.interceptor.JsfMdcInterceptor" scope="prototype"/>
<bean id="jsfLogMessageInterceptor" class="com.jd.datamill9n.common.spring.interceptor.JsfLogMessageInterceptor" scope="prototype"/>
 
<jsf:server id="date-asset-service-jsf-server" protocol="jsf"/>
 
<jsf:provider id="dataAssetMetricService"
  interface="com.jd.bdaa.arch.provider.service.DataAssetMetricService"
  alias="dataAssetMetricService-#{environment.VERSION == null ? '${jsf.alias.version:}' : environment.VERSION}#{environment.PROFILE == null ? '.${jsf.alias.profile:}' : '.'+environment.PROFILE}"
  ref="dataAssetMetricServiceImpl"
  serialization="hessian"
  timeout="20000"
  server="date-asset-service-jsf-server"
  filter="jsfMdcInterceptor,jsfLogMessageInterceptor">
</jsf:provider>

maven引入

<common.spring.version>1.3.1</common.spring.version>
<!-- jsf-filter组件：jsf filter通过配置jsf provider filter在日志中输出jsf服务被调用时的信息 -->
<dependency>
    <groupId>com.jd.datamill9n</groupId>
    <artifactId>common.spring</artifactId>
    <version>${common.spring.version}</version>
</dependency>

代码分析

public class JsfMdcInterceptor extends AbstractFilter {
 
    private static final Logger LOGGER = LoggerFactory.getLogger(JsfMdcInterceptor.class);
 
    private static final List<String> TRANS_ARGS_LIST = Arrays.asList("uuid", "pin", "erp", "PFTID");
 
    @Override
    public ResponseMessage invoke(RequestMessage request) {
        boolean isProvider = RpcContext.getContext().isProviderSide();
        boolean isConsumer = RpcContext.getContext().isConsumerSide();
        Invocation invocationBody = request.getInvocationBody();
        if (LOGGER.isDebugEnabled()) {
            LOGGER.debug("start transform uuid from RequestMessage. keys={isProvider,isConsumer,mdcMap,rpcMap}",
                         isProvider, isConsumer, MDC.getCopyOfContextMap(), invocationBody.getAttachments());
        }
        // 当作为服务提供者的角色时。RpcContext --> MDC
        if (isProvider) {
            Map<String, Object> mdcMapObj = invocationBody.getAttachments();
            if (mdcMapObj != null && mdcMapObj.size() > 0) {
                mdcMapObj.forEach((k, v) -> {
                    MDC.put(k, String.valueOf(v));
                });
            }
        }
        // 当作为服务消费者的角色时。MDC --> RpcContext
        if (isConsumer) {
            Map<String, String> mdcMap = MDC.getCopyOfContextMap();
            if (mdcMap != null && mdcMap.size() > 0) {
                mdcMap.forEach((k, v) -> {
                    if (TRANS_ARGS_LIST.contains(k)) {
                        invocationBody.addAttachment(k, v);
                    }
                });
            }
        }
        if (LOGGER.isDebugEnabled()) {
            LOGGER.debug("after transform uuid. keys={isProvider,isConsumer,mdcMap,rpcMap}",
                         isProvider, isConsumer, MDC.getCopyOfContextMap(), invocationBody.getAttachments());
        }
        try {
            return getNext().invoke(request);
        } finally {
            if (LOGGER.isDebugEnabled()) {
                LOGGER.debug("after invoke service. keys={isProvider,isConsumer,mdcMap,rpcMap}",
                             isProvider, isConsumer, MDC.getCopyOfContextMap(), invocationBody.getAttachments());
            }
            MDC.clear();
        }
    }
}

线程池之间的传输

// 引入方法同上
public class MdcTaskDecorator implements TaskDecorator {
    @Override
    public Runnable decorate(Runnable runnable) {
        Map<String, String> map = MDC.getCopyOfContextMap();
        return () -> {
            try {
                MDC.setContextMap(map);
                runnable.run();
            } finally {
                MDC.clear();
            }
        };
    }
}

服务与存储引擎

SELECT 1 settings log_comment = 'b2b56851715795ff92d1-181d32973a1';

参考链接： https://github.com/ClickHouse/ClickHouse/issues/18494

4.2 未来JavaAgent化

JavaAgent特性

• 完全非侵入式的进行代码埋点，进行系统监控
• 修改JAVA底层源码，进行JVM自定义
• 实现AOP动态代理

参考文档： https://www.jrebel.com/blog/how-write-javaagent

三、查询决策整体方案介绍

1.行业解决方案对比分析详情

在技术上这个属于“业务链路追踪”范畴的事情。在分布式场景下，链路追踪主流实现方式包括两类，一类是基于日志的ELK方案，一类是基于单次请求调用的会话跟踪方案，但是这两种方案均不适用于我们这种复杂的业务场景。

1.1 传统的ELK方案

方案流程：

1. 开发者编写程序时尽可能的打印日志；
2. 应用程序打印的日志被filebeat收集发送给logstash集群；
3. logstash解析日志存入ES集群。

排查问题：

根据uuid从ES集群搜索相关日志并分析。如果用户使用该种方式排查问题的话，该方式有如下痛点，无法实施；

1. 搜索慢：日志一般都是非结构化的文本，ES虽然能查，但涉及到分词并且频繁查询会很慢；
2. 噪音多：搜出来的日志可能会很多，也没有重点；对用户无疑大海捞针；
3. 分析难：搜索到的数据都是一条条离散的数据，必须充分懂业务逻辑，才能由人工对日志进行串联分析，尽可能地还原出请求的现场，对用户要求太高。

1.2 分布式链路追踪方案

在分布式系统，尤其是微服务系统中，业务场景的某次请求往往需要经过多个服务、多个中间件、多台机器的复杂链路处理才能完成。分布式链路追踪方案就是解决这种问题的，分布式链路追踪诞生的标志性事件就是 Google Dapper 论文的发表。市面上的同类型框架几乎都是以Google Dapper论文为基础进行实现，整体大同小异，集团内部的Pfinder也是。实现原理都是通过一个分布式全局唯一的id（即traceId），将分布在各个服务节点上的同一次请求串联起来，还原调用关系、追踪系统问题、分析调用数据、统计系统指标。

Dapper 首先明确了分布式链路追踪的两个目标：任意部署和持续监测。进而给出了三个具体的设计准则：
低开销： 确保核心系统不会因为额外的性能开销拒绝使用。
应用级透明： 对应用开发透明，无需开发人员的协助，降低接入门槛，提高迭代效率。
可扩展： 在未来相当长一段时间内，随着业务的高速发展，仍然可以有效运转。

分布式链路追踪是一种会话级别的追踪能力，主要作用是分析分布式系统的调用行为，并不能很好地应用于业务逻辑的追踪。举例：

一个审核业务场景的追踪案例，业务系统对外提供审核能力，待审对象的审核需要经过“初审”和“复审”两个环节（两个环节关联相同的taskId），因此整个审核环节的执行调用了两次审核接口。如图左侧所示，完整的审核场景涉及众多“业务逻辑”的执行，而分布式会话跟踪只是根据两次RPC调用生成了右侧的两条调用链路，并没有办法准确地描述审核场景业务逻辑的执行

指标服务平台的指标生产-消费链路同理，早晨6点生产的指标数据，和下午3点查询指标的请求不仅跨会话，对比上面例子更加复杂。所以这套方案有以下弊端：

1、无法同时追踪多条链路

会话级别的追踪。使用traceId串联，不同链路之间相互独立，给完整的业务追踪增加了难度。

1. 生产和消费链路独立；
2. 前端看板页面一次刷新，多次请求；

2、无法聚焦于当前业务系统的逻辑执行

分布式会话跟踪覆盖了单个请求流经的所有服务、组件、机器等等，不仅包含当前业务系统，还涉及了众多的下游服务。这些大部分都是噪音：

分布式链路追踪主要作用是分析分布式系统间的调用行为，解决的是服务和服务间、服务和中间件间的情况，但是对于我们来说服务内部各种优化决策逻辑和执行逻辑对用户才是最重要的。举个例子是分布式链路追踪只能看清人和人之间的情况，但是我们的业务场景是想看到每个人下面的毛细血管的情况。

总结

传统的ELK方案：

需要事后从大量离散的日志中搜集和筛选出需要的日志，并人工进行日志的串联分析，其过程必然耗时耗力。

分布式链路追踪：

在调用执行的同时，完成了链路的动态串联，但由于是会话级别且仅关注于调服务间用关系等问题，导致其无法很好地应用于业务追踪。

2.整体流程

3. 挑战点

3.1 SDK稳定性

目的：

• 不能影响主链路的正常使用，影响主要体现在自身错误发生时不影响、性能上不影响；
• 开箱即用，自动埋点和无侵入；

措施：

1. 大量的错误隔离机制：
2. 链路数据的批量异步传输：

结果：

1. 性能基准测试：接入SDK耗时在ns内；
2. 实际工程压测：影响可忽略不计；

3.2 跨会话的分布式链路串联

目的：让整体业务链路(生产、消费、页面)跨会话跨集群跨时间联系起来。

措施：单个会话复用分布式链路追踪技术做链路染色，完整的业务链路采用业务标识串联。

主要体现在以下几个场景上：

3.2.1 页面一次刷新发起多个查询

实现方式：调用方在调用时可传入batchId这一业务标识，batchId代表一次页面刷新事件标识，该次页面刷新请求的查询batchId都一样。

3.2.2 一次查询对应的SQL在CK集群扫描了多少行数据、用了多少内存、命中那些索引

实现方式：使用ClickHouse的log_comment能力可以将确定uuid的查询与CK执行的某次SQL关联起来，并结合CK的system.query_log、system.query_thread_log等表获取对应SQL执行用到的索引、扫描的数据行数、使用到的内存。

3.2.3 一次查询使用的数据是那个生产任务生产的，该生产任务表现怎么样？

实现方式：请求对应的集群 + 其他因素（时间、分区等）–> 对应生产任务及实例

3.3 高效的链路数据传输和查询

目的：链路数据低成本传输 + 高性能查询

措施：优化改造传统的ELK链路，解决日志非结构化、避免ES分词查询等弊端，采用ES嵌套结构+copy to进行关键词精准查询。

3.3.1 SDK结构化的日志打印

用户只需要引用SDK在日志打印中引入业务信息：

2023-11-08 00:00:04.994  uuid erp PFTID  INFO 281 --- [BatchSpanProcessor_WorkerThread-1] b.a.o.p.o.e.DataAssetLoggingSpanExporter : DataAssetLoggingSpanExporter invoke. param={"spanPointKey":"drive_execute_sql","routeContext_traceId":"c853cc931fd1644157332cecdcfee8d2","routeContext_spanId":"1d99cefda2cef59f","routeContext_parentSpanId":"46bcc37f30e1b010","其他链路信息":"······"}

3.3.2 将上述日志中param后的结构化数据放入ES嵌套结构并进行精准查询，提升查询性能

3.3.3 ES关键词精准查询，避免分词查询，大幅提升查询效率

3.4 业务关键点的灵活采集与展示

目的：新服务接入SDK添加新节点后，可观测展示端无需开发即可实时展示。

措施：通过设计SDK链路数据内容并结构化流程化的采集到ES中，可观测服务端拿到数据后，根据链路信息实时组织DAG与表格信息展示

3.4.1 链路数据内容

SDK自动传输的内容如下。以某个请求SQL执行步骤为例：

• 服务信息：SDK自动生成。包含正在处理该请求的服务信息，JDOS应用名 + JSF别名；
• 链路串联信息：SDK自动生成。用来动态生成链路的流程图结构；
• 步骤基础信息：SDK自动生成。包含请求的基础信息（如resAppKey、请求人等）和该步骤的基本信息（如是否成功、耗时多久等）
• 业务信息：用户指定传输。作为步骤额外补充的链路信息；

3.4.2 可观测服务根据“链路串联信息”动态构建流程图的规则文本，并通过组件展示

链路串联信息组合而成的流程图文本举例：

drive_receive_request_fc9: 定义驱动接收请求_编号4
index_service_request_atomic_service_f57: 指标服务请求原子服务_编号3
drive_execute_sql_7f9: 定义驱动执行SQL_编号5
drive_receive_request_0e3: 定义驱动接收请求_编号7
index_service_receive_request_f68: 指标服务接收请求_编号1
index_service_dissemble_8e2: 指标服务拆分:主从_编号2
index_service_request_atomic_service_32d: 指标服务请求原子服务_编号6
drive_execute_sql_7cd: 定义驱动执行SQL_编号8
ck_pub_183: CK183集群
ck_pub_183.shape: stored_data
ck_pub_106: CK106集群
ck_pub_106.shape: stored_data
index_service_request_atomic_service_f57 -> drive_receive_request_fc9
index_service_dissemble_8e2 -> index_service_request_atomic_service_f57: 主查询
drive_receive_request_fc9 -> drive_execute_sql_7f9
index_service_request_atomic_service_32d -> drive_receive_request_0e3
index_service_receive_request_f68 -> index_service_dissemble_8e2
index_service_dissemble_8e2 -> index_service_request_atomic_service_32d: 子查询 {style.stroke-dash: 3}
drive_receive_request_0e3 -> drive_execute_sql_7cd
drive_execute_sql_7f9 -> ck_pub_183
drive_execute_sql_7cd -> ck_pub_106

3.4.3 可观测根据步骤信息依次组织其他信息输出为详情表格

参考文献

Author：mazongguang(京东)

Filebeat介绍

Logstash-介绍

自建elk+filebeat+grafana日志收集平台