elk日誌大盤顯示和日誌監控報警配置實踐

阿新 • • 發佈：2019-02-03

1. Logstash

1.1具體內容詳解

1.2使用整體配置

1.2.1專案工程中的配置

在logback對應的配置檔案logback-spring.xml中配置如下內容：

注意encoder標籤中對應的三個欄位：requestUrl，traceId和clientIp，這三個是在要求在輸出日誌中增加這三個標籤，和logback對應的預設輸出合併，對應的輸出結果為：

1.2.2 日誌傳送到elasticsearch配置

具體對應conf配置如下：

input {

 file {

    type =>"XXX_kccf_pc_info"  ------對應的elasticsearch索引type值

   path => "/data/log/kccf_pc_info.*.log"     ------輸入日誌

   start_position => "beginning"

   codec => json {

            charset => "UTF-8"

       }

    }

}

filter {

   if[type] == "XXX_kccf_pc_info"

    {

       if "method=HEAD" in [requestUrl] {   ----------如果是心跳檢查（head）在不會將日誌拉取到elasticSearch

                drop {}

      }

          mutate {

                split =>["[requestUrl]",","]

         }

    }

   geoip {

           source => "ip"

           target => "geoip"

     }

}

filter {

   if[type] == "XXX_kccf_pc_info"

    {

        if [level] == "ERROR" and "PARAM_NO_PRIVILEGE" notin [message] and "ERROR-START" not in [message]

         {

          mutate {

              add_tag => "email"}

         }

     }

         geoip{

                     source=> "ip"

                     target => "geoip"

    }

}

output {

  if[type] == "XXX_kccf_pc_info"{

     elasticsearch {

       hosts => "XXXXX"

       index => "XXXX-%{+YYYY.MM.dd}"  ------elasticsearch對應的索引名稱

       sniffing => false

       manage_template => false

       flush_size => 3000

       idle_flush_time => 5

       user => "logstash"

       password => "logstash"

#       ssl => true

#       ssl_certificate_verification => false

#       truststore => "/etc/logstash/truststore.jks"

#       truststore_password => changeit

     }

  }

 if"email" in [tags] and [type] =="XXX_kccf_pc_info"

  {

   email{                ---------------------傳送錯誤日誌郵件

       port           =>    "587"

       address        =>    "smtp.XXXX.com"

       username       =>    " 
[email protected]"

       password       =>    "xxxxxx"

       authentication =>   "login"

       use_tls        =>    true

       from           =>    "[email protected]"

       subject        =>    "Waring: you have an error on host101.201.118.236(TYJ1)"

       to             =>    " 
[email protected]"

       via            =>    "smtp"

       body           =>    "you hava an error ofkccf_pc_info!  server_ip:XXX\n ERROR time: %{@timestamp}+08:00Hours ; \n ERROR message: %{message} ; \nLogger Name: %{logger_name}; \n Level:%{level} ; \n level_value:%{level_value}; \n Stack_trace : \n %{stack_trace}\n "

       }

  }

}

2. Elastic search

2.1具體內容詳解

Elasticsearch叢集可以包含多個索引（資料庫），這些索引又包含多個型別（表）。這些型別包含多個文件（行），每個文件都有多個欄位（列）。

2.2 elasticsearch head介紹

可以檢視elasticsearch伺服器上各種資訊

2.3 searchguard 外掛

從對應配置檔名稱就可以看出來，就主要有以下主要關係：

2.4 script field

就是可以根據指令碼動態的在ElasticSearch中生成對應的field，而不是一開始就在mapping中定義的好的，後面在kibana配置大盤的時候會用到。

2.5 painless 指令碼

指令碼語言詳解：https://www.elastic.co/guide/en/elasticsearch/reference/5.4/modules-scripting-painless.html

3.Kibana

3.1 具體內容詳解

3.2 searchguard 外掛

外掛下載地址和詳細說明參見：https://github.com/floragunncom/search-guard-kibana-plugin

3.3 sentinl 郵件報警外掛

注意：郵件配置的時候一定是授權碼，不是密碼，否則報授權失敗

4.日誌監控大盤配置

4.1 kibana日誌搜尋基本用法

AND，OR，NOT，範圍查詢語法支援

其實對於kibana查詢，還是可以使用filter進行查詢的（高版本已經支援了，但是我們使用的5.4需要經過下面的步驟來增加filter）：

4.2 增加索引scriptfield

之所以要增加scriptfields是因為比如從對應的下面日誌中抽取對應的耗時：

method=com.kccf.pc.controller.article.ArticleController.getArticleListcost 22 milliseconds

要從對應的日誌中擷取對應的耗時內容，所以要使用指令碼來進行擷取，並將擷取的值存放在對應的script fields欄位中，供後面的Visualize使用。

4.3 Visualize配置

單擊“加號”新增一個檢視：

並選擇對應的檢視型別，然後選擇要從哪個索引採集資料進行配置：

當然X，Y軸都可以配置多個統計，如下面的：

4.4 Dashboard大盤配置

大盤就是將上面配置的多個Visualized放在一個地方進行顯示，如下圖所示：

新增大盤，通過如下操作：

將Visualized放置在大盤中：

5. 監控郵件報警sentinl配置

5.1 新增watcher配置

（1） General：表示多長時間執行一次，定時遍歷日誌

（2） Input：對elasticSearch中的索引或者索引列表進行過濾採集，得到最終的輸入

（3） Condition：就是過濾之後的日誌滿足什麼條件才進行報警

（4） Transform：對結果進行轉換

（5） Actions：到達報警條件了，進行什麼樣的處理，一般就是傳送郵件通知

5.1.1 Input過濾：

下面是配置的Input過濾：在最近一個小時內，耗時大於20毫秒的記錄進行過濾

{

 "search": {

   "request": {

      "index":[

       "indu_kccf_pc_info-*"

      ],

     "body": {

       "query": {

         "bool": {

           "must": [

              {

               "script": {

                 "script": {

                   "lang": "painless",

                   "params": {

                     "costTimeThrehold":20

                   },

                   "inline":"if(null != doc['message.keyword'].value &&doc['message.keyword'].value.trim().startsWith(\"method=\")&&doc['message.keyword'].value.trim().endsWith(\"milliseconds\")){Number costTimeNum = NumberFormat.getInstance().parse(doc['message.keyword'].value.substring(doc['message.keyword'].value.indexOf('cost')+4,doc['message.keyword'].value.lastIndexOf('milliseconds')-1).trim());if(costTimeNum.longValue() > params.costTimeThrehold) true; else false;}elsefalse;"

                  }

                }

              },

              {

               "range":{

                 "@timestamp": {

                   "gte": "now-1h",

                   "lte": "now",

                   "format": "epoch_millis"

                  }

                }

              }

            ]

          }

        }

      }

    }

  }

}

注意：inline裡面的指令碼功能就是擷取如下日誌中對應的耗時method=com.kccf.pc.controller.article.ArticleController.getArticleListcost 22 milliseconds。Range對應的是時間範圍過濾。

5.1.2 Condition配置：

Condition配置就是對input過濾得到結果進行統計，如果統計的結果滿足設定的條件則進行觸發後面的行為：

上面的條件是Input對應的過濾結果的條數大於1則進行報警，這個指令碼是可以自己動態編寫的，因為sentinl外掛也是在elasticSearch中定義了自己對應的索引：watcher和watcher_alarms-時間，如下面在elasticSearch-head中對應的截圖所示：

{

    "_index":"watcher",

    "_type":"sentinl-watcher",

    "_id":"38eljz92mycwqi00ian55ng66r-0tqtq7q46valazkmg8yn20ggb9-im815vceg9gap6akbvjejnhfr",

    "_version":32,

    "_score":1,

    "_source":{

       "title":"所有",

       "disable":false,

       "report":false,

       "trigger":{

           "schedule":{

              "later":"every 1 secs"

           }

       },

       "input":{

           "search":{

              "request":{

                  "index":[

                     "cpcn-*"

                  ],

                  "body":{}

              }

           }

       },

       "condition":{

           "script":{

              "script":"payload.hits.total >0"

           }

       },

       "actions":{

           "error報警": {

              "throttle_period":"0h0m1s",

              "email":{

                  "to":"[email protected]",

                  "from":"[email protected]",

                  "subject":"報警",

                  "body":"event:{{payload.hits.total}} errorMsgConent:{{payload.hits.hits}}"

              }

           }

       }

    }

}

{

    "_index":"watcher",

    "_type":"sentinl-script",

    "_id":"8mmkkevvdr0g0sz3mmojn61or-s6hfac0a3y36ier49ju1sjor-keeuakonk0v51cntohp1ll3di",

    "_version":1,

    "_score":1,

    "_source":{

       "description":"input",

       "title":"error",

       "body":"{

       "search":{

           "request":{

              "index":["cpcn-*"],

              "body":{

                  "query":{

                     "bool":{

                         "must":[{

                                "wildcard":{

                                   "message":"*exception*"

                                }

                            },

                            {

                                "wildcard":{

                                   "message":"*error*"

                                }

                            },

                            {

                                "range":{

                                   "@timestamp":{

                                       "gte":"now-1h",

                                       "lte":"now",

                                       "format":"epoch_millis"

                                   }

                                }

                            }

                         ],

                         "must_not":[]

                     }

                  }

              }

           }

       }

    }

    "

}

}

（3）報警資訊內容對應的是索引watcher_alarms-時間型別為配置Action名稱的內容：

{

    "_index":"watcher_alarms-2018.04.09",

    "_type":"email_admin",------配置的action名稱

    "_id":"AWKphaIhjWbDkGCDylsw",

    "_version":1,

    "_score":1,

    "_source":{

       "@timestamp":"2018-04-09T08:28:50.080Z",

       "watcher":"watcher_title",

       "level":"high",

       "message":"Found 177 Events",

       "action":"email_admin",

       "payload":{

           "took":4,

           "timed_out":false,

           "_shards":{

              "total":25,

              "successful":25,

              "skipped":0,

              "failed":0

           },

           "hits":{

              "total":177,

              "max_score":1,

              "hits":[{

                  "_index":"cpcn-2018.01.17",

                  "_type":"logs",

                  "_id":"AWECHmRX1zYDH4X06oyZ",

                  "_score":1,

                  "_source":{

                     "@version":"1",

                     "host":"DESKTOP-7DN8E16",

                     "@timestamp":"2018-01-17T03:16:41.677Z",

                     "message":"qwe"

                  }

              }]

           }

       },

       "report":false

    }

}

注意上面的報警內容，正好是我們在Condition中配置的需要的資訊，如condition配置所示：

{

 "script": {

   "script": "payload.hits.total > 1"  -----正好對應的上面報警內容對應的記錄中的內容

  }

}

5.1.3 Actions配置：

Actions配置就配置在滿足報警條件的時候，如何進行處理，如下面報警條件滿足的時候進行郵件通知處理配置如下：

注意：body中的內容正是取索引watcher_alarms-時間型別為“方法耗時超過閾值報警”名稱的內容