一、ELK原理介绍
ELK可以将我们的系统日志、网站日志、应用系统日志等各种日志进行收集、过滤、清洗、然后进行集中存放并可用于实时检索、分析
1、Elasticsearch
Elasticsearch是一个实时的分布式搜索和分析引擎，它可以用于全文搜索，结构化搜索以及分析，采用Java语言编写 Elasticsearch 是一个分布式的 RESTful 风格的搜索和数据分析引擎
实时搜索，实时分析
分布式架构、实时文件存储，并将每一个字段都编入索引
文档导向，所有的对象全部是文档
高可用性，易扩展，支持集群（Cluster）、分片和复制（Shards和Replicas）
接口友好，支持JSON
可以扩展到上百台服务器，轻松处理PB级结构化或非结构化数据
基于Apache Lucene(TM)的开源搜索引擎，让全文搜索变得简单
2、 Logstash
Logstash是一款轻量级的、开源的日志收集处理框架，它可以方便的把分散的、多样化的日志搜集起来，并进行自定义过滤分析处理，然后传输到指定的位置，比如某个服务器或者文件。Logstash采用JRuby语言编写，而Jruby的基础又是Java开发而来
input：数据收集
filter：数据加工，如过滤，改写等
output：数据输出
Shipper：主要用来收集日志数据，负责监控本地日志文件的变化，及时把日志文件的最新内容收集起来，然后经过加工、过滤，输出到Broker
Broker：相当于日志Hub，用来连接多个Shipper和多个Indexer。
Indexer：从Broker读取文本，经过加工、过滤，输出到指定的介质（可以是文件、网络、elasticsearch等）中。Redis服务器是logstash官方推荐的broker，这个broker起数据缓存的作用，通过这个缓存器可以提高Logstash shipper发送日志到Logstash indexer的速度，同时避免由于突然断电等导致的数据丢失。可以实现broker功能的还有很多软件，例如kafka等。这里需要说明的是，在实际应用中，LogStash自身并没有什么角色，只是根据不同的功能、不同的配置给出不同的称呼而已，无论是Shipper还是Indexer，始终只做前面提到的三件事。这里需要重点掌握的是logstash中Shipper和Indexer的作用，因为这两个部分是logstash功能的核心
3、Kibana
Kibana是一个开源的数据分析可视化平台。使用Kibana可以为Logstash和ElasticSearch提供的日志数据进行高效的搜索、可视化汇总和多维度分析，还可以与Elasticsearch搜索引擎之中的数据进行交互。它基于浏览器的界面操作可以快速创建动态仪表板，实时监控ElasticSearch的数据状态与更改
一般都是在需要收集日志的所有服务上部署logstash，作为logstash shipper用于监控并收集、过滤日志，接着，将过滤后的日志发送给Broker，然后，Logstash Indexer将存放在Broker中的数据再写入Elasticsearch，Elasticsearch对这些数据创建索引，最后由Kibana对其进行各种分析并以图表的形式展示。
4、ELK工作流程说明
ELK工作流程说明：有些时候，如果收集的日志量较大，为了保证日志收集的性能和数据的完整性，logstash shipper和logstash indexer之间的缓冲器（Broker）也经常采用kafka来实现。
二、ELK环境部署
1、资源包下载地址
https://artifacts.elastic.co/downloads/logstash/logstash-7.5.1.tar.gz
https://artifacts.elastic.co/downloads/kibana/kibana-7.5.1-darwin-x86_64.tar.gz
https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-7.5.1-darwin-x86_64.tar.gz
2、Elasticsearch
验证启动结果
3、Shipper
4、Kibana
5、Indexer
indexer-logstash.conf
input {
redis {
type => "logback" host => "192.168.142.131" # redis主机地址 port => 6379 # redis端口号 db => 8 # redis数据库编号 data_type => "channel" # 使用发布/订阅模式 key => "sb-logback" # 发布通道名称 }
redis {
type => "nginx" host => "192.168.142.131" port => 6379
db => 8
data_type => "channel" key => "nginx" }
}
filter {
if [type] == "logback" {
# logback输出的日志 grok {
match => { "message" => "%{TIMESTAMP_ISO8601:time} \[%{NOTSPACE:threadName}\] %{LOGLEVEL:level} %{DATA:logger} %{NOTSPACE:applicationName} -(?:.*=%{NUMBER:timetaken}ms|)"}
}
}
if [type] == "nginx" {
# nginx输出的日志 grok {
match => { "message" => "%{IPV4:ip} \- \- \[%{HTTPDATE:time}\] \"%{NOTSPACE:method} %{URIPATH:requestUrl} HTTP/%{NUMBER:httpVersion}\" %{NUMBER:httpStatus} %{NUMBER:bytes} \"%{DATA:referer}\" \"%{DATA:agent}\""}
}
}
}
output {
stdout {}
if [type] == "logback" {
elasticsearch {
hosts => "localhost:9200" index => "logback" }
}
if [type] == "nginx" {
elasticsearch {
hosts => "localhost:9200" index => "nginx" }
}
}
6、shipper启动方式同上
shipper-logstash.conf
input {
file {
type => "logback" path => [
# 这里填写需要监控的文件 "/log/sb-log.log" ]
}
file {
type => "nginx" path => [
# 这里填写需要监控的文件 "/var/log/nginx/access.log" ]
}
}
output {
stdout { }
# 输出到redis if [type] == "logback" {
redis {
host => "localhost" # redis主机地址 port => 6379 # redis端口号 db => 8 # redis数据库编号 data_type => "channel" # 使用发布/订阅模式 key => "sb-logback" # 发布通道名称 }
}
if [type] == "nginx" {
redis {
host => "localhost" # redis主机地址 port => 6379 # redis端口号 db => 8 # redis数据库编号 data_type => "channel" # 使用发布/订阅模式 key => "nginx" # 发布通道名称 }
}}
Redis 的配置 data_type 可选值有 channel 和 list 两个。
channel 是 Redis 的发布/订阅通信模式，而 list 是 Redis 的队列数据结构，两者都可以用来实现系统间有序的消息异步通信。
channel 相比 list 的好处是，解除了发布者和订阅者之间的耦合。
举个例子，一个 Indexer 在持续读取 Redis 中的记录，现在想加入第二个 Indexer，如果使用 list，就会出现上一条记录被第一个 Indexer 取走，而下一条记录被第二个 Indexer 取走的情况，两个 Indexer 之间产生了竞争，导致任何一方都没有读到完整的日志。channel 就可以避免这种情况。
这里 Shipper 角色的配置文件和下面将要提到的 Indexer 角色的配置文件中都使用了 channel。
三、ELK使用场景实战
1、启动springboot项目
a、项目地址
https://github.com/mengfanxiao/spring-tutorial
b、配置说明
logback-spring.xml
<?xml version="1.0" encoding="UTF-8"?><configuration debug="false"> <contextName>Logback For demo Mobile</contextName> <property name="LOG_HOME" value="/log" /> <springProperty scope="context" name="appName" source="spring.application.name" defaultValue="localhost" /> <!-- 控制台输出 --> <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender"> <!-- encoder默认配置为PartternLayoutEncoder --> <encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder"> <pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{80} -%msg%n</pattern>
</encoder> <target>System.out</target>
</appender> <!-- 按照每天生成日志文件 --> <appender name="ROLLING_FILE" class="ch.qos.logback.core.rolling.RollingFileAppender"> <file>/log/sb-log.log</file> <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy"> <!--日志文件输出的文件名 ,每天保存（侧翻）一次 --> <FileNamePattern>${LOG_HOME}/%d{yyyy-MM-dd}.log</FileNamePattern> <!--日志文件保留天数 --> <MaxHistory>180</MaxHistory>
</rollingPolicy> <encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder"> <pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{25} ${appName} -%msg%n</pattern>
</encoder> <!--日志文件最大的大小 --> <triggeringPolicy class="ch.qos.logback.core.rolling.SizeBasedTriggeringPolicy"> <MaxFileSize>20MB</MaxFileSize>
</triggeringPolicy>
</appender> <!-- 全局，控制台遇到INFO及以上级别就进行输出 --> <root level="INFO"> <appender-ref ref="STDOUT" /> <appender-ref ref="ROLLING_FILE" />
</root></configuration>
访问测试接口成功
2、启动nginx
docker run -d --name nginx2 -p 7001:80 -v /Users/mengfanxiao/nginx/conf/nginx.conf:/etc/nginx/nginx.conf -v /Users/mengfanxiao/nginx/logs:/var/log/nginx -v /Users/mengfanxiao/nginx/conf.d/default.conf:/etc/nginx/conf.d/default.conf nginx
nginx.conf
user nginx;worker_processes 1;error_log /var/log/nginx/error.log warn;pid /var/run/nginx.pid;events {
worker_connections 1024;}
http {
include /etc/nginx/mime.types; default_type application/octet-stream; log_format main '$remote_addr - $remote_user [$time_local] "$request" '
'$status $body_bytes_sent "$http_referer" ' '"$http_user_agent" "$http_x_forwarded_for"'; access_log /var/log/nginx/access.log main; sendfile on; #tcp_nopush on; keepalive_timeout 65; #gzip on; include /etc/nginx/conf.d/*.conf; upstream mysvr {
server 192.168.50.100:8080; }
server {
keepalive_requests 120; #单连接请求上限次数。 listen 80; #监听端口 server_name 127.0.0.1; #监听地址 location ~*^.+$ { #请求的url过滤，正则匹配，~为区分大小写，~*为不区分大小写。
#root path; #根目录
#index vv.txt; #设置默认页 proxy_pass http://mysvr; #请求转向mysvr 定义的服务器列表 }
}
}
通过nginx访问测试接口成功
3、查看kibana
a、分别配置这2个数据源的索引
这里是已创建好的索引
b、通过索引查询数据
四、整体流程简要梳理
1、访问
a、访问nginx=>springboot 生成nginx日志和spirngboot日志文件
b、直接反问springboot 生成springboot日志文件
2、shipper logstash 读取到 nginx和springboot日志文件
3、将日志输出到redis中
4、indexer logstash 从redis中读取 输出给Elasticsearch
5、kibana读取Elasticsearch数据显示出来
备注：上述代码为截图形式 若想获取源码 有2种方式
1、下载demo项目 里面有
2、关注我的微信公众号-平凡人笔记 同名文章 里面是代码形式
网友评论