ELK完整搭建流程（小白入門記）

踏入社會的第一份工做，作了點ELK，記錄一下。

客官~下面直接上菜！

首先直接來張思惟導圖，演示大衆化的入門級的ELK搭建流程。

（注意~ELK的啓動順序按照 Elasticsearch Logstash Kibana 來啓動，不然kibana會報錯） （注意~其中的kafka要配合zookeeper使用，zookeeper是註冊管理中心，實現負載均衡）

First things first, 搭建架構，就如上圖咯；

Two, 在每臺機器上安裝對應的軟件

直接去官網下載，而後download，而後解壓，就能夠在bin目錄下啓動測試了。

須要注意的是，logstash須要在你採集日誌的機器上部署，kafka和elasticsearch根據本身的狀況來部署，kibana只須要在單臺機器上部署就OK（固然若是用戶多，並且這臺機器掛掉呢。so，最終仍是要部署多臺，單臺部署kibana 起步足夠）.

Three, 配置logstash，官網的這張圖不錯，直接盜用過來了

在input中，按需取log，通常去拿業務系統路徑下面的log； 在filter中，有大約200多個插件，可以實現不一樣的功能，下面的例子我用ruby插件實現過hash； 在output中，將信息傳送到消息隊列或者elasticsearch中；

個人兩個實例以下：

① shipper->broker：將消息從源送往緩存，減輕elasticsearch的壓力

input {
        file {
           codec => multiline {
             pattern => "^201"
             negate => true
             what => "previous"
          }
        type => "test-type"
        path => ["/path/to/yourlog/test1.log","/path/to/yourlog/test2.log"]
        exclude => ["*.gz"]
        tags => ["test-tags"]
        start_position => "beginning"
        }
    }#註釋——file插件，只須要指向日誌的path便可，本身定義type（保證惟一~在後面會用到；start_position是log掃描的起始位置，默認是尾部，這裏設成的是begining）
    #註釋———filter能對數據作200多種處理（由於有200多種插件），這裏只是傳輸數據，因此不必用filter處理數據，但會在下面用到
output {
    kafka {
      codec => plain {
                 format => "%{host} %{message}"
               }
      compression_type => "snappy"#註釋——kafka的壓縮格式，可以節省空間
      topic_id => "test-topic-id"#註釋——區分不用的消息隊列，每一個topic_id都要保證惟一性
      client_id => "xx.0.40.1"#註釋——相似於id，便於後期查找問題
      bootstrap_servers => "xx.0.49.1:9092,xx.0.49.2:9092,xx.0.49.3:9092,xx.0.49.4:9092,xx.0.49.5:9092"
   }#註釋——五臺kafka的機器用來作消息隊列，起到緩衝的做用
}

② broker->indexer->elasticsearch：將消息從緩存拿出來（input），而後用filter按咱們須要的處理，而後送到elasticsearch（output）

input{
        kafka {
          codec => plain {}
          zk_connect => "xx.0.49.1:2181,xx.0.49.2:2181,xx.0.49.3:2181,xx.0.49.4:2181,xx.0.49.5:2181"#註釋——注意這裏是zookeeper，不是kafka，因此，各位須要自行配置zk實現負載均衡等...
          group_id => "test-group-id"#註釋——區分用~避免之後爭搶同一topic_id
          topic_id => "test-topic-id"#註釋——注意這裏和上面的topic_id對應
          consumer_threads => 2  #註釋——這個線程數要和kafka的partitions數目保持一致來發揮最佳性能
          auto_offset_reset => "smallest"
        }
    }
filter{#註釋——在這裏能夠經過處理數據，作不少事情
        grok {
              match => ["message", "%{TIMESTAMP_ISO8601:logdate}"]
        }#註釋——grok這個插件自我感受主要就是用正則匹配

        date {
              match => [ "logdate", "yyyy-MM-dd HH:mm:ss,SSS" ]
              target => "logtime"
              remove_field => [ "logdate"]
        }#註釋——date插件用來處理時間，把我係統日誌的時間轉爲kibana能識別的時間
        ruby{#註釋——ruby插件用來：首先熊message中，匹配到id，而後將id對應的name用hash展現出來，咱們的原始數據是相似這樣的（一條，其中加黑加斜體部分是可變的，好比registerFrom:56，regiserFrom和loginFrom是信息類型，56和222是id，他們分別對應着中文名字——中文名字是咱們最終想要的結果）：bj2-lkb-tomcat2.quantgroup.cn 2016-04-19 21:38:09,975 [INFO ] c.q.x.c.external.user.AppController - 第三方用戶登陸成功, **_loginFrom:222_**, phoneNo:13976701637 咱們要將**_loginFrom:222_**提取成**_channelName:222對應的中文名字_**
                init => "[@kname](http://my.oschina.net/u/1384245) = ['channelName'];@pattern=/registerFrom:(\d+)|loginFrom:(\d+)/;@hashChannel={'56'=>'量化派','222'=>'現金巴士'}"#註釋——此處配置的ruby的hash來將匹配到的數字轉換爲 中文名字（提升在kibana的閱讀性）

                 code => "event.append(Hash[@kname.zip(Array(if @pattern.match(event['message'])[1] then @hashChannel[@pattern.match(event['message'])[1]]  elsif @pattern.match(event['message'])[2] then @hashChannel[@pattern.match(event['message'])[2]]  end ))]) if @pattern.match(event['message'])"#註釋——ruby代碼 ：最後面的if 至關於咱們的前提條件，若是匹配到，纔會去執行前面的代碼，避免ruby報exception（不然exception會顯示在kibana中）
            }
}
    output{
        elasticsearch {
        hosts => ["xx.0.9.1:9200","xx.0.9.2:9200","xx.0.9.3:9200", "xx.0.9.4:9200", "xx.0.9.5:9200"]#註釋——集羣，hosts能夠寫多個地址,有細心的朋友問不是能夠寫一個地址，而後ES實現自動分配嗎？正解：萬一 一個斷了，ES能夠分配給其他的節點（ES只會根據優先級挑選一個ES，不會發給全部的host），苦逼的運維思路~
        index => "logstash-test-%{+xxxx.ww}"#註釋——index索引，後面的xxxx.ww說明咱們是按周來生成索引，咱們能夠按需設置成按天/周/月...自我感受按周的適用性較高。另外特別注意的一點是沒什麼事老實的用logstash-起頭來定義你的index索引，由於這樣的話，logstash會自動去匹配一個優化好的模板，不然的話，你還須要本身配置映射，麻煩麻煩麻煩的很，[可參見這篇博文](http://my.oschina.net/yangchunlian/blog/660848)，我當時走過彎路。這裏就是配置中文索引字段的小技巧了，其中配置模板映射的模式能夠精準命名。raw的方式會同時包含一個analyzed的字段和一個not_analyzed的字段。強烈建議日誌前加 logstash-!!!!!!!!!！！！！！！
       }
    }

至此，logstash配置完畢。

可是，你們確定不會每次啓動logstash的時候都用 logstash -f xx.conf 的方式去啓動， 這裏，須要咱們配置supervisord,配置的教程你們看這兒吧，這裏要配置成自動拉起(須要注意的是，每臺機器都要配置哦~~），它還能夠直接圖形化界面的形式管理你的logstash啓動項（見下圖），好開心有木有？

在supervisor/conf/supervisord.conf中，咱們能夠修改相應的參數，添加相應的logstash的conf啓動項。

Four, 配置Elasticsearch

Elasticsearch須要配置什麼呢？ 我目前也就改改bin/elasticsearch.in.sh裏面的內存

if [ "x$ES_MIN_MEM" = "x" ]; then
        ES_MIN_MEM=256m
    fi
    if [ "x$ES_MAX_MEM" = "x" ]; then
        ES_MAX_MEM=1g
    fi

logstash說不定還會改改java環境變量。 再就是改改config/elasticsearch.yml裏面的名字神馬的，

cluster.name: my-application#註釋——集羣名字，一個集羣下的名字總該相同咯
    node.name: node-2#註釋——節點名字，一個集羣下，節點名字確定不一樣咯
    node.rack: r1
    path.data: /home/es/elasticsearch-2.2.0/data#註釋——索引存儲地址，因此這兒的文件最大咯
    path.logs: /home/es/elasticsearch-2.2.0/logs#註釋——es的log地址
    network.host: xx.0.9.2#註釋——配置host
    http.port: 9200#註釋——配置端口，建議默認哦
    discovery.zen.minimum_master_nodes: 3
    gateway.recover_after_nodes: 3

思前想後，也就這麼點東西吧，歡迎補充。

Five, 配置kibana

與logstash和elasticsearch不一樣，kibana咱們設置單節點便可（固然若是用戶多，也能夠配置集羣；另外，萬一咱們配置的機器若是掛了，也能從其它機器訪問嘛；but，入門階段，建議先配置一個就行了，機器哪那麼容易就掛掉）。

隨便找臺你的es機器，安裝啓動了kibana以後（啓動命令 nohup ./bin/kibana &），在網頁中打開host:5601就能打開kibana的web界面。

linux下 重啓kibana

ps -ef|grep kibana  
  
ps -ef|grep 5601  
  
都找不到   
  
嘗試 使用 fuser -n tcp 5601  
  
kill -9  端口  
  
啓動便可 nohup ./bin/kibana &

而就是這個web界面，可以極大提升程序員的bug查找效率。

好比最low的方式是去線上機器中查找log（tailf命令。。。。。。），而後用你的肉眼凡睛去定位你的問題。 可是用了kibana，能夠直接圖形化界面的形式查找。

再就是kibana可以根據日誌來造成圖形化界面，來觀測數據庫中和程序中看不到的數據。（其實kibana還能夠作一些權限控制等功能，這個你們入門後根據需求慢慢深刻）

此外，利用logstash的filter的二百多個插件，咱們能夠集成不少不一樣的功能，好比傳統的用ab + nmon來測系統的負載狀況，咱們能夠直接用filter的collectd插件集成進來，而後用kibana來圖形化，簡單！高效！開心！

so，開源的ELK就是利國利民的神器。