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.yamlfilesource 读取所有*.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_dirsVector 会在其中的绝对路径里查找如果没有设置就使用配置文件所在的目录。把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 列数组fieldsLua 序列从 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.yamlconsolesink 会输出结构化事件。文档示例输出时间戳以实际运行为准{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%官方建议尽可能使用remaplua只用于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），仅供参考