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" #合并上一行內容
}
}
}

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}" #指定一個索引
}
}
}

測試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"
}
}

以上所有配置文件配置完成后啟動方式同下：

/opt/logstash/bin/logstash -f /etc/logstash/conf.d/*.conf &

效果圖如下：

生產里面抓的一份數據，做分析統計，效果圖如下：

圖中可以清楚的看到訪問量大的IP，訪問返回狀態等等信息