ELK 小技巧 2：掌握 input 和 output 插值表达式

2023-09-07 16:04:59

输入和输出插值表达式（input and output interpolation） in Filebeat 8

让我们先从一个简单的例子开始。假设我们要将 Filebeat 收集的日志发送到 ElasticSearch 中，并将每条日志的时间戳作为文档的@timestamp字段。Filebeat 可以通过 output 插件将数据发送到 ElasticSearch，并且可以配置字段映射，将 Filebeat 中的字段映射到 ElasticSearch 中的字段。要将时间戳作为@timestamp字段，我们可以使用 input 和 output 插值表达式。

input:
  type: log

  paths:
    - /var/log/*.log

output:
  elasticsearch:
    hosts: ["localhost:9200"]
    index: "my-index-%{+yyyy.MM.dd}" # 在索引名称中使用时间戳插值
    template: "my-template"

    fields:
      @timestamp: "%{@timestamp}"  # 在字段映射中使用时间戳插值

在上面的配置中，我们在 input 中定义了日志文件的路径，在 output 中定义了 ElasticSearch 的地址和索引名称。我们还使用了插值表达式将时间戳作为索引名称的一部分和@timestamp字段的值。

插值表达式可以在 input 和 output 的任何地方使用，包括字段名称、字段值、索引名称、类型名称、甚至是管道配置中。例如，我们可以使用插值表达式来动态生成字段名称。

output:
  elasticsearch:
    hosts: ["localhost:9200"]
    index: "my-index"
    template: "my-template"

    fields:
      "#{[@metadata][file][path]}"："%{@message}"  # 使用文件路径作为字段名称

在上面的配置中，我们使用了插值表达式将文件路径作为字段名称。这允许我们将每条日志存储在一个单独的字段中，从而便于搜索和分析。

插值表达式还可以在管道配置中使用。例如，我们可以使用插值表达式来将数据发送到不同的索引或类型。

processors:
  - split:
      field: "log"
      when:
        - condition: "contains"
          value: "ERROR"
        - condition: "contains"
          value: "WARN"
    
  - add_tag:
      when:
        - condition: "contains"
          value: "ERROR"
        - condition: "contains"
          value: "WARN"
      tags: ["error", "warn"]

  - drop_fields:
      fields: ["log"]

output:
  elasticsearch:
    hosts: ["localhost:9200"]
    template: "my-template"

    # 将包含 ERROR 或 WARN 的日志发送到 error-log 索引
    condition: "or"
    cases:
      - when: "has_tag"
        args: ["error", "warn"]
        index: "error-log"
        
      # 将不包含 ERROR 或 WARN 的日志发送到 info-log 索引
      - when: "not_has_tag"
        args: ["error", "warn"]
        index: "info-log"