日誌主要包括系統日誌、應用程式日誌和安全日誌。系統運維和開發人員可以通過日誌瞭解伺服器軟硬體資訊、檢查配置過程中的錯誤及錯誤發生的原因。經常分析日誌可以瞭解伺服器的負荷，效能安全性，從而及時採取措施糾正錯誤。

通常，日誌被分散的儲存不同的裝置上。如果你管理數十上百臺伺服器，你還在使用依次登入每臺機器的傳統方法查閱日誌。這樣是不是感覺很繁瑣和效率低下。當務之急我們使用集中化的日誌管理，例如：開源的syslog，將所有伺服器上的日誌收集彙總。

集中化管理日誌後，日誌的統計和檢索又成為一件比較麻煩的事情，一般我們使用grep、awk和wc等Linux命令能實現檢索和統計，但是對於要求更高的查詢、排序和統計等要求和龐大的機器數量依然使用這樣的方法難免有點力不從心。

開源實時日誌分析ELK平臺能夠完美的解決我們上述的問題，ELK由ElasticSearch、Logstash和Kiabana三個開源工具組成。官方網站： https://www.elastic.co/products

• Elasticsearch是個開源分散式搜尋引擎，它的特點有：分散式，零配置，自動發現，索引自動分片，索引副本機制，restful風格介面，多資料來源，自動搜尋負載等。
• Logstash是一個完全開源的工具，他可以對你的日誌進行收集、過濾，並將其儲存供以後使用（如，搜尋）。
• Kibana 也是一個開源和免費的工具，它Kibana可以為 Logstash 和 ElasticSearch 提供的日誌分析友好的 Web 介面，可以幫助您彙總、分析和搜尋重要資料日誌。

Centos7 環境下的搭建

1.JDK

下載linux的1.8 JDK 包，解壓，進行如下配置：

vi ~/.bashrc
export JAVA_HOME=/usr/local/jdk1.8.0_101
export PATH=$PATH:$JAVA_HOME/bin
exportCLASSPATH=.:$JAVA_HOME/lib/tools.jar:$JAVA_HOME/lib/dt.jar:$CLASSPATH

2.ElasticSearch

安裝：

tar -zxvf elasticsearch-2.4.0.tar.gz
cd elasticsearch-2.4.0
 

安裝Head外掛:

./bin/plugin install mobz/elasticsearch-head

編輯ElasticSearch的配置檔案:

vi config/elasticsearch.yml
cluster.name=es_cluster
node.name=node0
path.data=/tmp/elasticsearch/data
path.logs=/tmp/elasticsearch/logs
#當前hostname或IP
network.host=192.168.245.129
network.port=9200

啟動elasticsearch

由於不能使用root身份啟動，所以需要先建立一個使用者，如：elk

adduser elk
passwd elk    #輸入兩次密碼。
chown -R elk /你的elasticsearch安裝目錄  #root給elk賦許可權
./bin/elasticsearch

可以看到，它跟其他的節點的傳輸埠為9300，接受HTTP請求的埠為9200。使用ctrl+C停止。當然，也可以使用後臺程序的方式啟動ES：

./bin/elasticsearch &

安裝的head外掛，它是一個用瀏覽器跟ES叢集互動的外掛，可以檢視叢集狀態、叢集的doc內容、執行搜尋和普通的Rest請求等。現在也可以使用它開啟 localhost:9200/_plugin/head 頁面來檢視ES叢集狀態：

3.Logstash

安裝：

tar -zxvf logstash-2.4.0.tar.gz
cd logstash-2.4.0.tar.gz

配置:

mkdir config #建立放置配置檔案的目錄
vi config/log_to_es.conf

加入input,output配置：

input { 
   syslog {
     codec => "json"
     port => 514
   }
}
output {
    elasticsearch {
       hosts => ["192.168.245.129:9200"]
    }
}

- input：使用syslog udp方式來接收logback傳輸過來的日誌,配置埠為514（預設就是514，可以根據需要隨意指定），使用 netstat -anptu可以檢視到啟動的監聽

    tcp6       0      0 :::514                  :::*                    LISTEN      2869/java                
    udp6       0      0 :::514                  :::*                                2869/java        

• output：輸出到elasticsearch，注意埠必須是http埠

啟動:

./bin/logstash agent -f config/log_to_es.conf

注意：要使用root身份進行啟動

4.Kibana

安裝：

tar -zxvf kibana-4.6.1-linux-x86_64.tar.gz
cd kibana-4.6.1-linux-x86_64

配置：

vi config/kibana.yml

server.port: 5601
server.host: 192.168.245.129
elasticsearch.url: http://192.168.245.129:9200
kibana.index: “.kibana”

啟動:

./bin/kibana

用瀏覽器開啟該地址 http://192.168.245.129:5601，為了後續使用Kibana，需要配置至少一個Index名字或者Pattern，它用於在分析時確定ElasticSearch中的Index。

5.logback配置

使用springboot+logback，將logback的記錄的日誌儲存到elk中。

自定義的屬性類：

@Component
@ConfigurationProperties(prefix = "myproperties", ignoreUnknownFields = false)
public class MyProperties {
    private final CorsConfiguration cors = new CorsConfiguration();

    public CorsConfiguration getCors() {
        return cors;
    }

    private final Logging logging = new Logging();

    public Logging getLogging() { return logging; }

    public static class Logging {

        private final Logstash logstash = new Logstash();

        public Logstash getLogstash() { return logstash; }

        public static class Logstash {

            private boolean enabled = false;

            private String host = "localhost";

            private int port = 5000;

            private int queueSize = 512;

            public boolean isEnabled() { return enabled; }

            public void setEnabled(boolean enabled) { this.enabled = enabled; }

            public String getHost() { return host; }

            public void setHost(String host) { this.host = host; }

            public int getPort() { return port; }

            public void setPort(int port) { this.port = port; }

            public int getQueueSize() { return queueSize; }

            public void setQueueSize(int queueSize) { this.queueSize = queueSize; }
        }

    }
}   

日誌配置類:

@Configuration  
public class LoggingConfiguration {

    private final Logger log = LoggerFactory.getLogger(LoggingConfiguration.class);

    private LoggerContext context = (LoggerContext) LoggerFactory.getILoggerFactory();

//    @Value("${spring.application.name}")
    private String appName="Example";

//    @Value("${server.port}")
    private String serverPort="8089";

    @Autowired
    private MyProperties myPro;

    @PostConstruct
    private void init() {
        if (myPro.getLogging().getLogstash().isEnabled()) {
            addLogstashAppender();
        }
    }

    public void addLogstashAppender() {
        log.info("Initializing Logstash logging");

        LogstashSocketAppender logstashAppender = new LogstashSocketAppender();
        logstashAppender.setName("LOGSTASH");
        logstashAppender.setContext(context);
        String customFields = "{\"app_name\":\"" + appName + "\",\"app_port\":\"" + serverPort + "\"}";

        // Set the Logstash appender config from JHipster properties
        logstashAppender.setSyslogHost(myPro.getLogging().getLogstash().getHost());
        logstashAppender.setPort(myPro.getLogging().getLogstash().getPort());
        logstashAppender.setCustomFields(customFields);

        //新增日誌功率，只有warn和error才記錄到elk stack中
        logstashAppender.addFilter(new Filter<ILoggingEvent>() {
            @Override
            public FilterReply decide(ILoggingEvent event) {
                if(event.getLevel() == Level.ERROR || event.getLevel()==Level.WARN){
                    return FilterReply.ACCEPT;
                }
                return FilterReply.DENY;
            }
        });
        // Limit the maximum length of the forwarded stacktrace so that it won't exceed the 8KB UDP limit of logstash
        ShortenedThrowableConverter throwableConverter = new ShortenedThrowableConverter();
        throwableConverter.setMaxLength(7500);
        throwableConverter.setRootCauseFirst(true);
        logstashAppender.setThrowableConverter(throwableConverter);

        logstashAppender.start();

        // Wrap the appender in an Async appender for performance
        AsyncAppender asyncLogstashAppender = new AsyncAppender();
        asyncLogstashAppender.setContext(context);
        asyncLogstashAppender.setName("ASYNC_LOGSTASH");
        asyncLogstashAppender.setQueueSize(myPro.getLogging().getLogstash().getQueueSize());
        asyncLogstashAppender.addAppender(logstashAppender);
        asyncLogstashAppender.start();

        context.getLogger("ROOT").addAppender(asyncLogstashAppender);
    }
}

注意：日誌過濾可以通過新增logback appender的filter方式，或是採用logstash filter的drop外掛,如：

filter {
      if ([message] =~ "^toElk") {
      }else {
         drop {}
      }
}

6.長期執行程式

使用supervisor來管理elk stack程序。

1.安裝：

先安裝setuptools:

wget https://bootstrap.pypa.io/ez_setup.py -O - | python

再安裝supervisor：

wget https://pypi.python.org/packages/80/37/964c0d53cbd328796b1aeb7abea4c0f7b0e8c7197ea9b0b9967b7d004def/supervisor-3.3.1.tar.gz
tar zxvf supervisor-3.3.1.tar.gz
cd supervisor-3.3.1
python setup.py install

2.配置:

建立supervisor配置檔案：

echo_supervisord_conf  >/etc/supervisord.conf
vi /etc/supervisord.conf

取消以下的註釋,並修改IP為0.0.0.0

[inet_http_server]         ; inet (TCP) server disabled by default
port=0.0.0.0:9001        ; (ip_address:port specifier, *:port for all iface)
username=user              ; (default is no username (open server))
password=123               ; (default is no password (open server))

3.supervisor啟動檔案:

vi /etc/init.d/supervisord

#! /bin/sh
PATH=/sbin:/bin:/usr/sbin:/usr/bin

PROGNAME=supervisord

DAEMON=/usr/bin/$PROGNAME

CONFIG=/etc/$PROGNAME.conf

PIDFILE=/tmp/$PROGNAME.pid

DESC="supervisord daemon"

SCRIPTNAME=/etc/init.d/$PROGNAME

# Gracefully exit if the package has been removed.

test -x $DAEMON || exit 0

start()

{
        echo -n "Starting $DESC: $PROGNAME"
        $DAEMON -c $CONFIG
        echo "..."
}

stop()

{
        echo -n "Stopping $DESC: $PROGNAME"
        supervisor_pid=$(cat $PIDFILE)
        kill -15 $supervisor_pid
        echo "..."
}

case "$1" in
  start)
        start
        ;;
  stop)
        stop
        ;;
  restart)
        stop
        start
        ;;
  *)
echo "Usage: $SCRIPTNAME {start|stop|restart}" >&2
exit 1
;;
esac
exit 0

4.設定管理的程序

vi /usr/local/supervisor_ini.ini

加入：

[program:elasticsearch]
directory=/usr/local/elasticsearch-2.4.0
command=/usr/local/elasticsearch-2.4.0/bin/elasticsearch
;process_name=elasticsearch ; process_name expr (default %(program_name)s)
user=elk
numprocs=1                    ; number of processes copies to start (def 1)
priority=1                  ; the relative start priority (default 999)
autostart=true                ; start at supervisord start (default: true)
;startsecs=1                   ; # of secs prog must stay up to be running (def. 1)
startretries=3                ; max # of serial start failures when starting (default 3)
autorestart=true        ; when to restart if exited after running (def: unexpected)
stopasgroup=true             ; send stop signal to the UNIX process group (default false)
killasgroup=true
redirect_stderr=true
stdout_logfile=/tmp/supervisor_elasticsearch.log


[program:logstash]
directory=/usr/local/logstash-2.4.0
command=/usr/local/logstash-2.4.0/bin/logstash agent -f /usr/local/logstash-2.4.0/config/to_es.conf
;process_name=logstash ; process_name expr (default %(program_name)s)
numprocs=1                    ; number of processes copies to start (def 1)
priority=2                  ; the relative start priority (default 999)
autostart=true                ; start at supervisord start (default: true)
;startsecs=1                   ; # of secs prog must stay up to be running (def. 1)
startretries=3                ; max # of serial start failures when starting (default 3)
autorestart=true        ; when to restart if exited after running (def: unexpected)
;exitcodes=0,2                 ; 'expected' exit codes used with autorestart (default 0,2)
;stopsignal=QUIT               ; signal used to kill process (default TERM)
;stopwaitsecs=10               ; max num secs to wait b4 SIGKILL (default 10)
stopasgroup=true             ; send stop signal to the UNIX process group (default false)
killasgroup=true             ; SIGKILL the UNIX process group (def false)
redirect_stderr=true          ; redirect proc stderr to stdout (default false)
stdout_logfile=/tmp/supervisor_logstash.log       ; stdout log path, NONE for none; default AUTO
;stdout_logfile_maxbytes=1MB   ; max # logfile bytes b4 rotation (default 50MB)
;stdout_logfile_backups=10     ; # of stdout logfile backups (default 10)
;stdout_capture_maxbytes=1MB   ; number of bytes in 'capturemode' (default 0)
;stdout_events_enabled=false   ; emit events on stdout writes (default false)
;stderr_logfile=/a/path        ; stderr log path, NONE for none; default AUTO
;stderr_logfile_maxbytes=1MB   ; max # logfile bytes b4 rotation (default 50MB)
;stderr_logfile_backups=10     ; # of stderr logfile backups (default 10)
;environment=A="1",B="2"       ; process environment additions (def no adds)

[program:kibana]
directory=/usr/local/kibana-4.6.1-linux-x86_64
command=/usr/local/kibana-4.6.1-linux-x86_64/bin/kibana
;process_name=kibana ; process_name expr (default %(program_name)s)
numprocs=1                    ; number of processes copies to start (def 1)
priority=3                  ; the relative start priority (default 999)
autostart=true                ; start at supervisord start (default: true)
;startsecs=1                   ; # of secs prog must stay up to be running (def. 1)
startretries=3                ; max # of serial start failures when starting (default 3)
autorestart=true        ; when to restart if exited after running (def: unexpected)
redirect_stderr=true
stdout_logfile=/tmp/supervisor_kibana.log


[group:elk_stack]
programs=elasticsearch,logstash,kibana  ; each refers to 'x' in [program:x] definitions
priority=1

修改/etc/supervisord.conf檔案，加入如下：

[include]
files = /usr/local/supervisor_ini/*.ini

5.啟停supervisor:

/etc/init.d/supervisor start
/etc/init.d/supervisor stop

6.通過web介面檢視：