# Retrofitting Filebeat into my ELK stack

**URL:** <https://discuss.elastic.co/t/retrofitting-filebeat-into-my-elk-stack/203282>\
**Category:** Beats\
**Tags:** filebeat\
**Created:** [October 11, 2019, 6:55pm UTC](https://discuss.elastic.co/t/retrofitting-filebeat-into-my-elk-stack/203282 "2019-10-11T18:55:03Z")\
**Posts on this page:** 10\
**Page:** 1

<div class="post-metadata">

**Author:** ![mhare](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/mhare/32/52213_2.png) [@mhare](https://discuss.elastic.co/u/mhare)\
**Post date:** [October 11, 2019, 6:55pm UTC](https://discuss.elastic.co/t/retrofitting-filebeat-into-my-elk-stack/203282/1 "2019-10-11T18:55:04Z")

</div>

I am ingesting 8 different CSV file schemas using 8 logstash pipelines and reading from a Windows file share. IT has been recommending to not read across the file share but use Filebeat.  
Not finding just a whole bunch of info in this area.  
Here are 2 of the 'inputs' section:

# --------------------------------------------------------------

# MemoryLeak Input

- type: log  
enabled: false  
paths:
  - C:\MemoryLeakTest\v03\*-memleak\*.log  
tags: ["memleak"]  
fields: {log\_type: memleak}

# --------------------------------------------------------------

# AM-300 TaskStat Input

- type: log  
enabled: false  
paths:
  - C:\MemoryLeakTest\v03\*\_am-300-taskstat\*.log  
tags: ["taskstat-am-300"]  
fields: {log\_type: taskstat-am-300}

The output is:  
output.logstash:

# The Logstash hosts

hosts: [":5044"]

Now, how the heck do I get this to work with my 8 pipeline files?

I put this at the top of one of the pipeline files:

input {  
beats {  
host =\> "0.0.0.0"  
port =\> "5044"  
client\_inactivity\_timeout =\> 180  
}  
}

My question is, does this look right so far?  
And then, how do I split the data across the 8 filters in the 8 pipeline files?

Thanks, Mike

---

<div class="post-metadata">

**Author:** ![rugenl](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/rugenl/32/12887_2.png) [@rugenl](https://discuss.elastic.co/u/rugenl)\
**Post date:** [October 11, 2019, 7:53pm UTC](https://discuss.elastic.co/t/retrofitting-filebeat-into-my-elk-stack/203282/2 "2019-10-11T19:53:19Z")

</div>

You could use [pipeline to pipeline communication](https://www.elastic.co/guide/en/logstash/current/pipeline-to-pipeline.html)

You would be using a "distributor". The if logic in the output section could reference the [fields][log\_type] value you have set in filebeat.

---

<div class="post-metadata">

**Author:** ![mhare](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/mhare/32/52213_2.png) [@mhare](https://discuss.elastic.co/u/mhare)\
**Post date:** [October 22, 2019, 6:03pm UTC](https://discuss.elastic.co/t/retrofitting-filebeat-into-my-elk-stack/203282/3 "2019-10-22T18:03:20Z")

</div>

OK, the Distributor pattern looks like what I need.  
In the description in your link, they have a set of decision points like:

```auto
output {
        if [type] == apache {
          pipeline { send_to => weblogs }
        } else if [type] == system {
          pipeline { send_to => syslog }
        } else {
          pipeline { send_to => fallback }
        }

```

I am assuming I need to set [type] in my pipeline. My question now is, how?  
In my example I posted above, I have:

```auto
tags: ["memleak"]
fields: {log_type: memleak}

```

I am guessing 'tag' is not needed and I can use a decision point as:

```auto
if [log_type] == memleak {
(do pipeline connection stuff here)
}

```

Nope.. didn't work. the 'fields' in my filbeat.yml is invalid.  
So my question remains, how do I set the 'type' value for the decision point?

I think I'm getting close, still don't understand how to set the [type] value to an arbitrary value

For filebeat.yml, I have:

```auto
- type: log
  enabled: true
  paths:
    - C:\Users\xyzzy\CommonLogs\*-memleak*.log
  fields:
    "log_type": "memleak"

```

My decision point in pipelines.yml is:

```auto
- pipeline.id: beats-server
  config.string: |
    input { beats { port => 5044 } }
    output { 
      if ["log_type"] == "memleak" {
        pipeline { send_to => memleakpl }
      }
    }

```

If I remove the decision point and just call the 'send\_to' it works. So i'm just trying to get the 'if' statement to work

---

<div class="post-metadata">

**Author:** ![mhare](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/mhare/32/52213_2.png) [@mhare](https://discuss.elastic.co/u/mhare)\
**Post date:** [October 23, 2019, 2:56pm UTC](https://discuss.elastic.co/t/retrofitting-filebeat-into-my-elk-stack/203282/4 "2019-10-23T14:56:41Z")

</div>

From what I have read, in my filebeat yaml I can either:

```auto
tags: ["memleak"]

```

- or -

```auto
fields:
   log_type: "memleak"

```

In my pipelines yaml I need to evaluate one of these and make a decision

```auto
output {
   if <something good goes here> {
      pipeline { send_to => memleakpl }
   }
}

```

I cannot find how to link either a tag or a field to the part of the conditional  
Can you help me see the light?

---

<div class="post-metadata">

**Author:** ![rugenl](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/rugenl/32/12887_2.png) [@rugenl](https://discuss.elastic.co/u/rugenl)\
**Post date:** [October 23, 2019, 3:01pm UTC](https://discuss.elastic.co/t/retrofitting-filebeat-into-my-elk-stack/203282/5 "2019-10-23T15:01:15Z")

</div>

Logstash syntax is:

```
If [fields][log_type] == "memleak" {
     do something
}

```

I think you are close 🙂

---

<div class="post-metadata">

**Author:** ![mhare](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/mhare/32/52213_2.png) [@mhare](https://discuss.elastic.co/u/mhare)\
**Post date:** [October 23, 2019, 4:21pm UTC](https://discuss.elastic.co/t/retrofitting-filebeat-into-my-elk-stack/203282/6 "2019-10-23T16:21:41Z")

</div>

I will give that a try.  
What I just completed was in filebeat.yml:

```auto
- type: log
  enabled: true
  tags: ["memleak"]
  paths:
    - C:\Users\xyzzy\CommonLogs\*-memleak*.log

- type: log
  enabled: true
  tags: ["rmc3_taskstat"]
  paths:
    - C:\Users\xyzzy\CommonLogs\*_rmc3-taskstat*.log

```

and in pipelines.yml:

```auto
- pipeline.id: beats-server
  config.string: |
    input { beats { port => 5044 } }
    output { 
      if "memleak" in [tags] {
        pipeline { send_to => memleakpl }
      } else if "rmc3_taskstat" in [tags] {
        pipeline { send_to => rmc3_taskstatpl }
      }
    }

```

Which seems to be working.  
I'll be adding another 6 logs or so in a similar fashion. Do you see a problem with this approach?

---

<div class="post-metadata">

**Author:** ![mhare](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/mhare/32/52213_2.png) [@mhare](https://discuss.elastic.co/u/mhare)\
**Post date:** [October 23, 2019, 5:02pm UTC](https://discuss.elastic.co/t/retrofitting-filebeat-into-my-elk-stack/203282/7 "2019-10-23T17:02:45Z")

</div>

Len, your solution also works.  
Is one better than the other?

---

<div class="post-metadata">

**Author:** ![rugenl](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/rugenl/32/12887_2.png) [@rugenl](https://discuss.elastic.co/u/rugenl)\
**Post date:** [October 24, 2019, 4:14pm UTC](https://discuss.elastic.co/t/retrofitting-filebeat-into-my-elk-stack/203282/8 "2019-10-24T16:14:40Z")

</div>

Fields vs tags? I'd say whichever is understandable, logical and maintainable in your environment. I doubt there is a measurable performance difference.

I guess the old coding technique of putting the most frequent if condition first would apply either way.

---

<div class="post-metadata">

**Author:** ![mhare](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/mhare/32/52213_2.png) [@mhare](https://discuss.elastic.co/u/mhare)\
**Post date:** [October 24, 2019, 4:47pm UTC](https://discuss.elastic.co/t/retrofitting-filebeat-into-my-elk-stack/203282/9 "2019-10-24T16:47:22Z")

</div>

Thanks!

---

<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:** [November 21, 2019, 4:47pm UTC](https://discuss.elastic.co/t/retrofitting-filebeat-into-my-elk-stack/203282/10 "2019-11-21T16:47:33Z")

</div>

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