rsyslog

1.更新rsyslog服务

因为系统默认的版本比较低，对功能支持比较有限，需要升级rsyslog到8.x版本。

# 下载官方yum源
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.创建收集配置文件

因为rsyslog在系统上还有其他很多种作用，在主配置文件中，会自动加载/etc/rsyslog.d/*.conf配置文件。

# 创建一个空的配置文件
touch afka.conf

配置文件内容如下：

# 加载两个模块
module(load="omkafka")
module(load="imfile")

# ruleset
ruleset(name="nginx-kafka") {
    #日志转发kafka
    action (
        type="omkafka"      # 要使用的模块
        # template="nginxAccessTemplate"    # 模板，这里没有，就注释掉
        confParam=["compression.codec=snappy", "queue.buffering.max.messages=400000"]
        # partitions.number="4" # 分区，在kafka为集群时会用到
        topic="test_nginx"  # 使用的topic
        broker="172.30.90.9:9092"   # kafka主机
        # 以下为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"
    )
}

# 定义消息来源及设置相关的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.启动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

启动

docker-compose up -d

5.部署logstash

input {
  kafka {
    bootstrap_servers => ["kafka:9092"] # kafka地址
    auto_offset_reset => "latest"   # 读取方式，只读取最新值
    consumer_threads => 1   # 线程数
    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
  }
}