logstash.conf

filter {
  mutate {
    copy => { '[srcip]' => '[source][address]' }
    copy => { '[srcip]' => '[source][ip]' }
    copy => { '[new_event][srcip]' => '[source][ip]' }
    rename => { '[srcport]' => '[source][port]' }
    convert => { '[source][port]' => 'integer' }
    copy => { '[destip]' => '[destination][address]' }
    copy => { '[destport]' => '[destination][port]' }
    convert => { '[destination][port]' => 'integer' }
    copy => { '[ts]' => '[timestamp]' }
    rename => { '[action]' => '[event][action]' }
    lowercase => [ '[event][action]' ]
    rename => { '[duration]' => '[event][duration]' }
    convert => { '[event][duration]' => 'float' }
    rename => { '[user_agent]' => '[user_agent][original]' }
    rename => { '[log_level]' => '[log][level]' }
    uppercase => [ '[log][level]' ]
    rename => { '[eventid]' => '[event][id]' }
    convert => { '[event][id]' => 'string' }
    rename => { '[hostip]' => '[host][ip]' }
  }

  date { 
    match => ["[some_timestamp_field]", "UNIX_MS"]
    target => "[@timestamp]" 
  }

    
  date { 
    match => ["[some_other_timestamp]", "UNIX_MS"]
    target => "[@timestamp]" 
  }

    
  date { 
    match => ["[some_new_timestamp]", "UNIX"]
    target => "[destination_timestamp]" 
  }

  if [host][ip] {
    ruby {
      code => "event.set('[host][ip]', Array(event.get('[host][ip]')) )"
    }
  }
}