logstash elasticsearch output

Question

I'm having issues while I've tried to change the naming convention for an elasticsearch index in a logstash conf file. I need to use the part of the file name which will be passed through a logstash pipeline, exactly the part which set the date of the data contained by the file. So, instead of using the standard naming convention, which is, as long I can read: logstash-%{+YYYY.MM.DD}, I need this: -.

I was trying to get the actual name of the file which is currently passed through the pipeline, but I do not know how to get it. Then I decided to use the year and the month of the current line which is being processed by in the filter section. This is the grok pattern I use:

  grok {
    match => [ "message", "%{IP:client} %{NOTSPACE:sep} %{NOTSPACE:ident} %{NOTSPACE:inbracket}%{MONTHDAY:day}/%{MONTH:month}/%{YEAR:year}:%{HOUR:hour}:%{MINUTE:minute}:%{SECOND:second} %{ISO8601_TIMEZONE:tz}%{NOTSPACE:outbracket} \"%{WORD:method} %{NOTSPACE:uri} %{NOTSPACE:http_version}\" %{NUMBER:code} %{NUMBER:size} %{NOTSPACE:action_hierarchy} %{NOTSPACE:content_type}" ]
    remove_field => ["sep"]
    remove_field => ["inbracket"]
    remove_field => ["outbracket"]
  }

As it can be seen, "year"and "month"are two of the fields I can recover after applying grok pattern. SO I thought I could do this:

elasticsearch {
    action => "index"
    index => "myindexname-%{year}.%{month}"
    index_type => "logs"
    node_name => "Node001"
}

but nor "year" nor "month" can be used in this section: there's no compilation problem in the conf file, simply that is not the way for getting those values. Maybe using ruby could be a way, but my attempts were all wrong. How can I achieve this?

So, for everyone who could have the same issue, this is a solution which works for me.

The code for my plugin is:

# Call this file 'ordinalmonth.rb' (in logstash/filters, as above)
require "logstash/filters/base"
require "logstash/namespace"

class LogStash::Filters::OrdinalMonth < LogStash::Filters::Base

# Setting the config_name here is required. This is how you
# configure this filter from your logstash config.
#
# filter {
#   ordinalmonth { ... }
# }
  config_name "ordinalmonth"

# New plugins should start life at milestone 1.
milestone 2

# Replace the message with this value.
config :month_field, :validate => :string, :default => "month"

public
def register
  # nothing to do
end # def register

public
def filter(event)
  # return nothing unless there's an actual filter event
  return unless filter?(event)
  if event[@month_field]
    # Replace the event message with our message as configured in the
    # config file.
    tmp = case event[@month_field]
      when "Jan" then "01"
      when "Feb" then "02"
      when "Mar" then "03"
      when "Apr" then "04"
      when "May" then "05"
      when 'Jun' then '06'
      when "Jul" then "07"
      when "Aug" then "08"
      when "Sep" then "09"
      when "Oct" then "10"
      when "Nov" then "11"
      when "Dec" then "12"
      else "Unknown"
      end
    event["month"] = tmp
  end
  # filter_matched should go in the last line of our successful code 
  filter_matched(event)
end # def filter
end # class LogStash::Filters::OrdinalMonth

Basically, the plugin receive the name of the filed which contain the name of the month with 3 letters, starting with a capital one. Then enters in the case statement, so the updating can be achieved. Then, it alters the old value contained in the field.

So, for this works in a expected manner, I had to change the code in the configuration file for my logstash job:

filter {
  if [type] == "nauta_navroom" {
      grok {
        match => [ "message", "%{IP:client} %{NOTSPACE:sep} %{NOTSPACE:ident} %{NOTSPACE:inbracket}%{NOTSPACE:day}/%{MONTH:month}/%{YEAR:year}:%{HOUR:hour}:%{MINUTE:minute}:%{SECOND:second} %{ISO8601_TIMEZONE:tz}%{NOTSPACE:outbracket} \"%{WORD:method} %{NOTSPACE:uri} %{NOTSPACE:http_version}\" %{NUMBER:code} %{NUMBER:size} %{NOTSPACE:action_hierarchy} %{NOTSPACE:content_type}" ]
        remove_field => ["sep"]
        remove_field => ["inbracket"]
        remove_field => ["outbracket"]
      }
      ordinalmonth {}
      kv {
        source => "@message"
      }
  }
}

Check out the invocation of the ordinalmonth plugin, without any parameters. The other magic thing was using the kv filter, who actually makes the changes visible outside the filter.

And that's it. I hope this can be useful to anyone who needs it.

Alcanzar · Accepted Answer · 2015-04-14 20:52:40Z

Logstash builds the index name from the time in the @timestamp field (which defaults to "now"). What you will want to do is parser the time from the file and use that to set your timestamp.

For you example, you have a pattern of %{MONTHDAY:day}/%{MONTH:month}/%{YEAR:year}:%{HOUR:hour}:%{MINUTE:minute}:%{SECOND:second} %{ISO8601_TIMEZONE:tz} in your file, so you could do something like this in your config file:

mutate {
  add_field => [ "timestamp", "%{year}-%{month}-%{day}T%{hour}:%{minute}:%{second}%{tz}" ]
}
date {
  match => [ "timestamp", "ISO8601" ]
  remove_field => ["timestamp" ]
}

Which adds a timestamp field to your event based on what you parsed out, then sets @timstamp based on it and removes the field that was added.

Then you'll just need to change your elasticsearch output to

elasticsearch {
    action => "index"
    index => "myindexname-%{+YYYY-MM}"
    index_type => "logs"
    node_name => "Node001"
}

Jorge Olmedo · Accepted Answer · 2015-04-17 21:58:37Z

So, for everyone who could have the same issue, this is a solution which works for me.

The code for my plugin is:

# Call this file 'ordinalmonth.rb' (in logstash/filters, as above)
require "logstash/filters/base"
require "logstash/namespace"

class LogStash::Filters::OrdinalMonth < LogStash::Filters::Base

# Setting the config_name here is required. This is how you
# configure this filter from your logstash config.
#
# filter {
#   ordinalmonth { ... }
# }
  config_name "ordinalmonth"

# New plugins should start life at milestone 1.
milestone 2

# Replace the message with this value.
config :month_field, :validate => :string, :default => "month"

public
def register
  # nothing to do
end # def register

public
def filter(event)
  # return nothing unless there's an actual filter event
  return unless filter?(event)
  if event[@month_field]
    # Replace the event message with our message as configured in the
    # config file.
    tmp = case event[@month_field]
      when "Jan" then "01"
      when "Feb" then "02"
      when "Mar" then "03"
      when "Apr" then "04"
      when "May" then "05"
      when 'Jun' then '06'
      when "Jul" then "07"
      when "Aug" then "08"
      when "Sep" then "09"
      when "Oct" then "10"
      when "Nov" then "11"
      when "Dec" then "12"
      else "Unknown"
      end
    event["month"] = tmp
  end
  # filter_matched should go in the last line of our successful code 
  filter_matched(event)
end # def filter
end # class LogStash::Filters::OrdinalMonth

Basically, the plugin receive the name of the filed which contain the name of the month with 3 letters, starting with a capital one. Then enters in the case statement, so the updating can be achieved. Then, it alters the old value contained in the field.

So, for this works in a expected manner, I had to change the code in the configuration file for my logstash job:

filter {
  if [type] == "nauta_navroom" {
      grok {
        match => [ "message", "%{IP:client} %{NOTSPACE:sep} %{NOTSPACE:ident} %{NOTSPACE:inbracket}%{NOTSPACE:day}/%{MONTH:month}/%{YEAR:year}:%{HOUR:hour}:%{MINUTE:minute}:%{SECOND:second} %{ISO8601_TIMEZONE:tz}%{NOTSPACE:outbracket} \"%{WORD:method} %{NOTSPACE:uri} %{NOTSPACE:http_version}\" %{NUMBER:code} %{NUMBER:size} %{NOTSPACE:action_hierarchy} %{NOTSPACE:content_type}" ]
        remove_field => ["sep"]
        remove_field => ["inbracket"]
        remove_field => ["outbracket"]
      }
      ordinalmonth {}
      kv {
        source => "@message"
      }
  }
}

Check out the invocation of the ordinalmonth plugin, without any parameters. The other magic thing was using the kv filter, who actually makes the changes visible outside the filter.

And that's it. I hope this can be useful to anyone who needs it. Thanks for your attention. Jorge.

Collectives™ on Stack Overflow

logstash elasticsearch output

2 Answers 2

Comments

Comments

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

2 Answers 2

Comments

Comments

Your Answer

Sign up or log in

Post as a guest

Related