logstash logs:
root@ip-172-31-0-150:/usr/share# tail /var/log/logstash/logstash.log
{:timestamp=>"2018-01-16T11:54:58.065000+0000", :message=>#<LogStash::PipelineReporter::Snapshot:0x5276e99 @data={:events_filtered=>1914, :events_consumed=>1914, :worker_count=>2, :inflight_count=>21, :worker_states=>[{:status=>"sleep", :alive=>true, :index=>0, :inflight_count=>12}, {:status=>"sleep", :alive=>true, :index=>1, :inflight_count=>9}], :output_info=>[{:type=>"elasticsearch", :config=>{"hosts"=>"localhost", "index"=>"influxCSVData", "document_type"=>"influxCSV_data_document_type"}, :is_multi_worker=>true, :events_received=>1914, :workers=><Java::JavaUtilConcurrent::CopyOnWriteArrayList:419037665 [<LogStash::Outputs::ElasticSearch hosts=>["localhost"], index=>"influxCSVData", document_type=>"influxCSV_data_document_type", codec=><LogStash::Codecs::Plain charset=>"UTF-8">, workers=>1, manage_template=>true, template_name=>"logstash", template_overwrite=>false, flush_size=>500, idle_flush_time=>1, doc_as_upsert=>false, max_retries=>3, script_type=>"inline", script_var_name=>"event", scripted_upsert=>false, retry_max_interval=>2, retry_max_items=>500, action=>"index", path=>"/", ssl_certificate_verification=>true, sniffing=>false, sniffing_delay=>5>, <LogStash::Outputs::ElasticSearch hosts=>["localhost"], index=>"influxCSVData", document_type=>"influxCSV_data_document_type", codec=><LogStash::Codecs::Plain charset=>"UTF-8">, workers=>1, manage_template=>true, template_name=>"logstash", template_overwrite=>false, flush_size=>500, idle_flush_time=>1, doc_as_upsert=>false, max_retries=>3, script_type=>"inline", script_var_name=>"event", scripted_upsert=>false, retry_max_interval=>2, retry_max_items=>500, action=>"index", path=>"/", ssl_certificate_verification=>true, sniffing=>false, sniffing_delay=>5>]>, :busy_workers=>2}, {:type=>"stdout", :config=>{}, :is_multi_worker=>false, :events_received=>1893, :workers=><Java::JavaUtilConcurrent::CopyOnWriteArrayList:-535213015 [<LogStash::Outputs::Stdout codec=><LogStash::Codecs::Line charset=>"UTF-8", delimiter=>"\n">, workers=>1>]>, :busy_workers=>0}, {:type=>"elasticsearch", :config=>{"hosts"=>["13.59.198.44:9200"], "index"=>"%{[@metadeta][beat]}-%{+YYYY.MM.dd}", "document_type"=>"%{[@metadata][type]}"}, :is_multi_worker=>true, :events_received=>1893, :workers=><Java::JavaUtilConcurrent::CopyOnWriteArrayList:1703971105 [<LogStash::Outputs::ElasticSearch hosts=>["13.59.198.44:9200"], index=>"%{[@metadeta][beat]}-%{+YYYY.MM.dd}", document_type=>"%{[@metadata][type]}", codec=><LogStash::Codecs::Plain charset=>"UTF-8">, workers=>1, manage_template=>true, template_name=>"logstash", template_overwrite=>false, flush_size=>500, idle_flush_time=>1, doc_as_upsert=>false, max_retries=>3, script_type=>"inline", script_var_name=>"event", scripted_upsert=>false, retry_max_interval=>2, retry_max_items=>500, action=>"index", path=>"/", ssl_certificate_verification=>true, sniffing=>false, sniffing_delay=>5>, <LogStash::Outputs::ElasticSearch hosts=>["13.59.198.44:9200"], index=>"%{[@metadeta][beat]}-%{+YYYY.MM.dd}", document_type=>"%{[@metadata][type]}", codec=><LogStash::Codecs::Plain charset=>"UTF-8">, workers=>1, manage_template=>true, template_name=>"logstash", template_overwrite=>false, flush_size=>500, idle_flush_time=>1, doc_as_upsert=>false, max_retries=>3, script_type=>"inline", script_var_name=>"event", scripted_upsert=>false, retry_max_interval=>2, retry_max_items=>500, action=>"index", path=>"/", ssl_certificate_verification=>true, sniffing=>false, sniffing_delay=>5>]>, :busy_workers