Прошедшее + совокупное прохождение пользовательских полей в Logstash

Question

Я использую истекший плагин, чтобы вычислить время и агрегировать плагин, чтобы затем отобразить его. Я добавил настраиваемые поля в прошедший фильтр. Вы можете увидеть это ниже:

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

Один статичен, другой динамично идет с событием. На выходе logstash отображаются только статические значения, а не динамические.

Он отображает%{byUser} для динамического. Но для идентификатора задачи и полей состояния работает просто отлично, и я получил правильные значения.

Есть идеи почему?

Немного больше кода

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

входной сигнал:

 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, было верно в моей исправленной проблеме конвейера. Но все еще задаюсь вопросом, почему.

Answer 1

Я также столкнулся с подобной проблемой сейчас, и нашел решение для нее. когда 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
      }
}

Таким образом, чтобы добавить настраиваемое поле к прошедшему событию, нам нужно объединить совокупный фильтр вместе с фильтром истечения