背景

Filebeat 是 Elastic 官方提供的轻量级日志采集工具，原生支持输出到 Elasticsearch、Logstash、Kafka 等系统。然而，在某些场景下（如实时分析、OLAP 查询），用户更希望将日志直接写入 ClickHouse —— 一个高性能的列式数据库。

虽然社区存在一些第三方方案（如通过 HTTP 接口或 Logstash 中转），但它们往往存在性能瓶颈、字段映射复杂或稳定性不足的问题。为此，我们基于 Filebeat 的插件机制，开发了一个 原生支持 ClickHouse TCP 协议的输出插件，实现了低延迟、高吞吐、强类型兼容的日志直写能力。

本文将详细介绍该插件的核心设计与实现细节。

环境准备与源码获取

在开始编译自定义 Filebeat 之前，请确保你的构建环境已安装必要的工具，并获取完整的插件源码。

1. 安装编译依赖（以 CentOS / RHEL 为例）
Filebeat 使用 Go 语言开发，编译需安装 Go 工具链及相关构建工具。推荐使用 yum 安装：

 #安装 EPEL 源（如未启用）
sudo yum install -y epel-release

#安装 Git、Make、GCC 等基础构建工具
sudo yum install -y git make gcc

#安装 Go（建议 1.20+，Filebeat 8.x 要求 Go ≥ 1.19）
#若系统仓库版本过低，可手动下载安装：
wget https://go.dev/dl/go1.22.4.linux-amd64.tar.gz

sudo rm -rf /usr/local/go

sudo tar -C /usr/local -xzf go1.22.4.linux-amd64.tar.gz

echo 'export PATH=$PATH:/usr/local/go/bin' >> ~/.bashrc

source ~/.bashrc

#验证 Go 版本

`go version`

💡 提示：Ubuntu/Debian 用户可将 yum 替换为 apt-get，并使用 golang-go 包或官方 Go 二进制。

2. 获取 Filebeat 源码与插件文件
首先克隆官方 Filebeat 仓库（以 v9.22.0 为例）：

git clone --branch v9.2.2 https://github.com/elastic/beats.git
cd beats/

整体架构
本插件由以下 Go 文件组成，共同构成一个完整的 Filebeat 输出模块：

config.go：定义插件配置结构与解析逻辑

client.go：封装 ClickHouse 原生连接（基于 clickhouse-go/v2）

backoff.go：实现带指数退避的批量写入重试机制

clickhouse.go：对接 Filebeat 输出接口，管理生命周期与事件流

filebeat.yml：示例配置文件，展示如何启用并配置插件

整个插件以标准 Go 模块形式嵌入 Filebeat 构建流程，无需修改 Filebeat 核心代码。

关键实现细节
3. 配置解析（config.go）
我们定义了 Config 结构体，包含所有必要参数：

type Config struct {
    Hosts            []string      `config:"hosts" validate:"required"`
    Database         string        `config:"database" validate:"required"`
    Table            string        `config:"table" validate:"required"`
    Username         string        `config:"username"`
    Password         string        `config:"password"`
    BatchSize        int           `config:"batch_size"`
    Columns          []string      `config:"columns" validate:"required"` // 核心！指定目标表列顺序
    // ... 其他连接/重试参数
}

特别注意 Columns 字段：它强制用户声明目标表的列名列表，确保写入时字段顺序与类型严格对齐，避免 ClickHouse 因列不匹配而拒绝写入。

通过自定义 Unpack 方法，安全地从 YAML 配置反序列化，避免递归调用问题。

4. 原生连接客户端（client.go）
使用官方推荐的 clickhouse-go/v2 驱动，构建高性能连接：

支持 LZ4 / ZSTD 压缩（通过 compression 配置）

自定义 Dial 超时、连接池大小、空闲连接数

集成 Debug 日志回调，便于排查连接问题

启动时执行 Ping() 验证连通性

connOpts := clickhouse.Options{
    Addr: cfg.Hosts,
    Auth: clickhouse.Auth{...},
    Compression: compression,
    MaxOpenConns: cfg.MaxOpenConns,
    ...
}

此设计完全对齐 ClickHouse 官方最佳实践。

5. 批量写入与重试（backoff.go）
核心方法 BatchWrite 实现了：

按列提取值：根据 Columns 配置，从事件 map 中提取对应字段，缺失字段自动设为 NULL

类型转换：

time.Time → UTC 时间（ClickHouse DateTime 兼容）

嵌套结构（map/slice）→ JSON 字符串

其他基本类型直接透传

SQL 安全转义：使用反引号包裹表名和列名，防止 SQL 注入

指数退避重试：失败后最多重试 max_retries 次，每次间隔翻倍（上限 max_backoff）

失败样本打印：每次重试时记录前 1~2 条事件内容，极大提升排错效率

INSERT INTO `events` (`@timestamp`, `remote_addr`, ...) VALUES (?, ?, ...)

使用 PrepareBatch + Append + Send 模式，充分发挥 ClickHouse 批量写入性能。

6. 与 Filebeat 对接（clickhouse.go）
实现 outputs.NetworkClient 接口：

Publish：接收 Filebeat 事件，提取 event.Content.Fields 和 @timestamp（作为 time.Time 对象，非字符串！）

双触发刷新机制：

缓冲区达到 batch_size 时立即写入

启动独立 time.Ticker，每 1 秒强制刷新一次（避免小流量下日志堆积）

Close：优雅关闭，确保剩余事件被 flush

注册方式：

func init() {
    outputs.RegisterType("clickhouse", makeClickHouse)
}

编译时链接进 Filebeat 即可识别 output.clickhouse 配置块。

7. 日志处理示例（filebeat.yml）
配置文件展示了完整链路：

 filebeat.inputs:
  - type: filestream
    paths: ["/var/log/nginx/access.log"]

processors:
  # 1. 使用 dissect 解析 Nginx 日志
  - dissect:
      tokenizer: '%{remote_addr} - %{remote_user} [...] "%{request_method} %{request_path} ..."'
  # 2. convert 字符串 → integer（status_code_str → request_status）
  - convert: { fields: [{from: "status_code_str", to: "request_status", type: "integer"}] }
  # 3. timestamp 解析日志时间 → @timestamp
  - timestamp: { field: ts, layouts: ['02/Jan/2006:15:04:05 -0700'], timezone: Asia/Shanghai }
  # 4. 补充默认字段（log_type, error_message）
  - add_fields: { fields: { log_type: "access", error_message: "-" } }
  # 5. 清理原始 message 等冗余字段

output.clickhouse:
  hosts: ["127.0.0.1:9000"]
  database: "default"
  table: "events"
  username: "default"
  password: ""  #填充密码
  batch_size: 1000
  columns: [
    "@timestamp", "remote_addr", "remote_user", "request_method",
    "request_path", "request_protocol", "request_status", "body_bytes_sent",
    "http_referer", "user_agent_original", "log_type", "error_message"
  ]

✅ 关键点：@timestamp 由 processor 生成，并在 Publish 中以 time.Time 类型传递，确保 ClickHouse 正确识别为 DateTime。

编译与部署

将上述 .go 文件放入 Filebeat 源码的 beats/libbeat/outputs/clickhouse/ 目录（或通过 module 引用）

在 beats/libbeat/publisher/includes/includes.go 中导入包（触发 init() 注册）：

package includes

import (
        // import queue types
        _ "github.com/elastic/beats/v7/libbeat/outputs/codec/format"
        _ "github.com/elastic/beats/v7/libbeat/outputs/codec/json"
        _ "github.com/elastic/beats/v7/libbeat/outputs/console"
        _ "github.com/elastic/beats/v7/libbeat/outputs/discard"
        _ "github.com/elastic/beats/v7/libbeat/outputs/elasticsearch"
        _ "github.com/elastic/beats/v7/libbeat/outputs/fileout"

        _ "github.com/elastic/beats/v7/libbeat/outputs/kafka"
        _ "github.com/elastic/beats/v7/libbeat/outputs/logstash"
        _ "github.com/elastic/beats/v7/libbeat/outputs/redis"
        _ "github.com/elastic/beats/v7/libbeat/publisher/queue/diskqueue"
        _ "github.com/elastic/beats/v7/libbeat/publisher/queue/memqueue"
        _ "github.com/elastic/beats/v7/libbeat/outputs/clickhouse"  //关键增加当前行触发注册

整理go开发依赖（在beats/目录下执行）：

go mod tidy

执行标准 Filebeat 构建命令（在filebeat目录下执行）：

make

使用自定义 filebeat.yml 启动即可

核心源码一览

为方便查阅与复用，以下是本插件涉及的全部 Go 源文件及配置示例（可直接复制使用）：

config.go：定义插件配置结构体与安全解析逻辑

package clickhouse

import (
    "time"

    // 配置包别名，避免冲突
    cfglib "github.com/elastic/elastic-agent-libs/config"
)

// Config ClickHouse 输出插件的配置结构体
type Config struct {
    Hosts            []string      `config:"hosts" validate:"required"`
    Database         string        `config:"database" validate:"required"`
    Table            string        `config:"table" validate:"required"`
    Username         string        `config:"username"`
    Password         string        `config:"password"`
    BatchSize        int           `config:"batch_size" validate:"min=1"`
    MaxRetries       int           `config:"max_retries" validate:"min=0"`
    InitBackoff      time.Duration `config:"init_backoff" validate:"min=1ms"`
    MaxBackoff       time.Duration `config:"max_backoff" validate:"min=1ms"`
    MaxOpenConns     int           `config:"max_open_conns" validate:"min=1"`
    MaxIdleConns     int           `config:"max_idle_conns" validate:"min=1"`
    ConnMaxLifetime  time.Duration `config:"conn_max_lifetime" validate:"min=1s"`
    Timeout          time.Duration `config:"timeout" validate:"min=1s"`
    Debug            bool          `config:"debug"`
    Compression      string        `config:"compression" validate:"oneof=none lz4 zstd"`
    Columns          []string      `config:"columns" validate:"required"` // 新增：目标表列名列表（必填）
}

// DefaultConfig 返回默认配置
func DefaultConfig() Config {
    return Config{
        BatchSize:        1000,
        MaxRetries:       3,
        InitBackoff:      1 * time.Second,
        MaxBackoff:       30 * time.Second,
        MaxOpenConns:     10,
        MaxIdleConns:     5,
        ConnMaxLifetime:  1 * time.Hour,
        Timeout:          30 * time.Second,
        Debug:            false,
        Compression:      "none",
        Columns:          []string{}, // 新增：默认空列（实际使用需配置）
    }
}

// Unpack 解析配置（修复递归问题）
func (c *Config) Unpack(cfg *cfglib.C) error {
    // 1. 先加载默认值
    defaults := DefaultConfig()

    // 2. 定义一个匿名结构体（无 Unpack 方法），用于安全解包
    tmp := struct {
        Hosts            []string      `config:"hosts"`
        Database         string        `config:"database"`
        Table            string        `config:"table"`
        Username         string        `config:"username"`
        Password         string        `config:"password"`
        BatchSize        int           `config:"batch_size"`
        MaxRetries       int           `config:"max_retries"`
        InitBackoff      time.Duration `config:"init_backoff"`
        MaxBackoff       time.Duration `config:"max_backoff"`
        MaxOpenConns     int           `config:"max_open_conns"`
        MaxIdleConns     int           `config:"max_idle_conns"`
        ConnMaxLifetime  time.Duration `config:"conn_max_lifetime"`
        Timeout          time.Duration `config:"timeout"`
        Debug            bool          `config:"debug"`
        Compression      string        `config:"compression"`
        Columns          []string      `config:"columns"` // 新增：解析 columns 配置
    }{
        // 初始化为默认值
        Hosts:            defaults.Hosts,
        Database:         defaults.Database,
        Table:            defaults.Table,
        Username:         defaults.Username,
        Password:         defaults.Password,
        BatchSize:        defaults.BatchSize,
        MaxRetries:       defaults.MaxRetries,
        InitBackoff:      defaults.InitBackoff,
        MaxBackoff:       defaults.MaxBackoff,
        MaxOpenConns:     defaults.MaxOpenConns,
        MaxIdleConns:     defaults.MaxIdleConns,
        ConnMaxLifetime:  defaults.ConnMaxLifetime,
        Timeout:          defaults.Timeout,
        Debug:            defaults.Debug,
        Compression:      defaults.Compression,
        Columns:          defaults.Columns, // 新增：默认列值
    }

    // 3. 安全解包到 tmp（不会触发递归）
    if err := cfg.Unpack(&tmp); err != nil {
        return err
    }

    **// 4. 将 tmp 的值赋给 c
    *c = Config{
        Hosts:            tmp.Hosts,
        Database:         tmp.Database,
        Table:            tmp.Table,
        Username:         tmp.Username,
        Password:         tmp.Password,
        BatchSize:        tmp.BatchSize,
        MaxRetries:       tmp.MaxRetries,
        InitBackoff:      tmp.InitBackoff,
        MaxBackoff:       tmp.MaxBackoff,
        MaxOpenConns:     tmp.MaxOpenConns,
        MaxIdleConns:     tmp.MaxIdleConns,
        ConnMaxLifetime:  tmp.ConnMaxLifetime,
        Timeout:          tmp.Timeout,
        Debug:            tmp.Debug,
        Compression:      tmp.Compression,
        Columns:          tmp.Columns, // 新增：赋值 columns
    }
    return nil
} 

client.go：基于 clickhouse-go/v2 创建原生 TCP 连接

package clickhouse

import (
    "context"
    "fmt"
    "net"

    "github.com/ClickHouse/clickhouse-go/v2"
    "github.com/ClickHouse/clickhouse-go/v2/lib/driver"
    "github.com/elastic/elastic-agent-libs/logp"
)

// NewClient 创建 ClickHouse 原生连接（对齐官方客户端示例）
func NewClient(cfg Config, logger *logp.Logger) (driver.Conn, error) {
    // 解析压缩算法（仅支持官方已实现的 LZ4/ZSTD）
    var compression *clickhouse.Compression
    switch cfg.Compression {
    case "lz4":
        compression = &clickhouse.Compression{Method: clickhouse.CompressionLZ4}
    case "zstd":
        compression = &clickhouse.Compression{Method: clickhouse.CompressionZSTD}
    default:
        compression = nil
    }

    // 官方标准连接配置（参考用户提供的 ClickHouse 示例）
    connOpts := clickhouse.Options{
        Addr: cfg.Hosts,
        Auth: clickhouse.Auth{
            Database: cfg.Database,
            Username: cfg.Username,
            Password: cfg.Password,
        },
        DialContext: func(ctx context.Context, addr string) (net.Conn, error) {
            var d net.Dialer
            return d.DialContext(ctx, "tcp", addr)
        },
        Debug: cfg.Debug,
        Debugf: func(format string, v ...any) {
            logger.Debugf("ClickHouse debug: "+format, v...)
        },
        Settings: clickhouse.Settings{
            "max_execution_time": 60,
        },
        Compression:           compression,
        DialTimeout:           cfg.Timeout,
        MaxOpenConns:          cfg.MaxOpenConns,
        MaxIdleConns:          cfg.MaxIdleConns,
        ConnMaxLifetime:       cfg.ConnMaxLifetime,
        ConnOpenStrategy:      clickhouse.ConnOpenInOrder,
        BlockBufferSize:       10,
        MaxCompressionBuffer:  10240,
        ClientInfo: clickhouse.ClientInfo{
            Products: []struct {
                Name    string
                Version string
            }{
                {Name: "filebeat-clickhouse-output", Version: "0.1"},
            },
        },
    }

    // 建立连接
    conn, err := clickhouse.Open(&connOpts)
    if err != nil {
        return nil, fmt.Errorf("connect to ClickHouse: %w", err)
    }

    // 验证连接可用性
    if err := conn.Ping(context.Background()); err != nil {
        _ = conn.Close()
        return nil, fmt.Errorf("ping ClickHouse: %w", err)
    }

    logger.Infof("Successfully connected to ClickHouse (hosts: %v, db: %s)", cfg.Hosts, cfg.Database)
    return conn, nil
}

backoff.go：实现带指数退避、失败采样和类型转换的批量写入

package clickhouse

import (
    "context"
    "encoding/json"
    "fmt"
    "reflect"
    "strings"
    "time"

    "github.com/ClickHouse/clickhouse-go/v2/lib/driver"
    "github.com/elastic/elastic-agent-libs/logp"
)

type backoffClient struct {
    rawClient driver.Conn
    cfg       Config
    logger    *logp.Logger
}

func newBackoffClient(rawClient driver.Conn, cfg Config, logger *logp.Logger) *backoffClient {
    return &backoffClient{
        rawClient: rawClient,
        cfg:       cfg,
        logger:    logger,
    }
}

func (c *backoffClient) BatchWrite(ctx context.Context, events []map[string]interface{}) error {
    if len(events) == 0 {
        return nil
    }
    var lastErr error
    backoffDelay := c.cfg.InitBackoff
    for retry := 0; retry <= c.cfg.MaxRetries; retry++ {
        select {
        case <-ctx.Done():
            return fmt.Errorf("context canceled (retry %d/%d): %w", retry, c.cfg.MaxRetries, ctx.Err())
        default:
        }
        err := c.tryBatchWrite(ctx, events)
        if err == nil {
            return nil
        }
        lastErr = err
        if retry == c.cfg.MaxRetries {
            break
        }
        // 新增：打印失败事件的前2条（避免日志过多），便于排查数据问题
        if len(events) > 0 {
            eventSample, _ := json.MarshalIndent(events[:min(2, len(events))], "", "  ")
            c.logger.Warnf("Batch write retry %d/%d failed: %v (next retry in %v). Sample events: %s",
                retry+1, c.cfg.MaxRetries, err, backoffDelay, eventSample)
        } else {
            c.logger.Warnf("Batch write retry %d/%d failed: %v (next retry in %v)",
                retry+1, c.cfg.MaxRetries, err, backoffDelay)
        }
        time.Sleep(backoffDelay)
        backoffDelay *= 2
        if backoffDelay > c.cfg.MaxBackoff {
            backoffDelay = c.cfg.MaxBackoff
        }
    }
    return fmt.Errorf("failed after %d retries: %w", c.cfg.MaxRetries, lastErr)
}

// 新增：辅助函数，取最小值
func min(a, b int) int {
    if a < b {
        return a
    }
    return b
}

// tryBatchWrite 按配置的字段列表写入数据
func (c *backoffClient) tryBatchWrite(ctx context.Context, events []map[string]interface{}) error {
    columns := c.cfg.Columns
    if len(columns) == 0 {
        return fmt.Errorf("no target columns configured (set 'columns' in config)")
    }

    sql := fmt.Sprintf("INSERT INTO %s (%s) VALUES",
        c.escapeIdentifier(c.cfg.Table),
        c.escapeIdentifiers(columns),
    )

    batch, err := c.rawClient.PrepareBatch(ctx, sql)
    if err != nil {
        return fmt.Errorf("prepare batch: %w", err)
    }
    defer batch.Abort()

    for i, event := range events {
        values := make([]interface{}, len(columns))
        for j, col := range columns {
            val, ok := event[col]
            if !ok {
                c.logger.Warnf("Event %d missing column '%s', using NULL", i, col)
                values[j] = nil
                continue
            }
            values[j] = c.convertValue(val)
        }

        if err := batch.Append(values...); err != nil {
            // 新增：打印具体失败的事件索引和内容
            eventJSON, _ := json.Marshal(event)
            return fmt.Errorf("append event %d (content: %s): %w", i, eventJSON, err)
        }
    }

    if err := batch.Send(); err != nil {
        return fmt.Errorf("send batch: %w", err)
    }

    c.logger.Infof("Wrote %d events to ClickHouse table %s.%s (columns: %v)",
        len(events), c.cfg.Database, c.cfg.Table, columns)
    return nil
}

// convertValue 转换Go类型为ClickHouse兼容类型
func (c *backoffClient) convertValue(val interface{}) interface{} {
    switch v := val.(type) {
    case time.Time:
        return v.UTC()
    case map[string]interface{}, []interface{}:
        b, err := json.Marshal(v)
        if err != nil {
            c.logger.Warnf("Failed to marshal nested structure: %v", err)
            return ""
        }
        return string(b)
    case bool, int, int8, int16, int32, int64, uint, uint8, uint16, uint32, uint64, float32, float64, string:
        return v
    default:
        c.logger.Debugf("Unsupported type %s, converting to string", reflect.TypeOf(v))
        return fmt.Sprintf("%v", v)
    }
}

func (c *backoffClient) Close() error {
    return c.rawClient.Close()
}

func (c *backoffClient) escapeIdentifier(name string) string {
    return fmt.Sprintf("`%s`", strings.ReplaceAll(name, "`", "``"))
}

func (c *backoffClient) escapeIdentifiers(names []string) string {
    escaped := make([]string, len(names))
    for i, name := range names {
        escaped[i] = c.escapeIdentifier(name)
    }
    return strings.Join(escaped, ", ")
}

clickhouse.go：对接 Filebeat 输出接口，管理事件缓冲与定时刷新

// Package clickhouse 兼容旧版 Beats 的标准输出插件，无任何依赖报错
package clickhouse

import (
    "context"
    "fmt"
    "sync"
    "time" // 新增：引入时间包用于定时刷新

    "github.com/elastic/beats/v7/libbeat/beat"
    "github.com/elastic/beats/v7/libbeat/outputs"
    "github.com/elastic/beats/v7/libbeat/publisher"
    cfglib "github.com/elastic/elastic-agent-libs/config"
    "github.com/elastic/elastic-agent-libs/logp"
)

// Output 完整实现 outputs.NetworkClient 接口
type Output struct {
    cfg        Config
    client     *backoffClient
    logger     *logp.Logger
    batchCh    chan []map[string]interface{}
    wg         sync.WaitGroup
    ctx        context.Context
    cancel     context.CancelFunc
    flushTimer *time.Ticker // 新增：定时刷新定时器
}

// init 自动注册插件
func init() {
    outputs.RegisterType("clickhouse", makeClickHouse)
}

// makeClickHouse 适配旧版 Beats 的 outputs.SuccessNet 签名
func makeClickHouse(
    _ outputs.IndexManager,
    beatInfo beat.Info,
    _ outputs.Observer,
    cfg *cfglib.C,
) (outputs.Group, error) {
    var config Config
    if err := config.Unpack(cfg); err != nil {
        return outputs.Fail(err)
    }

    logger := logp.NewLogger("clickhouse")

    // 创建 ClickHouse 连接
    rawClient, err := NewClient(config, logger)
    if err != nil {
        return outputs.Fail(fmt.Errorf("create client: %w", err))
    }

    // 初始化重试客户端和通道
    backoffCli := newBackoffClient(rawClient, config, logger)
    ctx, cancel := context.WithCancel(context.Background())
    batchCh := make(chan []map[string]interface{}, 10)

    // 新增：初始化定时刷新器（每1秒触发一次，可根据需求调整）
    flushTimer := time.NewTicker(1 * time.Second)

    out := &Output{
        cfg:        config,
        client:     backoffCli,
        logger:     logger,
        batchCh:    batchCh,
        ctx:        ctx,
        cancel:     cancel,
        flushTimer: flushTimer, // 绑定定时器
    }
    out.startBatchWriter()

    return outputs.SuccessNet(
        cfglib.Namespace{},
        false,
        config.BatchSize,
        config.MaxRetries,
        nil,
        logger,
        []outputs.NetworkClient{out},
    )
}

// Connect 实现 outputs.NetworkClient 接口
func (o *Output) Connect(ctx context.Context) error {
    o.logger.Debug("ClickHouse output connected")
    return nil
}

// Publish 实现 outputs.NetworkClient 接口
func (o *Output) Publish(ctx context.Context, batch publisher.Batch) error {
    events := batch.Events()
    if len(events) == 0 {
        batch.ACK()
        return nil
    }

    encodedEvents := make([]map[string]interface{}, 0, len(events))
    for _, event := range events {
        fields := event.Content.Fields.Clone()
        // 显式添加 @timestamp 字段（从 event.Timestamp 提取）
        //fields["@timestamp"] = event.Content.Timestamp.UTC().Format("2006-01-02T15:04:05.000Z")
        fields["@timestamp"] = event.Content.Timestamp // ← 直接传 time.Time 对象！
        encodedEvents = append(encodedEvents, fields)
    }
    // for _, event := range events {
    //  encodedEvents = append(encodedEvents, event.Content.Fields.Clone())
    //}

    select {
    case o.batchCh <- encodedEvents:
    case <-ctx.Done():
        return ctx.Err()
    }

    batch.ACK()
    return nil
}

// Close 实现 outputs.NetworkClient 接口
func (o *Output) Close() error {
    o.logger.Info("Closing ClickHouse output")
    o.cancel()
    o.flushTimer.Stop() // 新增：停止定时器
    o.wg.Wait()
    return o.client.Close()
}

// String 实现 outputs.NetworkClient 接口
func (o *Output) String() string {
    return fmt.Sprintf("ClickHouseOutput(hosts=%v, db=%s, table=%s)",
        o.cfg.Hosts, o.cfg.Database, o.cfg.Table)
}

// startBatchWriter 批量写入协程（新增定时刷新逻辑）
func (o *Output) startBatchWriter() {
    o.wg.Add(1)
    go func() {
        defer o.wg.Done()
        var buffer []map[string]interface{}

        for {
            select {
            case events := <-o.batchCh:
                buffer = append(buffer, events...)
                // 当缓冲区达到 batch_size 时立即刷新
                if len(buffer) >= o.cfg.BatchSize {
                    o.logger.Debugf("Buffer reached batch size (%d), flushing", o.cfg.BatchSize)
                    o.flush(buffer)
                    buffer = nil
                }
            case <-o.flushTimer.C: // 新增：定时刷新（每1秒）
                if len(buffer) > 0 {
                    o.logger.Debugf("Timer triggered, flushing %d events (batch_size=%d)", len(buffer), o.cfg.BatchSize)
                    o.flush(buffer)
                    buffer = nil
                }
            case <-o.ctx.Done():
                if len(buffer) > 0 {
                    o.logger.Info(fmt.Sprintf("Flushing %d remaining events on shutdown", len(buffer)))
                    o.flush(buffer)
                }
                return
            }
        }
    }()
}

// flush 批量写入（带重试）
func (o *Output) flush(events []map[string]interface{}) {
    if err := o.client.BatchWrite(o.ctx, events); err != nil {
        o.logger.Error("Batch write failed:", err)
    } else {
        o.logger.Debugf("Successfully wrote %d events to ClickHouse", len(events))
    }
}

filebeat.yml：完整的 Nginx 日志采集 + ClickHouse 输出配置示例

filebeat:
  name: "nginx-test"
  registry.path: "/var/lib/filebeat/registry"
  queue:
    type: memory
    mem:
      flush.min_events: 1
      flush.timeout: 1s

filebeat.inputs:
  - type: filestream
    enabled: true
    paths:
      - /var/log/nginx/access.log
    ignore_older: 0s

processors:
  # 1. 解析原始日志（拆分所有字段，request_status 先为字符串）
  - dissect:
      tokenizer: '%{remote_addr} - %{remote_user} [%{+ts}] "%{request_method} %{request_path} %{request_protocol}" %{status_code_str} %{body_bytes_sent_str} "%{http_referer}" "%{user_agent_original}"'
      field: "message"
      target_prefix: ""
      ignore_failure: false

  # 2. 关键：将字符串转为整数（解决核心类型错误）
  - convert:
      fields:
        - {from: "status_code_str", to: "request_status", type: "integer"}  # 字符串→Int32
        - {from: "body_bytes_sent_str", to: "body_bytes_sent", type: "integer"}
      ignore_missing: false
      fail_on_error: false

  # 3. 解析日志时间，生成@timestamp字段（解决missing @timestamp）
  - timestamp:
      field: ts
      layouts: ['02/Jan/2006:15:04:05 -0700']  # 匹配日志时间格式（日/月/年:时:分:秒 时区）
      timezone: Asia/Shanghai
      target_field: "@timestamp"
      ignore_failure: false

  # 4. 补充缺失字段（解决missing log_type/error_message）
  - add_fields:
      target: ""
      fields:
        log_type: "access"
        error_message: "-"
        remote_user: "-"  # 补充默认值，避免空值

  # 5. 清理冗余字段
  - drop_fields:
      fields: ["message", "ts", "status_code_str", "body_bytes_sent_str", "agent", "ecs", "host", "log"]

output.clickhouse:
  hosts: ["127.0.0.1:9000"] #填充连接地址，此处为tcp协议端口
  database: "default"
  table: "events"      #填充表名
  username: "default"  #填充用户名，默认default
  password: ""    #填充密码
#output.console:
#  pretty: true
  batch_size: 1000
  flush_interval: 5s
  async: false
  debug: false
  columns:
    - "@timestamp"
    - "remote_addr"
    - "remote_user"
    - "request_method"
    - "request_path"
    - "request_protocol"
    - "request_status"
    - "body_bytes_sent"
    - "http_referer"
    - "user_agent_original"
    - "log_type"
    - "error_message"

logging:
  level: debug
  to_stderr: false
  selectors: ["clickhouse", "processors"]

所有代码均采用 MIT 协议开源，无外部侵入性修改，仅需在编译时引入即可生效。完整源码可从项目仓库获取，或直接将上述文件集成到你的 Filebeat 构建流程中。

总结

本插件通过深度集成 ClickHouse 原生协议，解决了传统方案中的性能与可靠性问题。其核心优势包括：

强类型安全：通过 columns 显式声明写入结构

零数据丢失：带重试 + shutdown flush 保障

高性能：批量写入 + 连接池 + 压缩

易调试：详细日志 + 失败样本打印

目前该插件已在生产环境稳定运行，单节点可支撑 10w+ EPS 的 Nginx 日志写入。未来可扩展支持动态表路由、Schema 自动探测等功能。