Fluent Bit Filter 将 Unix Epoch 时间戳转换为人类可读的时间格式

Question

我有一个 Java 应用程序，我在其中使用 Log4j2 在 JSONLayout 中打印日志，这里是日志格式的示例：

{
    "thread": "TopicStatusThreadPool-thread-1",
    "level": "INFO",
    "loggerName": "My Kafka Logger",
    "message": "Topic Status List is empty, returning from summarize method",
    "endOfBatch": false,
    "loggerFqcn": "org.apache.logging.slf4j.Log4jLogger",
    "instant": {
        "epochSecond": 1587980988,
        "nanoOfSecond": 214241000
    },
    "threadId": 37,
    "threadPriority": 5
  }

以这种格式打印的日志随后被 Fluent Bit 提取并推送到 Elasticsearch。我正在使用 Kibana 可视化日志，但是当我看到日志和时间为 epochSecond 和 nonOfSecond 时，很难将其与实际应用程序日志关联起来，因为格式。

是否有任何 Fluent Bit Filter 可用于将这种时间格式修改为更易于阅读的格式。

目前我在 Fluent Bit 配置中使用 基本 JSON 解析器和 Kubernetes 过滤器 将 Kubernetes 信息添加到日志消息中。

更新：

我对 Log4j2 配置进行了更改，现在我得到了 timeMillis 字段，它在日志中以毫秒为单位打印时间。

{
    "thread": "TopicStatusThreadPool-thread-1",
    "level": "INFO",
    "loggerName": "My Kafka Logger",
    "message": "Topic Status List is empty, returning from summarize method",
    "endOfBatch": false,
    "loggerFqcn": "org.apache.logging.slf4j.Log4jLogger",
    "timeMillis": 1587980988213,
    "threadId": 37,
    "threadPriority": 5
  }

将它转换为人类可读格式的 lua 过滤器是什么。默认情况下，Fluent Bit 中的时间转换不支持以毫秒为单位的时间，它需要以秒为单位的时间。

我试过了：

[PARSER]
        Name   json
        Format json
        Time_Key timeMillis
        Time_Format %s
        Time_Keep On

但这并没有选择毫秒部分，以秒为单位处理时间。

Answer 1

你可以按如下方式使用 lua 过滤器：

-- test.lua
function append_converted_timestamp(tag, timestamp, record)
    new_record = record
    new_record["instant"]["recorded_time"] = os.date("%m/%d/%Y %I:%M:%S %p", record["instant"]["epochSecond"])
    return 2, timestamp, new_record
end

FluentBit 配置：

[FILTER]
    Name    lua
    Match   *
    script  test.lua
    call    append_converted_timestamp

它将在您的记录中附加一个带有人类可读日期的新字段 recorded_time。

更新

timeMillis 字段的 Lua 函数可以这样实现：

-- test.lua
function append_converted_timestamp(tag, timestamp, record)
    new_record = record
    new_record["recorded_time"] = os.date("%m/%d/%Y %I:%M:%S %p", record["timeMillis"]/1000)
    return 2, timestamp, new_record
end