Indexing question related to pipeline outputs

[_finack]

Issue: I have a Logstash pipeline output testing if [agent][type] == "metricbeat". I am expecting such documents to end up in indices matching the pattern metricbeat-*; however, some are ending up in indices matching the pattern indexme-*, despite the documents in indexme-* having an agent.type field set to "metricbeat". I don't understand why this is occurring.

The Scenario: I am testing Elastic Stack and right now I just have Metricbeat feeding data from the ES server and Winlogbeat feeding data from a Windows Event Collector. Both are going into Logstash.

I was messing around with the pipeline filters to create new document fields for Windows events and discovered the pipeline configurations from the HELK project. I loaded all of these plugins (except 0002-kafka-input.conf), fixed the output plugins to not use authentication and go to the correct ES server (localhost).

This resulted in all of the Metricbeat documents going to indexme-* indices (due to 9998-catch_all-output.conf). I created a new pipeline output titled 9930-metricbeat-output.conf that uses the basic output configuration from the ES getting started guide (see below) except with the addition of checking if [agent][type] == "metricbeat". Now I am getting some metricbeat-related documents in indexme-* and others in metricbeat-*. Documents in both indices contain "metricbeat" in a field named agent.type.

Please help me understand why this is happening.

I'm new to Elastic Search.

Supporting Information:

9930-metricbeat-output.conf

output {
  if [agent][type] == "metricbeat" {
    elasticsearch {
      hosts => "localhost:9200"
      manage_template => false
      index => "%{[@metadata][beat]}-%{[@metadata][version]}-%{+YYYY.MM.dd}"
      #user => "username"
      #password => "password"
    }
  }
}

9998-catch_all-output.conf

output {
  if [@metadata][helk_parsed] != "yes" and [source] != "/var/log/osquery/osqueryd.results.log" and [@metadata][kafka][topic] != "SYSMON_JOIN" and [@metadata][helk_input_source] != "mitre_attack"{
    elasticsearch {
      hosts => ["localhost:9200"]
      index => "indexme-%{+YYYY.MM.dd}"
      # document_id => "%{[@metadata][log_hash]}"
      #user => 'elastic'
      #password => 'elasticpassword'
    }
  }
}