ELK(ElasticSearch+Logstash+ Kibana)搭建实时日志分析平台

一、简介

      ELK 由三部分组成elasticsearch、logstash、kibana，elasticsearch是一个近似实时的搜索平台,它让你以前所未有的速度处理大数据成为可能。

      Elasticsearch所涉及到的每一项技术都不是创新或者革命性的，全文搜索，分析系统以及分布式数据库这些早就已经存在了。它的革命性在于将这些独立且有用的技术整合成一个一体化的、实时的应用。Elasticsearch是面向文档(document oriented)的，这意味着它可以存储整个对象或文档(document)。然而它不仅仅是存储，还会索引(index)每个文档的内容使之可以被搜索。在Elasticsearch中，你可以对文档（而非成行成列的数据）进行索引、搜索、排序、过滤。这种理解数据的方式与以往完全不同，这也是Elasticsearch能够执行复杂的全文搜索的原因之一。

      应用程序的日志大部分都是输出在服务器的日志文件中，这些日志大多数都是开发人员来看，然后开发却没有登陆服务器的权限，如果开发人员需要查看日志就需要到服务器来拿日志，然后交给开发；试想下，一个公司有10个开发，一个开发每天找运维拿一次日志，对运维人员来说就是一个不小的工作量，这样大大影响了运维的工作效率，部署ELKstack之后，开发任意就可以直接登陆到Kibana中进行日志的查看，就不需要通过运维查看日志，这样就减轻了运维的工作。

      日志种类多，且分散在不同的位置难以查找：如LAMP/LNMP网站出现访问故障，这个时候可能就需要通过查询日志来进行分析故障原因，如果需要查看apache的错误日志，就需要登陆到Apache服务器查看，如果查看数据库错误日志就需要登陆到数据库查询，试想一下，如果是一个集群环境几十台主机呢？这时如果部署了ELKstack就可以登陆到Kibana页面进行查看日志，查看不同类型的日志只需要电动鼠标切换一下索引即可。

Logstash：日志收集工具，可以从本地磁盘，网络服务（自己监听端口，接受用户日志），消息队列中收集各种各样的日志，然后进行过滤分析，并将日志输出到Elasticsearch中。

Elasticsearch：日志分布式存储/搜索工具，原生支持集群功能，可以将指定时间的日志生成一个索引，加快日志查询和访问。

Kibana：可视化日志Web展示工具，对Elasticsearch中存储的日志进行展示，还可以生成炫丽的仪表盘。

二、安装部署(因为我是测试环境，就将ElasticSearch+Logstash+ Kibana装在一台虚拟机上面了)

1、安装jdk

2、安装elasticsearch

3、安装kibana

4、安装logstash

logstash是ELK中负责收集和过滤日志的

编写配置文件如下：

解释：

logstash的配置文件须包含三个内容：

input{}：此模块是负责收集日志，可以从文件读取、从redis读取或者开启端口让产生日志的业务系统直接写入到logstash

filter{}：此模块是负责过滤收集到的日志，并根据过滤后对日志定义显示字段

output{}：此模块是负责将过滤后的日志输出到elasticsearch或者文件、redis等

output直接输出到Elasticsearch

本环境需处理两套业务系统的日志

type：代表类型，其实就是将这个类型推送到Elasticsearch，方便后面的kibana进行分类搜索，一般直接命名业务系统的项目名

path：读取文件的路径

这个是代表日志报错时，将报错的换行归属于上一条message内容

start_position => "beginning"是代表从文件头部开始读取

filter{}中的grok是采用正则表达式来过滤日志，其中%{TIMESTAMP_ISO8601}代表一个内置获取2016-11-05 00:00:03,731时间的正则表达式的函数，%{TIMESTAMP_ISO8601:date1}代表将获取的值赋给date1，在kibana中可以体现出来

本环境有两条grok是代表，第一条不符合将执行第二条

其中index是定义将过滤后的日志推送到Elasticsearch后存储的名字

%{type}是调用input中的type变量(函数)

启动logstash

5、然后再一次启动 elasticsearch 和 kibana ，并登录kibana:

其中api-app-*和api-cxb-*从来的，*代表所有

代表实时收集的日志条数

红色框内的就是在刚才filter过滤规则中定义的

三、案例测试

1、测试logstash 收集tomcat上某项目的日志文件：coaLog.conf

input {
    file {
        type => "coa_msg"
        path => "/usr/local/tomcat/apache-tomcat-8.0.48/logs/coa/coa_msg.log"
        start_position => beginning
    }
}
filter {
 
}
output {
     elasticsearch { 
        hosts => "10.10.10.34:9200"
        index => "%{type}-%{+YYYY.MM.dd}"
        template_name => "%{type}"
     }
}

启动logstash：sh logstash -f coaLog.conf  &

2、logstash配置mysql数据同步到elasticsearch：我参考的(https://blog.csdn.net/hatlonely/article/details/79945539)

获取 jdbc mysql 驱动：

wget https://cdn.mysql.com//Downloads/Connector-J/mysql-connector-java-5.1.46.zip
unzip mysql-connector-java-5.1.46.zip

可以把解压后的mysql-connector-java-5.1.46-bin.jar复制到bin目录，方便启动

在bin目录新建配置文件    vim mysql-logstash.cnf ，写入：

input {
jdbc {
jdbc_driver_library => "mysql-connector-java-5.1.46-bin.jar"
jdbc_driver_class => "com.mysql.jdbc.Driver"
jdbc_connection_string => "jdbc:mysql://ip:3306/db_name"
jdbc_user => "jdbc_user"
jdbc_password => "jdbc_password"
schedule => "* * * * *"
statement => "SELECT * FROM table WHERE 时间字段 >= :sql_last_value"
use_column_value => true
tracking_column_type => "timestamp"
tracking_column => "时间字段"
last_run_metadata_path => "syncpoint_table"
}
}


output {
elasticsearch {
hosts => "localhost:9200"
index => "索引名称"
document_id => "%{主键字段}"
}
}

需要先把logstash停了，在重启

netstat -ntlp  命令查看端口为9600的进程，

然后kill 线程id，我的是 kill 25437

 

使用命令 sh logstash -f mysql-logstash.cnf  --path.data=/home/elk/logstash-6.4.2/logs &

启动logstash

完了在kibana可以看到同步过去的数据，我的是这样的： 

另外附上两个教程：

elasticsearch集成head插件查看es的数据：https://blog.csdn.net/mjlfto/article/details/79772848

kibana基础教程：https://www.elastic.co/guide/cn/kibana/current/introduction.html