本文主要是介绍ELK系列之四---如何通过Filebeat和Logstash优化K8S集群的日志收集和展示，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！
前 言
上一篇文章《日志不再乱: 如何使用Logstash进行高效日志收集与存储》介绍了使用ELK收集通用应用的日志，在目前大多应用都已运行在K8S集群上的环境，需要考虑怎么收集K8S上的日志，本篇就介绍一下如何使用现有的ELK平台收集K8S集群上POD的日志。
K8S日志文件说明
一般情况下，容器中的日志在输出到标准输出(stdout)时，会以.log的命名方式保存在/var/log/pods/containerd_ID/目录中，例如：
K8S默认会在/var/log/containers/目录下创建软链接链接到/var/log/pods/containerd_id/目录下的具体日志文件，并规范化了软连接到文件名称，软链接的命名规范为
[podName]_[nameSpace]_[depoymentName]-[containerId].log
上面这个是deployment的命名方式，其他的会有些不同，例如：DaemonSet，StatefulSet等，不过所有的都有一个共同点，就是
每个软链接名称都是
*_[nameSpace]_*.log
所以根据这个日志特征我们可以很容易想到可以根据namespace来分别收集不同名称空间的日志。
一、日志收集架构
在ELK Stack组件中，有logstash和filebeat都可以实现日志收集，因为filebeat更轻量化，在K8S集群中通常使用filebeat来收集pod的日志，通过daemonSet控制器在每个K8S每个节点上运行一个filebeat来收集每个节点上的pod日志，filebeat官网图如下，filebeat输出支持对接Elasticsearch、Logstash、Kafka、Redis，本文选择对接Kafka，先缓存filebeat收集的日志，这样可以减少对ES的性能要求。
filebeat将收集的pod日志保存在K8S集群外的kafka集群，然后通过logstash再从kafka读取、过滤、整理之后再保存到elasticsearch集群中，然后通过kibana可视化管理日志，架构图如下
此架构中包含了K8S集群、Kafka集群、ElasticSearch集群，具体机器信息如下表
| 集群 | 主机名 | IP地址 | 应用 |
|---|---|---|---|
| k8s-cluster | k8s-master01 | 172.16.1.65 | k8s-control |
| k8s-cluster | k8s-node01 | 172.16.1.66 | k8s-work |
| k8s-cluster | k8s-node02 | 172.16.1.67 | k8s-work |
| k8s-cluster | k8s-node03 | 172.16.1.68 | k8s-work |
| kafka-cluster | kafka1 | 172.16.1.11 | zookeeper/kafka/kibana |
| kafka-cluster | kafka2 | 172.16.1.12 | zookeeper/kafka |
| kafka-cluster | kafka3 | 172.16.1.13 | zookeeper/kafka |
| ES-cluster | es1 | 172.16.1.101 | elasticsearch |
| ES-cluster | es2 | 172.16.1.102 | elasticsearch |
| ES-cluster | es3 | 172.16.1.103 | elasticsearch |
| Logstash | deploy | 172.16.1.70 | Lostash |
二、部署kafka集群
文章《在K8S上部署Cilium组件，看这一篇干货就够了》介绍了k8s-cluster集群的搭建过程，文章《可视化日志分析新境界：手把手教你搭建高效的ElasticSearch集群》介绍了ES-cluster集群的搭建过程
2.1、下载kafka
现在需要搭建kafka-cluster集群，在kafka-cluster集群的三个节点上下载最新版本kafka，下载二进制的压缩包
在每个节点上创建/opt目录，将kafka压缩包放到/opt目录下，解压并创建软连接
2.2、搭建zookeeper集群
最新版本kafka的taz包里已经包含了zookeeper的文件，所以在配置kafka之间先配置zookeeper，
三个几点上修改zookeeper配置文件如下
dataDir #指定zookeeper的数据目录
clientPort #指定zookeeper的监听端口
maxClientCnxns #限制最大client连接数
修改完成后，3个节点上同时开启zookeeper服务
~# /opt/kafka/bin/zookeeper-server-start.sh -daemon /opt/kafka/config/zookeeper.properties
查看三个几点都已监听2181端口即表示zookeeper集群搭建成功
2.3、搭建kafka集群
3个节点修改kafka的配置文件，如下图
broker.id #指定kafka的节点IP，每个节点都不相同
listeners #指定kafka的监听IP和端口，三个节点分别监听自己的IP
log.dirs #指定kafka的数据目录
zookeeper.connect #指定zookeeper的连接地址，三个地址都写上
其他选项默认即可
3个节点同时启动kafka
~# /opt/kafka/kafka-server-start.sh -daemon /opt/kafka/config/server.properties
3个几点同时监听2181和9092两个端口即搭建成功
2.4、测试验证kafka
kafka有一个客户端工具OffsetExplorer，可链接kafka
创建连接时填入任意一个kafka节点的IP即可连接，连接显示绿色表示集群状态正常
可以通过kafka的命令行客户端测试数据的写入的读取
#创建一个topic
~# /opt/kafka/bin/kafka-topic.sh --crate --topic quick-events --bootstrap-server 172.16.1.12:9092
#写入数据
~# /opt/kafka/bin/kafka-console-producer.sh --topic quick-events --bootstrap-server 172.16.1.12:9092
#读取数据
~# /opt/kafka/bin/kafka-console-consumer.sh --topic quick-events --bootstrap-server 172.16.1.12:9092
在OffsetExplorer上也可以看到刚才创建的topic和写入的数据
三、部署filebeat
在K8S集群上使用DaemonSet控制器部署filebeat，可以通过ELK Stack官网下载filebeat的部署文件filebeat-kubernetes.yaml
curl -L -O https://raw.githubusercontent.com/elastic/beats/8.15/deploy/kubernetes/filebeat-kubernetes.yaml
下载完成后修改yaml文件，添加创建elfk名称空间的指令，并修改下面所有资源的namespace为elfk
apiVersion: v1
kind: Namespace
metadata:name: elfklabels:name: elfk
修改ConfigMap部分，指定了按namespace收集
apiVersion: v1
kind: ConfigMap
metadata:name: filebeat-confignamespace: elfklabels:k8s-app: filebeat
data:filebeat.yml: |-filebeat.inputs:- type: containerenabled: truepaths:- /var/log/containers/*_kube-system_*logfields:log_topic: kube-systemenv: devmultiline.pattern: '(^\[[0-9]{4}-[0-9]{2}-[0-9]{2} [0-9]{2}:[0-9]{2}:[0-9]{2}\.[0-9]{3}\])|(^[0-9]{4}-[0-9]{2}-[0-9]{2} [0-9]{2}:[0-9]{2}:[0-9]{2}\.[0-9]{3})|(^[0-9]{2}:[0-9]{2}:[0-9]{2})'multiline.negate: truemultiline.match: aftermultiline.max_lines: 100- type: containerenabled: truepaths:- /var/log/containers/*_elfk_*logfields:log_topic: elfkenv: devmultiline.pattern: '(^\[[0-9]{4}-[0-9]{2}-[0-9]{2} [0-9]{2}:[0-9]{2}:[0-9]{2}\.[0-9]{3}\])|(^[0-9]{4}-[0-9]{2}-[0-9]{2} [0-9]{2}:[0-9]{2}:[0-9]{2}\.[0-9]{3})|(^[0-9]{2}:[0-9]{2}:[0-9]{2})'multiline.negate: truemultiline.match: aftermultiline.max_lines: 100- type: containerenabled: truepaths:- /var/log/containers/*_defaulte_*logfields:log_topic: defaulteenv: devmultiline.pattern: '(^\[[0-9]{4}-[0-9]{2}-[0-9]{2} [0-9]{2}:[0-9]{2}:[0-9]{2}\.[0-9]{3}\])|(^[0-9]{4}-[0-9]{2}-[0-9]{2} [0-9]{2}:[0-9]{2}:[0-9]{2}\.[0-9]{3})|(^[0-9]{2}:[0-9]{2}:[0-9]{2})'multiline.negate: truemultiline.match: aftermultiline.max_lines: 100- type: containerenabled: truepaths:- /var/log/containers/*_metallb-system_*logfields:log_topic: metallb-systemenv: devmultiline.pattern: '(^\[[0-9]{4}-[0-9]{2}-[0-9]{2} [0-9]{2}:[0-9]{2}:[0-9]{2}\.[0-9]{3}\])|(^[0-9]{4}-[0-9]{2}-[0-9]{2} [0-9]{2}:[0-9]{2}:[0-9]{2}\.[0-9]{3})|(^[0-9]{2}:[0-9]{2}:[0-9]{2})'multiline.negate: truemultiline.match: aftermultiline.max_lines: 100- type: containerenabled: truepaths:- /var/log/containers/*_redis_*logfields:log_topic: redisenv: devmultiline.pattern: '(^\[[0-9]{4}-[0-9]{2}-[0-9]{2} [0-9]{2}:[0-9]{2}:[0-9]{2}\.[0-9]{3}\])|(^[0-9]{4}-[0-9]{2}-[0-9]{2} [0-9]{2}:[0-9]{2}:[0-9]{2}\.[0-9]{3})|(^[0-9]{2}:[0-9]{2}:[0-9]{2})'multiline.negate: truemultiline.match: aftermultiline.max_lines: 100filebeat.config.modules:path: ${path.config}/modules.d/*.ymlreload.enabled: falseprocessors:#添加k8s元数据信息- add_kubernetes_metadata:host: ${NODE_NAME}matchers:- logs_path:logs_path: "/var/log/containers/"#移除多余的字段- drop_fields:fields:- host- ecs- log- agent- input- stream- container- kubernetes.pod.uid- kubernetes.namespace_uid- kubernetes.namespace_labels- kubernetes.node.uid- kubernetes.node.labels- kubernetes.replicaset- kubernetes.labels- kubernetes.node.nameignore_missing: true- script:lang: javascriptid: format_timetag: enablesource: |function process(event) {var str = event.Get("message");// 用括号提取时间戳var regex = /^\[(.*?)\]/;var match = str.match(regex);if (match && match.length > 1) {var time = match[1]; //提取的不带括号的时间戳event.Put("time", time);}// 提取不带括号的时间戳var regex2 = /^\d{2}:\d{2}:\d{2}/;var match2 = str.match(regex2);if (match2) {time = match2[0]; // Extracted timestampevent.Put("time", time);}}#优化层级结构- script:lang: javascriptid: format_k8stag: enablesource: |function process(event) {var k8s = event.Get("kubernetes");var newK8s = {podName: k8s.pod.name,nameSpace: k8s.namespace,imageAddr: k8s.container.name,hostName: k8s.node.hostname};event.Put("k8s", newK8s);}#添加时间，可以在logstash处理- timestamp:field: timetimezone: Asia/Shanghailayouts:- '2006-01-02 15:04:05'- '2006-01-02 15:04:05.999'test:- '2019-06-22 16:33:51'
修改daemonset部分，添加k8s-control-plan容忍，允许在master节点上运行filebeat
spec:tolerations:- key: node-role.kubernetes.io/control-planeoperator: Existseffect: NoSchedule
部署filebeat
~# kubectl apply -f filebeat-kubernetes.yaml
检查资源运行情况
~# kubectl get all -n elfk -o wide
4个节点上都运行filebeat的pod即表示部署成功
四、部署Logstash
4.1、检查kafka上topic
先通过OffsetExlporer查看kafka上是否成功收到了数据，产生了topic
上图可以看到以namespace为名称产生了topic，说明数据已经成功对接到了kafka集群
在kafka集群上不可很好的查看和检索日志数据，现在需要将kafka上的日志传输到ES集群上，这里需要使用Logstash对日志进行转发和过滤
4.2、部署Logstash
这里复用上一篇里部署的Logstash，新写一份对接kafka的配置文件即可，如下图
input #写入kafka集群的一个节点地址
topics #写入kafka集群上已经产生的topic
codec #使用json插件解码，生成多个日志字段
output #写入ES集群的连接地址
index #写入index名称
可以先测试一下配置文件是否正确
~# /usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/kafka-to-es.conf -t
重启logstash
~# systemctl restart logstash
4.3、检查ES上数据
可以通过elasticsearch head插件或者cerebro插件查看ES-cluster集群上数据是否成功产生
可以看到ES-cluster上产生jnlikai-k8s-container-log开头的index表示k8s的日志数据已经被转发到了ES集群
五、在Kibana上检索日志
在Kibana上创建数据视图
创建成功后可以看到此视图下有很多k8s的日志字段，检索日志时可以根据这些字段进行任意筛选检索
查看视图时也可以看到K8S上的节点、名称空间都是单独字段
欢迎关注作者的公众号，公众号每天分享运维干货文章
这篇关于ELK系列之四---如何通过Filebeat和Logstash优化K8S集群的日志收集和展示的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！
