Rsyslog负责高效采集与可靠传输系统日志，Logstash专注解析、清洗和结构化；配置Rsyslog启用TCP输入并转发RFC5424或JSON格式日志至Logstash，Logstash通过tcp插件接收并用dissect/date等过滤器处理，推荐Rsyslog端直接生成JSON以降低Logstash负载。
在Linux中，用Rsyslog将日志转发给Logstash，再由Logstash完成解析、清洗和结构化，是一套成熟且轻量的方案。核心思路是：Rsyslog负责高效采集与可靠传输（尤其适合系统日志），Logstash专注字段提取、条件过滤、格式转换等逻辑处理。
一、配置Rsyslog：启用TCP接收并转发日志
Rsyslog默认只处理本地日志，需开启网络输入，并将日志以标准格式（如RFC5424）发往Logstash监听端口。
- 编辑/etc/rsyslog.conf或新建/etc/rsyslog.d/50-logstash.conf，添加以下内容：
# 启用TCP输入模块
module(load="imtcp")
input(type="imtcp" port="5140" ruleset="to_logstash")
<h1>定义转发规则集</h1><p>ruleset(name="to_logstash") {
action(
type="omfwd"
protocol="tcp"
target="127.0.0.1"
port="5044"
template="RSYSLOG_SyslogProtocol23Format"
queue.type="LinkedList"
queue.size="10000"
queue.dequeuebatchsize="100"
action.resumeRetryCount="-1"
)
}- 重启服务：sudo systemctl restart rsyslog
- 确认5140端口已监听：ss -tlnp | grep 5140
二、配置Logstash：接收、解析与结构化日志
Logstash通过beats或tcp插件接收Rsyslog发来的日志，再用dissect、grok或kv等过滤器提取字段。
- 创建配置文件/etc/logstash/conf.d/rsyslog-to-es.conf：
input {
tcp {
port => 5044
codec => "json" # 若Rsyslog发送的是JSON（推荐），否则用plain并配合filter解析
}
}
<p>filter {</p><h1>若Rsyslog未转成JSON，可用dissect粗粒度拆分（比grok快）</h1><p>dissect {
mapping => { "message" => "%{timestamp} %{hostname} %{program}[%{pid}]: %{logmsg}" }
}</p><h1>清洗常见噪声（如多余空格、控制字符）</h1><p>mutate {
gsub => ["logmsg", "\x00", ""]
strip => ["logmsg", "hostname", "program"]
}</p><h1>结构化时间字段（Rsyslog时间格式示例：May 23 14:22:01）</h1><p>date {
match => ["timestamp", "MMM dd HH:mm:ss"]
target => "@timestamp"
}
}</p><p>output {
elasticsearch {
hosts => ["<a href="https://www.php.cn/link/fb7850115a917d3ab720269da3e667de">https://www.php.cn/link/fb7850115a917d3ab720269da3e667de</a>"]
index => "syslog-%{+YYYY.MM.dd}"
}
stdout { codec => rubydebug } # 调试时启用
}- 确保Logstash有权限读取配置、网络可达，并启动：sudo systemctl start logstash
- 验证是否收到数据：tail -f /var/log/logstash/logstash-plain.log
三、让Rsyslog直接发送结构化JSON（推荐进阶方式）
避免Logstash做繁重正则解析，可让Rsyslog在转发前就生成JSON，大幅提升吞吐与稳定性。
- 在/etc/rsyslog.d/50-logstash.conf中启用mmjsonparse模块，并定义JSON模板：
module(load="mmjsonparse")
template(name="json-template" type="list") {
constant(value="{")
constant(value="\"@timestamp\":\"") property(name="timereported" dateFormat="rfc3339")
constant(value="\",\"host\":\"") property(name="hostname")
constant(value="\",\"program\":\"") property(name="programname")
constant(value="\",\"pid\":\"") property(name="procid")
constant(value="\",\"message\":\"") property(name="msg" format="json")
constant(value="\"}")
}
<p>action(
type="omfwd"
protocol="tcp"
target="127.0.0.1"
port="5044"
template="json-template"
)- 此时Logstash input可直接用codec => "json"，filter中几乎无需grok，字段天然可用
- 注意：需确保Rsyslog版本 ≥ 8.16.0，且启用imjournal或imuxsock获取完整字段
四、验证与调优要点
实际部署后需关注连通性、字段完整性与性能瓶颈。
- 连通性检查：在Logstash服务器上运行nc -lvp 5044，再手动向Rsyslog发日志（logger "test message"），看是否收到
- 字段验证：启用Logstash的stdout { codec => rubydebug }，观察@timestamp、host、message等是否正确提取
- 性能调优：Rsyslog端开启队列（如示例中的queue.type="LinkedList"），Logstash端适当增加pipeline.workers和pipeline.batch.size
- 安全加固：生产环境建议用TLS加密TCP传输（Rsyslog + Logstash均支持TLS），或限制防火墙仅允许内网Rsyslog IP访问5044端口
