如何将 Filebeat 限制为仅在日志包含特定字段时将日志传送到 ELK

Question

我正在尝试使用 Filebeat 从 Kubernetes 节点收集日志，并且只有在日志来自特定 Kubernetes 命名空间时才将它们发送到 ELK。

到目前为止，我发现您可以定义我认为可以做到这一点的处理器。但是，无论我做什么，我都无法约束运送的日志。这看起来对吗？

嗯，这看起来对吗？

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']

尽管有这种配置，我仍然从所有命名空间获取日志。

Filebeat 在 Kubernetes 上作为 DaemonSet 运行。下面是扩展日志条目的示例：https://i.stack.imgur.com/HWMqF.png

Answer 1

您有数字选项来执行此操作：

1. 按文件节拍筛选数据

processors:
 - drop_event:
     when:
        contains:
           source: "field"

2. 在 elasticsearch 中使用摄取管道：

output.elasticsearch:
  hosts: ["localhost:9200"]
  pipeline: my_pipeline_id

然后将事件测试到管道中：

{
  "drop": {
    "if" : "ctx['field'] == null "
  }
}

3. 使用日志仓库的跌落过滤器：

filter {
  if ![field] {
    drop { }
  }
}

Answer 2

最后，我通过将拖放处理器从配置文件移动到输入配置文件来解决此问题。