ELK实时日志分析平台环境部署

时间 2020-01-09

标签 elk 实时日志分析平台环境部署繁體版

原文原文链接

1、概念介绍

日志主要包括系统日志、应用程序日志和安全日志。系统运维和开发人员能够经过日志了解服务器软硬件信息、检查配置过程当中的错误及错误发生的缘由。常常分析日志能够了解服务器的负荷，性能安全性，从而及时采起措施纠正错误。html

一般，日志被分散的储存不一样的设备上。若是你管理数十上百台服务器，你还在使用依次登陆每台机器的传统方法查阅日志。这样是否是感受很繁琐和效率低下。当务之急咱们使用集中化的日志管理，例如：开源的syslog，将全部服务器上的日志采集汇总。前端

集中化管理日志后，日志的统计和检索又成为一件比较麻烦的事情，通常咱们使用grep、awk和wc等Linux命令能实现检索和统计，可是对于要求更高的查询、排序和统计等要求和庞大的机器数量依然使用这样的方法不免有点力不从心。java

经过咱们须要对日志进行集中化管理，将全部机器上的日志信息采集、汇总到一块儿。完整的日志数据具备很是重要的做用：node

信息查找。经过检索日志信息，定位相应的bug，找出解决方案。
服务诊断。经过对日志信息进行统计、分析，了解服务器的负荷和服务运行状态，找出耗时请求进行优化等等。
数据分析。若是是格式化的log，能够作进一步的数据分析，统计、聚合出有意义的信息，好比根据请求中的商品id，找出TOP10用户感兴趣商品。

那么，有没有一种方法能解决日志采集、统计和检索？开源实时日志分析ELK平台就可以完美的解决咱们上述的问题。mysql

简单地来讲，ELK 就是 Elasticsearch、Logstash 和 Kibana 三个开源工具的首字母简写。linux

一、Elasticsearch

Elasticsearch是一个开源的基于Lucene的开源实时的分布式搜索分析引擎工具。它能让你以一个以前从未有过的速度和规模，去探索你的数据。它的特色有：分布式，零配置，自动发现，索引自动分片，集群配置方便等。索引副本机制，restful风格接口，多数据源，自动搜索负载等。它提供了一个分布式多用户能力的全文搜索引擎，基于RESTful web接口。Elasticsearch是用Java开发的，并做为Apache许可条款下的开放源码发布，是第二流行的企业搜索引擎。设计用于云计算中，可以达到实时搜索，稳定，可靠，快速，安装使用方便。在Elasticsearch中，全部节点的数据是均等的。ios

（1）关于集群配置nginx

discovery.zen.ping.unicast.hosts，Elasticsearch默认使用Zen Discovery来作节点发现机制，推荐使用unicast来作通讯方式，在该配置项中列举出Master节点。
discovery.zen.minimum_master_nodes，该参数表示集群中可工做的具备Master节点资格的最小数量，默认值是1。为了提升集群的可用性，官方推荐设置为(N/2)+1，其中N是具备Master资格的节点的数量。
discovery.zen.ping_timeout，表示节点在发现过程当中的等待时间，默认值是3秒，能够根据自身网络环境进行调整，必定程度上提供可用性。
```
discovery.zen.ping.unicast.hosts: ["host1", "host2"]
discovery.zen.minimum_master_nodes: 2
discovery.zen.ping_timeout: 10
```

（2）关于集群节点git

节点类型包括：候选Master节点、数据节点和Client节点。经过设置两个配置项node.master和node.data为true或false，来决定将一个节点分配为何类型的节点。
尽可能将候选Master节点和Data节点分离开，一般Data节点负载较重，须要考虑单独部署。

（3）关于内存
Elasticsearch默认设置的内存是1GB，对于任何一个业务部署来讲，这个都过小了。经过指定ES_HEAP_SIZE环境变量，能够修改其堆内存大小，服务进程在启动时候会读取这个变量，并相应的设置堆的大小。建议设置系统内存的一半给Elasticsearch，可是不要超过32GB。github

（4）关于硬盘空间
Elasticsearch将数据存储在自定义的路径下，随着数据的增加，必定会出现硬盘空间不够用的情形，此时就须要给机器挂载新的硬盘，并将Elasticsearch的路径配置到新硬盘的路径下。经过“path.data”配置项来进行设置，好比“path.data: /elk/elasticsearch/data”。须要注意的是，同一分片下的数据只能写入到一个路径下，所以仍是须要合理的规划。

（5）关于Index的划分和分片的个数
这个须要根据数据量来作权衡了，Index能够按时间划分，好比每个月一个或者天天一个，在Logstash输出时进行配置，shard的数量也须要作好控制。

（6）关于监控
这里我使用head和kopf两个监控插件。

二、Logstash

Logstash是一个开源的日志采集工具，是一个数据管道。主要用来采集、解析和分析大量结构化和非结构化的数据以及各类系统产生的事件。它能够对你的日志进行采集、过滤、分析，并将其存储供之后使用（如，搜索），咱们可使用它。说到搜索，logstash带有一个web界面，搜索和展现全部日志。

Logstash工做原理展现图：

（1）经常使用的输入input

file：文件输入插件用于将事件和日志文件输入给logstash。它会自动检测日志文件轮转，并从上次读取的位点读取数据。
syslog：在514端口上监听系统日志消息，并根据RFC3164标准进行解析
redis：从redis service中读取
beats：从filebeat中读取
Filters：数据中间处理，对数据进行操做。

（2）经常使用的过滤器filter

grok：解析任意文本数据，Grok 是 Logstash 最重要的插件。它的主要做用就是将文本格式的字符串，转换成为具体的结构化的数据，配合正则表达式使用。内置120多个解析语法。
mutate：对字段进行转换。例如对字段进行删除、替换、修改、重命名等。
drop：丢弃一部分events不进行处理。
clone：拷贝 event，这个过程当中也能够添加或移除字段。
geoip：添加地理信息（为前台kibana图形化展现使用）。
Outputs：outputs是logstash处理管道的最末端组件。一个event能够在处理过程当中通过多重输出，可是一旦全部的outputs都执行结束，这个event也就完成生命周期。

（3）常见的输出output

elasticsearch：elasticsearch插件是ELK技术栈中用到的最重要的插件，它用于将输出的结果存储到elasticsearch搜索引擎中供Kibana分析。
file：将event数据保存到文件中。
graphite：将event数据发送到图形化组件中，一个很流行的开源存储图形化展现的组件。
codecs：codecs 是基于数据流的过滤器，它能够做为input，output的一部分配置。Codecs能够帮助你轻松的分割发送过来已经被序列化的数据。

（4）常见的编解码codecs

json：使用json格式对数据进行编码/解码。
line：用于将每行输入日志做为一个事件，将每一个事件解码成一行。
multiline：将汇多个事件中数据汇总为一个单一的行。好比：java异常信息和堆栈信息。
plain：插件用于指示输入或输出事件时，不须要额外的编解码操做，对应的输入输出插件会本身处理编解码。对不少插件如redis、mongondb等而言，plain是默认的编解码类型。
rubydebug：只在输出事件时使用，它使用Ruby Awesome打印库打印输出事件。

（5）采集日志信息

pattern => "^\["             # 采集以"["开头的日志信息
pattern => "^2018"           # 采集以"2018"开头的日志信息
pattern => "^[a-zA-Z0-9]"      # 采集以字母（大小写）或数字开头的日志信息
pattern => "^[a-zA-Z0-9]|[^ ]+"  # 采集以字母（大小写）或数字或空格的日志信息

三、Kibana

Kibana是一个开源的基于浏览器页面的Elasticsearch前端展现工具。它Kibana能够为 Logstash 和 ElasticSearch 提供的日志分析友好的 Web 界面，能够帮助咱们汇总、分析和搜索重要数据日志，也能够经过直方图、地图、饼图、其它图形和表格等方式来表达数据。只须要点击鼠标，就能够完成搜索、聚合功能，生成炫丽的仪表板。
Kibana提供的是数据查询和显示的Web服务，有丰富的图表样板，能知足大部分的数据可视化需求，这也是不少人选择ELK的主要缘由之一。

而在5.0版本之后，Elastic公司将原来的 ELK Stack 称之为 Elastic Stack，缘由是引入了 Beats 套件。

官网：https://www.elastic.co/products

ELK工做原理展现图：

如上图所示：多个应用服务器上的日志经过Logstash采集器传输到一个集中化的索引器中，索引器将处理后的数据结果输出到ElasticSearch集群，而后Kibana经过下查询ElasticSearch集群中的日志数据建立仪表盘，作可视化展示。

2、ELK总体方案

ELK中的三个系统分别扮演着不一样的角色，组成了一个总体的解决方案。Logstash是一个ETL工具，负责从每台机器抓取日志数据，对数据进行格式转换和处理后，输出到Elasticsearch中存储。Elasticsearch是一个分布式搜索引擎和分析引擎，用于数据存储，可提供实时的数据查询。Kibana是一个数据可视化服务，根据用户的操做从Elasticsearch中查询数据，造成相应的分析结果，以图表的形式展示给用户。
ELK的安装很简单，能够按照“下载->修改配置文件->启动”方法分别部署三个系统，也可使用Docker来快速部署。下面来看一个常见的部署方案，以下图所示：

在每台生成日志文件的机器上，部署Logstash，做为Shipper的角色，负责从日志文件中提取数据，可是不作任何处理，直接将数据输出到Redis队列(list)中；
须要一台机器部署Logstash，做为Indexer的角色，负责从Redis中取出数据，对数据进行格式化和相关处理后，输出到Elasticsearch中存储；
部署Elasticsearch集群，固然取决于你的数据量了，数据量小的话可使用单台服务，若是作集群的话，最好是有3个以上节点，同时还须要部署相关的监控插件；
部署Kibana服务，提供Web可视化展示服务。

在前期部署阶段，主要工做是Logstash节点和Elasticsearch集群的部署，而在后期使用阶段，主要工做就是Elasticsearch集群的监控和使用Kibana来检索、分析日志数据了，固然也能够直接编写程序来消费Elasticsearch中的数据。

在上面的部署方案中，咱们将Logstash分为Shipper和Indexer两种角色来完成不一样的工做，中间经过Redis作数据管道，为何要这样作？为何不是直接在每台机器上使用Logstash提取数据、处理、存入Elasticsearch？

首先，采用这样的架构部署，有三点优点：

下降对日志所在机器的影响，这些机器上通常都部署着反向代理或应用服务，自己负载就很重了，因此尽量的在这些机器上少作事。
若是有不少台机器须要作日志采集，那么让每台机器都向Elasticsearch持续写入数据，必然会对Elasticsearch形成压力，所以须要对数据进行缓冲，同时，这样的缓冲也能够必定程度的保护数据不丢失。
将日志数据的格式化与处理放到Indexer中统一作，能够在一处修改代码、部署，避免须要到多台机器上去修改配置。

其次，咱们须要作的是将数据放入一个消息队列中进行缓冲，因此Redis只是其中一个选择，也能够是RabbitMQ、Kafka等等，在实际生产中，Redis与Kafka用的比较多。因为Redis集群通常都是经过key来作分片，没法对list类型作集群，在数据量大的时候必然不合适了，而Kafka天生就是分布式的消息队列系统。

3、ELK环境部署

一、基础环境介绍

# 系统： Centos7.3
[root@elk-01 ~]# cat /etc/redhat-release 
CentOS Linux release 7.3.1611 (Core)

# 防火墙： 关闭
[root@elk-01 ~]# systemctl stop firewalld && systemctl disable firewalld
Removed symlink /etc/systemd/system/multi-user.target.wants/firewalld.service.
Removed symlink /etc/systemd/system/dbus-org.fedoraproject.FirewallD1.service.

# Sellinux： 关闭
[root@elk-01 ~]# setenforce 0
[root@elk-01 ~]# sed -i 's/SELINUX=enforcing/SELINUX=disabled/g' /etc/selinux/config

# 机器环境： 两台
elk-01: 192.168.8.55       # master机器
elk-02: 192.168.8.66       # slave机器

说明：
master-slave模式：master采集到日志后，会把一部分数据碎片到salve上（随机的一部分数据）；同时，master和slave又都会各自作副本，并把副本放到对方机器上，这样就保证了数据不会丢失。
若是master宕机了，那么客户端在日志采集配置中将Elasticsearch主机指向改成slave，就能够保证ELK日志的正常采集和web展现

二、Elasticsearch安装配置

Elasticsearch是java程序，并且要跑在jdk1.8版本以上
elk-01和elk-02同时操做

（1）写hosts文件

[root@elk-01 ~]# echo "192.168.8.55 elk-01" >> /etc/hosts
[root@elk-02 ~]# echo "192.168.8.66 elk-02" >> /etc/hosts

（2）安装jdk1.8版本

[root@elk-01 ~]# yum -y install java-1.8.0
[root@elk-01 ~]# java -version
openjdk version "1.8.0_102"
OpenJDK Runtime Environment (build 1.8.0_102-b14)
OpenJDK 64-Bit Server VM (build 25.102-b14, mixed mode)

（3）安装Elasticsearch

[root@elk-01 ~]# wget https://download.elastic.co/elasticsearch/release/org/elasticsearch/distribution/tar/elasticsearch/2.4.6/elasticsearch-2.4.6.tar.gz
[root@elk-01 ~]# mkdir /elk
[root@elk-01 ~]# tar zxvf elasticsearch-2.4.6.tar.gz -C /elk
[root@elk-01 ~]# cd /elk/
[root@elk-01 elk]# ls
elasticsearch-2.4.6
[root@elk-01 elk]# mv ./elasticsearch-2.4.6 ./elasticsearch

（4）修改Elasticsearch配置文件

[root@elk-01 ~]# vim /elk/elasticsearch/config/elasticsearch.yml     #将配置文件内容修改以下，打开注释
17 cluster.name: test-ELK     # 配置集群名，两台服务器名保持一致
23 node.name: elk-01   # 配置单一节点名称，每一个节点惟一标识
33 path.data: /elk/elasticsearch/data   # data存储路径
37 path.logs: /elk/elasticsearch/logs   #log存储路径
43 bootstrap.memory_lock: true      # 锁住内存，不被使用到交换分区去
54 network.host: 0.0.0.0    # 监听地址
58 http.port: 9200          # 监听端口
68 discovery.zen.ping.unicast.hosts: ["192.168.8.66"]   # 集群节点发现列表，写另一台的ip
72 discovery.zen.minimum_master_nodes: 2   # 集群可作master的最小节点数

[root@elk-01 ~]# mkdir -p /elk/elasticsearch/{data,logs}

（5）修改系统参数

[root@elk-01 ~]# vim /etc/security/limits.conf     --文件最后添加系统参数，让系统用户打开文件无限制
# 添加配置
*        soft    nofile  65536
*        hard    nofile  131072
*        soft    nproc   2048
*        hard    nproc   4096
*        soft    memlock unlimited
*        hard    memlock unlimited

[root@elk-01 ~]# vim /etc/sysctl.conf   --在此文件里加入如下参数
# 添加配置
vm.max_map_count= 262144

[root@elk-01 ~]# sysctl -p  --刷新
vm.max_map_count = 262144

（6）开户启动Elasticsearch服务

当使用root帐户启动Elasticsearch会出现错误信息，这是由于处于系统安装考虑的设置，因为Elasticsearch能够接收用户输入的脚本而且执行，为了系统安全考虑，不容许root帐号启动，因此建议给Elasticsearch单首创建一个用户来运行Elasticsearch。

# 建立elk用户组及elk用户
[root@elk-01 ~]# groupadd elk
[root@elk-01 ~]# useradd elk -g elk -p 123
[root@elk-01 ~]# chown -R elk.elk /elk/elasticsearch
[root@elk-01 ~]# su - elk
上一次登陆：六 5月  5 11:08:30 CST 2018pts/5 上
[elk@elk-01 ~]$ cd /elk/elasticsearch/
[elk@elk-01 elasticsearch]$ nohup ./bin/elasticsearch &
[1] 3503
[elk@elk-01 elasticsearch]$ nohup: 忽略输入并把输出追加到"nohup.out"

[root@elk-01 ~]# netstat -antlp |egrep "9200|9300"
tcp6       0      0 :::9200                 :::*                    LISTEN      1608/java
tcp6       0      0 :::9300                 :::*                    LISTEN      1608/java
tcp6       0      1 192.168.8.66:47992      192.168.8.55:9300       SYN_SENT    1608/java
tcp6       0      0 192.168.8.66:9200       192.168.8.1:2015        ESTABLISHED 1608/java

若是启动失败，在系统日志（咱们的示例中为/ var/log/messages）中将会遇到与如下相似的错误：

Apr  9 15:39:09 elk-02 kernel: [ 3727]  1000  3727   784748   214244     617    70749             0 java
Apr  9 15:39:09 elk-02 kernel: Out of memory: Kill process 3727 (java) score 368 or sacrifice child
Apr  9 15:39:09 elk-02 kernel: Killed process 3727 (java) total-vm:3138992kB, anon-rss:849016kB, file-rss:7960kB, shmem-rss:0kB

这个错误的缘由是low memory耗尽。“内核使用low memory来跟踪全部的内存分配，一旦low memory耗尽，就会查杀进程，以保持系统的正常运转。说白了 OOM Killer 就是一层保护机制，用于避免 Linux 在内存不足的时候不至于出太严重的问题，把可有可无的进程杀掉。
最简单的解决方法就是增长内存空间。

（7）访问Elasticsearch

http://192.168.8.55:9200/
http://192.168.8.66:9200/

浏览器访问结果（访问时建议使用chrome、firefox浏览器）

三、安装Elasticsearch-head插件

elasticsearch-head是用于浏览Elasticsearch集群并与其进行交互的Web前端。也就是用于显示集群节点和数据信息。

Elasticsearch 5及以上版本（可自行研究）中的head插件已经不经过elasticsearch-plugin 来进行安装，已经成为了一个独立的服务，须要单独进行安装，安装方法可参考GitHub网站。

在Elasticsearch 5以前的版本，安装起来比较简单，在安装成功后能够在浏览器中经过http://ip:9200/_plugin/head 就能够访问了，这里的IP地址即为你的Elasticsearch服务器地址。查看官方文档。具体安装以下：

[root@elk-01 ~]# /elk/elasticsearch/bin/plugin install mobz/elasticsearch-head
-> Installing mobz/elasticsearch-head...
Trying https://github.com/mobz/elasticsearch-head/archive/master.zip ...
Downloading .............................................................................................................................DONE
Verifying https://github.com/mobz/elasticsearch-head/archive/master.zip checksums if available ...
NOTE: Unable to verify checksum for downloaded plugin (unable to find .sha1 or .md5 file to verify)
Installed head into /elk/elasticsearch/plugins/head

插件安装的目录：/elk/elasticsearch/plugins
在线安装完成以后，进行授予权限

[root@elk-01 ~]# chown -R elk:elk /elk/elasticsearch/plugins
[root@elk-01 ~]# ll /elk/elasticsearch/plugins/head/

插件访问（最好提早将elk-02节点的配置和插件都安装后，再来进行访问和数据插入测试）

http://192.168.8.55:9200/_plugin/head/
http://192.168.8.66:9200/_plugin/head/

注：ES集群健康度分为：红、黄、绿三种颜色
红色：个别分片副本不可用
×××：个别副本不可用
绿色：为健康

两台机器看到的内容彻底一致，test-ELK是集群名称，集群健康值为绿色。

测试一
插入数据实例，看界面能不能正常显示
以下：点击“复合查询”，在POST选项下，任意输入如/2018/test而后在下面输入数据（注意内容之间换行的逗号不要漏掉）；
数据输入好以后(以下输入"user":"xiaozuo","mess":"test Elasticsearch"内容)，下面点击“验证JSON”->“提交请求”，提交成功后，观察右栏内出现的信息：有index、type、version等信息，failed:0（成功消息）

测试二
以下：点击“复合查询”，选择GET选项，在/2018/test/后面输入上面POST结果中的id号，不输入内容，即{}括号里为空！
而后点击“验证JSON”->“提交请求”，观察右栏内就有了上面插入的数据了（即xiaozuo，test Elasticsearch）

点击“基本查询”，查看数据，以下图所示，便可查询到上面插入的数据：

点击“数据浏览”，也能查看到插入的数据：

点击“概览”，主界面也会更新，显示刚才建立类型，也会显示数据碎片存取的位置：

每一个索引都有5个分片，粗线宽的分片是主节点分片，细线宽的是副本。这样kibana在搜索的时候就能够从多个Easticsearch服务器上读取，压力也按比例分布在各个集群节点中。

说明：
必定要提早在elk-02节点上也完成配置（配置内容同上）。若是你尚未创建集群，那么只能看到一个节点，并且上面插入数据后，Elasticsearch集群状态会呈现×××yellow状态，elk-02完成配置加入到集群里后就会恢复到正常的绿色状态。
当数据很少的时候经过一台Elasticsearch服务器也能完成任务。可是数据随着时间的推移而增多，尤为须要查询6个月、一年甚至更长时间跨度的数据时，你会发现集群有多么重要。

四、安装kopf监控插件

Kopf展现了ElasticSearch的节点分片、文档、所占用空间、大小等信息，而且能够查询文档的索引。

[root@elk-01 ~]# /elk/elasticsearch/bin/plugin install lmenezes/elasticsearch-kopf
-> Installing lmenezes/elasticsearch-kopf...
Trying https://github.com/lmenezes/elasticsearch-kopf/archive/master.zip ...
Downloading .............................................................................................................................DONE
Verifying https://github.com/lmenezes/elasticsearch-kopf/archive/master.zip checksums if available ...
NOTE: Unable to verify checksum for downloaded plugin (unable to find .sha1 or .md5 file to verify)
Installed kopf into /elk/elasticsearch/plugins/kopf

# 授予权限
[root@elk-01 ~]# chown -R elk:elk /elk/elasticsearch/plugins
[root@elk-01 ~]# ll /elk/elasticsearch/plugins/kopf/

访问插件：（以下，一样要提早安装好elk-02节点上的插件，不然访问时会出现集群节点为×××的yellow告警状态）

http://192.168.8.55:9200/_plugin/kopf/#!/cluster
http://192.168.8.66:9200/_plugin/kopf/#!/cluster

到此Elasticsearch软件包安装完成。

五、Logstash安装配置

elk-01和elk-02都要安装。同时，Logstash部署在每台生成日志文件的机器上，采集到的数据写入到Elasticsearch里，就能够登录Logstash界面查看到了）

（1）安装Logstash

[root@elk-01 ~]# wget https://download.elastic.co/logstash/logstash/logstash-2.4.1.tar.gz
[root@elk-01 ~]# tar zxvf logstash-2.4.1.tar.gz -C /elk/
[root@elk-01 ~]# mv /elk/logstash-2.4.1/ /elk/logstash/
[root@elk-01 ~]# useradd logstash -s /sbin/nologin 
[root@elk-01 ~]# chown -R logstash.logstash /elk/logstash/

（2）数据的测试

测试一
基本的输入输出，验证logstash可否接收到数据。

[root@elk-01 ~]# /elk/logstash/bin/logstash -e "input { stdin{ } } output { stdout{} }"
Settings: Default pipeline workers: 1
Pipeline main started
xiaozuo      # 输入的内容
2018-05-06T07:15:50.140Z elk-01.com xiaozuo   # 输出的内容
hello        # 输入的内容
2018-05-06T07:16:09.850Z elk-01.com hello     # 输出的内容

在这个例子中，咱们经过输入插件stdin和输出插件stdout来运行Logstash，因此无论你输入什么，都会输出一样的信息。使用-e参数能够在命令行中快速地测试配置是否正确。

测试二
使用rubydebug详细输出

[root@elk-01 ~]# /elk/logstash/bin/logstash -e "input { stdin{} } output { stdout{ codec => rubydebug} }"
Settings: Default pipeline workers: 1
Pipeline main started
hello    # 输入的内容
{       # 输出下面信息
       "message" => "hello",
      "@version" => "1",
    "@timestamp" => "2018-05-06T07:23:16.933Z",
          "host" => "elk-01.com"
}
xiaozuoxiansen   # 输入的内容
{       # 输出下面信息
       "message" => "xiaozuoxiansen",
      "@version" => "1",
    "@timestamp" => "2018-05-06T07:23:48.631Z",
          "host" => "elk-01.com"
}

测试三
把内容写到Elasticsearch中

[root@elk-01 ~]# /elk/logstash/bin/logstash -e "input { stdin{} } output { elasticsearch { hosts => ['192.168.8.55:9200']} }"
Settings: Default pipeline workers: 1
Pipeline main started
123456     # 随意输入内容
xiaozuo
hello

说明：
使用rubydebug和写到elasticsearch中的区别：其实就在于后面标准输出的区别，前者使用codec；后者使用elasticsearch。

写到Elasticsearch中在Logstash中查看，以下图所示：

注意：
master采集到日志后，会把一部分数据碎片到salve上（随机的一部分数据），master和slave又都会各自作副本，并把副本放到对方机器上，这样就保证了数据不会丢失。
以下，master采集到的数据放到了本身的第0、二、4分片上，其余的放到了slave的第一、3分片上。

再点击“数据浏览”，能够看到刚才输入的内容：

（3） Logstash的配置和文件的编写

1）Logstash的配置
简单的配置方式：

[root@elk-01 ~]# mkdir /elk/logstash/conf.d/
[root@elk-01 ~]# vim /elk/logstash/conf.d/01-logstash.conf
input { stdin { } }
output {
        elasticsearch { hosts => ["192.168.8.55:9200"]}
        stdout { codec => rubydebug }
}

执行结果：

[root@elk-01 ~]# /elk/logstash/bin/logstash -f /elk/logstash/conf.d/01-logstash.conf 
Settings: Default pipeline workers: 1
Pipeline main started
ShenZhen       # 随意输入内容
{       # 输出下面的信息
       "message" => "ShenZhen",
      "@version" => "1",
    "@timestamp" => "2018-05-06T07:59:40.650Z",
          "host" => "elk-01.com"
}

再击“数据浏览”，能够看到刚才输入的内容：

参考官方文档：
https://www.elastic.co/guide/en/logstash/current/configuration.html
https://www.elastic.co/guide/en/logstash/current/configuration-file-structure.html

2）采集系统日志

[root@elk-01 ~]# vim /elk/logstash/conf.d/file.conf
input {            # 指定输出
    file {         # 输出的内容为文件
      path => "/var/log/messages"   # 文件路径
      type => "system"            # 给该类日志内容定一个名称，可自定义
      start_position => "beginning"  # 表示从哪里开始读取日志,beginning是全部都读
    }
}

output {               # 指定输出到哪里
    elasticsearch {      # 指定输出到elasticaearch服务里
       hosts => ["192.168.8.55:9200"]     # 指定链接elasticaearch服务的IP和端口
       index => "system-%{+YYYY.MM.dd}"   # 把日志按日期进行分类
    }
}

执行上面日志信息的采集，以下图所示，这个命令会一直在执行中，表示日志在监控采集中；若是中断，就表示日志不在采集！因此须要放在后台执行&

[root@elk-01 ~]# /elk/logstash/bin/logstash -f /elk/logstash/conf.d/file.conf &

登录Elasticsearch界面，查看本机系统日志的信息：

参考官方文档：
https://www.elastic.co/guide/en/logstash/current/plugins-outputs-elasticsearch.html
https://www.elastic.co/guide/en/logstash/current/event-dependent-configuration.html

从上面采集系统日志的“数据浏览”中能够看出，每一个日志都给采集成一行了，不是按照一个日志，一个事件模块采集的。

（4）将行换成事件的方式展现

[root@elk-01 ~]# vim /elk/logstash/conf.d/multiline.conf
input {
    stdin {
       codec => multiline {
          pattern => "^\["
          negate => true
          what => "previous"
        }
    }
}
output {
    stdout {
      codec => "rubydebug"
     }
}

执行命令：

[root@elk-01 ~]# /elk/logstash/bin/logstash -f /elk/logstash/conf.d/multiline.conf

说明：
在没有遇“[”的时候，系统不会采集，只有碰见“[”的时候，才算是一个事件，才采集起来。

参考官方文档：
https://www.elastic.co/guide/en/logstash/current/plugins-codecs-multiline.html

六、Kibana安装配置

（1）安装Kibana

[root@elk-01 ~]# wget https://download.elastic.co/kibana/kibana/kibana-4.6.6-linux-x86_64.tar.gz
[root@elk-01 ~]# tar zxvf kibana-4.6.6-linux-x86_64.tar.gz -C /elk/
[root@elk-01 ~]# mv /elk/kibana-4.6.6-linux-x86_64/ /elk/kibana/

（2）修改配置文件

[root@elk-01 ~]# cd /elk/kibana/config/
[root@elk-01 config]# pwd
/elk/kibana/config
[root@elk-01 config]# ls
kibana.yml
[root@elk-01 config]# mv kibana.yml kibana.yml.bak
[root@elk-01 config]# vim kibana.yml
server.port: 5601
server.host: "0.0.0.0"
elasticsearch.url: "http://192.168.8.55:9200"
kibana.index: ".kibana"  # 注意这个.Kibana的index索引配置必定要有，它是将ES数据经过kibana进行web展现的关键。这个配置后，在ES的web界面里就会看到这个.kibana索引。

（3）启动Kibana

由于Kibana一直运行在前台，因此能够放在后台执行&

[root@elk-01 ~]# cd /elk/kibana/
[root@elk-01 kibana]# nohup ./bin/kibana &
[1] 10314
[root@elk-01 kibana]# nohup: 忽略输入并把输出追加到"nohup.out"
[root@elk-01 kibana]# lsof -i:5601
COMMAND   PID USER   FD   TYPE DEVICE SIZE/OFF NODE NAME
node    10314 root   11u  IPv4 107473      0t0  TCP *:esmagent (LISTEN)

（4）访问Kibana

http://192.168.8.55:5601

（5）建立日志采集项

建立一个日志采集项，该名称要和前面写的file.conf配置文件里的type字段一致。好比添加system系统日志。注意后面的*不要忘了。

建立完成后，再点击“Discover”，在Discover中查看：

查看日志登录，须要点击“Discover”-->“message”，点击它后面的“add”
说明：
须要右边查看日志内容时带什么属性，就在左边点击相应属性后面的“add”
以下图，添加了message和path的属性：

这样，右边显示的日志内容的属性就带了message和path。

点击右边日志内容属性后面隐藏的<<，就可将内容向前缩进：

添加新的日志采集项，点击“Settings”->“ +Add New”，好比添加logstash-日志。注意后面的*不要忘了。

删除Kibana里的日志采集项，点击删除图标便可，以下图所示：

若是打开Kibana查看日志，发现没有日志内容，出现“No results found”，以下图所示，这说明要查看的日志在当前时间没有日志信息输出，能够点击右上角的时间钟来调试日志信息的查看。

（6）采集nginx的访问日志

[root@elk-01 ~]# wget https://nginx.org/download/nginx-1.14.0.tar.gz
[root@elk-01 ~]# tar zxf nginx-1.14.0.tar.gz -C /usr/src/
[root@elk-01 ~]# cd /usr/src/nginx-1.14.0/
[root@elk-01 ~]# ./configure --prefix=/usr/local/nginx
[root@elk-01 ~]# make && make install
[root@elk-01 ~]# useradd nginx -s /sbin/nologin  --建立nginx用户，-s制定shell，nginx用户不须要登陆系统
[root@elk-01 ~]# id nginx
uid=1002(nginx) gid=1002(nginx) 组=1002(nginx)
[root@elk-01 ~]# chown -R nginx.nginx /usr/local/nginx/
[root@elk-01 ~]# vim /usr/local/nginx/conf/nginx.conf  --打开如下注释（根据需求）
  2 user nginx nginx;  # 运行的用户和组，打开注释，使用nginx身份
  7 error_log  logs/error.log  info;  # 错误日志以及日志等级，打开等级为info级别的日志
  9 pid        logs/nginx.pid;  # pid文件
33     gzip  on;  # 在服务器压缩数据，进行传输
37         server_name  192.168.8.55;  # 主机名或者IP
39         charset utf8;  # 字符集,改为utf8
48         error_page  404              /404.html;  # 错误页面

Nginx安装配置完成后，先查看一下80端口是否被占用，占用的话先停掉Apache（因Apache默认是80端口，若是是其它程序占用，则停掉占用80端口的程序）

[root@elk-01 ~]# lsof -i:80   --80端口没有占用
[root@elk-01 ~]# /usr/local/nginx/sbin/nginx 
[root@elk-01 ~]# lsof -i:80   --Nginx启动成功
COMMAND   PID   USER   FD   TYPE DEVICE SIZE/OFF NODE NAME
nginx   13375   root    6u  IPv4 166972      0t0  TCP *:http (LISTEN)
nginx   13376 nobody    6u  IPv4 166972      0t0  TCP *:http (LISTEN)

浏览器访问Nginx

http://192.168.8.55/

Nginx安装成功以后，修改Nginx的配置文件，分别在nginx.conf的http和server配置区域添加下面内容：

# http 标签中
    log_format json '{"@timestamp":"$time_iso8601",'
                      '"@version":"1",'
                      '"client":"$remote_addr",'
                      '"url":"$uri",'
                      '"status":"$status",'
                      '"domain":"$host",'
                      '"host":"$server_addr",'
                      '"size":$body_bytes_sent,'
                      '"responsetime":$request_time,'
                      '"referer": "$http_referer",'
                      '"ua": "$http_user_agent"'
                      '}';

# server标签中
        access_log /var/log/nginx/access_json.log json;

截图以下：

启动Nginx服务：

[root@elk-01 ~]# mkdir /var/log/nginx/
[root@elk-01 ~]# touch /var/log/nginx/access_json.log
[root@elk-01 ~]# /usr/local/nginx/sbin/nginx -s stop
[root@elk-01 ~]# /usr/local/nginx/sbin/nginx

编写日志采集文件：
使用的是json的方式采集：

[root@elk-01 ~]# vim json.conf
input {
   file {
      path => "/var/log/nginx/access_json.log"
      codec => "json"
   }
}

output {
   stdout {
      codec => "rubydebug"
   }
}

启动日志采集程序：

[root@elk-01 ~]# /elk/logstash/bin/logstash -f /root/json.conf &   --加个&放在后台执行

访问nginx页面（在elk-01的主机上执行访问页面的命令：curl http://192.168.8.55）就会出现如下图所示：

注意：
上面的json.conf配置只是将nginx日志输出，尚未输入到Elasticsearch里，因此这个时候在Elasticsearch界面里是采集不到nginx日志的。
须要配置一下，将Nginx日志输入到Elasticsearch中，将其汇总到总文件file.conf里，以下也将nginx-log日志输入到Elasticserach里：（后续就能够只用这个汇总文件，把要追加的日志汇总到这个总文件里便可）

[root@elk-01 ~]# cat file.conf 
input {
    file {
      path => "/var/log/messages"
      type => "system"
      start_position => "beginning"
    }

    file {
       path => "/var/log/nginx/access_json.log"
       codec => json
       start_position => "beginning"
       type => "nginx-log"
    }
}

output {

    if [type] == "system"{
        elasticsearch {
           hosts => ["192.168.8.55:9200"]
           index => "system-%{+YYYY.MM.dd}"
        }
    }

    if [type] == "nginx-log"{
        elasticsearch {
           hosts => ["192.168.8.55:9200"]
           index => "nignx-log-%{+YYYY.MM.dd}"
        }
    }
}

在执行日志采集命令时，能够加上--configtest参数，测试下配置文件是否有语法错误或配置不当的地方，这个很重要！！

[root@elk-01 ~]# /elk/logstash/bin/logstash -f file.conf --configtest
Configuration OK

而后接着执行Logstash命令（因为上面已经将这个执行命令放到了后台，因此这里其实不用执行，也能够先kill以前的，再放后台执行），而后能够再访问Nginx界面测试下

[root@elk-01 ~]# /elk/logstash/bin/logstash -f /root/file.conf &

登录Elasticsearch主界面查看：

登录Kibana主界面查看：

（7）采集系统日志

编写日志采集文件：

[root@elk-01 ~]# cat syslog.conf 
input {
    syslog {
        type => "system-syslog"
        host => "192.168.8.55"
        port => "514"
    }
}

output {
    stdout {
        codec => "rubydebug"
    }
}

对上面的采集文件进行执行：

[root@elk-01 ~]# /elk/logstash/bin/logstash -f syslog.conf

从新开启一个窗口，查看服务是否启动：

[root@elk-01 ~]# netstat -antlp | grep 514
tcp6       0      0 192.168.8.55:514        :::*                    LISTEN      14101/java          
[root@elk-01 ~]# vim /etc/rsyslog.conf
[root@elk-01 ~]# systemctl restart rsyslog

回到原来的窗口（即上面采集文件的执行终端），就会出现数据：

[root@elk-01 ~]# /elk/logstash/bin/logstash -f syslog.conf
Settings: Default pipeline workers: 1
Pipeline main started
{
           "message" => "[origin software=\"rsyslogd\" swVersion=\"7.4.7\" x-pid=\"14155\" x-info=\"http://www.rsyslog.com\"] start\n",
          "@version" => "1",
        "@timestamp" => "2018-05-07T08:30:05.000Z",
              "type" => "system-syslog",
              "host" => "192.168.8.55",
          "priority" => 46,
         "timestamp" => "May  7 16:30:05",
         "logsource" => "elk-01",
           "program" => "rsyslogd",
          "severity" => 6,
          "facility" => 5,
    "facility_label" => "syslogd",
    "severity_label" => "Informational"
}
........
........

再次把syslog.conf文件的内容添加到总文件file.conf中：

[root@elk-01 ~]# cat file.conf 
input {
    file {
      path => "/var/log/messages"
      type => "system"
      start_position => "beginning"
    }

    file {
       path => "/var/log/nginx/access_json.log"
       codec => json
       start_position => "beginning"
       type => "nginx-log"
    }
    syslog {
        type => "system-syslog"
        host => "192.168.8.55"
        port => "514"
    }
}

output {
    if [type] == "system"{
        elasticsearch {
           hosts => ["192.168.8.55:9200"]
           index => "system-%{+YYYY.MM.dd}"
        }
    }

    if [type] == "nginx-log"{
        elasticsearch {
           hosts => ["192.168.8.55:9200"]
           index => "nignx-log-%{+YYYY.MM.dd}"
        }
    }
        ``
    if [type] == "system-syslog"{
        elasticsearch {
           hosts => ["192.168.8.55:9200"]
           index => "system-syslog-%{+YYYY.MM.dd}"
        }
    }
}

执行总文件（先测试下总文件file.conf配置是否有误，而后先kill以前在后台启动的file.conf文件，再次执行）：

[root@elk-01 ~]# /elk/logstash/bin/logstash -f /root/file.conf --configtest
Configuration OK
[root@elk-01 ~]# jobs -l
[1]  13511 运行中               nohup ./bin/kibana &(工做目录：/elk/kibana)
[2]- 13645 运行中               /elk/logstash/bin/logstash -f /root/json.conf &
[3]+ 13815 运行中               /elk/logstash/bin/logstash -f /root/file.conf &
[root@elk-01 ~]# kill -9 13815
[root@elk-01 ~]# 
[3]+  已杀死               /elk/logstash/bin/logstash -f /root/file.conf
[root@elk-01 ~]# /elk/logstash/bin/logstash -f /root/file.conf &

测试：
向日志中添加数据，查看Elasticsearch和Kibana的变化：

[root@elk-01 ~]# logger "Hello World_01"
[root@elk-01 ~]# logger "Hello World_02"
[root@elk-01 ~]# logger "Hello World_03"
[root@elk-01 ~]# logger "Hello World_04"
[root@elk-01 ~]# logger "Hello World_05"

（8）TCP日志的采集

编写日志采集文件，并执行：（有须要的话，能够将下面采集文件的配置汇总到上面的总文件file.conf里，进而输入到Elasticsearch界面里和Kibana里查看）

[root@elk-01 ~]# cat tcp.conf 
input {
    tcp {
        host => "192.168.8.55"
        port => "6666"
    }
}

output {
    stdout {
        codec => "rubydebug"
    }
}
# 执行
[root@elk-01 ~]# /elk/logstash/bin/logstash -f /root/tcp.conf

从新开启一个窗口，执行以下命令：
测试一（安装nc命令）：

[root@elk-01 ~]# yum install -y nc
[root@elk-01 ~]# nc 192.168.8.55 6666 < /etc/resolv.conf

回到原来的窗口（即上面采集文件的执行终端），就会出现数据：

[root@elk-01 ~]# /elk/logstash/bin/logstash -f /root/tcp.conf 
Settings: Default pipeline workers: 1
Pipeline main started
{                  
       "message" => "# Generated by NetworkManager",
      "@version" => "1",
    "@timestamp" => "2018-05-07T08:59:11.599Z",
          "host" => "192.168.8.55",
          "port" => 58622
}
{
       "message" => "search elk-01.com",
      "@version" => "1",
    "@timestamp" => "2018-05-07T08:59:11.601Z",
          "host" => "192.168.8.55",
          "port" => 58622
}
{
       "message" => "nameserver 192.168.8.1",
      "@version" => "1",
    "@timestamp" => "2018-05-07T08:59:11.614Z",
          "host" => "192.168.8.55",
          "port" => 58622
}

测试二

[root@elk-01 ~]# echo "hello" | nc 192.168.8.55 6666
[root@elk-01 ~]# echo "hello" > /dev/tcp/192.168.8.55/6666

回到以前的执行端口，再去查看，就会显示出数据：

[root@elk-01 ~]# /elk/logstash/bin/logstash -f /root/tcp.conf 
Settings: Default pipeline workers: 1
Pipeline main started
{
       "message" => "hello",
      "@version" => "1",
    "@timestamp" => "2018-05-07T09:00:51.130Z",
          "host" => "192.168.8.55",
          "port" => 58970
}
{
       "message" => "hello",
      "@version" => "1",
    "@timestamp" => "2018-05-07T09:01:13.536Z",
          "host" => "192.168.8.55",
          "port" => 59048
}

（9）使用filter

编写日志采集文件：

[root@elk-01 ~]# cat grok.conf 
input {
    stdin { }
}
filter {
    grok {
        match => { "message" => "%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration}" }
    }
}

output {
    stdout{
        codec => "rubydebug"
    }
}

对上面的采集文件进行执行：

[root@elk-01 ~]# /elk/logstash/bin/logstash -f grok.conf 
Settings: Default pipeline workers: 1
Pipeline main started
55.3.244.1 GET /index.html 15824 0.043   # 输入左边内容，下面就会自动生成字典的形式
{
       "message" => "55.3.244.1 GET /index.html 15824 0.043",
      "@version" => "1",
    "@timestamp" => "2018-05-07T09:14:56.201Z",
          "host" => "elk-01.com",
        "client" => "55.3.244.1",
        "method" => "GET",
       "request" => "/index.html",
         "bytes" => "15824",
      "duration" => "0.043"
}

其实上面使用的那些正则在程序中都有定义：

[root@elk-01 patterns]# cd /elk/logstash/vendor/bundle/jruby/1.9/gems/logstash-patterns-core-2.0.5/patterns
[root@elk-01 patterns]# ls
aws     bro   firewalls      haproxy  junos         mcollective           mongodb  postgresql  redis
bacula  exim  grok-patterns  java     linux-syslog  mcollective-patterns  nagios   rails       ruby
[root@elk-01 patterns]# cat grok-patterns

（10）mysql慢查询

编写日志采集文件：

[root@elk-01 ~]# cat mysql-slow.conf 
input {
    file {
        path => "/var/lib/mysql/elk-01-slow.log"
        type => "mysql-slowlog"
        codec => multiline {
            pattern => "^# User@Host"
            negate => true
            what => "previous"
        }
    }
}

filter {
    # drop sleep events
    grok {
        match => { "message" =>"SELECT SLEEP" }
        add_tag => [ "sleep_drop" ]
        tag_on_failure => [] # prevent default _grokparsefailure tag on real records
    }
    if "sleep_drop" in [tags] {
        drop {}
    }
    grok {
        match => [ "message", "(?m)^# User@Host: %{USER:user}\[[^\]]+\] @ (?:(?<clienthost>\S*) )?\[(?:%{IP:clientip})?\]\s+Id: %{NUMBER:row_id:int}\s*# Query_time: %{NUMBER:query_time:float}\s+Lock_time: %{NUMBER:lock_time:float}\s+Rows_sent: %{NUMBER:rows_sent:int}\s+Rows_examined: %{NUMBER:rows_examined:int}\s*(?:use %{DATA:database};\s*)?SET timestamp=%{NUMBER:timestamp};\s*(?<query>(?<action>\w+)\s+.*)\n#\s*" ]
    }
    date {
        match => [ "timestamp", "UNIX" ]
        remove_field => [ "timestamp" ]
    }
}

output {
    stdout {
       codec =>"rubydebug"
    }
}

安装好mysql数据库以后，在配置文件my.cnf

[root@elk-01 ~]# vim /etc/my.cnf   --在[mysqld]组里面添加如下内容
slow_query_log      # 打开慢查询日志
slow_query_log_file=/var/lib/mysql/elk-01-slow.log   # 指定慢查询日志所在文件
long_query_time=2               # 设一个阈值，要大于这个值才会记录，等于该值时不记录。
log_queries_not_using_indexes   # 若是运行的SQL语句没有使用索引，则MySQl数据库一样会将这条SQL语句记录到慢查询日志文件

对上面的采集文件进行执行：

[root@elk-01 ~]# /elk/logstash/bin/logstash -f /root/mysql-slow.conf
Settings: Default pipeline workers: 1
Pipeline main started
{
    "@timestamp" => "2018-05-08T04:19:49.871Z",
       "message" => "/usr/libexec/mysqld, Version: 5.5.52-MariaDB (MariaDB Server). started with:\nTcp port: 0  Unix socket: /var/lib/mysql/mysql.sock\nTime                 Id Command    Argument\n# Time: 180508 12:19:47",
      "@version" => "1",
          "tags" => [
        [0] "multiline",
        [1] "_grokparsefailure"
    ],
          "path" => "/var/lib/mysql/elk-01-slow.log",
          "host" => "elk-01.com",
          "type" => "mysql-slowlog"
}

再次把mysql-slow.conf文件的内容添加到总文件file.conf中，或者修改mysql-slow.conf文件，让数据展现在ES和Kibana中：

[root@elk-01 ~]# cat mysql-slow.conf 
input {
    file {
        path => "/var/lib/mysql/elk-01-slow.log"
        type => "mysql-slowlog"
        start_position => "beginning"
        codec => multiline {
            pattern => "^# User@Host"
            negate => true
            what => "previous"
        }
    }
}

filter {
    # drop sleep events
    grok {
        match => { "message" =>"SELECT SLEEP" }
        add_tag => [ "sleep_drop" ]
        tag_on_failure => [] # prevent default _grokparsefailure tag on real records
    }
    if "sleep_drop" in [tags] {
        drop {}
    }
    grok {
        match => [ "message", "(?m)^# User@Host: %{USER:user}\[[^\]]+\] @ (?:(?<clienthost>\S*) )?\[(?:%{IP:clientip})?\]\s+Id: %{NUMBER:row_id:int}\s*# Query_time: %{NUMBER:query_time:float}\s+Lock_time: %{NUMBER:lock_time:float}\s+Rows_sent: %{NUMBER:rows_sent:int}\s+Rows_examined: %{NUMBER:rows_examined:int}\s*(?:use %{DATA:database};\s*)?SET timestamp=%{NUMBER:timestamp};\s*(?<query>(?<action>\w+)\s+.*)\n#\s*" ]
    }
    date {
        match => [ "timestamp", "UNIX" ]
        remove_field => [ "timestamp" ]
    }
}

output {
    if [type] == " mysql-slowlog"{
        elasticsearch {
            hosts => ["192.168.8.55:9200"]
            index => "mysql-slowlog-%{+YYYY.MM.dd}"
        }
    }
}

对上面的采集文件进行执行：

[root@elk-01 ~]# /elk/logstash/bin/logstash -f /root/mysql-slow.conf --configtest
Configuration OK
[root@elk-01 ~]# /elk/logstash/bin/logstash -f /root/mysql-slow.conf &

登录Elasticsearch主界面查看：

登录Kibana主界面查看：

接下来，咱们试想一个问题：
若是咱们的Elasticsearch出现了问题，那就不能进行日志采集处理了！
这种状况下该怎么办呢？
解决方案：
能够在Client和Elasticsearch之间添加一个中间件做为缓存，先将采集到的日志内容写到中间件上，而后再从中间件输入到Elasticsearch中。
这样，就完美的解决了上述的问题了。

七、ELK中使用redis做为中间件，缓存日志采集数据

（1）redis的配置和启动

# 下载解压
[root@elk-01 ~]# wget http://download.redis.io/releases/redis-4.0.8.tar.gz
[root@elk-01 ~]# tar zxvf redis-4.0.8.tar.gz -C /usr/src/
[root@elk-01 ~]# cd /usr/src/redis-4.0.8/
# 进入解压后的文件目录，以后直接编译便可（redis安装相对简单）
[root@elk-01 redis-4.0.8]# make
[root@elk-01 redis-4.0.8]# make install
[root@elk-01 redis-4.0.8]# cd utils
[root@elk-01 utils]# ./install_server.sh
Selected config:
Port           : 6379
Config file    : /etc/redis/6379.conf
Log file       : /var/log/redis_6379.log
Data dir       : /var/lib/redis/6379
Executable     : /usr/local/bin/redis-server
Cli Executable : /usr/local/bin/redis-cli

# 修改配置文件和启动
[root@elk-01 ~]# vim /etc/redis/6379.conf   --修改下面内容
# bind 127.0.0.1   # 注释掉
protected-mode no  # 保护模式为no
daemonize yes  # 守护进程模式为yes

# 启动
[root@elk-01 ~]# redis-server /etc/redis/6379.conf
[root@elk-01 ~]# lsof -i:6379
COMMAND     PID USER   FD   TYPE DEVICE SIZE/OFF NODE NAME
redis-ser 27502 root    6u  IPv6 380301      0t0  TCP *:6379 (LISTEN)
redis-ser 27502 root    7u  IPv4 380302      0t0  TCP *:6379 (LISTEN)
# 登录
[root@elk-01 ~]# redis-cli -h 192.168.8.55 
192.168.8.55:6379> info
# Server
redis_version:4.0.8
……

（2）编写从Client端采集数据的文件

编写日志采集文件：

[root@elk-01 ~]# cat redis-out.conf
input {
   stdin {}
}

output {
   redis {
      host => "192.168.8.55"
      port => "6379"
      db => "6"
      data_type => "list"
      key => "demo"
   }
}

（3）执行采集数据的文件，并输入数据hello redis

[root@elk-01 ~]# /elk/logstash/bin/logstash -f redis-out.conf
Settings: Default pipeline workers: 1
Pipeline main started    # 下面输入数据hello redis
hello redis

（4）在redis中查看数据

从新开启一个窗口，在redis中查看数据：

[root@elk-01 ~]# redis-cli -h 192.168.8.55
192.168.8.55:6379> info  # 输入内容
# Server
redis_version:4.0.8
……
……
# Keyspace   # 在最下面一行，显示是db6
db6:keys=1,expires=0,avg_ttl=0
192.168.8.55:6379> select 6  # 输入内容
OK
192.168.8.55:6379[6]> keys *  # 输入内容
1) "demo"
192.168.8.55:6379[6]> LINDEX demo -1  # 输入内容
"{\"message\":\"hello redis\",\"@version\":\"1\",\"@timestamp\":\"2018-05-08T08:53:16.806Z\",\"host\":\"elk-01.com\"}"

（5）继续自定义输入数据

[root@elk-01 ~]# /elk/logstash/bin/logstash -f redis-out.conf 
Settings: Default pipeline workers: 1
Pipeline main started
hello redis
123456
xiaozuo
51cto
test
haha

（6）在redis中查看

在redis中查看长度：

[root@elk-01 ~]# redis-cli -h 192.168.8.55
192.168.8.55:6379> info  # 输入内容
# Server
redis_version:4.0.8
……
……
# Keyspace
db6:keys=1,expires=0,avg_ttl=0  # 显示是db6
192.168.8.55:6379> select 6  # 输入内容
OK
192.168.8.55:6379[6]> keys *  # 输入内容
1) "demo"
192.168.8.55:6379[6]> LLEN demo  # 输入内容
(integer) 6  # 刚才自定义输入了6个字符串

（7）将redis中的内容写到ES中

编写日志采集文件：

[root@elk-01 ~]# cat redis-in.conf 
input {
    redis {
        host => "192.168.8.55"
        port => "6379"
        db => "6"
        data_type => "list"
        key => "demo"
    }
}

output {
    elasticsearch {
        hosts => ["192.168.8.55:9200"]
        index => "redis-in-%{+YYYY.MM.dd}"
    }
}

对上面的采集文件进行执行：

[root@elk-01 ~]# /elk/logstash/bin/logstash -f /root/redis-in.conf --configtest
Configuration OK
[root@elk-01 ~]# /elk/logstash/bin/logstash -f /root/redis-in.conf &

在redis中查看，发现数据已被读出：

[root@elk-01 ~]# redis-cli -h 192.168.8.55
192.168.8.55:6379> LLEN demo
(integer) 0

登录Elasticsearch主界面查看：

（8）将采集到的全部日志写入到redis中

我这里从新定义一个添加redis缓存后的总文件shipper.conf。（能够将以前执行的总文件file.conf停掉）
编写总采集shipper.conf文件：

[root@elk-01 ~]# cat shipper.conf 
input {
    file {
        path => "/var/log/messages"
        type => "system"
        start_position => "beginning"
    }
    file {
        path => "/var/log/nginx/access_json.log"
        codec => json
        start_position => "beginning"
        type => "nginx-log"
    }
    syslog {
        type => "system-syslog"
        host => "192.168.8.55"
        port => "514"
    }
}

output {
    if [type] == "system"{
        redis {
            host => "192.168.8.55"
            port => "6379"
            db => "6"
            data_type => "list"
            key => "system"
        }
    }
    if [type] == "nginx-log"{   
        redis {
            host => "192.168.8.55"
            port => "6379"
            db => "6"
            data_type => "list"
            key => "nginx-log"
        }
    }
    if [type] == "system-syslog"{
        redis {
            host => "192.168.8.55"
            port => "6379"
            db => "6"
            data_type => "list"
            key => "system-syslog"
        }   
    }
}

对上面的采集文件进行执行（提早将上面以前启动的file.conf文件的执行给结束掉！）

[root@elk-01 ~]# /elk/logstash/bin/logstash -f /root/shipper.conf --configtest
Configuration OK
[root@elk-01 ~]# /elk/logstash/bin/logstash -f /root/shipper.conf &

在redis中查看：

[root@elk-01 ~]# redis-cli -h 192.168.8.55
192.168.8.55:6379> info  # 输入内容
# Server
redis_version:4.0.8
……
……
# Keyspace
db6:keys=1,expires=0,avg_ttl=0  # 显示是db6
192.168.8.55:6379> select 6   # 输入内容
OK
192.168.8.55:6379[6]> keys *  # 输入内容
1) "system"
192.168.8.55:6379[6]> keys *  # 输入内容
1) "nginx-log"
2) "system"

从新开启一个窗口，自定义添加日志数据：

[root@elk-01 ~]# logger "Hello World"
[root@elk-01 ~]# logger "123456"
[root@elk-01 ~]# logger "test"
[root@elk-01 ~]# logger "hello"
[root@elk-01 ~]# logger "hello"
[root@elk-01 ~]# logger "hello"

回到原来的窗口（即上面执行redis命令的终端），查看内容：

192.168.8.55:6379[6]> keys *  # 输入内容
1) "system-syslog"   # 新增的日志
2) "nginx-log"
3) "system"

其实能够在任意的一台ES中，将数据从redis读取到ES中。
下面咱们在elk-02节点上，将数据从redis读取到ES中：
编写日志采集文件：

[root@elk-02 ~]# cat file.conf 
input {
    redis {
        type => "system"
        host => "192.168.8.55"
        port => "6379"
        db => "6"
        data_type => "list"
        key => "system"
    }
    redis {
        type => "nginx-log"
        host => "192.168.8.55"
        port => "6379"
        db => "6"
        data_type => "list"
        key => "nginx-log"
    }
    redis {
        type => "system-syslog"
        host => "192.168.8.55"
        port => "6379"
        db => "6"
        data_type => "list"
        key => "system-syslog"
    }   
}

output { 
    if [type] == "system"{
        elasticsearch {
            hosts => ["192.168.8.55:9200"]
            index => "system-%{+YYYY.MM.dd}"
        }
    }
    if [type] == "nginx-log"{
        elasticsearch {
            hosts => ["192.168.8.55:9200"]
            index => "nignx-log-%{+YYYY.MM.dd}"
        }
    }
    if [type] == "system-syslog"{
        elasticsearch {
            hosts => ["192.168.8.55:9200"]
            index => "system-syslog-%{+YYYY.MM.dd}"
        }
    }
}

对上面的日志采集文件进行执行：

[root@elk-02 ~]# /elk/logstash/bin/logstash -f /root/file.conf --configtest
Configuration OK
[root@elk-02 ~]# /elk/logstash/bin/logstash -f /root/file.conf &

在redis中查看，发现数据已经被读出到Elasticsearch中了。

[root@elk-01 ~]# redis-cli -h 192.168.8.55
192.168.8.55:6379> keys *
(empty list or set)

同时登录Logstash和Kibana看，发现能够正常采集到日志了。

也能够启动多个redis写到ES中，具体根据本身的实际状况而定。

当客户机的日志信息收集后，通过redis刚读到ES数据库里后，若是没有新数据写入，则默认在ES的访问界面里是看不到数据的。只有当日志文件里有新的日志写入后才会触发数据展现的动做，即ES的访问界面（http://192.168.8.55:9200/_plugin/head/）里才能看到日志数据的展现效果。

4、配置示例

# 192.168.8.55为ELK的master节点，同时也是redis节点
# 编写日志采集文件
[root@elk-01 ~]# cat /conf/tomacat-log.conf 
input {
    file {
        path => "/usr/local/tomcat/logs/catalina.out"
        type => "tomcat-logs"
        start_position => "beginning"
        codec => multiline {
            pattern => "^\["           # 表示采集以“[”开头的日志信息
            negate => true
            what => "previous"
        }
    }
}

output {
    if [type] == "tomcat-logs"{
        redis {
            host => "192.168.8.55"
            port => "6379"
            db => "1"
            data_type => "list"
            key => "tomcat-logs"
        }
    }
}

[root@elk-01 ~]# cat /conf/system-log.conf 
input {
    file {
        path => "/var/log/messages"
        type => "systemlog"
        start_position => "beginning"
        stat_interval => "2"
    }
}

output {
    if [type] == "systemlog" {
        redis {
            data_type => "list"
            host => "192.168.10.44"
            db => "2"
            port => "6379"
            key => "systemlog"
        }
    }
}

[root@elk-01 ~]# cat /conf/file.conf 
input {
    redis {
        type => "tomcat8-logs"
        host => "192.168.10.44"
        port => "6379"
        db => "1"
        data_type => "list"
        key => "tomcat8-logs"
    }

    redis {
        type => "systemlog"
        host => "192.168.10.44"
        port => "6379"
        db => "2"
        data_type => "list"
        key => "systemlog"
    }
}

output {
    if [type] == "tomcat8-logs"{
        elasticsearch {
            hosts => ["192.168.10.44:9200"]
            index => "elk-node2-tomcat8-logs-%{+YYYY.MM.dd}"
        }
    }

    if [type] == "systemlog"{
        elasticsearch {
            hosts => ["192.168.10.44:9200"]
            index => "elk-node2-systemlog-%{+YYYY.MM.dd}"
        }
    }
}

# 验证
[root@elk-01 ~]# /elk/logstash/bin/logstash -f /conf/tomacat-log.conf --configtest
Configuration OK
[root@elk-01 ~]# /elk/logstash/bin/logstash -f /conf/system-log.conf --configtest
Configuration OK
[root@elk-01 ~]# /elk/logstash/bin/logstash -f /conf/file.conf --configtest
Configuration OK

# 执行
[root@elk-01 ~]# /elk/logstash/bin/logstash -f /conf/tomacat-log.conf &
[root@elk-01 ~]# /elk/logstash/bin/logstash -f /conf/system-log.conf &
[root@elk-01 ~]# /elk/logstash/bin/logstash -f /conf/file.conf &

当/usr/local/tomcat/logs/catalina.out和/var/log/messages文件里有新日志信息写入时，就会触发动做，在redis里就能查看到相关信息，并查看写入到ES里。