# Logstash ingest pipeline - Data from one pipeline going to another

**URL:** <https://discuss.elastic.co/t/logstash-ingest-pipeline-data-from-one-pipeline-going-to-another/289131>\
**Category:** Logstash\
**Tags:** ingest-pipeline\
**Created:** [November 14, 2021, 9:08am UTC](https://discuss.elastic.co/t/logstash-ingest-pipeline-data-from-one-pipeline-going-to-another/289131 "2021-11-14T09:08:06Z")\
**Posts on this page:** 10\
**Page:** 1

<div class="post-metadata">

**Author:** ![parthmaniar](https://avatars.discourse-cdn.com/v4/letter/p/71e660/32.png) [@parthmaniar](https://discuss.elastic.co/u/parthmaniar)\
**Post date:** [November 14, 2021, 9:08am UTC](https://discuss.elastic.co/t/logstash-ingest-pipeline-data-from-one-pipeline-going-to-another/289131/1 "2021-11-14T09:08:06Z")

</div>

Hello,

I hope my message finds the members of the community and their loved ones safe and healthy.

I am running logstash (7.15.2) on a Raspberry Pi 4B running Ubuntu 20.04.3 LTS.

I have seven pipelines, all **listening on different ports** defined in pipelines.yml.

I have data going on pipeline into an index of another pipeline & also in its original pipeline/index. **In both cases, the source is unique filebeat instances running on the same host.**

Following are the details:

**1. Source and Pipeline details:**

**Source** : Filebeat on the remote host defined in and **not** installed as a service & running through /etc/rc.local. It reads a particular **.log file from /var/log**

**Pipeline (destination):**

**Port: 5055**  
Configuration:

```auto
input {
  beats {
    port => 5055
    type => "logs"
  }
}

filter {
 
REMOVED
}

output {
  elasticsearch {
   hosts => ["IP REDACTED"]
   index => "cowrie-firewall-logstash-%{+yyyy.MM.dd}"
   ssl => true
   user => ' **REDACTED**'
   password => ' **REDACTED**'
   cacert => '/etc/logstash/elasticsearch-ca.pem'
   ssl_certificate_verification => true
   ilm_enabled => auto
   ilm_rollover_alias => "cowrie-firewall-logstash"
  }
}

```

**2. Source and Pipeline details:**

**Source** : Filebeat installed as a service reading a **.json file from /srv/cowrie/var/log/cowrie/**

**Pipeline (destination):**

**Port: 5045**  
Configuration:

```auto
input {
       # filebeats
       beats {
             port => 5054
             type => "cowrie"
             #id => "honeypot_ingest"
       }

       # if you don't want to use filebeat: this is the actual live log file to monitor
       #file {
       # path => ["/home/cowrie/cowrie-git/log/cowrie.json"]
       # codec => json
       # type => "cowrie"
       #}
}

REMOVED

output {
    if [type] == "cowrie" {
        elasticsearch {
             hosts => ["IP REDACTED"]
            #data_stream => true #Causes Errors: added after reading this: https://www.elastic.co/guide/en/logstash/current/plugins-outputs-elasticsearch.html#plugins-outputs-elasticsearch-data_streamwhile diagnosing cowrie ingestion causing data duplication.
            index => "cowrie-logstash-%{+yyyy.MM.dd}"
            ssl => true
            user => ' **REDACTED**'
            password => ' **REDACTED**'
            cacert => '/etc/logstash/elasticsearch-ca.pem'
            ssl_certificate_verification => true
            ilm_enabled => auto
            ilm_rollover_alias => "cowrie-logstash"
        }
        #file {
        # path => "/tmp/cowrie-logstash.log"
        # codec => json
        #}
        #stdout {
            #codec => rubydebug
        #}
    }
}

```

Even though logs from the pipeline 1 - cowrie-firewall should are pointed to port 5055 they are also present in pipeline 2 - cowrie-logs which is listening on 5045 port.

How can i remove the duplication?

---

<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:** [November 14, 2021, 12:27pm UTC](https://discuss.elastic.co/t/logstash-ingest-pipeline-data-from-one-pipeline-going-to-another/289131/2 "2021-11-14T12:27:21Z")

</div>

what does pipelines.yml look like?

---

<div class="post-metadata">

**Author:** ![parthmaniar](https://avatars.discourse-cdn.com/v4/letter/p/71e660/32.png) [@parthmaniar](https://discuss.elastic.co/u/parthmaniar)\
**Post date:** [November 14, 2021, 12:45pm UTC](https://discuss.elastic.co/t/logstash-ingest-pipeline-data-from-one-pipeline-going-to-another/289131/3 "2021-11-14T12:45:31Z")

</div>

Hello, thank you for your reply. Here is the pipelines.yml

```auto
# This file is where you define your pipelines. You can define multiple.
# For more information on multiple pipelines, see the documentation:
# https://www.elastic.co/guide/en/logstash/current/multiple-pipelines.html
#09-06-2021: added new pipeline for pihole
#11-10-2021: Added mew pipeline for cowrie_firewall logs.

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

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

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

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

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

#- pipeline.id: vmware_ingest
# path.config: "/etc/logstash/conf.d/vmware_vsphere.conf"

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

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

```

---

<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:** [November 14, 2021, 1:56pm UTC](https://discuss.elastic.co/t/logstash-ingest-pipeline-data-from-one-pipeline-going-to-another/289131/4 "2021-11-14T13:56:28Z")

</div>

that config shouldn’t mix afaik.

if you have verified that each filebeat instance sends to a the correct port, then maybe file an issue in github? i can’t think of any other reason, and haven’t encountered such problem with specific config file

---

<div class="post-metadata">

**Author:** ![leandrojmp](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/leandrojmp/32/107231_2.png) [@leandrojmp](https://discuss.elastic.co/u/leandrojmp)\
**Post date:** [November 14, 2021, 3:16pm UTC](https://discuss.elastic.co/t/logstash-ingest-pipeline-data-from-one-pipeline-going-to-another/289131/5 "2021-11-14T15:16:38Z")

</div>

How are you running logstash? As a service?

---

<div class="post-metadata">

**Author:** ![parthmaniar](https://avatars.discourse-cdn.com/v4/letter/p/71e660/32.png) [@parthmaniar](https://discuss.elastic.co/u/parthmaniar)\
**Post date:** [November 14, 2021, 5:04pm UTC](https://discuss.elastic.co/t/logstash-ingest-pipeline-data-from-one-pipeline-going-to-another/289131/6 "2021-11-14T17:04:54Z")

</div>

Sure let me have a look at filing a bug.

---

<div class="post-metadata">

**Author:** ![parthmaniar](https://avatars.discourse-cdn.com/v4/letter/p/71e660/32.png) [@parthmaniar](https://discuss.elastic.co/u/parthmaniar)\
**Post date:** [November 14, 2021, 5:05pm UTC](https://discuss.elastic.co/t/logstash-ingest-pipeline-data-from-one-pipeline-going-to-another/289131/7 "2021-11-14T17:05:20Z")

</div>

Hello,

Logstash is installed via APT on the Raspberry Pi. It is running a service.

 ![image](https://us1.discourse-cdn.com/elastic/original/3X/2/9/29b8a8749f14ec02c5500eff5ac5f5bb6c273fb9.png)

Here are the two different installs of filebeat (one through APT and one using the compressed file)

 ![image](https://us1.discourse-cdn.com/elastic/original/3X/0/6/0625e220f71be5f9369625abd89cb7cef5f0fb04.png)

They are running on different configurations.

---

<div class="post-metadata">

**Author:** ![parthmaniar](https://avatars.discourse-cdn.com/v4/letter/p/71e660/32.png) [@parthmaniar](https://discuss.elastic.co/u/parthmaniar)\
**Post date:** [November 14, 2021, 5:07pm UTC](https://discuss.elastic.co/t/logstash-ingest-pipeline-data-from-one-pipeline-going-to-another/289131/8 "2021-11-14T17:07:21Z")

</div>

Yes I just double checked the same and it clearly shows the correct port.

Assuming the port is mixed up, it would be odd that the index created for firewall logs (collected via port 5055) is having new events, right?

---

<div class="post-metadata">

**Author:** ![parthmaniar](https://avatars.discourse-cdn.com/v4/letter/p/71e660/32.png) [@parthmaniar](https://discuss.elastic.co/u/parthmaniar)\
**Post date:** [December 12, 2021, 9:35am UTC](https://discuss.elastic.co/t/logstash-ingest-pipeline-data-from-one-pipeline-going-to-another/289131/9 "2021-12-12T09:35:42Z")

</div>

Issue created:

> <https://github.com/logstash-plugins/logstash-input-beats/issues/437>
>
> \<!--
> GitHub is reserved for bug reports and feature requests; it is not the pla…ce
> for general questions. If you have a question or an unconfirmed bug , please
> visit the \[forums\](https://discuss.elastic.co/c/logstash). Please also
> check your OS is \[supported\](https://www.elastic.co/support/matrix#show\_os).
> If it is not, the issue is likely to be closed.
> 
> Logstash is located in a different organization: \[logstash\](https://github.com/elastic/logstash). For bugs specific to Logstash and not related to Logstash Plugins, please open it in the respective Logstash repository.
> 
> For security vulnerabilities please only send reports to security@elastic.co.
> See https://www.elastic.co/community/security for more information.
> 
> Please fill in the following details to help us reproduce the bug:
> \--\>
> 
> \*\*Logstash information\*\*:
> 
> Please include the following information:
> 
> 1. Logstash version: 7.16.0 (running on Raspberry Pi 4 Model B Rev 1.1 with AArch64 \[ARM64\] using Ubuntu 20.04 LTS)
> 2. Logstash installation source: APT 
> 3. How is Logstash being run: As a service using systemd 
> 4. How was the Logstash Plugin installed: Default plugin
> 
> \*\*JVM\*\*:
> openjdk version "11.0.13" 2021-10-19
> OpenJDK Runtime Environment Temurin-11.0.13+8 (build 11.0.13+8)
> OpenJDK 64-Bit Server VM Temurin-11.0.13+8 (build 11.0.13+8, mixed mode)
> 
> \*\*OS version\*\*: Ubuntu 20.04 LTS \_(5.4.0-1047-raspi #52-Ubuntu SMP PREEMPT Wed Nov 24 08:16:38 UTC 2021 aarch64 aarch64 aarch64 GNU/Linux)\_
> 
> \*\*Description of the problem including expected versus actual behavior\*\*:
> Expected: Input data should go its respective index
> Error: Data from one pipeline which has a specific index is also going to another index. (two copies being created)
> 
> There are two different instances of Filebeat running on a single host. Both have different installation and configuration mechanisms. 
> 
> \*\*Input is via filebeat:\*\*
> 
> \*\*Installation and persistence:\*\*
> 1. Filebeat installed via APT (running configuration: /etc/filebeat/filebeat.yml) && referred to as cowrie-\*
> 2. Filebeat unzipped and made persistent via /etc/rc.local (running configuration: /home/user/filebeat2/filebeat.yml) && referred to as cowrie-firewall-\*
> 
> \*\*Configuration:\*\*
> 1. Both filebeat instances have their unique configurations. They are configured to send logs to Logstash on \*\*different ports\*\*
> 
> \*\*Logstash\*\*
> 1. Logstash is running on a single host with different pipelines for each ingest. 
> 2. Logs being sent to pipeline "cowrie-\*" on port 5045 are visible in the index of pipeline "cowrie-logstash-\*" (pipeline.id: honeypot\_ingest)
> 3. Logs being sent to pipeline "cowrie-firewall\*" on port 5055 are visible in the index of pipeline "cowrie-logstash-\*" (pipeline.id: cowrie\_firewall\_ingest)
> 3. Output section of the configuration for each configuration:
> 
> A. \*\*cowrie-\*\*\*
> 
> \`\`\`
> input {
> # filebeats
> beats {
> port =\> 5054
> type =\> "cowrie"
> #id =\> "honeypot\_ingest"
> }
> \`\`\`
> \`\`\`
> filter {
> if \[type\] == "cowrie" {
> json {
> \`\`\`
> 
> \`\`\`
> output {
> if \[type\] == "cowrie" {
> elasticsearch {
> hosts =\> \["REDACTED","REDACTED"\]
> #data\_stream =\> true #Causes Errors: added after reading this: https://www.elastic.co/guide/en/logstash/current/plugins-outputs-elasticsearch.html#plugins-outputs-elasticsearch-data\_streamwhile diagnosing cowrie ingestion causing data duplication.
> index =\> "cowrie-logstash-%{+yyyy.MM.dd}"
> ssl =\> true
> user =\> 'REDACTED'
> password =\> 'REDACTED'
> cacert =\> '/etc/logstash/elasticsearch-ca.pem'
> ssl\_certificate\_verification =\> true
> ilm\_enabled =\> auto
> ilm\_rollover\_alias =\> "cowrie-logstash"
> }
> #file {
> # path =\> "/tmp/cowrie-logstash.log"
> # codec =\> json
> #}
> #stdout {
> #codec =\> rubydebug
> #}
> }
> }
> \`\`\`
> 
> \*\*B. Cowrie-Firewall-\*\*\*
> 
> \`\`\`
> input {
> beats {
> port =\> 5055
> type =\> "logs"
> }
> }
> 
> \`\`\`
> \`\`\`
> filter {
> grok {
> patterns\_dir =\> \["/etc/logstash/patterns/"\]
> match=\> \["message", "%{HOSTNAME:hostname}.\*?SRC=%{IPV4:source\_ip} DST=%{IPV4:destination\_ip} LEN=%{DATA:length} TOS=%{DATA:type\_of\_service} PREC=%{DATA:precedence} TTL=%{DATA:ttl} ID=%{DATA:unique\_id} PROTO=%{DATA:protocol} SPT=%{DATA:source\_port} DPT=%{DATA:destination\_port} WINDOW=%{DATA:tcp\_receive\_size} RES=%{DATA:reserved\_bits} %{DATA:tcp\_flag} URGP=%{GREEDYDATA:urgent\_flag}",
> "message","%{HOSTNAME:hostname}.\*?SRC=%{IPV4:source\_ip} DST=%{IPV4:destination\_ip} LEN=%{DATA:length} TOS=%{DATA:type\_of\_service} PREC=%{DATA:precedence} TTL=%{DATA:ttl} ID=%{DATA:unique\_id} %{DATA:tcp\_flag} PROTO=%{DATA:protocol} SPT=%{DATA:source\_port} DPT=%{DATA:destination\_port} WINDOW=%{DATA:tcp\_receive\_size} RES=%{DATA:reserved\_bits} %{DATA:packet\_flag} URGP=%{GREEDYDATA:urgent\_flag}",
> "message","%{HOSTNAME:hostname}.\*?SRC=%{IPV4:source\_ip} DST=%{IPV4:destination\_ip} LEN=%{DATA:length} TOS=%{DATA:type\_of\_service} PREC=%{DATA:precedence} TTL=%{DATA:ttl} ID=%{NUMBER:unique\_id} %{DATA:udp\_flag} PROTO=%{DATA:protocol} SPT=%{DATA:source\_port} DPT=%{DATA:destination\_port} LEN=%{NUMBER:length}",
> "message","%{HOSTNAME:hostname}.\*?SRC=%{IPV4:source\_ip} DST=%{IPV4:destination\_ip} LEN=%{DATA:length} TOS=%{DATA:type\_of\_service} PREC=%{DATA:precedence} TTL=%{DATA:ttl} ID=%{NUMBER:unique\_id} PROTO=%{DATA:protocol} SPT=%{DATA:source\_port} DPT=%{DATA:destination\_port} LEN=%{NUMBER:length}"\]
> }
> 
> geoip {
> source =\> "source\_ip"
> target =\> "geoip"
> database =\> "/opt/logstash/vendor/geoip/GeoLite2-City.mmdb"
> 
> }
> }
> \`\`\`
> \`\`\`
> output {
> elasticsearch {
> hosts =\> \["REDACTED","REDACTED"\]
> index =\> "cowrie-firewall-logstash-%{+yyyy.MM.dd}"
> ssl =\> true
> user =\> 'REDACTED'
> password =\> 'REDACTED'
> cacert =\> '/etc/logstash/elasticsearch-ca.pem'
> ssl\_certificate\_verification =\> true
> ilm\_enabled =\> auto
> ilm\_rollover\_alias =\> "cowrie-firewall-logstash"
> }
> }
> \`\`\`
> 
> 
> \*\*pipelines.yml\*\*
> 
> \`\`\`
> 
> \- pipeline.id: honeypot\_ingest
> path.config: "/etc/logstash/conf.d/cowrie.conf"
> 
> \- pipeline.id: cowrie\_firewall\_ingest
> path.config: "/etc/logstash/conf.d/cowrie\_firewall.conf"
> 
> \`\`\`
> 
> \*\*Steps to reproduce\*\*:
> 
> 1. On the ISP router configured inbound from ANY to port 5000-6000 send to logstash IP port 5000-6000
> 2. Installed two distinct instances of Filebeat - one via APT and second via decompressing folder and using /etc/rc.local
> 3. Verify that both filebeat instances are using their own configurations:
> \`\`\`
> 
> root 510 1 0 01:12 ? 00:00:18 /usr/share/filebeat/bin/filebeat --environment systemd -c /etc/filebeat/filebeat.yml --path.home /usr/share/filebeat --path.config /etc/filebeat --path.data /var/lib/filebeat --path.logs /var/log/filebeat
> root 529 1 0 01:12 ? 00:00:37 /home/user/filebeat2/filebeat -c /home/user/filebeat2/filebeat.yml
> root 9429 9113 0 08:57 pts/1 00:00:00 grep --color=auto filebeat
> \`\`\`
> 
> 4. Verify \*\*input\*\* configuration of each filebeat instance:
> 
> A. \*\*cowrie-logstash-\*\*\* (filepath: /etc/filebeat/filebeat.yml)
> \`\`\`
> # Paths that should be crawled and fetched. Glob based paths.
> paths:
> #- /var/log/\*.log
> - /srv/cowrie/var/log/cowrie/\*.json
> - /home/ubuntu/logs/\*.json
> #- c:\\programdata\\elasticsearch\\logs\\\*
> \`\`\`
> 
> 
> B. \*\*Cowrie-firewall-\*\*\* (filepath: /home/user/filebeat2/filebeat.yml)
> 
> \`\`\`
> # Paths that should be crawled and fetched. Glob based paths.
> paths:
> #- /var/log/\*.log
> - /var/log/dshield.log
> #- /home/ubuntu/logs/\*.json
> #- c:\\programdata\\elasticsearch\\logs\\\*
> \`\`\`
> 
> 4. Verify output configurations for filebeat:
> A. \*\*cowrie-logstash-\*\*\* (filepath: /etc/filebeat/filebeat.yml)
> \`\`\`
> output.logstash:
> # The Logstash hosts
> #hosts: \["localhost:5044"\]
> hosts: \["IP REDACTED:5054"\]
> \`\`\`
> 
> B. \*\*Cowrie-firewall-\*\*\* (filepath: /home/user/filebeat2/filebeat.yml)
> \`\`\`
> output.logstash:
> # The Logstash hosts
> #hosts: \["localhost:5044"\]
> hosts: \["IP REDACTED:5055"\]
> \`\`\`
> 
> \*\*Provide logs (if relevant)\*\*:
> I can email the logs if need be.
> 
> Please refer to the thread: https://discuss.elastic.co/t/logstash-ingest-pipeline-data-from-one-pipeline-going-to-another/289131

---

<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:** [January 9, 2022, 9:35am UTC](https://discuss.elastic.co/t/logstash-ingest-pipeline-data-from-one-pipeline-going-to-another/289131/10 "2022-01-09T09:35:51Z")

</div>

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