ELFK基础搭建流程及在SpringBoot项目中进行日志采集的简单实践

一、前言

二、ELK简介

三、ELK常见的几种架构

四、Docker安装ELFK的详细流程

4.1环境说明。

4.2ElasticSearch搭建

配置ElasticSearch

启动ElasticSearch

4.3Kibana搭建

配置kibana

启动kibana

4.4LogStash搭建

配置LogStash

启动LogStash

4.5Filebeat搭建

配置Filebeat

启动Filebeat

五、SpringBoot生产日志配置

六、日志可视化展示

七、ELFK集成中间件Redis

八、后记

一、前言

在软件开发和维护中，日志总是DevOps们避不开一块。日志主要包括系统日志和应用程序日志，通过日志运维和开发人员可以了解服务器中软硬件的信息，检查应用程序或系统的故障，了解故障出现的原因，以便解决问题。分析日志可以更清楚的了解服务器的状态和系统安全状况，从而可以维护服务器稳定运行。

但当项目规模较大、日志量多而复杂的场景中，总会面临日志量太大如何归档、日志搜索太慢怎么办、如何多维度查询日志等问题。大型系统通常都是一个分布式部署的架构，不同的服务模块部署在不同的服务器上，问题出现时，大部分情况需要根据问题暴露的关键信息，定位到具体的服务器和服务模块。这就需要我们对日志信息进行集中化管理，常见解决思路是建立集中式日志收集系统，将所有节点上的日志统一收集，管理，访问。构建一套集中式日志系统，可以提高定位问题的效率。

一个完整的集中式日志系统，需要包含以下几个主要特点：

1）收集－能够采集多种来源的日志数据

2）传输－能够稳定的把日志数据传输到中央系统

3）存储－如何存储日志数据

4）分析－可以支持 UI 分析

5）警告－能够提供错误报告，监控机制

而ELFK则提供了一整套解决方案，并且都是开源软件，之间互相配合使用，完美衔接，高效的满足了很多场合的应用。是目前主流的一种日志系统。

二、ELK简介

ELK主要由ElasticSearch、Logstash和Kibana三个开源工具组成，还有其他专门由于收集数据的轻量型数据采集器Beats。

Elasticsearch ：分布式搜索引擎。具有高可伸缩、高可靠、易管理等特点。可以用于全文检索、结构化检索和分析，并能将这三者结合起来。Elasticsearch 是用Java 基于 Lucene 开发，现在使用最广的开源搜索引擎之一，Wikipedia 、StackOverflow、Github 等都基于它来构建自己的搜索引擎。在elasticsearch中，所有节点的数据是均等的。
Logstash ：数据收集处理引擎。支持动态的从各种数据源搜集数据，并对数据进行过滤、分析、丰富、统一格式等操作，然后存储以供后续使用。
Kibana ：可视化化平台。它能够搜索、展示存储在 Elasticsearch 中索引数据。使用它可以很方便的用图表、表格、地图展示和分析数据。
Beats在ELK中是一个轻量级日志采集器，其实Beats家族有6个成员，早期的ELK架构中使用Logstash收集、解析日志，但是Logstash对内存、cpu、io等资源消耗比较高。相比 Logstash，Beats所占系统的CPU和内存几乎可以忽略不计

所以这里也额外介绍其中一种Beats：

Filebeat：轻量级数据收集引擎。相对于Logstash所占用的系统资源来说，Filebeat 所占用的系统资源几乎是微乎及微。它是基于原先 Logstash-fowarder 的源码改造出来。换句话说：Filebeat就是新版的 Logstash-fowarder，也会是 ELK Stack 在 Agent 的第一选择。

三、ELK常见的几种架构

1 Elasticsearch + Logstash + Kibana

这是一种最简单的架构。这种架构，通过logstash收集日志，Elasticsearch分析日志，然后在Kibana(web界面)中展示。这种架构虽然是官网介绍里的方式，但是往往在生产中很少使用。

2 Elasticsearch + Logstash + filebeat + Kibana

这也就是所谓的ELFK，与上一种架构相比，这种架构增加了一个filebeat模块。filebeat是一个轻量的日志收集代理，用来部署在客户端，优势是消耗非常少的资源(较logstash)，所以生产中，往往会采取这种架构方式。它的好处在于：

Filebeat 可以将日志数据直接发送给 Elasticsearch。
适用于高吞吐量的场景，且对实时性要求较高。
可以在每个日志源上安装 Filebeat，减少 Logstash 的压力和资源消耗。

但是这种架构有一个缺点，当logstash出现故障，会造成日志的丢失。

3 Elasticsearch + Logstash + filebeat + redis(也可以是其他中间件，比如RabbitMQ) + Kibana

这种架构是上面那个架构的完善版，通过增加中间件，来避免数据的丢失。当Logstash出现故障，日志还是存在中间件中，当Logstash再次启动，则会读取中间件中积压的日志。

四、Docker安装ELFK的详细流程

4.1环境说明。

在生产环境使用，vm.max_map_count 内核设置必须至少为 262144。

vm.max_map_count 应该永久设置在 /etc/sysctl.conf：

root@master01:/data# grep vm.max_map_count /etc/sysctl.conf
vm.max_map_count=262144

在运行的系统中应用此配置，执行：

root@master01:/data#sysctl -w vm.max_map_count=262144

除此以外，建议增加limit，调整文件句柄数到 56635。因为Elasticsearch在处理大量数据时会打开许多文件。默认的文件句柄限制可能不足以满足Elasticsearch的需求，因此需要进行相应的配置。但是笔者这块仅作为测试搭建使用所以，临时改一下就ok了。

root@master01:/data# ulimit -n 65535

为了方便ELFK的搭建，这里笔者所用的ELFK组件一律用docker容器化进行搭建。

版本说明：

Elasticsearch、Logstash、Kibana、Filebeat安装的版本号必须全部一致,不然会出现kibana无法显示web页面。

创建ELFK的挂载目录：

root@master01:/data/elk/# mkdir -p /data/elk/es/{config,data,logs}
root@master01:/data/elk/# mkdir -p /data/elk/kibana/config
root@master01:/data/elk/# mkdir -p /data/elk/logstash/{config,data,pipeline}
root@master01:/data/elk/# mkdir -p /data/elk/filebeat/{config,logs}

4.2ElasticSearch搭建

给es目录授权

root@master01:/data/elk/es#chmod -R 777 elk/es

这里如果没有进行授权操作的话，后面当ES容器启动的时候会报错：

Exception in thread "main" java.lang.RuntimeException: starting java failed with [1]
output:
[0.001s][error][logging] Error opening log file 'logs/gc.log': Permission denied
[0.001s][error][logging] Initialization of output 'file=logs/gc.log' using options 'filecount=32,filesize=64m' failed.
error:
Invalid -Xlog option '-Xlog:gc*,gc+age=trace,safepoint:file=logs/gc.log:utctime,level,pid,tags:filecount=32,filesize=64m', see error log for details.
Error: Could not create the Java Virtual Machine.
Error: A fatal exception has occurred. Program will exit.
    at org.elasticsearch.server.cli.JvmOption.flagsFinal(JvmOption.java:120)
    at org.elasticsearch.server.cli.JvmOption.findFinalOptions(JvmOption.java:87)
    at org.elasticsearch.server.cli.MachineDependentHeap.determineHeapSettings(MachineDependentHeap.java:59)
    at org.elasticsearch.server.cli.JvmOptionsParser.jvmOptions(JvmOptionsParser.java:138)
    at org.elasticsearch.server.cli.JvmOptionsParser.determineJvmOptions(JvmOptionsParser.java:91)
    at org.elasticsearch.server.cli.ServerProcess.createProcess(ServerProcess.java:208)
    at org.elasticsearch.server.cli.ServerProcess.start(ServerProcess.java:104)
    at org.elasticsearch.server.cli.ServerProcess.start(ServerProcess.java:88)
    at org.elasticsearch.server.cli.ServerCli.startServer(ServerCli.java:239)
    at org.elasticsearch.server.cli.ServerCli.execute(ServerCli.java:100)
    at org.elasticsearch.common.cli.EnvironmentAwareCommand.execute(EnvironmentAwareCommand.java:54)
    at org.elasticsearch.cli.Command.mainWithoutErrorHandling(Command.java:85)
    at org.elasticsearch.cli.Command.main(Command.java:50)
    at org.elasticsearch.launcher.CliToolLauncher.mai(CliToolLauncher.java:64)

表示挂载的目录没有授权，因此这里需要对挂载目录进行简单的授权处理，避免启动报错。

配置ElasticSearch

进入es配置目录新建一个elasticsearch.yml的配置文件，并简单配置一下es的节点名称、主机网络和端口。

[root@localhost data]# vi elk/es/config/elasticsearch.yml
[root@localhost data]# cat elk/es/config/elasticsearch.yml
#集群名称
cluster.name: "es-master"
# 0.0.0.0为不限制，生产环境请设置为固定IP
network.host: 0.0.0.0
#设置对外服务的http端口，默认为9200
http.port: 9200
http.cors.enabled: true
#允许REST请求来自何处
http.cors.allow-origin: "*"
# 开启x-pack安全验证 访问时需要密码，先关闭保证访问可测试
xpack.security.enabled: false

这里的需要注意的是：在不进行密码验证时，xpack.security.enabled: false必须设置为false，而不是直接注释掉。因为elasticsearch默认是开启密码验证的如果这里注释掉，es运行后仍然会进行访问密码认证。所以，必须用false关闭掉才能进行curl或者网页ip地址访问。

启动ElasticSearch

root@master01:~#docker run -d -p 9200:9200 -p 9300:9300 --name es --network host -e ES_JAVA_OPTS="-Xms256m -Xmx256m"  -e "discovery.type=single-node" --restart=unless-stopped -v /data/elk/es/config/elasticsearch.yml:/usr/share/elasticsearch/config/elasticsearch.yml -v /data/elk/es/data:/usr/share/elasticsearch/data -v /data/elk/es/logs:/usr/share/elasticsearch/logs docker.elastic.co/elasticsearch/elasticsearch:8.15.0
Unable to find image 'docker.elastic.co/elasticsearch/elasticsearch:8.15.0' locally
8.15.0: Pulling from elasticsearch/elasticsearch
bef9b66d64c1: Already exists 
1fd631a7f77b: Pull complete 
be0505076ce2: Pull complete 
4ca545ee6d5d: Pull complete 
dc9db6ea5ff2: Pull complete 
dd7fea410473: Pull complete 
7aaf9e867095: Pull complete 
8c9e11efa7e5: Pull complete 
8d0f979b52e8: Pull complete 
8cadc84c16df: Pull complete 
Digest: sha256:84a73ced8390c059e7bc2858595c68dc36e6f8bdb98895dcab0074eda35ac96e
Status: Downloaded newer image for docker.elastic.co/elasticsearch/elasticsearch:8.15.0
WARNING: Published ports are discarded when using host network mode
8d9c813f83db403852c0a1d5b8996437cafc54ea0c9a3a1619c7216ef8066c1f

Tips：在Docker中，为了增强容器的安全性，不建议使用root用户来运行容器。使用root用户会给恶意用户提供更多的权限，可能导致容器内的应用程序被滥用或攻击。因此，为了保护容器和宿主机的安全，Docker默认以非特权用户的身份运行容器。

在Elasticsearch的官方Docker镜像中，默认情况下不允许使用root用户来启动Elasticsearch。这是为了确保运行Elasticsearch的容器具有更高的安全性。如果尝试使用root用户启动Elasticsearch容器，会得到一个警告，并且容器将停止运行。

为了解决这个问题，可以通过指定非root用户来运行Elasticsearch容器:

我们需要创建一个新的非root用户，并且禁用本地的root用户。我们可以使用以下命令创建新用户：

root@master01:/data/elk/es# adduser elasticsearch
正在添加用户"elasticsearch"...
正在添加新组"elasticsearch" (1001)...
正在添加新用户"elasticsearch" (1001) 到组"elasticsearch"...
创建主目录"/home/elasticsearch"...
正在从"/etc/skel"复制文件...
新的密码： 
无效的密码： 密码未通过字典检查 - ????????????/?????????
重新输入新的密码： 
passwd：已成功更新密码
正在改变 elasticsearch 的用户信息
请输入新值，或直接敲回车键以使用默认值
	全名 []: 
	房间号码 []: 
	工作电话 []: 
	家庭电话 []: 
	其它 []: 
这些信息是否正确？ [Y/n] y

这条命令将新用户的用户名指定为。然后，按照提示输入密码和其他信息即可创建新用户。

然后给该用户授权，并将es所属的文件目录访问权给到新建的elasticsearch用户：

root@master01:/data/elk/es#chown -R elasticsearch:elasticsearch /data/elk/es
root@master01:/data/elk/es# ls -l /data/elk/es
总计 12
drwxrwxrwx 2 elasticsearch elasticsearch 4096  8月 18 10:30 config
drwxrwxrwx 4 elasticsearch elasticsearch 4096  8月 18 10:41 data
drwxrwxrwx 2 elasticsearch elasticsearch 4096  8月 18 10:30 logs

然后切换到elasticsearch用户，以这个用户启动es，通过镜像，启动es容器，并将9200和9300端口映射到本机（elasticsearch的默认端口是9200，我们把宿主环境9200端口映射到Docker容器中的9200端口）。此处建议给容器设置固定ip，不然后面容器重启的时候可能出现ipaddress变动的情况。

elasticsearch@master01:~#sudo docker run -d -p 9200:9200 -p 9300:9300 --name es --network host -e ES_JAVA_OPTS="-Xms256m -Xmx256m"  -e "discovery.type=single-node" --restart=unless-stopped -v /data/elk/es/config/elasticsearch.yml:/usr/share/elasticsearch/config/elasticsearch.yml -v /data/elk/es/data:/usr/share/elasticsearch/data -v /data/elk/es/logs:/usr/share/elasticsearch/logs docker.elastic.co/elasticsearch/elasticsearch:8.15.0
[sudo] elasticsearch 的密码:
elasticsearch不在sudoers文件中，此事将被报告

但是这里出现个elasticsearch不在sudoers文件中，此事将被报告 的问题。需要处理一下，不然没法使用命令。处理流程如下：

#先切换至root用户
elasticsearch@master01:~#su root
密码:
#查看/etc/sudoers 文件权限，如果只读权限，修改为可写权限
root@master01:/data/elk/es# ls -l /etc/sudoers
-r--r----- 1 root root 1700  8月 17 22:46 /etc/sudoers
#增加修改权限
root@master01:/data/elk/es#chmod u+w /etc/sudoers
#添加用户elasticsearch的允许sudo操作
root@master01:/data/elk/es#vi /etc/sudoers
#查看
root@master01:/data/elk/es# cat /etc/sudoers
#
# This file MUST be edited with the 'visudo' command as root.
#
# Please consider adding local content in /etc/sudoers.d/ instead of
# directly modifying this file.
#
# See the man page for details on how to write a sudoers file.
#
Defaults        env_reset
Defaults        mail_badpass
Defaults        secure_path="/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin:/snap/bin"
Defaults        use_pty

# This preserves proxy settings from user environments of root
# equivalent users (group sudo)
#Defaults:%sudo env_keep += "http_proxy https_proxy ftp_proxy all_proxy no_proxy"

# This allows running arbitrary commands, but so does ALL, and it means
# different sudoers have their choice of editor respected.
#Defaults:%sudo env_keep += "EDITOR"

# Completely harmless preservation of a user preference.
#Defaults:%sudo env_keep += "GREP_COLOR"

# While you shouldn't normally run git as root, you need to with etckeeper
#Defaults:%sudo env_keep += "GIT_AUTHOR_* GIT_COMMITTER_*"

# Per-user preferences; root won't have sensible values for them.
#Defaults:%sudo env_keep += "EMAIL DEBEMAIL DEBFULLNAME"

# "sudo scp" or "sudo rsync" should be able to use your SSH agent.
#Defaults:%sudo env_keep += "SSH_AGENT_PID SSH_AUTH_SOCK"

# Ditto for GPG agent
#Defaults:%sudo env_keep += "GPG_AGENT_INFO"

# Host alias specification

# User alias specification

# Cmnd alias specification

# User privilege specification
root    ALL=(ALL:ALL) ALL
elasticsearch   ALL=(ALL) ALL
# Members of the admin group may gain root privileges
%admin ALL=(ALL) ALL

# Allow members of group sudo to execute any command
%sudo   ALL=(ALL:ALL) ALL

# See sudoers(5) for more information on "@include" directives:

@includedir /etc/sudoers.d
#恢复/etc/sudoers的权限
root@master01:/data/elk/es#chmod 440 /etc/sudoers
#查看
root@master01:/data/elk/es# ls -l /etc/sudoers
-r--r----- 1 root root 1700  8月 17 22:46 /etc/sudoers
#重新切换为elasticsearch用户，重新运行es容器
root@master01:/data/elk/es# su elasticsearch
elasticsearch@master01:/data/elk/es$ sudo docker run -d -p 9200:9200 -p 9300:9300 --name es --network host -e ES_JAVA_OPTS="-Xms256m -Xmx256m"  -e "discovery.type=single-node" --restart=unless-stopped -v /data/elk/es/config/elasticsearch.yml:/usr/share/elasticsearch/config/elasticsearch.yml -v /data/elk/es/data:/usr/share/elasticsearch/data -v /data/elk/es/logs:/usr/share/elasticsearch/logs docker.elastic.co/elasticsearch/elasticsearch:8.15.0
Unable to find image 'docker.elastic.co/elasticsearch/elasticsearch:8.15.0' locally
8.15.0: Pulling from elasticsearch/elasticsearch
bef9b66d64c1: Already exists 
1fd631a7f77b: Pull complete 
be0505076ce2: Pull complete 
4ca545ee6d5d: Pull complete 
dc9db6ea5ff2: Pull complete 
dd7fea410473: Pull complete 
7aaf9e867095: Pull complete 
8c9e11efa7e5: Pull complete 
8d0f979b52e8: Pull complete 
8cadc84c16df: Pull complete 
Digest: sha256:84a73ced8390c059e7bc2858595c68dc36e6f8bdb98895dcab0074eda35ac96e
Status: Downloaded newer image for docker.elastic.co/elasticsearch/elasticsearch:8.15.0
WARNING: Published ports are discarded when using host network mode
8d9c813f83db403852c0a1d5b8996437cafc54ea0c9a3a1619c7216ef8066c1f

最后验证ES运行是否成功：

可通过curl的形式进行验证：curl http://localhost:9200

访问ES端口如出现下面内容：

root@master01:/data/elk/es# curl http://192.168.1.200:9200
{
  "name" : "master01",
  "cluster_name" : "es-master",
  "cluster_uuid" : "SPlA9bg2Sly4fKzvbQbzTA",
  "version" : {
    "number" : "8.15.0",
    "build_flavor" : "default",
    "build_type" : "docker",
    "build_hash" : "1a77947f34deddb41af25e6f0ddb8e830159c179",
    "build_date" : "2024-08-05T10:05:34.233336849Z",
    "build_snapshot" : false,
    "lucene_version" : "9.11.1",
    "minimum_wire_compatibility_version" : "7.17.0",
    "minimum_index_compatibility_version" : "7.0.0"
  },
  "tagline" : "You Know, for Search"
}

则证明ES容器运行正常了。

4.3Kibana搭建

配置kibana

新建kibana的配置文件并配置下列信息：

root@young-virtual-machine:/data/elk/es# mkdir -p /data/elk/kibana
root@young-virtual-machine:/data/elk/es# vim ../kibana/kibana.yml
root@young-virtual-machine:/data/elk/es# cat ../kibana/kibana.yml
server.name: kibana
server.port: 5601
server.host: "0.0.0.0"
xpack.monitoring.ui.container.elasticsearch.enabled: true
#中文设置
i18n.locale: "zh-CN"
#关闭遥测
telemetry.enabled:false

这里先不用配置es的密码配置，后面当es设置好密码验证后，可以手动配置后自动生成。

启动kibana

保存配置文件后即可在docker上拉取镜像并运行kibana容器：

root@master01:/data/elk/kibana# docker run -d --restart=unless-stopped --log-driver json-file --log-opt max-size=100m --log-opt max-file=2 -e TZ=Asia/Shanghai --name kibana -p 5601:5601 --network host -v /data/elk/kibana/config/kibana.yml:/usr/share/kibana/config/kibana.yml -v /etc/localtime:/etc/localtime docker.elastic.co/kibana/kibana:8.15.0
Unable to find image 'docker.elastic.co/kibana/kibana:8.15.0' locally
8.15.0: Pulling from kibana/kibana
bef9b66d64c1: Already exists 
f6b84247827c: Pull complete 
b1a5a823635f: Pull complete 
7ff3c1349574: Pull complete 
acaf211e68f0: Pull complete 
e6482380a03e: Pull complete 
4ca545ee6d5d: Pull complete 
5d37849b8bfb: Pull complete 
38a4f0ace1b9: Pull complete 
93469ea36cfe: Pull complete 
40d69daa44b4: Pull complete 
9bcdcf41232d: Pull complete 
50c27f863681: Pull complete 
Digest: sha256:ff5f6b9a49f410658b74b337b102c302bbeb52b470efe1f0e3af3c7526fbe0e7
Status: Downloaded newer image for docker.elastic.co/kibana/kibana:8.15.0
WARNING: Published ports are discarded when using host network mode
26ace53f1cf62f658e3547d99559a74cba5ee08ad37667c1d11d1741bcc5525e

通过观察日志，等待启动成功：

只要日志中没有严重影响服务运行的错误，就可以尝试着访问kibana页面试试了。偶尔一些插件异常的问题暂时可以忽略，毕竟咱们也没配置什么插件。

可以看到目前访问正常~

4.4LogStash搭建

配置LogStash

首先仍然在启动前先将建立一些需要的文件目录和配置文件，一般与需要挂载的文件对应。在logstash中一般需要data、yml配置文件和输出的conf配置文件。

Logstash包括以下设置文件:

logstash.yml：包含Logstash配置标志。您可以在这个文件中设置标志，而不是在命令行传递标志。在命令行设置的任何标志都会覆盖logstash.yml文件中的相应设置。
pipelines.yml：包含在单个Logstash实例中运行多个管道的框架和说明。
jvm.options：选项包含JVM配置标志。使用该文件设置总堆空间的初始值和最大值。我们还可以使用该文件来设置Logstash的区域设置。
log4j2.properties：包含log4j 2库的默认设置。

这里为了方便对不同的数据源进行采集，我们可以采用管道pipeline进行分别配置。并对一些常见配置进行设置：

#先授权
root@master01:/data/elk/logstash# chmod 777 -R ./
#配置运行jvm堆内存大小
root@master01:/data/elk/logstash# cat >>config/jvm.options<<EOF
> -Xmx128m
> -Xms128m
> EOF
#配置logstash
root@master01:/data/elk/logstash# cat >>config/logstash.yml<<EOF
> http.host: "0.0.0.0"
> xpack.management.pipeline.id: ["test"]
> EOF
#配置管道信息和映射信息
root@master01:/data/elk/logstash# cat >>config/pipelines.yml<<EOF
> - pipeline.id: test
>   path.config: "/usr/share/logstash/pipeline"
> EOF

然后我们先配置filebeat作为中间件，用于收集SpingBoot服务上的日志的信息记录作为输入信息。

root@master01:/data/elk/logstash# vi pipeline/pipeline-springboot-log.conf
root@master01:/data/elk/logstash# cat pipeline/pipeline-springboot-log.conf
# 输入配置
input {
  beats {
    port => 5044
    client_inactivity_timeout => 36000
  }
}

# 过滤配置
filter {
  grok {
    match => { "message" => "%{SYSLOGTIMESTAMP:syslog_timestamp} %{SYSLOGHOST:syslog_hostname} %{DATA:syslog_program}(?:$%{POSINT:syslog_pid}$)?: %{GREEDYDATA:syslog_message}" }
  }
  date {
    match => [ "syslog_timestamp", "MMM  d HH:mm:ss", "MMM dd HH:mm:ss" ]
  }
}

# 输出配置
output {
  elasticsearch {
    hosts => ["http://192.168.1.200:9200"]
    index => "logstash-%{+YYYY.MM.dd}"
  }
  stdout {
    codec => rubydebug
  }
}

启动LogStash

root@master01:/data/elk/logstash# docker run -d --restart=unless-stopped --network host  -p 5044:5044 -p 9600:9600 -e TZ=Asia/Shanghai -v /etc/localtime:/etc/localtime -v /data/elk/logstash/config:/usr/share/logstash/config -v /data/elk/logstash/data:/usr/share/logstash/data -v /data/elk/logstash/pipeline:/usr/share/logstash/pipeline --name logstash docker.elastic.co/logstash/logstash:8.15.0
Unable to find image 'docker.elastic.co/logstash/logstash:8.15.0' locally
8.15.0: Pulling from logstash/logstash
bef9b66d64c1: Already exists 
bbee25ee946c: Pull complete 
26488eb933d4: Pull complete 
45b1fd29c944: Pull complete 
4ca545ee6d5d: Pull complete 
12a8edbceb65: Pull complete 
7d5f2a507bae: Pull complete 
6eecdc48a52f: Pull complete 
46ef65f74cf8: Pull complete 
1ebb4dc57ea8: Pull complete 
d0fca5b18401: Pull complete 
db33482f1e04: Pull complete 
Digest: sha256:73a30fb57f305c0a6e0018ef37aeda016b5a3578a95530af13579382938cc3d6
Status: Downloaded newer image for docker.elastic.co/logstash/logstash:8.15.0
WARNING: Published ports are discarded when using host network mode
9f35c817f92750a444e3acd3b5efaf3997de96335e5153441ce7a181263685e6

此时等待启动完成后就可以配置filebeat收集日志的地址了，目前已经启动成功。

4.5Filebeat搭建

配置Filebeat

这里首先仍然需要给该目录授权，以便应用能进行数据读取。然后创建配置目录及文件、日志目录等等。

root@master01:/data/elkt# mkdir -p filebeat/{config,logs}
root@master01:/data/elk# chmod 777 -R /data/elk/filebeat
root@master01:/data/elk/filebeat# vi config/filebeat.yml
root@master01:/data/elk/filebeat# cat config/filebeat.yml 
filebeat.inputs:
- type: log
  enabled: true

  paths:
    #docker部署则指定容器内部路径，并且需做映射
    - /usr/share/filebeat/logs/*.log
  #自定义字段,可用于提供给Logstash区分日志来源
  fields:
    log_name: server1
      #开启自定义字段
    fields_under_root: true

processors:
    - drop_fields:
            fields: ["log","input","host","agent","ecs"] # 过滤不需要的字段

output.logstash:
  hosts: ["192.168.1.200:5044"]


#output.elasticsearch:
# hosts: ["192.168.1.200:9200"]
# indices:
#  - index: "filebeat-%{+yyyy.MM.dd}"
#output.redis:
# hosts: ["192.168.1.200:6379"]
# key: "filebeat-redis"
# db: 1
# timeout: 60

启动Filebeat

root@master01:/data/elk/filebeat# docker run -d --name filebeat --network host --restart=unless-stopped --log-driver json-file --log-opt max-size=100m --log-opt max-file=2 -v /etc/localtime:/etc/localtime -v /data/elk/filebeat/config/filebeat.yml:/usr/share/filebeat/filebeat.yml -v /data/elk/filebeat/logs/:/usr/share/filebeat/logs/ docker.elastic.co/beats/filebeat:8.15.0
Unable to find image 'docker.elastic.co/beats/filebeat:8.15.0' locally
8.15.0: Pulling from beats/filebeat
bef9b66d64c1: Already exists 
12eb5912ba82: Pull complete 
cb21ead838f7: Pull complete 
357c868fca87: Pull complete 
678d81025334: Pull complete 
32dc2e2450c7: Pull complete 
285151d7dd89: Pull complete 
72f56439d14c: Pull complete 
8a3d0952a428: Pull complete 
bb370a452652: Pull complete 
6f7cb028dc1f: Pull complete 
4ca545ee6d5d: Pull complete 
Digest: sha256:c2cb458cba1eff0536639f7f101d8ddf570a4cf1973ed79d95db243077a39e7c
Status: Downloaded newer image for docker.elastic.co/beats/filebeat:8.15.0
WARNING: Published ports are discarded when using host network mode
ea41de80baae3ccadf43644aa12dbd70d796e9e70e454c47a29d3ee2173a0071

到此，ELFK最最基础款就算搭建完成了，下面需要配置SpringBoot项目的生产日志生成。

五、SpringBoot生产日志配置

一般生产环境下都需要记录项目日志，以此来排查在项目运行过程中的问题。而对于SpringBoot项目，它默认集成了Logback，Logback是一个流行的Java日志框架，这个日志框架可以提供高效的日志记录功能。可以通过在src/main/resources目录下创建logback-spring.xml或者logback.xml文件来自定义Logback配置。Spring Boot推荐使用logback-spring.xml，因为它支持Spring的环境变量和占位符。

这里我们根据生产环境的日志存放地编写一个logback-spring.xml文件来产生并记录日志文件，将至其指定到filebeat可采集的目录下：

<configuration>
    <!--log日志目录-->
    <property name="LOG_PATH" value="/data/elk/filebeat/logs/"/>
<!--    正常日志名称-->
    <property name="LOG_FILE_NAME" value="application.log"/>
<!--    错误日志名称-->
    <property name="LOG_FILE_ERROR_NAME" value="application-error.log"/>
    <!-- 控制台日志 -->
    <appender name="consoleLog" class="ch.qos.logback.core.ConsoleAppender">
        <encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
            <!--格式化输出（配色）：%d表示日期，%thread表示线程名，%-5level：级别从左显示5个字符宽度%msg：日志消息，%n是换行符-->
            <pattern>%yellow(%d{yyyy-MM-dd HH:mm:ss.SSS}) %red([%thread]) %highlight(%-5level) %cyan(%logger{50}) - %magenta(%msg) %n</pattern>
            <charset>UTF-8</charset>
        </encoder>
    </appender>

    <!--根据日志级别分离日志，分别输出到不同的文件-->
    <!--appender用来格式化日志输出节点，有俩个属性name和class，class用来指定哪种输出策略，常用就是控制台输出策略和文件输出策略。-->
    <appender name="fileInfoLog" class="ch.qos.logback.core.rolling.RollingFileAppender">
        <file>${LOG_PATH}/${LOG_FILE_NAME}</file>
        <encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
            <pattern>
                %d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{50} - %msg%n
            </pattern>
            <charset>UTF-8</charset>
        </encoder>
        <!--滚动策略-->
        <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
            <!--按时间保存日志 修改格式可以按小时、按天、月来保存-->
            <fileNamePattern>${LOG_PATH}/application.%d{yyyy-MM-dd}.log.zip</fileNamePattern>
            <!--保存时长-->
            <MaxHistory>30</MaxHistory>
            <!--文件大小-->
            <totalSizeCap>100MB</totalSizeCap>
        </rollingPolicy>
    </appender>k

    <!--ERROR-->
    <appender name="fileErrorLog" class="ch.qos.logback.core.rolling.RollingFileAppender">
        <file>${LOG_PATH}/${LOG_FILE_ERROR_NAME}</file>
        <filter class="ch.qos.logback.classic.filter.ThresholdFilter">
            <level>ERROR</level>
        </filter>
        <encoder>
            <pattern>
                %d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{50} - %msg%n
            </pattern>
        </encoder>
        <!--滚动策略-->
        <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
            <!--路径-->
            <fileNamePattern>${LOG_PATH}/application-error.%d{yyyy-MM-dd}.log.zip</fileNamePattern>
            <MaxHistory>90</MaxHistory>
        </rollingPolicy>
    </appender>

    <root level="info">
        <!-- 打印到控制台 -->
        <appender-ref ref="consoleLog"/>
        <appender-ref ref="fileInfoLog"/>
        <appender-ref ref="fileErrorLog"/>
    </root>
</configuration>

然后在服务器上重新运行SpringBoot项目。可以检查随着项目启动，调用相关服务接口后，日志文件已经产生了。：

root@master01:/data/elk/filebeat# ls -l logs/
总计 388
-rw-r--r-- 1 root  root     208  8月 19 17:36 application-error.log
-rw-r--r-- 1 root  root    2977  8月 19 17:36 application.log
-rw------- 1 young young  29946  8月 18 14:54 filebeat-20240818-3.ndjson
-rw------- 1 young young  61440  8月 18 15:14 filebeat-20240818-4.ndjson
-rw------- 1 young young  22127  8月 18 15:29 filebeat-20240818-5.ndjson
-rw------- 1 young young  20800  8月 18 21:14 filebeat-20240818-6.ndjson
-rw------- 1 young young  42152  8月 19 18:15 filebeat-20240819-1.ndjson
-rw------- 1 young young  30174  8月 19 18:19 filebeat-20240819-2.ndjson
-rw------- 1 young young  36331  8月 19 18:25 filebeat-20240819-3.ndjson
-rw------- 1 young young 104668  8月 19 18:06 filebeat-20240819.ndjson

六、日志可视化展示

等待logstash处理一段时间后，进入kibana可视化界面，找到如下地址即可在页面上看到生成的一条索引：

因为我这块当初在logstash配置上设置的索引是index => "logstash-%{+YYYY.MM.dd}"，因此filebeat收集后传给logstas就是一条过滤后的索引信息。然后点击这条索引，进入后点击“Discover索引”。

就可以在查看日志文件中的日志信息了。只不过现有的信息是经过logstash过滤后的：

并且kibana会根据时间间隔同步记录日志生产信息。至于具体的操作，感兴趣的小伙伴可以后面依照这个方法采集日志后慢慢研究。

七、ELFK集成中间件Redis

前面也提到了，单纯的ELFK运行当logstash出现故障，会造成日志的丢失。而Redis作为一个高性能的内存数据存储，可以用作日志数据的缓冲区，确保在Logstash处理能力不足或临时故障时，日志数据不会丢失。当然这里也可以用其他的哈，但是逻辑差不多。

这里redis笔者同样使用docker运行，具体流程就不重复再说了。仅看看需要如何配置：

首先{关闭logstash（因为redis仅是作为缓冲区，如果开启会看不到redis中的存放日志数据，直接被logstash解析到es），如果不需要看数据可以略过这步。}在filebeat上修改一下输出的地方，原来是输出到logstash，现在修改为redis并重启：

filebeat.inputs:
- type: log
  enabled: true
  paths:
#更换一个采集路径
    - /usr/share/filebeat/logs/springboot-test-log2/*.log
  fields:
    log_name: springboot-test2
  fields_under_root: true
  tags: ["test2"]

processors:
  - drop_fields:
      fields: ["log", "input", "host", "agent", "ecs"]

output.redis:
  hosts: ["192.168.1.200:6379"]
  key: "filebeat-redis"
  db: 1
  timeout: 5

这里有个注意点需要关注一下：docker容器启动filebeat的时候不知道是不是笔者没有挂载filebeat日志的原因。容器启动时看不到日志。因此有问题的话很难排查。咱们可以通过filebeat命令进行调试，比如：filebeat -e -d "*"。在docker上可以使用docker exec -it filebeat filebeat -e -d "*"
方便查看filebeat的日志信息采集过程。

然后可以到redis看看数据是否存在：

然后在logstash上也需要修改输入点，由原来的beats改为redis：

root@master01:/data/elk/logstash# vi pipeline/pipeline-springboot-log.conf 
root@master01:/data/elk/logstash# cat pipeline/pipeline-springboot-log.conf 
# 输入配置
input {
#  beats {
#    port => 5044
#    client_inactivity_timeout => 36000
#  }
   redis {
    host => "192.168.1.200"
    port => 6379
    data_type => "list"
    key => "filebeat-redis"
    batch_count => 100
  }
}

# 过滤配置
filter {
  grok {
    match => { "message" => "%{SYSLOGTIMESTAMP:syslog_timestamp} %{SYSLOGHOST:syslog_hostname} %{DATA:syslog_program}(?:$%{POSINT:syslog_pid}$)?: %{GREEDYDATA:syslog_message}" }
  }
  date {
    match => [ "syslog_timestamp", "MMM  d HH:mm:ss", "MMM dd HH:mm:ss" ]
  }
}

# 输出配置
output {
  elasticsearch {
    hosts => ["http://192.168.1.200:9200"]
    index => "logstash-addredis-%{+YYYY.MM.dd}"
  }
  stdout {
    codec => rubydebug
  }
}

这里为了区分原来的日志索引，所以将es中的索引名添加上了addRedis以作区分。最后重新运行一下logstash。重新在kibana上查看日志索引是否创建成功：