Skip to content

Latest commit

 

History

History
147 lines (106 loc) · 5.29 KB

通过logstash给kafka推送日志（适于Redhat&Centos）.md

File metadata and controls

147 lines (106 loc) · 5.29 KB
Raw

通过logstash给kafka推送日志（适于Redhat/Centos）

说明：

1、ATD支持多种给kafka推送日志的方式

针对Redhat、Centos支持如下方式（按推荐顺序排列）：
（1）通过kafkacat给kafka推送日志
（2）通过filebeat给kafka推送日志
（3）通过rsyslog给kafka推送日志
（4）通过logstash给kafka推送日志

针对Ubuntu支持如下方式：
（1）通过filebeat给kafka推送日志

针对Windows系统支持如下方式：
（1）通过nxlog给kafka推送日志

2、注意：您需要将日志推送到kafka相应的TopicName中，但是，请勿将不同格式的域名日志推到同一个TopicName下，否则ATD将无法完成日志解析。

以下是通过logstash给kafka推送日志的操作文档（适于Redhat/Centos）：

1.下载logstash-5.6:

  • 配置yum源，编辑/etc/yum.repos.d/bsc-juhe.repo文件：

[bsc-juhe]

name=juhe

baseurl=http://bsc-juhe:[email protected]/repo/bsc/el$releasever/$basearch/

gpgcheck=0

enabled=1

  • 编辑完成后清除一下缓存:

      yum clean all

  • 安装:

      yum install logstash-5.6.4-1.noarch

2.绑定kafka-server机器的hostname到ip的映射：

  • 绑定kafka server,对应的主机名和ip如下，那么就需要在日志服务器上绑定这些hosts，在/etc/hosts文件中追加：

      vim /etc/hosts

{{ip}} {{hostname}}

  • 例：

      vim /etc/hosts

172.18.1.2 ATD1
172.18.1.3 ATD2
172.18.1.4 ATD3

3.编辑logstash的配置文件:

  • 编辑文件/etc/logstash/conf.d/file-kafka.conf :
input {
    file {
        path => "/tmp/*.log"
        discover_interval => 1
        start_position => "end"
        sincedb_path => "/dev/null"
    }
}

output{
    kafka {
            codec => plain {
                format => "%{message}"
            }
            bootstrap_servers => "kafka1:6667"
            topic_id => "juhe-test"
            compression_type => "snappy"
    }
}

  • 注意：
    1）paths为本地文件的路径，支持多个文件的导入，支持通配符，支持文件句柄释放后新的同名文件的句柄引入。
    2）bootstrap_servers为kafka-server的地址，格式为”ATD机器hostname:kafk-server的端口号”。
    3）以上两个参数都可以设置多个，例：

      path => [“/tmp/*.log”, ”/tmp/2.log.b”]
  bootstrap_servers => [“kafka1:6667”, “kafka2:6667”, “kafka3:6667”]

  • 完成logstash配置文件的编辑操作后，启动logstash：

      /usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/file-kafka.conf

  • 如果因为日志量过大而出现日志积压等情况，可以给logstash加pipeline.workers和 pipeline.batch.size，参数为在命令行中添加”-w”及”-b”，具体参考 /usr/share/logstash/bin/logstash --help 输出的结果。

4、常见的问题及解决办法：

（1）确认是否有新产生的日志进入到kafka中： 登陆到ATD部署机器192.168.0.89，消费对应kafka的topic数据，如果日志源有新日志产生且推送日志流程正常，使用如下命令能看到日志：

# /usr/hdp/2.6.2.0-205/kafka/bin/kafka-console-consumer.sh --bootstrap-server $(hostname):6667 --topic juhe-1710116uSh

（2）如果（1）步骤中没有消费到日志，则自查如下：

查看推送日志的机器到kafka机器的网络是否连通：
# telnet 192.168.0.89 6667

（3）supervisor启动失败： 在/etc/supervisord.conf文件中查看是否有以下内容，如果没有，请添加。

[include]
files = supervisord.d/*.ini

快速链接

1、官方参考文档：
https://www.elastic.co/guide/en/logstash/5.6/configuration.html

2、安装拦截器文档：
（1）安装拦截器文档（适于Redhat/Centos）：https://github.com/yunjuhe/atd/blob/master/%E5%AE%89%E8%A3%85%E6%8B%A6%E6%88%AA%E5%99%A8%E6%96%87%E6%A1%A3%EF%BC%88%E9%80%82%E4%BA%8ERedhat%26Centos).md
（2）安装拦截器文档（适于Ubuntu）：https://github.com/yunjuhe/atd/blob/master/%E5%AE%89%E8%A3%85%E6%8B%A6%E6%88%AA%E5%99%A8%E6%96%87%E6%A1%A3%EF%BC%88%E9%80%82%E4%BA%8EUbuntu).md