[ECK]elasticsearch ingest pipeline

前言

用filebeat 蒐集 nginx的資料，取得的Nginx資料沒再經過分解，
導致沒辦法運用在efk上面。

這邊說的運用，指的是直接在Dscover上面直接用fileter的方式查詢到需要的資料。

因為資料都是一個欄位[Message]，所以必須分解後指定他們資料對應的欄位。

正文

先用UI講解步驟。

先來看一下nginx的message，然後再分析裡面的欄位

127.0.0.1 - - [06/May/2022:06:44:01 +0000] , http-host: \"abc.boxing.com\" , URL: \"GET /static/tpl/analytics/index.html?v=51817715.1 HTTP/1.1\"  , request-status : \"200\"  ,   body-byte: 0  ,http-referer: \"https://abc.boxing.com/mobile\"  ,user-agent: \"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/100.0.4896.127 Safari/537.36\" , X-Forwarded-For : \"132.235.211.18,10.107.10.14\" ,   request-time: \"0.000\"  , response_time : \"-\"

這邊的格式有特別去調整過，詳情可參考，[[blog.56. GKE記錄 nginx log]]。

2. ingest pipeline
   UI畫面，Stack Management -> Ingest Pipelines-> Create pipeline
   

   
   
   
   

3. Add a processor
   

   
   
   
   Processor ： Grok
   Field ： message
   patterns:

%{IPORHOST:nginx.source.ip} %{USER:nginx.user.id} %{USER:nginx.user.name} \\[%{HTTPDATE:nginx.@timestamp}\\] , http-host: \"%{DATA:nginx.url.host}\" , URL: \"%{WORD:nginx.http.request.method} %{DATA:nginx.url.path} HTTP/%{NUMBER:nginx.http.version}\"  , request-status : \"%{NUMBER:nginx.http.response.status_code:int}\"  ,   body-byte: %{NUMBER:nginx.http.response.body.bytes:int}  ,http-referer: \"%{DATA:nginx.http.request.referer}\"  ,user-agent: %{QS:nginx.user_agent} , X-Forwarded-For : \"%{DATA:nginx.http.request.xff}\" ,   request-time: \"%{NUMBER:nginx.http.request.time}\"  , response_time : \"(?:-|%{NUMBER:nginx.http.request.time:int})\"    

此處的patterns是符合我上面(1)的訊息，所產生的。

ps. Grok Debugger 記得要8.1後才有，如果版本過低，可以參考下方，用DevTools下指令的方式除錯。

Grok簡單說，

%{型態：要存的欄位名稱}

型態有哪些，參考連結『 grok-patterns'

ref.

• Grok Debugger
• Grok processor
• Grokking grok
• grok-patterns

4. 測試
   寫好了後，不測試一下也不知道是對或錯，
   到新建的畫面，右邊的中間，有Add Document的選項。
   
   
   
   
   之後輸入要測試的範例
   
   
   
   
   內容為：

    {
      "_source": {
        "message": "127.0.0.1 - - [06/May/2022:06:44:01 +0000] , http-host: \"abc.boxing.com\" , URL: \"GET /static/tpl/analytics/index.html?v=51817715.1 HTTP/1.1\"  , request-status : \"200\"  ,   body-byte: 0  ,http-referer: \"https://abc.boxing.com/mobile\"  ,user-agent: \"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/100.0.4896.127 Safari/537.36\" , X-Forwarded-For : \"132.235.211.18,10.107.10.14\" ,   request-time: \"0.000\"  , response_time : \"-\""
      }
    }

按下 Run the pipeline，有成功就會跑結果出來了。

失敗的話，會變成下面的圖。

此處皆參考官方範例
ref. Example: Parse logs in the Common Log Format

使用api測試

在UI上面測試 Grok實在很累，除了用剛剛提到的 Grok Debug以外。
就是直接用api方式直接驗證了。

Note, 8.1 我用Grok Debugger 有些特殊字元會無法match ，但用API沒問題。

Nginx 為 ingest Pipeline Name

以下API 皆在 DevTools驗證

1. 驗證Document

POST _ingest/pipeline/Nginx/_simulate
{
  "docs": [
    {
      "_source": {
        "message": "127.0.0.1 - - [06/May/2022:06:44:01 +0000] , http-host: \"abc.boxing.com\" , URL: \"GET /static/tpl/analytics/index.html?v=51817715.1 HTTP/1.1\"  , request-status : \"200\"  ,   body-byte: 0  ,http-referer: \"https://abc.boxing.com/mobile\"  ,user-agent: \"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/100.0.4896.127 Safari/537.36\" , X-Forwarded-For : \"132.235.211.18,10.107.10.14\" ,   request-time: \"0.000\"  , response_time : \"-\""
      }
    }
  ]
}

2. 修改grok

PUT _ingest/pipeline/Nginx
{
  "description": "My optional pipeline description",
  "processors": [
    {
      "grok": {
        "description": "Extract fields from 'message'",
        "field": "message",
        "patterns": ["%{IPORHOST:nginx.source.ip} %{USER:nginx.user.id} %{USER:nginx.user.name} \\[%{HTTPDATE:nginx.@timestamp}\\] , http-host: \"%{DATA:nginx.url.host}\" , URL: \"%{WORD:nginx.http.request.method} %{DATA:nginx.url.path} HTTP/%{NUMBER:nginx.http.version}\"  , request-status : \"%{NUMBER:nginx.http.response.status_code:int}\"  ,   body-byte: %{NUMBER:nginx.http.response.body.bytes:int}  ,http-referer: \"%{DATA:nginx.http.request.referer}\"  ,user-agent: %{QS:nginx.user_agent} , X-Forwarded-For : \"%{DATA:nginx.http.request.xff}\" ,   request-time: \"%{NUMBER:nginx.http.request.time}\"  , response_time : \"(?:-|%{NUMBER:nginx.http.request.time:int})\""],
         "ignore_failure": false
      }
    }
  ]
}

ignore_failure 為忽略錯誤，false的話，只要有一個沒過，
ingest pipeline就會停止。

結論

上述兩種方法，都可以建立且測試，就看哪種習慣。
但要將ingest 套用到data stream 或 index的話，
又是另一篇了…。

不趕時間的方法，到 Index Managemenet 修改 Index Templates

到 index settings 將 default_pipeline 修改為你的 ingest pipeline name

然後就等 lifecycle條件達到時，建立新的index，就會套用上去了。

趕時間的話，請看下一篇。

ref.

• set-default-pipeline
• index.default_pipeline
• Elasticsearch：Elastic可觀測性 - 運用 pipeline 使數據結構化