Can't see Suricata events in Kibana

Elastic Stack Logstash
1

I'm trying to send Suricata event to the ELK stack . I have the ELK stack on a server and on an other server I have Suricata and Filebeat. I activated the filebeat suricata module and Filebeat send event from eve.json file to Logstash.

At the begining I saw this message on the Kibana logs panel :
failed to format message from /var/log/suricata/eve.json
But i saw on the suricata events overviws dashboard that events are send but I don't saw analyse of this events.

After some research and after change config file, I don't have any more the previous message but unfortunately, I have no more events on the dashboard.

My differents config file :

input {
  beats {
    port => 5044
    codec => "json_lines"
  }
}

filter {
  if [application] == "suricata" {
    date {
      match => [ "timestamp", "ISO8601" ]
    }
    ruby {
      code => "
      if event.get('[event_type]') == 'fileinfo'
         event.set('[fileinfo][type]', event.get('[fileinfo][magic]').to_s.split(',')[0])
      end
      "
    }
    if [src_ip]  {
    geoip {
      source => "src_ip"
      target => "geoip"
      database => "/etc/logstash/GeoLite2-City.mmdb"
      add_field => [ "[geoip][coordinates]", "%{[geoip][longitude]}" ]
      add_field => [ "[geoip][coordinates]", "%{[geoip][latitude]}"  ]
    }
    mutate {
      convert => [ "[geoip][coordinates]", "float" ]
    }
    if ![geoip.ip] {
      if [dest_ip]  {
        geoip {
          source => "dest_ip"
          target => "geoip"
          database => "/etc/logstash/GeoLite2-City.mmdb"
          add_field => [ "[geoip][coordinates]", "%{[geoip][longitude]}" ]
          add_field => [ "[geoip][coordinates]", "%{[geoip][latitude]}"  ]
        }
        mutate {
          convert => [ "[geoip][coordinates]", "float" ]
        }
      }
    }
  }
 }
}

{
  "template" : "suricata-*",
  "version" : 50001,
  "settings" : {
    "number_of_replicas": 0,
    "index.refresh_interval" : "5s"
  },
  "mappings" : {
    "_default_" : {
      "_all" : {"enabled" : true, "norms" : false},
      "dynamic_templates" : [ {
        "message_field" : {
          "path_match" : "message",
          "match_mapping_type" : "string",
          "mapping" : {
            "type" : "text",
            "norms" : false
          }
        }
      }, {
        "string_fields" : {
          "match" : "*",
          "match_mapping_type" : "string",
          "mapping" : {
            "type" : "text", "norms" : false,
            "fields" : {
              "keyword" : { "type": "keyword", "index": "not_analyzed", "ignore_above": 256 },
              "raw" : { "type": "keyword", "index": "not_analyzed", "ignore_above": 256 }
            }
          }
        }
      } ],
      "properties" : {
        "@timestamp": { "type": "date", "include_in_all": false },
        "@version": { "type": "keyword", "include_in_all": false },
        "geoip"  : {
          "dynamic": true,
          "properties" : {
            "ip": { "type": "ip" },
            "location" : { "type" : "geo_point" },
            "latitude" : { "type" : "half_float" },
            "longitude" : { "type" : "half_float" }
          }
        },
        "dest_ip": {
            "type": "ip",
            "fields": {
                "raw": {"index": "not_analyzed", "type": "keyword"},
                "keyword": {"index": "not_analyzed", "type": "keyword"}
             }
        },
        "src_ip": {
            "type": "ip",
            "fields": {
                "raw": {"index": "not_analyzed", "type": "keyword"},
                "keyword": {"index": "not_analyzed", "type": "keyword"}
             }
        }
      }
    }
  }
}


output {
 if [application] == "suricata" {
  elasticsearch {
   hosts => ["localhost:9200"]
   index => "%{[@metadata][beat]}-%{[@metadata][version]}-%{+YYYY.MM.dd}"
   #document_type => "suricata"
   template => "/etc/logstash/elasticsearch5-template.json"
   template_overwrite => false
  }
 }
}

Thank you in advance !

2

Usually you want to use the stdout plugin in Logstash to see if events are processed properly.