Fluentd介紹

Fluentd是一個(gè)是一個(gè)開(kāi)源的日志收集和傳輸工具，旨在解決日志數(shù)據(jù)的收集、傳輸和處理問(wèn)題，它可以收集來(lái)自于各種系統(tǒng)或應(yīng)用的日志，轉(zhuǎn)化為用戶指定的格式后，轉(zhuǎn)發(fā)到用戶所指定的日志存儲(chǔ)系統(tǒng)之中。
用圖來(lái)說(shuō)明問(wèn)題的話，在沒(méi)有使用之前Fluentd，日志采集過(guò)濾存儲(chǔ)流程如下所示：

使用Fluentd之后，日志系統(tǒng)如下所示：

Fluentd負(fù)責(zé)從服務(wù)器收集日志信息，將數(shù)據(jù)流交給后續(xù)存儲(chǔ)服務(wù)和查詢工具。換句話說(shuō)，F(xiàn)luentd不負(fù)責(zé)生產(chǎn)數(shù)據(jù)、不負(fù)責(zé)存儲(chǔ)數(shù)據(jù)，只是數(shù)據(jù)的搬運(yùn)工。

Fluentd特點(diǎn)

使用JSON進(jìn)行統(tǒng)一日志記錄：Fluentd嘗試盡可能地將數(shù)據(jù)結(jié)構(gòu)化為JSON：這允許Fluentd 統(tǒng)一處理日志數(shù)據(jù)的所有方面：收集，過(guò)濾，緩沖和跨多個(gè)源和目標(biāo)（統(tǒng)一日志層）輸出日志。使用JSON可以更輕松地進(jìn)行下游數(shù)據(jù)處理，因?yàn)樗哂凶銐虻慕Y(jié)構(gòu)，可以在保留靈活模式的同時(shí)進(jìn)行訪問(wèn)。
可插拔架構(gòu)：Fluentd擁有靈活的插件系統(tǒng)，允許社區(qū)擴(kuò)展其功能。我們的500多個(gè)社區(qū)貢獻(xiàn)插件連接了數(shù)十個(gè)數(shù)據(jù)源和數(shù)據(jù)輸出。通過(guò)利用插件，您可以立即開(kāi)始更好地使用日志。
所需的資源較少： Fluentd是用C語(yǔ)言和Ruby組合編寫(xiě)的，只需要很少的系統(tǒng)資源。vanilla實(shí)例運(yùn)行30-40MB內(nèi)存，可處理13,000個(gè)事件/秒/核心。
內(nèi)置可靠性：Fluentd支持基于內(nèi)存和文件的緩沖，以防止節(jié)點(diǎn)間數(shù)據(jù)丟失。Fluentd還支持強(qiáng)大的故障轉(zhuǎn)移功能，可以設(shè)置為高可用性。

Fluentd對(duì)比Filebeat

Fluentd 是一個(gè)日志收集和聚合的工具，它支持多種數(shù)據(jù)輸入和輸出，并提供了豐富的插件和過(guò)濾器來(lái)處理和轉(zhuǎn)換數(shù)據(jù)。Fluentd 通過(guò)流水線的方式處理日志數(shù)據(jù)，可以靈活地構(gòu)建復(fù)雜的數(shù)據(jù)處理流程。Fluentd 還提供了內(nèi)置的緩存機(jī)制，可以減少數(shù)據(jù)丟失的風(fēng)險(xiǎn)。同時(shí)，F(xiàn)luentd 是一個(gè)輕量級(jí)的工具，對(duì)系統(tǒng)資源的占用較小。
Filebeat 是 Elastic 公司開(kāi)源的一款輕量級(jí)日志數(shù)據(jù)采集器，旨在收集、解析和轉(zhuǎn)發(fā)各種類(lèi)型的日志數(shù)據(jù)。它支持多種數(shù)據(jù)源采集并內(nèi)置了七十多種場(chǎng)景服務(wù)日志過(guò)濾處理模塊，并且能夠自動(dòng)發(fā)現(xiàn)新的日志文件。雖然也支持?jǐn)?shù)據(jù)解析，但解析器目前僅有四十多種，僅能完成簡(jiǎn)單的數(shù)據(jù)解析操作。Filebeat 具有較低的資源消耗，并且支持高可用性和故障轉(zhuǎn)移。Filebeat 可以將日志數(shù)據(jù)轉(zhuǎn)發(fā)到多個(gè)輸出目的地，如 Elasticsearch、Logstash 或 Kafka 等。
兩者在設(shè)計(jì)理念上的區(qū)別主要在于 Fluentd 更側(cè)重于數(shù)據(jù)處理和轉(zhuǎn)換，而 Filebeat 更側(cè)重于數(shù)據(jù)采集和傳輸。Fluentd 更適用于需要進(jìn)行復(fù)雜數(shù)據(jù)處理和轉(zhuǎn)換的場(chǎng)景，而 Filebeat 則更適用于簡(jiǎn)單的日志數(shù)據(jù)采集和轉(zhuǎn)發(fā)場(chǎng)景。如果你需要一個(gè)輕量級(jí)的工具來(lái)處理海量日志數(shù)據(jù)，F(xiàn)luentd 可能會(huì)更合適；如果你只需要一個(gè)簡(jiǎn)單的日志采集器來(lái)快速收集和轉(zhuǎn)發(fā)日志數(shù)據(jù)，那么 Filebeat 可能會(huì)更適合你的需求。

Fluentd對(duì)比Logstash

1. 設(shè)計(jì)哲學(xué)：

• Fluentd的設(shè)計(jì)目標(biāo)是簡(jiǎn)單、可擴(kuò)展和靈活。它采用輕量級(jí)的標(biāo)簽（tag）和記錄（record）的概念來(lái)處理日志數(shù)據(jù)。Fluentd提供了豐富的輸入和輸出插件，可以與各種數(shù)據(jù)源和目標(biāo)進(jìn)行集成。
• Logstash的設(shè)計(jì)目標(biāo)是可編排、易使用和功能全面。它使用基于管道（pipeline）的配置模型，通過(guò)輸入、過(guò)濾和輸出插件來(lái)處理日志數(shù)據(jù)。Logstash提供了很多內(nèi)置插件，可以方便地與各種系統(tǒng)進(jìn)行集成。

2. 插件生態(tài)系統(tǒng)：

• Fluentd擁有一個(gè)龐大的插件生態(tài)系統(tǒng)，支持?jǐn)?shù)百種輸入和輸出插件。這些插件覆蓋了各種數(shù)據(jù)源和目標(biāo)，包括文件、數(shù)據(jù)庫(kù)、消息隊(duì)列、云服務(wù)等。此外，F(xiàn)luentd還支持自定義插件的開(kāi)發(fā)。
• Logstash也有一個(gè)廣泛的插件生態(tài)系統(tǒng)，包括大量的輸入、過(guò)濾和輸出插件。它支持與許多數(shù)據(jù)源和目標(biāo)的集成，如文件、數(shù)據(jù)庫(kù)、消息隊(duì)列、Web服務(wù)等。

3. 性能和資源消耗：

• Fluentd在性能方面表現(xiàn)出色，它使用事件驅(qū)動(dòng)的架構(gòu)，能有效地處理大量的日志數(shù)據(jù)。Fluentd還支持多線程和異步處理，以提高吞吐量和響應(yīng)性能。
• Logstash在處理大量數(shù)據(jù)時(shí)可能會(huì)遇到性能和資源消耗的挑戰(zhàn)。它使用JRuby作為運(yùn)行環(huán)境，相對(duì)于Fluentd的C擴(kuò)展，可能需要更多的內(nèi)存和處理資源。

Fluentd對(duì)比FluentBit

Fluent Bit 是一個(gè)開(kāi)源的多平臺(tái)日志采集器，旨在打造日志采集處理和分發(fā)的通用利器。2014 年,Fluentd 團(tuán)隊(duì)預(yù)測(cè)對(duì)于嵌入式 Linux 和 Gateways 等受約束的環(huán)境，需要更輕量級(jí)的日志處理器，于是便開(kāi)發(fā)了Fluent Bit，并把該項(xiàng)目建設(shè)成為Fluentd 生態(tài)系統(tǒng)的一部分。

出于性能和兼容性原因，推薦從Fluentd切換到Fluent Bit。Fluent Bit現(xiàn)在被認(rèn)為是下一代解決方案。

Fluentd部署與配置

部署調(diào)試

下載地址：https://www.fluentd.org/download/fluent_package

# wget https://s3.amazonaws.com/packages.treasuredata.com/lts/5/redhat/8/x86_64/fluent-package-5.0.1-1.el8.x86_64.rpm
# rpm -ivh fluent-package-5.0.1-1.el8.x86_64.rpm
# 添加配置文件，讀取/var/log/test.log文件內(nèi)容，輸出到終端
# cat > /etc/fluent/test.conf << EOF
<source>
  @type tail
  path /var/log/test.log
  pos_file /var/log/fluent/test.log.pos
  tag test.logs
  read_from_head true
  <parse>
    @type none
  </parse>
</source>

<match test.logs>
  @type stdout
</match>
EOF
# 指定配置文件啟動(dòng)fluentd
# fluentd -c /etc/fluent/test.conf

# 新建終端追加內(nèi)容到test.log文件
# echo "hello fluentd" >> /var/log/test.log

# 觀察fluentd控制臺(tái)輸出
2023-09-05 15:40:57.992178132 +0800 test.logs: {"message":"hello fluent"}

Fluentd事件

Fluentd是一個(gè)日志收集系統(tǒng)，那么一條日志消息，在Fluentd里就認(rèn)為是一個(gè)事件
Fluentd的事件由下面三部分組成

• 標(biāo)簽(tag): 用于說(shuō)明這個(gè)事件是哪里產(chǎn)生的，可用于后面的事件路由，例如test.logs
• 時(shí)間(time): 事件是什么時(shí)候發(fā)生的，例如2023-09-05 15:40:57.992178132 +0800
• 記錄(record): 事件內(nèi)容本身，JSON格式，例如{“message”:“hello fluent”}

當(dāng)fluentd收到一個(gè)事件之后，會(huì)經(jīng)過(guò)一系列的處理流程:

• 如修改事件的相關(guān)字段
• 過(guò)濾掉一些不關(guān)心的事件
• 路由事件輸出到不同的地方

Fluentd配置文件插件組成

• source：指定輸入源，可以有一個(gè)或多個(gè)
• filter：指定事件處理流程
• match：指定輸出，可以有一個(gè)或多個(gè)
• parse：解析器插件，可以在source、match、filter中使用
• format：格式化插件，可以在match、filter中使用
• buffer：緩沖區(qū)插件，可以在match中使用

配置文件通配符

在前面的配置中，我們?cè)谳斎朐粗付╰ag test.logs，然后在輸出中指定match test.logs，以此實(shí)現(xiàn)指定輸入與輸出關(guān)聯(lián)。fluented主要根據(jù)事件的tag來(lái)分區(qū)不同的處理流程，在filter和match過(guò)程中除了可以指定tag名稱(chēng)外，也可以使用通配符匹配
tag通常是一個(gè)字符串，由.分隔，比如myapp.access

• *: 匹配滿足一個(gè)tag部分的事件, 比如: a.*, 它將匹配a.b這樣的tag, 但是不會(huì)處理a或者a.b.c這類(lèi)tag
• **: 匹配滿足0個(gè)或多個(gè)tag部分，比如: a.**, 它將匹配a, a.b, a.b.c這三種tag
• {X, Y, Z}: 匹配滿足X,Y或者Z的tag, 比如: {a, b}將匹配a或者b,但是不會(huì)匹配c。
  這種格式也可以和通配符組合使用,比如a.{b.c}.*或a.{b.c}.**
• #{...} 會(huì)把花括號(hào)內(nèi)的字符串當(dāng)做是ruby的表達(dá)式處理。比如

<match "app.#{ENV['FLUENTD_TAG']}">
  @type stdout
</match>

如果設(shè)置了環(huán)境變量FLUENTD_TAG為dev,那上面等價(jià)于app.dev

• 當(dāng)指定了多個(gè)模式時(shí)（使用一個(gè)或多個(gè)空格分開(kāi)）,只要滿足其中任意一個(gè)就行。
  比如:
  <match a b>匹配a和b
<match a.** b.*>匹配a, a.b, a.b.c, b.d等

標(biāo)簽路由

fluentd的處理流程是根據(jù)我們?cè)谂渲梦募械亩x，從上到下依次執(zhí)行的。假如我們?cè)谂渲梦募锒x了比較多輸入源，同時(shí)針對(duì)不同的輸入源需要使用不同的filters和輸出時(shí)，如果仍然按照從上到下執(zhí)行的順序的話，由于不同的處理需求，我們的配置文件可能變得非常復(fù)雜。此時(shí)可以通過(guò)label標(biāo)簽，使配置文件結(jié)構(gòu)更加清晰便于配置和閱讀。
配置文件示例

<source>
  @type http
  bind 0.0.0.0
  port 9999
  @label @STAGING
</source>

<filter test.cycle>
  @type grep
  <exclude>
    key action
    pattern ^login$
  </exclude>
</filter>

<label @STAGING>
  <filter test.cycle>
    @type grep
    <exclude>
      key action
      pattern ^logout$
    </exclude>
  </filter>

  <match test.cycle>
    @type stdout
  </match>
</label>

在source輸入中，定義一個(gè)名為STAGING的標(biāo)簽，然后在filter中過(guò)濾了login事件，接著又在label中過(guò)濾了logout事件，接下來(lái)訪問(wèn)測(cè)試

# curl -i -X POST -d 'json={"action":"login","user":2}' http://localhost:9999/test.cycle
# curl -i -X POST -d 'json={"action":"logout","user":2}' http://localhost:9999/test.cycle
# curl -i -X POST -d 'json={"action":"request","user":2}' http://localhost:9999/test.cycle

控制臺(tái)打印結(jié)果如下：

2023-09-07 10:53:37.288951438 +0800 test.cycle: {"action":"login","user":2}
2023-09-07 10:54:18.860724760 +0800 test.cycle: {"action":"request","user":2}

可以看到，只有logout事件被過(guò)濾了，原因是我們?yōu)檩斎朐O(shè)置了label，因此fluentd會(huì)在執(zhí)行時(shí)跳過(guò)其余的filter，只運(yùn)行l(wèi)abel標(biāo)簽塊中的filter規(guī)則。

輸入插件（source）

輸入源可以一次指定多個(gè)， @type參數(shù)指定使用哪一個(gè)輸入插件。
fluentd支持各種輸入插件, 比如:

• tail：從文本文件讀取數(shù)據(jù)
• forward：從TCP/UDP接收事件流，主要用于從其他fluentd實(shí)例接收數(shù)據(jù)。
• udp：從UDP中接收數(shù)據(jù)
• tcp：從TCP中接收數(shù)據(jù)
• unix：從unix中接收數(shù)據(jù)
• http：通過(guò)HTTP請(qǐng)求接收數(shù)據(jù)
• syslog：通過(guò)TCP或UDP接收syslog數(shù)據(jù)
• exec：執(zhí)行外部程序來(lái)接收事件日志
• sample：接收示例數(shù)據(jù)，主要用于調(diào)試使用
• monitor_agent：通過(guò) REST API 導(dǎo)出 Fluentd 的內(nèi)部指標(biāo)
• windows_eventlog：從windows事件日志中讀取數(shù)據(jù)

插件的具體使用可以參考文檔:
https://docs.fluentd.org/input

http輸入

配置文件如下所示：

• @type http指定輸入類(lèi)型為http
• port和bind：指定監(jiān)聽(tīng)地址和端口
• 在match中指定了要匹配的標(biāo)簽名稱(chēng)，并將匹配的事件發(fā)送到stdout插件進(jìn)行輸出

<source>
  @type http
  port 9999
  bind 0.0.0.0
</source>

<match http.logs>
  @type stdout
</match>

測(cè)試驗(yàn)證

# curl -i -X POST -d 'json={"action":"hello","fluent":1}' http://127.0.0.1:9999/http.logs
# 控制臺(tái)輸出如下
2023-09-05 16:08:05.238093979 +0800 http.logs: {"action":"hello","fluent":1}

單個(gè)文件輸入

<source>
  @type tail
  path /var/log/test.log
  pos_file /var/log/fluent/test.log.pos
  tag test.logs
  read_from_head true
  refresh_interval 5 
  <parse>
    @type none
  </parse>
</source>

<match test.logs>
  @type stdout
</match>

• @type tail：插件類(lèi)型，指定為tail，從文件輸入
• path /var/log/test.log：指定要讀取的文件路徑，可以使用*通配符匹配。
• pos_file /var/log/fluent/test.log.pos：用于跟蹤讀取位置的位置文件路徑
• tag test.logs：為讀取的事件設(shè)置的標(biāo)簽名稱(chēng)
• read_from_head true：表示讀取整個(gè)文件，而不只是新的日志行。
• refresh_interval 5：刷新文件以檢查新事件的時(shí)間間隔（秒），默認(rèn)為5秒
• parse @type none：事件解析器的類(lèi)型，none表示不進(jìn)行解析

多個(gè)文件輸入

<source>
  @type tail
  path /var/log/test.log
  pos_file /var/log/fluent/test.log.pos
  tag test.logs
  read_from_head true
  refresh_interval 5 
  <parse>
    @type none
  </parse>
</source>
<source>
  @type tail
  path /var/log/json.log
  pos_file /var/log/fluent/json.log.pos
  tag json.logs
  read_from_head true
  <parse>
    @type json
    time_format %Y-%m-%dT%H:%M:%S.%NZ
    time_type string
  </parse>
</source>

<match *.logs>
  @type stdout
</match>

• parse @type json：事件解析器的類(lèi)型，解析為json內(nèi)容
• time_format、time_type：指定時(shí)間戳字段格式和內(nèi)容
• match *.logs：匹配所有以.logs結(jié)尾的事件發(fā)送到stdout插件進(jìn)行輸出

測(cè)試驗(yàn)證

# echo "123.com" >> /var/log/test.log
# echo '{"time":"2023-9-5T01:01:01.123Z","text":"hello","age":18}' >> /var/log/json.log

# 控制臺(tái)查看輸出
2023-09-05 21:36:15.853468713 +0800 test.logs: {"message":"123.com"}
2023-09-05 01:01:01.123000000 +0800 json.logs: {"text":"hello","age":18}

輸出插件（match）

輸出也可以一次指定多個(gè)， @type參數(shù)指定使用哪一個(gè)輸出插件。
fluentd支持各種輸出插件, 比如:

• file：將事件寫(xiě)入文件，默認(rèn)情況下，每天（大約 00：10）創(chuàng)建文件，首次使用該插件導(dǎo)入記錄時(shí)，不會(huì)立即創(chuàng)建任何文件。
• forward：用于將事件轉(zhuǎn)發(fā)到其他fluentd中。
• http：HTTP / HTTPS寫(xiě)入記錄
• exec：事件傳遞給外部程序
• copy：事件復(fù)制到多個(gè)輸出
• relabel：重新標(biāo)記事件
• roundrobin：加權(quán)循環(huán)算法將事件分發(fā)到多個(gè)輸出
• stdout：終端輸出，主要用于調(diào)試
• null：丟棄指定事件
• s3
• kafka
• elasticsearch
• opensearch
• mongo
• mongo_replset
• rewrite_tag_filter
• webhdfs

插件的具體使用可以參考文檔:
https://docs.fluentd.org/output

輸出到文件

<source>
  @type sample
  sample {"hello":"world"}
  tag sample.logs
</source>

<match sample.logs>
  @type file
  path /var/log/sample.logs
  append true
  compress gzip
</match>

• @type file：輸出插件類(lèi)型，指定為file
• path：輸出文件的路徑
• append：是否將事件追加到文件末尾，如果設(shè)置為false，則會(huì)覆蓋現(xiàn)有文件，默認(rèn)為true
• compress：指定文件壓縮格式

查看驗(yàn)證

# cat /var/log/sample.logs/buffer.b60498a2519b1bb278da14031742109bc.log
2023-09-05T16:44:25+08:00       sample.logs     {"hello":"world"}

輸出到elasticsearch

如果需要輸出到elasticsearch需要提前安裝插件

fluent-gem install fluent-plugin-elasticsearch

配置文件示例

<source>
  @type sample
  sample {"hello":"world"}
  tag sample.logs
</source>

<match sample.logs>
  @type elasticsearch
  host 172.18.0.100
  port 9200
  default_elasticsearch_version 8
  user elastic
  password 0IgIjSHiIhGWp_TWqoJk
  scheme https
  ssl_verify false
  index_name fluentd-${tag}
  # api_key "ElasticsearchAPIKEY" # 或者使用api_key方式驗(yàn)證
</match>

• host：es主機(jī)地址，可以有多個(gè)地址，例如：hosts https://customhost.com:443/path,https://username:password@host-failover.com:443
• user、password：連接es的賬號(hào)和密碼
• default_elasticsearch_version：指定es版本
• scheme：連接方式是http還是https
• ssl_verify：跳過(guò)ca證書(shū)驗(yàn)證
• index_name：指定索引名稱(chēng)

輸出到kafka

如果需要輸出到kafka需要提前安裝插件

fluent-gem install fluent-plugin-kafka

配置文件示例

<source>
  @type sample
  sample {"hello":"world"}
  tag sample.logs
</source>

<match sample.logs>
  @type kafka2

  brokers <broker1_host>:<broker1_port>,<broker2_host>:<broker2_port>
  use_event_time true

  <buffer topic>
    @type file
    path /var/log/td-agent/buffer/td
    flush_interval 3s
  </buffer>
 
  <format>
    @type json
  </format>

  topic_key topic
  default_topic messages
  required_acks -1
  compression_codec gzip
</match>

• @type kafka2：指定輸出到kafka
• brokers：kafka集群broker列表
• use_event_time：將fluentd事件設(shè)置為kafka的createTime，默認(rèn)為false使用當(dāng)前時(shí)間
• topic_key：默認(rèn)topic名稱(chēng)
• required_acks：請(qǐng)求確認(rèn)次數(shù)，默認(rèn)-1不確認(rèn)
• compression_codec：壓縮消息編碼，默認(rèn)不壓縮

過(guò)濾插件（filter）

過(guò)濾也可以一次指定多個(gè)， @type參數(shù)指定使用哪一個(gè)過(guò)濾插件。
fluentd支持各種輸出插件, 比如:

• record_transformer：添加、刪除、修改事件內(nèi)容
• grep：過(guò)濾指定字段值的事件
• parser：字段值數(shù)據(jù)格式化。
• geoip：使用Maxmind GeoIP數(shù)據(jù)庫(kù)將地理位置信息添加到日志中
• stdout：將事件打印到標(biāo)準(zhǔn)輸出，主要用于調(diào)試

插件的具體使用可以參考文檔:
https://docs.fluentd.org/filter

字段添加、刪除、修改、替換

配置示例：

<source>
  @type sample
  sample {"total":100, "count":10,"message":"hello world!","name":"alex","pod":{"pod_name":"mypod","pod_id":"b1187408"}}
  tag sample.logs
</source>

<filter sample.logs>
  @type record_transformer
  enable_ruby
  <record>
    avg ${record["total"] / record["count"]}
    hostname "#{Socket.gethostname}"
    tag ${tag}
    message fluentd, ${record["message"]}
    name ${record["name"].gsub("alex", "tom")}
  </record>
  remove_keys name,$.pod.pod_id
</filter>

<match sample.logs>
  @type stdout
</match>

• avg ${record[“total”] / record[“count”]}：新增avg字段，值為record[“total”] / record[“count”]
• hostname “#{Socket.gethostname}”：新增hostname字段，值為#{Socket.gethostname}
• message fluentd, ${record[“message”]}：修改message字段，在前面添加fluentd
• remove_keys $.pod.pod_id：刪除pod下的pod_id字段
• name ${record[“name”].gsub(“alex”, “tom”)}：將name字段中的alex值替換為tom

執(zhí)行結(jié)果如下：

2023-09-05 18:00:36.097785459 +0800 sample.logs: {"total":100,"count":10,"message":"fluentd, hello world!","name":"tom","pod":{"pod_name":"mypod"},"avg":10,"hostname":"huanbao","tag":"sample.logs"}

過(guò)濾事件

配置示例：

<source>
  @type sample
  sample {"message":"I am cool but you are uncool", "hostname":"db001.example.com"}
  sample {"hostname":"web001.example.com"}
  sample {"message":"It's cool outside today"}
  sample {"message":"It's cool outside today", "hostname":"web001.example.com"}
  sample {"message":"That's not cool", "hostname":"web1337.example.com"}
  tag sample.logs
</source>

<filter sample.logs>
  @type grep
  <regexp>
    key message
    pattern /cool/
  </regexp>

  <regexp>
    key hostname
    pattern /^web\d+\.example\.com$/
  </regexp>

  <exclude>
    key message
    pattern /uncool/
  </exclude>
</filter>

<match sample.logs>
  @type stdout
</match>

• key message，pattern /cool/：message字段的值包含 cool
• key hostname：hostname字段的值與 web.example.com 匹配。
• key message，pattern /uncool/：message字段的值不包含 uncool。

執(zhí)行結(jié)果如下：

2023-09-05 18:07:44.097470768 +0800 sample.logs: {"message":"That's not cool","hostname":"web1337.example.com"}

數(shù)據(jù)解析

利用解析器插件，我們可以將數(shù)據(jù)解析并格式化為指定格式的內(nèi)容
配置示例：

<source>
  @type sample
  sample {"log":"192.168.0.1 - - [05/Feb/2023:09:00:00 +0900] \"GET / HTTP/1.1\" 200 777"}
  tag sample.logs
</source>

<filter sample.logs>
  @type parser
  key_name log
  <parse>
    @type regexp
    expression /^(?<host>[^ ]*) [^ ]* (?<user>[^ ]*) \[(?<time>[^\]]*)\] "(?<method>\S+)(?: +(?<path>[^ ]*) +\S*)?" (?<code>[^ ]*) (?<size>[^ ]*)$/
    time_format %d/%b/%Y:%H:%M:%S %z
  </parse>
</filter>

<match sample.logs>
  @type stdout
</match>

• key_name log：解析log字段內(nèi)容
• @type regexp：使用正則解析器
• expression：解析表達(dá)式
• time_format：時(shí)間字段格式

執(zhí)行結(jié)果如下：

2023-09-05 11:00:00.000000000 +0800 sample.logs: {"host":"192.168.0.1","user":"-","method":"GET","path":"/","code":"200","size":"777"}

地理位置信息解析

安裝地理IP解析庫(kù)

dnf group install "Development Tools"
dnf yum install geoip-devel --enablerepo=epel

fluentd安裝解析插件

fluent-gem install fluent-plugin-geoip

配置示例：

<source>
  @type sample
  sample {"host":"66.102.9.80"}
  tag sample.logs
</source>

<filter sample.logs>
  @type geoip
  geoip_lookup_keys host
  geoip2_database /etc/fluent/GeoLite2-City.mmdb
  backend_library geoip2_c
  <record>
    city            ${city.names.en["host"]}
    latitude        ${location.latitude["host"]}
    longitude       ${location.longitude["host"]}
    country         ${country.iso_code["host"]}
    country_name    ${country.names.en["host"]}
    postal_code     ${postal.code["host"]}
    region_code     ${subdivisions.0.iso_code["host"]}
    region_name     ${subdivisions.0.names.en["host"]}
  </record>
</filter>

<match sample.logs>
  @type stdout
</match>

• geoip_lookup_keys：指定解析ip的字段
• geoip2_database：本地ip數(shù)據(jù)庫(kù)地址
• backend_library：ip庫(kù)類(lèi)型，可選值為geoip, geoip2_compat, geoip2_c

執(zhí)行結(jié)果：

2023-09-05 20:19:20.099538026 +0800 sample.logs: {"host":"66.102.9.80","city":null,"latitude":37.751,"longitude":-97.822,"country":"US","country_name":"United States","postal_code":null,"region_code":null,"region_name":null}

解析器插件（parse）

在fluentd輸入、過(guò)濾、輸出過(guò)程中，均可以使用多種解析器插件，目前支持的解析器插件如下：

• regexp：正則匹配，類(lèi)似logstash的grok插件
• apache2：解析Apache日志
• apache_error：解析Apache error日志
• nginx：解析NGINX日志
• syslog：解析syslog日志
• csv：解析csv格式數(shù)據(jù)
• tsv：解析tsv格式數(shù)據(jù)
• ltsv：解析ltsv格式數(shù)據(jù)
• json：將字符串解析為json數(shù)據(jù)
• msgpack：解析MessagePack數(shù)據(jù)
• multiline：解析多行日志
• none：不解析處理

更多解析器詳細(xì)信息參考文檔：https://docs.fluentd.org/parser

正則匹配解析

在線調(diào)試正則地址：http://fluentular.herokuapp.com/
常用正則匹配參數(shù)

• 忽略大小寫(xiě)：expression /…/i
• 多行匹配：expression /…/m
• 同時(shí)指定參數(shù)：expression /…/im

配置示例

<source>
  @type sample
  sample {"log":"[2013-02-28 12:00:00 +0900] alice engineer 1"}
  tag sample.logs
</source>

<filter sample.logs>
  @type parser
  key_name log
  <parse>
    @type regexp
    expression /^\[(?<logtime>[^\]]*)\] (?<name>[^ ]*) (?<title>[^ ]*) (?<id>\d*)$/
    time_key logtime
    time_format %Y-%m-%d %H:%M:%S %z
    types id:integer
  </parse>
</filter>

<match sample.logs>
  @type stdout
</match>

執(zhí)行結(jié)果

2013-02-28 11:00:00.000000000 +0800 sample.logs: {"name":"alice","title":"engineer","id":1}

json解析

配置示例

<source>
  @type sample
  sample {"log":"{\"key1\": \"value1\", \"key2\": 2}"}
  tag sample.logs
</source>

<filter sample.logs>
  @type parser
  key_name log
  reserve_data true
  remove_key_name_field true
  <parse>
    @type json
    json_parser json
  </parse>
</filter>

<match sample.logs>
  @type stdout
</match>

• key_name：需要轉(zhuǎn)換的字段
• reserve_data：是否保留轉(zhuǎn)換前除log之外的其他字段
• remove_key_name_field：是否刪除轉(zhuǎn)換前的log字段
• json_parser：json解析器可選oj, yajl, json

執(zhí)行結(jié)果

2023-09-09 09:38:47.016389000 +0800 sample.logs: {"key1":"value1","key2":2}

多行正則解析

多行解析只能在in_tail中使用
例如java異常日志如下

2023-3-03 14:27:33 [main] INFO  Main - Start
2023-3-03 14:27:33 [main] ERROR Main - Exception
javax.management.RuntimeErrorException: null
    at Main.main(Main.java:16) ~[bin/:na]
2023-3-03 14:27:33 [main] INFO  Main - End

配置示例

<source>
  @type tail
  path /var/log/test.log
  pos_file /var/log/fluent/test.log.pos
  tag test.logs
  read_from_head true
  <parse>
    @type multiline
    format_firstline /\d{4}-\d{1,2}-\d{1,2}/
    format1 /^(?<time>\d{4}-\d{1,2}-\d{1,2} \d{1,2}:\d{1,2}:\d{1,2}) \[(?<thread>.*)\] (?<level>[^\s]+)(?<message>.*)/
  </parse>
</source>

<match test.logs>
  @type stdout
</match>

• format_firstline：匹配多行日志起始位置
• formatN：每行日志匹配表達(dá)式

執(zhí)行結(jié)果

2023-03-03 14:27:33.000000000 +0800 test.logs: {"thread":"main","level":"INFO","message":"  Main - Start"}
2023-03-03 14:27:33.000000000 +0800 test.logs: {"thread":"main","level":"ERROR","message":" Main - Exception\njavax.management.RuntimeErrorException: null\n    at Main.main(Main.java:16) ~[bin/:na]"}

其他插件

格式化插件（format）

格式化插件配置在match輸出中，主要用于將自定義格式輸出
目前內(nèi)置的格式化插件如下：

• out_file：將事件輸出格式化為由分隔符分隔的時(shí)間、標(biāo)簽和 json 記錄
• json：將事件輸出格式化為 JSON
• ltsv：將事件輸出格式化為ltsv
• csv：將事件輸出格式化為csv
• msgpack：將事件輸出格式化為msgpack
• hash：將事件輸出格式化為哈希值
• single_value：輸出單個(gè)字段的值，而不是整個(gè)記錄
• tsv：將事件輸出格式化為tsv

更多格式化插件參考文檔：https://docs.fluentd.org/formatter

json

輸出為標(biāo)準(zhǔn)json格式，需要注意的是json 格式化后結(jié)果不包含tag和time字段。
配置示例：

<source>
  @type sample
  sample {"hello":"world"}
  tag sample.logs
</source>

<match sample.logs>
  @type stdout
  <format>
    @type json
  </format>
</match>

執(zhí)行結(jié)果

{"hello":"world"}

緩沖區(qū)插件（buffer）

緩沖區(qū)插件配置在match輸出中，主要用于暫時(shí)存儲(chǔ)傳入流數(shù)據(jù)，使用buffer插件可有效提高數(shù)據(jù)傳輸過(guò)程中的性能和可靠性。
目前內(nèi)置的格式化插件如下：

• memory：使用內(nèi)存來(lái)存儲(chǔ)緩沖區(qū)塊
• file：使用磁盤(pán)文件存儲(chǔ)緩沖區(qū)塊

更多格式化插件參考文檔：https://docs.fluentd.org/buffer

內(nèi)存緩沖

配置示例

<source>
  @type sample
  sample {"hello":"world"}
  tag sample.logs
</source>

<match sample.logs>
  @type file
  path /var/log/sample.logs
  append true
  compress gzip
  <buffer>
    @type memory
  </buffer>
</match>

文件緩沖

配置示例

<source>
  @type sample
  sample {"hello":"world"}
  tag sample.logs
</source>

<match sample.logs>
  @type file
  path /var/log/sample.logs
  append true
  compress gzip
  <buffer>
    @type file
    path /var/log/fluent/buf/sample.*
  </buffer>
</match>

結(jié)果驗(yàn)證

[root@huanbao buf]# pwd
/var/log/fluent/buf
[root@huanbao buf]# ls
sample.b604bb35d921c75b350f722e963dacbef.log  sample.b604bb35d921c75b350f722e963dacbef.log.meta

存儲(chǔ)插件（storage）

有時(shí)，輸入/過(guò)濾器/輸出插件需要將其內(nèi)部狀態(tài)保存在內(nèi)存、存儲(chǔ)或鍵值存儲(chǔ)中。Fluentd有一個(gè)名為Storage的插件，它允許插件存儲(chǔ)并重用其內(nèi)部狀態(tài)作為鍵值對(duì)。

本地存儲(chǔ)

配置示例

<source>
  @type sample
  sample {"hello":"world"}
  tag sample.logs
  <storage>
    @type local
    path /var/log/fluent/storage/sample.json
  </storage>
</source>

<match sample.logs>
  @type stdout
</match>

結(jié)果驗(yàn)證

[root@huanbao buf]# cd /var/log/fluent/storage/
[root@huanbao storage]# ls
sample.json
[root@huanbao storage]# cat sample.json 
{"increment_value":0,"dummy_index":0}

Fluentd采集解析日志實(shí)踐

調(diào)試fluentd配置

原始日志格式

2023-07-23 09:35:18.987 | INFO     | __main__:debug_log:49 - {'access_status': 200, 'request_method': 'GET', 'request_uri': '/account/', 'request_length': 67, 'remote_address': '186.196.110.240', 'server_name': 'cu-36.cn', 'time_start': '2023-07-23T09:35:18.879+08:00', 'time_finish': '2023-07-23T09:35:19.638+08:00', 'http_user_agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_12_3) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.2999.0 Safari/537.36'}

1. fluentd添加source配置，從文件中讀取數(shù)據(jù)，新增match配置，輸出到控制臺(tái)，fluentd配置文件如下所示

<source>
  @type tail
  path /opt/log_demo/log/info.log
  pos_file /var/log/fluent/demo.logs.pos
  tag demo.logs
  read_from_head true
  <parse>
    @type none
  </parse>
</source>
<match demo.logs>
  @type stdout
</match>

觀察控制臺(tái)輸出內(nèi)容，已正確采集到了/opt/log_demo/log/info.log文件中的日志內(nèi)容并添加到了message字段中

2023-09-09 09:20:10.439369419 +0800 demo.logs: {"message":"2023-09-09 09:20:10.438 | WARNING  | __main__:debug_log:48 - {'access_status': 404, 'request_method': 'POST', 'request_uri': '/login/', 'request_length': 89, 'remote_address': '154.87.205.94', 'server_name': 'cm-3.cn', 'time_start': '2023-09-09T09:20:09.729+08:00', 'time_finish': '2023-09-09T09:20:10.500+08:00', 'http_user_agent': 'Chrome 9\\tMozilla/5.0 (Macintosh; Intel Mac OS X 10_13_2) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/94.0.4606.81 Safari/537.36'}"}

2. 接下來(lái)添加filter過(guò)濾規(guī)則，通過(guò)regexp正則匹配解析，將日志內(nèi)容解析為logtime、level、class、content四部分，并指定logtime為事件時(shí)間字段

<filter demo.logs>
  @type parser
  key_name message
  <parse>
    @type regexp
    expression /^(?<logtime>[^|]+) \| (?<level>[A-Z]*) *\| __main__:(?<class>\D*:\d*) - (?<content>.*)$/
    time_key logtime
    time_format %Y-%m-%d %H:%M:%S.%L
  </parse>
</filter>

添加正則解析規(guī)則后，控制臺(tái)輸出內(nèi)容如下，觀察可知事件時(shí)間已替換為正確的日志生成時(shí)間，并提取到了level、class、content相關(guān)內(nèi)容

2023-09-09 09:24:20.979000000 +0800 demo.logs: {"level":"INFO","class":"debug_log:53","content":"{'access_status': 200, 'request_method': 'GET', 'request_uri': '/login/', 'request_length': 46, 'remote_address': '213.225.29.116', 'server_name': 'cm-4.cn', 'time_start': '2023-09-09T09:24:20.323+08:00', 'time_finish': '2023-09-09T09:24:21.775+08:00', 'http_user_agent': 'Chrome 9\\tMozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'}"}

3. 由于content內(nèi)容不是標(biāo)準(zhǔn)的json格式，因此在json解析前我們需要使用record_transformer字符處理插件，將其中的’替換為"。

<filter demo.logs>
  @type record_transformer
  enable_ruby
  <record>
    content ${record["content"].gsub("'", '"')}
  </record>
</filter>

添加字符替換規(guī)則后，控制臺(tái)輸出如下，已成功將content字段的值中的所有’替換為"

2023-09-09 09:27:44.646000000 +0800 demo.logs: {"level":"INFO","class":"debug_log:71","content":"{\"access_status\": 301, \"request_method\": \"DELETE\", \"request_uri\": \"/login/\", \"request_length\": 54, \"remote_address\": \"128.46.31.93\", \"server_name\": \"cu-8.cn\", \"time_start\": \"2023-09-09T09:27:44.331+08:00\", \"time_finish\": \"2023-09-09T09:27:45.052+08:00\", \"http_user_agent\": \"Firefox 7\\tMozilla/5.0 (Windows NT 10.0; Win64; x64; rv:78.0) Gecko/20100101 Firefox/78.0\"}"}

4. 現(xiàn)在content字段值已經(jīng)是標(biāo)準(zhǔn)的json格式，接下來(lái)可以使用json解析器解析。

<filter demo.logs>
  @type parser
  key_name content
  <parse>
    @type json
  </parse>
</filter>

接下來(lái)觀察控制臺(tái)輸出，已將content中的內(nèi)容解析并添加到事件內(nèi)容中

2023-09-09 09:39:29.342134038 +0800 demo.logs: {"level":"INFO","class":"debug_log:71","access_status":200,"request_method":"GET","request_uri":"/account/","request_length":12,"remote_address":"170.4.117.203","server_name":"cu-4.cn","time_start":"2023-09-09T09:39:28.871+08:00","time_finish":"2023-09-09T09:39:30.303+08:00","http_user_agent":"Firefox 7\tMozilla/5.0 (Windows NT 10.0; Win64; x64; rv:78.0) Gecko/20100101 Firefox/78.0"}

5. 我們已經(jīng)成功解析到了remote_address字段，接下來(lái)使用geoip插件從數(shù)據(jù)庫(kù)中查詢ip的地理位置信息。

<filter demo.logs>
  @type geoip
  geoip_lookup_keys remote_address
  geoip2_database /etc/fluent/GeoLite2-City.mmdb
  backend_library geoip2_c
  <record>
    geoip_city            ${city.names.en["remote_address"]}
    geoip_latitude        ${location.latitude["remote_address"]}
    geoip_longitude       ${location.longitude["remote_address"]}
    geoip_country         ${country.iso_code["remote_address"]}
    geoip_country_name    ${country.names.en["remote_address"]}
    geoip_postal_code     ${postal.code["remote_address"]}
    geoip_region_name     ${subdivisions.0.names.en["remote_address"]}
  </record>
</filter>

觀察控制臺(tái)輸出，已經(jīng)成功通過(guò)remote_address字段的ip地址獲取到了地理位置信息內(nèi)容。

2023-09-09 09:45:17.910753124 +0800 demo.logs: {"level":"INFO","class":"debug_log:54","access_status":200,"request_method":"GET","request_uri":"/account/","request_length":94,"remote_address":"176.146.11.173","server_name":"cm-4.cn","time_start":"2023-09-09T09:45:17.727+08:00","time_finish":"2023-09-09T09:45:18.115+08:00","http_user_agent":"Safari 11\tMozilla/5.0 (Macintosh; Intel Mac OS X 10_13_1) AppleWebKit/604.3.5 (KHTML, like Gecko) Version/11.0.1 Safari/604.3.5","geoip_city":"Chassieu","geoip_latitude":45.743,"geoip_longitude":4.97,"geoip_country":"FR","geoip_country_name":"France","geoip_postal_code":"69680","geoip_region_name":"Auvergne-Rhone-Alpes"}

6. 數(shù)據(jù)解析完成后，我們配置match輸出到elasticsearch中。

<match demo.logs>
  @type elasticsearch
  host 172.18.0.100
  port 9200
  default_elasticsearch_version 8
  user elastic
  scheme https
  password 0IgIjSHiIhGWp_TWqoJk
  ssl_verify false
  index_name fluentd-${tag}
  <buffer>
    @type file
    path /fluentd/buf/demo.logs.*
  </buffer>
</match>

此時(shí)觀察kibana索引，日志數(shù)據(jù)已成功寫(xiě)入es中。

構(gòu)建fluentd鏡像

由于fluentd鏡像未安裝elasticsearch和geoip插件，如果在pod的init階段在線安裝插件經(jīng)常會(huì)導(dǎo)致拉取超時(shí)無(wú)法正常啟動(dòng)。因此需要提前構(gòu)建包含elasticsearch和geoip插件的fluentd鏡像，并上傳至harbor倉(cāng)庫(kù)中。

[root@tiaoban fluent]# ls
Dockerfile  fluentd-conf.yaml  GeoLite2-City.mmdb  log-demo.yaml
[root@tiaoban fluent]# cat Dockerfile 
FROM fluentd:v1.16-debian-1
USER root
RUN apt-get update && apt-get install -y build-essential libgeoip-dev autoconf automake libtool libffi-dev && apt-get autoclean
ADD GeoLite2-City.mmdb /fluentd/GeoLite2-City.mmdb
RUN fluent-gem install fluent-plugin-elasticsearch && fluent-gem install fluent-plugin-geoip && rm -rf /var/lib/apt/lists/* /var/cache/debconf/* /var/log/* /var/tmp/*
[root@tiaoban fluent]# docker build -t harbor.local.com/elk/fluentd:v1.16 .
[root@tiaoban fluent]# docker push harbor.local.com/elk/fluentd:v1.16

k8s部署資源

[root@tiaoban fluentd]# kubectl apply -f .
configmap/fluentd-conf created
deployment.apps/log-demo created
[root@tiaoban fluentd]# kubectl get pod -o wide
NAME                        READY   STATUS    RESTARTS      AGE     IP             NODE    NOMINATED NODE   READINESS GATES
log-demo-59b8687b8c-428qv   2/2     Running   0             5m11s   10.244.3.186   work3   <none>           <none>
log-demo-59b8687b8c-rft4f   2/2     Running   0             5m11s   10.244.4.210   work2   <none>           <none>

訪問(wèn)驗(yàn)證

訪問(wèn)kibana索引管理，已創(chuàng)建名為fluentd-demo.logs的索引

查看索引數(shù)據(jù)內(nèi)容，已完成對(duì)數(shù)據(jù)的過(guò)濾清洗操作，并成功加載地理位置信息。

接下來(lái)調(diào)整索引模板setting和mapping以及ILM策略等配置，具體內(nèi)容可參考文檔：https://www.cuiliangblog.cn/detail/article/65。

完整資源清單

本實(shí)驗(yàn)案例所有yaml文件已上傳至git倉(cāng)庫(kù)。訪問(wèn)地址如下：

github

https://github.com/cuiliang0302/blog-demo

gitee

https://gitee.com/cuiliang0302/blog_demo

參考文檔

fluentd輸入插件：https://docs.fluentd.org/input
fluentd輸出插件：https://docs.fluentd.org/output
fluentd過(guò)濾插件：https://docs.fluentd.org/filter

查看更多

微信公眾號(hào)

微信公眾號(hào)同步更新，歡迎關(guān)注微信公眾號(hào)《崔亮的博客》第一時(shí)間獲取最近文章。

博客網(wǎng)站

崔亮的博客-專(zhuān)注devops自動(dòng)化運(yùn)維，傳播優(yōu)秀it運(yùn)維技術(shù)文章。更多原創(chuàng)運(yùn)維開(kāi)發(fā)相關(guān)文章，歡迎訪問(wèn)https://www.cuiliangblog.cn文章來(lái)源地址http://www.zghlxwxcb.cn/news/detail-704937.html

到了這里，關(guān)于ES8生產(chǎn)實(shí)踐——pod日志采集（Fluentd方案）的文章就介紹完了。如果您還想了解更多內(nèi)容，請(qǐng)?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！