作者 观测云 系统开发工程师 李国壮
前言
日志采集系统的执行过程，从 “定位日志” 开始，然后是 “数据采集和处理”，最后则是 “同步采集状态”。本文主要介绍第二项，即数据的采集和解析，其中包含了很多细节处理，将会影响到采集效率、解析结果等各个方面。
数据采集和解析
读取数据并分割成行
提到读取日志数据，大部分情况都会先想到类似 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}"
如果不进行处理，不删除颜色字符，那么最终日志数据也会带有 \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"} -
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拼接之后的日志数据是
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
如果没有多行处理，那么最终数据就是以上 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
可以看到，现在每行日志数据都以 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
pipeline 脚本：
grok(_, "%{date:time} %{NOTSPACE:status} %{GREEDYDATA:msg}")
default_time(time)
最终结果：
{
"message": "2020-10-23 06:41:56,688 INFO demo.py 1.0",
"msg": "demo.py 1.0",
"status": "info",
"time": 1603435316688000000
}
注意：Pipeline 切割后的 status 字段是 INFO，但是 DataKit 有做映射处理，所以严谨起见显示为小写的 info
Pipeline 是日志数据处理最后一步，DataKit 会使用 Pipeline 的结果构建行协议，序列化对象并准备打包发送给 Dataway。
结尾
日志数据处理的细节又多又杂，在此只简单描述了一些基础的执行策略，后续会再进行内容补全。
