# Parsing data from multiple application servers through logstash

**URL:** <https://discuss.elastic.co/t/parsing-data-from-multiple-application-servers-through-logstash/285069>\
**Category:** Logstash\
**Created:** [September 24, 2021, 10:48am UTC](https://discuss.elastic.co/t/parsing-data-from-multiple-application-servers-through-logstash/285069 "2021-09-24T10:48:34Z")\
**Posts on this page:** 12\
**Page:** 1

<div class="post-metadata">

**Author:** ![prat](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/prat/32/79978_2.png) [@prat](https://discuss.elastic.co/u/prat)\
**Post date:** [September 24, 2021, 10:48am UTC](https://discuss.elastic.co/t/parsing-data-from-multiple-application-servers-through-logstash/285069/1 "2021-09-24T10:48:34Z")

</div>

Hi Team,

I have application running on 2 servers and application logs are getting logged on both the servers, so i want to parse logs from both servers.

`filebeat` is installed on two application servers,  
`logstash` is installed on separate two servers and  
`elasticsearch` is installed on three servers (2 of which are `logstash` servers also)

`filebeat.yml` is like below,

`Application Server1` -

```auto
filebeat.inputs:
      - type: log
         fields_under_root: true
         fields:
           log_type: federate_server1
           app_id: pf
         multiline.pattern: ^[[:space:]]+(at|\.{3})\b|^Caused by:|^java|^...|^-
         multiline.negate: true
         multiline.match: after
         paths:
           - /opt/federate-0.2.0/federate/log/*

output.logstash:
   hosts: ['logstash1:5044'], ['logstash2:5044']
   loadbalance: true

```

`Application Server 2` -

```auto
filebeat.inputs:
      - type: log
         fields_under_root: true
         fields:
           log_type: federate_server2
           app_id: pf
         multiline.pattern: ^[[:space:]]+(at|\.{3})\b|^Caused by:|^java|^...|^-
         multiline.negate: true
         multiline.match: after
         paths:
           - /opt/federate-0.2.0/federate/log/*

output.logstash:
   hosts: ['logstash1:5044'], ['logstash2:5044']
   loadbalance: true

```

`logstash.yml` -

`logstash server1`

```auto
input {
  beats {
    port => 5044
  }
}

filter {
if [log_type] == "federate_server" and [app_id] == "pf"
  {
    mutate { gsub => ["message","\|"," "] } grok { patterns_dir => ["/etc/logstash/patterns"] match => { "message" => "%{MY_DATE_PATTERN:timestamp}%{SPACE}%{LOGLEVEL:level}%{SPACE}%{UUID:ConsentID}%{SPACE}%{WORD:TransactionID}%{SPACE}%{WORD:TraceID}%{SPACE}%{GREEDYDATA:messagetext}" } }
    mutate {
             replace => {
               "[type]" => "federate_server"
             }
           }
  }

output {
  if [log_type] == "federate_server" {
  elasticsearch {
    hosts => ['http://es1:9200', 'http://es2:9200', 'http://es3:9200']
        user => elastic
    password => "${es_pwd}"
     index => "federate"
     template_name => "federate"
     template_overwrite => "false"
      }
 }
  elasticsearch {
    hosts => ['http://es1:9200', 'http://es2:9200', 'http://es3:9200']
    index => "%{[@metadata][beat]}-%{[@metadata][version]}-%{+YYYY.MM}"
    user => elastic
    password => "${es_pwd}"
  }
}

```

`logstash server 2`

```auto
input {
  beats {
    port => 5044
  }
}

filter {
if [log_type] == "federate_server" and [app_id] == "pf"
  {
    mutate { gsub => ["message","\|"," "] } grok { patterns_dir => ["/etc/logstash/patterns"] match => { "message" => "%{MY_DATE_PATTERN:timestamp}%{SPACE}%{LOGLEVEL:level}%{SPACE}%{UUID:ConsentID}%{SPACE}%{WORD:TransactionID}%{SPACE}%{WORD:TraceID}%{SPACE}%{GREEDYDATA:messagetext}" } }
    mutate {
             replace => {
               "[type]" => "federate_server"
             }
           }
  }
output {
  if [log_type] == "federate_server" {
  elasticsearch {
    hosts => ['http://es1:9200', 'http://es2:9200', 'http://es3:9200']
        user => elastic
    password => "${es_pwd}"
     index => "federate"
     template_name => "federate"
     template_overwrite => "false"
      }
 }
elasticsearch {
    hosts => ['http://es1:9200', 'http://es2:9200', 'http://es3:9200']
    index => "%{[@metadata][beat]}-%{[@metadata][version]}-%{+YYYY.MM}"
    user => elastic
    password => "${es_pwd}"
  }
}

```

I know currently `log_type` in both `filebeat.yml` is not matching with `log_type` in both `logstash.yml` file.

Since both the `logstash` are mentioned in `filebeat.yml` and `loadbalance` is true so `filebeat` will send events either of the `logstash` servers at a time but two receive events on `logstash` end how can i add the other `log_type` in `logstash.yml`?. currently only one is specified.

1. i.e Can I change `log_type` as below on both server's `logstash.yml` to receive events from both the application server's `filebeat`?

filter {  
if [log\_type] == "federate\_server1" **or if [log\_type] == "federate\_server2"** and [app\_id] == "pf"

output {  
if [log\_type] == "federate\_server" **or if [log\_type] == "federate\_server2"** {  
Elasticsearch {

Is the above `or` condition correct? if yes, what will come at below `[type] =>` line

```auto
 mutate {
             replace => {
               "[type]" => "federate_server"
             }
           }

```

I just want to parse logs from both the application servers which will be send by `filebeat` to any `logstash` server but this currently above config seems to be incorrect as only one `log_type` will matched as only one if condition is mentioned.

1. Do we need to mentioned all es hosts in `output` section (like above its mentioned `3 es ` hosts or only one is enough and that will forward the requests to other two es nodes in cluster)

Thanks,

---

<div class="post-metadata">

**Author:** ![prat](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/prat/32/79978_2.png) [@prat](https://discuss.elastic.co/u/prat)\
**Post date:** [September 24, 2021, 2:44pm UTC](https://discuss.elastic.co/t/parsing-data-from-multiple-application-servers-through-logstash/285069/2 "2021-09-24T14:44:57Z")

</div>

Hi All,

I just tried adding or condition in `logstash.yml` but `logstash file validation` command is giving error and due to this `logstash` service is also getting restarted again and again.

```auto
filter {
if [log_type] == "federate_server1" or if [log_type] == "federate_server2" and [app_id] == "pf"
  {
    mutate { gsub => ["message","\|"," "] } grok { patterns_dir => ["/etc/logstash/patterns"] match => { "message" => "%{MY_DATE_PATTERN:timestamp}%{SPACE}%{LOGLEVEL:level}%{SPACE}%{UUID:ConsentID}%{SPACE}%{WORD:TransactionID}%{SPACE}%{WORD:TraceID}%{SPACE}%{GREEDYDATA:messagetext}" } }
    mutate {
             replace => {
               "[type]" => "federate_server"
             }
           }
  }
 output {
  if [log_type] == "federate_server1" or if [log_type] == "federate_server2" {
  elasticsearch {
    hosts => ['http://es1:9200', 'http://es2:9200', 'http://es3:9200']
        user => elastic
    password => "${es_pwd}"
     index => "federate"
     template_name => "federate"
     template_overwrite => "false"
      }
}

```

`logstash` config validation command showing error is due to above config.

```auto
if [log_type] == "federate_server1" or if
[2021-09-24T17:29:28,070][FATAL][org.logstash.Logstash] Logstash stopped processing because of an error: (SystemExit) exit
org.jruby.exceptions.SystemExit: (SystemExit) exit
        at org.jruby.RubyKernel.exit(org/jruby/RubyKernel.java:747) ~[jruby-complete-9.2.19.0.jar:?]
        at org.jruby.RubyKernel.exit(org/jruby/RubyKernel.java:710) ~[jruby-complete-9.2.19.0.jar:?]
        at usr.share.logstash.lib.bootstrap.environment.<main>(/usr/share/logstash/lib/bootstrap/environment.rb:89) ~[?:?]

```

`logstash` logs -

``  
[2021-09-24T17:32:16,364][ERROR][logstash.agent] Failed to execute action {:action=\>LogStash::PipelineAction::Create/pipeline\_id:main, :exception=\>"LogStash::ConfigurationError", :message=\>"Expected one of [\t\r\n], "#", "(" at line 73, column 50 (byte 2916) after filter {\nif [log\_type] == "developer-portal-api\_app\_server" and [app\_id] == "node"\n {\n grok { match =\> { "message" =\> "%{SYSLOGBASE} %{GREEDYDATA:json\_message}" } } json { source =\> "json\_message" }\n mutate {\n replace =\> {\n "[type]" =\> "developer-portal-api\_app\_server"\n }\n }\n }\nif [log\_type] == "developer-portal-spa\_app\_server" and [app\_id] == "node"\n {\n grok { match =\> { "message" =\> "%{SYSLOGBASE} %{GREEDYDATA:json\_message}" } } json { source =\> "json\_message" }\n mutate {\n replace =\> {\n "[type]" =\> "developer-portal-spa\_app\_server"\n }\n }\n }\nif [log\_type] == "ob-admin-api\_app\_server" and [app\_id] == "node"\n {\n grok { match =\> { "message" =\> "%{SYSLOGBASE} %{GREEDYDATA:json\_message}" } } json { source =\> "json\_message" }\n mutate {\n replace =\> {\n "[type]" =\> "ob-admin-api\_app\_server"\n }\n }\n }\nif [log\_type] == "ob-admin-spa\_app\_server" and [app\_id] == "node"\n {\n grok { match =\> { "message" =\> "%{SYSLOGBASE} %{GREEDYDATA:json\_message}" } } json { source =\> "json\_message" }\n mutate {\n replace =\> {\n "[type]" =\> "ob-admin-spa\_app\_server"\n }\n }\n }\nif [log\_type] == "consent-spa\_app\_server" and [app\_id] == "node"\n {\n grok { match =\> { "message" =\> "%{SYSLOGBASE} %{GREEDYDATA:json\_message}" } } json { source =\> "json\_message" }\n mutate {\n replace =\> {\n "[type]" =\> "consent-spa\_app\_server"\n }\n }\n }\nif [log\_type] == "obie-api\_app\_server" and [app\_id] == "app"\n {\n mutate { gsub =\> ["message","\|"," "] } grok { patterns\_dir =\> ["/etc/logstash/patterns"] match =\> { "message" =\> "%{MY\_DATE\_PATTERN:timestamp}%{SPACE}%{LOGLEVEL:level}%{SPACE}%{UUID:ConsentID}%{SPACE}%{WORD:TraceID}%{SPACE}%{WORD:TransactionID}%{SPACE}%{GREEDYDATA:messagetext}" } }\n mutate {\n replace =\> {\n "[type]" =\> "obie-api\_app\_server"\n }\n }\n }\nif [log\_type] == "access\_server" and [app\_id] == "pa"\n {\n grok { match =\> { "message" =\> "%{YEAR}-%{MONTHNUM}-%{MONTHDAY}[T]%{HOUR}:%{MINUTE}(?::?%{SECOND})\| %{USERNAME:exchangeId}\| %{DATA:trackingId}\| %{NUMBER:RoundTrip:int}%{SPACE}ms\| %{NUMBER:ProxyRoundTrip:int}%{SPACE}ms\| %{NUMBER:UserInfoRoundTrip:int}%{SPACE}ms\| %{DATA:Resource}\| %{DATA:subject}\| %{DATA:authmech}\| %{DATA:scopes}\| %{IPV4:Client}\| %{WORD:method}\| %{DATA:Request\_URI}\| %{INT:response\_code}\| %{DATA:failedRuleType}\| %{DATA:failedRuleName}\| %{DATA:APP\_Name}\| %{DATA:Resource\_Name}\| %{DATA:Path\_Prefix}" } }\n mutate {\n replace =\> {\n "[type]" =\> "access\_server"\n }\n }\n }\nif [log\_type] == "federate\_server1" or if ", :backtrace=\>["/usr/share/logstash/logstash-core/lib/logstash/compiler.rb:32:in `compile_imperative'", "org/logstash/execution/AbstractPipelineExt.java:187:in `initialize'", "org/logstash/execution/JavaBasePipelineExt.java:72:in `initialize'", "/usr/share/logstash/logstash-core/lib/logstash/java_pipeline.rb:47:in `initialize'", "/usr/share/logstash/logstash-core/lib/logstash/pipeline\_action/create.rb:52:in `execute'", "/usr/share/logstash/logstash-core/lib/logstash/agent.rb:391:in `block in converge\_state'"]}  
.  
.  
.  
.

[2021-09-24T17:32:18,079][INFO][logstash.javapipeline][.monitoring-logstash] Starting pipeline {:pipeline\_id=\>".monitoring-logstash", "pipeline.workers"=\>1, "pipeline.batch.size"=\>2, "pipeline.batch.delay"=\>50, "pipeline.max\_inflight"=\>2, "pipeline.sources"=\>["monitoring pipeline"], :thread=\>"#\<Thread:0x7796125 run\>"}  
[2021-09-24T17:32:18,895][INFO][logstash.javapipeline][.monitoring-logstash] Pipeline Java execution initialization time {"seconds"=\>0.81}  
[2021-09-24T17:32:18,928][INFO][logstash.javapipeline][.monitoring-logstash] Pipeline started {"pipeline.id"=\>".monitoring-logstash"}  
[2021-09-24T17:32:20,978][INFO][logstash.javapipeline][.monitoring-logstash] Pipeline terminated {"pipeline.id"=\>".monitoring-logstash"}  
[2021-09-24T17:32:21,153][INFO][logstash.runner] Logstash shut down.  
``

`logstash` service is getting restarted continuously.

no docs are getting indexed due to this problem.

```auto
green open federate-000001 FTjBayPLQreqk3PMtb4LJg 2 1 0 0 832b 416b

```

Can someone please point out how to correctly do this?

Thanks,

---

<div class="post-metadata">

**Author:** ![prat](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/prat/32/79978_2.png) [@prat](https://discuss.elastic.co/u/prat)\
**Post date:** [September 24, 2021, 4:54pm UTC](https://discuss.elastic.co/t/parsing-data-from-multiple-application-servers-through-logstash/285069/3 "2021-09-24T16:54:05Z")

</div>

After removing second if (after `or`), validation command gave ok output. also can see indexed getting docs and growing in size but not sure how can i confirm that its parsing logs from both the App servers and not just from any one server.

---

<div class="post-metadata">

**Author:** ![ptamba](https://avatars.discourse-cdn.com/v4/letter/p/7feea3/32.png) [@ptamba](https://discuss.elastic.co/u/ptamba)\
**Post date:** [September 28, 2021, 12:47am UTC](https://discuss.elastic.co/t/parsing-data-from-multiple-application-servers-through-logstash/285069/4 "2021-09-28T00:47:40Z")

</div>

since you already add field type from each server, you can verify whether log comes from both server by filtering each type. you might need to remove this config though, since it replaces the identifier

> [@prat](#):
>
> ```auto
> mutate {
> replace => {
> "[type]" => "federate_server"
> }
> }
> 
> ```

---

<div class="post-metadata">

**Author:** ![prat](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/prat/32/79978_2.png) [@prat](https://discuss.elastic.co/u/prat)\
**Post date:** [September 28, 2021, 8:09pm UTC](https://discuss.elastic.co/t/parsing-data-from-multiple-application-servers-through-logstash/285069/5 "2021-09-28T20:09:05Z")

</div>

Hi @ptamba,

Thanks for your reply.

You mean to say it will replace `federate_server1` or `federate_server2` with only `federate_server` ?

Can you please check and confirm once again.

I can see value of `type` as `log` (`type: log`) so it will replace `log` as `federate_server` right?

The intention is to parse the logs from both the server so that it can be identified from which server they have came when they are indexed into es.

Thanks,

---

<div class="post-metadata">

**Author:** ![ptamba](https://avatars.discourse-cdn.com/v4/letter/p/7feea3/32.png) [@ptamba](https://discuss.elastic.co/u/ptamba)\
**Post date:** [September 28, 2021, 11:42pm UTC](https://discuss.elastic.co/t/parsing-data-from-multiple-application-servers-through-logstash/285069/6 "2021-09-28T23:42:48Z")

</div>

> [@prat](#):
>
> ```auto
> filebeat.inputs:
> - type: log
> fields_under_root: true
> fields:
> log_type: federate_server1
> 
> ```

you already have this in your filebeat. i assume server2, will have log\_type value of federate\_server2

if you don’t need to do anything else to the log on logstash , then just send them to output, and search for log\_type fields in ES. you should have federate\_server1 and federate\_server2 if logs are coming from both server

if you need to do different things based on source , then on logstash

```auto
if [log_type] == ‘federate_server1” {
  #do something to log from server1
}

if [log_type] == ‘federate_server2”
  #do something to log from server2”
}

```

on logstash output, unless you want logs go to different index or different output, there is no need to use conditionals. but if you have other logstash config in place you can do

```auto
output {
  if [log_type] == “federate_server1” or [log_type] == “federate_server2” {
     #output config 
  }

}

```

---

<div class="post-metadata">

**Author:** ![prat](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/prat/32/79978_2.png) [@prat](https://discuss.elastic.co/u/prat)\
**Post date:** [October 2, 2021, 4:35pm UTC](https://discuss.elastic.co/t/parsing-data-from-multiple-application-servers-through-logstash/285069/7 "2021-10-02T16:35:43Z")

</div>

@ptamba, Thanks for your reply and sorry for delay.

> [@ptamba](#):
>
> you already have this in your filebeat. i assume server2, will have log\_type value of federate\_server2

Yes, as you can see in very first comment above under `Application Server 2` output.

> [@ptamba](#):
>
> if you don’t need to do anything else to the log on logstash , then just send them to output, and search for log\_type fields in ES. you should have federate\_server1 and federate\_server2 if logs are coming from both server

I do not want do anything other than parsing logs from both application servers through `logstash` and be able to `identify` them on `kibana` i.e which log is from which application servers out of two.

So in this case, is below correct? but as you said it's removing the `identifier` by `mutate { replace` config below.

I still not get this, my identifier are `log_type` and not `type` and in the mutate config, its written as replace `type`. don't you think either,  
i) it should be `log_type` there (if want to remove the `identifier`)  
ii) or as you said, need to remove the config from `mutate` as it's removing the identifier.

```auto
filter {
if [log_type] == "federate_server1" or [log_type] == "federate_server2" and [app_id] == "pf"
  {
    mutate { gsub => ["message","\|"," "] } grok { patterns_dir => ["/etc/logstash/patterns"] match => { "message" => "%{MY_DATE_PATTERN:timestamp}%{SPACE}%{LOGLEVEL:level}%{SPACE}%{UUID:ConsentID}%{SPACE}%{WORD:TransactionID}%{SPACE}%{WORD:TraceID}%{SPACE}%{GREEDYDATA:messagetext}" } }
    mutate {
             replace => {
               "[type]" => "federate_server"
             }
           }
  }

```

> [@ptamba](#):
>
> on logstash output, unless you want logs go to different index or different output, there is no need to use conditionals. but if you have other logstash config in place you can do
> 
> ```auto
> output {
> if [log_type] == “federate_server1” or [log_type] == “federate_server2” {
> #output config 
> }
> 
> }
> 
> ```

here also i do not want to do anything other than creating index based on the `log_type`, apply `index_template` and ultimately send them to `elasticsearch` .

If i correctly understand your reply on `use of conditionals`, the `if` statements are there as there are different applications logs (see below output) getting parsed through `logstash` (though i have only mentioned one above), may be that is why you are saying not to use conditionals.

Below is the example of three `applications` config in `logstash`.

```auto
filter {
if [log_type] == "portal-api_app_server" and [app_id] == "node"
  {
    grok { match => { "message" => "%{SYSLOGBASE} %{GREEDYDATA:json_message}" } } json { source => "json_message" }
    mutate {
             replace => {
               "[type]" => "portal-api_app_server"
             }
           }
  }
if [log_type] == "federate_ping_server" and [app_id] == "pf"
  {
    mutate { gsub => ["message","\|"," "] } grok { patterns_dir => ["/etc/logstash/patterns"] match => { "message" => "%{MY_DATE_PATTERN:timestamp}%{SPACE}%{LOGLEVEL:level}%{SPACE}%{UUID:ConsentID}%{SPACE}%{WORD:TransactionID}%{SPACE}%{WORD:TraceID}%{SPACE}%{GREEDYDATA:messagetext}" } }
    mutate {
             replace => {
               "[type]" => "federate_ping_server"
             }
           }
  }
if [log_type] == "directory_ping_server" and [app_id] == "pd"
  {
    mutate { gsub => ["message","\|"," "] } grok { patterns_dir => ["/etc/logstash/patterns"] match => { "message" => "%{MY_DATE_PATTERN:timestamp}%{SPACE}%{LOGLEVEL:level}%{SPACE}%{UUID:ConsentID}%{SPACE}%{WORD:TransactionID}%{SPACE}%{WORD:TraceID}%{SPACE}%{GREEDYDATA:messagetext}" } }
    mutate {
             replace => {
               "[type]" => "directory_ping_server"
             }
           }
  }
}  
output {
 if [log_type] == "portal-api_app_server" {
  elasticsearch {
    hosts => ['http://10.10.10.242:9200']
        user => elastic
    password => "${es_pwd}"
     index => "portal-api"
     template_name => "portal-api"
     template_overwrite => "false"
      }
 }
   if [log_type] == "federate_ping_server" {
  elasticsearch {
    hosts => ['http://10.10.10.242:9200']
        user => elastic
    password => "${es_pwd}"
     index => "federate"
     template_name => "federate"
     template_overwrite => "false"
      }
 }
 if [log_type] == "directory_ping_server" {
  elasticsearch {
    hosts => ['http://10.10.10.242:9200']
        user => elastic
    password => "${es_pwd}"
     index => "directory"
     template_name => "directory"
     template_overwrite => "false"
      }
 }
 elasticsearch {
    hosts => ['http://10.10.10.242:9200']
    index => "%{[@metadata][beat]}-%{[@metadata][version]}-%{+YYYY.MM}"
    user => elastic
    password => "${es_pwd}"
  }
}

```

Thanks,

---

<div class="post-metadata">

**Author:** ![prat](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/prat/32/79978_2.png) [@prat](https://discuss.elastic.co/u/prat)\
**Post date:** [October 4, 2021, 3:03pm UTC](https://discuss.elastic.co/t/parsing-data-from-multiple-application-servers-through-logstash/285069/8 "2021-10-04T15:03:55Z")

</div>

Hi @ptamba,

Could you please confirm.

Thanks,

---

<div class="post-metadata">

**Author:** ![ptamba](https://avatars.discourse-cdn.com/v4/letter/p/7feea3/32.png) [@ptamba](https://discuss.elastic.co/u/ptamba)\
**Post date:** [October 4, 2021, 3:09pm UTC](https://discuss.elastic.co/t/parsing-data-from-multiple-application-servers-through-logstash/285069/9 "2021-10-04T15:09:48Z")

</div>

if i understand correctly

on input server1

- log\_type: federate\_server1
- app\_id: pf

on input server2

- log\_type: federate\_server2
- app\_id: pf

why do you have

- if [log\_type] == “federate\_server” ?

based on that input i will do

```auto
filter { 
  if [app_id] == “pf” { 
     # parse pf logs 
  } 
} 

output { 
  if [app_id] == “pf” { 
    # output config for pf 
  }
}

```

which will match all logs with [app\_id] == “pf” regardless of the source

---

<div class="post-metadata">

**Author:** ![prat](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/prat/32/79978_2.png) [@prat](https://discuss.elastic.co/u/prat)\
**Post date:** [October 4, 2021, 3:26pm UTC](https://discuss.elastic.co/t/parsing-data-from-multiple-application-servers-through-logstash/285069/10 "2021-10-04T15:26:32Z")

</div>

Hi @ptamba,

Thanks for your reply.

Got your point on use of `app_id: pf` instead of `log_type: federate_server1` and `log_type: federate_server2`.  
`app_id: pf` will anyway match logs coming from both the application servers.

In this case, can you please update `mutate` config. Does it requries?

```auto
mutate {
             replace => {
               "[type]" => "federate_server"
             }
           }
  }

```

> [@prat](#):
>
> > [@ptamba](#):
> >
> > if you don’t need to do anything else to the log on logstash , then just send them to output, and search for log\_type fields in ES. you should have federate\_server1 and federate\_server2 if logs are coming from both server
> 
> I do not want do anything other than parsing logs from both application servers through `logstash` and be able to `identify` them on `kibana` i.e which log is from which application servers out of two.
> 
> So in this case, is below correct? but as you said it's removing the `identifier` by `mutate { replace` config below.
> 
> I still not get this, my identifier are `log_type` and not `type` and in the mutate config, its written as replace `type` . don't you think either,  
> i) it should be `log_type` there (if want to remove the `identifier` )  
> ii) or as you said, need to remove the config from `mutate` as it's removing the identifier.
> 
> ```auto
> filter {
> if [log_type] == "federate_server1" or [log_type] == "federate_server2" and [app_id] == "pf"
> {
> mutate { gsub => ["message","\|"," "] } grok { patterns_dir => ["/etc/logstash/patterns"] match => { "message" => "%{MY_DATE_PATTERN:timestamp}%{SPACE}%{LOGLEVEL:level}%{SPACE}%{UUID:ConsentID}%{SPACE}%{WORD:TransactionID}%{SPACE}%{WORD:TraceID}%{SPACE}%{GREEDYDATA:messagetext}" } }
> mutate {
> replace => {
> "[type]" => "federate_server"
> }
> }
> }
> 
> ```

---

<div class="post-metadata">

**Author:** ![ptamba](https://avatars.discourse-cdn.com/v4/letter/p/7feea3/32.png) [@ptamba](https://discuss.elastic.co/u/ptamba)\
**Post date:** [October 4, 2021, 3:27pm UTC](https://discuss.elastic.co/t/parsing-data-from-multiple-application-servers-through-logstash/285069/11 "2021-10-04T15:27:56Z")

</div>

> [@prat](#):
>
> In this case, can you please update `mutate` config. Does it requries?

depends, what do you want to do with that filter

---

<div class="post-metadata">

**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\
**Post date:** [November 1, 2021, 3:28pm UTC](https://discuss.elastic.co/t/parsing-data-from-multiple-application-servers-through-logstash/285069/12 "2021-11-01T15:28:26Z")

</div>

This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.
