# Logstash sending all data to elasticsearch via wrong pipeline

**URL:** <https://discuss.elastic.co/t/logstash-sending-all-data-to-elasticsearch-via-wrong-pipeline/165799>\
**Category:** Logstash\
**Created:** [January 25, 2019, 3:39pm UTC](https://discuss.elastic.co/t/logstash-sending-all-data-to-elasticsearch-via-wrong-pipeline/165799 "2019-01-25T15:39:02Z")\
**Posts on this page:** 12\
**Page:** 1

<div class="post-metadata">

**Author:** ![Wellguys](https://avatars.discourse-cdn.com/v4/letter/w/8e7dd6/32.png) [@Wellguys](https://discuss.elastic.co/u/Wellguys)\
**Post date:** [January 25, 2019, 3:39pm UTC](https://discuss.elastic.co/t/logstash-sending-all-data-to-elasticsearch-via-wrong-pipeline/165799/1 "2019-01-25T15:39:02Z")

</div>

Hello everyone.

I'm having an issue with pipelines, I think I caused the problem but can't find where it is. I have two servers, one running ELK and Filebeat, the other running Apache and Filebeat.

What I am trying to do is to get all the logs from /var/log/messages to one pipeline, and the ones from /varr/log/httpd/access\_log to another one. My goal is to have two seperate indexes.  
But unfortunately, all the logs are still going through the first pipeline and not the second one.

here are the config files:

pipe1.conf

```
#######
#INPUT#
#######

input {
  beats {
    #port d'ecoute de logstash
    port => 5044
    #pas d'authentification par certificat
    ssl => false
  }
  file {
    path => "/var/log/messages"

  }
}

########
#FILTER#
########

filter {
    grok {
      match => { "message" => "%{SYSLOGTIMESTAMP:syslog_timestamp} %{SYSLOGHOST:syslog_hostname} %{WORD:process}(?:\[%{POSINT:syslog_pid}\])?: %{GREEDYDATA:syslog_message}" }
    }

    date {
      match => ["timestamp", "MMM d HH:mm:ss", "MMM dd HH:mm:ss"]
    }
}

########
#OUTPUT#
########

output {
 elasticsearch {
  hosts => localhost
  index => "indextest"
       }
stdout {
    codec => rubydebug
       }
}

```

pipe2.conf

```
#######
#INPUT#
#######

input {
  file {
    path => "/var/log/httpd/access_log"
  }
  beats {
   #port d'ecoute de logstash
   port => 5044
   #pas d'authentification par certificat
   ssl => false
   }
}

########
#FILTER#
########

filter {
    grok {
      match => { "message" => "%{COMBINEDAPACHELOG}" }
    }
    date {
      match => ["timestamp" , "dd/MMM/yyyy:HH:mm:ss Z"]
    }
}

########
#OUTPUT#
########

output {
 elasticsearch {
  hosts => "localhost"
  index => "apache"
       }
stdout {
    codec => rubydebug
       }
}

```

I looked at the logstash logs and everything is running fine.  
It's surely a onditional problem.

---

<div class="post-metadata">

**Author:** ![danhermann](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/danhermann/32/33024_2.png) [@danhermann](https://discuss.elastic.co/u/danhermann)\
**Post date:** [January 25, 2019, 3:43pm UTC](https://discuss.elastic.co/t/logstash-sending-all-data-to-elasticsearch-via-wrong-pipeline/165799/2 "2019-01-25T15:43:05Z")

</div>

@Wellguys, if you have just separated your Logstash configuration into two different files, you aren't running in two separate pipelines. You have to configure that in your `pipelines.yml` file. If not, all your Logstash `*.conf` files are concatenated together and run in a single pipeline. That's almost certainly why you're seeing logs in the other ES index.

See [here](https://www.elastic.co/guide/en/logstash/6.4/multiple-pipelines.html) for an example.

---

<div class="post-metadata">

**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:** [January 25, 2019, 4:10pm UTC](https://discuss.elastic.co/t/logstash-sending-all-data-to-elasticsearch-via-wrong-pipeline/165799/3 "2019-01-25T16:10:21Z")

</div>

> [@Wellguys](#):
>
> I looked at the logstash logs and everything is running fine.

That seems unlikely. You have two beats inputs listening on port 5044. One of them should be getting "Error: Address already in use".

You might want to run with "--config.test\_and\_exit --log.level debug" to verify it is picking up the configuration that you expect.

---

<div class="post-metadata">

**Author:** ![Wellguys](https://avatars.discourse-cdn.com/v4/letter/w/8e7dd6/32.png) [@Wellguys](https://discuss.elastic.co/u/Wellguys)\
**Post date:** [January 28, 2019, 8:18am UTC](https://discuss.elastic.co/t/logstash-sending-all-data-to-elasticsearch-via-wrong-pipeline/165799/4 "2019-01-28T08:18:20Z")

</div>

My pipelines.yml is like this:

```
- pipeline.id: pipe1
  path.config: "/etc/logstash/conf.d/pipe1.conf"

- pipeline.id: pipe2
  path.config: "/etc/logstash/conf.d/pipe2.conf"

```

I restarted the service and it is true that my second pipeline is not running well. I cant't find the index in kibana either.

```
[2019-01-28][ERROR][logstash.pipeline] A plugin had an unrecoverable error. Will restart this plugin.
  Pipeline_id:pipe2

```

I'm using centos7 and I don't really know how to use "--config.test\_and\_exit --log.level debug". I enabled it in logstash.yml and restarted logstash but nothing seems to happen. I'm not getting any logs either.

---

<div class="post-metadata">

**Author:** ![Wellguys](https://avatars.discourse-cdn.com/v4/letter/w/8e7dd6/32.png) [@Wellguys](https://discuss.elastic.co/u/Wellguys)\
**Post date:** [January 28, 2019, 8:27am UTC](https://discuss.elastic.co/t/logstash-sending-all-data-to-elasticsearch-via-wrong-pipeline/165799/5 "2019-01-28T08:27:43Z")

</div>

I restarted the service normally again and this time pipe1 is having a problem. Meanwhile, pipe2 is taking all the logs.  
Is it not possible to send all data via one port, and then seperate them depending on log source file ?

---

<div class="post-metadata">

**Author:** ![Christian\_Dahlqvist](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/christian_dahlqvist/32/4617_2.png) [@Christian\_Dahlqvist](https://discuss.elastic.co/u/Christian_Dahlqvist)\
**Post date:** [January 28, 2019, 8:35am UTC](https://discuss.elastic.co/t/logstash-sending-all-data-to-elasticsearch-via-wrong-pipeline/165799/6 "2019-01-28T08:35:45Z")

</div>

You can not have two input plugins listening to the same port, which is why one of the pipelines always fails. You could however put the beats input plugin in a third separate pipeline and then use conditionals to direct the data to your other pipelines using [pipeline-to-pipeline communication](https://www.elastic.co/guide/en/logstash/6.5/pipeline-to-pipeline.html).

---

<div class="post-metadata">

**Author:** ![Wellguys](https://avatars.discourse-cdn.com/v4/letter/w/8e7dd6/32.png) [@Wellguys](https://discuss.elastic.co/u/Wellguys)\
**Post date:** [January 28, 2019, 9:49am UTC](https://discuss.elastic.co/t/logstash-sending-all-data-to-elasticsearch-via-wrong-pipeline/165799/7 "2019-01-28T09:49:50Z")

</div>

Oh yes thats perfect, it works great with a third pipeline. I'll send all my configuration for people who would have the same problem. Thanks for the help everyone!

pipeline.yml

```
- pipeline.id: analyzer
  path.config: "/etc/logstash/conf.d/analyzer.conf"

- pipeline.id: pipe1
  path.config: "/etc/logstash/conf.d/pipe1.conf"
- pipeline.id: pipe2
  path.config: "/etc/logstash/conf.d/pipe2.conf"

```

analyser.conf

```
#INPUT#
input {
  beats {
    port => 5044
    ssl => false
  }
}

#OUTPUT#
output {
  if [source] == "/var/log/messages" {
    pipeline { send_to => pipe1 }
  }
  else if [source] == "/var/log/httpd/access_log" {
    pipeline { send_to => pipe2 }
  }
}

```

pipe2.conf

```
#INPUT#
input {
  pipeline {
    address => pipe2
  }
}

#FILTER#
filter {
    grok {
      match => { "message" => "%{COMBINEDAPACHELOG}" }
    }
    date {
      match => ["timestamp" , "dd/MMM/yyyy:HH:mm:ss Z"]
    }
}

#OUTPUT#
output {
 elasticsearch {
  hosts => "localhost"
  index => "apache"
       }
stdout {
    codec => rubydebug
       }
}

```

pipe1.conf

```
#INPUT#
input {
  pipeline {
    address => pipe1
  }
}

#FILTER#
filter {
    grok {
      match => { "message" => "%{SYSLOGTIMESTAMP:syslog_timestamp} %{SYSLOGHOST:syslog_hostname} %{WORD:process}(?:\[%{POSINT:syslog_pid}\])?: %{GREEDYDATA:syslog_message}" }
    }

    date {
      match => ["timestamp", "MMM d HH:mm:ss", "MMM dd HH:mm:ss"]
    }
}

#OUTPUT#
output {
 elasticsearch {
  hosts => localhost
  index => "indextest"
       }
stdout {
    codec => rubydebug
       }
}
```

---

<div class="post-metadata">

**Author:** ![Nat](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/nat/32/40287_2.png) [@Nat](https://discuss.elastic.co/u/Nat)\
**Post date:** [January 28, 2019, 11:33am UTC](https://discuss.elastic.co/t/logstash-sending-all-data-to-elasticsearch-via-wrong-pipeline/165799/8 "2019-01-28T11:33:52Z")

</div>

Hello, I used your configuration for the pipeline. Though i have two different metricbeats in two different servers. So in the analyzer.conf I have 2 beats input listening to different port. I tried to send output to different pipelines based on port but it doesn't work. Is it even possible to use  
if [port] == "5045" { do something} . I am aiming to create different indeces based on which server the logs come from. I also added tags to each beat input to send to the pipeline based on tag but still doesn't work. i can't see the new indeces in kibana.

---

<div class="post-metadata">

**Author:** ![Wellguys](https://avatars.discourse-cdn.com/v4/letter/w/8e7dd6/32.png) [@Wellguys](https://discuss.elastic.co/u/Wellguys)\
**Post date:** [January 28, 2019, 1:17pm UTC](https://discuss.elastic.co/t/logstash-sending-all-data-to-elasticsearch-via-wrong-pipeline/165799/9 "2019-01-28T13:17:44Z")

</div>

If you want to create different indexes depending on the server you should use something like if [beathostname]. From what I understand, things in "[word]" are fields in logs so It won't work with port number. If you want to use two different ports you only need two pipelines, with a different input port in both.

---

<div class="post-metadata">

**Author:** ![Nat](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/nat/32/40287_2.png) [@Nat](https://discuss.elastic.co/u/Nat)\
**Post date:** [January 28, 2019, 2:26pm UTC](https://discuss.elastic.co/t/logstash-sending-all-data-to-elasticsearch-via-wrong-pipeline/165799/10 "2019-01-28T14:26:34Z")

</div>

It doesn't work with the hostname. But I think the problem is that both servers write to the same registry. I will fix that and see if indexing is working.

---

<div class="post-metadata">

**Author:** ![Nat](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/nat/32/40287_2.png) [@Nat](https://discuss.elastic.co/u/Nat)\
**Post date:** [January 28, 2019, 3:59pm UTC](https://discuss.elastic.co/t/logstash-sending-all-data-to-elasticsearch-via-wrong-pipeline/165799/11 "2019-01-28T15:59:52Z")

</div>

It works now . Each server writes to different registry and I can see the different indeces in kibana. Thanks Wellguys.

---

<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:** [February 25, 2019, 3:59pm UTC](https://discuss.elastic.co/t/logstash-sending-all-data-to-elasticsearch-via-wrong-pipeline/165799/12 "2019-02-25T15:59:55Z")

</div>

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