Heka项目中的过渡性Filter和Output API解析

概述

Heka作为一款高效的数据处理工具，在其v0.9.X系列到v0.10.0版本间经历了重大的API变更。为了帮助开发者平滑过渡，项目特别保留了过渡性API支持，让已有自定义插件能够以最小修改量适配新版本的磁盘缓冲功能。

过渡性API的核心变化

过渡性API与v0.9.X系列基本保持一致，但有两个关键改进点：

1. 游标更新机制：插件现在需要显式调用runner的UpdateCursor方法来推进磁盘缓冲的游标位置，这与新版API的要求一致。

2. 消息回收机制增强：pack.Recycle方法现在需要接收一个error参数，用于指示消息处理状态：

   • nil表示处理成功
   • 普通error表示处理失败且不应重试
   • RetryMessageError表示应尝试重新处理

过渡性Filter API详解

基础接口

过渡性Filter插件只需实现一个简单接口：

type OldFilter interface {
    Run(r FilterRunner, h PluginHelper) (err error)
}

核心工作流程

1. 消息获取：通过runner.InChan()获取输入通道，从中读取PipelinePack对象。

2. 消息处理：从PipelinePack中提取Message结构体进行业务处理。

3. 新消息生成：需要生成新消息时，调用PluginHelper.PipelinePack()获取新的PipelinePack，填充后通过FilterRunner.Inject()注入路由系统。

4. 定时任务：配置ticker_interval后，可通过FilterRunner.Ticker()获取定时通道，实现周期性任务。

5. 资源回收：必须调用pack.Recycle()释放资源，并传递适当的错误状态。

关键注意事项

• 游标管理：使用缓冲时必须正确处理QueueCursor和调用UpdateCursor
• 防消息循环：系统会自动防止插件注入自己会匹配的消息
• 优雅退出：输入通道关闭时应做好清理工作并返回nil

过渡性Output API详解

基础接口

type OldOutput interface {
    Run(or OutputRunner, h PluginHelper) (err error)
}

核心差异点

虽然与Filter API相似，但Output插件专注于：

1. 数据输出：将消息序列化后发送到外部系统

2. 编码处理：通过OutputRunner提供的编码方法：

   • Encode()：自动处理消息编码和帧包装
   • UsesFraming()：检查是否启用了帧包装
   • Encoder()：获取底层编码器实例

共同特性

• 同样需要处理输入通道和定时任务
• 必须正确调用pack.Recycle()
• 需要处理缓冲游标更新

最佳实践建议

1. 错误处理：根据业务场景合理选择Recycle的错误参数：

   • 瞬时错误使用RetryMessageError
   • 永久错误使用普通error
   • 成功则传递nil

2. 性能优化：对于高频处理场景，注意：

   • 及时回收PipelinePack
   • 合理设置缓冲大小
   • 避免在消息处理中进行耗时操作

3. 兼容性考虑：过渡性API将在v0.10.X系列后移除，建议新开发直接使用新版API

总结

Heka的过渡性API为现有插件升级提供了便利通道，开发者只需关注几个关键变更点即可适配新版本的缓冲功能。理解这些API的设计理念和使用规范，将帮助开发者构建更健壮、高效的数据处理插件。