# How to use multiple inputs in Logstash?
**URL:** https://discuss.elastic.co/t/how-to-use-multiple-inputs-in-logstash/36247
**Category:** Logstash
**Created:** [December 3, 2015, 6:35am UTC](https://discuss.elastic.co/t/how-to-use-multiple-inputs-in-logstash/36247 "2015-12-03T06:35:09Z")
**Posts on this page:** 13
**Page:** 1
### Author:  [@thyfere](https://discuss.elastic.co/u/thyfere)
#### Post date: [December 3, 2015, 6:35am UTC](https://discuss.elastic.co/t/how-to-use-multiple-inputs-in-logstash/36247/1 "2015-12-03T06:35:09Z")
Hi,
My setup: Two Windows boxes; one is for Elasticsearch and Kibana, and other is for Logstash. I am already dumping Windows events to my ELK setup.
Now, I want to collect Microsoft Exchange logs too. How can I control multiple inputs in logstash or how will it work in ELK stack? I will use this guide to collect Exchange log: [https://elijahpaul.co.uk/analysing-exchange-2013-message-tracking-logs-using-elk-elasticsearch-logstash-kibana/](https://elijahpaul.co.uk/analysing-exchange-2013-message-tracking-logs-using-elk-elasticsearch-logstash-kibana/)
As per above-mentioned guide, Exchange logs will be collected in syslog format but I am collecting Windows events in Jason, will it be okay?
---
### Author:  [@magnusbaeck](https://discuss.elastic.co/u/magnusbaeck)
#### Post date: [December 3, 2015, 7:01am UTC](https://discuss.elastic.co/t/how-to-use-multiple-inputs-in-logstash/36247/2 "2015-12-03T07:01:54Z")
Yes. In each input, use the `type` field and/or tags so that you can distinguish between different kinds of messages in subsequent filters and/or outputs.
---
### Author:  [@thyfere](https://discuss.elastic.co/u/thyfere)
#### Post date: [December 3, 2015, 7:04am UTC](https://discuss.elastic.co/t/how-to-use-multiple-inputs-in-logstash/36247/3 "2015-12-03T07:04:25Z")
Thanks Magnus,
So, how will they be distinguished in Kibana? Do I need to be create new index for it? Maybe I am asking a lame question but please bear with me.
---
### Author:  [@magnusbaeck](https://discuss.elastic.co/u/magnusbaeck)
#### Post date: [December 3, 2015, 7:42am UTC](https://discuss.elastic.co/t/how-to-use-multiple-inputs-in-logstash/36247/4 "2015-12-03T07:42:47Z")
You can use different indexes or distinguish between them using the type and/or tags. I prefer the latter since each index has a fixed RAM overhead.
---
### Author:  [@thyfere](https://discuss.elastic.co/u/thyfere)
#### Post date: [December 9, 2015, 7:48am UTC](https://discuss.elastic.co/t/how-to-use-multiple-inputs-in-logstash/36247/5 "2015-12-09T07:48:41Z")
Hi Magnus,
Thanks for your response. Can I use the same port while configuring new INPUT in logstash for Microsoft Exchange?
---
### Author:  [@magnusbaeck](https://discuss.elastic.co/u/magnusbaeck)
#### Post date: [December 9, 2015, 8:59am UTC](https://discuss.elastic.co/t/how-to-use-multiple-inputs-in-logstash/36247/6 "2015-12-09T08:59:11Z")
If you want to have different kinds of messages arriving on the same port you need to use some other kind of mechanism to distinguish events and modify the `type` field and/or the event's tags array. This could e.g. be done with the help of regexp matches on the event contents. If that's possible to do then yes you can use the same port.
---
### Author:  [@thyfere](https://discuss.elastic.co/u/thyfere)
#### Post date: [December 9, 2015, 10:48am UTC](https://discuss.elastic.co/t/how-to-use-multiple-inputs-in-logstash/36247/7 "2015-12-09T10:48:18Z")
I am a newbie to ELK stack. Could you please help me in writing the logstash-indexer for Microsoft Events and Exchange? Here is my config file:
> input {
> tcp {
> codec =\> "json"
> port =\> 5544
> tags =\> ["windows","nxlog"]
> type =\> "nxlog-json"
> }
> } # end input
>
> filter {
> if [type] == "nxlog-json" {
> date {
> match =\> ["[EventTime]", "YYYY-MM-dd HH:mm:ss"]
> # timezone =\> "Europe/London"
> }
> mutate {
> rename =\> ["AccountName", "user"]
> rename =\> ["AccountType", "[eventlog][account\_type]" ]
> rename =\> ["ActivityId", "[eventlog][activity\_id]" ]
> rename =\> ["Address", "ip6"]
> rename =\> ["ApplicationPath", "[eventlog][application\_path]" ]
> rename =\> ["AuthenticationPackageName", "[eventlog][authentication\_package\_name]" ]
> rename =\> ["Category", "[eventlog][category]" ]
> rename =\> ["Channel", "[eventlog][channel]" ]
> rename =\> ["Domain", "domain"]
> rename =\> ["EventID", "[eventlog][event\_id]" ]
> rename =\> ["EventType", "[eventlog][event\_type]" ]
> rename =\> ["File", "[eventlog][file\_path]" ]
> rename =\> ["Guid", "[eventlog][guid]" ]
> rename =\> ["Hostname", "hostname"]
> rename =\> ["Interface", "[eventlog][interface]" ]
> rename =\> ["InterfaceGuid", "[eventlog][interface\_guid]" ]
> rename =\> ["InterfaceName", "[eventlog][interface\_name]" ]
> rename =\> ["IpAddress", "ip"]
> rename =\> ["IpPort", "port"]
> rename =\> ["Key", "[eventlog][key]" ]
> rename =\> ["LogonGuid", "[eventlog][logon\_guid]" ]
> rename =\> ["LogonType", "[eventlog][logon\_type]" ]
> rename =\> ["Message", "message"]
> rename =\> ["ModifyingUser", "[eventlog][modifying\_user]" ]
> rename =\> ["NewProfile", "[eventlog][new\_profile]" ]
> rename =\> ["OldProfile", "[eventlog][old\_profile]" ]
> rename =\> ["Port", "port"]
> rename =\> ["PrivilegeList", "[eventlog][privilege\_list]" ]
> rename =\> ["ProcessID", "pid"]
> rename =\> ["ProcessName", "[eventlog][process\_name]" ]
> rename =\> ["ProviderGuid", "[eventlog][provider\_guid]" ]
> rename =\> ["ReasonCode", "[eventlog][reason\_code]" ]
> rename =\> ["RecordNumber", "[eventlog][record\_number]" ]
> rename =\> ["ScenarioId", "[eventlog][scenario\_id]" ]
> rename =\> ["Severity", "level"]
> rename =\> ["SeverityValue", "[eventlog][severity\_code]" ]
> rename =\> ["SourceModuleName", "nxlog\_input"]
> rename =\> ["SourceName", "[eventlog][program]" ]
> rename =\> ["SubjectDomainName", "[eventlog][subject\_domain\_name]" ]
> rename =\> ["SubjectLogonId", "[eventlog][subject\_logonid]" ]
> rename =\> ["SubjectUserName", "[eventlog][subject\_user\_name]" ]
> rename =\> ["SubjectUserSid", "[eventlog][subject\_user\_sid]" ]
> rename =\> ["System", "[eventlog][system]" ]
> rename =\> ["TargetDomainName", "[eventlog][target\_domain\_name]" ]
> rename =\> ["TargetLogonId", "[eventlog][target\_logonid]" ]
> rename =\> ["TargetUserName", "[eventlog][target\_user\_name]" ]
> rename =\> ["TargetUserSid", "[eventlog][target\_user\_sid]" ]
> rename =\> ["ThreadID", "thread"]
> }
> mutate {
> remove\_field =\> [
> "CurrentOrNextState",
> "Description",
> "EventReceivedTime",
> "EventTime",
> "EventTimeWritten",
> "IPVersion",
> "KeyLength",
> "Keywords",
> "LmPackageName",
> "LogonProcessName",
> # "LogonType",
> "Name",
> "Opcode",
> "OpcodeValue",
> "PolicyProcessingMode",
> "Protocol",
> "ProtocolType",
> "SourceModuleType",
> "State",
> "Task",
> "TransmittedServices",
> "Type",
> "UserID",
> "Version"
> ]
> }
> }
> }
> output {
> elasticsearch {
> host =\> "ip\_address"
> protocol =\> "http"
> }
> } # end output
---
### Author:  [@magnusbaeck](https://discuss.elastic.co/u/magnusbaeck)
#### Post date: [December 9, 2015, 11:41am UTC](https://discuss.elastic.co/t/how-to-use-multiple-inputs-in-logstash/36247/8 "2015-12-09T11:41:12Z")
You're not saying what the input events look like (are they also sent via NXLog?) or what you expect Logstash to do with those events so it's impossible to help.
Have your Exchange events sent to Logstash and use a `stdout { codec => rubydebug }` output to inspect what you get. Then start applying filters, one by one, to morph the events into what you want to have.
---
### Author:  [@thyfere](https://discuss.elastic.co/u/thyfere)
#### Post date: [December 9, 2015, 1:12pm UTC](https://discuss.elastic.co/t/how-to-use-multiple-inputs-in-logstash/36247/9 "2015-12-09T13:12:31Z")
Yes, they are sent via nxlog. The config file, I copied earlier was for Windows Events. I guess, for Windows events, it's mentioned what to do with them.
For exchange I am planning for following config but I don't know how to merge both inputs in one file.
input
```
#udp syslogs stream via 5141
input {
udp {
type => "Exchange"
port => 5141
}
}
```
filter
```
filter {
if [type] == "Exchange" {
csv {
add_tag => ['exh_msg_trk']
columns => ['logdate', 'client_ip', 'client_hostname', 'server_ip', 'server_hostname', 'source_context', 'connector_id', 'source', 'event_id', 'internal_message_id', 'message_id', 'network_message_id', 'recipient_address', 'recipient_status', 'total_bytes', 'recipient_count', 'related_recipient_address', 'reference', 'message_subject', 'sender_address', 'return_path', 'message_info', 'directionality', 'tenant_id', 'original_client_ip', 'original_server_ip', 'custom_data']
remove_field => ["logdate"]
}
grok {
match => ["message", "%{TIMESTAMP_ISO8601:timestamp}"]
}
mutate {
convert => ["total_bytes", "integer"]
convert => ["recipient_count", "integer"]
split => ["recipient_address", ";"]
split => ["source_context", ";"]
split => ["custom_data", ";"]
}
date {
match => ["timestamp", "ISO8601"]
timezone => "Europe/London"
remove_field => ["timestamp"]
}
if "_grokparsefailure" in [tags] {
drop { }
}
}
}
```
output
```
output {
elasticsearch { host => localhost }
stdout { codec => rubydebug }
}
```
---
### Author:  [@magnusbaeck](https://discuss.elastic.co/u/magnusbaeck)
#### Post date: [December 9, 2015, 1:42pm UTC](https://discuss.elastic.co/t/how-to-use-multiple-inputs-in-logstash/36247/10 "2015-12-09T13:42:13Z")
> For exchange I am planning for following config but I don't know how to merge both inputs in one file.
You mean you don't know how to listen to messages on a single UDP port? It seems the eventlog stuff arrives as JSON and the eventlog messages are CSV. Would a simple condition that checks if the first character of the payload is a left curly brace do? If there's a curly brace, use the json filter to unmarshal the JSON payload.
Who or what will be sending the Exchange messages?
---
### Author:  [@thyfere](https://discuss.elastic.co/u/thyfere)
#### Post date: [December 13, 2015, 7:28am UTC](https://discuss.elastic.co/t/how-to-use-multiple-inputs-in-logstash/36247/11 "2015-12-13T07:28:27Z")
Hi Again Magnus,
Thanks a lot for your reply but to be very honest, I didn't understand your reply completely 🙂 I apologize for that. But lemme brief again. We got 3 Mail and 2 CAS Exchange servers. I want to dump their message tracking logs to our ELK stack servers. I already know which logs need to be collected. But now the confusion is at logstash config file which is already configured for Event Logs as I have mentioned earlier. Yes, Event logs are coming as Jason via Nxlog. Exchange message tracking logs will also push via NXLOG. Here is its output lines:
```
Path in_exchange => out_exchange
```
Now, how can I deal both incoming Windows Event and Exchange logs in a single logstash config file. I think, now you may get a whole idea.
---
### Author:  [@magnusbaeck](https://discuss.elastic.co/u/magnusbaeck)
#### Post date: [December 13, 2015, 3:52pm UTC](https://discuss.elastic.co/t/how-to-use-multiple-inputs-in-logstash/36247/12 "2015-12-13T15:52:13Z")
So, the email servers send datagrams with a single JSON payload? And the Exchange logs are not JSON? In that case it sounds like you'll be receiving UDP datagrams with payloads that more or less look like either of these two:
```
{"message": "some message", ...}
<123>Dec 13 16:48:11 some-hostname some-daemon[3142]: some message
```
If this is true you can take a peek as the message payload and see if it looks like JSON. If it does, parse it as JSON. Otherwise assume it's syslog and use a grok filter to extract fields from the syslog message.
```auto
filter {
if [message] =~ /^\{/ {
json {
source => "message"
}
} else {
grok {
...
}
}
}
```
This would be easier if you'd use different ports or wrap the Exchange messages as JSON instead of sending them as syslog messages.
---
### Author:  [@system](https://discuss.elastic.co/u/system)
#### Post date: [July 6, 2017, 5:18am UTC](https://discuss.elastic.co/t/how-to-use-multiple-inputs-in-logstash/36247/13 "2017-07-06T05:18:41Z")