Making Logstash play nice with ECS (and filebeat)

While trying to test filebeat’s suricata module to parse suricata events and then pass the events on to logstash, I discovered a potential conflict between filebeat and logstash when using ECS. Hopefully this helps someone else.

BLUF:

Filebeat will apply ECS format whereas logstash will not.
If they both write to the same index in elasticsearch this can cause a type conflict especially for the value of “host”.

To fix this, either:
(1) index these events to separate indices or
(2) convert logstash host fields to beECS compliant.

I chose #2 with the following filter:

filter {
  if ![ecs] {
    mutate {
      rename => ["host", "[host][name]" ]
      update => { "[ecs][version]" => "1.5.0" }
      add_tag => [ "ecs_converted" ]
    }
  }
}

The longer story:

I’ve recently upgraded to elk 7.x by way of using Elastic Cloud.

I’m using filebeat -> logstash -> elasticsearch (cloud)

I kept running into an issue where my test log entries would not index to elasticsearch. I would receive the following error:

[2020-06-01T13:50:43,375][WARN ][logstash.outputs.elasticsearch][main][0ad4de10f2da5a7f63fca634e4229c098694dea6c5fbaa116636672b674099d4] Could not index event to Elasticsearch. {:status=>400, :action=>["index", {:_id=>nil, :_index=>"logstash-test3-2020.06.01", :routing=>nil, :_type=>"_doc"}, #<LogStash::Event:0xd971a0c>], :response=>{"index"=>{"_index"=>"logstash-test3-2020.06.01", "_type"=>"_doc", "_id"=>"fMYmcHIB0zHVy2MkFP0u", "status"=>400, "error"=>{"type"=>"mapper_parsing_exception", "reason"=>"object mapping for [host.hostname] tried to parse field [host.hostname] as object, but found a concrete value"}}}}

The reason was because I had originally used filebeat to add my first entry to the index. Dynamic mapping assigned “host” to be an object since filebeat starting in 7, host is an object as per ECS spec. This is what the entry looks like in filebeat:

"host" => {"name" => "0284934ff28b"},

But when I later added a test entry from logstash to the same index, logstash tried to add it as a text value which looks like this:

"host" => "dev-datapipeline",

Of course this conflicts with the mapping. This behavior changed when filebeat started using ECS. This seems like a major conflict for elastic to fix, at least an option to make logstash ECS compatible.

To work around this I added a mutate filter to convert the host field from a text value to an object:

filter {
  mutate { rename => ["host", "[host][name]" ]}
}

This fixed logstash, but then any filebeat entry was also converted:

"host" => {
  "name" => {
    "name" => "0284934ff28b"
  }
}

Doh!

At first I tried to check if the type of host is text vs an object, but unfortunately you cannot do that in logstash without adding ruby code. For my case, I decided on a simpler method: Check if the the event is already in ECS format, if not convert it to ECS, and add a tag indicating the conversion plus the ECS version

Filebeat flags that it us using ECS by adding an ecs field:

ecs.version => 1.5.0

I check to see if ecs is not set:

filter {
  if ![ecs] {
    mutate {
      rename => ["host", "[host][name]" ]
      update => { "[ecs][version]" => "1.5.0" }
      add_tag => [ "ecs_converted" ]
    }
  }
}

New end result with this filter added (see the entries in bold):

Logstash:

{
  "path" => "/usr/share/logstash/data/test/hello3.txt",
  "@timestamp" => 2020-06-01T17:34:21.069Z,
  "@metadata" => {
    "host" => "dev-datapipeline",
    "index-header" => "log",
    "path" => "/usr/share/logstash/data/test/hello3.txt"
  },
  "host" => {
    "name" => "dev-datapipeline"
  },
  "@version" => "1",
  "message" => "Hello from logstash!",
  "tags" => [
   [0] "test",
   [1] "ecs_converted"
   ]
}

Filebeat:

{
  "input" => {
    "type" => "log"
  },
  "agent" => {
    "hostname" => "0284934ff28b",
    "id" => "dcbc70af-f7f8-421a-a17d-a36d4b112106",
    "type" => "filebeat",
    "ephemeral_id" => "d66c7a57-146f-4751-b958-b661c4a14159",
    "version" => "7.7.0"
  },
  "@timestamp" => 2020-06-01T17:37:11.097Z,
  "ecs" => {
    "version" => "1.5.0"
  },
  "log" => {
    "file" => {
      "path" => "/usr/share/filebeat/data/test/fb-test.txt"
    },
    "offset" => 127
  },
  "@metadata" => {
    "ip_address" => "172.25.0.3",
    "index-header" => "log-beat",
    "type" => "_doc",
    "beat" => "filebeat",
    "version" => "7.7.0"
  },
  "host" => {
    "name" => "0284934ff28b"
  },
  "@version" => "1",
  "message" => "Hello from filebeat!",
  "tags" => [
    [0] "test",
    [1] "beat",
    [2] "beats_input_codec_plain_applied"
  ]
}

Success!