logstash-indexer.conf

# this configuration is to parse the syslog files and then put them into redis.

input {
    file {
        # this input is a dump of sendmail and proofpoint logs, it's really messy
        type => "proofpoint_raw_type"
        tags => "proofpoint_raw"
        path => ["/home/conrad/testlog.log"]
    }
    #stdin {
    #    type => "proofpoint_raw_type"
    #    tags => "proofpoint_raw"
    #}
}

filter {
    grok {
        tags => "proofpoint_raw"
        # this matches the 'user unknown' messages from sendmail
        # match May 21 00:01:18 pp-serve01 sendmail[16212]: r4L41GBT016212: <[email protected]>... User unknown
        pattern => "%{SYSLOGTIMESTAMP}%{SPACE}%{HOSTNAME:hostname}%{SPACE}%{WORD:logtype}\[%{INT:process_pid}\]:%{SPACE}%{WORD:smtp_messageid}:%{SPACE}<%{GREEDYDATA:user_unknown}>"
        add_tag => "user_unknown"
        tag_on_failure => "not_user_unknown"
        remove_tag => "proofpoint_raw"
    }

    grok {
        tags => "proofpoint_raw"
        # "keys" - not really key/value since it's both key/value & csv
        # arg1/relay/reject
        # match: May 21 01:02:52 pp-serve01 sendmail[28155]: r4L52peu028155: ruleset=check_mail, arg1=<[email protected]>, relay=2.32.53.48.megaline.example.kz [2.32.53.48] (may be forged), reject=553 5.1.8 <[email protected]>... Domain of sender address [email protected] does not exist
        # grok: %{SYSLOGTIMESTAMP}%{SPACE}%{HOSTNAME:hostname}%{SPACE}%{WORD:logtype}\[%{INT:process_pid}\]:%{SPACE}%{WORD:smtp_messageid}:%{SPACE}ruleset=%{DATA:ruleset},%{SPACE}arg1=<%{DATA:arg1}>,%{SPACE}relay=%{DATA:relay},%{SPACE}reject=%{GREEDYDATA:reject}
        pattern => "%{SYSLOGTIMESTAMP}%{SPACE}%{HOSTNAME:hostname}%{SPACE}%{WORD:logtype}\[%{INT:process_pid}\]:%{SPACE}%{WORD:smtp_messageid}:%{SPACE}ruleset=%{DATA:ruleset},%{SPACE}arg1=<%{DATA:arg1}>,%{SPACE}relay=%{DATA:relay},%{SPACE}reject=%{GREEDYDATA:reject}"
        add_tag => "sendmail_message"
        tag_on_failure => "not_sendmail_message"
        remove_tag => "proofpoint_raw"
    }

    grok {
        tags => "proofpoint_raw"
        # it'd be too easy if sendmail sent a single log type.
        # This should pick up remaining sendmail logs
        # orig pattern => "%{SYSLOGTIMESTAMP}%{SPACE}%{HOSTNAME:hostname}%{SPACE}%{WORD:logtype}\[%{INT:process_pid}\]:%{SPACE}%{WORD:smtp_messageid}:%{SPACE}%{GREEDYDATA:tokvparse}"
        pattern => "%{SYSLOGTIMESTAMP}%{SPACE}%{HOSTNAME:hostname}%{SPACE}%{WORD:logtype}\[%{INT:process_pid}\]:%{SPACE}%{WORD:smtp_messageid}:%{SPACE}%{GREEDYDATA:sendmail_message}"
        add_tag => "sendmail_message"
        tag_on_failure => "not_sendmail_message2"
        remove_tag => "proofpoint_raw"
    }

    mutate {
        tags => "sendmail_message"
        gsub => [ "arg1", ",", "" ]
        gsub => [ "relay", ",", "" ]
        gsub => [ "reject", ",", "" ]
    }

    grok {
        tags => "proofpoint_raw"
        # match May 21 00:00:00 pp-serve01 filter_instance1[5614]: rprt s=1cdv4jjmcx mod=session cmd=dispose module=access rule=netmlx action=reject value="550 5.7.0 Local Policy Violation - ${DnsblResult_netmlx}" 
        pattern => "%{SYSLOGTIMESTAMP}%{SPACE}%{HOSTNAME:hostname}%{SPACE}%{WORD:logtype}\[%{INT:process_pid}\]:%{SPACE}rprt%{SPACE}%{GREEDYDATA:tokvparse}"
        add_tag => "instance_message"
        tag_on_failure => "not_instance_message"
        remove_tag => "proofpoint_raw"
    }

    kv {
        tags => "instance_message"
        source => "tokvparse"
        value_split => "="
    }

    mutate {
        tags => "instance_message"
        remove => ["tokvparse"]
    }

    mutate {
        tags => "sendmail_message"
        remove => ["tokvparse"]
    }

    mutate {
        tags => "instance_message"
        add_field => ["user_unknown", "false"]
    }

    mutate {
        tags => "sendmail_message"
        add_field => ["user_unknown", "false"]
    }

    mutate {
        replace => [ "@source_host", "%{hostname}" ]
    }

    mutate {
        # at this piont all that should be left are logs that logstash somehow missed.
        tags => "proofpoint_raw"
        add_tag => "logstash_missed"
    }

    mutate {
        # this will remove the not_ tags we created above just to keep thing neat at this point. _grokfailurs should still bubble up.
        # There is probably a better way to do this.
        # note, the array in this context is to 'match all'
        tags => "sendmail_message"
        remove_tag => ["not_sendmail_message2", "not_user_unknown", "not_instance_message", "not_sendmail_message"]
    }

    mutate {
        tags => "instance_message"
        remove_tag => ["not_sendmail_message2", "not_user_unknown", "not_instance_message", "not_sendmail_message"]
    }
    mutate {
        tags => "user_unknown"
        remove_tag => ["not_sendmail_message2", "not_user_unknown", "not_instance_message", "not_sendmail_message"]
    }

}

output {
#  stdout { 
#    debug => true
#    debug_format => "json"
#  }

  elasticsearch {
    tags => "sendmail_message"
    embedded => false
    index => "logstash-proofpoint-%{+YYYY.MM.dd}"
    host => "localhost"
  }
  elasticsearch {
    tags => "instance_message"
    embedded => false
    index => "logstash-proofpoint-%{+YYYY.MM.dd}"
    host => "localhost"
  }
  elasticsearch {
    tags => "user_unknown"
    embedded => false
    index => "logstash-proofpoint-%{+YYYY.MM.dd}"
    host => "localhost"
  }
  elasticsearch {
    tags => "logstash_missed"
    embedded => false
    index => "logstash-proofpoint-%{+YYYY.MM.dd}"
    host => "localhost"
  }
}

this configuration is to parse the syslog files and then put them into redis.

https://gist.github.com/timconradinc/5636363#file-logstash-indexer-conf-L1

Where in your configuration are you pushing data into redis?