如何配置Fluentd将EKS应用Pod日志采集至AWS CloudWatch

仅能采集kube-system命名空间日志是3个部署/配置错误导致的：

• Fluentd以DaemonSet形式运行时默认部署在kube-system命名空间，你将采集配置ConfigMap创建到了业务命名空间stellacenter-dev，DaemonSet无法挂载读取这份自定义配置，实际生效的还是kube-system下默认仅采集系统组件日志的旧配置
• Fluentd绑定的服务账号RBAC权限不足，kubernetes_metadata过滤器无法拉取非kube-system命名空间下的业务Pod元数据，导致业务日志匹配失败被丢弃
• 原有容器日志流的字段提取规则不完整，未对齐kubectl logs的输出逻辑，日志流命名无业务维度标识，即使采集到也难以检索

1. 迁移并更新ConfigMap到Fluentd所在命名空间 #

先删除stellacenter-dev下创建的错误ConfigMap，在kube-system命名空间下创建调整后的配置，核心修改containers.conf段的日志匹配、字段提取规则：

apiVersion: v1
data:
  containers.conf: |
    <source>
      @type tail
      @id in_tail_container_logs
      @label @containers
      path /var/log/containers/*.log
      pos_file /var/log/fluentd-containers.log.pos
      tag kubernetes.*
      read_from_head true
      <parse>
        @type json
        time_format %Y-%m-%dT%H:%M:%S.%NZ
      </parse>
    </source>

    <label @containers>
      <filter kubernetes.**>
        @type kubernetes_metadata
        @id filter_kube_metadata
        <!-- 不需要采集kube-system日志就保留下面3行，需要的话直接删除即可 -->
        <exclude>
          key $.kubernetes.namespace_name
          pattern ^kube-system$
        </exclude>
      </filter>

      <filter kubernetes.**>
        @type record_transformer
        @id filter_containers_stream_transformer
        enable_ruby true
        <record>
          # 日志流按「命名空间/Pod名/容器名」命名，和kubectl查询到的资源完全对齐
          stream_name "${record.dig('kubernetes','namespace_name')}/${record.dig('kubernetes','pod_name')}/${record.dig('kubernetes','container_name')}"
          # 直接提取原始日志内容，和kubectl logs输出完全一致
          log_content ${record["log"]}
        </record>
      </filter>

      <match kubernetes.**>
        @type cloudwatch_logs
        @id out_cloudwatch_logs_containers
        region "#{ENV.fetch('AWS_REGION')}"
        log_group_name "/k8s-nest/#{ENV.fetch('AWS_EKS_CLUSTER_NAME')}/application-logs"
        log_stream_name_key stream_name
        remove_log_stream_name_key true
        auto_create_stream true
        <buffer>
          flush_interval 2
          chunk_limit_size 2m
          queued_chunks_limit_size 32
          retry_forever true
        </buffer>
      </match>
    </label>
  fluent.conf: |
    @include containers.conf
    @include systemd.conf

    <match fluent.**>
      @type null
    </match>
  systemd.conf: |
    <source>
      @type systemd
      @id in_systemd_kubelet
      @label @systemd
      filters [{ "_SYSTEMD_UNIT": "kubelet.service" }]
      <entry>
        field_map {"MESSAGE": "message", "_HOSTNAME": "hostname", "_SYSTEMD_UNIT": "systemd_unit"}
        field_map_strict true
      </entry>
      path /run/log/journal
      pos_file /var/log/fluentd-journald-kubelet.pos
      read_from_head true
      tag kubelet.service
    </source>

    <source>
      @type systemd
      @id in_systemd_kubeproxy
      @label @systemd
      filters [{ "_SYSTEMD_UNIT": "kubeproxy.service" }]
      <entry>
        field_map {"MESSAGE": "message", "_HOSTNAME": "hostname", "_SYSTEMD_UNIT": "systemd_unit"}
        field_map_strict true
      </entry>
      path /run/log/journal
      pos_file /var/log/fluentd-journald-kubeproxy.pos
      read_from_head true
      tag kubeproxy.service
    </source>

    <source>
      @type systemd
      @id in_systemd_docker
      @label @systemd
      filters [{ "_SYSTEMD_UNIT": "docker.service" }]
      <entry>
        field_map {"MESSAGE": "message", "_HOSTNAME": "hostname", "_SYSTEMD_UNIT": "systemd_unit"}
        field_map_strict true
      </entry>
      path /run/log/journal
      pos_file /var/log/fluentd-journald-docker.pos
      read_from_head true
      tag docker.service
    </source>

    <label @systemd>
      <filter **>
        @type record_transformer
        @id filter_systemd_stream_transformer
        <record>
          stream_name ${tag}-${record["hostname"]}
        </record>
      </filter>

      <match **>
        @type cloudwatch_logs
        @id out_cloudwatch_logs_systemd
        region "#{ENV.fetch('AWS_REGION')}"
        log_group_name "/k8s-nest/#{ENV.fetch('AWS_EKS_CLUSTER_NAME')}/systemd"
        log_stream_name_key stream_name
        auto_create_stream true
        remove_log_stream_name_key true
        <buffer>
          flush_interval 5
          chunk_limit_size 2m
          queued_chunks_limit_size 32
          retry_forever true
        </buffer>
      </match>
    </label>
kind: ConfigMap
metadata:
  labels:
    k8s-app: fluentd-cloudwatch
  name: fluentd-config
  namespace: kube-system

2. 补全Fluentd服务账号的RBAC权限 #

确保Fluentd绑定的ClusterRole有全命名空间的Pod、元数据读取权限，应用以下配置：

apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
  name: fluentd-cloudwatch-role
rules:
- apiGroups: [""]
  resources: ["pods", "namespaces"]
  verbs: ["get", "list", "watch"]
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRoleBinding
metadata:
  name: fluentd-cloudwatch-binding
subjects:
- kind: ServiceAccount
  name: fluentd-cloudwatch
  namespace: kube-system
roleRef:
  kind: ClusterRole
  name: fluentd-cloudwatch-role
  apiGroup: rbac.authorization.k8s.io

3. 重启Fluentd DaemonSet生效配置 #

执行命令滚动重启所有Fluentd Pod：
kubectl rollout restart daemonset fluentd-cloudwatch -n kube-system

• 等所有Fluentd Pod启动完成后，查看日志确认无权限类报错：kubectl logs -n kube-system <fluentd-pod-name>
• 进入CloudWatch控制台找到名为/k8s-nest/<你的集群名>/application-logs的日志组，下方日志流按「业务命名空间/Pod名/容器名」分层，点入后log_content字段的内容和执行kubectl logs -f <pod名>输出的内容完全一致，延迟在2秒左右。

内容的提问来源于stack exchange，提问作者Cyril