# How to split linux and windows data in two different indexes

**URL:** https://discuss.elastic.co/t/how-to-split-linux-and-windows-data-in-two-different-indexes/143697
**Category:** Logstash
**Created:** [August 9, 2018, 12:57pm UTC](https://discuss.elastic.co/t/how-to-split-linux-and-windows-data-in-two-different-indexes/143697 "2018-08-09T12:57:59Z")
**Posts on this page:** 11
**Page:** 1

<div class="post-metadata">

### Author: ![gabberoid](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/gabberoid/32/33466_2.png) [@gabberoid](https://discuss.elastic.co/u/gabberoid)
#### Post date: [August 9, 2018, 12:57pm UTC](https://discuss.elastic.co/t/how-to-split-linux-and-windows-data-in-two-different-indexes/143697/1 "2018-08-09T12:57:59Z")

</div>

Hi.  
I created two conf file in /etc/logstash/conf.d for two hosts: Linux hosts with filebeat and windows hosts with winlogbeat.  
In output part I wrote two dofferent indexes in pattern: "linux-%{+YYYY.MM.dd}" and "windows-%{+YYYY.MM.dd}".  
In input part I wrote different ports for each hosts: 5045 for linux and 5046 for windows with host address "0.0.0.0" in both files.  
Now the whole data from both hosts is in the only one index: windows-2018.08.09.  
What I did wrong?

---

<div class="post-metadata">

### Author: ![magnusbaeck](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/magnusbaeck/32/44943_2.png) [@magnusbaeck](https://discuss.elastic.co/u/magnusbaeck)
#### Post date: [August 9, 2018, 1:29pm UTC](https://discuss.elastic.co/t/how-to-split-linux-and-windows-data-in-two-different-indexes/143697/2 "2018-08-09T13:29:15Z")

</div>

You've configured Logstash with a single pipeline. Unless you use conditionals all events will reach all filters and outputs. Logstash will concatenate all files in /etc/logstash/conf.d.

---

<div class="post-metadata">

### Author: ![gabberoid](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/gabberoid/32/33466_2.png) [@gabberoid](https://discuss.elastic.co/u/gabberoid)
#### Post date: [August 9, 2018, 1:42pm UTC](https://discuss.elastic.co/t/how-to-split-linux-and-windows-data-in-two-different-indexes/143697/3 "2018-08-09T13:42:53Z")

</div>

Can you give an example for these conditionals?  
What is a better way to configure: create two pipeline for linux hosts and windows hosts or create a conditional in \*.conf file?  
Thanks

---

<div class="post-metadata">

### Author: ![magnusbaeck](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/magnusbaeck/32/44943_2.png) [@magnusbaeck](https://discuss.elastic.co/u/magnusbaeck)
#### Post date: [August 9, 2018, 1:46pm UTC](https://discuss.elastic.co/t/how-to-split-linux-and-windows-data-in-two-different-indexes/143697/4 "2018-08-09T13:46:31Z")

</div>

> Can you give an example for these conditionals?

> **[Accessing event data and fields | Logstash Reference \[8.11\] | Elastic](https://www.elastic.co/guide/en/logstash/current/event-dependent-configuration.html#conditionals)**

> What is a better way to configure: create two pipeline for linux hosts and windows hosts or create a conditional in \*.conf file?

I'm struggling to come up with an example where it makes any significant difference.

---

<div class="post-metadata">

### Author: ![gabberoid](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/gabberoid/32/33466_2.png) [@gabberoid](https://discuss.elastic.co/u/gabberoid)
#### Post date: [August 9, 2018, 1:51pm UTC](https://discuss.elastic.co/t/how-to-split-linux-and-windows-data-in-two-different-indexes/143697/5 "2018-08-09T13:51:51Z")

</div>

I still didn't recognize how I can configure Logstash. I'll appreciate if you can explain.  
I have 100 servers (10 linux and 90 windows). I want to split all logs into 2 indexes : linux and windows.  
Do I need to create a huge \*.conf file in /etc/logstash/conf.d with inputs from 100 servers with different ports under conditions you've suggested or I need to create many pipelines with many \*.conf files?  
May be it will be more convenient and appropriate to concatenate all events into one kind of index splitted by days (I need also to delete indexes older than 1 year)?

---

<div class="post-metadata">

### Author: ![magnusbaeck](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/magnusbaeck/32/44943_2.png) [@magnusbaeck](https://discuss.elastic.co/u/magnusbaeck)
#### Post date: [August 9, 2018, 2:17pm UTC](https://discuss.elastic.co/t/how-to-split-linux-and-windows-data-in-two-different-indexes/143697/6 "2018-08-09T14:17:08Z")

</div>

> I have 100 servers (10 linux and 90 windows). I want to split all logs into 2 indexes : linux and windows.  
> Do I need to create a huge \*.conf file in /etc/logstash/conf.d with inputs from 100 servers with different ports under conditions you've suggested or I need to create many pipelines with many \*.conf files?

One Beats input is enough but you can have more if you want. See the example at [elasticsearch - Make logstash add different inputs to different indices - Stack Overflow](https://stackoverflow.com/a/27147688/414355).

> May be it will be more convenient and appropriate to concatenate all events into one kind of index splitted by days (I need also to delete indexes older than 1 year)?

You'll definitely want to use time-based indexes (though not necessarily daily; depending on the log volumes monthly indexes might be better).

---

<div class="post-metadata">

### Author: ![gabberoid](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/gabberoid/32/33466_2.png) [@gabberoid](https://discuss.elastic.co/u/gabberoid)
#### Post date: [August 9, 2018, 2:21pm UTC](https://discuss.elastic.co/t/how-to-split-linux-and-windows-data-in-two-different-indexes/143697/7 "2018-08-09T14:21:33Z")

</div>

The monthly index will be with this pattern in an output:  
"%{type}-%{+YYYY.MM}"?

---

<div class="post-metadata">

### Author: ![magnusbaeck](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/magnusbaeck/32/44943_2.png) [@magnusbaeck](https://discuss.elastic.co/u/magnusbaeck)
#### Post date: [August 9, 2018, 2:33pm UTC](https://discuss.elastic.co/t/how-to-split-linux-and-windows-data-in-two-different-indexes/143697/8 "2018-08-09T14:33:06Z")

</div>

Yes.

---

<div class="post-metadata">

### Author: ![gabberoid](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/gabberoid/32/33466_2.png) [@gabberoid](https://discuss.elastic.co/u/gabberoid)
#### Post date: [August 9, 2018, 2:43pm UTC](https://discuss.elastic.co/t/how-to-split-linux-and-windows-data-in-two-different-indexes/143697/9 "2018-08-09T14:43:47Z")

</div>

Thank you!

But I still don't understand :

> [@magnusbaeck](#):
>
> One Beats input is enough but you can have more if you want. See the example at

How I can split the data into different indexes if in many conf files with conditions ES concatenates all events. For example these are two different conf files in conf.d:

For sys log with 5044 port:

input {  
beats {  
port =\> 5044  
host =\> "0.0.0.0"  
}  
}

filter {  
if [fileset][module] == "system" {  
if [fileset][name] == "auth" {  
grok {  
match =\> { "message" =\> ["%{SYSLOGTIMESTAMP:[system][auth][timestamp]} %{SYSLOGHOST:[system][auth][hostname]} sshd(?:[%{POSINT:[system][auth][pid]}])?: %{DATA:[system][auth][ssh][event]} %{DATA:[system][auth][ssh][method]} for (invalid user )?%{DATA:[system][auth][user]} from %{IPORHOST:[system][auth][ssh][ip]} port %{NUMBER:[system][auth][ssh][port]} ssh2(: %{GREEDYDATA:[system][auth][ssh][signature]})?",  
"%{SYSLOGTIMESTAMP:[system][auth][timestamp]} %{SYSLOGHOST:[system][auth][hostname]} sshd(?:[%{POSINT:[system][auth][pid]}])?: %{DATA:[system][auth][ssh][event]} user %{DATA:[system][auth][user]} from %{IPORHOST:[system][auth][ssh][ip]}",  
"%{SYSLOGTIMESTAMP:[system][auth][timestamp]} %{SYSLOGHOST:[system][auth][hostname]} sshd(?:[%{POSINT:[system][auth][pid]}])?: Did not receive identification string from %{IPORHOST:[system][auth][ssh][dropped\_ip]}",  
"%{SYSLOGTIMESTAMP:[system][auth][timestamp]} %{SYSLOGHOST:[system][auth][hostname]} sudo(?:[%{POSINT:[system][auth][pid]}])?: \s\*%{DATA:[system][auth][user]} ☹ %{DATA:[system][auth][sudo][error]} ;)? TTY=%{DATA:[system][auth][sudo][tty]} ; PWD=%{DATA:[system][auth][sudo][pwd]} ; USER=%{DATA:[system][auth][sudo][user]} ; COMMAND=%{GREEDYDATA:[system][auth][sudo][command]}",  
"%{SYSLOGTIMESTAMP:[system][auth][timestamp]} %{SYSLOGHOST:[system][auth][hostname]} groupadd(?:[%{POSINT:[system][auth][pid]}])?: new group: name=%{DATA:system.auth.groupadd.name}, GID=%{NUMBER:system.auth.groupadd.gid}",  
"%{SYSLOGTIMESTAMP:[system][auth][timestamp]} %{SYSLOGHOST:[system][auth][hostname]} useradd(?:[%{POSINT:[system][auth][pid]}])?: new user: name=%{DATA:[system][auth][user][add][name]}, UID=%{NUMBER:[system][auth][user][add][uid]}, GID=%{NUMBER:[system][auth][user][add][gid]}, home=%{DATA:[system][auth][user][add][home]}, shell=%{DATA:[system][auth][user][add][shell]}$",  
"%{SYSLOGTIMESTAMP:[system][auth][timestamp]} %{SYSLOGHOST:[system][auth][hostname]} %{DATA:[system][auth][program]}(?:[%{POSINT:[system][auth][pid]}])?: %{GREEDYMULTILINE:[system][auth][message]}"] }  
pattern\_definitions =\> {  
"GREEDYMULTILINE"=\> "(.|\n)_"  
}  
remove\_field =\> "message"  
}  
date {  
match =\> ["[system][auth][timestamp]", "MMM d HH:mm:ss", "MMM dd HH:mm:ss" ]  
}  
geoip {  
source =\> "[system][auth][ssh][ip]"  
target =\> "[system][auth][ssh][geoip]"  
}  
}  
else if [fileset][name] == "syslog" {  
grok {  
match =\> { "message" =\> ["%{SYSLOGTIMESTAMP:[system][syslog][timestamp]} %{SYSLOGHOST:[system][syslog][hostname]} %{DATA:[system][syslog][program]}(?:[%{POSINT:[system][syslog][pid]}])?: %{GREEDYMULTILINE:[system][syslog][message]}"] }  
pattern\_definitions =\> { "GREEDYMULTILINE" =\> "(.|\n)_" }  
remove\_field =\> "message"  
}  
date {  
match =\> ["[system][syslog][timestamp]", "MMM d HH:mm:ss", "MMM dd HH:mm:ss" ]  
}  
}  
}  
}

output {  
elasticsearch {  
sniffing =\> true  
hosts =\> ["uk1lv8702:9200", "uk1lv8703:9200", "uk1lv8704:9200"]  
index =\> "linux-%{+YYYY.MM}" # days index  
manage\_template =\> false  
#index =\> "%{[@metadata][beat]}-%{+YYYY.MM.dd}"  
}  
}

For eventlog with 5046 port:

input {  
beats {  
port =\> 5046  
host =\> "0.0.0.0"  
}  
}

filter {  
grok {  
match =\> { "message" =\> "%{GREEDYDATA:message}" }  
}  
}

output {  
elasticsearch {  
sniffing =\> true  
hosts =\> ["uk1lv8702:9200", "uk1lv8703:9200", "uk1lv8704:9200"]  
index =\> "windows-%{+YYYY.MM.dd}"  
manage\_template =\> false  
}  
}

---

<div class="post-metadata">

### Author: ![magnusbaeck](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/magnusbaeck/32/44943_2.png) [@magnusbaeck](https://discuss.elastic.co/u/magnusbaeck)
#### Post date: [August 9, 2018, 2:49pm UTC](https://discuss.elastic.co/t/how-to-split-linux-and-windows-data-in-two-different-indexes/143697/10 "2018-08-09T14:49:50Z")

</div>

You must have some characteristic (tag or field) that sets apart your different kinds of logs (you haven't showed example events so I can't get more concrete than that). Use that to select which filters and outputs to apply. You can set any fields and tags you want in the Winlogbeat and Filebeat configurations. Let's say you set `type` to "winevent" in Winlogbeat; then you can wrap the current contents of the filter and output sections in your Winlogbeat file with

```nohighlight
if [type] == "winevent" {
  ...
}

```

---

<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: [September 6, 2018, 2:49pm UTC](https://discuss.elastic.co/t/how-to-split-linux-and-windows-data-in-two-different-indexes/143697/11 "2018-09-06T14:49:55Z")

</div>

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