Serilog HTTP sink + Logstash: разбиение массива сообщений Serilog на отдельные события журнала

Question

Мы используем Serilog HTTP sink для отправки сообщений в Logstash. Но тело сообщения HTTP таково:

{ 
    "events": [ 
    { 
     "Timestamp": "2016-11-03T00:09:11.4899425+01:00", 
     "Level": "Debug", 
     "MessageTemplate": "Logging {@Heartbeat} from {Computer}", 
     "RenderedMessage": "Logging { UserName: \"Mike\", UserDomainName: \"Home\" } from \"Workstation\"", 
     "Properties": { 
     "Heartbeat": { 
      "UserName": "Mike", 
      "UserDomainName": "Home" 
     }, 
     "Computer": "Workstation" 
     } 
    }, 
    { 
     "Timestamp": "2016-11-03T00:09:12.4905685+01:00", 
     "Level": "Debug", 
     "MessageTemplate": "Logging {@Heartbeat} from {Computer}", 
     "RenderedMessage": "Logging { UserName: \"Mike\", UserDomainName: \"Home\" } from \"Workstation\"", 
     "Properties": { 
     "Heartbeat": { 
      "UserName": "Mike", 
      "UserDomainName": "Home" 
     }, 
     "Computer": "Workstation" 
     } 
    } 
    ] 
} 

ie. события регистрации группируются в массиве. Можно отправлять сообщения один за другим, но он по-прежнему является массивом из одного элемента.

событие будет отображаться в Kibana как имеющий поле message со значением

{ 
    "events": [ 
    { 
     // ... 
    }, 
    { 
     // ... 
    } 
    ] 
} 

т.е.. буквально то, что исходило от ввода HTTP.

Как я могу разделить элементы в events массива отдельных событий лесозаготовительных и «подтянуть» свойства до верхнего уровня, так что я бы две записи событий в ElasticSearch:

---

"Timestamp": "2016-11-03T00:09:11.4899425+01:00", 
    "Level": "Debug", 
    "MessageTemplate": "Logging {@Heartbeat} from {Computer}", 
    "RenderedMessage": "Logging { UserName: \"Mike\", UserDomainName: \"Home\" } from \"Workstation\"", 
    "Properties": { 
    "Heartbeat": { 
     "UserName": "Mike", 
     "UserDomainName": "Home" 
    }, 
    "Computer": "Workstation" 
    } 

---

"Timestamp": "2016-11-03T00:09:12.4905685+01:00", 
    "Level": "Debug", 
    "MessageTemplate": "Logging {@Heartbeat} from {Computer}", 
    "RenderedMessage": "Logging { UserName: \"Mike\", UserDomainName: \"Home\" } from \"Workstation\"", 
    "Properties": { 
    "Heartbeat": { 
     "UserName": "Mike", 
     "UserDomainName": "Home" 
    }, 
    "Computer": "Workstation" 
    } 

---

Я попытался Logstash json и split, но я не могу это сделать Работа.

Answer 1

После обновления к Logstash 5.0 Val's solution прекратил работу из-за изменения Event API: обновление event.to_hash не было отражено в оригинале event. Для Logstash 5.0+ event.get('field') и event.set('field', value) должны использоваться аксессоры.

Обновленное решение теперь:

input { 
    http { 
    port => 8080 
    codec => json 
    } 
} 

filter { 
    split { 
    field => "events" 
    } 
    ruby { 
    code => " 
     event.get('events').each do |k, v| 
     event.set(k, v) 
     end 
    " 
    } 
    mutate { 
    remove_field => [ "events" ] 
    } 
} 

Answer 2

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

filter { 
    split { 
    field => "events" 
    } 
    ruby { 
    code => " 
     event.to_hash.update(event['events'].to_hash) 
     event.to_hash.delete_if {|k, v| k == 'events'}  
    " 
    } 
} 

В результате событие будет выглядеть следующим образом:

{ 
      "@version" => "1", 
     "@timestamp" => "2017-01-20T04:51:39.223Z", 
       "host" => "iMac.local", 
      "Timestamp" => "2016-11-03T00:09:12.4905685+01:00", 
       "Level" => "Debug", 
    "MessageTemplate" => "Logging {@Heartbeat} from {Computer}", 
    "RenderedMessage" => "Logging { UserName: \"Mike\", UserDomainName: \"Home\" } from \"Workstation\"", 
     "Properties" => { 
     "Heartbeat" => { 
        "UserName" => "Mike", 
      "UserDomainName" => "Home" 
     }, 
     "Computer" => "Workstation" 
    } 
}