Configuring filebeat logstash plugin to transform one log, but not another

Elastic Stack Logstash
1

I have hosts that have log files in different formats, Vision, and Triad. They are being collected by Filebeat and shipped to Logstash. I have a Logstash config file setup for Beats as shown in the docs. At this point its pretty much a pass through (See below). I want to be able to transform the Triad logs, but leave the Vision logs alone.

So I am a little confused about how tell Logstash to differentiate between the two types of logs that it will be seeing. Its not clear to me from the Beats input plugin docs on how to do that in the input section.

input {
  beats {
    port => 5044
  }
}

filter {
} 

output {
  elasticsearch {
    hosts => "localhost:9200"
    manage_template => false
    index => "%{[@metadata][beat]}-%{[@metadata][version]}-%{+YYYY.MM.dd}"
 }
}
2

This is my Filebeat configuration for the inputs

filebeat.inputs:

# Each - is an input. Most options can be set at the input level, so
# you can use different inputs for various configurations.
# Below are the input specific configurations.

- type: log
  enabled: true
  paths:
     - /usr/cbridge/msg/LaunchPad/*.log
     - /usr/cbridge/msg/Manager/*.log
     - /usr/cbridge/msg/Reporting/*.log
     - /usr/cbridge/msg/WorkOrderPrinting/*.log
  # Handle Java exceptions and put them on one line
  multiline.pattern: '^[[:space:]]+(at|\.{3})\b|^Caused by:'  
  multiline.negate: true
  multiline.match: after

- type: log
  enabled: true
  paths:
     - /usr/cbridge/msg/current.msg
  # Exlude the line of dashes TODO this seems to get ignored when multiline is working. Probably need to strip it in logstash
  exclude_lines: ['^-+$']
  # Setup the pattern to harvest the multiline
  multiline.pattern: '^[A-Z]+: '
  multiline.negate: true
  multiline.match: after    filebeat.inputs:

# Each - is an input. Most options can be set at the input level, so
# you can use different inputs for various configurations.
# Below are the input specific configurations.

- type: log
  enabled: true
  paths:
     - /usr/cbridge/msg/LaunchPad/*.log
     - /usr/cbridge/msg/Manager/*.log
     - /usr/cbridge/msg/Reporting/*.log
     - /usr/cbridge/msg/WorkOrderPrinting/*.log
  # Handle Java exceptions and put them on one line
  multiline.pattern: '^[[:space:]]+(at|\.{3})\b|^Caused by:'  
  multiline.negate: true
  multiline.match: after

- type: log
  enabled: true
  paths:
     - /usr/cbridge/msg/current.msg
  # Exlude the line of dashes TODO this seems to get ignored when multiline is working. Probably need to strip it in logstash
  exclude_lines: ['^-+$']
  # Setup the pattern to harvest the multiline
  multiline.pattern: '^[A-Z]+: '
  multiline.negate: true
  multiline.match: after
3

I'm pretty new to Elastic so I could be way off base here but what you might be looking to do is filter the data. I'd take a look at, https://www.elastic.co/guide/en/logstash/6.5/config-examples.html and https://www.elastic.co/guide/en/logstash/6.5/filter-plugins.html. Again the experts know a lot more about this than me but hopefully I'm pointing you in the right direction. Sorry if this doesn't help.

This may be a useful link as well, https://www.elastic.co/guide/en/logstash/6.5/transformation.html.

4

Thanks for the response Ryan, I appreciate it. I think the basic issue is that I have one pipeline definition "vision-logs-pipeline.conf". It has one input section to listen for Beats. The Beat sends both log types to Logstash, and they appear at the input section. Once I have filtered I dont know how to send the filtered data to the correct output.

Vision logs = A
Triad logs = B

Filebeat(A, B) -> LogstashInput() -> Filter( A -> output1, B-> output2) -> Output1(A) -> ElasticIndex1
                                                                        -> Output2(B) -> ElasticIndex2

Or something like that anyway.

1 Like
5

I may be misunderstanding you but is this what you're looking to do:

6

This is the solution I came up with. It appears to be working. I have no clue whether its efficient or not.

input {
  beats {
    port => 5044
  }
}

output {
  if [source] == '/usr/cbridge/msg/current.msg' {
    elasticsearch {
      hosts => "localhost:9200"
      manage_template => false
      index => "triad-%{[@metadata][beat]}-%{[@metadata][version]}-%{+YYYY.MM.dd}"
    }
  } else {
    elasticsearch {
      hosts => "localhost:9200"
      manage_template => false
      index => "%{[@metadata][beat]}-%{[@metadata][version]}-%{+YYYY.MM.dd}"
    }
  }
}
7

Glad to hear its working, hopefully someone a little more advanced than me can help you streamline it if it even needs to be. Your many levels above where I'm at, good for you!!!

8

Haha, well after all that, I sort of realized that I probably dont want to split it into separate indexes. Ah well at least I learned something interesting.

9

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