輕量級的日志采集組件 Filebeat 講解與實戰操作
一、概述
Filebeat是一個輕量級的日志數據收集工具，屬于Elastic公司的Elastic Stack（ELK Stack）生態系統的一部分。它的主要功能是從各種來源收集日志數據，將數據發送到Elasticsearch、Logstash或其他目標，以便進行搜索、分析和可視化。
以下是Filebeat的主要概述和特點：
- 輕量級：Filebeat是一個輕量級的代理，對系統資源的消耗非常低。它設計用于高性能和低延遲，可以在各種環境中運行，包括服務器、容器和虛擬機。
- 多源收集：Filebeat支持從各種來源收集數據，包括日志文件、系統日志、Docker容器日志、Windows事件日志等。它具有多個輸入模塊，可以輕松配置用于不同數據源的數據收集。
- 模塊化：Filebeat采用模塊化的方式組織配置，每個輸入類型都可以作為一個模塊，易于擴展和配置。這使得添加新的數據源和日志格式變得更加簡單。
- 自動發現：Filebeat支持自動發現服務，可以在容器化環境中自動識別新的容器和服務，并開始收集其日志數據。
- 安全性：Filebeat支持安全傳輸，可以使用TLS/SSL加密協議將數據安全地傳輸到目標。它還支持基于令牌的身份驗證。
- 數據處理：Filebeat可以對數據進行簡單的處理，如字段分割、字段重命名和數據過濾，以確保數據適合進一步處理和分析。
- 目標輸出：Filebeat可以將數據發送到多個目標，最常見的是將數據發送到Elasticsearch，以便進行全文搜索和分析。此外，還可以將數據發送到Logstash、Kafka等目標。
- 實時性：Filebeat可以以實時方式收集和傳輸數據，確保日志數據及時可用于分析和可視化。
- 監控和管理：Filebeat具有自身的監控功能，可以監視自身的狀態和性能，并與Elasticsearch、Kibana等工具集成，用于管理和監控數據收集。
工作的流程圖如下：
圖片
Filebeat的采集原理的主要步驟
- 數據源檢測：
Filebeat首先配置要監視的數據源，這可以是日志文件、系統日志、Docker容器日志、Windows事件日志等。Filebeat可以通過輸入模塊配置來定義數據源。
- 數據收集：
一旦數據源被定義，Filebeat會定期輪詢這些數據源，檢查是否有新的數據產生。
如果有新數據，Filebeat將讀取數據并將其發送到后續處理階段。
數據處理：
Filebeat可以對采集到的數據進行一些簡單的處理，例如字段分割、字段重命名、數據解析等。這有助于確保數據格式適合進一步的處理和分析。
數據傳輸：
采集到的數據將被傳輸到一個或多個目標位置，通常是Elasticsearch、Logstash或Kafka等。
Filebeat可以配置多個輸出目標，以便將數據復制到多個地方以增加冗余或分發數據。
安全性和可靠性：
Filebeat支持安全傳輸，可以使用TLS/SSL協議對數據進行加密。它還具有數據重試機制，以確保數據能夠成功傳輸到目標位置。
數據目的地：
數據被傳輸到目標位置后，可以被進一步處理、索引和分析。目標位置通常是Elasticsearch，用于全文搜索和分析，或者是Logstash用于進一步的數據處理和轉換，也可以是Kafka等其他消息隊列。
實時性和監控：
Filebeat可以以實時方式監視數據源，確保新數據能夠快速傳輸和處理。
Filebeat還可以與監控工具集成，以監控其自身的性能和狀態，并將這些數據發送到監控系統中。
總的來說，Filebeat采集原理是通過輪詢監視數據源，將新數據采集并發送到目標位置，同時確保數據的安全傳輸和可靠性。它提供了一種高效且靈活的方式來處理各種類型的日志和事件數據，以便進行后續的分析和可視化。
二、Kafka 安裝
為了快速部署，這里選擇通過docker-compose部署，可以參考我這篇文章：【中間件】通過 docker-compose 快速部署 Kafka 保姆級教程
# 先安裝 zookeeper
git clone https://gitee.com/hadoop-bigdata/docker-compose-zookeeper.git
cd docker-compose-zookeeper
docker-compose -f docker-compose.yaml up -d
# 安裝kafka
git clone https://gitee.com/hadoop-bigdata/docker-compose-kafka.git
cd docker-compose-kafka
docker-compose -f docker-compose.yaml up -d如果僅僅只是為測試也可以部署一個單機kafka官方下載地址：http://kafka.apache.org/downloads
### 1、下載kafka
wget https://downloads.apache.org/kafka/3.4.1/kafka_2.12-3.4.1.tgz --no-check-certificate
### 2、解壓
tar -xf kafka_2.12-3.4.1.tgz
### 3、配置環境變量
# ~/.bashrc添加如下內容：
export PATH=$PATH:/opt/docker-compose-kafka/images/kafka_2.12-3.4.1/bin
### 4、配置zookeeper 新版Kafka已內置了ZooKeeper，如果沒有其它大數據組件需要使用ZooKeeper的話，直接用內置的會更方便維護。
# vi kafka_2.12-3.4.1/config/zookeeper.properties
#注釋掉
#maxClientCnxns=0
#設置連接參數，添加如下配置
#為zk的基本時間單元，毫秒
tickTime=2000
#Leader-Follower初始通信時限 tickTime*10
initLimit=10
#Leader-Follower同步通信時限 tickTime*5
syncLimit=5
#設置broker Id的服務地址
#hadoop-node1對應于前面在hosts里面配置的主機映射,0是broker.id， 2888是數據同步和消息傳遞端口，3888是選舉端口
server.0=local-168-182-110:2888:3888
### 5、配置kafka
# vi kafka_2.12-3.4.1/config/server.properties
#添加以下內容：
broker.id=0
listeners=PLAINTEXT://local-168-182-110:9092
# 上面容器的zookeeper
zookeeper.cnotallow=local-168-182-110:2181
# topic不存在的，kafka就會創建該topic。
#auto.create.topics.enable=true
### 6、啟動服務
./bin/zookeeper-server-start.sh -daemon config/zookeeper.properties
./bin/kafka-server-start.sh -daemon config/server.properties
### 7、測試驗證
#創建topic
kafka-topics.sh --bootstrap-server local-168-182-110:9092 --create --topic topic1 --partitions 8 --replication-factor 1
#列出所有topic
kafka-topics.sh --bootstrap-server local-168-182-110:9092 --list
#列出所有topic的信息
kafka-topics.sh --bootstrap-server local-168-182-110:9092 --describe
#列出指定topic的信息
kafka-topics.sh --bootstrap-server local-168-182-110:9092 --describe --topic topic1
#生產者（消息發送程序）
kafka-console-producer.sh --broker-list local-168-182-110:9092 --topic topic1
#消費者（消息接收程序）
kafka-console-consumer.sh --bootstrap-server local-168-182-110:9092 --topic topic1三、Filebeat 安裝
1）下載 Filebeat
官網地址：https://www.elastic.co/cn/downloads/past-releases#filebeat
curl -L -O https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-7.6.2-linux-x86_64.tar.gz
tar -xf filebeat-7.6.2-linux-x86_64.tar.gz2）Filebeat 配置參數講解
Filebeat的配置文件通常是YAML格式，包含各種配置參數，用于定義數據源、輸出目標、數據處理和其他選項。以下是一些常見的Filebeat配置參數及其含義：
- filebeat.inputs：指定要監視的數據源。可以配置多個輸入，每個輸入定義一個數據源。每個輸入包括以下參數：
type：數據源的類型，例如日志文件、系統日志、Docker日志等。
paths：要監視的文件路徑或者使用通配符指定多個文件。
enabled：是否啟用該輸入。
示例：
filebeat.inputs:
- type: log
paths:
- /var/log/*.log
- type: docker
enabled: true- filebeat.modules：定義要加載的模塊，每個模塊用于解析特定類型的日志或事件數據。每個模塊包括以下參數：
module：模塊名稱。
enabled：是否啟用模塊。
var：自定義模塊變量。
示例：
filebeat.modules:
- module: apache
access:
enabled: true
error:
enabled: true- output.elasticsearch：指定將數據發送到Elasticsearch的配置參數，包括Elasticsearch主機、索引名稱等。
hosts：Elasticsearch主機列表。
index：索引名稱模板。
username和password：用于身份驗證的用戶名和密碼。
pipeline：用于數據預處理的Ingest節點管道。
示例：
output.elasticsearch:
hosts: ["localhost:9200"]
index: "filebeat-%{[agent.version]}-%{+yyyy.MM.dd}"
username: "your_username"
password: "your_password"- output.logstash：指定將數據發送到Logstash的配置參數，包括Logstash主機和端口等。
hosts：Logstash主機列表。
index：索引名稱模板。
ssl：是否使用SSL/TLS加密傳輸數據。
示例：
output.logstash:
hosts: ["localhost:5044"]
index: "filebeat-%{[agent.version]}-%{+yyyy.MM.dd}"
ssl.enabled: true- processors：定義對數據的預處理步驟，包括字段分割、重命名、添加字段等。
add_fields：添加字段到事件數據。
decode_json_fields：解碼JSON格式的字段。
drop_fields：刪除指定字段。
rename：重命名字段。
示例：
processors:
- add_fields:
target: "my_field"
value: "my_value"
- drop_fields:
fields: ["field1", "field2"]- filebeat.registry.path：指定Filebeat用于跟蹤已經讀取的文件和位置信息的注冊文件的路徑。
- filebeat.autodiscover：自動發現數據源，特別是用于容器化環境，配置自動檢測新容器的策略。
- logging.level：指定Filebeat的日志級別，可選項包括info、debug、warning等。
這些是 Filebeat 的一些常見配置參數，具體的配置取決于您的使用場景和需求。您可以根據需要自定義配置文件，以滿足您的數據采集和處理需求。詳細的配置文檔可以在Filebeat官方文檔中找到。
3）filebeat.prospectors 推送kafka完整配置
這里主要用到幾個核心字段：filebeat.prospectors、processors、output.kafka
1、filebeat.prospectors
filebeat.prospectors：用于定義要監視的數據源和采集規則。每個 prospector 包含一個或多個輸入規則，它們指定要監視的文件或數據源以及如何采集和解析數據。
以下是一個示例 filebeat.prospectors 部分的配置：
filebeat.prospectors:
- type: log
enabled: true
paths:
- /var/log/*.log
exclude_files:
- "*.gz"
multiline.pattern: '^\['
multiline.negate: false
multiline.match: after
tags: ["tag1", "tag2"]
tail_files: true
fields:
app: myapp
env: production在上述示例中，我們定義了一個 filebeat.prospectors 包含一個 type: log 的 prospector，下面是各個字段的解釋：
- type（必需）：數據源的類型。在示例中，類型是 log，表示監視普通文本日志文件。Filebeat支持多種類型，如 log、stdin、tcp、udp 等。
- enabled：是否啟用此 prospector。如果設置為 true，則啟用，否則禁用。默認為 true。
- paths（必需）：要監視的文件或文件模式，可以使用通配符指定多個文件。在示例中，Filebeat將監視 /var/log/ 目錄下的所有以 .log 結尾的文件。
- exclude_files：要排除的文件或文件模式列表。這里排除了所有以 .gz 結尾的文件。可選字段。
- multiline.pattern：多行日志的起始模式。如果您的日志事件跨越多行，此選項可用于合并多行日志事件。例如，設置為 'pattern' 將根據以 'pattern' 開頭的行來合并事件。
- multiline.negate：是否取反多行日志模式。如果設置為 true，則表示匹配不包含多行日志模式的行。可選字段，默認為 false。
- multiline.match：多行匹配模式，可以是 before（與上一行合并）或 after（與下一行合并）。如果設置為 before，則當前行與上一行合并為一個事件；如果設置為 after，則當前行與下一行合并為一個事件。可選字段，默認為 after。
- tags：為采集的事件添加標簽，以便后續的數據處理。標簽是一個字符串數組，可以包含多個標簽。在示例中，事件將被標記為 "tag1" 和 "tag2"。可選字段。
- tail_files：用于控制Filebeat是否應該跟蹤正在寫入的文件（tail文件）。當 tail_files 設置為 true 時，Filebeat將監視正在被寫入的文件，即使它們還沒有完成。這對于實時監視日志文件非常有用，因為它允許Filebeat立即處理新的日志行。默認情況下，tail_files 是啟用的，因此只有在特殊情況下才需要顯式設置為 false。
- fields：為事件添加自定義字段。這是一個鍵值對，允許您添加額外的信息到事件中。在示例中，事件將包含 "app" 字段和 "env" 字段，分別設置為 "myapp" 和 "production"。可選字段。
這些字段允許您配置Filebeat以滿足特定的數據源和采集需求。您可以根據需要定義多個 prospector 來監視不同類型的數據源，每個 prospector 可以包含不同的參數。通過靈活配置 filebeat.prospectors，Filebeat可以適應各種日志和數據采集場景。
2、processors
processors 是Filebeat配置中的一個部分，用于定義在事件傳輸到輸出目標之前對事件數據進行預處理的操作。您可以使用 processors 來修改事件數據、添加字段、刪除字段，以及執行其他自定義操作。以下是一些常見的 processors 配置示例和說明：
- 添加字段（Add Fields）：可以使用 add_fields 處理器將自定義字段添加到事件中，以豐富事件的信息。例如，將應用程序名稱和環境添加到事件中：
processors:
- add_fields:
fields:
app: myapp
env: production- 刪除字段（Drop Fields）：使用 drop_fields 處理器可以刪除事件中的指定字段。以下示例刪除名為 "sensitive_data" 的字段：
processors:
- drop_fields:
fields: ["sensitive_data"]- 解碼 JSON 字段（Decode JSON Fields）：如果事件中包含JSON格式的字段，您可以使用 decode_json_fields 處理器將其解碼為結構化數據。以下示例將名為 "json_data" 的字段解碼為結構化數據：
processors:
- decode_json_fields:
fields: ["json_data"]
target: ""- 字段重命名（Rename Fields）：可以使用 rename 處理器重命名事件中的字段。例如，將 "old_field" 重命名為 "new_field"：
processors:
- rename:
fields:
- from: old_field
to: new_field- 條件處理（Conditional Processing）：使用 if 條件可以根據事件的特定字段或屬性來選擇是否應用某個處理器。以下示例根據事件中的 "log_level" 字段，僅在 "error" 日志級別時添加 "error" 標簽：
processors:
- add_tags:
tags: ["error"]
when:
equals:
log_level: "error"- 多個處理器（Multiple Processors）：您可以配置多個處理器，它們將按照順序依次應用于事件數據。例如，您可以先添加字段，然后刪除字段，最后重命名字段。
processors 部分允許您對事件數據進行復雜的處理和轉換，以適應特定的需求。您可以根據需要組合不同的處理器來執行多個操作，以確保事件數據在傳輸到輸出目標之前滿足您的要求。
3、output.kafka
output.kafka 是Filebeat配置文件中的一個部分，用于配置將事件數據發送到Kafka消息隊列的相關設置。以下是 output.kafka 部分的常見參數及其解釋：
output.kafka:
hosts: ["kafka-broker1:9092", "kafka-broker2:9092"]
topic: "my-log-topic"
partition.round_robin:
reachable_only: false
required_acks: 1
compression: gzip
max_message_bytes: 1000000以下是各個參數的詳細解釋：
- hosts（必需）：Kafka broker 的地址和端口列表。在示例中，我們指定了兩個Kafka broker：kafka-broker1:9092 和 kafka-broker2:9092。Filebeat將使用這些地址來連接到Kafka集群。
- topic（必需）：要發送事件到的Kafka主題（topic）的名稱。在示例中，主題名稱為 "my-log-topic"。Filebeat將會將事件發送到這個主題。
- partition.round_robin：事件分區策略的配置。這里的配置是將事件平均分布到所有分區，不僅僅是可達的分區。reachable_only 設置為 false，表示即使分區不可達也會發送數據。如果設置為 true，則只會發送到可達的分區。
- required_acks：Kafka的確認機制。指定要等待的確認數，1 表示只需要得到一個分區的確認就認為消息已經成功發送。更高的值表示更多的確認。通常，1 是常見的設置，因為它具有較低的延遲。
- compression：數據的壓縮方式。在示例中，數據被gzip壓縮。這有助于減小傳輸數據的大小，降低網絡帶寬的使用。
- max_message_bytes：Kafka消息的最大字節數。如果事件的大小超過此限制，Filebeat會將事件拆分為多個消息。
以上是常見的 output.kafka 參數，您可以根據您的Kafka集群配置和需求來調整這些參數。確保配置正確的Kafka主題和分區策略以滿足您的數據傳輸需求。同時，要確保Filebeat服務器可以連接到指定的Kafka broker地址。
以下是一個完整的Filebeat配置文件示例，其中包括了 filebeat.prospectors、processors 和 output.kafka 的配置部分，以用于從日志文件采集數據并將其發送到Kafka消息隊列：
4）filebeat.inputs 與 filebeat.prospectors區別
Filebeat 從 7.x 版本開始引入了新的配置方式 filebeat.inputs，以提供更靈活的輸入配置選項，同時保留了向后兼容性。以下是 filebeat.inputs 和 filebeat.prospectors 之間的主要區別：
- filebeat.inputs：
filebeat.inputs 是較新版本的配置方式，用于定義輸入配置。
允許您以更靈活的方式配置不同類型的輸入。您可以在配置文件中定義多個獨立的輸入塊，每個塊用于配置不同類型的輸入。
每個輸入塊可以包含多個字段，用于定制不同輸入類型的配置，如 type、enabled、paths、multiline 等。
使配置更具可讀性，因為每個輸入類型都有自己的配置塊。
示例：
filebeat.inputs:
- type: log
enabled: true
paths:
- /var/log/app/*.log
- type: syslog
enabled: true
port: 514
protocol.udp: true- filebeat.prospectors：
filebeat.prospectors 是舊版配置方式，用于定義輸入配置。
所有的輸入類型（如日志文件、系統日志、stdin 等）都需要放在同一個部分中。
需要在同一個配置塊中定義不同輸入類型的路徑等細節。
舊版配置方式，不如 filebeat.inputs 配置方式那么靈活和可讀性好。
以下是一些常見的 type 值以及它們的含義：
- log（常用）：用于監視和收集文本日志文件，例如應用程序日志。
- type: log
paths:
- /var/log/*.log- stdin：用于從標準輸入（stdin）收集數據。
- type: stdin- syslog：用于收集系統日志數據，通常是通過UDP或TCP協議從遠程或本地 syslog 服務器接收。
- type: syslog
port: 514
protocol.udp: true- filestream：用于收集 Windows 上的文件日志數據。
- type: filestream
enabled: true- httpjson：用于通過 HTTP 請求從 JSON API 收集數據。
- type: httpjson
enabled: true
urls:
- http://example.com/api/data- tcp 和 udp：用于通過 TCP 或 UDP 協議收集網絡數據。
- type: tcp
enabled: true
host: "localhost"
port: 12345- type: udp
enabled: true
host: "localhost"
port: 12345總的來說，filebeat.inputs 提供了更靈活的方式來配置不同類型的輸入，更容易組織和管理配置。如果您使用的是較新版本的 Filebeat，推薦使用 filebeat.inputs 配置方式。但對于向后兼容性，舊版的 filebeat.prospectors 仍然可以使用。
5）filebeat.yml 配置
filebeat.yml
filebeat.inputs:
- type: log
enabled: true
paths:
- /var/log/*.log
multiline.pattern: '^\['
multiline.negate: false
multiline.match: after
tail_files: true
fields:
app: myapp
env: production
topicname: my-log-topic
- type: log
enabled: true
paths:
- /var/log/messages
multiline.pattern: '^\['
multiline.negate: false
multiline.match: after
tail_files: true
fields:
app: myapp
env: production
topicname: my-log-topic
processors:
- add_fields:
fields:
app: myapp
env: production
- drop_fields:
fields: ["sensitive_data"]
output.kafka:
hosts: ["local-168-182-110:9092"]
#topic: "my-log-topic"
# 這里也可以應用上面filebeat.prospectors.fields的值
topic: '%{[fields][topicname]}'
partition.round_robin:
reachable_only: false
required_acks: 1
compression: gzip
max_message_bytes: 10000006）啟動 Filebeat 服務
nohup ./filebeat -e -c filebeat.yml >/dev/null 2>&1 &
# -e 將啟動信息輸出到屏幕上
# filebeat本身運行的日志默認位置${install_path}/logs/filebeat要修改filebeat的日子路徑，可以添加一下內容在filebeat.yml配置文件：
#logging.level :debug 日志級別
path.logs: /var/log/使用 systemctl 啟動 filebeat
# vi /usr/lib/systemd/system/filebeat.service
[Unit]
Descriptinotallow=filebeat server daemon
Documentatinotallow=/opt/filebeat-7.6.2-linux-x86_64/filebeat -help
Wants=network-online.target
After=network-online.target
[Service]
User=root
Group=root
Envirnotallow="BEAT_CONFIG_OPTS=-c /opt/filebeat-7.6.2-linux-x86_64/filebeat.yml"
ExecStart=/opt/filebeat-7.6.2-linux-x86_64/filebeat $BEAT_CONFIG_OPTS
Restart=always
[Install]
WantedBy=multi-user.target【溫馨提示】記得更換自己的 filebeat 目錄。
systemctl 啟動 filebeat 服務
#刷新一下配置文件
systemctl daemon-reload
# 啟動
systemctl start filebeat
# 查看狀態
systemctl status filebeat
# 查看進程
ps -ef|grep filebeat
# 查看日志
vi logs/filebeat7）檢測日志是否已經采集到 kafka
# 設置環境變量
export KAFKA_HOME=/opt/docker-compose-kafka/images/kafka_2.12-3.4.1
# 查看topic列表
${KAFKA_HOME}/bin/kafka-topics.sh --bootstrap-server local-168-182-110:9092 --list
# 查看topic列表詳情
${KAFKA_HOME}/bin/kafka-topics.sh --bootstrap-server local-168-182-110:9092 --describe
# 指定topic
${KAFKA_HOME}/bin/kafka-topics.sh --bootstrap-server local-168-182-110:9092 --describe --topic my-log-topic
# 查看kafka數據
${KAFKA_HOME}/bin/kafka-console-consumer.sh --topic my-log-topic --bootstrap-server local-168-182-110:9092
#上述命令會連接到指定的Kafka集群并打印my_topic主題上的所有消息。如果要查看特定數量的最新消息，則應將“--from-beginning”添加到命令中。
# 在較高版本的 Kafka 中（例如 Kafka 2.4.x 和更高版本），消費者默認需要明確指定要消費的分區。
#以下是查看特定最新消息數量的示例：
${KAFKA_HOME}/bin/kafka-console-consumer.sh --topic my-log-topic --bootstrap-server local-168-182-110:9092 --from-beginning --max-messages 10 --partition 0
# 查看kafka數據量，在較高版本的 Kafka 中（例如 Kafka 2.4.x 和更高版本），消費者默認需要明確指定要消費的分區。
${KAFKA_HOME}/bin/kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list local-168-182-110:9092 --topic my-log-topic --time -1
# 消費數據查看數據，這里指定一個分區
${KAFKA_HOME}/bin/kafka-console-consumer.sh --bootstrap-server local-168-182-110:9092 --topic my-log-topic --partition 0 --offset 100
# 也可以通過消費組消費，可以不指定分區
${KAFKA_HOME}/bin/kafka-console-consumer.sh --topic my-log-topic --bootstrap-server local-168-182-110:9092 --from-beginning --group my-group這將返回主題 <topic_name> 的分區和偏移量信息，您可以根據這些信息計算出數據量。
