Serilog HTTP接收器+ 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爲具有值

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

即字段message。字面上來自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" 
    } 
}