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

Question

A Jobber Docker container（运行定期任务）在stdout上输出，由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”字段下！）！ 您能指导我实现该方法吗？

Answer 1

Filebeat提供processors来处理这些任务。

下面是处理需求的配置“在“消息”字段中解码json”，“在内部的“ stdout”中解码json”（都使用decode_json_fields处理器）以及其他与Jobber相关的需求。] >

[请注意，给出的示例使用托管给Jobber进程的Docker容器的'custom-tag'标签来过滤通过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。