一、 简介
1.1 介绍
ELK是三个开源工具组成，简单解释如下：
Elasticsearch是个开源分布式搜索引擎，它的特点有：分布式，零配置，自动发现，索引自动分片，索引副本机制，restful风格接口，多数据源，自动搜索负载等。
Logstash是一个完全开源的工具，它可以对你的日志进行收集、过滤，并将其存储供以后使用（如，搜索）。
Kibana 也是一个开源和免费的工具，它可以为 Logstash 和 ElasticSearch 提供的日志分析友好的 Web 界面，可以帮助您汇总、分析和搜索重要数据日志。
1.2 场景分析
日志主要包括系统日志、应用程序日志和安全日志等等。运维人员和开发人员可以通过日志了解服务器软硬件信息、检查配置过程中的错误及错误发生的原因。经常分析日志可以了解服务器的负荷，性能安全性，从而及时采取措施纠正错误。
通常，日志被分散的储存不同的设备上。如果你管理数十上百台服务器，你还在使用依次登录每台机器的传统方法查阅日志。这样是不是感觉很繁琐和效率低下。当务之急我们使用集中化的日志管理，例如：开源的syslog，将所有服务器上的日志收集汇总。
集中化管理日志后，日志的统计和检索又成为一件比较麻烦的事情，一般我们使用grep、awk和wc等Linux命令能实现检索和统计，但是对于要求更高的查询、排序和统计等要求和庞大的机器数量依然使用这样的方法难免有点力不从心。
这里采用开源实时日志分析ELK平台能够完美的解决我们上述的问题，当然也还有别的平台或者工具可以使用，这里只讨论ELK，官方网站：https://www.elastic.co
二、安装Elasticsearch
2.1 安装jdk
|
1
2
3
4
|
java version
"1.8.0_121"
Java(TM) SE Runtime Environment (build
1.8
.
0_121
-
b13)
Java HotSpot(TM)
64
-
Bit Server VM (build
25.121
-
b13, mixed mode)
|
2.2 安装Elasticsearch
编辑配置文件
|
1
2
3
4
5
6
7
8
9
|
cluster.name: elk
-
application
node.name: node
-
1
path.data:
/
data
/
elasticsearch
/
data
path.logs:
/
data
/
elasticsearch
/
logs
network.host:
172.16
.
220.248
http.port:
9200
discovery.zen.ping.unicast.hosts: [
"node-1"
]
discovery.zen.minimum_master_nodes:
1
|
添加elasticsearch用户，不能使用root启动
修改sysctl.conf文件
|
1
2
3
4
|
vm.max_map_count
=
262144
|
修改/etc/security/limits.conf文件，修改打开文件句柄
|
1
2
3
4
|
*
soft nofile
100000
*
hard nofile
100000
*
soft nproc
100000
*
hard nproc
100000
|
添加hosts文件
|
1
2
|
172.16
.
220.248
node
-
1
|
启动
查看是否启动
![]()
简单的curl测试
![]()
三、安装Logstash和filebeat
filebeat用于在各个服务器上获取数据，发送到logstash上，再由logstash处理数据。
3.1 安装logstash
3.2 安装filebeat
下载filebeat并启动，通过它来监听数据源文件的新增内容经过logstash处理后上传到es里面
编辑filebeat.yml文件
|
1
2
3
4
5
6
|
filebeat.prospectors:
-
input_type: log
paths:
-
/
var
/
log
/
message
-
log
output.logstash:
hosts: [
"172.16.220.248:5044"
]
|
启动filebeat服务
查看启动，filebeat没有监听端口，主要看日志和进程
![]()
filebeat监听的文件记录信息在/data/filebeat/data/registry
新建一个本地文件message-log，可以取几条本机系统的messages文件
![]()
3.3 启动logstash
最后新建一个logstash的启动指定test.conf配置文件，内容如下：
|
1
2
3
4
5
6
7
8
9
10
11
|
input
{
beats {
port
=
>
"5044"
}
}
output {
elasticsearch {
hosts
=
>
"172.16.220.248:9200"
}
stdout { codec
=
> rubydebug }
}
|
Logstash默认有input、filter、output三个区域，一般最少需要配置input和output即可！
logstash的本身默认的logstash.yml配置文件选择不修改即可！
简单测试一下logstash不指定配置文件启动
![]()
我们手动输入 hello world，它也会输出 hello world
指定配置文件启动logstash
查看5044端口和9600端口是否开启
![]()
等待一会后应该会出现如下信息输出，这也就是test.conf里面最后一行定义输出到屏幕上
![]()
四、安装kibana
编辑kibana.yml配置文件
|
1
2
3
4
|
server.port:
5601
server.host:
"172.16.220.248"
elasticsearch.url:
"http://172.16.220.248:9200"
|
启动kinbana
查看端口
![]()
浏览器登入查看
![]()
点击create按钮后，然后点击上面的discover按钮，注意如果没数据的话，注意看看导入的时间@timestamp和现在的时间对比一下，kibana默认只显示最近15分钟的数据，如果超出15分钟请选择适当的时间，从kibana可以看到messages-log里面的15条数据都正常导入了。这就也完成我们的实现的第一个效果。但是这仅仅是把流程跑通了，接下来我们需要做的事情还有更多。注意只能先导入数据到es后才能在kibana创建索引。
![]()
五、获取Nginx access日志
Nginx日志格式在logstash的grok里面默认是没有的，需要我们手动配置，可以通过http://grokdebug.herokuapp.com/ 在线工具来判断配置是否正确。
5.1 在nginx服务器上安装filebeat
服务器： 172.16.200.160
|
1
2
3
4
|
# tar -zxvf filebeat-5.6.3-linux-x86_64.tar.gz
# mv filebeat-5.6.3-linux-x86_64 /data/filebeat
# cd /data/filebeat
# cp filebeat.yml filebeat.yml.bak
|
修改filebeat配置文件
|
1
2
3
4
5
6
7
8
|
# cat filebeat.yml | grep -v ^$ | grep -v ^# | grep -v "#"
filebeat.prospectors:
- input_type: log
paths:
- /data/nginx/logs/160_access.log
document_type: nginx_access
output.logstash:
hosts: ["172.16.220.248:5044"]
|
启动filebeat
5.2 重新配置logstash启动配置文件
nginx日志格式，根据业务要求，我们这做了一些修改，比如增加cookie等，修改access.log日志时间格式等，这个会在另外的博客中写出来，会给出链接的。
nginx日志中添加cookie信息
nginx改变access.log中的时间格式
Nginx日志格式
|
1
2
3
4
5
6
7
8
|
log_format main
'[$time_local] - $remote_addr:$remote_port - $upstream_addr $upstream_status $upstream_response_time - '
'"$request" $status $bytes_sent $request_time '
'"$http_referer" - "$http_user_agent" - '
'"$customerTag_cookie" - "$ym_cookie" - "$http_cookie" '
'"$http_x_forwarded_for"';
# 这里只是我们自己的格式，各位可以根据自己要求增删
|
grok使用表达式
可能我理解不是很到位，写的也比较复杂，我会把匹配对应项一一写出来，大家可以自己理解，然后为自己的项目配置
|
1
|
%{SYSLOG5424SD} - %{IPV4:clientip}:%{NUMBER:clientport} - %{IPV4:hostip}:%{NUMBER:itemport} %{INT:upstream_status} %{NUMBER:response_time} - \"%{WORD:method} %{URIPATHPARAM:request} HTTP/%{NUMBER:http_version}\" %{INT:status} %{INT:body_bytes_sent} %{NUMBER:request_time} %{QS:url} - %{QS:user_agent} - %{QS:customerTag} - %{QS:ym_traffic_session_id} - %{QS:all_cookies} %{IPV4:realip}
|
grok匹配说明
|
|
时间格式 |
|
|
获取ip,clientip是自己命名的 |
|
|
NUMBER匹配数字 |
|
|
INT整形 |
|
|
WORD单词 |
|
|
获取请求内容request |
|
|
QS可以获取一段字符串 |
修改logstash启动配置文件
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
|
input {
beats {
port => "5044"
}
}
filter {
if [type] == "nginx_access" {
grok {
match => {"message" => "%{SYSLOG5424SD} - %{IPV4:clientip}:%{NUMBER:clientport} - %{IPV4:hostip}:%{NUMBER:itemport} %{INT:upstream_status} %{NUMBER:response_time} - \"%{WORD:method} %{URIPATHPARAM:request} HTTP/%{NUMBER:http_version}\" %{INT:status} %{INT:body_bytes_sent} %{NUMBER:request_time} %{QS:url} - %{QS:user_agent} - %{QS:customerTag} - %{QS:ym_traffic_session_id} - %{QS:all_cookies} %{IPV4:realip}"}
}
mutate {
remove_field => "message" # 把message字段去掉，它会将上面单项获取的信息做个汇总，这样就重复了
}
}
}
output {
elasticsearch {
hosts => "172.16.220.248:9200"
}
#stdout { codec => rubydebug }
}
|
测试一下配置文件
|
1
|
./logstash -t -f ../config/logstash.conf
|
![]()
重启logstash
不出问题elasticsearch 和kibana中就会有数据了
![]()
