如何配置Fluent-Bit按时间或指定条数触发日志批量写入ClickHouse？

我以systemd作为Fluent-Bit的输入源，输出到ClickHouse，且通过Go语言开发了对应的输出插件，核心代码片段如下：

func FLBPluginFlushCtx(ctx, data unsafe.Pointer, length C.int, tag *C.char) int {
    dec := output.NewDecoder(data, int(length))

    batch, err := client.connect.PrepareBatch(context.Background(), insertSql)
    if err != nil {
        return output.FLB_PROXY_OUTPUT_PLUGIN
    }

    count := 0
    startFlushTime := time.Now()
    for {
        ret, ts, record := output.GetRecord(dec)
        // 将ret转换为rowPtr的逻辑
        err = batch.AppendStruct(rowPtr)
        if err != nil {
            log.Error(nil, "batch append error", zap.Error(err), zap.Any("rowPtr", rowPtr))
        }
    }
    err = batch.Send()
    if err != nil {
        log.Error(nil, "batch send error", zap.Error(err))
        return output.FLB_PROXY_OUTPUT_PLUGIN
    }
}

当Fluent-Bit调用FLBPluginFlushCtx函数时，日志会被写入ClickHouse。

当前我的Fluent-Bit配置如下：

[SERVICE]
    Flush        5
    Daemon       Off
    Log_Level    info
    Health_Check On
    HTTP_Server  On
    HTTP_Listen  0.0.0.0
    HTTP_Port    2020
    Parsers_File parsers.conf
    Parsers_File parsers_custom.conf

[INPUT]
    Name            systemd
    Systemd_Filter  _COMM=***
    DB              /fluent-bit/flag/flag

[OUTPUT]
    Name                  clickhouse
    Match                 *
    Address               ***
    Database              ***
    Table                 ***
    Username              ***
    Password              ***

其中[SERVICE]中的Flush 5表示每5秒调用一次FLBPluginFlushCtx函数，但日志量较大时单次仅处理约1500条日志。我希望配置成每5秒触发一次，或累计达到20000条日志时触发，请问该如何设置这个条数阈值？

在Fluent-Bit的[SERVICE]配置段中添加Flush_Threshold参数即可实现需求，该参数用于设置触发Flush操作的日志条数阈值，和Flush参数配合时，满足任意一个条件就会触发日志批量写入。

修改后的[SERVICE]配置如下：

[SERVICE]
    Flush        5
    Flush_Threshold 20000
    Daemon       Off
    Log_Level    info
    Health_Check On
    HTTP_Server  On
    HTTP_Listen  0.0.0.0
    HTTP_Port    2020
    Parsers_File parsers.conf
    Parsers_File parsers_custom.conf

参数说明 #

• Flush 5：保持原配置，每5秒强制触发一次Flush操作
• Flush_Threshold 20000：当Fluent-Bit缓存的日志条数达到20000条时，立即触发Flush操作，无需等待5秒周期

这样配置后，Fluent-Bit会在时间到5秒或累计日志数到20000条这两个条件满足任意一个时，调用你的FLBPluginFlushCtx函数处理批量日志。

内容的提问来源于stack exchange，提问作者zhang_