Используя ELK, я не уверен, как использовать правильный фильтр в файле конфигурации logstash для пар

Question

У меня есть термин в долларах, заполненный вещами в этом формате ...

Сообщение 1: Stuff1, Header1 | Value1 | Header2 | Value2 | Header3 | Value3 |

Сообщение 2: Stuff1, Заголовок2 | Значение2 | Заголовок4 | Значение4 | Заголовок5 | Значение5 |

Сообщение 3: Материал1, Заголовок3 | Значение3 | Заголовок4 | Значение4 | Заголовок5 | Значение5 |

И т.д.

Мой текущий фильтр выглядит так:

filter {
    dissect {
      mapping => {
        "message" => "%{stuff1},%{dollarVars}"
                 }
            }

    mutate {
    convert => {
      "stuff1"=>"integer"
                }
           }

     date{ 
      match => ["dateTime", "yyyy/MM/dd HH:mm:ss"]
         }
 }

Что я могу сделать, чтобы взять мои расчлененные доллары и разделить их на заголовок с соответствующим значением (по одному справа от каждого заголовка)?

Я подозреваю, что это будет использовать фильтр CSV, используя "|"вместо "," и как-то говоря, каждый нечетный столбец dollarVar является заголовком, а каждый четный столбец идет со своим соответствующим (n-1) нечетным столбцом.

Answer 1

Вот что исправило то, что я пытался сделать.Нет необходимости изменять, только этот код рубина.

   ruby { 
    code => "
    i = 0
    vars = event.get('dollarVars')
    v_array = vars.split('|')
    while i < v_array.count
      event.set(v_array[i],v_array[i+1])
      i += 2
    end  
  "
          }

Answer 2

Я думаю, вы захотите разделить эти доллары, используя функцию разделения в mutate .

Оттуда я бы перешел на ruby ​​через фильтр ruby ​​, чтобы пройти через результирующий массив split и создать новые поля:

ruby {
  code => "
    i = 0
    while i < dollarVars.count
      event.set(dollarVars[i++]) = dollarVars[i++]
    end  
  "
}