parser_syslog: accept a space after RFC3164 priority (#5449)

<!--
Thank you for contributing to Fluentd!
Your commits need to follow DCO: https://probot.github.io/apps/dco/
And please provide the following information to help us make the most of
your pull request:
-->

**Which issue(s) this PR fixes**: 
Fixes #4158

**What this PR does / why we need it**: 

Some syslog senders emit RFC3164 records with an ASCII space between the
priority and header. Depending on whether `in_syslog` or the parser owns
priority extraction, the RFC3164 parser sees that space either
immediately after `>` or as the first character of its input, and
currently attempts to parse it as the start of the timestamp.

This change accepts the RFC3164 boundary space in `rfc3164` and the
RFC3164 path of `auto`, across both parser engines and both
priority-ownership models. Canonical records without the space continue
to parse unchanged, malformed priorities remain invalid, and RFC5424
parsing is unchanged.

For the default RFC3164 time format, regression coverage verifies that
two boundary spaces and tabs remain invalid. An explicit RFC3164
`time_format` beginning with a space treats that leading space as
timestamp syntax, so this change does not impose or claim a one-space
maximum for that configuration.

This also corrects the string parser's dynamic timestamp slice cursor
and the field-count accounting needed by RFC3164 time formats that begin
with a space. Truncated subsecond records with the boundary space are
rejected through the normal `[nil, nil]` path instead of raising while
looking for the missing field delimiter.

**Docs Changes**:

None. This broadens input compatibility without adding or changing
configuration.

**Release Note**: 

parser_syslog: accept one optional ASCII space between RFC3164 priority
and header.

**Testing**:

- `bundle exec ruby -Itest test/plugin/test_parser_syslog.rb --name
test_truncated_subsecond_rfc3164_is_rejected_without_raising` — 4 tests,
12 assertions, 0 failures, 0 errors across `rfc3164`/`auto` and
parser-owned/input-owned priority
- `bundle exec ruby -Itest test/plugin/test_parser_syslog.rb` — 132
tests, 521 assertions, 0 failures, 0 errors
- `bundle exec ruby -Itest test/plugin/test_in_syslog.rb` — 41 tests,
280 assertions, 0 failures, 0 errors
- `bundle exec rake test` — 4,401 tests, 16,154 assertions, 0 failures,
0 errors, 3 pendings, 36 omissions
- `gem exec rubocop` — 460 files inspected, 0 offenses
- Balanced microbenchmark with 8 samples per revision and case — all 8
prespecified fixed regression gates passed. The narrowest throughput
ratio was 0.953; this was used only as a regression gate, and no
performance improvement is claimed.
- Installed-gem UDP validation — 20/20 scenarios passed. Both baseline
and candidate gems were built and installed into isolated environments,
real `fluentd --no-supervisor` processes were started, and packets were
sent by external UDP clients. The original 16-scenario matrix covered
both parser engines, both priority-ownership models, `rfc3164`/`auto`,
canonical and spaced packets, whitespace rejection, RFC5424
compatibility, and explicit leading-space time formats. Four additional
string-parser scenarios verified that the truncated subsecond packet
takes the normal unmatched path without an exception and that a valid
`.5` sibling is parsed by the same live process across `rfc3164`/`auto`
and both priority-ownership modes.

---------

Signed-off-by: Akash Kumar <meakash7902@gmail.com>
Signed-off-by: Shizuo Fujita <fujita@clear-code.com>
Co-authored-by: Shizuo Fujita <fujita@clear-code.com>
Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>

Author: 3 people

lib/fluent/plugin/parser_syslog.rb

@@ -71,6 +71,7 @@ def initialize
         @time_parser_rfc5424 = nil
         @space_count_rfc3164 = nil
         @space_count_rfc5424 = nil
+        @time_format_starts_with_space_rfc3164 = false
         @skip_space_count_rfc3164 = false
         @skip_space_count_rfc5424 = false
         @time_parser_rfc5424_without_subseconds = nil
@@ -123,10 +124,12 @@ class << self
 
       def setup_time_parser_3164(time_fmt)
         @time_parser_rfc3164 = time_parser_create(format: time_fmt)
+        @time_format_starts_with_space_rfc3164 = time_fmt.start_with?(SPLIT_CHAR)
         if ['%b %d %H:%M:%S', '%b %d %H:%M:%S.%N'].include?(time_fmt)
           @skip_space_count_rfc3164 = true
         end
         @space_count_rfc3164 = time_fmt.squeeze(' ').count(' ') + 1
+        @space_count_rfc3164 -= 1 if @time_format_starts_with_space_rfc3164
       end
 
       def setup_time_parser_5424(time_fmt)
@@ -162,6 +165,8 @@ def parse_auto(text, &block)
       end
 
       SPLIT_CHAR = ' '.freeze
+      DOT_CHAR = '.'.freeze
+      RFC3164_PRI_DIGITS_REGEXP = /\A[0-9]{1,3}\z/
 
       def parse_rfc3164_regex(text, &block)
         idx = 0
@@ -180,13 +185,26 @@ def parse_rfc3164_regex(text, &block)
 
         i = idx - 1
         sq = false
+        first_time_field = true
         @space_count_rfc3164.times do
           while text[i + 1] == SPLIT_CHAR
+            if first_time_field
+              unless @time_format_starts_with_space_rfc3164
+                idx += 1
+                i += 1
+                break
+              end
+            end
             sq = true
             i += 1
           end
 
+          first_time_field = false
           i = text.index(SPLIT_CHAR, i + 1)
+          unless i
+            yield nil, nil
+            return
+          end
         end
 
         time_str = sq ? text.slice(idx, i - idx).squeeze(SPLIT_CHAR) : text.slice(idx, i - idx)
@@ -287,16 +305,30 @@ def parse_rfc3164(text, &block)
           end
         end
 
+        if text[cursor] == SPLIT_CHAR
+          cursor = rfc3164_space_cursor(text, cursor)
+          unless cursor
+            yield nil, nil
+            return
+          end
+        end
+
         if @skip_space_count_rfc3164
           # header part
           time_size = 15 # skip Mmm dd hh:mm:ss
-          time_end = text[cursor + time_size]
+          time_end_index = cursor + time_size
+          time_end = text[time_end_index]
+
           if time_end == SPLIT_CHAR
             time_str = text.slice(cursor, time_size)
             cursor += 16 # time + ' '
-          elsif time_end == '.'.freeze
+          elsif time_end == DOT_CHAR
             # support subsecond time
-            i = text.index(SPLIT_CHAR, time_size)
+            i = text.index(SPLIT_CHAR, time_end_index)
+            unless i
+              yield nil, nil
+              return
+            end
             time_str = text.slice(cursor, i - cursor)
             cursor = i + 1
           else
@@ -312,14 +344,18 @@ def parse_rfc3164(text, &block)
               i += 1
             end
             i = text.index(SPLIT_CHAR, i + 1)
+            unless i
+              yield nil, nil
+              return
+            end
           end
 
-          time_str = sq ? text.slice(idx, i - cursor).squeeze(SPLIT_CHAR) : text.slice(cursor, i - cursor)
+          time_str = sq ? text.slice(cursor, i - cursor).squeeze(SPLIT_CHAR) : text.slice(cursor, i - cursor)
           cursor = i + 1
         end
 
         i = text.index(SPLIT_CHAR, cursor)
-        if i.nil?
+        unless i
           yield nil, nil
           return
         end
@@ -363,12 +399,23 @@ def parse_rfc3164(text, &block)
         msg.chomp!
         record['message'] = msg
 
-        time = @time_parser_rfc3164.parse(time_str)
+        begin
+          time = @time_parser_rfc3164.parse(time_str)
+        rescue Fluent::TimeParser::TimeParseError
+          yield nil, nil
+          return
+        end
         record['time'] = time_str if @keep_time_key
 
         yield time, record
       end
 
+      def rfc3164_space_cursor(text, cursor)
+        return if @with_priority && !RFC3164_PRI_DIGITS_REGEXP.match?(text.slice(1, cursor - 2))
+
+        @time_format_starts_with_space_rfc3164 ? cursor : cursor + 1
+      end
+
       NILVALUE = '-'.freeze
 
       def parse_rfc5424(text, &block)

test/plugin/test_in_syslog.rb

@@ -124,6 +124,79 @@ def test_time_format(data)
     }
   end
 
+  data(
+    'regexp/rfc3164/parser priority' => ['regexp', 'rfc3164', true],
+    'string/rfc3164/parser priority' => ['string', 'rfc3164', true],
+    'regexp/auto/parser priority' => ['regexp', 'auto', true],
+    'string/auto/parser priority' => ['string', 'auto', true],
+    'regexp/rfc3164/input priority' => ['regexp', 'rfc3164', false],
+    'string/rfc3164/input priority' => ['string', 'rfc3164', false],
+    'regexp/auto/input priority' => ['regexp', 'auto', false],
+    'string/auto/input priority' => ['string', 'auto', false],
+  )
+  def test_space_between_rfc3164_priority_and_header(data)
+    parser_engine, message_format, with_priority = data
+    d = create_driver([
+      ipv4_config,
+      'severity_key severity',
+      'facility_key facility',
+      '<parse>',
+      "  parser_engine #{parser_engine}",
+      "  message_format #{message_format}",
+      "  with_priority #{with_priority}",
+      '</parse>',
+    ].join("\n"))
+
+    message = 'Apr 25 16:43:29 PAA-SW1-1 General[procLOG]: main.c(257) 272264 %% Stopping System API  application'
+    d.run(expect_emits: 2) do
+      u = UDPSocket.new
+      u.connect('127.0.0.1', @port)
+      u.send("<14>#{message}", 0)
+      u.send("<14> #{message}", 0)
+    end
+
+    assert_equal(2, d.events.size)
+    assert_equal(d.events[0][1], d.events[1][1])
+    assert_equal(d.events[0][2], d.events[1][2])
+    d.events.each do |tag, time, record|
+      assert_equal('syslog.user.info', tag)
+      assert_equal(event_time('Apr 25 16:43:29', format: '%b %d %H:%M:%S'), time)
+      assert_equal('user', record['facility'])
+      assert_equal('info', record['severity'])
+      assert_equal('PAA-SW1-1', record['host'])
+      assert_equal('General', record['ident'])
+      assert_equal('main.c(257) 272264 %% Stopping System API  application', record['message'])
+    end
+  end
+
+  data('regexp' => 'regexp', 'string' => 'string')
+  def test_space_after_input_owned_priority_preserves_input_priority_syntax(parser_engine)
+    d = create_driver([
+      ipv4_config,
+      'emit_unmatched_lines true',
+      '<parse>',
+      "  parser_engine #{parser_engine}",
+      '  with_priority false',
+      '</parse>',
+    ].join("\n"))
+
+    messages = [
+      '<1234>Apr 25 16:43:29 host app: message',
+      '<1234> Apr 25 16:43:29 host app: message',
+      '<ab> Apr 25 16:43:29 host app: message',
+    ]
+    d.run(expect_emits: 3) do
+      u = UDPSocket.new
+      u.connect('127.0.0.1', @port)
+      messages.each { |message| u.send(message, 0) }
+    end
+
+    assert_equal(3, d.events.size)
+    assert_equal(d.events[0], d.events[1])
+    assert_equal('syslog.unmatched', d.events[2][0])
+    assert_equal(messages[2], d.events[2][2]['unmatched_line'])
+  end
+
   def test_msg_size
     d = create_driver
     tests = create_test_case