ELK环境部署（elk使用场景）

一、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、关注我的微信公众号-平凡人笔记 同名文章 里面是代码形式