# Multiple filebeats with different log type specification on logstash **URL:** \ **Category:** Logstash\ **Created:** [March 26, 2019, 7:01am UTC](https://discuss.elastic.co/t/multiple-filebeats-with-different-log-type-specification-on-logstash/173867 "2019-03-26T07:01:09Z")\ **Posts on this page:** 4\ **Page:** 1
**Author:** ![cenginz](https://avatars.discourse-cdn.com/v4/letter/c/e495f1/32.png) [@cenginz](https://discuss.elastic.co/u/cenginz)\ **Post date:** [March 26, 2019, 7:01am UTC](https://discuss.elastic.co/t/multiple-filebeats-with-different-log-type-specification-on-logstash/173867/1 "2019-03-26T07:01:09Z")
Hi all, I've two filebeats located on different servers with different purposes. One of them I've been using with the logstash configuration I give below to ship docker logs. And I've a new filebeat for shipping hazelcast logs with the given configuration. My ELK stack version is 5.5.1 So my question is can i configure logstash to listen on the same port for multiple beats? And what filter would be good to use in order to ingest the log below? I couldn't quite wrap my head around that. My example hazelcast log is below: 2019-03-26 09:43:37.617 INFO 31828 --- [d.HealthMonitor] c.h.internal.diagnostics.HealthMonitor : [ipv4]:5701 [servr-prod] [3.8.3] processors=8, physical.memory.total=31.2G, physical.memory.free=1001.0M, swap.space.total=16.0G, swap.space.free=14.9G, heap.memory.used=6.2G, heap.memory.free=5.2G, heap.memory.total=11.4G, heap.memory.max=11.4G, heap.memory.used/total=54.22%, heap.memory.used/max=54.22%, minor.gc.count=329, minor.gc.time=65348ms, major.gc.count=2, major.gc.time=109ms, load.process=0.00%, load.system=0.50%, load.systemAverage=54.00%, thread.count=74, thread.peakCount=79, cluster.timeDiff=-16228, event.q.size=0, executor.q.async.size=0, executor.q.client.size=0, executor.q.query.size=0, executor.q.scheduled.size=0, executor.q.io.size=0, executor.q.system.size=0, executor.q.operations.size=1, executor.q.priorityOperation.size=0, operations.completed.count=390597068, executor.q.mapLoad.size=0, executor.q.mapLoadAllKeys.size=0, executor.q.cluster.size=0, executor.q.response.size=0, operations.running.count=0, operations.pending.invocations.percentage=0.00%, operations.pending.invocations.count=1, proxy.count=0, clientEndpoint.count=19, connection.active.count=21, client.connection.count=19, connection.count=2 The one to deliver hazelcast logs' configuration is below: filebeat.prospectors: - input\_type: log paths: - /products/logs/hzcluster.log multiline.pattern: '^[' multiline.negate: true multiline.match: after output.logstash: hosts: ["ip:5044"] logging.level: debug The one responsible for delivering docker logs is below: filebeat: prospectors: - type: log paths: - /var/lib/docker/containers/_/_.log json: message\_key: log keys\_under\_root: true multiline: pattern: '^APP||^APM||^20\d\d-' negate: true match: after processors: - add\_docker\_metadata: ~ output: logstash: hosts: ["ip:5044"] compression\_level: 0 logging: metrics: enabled: false My logtash config is below: input { beats { port =\> 5044 type =\> "docker\_logs" } } filter { mutate { copy =\> { "log" =\> "[@metadata][log]" }} if [log] =~ /^APM/ { mutate { add\_field =\> { "[@metadata][logType]" =\> "APM" } } grok { match =\> { "log" =\> "APM|(?[^|]_)|(?._)" } } mutate { # The new line here is the only why to put a new line ... gsub =\> ["apmJson", "", ""] } json { source =\> "apmJson" } } else if [log] =~ /^APP/ { mutate { add\_field =\> { "[@metadata][logType]" =\> "APP" } } grok { match =\> { "log" =\> "APP|(?[^|]_)|(?[^|]_)|(?[^|]_)|(?[^|]_)|(?[^|]_)|(?[^|]_)|(?[^|]_)|(?[.a-zA-Z0-9]_) --- (?.\*)" } } } date { match =\> ["logTime" , "yyyy-MM-dd HH:mm:ss.SSS"] timezone =\> "UTC" } mutate { remove\_field =\> ["apmJson","beat","stream","docker","log","source","type","prospector","host","logTime","logLevel","logger"] } } output { if [@metadata][logType] == "APM" { elasticsearch { hosts =\> ["ip:9200"] index =\> "monitoring" document\_type =\> "metric" } } else { if [@metadata][logType] == "APP" { elasticsearch { hosts =\> ["ip:9200"] index =\> "logstash-%{+YYYY.MM.dd}" document\_type =\> "app-log" } } file { path =\> "/products/logstash/logs/vf-docker-%{+YYYY.MM.dd}.log" codec =\> line { format =\> "%{[@metadata][log]}" } } } } Any help is highly appreciated. Thanks ---
**Author:** ![Badger](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/badger/32/25190_2.png) [@Badger](https://discuss.elastic.co/u/Badger)\ **Post date:** [March 26, 2019, 1:36pm UTC](https://discuss.elastic.co/t/multiple-filebeats-with-different-log-type-specification-on-logstash/173867/2 "2019-03-26T13:36:56Z")
> [@cenginz](#): > > So my question is can i configure logstash to listen on the same port for multiple beats? Yes. You may want to add a field in filebeat to identify the type of log if you do that. For that hazelcast log I would use ``` mutate { add_field => { "[@metadata][copy]" => "%{[message]}" } } mutate { gsub => ["[@metadata][copy]", ".*\]", "" ] } kv { source => "[@metadata][copy]" field_split => ", " } ``` ---
**Author:** ![cenginz](https://avatars.discourse-cdn.com/v4/letter/c/e495f1/32.png) [@cenginz](https://discuss.elastic.co/u/cenginz)\ **Post date:** [March 26, 2019, 3:43pm UTC](https://discuss.elastic.co/t/multiple-filebeats-with-different-log-type-specification-on-logstash/173867/3 "2019-03-26T15:43:47Z")
Hey Badger, Thanks for the reply. I'll give it a go. ---
**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\ **Post date:** [April 23, 2019, 3:43pm UTC](https://discuss.elastic.co/t/multiple-filebeats-with-different-log-type-specification-on-logstash/173867/4 "2019-04-23T15:43:52Z")
This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.