寫在開篇

本文只分享各個鏈路環節的配置對接，關於環境的搭建，比如kafka集羣、es集羣的搭建等請自行完成。還有，業務應用的日誌可以是你的其他業務日誌，希望本文可以起到拋磚引用的效果。

測試架構

這個架構描述了一個將來自不同數據源的數據通過 Kafka 中轉，然後使用 Logstash 將數據從 Kafka 中讀取並處理，最終將處理後的數據再寫回到 Kafka 中，以供 Elasticsearch 進行存儲和分析的過程。通過使用 Kafka 和 Logstash，可以將來自不同數據源的數據進行集中管理和處理，並將數據以可靠的方式發送到 Elasticsearch 進行存儲和分析。這種架構具有高可用性和可伸縮性，並且可以在處理大量數據時提供良好的性能。同時，由於 Logstash 可以從多種來源讀取數據，因此可以適應各種數據集成方案的需求。

注意：kafka集羣a接收來自filebeat的消息，並由logstash進行消費。kafka集羣b接收來自logstash的消息，並由es或者其他業務進行消費。

機器規劃

注意：在同一個網段中，兩個 Kafka 集羣必須要使用不同的 broker.id，否則會導致衝突。因為 broker.id 是 Kafka 集羣中唯一標識一個 Broker 的參數，同一個網段中不能存在兩個具有相同 broker.id 的 Broker。

實戰開擼

創建kafka主題

在kafka集羣a中創建主題

bin/kafka-topics.sh --create --zookeeper 192.168.11.247:2181 --replication-factor 2 --partitions 3 --topic wordpress-nginx-log

安裝和配置filebeat

在安裝了wordpress web的業務主機上進行安裝

1. 下載和安裝

wget https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-8.6.2-x86_64.rpm
rpm -ivh filebeat-8.6.2-x86_64.rpm 
filebeat version

2. 配置filebeat讀取日誌 打開和編輯/etc/filebeat/filebeat.yml配置文件，添加以下內容，使其讀取Nginx訪問日誌文件：

filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /usr/local/nginx/logs/wordpress.access.log
  fields:
    log_type: wordpress_access

output.kafka:
  hosts: ["192.168.11.247:9092",192.168.11.248:9092,192.168.11.249:9092]
  topic: "wordpress-nginx-log"

3. 啓動filebeat

systemctl start filebeat

配置logstash01，消費kafka集羣a的消息

在logstash01主機上配置logstash，使其能夠消費kafka集羣a中主題為"wordpress-nginx-log"的消息。

1. 安裝kafka插件

bin/logstash-plugin install logstash-input-kafka

注意：如果安裝的時候提示：ERROR: Installation aborted, plugin 'logstash-input-kafka' is already provided by 'logstash-integration-kafka'，這個錯誤提示'logstash-input-kafka'插件已經被Logstash集成插件'logstash-integration-kafka'提供了，可以直接使用 logstash-integration-kafka 插件消費 Kafka 消息

2. 在 Logstash 的配置文件中使用 kafka 輸入插件

配置之前，先説明下我的nginx日誌自定義的格式：

log_format my_log_format '$remote_addr - $remote_user [$time_local] "$request" '
                              '$status $body_bytes_sent "$http_referer" '
                              '"$http_user_agent" "$http_x_forwarded_for"';

創建和編輯配置文件config/wordpress-logstash.conf，添加以下內容：

input {
  kafka {
    bootstrap_servers => "192.168.11.247:9092,192.168.11.248:9092,192.168.11.249:9092"
    topics => ["wordpress-nginx-log"]
  }
}

filter {
  # 解析Nginx日誌行
  grok {
    match => { "message" => '%{IPORHOST:clientip} - %{USERNAME:remote_user} [%{HTTPDATE:timestamp}] "%{WORD:method} %{URIPATHPARAM:request} HTTP/%{NUMBER:httpversion}" %{NUMBER:status} %{NUMBER:body_bytes_sent} "%{DATA:http_referer}" "%{DATA:user_agent}"' }
  }
  # 將時間戳轉換為ISO 8601格式
  date {
    match => [ "timestamp", "dd/MMM/yyyy:HH:mm:ss Z" ]
    target => "@timestamp"
  }
}

output {
  stdout { codec => rubydebug }
}

3. 啓動 Logstash

bin/logstash -f config/wordpress-logstash.conf 

在這裏，先讓其輸出到屏幕，看看是否正常。確保logstash消費kafka集羣a的消息沒問題、以及確保過濾沒問題可以正常打印到屏幕之後，就可以繼續下面的步驟了。

配置logstash01，過濾後的消息寫入到kafka集羣b

繼續在logstash01上配置，從kafka集羣a中消費數據並過濾，處理後寫入到kafka集羣b中的主題wordpress-web-log。

1. 在kafka集羣b中創建主題

bin/kafka-topics.sh --create --zookeeper 192.168.11.40:2181 --replication-factor 2 --partitions 3 --topic wordpress-web-log

2. 編輯配置文件config/wordpress-logstash.conf，配置output

input {
  kafka {
    bootstrap_servers => "192.168.11.247:9092,192.168.11.248:9092,192.168.11.249:9092"
    topics => ["wordpress-nginx-log"]
  }
}

filter {
  # 解析Nginx日誌行
  grok {
    match => { "message" => '%{IPORHOST:clientip} - %{USERNAME:remote_user} [%{HTTPDATE:timestamp}] "%{WORD:method} %{URIPATHPARAM:request} HTTP/%{NUMBER:httpversion}" %{NUMBER:status} %{NUMBER:body_bytes_sent} "%{DATA:http_referer}" "%{DATA:user_agent}"' }
  }
  # 將時間戳轉換為ISO 8601格式
  date {
    match => [ "timestamp", "dd/MMM/yyyy:HH:mm:ss Z" ]
    target => "@timestamp"
  }
}

output {
  kafka {
    bootstrap_servers => "192.168.11.40:9092,192.168.11.41:9092,192.168.11.42:9092"
    topic_id => "wordpress-web-log"
  }
}

編輯完成後，記得重啓logstash哦。

3. 臨時啓動一個消費者，驗證從kafka集羣b消費主題wordpress-web-log的消息

bin/kafka-console-consumer.sh --bootstrap-server 192.168.11.40:9092 --topic wordpress-web-log

如果能正常消費，讀取到的消息打印到控制枱上，就可以繼續下面的步驟了。

配置logstash02，消費kafka集羣a的消息

在logstash02主機上配置logstash，使其能夠消費kafka集羣b中主題為"wordpress-web-log"的消息，並寫入到ES集羣

打開並編輯config/logstash.conf，添加以下內容：

input {
  kafka {
    bootstrap_servers => "192.168.11.40:9092,192.168.11.41:9092,192.168.11.42:9092"
    topics => ["wordpress-web-log"]
  }
}

output {
  elasticsearch {
    hosts => ["http://192.168.11.243:9200","http://192.168.11.244:9200","http://192.168.11.245:9200"]
    index => "wordpress-web-log-%{+YYYY.MM.dd}"
  }
}

寫在最後

所有環節對接完畢，看看最終成果。

關於如何將logstash部署到K8S，感興趣？請保持高度關注，有空了再分享。

本文轉載於WX公眾號：不背鍋運維（喜歡的盆友關注我們）：https://mp.weixin.qq.com/s/5uGpFBEg67yxRCNxLpyUmw