在 ELK 上可视化 Jobber 任务（通过 Filebeat）答案

【问题标题】：Visualize Jobber tasks on ELK (via Filebeat)在 ELK 上可视化 Jobber 任务（通过 Filebeat）
【发布时间】：2020-01-20 17:52:48
【问题描述】：

一个Jobber Docker container（运行周期性任务）在标准输出上输出，由 Filebeat（Docker 容器自动发现标志打开）捕获，然后发送到 Logstash（在 ELK 堆栈内）或直接发送到 Elasticsearch。

现在在 Kibana 上，文档如下所示：

@timestamp  Jan 20, 2020 @ 20:15:07.752
...
agent.type  filebeat
container.image.name    jobber_jobber
...
message {
          "job": {
                    "command":"curl http://my.service/run","name":"myperiodictask",
                    "status":"Good",
                    "time":"0 */5 * * * *"
          },
          "startTime":1579540500,
          "stdout":"{\"startDate\":\"2020-01-20T16:35:00.000Z\",\"endDate\":\"2020-01-20T17:00:00.000Z\",\"zipped\":true,\"size\":3397}",
          "succeeded":true,
          "user":"jobberuser",
          "version":"1.4"
        }
...

注意：上面的'message'字段是一个反映json对象的简单字符串；上面显示的格式是为了更清晰的可读性。

我的目标是能够在消息字段上请求 Elastic，例如，我可以按 Jobber 任务进行过滤。

我怎样才能做到这一点？ 我知道 Filebeat 使用插件和容器标签来应用这个或那个过滤器：有适用于 Jobber 的吗？如果没有，该怎么做？

更好的是能够利用 Jobber 任务结果的字段（在“stdout”字段下）！ 能否请您指导我如何实现它？

【问题讨论】：

标签： elastic-stack filebeat

【解决方案1】：

Filebeat 提供processors 来处理此类任务。

下面是一个配置来处理“解码'message'字段中的json”、“解码'stdout'内的json”（都使用decode_json_fields处理器）和其他与Jobber相关的需求。

请注意，给定的示例通过提供给托管 Jobber 进程的 Docker 容器的“自定义标签”标签过滤通过 Filebeat 的事件。 docker.container.labels.custom-tag: jobber 条件应根据您的用例替换。

filebeat.yml：

processors:

# === Jobber events processing ===
- if:
    equals:
      docker.container.labels.custom-tag: jobber
  then:
    # Drop Jobber events which are not job results
    - drop_event:
        when:
          not:
            regexp:
              message: "{.*"
    # Json-decode event's message part
    - decode_json_fields:
        when:
          regexp:
            message: "{.*"
        fields: ["message"]
        target: "jobbertask"
    # Json-decode message's stdout part
    - decode_json_fields:
        when:
          has_fields: ["jobbertask.stdout"]
        fields: ["jobbertask.stdout"]
        target: "jobbertask.result"
    # Drop event's decoded fields
    - drop_fields:
        fields: ["message"]
    - drop_fields:
        when:
          has_fields: ["jobbertask.stdout"]
        fields: ["jobbertask.stdout"]

解码后的字段被放置在“jobbertask”字段中。这是为了避免根字段上的索引映射冲突。随意用任何其他字段名称替换“jobbertask”，注意映射冲突。

就我而言，无论 Filebeat 将事件发送到 Logstash 还是直接发送到 Elasticsearch，这都有效。

【讨论】：