CentOS上装个filebeat，弄日志发到kafka去收集一下吧

（根据网络技术博客、Elastic官方文档及Kafka使用指南综合整理）

先在CentOS机器上，确保你是用root用户登录的，或者有sudo权限的普通用户也行，第一步，得把filebeat这个软件包给弄到服务器上，因为CentOS通常用yum来管理软件包，所以最直接的方法就是添加Elastic的官方软件源，打开一个终端窗口，输入命令：sudo rpm --import https://artifacts.elastic.co/GPG-KEY-elasticsearch，这行命令的意思是下载并导入Elastic公司的GPG密钥，用来验证接下来要安装的软件包是不是官方的、有没有被篡改过,这是个安全措施。

需要创建一个新的软件源配置文件，可以用vi或者nano这些文本编辑器来操作，比如输入命令：sudo vi /etc/yum.repos.d/elastic.repo,然后在这个新文件里写入以下几行内容：

[elastic-8.x] name=Elastic repository for 8.x packages baseurl=https://artifacts.elastic.co/packages/8.x/yum gpgcheck=1 gpgkey=https://artifacts.elastic.co/GPG-KEY-elasticsearch enabled=1 autorefresh=1 type=rpm-md

写完之后保存退出，这里面的baseurl指明了软件包从哪里下载，gpgcheck=1就是要求进行GPG验证，enabled=1表示启用这个软件源,这些配置信息在Elastic官网的安装指南里都能找到。

现在软件源配好了，就可以正式安装filebeat了，输入命令：sudo yum install filebeat -y，yum会自动处理依赖关系，下载filebeat的安装包并进行安装，那个-y参数是为了让安装过程自动确认，不用我们再手动输入yes，等命令执行完毕，屏幕上出现Complete!之类的提示,就说明filebeat已经安装到系统里了。

安装完不是马上就能用的，得对它进行配置，告诉它要收集哪些日志、以及收集之后发送到哪里去，filebeat的主要配置文件默认在/etc/filebeat/filebeat.yml，在修改这个文件之前，最好先做个备份，万一配错了还能恢复，可以输入命令：sudo cp /etc/filebeat/filebeat.yml /etc/filebeat/filebeat.yml.bak。

然后用文本编辑器打开这个配置文件：sudo vi /etc/filebeat/filebeat.yml，这个文件内容挺长的，有很多注释行，都是以井号#开头的，我们不需要一下子理解所有内容，重点关注几个关键部分，首先找到关于日志输入的部分，配置项通常叫filebeat.inputs，默认情况下，配置可能是禁用的，或者只配置了系统日志，要收集系统日志（/var/log/messages）和安全日志（/var/log/secure），可以找到类似下面的配置段,把enable改成true：

• type: log enabled: true paths:
  • /var/log/messages
  • /var/log/secure

如果要收集自己应用的日志，比如日志文件在/opt/myapp/logs/app.log，那就再添加一段类似的配置，把路径指过去就行了，paths下面的内容支持通配符，opt/myapp/logs/*.log，这样就会收集该目录下所有以.log结尾的文件。

接下来是最关键的一步，配置输出目的地，也就是Kafka，在配置文件里找到output部分，默认可能是输出到Elasticsearch，那些行都被注释掉了，我们需要找到output.kafka这个配置段，把注释去掉，并填写正确的参数,大概需要配置这些：

output.kafka: hosts: [“kafka-server1:9092", “kafka-server2:9092"] # 这里换成你真实的Kafka服务器地址和端口，如果有多个就用逗号隔开 topic: “centos-logs” # 指定一个主题名字，Kafka上需要提前创建好这个主题 required_acks: 1 # 这个参数关系到消息确认的可靠性，1是个常用值，表示leader副本写入成功就确认 compression: gzip # 可选，启用压缩可以减少网络流量 codec.json: # 可选，将日志事件转换成JSON格式发送，通常更易于处理 pretty: false

这里面的hosts一定要填对，不然数据发不出去，topic的名字可以自己定，但必须和Kafka上已有的主题一致，关于Kafka集群地址和主题创建的信息,得参考你的Kafka运维文档或者问相关负责的同学。

配置保存之后，先别急着启动，最好检查一下配置文件语法对不对，可以运行：sudo filebeat test config -c /etc/filebeat/filebeat.yml，如果输出显示Config OK,那就说明配置文件没有语法错误。

检查通过后，就可以启动filebeat服务了，执行命令：sudo systemctl daemon-reload （重新加载系统服务配置），然后执行sudo systemctl enable filebeat （设置开机自启），最后执行sudo systemctl start filebeat （启动服务）。

怎么知道它有没有成功运行呢？可以看服务状态：sudo systemctl status filebeat，如果看到active (running)的字样，说明服务启动成功了，还可以查看filebeat自己的日志来排错：tail -f /var/log/filebeat/filebeat，如果看到类似“Kafka connection established”或者成功发送一批事件（publish events）的日志,那通常就说明数据已经在往Kafka发了。

还有一个方法可以快速测试一下，就是手动生成一些日志条目，比如用logger命令：logger -t test “This is a test message from CentOS”，然后去看看Kafka的消费者那边能不能收到这条消息，到Kafka服务器上，用kafka-console-consumer.sh脚本来监听那个主题，命令大概长这样：bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic centos-logs --from-beginning，如果配置都正确,应该能在终端上看到刚刚发出去的那条测试日志。

如果遇到问题，排查思路一般是几步：首先确认网络通不通，CentOS服务器能不能ping通Kafka的地址和端口，可以用telnet kafka-server 9092试试，其次检查Kafka主题是不是真的存在，权限够不够，再看filebeat的日志文件，里面通常会有比较详细的错误信息，还有可能遇到版本兼容性问题,确保filebeat版本和Kafka集群版本不要太老。

（以上步骤综合了Linux系统管理常规操作、Elastic官网Filebeat安装与配置文档、以及Apache Kafka生产者客户端配置的常见实践）