90-manageiq-input.conf

filter {
  if "cloudforms" in [tags] {

    grok {
      patterns_dir => ["/etc/logstash/patterns"]
      # number of metric captures on the queue for a zone
      match => { "message" => "%{MIQ_PREAMBLE}%{NUMBER:miq_capture_count:int} \"%{WORD:miq_capture_type}\"%{DATA}\[%{WORD:miq_capture_zone}\]" }
      add_tag => ["miq_capture_healthcheck", "grokked" ]
    }

    # EVM events - worker start, stop, vm start, etc.
    if "grokked" not in [tags] {
      grok {
        patterns_dir => ["/etc/logstash/patterns"]
        match => { "message" => "%{MIQ_PREAMBLE}%{DATA}MiqEvent.raise_evm_event%{DATA}Args: \[\[\"%{DATA:miq_event_source}\", %{NUMBER:miq_event_source_id}\], \"%{WORD:miq_event}\", {%{DATA}:type=>\"%{DATA:miq_event_type}\"}" }
        add_tag => [ "grokked" ]
      }
    }

    # dequeue times for tasks popped off the queue
    if "grokked" not in [tags] {
      grok {
        patterns_dir => ["/etc/logstash/patterns"]
        match => { "message" => "%{MIQ_PREAMBLE}%{DATA}Zone: \[%{WORD:miq_message_zone}?\]%{DATA}Role: \[%{WORD:miq_message_role}?\]%{DATA}Command: \[%{DATA:miq_message_command}\]%{DATA}Deliver On: \[%{DATA:miq_message_deliver_on}?\]%{DATA}Dequeued in: \[%{NUMBER:miq_message_dequeue_in:float}\]" }
        add_tag => [ "miq_dequeue_message", "grokked" ]
      }
    }

    # some extra post-processing for tags
    # if they have a deliver on time, let's calculate the
    # time between deliver_on and the generation of the log message
    if "miq_dequeue_message" in [tags] {
      ruby {
        code => "
          require 'date'
          if event.get('miq_message_deliver_on').nil?
            event.set('miq_message_latency', event.get('miq_message_dequeue_in'))

          else
            deliver_on = DateTime.parse(event.get('miq_message_deliver_on'))
            msg_timestamp = DateTime.parse(event.get('miq_timestamp'))
            delta = msg_timestamp - deliver_on

            # take the better of the two - seen situations where the dequeue in
            # seconds is less than delta, probably because there's a delay writing the log
            # message. buffering?
            delta = [event.get('miq_message_dequeue_in'), (delta*3600*24)].min
            event.set('miq_message_latency',delta.to_f)
          end
        "
      }

      # cleanup - we don't need this
      mutate {
        remove_field => ['miq_message_deliver_on']
      }
    }

    # this message is for EMS refresh timings
    # we do extra grokking for this, returning further events.
    if "grokked" not in [tags] {
      grok {
        patterns_dir => ["/etc/logstash/patterns"]
        # EMS refresh timings
        match => { "message" => "%{MIQ_PREAMBLE}EMS:%{SPACE}\[%{DATA:miq_refresh_ems}\]%{DATA}Refreshing targets for EMS...Complete - Timings%{SPACE}{%{DATA:miq_refresh_timings}}" }
        add_tag => [ "miq_refresh_timings", "grokked" ]
      }
    }

    # further parse the EMS refresh timings with our ruby script
    if "miq_refresh_timings" in [tags] {
      ruby {
        code => "
          timings = eval('{' + event.get('miq_refresh_timings') + '}')
          timings.each do |k, v|
            event.set('miq_refresh_timing_' + k.to_s, v)
          end
        "
      }

      # cleanup - we don't need this
      mutate {
        remove_field => ['miq_refresh_timings']
      }

    }

    # catch all for everything else.
    if "grokked" not in [tags] {
      grok {
        patterns_dir => [ "/etc/logstash/patterns" ]
        match => { "message" => "%{MIQ_PREAMBLE}%{GREEDYDATA:miq_msg}" }
        add_tag => [ "grokked" ]
      }
    }

    # cleanup.
    if "grokked" in [tags] {

      # replace the logstash @timestamp with the timestamp from the log line
      date {
        match => [ "miq_timestamp", "ISO8601" ]
        timezone => "UTC"
      }

      mutate {
        remove_field => ["fields", "input_type", "offset", "program"]
        remove_tag => ["beats_input_codec_plain_applied", "_grokparsefailure", "grokked"]
      }
    }
  }
}