Metricbeat not working through Logstash

Hello Everyone, i was wondering if anyone could provide some assistance in the following:

We have a setup in which we output from Multiple beats into logstash which outputs into an Elasticsearch Cluster, and Graylog Cluster simultaneously as an additional layer of redundancy. the setup is Beats > Logstash > Graylog / Elasticsearch. now all the beats besides Metricbeat can successfully send data and i can see the messages in the inputs coming in with no issues. i know that the beat can connect and send data, as when i point it directly to graylog and set up a Beats input and not a GELF UDP input it works fine. This is the configuration:

Metricbeat:

metricbeat.config.modules:
path: ${path.config}/modules.d/*.yml
reload.enabled: false

setup.template.settings:
index.number_of_shards: 1
index.codec: best_compression

setup.kibana:
host: “10.226.100.30:5601”

output.logstash:
hosts: [“logstash01:6969”]

Logstash

input {

beats {
port => 6969
}

}

output {

elasticsearch {

hosts => ["elasticnode01", "elasticnode02"]
index => "%{[@metadata][beat]}-%{[@metadata][version]}"
ilm_enabled => true

}

gelf {

host => "graylognode01"
port => 12201
short_message => "Output to Node 1"

}
gelf {

host => "graylognode02"
port => 12201
short_message => "Output to Node 2"

}

}

And for Gelf input i set up a Global GELF UDP input bound to 0.0.0.0:12201 and have turned off Firewalld and SElinux to see if the problem is there by any chance. All the machines are running on Centos7 if it makes any difference. Any information would be more than Welcome as i cant seem to figure out why this is happening with only one of the beats

did you checked the graylog server.log or the elasticsearch log (from the ES cluster that is used by Graylog) for error messages?

I checked the Graylog log file and the end of it only shows the following

2019-09-04T09:48:22.941Z WARN [UdpTransport] receiveBufferSize (SO_RCVBUF) for input GELFUDPInput{title=GELF UDP Input, type=org.graylog2.inputs.gelf.udp.GELFUDPInput, nodeId=null} (channel
[id: 0xcd7b5640, L:/0:0:0:0:0:0:0:0%0:12201]) should be 262144 but is 425984.
2019-09-04T09:48:22.944Z WARN [UdpTransport] receiveBufferSize (SO_RCVBUF) for input GELFUDPInput{title=GELF UDP Input, type=org.graylog2.inputs.gelf.udp.GELFUDPInput, nodeId=null} (channel
[id: 0xe908892b, L:/0:0:0:0:0:0:0:0%0:12201]) should be 262144 but is 425984.
2019-09-04T09:48:22.948Z INFO [InputStateListener] Input [GELF UDP/5d6aba27cb5ebe07ad390283] is now RUNNING
2019-09-09T09:37:02.863Z INFO [connection] Opened connection [connectionId{localValue:19, serverValue:65}] to graylog01.log.lab.aginion.net:27017

And Elasticsearch comes up with

Caused by: java.lang.IllegalArgumentException: Expected numeric type on field [beats_type], but got [keyword]
at org.elasticsearch.search.aggregations.support.ValuesSourceConfig.numericField(ValuesSourceConfig.java:309) ~[elasticsearch-6.8.2.jar:6.8.2]
at org.elasticsearch.search.aggregations.support.ValuesSourceConfig.originalValuesSource(ValuesSourceConfig.java:292) ~[elasticsearch-6.8.2.jar:6.8.2]
at org.elasticsearch.search.aggregations.support.ValuesSourceConfig.toValuesSource(ValuesSourceConfig.java:249) ~[elasticsearch-6.8.2.jar:6.8.2]
at org.elasticsearch.search.aggregations.support.ValuesSourceAggregatorFactory.createInternal(ValuesSourceAggregatorFactory.java:55) ~[elasticsearch-6.8.2.jar:6.8.2]
at org.elasticsearch.search.aggregations.AggregatorFactory.create(AggregatorFactory.java:216) ~[elasticsearch-6.8.2.jar:6.8.2]
at org.elasticsearch.search.aggregations.AggregatorFactories.createSubAggregators(AggregatorFactories.java:201) ~[elasticsearch-6.8.2.jar:6.8.2]
at org.elasticsearch.search.aggregations.AggregatorBase.(AggregatorBase.java:76) ~[elasticsearch-6.8.2.jar:6.8.2]
at org.elasticsearch.search.aggregations.bucket.BucketsAggregator.(BucketsAggregator.java:47) ~[elasticsearch-6.8.2.jar:6.8.2]
at org.elasticsearch.search.aggregations.bucket.histogram.DateHistogramAggregator.(DateHistogramAggregator.java:74) ~[elasticsearch-6.8.2.jar:6.8.2]
at org.elasticsearch.search.aggregations.bucket.histogram.DateHistogramAggregatorFactory.createAggregator(DateHistogramAggregatorFactory.java:80) ~[elasticsearch-6.8.2.jar:6.8.2]
at org.elasticsearch.search.aggregations.bucket.histogram.DateHistogramAggregatorFactory.doCreateInternal(DateHistogramAggregatorFactory.java:74) ~[elasticsearch-6.8.2.jar:6.8.2]
at org.elasticsearch.search.aggregations.bucket.histogram.DateHistogramAggregatorFactory.doCreateInternal(DateHistogramAggregatorFactory.java:38) ~[elasticsearch-6.8.2.jar:6.8.2]
at org.elasticsearch.search.aggregations.support.ValuesSourceAggregatorFactory.createInternal(ValuesSourceAggregatorFactory.java:59) ~[elasticsearch-6.8.2.jar:6.8.2]
at org.elasticsearch.search.aggregations.AggregatorFactory.create(AggregatorFactory.java:216) ~[elasticsearch-6.8.2.jar:6.8.2]
at org.elasticsearch.search.aggregations.AggregatorFactories.createTopLevelAggregators(AggregatorFactories.java:217) ~[elasticsearch-6.8.2.jar:6.8.2]
at org.elasticsearch.search.aggregations.AggregationPhase.preProcess(AggregationPhase.java:55) ~[elasticsearch-6.8.2.jar:6.8.2]
at org.elasticsearch.search.query.QueryPhase.execute(QueryPhase.java:112) ~[elasticsearch-6.8.2.jar:6.8.2]
at org.elasticsearch.search.SearchService.loadOrExecuteQueryPhase(SearchService.java:350) ~[elasticsearch-6.8.2.jar:6.8.2]
at org.elasticsearch.search.SearchService.executeQueryPhase(SearchService.java:394) ~[elasticsearch-6.8.2.jar:6.8.2]
at org.elasticsearch.search.SearchService.access$100(SearchService.java:126) ~[elasticsearch-6.8.2.jar:6.8.2]
at org.elasticsearch.search.SearchService$2.onResponse(SearchService.java:359) ~[elasticsearch-6.8.2.jar:6.8.2]

so what is in your index at the field [beats_type]?

You might want to split indices by ingest type or force with a custom template a specific store.

Sorry i didnt mention that i am really new with these systems so i dont know how to check the field in the index

Ok after some digning around, by pointing the beats to graylog directly i managed to find that [beats_type] filed holds a string metricbeat