前言

本篇文章主要是说ElasticSearch对Logstash、FileBeat、Kibana整合形成ELK的架构，为什么需要整合这个架构呢？一个很重要的原因就是我们开发过程中有相当多的日志需要进行查看，如果我们要查找一个问题需要到多台服务器进行查看那是相当麻烦的，因为有了这个架构的出现

本篇依赖的ElasticSearch、Kibana安装可以参考我之前写的另一篇文章，我已经详细的介绍了这两个软件的安装：https://blog.csdn.net/zxc_user/article/details/128666834

ELK常用的两种架构

经典的ELK架构

首先通过beats收集数据，然后发到logstash，接着再存到elasticsearch，最后再使用Kibana进行查看，但是如果beats收集数据过快把logstash服务打挂掉，那么有可能会导致数据的丢失，所以我们一遍会在logstash前面加一层Mq，利用MQ的流量削峰功能可以解决这个问题

MQ模式的架构

核心组件的介绍

Logstash

什么是Logstash呢？官方称之为管道，简单理解就是一个数据经过Logstash然后进行一定的处理再转发到其他数据源(常用的就是放到elasticsearch中)

官网地址为：https://www.elastic.co/cn/logstash/

Logstash数据传输原理

主要设计以下三个概念

input：数据输入源，也就是源数据，可以来自于tomcat,nginx,java应用等

filter: 过滤器，logstash对数据的一些中间处理，可以有也可以没有，logstash也会有一些默认的处理

output: 数据输出地，把来源数据处理完成以后要存储的位置

官方提供了大量的插件，地址如下

input 插件：https://www.elastic.co/guide/en/logstash/7.17/input-plugins.html

output插件：https://www.elastic.co/guide/en/logstash/7.17/output-plugins.html

filter插件：https://www.elastic.co/guide/en/logstash/7.17/filter-plugins.html

Codec 插件： https://www.elastic.co/guide/en/logstash/7.17/codec-plugins.html，也就是定义数据输入的格式

Logstash安装

这里只讨论linux的安装，windows版本的请自行百度，注，我这里用的是7.3.13版本的

官网地址： https://www.elastic.co/guide/en/logstash/7.17/installing-logstash.html

下载地址：https://www.elastic.co/cn/downloads/past-releases#logstash

下载完以后把 logstash-7.17.3-linux-x86_64.tar.gz 上传到你的服务器上，如下

然后执行解压命令，如下

tar -zxvf logstash-7.17.3-linux-x86_64.tar.gz

接着进入到解压目录，执行下面的命令，可以先验证下怎么用

#-e选项表示，直接把配置放在命令中，这样可以有效快速进行测试
bin/logstash -e 'input { stdin { } } output { stdout {} }'

这个启动可能比较久，要耐心等一等，记得要配置JAVA_HOME，也就是需要jdk，看到下面的日志就成功了，如下

注意启动日志：DEPRECATION: The use of JAVA_HOME is now deprecated and will be removed starting from 8.0. Please configure LS_JAVA_HOME instead. 意思是说建议你用LS_JAVA_HOME代替JAVA_HOME，以后版本可能会用，不过至少在7,13版本JAVA_HOME还是可以用的

启动完成后随便输点如下，如果返回如下的信息，就说明是成功了

上面是最简单的文件模式，接下来我们验证下codec，使用json格式化数据进行存储，这些都是单行模式的

# single line
##这个就是上面测试那个
bin/logstash -e "input{stdin{codec=>line}}output{stdout{codec=> rubydebug}}"
##这个需要json格式
bin/logstash -e "input{stdin{codec=>json}}output{stdout{codec=> rubydebug}}"

多行模式

上面的数据输入都是单行的，但是如果我们有以下的错误的时候仍然用单行模式会产生什么样的情况呢，比如我们有这样的异常日志

Exception in thread "main" java.lang.NullPointerException
        at com.example.zxc.User.getTitle(Book.java:25)
        at com.example.zxc.Author.getBookTitles(Author.java:45)
        at com.example.zxc.Bootstrap.main(Bootstrap.java:23)

如果用单行模式输入会变成这样，如下

它会变成多条数据展示，很明显这样是有问题的，因为我们要查日志的时候异常肯定是要在一起的，这样压根就查不了，所以我们需要用多行模式，这就是其中的一个插件使用

Codec Plugin —— Multiline

设置参数:

pattern: 设置行匹配的正则表达式，就是正则模式

what : 如果匹配成功，那么匹配行属于上一个事件还是下一个事件，取值有 previous / next

negate : 是否对pattern结果取反，取值有true / false

首先创建multiline-test.conf文件，内容如下

input {
  stdin {
    codec => multiline {
      pattern => "^\s"
      what => "previous"
    }
  }
}

filter {}

output {
  stdout { codec => rubydebug }
}

主要是input里面的内容， ^\s表示前面不是空格，如果是空格那么该内容属于上面的previous

然后启动logstash时需要指定文件路径，如下，通过-f指定你文件所在的位置

 bin/logstash -f test/multiline-test.conf

这个时候再输入上面的数据，返回如下，它变成了一行数据，如图

它变成了一行数据

Exception in thread "main" java.lang.NullPointerException
        at com.example.zxc.User.getTitle(Book.java:25)
        at com.example.zxc.Author.getBookTitles(Author.java:45)
        at com.example.zxc.Bootstrap.main(Bootstrap.java:23)
注意输入以上内容回车后再随便输入其他文本才会出来，这跟这个插件有关系，也就是你当次回车是上次的内容

从文件读取数据

这个是Input Plugin —— File插件提供的功能，它可以从一个文件读取数据然后进行存储，下面举一个从文件获取数据然后转发到elasticsearch的例子

首先，准备一个zxc.csv文件，内容如下

1,zxc,28
2,ldh,29
3,cbx,30
4,hzh,31

其次，启动elasticsearch，记得用普通用户启动，参考上篇文章

再次，启动Kibana,也是普通用户启动，参考上篇文章

访问/zxcfile索引，目前是没数据的，如下

##创建索引
PUT /zxcfile
##查询数据
GET /zxcfile/_search

然后创建zxc-file.conf文件，内容如下

input {
  file {
    path => "/opt/logstash-7.17.3/zxc.csv"
    start_position => "beginning"
    sincedb_path => "/dev/null"
  }
}
filter {
  csv {
    separator => ","
    columns => ["id","name","age"]
  }

  mutate {
    remove_field => ["path", "host","@timestamp","message"]
  }
}
output {
   elasticsearch {
     hosts => "http://localhost:9200"
     index => "zxcfile"
     document_id => "%{id}"
   }
  stdout {}
}

插件使用参考：https://www.elastic.co/guide/en/logstash/7.17/plugins-inputs-file.html

使用命令启动

bin/logstash -f test/zxc-file.conf

启动完以后它输到控制台，同时传输到elasticsearch中一份，查看如下

注意文件插入就成功了

注：这里启动Kibana的时候一直报异常，刚开始以为是内存问题，但是改了内存也不行，后来发现是我之前测试的时候elasticsearch集群信息有了Kibana的信息，要么把elasticsearch节点的data数据清掉，要么用一个新的单节点的elasticsearch来给Kibana链接，因为我这里为了方便，只是启动了一台elasticsearch，下面的操作一直失败，所以我是直接搞了一台单独的elasticsearch了

删除Kibana在elastic search中的信息
在kibana服务器中执行，目标es集群删除信息
curl -X DELETE http://localhost:9200/.kibana*

Beats

Beats也是一个采集日志的组件，不过它比Logstash轻量，使用go语言开发的，但是它没有filter功能，只有输入和输出

Beats包含了很多内容，但是我们最经常用的还是FileBeat

Filebeat安装

文档地址：https://www.elastic.co/guide/en/beats/libbeat/7.17/index.html

安装地址：https://www.elastic.co/guide/en/beats/filebeat/7.17/filebeat-installation-configuration.html

下载地址：https://www.elastic.co/cn/downloads/past-releases#filebeat

我这里还是采用了7.17.3版本，下载完filebeat-7.17.3-linux-x86_64.tar.gz文件，上传到你的服务器，如下,注意，版本号要跟Kibana一致，否则仪表盘可能会有问题

先进行解压

tar -zxvf filebeat-7.17.3-linux-x86_64.tar.gz

然后编辑配置，修改 filebeat.yml，如下，（192.168.0.90为我本地虚拟机地址）

output.elasticsearch:
  hosts: ["192.168.0.90"]
setup.kibana:
  host: "192.168.0.90:5601"

启用和配置数据收集模块

# 查看可以模块列表
./filebeat modules list

#启用nginx模块
./filebeat modules enable nginx
#如果需要更改nginx日志路径,修改modules.d/nginx.yml
- module: nginx
  access:
    var.paths: ["/var/log/nginx/access.log*"]

#启用 Logstash 模块
./filebeat modules enable logstash
#在 modules.d/logstash.yml 文件中修改设置
- module: logstash
  log:
    enabled: true
    var.paths: ["/opt/logstash-7.17.3/logs/*.log"]

还可以查看有哪些使用命令，不用特意去背

./filebeat modules --help

./filebeat modules list

可以看到filebeat支持很多收集日志，如下

开启模块会到modules.d 里面进行配置文件

启动 Filebeat

# setup命令加载Kibana仪表板。 如果仪表板已经设置，则忽略此命令。 
./filebeat setup
# 启动Filebeat
./filebeat -e

实际案例，采集Tomcat日志

首先，你的服务器得安装一个tomcat,这个就不说了

创建配置文件filebeat-logstash.yml

创建配置文件filebeat-logstash.yml，内容如下

vim filebeat-logstash.yml
chmod 644 filebeat-logstash.yml
#因为Tomcat的web log日志都是以IP地址开头的，所以我们需要修改下匹配字段。
# 不以ip地址开头的行追加到上一行
filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /home/es/apache-tomcat-8.5.33/logs/*access*.*
  multiline.pattern: '^\\d+\\.\\d+\\.\\d+\\.\\d+ '
  multiline.negate: true
  multiline.match: after

output.logstash:
  enabled: true
  hosts: ["192.168.0.90:5044"]

启动FileBeat，并指定使用指定的配置文件

./filebeat -e -c filebeat-logstash.yml

配置Logstash接收FileBeat收集的数据并打印

vim test/filebeat-console.conf
# 配置从FileBeat接收数据
input {
    beats {
      port => 5044
    }
}

output {
    stdout {
      codec => rubydebug
    }
}

启动logstash

# reload.automatic：修改配置文件时自动重新加载
bin/logstash -f test/filebeat-console.conf --config.reload.automatic

测试访问tomcat，logstash是否接收到了Filebeat传过来的tomcat日志,如果有则说明成功了，如下

你可能要稍微等等，我的虚拟机内存就2,5g，启动了elasticsearch,kibnana,logstash,filebeat,tomcat,所以非常卡顿,,,

Logstash输出数据到Elasticsearch

我们可以是不想输出在控制台的，那么对文件进行修改，输出到elastcisearch中，如下

input {
    beats {
      port => 5044
    }
}

output {
  elasticsearch {
    hosts => ["http://localhost:9200"]
  }
  stdout{
    codec => rubydebug
  }

添加了输出到elasticsearch的配置，ES中会生成一个以logstash开头的索引，测试日志是否保存到了ES。如图所示

使用Kibana查看索引

注意，时间戳字段要勾选@timestamp，不然没法根时间查询，创建完进行保存，去到

我这里只有一个message字段，所以我勾选的是没有时间选项的，如下，数据出来了

如果我勾选时间，但是因为文档里面没有时间过滤，会展示为空，需要用Grok模式插件进行分词，以下是官方文档：https://help.aliyun.com/document_detail/129387.html?scm=20140722.184.2.173

添加logstash配置如下即可

filter {
  grok {
    match => { "message" => "%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration}" }
  }
}

配置完再启动应该就行了，不过我的虚拟机时间有问题，里面是1月16号的，但是现在才15号，所以时间过滤的仍然查不到

不过配置大概就这么配置了

最后建议把数据写到指定的索引上，如下，就是添加了一句zxctomcat这个索引上,注意索引不能有大写字母

input {
    beats {
      port => 5044
    }
}
filter {
  grok {
    match => { "message" => "%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration}" }
  }
}
output {
  elasticsearch {
    hosts => ["http://localhost:9200"]
    index => "zxctomcat"
  }
  stdout{
    codec => rubydebug
  }
}