配置Fluentd采集K8s中Spring Boot应用日志发送至Elasticsearch

你的Logback配置已经用LogstashEncoder输出结构化JSON到容器标准输出，是K8s场景下日志采集最友好的形式，不需要额外做多行切割、正则匹配，按以下步骤配置即可：

采用DaemonSet模式在K8s集群每个节点上部署Fluentd，直接读取节点上的容器日志文件，自动注入Pod元数据后转发到Elasticsearch，不需要对业务应用做任何侵入修改。

Fluentd需要读取节点日志目录、访问K8s API拉取Pod/命名空间元数据，先创建对应的服务账户和权限：

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

创建ConfigMap存Fluentd核心配置，注意需要做两层JSON解析：第一层解析K8s容器日志的外层包装格式，第二层解析LogstashEncoder输出的业务日志内容：

apiVersion: v1
kind: ConfigMap
metadata:
  name: fluentd-config
  namespace: kube-system
data:
  fluent.conf: |
    # 输入源：读取节点上所有容器的日志文件
    <source>
      @type tail
      path /var/log/containers/*.log
      pos_file /var/log/fluentd-containers.log.pos
      tag kubernetes.*
      read_from_head true
      <parse>
        @type json
        time_key time
        time_format %Y-%m-%dT%H:%M:%S.%NZ
      </parse>
    </source>

    # 注入K8s元数据：自动附加Pod名、命名空间、标签等信息
    <filter kubernetes.**>
      @type kubernetes_metadata
      skip_master_url true
    </filter>

    # 过滤：只保留你部署的Spring Boot应用日志，按需修改匹配规则
    <filter kubernetes.**>
      @type grep
      <regexp>
        key $.kubernetes.labels.app
        pattern /spring-boot/ # 匹配你Spring Boot应用的app标签值
      </regexp>
    </filter>

    # 第二层解析：把LogstashEncoder输出的JSON字符串解析成结构化字段
    <filter kubernetes.**>
      @type parser
      key_name log
      reserve_data false
      emit_invalid_record_to_error false
      <parse>
        @type json
        time_key timestamp
        time_format %Y-%m-%dT%H:%M:%S.%NZ
      </parse>
    </filter>

    # 字段规整：打平常用字段，删除冗余元数据
    <filter kubernetes.**>
      @type record_transformer
      enable_ruby true
      <record>
        app_name ${record.dig("kubernetes", "labels", "app")}
        namespace ${record.dig("kubernetes", "namespace_name")}
        pod_name ${record.dig("kubernetes", "pod_name")}
        log_level ${record.dig("level")}
      </record>
      remove_keys kubernetes, docker, stream
    </filter>

    # 输出到Elasticsearch
    <match kubernetes.**>
      @type elasticsearch
      host elasticsearch.logging.svc.cluster.local # 替换为集群内ES服务地址
      port 9200
      index_name springboot-logs-%Y%m%d # 按天滚动生成索引
      logstash_format false
      flush_interval 5s
      # ES开启认证时放开下面配置
      # user elastic
      # password your-es-password
      <buffer>
        @type file
        path /var/log/fluentd-buffer
        flush_thread_count 4
        chunk_limit_size 8M
        queue_limit_length 64
        retry_forever true
        retry_max_interval 30
      </buffer>
    </match>

创建DaemonSet，注意挂载对应节点的日志目录，根据你的容器运行时调整挂载路径：

apiVersion: apps/v1
kind: DaemonSet
metadata:
  name: fluentd
  namespace: kube-system
  labels:
    app: fluentd
spec:
  selector:
    matchLabels:
      app: fluentd
  template:
    metadata:
      labels:
        app: fluentd
    spec:
      serviceAccountName: fluentd
      tolerations:
      - key: node-role.kubernetes.io/master
        effect: NoSchedule
      - key: node-role.kubernetes.io/control-plane
        effect: NoSchedule
      containers:
      - name: fluentd
        # 选择和你ES大版本匹配的镜像，比如ES7用elasticsearch7标签，ES8用elasticsearch8标签
        image: fluent/fluentd-kubernetes-daemonset:v1.16-elasticsearch7
        env:
        - name: FLUENT_ELASTICSEARCH_SED_DISABLE
          value: "true"
        resources:
          limits:
            memory: 512Mi
          requests:
            cpu: 100m
            memory: 200Mi
        volumeMounts:
        - name: varlog
          mountPath: /var/log
        # Docker运行时挂载下面这个目录
        - name: docker-containers
          mountPath: /var/lib/docker/containers
          readOnly: true
        # containerd运行时替换上面的docker-containers挂载为下面的配置
        # - name: containerd-pods
        #   mountPath: /var/log/pods
        #   readOnly: true
        - name: fluentd-config
          mountPath: /fluentd/etc/fluent.conf
          subPath: fluent.conf
      volumes:
      - name: varlog
        hostPath:
          path: /var/log
      - name: docker-containers
        hostPath:
          path: /var/lib/docker/containers
      # - name: containerd-pods
      #   hostPath:
      #     path: /var/log/pods
      - name: fluentd-config
        configMap:
          name: fluentd-config

• 因为你用了LogstashEncoder，异常栈信息会被序列化到stack_trace字段作为单行JSON输出，不需要额外配置Fluentd的multiline多行插件，不会出现日志拆分问题。
• 业务代码里放到MDC的自定义字段（比如traceId、userId）会被LogstashEncoder自动输出到JSON顶层，Fluentd解析后会直接同步到ES，不需要额外配置。
• 建议给ES配置ILM索引生命周期策略，自动删除超过保留期的旧日志，避免磁盘占满。
• 部署完成后可以通过kubectl logs -n kube-system <fluentd-pod名>查看运行日志，确认没有解析错误、ES连接错误；再到ES里检查springboot-logs-*索引是否生成，日志字段是否完整。

内容的提问来源于stack exchange，提问作者Алина Чебакова