虚拟化运维LogKubernetes
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
代码语言：javascript复制# 安装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.yamlLocalpath本地存储
代码语言：javascript复制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
代码语言：javascript复制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/hostnameapiVersion: 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
代码语言：javascript复制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"代码语言：javascript复制由于新版默认为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 节点上的所有容器日志，就需要做如下的日志源配置：
代码语言：javascript复制<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：
代码语言：javascript复制<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
代码语言：javascript复制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>代码语言：javascript复制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 节点的日志，则需要添加上容忍：
代码语言：javascript复制tolerations:
- operator: Exists代码语言：javascript复制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
代码语言：javascript复制kubectl apply -f cm.yaml
kubectl apply -f rbac.yaml
kubectl apply -f ds.yamlFluentd 启动成功后，这个时候就可以发送日志到 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 52sPod 创建并运行后，回到 Kibana Dashboard 页面，点击左侧最下面的 Management -> Stack Management，进入管理页面，点击左侧 数据 下面的 索引管理 就会发现索引数据：
点击左侧 Kibana 下面的 试图数据 点击 创建试图数据 开始导入索引数据：：
在该页面中配置使用哪个字段按时间过滤日志数据，在下拉列表中，选择@timestamp字段，然后点击 创建索引模式，创建完成后，点击左侧导航菜单中的 Discover，然后就可以看到一些直方图和最近采集到的日志数据了：
