Multiple Logstash Pipelines outputting into same index

Question

I have two Filebeat pipes inputting into Logstash. One parses out log errors that I actually care about from one service while the other takes each line in order to keep track of the health of another service that often crashes by just hanging indefinitely. Originally, I was only running the exceptions pipe which was working and filtering correctly.

Yesterday, I added the second pipe. I used the pipelines.yml file. The Logstash log shows that both pipelines are initialized correctly at startup, shows that there are two pipelines running. They are running the inputs on separate ports as required. Filebeat side is also configured to run on the correct ports.

The problem is that they are outputting to the same index and now the filtering for the exception logs is being ignored. It's like the exception log config file is being completely disregarded except for the part about which port to receive from Filebeat on.

Has anyone had this issue, or should this be a bug report to the ES team? Below are my config file input/outputs and pipelines.yml setup, just in case anyone can glance and see that I'm missing something. It's completely perplexing me because it's an extremely simple setup.

Config1:

input {
    beats {
        port => "5044"

    }
}
...
output {
    elasticsearch {
        hosts => ["http://****:9200"]
        index => "logtailing-%{[@metadata][version]}-%{+YYYY.MM.dd}" 
    }
}

Config2:

input {
    beats {
        port => "5043"

    }
}
...
output {
    elasticsearch {
        hosts => ["http://****:9200"]
        index => "releaserexceptions-%{[@metadata][version]}-%{+YYYY.MM.dd}" 
    }
}

pipelines.yml:

 - pipeline.id: logtailing
   path.config: "../config/logtailing.conf"
 - pipeline.id: releaserexceptions
   path.config: "../config/releaserexceptions.conf"

Logstash logs look like this so it seems to me it should be working with no issues:

[2019-01-04T12:46:26,256][INFO ][logstash.runner          ] Starting Logstash {"logstash.version"=>"6.4.0"}
[2019-01-04T12:46:31,249][INFO ][logstash.pipeline        ] Starting pipeline {:pipeline_id=>"logtailing", "pipeline.workers"=>1, "pipeline.batch.size"=>125, "pipeline.batch.delay"=>50}
[2019-01-04T12:46:32,138][INFO ][logstash.outputs.elasticsearch] Elasticsearch pool URLs updated {:changes=>{:removed=>[], :added=>[http://****:9200/]}}
[2019-01-04T12:46:32,151][INFO ][logstash.outputs.elasticsearch] Running health check to see if an Elasticsearch connection is working {:healthcheck_url=>http://****:9200/, :path=>"/"}
[2019-01-04T12:46:32,644][INFO ][logstash.pipeline        ] Starting pipeline {:pipeline_id=>"releaserexceptions", "pipeline.workers"=>1, "pipeline.batch.size"=>125, "pipeline.batch.delay"=>50}
[2019-01-04T12:46:32,670][INFO ][logstash.outputs.elasticsearch] Elasticsearch pool URLs updated {:changes=>{:removed=>[], :added=>[http://****:9200/]}}
[2019-01-04T12:46:32,672][INFO ][logstash.outputs.elasticsearch] Running health check to see if an Elasticsearch connection is working {:healthcheck_url=>http://****:9200/, :path=>"/"}
[2019-01-04T12:46:32,741][WARN ][logstash.outputs.elasticsearch] Restored connection to ES instance {:url=>"http://****:9200/"}
[2019-01-04T12:46:32,806][INFO ][logstash.outputs.elasticsearch] ES Output version determined {:es_version=>6}
[2019-01-04T12:46:32,811][WARN ][logstash.outputs.elasticsearch] Detected a 6.x and above cluster: the `type` event field won't be used to determine the document _type {:es_version=>6}
[2019-01-04T12:46:32,842][INFO ][logstash.outputs.elasticsearch] New Elasticsearch output {:class=>"LogStash::Outputs::ElasticSearch", :hosts=>["http://****:9200"]}
[2019-01-04T12:46:32,869][INFO ][logstash.outputs.elasticsearch] Using mapping template from {:path=>nil}
[2019-01-04T12:46:32,900][INFO ][logstash.outputs.elasticsearch] Attempting to install template {:manage_template=>{"template"=>"logstash-*", "version"=>60001, "settings"=>{"index.refresh_interval"=>"5s"}, "mappings"=>{"_default_"=>{"dynamic_templates"=>[{"message_field"=>{"path_match"=>"message", "match_mapping_type"=>"string", "mapping"=>{"type"=>"text", "norms"=>false}}}, {"string_fields"=>{"match"=>"*", "match_mapping_type"=>"string", "mapping"=>{"type"=>"text", "norms"=>false, "fields"=>{"keyword"=>{"type"=>"keyword", "ignore_above"=>256}}}}}], "properties"=>{"@timestamp"=>{"type"=>"date"}, "@version"=>{"type"=>"keyword"}, "geoip"=>{"dynamic"=>true, "properties"=>{"ip"=>{"type"=>"ip"}, "location"=>{"type"=>"geo_point"}, "latitude"=>{"type"=>"half_float"}, "longitude"=>{"type"=>"half_float"}}}}}}}}
[2019-01-04T12:46:33,461][WARN ][logstash.outputs.elasticsearch] Restored connection to ES instance {:url=>"http://****:9200/"}
[2019-01-04T12:46:33,466][INFO ][logstash.outputs.elasticsearch] ES Output version determined {:es_version=>6}
[2019-01-04T12:46:33,466][WARN ][logstash.outputs.elasticsearch] Detected a 6.x and above cluster: the `type` event field won't be used to determine the document _type {:es_version=>6}
[2019-01-04T12:46:33,468][INFO ][logstash.outputs.elasticsearch] New Elasticsearch output {:class=>"LogStash::Outputs::ElasticSearch", :hosts=>["http://****:9200"]}
[2019-01-04T12:46:33,474][INFO ][logstash.outputs.elasticsearch] Using mapping template from {:path=>nil}
[2019-01-04T12:46:33,499][INFO ][logstash.outputs.elasticsearch] Attempting to install template {:manage_template=>{"template"=>"logstash-*", "version"=>60001, "settings"=>{"index.refresh_interval"=>"5s"}, "mappings"=>{"_default_"=>{"dynamic_templates"=>[{"message_field"=>{"path_match"=>"message", "match_mapping_type"=>"string", "mapping"=>{"type"=>"text", "norms"=>false}}}, {"string_fields"=>{"match"=>"*", "match_mapping_type"=>"string", "mapping"=>{"type"=>"text", "norms"=>false, "fields"=>{"keyword"=>{"type"=>"keyword", "ignore_above"=>256}}}}}], "properties"=>{"@timestamp"=>{"type"=>"date"}, "@version"=>{"type"=>"keyword"}, "geoip"=>{"dynamic"=>true, "properties"=>{"ip"=>{"type"=>"ip"}, "location"=>{"type"=>"geo_point"}, "latitude"=>{"type"=>"half_float"}, "longitude"=>{"type"=>"half_float"}}}}}}}}
[2019-01-04T12:46:33,813][INFO ][logstash.inputs.beats    ] Beats inputs: Starting input listener {:address=>"0.0.0.0:5043"}
[2019-01-04T12:46:33,831][INFO ][logstash.inputs.beats    ] Beats inputs: Starting input listener {:address=>"0.0.0.0:5044"}
[2019-01-04T12:46:33,903][INFO ][logstash.pipeline        ] Pipeline started successfully {:pipeline_id=>"logtailing", :thread=>"#<Thread:0x132b2cfb run>"}
[2019-01-04T12:46:33,903][INFO ][logstash.pipeline        ] Pipeline started successfully {:pipeline_id=>"releaserexceptions", :thread=>"#<Thread:0xaf8b522 run>"}
[2019-01-04T12:46:33,998][INFO ][logstash.agent           ] Pipelines running {:count=>2, :running_pipelines=>[:logtailing, :releaserexceptions], :non_running_pipelines=>[]}
[2019-01-04T12:46:34,013][INFO ][org.logstash.beats.Server] Starting server on port: 5043
[2019-01-04T12:46:34,013][INFO ][org.logstash.beats.Server] Starting server on port: 5044
[2019-01-04T12:46:34,453][INFO ][logstash.agent           ] Successfully started Logstash API endpoint {:port=>9600}

Answer 1

"I have two Filebeat pipes inputting into Logstash." ---I assume this means that you have conditional logic and/or prospectors in your filebeat config to ship to the multiple logstash (5044 and 5043)ports? I'd be interested to see that config.

Here's another approach that might acheive your goal: Use the source value in a conditional output. An example of logstash conditional outputs can be found here: https://discuss.elastic.co/t/conditional-statement-in-output/74672

If, for some reason filebeat's 'source' field will not suffice: you might inject your own fields, per prospector, as described by andrewkroh midway through the thread here: https://discuss.elastic.co/t/how-to-tag-log-files-in-filebeat-for-logstash-ingestion/44713/6. Then you could do conditional output on your custom fields. The example Andrew provided for adding custom fields by prospector is as follows:

```filebeat:
```  prospectors:
```    - paths:
```        - /path/to/logs/access.log
```      fields:  {log_type: access}
```    - paths:
```        - /path/to/other/logs/errors.log
```      fields: {log_type: errors}