一、ELK 是什么?
ELK 实际上是三个工具的集合,Elasticsearch + Logstash + Kibana,这三个工具组合形成了一套实用、易用的监控架构,很多公司利用它来搭建可视化的海量日志分析平台。
-
ElasticSearch
ElasticSearch 是一个基于 Lucene 的搜索服务器。它提供了一个分布式多用户能力的全文搜索引擎,基于 RESTful web 接口。Elasticsearch 是用 Java 开发的,并作为 Apache 许可条款下的开放源码发布,是当前流行的企业级搜索引擎。设计用于云计算中,能够达到实时搜索,稳定,可靠,快速,安装使用方便。 -
Logstash
Logstash 是一个用于管理日志和事件的工具,你可以用它去收集日志、转换日志、解析日志并将他们作为数据提供给其它模块调用,例如搜索、存储等。 -
Kibana
Kibana 是一个优秀的前端日志展示框架,它可以非常详细的将日志转化为各种图表,为用户提供强大的数据可视化支持。
二、ELK 有何优势?
-
强大的搜索功能,elasticsearch 可以以分布式搜索的方式快速检索,而且支持 DSL 的语法来进行搜索,简单的说,就是通过类似配置的语言,快速筛选数据。
-
完美的展示功能,可以展示非常详细的图表信息,而且可以定制展示内容,将数据可视化发挥的淋漓尽致。
-
分布式功能,能够解决大型集群运维工作很多问题,包括监控、预警、日志收集解析等。
三、ELK 一般用来做啥?
ELK 组件在海量日志系统的运维中,可用于解决:
-
分布式日志数据集中式查询和管理
-
系统监控,包含系统硬件和应用各个组件的监控
-
故障排查
-
安全信息和事件管理
-
报表功能
ELK 组件在大数据运维系统中,主要可解决的问题如下:
-
日志查询,问题排查,上线检查
-
服务器监控,应用监控,错误报警,Bug 管理
-
性能分析,用户行为分析,安全漏洞分析,时间管理
四、ELK 搭建
1、版本介绍
这里笔者采用 win11 进行搭建,其余版本如下
elasticsearch-7.12.0
kibana-7.12.0-windows-x86_64
logstash-7.12.0-windows-x86_64
最后奉上网址:下载中心 - Elastic 中文社区 (elasticsearch.cn)
看清楚喽,是 windows 下,版本要一致
2、启动 elasticsearch
ok,下载解压之后开始我们的文件配置,首先打开我们的 elasticsearch 文件夹,进入 config 文件夹,编辑 elasticsearch.yml 文件,在文件中引入
cluster.name: my-application #集群名:类似于数据库名
path.data: D:/log #数据目录
path.logs: D:/log #日志目录
node.name: node-1
node.attr.rack: r1
cluster.initial_master_nodes: node-1
node.master: true
node.data: true
bootstrap.memory_lock: true
network.host: 0.0.0.0
http.compression: true
transport.tcp.compress: true
indices.query.bool.max_clause_count: 10240
xpack.watcher.execution.default_throttle_period: 1s
http.cors.enabled: true
http.cors.allow-origin: "*"
保存配置后,启动 ElasticSearch,进入 bin 目录,elasticsearch.bat 启动成功,打开浏览器访问:http://127.0.0.1:9200,出现下面界面,说明 ElasticSearch 启动成功
3、安装 elasticsearch-head 插件
安装 elasticsearch-head 插件(这是 elasticsearch 的一个 web 端管理插件,通过管理插件可以查看 ElsasticSearch 的客户端工具,由 node.js 构建的前端工程。)git 地址:https://github.com/mobz/elasticsearch-head 。拉去过慢,可以使用 gitclone 进行拉取 https://gitclone.com/github.com/mobz/elasticsearch-head
拉取成功后,可以使用进入使用 vscode 打开或者其他编译器,打开终端,
npm install
npm run start
这里 npm install 可能会有一个包 npm install 时 phantomjs 包下载不下来,单独下载就可以批量下载时,
npm install phantomjs@2.1.1 --ignore-scripts
然后再重复上述操作即可,启动之后,访问 http://localhost:9100,界面如下:
4、启动 logstash
进入 logstash/config 中新建配置文件 logstash.conf, 内容如下:
input {
file {
type => "logstash"
//path用来指定要加载到logstash中的文件目录,
path => ["D:/logs/*/error/*.log","D:/logs/*/warn/*.log","D:/logs/*/info/*.log"]
start_position => "beginning"
}
}
output {
elasticsearch {
hosts => ["127.0.0.1:9200"]
//index用来指定在es中自动创建的索引库
index => "logstash-%{+YYYY.MM.dd}"
}
stdout {
//以JSON的形式传输到es中
codec => json_lines
}
}
启动 logstash, 进入 logstash/bin 执行:.\logstash.bat -f …/config/logstash.conf
以下界面就是启动成功了
5、启动 Kibana
进入 Kibana 下 config 下的 kibana.yml 修改为:
server.port: 5601
server.host: "127.0.0.1"
elasticsearch.hosts: ["http://localhost:9200"]
xpack.encryptedSavedObjects.encryptionKey: encryptedSavedObjects12345678909876543210
xpack.security.encryptionKey: encryptionKeysecurity12345678909876543210
xpack.reporting.encryptionKey: encryptionKeyreporting12345678909876543210
xpack.reporting.capture.browser.chromium.disableSandbox: true
启动 Kibana,进入 Kibana/bin 执行:kibana.bat,打开浏览器访问 http://localhost:5601/
五、与 springboot 集成
- 引入依赖
<!--logstash-->
<dependency>
<groupId>net.logstash.logback</groupId>
<artifactId>logstash-logback-encoder</artifactId>
<version>5.3</version>
</dependency>
<!--web-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
</dependency>
yml 配置文件中写入
spring.application.name=myElk
server.port = 8081
在 resource 下创建 logback-spring.xml 文件,内容如下:
<?xml version="1.0" encoding="UTF-8"?>
<configuration>
<property name="LOG_CONTEXT_NAME" value="log"/>
<!--定义日志文件的存储地址 勿在 LogBack 的配置中使用相对路径-->
<property name="LOG_HOME" value="D:/logs" />
<!-- 定义日志上下文的名称 -->
<contextName>${LOG_CONTEXT_NAME}</contextName>
<!-- 控制台输出 -->
<!--<appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
<encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
<!–格式化输出:%d表示日期,%thread表示线程名,%-5level:级别从左显示5个字符宽度%msg:日志消息,%n是换行符–>
<pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %highlight(%-5level) %cyan(%logger{50}:%L) - %msg%n</pattern>
<charset>utf-8</charset>
</encoder>
<filter class="ch.qos.logback.classic.filter.ThresholdFilter">
<level>INFO</level>
</filter>
</appender>-->
<!-- 彩色日志依赖的渲染类 -->
<conversionRule conversionWord="clr" converterClass="org.springframework.boot.logging.logback.ColorConverter" />
<conversionRule conversionWord="wex" converterClass="org.springframework.boot.logging.logback.WhitespaceThrowableProxyConverter" />
<conversionRule conversionWord="wEx" converterClass="org.springframework.boot.logging.logback.ExtendedWhitespaceThrowableProxyConverter" />
<!-- 彩色日志格式 -->
<property name="CONSOLE_LOG_PATTERN" value="${CONSOLE_LOG_PATTERN:-%clr(%d{yyyy-MM-dd HH:mm:ss.SSS}){faint} %clr(${LOG_LEVEL_PATTERN:-%5p}) %clr(${PID:- }){magenta} %clr(---){faint} %clr([%15.15t]){faint} %clr(%-40.40logger{39}){cyan} %clr(:){faint} %m%n${LOG_EXCEPTION_CONVERSION_WORD:-%wEx}}"/>
<!--1. 输出到控制台-->
<appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
<!--此日志appender是为开发使用,只配置最底级别,控制台输出的日志级别是大于或等于此级别的日志信息-->
<filter class="ch.qos.logback.classic.filter.ThresholdFilter">
<level>INFO</level>
</filter>
<encoder>
<Pattern>${CONSOLE_LOG_PATTERN}</Pattern>
<!-- 设置字符集 -->
<charset>UTF-8</charset>
</encoder>
</appender>
<!--info日志统一输出到这里-->
<appender name="file.info" class="ch.qos.logback.core.rolling.RollingFileAppender">
<Prudent>true</Prudent>
<rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
<!--日志文件输出的文件名,按小时生成-->
<FileNamePattern>${LOG_HOME}/%d{yyyy-MM-dd}/info/info.%d{yyyy-MM-dd-HH}.%i.log</FileNamePattern>
<!--日志文件保留天数-->
<MaxHistory>30</MaxHistory>
<timeBasedFileNamingAndTriggeringPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP">
<!-- 除按日志记录之外,还配置了日志文件不能超过10M(默认),若超过10M,日志文件会以索引0开始, -->
<maxFileSize>10MB</maxFileSize>
</timeBasedFileNamingAndTriggeringPolicy>
</rollingPolicy>
<encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
<!--格式化输出:%d表示日期,%thread表示线程名,%-5level:级别从左显示5个字符宽度 %method 方法名 %L 行数 %msg:日志消息,%n是换行符-->
<pattern> %d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{56}.%method:%L - %msg%n</pattern>
<charset>utf-8</charset>
</encoder>
<!-- 此日志文件只记录info级别的 -->
<filter class="ch.qos.logback.classic.filter.LevelFilter">
<level>INFO</level>
<onMatch>ACCEPT</onMatch>
<onMismatch>DENY</onMismatch>
</filter>
</appender>
<!--错误日志统一输出到这里-->
<appender name="file.error" class="ch.qos.logback.core.rolling.RollingFileAppender">
<Prudent>true</Prudent>
<rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
<!--日志文件输出的文件名,按天生成-->
<FileNamePattern>${LOG_HOME}/%d{yyyy-MM-dd}/error/error.%d{yyyy-MM-dd}.%i.log</FileNamePattern>
<!--日志文件保留天数-->
<MaxHistory>30</MaxHistory>
<timeBasedFileNamingAndTriggeringPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP">
<!-- 除按日志记录之外,还配置了日志文件不能超过10M(默认),若超过10M,日志文件会以索引0开始, -->
<maxFileSize>10MB</maxFileSize>
</timeBasedFileNamingAndTriggeringPolicy>
</rollingPolicy>
<encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
<!--格式化输出:%d表示日期,%thread表示线程名,%-5level:级别从左显示5个字符宽度 %method 方法名 %L 行数 %msg:日志消息,%n是换行符-->
<pattern> %d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{56}.%method:%L - %msg%n</pattern>
<charset>utf-8</charset>
</encoder>
<!-- 此日志文件只记录error级别的 -->
<filter class="ch.qos.logback.classic.filter.LevelFilter">
<level>ERROR</level>
<onMatch>ACCEPT</onMatch>
<onMismatch>DENY</onMismatch>
</filter>
</appender>
<!--warn日志统一输出到这里-->
<appender name="file.warn" class="ch.qos.logback.core.rolling.RollingFileAppender">
<Prudent>true</Prudent>
<rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
<FileNamePattern>${LOG_HOME}/%d{yyyy-MM-dd}/warn/warn.%d{yyyy-MM-dd}.%i.log</FileNamePattern>
<!--日志文件保留天数-->
<MaxHistory>30</MaxHistory>
<timeBasedFileNamingAndTriggeringPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP">
<!-- 除按日志记录之外,还配置了日志文件不能超过10M(默认),若超过10M,日志文件会以索引0开始, -->
<maxFileSize>10MB</maxFileSize>
</timeBasedFileNamingAndTriggeringPolicy>
</rollingPolicy>
<encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
<!--格式化输出:%d表示日期,%thread表示线程名,%-5level:级别从左显示5个字符宽度 %method 方法名 %L 行数 %msg:日志消息,%n是换行符-->
<pattern> %d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{56}.%method:%L - %msg%n</pattern>
<charset>utf-8</charset>
</encoder>
<!-- 此日志文件只记录warn级别的 -->
<filter class="ch.qos.logback.classic.filter.LevelFilter">
<level>WARN</level>
<onMatch>ACCEPT</onMatch>
<onMismatch>DENY</onMismatch>
</filter>
</appender>
<!--输出到logstash的appender-->
<appender name="LOGSTASH" class="net.logstash.logback.appender.LogstashTcpSocketAppender">
<!--可以访问的logstash日志收集端口-->
<destination>127.0.0.1:4560</destination>
<encoder charset="UTF-8" class="net.logstash.logback.encoder.LogstashEncoder"/>
</appender>
<!-- 日志输出级别 -->
<root level="DEBUG">
<appender-ref ref="STDOUT" />
<appender-ref ref="file.error" />
<appender-ref ref="file.info" />
<appender-ref ref="file.warn" />
<appender-ref ref="LOGSTASH"/>
</root>
</configuration>
ok,接下来创建测试类
package com.example.myelk.controller;
import lombok.extern.slf4j.Slf4j;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import java.util.Date;
@Slf4j
@RestController
@RequestMapping("/test")
public class TestController {
@GetMapping("/success")
public String test (){
log.info("###test:{}", new Date());
return "success";
}
}
访问接口后,在 Kibana 中创建索引 logstash-*, 时间选择 I dont want to use te time filter 就行
点击 create index pattern 即可创建成功
创建成功之后,可以选择上方搜索框,搜索 Discover, 可以选择 table 或者 json
IDEA 中控制台打印的原日志内容是下面内容。Logstash 作用就是把下面内容转换为上面 Elasticsearch 存储的内容。在中间做了数据格式转换,收集数据放入 Elasticsearch 中的工作。