在 GreptimeDB v0.9 版本我们加入了对日志相关的支持：Pipeline 引擎和全文索引。GreptimeDB 致力于成为统一处理指标（Metric）、日志（Log）、事件（Event）和追踪（Trace）的时序数据库。在 v0.9 之前，用户虽然可以写入文本（string）类型的数据，但无法进行专门的解析和查询。有了 Pipeline 引擎和全文索引之后，用户可以直接使用 GreptimeDB 完成日志数据的处理，并极大提升数据的压缩率，并且支持通过模糊查询语法快速检索目标日志数据。
本文会从设计思路出发，简单介绍 GreptimeDB 中 Pipeline 引擎的实现原理和方案步骤。
明确设计目标和优势
提到日志，我们会首先想到一个长字符串。以下是一行非常经典的 nginx access log：
192.168.97.8 - - [15/Oct/2024:08:41:09 +0000] "GET /query/myelosyphilis-anatomicopathologic-polarography-b8be0a5b-8a68-48a4-8a4e-e92f9fcb0a38 HTTP/1.1" 200 664 "https://www.github.com" "Mozilla/5.0 (Windows NT 6.2; WOW64; rv:116.0) Gecko/20100101 Firefox/116.0"
不难发现，虽然这行日志整体是一个大的字符串，但是其中已经包含了一些结构化的信息，例如 IP（192.168.97.8）、时间戳（[15/Oct/2024:08:41:09 +0000]）、请求方法、请求路径、HTTP 协议版本号、HTTP 状态码等等。
尽管日志本质上是非结构化的数据，但在实际应用中，我们常见的日志大多由系统日志中间件打印，而日志中间件通常会在日志的前面附带上一些特定的信息，例如日志的时间戳，日志等级，以及一些特定的标签（例如应用名或者方法名）。
如果我们将这个字符串视为一个整体，保存在数据库的一列中，那么后续只能通过 like 语法进行模糊查询。这样，若用户需要查询某特定请求路径下所有 HTTP 状态码为 200 的日志，过程会十分繁琐且低效。
如果我们可以在接收到日志的时候，直接将日志内容进行解析成不同的列，会使得写入和查询的效率大大提高。有的读者可能已经想到了，这就是 ETL 的流程。目前市面上有一些产品支持将输入的文本行进行转换并输出，但是这些产品大多需要独立部署，也就是在数据库写入流程的前面多部署一个组件。这不仅会带来资源的开销，也提升了运维的复杂度。
到这里，我们稍微明确了我们的目标。我们希望在 GreptimeDB 接收到日志数据库的时候，增加一个简单的处理流程，能使得一个文本行日志，能被提取和转换成不同的数据类型和字段值，并将这些字段值分别保存同一张数据库表的不同列中。
当然最重要的是，这个转换的规则是可以用配置文件来描述的，不同的日志可以用不同的转换规则来处理。
相比于直接保存字符串文本，先解析再入库带来了两个显著的好处：
提取并保存带有语义的数据，提高查询效率。比如我们可以将 HTTP 状态码单独保存成一个列，这样后续需要查询所有状态码是 200 的日志行的时候就会非常方便。
提高数据压缩率。对于文本的压缩，我们可以使用 gzip 等工具得到一个近似极限的压缩率；数据库对于纯文本的压缩和保存大概率不会比这个压缩率更优。而通过解析日志中的字段并转换成对应的数据类型（例如将 HTTP 状态码转换成 int 类型），数据库可以通过 Run-Length Encoding （游程编码）、列式压缩等技术手段，进一步提高压缩率。通过我们的实测，容量占用相比通用文本压缩可以至少降低 50%。
方案设计
我们依然使用上述这条日志作为示例来展示处理流程。
192.168.97.8 - - [15/Oct/2024:08:41:09 +0000] "GET /query/myelosyphilis-anatomicopathologic-polarography-b8be0a5b-8a68-48a4-8a4e-e92f9fcb0a38 HTTP/1.1" 200 664 "https://www.github.com" "Mozilla/5.0 (Windows NT 6.2; WOW64; rv:116.0) Gecko/20100101 Firefox/116.0"
在配置规则的选择和设计上，我们参考了 Elasticsearch 的 Ingest processor。每个 Processor 足够独立，便于扩展；同时我们可以使用 Processor 组合来应对复杂的情况。
Processor 用来将字段进行初步处理，例如切分，从而获得子字符串，例如 HTTP 状态码 "200"。我们希望进一步将字符串转换成更高效的类型，例如数值类型。因此我们需要一种能将解析后的子字符串字段转换成数据库可以支持的数据类型的处理方式。在这里我们需要引入一个简单的内置类型系统和转换处理器 Transform 用来处理这种情况。
根据我们观测到的日志行的结构，我们可以大概写出以下配置规则:
processors:
- dissect:
fields:
- line
patterns:
- '%{ip} %{?ignored} %{?ignored} [%{ts}] "%{method} %{path} %{protocol}" %{status} %{size} "%{referer}" "%{ua}"' - date:
fields:
- ts
formats:
- "%d/%b/%Y:%H:%M:%S %Z"
transform:
- fields:
- status
- size
type: int32
- fields:
- ip
- method
- path
- protocol
- referer
- ua
type: string
- field: ts
type: time
index: time
首先，我们使用 Processor 对数据进行处理。使用 Dissect Processor 将一行日志提取出不同的字段，并使用 Date Processor 将日期时间戳文本解析成 timestamp。然后，我们使用 Transform 将提取解析出来的字段转换成数据库支持的数据类型，例如 int32 和 string。需要注意的是，在 Transform 中设定的字段名会被作为数据库的列名。最后，我们可以在 Transform 中指定字段是否需要设置为索引，即上述配置中的 index: time，会将 ts 字段设置成 GreptimeDB 中的 time index。
数据处理的流程简洁明了，如下图所示:
实现细节
有了方案之后我们就可以开工了！
我们的接口支持一次接受多行日志，即一个日志行的数组。实际上对于数组，我们只需要循环对每一行进行处理即可，并没有什么特殊的操作。我们在下文中还是以上述日志行为例，介绍数据的处理流程。
对于一次处理，我们在空间上将整体逻辑分为两部分：数据空间（上下文）和“代码”。
数据空间是一个上下文，数据在其中通过 key-value 结构存储：每个数据有它的名称（key）和值（value）。数据的初始状态即原始的日志输入行，为了方便我们直接用 JSON 格式来表示数据空间，示意如下：
{
"line": "192.168.97.8 - - [15/Oct/2024:08:41:09 +0000] "GET /query/myelosyphilis-anatomicopathologic-polarography-b8be0a5b-8a68-48a4-8a4e-e92f9fcb0a38 HTTP/1.1" 200 664 "https://www.github.com" "Mozilla/5.0 (Windows NT 6.2; WOW64; rv:116.0) Gecko/20100101 Firefox/116.0""
}
而代码就是我们通过配置文件定义的 Processor 和 Transform。首先我们需要将配置文件进行解析导入到程序中，这部分的逻辑主要是对配置文件的定义解析并加载，不涉及日志的处理，因此不在本文中展开。我们以 Date Processor 为例，代码结构大体如下：
pub struct DateProcessor {
// input fields
fields: Fields,
// the format for parsing date string
formats: Formats,
// optional timezone param for parsing
timezone: Option
}
Processor 最主要的方法如下：
pub trait Processor {
// execute processor
fn exec_field(&self, val: &Value) -> Result<Map, String>;
}
对于每一个 Processor，我们调用 exec_field 方法对数据空间中的数据进行处理。Processor 中记录了配置文件中指定的 field 名称（即数据空间中的 key），因此我们可以通过这个 key 在获取到对应的 value。我们使用 Processor 的代码处理完成这个 value 后，将它重新放置回数据空间中。这样，我们就完成了一个 Processor 的处理流程。
我们以第一个 Dissect Processor 为例，简单描述一下处理流程。在初始状态下，数据空间中只含有原始输入，如下所示：
{
"line": "192.168.97.8 - - [15/Oct/2024:08:41:09 +0000] "GET /query/myelosyphilis-anatomicopathologic-polarography-b8be0a5b-8a68-48a4-8a4e-e92f9fcb0a38 HTTP/1.1" 200 664 "https://www.github.com" "Mozilla/5.0 (Windows NT 6.2; WOW64; rv:116.0) Gecko/20100101 Firefox/116.0""
}
以 Pipeline 中的第一个 Dissect Processor 为例，规则定义如下：
processors:
- dissect:
fields:
- line
patterns:
- '%{ip} %{?ignored} %{?ignored} [%{ts}] "%{method} %{path} %{protocol}" %{status} %{size} "%{referer}" "%{ua}"'
Processor 处理的流程也很简单，我们首先通过 fields 指定的 key 从数据空间中获取对应的值，然后使用该 Processor 来处理这个值，得到一个或者多个输出，最后我们将输出的结果保存回到数据空间中，这个 Processor 就处理完成了。
Dissect Processor 的作用是根据空格或者简单的标点符号对文本进行分割，并将分割形成的子字符串通过格式中指定的 key 名进行关联，最后保存在数据空间中。在这个例子中，首先通过 line 这个 key 从数据空间中获取到对应的值，然后对文本进行切分，将切分后的结果依次赋值给 ip、ts 等字段，最后将这些字段保存回数据空间中。此时数据空间的状态示例如下：
{
"line": "192.168.97.8 - - [15/Oct/2024:08:41:09 +0000] "GET /query/myelosyphilis-anatomicopathologic-polarography-b8be0a5b-8a68-48a4-8a4e-e92f9fcb0a38 HTTP/1.1" 200 664 "https://www.github.com" "Mozilla/5.0 (Windows NT 6.2; WOW64; rv:116.0) Gecko/20100101 Firefox/116.0"",
"ip": "192.168.97.8",
"ts": "15/Oct/2024:08:41:09 +0000",
"method": "GET",
"path": "/query/myelosyphilis-anatomicopathologic-polarography-b8be0a5b-8a68-48a4-8a4e-e92f9fcb0a38",
"protocol": "HTTP/1.1",
"status": "200",
"size": "664",
"referer": "https://www.github.com",
"ua": "Mozilla/5.0 (Windows NT 6.2; WOW64; rv:116.0) Gecko/20100101 Firefox/116.0"
}
cqgn.ousnled.com,cqgn.syshuangyihe.com,cqgn.eyeql.com
cqgn.xyfhm.com,cqgn.nc-lh.com
依次执行定义在规则配置中的 Processor，我们就完成了对数据的处理流程。
经过 Processor 处理的数据，有时候依然不是我们想要的最终结果。例如上面的例子中，虽然我们通过分本切分得到了 status 这个字段，但它依然是一个字符串。如果我们能将它转换成一个数字进行存储，不管是存储效率还是查询效率都会得到提升。我们把这一部分的处理定义为 Transform。
Transform 的结构大致如下：
pub struct Transforms {
transforms: Vec
}
pub struct Transform {
// input fields
pub fields: Fields,
// target datatype for database
pub type_: Value,
// database index hint
pub index: Option
}
同样非常简单明了。对于每个 Transform，同样我们通过 fields 获取到数据空间中的值，转换成 type_ 指定的数据类型，然后放回到数据空间中。
到这一步，我们就完成了对原始输入的非结构化的日志行解析，获得了相对结构化的字段（列）和对应的值。大致结果如下：
{
"line": "192.168.97.8 - - [15/Oct/2024:08:41:09 +0000] "GET /query/myelosyphilis-anatomicopathologic-polarography-b8be0a5b-8a68-48a4-8a4e-e92f9fcb0a38 HTTP/1.1" 200 664 "https://www.github.com" "Mozilla/5.0 (Windows NT 6.2; WOW64; rv:116.0) Gecko/20100101 Firefox/116.0"",
"status": 200,
"size": 664,
"ip": "192.168.97.8",
"method": "GET",
"path": "/query/myelosyphilis-anatomicopathologic-polarography-b8be0a5b-8a68-48a4-8a4e-e92f9fcb0a38",
"protocol": "HTTP/1.1",
"referer": "https://www.github.com",
"ua": "Mozilla/5.0 (Windows NT 6.2; WOW64; rv:116.0) Gecko/20100101 Firefox/116.0",
"ts": 1728981669000000000
}
cqgn.xpdahan.com,cqgn.yubingame.com,cqgn.lhfeshop.com
cqgn.juwanci.com,cqgn.gztdzk.com
至此，我们已经完成了对数据的提取和处理。剩下的只是将这个数据转换成插入请求写入到数据库中了，这部分就不在本文中展开。
感兴趣的小伙伴可以在此查看该版本的代码，了解详细的代码执行过程。
结束语
本文简单介绍了 GreptimeDB v0.9.0 中引入的 Pipeline 引擎的设计思路和实现原理。联想力丰富的读者可以发现整个过程其实是一个非常简单的 interpreter 实现，对此感兴趣的读者可以访问参考此处教程进行进一步的了解。而在实际中我们针对 Pipeline 的执行进行了多次优化和重构，目前的实现相比较于原版的可以说已经是“面目全非”了。对此感兴趣的读者可以期待我们的后续文章。
关于 Greptime
Greptime 格睿科技专注于为可观测、物联网及车联网等领域提供实时、高效的数据存储和分析服务，帮助客户挖掘数据的深层价值。目前基于云原生的时序数据库 GreptimeDB 已经衍生出多款适合不同用户的解决方案，更多信息或 demo 展示请联系下方小助手（微信号：greptime）。
欢迎对开源感兴趣的朋友们参与贡献和讨论，从带有 good first issue 标签的 issue 开始你的开源之旅吧～期待在开源社群里遇见你！添加小助手微信即可加入“技术交流群”与志同道合的朋友们面对面交流哦~
Star us on GitHub Now: https://github.com/GreptimeTeam/greptimedb
官网：https://greptime.cn/
文档：https://docs.greptime.cn/
Twitter: https://twitter.com/Greptime
Slack: https://greptime.com/slack
LinkedIn: https://www.linkedin.com/company/greptime/
