Kafka Server日志无法解析至RSYS及Rsyslog发送JSON日志至Fluentd的格式修正问询

Looks like the issue boils down to two key problems with your Rsyslog setup: it’s tacking on extra syslog metadata (hostname, tag) to your JSON payload, and it’s automatically escaping quotes in the message field. Let’s fix this step by step.

1. Create a Rsyslog Template for Raw JSON #

Your current "elastic" template appends extra fields that break the pure JSON structure. Instead, make a template that sends only the raw log line from Kafka’s log file, with no additions or escaping.

Add this to your rsyslog.conf:

template(name="RawJsonTemplate" type="string" string="%msg:::drop-last-lf%\n")

• %msg:::drop-last-lf% grabs the exact message content (removing the trailing newline)
• No extra metadata means Fluentd gets the unmodified JSON from Kafka
• This template doesn’t escape quotes or special characters

2. Update Rsyslog Forwarding to Use the New Template #

Modify your file monitor and forwarding rules to leverage the raw template. Replace your existing rule block with this:

$InputFilePollInterval 1
$InputFileName /var/log/kafka/server.log
$InputFileTag kafkalogs:
$InputFileStateFile kafkalogs
$InputFileFacility local0
$InputRunFileMonitor

:syslogtag, isequal, "kafkalogs:" {
    :msg, contains, "ERROR" {
        local0.* /var/log/kafkalog_error.log
        # Attach the raw template to forward unmodified JSON
        local0.* @fluentdvmip:5144;RawJsonTemplate
    }
    stop
}

The critical change here is ;RawJsonTemplate at the end of the forwarding line — this tells Rsyslog to use our custom template instead of the default syslog format.

3. Fix Fluentd Tag Matching #

I spotted a mismatch: your Fluentd source uses tag kafkalogs, but your match block is listening for kafka.**. Even if the JSON is fixed, Fluentd won’t process the logs until this is corrected. Update the match block in td-agent.conf:

<match kafkalogs.**>
    @type stdout
</match>

4. Disable Automatic Escape (Optional) #

If you still see escaped quotes after the above changes, add this line to rsyslog.conf to turn off default character escaping:

$EscapeControlCharactersOnReceive off

Final Steps #

Restart both services to apply the changes:

# Restart Rsyslog
systemctl restart rsyslog

# Restart td-agent/Fluentd
systemctl restart td-agent

Now check your td-agent.log — you should see the raw Kafka JSON without extra escapes, and Fluentd will parse it correctly into structured data.

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