Syslog parsing strategy

[dspruell]

• fluentd 1.14.3
• ruby 2.6.9p207

I'm running the following configuration: Fluentd is deployed as a syslog receiver for multiple client systems that have configured their syslogd to simply forward all messages to a in_syslog input. Logs are ingested and forwarded to a log collection system.

*.*     @fluent-log.example.net:5140

Now I wish to begin to parse a number of these log types in Fluentd prior to forwarding them. A small portion of the logs will be parsed to fields (filter, with parser of type regexp) and then forwarded, and all of the rest of the logs that aren't parsed should simply be forwarded.

What is the correct pattern for this? If I try implementing a filter section in the pipeline I can successfully parse and forward the logs I explicitly parse, but all other logs are dropped with a ParserError:

2022-02-25 00:43:58 -0700 [warn]: #0 dump an error event: error_class=Fluent::Plugin::Parser::ParserError error="pattern not matched with data 'my redacted message data that fails to parse'" location=nil tag="test.etc.fluent.testing.parsing.txt" time=2022-02-25 00:43:58.229254979 -0700 record={"message"=>"my redacted message data that fails to parse"}

I can isolate the syslog events I want to parse by matching on their tag (syslog.daemon.info) and ident (unbound) values. But I can't figure out how to only apply a parser to those logs and allow all other logs to flow uninterrupted.

[dspruell]

I prototyped the below, based on out_rewrite_tag_filter:

<source>
    @type syslog
    @label @SYSLOG
    tag syslog
    port 5140
    bind 0.0.0.0
</source>

<label @SYSLOG>
    <filter **>
        @type record_transformer
        <record>
            host "#{Socket.gethostname}"
            tag ${tag}
        </record>
    </filter>

    <match syslog.**>
        @type rewrite_tag_filter

        # Unbound query logs
        <rule>
            key message
            # "[62425:0] info: x.x.1.46 www.example.com. AAAA IN"
            pattern /info: \S+ \S+\. \S+ IN$/
            tag unbound.query
        </rule>

        # Unbound RPZ logs
        <rule>
            key message
            # [8113:0] info: RPZ applied [rpz-stevenblack] api.mixpanel.com. nxdomain x.x.7.26@1565 api.mixpanel.com. A IN
            pattern /info: RPZ applied/
            tag unbound.rpz
        </rule>

        # Fallthrough
        <rule>
            key message
            pattern /.+/
            tag passthru
        </rule>
    </match>

    # Unbound query logs
    <filter unbound.query>
        @type parser
        key_name message
        <parse>
            @type regexp
            expression /(?<severity>info): (?<src_ip>\S+) (?<query>\S+)\. (?<query_type>\S+) (?<query_class>IN)$/
        </parse>
        reserve_data true
    </filter>

    # Unbound RPZ logs
    <filter unbound.rpz>
        @type parser
        key_name message
        <parse>
            @type regexp
            expression /(?<severity>info): (?<action>RPZ applied) \[(?<rpz_source>[^\]]+)\] (?<query>\S+)\. (?<reply_code>\S+) (?<src_ip>\S+)@(?<src_port>\d+) \S+ (?<query_type>\S+) (?<query_class>IN)$/
        </parse>
        reserve_data true
    </filter>

    <match **>
        @type stdout
    </match>
</label>

This meets the requirements as it allows specific events to be parsed while others fall through, and all are picked up by the output. It doesn't preserve the original tags (for syslog input, the facility and priority) and I imagine there are ways this can be improved.

[fujimotos]

I prototyped the below, based on out_rewrite_tag_filter:

Yes, this is the common way to do this. You use rewrite_tagfilter to
"split" the incoming data stream, and process branches separately.

This meets the requirements as it allows specific events to be parsed while others fall through, and all are picked up by the output. It doesn't preserve the original tags (for syslog input, the facility and priority) and I imagine there are ways this can be improved.

Another approach is to use out_copy and out_relabel to clone the
stream and process cloned streams. Basically like this:

<match **>
  @type copy
  <store>
     @type relabel
     @label @filtered
  </store>
  <store>
     @type relabel
     @label @original
  </store>
</match>

<label @filtered>
...
</label>

<label @original>
...
</label>