ELK日志分析系統的搭建

一、環境準備

1.安裝java環境：

yum install java-1.8.0-openjdk* -y

2.添加elk執行用戶：

groupadd -g 77 elk
useradd -u 77 -g elk -d /home/elk -s /bin/bash elk

3.在 /etc/security/limits.conf 追加以下內容：

elk soft memlock unlimited
elk hard memlock unlimited
* soft nofile 65536
* hard nofile 131072

4.執行生效

sysctl -p

5.配置主機名

hostnamectl set-hostname monitor-elk
echo "10.135.3.135     monitor-elk" >> /etc/hosts

二、服務部署

1.服務端：

1）下載ELK相關的源碼包：

wget "https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-5.2.2.tar.gz"
wget "https://artifacts.elastic.co/downloads/logstash/logstash-5.2.2.tar.gz"
wget "https://artifacts.elastic.co/downloads/kibana/kibana-5.2.2-linux-x86_64.tar.gz"
wget "http://mirror.bit.edu.cn/apache/kafka/0.10.2.0/kafka_2.12-0.10.2.0.tgz"
wget "http://mirror.bit.edu.cn/apache/zookeeper/zookeeper-3.4.9/zookeeper-3.4.9.tar.gz"

2）創建elk目錄，并將以上源碼包解壓至該目錄：

mkdir /usr/local/elk
mkdir -p /data/elasticsearch/
chown -R elk.elk /data/elasticsearch/
mkdir -p /data/{kafka,zookeeper}
mv logstash-5.2.2 logstash && mv kibana-5.2.2-linux-x86_64 kibana && mv elasticsearch-5.2.2 elasticsearch && mv filebeat-5.2.2-linux-x86_64 filebeat && mv kafka_2.12-0.10.2.0 kafka && mv zookeeper-3.4.9 zookeeper
chown -R elk.elk /usr/local/elk/

程序目錄列表如下：

3）修改以下程序的相應配置文件

①kibana：

[root@monitor-elk ~]# cat /usr/local/elk/kibana/config/kibana.yml |grep -v "^#\|^$"
server.host: "localhost"
elasticsearch.url: "http://localhost:9200"
elasticsearch.requestTimeout: 30000
logging.dest: /data/elk/logs/kibana.log
[root@monitor-elk ~]#

②elasticsearch：

[root@monitor-elk ~]# cat /usr/local/elk/elasticsearch/config/elasticsearch.yml |grep -v "^#\|^$"
node.name: node01
path.data: /data/elasticsearch/data
path.logs: /data/elk/logs/elasticsearch
bootstrap.memory_lock: true
network.host: 127.0.0.1
http.port: 9200
[root@monitor-elk ~]# /usr/local/elk/elasticsearch/config/jvm.options
#修改以下參數
-Xms1g
-Xmx1g

③logstash：

[root@monitor-elk ~]# cat /usr/local/elk/logstash/config/logs.yml

input {
    #使用kafka的數據作為日志數據源
    kafka
    {
        bootstrap_servers => ["127.0.0.1:9092"]
        topics => "beats"
        codec => json
    }
}

filter {
   #過濾數據，如果日志數據里面包含有該IP地址，將會被丟棄
   if [message] =~ "123.151.4.10" {
       drop{}
   }

# 轉碼，轉成正常的url編碼，如中文
#   urldecode {
#       all_fields => true
#   }

# nginx access
   #通過type來判斷傳入的日志類型
   if [type] == "hongbao-nginx-access" or [type] == "pano-nginx-access" or [type] == "logstash-nginx-access" {
    grok {
        #指定自定義的grok表達式路徑
        patterns_dir => "./patterns"
        #指定自定義的正則表達式名稱解析日志內容，拆分成各個字段
        match => { "message" => "%{NGINXACCESS}" }
        #解析完畢后，移除默認的message字段
        remove_field => ["message"]
    }
    #使用geoip庫解析IP地址
    geoip {
        #指定解析后的字段作為數據源
        source => "clientip"
        fields => ["country_name", "ip", "region_name"]
    }
    date {
         #匹配日志內容里面的時間，如 05/Jun/2017:03:54:01 +0800
         match => ["timestamp","dd/MMM/yyyy:HH:mm:ss Z"]
         #將匹配到的時間賦值給@timestamp字段
         target => "@timestamp"
         remove_field => ["timestamp"]
    }
   }

#  tomcat access
  if [type] == "hongbao-tomcat-access" or [type] == "ljq-tomcat-access" {
    grok {
        patterns_dir => "./patterns"
        match => { "message" => "%{TOMCATACCESS}" }
        remove_field => ["message"]
    }
    geoip {
        source => "clientip"
        fields => ["country_name", "ip", "region_name"]
    }
    date {
         match => ["timestamp","dd/MMM/yyyy:HH:mm:ss Z"]
         target => "@timestamp"
         remove_field => ["timestamp"]
    }
   }

# tomcat catalina
  if [type] == "hongbao-tomcat-catalina" {
    grok {
     match => {
                "message" => "^(?<log_time>\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2},\d{3}) (?<level>\w*) (?<log_data>.+)"
            }        
        remove_field => ["message"]
    }
    date {
         match => ["log_time","yyyy-MM-dd HH:mm:ss,SSS"]
         target => "@timestamp"
         remove_field => ["log_time"]
    }
   }


}
 
output {

    #將解析失敗的記錄寫入到指定的文件中
    if "_grokparsefailure" in [tags] {
        file {
             path => "/data/elk/logs/grokparsefailure-%{[type]}-%{+YYYY.MM}.log"
        }
    }
   
# nginx access
    #根據type日志類型分別輸出到elasticsearch不同的索引
    if [type] == "hongbao-nginx-access" {
            #將處理后的結果輸出到elasticsearch
        elasticsearch {
            hosts => ["127.0.0.1:9200"]
            #指定輸出到當天的索引
            index => "hongbao-nginx-access-%{+YYYY.MM.dd}"
        }
    }

    if [type] == "pano-nginx-access" {
        elasticsearch {
            hosts => ["127.0.0.1:9200"]
            index => "pano-nginx-access-%{+YYYY.MM.dd}"
        }
    }

    if [type] == "logstash-nginx-access" {
        elasticsearch {
            hosts => ["127.0.0.1:9200"]
            index => "logstash-nginx-access-%{+YYYY.MM.dd}"
        }
    }


# tomcat access
    if [type] == "hongbao-tomcat-access" {
        elasticsearch {
            hosts => ["127.0.0.1:9200"]
            index => "hongbao-tomcat-access-%{+YYYY.MM.dd}"
        }
    }

    if [type] == "ljq-tomcat-access" {
        elasticsearch {
            hosts => ["127.0.0.1:9200"]
            index => "ljq-tomcat-access-%{+YYYY.MM.dd}"
        }
    }

# tomcat catalina
    if [type] == "hongbao-tomcat-catalina" {
        elasticsearch {
            hosts => ["127.0.0.1:9200"]
            index => "hongbao-tomcat-catalina-%{+YYYY.MM.dd}"
        }
    }

}

[root@monitor-elk ~]#

配置正則表達式
[root@monitor-elk ~]# cp /usr/local/elk/logstash/vendor/bundle/jruby/1.9/gems/logstash-patterns-core-4.0.2/patterns/grok-patterns /usr/local/elk/logstash/config/patterns
[root@monitor-elk  ~]# tail -5 /usr/local/elk/logstash/config/patterns
# Nginx
NGINXACCESS %{COMBINEDAPACHELOG} %{QS:x_forwarded_for}

# Tomcat
TOMCATACCESS %{COMMONAPACHELOG}
[root@monitor-elk ~]# chown elk.elk /usr/local/elk/logstash/config/patterns

4）配置zookeeper：

cp /usr/local/elk/zookeeper/conf/zoo_sample.cfg /usr/local/elk/zookeeper/conf/zoo.cfg

修改配置文件中的數據存儲路徑

vim /usr/local/elk/zookeeper/conf/zoo.cfg
dataDir=/data/zookeeper

備份并修改腳本 /usr/local/elk/zookeeper/bin/zkEnv.sh

修改以下變量的參數

ZOO_LOG_DIR="/data/zookeeper-logs"
ZOO_LOG4J_PROP="INFO,ROLLINGFILE"

備份并修改日志配置  /usr/local/elk/zookeeper/conf/log4j.properties

修改以下變量的參數

zookeeper.root.logger=INFO, ROLLINGFILE
log4j.appender.ROLLINGFILE=org.apache.log4j.DailyRollingFileAppender# 每天輪轉日志

啟動zookeeper：

/usr/local/elk/zookeeper/bin/zkServer.sh start

5）配置kafka：

修改配置文件 /usr/local/elk/kafka/config/server.properties 的以下參數

log.dirs=/data/kafka
zookeeper.connect=localhost:2181

備份并修改腳本 /usr/local/elk/kafka/bin/kafka-run-class.sh 

在“base_dir=$(dirname $0)/.. ”的下一行追加LOG_DIR變量，并指定日志輸出路徑

LOG_DIR=/data/kafka-logs

創建日志存儲目錄：

mkdir -p /data/kafka-logs
mkdir -p /data/elk/logs
chown -R elk.elk /data/elk/logs

啟動kafka：

nohup /usr/local/elk/kafka/bin/kafka-server-start.sh /usr/local/elk/kafka/config/server.properties &>> /data/elk/logs/kafka.log &

需要注意的是主機名一定要有配置在/etc/hosts文件中，否則kafka會無法啟動

[root@monitor-elk ~]# cat /etc/hosts
127.0.0.1  localhost  localhost.localdomain
::1         localhost localhost.localdomain localhost6 localhost6.localdomain6
10.135.3.135     monitor-elk

6）配置supervisor

①安裝supervisor：

yum install supervisor -y

設置服務開機自啟動(server程序也會一起啟動)：

systemctl enable supervisord.service

②修改配置

a.創建日志存儲路徑：

mkdir -p /data/supervisor
chown -R elk.elk /data/supervisor/

b.修改主配置文件 /etc/supervisord.conf

logfile=/data/supervisor/supervisord.log

c.創建elk程序對應的supervisor配置文件，并添加以下配置內容：

[root@monitor-elk ~]# cat /etc/supervisord.d/elk.ini 
[program:elasticsearch]
directory=/usr/local/elk/elasticsearch
command=su -c "/usr/local/elk/elasticsearch/bin/elasticsearch" elk
autostart=true
startsecs=5
autorestart=true
startretries=3
priority=10

[program:logstash]
directory=/usr/local/elk/logstash
command=/usr/local/elk/logstash/bin/logstash -f /usr/local/elk/logstash/config/logs.yml
user=elk
autostart=true
startsecs=5
autorestart=true
startretries=3
redirect_stderr=true
stdout_logfile=/data/elk/logs/logstash.log
stdout_logfile_maxbytes=1024MB
stdout_logfile_backups=10
priority=11

[program:kibana]
directory=/usr/local/elk/kibana
command=/usr/local/elk/kibana/bin/kibana
user=elk
autostart=true
startsecs=5
autorestart=true
startretries=3
priority=12
[root@monitor-elk ~]#

③啟動supervisor：

systemctl start supervisord

查看程序進程和日志：

ps aux|grep -v grep|grep "elasticsearch\|logstash\|kibana"

tip：

重啟配置的單個程序，如：

supervisorctl restart logstash

重啟配置的所有程序：

supervisorctl restart all

重載配置（只重啟配置變動的對應程序，其他配置未變動的程序不重啟）：

supervisorctl update

7）配置nginx

①安裝nginx

yum install nginx -y

②配置nginx代理:

[root@monitor-elk ~]# cat /etc/nginx/conf.d/kibana.conf 
upstream kibana {
    server 127.0.0.1:5601 max_fails=3 fail_timeout=30s;
}
server {
    listen       8080;
    server_name  localhost;
    location / {
        proxy_pass http://kibana/;
        index index.html index.htm;
        #auth
        auth_basic "kibana Private";
        auth_basic_user_file /etc/nginx/.htpasswd;
    }
 }
[root@monitor-elk ~]# grep listen /etc/nginx/nginx.conf
listen       8000 default_server;
listen       [::]:8000 default_server;
[root@monitor-elk ~]#

③創建nginx認證：

[root@monitor-elk ~]# yum install httpd -y
[root@monitor-elk ~]# htpasswd -cm /etc/nginx/.htpasswd elk
New password: 
Re-type new password: 
Adding password for user elk
[root@monitor-elk ~]# systemctl start nginx
[root@monitor-elk ~]# systemctl enable nginx

8）配置ik中文分詞：

①安裝maven：

wget "http://mirror.bit.edu.cn/apache/maven/maven-3/3.3.9/binaries/apache-maven-3.3.9-bin.tar.gz"
tar -zxf apache-maven-3.3.9-bin.tar.gz
mv apache-maven-3.3.9 /usr/local/maven
echo "export MAVEN_HOME=/usr/local/maven" >> /etc/bashrc
echo "export PATH=$PATH:$MAVEN_HOME/bin" >> /etc/bashrc
. /etc/bashrc

②編譯安裝ik（注意下載對應版本）：

wget "https://github.com/medcl/elasticsearch-analysis-ik/archive/v5.2.2.zip"
unzip v5.2.2.zip
cd elasticsearch-analysis-ik-5.2.2/
mvn package
mkdir /usr/local/elk/elasticsearch/plugins/ik
cp target/releases/elasticsearch-analysis-ik-5.2.2.zip /usr/local/elk/elasticsearch/plugins/ik/
cd /usr/local/elk/elasticsearch/plugins/ik/
unzip elasticsearch-analysis-ik-5.2.2.zip 
rm -f elasticsearch-analysis-ik-5.2.2.zip
chown -R elk.elk ../ik
supervisorctl restart elasticsearch

③創建索引模板：

要使用ik分詞，需要在創建指定的索引前（不管是通過命令手動還是logstash配置來創建）先創建索引模板，否則使用默認的模板即可：

cd /usr/local/elk/logstash

創建并編輯文件 logstash.json ，添加以下內容：

{
    "order" : 1,
    "template" : "tomcatcat-*",
    "settings" : {
      "index" : {
        "refresh_interval" : "5s"
      }
    },
    "mappings" : {
      "_default_" : {
        "dynamic_templates" : [
          {
            "string_fields" : {
              "mapping" : {
                "norms" : false,
                "type" : "text",
                "analyzer": "ik_max_word",
                "search_analyzer": "ik_max_word"
              },
              "match_mapping_type" : "text",
              "match" : "*"
            }
          }
        ],
        "_all" : {
          "norms" : false,
          "enabled" : true
        },
        "properties" : {
          "@timestamp" : {
            "include_in_all" : false,
            "type" : "date"
          },
          "log_data": {
            "include_in_all" : true,
            "type" : "text",
            "analyzer": "ik_max_word",
            "search_analyzer": "ik_max_word",
            "boost" : 8
          },
          "@version" : {
            "include_in_all" : false,
            "type" : "keyword"
          }
        }
      }
    },
    "aliases" : { }
}'

添加完畢后，執行curl命令創建索引模板

curl -XPUT 'http://localhost:9200/_template/tomcatcat' -d @logstash.json

執行成功后會返回結果 {"acknowledged":true}

④熱更新配置：

有些詞語ik無法識別分詞，如公司名稱、服務名稱之類

curl -XGET 'http://localhost:9200/_analyze?pretty&analyzer=ik_smart' -d '
騰訊云'

這時需要自己自定義詞庫，ik支持分詞熱更新的方式（不需要重啟elasticsearch），每分鐘自動檢測一次

在nginx根路徑下創建一個utf8格式的文本文件 ik.txt ，將自己需要分詞的詞語寫入ik.txt，一行一詞：

然后修改/usr/local/elk/elasticsearch/plugins/ik/config/IKAnalyzer.cfg.xml

<!--用戶可以在這里配置遠程擴展字典 -->
<entry key="remote_ext_dict">http://127.0.0.1:8000/ik.txt</entry>

配置完畢重啟elasticsearch，再次獲取分詞結果：

2.客戶端：

1）下載filebeat:

wget "https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-5.2.2-linux-x86_64.tar.gz"

解壓filebeat-5.2.2-linux-x86_64.tar.gz至/usr/local/elk/目錄，并重命名為filebeat

mkdir /usr/local/elk/
mkdir -p /data/elk/logs/
echo "10.135.3.135     elk" >> /etc/hosts

2）配置filebeat：

[root@test2 filebeat]# cat logs.yml
filebeat.prospectors:
-
#指定需要監控的日志文件路徑，可以使用*匹配
paths:
- /data/nginx/log/*_access.log
#指定文件的輸入類型為log（默認）
input_type: log
#設定日志類型
document_type: pano-nginx-access
#從文件的末尾開始監控文件新增的內容，并按行依次發送
tail_files: true
#將日志內容輸出到kafka
output.kafka:
hosts: ["10.135.3.135:9092"]
topic: beats
compression: Snappy
[root@test2 filebeat]#


[root@test3 filebeat]# cat logs.yml
filebeat.prospectors:
-
  paths:
    - /usr/local/tomcat/logs/*access_log.*.txt
  input_type: log
  document_type: hongbao-tomcat-access
  tail_files: true
-
 paths:
    - /usr/local/tomcat/logs/catalina.out
  input_type: log
  document_type: hongbao-tomcat-catalina
  #多行匹配模式，后接正則表達式，這里表示匹配時間，如 2017-06-05 10:00:00,713
  multiline.pattern: '^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2},\d{3}'
  #將未匹配到的行合并到上一行，如java的錯誤日志
  multiline.negate: true
  #將未匹配到的行添加到上一行的末尾
  multiline.match: after
  tail_files: true
output.kafka:
hosts: ["10.135.3.135:9092"]
topic: beats
compression: Snappy
[root@test3 filebeat]#

3）啟動filebeat

nohup /usr/local/elk/filebeat/filebeat -e -c /usr/local/elk/filebeat/logs.yml -d "publish" &>> /data/elk/logs/filebeat.log &

三、kibana web端配置

1.瀏覽器訪問kibana地址，并輸入前面nginx設置的賬號密碼：

http://10.135.3.135:8080

在訪問 Kibana 時，默認情況下將加載 Discover（發現） 頁面，并選擇默認的索引模式（logstash-*）。 time filter（時間過濾器）默認設置為 last 15 minutes（最近 15 分鐘），搜索查詢默認設置為 match-all（*）。

服務器資源狀態頁：

http://10.135.3.135:8080/status

2.建立索引模式

注意，索引模式的名稱要和logstash的output生成的索引（也就是說必須存在于Elasticsearch中，而且必須包含有數據）進行匹配，如logstash-*可與logstash-20170330匹配，還可以匹配多個索引（所有以logstash-開頭的索引）。

*匹配索引名稱中的零個或多個字符

3.索引建立完畢后，點擊Discover中的索引模式，即可看到Elasticsearch的日志數據

4.創建可視化圖表

繪制可視化圖表，將拆分出來的nginx或tomcat訪問日志中的字段response狀態碼進行聚合顯示，以圖表的形式直觀顯示各狀態碼（如200、400等）的統計情況

1）點擊 Visualize 中 Vertical Bar Charts（垂直條形圖）

2）選擇其中一個索引模式，如 pano-*

3）通過字段 response.keyword 指定 terms（詞條）聚合，按從大到小的順序來顯示前五列狀態碼的總數數據，然后點擊 Apply changes 圖標生效。

圖表中，X軸顯示的是狀態碼，Y軸顯示的是對應的狀態碼總數。

4）最后點擊右上角的 Save 保存，同時輸入一個可視化的名稱。

5.創建儀表盤

可以將相同業務或類型的可視化對象，集中顯示在同一個儀表盤中。

1）點擊 add 添加可視化對象到儀表盤，

2）點擊創建好的可視化對象，將會排列在在儀表盤的窗口中。對其可視化對象的窗口大小進行合適的調整。

3）添加和調整完畢后，點擊右上角的 Save 保存，同時輸入一個儀表盤的名稱。

4）顯示的結果

四、服務監控腳本

1.服務端

1)kafka

[root@monitor-elk  ~]# cat /usr/local/scripts/monitor_kafka.sh
#!/bin/bash
#
#############################################
# author:Ellen
# describes:Check kafka program
# version:v1.0
# updated:20170407
#############################################
#
# Configuration information
program_dir=/usr/local/elk/kafka
logfile=/usr/local/scripts/log/monitor_kafka.log
# Check executed user
if [ `whoami` != "root" ];then
echo "Please use root run script!!!"
exit 1
fi
# Check kafka program
num=`ps aux|grep -w $program_dir|grep -vw "grep\|vim\|vi\|mv\|scp\|cat\|dd\|tail\|head\|script\|ls\|echo\|sys_log\|logger\|tar\|rsync\|ssh"|wc -l`
if [ ${num} -eq 0 ];then
echo "[`date +'%F %T'`] [CRITICAL] Kafka program dost not start!!!"|tee -a $logfile
# Send alarm information
#cagent_tools是騰訊云服務器自帶的報警插件，該插件可發送短信或郵箱告警，如不需要可注釋
/usr/bin/cagent_tools alarm "Kafka program dost not start!!!"
echo "[`date +'%F %T'`] [  INFO  ] Begin start kafka program..."|tee -a $logfile
nohup /usr/local/elk/kafka/bin/kafka-server-start.sh /usr/local/elk/kafka/config/server.properties &>> /data/elk/logs/kafka.log &
if [ $? -eq 0 ];then
echo "[`date +'%F %T'`] [  INFO  ] Kafka program start successful."|tee -a $logfile
/usr/bin/cagent_tools alarm "Kafka program start successful"
exit 0
else
echo "[`date +'%F %T'`] [CRITICAL] Kafka program start failed!!!"|tee -a $logfile
/usr/bin/cagent_tools alarm "Kafka program start failed!!!Please handle it!!!"
exit 6
fi
else
echo "[`date +'%F %T'`] [  INFO  ] Kafka program is running..."|tee -a $logfile
exit 0
fi
[root@monitor-elk ~]#

2)zookeeper

[root@monitor-elk ~]# cat /usr/local/scripts/monitor_zookeeper.sh
#!/bin/bash
#
#############################################
# author:Ellen
# describes:Check zookeeper program
# version:v1.0
# updated:20170407
#############################################
#
# Configuration information
program_dir=/usr/local/elk/zookeeper
logfile=/usr/local/scripts/log/monitor_zookeeper.log
# Check executed user
if [ `whoami` != "root" ];then
echo "Please use root run script!!!"
exit 1
fi
# Check zookeeper program
num=`ps aux|grep -w $program_dir|grep -vw "grep\|vim\|vi\|mv\|scp\|cat\|dd\|tail\|head\|ls\|echo\|sys_log\|tar\|rsync\|ssh"|wc -l`
if [ ${num} -eq 0 ];then
echo "[`date +'%F %T'`] [CRITICAL] Zookeeper program dost not start!!!"|tee -a $logfile
# Send alarm information
/usr/bin/cagent_tools alarm "Zookeeper program dost not start!!!"
echo "[`date +'%F %T'`] [  INFO  ] Begin start zookeeper program..."|tee -a $logfile
/usr/local/elk/zookeeper/bin/zkServer.sh start
if [ $? -eq 0 ];then
echo "[`date +'%F %T'`] [  INFO  ] Zookeeper program start successful."|tee -a $logfile
/usr/bin/cagent_tools alarm "Zookeeper program start successful"
exit 0
else
echo "[`date +'%F %T'`] [CRITICAL] Zookeeper program start failed!!!"|tee -a $logfile
/usr/bin/cagent_tools alarm "Zookeeper program start failed!!!Please handle it!!!"
exit 6
fi
else
echo "[`date +'%F %T'`] [  INFO  ] Zookeeper program is running..."|tee -a $logfile
exit 0
fi
[root@monitor-elk ~]#

3)添加crontab定時任務

0-59/5 * * * * /usr/local/scripts/monitor_kafka.sh &> /dev/null
0-59/5 * * * * /usr/local/scripts/monitor_zookeeper.sh &> /dev/null

2.客戶端：

[root@test2 ~]# cat /usr/local/scripts/monitor_filebeat.sh
#!/bin/bash
#
#############################################
# author:Ellen
# describes:Check filebeat program
# version:v1.0
# updated:20170407
#############################################
#
# Configuration information
program_dir=/usr/local/elk/filebeat
logfile=/usr/local/scripts/log/monitor_filebeat.log
# Check executed user
if [ `whoami` != "root" ];then
echo "Please use root run script!!!"
exit 1
fi
# Check filebeat program
num=`ps aux|grep -w $program_dir|grep -vw "grep\|vim\|vi\|mv\|cp\|scp\|cat\|dd\|tail\|head\|script\|ls\|echo\|sys_log\|logger\|tar\|rsync\|ssh"|wc -l`
if [ ${num} -eq 0 ];then
echo "[`date +'%F %T'`] [CRITICAL] Filebeat program dost not start!!!"|tee -a $logfile
# Send alarm information
/usr/bin/cagent_tools alarm "Filebeat program dost not start!!!"
echo "[`date +'%F %T'`] [  INFO  ] Begin start filebeat program..."|tee -a $logfile
nohup /usr/local/elk/filebeat/filebeat -e -c /usr/local/elk/filebeat/logs.yml -d "publish" &>> /data/elk/logs/filebeat.log &
if [ $? -eq 0 ];then
echo "[`date +'%F %T'`] [  INFO  ] Filebeat program start successful."|tee -a $logfile
/usr/bin/cagent_tools alarm "Filebeat program start successful"
exit 0
else
echo "[`date +'%F %T'`] [CRITICAL] Filebeat program start failed!!!"|tee -a $logfile
/usr/bin/cagent_tools alarm "Filebeat program start failed!!!Please handle it!!!"
exit 6
fi
else
echo "[`date +'%F %T'`] [  INFO  ] Filebeat program is running..."|tee -a $logfile
exit 0
fi
[root@test2 ~]#

3)添加crontab定時任務

0-59/5 * * * * /usr/local/scripts/monitor_filebeat.sh &> /dev/null

五、注意事項

1.數據流向

--------------------------------------------------------------------------------------------------

log_files  ->  filebeat ->  kafka->  logstash  ->  elasticsearch  ->  kibana

--------------------------------------------------------------------------------------------------

2.每天定時清理elasticsearch索引，只保留30天內的索引

1）編寫腳本

[root@monitor-elk ~]# cat /usr/local/scripts/del_index.sh
#!/bin/bash
#
#############################################
# author:Ellen
# describes:Delete elasticsearch history index.
# version:v1.0
# updated:20170407
#############################################
#
# Configuration information
logfile=/usr/local/scripts/log/del_index.log
tmpfile=/tmp/index.txt
host=localhost
port=9200
deldate=`date -d '-30days' +'%Y.%m.%d'`
# Check executed user
if [ `whoami` != "root" ];then
echo "Please use root run script!!!"
exit 1
fi
# Delete elasticsearch index
curl -s "$host:$port/_cat/indices?v"|grep -v health|awk {'print $3'}|grep "$deldate" > $tmpfile
if [ ! -s $tmpfile ];then
echo "[`date +'%F %T'`] [WARNING] $tmpfile is a empty file."|tee -a $logfile
exit 1
fi
for i in `cat /tmp/index.txt`
do
curl -XDELETE http://$host:$port/$i
if [ $? -eq 0 ];then
echo "[`date +'%F %T'`] [  INFO  ] Elasticsearch index $i delete successful."|tee -a $logfile
else
echo "[`date +'%F %T'`] [CRITICAL] Elasticsearch index $i delete failed!!!"|tee -a $logfile
/usr/bin/cagent_tools alarm "Elasticsearch index $i delete failed!!!"
exit 6
fi
done
[root@monitor-elk ~]#

2）添加crontab定時任務

00 02 * * * /usr/local/scripts/del_index.sh &> /dev/null

3.按業務進行建立索引

如hongbao、pano等

4.nginx和tomcat等訪問日志使用默認格式

六、相關命令參考

1.列出所有索引

curl -s 'http://localhost:9200/_cat/indices?v'

2.列出節點列表

curl 'localhost:9200/_cat/nodes?v'

3.查詢集群健康信息

curl 'localhost:9200/_cat/health?v'

4.查看指定的索引數據（默認返回十條結果）

curl -XGET 'http://localhost:9200/logstash-nginx-access-2017.05.20/_search?pretty'

5.刪除指定的索引

curl -XDELETE 
http://localhost:9200/logstash-nginx-access-2017.05.20

6.查詢模板

curl -s 'http://localhost:9200/_template'