CODEWITHSUNDEEP
elasticsearchkuberneteselastic-stackfilebeat
TJ Zimmerman
TJ Zimmerman

Reputation: 3484

How to constrain Filebeat to only ship logs to ELK if they contain a specific field?

I’m trying to collect logs from Kubernetes nodes using Filebeat and ONLY ship them to ELK IF the logs originate from a specific Kubernetes Namespace.

So far I’ve discovered that you can define Processors which I think accomplish this. However, no matter what I do I can not get the shipped logs to be constrained. Does this look right?

Hm, does this look correct then?

filebeat.config:
  inputs:
    path: ${path.config}/inputs.d/*.yml
    reload.enabled: true
    reload.period: 10s
    when.contains:
      kubernetes.namespace: "NAMESPACE"
  modules:
    path: ${path.config}/modules.d/*.yml
    reload.enabled: false
  processors:
    - add_kubernetes_metadata:
      namespace: "NAMESPACE"
xpack.monitoring.enabled: true
output.elasticsearch:
  hosts: ['elasticsearch:9200']

Despite this configuration I still get logs from all of the namespaces.

Filebeat is running as a DaemonSet on Kubernetes. Here is an example of an expanded log entry: https://i.sstatic.net/HWMqF.png

Upvotes: 0

Views: 3046

Answers (2)

TJ Zimmerman
TJ Zimmerman

Reputation: 3484

In the end, I resolved this by moving the drop processor to the input configuration file from the configuration file.

Upvotes: 0

ozlevka
ozlevka

Reputation: 2146

You have number options to do it:

  1. Filter data by filebeat
processors:
 - drop_event:
     when:
        contains:
           source: "field"
  1. Use ingest pipeline into elasticsearch:
output.elasticsearch:
  hosts: ["localhost:9200"]
  pipeline: my_pipeline_id

And then test events into pipeline:

{
  "drop": {
    "if" : "ctx['field'] == null "
  }
}
  1. Use drop filter of logstash:
filter {
  if ![field] {
    drop { }
  }
}

Upvotes: 2

Related Questions