Fluentd 根据字段值创建标签

Question

我有一个Kubernetes 集群，我试图在其中聚合节点上的容器日志并将它们发送到MongoDB。但是我需要能够根据它自己的日志记录中的值将日志记录发送到不同的 MongoDB 服务器。

我正在使用fluent-plugin-kubernetes_metadata_filter 插件将来自 Kubernetes 的附加信息附加到日志记录中。其中一个字段是kubernetes_namespace_name。是否可以使用该字段来创建一个标签，我可以使用它来匹配 mongodb 输出插件。

例如。下面我只使用一个输出，但想法是有多个并让流利根据kubernetes_namespace_name字段中的值将日志发送到该mongodb数据库：

<source>
  @type tail
  @label @KUBERNETES
  path /var/log/containers/*.log
  pos_file /var/log/es-containers.log.pos
  time_format %Y-%m-%dT%H:%M:%S
  tag kubernetes.*
  format json
  keep_time_key true
  read_from_head true
</source>

<label @KUBERNETES>
  <filter kubernetes.**>
    @type kubernetes_metadata
    kubernetes_url "#{ENV['K8S_HOST_URL']}"
    bearer_token_file /var/run/secrets/kubernetes.io/serviceaccount/token
    ca_file /var/run/secrets/kubernetes.io/serviceaccount/ca.crt
    include_namespace_id true
  </filter>
  <filter kubernetes.**>
    @type flatten_hash
    separator _
  </filter>

  # < Tag 'kubernetes.namespace.default' is created here somehow >

  <match kubernetes.namespace.default>
    @type mongo
    host "#{ENV['MONGO_HOST']}"
    port "#{ENV['MONGO_PORT']}"
    database "#{ENV['MONGO_DATABASE']}"
    collection "#{ENV['MONGO_COLLECTION']}"
    capped
    capped_size 1024m
    user "#{ENV['MONGO_USER']}"
    password "#{ENV['MONGO_PASSWORD']}"
    time_key time
    flush_interval 10s
  </match>
</label>

Answer 1

您可以使用 Fluentd 的 grep 过滤器来使用消息内容进行过滤，而不是使用标签。您可以在 kubernetes 元数据过滤器之后和数据扁平化器之前添加过滤器。这允许您指定键 kubernetes_namespace_name ，然后根据其中的值进行路由。因为您可能有额外的 MongoDB 输出，使用标签可以帮助分离流程工作流。

文档：https://docs.fluentd.org/v0.12/articles/filter_grep

例子：

<filter kubernetes.**>
  @type grep
  <regexp>
    key kubernetes_namespace_name
    pattern cool
  </regexp>
</filter>

<YOUR MONGO CONFIG HERE>