logstash 收集日志传到另外一台机器的logstash并落地到磁盘

收集端：
input {
  file {
    path => "/home/admin/local/uae_agent/apps/InProcessorOL/log/a4_comment_processor/comment_processor.log*"
    start_position => "beginning"
    type => "processor_10.40.87.141"
    codec => multiline {
      pattern => "^\[%{TIMESTAMP_ISO8601} "
      negate => true
      what => "previous"
    }
  }
  file {
    path => "/home/admin/local/uae_agent/apps/InProcessorOL/log/a4_comment_processor/rank.log*"
    start_position => "beginning"
    type => "rank_10.40.87.141"
    codec => multiline {
      pattern => "^\[%{TIMESTAMP_ISO8601} "
      negate => true
      what => "previous"
    }
  }
  file {
    path => "/home/admin/local/uae_agent/apps/InDispatherOL/log/a4_comment_processor/access.log*"
    start_position => "beginning"
    type => "disaptcher_10.40.87.141"
    codec => multiline {
      pattern => "^\[%{TIMESTAMP_ISO8601} "
      negate => true
      what => "previous"
    }
  }
}

filter {
  #grok {
  #  remove_field => [ "@timestamp", "@version","path" ]
  #}
  mutate {
    replace => { "message" => "%{type}###%{message}" }
  }
}

output {
  tcp {
    host => "10.40.87.147"
    port => 12345
    #codec => line { format => "%{message}-----%{%fn}"}
    codec => json_lines
  }
}

汇总落磁盘端：
input {
  tcp {
    port => 12345
    codec => json_lines 
  }
}

filter {
  grok {
     patterns_dir => ["./patterns"]
     match => { "message" => [ "^%{USERNAME:fn}###\[%{MYDATE:today}[T ]%{HOUR:hours}:%{GREEDYDATA:msg}"] } 
  }
  mutate {
    replace => { "message" => "[%{today} %{hours}:%{msg}" }
  } 
}

output {
  file {
    path => "/home1/tanzhenghai/logs/%{today}/%{fn}_%{hours}"
    flush_interval => 5
    codec => line { format => "%{message}"}
  }
}