背景

1，filebeat直連Elasticsearch，需要對massage提取一些特定的字段。

2，如果你對數據需要處理的比較多還是建議用logstash，logstash更強大一些。

pipeline 簡單介紹

pipeline 我最開始了解的時候是使用jenkins的時候，即交付流水線。

平時寫的jenkinsfile就是用的pipeline。

我理解的pipeline就是把你在界面上配置的自動化變成代碼腳本。即Pipeline as Code。

一、原始massage

原始的日志內容如下，

我需要提取：4975481caf96a734bdb300acc118b5b5

2021-06-17 20:50:39|ERROR|4975481caf96a734bdb300acc118b5b5|1623934239.079|20003|user service error|/var/www/html/vendor/mdp/java-client/src/JavaClientResponse.php

二、操作

想要使用filebeat提取massage字段

1，得先在es里面創建一個pipeline腳本。

2，配置filebeat使用es的pipeline腳本。

2.1 創建pipeline.json文件

就是隨便在linux目錄創建一個文件。

文件名稱隨便起，后期能找到就行。方便以后做更改。
vim extract-traceid-pipeline.json
{
  "description" : "extract-traceid-pipeline",    # 描述，可以描述一下這個pipeline的作用
  "processors" : [
    {
      "grok" :{            # grok 語法
        "field" : "message",    # 需要對你采集的信息的哪個字段進行處理。其實就是filebeat采集完一條日志之后 來執行這個處理。你可以對采集的任何字段的值進行處理
        "patterns" : ["ERROR\\|%{DATA:trace_id}\\|"]    # 這個就是從message信息中提取信息的語法了。
      }
    }
  ]
}

# [""] 這個是固定寫法，里面的才是正則
# 2021-06-17 20:50:39|ERROR|4975481caf96a734bdb300acc118b5b5|1623934239.079......
# 提取4975481caf96a734bdb300acc118b5b5,可以看到這段數據之前的是ERROR|
# 那我前面就寫ERROR\\| |需要轉義但是es也需要轉義一層，所以就是兩個\\
# %{} 這個是固定的寫法，就是你想要提取哪個信息然后 使你提取出來的信息成為一個字段key:value。
# DATA 是grok的語法，內置字段類型，比如
                               DATE：表示你提取的信息是日期
                               NUMBER：表示為數值
                               DATA：表示為字符串
                               IP：表示為一個ip地址
                               你在這標識的類型會在kibana上面把key的類型標識出來。
# ：trace_id 這個表示你給 你提取出來的信息定義一個Key的名稱，比如
                               trace_id：4975481caf96a734bdb300acc118b5b5
                               到時候你收集到的這條日志信息里面就會多出這一個字段
                               你在kibana就能搜索到。
                               這個名稱你隨便定義就行。

2.2 將json文件傳到es

在你json文件所在的目錄執行
curl -H "Content-Type: application/json" -XPUT 'http://192.168.31.61:9200/_ingest/pipeline/extract-traceid-pipeline' -d@extract-traceid-pipeline.json

# extract-traceid-pipeline 這個名稱你可以隨便起，但是在filebeat配置的時候需要寫這個名稱
# extract-traceid-pipeline.json 這個就是你自己的json文件

2.3 查看pipeline腳本

可以看到es里面有我們上傳的pipeline規則了

訪問：http://192.168.31.61:9200/_ingest/pipeline，我沒安裝google瀏覽器json插件。安裝一個就格式化了。

還有就是可以用kibana的開發工具看到：

2.4 filebeat配置

# 編輯filebeat.yml配置文件

output.elasticsearch:
  hosts: ["192.168.31.61:9200","192.168.31.62:9200","192.168.31.63:9200"]
  pipelines:                                  # pipelines 就是你下面可以放多個pipeline
    - pipeline: "extract-traceid-pipeline"    # 這個名稱就是你curl put到es里面 你后面寫的那個名稱
      when.contains:                          # 這里是一個filebeat的配置語法，可以看看filebeat的官方文檔，大致的意思就是if 條件判斷
        log_topics: "php-errlogs"             # 就是當log_topics："php-errlogs" 的時候才用這個pipeline 
        
        
   # when 就是if
   # contains 就是 包含的意思 還有其他的比如：and,or,not,equals(必須相等)