Getting mutate_error while trying to use ECS schema

Elastic Stack Logstash
1

I was trying to ingest Fortigate firewall logs and under ECS schema. Here is my configuration:

input {
  udp {
    host => "0.0.0.0"
    port => 10514
    type => "log"
  }
}


filter {
 if [type] == "log" {
 kv {
      source => "message"
        exclude_keys => [ "type", "subtype" ] 
}
        geoip { source => "dst" }
        geoip { source => "dstip" }
        geoip { source => "src" }
        geoip { source => "srcip" }

        mutate {

            rename => [ "zone", "src_intf" ]
            rename => [ "devid", "[host][id]" ]
            rename => [ "device_id", "[host][name]" ]
            rename => [ "host", "[host][ip]" ]
            rename => [ "level", "[log][level]" ]
            convert => ["[destination][bytes]", "integer"]
            convert => ["[source][bytes]", "integer"]
            remove_field => [ "msg" ]
    }
  }
}

output {
if [type] == "log" {
 elasticsearch {
            cloud_id => "XXXXX"
            cloud_auth => "YYYY"
            index => "logs-%{+YYYY.MM.dd}"
       }
}
}

But running this, I am seeing a message in the logstash console like this:

[WARN ] 2020-07-10 12:22:55.572 [[main]>worker1] mutate - Exception caught while applying mutate filter {:exception=>"Could not set field 'id' on object '222.66.327.233' to value 'FGT3172132213'.This is probably due to trying to set a field like [foo][bar] = someValuewhen [foo] is not either a map or a string"}

Also here is a sample document. The field structure is not coming properly as per the Logstash configuration. Example, device_id is not mutated to host.name

{
    "<189>date": "2020-07-09",
    "action": "deny",
    "app": "tcp/17167",
    "appcat": "unscanned",
    "bytes_recieved": 0,
    "bytes_sent": 0,
    "craction": "262144",
    "crlevel": "low",
    "crscore": "5",
    "device_id": "StarWars",
    "devid": "FGT3172132213",
    "dst_ip": "222.66.327.233",
    "dst_port": "12343",
    "dstcountry": "India",
    "dstintf": "root",
    "dstintfrole": "undefined",
    "duration": "0",
    "host": "222.66.327.233",
    "level": "notice",
    "logid": "0023000014",
    "logtime": "1594296011",
    "policyid": "0",
    "policytype": "local-in-policy",
    "proto": "6",
    "sentpkt": "0",
    "service": "tcp/17167",
    "sessionid": "34323342",
    "src_intf": "wan",
    "src_ip": "49.51.51.58"
}
2

At this point host is still a string. Therefore it cannot have a subfield.
And you should correct your syntax like this instead of repeating the same filter option again and again and again:

mutate {
  rename => {
    "zone" => "src_intf"
    "action" => "[event][action]"
    ...
  }
}
1 Like
3

@Jenni
would try this and get back to you

4

@curiousmind

The ordering of fields in the mutate plugin is creating your issue.
You can do like this:

rename => {
                "zone"=> "src_intf" 
                "host" => "[host][ip]" 
                "devid"=> ""[host][id]" 
                "device_id"=> "[host][name]" 
                "level" => "[log][level]" 
          }
  convert => ["[destination][bytes]", "integer"]
  convert => ["[source][bytes]", "integer"]
  remove_field => [ "msg" ]

What is different is that, the "host" field gets mutated first as an object, then the other operations on the "host" field will happen smoothly.
In your configuration the "host" field is registred as a string field and then you are trying to set inner objects inside it, which will not be allowed.

3 Likes
5

Okay got it.
Now the configuration is working.

Thanks

6

This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.