Logsta sh для объединения неструктурированных строк журнала в одно событие - PullRequest
Новая
       78

Logsta sh для объединения неструктурированных строк журнала в одно событие

0 голосов
/ 08 мая 2020

У меня такой формат журнала 

[23:59:43] TYPE[student03]
[23:59:43] Write Ack!
[23:59:43] <application><status>Active</status></application>
[23:59:43] --------PushToFile Start--------
[23:59:43] --------PushToFile Name --------
[23:59:43] --------PushToFile End--------
[23:59:47] --------on parser--------
[23:59:47] ClientIPAddress > 123.12.123.2
[23:59:43] TYPE[student04]
[23:59:43] Write Ack!
[23:59:43] FGHSFHG3453D56HJ3456FJ372GJ1387FFKJH
[23:59:43] --------PushToFile Start--------
[23:59:43] --------PushToFile Name --------
[23:59:43] --------PushToFile End--------
[23:59:47] --------on parser--------
[23:59:47] ClientIPAddress > 123.12.123.2
[23:59:43] TYPE[student03]
[23:59:43] Write Ack!
[23:59:43] FGHSFHG3453D56HJ3456FJ372GJ1387FFKJH
[23:59:43] --------PushToFile Start--------
[23:59:43] --------PushToFile Name --------
[23:59:43] --------PushToFile End--------
[23:59:47] --------on parser--------
[23:59:47] ClientIPAddress > 123.12.123.2

Я хочу проанализировать эти журналы, чтобы получить только ТИП [student03] и статус студента. Остальные строчки сбрасываю фильтрами losta sh. Теперь мне нужен ТИП и статус как одно событие, чтобы я мог отправить его в rabbitmq как одно сообщение, например

[23:59:43]-student03-<application><status>Active</status></application>

Как мы можем добиться этого с помощью агрегированных фильтров? Я пробовал агрегатные фильтры. Но я думаю, что нам нужен один и тот же шаблон в обеих строках журнала для слияния с агрегатом. или любые альтернативы для этого

Пожалуйста, помогите!

My logsta sh файл конфигурации 

    beats {
     port => 5044
   }
}
filter {

  if ([message] =~ "Write Ack!"){
   drop{}
  }
  else if ([message] =~ "PushToFile Start"){
    drop{}
  }
  else if ([message] =~ "PushToFile Name"){
    drop{}
  }
  else if ([message] =~ "PushToFile End"){
    drop{}
  }
  else if ([message] =~ "on parser"){
     drop{}
  }  
  else if ([message] =~ "ClientIPAddress") {
    drop { }
  }                     
}
output {
 stdout { codec => rubydebug }
}

1 Ответ

0 голосов
/ 09 мая 2020

Вы можете использовать фильтр Grok (см. Ниже) для анализа только двух требуемых строк, а затем использовать приведенный ниже Ruby фильтр для выполнения необходимых строковых операций и сохранения последнее сообщение в поле и используйте его.

Grok filter 

grok{
    match => {"message" => "%{TIME}] TYPE.%{GREEDYDATA:STATUS}."}
    match => {"message" => "%{TIME}] <application><status>Active</status></application>"}
    }
if "_grokparsefailure" in [tags]{drop{}}

Ruby filter 

ruby{
    code => "
        if event.get('STATUS')
            @save_status = event.get('STATUS')
            event.cancel()
        else
            event.set('final_message', event.get('message').sub(' ', '-'+@save_status+'-'))
        end
    "
}

Обратите внимание, что это работает, только если ваши журналы идут последовательно.

...