ELK——日志处理界的瑞士军刀

引言

一、ELK简介

（一）基本概述

1.Elasticsearch服务

2.Logstash服务

2.2 logstash关键组件

2.2 logstash数据流向

3.Kibana服务

（二）ELK工作流程

（三）ELK的应用价值

二、部署搭建ELK

（一）搭建Elasticsearch集群

1.安装JDK

2.安装elasticsearch

3.修改配置文件

4.使用浏览访问节点信息

5.安装插件

6.插入索引

（二）搭建Logstash

1.下载http服务

2.安装java环境

3.安装logstash

4.测试logstash

5.定义logstash配置文件

（三）安装kibana

（四）监控Apache服务器日志

1.开启apache服务

2.添加监听信息

3.加载配置文件

三、搭建ELK+filebeat

（一）Filebeat基本介绍

（二）部署ELK+filebeat

1.下载安装filebeat

2.修改yml文件

3.修改logstash服务配文文件

4.查看信息

总结

引言

在大数据时代，日志管理与分析已经成为运维、开发和业务优化中不可或缺的一环。而ELK ，即Elasticsearch、Logstash、Kibana三者组成的开源解决方案，以其强大的实时数据分析能力，在业界广受好评，被誉为“日志处理界的瑞士军刀”。

一、ELK简介

ELK平台是一套完整的日志集中处理解决方案，将 ElasticSearch、Logstash 和 Kiabana 三个开源工具配合使用，完成更强大的用户对日志的查询、排序、统计需求

（一）基本概述

1.Elasticsearch服务

Elasticsearch: 作为一款分布式、RESTful 风格的搜索引擎，Elasticsearch 提供了近实时的全文搜索和分析功能。它能高效地存储、检索和分析海量数据，并且支持水平扩展，以满足大规模数据处理的需求。

2.Logstash服务

Logstash: Logstash 是一个开源的数据收集引擎，可以同时从多个来源采集数据，然后进行过滤、转换，最终将结构化或非结构化的数据输出到你希望的目的地（如 Elasticsearch）。无论是系统日志、数据库记录还是应用程序日志，Logstash 都能轻松应对。

2.2 logstash关键组件

logstash有三个关键组件，分别为：input、filter、output

1. Input（输入） Input 组件是 Logstash 数据处理的第一环节，它负责从不同的数据源捕获和收集数据。这些数据源可以是本地文件、网络socket、数据库、消息队列（如 Kafka）、 syslog 服务、Beats 系列工具（如 Filebeat）等等。Logstash 支持多种输入插件，使得它能够从多样化的环境中实时抓取数据。

2. Filter（过滤器） Filter 组件位于 Input 和 Output 之间，它是数据处理的核心部分，用于对从 Input 获取的原始事件进行清洗、转换和增强。Filter 可以用来解析日志格式、去除无关信息、添加标签、进行地理编码、加密解密数据、甚至进行条件判断和数据映射等操作。常见的过滤器包括 grok（用于解析非结构化日志）、mutate（修改字段值或删除字段）、geoip（地理位置信息处理）等。

3. Output（输出） Output 组件是 Logstash 流水线的最后一个阶段，负责将经过 Filter 处理后的结构化数据发送到目标存储或系统。常见的 Output 包括 Elasticsearch（用于日志分析和搜索）、Redis（缓存或消息队列）、MySQL 或其他关系型数据库（持久化存储）、甚至是邮件服务器（发送报警通知）等。

2.2 logstash数据流向

数据在 Logstash 中的流转过程如下

① 从某个数据源（如日志文件）通过 file 输入插件收集原始日志数据。

② 收集的数据进入 Filter 阶段，通过 grok 过滤器解析日志格式，提取有用信息，并可能通过 date 过滤器标准化时间戳。

③ 经过处理的事件被送至 Output 阶段，通过 elasticsearch 输出插件将结构化日志数据推送到 Elasticsearch 集群，以便后续的搜索、分析和可视化。

3.Kibana服务

Kibana: Kibana 是一个可视化平台，它可以对 Elasticsearch 中的数据进行深度分析和展示。通过构建丰富的仪表板和交互式图表，用户能够以直观的方式对日志数据进行实时监控、问题定位和趋势分析。

（二）ELK工作流程

通常情况下，ELK的工作流程如下

1.在所有需要收集日志的服务器上部署Logstash；或者先将日志进行集中化管理在日志服务器上，在日志服务器上部署 Logstash。

2.Logstash 负责收集来自不同源的日志数据，将这些原始日志数据经过解析、过滤。而后存入 Elasticsearch

3.Elasticsearch 进行索引和存储

4.最后，用户可以通过 Kibana 对这些已结构化的日志数据进行查询、分析并生成可视化报告。并进行前端数据的展示。

（三）ELK的应用价值

ELK Stack的整体应用价值在于

构建集中式日志管理系统，提升运维效率。

实现实时监控和告警，保障系统稳定性。

助力企业实现大数据分析和BI需求，推动业务发展。

二、部署搭建ELK

环境准备

主机名	IP地址	安装服务
node1	192.168.83.30	elasticsearch-6.6.1 elasticsearch-head
node2	192.168.83.40	elasticsearch-6.6.1
http	192.168.83.60	logstash-6.6.1 httpd
kibana	192.1768.83.100	kibana-6.6.1

（一）搭建Elasticsearch集群

环境准备

IP地址	主机名	安装软件
192.168.83.30	node1	elasticsearch-6.6.1
192.168.83.40	node1	elasticsearch-6.6.1

1.安装JDK

最小化安装的需要使用yum install java -y 下载安装java环境，图形界面默认安装的java环境，不过默认的JDK环境是OPENJDK，建议安装jdk

使用脚本进行安装

#!/bin/bash
JAVA=`find  / -name  *jdk*tar*  -exec dirname {} \;`
#找到jdk的压缩包所在目录，并将设置为变量JAVA
JDK=`find  /  -name  *jdk*tar* 2>>/dev/null |awk  -F/ '{print $NF}'`
#找打jdk压缩包的名字，设置为变量JDK
cd ${JAVA}
tar xf ${JDK} -C /usr/local/
#切换到压缩包所在目录，并指定解压到/usr/local/目录下
JDKAPP=`find  /usr/local/   -maxdepth 1  -type  d  |grep  jdk|awk  -F/  '{print  $NF}'`
#找到解压后的目录名，并设置为变量名JDKAPP
ln  -s  /usr/local/${JDKAPP}  /usr/local/jdk
#做软链接，便于shell环境识别命令
cat > /etc/profile.d/jdk.sh  <<EOF
export JAVA_HOME=/usr/local/jdk
export PATH=\$JAVA_HOME/bin:\$PATH
export JRE_HOME=\$JAVA_HOME/jre 
export CLASSPATH=\$JAVA_HOME/lib/:\$JRE_HOME/lib/
EOF
#修改环境变量
echo "请执行  source /etc/profile.d/jdk.sh 命令，刷新文件

2.安装elasticsearch

可以直接使用rmp包进行安装

3.修改配置文件

安装完毕后先修改配置文件再启动服务

两台服务器同样操作

cluster.name: elk-cluster
#设置Elasticsearch集群的名称为 "elk-cluster"。
#这意味着所有带有相同cluster.name配置的Elasticsearch节点将会尝试加入同一个集群。node.name: node1
#指定当前节点的名字为 "node1"，进行区分，便于集群内部管理和监控集群中的各个节点。path.data: /data/elk_data
#设置Elasticsearch数据存储路径，这是Elasticsearch存放索引数据的地方。path.logs: /var/log/elasticsearch
#设置Elasticsearch日志文件的存放路径，Elasticsearch的所有日志都会写入到这个目录下的文件bootstrap.memory_lock: false
#设置为 "false" 表示不锁定内存。
#若设为 true，Elasticsearch将尝试锁定全部分配给它的内存，防止在操作系统层面被交换到磁盘。
#在生产环境中，通常建议开启内存锁定以获得更好的性能和稳定性，但需要有足够的权限，并确保物理内存充足。network.host: 0.0.0.0
#设置为 "0.0.0.0"，表示Elasticsearch节点监听所有可用网络接口上的连接请求，对外提供服务。http.port: 9200
#设置HTTP协议的服务端口为 "9200"discovery.zen.ping.unicast.hosts: ["node1", "node2"]
#设置Elasticsearch集群发现机制的初始节点列表，这里包含了 "node1" 和 "node2" 两个节点地址。
#集群中的节点通过互相ping这些地址来发现彼此并组建集群。
#在启动过程中，节点会尝试联系这个列表中的其他节点来参与集群。

修改完毕后启动服务

4.使用浏览访问节点信息

"name": "node1"
表明此节点在Elasticsearch集群中的名称为"node1"。"cluster_name": "elk-cluster"
表示此节点属于名为"elk-cluster"的Elasticsearch集群。"cluster_uuid": "DyYgn-Z2TumAOAs1xPU7iQ"
是集群的唯一标识符，用于区分不同的Elasticsearch集群。"version"
对象中包含有关Elasticsearch版本的具体信息："number": "6.6.1"
表示Elasticsearch的版本号为6.6.1。
"build_flavor": "default"
说明构建类型为默认版本。
"build_type": "rpm"
指出此版本是通过RPM包构建的，适用于基于RPM包管理系统的Linux发行版，如Red Hat或CentOS。
"build_hash": "1fd8f69"
构建哈希值，用于标识具体的构建版本。
"build_date": "2019-02-13T17:10:04.160291Z"
构建日期和时间，说明此版本是在2019年2月13日17点10分04秒UTC时间构建的。
"build_snapshot": false
表示这是一个正式发布的版本，而非snapshot快照版本。
"lucene_version": "7.6.0"
使用的Lucene搜索引擎内核版本为7.6.0。
"minimum_wire_compatibility_version": "5.6.0"
标明此Elasticsearch版本支持的最低网络通信兼容版本为5.6.0，即与5.6.0及以上版本的节点可以正常通信。
"minimum_index_compatibility_version": "5.0.0"
表示此Elasticsearch版本能够处理的最小索引兼容版本为5.0.0，即可以读取和管理5.0.0及以上版本创建的索引。
"tagline": "You Know, for Search"
Elasticsearch的标志性标语，强调它是一款强大的搜索工具

浏览器访问 http://192.168.10.13:9200/_cluster/health?pretty 查看健康情况

绿色：健康数据和副本全都没有问题
红色：数据都不完整
黄色：数据完整，但副本有问题

5.安装插件

为了更加直观的查看节点信息，可以安装elasticsearch-head插件

Elasticsearch-Head 是一个针对 Elasticsearch 的开源 web 界面管理工具，它提供了简洁直观的方式来查看和管理 Elasticsearch 集群的各项信息，包括但不限于索引列表、文档内容、节点状态、分片分布、集群健康状况、性能统计等。这个工具特别适合于日常的 Elasticsearch 集群运维和监控任务，因为它提供了比原生 REST API 更为友好的图形化界面。

Elasticsearch 在 5.0 版本后，Elasticsearch-head 插件需要作为独立服务进行安装，需要使用npm工具（NodeJS的包管理工具）安装。

安装 Elasticsearch-head 需要提前安装好依赖软件 node 和 phantomjs。

node：是一个基于 Chrome V8 引擎的 JavaScript 运行环境。

phantomjs：是一个基于 webkit 的JavaScriptAPI，可以理解为一个隐形的浏览器，任何基于 webkit 浏览器做的事情，它都可以做到。

[root@node1 opt]#ls
elasticsearch-6.6.1.rpm  elasticsearch-head.tar.gz node-v8.2.1.tar.gz 
phantomjs-2.1.1-linux-x86_64.tar.bz2
#准备好elasticsearch-head、node、phantomjs三个软件包

5.1 安装node

[root@node1 opt]#yum install gcc gcc-c++ make -y
已加载插件：fastestmirror, langpacks
Loading mirror speeds from cached hostfile
....................
#安装编译环境
[root@node1 opt]#tar xf node-v8.2.1.tar.gz
#解压软件包
[root@node1 opt]#cd  node-v8.2.1/
[root@node1 node-v8.2.1]#./configure
......
[root@node1 node-v8.2.1]#make -j 4 && make install
......
#安装过程会有点久，需要耐心等待

5.2 安装phantomjs

[root@node1 opt]#tar xf phantomjs-2.1.1-linux-x86_64.tar.bz2 -C /usr/local/src/
#解压至/usr/local/src/
[root@node1 opt]#cd /usr/local/src/phantomjs-2.1.1-linux-x86_64/bin
[root@node1 bin]#ls
phantomjs
[root@node1 bin]#cp phantomjs /usr/local/bin/
#将命令拷贝到/usr/local/bin/，使phantomjs命令在任何目录下都可以使用

5.3 安装Elasticsearch-head

[root@node1 bin]#cd  /opt
[root@node1 opt]#tar xf elasticsearch-head.tar.gz -C /usr/local/src/
[root@node1 opt]#cd  /usr/local/src/elasticsearch-head/
[root@node1 elasticsearch-head]#ls
Dockerfile                          grunt_fileSets.js  package.json                  README.textile
Dockerfile-alpine                   index.html         package-lock.json             _site
elasticsearch-head.sublime-project  LICENCE            plugin-descriptor.properties  src
Gruntfile.js                        node_modules       proxy                         test
[root@node1 elasticsearch-head]#npm  install
#执行npm  install，安装所有依赖项(node_modules目录下)
......

5.4 修改Elasticsearch主配置文件

http.cors.enabled: true				#开启跨域访问支持，默认为 false
http.cors.allow-origin: "*"			#指定跨域访问允许的域名地址为所有

5.5 启动elasticsearch-head服务

必须在解压后的 elasticsearch-head 目录下启动服务，进程会读取该目录下的 gruntfile.js 文件，否则可能启动失败

5.6 使用浏览器访问

首先使用浏览器访问服务器的9100端口，因为elasticsearch-head 监听的端口是 9100

6.插入索引

curl
#是一个用于发送HTTP请求的命令行工具。-X PUT
#指定了HTTP方法为PUT，这里用于在Elasticsearch中创建或更新文档。'localhost:9200/index-demo1/test/1'
#请求的目标URL，其中：
#localhost:9200 是Elasticsearch服务的主机地址和端口。
#index-demo1 是要操作的索引名称。
#test 是索引下的类型
#1 是待插入或更新的文档ID。?pretty&pretty
#两个pretty参数用于美化返回的结果，使其更易阅读。-H 'Content-Type: application/json'
#指定请求头中的Content-Type为application/json，说明传输的数据格式是JSON。-d '{"user":"lisi","mesg":"hello"}'
#-d 参数后面跟着的是要发送的实际数据，这里是一个简单的JSON对象，
#表示要插入的文档内容，包含两个字段：user 和 mesg。综合起来，这个命令的作用是在Elasticsearch的index-demo1索引中（如果索引不存在，
Elasticsearch将在插入文档时自动创建索引），创建或更新一个ID为1的文档，文档内容
是{"user": "lisi", "mesg": "hello"}。如果索引已存在且ID为1的文档也已存在，则
会更新该文档内容；否则，会创建一个新的文档。

在浏览器访问查看索引信息，可以看见索引默认被分片5个，并且有一个副本(在node2)备份文件

（二）搭建Logstash

环境准备

IP地址

主机名

安装软件

192.168.83.30

http

logstash-6.6.1

httpd

Logstash 一般部署在需要监控其日志的服务器。Logstash部署在 Apache 服务器上，用于收集 Apache 服务器的日志信息并发送到 Elasticsearch。

1.下载http服务

yum install httpd -y

2.安装java环境

同样使用脚本进行安装

3.安装logstash

准备好rpm包，直接使用rpm命令安装

4.测试logstash

Logstash 命令常用选项：
-f：通过这个选项可以指定 Logstash 的配置文件，根据配置文件配置 Logstash 的输入和输出流。
-e：从命令行中获取，输入、输出后面跟着字符串，该字符串可以被当作 Logstash 的配置（如果是空，则默认使用 stdin 作为输入，stdout 作为输出）。
-t：测试配置文件是否正确，然后退出。

命令解释

logstash -e 'input { stdin{} } output { stdout{} }'logstash   、
#这是Logstash命令行工具，用于处理和传输数据。-e
#这是一个命令行参数，允许用户在命令行中直接指定Logstash的配置字符串，而不是从配置文件中读取。配置字符串内容详解：input { stdin{} }
#定义了一个输入插件（input plugin），这里是stdin，表示从标准输入（键盘输入）读取数据。
#当运行这个命令后，可以开始在终端窗口输入数据，这些数据会被Logstash捕获。output { stdout{} }
#定义了一个输出插件（output plugin），这里是stdout，表示将处理过的数据输出到标准输出（屏幕）
#因此，当在终端输入数据后，Logstash会捕获这些数据并将其打印回同一终端窗口

输出信息解释

"host"=> "http"
#这是一个字段，表示产生这条日志事件的源头主机名或地址。@timestamp => 2024-04-10T10:29:14.908Z,
#这是Elasticsearch和Logstash中通用的时间戳字段，记录了该事件发生的时间。
#在这个例子中，事件发生在2024年4月10日10点29分14.908秒（协调世界时，UTC）。"message"=> "192.168.83.40"
#这是日志消息的主要内容，即原始日志文本，此处记录的是一个具体的IPv4地址192.168.83.40。@version"=> "1"
#表示日志事件的版本信息，这里标记为1，是Logstash标准化事件格式的一部分。

在实际生产环境中，Logstash通常会连接到更复杂的数据源（如文件、数据库、网络接口等）并将数据输出到长期存储（如Elasticsearch）或其他系统（如消息队列、文件等）

使用logstash将信息写入Elasticsearch中

通过浏览器可以查看到索引信息

5.定义logstash配置文件

Logstash 配置文件基本由三部分组成：input、output 以及 filter（可选，根据需要选择使用）。
input：表示从数据源采集数据，常见的数据源如Kafka、日志文件等
filter：表示数据处理层，包括对数据进行格式化处理、数据类型转换、数据过滤等，支持正则表达式
output：表示将Logstash收集的数据经由过滤器处理之后输出到Elasticsearch

input {                                  #定义input部分file{                                #指定路径读取日志数据path =>"/var/log/messages"       #监控的日志文件路径type =>"logs"               #设置了输入数据的类型为"logs"，自定义start_position =>"beginning"     
#指定从文件的开头开始读取日志，即使Logstash在此之前已经运行过，也会重新读取整个文件。}   
#注意空格位置，与file插件的f字母对齐，运行在java环境下的程序，对空格与缩进要求比较高
}
output {                                    #定义input部分elasticsearch {                         #将处理过的数据发送到Elasticsearch集群。hosts => ["192.168.83.30:9200"]     #指定了Elasticsearch集群的地址和端口index =>"logs-%{+YYYY.MM.dd}" 
#设置了输出索引的名称，其中 %{+YYYY.MM.dd} 是一个动态时间格式化表达式，
#表示每天创建一个新的索引，索引名称会随着日期的变化而变化，例如："logs-2024.04.11"。}
#同样注意对齐
}

（三）安装kibana

环境准备

IP地址	主机名	安装软件
192.168.83.30	kibana	kibana-6.6.1

将kibanaan安装在节点1机器上，也可以使用新的机器进行安装

修改配置文件

server.port: 5601
#指定了Kibana服务监听的HTTP端口，Kibana Web界面将通过这个端口提供服务server.host: "0.0.0.0"
#配置了Kibana服务绑定的主机地址。这里的"0.0.0.0"意味着Kibana将监听所有的网络接口elasticsearch.hosts: ["http://192.168.83.30:9200"]
#指定了Kibana与之通信的Elasticsearch集群地址和端口。在这个配置下，
#Kibana将连接到IP地址为192.168.83.30的服务器上的Elasticsearch服务，该服务监听9200端口。kibana.index: ".kibana"
#设定Kibana在Elasticsearch中用于存储其元数据、配置信息和索引模式的索引名称，
#默认为".kibana"。这个索引是Kibana的核心配置和状态存储库。

使用浏览器访问elasticsearch服务器

可以看到已经存在了.kibana的索引信息，而后使用浏览器访问kinaba服务器

创建模块索引

使用模块索引查看信息

也可以指定查看字段信息

（四）监控Apache服务器日志

将 Apache 服务器的日志（访问的、错误的）添加到 Elasticsearch 并通过 Kibana 显示

1.开启apache服务

2.添加监听信息

3.加载配置文件

执行：logstash -f /etc/logstash/conf.d/apache.conf命令

启动Logstash，并加载指定的配置文件来运行数据处理管道

使用浏览器访问elasticsearch服务

同样可以在kibana服务中创建模块索引进行查看

三、搭建ELK+filebeat

环境准备

主机名	IP地址	安装服务
node1	192.168.83.30	elasticsearch-6.6.1 elasticsearch-head
node2	192.168.83.40	elasticsearch-6.6.1
http	192.168.83.60	logstash-6.6.1 httpd
kibana	192.1768.83.100	kibana-6.6.1
filebeat	192.168.83.50	filebeat-6.6.1

（一）Filebeat基本介绍

Filebeat是Elastic Beats系列中的一个轻量型日志采集器，它主要用于从服务器上的日志文件中读取数据，并将这些数据转发到Elasticsearch、Logstash或其他输出目的地进行存储、分析和可视化。Filebeat设计目标是在尽可能减少资源消耗的前提下高效地收集日志数据。

使用Filebeat的好处包括：

轻量级：Filebeat占用系统资源少，适合部署在生产环境中进行实时日志收集。

模块化：Filebeat包含许多预配置的模块，可以直接使用这些模块快速地从常见服务（如Apache、Nginx、MySQL等）的日志文件中收集数据。

灵活性：支持自定义配置，可以根据实际需求配置哪些文件需要收集，以及如何解析这些文件中的日志内容。

高效性：Filebeat仅读取和传输日志文件中的新增内容，而非全量读取，降低了对系统IO的影响。

可靠传输：Filebeat具有内置的重试和缓冲机制，确保日志数据能够安全到达目标系统，如Elasticsearch或Logstash。

扩展性：可以与Elasticsearch、Logstash、Kibana等工具紧密结合，形成一套完整的日志收集、处理、存储和分析解决方案。

（二）部署ELK+filebeat

1.下载安装filebeat

2.修改yml文件

该配置文件主要分为三个部分

filebeat.inputs:   #输入配置- type: log        #Filebeat被配置为从磁盘上的日志文件收集数据，类型(type)为"log"。
enabled: true      #表示启用这个输入模块。
paths              #列表中指定了两个日志文件路径，表示从这两个文件中获取日志信息
／var／log／messages  #这是Linux系统中的系统日志文件。
／var／log／*.log     #这个通配符路径会匹配／var／log目录下所有以.log结尾的日志文件。fields:            #附加字段
#在每个收集到的日志事件上添加额外的上下文信息，方便在Elasticsearch中做进一步的筛选和分析service_name: filebeat_server
#定义了一个名为service_name的字段，并赋值为filebeat_server。
log_type: log     
#定义了一个名为log_type的字段，并赋值为log。
service_id: 192.168.83.50
#定义了一个名为service_id的字段，并赋值为192.168.83.50。#输出配置。
output.logstash:     
#表示输出类型为Logstash。Filebeat被配置将收集到的数据发送到Logstash
hosts: ["192.168.83.60:5044"] 
#指定了Logstash服务器的地址和监听端口，Filebeat将会把收集到的日志事件推送到这个地址。

3.修改logstash服务配文文件

input {
#输入部分，这部分配置了Logstash接收来自Beats（如Filebeat、Metricbeat等）的数据输入方式：beats {port => "5044"}
}
#在这段配置中，Logstash监听5044端口，等待接收来自Beats客户端的数据。
#这意味着Filebeat等Beats产品在收集到日志或监控数据后，会通过TCP协议将数据发送到这个端口。output {
#输出配置： 这部分配置了Logstash将处理完的数据输出到Elasticsearch：elasticsearch {hosts => ["192.168.83.30:9200"]
#指定输出的elasticsearch集群的IP地址及端口号index => "%{[fields][service_name]}-%{+YYYY.MM.dd}"
#index字段指定了Elasticsearch中索引的名称格式，采用动态索引命名策略，
#索引名称由fields字段下的service_name字段值与当前日期（格式为YYYY.MM.dd）拼接而成}
}
#在这段配置中，Logstash将处理后的数据发送到Elasticsearch集群，集群地址为192.168.83.30，
#端口为9200。这意味着每条数据将根据其携带的service_name字段值和数据处理当天的日期生成
#唯一的索引名称。#综上所述，Logstash通过这段配置接收来自Beats的数据，并将其输出到指定的Elasticsearch集群，
#同时根据数据中的service_name字段值按天划分索引，便于后期日志管理和分析。