Vector Lua Transform 实战：用 Lua 解析 CSV 日志（以 PostgreSQL csvlog 为例）

Vector Lua Transform 实战用 Lua 解析 CSV 日志以 PostgreSQL csvlog 为例【免费下载链接】vectorA high-performance observability data pipeline.项目地址: https://gitcode.com/GitHub_Trending/vect/vector本指南讲解如何在 Vector 数据管道中使用lua可编程转换transform解析自定义格式的 CSV 日志。全文以 PostgreSQL 的csvlog输出为示例从零搭建一条file源 →lua解析 →console输出的最小管线并逐步引入外部 Lua 模块lua-csv、实现字段映射与类型转换。读完本文你将掌握lua转换 v2 API 的source与hooks.process用法、外部模块加载机制以及处理多行与类型转换的进阶手段。前置条件了解 Lua 转换的基本概念lua转换允许你用 Lua 编程语言改写事件数据。了解 Vector 的基本概念并知道如何搭建一条最简单的数据管道源 → 转换 → 目标。为什么需要可编程转换Vector 内置了大量针对常见结构化格式JSON、GELF、protobuf 等的解析器。但当日志格式是应用自定义的、或者像 PostgreSQLcsvlog这样字段高度专有时内置解析器往往无法直接覆盖。此时lua转换提供了最大的灵活性你可以编写任意 Lua 代码对每个事件进行解析、改写、拆分甚至生成新事件。本指南的场景是读取 PostgreSQL 产生的csvlog文件把它解析为字段齐全的结构化事件。搭建最小管线file → lua → console示例日志PostgreSQL csvlog假设待读取的日志文件log.csv内容如下这是 PostgreSQLcsvlog的典型输出字段用逗号分隔、带引号的字段内部可以包含逗号和引号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,,,,,,,,,注意第三条记录中message字段内部既有逗号又有双引号Future log output will go to log destination csvlog.这正是裸用字符串切分无法正确处理、必须依赖真正 CSV 解析器的原因。初始配置先起草一版 Vector 配置此时lua转换只做透传pass-through用于确认管线本身能跑通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这条配置定义了一个管线file源读取*.csv文件 →lua转换当前原样放行事件→console目标以 JSON 格式打印事件。version: 2是必须的lua转换同时存在 v1 与 v2 两套 APIv1 已标记为废弃v2 采用hooks生命周期钩子source初始化代码的模型。在 src/transforms/lua/mod.rs 中可以看到LuaConfig是一个 untagged 枚举根据version字段分派到LuaConfigV1/LuaConfigV2且官方generate_config默认生成的就是version: 2加空的hooks.process。运行vector --config vector.yaml输出如下每行 CSV 被file源作为一条独立事件原始文本整体存放在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} {file:log.csv,host:localhost,message: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\,,,,,,,,,\\,timestamp:2020-04-09T14:33:28Z} {file:log.csv,host:localhost,message: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\,,,,,,,,,\\,timestamp:2020-04-09T14:33:28Z}除file、host、timestamp等由 Vector 自动添加的元数据字段外整个 CSV 行还只是一串原始字符串——接下来我们把它解析开。引入 CSV 解析模块要做真正的 CSV 解析正确处理带引号、含逗号和转义引号的字段可以直接复用成熟的纯 Lua 库lua-csv。该库只由一个文件构成把它下载到vector.yaml所在的目录即可curl -o csv.lua https://raw.githubusercontent.com/geoffleyland/lua-csv/d20cd42d61dc52e7f6bcb13b596ac7a7d4282fbf/lua/csv.lua然后在lua转换的source配置段中用 Lua 标准库函数require加载它source: | csv require(csv)source段在整个转换构建时执行一次因此csv模块会在 Vector 启动或配置热加载、该转换被动态加入时被加载之后作为全局变量csv在生命周期钩子中直接可用。模块查找机制search_dirsrequire(csv)能命中与vector.yaml同目录的csv.lua是因为 v2 的LuaConfig提供了search_dirs配置项且其默认值就是配置文件的所在目录。在 src/transforms/lua/v2/mod.rs 中search_dirs: VecPathBuf指定require查找 Lua 文件的目录列表未配置时使用default_config_paths()即从当前加载的配置路径推导出的目录。构建时Lua::newsrc/transforms/lua/v2/mod.rs会把search_dirs中每个目录格式化为{dir}/?.lua追加到 Lua 的package.path前部因此require(csv)会优先解析到配置目录/csv.lua。如果你把 Lua 模块放在其他目录可以通过显式配置指向它transforms: lua: type: lua version: 2 search_dirs: - /etc/vector/lua source: | csv require(csv)这一机制在 src/transforms/lua/v2/mod.rs 的lua_load_file测试中有完整验证测试在临时目录写入script2.lua通过search_dirs指向该目录随后在process钩子中require(script2)并调用其导出的函数。实现自定义解析解析逻辑从 message 到多个字段有了csv模块把hooks.process改成真正的解析逻辑hooks: process: | function (event, emit) fields csv.openstring(event.log.message):lines()() -- parse the message field event.log.message nil -- drop the message field column_names { -- a sequence containing CSV column names -- ... } for column, value in ipairs(fields) do -- iterate over CSV columns column_name column_names[column] -- get column name 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()()把event.log.message当作 CSV 字符串解析并取回第一行也是唯一一行的字段序列event.log.message nil删除原始message字段在 Lua 侧给字段赋nil即删除该字段按位置把字段值写入以column_names命名的字段最后必须调用emit(event)把处理完的事件交给下游否则事件会被丢弃。把列名表提升到 source 段column_names对每个事件都是常量无需在每次process调用时重建。把它挪到source段只初始化一次能减少每事件的开销。结合 PostgreSQL 文档给出的csvlog列定义完整的转换配置如下# ... 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 # ...列名表来自 PostgreSQLcsvlog的官方字段顺序。注意backend_type、leader_pid、query_id三个字段仅在 PostgreSQL 13 以上的版本存在若你的数据库版本更低需要从表中移除它们否则列与值会错位。表驱动思路与源码实现一致hooks.process在 v2 中是一个必填的 Lua 闭包或函数签名为function (event, emit)source是可选初始化程序用于导入外部依赖和定义钩子所用的辅助数据。参见 src/transforms/lua/v2/mod.rs 中LuaConfig.source与HooksConfig的字段文档。运行结果结构化事件用同一输入文件再次运行vector --config vector.yaml现在输出的是结构化的 JSON 事件{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:} {application_name:,backend_type:not initialized,command_tag:,connection_from:,context:,database_name:,detail:,error_severity:LOG,file:log.csv,hint:,host:localhost,internal_query:,internal_query_pos:,leader_pid:,location:,log_time:2020-04-09 12:48:49.669 UTC,message:database system was shut down at 2020-04-09 12:48:25 UTC,process_id:27,query:,query_id:0,query_pos:,session_id:localhost.1b,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:} {application_name:,backend_type:not initialized,command_tag:,connection_from:,context:,database_name:,detail:,error_severity:LOG,file:log.csv,hint:,host:localhost,internal_query:,internal_query_pos:,leader_pid:,location:,log_time:2020-04-09 12:48:49.683 UTC,message:database system is ready to accept connections,process_id:1,query:,query_id:0,query_pos:,session_id:localhost.1,session_line_num:2,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:}对其中一条事件做美化展示{ 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: }可以看到原先挤在message里的原始行被拆解为log_time、user_name、error_severity、sql_state_code、message、hint、session_id等 20 余个语义化字段带引号、含内部逗号与转义引号的字段如hint被正确还原为原始内容空的 CSV 字段映射为空字符串file、host、timestamp等由源与 Vector 运行时添加的元数据保持不变。进一步改进解析任务完成后还有两个常见的方向可以继续完善。支持多行字符串CSV 规范允许字段内含换行但file源默认按行切分事件——一个被引号包裹的多行字段会被拆成多条独立事件。有两种处理方式简单场景在file源上启用multiline聚合配置让源在切分前先把属于同一条逻辑记录的连续行合并。multiline是file源内置的聚合选项默认关闭配置项定义见 src/sources/file.rs可以设置模式匹配与超时等参数。复杂场景当multiline无法覆盖例如需要跨多条事件做条件拼接时可以在 Lua 代码中缓冲事件、按条件拼接后再emit这类事件聚合的通用做法可以参考仓库中的自定义聚合指南custom-aggregations-with-lua。注意hooks.process收到的是单条事件聚合场景需要自行持有状态变量定义在source段的变量在钩子间共享。字段类型转换默认情况下CSV 解析出的所有字段都是字符串。可以通过两种方式转换为其他类型在 Lua 内转换使用 Lua 内建函数如tonumber把数值型字段就地转为数字。例如event.log.process_id tonumber(event.log.process_id)。Vector 的事件数据模型在 Lua 侧支持字符串、数字、布尔、数组、对象表等类型。在管线中追加coercer转换在lua转换之后串联coercer用声明式配置完成类型强制转换例如把log_time、session_start_time解析为时间戳或把数值字段转成整数/浮点数。这种方式把解析与定型解耦更利于配置复用与维护。源码级的运行原理补充最后从 src/transforms/lua/v2/mod.rs 的视角总结lua转换在 Vector 内部是如何工作的这有助于你在更大规模场景下预估行为构建阶段LuaConfig::build→Lua::new创建一个mlua运行时实例先把search_dirs格式化为?.lua路径并合并进package.path然后依次求值source代码、编译hooks.init/hooks.process/hooks.shutdown三个钩子均转为RegistryKey保存并注册定时器处理器。运行阶段RuntimeTransform实现hook_process对每条入站事件调用注册的process函数传入包装成LuaEvent的事件对象与emit回调emit会把事件写回 Vector 运行时并延续其source_id元数据见 src/transforms/lua/v2/mod.rs。内存管理在高事件速率下Lua 的自动 GC 可能不及时回收v2 实现引入了GC_INTERVAL 16的阈值——每处理 16 个事件后手动触发一次gc_collect并通过内部事件LuaGcTriggered上报内存占用见 src/transforms/lua/v2/mod.rs 与attempt_gc。如果你的场景内存曲线异常可以关注这一行为。借助以上机制lua转换足以支撑自定义格式 → 结构化事件的解析需求而source预加载模块 hooks.process逐事件处理的组合也让解析逻辑具备良好的可维护性与性能基础。【免费下载链接】vectorA high-performance observability data pipeline.项目地址: https://gitcode.com/GitHub_Trending/vect/vector创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考