# Multiple filebeat inputs and multiple logstash pipelines

**URL:** <https://discuss.elastic.co/t/multiple-filebeat-inputs-and-multiple-logstash-pipelines/244626>\
**Category:** Logstash\
**Created:** [August 11, 2020, 7:58pm UTC](https://discuss.elastic.co/t/multiple-filebeat-inputs-and-multiple-logstash-pipelines/244626 "2020-08-11T19:58:35Z")\
**Posts on this page:** 6\
**Page:** 1

<div class="post-metadata">

**Author:** ![nino](https://avatars.discourse-cdn.com/v4/letter/n/b9bd4f/32.png) [@nino](https://discuss.elastic.co/u/nino)\
**Post date:** [August 11, 2020, 7:58pm UTC](https://discuss.elastic.co/t/multiple-filebeat-inputs-and-multiple-logstash-pipelines/244626/1 "2020-08-11T19:58:35Z")

</div>

Dear all,

this is my scenario:

one directory with two types of files that i want to proccess with one pipeline each. File types are identified by his name. I am very new to pipeline logstash, i usually go with a single logstash configuration but things are getting complex and i would like to use different pipelines for each type of file to separate logic and a better maintenance

```
filebeat.inputs:

- type: log
  enabled: true
  paths: C:\files\*apache-log*.txt
  tags: ["type1"]

- type: log
  enabled: true
  paths: C:\files\*ngnix-log*.txt
  tags: ["type2"]

```

How can apply one pipeline base on the file type using multiple pipelines mode?

```
- pipeline.id: pipeline-for-type1-files
  path.config: "/etc/path/to/type1-pipeline.config"
- pipeline.id: pipeline-for-type2-files
  path.config: "/etc/different/path/type2-pipeline.config"

```

If logstash recieves type1 files apply pipeline-for-type1-files and if recieves type2 apply pipeline-for-type2-files. Could you give some sample code of how to handle this?

Best regards

---

<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:** [August 11, 2020, 9:12pm UTC](https://discuss.elastic.co/t/multiple-filebeat-inputs-and-multiple-logstash-pipelines/244626/2 "2020-08-11T21:12:56Z")

</div>

Look at the examples for the [distributor](https://www.elastic.co/guide/en/logstash/current/pipeline-to-pipeline.html#distributor-pattern) pattern in the pipeline-to-pipeline communication documentation.

---

<div class="post-metadata">

**Author:** ![nino](https://avatars.discourse-cdn.com/v4/letter/n/b9bd4f/32.png) [@nino](https://discuss.elastic.co/u/nino)\
**Post date:** [August 11, 2020, 10:37pm UTC](https://discuss.elastic.co/t/multiple-filebeat-inputs-and-multiple-logstash-pipelines/244626/3 "2020-08-11T22:37:20Z")

</div>

Thank you so much badger. Just to simplify a little and while i learn how distributor pattern works, should this simple configuration could work?.

If i understand well how pipeline config files work, if i filter by tag with and if statement in each pipeline config should filter and apply this pipeline only in the events from the file tagged as "type1". Am i correct Badger?

type1-pipeline.config

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

filter {
  if [tags] == "type1" {

  }
}

```

type2-pipeline.config

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

    filter {
      if [tags] == "type2" {

      }
    }

```

Thanks again

---

<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:** [August 11, 2020, 10:46pm UTC](https://discuss.elastic.co/t/multiple-filebeat-inputs-and-multiple-logstash-pipelines/244626/4 "2020-08-11T22:46:16Z")

</div>

Not quite. Go back and read the documentation. They use config.string rather than config.path, but you could do the same with config.path

```
- pipeline.id: distributor
  config.string: |
    input { beats { port => 5044 } }
    output {
        if "type1" in [tags] {
          pipeline { send_to => type1 }
        } else if "type2" in [tags] {
          pipeline { send_to => type2 }
        } else {
          pipeline { send_to => fallback }
        }
    }
- pipeline.id: type1
  config.string: |
    input { pipeline { address => type1 } }
    filter {
       # type1 filter statements here...
    }
    output {
      # type1 output here...
    }
- pipeline.id: type2
  config.string: |
    input { pipeline { address => type2 } }
    filter {
       # type2 filter statements here...
    }
    output {
      # type2 output here...
    }
- pipeline.id: fallback
  config.string: |
    input { pipeline { address => fallback } }
    filter {
       # fallback filter statements here, if any
    }
    output {
      # fallback output here...
    }

```

Having another pipeline for events that are neither type1 nor type2 is not mandatory, you could just make 'pipeline { send\_to =\> type2 }' the 'else' option.

---

<div class="post-metadata">

**Author:** ![nino](https://avatars.discourse-cdn.com/v4/letter/n/b9bd4f/32.png) [@nino](https://discuss.elastic.co/u/nino)\
**Post date:** [August 12, 2020, 7:30am UTC](https://discuss.elastic.co/t/multiple-filebeat-inputs-and-multiple-logstash-pipelines/244626/5 "2020-08-12T07:30:23Z")

</div>

Thank you so much @Badger!

---

<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 9, 2020, 7:30am UTC](https://discuss.elastic.co/t/multiple-filebeat-inputs-and-multiple-logstash-pipelines/244626/6 "2020-09-09T07:30:36Z")

</div>

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