ELK日志分析系统搭建配置

时间：2017-06-21 19:48:37 阅读：236 评论：0 收藏：0 [点我收藏+]

标签：elk

我们主要用ELK日志分析系统来分析Nginx访问日志，mysql慢查询日志，tomcat运行日志以及系统日志等。

介绍：
ELK：ElasticSearch+LogStash+Kibana=ElkStack
ElasticSearch:存储、收索、分析（可以用solr替代）
LogStash：收集器，输入，处理分析，存储到ES
Kibana：展示
备注：ElasticSearch支持集群功能，日志收集后会在每个节点存放一份（可以选择）

1、安装jdk
wget http://sg-new.oss-cn-hangzhou.aliyuncs.com/jdk1.8.0_102.tgz
tar -zxvf jdk1.8.0_102.tgz -C /App/java
----------------------------------------------------
vim /etc/profile
#set for java
export JAVA_HOME=/App/java/jdk1.8.0_102
export PATH=$JAVA_HOME/bin:$PATH
export CLASSPATH=.:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar
export LD_LIBRARY_PATH=$LD_LIBRARY_PATH:/usr/local/apr/lib
----------------------------------------------------
source /etc/profile
java -version

2、下载安装elasticsearch（可以部署分布式），启动
rpm --import https://packages.elastic.co/GPG-KEY-elasticsearch
echo "
[elasticsearch-2.x]
name=Elasticsearch repository for 2.x packages
baseurl=http://packages.elastic.co/elasticsearch/2.x/centos
gpgcheck=1
gpgkey=http://packages.elastic.co/GPG-KEY-elasticsearch
enabled=1" >> /etc/yum.repos.d/elasticsearch.repo
yum install elasticsearch -y

mkdir /data/elk/{data,logs} -p
vi /etc/elasticsearch/elasticsearch.yml
cluster.name: es #集群名称（一个集群必须是同一个名称）
node.name: es-node1 #节点名称
path.data: /data/elk/data
path.logs: /data/elk/logs
bootstrap.mlockall: true #设置成ture，锁住内存（不交互到swap）
network.host: 0.0.0.0
http.port: 9200
#discovery.zen.ping.unicast.hosts: ["192.168.2.215", "host2"]
启动：
启动前注意文件夹权限
/etc/init.d/elasticsearch start
-----------------------------
测试：此时可以访问：http://192.168.88.48:9200/
访问结果：
{
"name" : "Bombshell",
"cluster_name" : "es",
"cluster_uuid" : "Rueqwrx2TjaKp24QJDt4wg",
"version" : {
"number" : "2.4.5",
"build_hash" : "c849dd13904f53e63e88efc33b2ceeda0b6a1276",
"build_timestamp" : "2017-04-24T16:18:17Z",
"build_snapshot" : false,
"lucene_version" : "5.5.4"
},
"tagline" : "You Know, for Search"
}

3、安装elasticsearch插件
安装head插件（集群管理插件）
cd /usr/share/elasticsearch/bin/
./plugin install mobz/elasticsearch-head
ll /usr/share/elasticsearch/plugins/head
测试插件：
http://192.168.88.48:9200/_plugin/head/

安装插件kopf（集群资源查看监控和查询插件）
/usr/share/elasticsearch/bin/plugin install lmenezes/elasticsearch-kopf
http://192.168.88.48:9200/_plugin/kopf
重启elasticearch
/etc/init.d/elasticsearch restart

重点：
如果做集群，其他配置一样
mkdir /data/elk/{data,logs}
vi /etc/elasticsearch/elasticsearch.yml
cluster.name: es #集群名称（一个集群必须是同一个名称）
node.name: es-node2 #节点名称
path.data: /data/elk/data
path.logs: /data/elk/logs
bootstrap.mlockall: true #设置成ture，锁住内存（不交互到swap）
network.host: 0.0.0.0
http.port: 9200
#discovery.zen.ping.unicast.hosts: ["192.168.2.215", "host2"]
-------------------------------------
出现群集连接不上问题（只能出现一个节点，一个丢失），一个数据被分片成5份
问题1、锁住内存，因为是普通用户，所以使用内存有限制
vim /etc/security/limits.conf
elasticsearch soft memlock unlimited
elasticsearch hard memlock unlimited
注意用户打开文件数ulimit值 65536
问题2：主播方式，默认是组播，连接集群会出现问题，改成单播
discovery.zen.ping.multicast.enabled: false
discovery.zen.ping.unicast.hosts: ["192.168.2.215", "host2"]
问题3：权限问题
chown -R elasticsearch:elasticsearch /data/elk/
此时集群功能完成

4、安装kibana
wget https://download.elastic.co/kibana/kibana/kibana-4.5.1-linux-x64.tar.gz
tar zxvf kibana-4.5.1-linux-x64.tar.gz
mv kibana-4.5.1-linux-x64 /usr/local/kibana
vi /etc/rc.local
/usr/local/kibana/bin/kibana > /var/log/kibana.log 2>&1 &
vi /usr/local/kibana/config/kibana.yml
server.port: 5601
server.host: "192.168.88.48"
elasticsearch.url: "http://192.168.88.48:9200"
每个版本下面有这么一行内容，一定要注意这些内容
启动服务
/usr/local/kibana/bin/kibana &

5、安装logstash
在logstash中，包括了三个阶段:
输入input --> 处理filter（不是必须的） --> 输出output
rpm --import https://packages.elastic.co/GPG-KEY-elasticsearch
echo "
[logstash-2.1]
name=Logstash repository for 2.1.x packages
baseurl=http://packages.elastic.co/logstash/2.1/centos
gpgcheck=1
gpgkey=http://packages.elastic.co/GPG-KEY-elasticsearch
enabled=1" >> /etc/yum.repos.d/logstash.repo
yum install logstash -y
通过配置验证Logstash的输入和输出
测试语法：-e输入命令，前台运行
/opt/logstash/bin/logstash -e ‘input{stdin{}}output{stdout{codec=>rubydebug}}‘
输入my name is caicai. 回车

测试1：基于屏幕输入测试，同上面一样的，只是配置使用配置文件
vim /etc/logstash/conf.d/stdout.conf
input {
stdin {}
}

output {
stdout {
codec => "rubydebug"
}
}
启动：/opt/logstash/bin/logstash -f /etc/logstash/conf.d/stdout.conf
------------------------------------------------------------

测试2：logstash结合es，数据写入到es：注意端口（老版本不需要）
vim /etc/logstash/conf.d/stdout.conf
input {
stdin {}
}

output {
elasticsearch {
hosts => ["192.168.88.48:9200","hosts节点2IP"]
#protocol => "http"(此版本没有协议一说)
}
stdout { codec=> rubydebug } #可以不要，标准输出到屏幕
}
此时http://192.168.88.48:9200/_plugin/head/中可以看到索引和具体数据

测试3：搜集系统日志：
input {
file {
path => "/var/log/messages"
type => "system"
start_position => "beginning" #从头开始收集
}
}

output {
elasticsearch {
hosts => ["192.168.88.48:9200"]
index => "system-%{+YYYY.MM.dd}" #指定一个索引
}
}

测试4：收集java异常日志，接着上面的,做判断
input {
file {
path => "/var/log/messages"
type => "system"
start_position => "beginning" #从头开始收集
}
file {
path => "/logs/java/java.log"
type => "es-error"
start_position => "beginning" #从头开始收集
codec => multilinc { #默认收集是一行一个事件，加参数后合并成一个事件
pattern => "^\[" #分隔符
negate => ture
what => "previous" #合并上一行内容
}
}
}

测试5：收集nginx日志
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"‘
‘}‘;

------------------------------------------------
input {
file {
path => "/var/log/messages"
type => "system"
start_position => "beginning" #从头开始收集
}
file {
path => "/logs/nginx/lux.cngold.org.access.log"
codec => "json"

start_position => "beginning" #从头开始收集
type => "nginx-log"
}
file {
path => "/logs/java/java.log"
type => "es-error"
start_position => "beginning" #从头开始收集
codec => multilinc { #默认收集是一行一个事件，加参数后合并成一个事件
pattern => "^\[" #分隔符
negate => ture
what => "previous" #合并上一行内容
}
}
}

output {
if [type] == "system"{
elasticsearch {
hosts => ["192.168.88.48:9200"]
index => "system-%{+YYYY.MM.dd}" #指定一个索引
}
}
if [type] == "es-error"{
elasticsearch {
hosts => ["192.168.88.48:9200"]
index => "es-error-%{+YYYY.MM.dd}" #指定一个索引
}
}
if [type] == "nginx-log"{
elasticsearch {
hosts => ["192.168.88.48:9200"]
index => "nginx-log-%{+YYYY.MM.dd}" #指定一个索引
}
stdout {
codec=> rubydebug
}
}
}

出问题测试用的：------------------------------------
nput {
file {
path => ["/logs/nginx/80-access.log"]
codec => "json"
start_position => "beginning" #从头开始收集
type => "nginx-log"
}
}

output {
if [type] == "nginx-log"{
elasticsearch {
hosts => ["192.168.88.48:9200"]
index => "nginx-80-log-%{+YYYY.MM.dd}" #指定一个索引
}
}
stdout {
codec=> rubydebug
}
}
------------------------------------------------------

测试6：使用syslog收集系统日志
vim /etc/rsyslog.conf 设置让文件发送到514端口上
*.* @@192.168.88.48:514 #将日志发送给这个主机的这个端口
/etc/init.d/rsyslog restart
配置文件
vim /etc/logstash/conf.d/04-syslog.conf
input {
syslog {
type => "system-syslog"
host => "192.168.88.48"
port => "514"
}
}

output {
if [type] == "system-syslog" {
elasticsearch {
hosts => ["192.168.88.48:9200"]
index => "system-syslog-%{+YYYY.MM.dd}"
}
stdout {
codec=> rubydebug
}
}
}
重启rsyslog就会有输出了

测试7：tcp日志收集
vim /etc/logstash/conf.d/05-tcp.conf
input {
tcp {
host => "192.168.88.48"
port => "6666"
}
}
output {
stdout {
codec => "rubydebug"
}
}
使用nc对6666端口写入数据
nc 192.168.88.48 6666 </var/log/yum.log
将信息输入到tcp的伪设备中
echo "chuck" >/dev/tcp/192.168.88.48/6666

----------------------------------------------
apache不支持json，所以引入grok正则表达式
使用grok必须要保证有插件：位置
/opt/logstash/vendor/bundle/jruby/1.9/gems/logstash-patterns-core-2.0.2/patterns
[root@linux-node1 ~]# cat grok.conf
input {
stdin {}
}
filter {
grok {
match => { "message" => "%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration}" }
}
}
output {
stdout {
codec => "rubydebug"
}
}

输入测试：55.3.244.1 GET /index.html 15824 0.043，此时有输出，格式为正则格式

测试8，使用logstash正则表达式收集mysql的slowlog（慢查询）mysql5.6.21版本
问题：多行合并插件codec => multilinc
vim /etc/logstash/conf.d/07-mysql-slow.conf
input{
file {
path => "/root/slow.log"
type => "mysql-slow-log"
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 {
stdout{
codec => "rubydebug"
}
}