rsyslog

1.更新rsyslog服務(wù)

因?yàn)橄到y(tǒng)默認(rèn)的版本比較低，對(duì)功能支持比較有限，需要升級(jí)rsyslog到8.x版本呻疹。

# 下載官方y(tǒng)um源
wget -O /etc/yum.repos.d/rsyslog.repo http://rpms.adiscon.com/v8-stable/rsyslog.repo
# 安裝rsyslog
yum install rsyslog -y
# 安裝kafka插件出革，安裝此插件以后
yum -y install rsyslog-kafka

2.創(chuàng)建收集配置文件

因?yàn)閞syslog在系統(tǒng)上還有其他很多種作用柏靶，在主配置文件中冗美，會(huì)自動(dòng)加載/etc/rsyslog.d/*.conf配置文件兽赁。

# 創(chuàng)建一個(gè)空的配置文件
touch afka.conf

配置文件內(nèi)容如下：

# 加載兩個(gè)模塊
module(load="omkafka")
module(load="imfile")

# ruleset
ruleset(name="nginx-kafka") {
    #日志轉(zhuǎn)發(fā)kafka
    action (
        type="omkafka"      # 要使用的模塊
        # template="nginxAccessTemplate"    # 模板袁余，這里沒有擎勘，就注釋掉
        confParam=["compression.codec=snappy", "queue.buffering.max.messages=400000"]
        # partitions.number="4" # 分區(qū)，在kafka為集群時(shí)會(huì)用到
        topic="test_nginx"  # 使用的topic
        broker="172.30.90.9:9092"   # kafka主機(jī)
        # 以下為kafka收集常用配置颖榜，不用修改
        queue.spoolDirectory="/tmp" 
        queue.filename="test_nginx_kafka"
        queue.size="360000"
        queue.maxdiskspace="2G"
        queue.highwatermark="216000"
        queue.discardmark="350000"
        queue.type="LinkedList"
        queue.dequeuebatchsize="4096"
        queue.timeoutenqueue="0"
        queue.maxfilesize="10M"
        queue.saveonshutdown="on"
        queue.workerThreads="4"
    )
}

# 定義消息來源及設(shè)置相關(guān)的action（一下都為必須字段）
input(type="imfile" tag="nginx" File="/var/log/nginx/access.log" Ruleset="nginx-kafka")

3.重啟rsyslog

# /etc/init.d/rsyslog restart
Shutting down system logger:                               [  OK  ]
Starting system logger:                                    [  OK  ]

4.啟動(dòng)kafka

docker-compose.cfg文件

version: '2'
services:
  zk:
    image: zookeeper:3.3.6
    ports:
      - '2181:2181'
  kafka:
    image: wurstmeister/kafka
    ports:
      - "9092:9092"
    environment:
      KAFKA_ADVERTISED_HOST_NAME: 172.30.90.9
      # zookeeper地址
      KAFKA_ZOOKEEPER_CONNECT: zk:2181
    volumes:
      - /var/run/docker.sock:/var/run/docker.sock

啟動(dòng)

docker-compose up -d

5.部署logstash

input {
  kafka {
    bootstrap_servers => ["kafka:9092"] # kafka地址
    auto_offset_reset => "latest"   # 讀取方式棚饵，只讀取最新值
    consumer_threads => 1   # 線程數(shù)
    topics => ["test_nginx"]    # topic 地址
    type => "web"
  }
}
output {
  elasticsearch {
    hosts => ["192.168.2.194:9200", "192.168.2.192:9200"]
    index => "%{type}-%{+YYYY.MM.dd}"
    timeout => 300
  }
}