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"
      }
    }
  ]
}

即。日志记录事件在一个数组中批处理。消息是可以一条一条发送的，但那还是单项数组。

然后该事件在 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

您可以使用额外的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"
    }
}

Answer 2

升级到 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 3

您现在可以通过设置 batchFormatter 来实现。默认批处理格式化程序会创建错误事件，但 ArrayBatchFormatter 会解决此问题：

 logger.WriteTo.DurableHttpUsingFileSizeRolledBuffers(
                    requestUri: new Uri($"http://{elasticHost}:{elasticPort}").ToString(),
                    batchFormatter: new ArrayBatchFormatter());