论一个优秀的日志采集系统是如何设计和实现数据处理的

作者 观测云 系统开发工程师 李国壮

前言

日志采集系统的执行过程，从 “定位日志” 开始，然后是 “数据采集和处理”，最后则是 “同步采集状态”。本文主要介绍第二项，即数据的采集和解析，其中包含了很多细节处理，将会影响到采集效率、解析结果等各个方面。

数据采集和解析

读取数据并分割成行

提到读取日志数据，大部分情况都会先想到类似 Readline() 这种方法函数，每次调用都会返回完整的一行日志。但是在 DataKit 没有这样实现。

为了确保更细致的操控和更高的性能，DataKit 只使用了最基础的 Read() 方法，每次读取 4KiB 数据（buff 大小是 4KiB，实际读取到可能更少），手动将这 4KiB 数据通过换行符 \n 分割成 N 份。这样会出现两种情况：

• 这 4KiB 数据最后一个字符刚好是换行符，可以分割成 N 份，没有剩余
• 这 4KiB 数据最后一个字符不是换行符，对比上文，本次的分割只有 N-1 份，有剩余部分，这段剩余的部分将补充到下一个 4KiB 数据的首部，依次类推

在 DataKit 的代码中，此处对同一个 buff 不断进行 update CursorPosition、copy 和 truncate，以实现最大化的内存复用。

经过处理，读取到的数据已经变成一行一行，可以走向执行流的下一层也就是转码和特殊字符处理。

转码和特殊字符处理

转码和特殊字符的处理要在数据成型之后再进行，否则会出现从字符中间截断、对一段截断的数据做处理的情况。比如一个 UTF-8 的中文字符占 3 字节，在采集到第 1 个字节时就做转码处理，这属于 Undefined 行为。

数据转码是很常见的行为，需要指定编码类型和大端小端（如果有），本文重点讲述一下 “特殊字符处理”。

“特殊字符” 在此处代指数据中的颜色字符，比如以下命令会在命令行终端输出一个红色 rea 单词：

$ RED='\033[0;31m' && NC='\033[0m' && print "${RED}red${NC}"
1

如果不进行处理，不删除颜色字符，那么最终日志数据也会带有 \033[0;31m，不仅缺乏美观、占用存储，而且可能对后续的数据处理产生负影响。所以要在此处筛除特殊颜色字符。

开源社区有许多案例，大部分都使用正则表达式进行实现，性能不是很出色。

但是对于一个成熟的日志输出框架，一定有关闭颜色字符的方法，DataKit 更推荐这种做法，由日志产生端从避免打印颜色字符。

解析行数据

“解析行数据” 主要是针对容器 Stdout/Stderr 日志。容器 runtime 管理和落盘日志时会添加一些额外的信息字段，比如产生时间，来源是 stdout 还是 stderr，本条日志是否被截断等等。DataKit 需要对这种数据做解析，提取对应字段。

• Docker json-file 日志单条格式如下，是 JSON 格式，正文在 log 字段中。如果 log 内容的结尾是 \n 表示这一行数据是完整的，没有被截断；如果不是 \n，则表明数据太长超过 16KB 被截断了，其剩余部分在下一个 JSON 中。

  {"log":"2022/09/14 15:11:11 Bash For Loop Examples. Hello, world! Testing output.\n","stream":"stdout","time":"2022-09-14T15:11:11.125641305Z"}
  1

• Containerd（CRI）单条日志格式如下，各项字段以空格分割。和 Docker 相同的是，Containerd（CRI）也有日志截断的标记，即第三个字段 P，此外还有 F。P 表示 Partial，即不完整的、被截断的；F 表示 Full。

  2016-10-06T00:17:09.669794202Z stdout P log content 1
  2016-10-06T00:17:09.669794202Z stdout F log content 2
  1
  2

  拼接之后的日志数据是 log content 1 log content 2。

通过解析行数据，可以获得日志正文、stdout/sterr 等信息，根据标记确定是否是不完整的截断日志，要进行日志拼接。在普通日志文件中不存在截断，文件中的单行数据理论上可以无限长。

此外，日志单行被截断，拼接之后也属于一行日志，而不是下文要提到的多行日志，这是两个不同的概念。

数据处理

上文描述的过程，是将数据从文件中读取，并通过解码和解析，组织成一行可读的文本。从现在开始讲针对这一行文本做各种处理，包括多行拼接、字段切割。

多行数据

多行处理是日志采集非常重要的一项，它将一些不符合特征的数据，在不丢失数据的前提下变得符合特征。比如日志文件中有以下数据，这是一段常见的 Python 栈打印：

2020-10-23 06:41:56,688 INFO demo.py 1.0
2020-10-23 06:54:20,164 ERROR /usr/local/lib/python3.6/dist-packages/flask/app.py Exception on /0 [GET]
Traceback (most recent call last):
  File "/usr/local/lib/python3.6/dist-packages/flask/app.py", line 2447, in wsgi_app
    response = self.full_dispatch_request()
2020-10-23 06:41:56,688 INFO demo.py 5.0
1
2
3
4
5
6

如果没有多行处理，那么最终数据就是以上 7 行，和原文一模一样。这不利于后续的 Pipeline 切割，因为像第 3 行的 Traceback (most recent call last): 或 第 4 行的 File "/usr/local/lib/python3.6/dist-packages/flask/app.py", line 2447, in wsgi_app 都不是固定格式。

如果经过有效的多行处理，这 7 行数据会变成 3 行，结果如下：

2020-10-23 06:41:56,688 INFO demo.py 1.0
2020-10-23 06:54:20,164 ERROR /usr/local/lib/python3.6/dist-packages/flask/app.py Exception on /0 [GET]\nTraceback (most recent call last):\n  File "/usr/local/lib/python3.6/dist-packages/flask/app.py", line 2447, in wsgi_app\n    response = self.full_dispatch_request()
2020-10-23 06:41:56,688 INFO demo.py 5.0
1
2
3

可以看到，现在每行日志数据都以 2020-10-23 这样的特征字符串开头，原文中不符合特征的第 3、4、5 行被追加到第 2 行的末尾。这样看起来要美观很多，而且有利于后续的 Pipeline 字段切割。

这一功能并不复杂，只需要指定特征字符串的正则表达式即可。

在 DataKit logging 采集器配置有 multiline_match 项，以上文的例子，该项的配置应该是 ^\d{4}-\d{2}-\d{2}，即匹配形如 2020-10-23 这样的行首。

具体实现上类似一个数据结构中的栈（stack）结构，符合特征就将前一条出栈并再把自己入栈进去，不符合特征就只将自己入栈追加到前一条末尾，这样从外面收到的出栈数据都是符合特征的。

此外，DataKit 还支持自动多行，在 logging 采集器的配置项中是 auto_multiline_detection 和 auto_multiline_extra_patterns，它的逻辑非常简单，就是提供一组的 multiline_match，根据原文遍历匹配所有规则，匹配成功就提高它的权重以便下次优先选择它。

自动多行是简化配置的一种方式，除了用户配置外，还提供 “默认自动多行规则列表”，详情链接见文章末尾。

Pipeline 切割和日志 status

Pipeline 是一种简单的脚本语言，提供各种函数和语法，用以编写对一段文本数据的执行规则，主要用于切割非结构化的文本数据，例如把一行字符串文本切割出多个有意义的字段，或者用于从结构化的文本中（如 JSON）提取部分信息。

Pipeline 的实现比较复杂，它由抽象语法树（AST）和一系列内部状态机、功能纯函数组成，此处不过多描述。

只看使用场景，举个简单的例子，原文如下：

2020-10-23 06:41:56,688 INFO demo.py 1.0
1

pipeline 脚本：

grok(_, "%{date:time} %{NOTSPACE:status} %{GREEDYDATA:msg}")
default_time(time)
1
2

最终结果：

{
    "message": "2020-10-23 06:41:56,688 INFO demo.py 1.0",
    "msg": "demo.py 1.0",
    "status": "info",
    "time": 1603435316688000000
}
1
2
3
4
5
6

注意：Pipeline 切割后的 status 字段是 INFO，但是 DataKit 有做映射处理，所以严谨起见显示为小写的 info

Pipeline 是日志数据处理最后一步，DataKit 会使用 Pipeline 的结果构建行协议，序列化对象并准备打包发送给 Dataway。

结尾

日志数据处理的细节又多又杂，在此只简单描述了一些基础的执行策略，后续会再进行内容补全。