1. 使用 fluent-bit 采集文件

簡介

Fluent Bit是一款快速、靈活的日志處理器，旨在收集、解析、過濾日志，並將日志發送到遠程數據庫，以便執行數據分析。

數據分析通常發生在數據存儲和數據庫索引之后，但對於實時和復雜的分析需求，在日志處理器中處理仍在運行的數據會帶來很多好處，這種方法被稱為邊緣流處理(Stream Processing on the Edge)。

流式處理引擎架構

fluent-bit之所以這么說，其實是因為其架構設計了一個流式的處理引擎：

默認實現組件

fluent-bit實現了不少默認的組件：

• INPUT
  collectd, cpu-metrics, disk-io-metrics, docker-events, docker-metrics, dummy, exec, fluentbit-metrics, forward, head, health, http, kernel-logs, memory-metrics, mqtt, network-io-metrics, nginx, node-exporter-metrics, process, random, serial-interface, standard-input, statsd, syslog, systemd, tail, tcp, thermal, windows-event-log, windows-event-log-winevtlog, windows-exporter-metrics,

• OUTPUT
  new-relic, forward, prometheus-remote-write, treasure-data, opensearch, skywalking, prometheus-exporter, azure, azure_blob, postgresql, bigquery, loki, elasticsearch, counter, null, cloudwatch, standard-output, syslog, websocket, flowcounter, logdna, firehose, influxdb, tcp-and-tls, kinesis, stackdriver, kafka-rest-proxy, opentelemetry, stackdriver_special_fields, slack, gelf, s3, datadog, splunk, file, http, kafka, nats

• FILTERS
  grep, aws-metadata, nightfall, lua, parser, type-converter, nest, record-modifier, standard-output, throttle, multiline-stacktrace, expect, tensorflow, geoip2-filter, modify, checklist, kubernetes, rewrite-tag

• PARSERS
  ltsv, configuring-parser, regular-expression, decoders, json, logfmt

該系列文章將介紹如何使用Fluent Bit，進行數據采集、處理、分發的過程。

本節將以采集文本文件入手，並結合監控方法來逐步為您展開介紹；

編譯部署

前置條件

fluent-bit 采用c語言編寫，可以通過容器或者二進制進行部署安裝；其占用較少的CPU和內存資源，目前能夠兼容絕大部分基於x86、x86_64、arm32v7和arm64v8的系統平台；

為了能夠進行在本地進行測試編譯，你需要如下依賴環境：

• GCC or Clang
• CMake
• Flex & Bison: 僅當你需要流處理和記錄訪問器特性(這兩種功能默認也是開啟的)

編譯

# clone 代碼
git clone https://github.com/fluent/fluent-bit.git

# 切換到當前最新的一個發布分支
git checkout -b v1.9.1 v.9.1

# 編譯
cd build
cmake ..
make

# 安裝
# sudo make install 先不着急安裝，我們來測試一下

測試

1. 先來創建一個測試路徑：

mkdir ~/iSoft/fluent-bit -p
mkdir ~/isoft/fluent-bit/bin
mkdir ~/isoft/fluent-bit/conf
mkdir ~/isoft/fluent-bit/db
mkdir ~/isoft/fluent-bit/data
mkdir ~/isoft/fluent-bit/tmp

# 先將我們編譯路徑build下bin子路徑內的東西全部復制過去
cp build/bin/* ~/iSoft/fluent-bit/bin/

2. 在conf目錄下，創建三個配置文件：

fluent-bit.conf

[SERVICE]
    # Flush
    # =====
    # set an interval of seconds before to flush records to a destination
    flush        1

    # Daemon
    # ======
    # instruct Fluent Bit to run in foreground or background mode.
    daemon       Off

    # Log_Level
    # =========
    # Set the verbosity level of the service, values can be:
    #
    # - error
    # - warning
    # - info
    # - debug
    # - trace
    #
    # by default 'info' is set, that means it includes 'error' and 'warning'.
    log_level    info

    # Parsers File
    # ============
    # specify an optional 'Parsers' configuration file
    parsers_file parsers.conf

    # Plugins File
    # ============
    # specify an optional 'Plugins' configuration file to load external plugins.
    # plugins_file plugins.conf

    # HTTP Server
    # ===========
    # Enable/Disable the built-in HTTP Server for metrics
    http_server  On
    http_listen  0.0.0.0
    http_port    2020

    # Storage
    # =======
    # Fluent Bit can use memory and filesystem buffering based mechanisms
    #
    # - https://docs.fluentbit.io/manual/administration/buffering-and-storage
    #
    # storage metrics
    # ---------------
    # publish storage pipeline metrics in '/api/v1/storage'. The metrics are
    # exported only if the 'http_server' option is enabled.
    #
    #storage.metrics on

    # storage.path
    # ------------
    # absolute file system path to store filesystem data buffers (chunks).
    #
    # storage.path /tmp/storage

    # storage.sync
    # ------------
    # configure the synchronization mode used to store the data into the
    # filesystem. It can take the values normal or full.
    #
    # storage.sync normal

    # storage.checksum
    # ----------------
    # enable the data integrity check when writing and reading data from the
    # filesystem. The storage layer uses the CRC32 algorithm.
    #
    # storage.checksum off

    # storage.backlog.mem_limit
    # -------------------------
    # if storage.path is set, Fluent Bit will look for data chunks that were
    # not delivered and are still in the storage layer, these are called
    # backlog data. This option configure a hint of maximum value of memory
    # to use when processing these records.
    #
    # storage.backlog.mem_limit 5M

[INPUT]
    Name   tail
    Path   /home/etl/iSoft/fluent-bit/data/*.txt
    DB     /home/etl/iSoft/fluent-bit/db/tail.db


[OUTPUT]
    Name  stdout
    Match *

parser.conf

[PARSER]
    Name   json
    Format json
    Time_Key time
    Time_Format %d/%b/%Y:%H:%M:%S %z%

plugins.conf
這個文件其實暫時還用不上，是為我們自定義擴展插件的配置文件

[PLUGINS]
    # Path /path/to/out_gstdout.so

3. 啟動fluent-bit
先進入到我們自己創建的tmp目錄, 創建一些測試數據

cd ~/iSoft/fluent-bit/tmp

cat <<EOF > test-data.txt
{"log":"Done","stream":"stdout","time":"2018-02-19T23:25:29.1845622Z"}
{"log":"Done","stream":"stdout","time":"2018-02-19T23:25:29.1845622Z"}
{"log":"Done","stream":"stdout","time":"2018-02-19T23:25:29.1845622Z"}
{"log":"Done","stream":"stdout","time":"2018-02-19T23:25:29.1845622Z"}
{"log":"Done","stream":"stdout","time":"2018-02-19T23:25:29.1845622Z"}
{"log":"Done","stream":"stdout","time":"2018-02-19T23:25:29.1845622Z"}
{"log":"Done","stream":"stdout","time":"2018-02-19T23:25:29.1845622Z"}
{"log":"Done","stream":"stdout","time":"2018-02-19T23:25:29.1845622Z"}
{"log":"Done","stream":"stdout","time":"2018-02-19T23:25:29.1845622Z"}
{"log":"Done","stream":"stdout","time":"2018-02-19T23:25:29.1845622Z"}
EOF

根據上面fluent-bit.conf配置文件當中的INPUT/OUTPUT可以看出，我們利用了tail組件來監聽data目錄中的txt文件，並將結果輸出到標准輸出上來；

所以，啟動程序

./fluent-bit -c ../conf/fluent-bit.conf 
Fluent Bit v1.9.1
* Git commit: 619277847c6343dea9e4215deacd36cf61caf0a3
* Copyright (C) 2015-2021 The Fluent Bit Authors
* Fluent Bit is a CNCF sub-project under the umbrella of Fluentd
* https://fluentbit.io

[2022/03/20 15:23:43] [ info] [engine] started (pid=15543)
[2022/03/20 15:23:43] [ info] [storage] version=1.1.6, initializing...
[2022/03/20 15:23:43] [ info] [storage] in-memory
[2022/03/20 15:23:43] [ info] [storage] normal synchronization mode, checksum disabled, max_chunks_up=128
[2022/03/20 15:23:43] [ info] [cmetrics] version=0.3.0
[2022/03/20 15:23:43] [ info] [output:stdout:stdout.0] worker #0 started
[2022/03/20 15:23:43] [ info] [http_server] listen iface=0.0.0.0 tcp_port=2020
[2022/03/20 15:23:43] [ info] [sp] stream processor started

ok, fluent-bit已經阻塞開始監聽了

4. 開始測試

重新開啟一個終端創口，進入到創建好的tmp目錄，將測試數據手動批量的輸出到data目錄下：

cd ~/iSoft/fluent-bit/tmp
cat test-data.txt >> ../data/test.txt
cat test-data.txt >> ../data/test.txt
cat test-data.txt >> ../data/test.txt
cat test-data.txt >> ../data/test.txt
cat test-data.txt >> ../data/test.txt
cat test-data.txt >> ../data/test.txt
cat test-data.txt >> ../data/test.txt

這時可以看到fluent-bit阻塞的窗口已經在一直輸出了

Fluent Bit v1.9.1
* Git commit: 619277847c6343dea9e4215deacd36cf61caf0a3
* Copyright (C) 2015-2021 The Fluent Bit Authors
* Fluent Bit is a CNCF sub-project under the umbrella of Fluentd
* https://fluentbit.io

[2022/03/20 15:23:43] [ info] [engine] started (pid=15543)
[2022/03/20 15:23:43] [ info] [storage] version=1.1.6, initializing...
[2022/03/20 15:23:43] [ info] [storage] in-memory
[2022/03/20 15:23:43] [ info] [storage] normal synchronization mode, checksum disabled, max_chunks_up=128
[2022/03/20 15:23:43] [ info] [cmetrics] version=0.3.0
[2022/03/20 15:23:43] [ info] [output:stdout:stdout.0] worker #0 started
[2022/03/20 15:23:43] [ info] [http_server] listen iface=0.0.0.0 tcp_port=2020
[2022/03/20 15:23:43] [ info] [sp] stream processor started
[2022/03/20 15:23:43] [ info] [input:tail:tail.0] inotify_fs_add(): inode=11409520 watch_fd=1 name=/home/etl/iSoft/fluent-bit/data/test.txt
[0] tail.0: [1647761052.113552720, {"log"=>"{"log":"Done","stream":"stdout","time":"2018-02-19T23:25:29.1845622Z"}"}]
[0] tail.0: [1647763863.114999757, {"log"=>"{"log":"Done","stream":"stdout","time":"2018-02-19T23:25:29.1845622Z"}"}]
[1] tail.0: [1647763863.403883305, {"log"=>"{"log":"Done","stream":"stdout","time":"2018-02-19T23:25:29.1845622Z"}"}]
[0] tail.0: [1647763893.545292283, {"log"=>"{"log":"Done","stream":"stdout","time":"2018-02-19T23:25:29.1845622Z"}"}]
[0] tail.0: [1647763954.515264556, {"log"=>"{"log":"Done","stream":"stdout","time":"2018-02-19T23:25:29.1845622Z"}"}]
[1] tail.0: [1647763954.515268143, {"log"=>"{"log":"Done","stream":"stdout","time":"2018-02-19T23:25:29.1845622Z"}"}]
[2] tail.0: [1647763954.515268725, {"log"=>"{"log":"Done","stream":"stdout","time":"2018-02-19T23:25:29.1845622Z"}"}]
[3] tail.0: [1647763954.515269171, {"log"=>"{"log":"Done","stream":"stdout","time":"2018-02-19T23:25:29.1845622Z"}"}]
[4] tail.0: [1647763954.515269621, {"log"=>"{"log":"Done","stream":"stdout","time":"2018-02-19T23:25:29.1845622Z"}"}]
[5] tail.0: [1647763954.515270065, {"log"=>"{"log":"Done","stream":"stdout","time":"2018-02-19T23:25:29.1845622Z"}"}]
[6] tail.0: [1647763954.515270512, {"log"=>"{"log":"Done","stream":"stdout","time":"2018-02-19T23:25:29.1845622Z"}"}]
[7] tail.0: [1647763954.515270965, {"log"=>"{"log":"Done","stream":"stdout","time":"2018-02-19T23:25:29.1845622Z"}"}]
[8] tail.0: [1647763954.515271452, {"log"=>"{"log":"Done","stream":"stdout","time":"2018-02-19T23:25:29.1845622Z"}"}]
[9] tail.0: [1647763954.515276040, {"log"=>"{"log":"Done","stream":"stdout","time":"2018-02-19T23:25:29.1845622Z"}"}]

小結

通過上面的配置文件fluent-bit.conf看到，我們為INPUT tail組件，還配置了一個DB參數，那是因為tail組件采用了嵌入式的sqlite3的數據庫來記錄監聽文件的偏移量，我們可以去查看一下：

cd ~/iSoft/fluent-bit/db
sqlite3 tail.db 
sqlite> .schema
CREATE TABLE in_tail_files (  id      INTEGER PRIMARY KEY,  name    TEXT NOT NULL,  offset  INTEGER,  inode   INTEGER,  created INTEGER,  rotated INTEGER DEFAULT 0);
sqlite> 
sqlite> select * from in_tail_files ;
1|/home/etl/iSoft/fluent-bit/data/test.txt|77520|11409520|1647761023|0
sqlite> 

通過字段名稱可以看出來，分別是文件全路徑名稱、讀取到的文件偏移量(字節數)、文件inode、文件創建時間、文件滾動標記

監控

根據fluent-bit官網的介紹，我們在上面的配置文檔中，將fluent-bit進程的http_server設置為On，並且暴露在2020端口上，那么我們可以使用如下兩個接口來查看其輸出情況：

curl -s http://127.0.0.1:2020/api/v1/uptime | jq
{
  "uptime_sec": 391,
  "uptime_hr": "Fluent Bit has been running:  0 day, 0 hour, 6 minutes and 31 seconds"
}


curl -s http://127.0.0.1:2020/api/v1/metrics | jq
{
  "input": {
    "tail.0": {
      "records": 1,
      "bytes": 88,
      "files_opened": 1,
      "files_closed": 0,
      "files_rotated": 0
    }
  },
  "filter": {},
  "output": {
    "stdout.0": {
      "proc_records": 1,
      "proc_bytes": 88,
      "errors": 0,
      "retries": 0,
      "retries_failed": 0,
      "dropped_records": 0,
      "retried_records": 0
    }
  }
}

拉取promethues grafana鏡像

fluent-bit為promethues提供了監控接口，那么我們來嘗試一下：

在本地拉取promethues和grafana的鏡像（這里就簡單的將這兩位泡在容器里）

docker pull prom/prometheus
docker pull grafana/grafana

配置promethues

mkdir ~/isoft/prometheus
vim ~/isoft/prometheus/fluent-bit-prom.yaml

這里為prometheus增加了一個job，因為是運行在docker里面，所以選擇了另外一個宿主機IP, 以能否訪問運行在宿主機上的fluent-bit

# my global config
global:
  scrape_interval: 15s # Set the scrape interval to every 15 seconds. Default is every 1 minute.
  evaluation_interval: 15s # Evaluate rules every 15 seconds. The default is every 1 minute.
  # scrape_timeout is set to the global default (10s).

# Alertmanager configuration
alerting:
  alertmanagers:
    - static_configs:
        - targets:
          # - alertmanager:9093

# Load rules once and periodically evaluate them according to the global 'evaluation_interval'.
rule_files:
  # - "first_rules.yml"
  # - "second_rules.yml"

# A scrape configuration containing exactly one endpoint to scrape:
# Here it's Prometheus itself.
scrape_configs:
  # The job name is added as a label `job=<job_name>` to any timeseries scraped from this config.
  - job_name: "prometheus"

    # metrics_path defaults to '/metrics'
    # scheme defaults to 'http'.

    static_configs:
      - targets: ["localhost:9090"]
  - job_name: "fluent-bit"
    metrics_path: '/api/v1/metrics/prometheus' # metrics_path defaults to '/metrics'
    # scheme defaults to 'http'.

    static_configs:
      # 宿主機IP
      - targets: ["192.168.241.1:2020"]

啟動

docker run -d -p 9090:9090 --name prom -v /home/etl/iSoft/prometheus/fluent-bit-prom.yaml:/etc/prometheus/prometheus.yml prom/prometheus

通過瀏覽器訪問 http://localhost:9090/

配置grafana

mkdir ~/isoft/grafana/storage -p
chmod 777 ~/isoft/grafana/storage

啟動

docker run -d -p 3000:3000 --name grafana -v /home/etl/iSoft/grafana/storage:/var/lib/grafana grafana/grafana

添加promethues數據源

導入fluent-bit提供的默認dashboard
fluent-bit-dashboard.json

查看監控界面
輸入項

輸出項

發現資源都叫 tail.0，所以根據文檔，將配置文件INPUT/OUTPUT處增加兩個別名，以區分可能監控到的多個配置

[INPUT]
   Name   tail
   Path   /home/etl/iSoft/fluent-bit/data/*.txt
   DB     /home/etl/iSoft/fluent-bit/db/tail.db
   Alias  monitor_txt_file
[OUTPUT]
   Name  stdout
   Match *
   Alias output_txt_file

再次查看監控界面
輸入項

輸出項

總結

至此，先完成初步測試，感興趣的同學可以持續關注，后續會逐個介紹異構數據處理和發送（盡量多介紹相關組件和場景的使用）、如何使用pipline streaming的窗口SQL統計、為fluent-bit開發擴展組件等功能；

至於windows環境編譯較為復雜（需要微軟編譯環境，我已測試過），感興趣的同學，可以直接從github下載對應的win版本來進行測試；（我測試是OK的）