Kubernetes集群监控-使用ELK实现日志监控和分析

王先森sec

发布于 2023-11-20 08:48:15

5180

发布于 2023-11-20 08:48:15

文章被收录于专栏：王先森王先森

虚拟化运维 Log Kubernetes

Kubernetes集群监控-使用ELK实现日志监控和分析

王先森2023-11-192023-11-19

日志收集架构

日志对于调试问题和监视集群情况也是非常有用的。而且大部分的应用都会有日志记录，对于传统的应用大部分都会写入到本地的日志文件之中。对于容器化应用程序来说则更简单，只需要将日志信息写入到 stdout 和 stderr 即可，容器默认情况下就会把这些日志输出到宿主机上的一个 JSON 文件之中，同样也可以通过 docker logs 或者 kubectl logs 来查看到对应的日志信息。

但是，通常来说容器引擎或运行时提供的功能不足以记录完整的日志信息，比如，如果容器崩溃了、Pod 被驱逐了或者节点挂掉了，仍然也希望访问应用程序的日志。所以，日志应该独立于节点、Pod 或容器的生命周期，这种设计方式被称为 cluster-level-logging，即完全独立于 Kubernetes 系统，需要自己提供单独的日志后端存储、分析和查询工具。

Kubernetes 日志收集

Kubernetes 集群本身不提供日志收集的解决方案，一般来说有主要的 3 种方案来做日志收集：

在每个节点上运行的节点级日志收集代理。
在应用程序的 pod 中，包含专门记录日志的 sidecar 容器。
将日志直接从应用程序中推送到日志记录后端。

使用节点级日志代理

优势：

部署方便，使用DaemonSet类型控制器来部署agent即可
对业务应用的影响最小，没有侵入性

劣势:

只能收集标准和错误输出，对于容器内的文件日志，暂时收集不到

以 sidecar 容器收集日志

上面的图可以看到有一个明显的问题就是采集的日志都是通过输出到容器的 stdout 和 stderr 里面的信息，这些信息会在本地的容器对应目录中保留成 JSON 日志文件，所以直接在节点上运行一个 agent 就可以采集到日志。但是如果应用程序的日志是输出到容器中的某个日志文件的话，这种日志数据显然只通过上面的方案是采集不到的了。

方式一

sidecar 容器将应用程序日志传送到自己的标准输出。

在pod中启动一个sidecar容器，把容器内的日志文件吐到标准输出，由宿主机中的日志收集agent进行采集。

$ cat count-pod.yaml
apiVersion: v1
kind: Pod
metadata:
  name: counter
spec:
  containers:
  - name: count
    image: busybox
    args:
    - /bin/sh
    - -c
    - >
      i=0;
      while true;
      do
        echo "$i: $(date)" >> /var/log/1.log;
        echo "$(date) INFO $i" >> /var/log/2.log;
        i=$((i+1));
        sleep 1;
      done
    volumeMounts:
    - name: varlog
      mountPath: /var/log
  - name: count-log-1
    image: busybox
    args: [/bin/sh, -c, 'tail -n+1 -f /var/log/1.log']
    volumeMounts:
    - name: varlog
      mountPath: /var/log
  - name: count-log-2
    image: busybox
    args: [/bin/sh, -c, 'tail -n+1 -f /var/log/2.log']
    volumeMounts:
    - name: varlog
      mountPath: /var/log
  volumes:
  - name: varlog
    emptyDir: {}
    
$ kubectl apply -f counter-pod.yaml
$ kubectl logs -f counter -c count-log-1

优势：

可以实现容器内部日志收集
对业务应用的侵入性不大

劣势：

每个业务pod都需要做一次改造
增加了一次日志的写入，对磁盘使用率有一定影响

方式二

sidecar 容器运行一个日志代理，配置该日志代理以便从应用容器收集日志。不过这样虽然更加灵活，但是在 sidecar 容器中运行日志采集代理程序会导致大量资源消耗，因为你有多少个要采集的 Pod，就需要运行多少个采集代理程序，另外还无法使用 kubectl logs 命令来访问这些日志，因为它们不受 kubelet 控制。

apiVersion: v1
kind: ConfigMap
metadata:
  name: fluentd-config
data:
  fluentd.conf: |
    <source>
      type tail
      format none
      path /var/log/1.log
      pos_file /var/log/1.log.pos
      tag count.format1
    </source>

    <source>
      type tail
      format none
      path /var/log/2.log
      pos_file /var/log/2.log.pos
      tag count.format2
    </source>

    <match **>
      type google_cloud
    </match>
---
apiVersion: v1
kind: Pod
metadata:
  name: counter
spec:
  containers:
    - name: count
      image: busybox
      args:
        - /bin/sh
        - -c
        - >
          i=0;
          while true;
          do
            echo "$i: $(date)" >> /var/log/1.log;
            echo "$(date) INFO $i" >> /var/log/2.log;
            i=$((i+1));
            sleep 1;
          done
      volumeMounts:
        - name: varlog
          mountPath: /var/log
    - name: count-agent
      image: k8s.gcr.io/fluentd-gcp:1.40
      env:
        - name: FLUENTD_ARGS
          value: -c /etc/fluentd-config/fluentd.conf
      volumeMounts:
        - name: varlog
          mountPath: /var/log
        - name: config-volume
          mountPath: /etc/fluentd-config
  volumes:
    - name: varlog
      emptyDir: {}
    - name: config-volume
      configMap:
        name: fluentd-config

上面的 Pod 创建完成后，容器 count-agent 就会将 count 容器中的日志进行收集然后上传。当然，这只是一个简单的示例，完全可以使用其他的任何日志采集工具来替换 fluentd，比如 logstash、fluent-bit 等等。

优势：不用往宿主机存储日志，本地日志完全可以收集

劣势：每个业务应用额外启动一个日志agent，带来额外的资源损耗

直接从应用程序收集日志

除了上面的几种方案之外，完全可以通过直接在应用程序中去显示的将日志推送到日志后端，但是这种方式需要代码层面的实现，也超出了 Kubernetes 本身的范围。

安装 EFK 集群

参考官网：https://www.elastic.co/guide/en/cloud-on-k8s/current/k8s-deploy-elasticsearch.html

# 安装ECK https://www.elastic.co/guide/en/cloud-on-k8s/current/k8s-deploy-eck.html
kubectl create -f https://download.elastic.co/downloads/eck/2.10.0/crds.yaml
kubectl apply -f 
https://download.elastic.co/downloads/eck/2.10.0/operator.yaml

Localpath本地存储

kind: StorageClass
apiVersion: storage.k8s.io/v1
metadata:
  name: local-storage
  labels:
    app: prometheus
provisioner: kubernetes.io/no-provisioner
volumeBindingMode: WaitForFirstConsumer
---
apiVersion: v1
kind: PersistentVolume
metadata:
  name: elastic-local
  labels:
    app: elastic
spec:
  accessModes:
    - ReadWriteOnce
  capacity:
    storage: 10Gi
  storageClassName: local-storage
  local:
    path: /data/k8s/elastic             # 在节点上创建此目录
  nodeAffinity:                         # 多个node节点不需要配置
    required:
      nodeSelectorTerms:
        - matchExpressions:
            - key: kubernetes.io/hostname
              operator: In
              values:
                - k8s-node1                 # 固定到k8s-node1节点上
  persistentVolumeReclaimPolicy: Retain

部署 Elasticsearch

ElasticsearchIngress

apiVersion: v1
kind: Secret
metadata:
  name: elastic-auth
  namespace: logging
# type: kubernetes.io/basic-auth
stringData:
  username: elastic     # Elasticsearch用户
  password: admin123    # Elasticsearch 用户密码
  roles: superuser,kibana_admin,ingest_admin  # Elasticsearch用户拥有那些角色权限
---
apiVersion: elasticsearch.k8s.elastic.co/v1
kind: Elasticsearch
metadata:
  name: elasticsearch
  namespace: logging
spec:
  version: 8.11.1
  auth:
    fileRealm:
    - secretName: elastic-auth
  nodeSets:
  - name: logging
    count: 1                                                   # 副本数(建议设置为3，我这里资源不足只用了1个副本)
    config:
      node.roles: ["master", "data", "remote_cluster_client"]  # 设置角色
      node.store.allow_mmap: false
    volumeClaimTemplates:
    - metadata:
        name : elasticsearch-data
      spec:
        storageClassName: local-storage                        # 设置存储
        accessModes:
        - ReadWriteOnce
        selector:
          matchLabels:
            app: elastic                                       # 通过这个标签选择存储pv
        resources:
          requests:
            storage: 10G
    podTemplate:
      spec:
        initContainers:
        - name: sysctl
          securityContext:
            privileged: true
            runAsUser: 0
          command: ['sh', '-c', 'sysctl -w vm.max_map_count=262144']
        containers:
        - name: elasticsearch
          resources:
            requests:
              memory: "1Gi"
              cpu: "800m"
            limits:
              memory: "1Gi"
              cpu: "1000m"
         #affinity:                                   # 多个几点则需要使用 affinity 亲和性保证每个副本调度到不同的node节点（因为pv用的是相同路径)
         # podAntiAffinity:
         #   requiredDuringSchedulingIgnoredDuringExecution:
         #   - labelSelector:
         #       matchLabels:
         #         elasticsearch.k8s.elastic.co/cluster-name: local-es
         #     topologyKey: kubernetes.io/hostname

apiVersion: traefik.containo.us/v1alpha1
kind: IngressRoute
metadata:
  name: elasticsearch-web
  namespace: logging
spec:
  entryPoints:
    - web
  routes:
    - match: Host(`elastic.od.com`)
      kind: Rule
      services:
        - name: elasticsearch-es-internal-http
          port: 9200

浏览器访问 http://elastic.od.com/ 用户elastic 密码：admin123

安装 Kibana

KibanaIngress

apiVersion: kibana.k8s.elastic.co/v1
kind: Kibana
metadata:
  name: kibana
  namespace: logging
spec:
  version: 8.11.1
  count: 1
  elasticsearchRef:
    name: elasticsearch                            # 关联elasticsearch 名称与es配置一致
    namespace: logging                             # 存在那个名称空间
  podTemplate:
    spec:
      containers:
      - name: kibana
        env:
          - name: NODE_OPTIONS
            value: "--max-old-space-size=512"
          - name: I18N_LOCALE                         # kibana 配置中添加语言配置，设置 kibana 为中文
            value: "zh-CN"
        resources:
          requests:
            memory: 512Mi
            cpu: "500m"
          limits:
            memory: 1Gi
            cpu: "1000m"

由于新版默认为HTTPS协议所以需要创建tls证书: openssl req -x509 -nodes -days 365 -newkey rsa:2048 -keyout tls.key -out tls.crt -subj "/CN=kibana.od.com/O=BoySefa/OU=wangxiansen" kubectl create secret tls tls --cert=tls.crt --key=tls.key -n logging

apiVersion: traefik.containo.us/v1alpha1
kind: IngressRoute
metadata:
  name: kibana-web
  namespace: logging
spec:
  entryPoints:
    - websecure
  routes:
    - match: Host(`kibana.od.com`)
      kind: Rule
      services:
        - name: kibana-kb-http
          port: 5601
  tls:
    secretName: tls

浏览器访问 https://kibana.od.com/ 同样也是用户elastic 密码：admin123

部署 Fluentd

一个针对日志的收集、处理、转发系统。通过丰富的插件系统，可以收集来自于各种系统或应用的日志，转化为用户指定的格式后，转发到用户所指定的日志存储系统之中。

Fluentd 通过一组给定的数据源抓取日志数据，处理后（转换成结构化的数据格式）将它们转发给其他服务，比如 Elasticsearch、对象存储、kafka等等。Fluentd 支持超过300个日志存储和分析服务，所以在这方面是非常灵活的。主要运行步骤如下

首先 Fluentd 从多个日志源获取数据
结构化并且标记这些数据
然后根据匹配的标签将数据发送到多个目标服务

配置

一般是通过一个配置文件来告诉 Fluentd 如何采集、处理数据的，下面简单和大家介绍下 Fluentd 的配置方法。

日志源配置

比如这里为了收集 Kubernetes 节点上的所有容器日志，就需要做如下的日志源配置：

<source>
  @id fluentd-containers.log
  @type tail                             # Fluentd 内置的输入方式，其原理是不停地从源文件中获取新的日志。
  path /var/log/containers/*.log         # 挂载的宿主机容器日志地址
  pos_file /var/log/es-containers.log.pos
  tag raw.kubernetes.*                   # 设置日志标签
  read_from_head true
  <parse>                                # 多行格式化成JSON
    @type multi_format                   # 使用 multi-format-parser 解析器插件
    <pattern>
      format json                        # JSON 解析器
      time_key time                      # 指定事件时间的时间字段
      time_format %Y-%m-%dT%H:%M:%S.%NZ  # 时间格式
    </pattern>
    <pattern>
      format /^(?<time>.+) (?<stream>stdout|stderr) [^ ]* (?<log>.*)$/
      time_format %Y-%m-%dT%H:%M:%S.%N%:z
    </pattern>
  </parse>
</source>

上面配置部分参数说明如下：

id：表示引用该日志源的唯一标识符，该标识可用于进一步过滤和路由结构化日志数据
type：Fluentd 内置的指令，tail 表示 Fluentd 从上次读取的位置通过 tail 不断获取数据，另外一个是 http 表示通过一个 GET 请求来收集数据。
path：tail 类型下的特定参数，告诉 Fluentd 采集 /var/log/containers 目录下的所有日志，这是 docker 在 Kubernetes 节点上用来存储运行容器 stdout 输出日志数据的目录。
pos_file：检查点，如果 Fluentd 程序重新启动了，它将使用此文件中的位置来恢复日志数据收集。
tag：用来将日志源与目标或者过滤器匹配的自定义字符串，Fluentd 匹配源/目标标签来路由日志数据。

路由配置

上面是日志源的配置，接下来看看如何将日志数据发送到 Elasticsearch：

<match **>
  @id elasticsearch
  @type elasticsearch
  @log_level info
  include_tag_key true
  type_name fluentd
  host "#{ENV['OUTPUT_HOST']}"
  port "#{ENV['OUTPUT_PORT']}"
  logstash_format true
  <buffer>
    @type file
    path /var/log/fluentd-buffers/kubernetes.system.buffer
    flush_mode interval
    retry_type exponential_backoff
    flush_thread_count 2
    flush_interval 5s
    retry_forever
    retry_max_interval 30
    chunk_limit_size "#{ENV['OUTPUT_BUFFER_CHUNK_LIMIT']}"
    queue_limit_length "#{ENV['OUTPUT_BUFFER_QUEUE_LIMIT']}"
    overflow_action block
  </buffer>
</match>

match：标识一个目标标签，后面是一个匹配日志源的正则表达式，这里想要捕获所有的日志并将它们发送给 Elasticsearch，所以需要配置成**。
id：目标的一个唯一标识符。
type：支持的输出插件标识符，这里要输出到 Elasticsearch，所以配置成 elasticsearch，这是 Fluentd 的一个内置插件。
log_level：指定要捕获的日志级别，这里配置成 info，表示任何该级别或者该级别以上（INFO、WARNING、ERROR）的日志都将被路由到 Elsasticsearch。
host/port：定义 Elasticsearch 的地址，也可以配置认证信息，的 Elasticsearch 不需要认证，所以这里直接指定 host 和 port 即可。
logstash_format：Elasticsearch 服务对日志数据构建反向索引进行搜索，将 logstash_format 设置为 true，Fluentd 将会以 logstash 格式来转发结构化的日志数据。
Buffer： Fluentd 允许在目标不可用时进行缓存，比如，如果网络出现故障或者 Elasticsearch 不可用的时候。缓冲区配置也有助于降低磁盘的 IO。

过滤

由于 Kubernetes 集群中应用太多，也还有很多历史数据，所以可以只将某些应用的日志进行收集，比如只采集具有 logging=true 这个 Label 标签的 Pod 日志，这个时候就需要使用 filter，如下所示：

# 删除无用的属性
<filter kubernetes.**>
  @type record_transformer
  remove_keys $.docker.container_id,$.kubernetes.container_image_id,$.kubernetes.pod_id,$.kubernetes.namespace_id,$.kubernetes.master_url,$.kubernetes.labels.pod-template-hash
</filter>
# 只保留具有logging=true标签的Pod日志
<filter kubernetes.**>
  @id filter_log
  @type grep
  <regexp>
    key $.kubernetes.labels.logging
    pattern ^true$
  </regexp>
</filter>

安装

要收集 Kubernetes 集群的日志，直接用 DasemonSet 控制器来部署 Fluentd 应用，这样，它就可以从 Kubernetes 节点上采集日志，确保在集群中的每个节点上始终运行一个 Fluentd 容器。当然可以直接使用 Helm 来进行一键安装，为了能够了解更多实现细节，这里还是采用手动方法来进行安装。

可以直接使用官方的对于 Kubernetes 集群的安装文档: https://docs.fluentd.org/container-deployment/kubernetes。

ConfigMapDaemonSetRBAc

vim cm.yaml

kind: ConfigMap
apiVersion: v1
metadata:
  name: fluentd-conf
  namespace: logging
data:
  # 容器日志
  containers.input.conf: |-
    <source>
      @id fluentd-containers.log
      @type tail                              # Fluentd 内置的输入方式，其原理是不停地从源文件中获取新的日志
      path /var/log/containers/*.log          # Docker 容器日志路径
      pos_file /var/log/es-containers.log.pos  # 记录读取的位置
      tag raw.kubernetes.*                    # 设置日志标签
      read_from_head true                     # 从头读取
      <parse>                                 # 多行格式化成JSON
        # 可以使用我们介绍过的 multiline 插件实现多行日志
        @type multi_format                    # 使用 multi-format-parser 解析器插件
        <pattern>
          format json                         # JSON解析器
          time_key time                       # 指定事件时间的时间字段
          time_format %Y-%m-%dT%H:%M:%S.%NZ   # 时间格式
        </pattern>
        <pattern>
          format /^(?<time>.+) (?<stream>stdout|stderr) [^ ]* (?<log>.*)$/
          time_format %Y-%m-%dT%H:%M:%S.%N%:z
        </pattern>
      </parse>
    </source>

    # 在日志输出中检测异常(多行日志)，并将其作为一条日志转发
    # https://github.com/GoogleCloudPlatform/fluent-plugin-detect-exceptions
    <match raw.kubernetes.**>           # 匹配tag为raw.kubernetes.**日志信息
      @id raw.kubernetes
      @type detect_exceptions           # 使用detect-exceptions插件处理异常栈信息
      remove_tag_prefix raw             # 移除 raw 前缀
      message log
      stream stream
      multiline_flush_interval 5
    </match>

    <filter **>  # 拼接日志
      @id filter_concat
      @type concat                # Fluentd Filter 插件，用于连接多个日志中分隔的多行日志
      key message
      multiline_end_regexp /\n$/  # 以换行符“\n”拼接
      separator ""
    </filter>

    # 添加 Kubernetes metadata 数据
    <filter kubernetes.**>
      @id filter_kubernetes_metadata
      @type kubernetes_metadata
    </filter>

    # 修复 ES 中的 JSON 字段
    # 插件地址：https://github.com/repeatedly/fluent-plugin-multi-format-parser
    <filter kubernetes.**>
      @id filter_parser
      @type parser                # multi-format-parser多格式解析器插件
      key_name log                # 在要解析的日志中指定字段名称
      reserve_data true           # 在解析结果中保留原始键值对
      remove_key_name_field true  # key_name 解析成功后删除字段
      <parse>
        @type multi_format
        <pattern>
          format json
        </pattern>
        <pattern>
          format none
        </pattern>
      </parse>
    </filter>

    # 删除一些多余的属性
    <filter kubernetes.**>
      @type record_transformer
      remove_keys $.docker.container_id,$.kubernetes.container_image_id,$.kubernetes.pod_id,$.kubernetes.namespace_id,$.kubernetes.master_url,$.kubernetes.labels.pod-template-hash
    </filter>

    # 只保留具有logging=true标签的Pod日志
    <filter kubernetes.**>
      @id filter_log
      @type grep
      <regexp>
        key $.kubernetes.labels.logging
        pattern ^true$
      </regexp>
    </filter>
    <filter **>
      @type stdout
    </filter>

  ###### 监听配置，一般用于日志聚合用 ######
  forward.input.conf: |-
    # 监听通过TCP发送的消息
    <source>
      @id forward
      @type forward
    </source>

  output.conf: |-
    <match **>
      @id elasticsearch
      @type elasticsearch
      @log_level info
      include_tag_key true
      host "#{ENV['FLUENT_ELASTICSEARCH_HOST']}"              # 通过pod.env传递Elasticsearch地址变量
      port "#{ENV['FLUENT_ELASTICSEARCH_PORT'] || 9200}"      # 通过pod.env传递Elasticsearch端口变量，默认9200
      scheme "#{ENV['FLUENT_ELASTICSEARCH_SCHEME'] || 'http'}" # 访问协议，默认http
      ssl_verify "#{ENV['FLUENT_ELASTICSEARCH_SSL_VERIFY'] || 'true'}"  # 是否开启ssl证书
      ssl_version "#{ENV['FLUENT_ELASTICSEARCH_SSL_VERSION'] || 'TLSv1'}" # ssl协议访问版本
      user "#{ENV['FLUENT_ELASTICSEARCH_USER']}"                 # Elasticsearch用户名
      password "#{ENV['FLUENT_ELASTICSEARCH_PASSWORD']}"         # Elasticsearch 密码
      reload_connections "#{ENV['FLUENT_ELASTICSEARCH_RELOAD_CONNECTIONS'] || 'true'}"
      logstash_prefix "#{ENV['FLUENT_ELASTICSEARCH_LOGSTASH_PREFIX'] || 'logstash'}"
      logstash_format true
      request_timeout 30s
      <buffer>
        @type file
        path /var/log/fluentd-buffers/kubernetes.system.buffer
        flush_mode interval
        retry_type exponential_backoff
        flush_thread_count 2
        flush_interval 5s
        retry_forever
        retry_max_interval 30
        chunk_limit_size 2M
        queue_limit_length 8
        overflow_action block
      </buffer>
    </match>

vim ds.yaml

apiVersion: apps/v1
kind: DaemonSet
metadata:
  name: fluentd
  namespace: logging
  labels:
    app: fluentd
    kubernetes.io/cluster-service: 'true'
spec:
  selector:
    matchLabels:
      app: fluentd
  template:
    metadata:
      labels:
        app: fluentd
        kubernetes.io/cluster-service: 'true'
    spec:
      tolerations:
          operator: "Exists"
      serviceAccountName: fluentd-es
      containers:
        - name: fluentd
          image: quay.io/fluentd_elasticsearch/fluentd:v4.6.2
          # image: fluent/fluentd-kubernetes-daemonset:v1-debian-elasticsearch
          env:
          - name: K8S_NODE_NAME
            valueFrom:
              fieldRef:
                fieldPath: spec.nodeName
          - name:  FLUENT_ELASTICSEARCH_HOST      # Elasticsearch SVC地址
            value: "elasticsearch-es-http"
          - name:  FLUENT_ELASTICSEARCH_PORT      # Elasticsearch 端口
            value: "9200"
          - name: FLUENT_ELASTICSEARCH_SCHEME     # 访问协议
            value: "https"
          - name: FLUENT_ELASTICSEARCH_SSL_VERIFY # 忽略证书
            value: "false"
          - name: FLUENT_ELASTICSEARCH_USER       # 用户名
            valueFrom:
              secretKeyRef:
                key: username
                name: elastic-auth
          - name: FLUENT_ELASTICSEARCH_PASSWORD  # 密码
            valueFrom:
              secretKeyRef:
                key: password
                name: elastic-auth
          - name: FLUENT_ELASTICSEARCH_LOGSTASH_PREFIX # 日志以k8s开头，默认是logstash
            value: "k8s"
          volumeMounts:
            - name: fluentconfig
              mountPath: /etc/fluent/config.d
            - name: varlog
              mountPath: /var/log
      volumes:
        - name: fluentconfig
          configMap:
            name: fluentd-conf
        - name: varlog
          hostPath:
            path: /var/log

为了能够灵活控制哪些节点的日志可以被收集，还可以添加了一个 nodSelector 属性：

nodeSelector:
  beta.kubernetes.io/fluentd-ds-ready: 'true'

意思就是要想采集节点的日志，那么我们就需要给节点打上上面的标签。

如果你需要在其他节点上采集日志，则需要给对应节点打上标签，使用如下命令：kubectl label nodes node名 beta.kubernetes.io/fluentd-ds-ready=true。

默认情况下 master 节点有污点，所以如果要想也收集 master 节点的日志，则需要添加上容忍：

tolerations:
  - operator: Exists

vim rbac.yaml

apiVersion: v1
kind: ServiceAccount
metadata:
  name: fluentd-es
  namespace: logging
  labels:
    k8s-app: fluentd-es
    kubernetes.io/cluster-service: 'true'
    addonmanager.kubernetes.io/mode: Reconcile
---
kind: ClusterRole
apiVersion: rbac.authorization.k8s.io/v1
metadata:
  name: fluentd-es
  labels:
    k8s-app: fluentd-es
    kubernetes.io/cluster-service: 'true'
    addonmanager.kubernetes.io/mode: Reconcile
rules:
  - apiGroups:
      - ''
    resources:
      - 'namespaces'
      - 'pods'
    verbs:
      - 'get'
      - 'watch'
      - 'list'
---
kind: ClusterRoleBinding
apiVersion: rbac.authorization.k8s.io/v1
metadata:
  name: fluentd-es
  labels:
    k8s-app: fluentd-es
    kubernetes.io/cluster-service: 'true'
    addonmanager.kubernetes.io/mode: Reconcile
subjects:
  - kind: ServiceAccount
    name: fluentd-es
    namespace: logging
    apiGroup: ''
roleRef:
  kind: ClusterRole
  name: fluentd-es
  apiGroup: ''

分别创建上面的 ConfigMap 对象和 DaemonSet

kubectl apply -f cm.yaml
kubectl apply -f rbac.yaml
kubectl apply -f ds.yaml

Fluentd 启动成功后，这个时候就可以发送日志到 ES 了，但是我们这里是过滤了只采集具有 logging=true 标签的 Pod 日志，所以现在还没有任何数据会被采集。

验证

下面我们部署一个简单的测试应用，新建 counter.yaml 文件，文件内容如下：

apiVersion: v1
kind: Pod
metadata:
  name: counter
  labels:
    logging: 'true' # 一定要具有该标签才会被采集
spec:
  containers:
    - name: count
      image: busybox
      args:
        [
          /bin/sh,
          -c,
          'i=0; while true; do echo "$i: $(date)"; i=$((i+1)); sleep 1; done',
        ]

该 Pod 只是简单将日志信息打印到 stdout，所以正常来说 Fluentd 会收集到这个日志数据，在 Kibana 中也就可以找到对应的日志数据了，使用 kubectl 工具创建该 Pod：

$ kubectl create -f counter.yaml
$ kubectl get pods
NAME                             READY   STATUS    RESTARTS   AGE
counter                          1/1     Running   0          52s

Pod 创建并运行后，回到 Kibana Dashboard 页面，点击左侧最下面的 Management -> Stack Management，进入管理页面，点击左侧 数据 下面的 索引管理 就会发现索引数据：