Output to different index based on field

Elastic Stack Logstash
1

Hi there,

I have a couple of use-cases where I want to ingest filebeat module data, send it through Logstash, and then put it into different indexes based on what filebeat module that it used.

So for example, the two I have currently are NetFlow and ThreatIntel -

I want to be able to send NetFlow to index-NetFlow and ThreatIntel to index-ThreatIntel in Elastic/Kibana.

Now i've tried a number of different configs, below is one such example:

input {
  beats {
    port => 5044
  }
}
filter {
}
output {
  stdout { codec => rubydebug }
  elasticsearch {
    hosts => ["localhost:9200"]
    index => "index-%{[fields][events.module]}"

This results in just naming the index exactly what is there, it doesn't pull the fields from the message at all. I've tried without the [fields] part as well, you can see my attempts in Kibana here:

image
image863×220 21.4 KB

Can anyone give any pointers on what I'm doing wrong? I managed to pull fields from Metadata but the Metadata fields are the same for Type which is _doc, so close to useless. I then tried to convert a standard field into a metadata field and then pull it, but that didn't appear to work either.

Thanks

2

I don't know the way to access filebeat module name from logstash.

How about configuring the logstash output of filebeat as a workaround? According to the reference, you can set index parameter and it can be accessed in Logstash's output section as %{[@metadata][beat]} .

3

I'm not as familiar with Filebeat configuration, how do I set the index parameter? I guess I set it as if I was ingesting it directly to Elasticsearch but actually not?

4

Tried this, didn't work:

output.logstash:
  # The Logstash hosts
  hosts: ["localhost:5044"]
  index: "filebeat-%{event.module}"
5

I'm not as familiar with Filebeat configuration, how do I set the index parameter?

I recommend to read the reference linked in the previous post carefully.

I guess I set it as if I was ingesting it directly to Elasticsearch but actually not?

I'm not sure about your filebeat setting (you did not share) but it seems you configured the filebeat to output to logstash because indices were created and documents were indexed. Of course you can configure filebeat to output directly to elasticsearch if you want. Please see the elasticsearch part of configurein the output of Filebeat.

6

Hi Tomo,

I'm trying to use Logstash where I can for now, simply because I'm more familiar with it.

I've added some additional replies above, it appears all my dynamic references to fields within my filebeat.yml just don't resolve, they all get passed through as literals and my indexes get named %{[fields.event]} etc.

7

Isn't it just that unresolved dynamic references were treated as literals?

I suppose there are no such fields in Netflow module output.

And also you have to check the format of field reference in logstash configuration. %{[fields][event]} is the way to access the field fields.event.

I really recommend you to read these references carefully.

8

If you are using filebeat modules, the parsing and processing that will create the event object and other fields from the module will be done using a ingest pipeline in Elasticseach, those fields do not exist in Logstash, that is the reason you are getting the literal %{field.name}.

I think that you will need to edit the module configuration and add the fields there, for each module, but I'm not sure as do not use the filebeat modules much.

9

It's certainly a field within the output shown in Kibana:

image

And I tried [fields][event], same scenario.

10

Also, I don't need a field that exists within NetFlow, I need a field that exists within EVERY Filebeat output...not sure if this is even possible.

Maybe I need to go back to the drawing board but essentially how do I take two filebeat modules and parse them into two separate indexes?

11

I think you are right, your document probably has the event.module field already, the issue could be that you are referring to it in a wrong way.

Try the following in your Logstash output:

index => "index-%{[event][module]}"

This is the correct way to make reference to nested fields in logstash, using event.module makes a reference to a field with a literal dot in the name.

I'm not sure that this will work because you are not parsing the message in Logstash.

Filebeat modules work better when you send the data directly to Elasticsearch, putting Logstash in between them can be confusing some times as you need to adjuste some things and since you are not using any filter in your Logstash pipeline, why not send it directly to Elasticsearch from filebeat?

You would be able to use something like this to save in different indices:

output.elasticsearch:
  hosts: ["http://localhost:9200"]
  index: "index-%{[event.module]}-%{+yyyy.MM.dd}"
12

In the end, I gave up trying to use logstash and used this method of achieving what I desired:


output.elasticsearch:
  # Array of hosts to connect to.
  hosts: ["localhost:9200"]
  # Protocol - either `http` (default) or `https`.
  #protocol: "https"
  indices:
    - index: "filebeat-netflow"
      when.equals:
        event.module: "netflow"
    - index: "filebeat-threatintel"
      when.equals:
        event.module: "threatintel"
1 Like
13

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