Newbie trying to use syslog date and time as timestamp in Logstash

Elastic Stack Logstash
1

Hello!

I know that I'm trying to do something simple, and I'm just getting started with Elasticsearch/Logstash, but I just can't seem to get the right combination, so I was hoping that someone might point me in the correct direction.

I am trying to import a bunch of old syslog messages into Elasticsearch using Logstash since I already have a Logstash configuration file which works for current syslog messages. The problem is that my prior syslog files require that the date be parsed and used for the timestamp. My problem is that I don't know how/where to put the date conversion code in my configuration file.

The logs that I want to ingest look like this:
2020-02-19 23:59:59 Local7.Notice 172.25.0.1 date=2020-02-19 time=23:59:59 devname="firewall" ...

Here is my Logstash configuration file:

input {
   file {
     path => "/home/templogs/parse.txt"
     type => "forti_log"
     start_position => "beginning"
     ignore_older => 3000000
  }
}

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

      mutate {

            rename => [ "dst", "dst_ip" ]
            rename => [ "dstip", "dst_ip" ]
            rename => [ "dstport", "dst_port" ]
            rename => [ "devname", "device_id" ]
            rename => [ "status", "action" ]
            rename => [ "src", "src_ip" ]
            rename => [ "srcip", "src_ip" ]
            rename => [ "zone", "src_intf" ]
            rename => [ "srcintf", "src_intf" ]
            rename => [ "srcport", "src_port" ]
            rename => [ "rcvd", "byte_recieved" ]
            rename => [ "rcvdbyte", "bytes_recieved" ]
            rename => [ "sentbyte", "bytes_sent" ]
            rename => [ "sent", "bytes_sent" ]
            convert => ["bytes_recieved", "integer"]
            convert => ["bytes_sent", "integer"]
            remove_field => [ "msg" ]
      }
  }
}

output {
  if [type] == "forti_log" {
    stdout { codec => rubydebug }
    elasticsearch {
      hosts => "localhost:9200"
      index => "forti-%{+YYYY.MM.dd}"
    }
  }
}

Can someone please show me where I should put the date match commands?

And far less importantly, it seems Logstash only reads newly written lines in my parse.txt file. Of course I can fake it out with "cat syslog >> parse.txt" after Logstash is running, but I wondered if there is a way to just have Logstash read through the file from top to bottom and then exit when the input file has been parsed.

Thanks for any offered help!

2

cat the file into a stdin input instead of using a file input.

Just before the kv filter you could add

dissect { mapping => { "message" => "%{[@metadata][timestamp]} %{+[@metadata][timestamp]} %{}" } }
date { match => [ "[@metadata][timestamp]", "YYYY-MM-dd HH:mm:ss" ] }
3

Thank you so much! This was very helpful and the datestamp worked perfectly, however I had a problem with the stdin.

input {
    stdin {}
}

And then ran the logstash command with the following output:

cat 2020-02-17.txt | /usr/share/logstash/bin/logstash --path.settings /home/templogs -f fortigate.conf --verbose

[INFO ] 2020-04-07 14:43:28.621 [Ruby-0-Thread-5: :1] elasticsearch - Using default mapping template
[INFO ] 2020-04-07 14:43:28.774 [Ruby-0-Thread-5: :1] elasticsearch - Attempting to install template {:manage_template=>{"index_patterns"=>"logstash-*", "version"=>60001, "settings"=>{"index.refresh_interval"=>"5s", "number_of_shards"=>1}, "mappings"=>{"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", "ignore_above"=>256}}}}}], "properties"=>{"@timestamp"=>{"type"=>"date"}, "@version"=>{"type"=>"keyword"}, "geoip"=>{"dynamic"=>true, "properties"=>{"ip"=>{"type"=>"ip"}, "location"=>{"type"=>"geo_point"}, "latitude"=>{"type"=>"half_float"}, "longitude"=>{"type"=>"half_float"}}}}}}}
[WARN ] 2020-04-07 14:43:28.866 [[main]-pipeline-manager] LazyDelegatingGauge - A gauge metric of an unknown type (org.jruby.specialized.RubyArrayOneObject) has been create for key: cluster_uuids. This may result in invalid serialization.  It is recommended to log an issue to the responsible developer/development team.
[INFO ] 2020-04-07 14:43:28.870 [[main]-pipeline-manager] javapipeline - Starting pipeline {:pipeline_id=>"main", "pipeline.workers"=>2, "pipeline.batch.size"=>125, "pipeline.batch.delay"=>50, "pipeline.max_inflight"=>250, "pipeline.sources"=>["/home/templogs/fortigate.conf"], :thread=>"#<Thread:0x548bd1a3 run>"}
[INFO ] 2020-04-07 14:43:30.393 [[main]-pipeline-manager] javapipeline - Pipeline started {"pipeline.id"=>"main"}
[INFO ] 2020-04-07 14:43:30.521 [Agent thread] agent - Pipelines running {:count=>1, :running_pipelines=>[:main], :non_running_pipelines=>[]}
[INFO ] 2020-04-07 14:43:31.026 [Api Webserver] agent - Successfully started Logstash API endpoint {:port=>9601}
[INFO ] 2020-04-07 14:43:31.798 [LogStash::Runner] runner - Logstash shut down.

Logstash started up then shutdown, but I didn't have any console output of the processed data (my configuration file did show the output when using the file input), nor was I able to find these records in a Kibana query. Any thoughts?

Thanks again!

4

I don't know why that would happen.

5

I'm sorry, it's my mistake. With my config file I needed to specify the type:

input {
    stdin {
      type => "forti_log"
    }
}

Thanks again! I truly appreciate your help!
dave

6

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