使用Fluentd進行簡單流處理

在某些日志采集場景中，我們需要對資料流進行一些轉換。比如，我們可能需要從日志記錄中提取某些字段以進行錯誤告警，或向日志記錄中插入新的字段用以後續的分析。

本文簡單介紹一下使用Fluentd進行資料操作的技術細節。

1. 根據日志字段取值來過濾事件

   談到過濾，我們通常會想到正規表達式，在linux中通常使用grep來進行文本查找和過濾。Fluentd内置了filter_grep過濾插件，可對資料流進行正則過濾。

   假設我們正在使用一個web服務，比如Apache，我們需要對其通路日志進行監控。由輸入插件産生的事件類似如下結構：

{
  "host": "192.168.1.1",
  "method": "GET",
  "path": "/index.html",
  "code": 200,
  "size": 2344,
  "referer": null
}
           

這其中的code字段表示使用者請求狀态，我們可能對狀态為2xx的請求不太關心，這樣就可以将這類事件過濾掉，專門處理使用者請求可能發生的異常情況。

我們可以通過在Fluentd中增加如下

<filter>

配置來實作事件過濾。

<filter apache.**>
  @type grep
  <exclude>
    key code
    pattern ^2\d\d$
  </exclude>
</filter>
           

使用grep過濾插件，通過key指定code字段為過濾字段，通過pattern比對code值為2xx的事件，将這些事件排除（exclude）掉。

<filter apache.**>
  @type grep
  <regexp>
    key code
    pattern ^5\d\d$
  </regexp>
  <exclude>
    key path
    pattern ^/test/
  </exclude>
</filter>
           

1. 向事件中插入定制字段

   我們可以在某個處理階段向日志記錄中插入一些字段，供後續使用。這可以通過Fluentd内置的filter_record_transformer過濾器插件來實作。

   假設我們是以叢集的方式來部署web服務的，我們可能需要标記使用者請求是由哪台伺服器來處理的。

   在Fluentd中進行如下配置即可實作此類需求：

<filter apache.**>
  @type record_transformer
  <record>
    server "${hostname}"
  </record>
</filter>
           

{
  "host": "192.168.1.1",
  "method": "GET",
  "path": "/index.html",
  "code": 200,
  "size": 2344,
  "referer": null,
  "server": "app1"
}