gsub config 2

gistfile1.rb

input {
  file {
    path => ["/var/log/network.log"]
    sincedb_path => "/var/log/logstash"
    start_position => "beginning"
    type => "syslog"
    tags => [ "netsyslog" ]
  }

  udp {
    type => "netflow"
    tags => [ "nProbe" ]
    port => 2055
    codec => netflow {
             }
  }

} #end input block

filter {

  if [type] == "syslog" {
    grok {
      match => [ "message", "%{TIMESTAMP_ISO8601:syslog_datetime} %{IPORHOST:syslog_host} %{POSINT}?: %{WORD} %{POSINT} %{TIME}: %%{WORD:facility}-%{POSINT:severity}-%{NOTSPACE:mneumonic}: %{GREEDYDATA:raw_message}",
                 "message", "%{TIMESTAMP_ISO8601:syslog_datetime} %{IPORHOST:syslog_host} : %{YEAR} %{SYSLOGTIMESTAMP} %{TZ}: %%{WORD:facility}-%{POSINT:severity}-%{NOTSPACE:mneumonic}: %{GREEDYDATA:raw_message}",
                 "message", "%{TIMESTAMP_ISO8601:syslog_datetime} %{IPORHOST:syslog_host} : %%{WORD:facility}-%{POSINT:severity}-%{NOTSPACE:mneumonic}: %{GREEDYDATA:raw_message}" ]
      add_tag => [ "grok_ran" ]
    }

    date {
      match => [ "syslog_datetime", "ISO8601" ]
    }
  }

  if "grok_ran" in [tags] and "_grokparsefailure" not in [tags] {
      mutate {
        remove_field => [ "message", "syslog_datetime" ]
      }
    }

  if [type] == "netflow" {
    date {
      match => [ "netflow.first_switched", "ISO8601" ]
    }

    mutate {
      gsub => [ "[netflow][protocol]", "6", "TCP" ]
      gsub => [ "[netflow][protocol]", "17", "UDP" ]
      gsub => [ "[netflow][protocol]", "1", "ICMP" ]
      convert => [ "[netflow][in_bytes]", "integer"]
      convert => [ "[netflow][in_pkts]", "integer"]

    }

    noop {
      add_tag => [ "netflow_if_statement_hit" ]
    }
  }

} #end filter block

output {
  elasticsearch {
    protocol => "node"
    host => "127.0.0.1"
    cluster => "elasticsearch"
  }
} #end output block