如何实现rsyslog向Kafka发送日志？适配Ansible脚本输出场景

我编写了一个Ansible剧本，用于禁用主机的SELinux和ufw，并运行一个bash脚本。该bash脚本接收整数输入，输出指定行数的内容（包含时间戳和32个随机字符）。现在我希望将这些输出发送到UDP端口，再由rsyslog将日志发送至Kafka，但不清楚如何完成最后这一步，恳请帮助。

现有Ansible Playbook #

---
- name: My playbook
  hosts: all
  become: yes

  tasks:
    - name: Disabling SELinux state
      selinux:
        state: disabled

    - name: Stop and disable firewalld.
      service:
         name: ufw
         state: stopped
         enabled: False

    - name: Copy unit file
      ansible.builtin.copy:
        src: /home/AnsiblePlaybooks/files/mytest.service
        dest: /etc/systemd/system/mytest.service
        owner: root
        group: root
        mode: '0644'

    - name: Copy app file    
      template:
        src: /home/ansible-test/AnsiblePlaybooks/app.sh
        dest: /home/app.sh
        owner: root
        group: root
        mode: '0644'

    - name: reload daemons
      ansible.builtin.systemd:
        daemon_reload: yes

    - name: Enable mytest.service
      ansible.builtin.systemd:
        name: mytest.service
        state: started
        enabled: yes

现有Bash脚本 #

#!/usr/bin/env bash

num = {{ input_var }} 
for i in $(seq 1 $num) ; do
    echo =============================
    echo "$i: $(date +%Y-%m-%d-%H:%M:%S) $(openssl rand -hex 16)"
    sleep 0.5
done

步骤1：修改Bash脚本，将输出发送到UDP端口 #

先修正脚本语法错误（变量赋值不能有空格），再添加UDP发送逻辑：

#!/usr/bin/env bash

# 修正变量赋值语法，移除空格
num={{ input_var }}
# 定义UDP目标地址和端口，可按需修改
UDP_HOST="localhost"
UDP_PORT="5140"

for i in $(seq 1 $num); do
    log_line="$i: $(date +%Y-%m-%d-%H:%M:%S) $(openssl rand -hex 16)"
    # 用bash内置的/dev/udp发送日志，无需额外依赖
    echo "$log_line" > /dev/udp/${UDP_HOST}/${UDP_PORT}
    sleep 0.5
done

如果系统不支持/dev/udp，可替换为echo "$log_line" | nc -u $UDP_HOST $UDP_PORT，但需确保已安装netcat。

步骤2：配置rsyslog接收UDP日志并转发到Kafka #

2.1 安装rsyslog Kafka模块

在Ansible中添加任务，根据系统包管理器安装对应模块：

- name: Install rsyslog Kafka module (Debian/Ubuntu)
  apt:
    name: rsyslog-kafka
    state: present
  when: ansible_os_family == 'Debian'

- name: Install rsyslog Kafka module (RHEL/CentOS)
  yum:
    name: rsyslog-kafka
    state: present
  when: ansible_os_family == 'RedHat'

2.2 添加rsyslog转发配置

创建配置文件/etc/rsyslog.d/90-kafka-forward.conf，内容如下：

# 加载UDP输入模块，监听5140端口
module(load="imudp")
input(type="imudp" port="5140")

# 加载Kafka输出模块
module(load="omkafka")

# 定义日志格式，直接使用原始日志内容
template(name="kafkaFormat" type="string" string="%msg%\n")

# 仅转发本地UDP日志到Kafka，替换为你的Kafka集群地址和主题
if $fromhost-ip == '127.0.0.1' then {
    action(
        type="omkafka"
        broker=["kafka-broker-1:9092,kafka-broker-2:9092"]
        topic="my-test-logs"
        template="kafkaFormat"
        queue.type="LinkedList"
        queue.size="10000"
        queue.dequeueBatchSize="100"
        resubmitOnFailure="on"
    )
}

2.3 重启rsyslog服务

在Ansible中添加任务：

- name: Restart rsyslog service
  service:
    name: rsyslog
    state: restarted
    enabled: yes

步骤3：调整Systemd服务配置 #

确保mytest.service能正确运行脚本，示例内容：

[Unit]
Description=My Test Log Generator

[Service]
Type=simple
ExecStart=/home/app.sh
Restart=always  # 若只需运行一次，改为Restart=no
User=root

[Install]
WantedBy=multi-user.target

步骤4：更新Ansible Playbook #

将上述新增任务整合到原Playbook中，同时修正脚本权限（需改为可执行）：

---
- name: My playbook
  hosts: all
  become: yes

  tasks:
    - name: Disabling SELinux state
      selinux:
        state: disabled

    - name: Stop and disable ufw
      service:
         name: ufw
         state: stopped
         enabled: False

    - name: Install rsyslog Kafka module (Debian/Ubuntu)
      apt:
        name: rsyslog-kafka
        state: present
      when: ansible_os_family == 'Debian'

    - name: Install rsyslog Kafka module (RHEL/CentOS)
      yum:
        name: rsyslog-kafka
        state: present
      when: ansible_os_family == 'RedHat'

    - name: Copy rsyslog Kafka forward config
      ansible.builtin.copy:
        src: /home/AnsiblePlaybooks/files/90-kafka-forward.conf
        dest: /etc/rsyslog.d/90-kafka-forward.conf
        owner: root
        group: root
        mode: '0644'

    - name: Restart rsyslog service
      service:
        name: rsyslog
        state: restarted
        enabled: yes

    - name: Copy unit file
      ansible.builtin.copy:
        src: /home/AnsiblePlaybooks/files/mytest.service
        dest: /etc/systemd/system/mytest.service
        owner: root
        group: root
        mode: '0644'

    - name: Copy app file    
      template:
        src: /home/ansible-test/AnsiblePlaybooks/app.sh
        dest: /home/app.sh
        owner: root
        group: root
        mode: '0755'  # 修正为可执行权限

    - name: reload daemons
      ansible.builtin.systemd:
        daemon_reload: yes

    - name: Enable mytest.service
      ansible.builtin.systemd:
        name: mytest.service
        state: started
        enabled: yes

验证方法 #

1. 运行Ansible剧本，确认所有任务执行成功
2. 查看rsyslog日志（/var/log/syslog或/var/log/messages），确认无报错
3. 用Kafka消费者工具验证日志是否送达：

kafka-console-consumer.sh --bootstrap-server <你的Kafka地址> --topic my-test-logs --from-beginning

内容的提问来源于stack exchange，提问作者chicho