Vector 如何使用 Lua 转换器解析 PostgreSQL csvlog 格式的 CSV 日志文件-育师

Vector 如何使用 Lua 转换器解析 PostgreSQL csvlog 格式的 CSV 日志文件

【免费下载链接】vectorA high-performance observability data pipeline.项目地址: https://gitcode.com/GitHub_Trending/vect/vector

当 PostgreSQL 以csvlog格式输出日志时，日志文件里每一行是一个完整的事件字符串，而其中真正的message字段本身又是按逗号分隔的 CSV。Vector 的内置解析器不覆盖这种“行内再套一层 CSV”的结构。这个任务的操作路径是：用filesource 读取 CSV 日志文件，再用luatransform 加载外部的lua-csv模块，把message字段按 PostgreSQL csvlog 的列顺序拆分成独立的字段，最后通过consolesink 输出结构化事件。

开始前需要两个前提（来源文档列出的 Pre-requisites）：了解 Lua transform 的基本概念，以及了解如何搭建一条 Vector 管道。

准备输入与初始管道

假设要读取的日志由 PostgreSQL 的csvlog产生，例如log.csv（文档示例）：

2020-04-09 12:48:49.661 UTC,,,1,,localhost.1,1,,2020-04-09 12:48:49 UTC,,0,LOG,00000,"ending log output to stderr",,"Future log output will go to log destination ""csvlog"".",,,,,,,"" 2020-04-09 12:48:49.669 UTC,,,27,,localhost.1b,1,,2020-04-09 12:48:49 UTC,,0,LOG,00000,"database system was shut down at 2020-04-09 12:48:25 UTC",,,,,,,,,"" 2020-04-09 12:48:49.683 UTC,,,1,,localhost.1,2,,2020-04-09 12:48:49 UTC,,0,LOG,00000,"database system is ready to accept connections",,,,,,,,,""

先写一版初始配置vector.yaml：filesource 读取所有*.csv文件，luatransform 暂时只做透传（emit(event)），consolesink 以 JSON 编码输出。这是后续判断“解析前长什么样、解析后长什么样”的基线。

data_dir: "." sources: file: type: "file" include: ["*.csv"] ignore_checkpoints: true transforms: lua: inputs: ["file"] type: "lua" version: "2" hooks: process: | function (event, emit) -- to be expanded emit(event) end sinks: console: inputs: ["lua"] type: "console" encoding: codec: "json"

在vector.yaml所在目录运行vector --config vector.yaml，此时每行日志会原样落进message字段。文档示例输出如下（时间戳以实际运行为准）：

{"file":"log.csv","host":"localhost","message":"2020-04-09 12:48:49.661 UTC,,,1,,localhost.1,1,,2020-04-09 12:48:49 UTC,,0,LOG,00000,\"ending log output to stderr\",,\"Future log output will go to log destination \"\"csvlog\"\".\",,,,,,,\"\"","timestamp":"2020-04-09T14:33:28Z"}

说明解析尚未发生：整行 CSV 只是一个字符串。

引入 lua-csv 模块

实际解析依赖单文件模块lua-csv。文档给出的做法是把它下载到vector.yaml所在目录（需要可访问该下载地址的网络环境）：

curl -o csv.lua https://raw.githubusercontent.com/geoffleyland/lua-csv/d20cd42d61dc52e7f6bcb13b596ac7a7d4282fbf/lua/csv.lua

然后在 transform 的source配置段中用 Lua 的require函数加载它：

source: | csv = require("csv")

source段在 Vector 启动时（或在配置自动重载后新增该luatransform 时）执行一次，之后csv就成了全局可用的模块变量。require的查找路径由search_dirs选项控制：如果设置了search_dirs，Vector 会在其中的绝对路径里查找；如果没有设置，就使用配置文件所在的目录。把csv.lua放在vector.yaml同目录下，正是利用了这个默认行为。

编写完整的解析 transform

把解析逻辑放进hooks.process，完整 transform 如下。column_names是按 PostgreSQL 文档列出的 csvlog 列名顺序，建议放在source段里只创建一次，而不是每个事件都重建，以加快速度。

transforms: lua: inputs: ["file"] type: "lua" version: "2" source: | csv = require("csv") -- load external module for parsing CSV column_names = { -- a sequence containing CSV column names "log_time", "user_name", "database_name", "process_id", "connection_from", "session_id", "session_line_num", "command_tag", "session_start_time", "virtual_transaction_id", "transaction_id", "error_severity", "sql_state_code", "message", "detail", "hint", "internal_query", "internal_query_pos", "context", "query", "query_pos", "location", "application_name", -- available only in postgres > 13, to remove for postgres <= 13 "backend_type", "leader_pid", "query_id" } hooks: process: | function (event, emit) fields = csv.openstring(event.log.message):lines()() -- parse the `message` field event.log.message = nil -- drop the `message` field for column, column_name in ipairs(column_names) do -- iterate over column names value = fields[column] -- get field value event.log[column_name] = value -- set the corresponding field in the event end emit(event) -- emit the transformed event end

处理逻辑逐行看：

csv.openstring(event.log.message):lines()()把message字段解析为 CSV 列数组fields（Lua 序列从 1 开始索引）；
event.log.message = nil删掉原始字段——在 Lua 数据模型中，把字段设为nil即表示删除该字段；
用ipairs遍历column_names，把第 N 列的值写入事件字段column_names[N]，即每个 CSV 位置都获得一个语义化的列名；
emit(event)发出变换后的事件。

一个版本相关的限制需要在这里处理：序列末尾的backend_type、leader_pid、query_id三列只在 PostgreSQL 高于 13 的版本中出现在 csvlog 输出里，如果日志来自 PostgreSQL 13 或更低版本，要把这三列从column_names中移除，否则后面的列会整体错位。

验证解析结果

再次运行vector --config vector.yaml，consolesink 会输出结构化事件。文档示例输出（时间戳以实际运行为准）：

{"application_name":"","backend_type":"not initialized","command_tag":"","connection_from":"","context":"","database_name":"","detail":"","error_severity":"LOG","file":"log.csv","hint":"Future log output will go to log destination \"csvlog\".","host":"localhost","internal_query":"","internal_query_pos":"","leader_pid":"","location":"","log_time":"2020-04-09 12:48:49.661 UTC","message":"ending log output to stderr","process_id":"1","query":"","query_id":"0","query_pos":"","session_id":"localhost.1","session_line_num":"1","session_start_time":"2020-04-09 12:48:49 UTC","sql_state_code":"00000","timestamp":"2020-04-09T19:49:07Z","transaction_id":"0","user_name":"","virtual_transaction_id":""}

对第一个事件做美化后可以看到字段结构：log_time、error_severity、message、hint等列名各自独立，原样整行的message已被替换为 csvlog 的message列内容：

{ "application_name": "", "backend_type": "not initialized", "command_tag": "", "connection_from": "", "context": "", "database_name": "", "detail": "", "error_severity": "LOG", "file": "log.csv", "hint": "Future log output will go to log destination \"csvlog\".", "host": "localhost", "internal_query": "", "internal_query_pos": "", "leader_pid": "", "location": "", "log_time": "2020-04-09 12:48:49.661 UTC", "message": "ending log output to stderr", "process_id": "1", "query": "", "query_id": "0", "query_pos": "", "session_id": "localhost.1", "session_line_num": "1", "session_start_time": "2020-04-09 12:48:49 UTC", "sql_state_code": "00000", "timestamp": "2020-04-09T19:49:07Z", "transaction_id": "0", "user_name": "", "virtual_transaction_id": "" }

判断标准：输出事件里出现log_time、error_severity等独立字段、且不再存在整行 CSV 字符串，就说明luatransform 的解析已生效。

后续处理：字段类型与多行字符串

解析完成后，文档给出了两个明确的改进方向，都属于可选的后续步骤。

转换字段类型。CSV 解析出来的列默认全部是字符串。要改成其他数据类型，可以直接在 Lua 代码里用tonumber等内置函数转换；也可以在luatransform 之后追加coercertransform，例如用它来解析log_time这类时间戳字段。

处理多行字符串。CSV 允许字符串里包含换行，但filesource 默认每行生成一个独立事件，多行值会被截断。文档给出两条路线：简单情况下用filesource 的multiline配置项；复杂情况下在 Lua 代码里把多个事件的消息条件式拼接起来，可参考 custom aggregations with Lua 指南。

限制与取舍说明

lua transform 的参考文档明确标注了它的设计定位：luatransform 比remaptransform 慢约 60%，官方建议尽可能使用remap，lua只用于remap覆盖不到的边缘场景。本文的 csvlog 场景正是这种“行内嵌套 CSV、需要外部模块”的定制解析需求，才落到luatransform 上。luatransform 内嵌的是 Lua 5.4 引擎，source段只执行一次，因此适合在其中做column_names这类一次性初始化。

完整操作路径可对照仓库中的 Parsing CSV logs with Lua 指南：建基线配置确认原始输出 → 下载并require加载lua-csv模块 → 在source/hooks.process中实现按列解析（按 PostgreSQL 版本裁剪列名）→ 用 console 输出验证字段结构，再按需追加类型转换或多行拼接。

【免费下载链接】vectorA high-performance observability data pipeline.项目地址: https://gitcode.com/GitHub_Trending/vect/vector

创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考