Elasticsearch+Fluentd+Kafka搭建日誌系統

前言

因爲logstash內存佔用較大,靈活性相對沒那麼好,ELK正在被EFK逐步替代.其中本文所講的EFK是Elasticsearch+Fluentd+Kfka,實際上K應該是Kibana用於日誌的展現,這一塊不作演示,本文只講述數據的採集流程.

前提

1. docker
2. docker-compose
3. apache kafka服務

架構

數據採集流程

數據的產生使用cadvisor採集容器的監控數據並將數據傳輸到Kafka.

數據的傳輸鏈路是這樣: Cadvisor->Kafka->Fluentd->elasticsearch

每個服務均可以橫向擴展,添加服務到日誌系統中.

配置文件

docker-compose.yml

version: "3.7"

services:
  
  elasticsearch:
   image: elasticsearch:7.5.1
   environment:
    - discovery.type=single-node  #使用單機模式啓動
   ports:
    - 9200:9200

  cadvisor:
    image: google/cadvisor
    command: -storage_driver=kafka -storage_driver_kafka_broker_list=192.168.1.60:9092(kafka服務IP:PORT) -storage_driver_kafka_topic=kafeidou
    depends_on:
      - elasticsearch

  fluentd:
   image: lypgcs/fluentd-es-kafka:v1.3.2
   volumes:
    - ./:/etc/fluent
    - /var/log/fluentd:/var/log/fluentd

其中:

1. cadvisor產生的數據會傳輸到192.168.1.60這臺機器的kafka服務,topic爲kafeidou
2. elasticsearch指定爲單機模式啓動(discovery.type=single-node環境變量),單機模式啓動是爲了方便實驗總體效果

fluent.conf

#<source>
#  type http
#  port 8888
#</source>

<source>
  @type kafka
  brokers 192.168.1.60:9092
  format json
  <topic>
    topic     kafeidou
  </topic>
</source>

<match **>
  @type copy

#  <store>
#   @type stdout
#  </store>

  <store>
  @type elasticsearch
  host 192.168.1.60
  port 9200
  logstash_format true
  #target_index_key machine_name
  logstash_prefix kafeidou
  logstash_dateformat %Y.%m.%d   
  
  flush_interval 10s
  </store>
</match>

其中:

1. type爲copy的插件是爲了可以將fluentd接收到的數據複製一份,是爲了方便調試,將數據打印在控制檯或者存儲到文件中,這個配置文件默認關閉了,只提供必要的es輸出插件.

須要時能夠將@type stdout這一塊打開,調試是否接收到數據.

1. 輸入源也配置了一個http的輸入配置,默認關閉,也是用於調試,往fluentd放入數據.

能夠在linux上執行下面這條命令:

curl -i -X POST -d 'json={"action":"write","user":"kafeidou"}' http://localhost:8888/mytag

1. target_index_key參數,這個參數是將數據中的某個字段對應的值做爲es的索引,例如這個配置文件用的是machine_name這個字段內的值做爲es的索引.

開始部署

在包含docker-compose.yml文件和fluent.conf文件的目錄下執行:
``
docker-compose up -d
``

在查看全部容器都正常工做以後能夠查看一下elasticsearch是否生成了預期中的數據做爲驗證,這裏使用查看es的索引是否有生成以及數據數量來驗證:

-bash: -: 未找到命令
[root@master kafka]# curl http://192.168.1.60:9200/_cat/indices?v
health status index                                uuid                   pri rep docs.count docs.deleted store.size pri.store.size
yellow open   55a4a25feff6                         Fz_5v3suRSasX_Olsp-4tA   1   1       1            0      4kb            4kb

也能夠直接在瀏覽器輸入 http://192.168.1.60:9200/_cat/indices?v查看結果,會更方便.

能夠看到我這裏是用了machine_name這個字段做爲索引值,查詢的結果是生成了一個叫55a4a25feff6的索引數據,生成了1條數據(docs.count)

到目前爲止kafka->fluentd->es這樣一個日誌收集流程就搭建完成了.

固然了,架構不是固定的.也能夠使用fluentd->kafka->es這樣的方式進行收集數據.這裏不作演示了,無非是修改一下fluentd.conf配置文件,將es和kafka相關的配置作一下對應的位置調換就能夠了.

鼓勵多看官方文檔,在github或fluentd官網上均可以查找到fluentd-es插件和fluentd-kafka插件.

始發於 四顆咖啡豆 ,轉載請聲明出處.
關注公衆號->[四顆咖啡豆] 獲取最新內容