0

我正在使用经过的插件来计算时间和聚合插件然后显示它。我将自定义字段添加到经过的过滤器您可以在下面看到它：

 add_field => {
       "status" => "Status"
        "User" => "%{byUser}"
    }

一种是静态的，另一种是动态的，随事件而来。在输出logstash时，它只显示静态值而不是动态值..

它为动态显示 %{byUser}。但是对于任务 id 和状态字段来说工作得很好，我得到了正确的值。

知道为什么吗？

多一点代码

elapsed {
    unique_id_field => "assetId"
    start_tag => "tag1:tag2"
    end_tag => "tag3:tag4"
    add_field => {
       "wasInStatus" => "tag3"
       "User" => "%{byUser}"
    }
    add_tag => ["CustomTag"]
  }

grok 输入：

 grok {
        match => [
            "message", "%{TIMESTAMP_ISO8601:timestamp} %{NUMBER:assetId} %{WORD:event}:%{WORD:event1} User:%{USERNAME:byUser}"]


if "CustomTag" in [tags] and "elapsed" in [tags] {
    aggregate {
      task_id => "%{assetId}"
       code => "event.to_hash.merge!(map)"
        map_action => "create_or_update"
    }
  }

问题与：经过的过滤器有关：

new_event_on_match => true/false

在我的管道修复问题中将 new_event_on_match 更改为 false 是真的。但仍然想知道为什么。

4

1 回答 1

0

我现在也遇到了类似的问题，并找到了解决方法。使用时new_event_on_match => true，经过的事件将从原始日志中分离出来，并将新的经过的事件输入到 ElasticSearch 中，如下所示

{
  "_index": "elapsed_index_name",
  "_type": "doc",
  "_id": "DzO03mkBUePwPE-nv6I_",
  "_version": 1,
  "_score": null,
  "_source": {
    "execution_id": "dfiegfj3334fdsfsdweafe345435",
    "elapsed_timestamp_start": "2019-03-19T15:18:34.218Z",
    "tags": [
      "elapsed",
      "elapsed_match"
    ],
    "@timestamp": "2019-04-02T15:39:40.142Z",
    "host": "3f888b2ddeec",
    "cus_code": "Custom_name", [This is a custom field]
    "elapsed_time": 41.273,
    "@version": "1"
  },
  "fields": {
    "@timestamp": [
      "2019-04-02T15:39:40.142Z"
    ],
    "elapsed_timestamp_start": [
      "2019-03-19T15:18:34.218Z"
    ]
  },
  "sort": [
    1554219580142
  ]
}

为了将“cus_code”添加到原始日志中的已用事件对象（检测到已用过滤器结束标记的日志），我添加了一个聚合过滤器，如下所示：

if "elapsed_end_tag" in [tags] {
    aggregate {
       task_id => "%{execution_id}"
       code => "map['cus_code'] = event.get('custom_code_field_name')"
       map_action => "create"
    }
}

并通过验证“elapsed”标签添加聚合的结束块

if "elapsed" in [tags] {
       aggregate {
          task_id => "%{execution_id}"
          code => "event.set('cus_code', map['cus_code'])"
          map_action => "update"
          end_of_task => true
          timeout => 400
      }
}

因此，要将自定义字段添加到 elapsed 事件，我们需要将聚合过滤器与 elapse 过滤器结合起来

于 2019-04-02T16:02:02.817 回答