in_kafka: time_source record drops the time parsed by apache2, syslog and some regexp parsers

Open Beginner friendly
  • Is something wrong?
#612 0 comments 0 reactions 0 assignees View on GitHub

Maintainers usually reply within 2 days

Nobody has claimed this yet.

Assessment

Difficulty
2/5
Estimated time
1-3 hours
Newbie friendliness
84/100
Issue type
Bug
Clarity
Clearly specified
Activity status
Active
Tech stack
kafka, ruby
Domain
backend, stream-processing

Research direction

Start in lib/fluent/plugin/in_kafka.rb around the configure-time consumes_record_time_key check described in the issue, then compare how #598 handles parsed_time in in_kafka_group and in_rdkafka_group. Reproduce with the included script or the stated apache2, syslog, and regexp cases; done means each event uses the parsed message time when the record has no configured time field.

Written by the indexing model from the issue text.

Description

bug
Describe the bug

With time_source record and a custom parser, in_kafka drops the time that the parser returns in some cases. The event then gets the current time. No warning is logged.

in_kafka decides at configure time whether the parser consumes the record time:

consumes_record_time_key = @custom_parser.time_key == @record_time_key && !@custom_parser.keep_time_key

https://github.com/fluent/fluent-plugin-kafka/blob/4a408fb/lib/fluent/plugin/in_kafka.rb#L190

This guess is false in the following cases, although the parser removes the time field from the record:

  • format apache2 and format syslog. Their time_key is nil.
  • format regexp with time_key ts, while record_time_key is time (the default).

So the parsed time is not passed on. With time_format set, the record has no time field, and the event gets Fluent::Engine.now.

#598 changed in_kafka_group and in_rdkafka_group to decide this at run time (parsed_time && !record.has_key?(@record_time_key)). in_kafka still uses the guess added in #592.

This affects master only. #592 is not released yet (0.19.9 was released before it).

To Reproduce

Use the configuration below and send this message to the topic:

192.168.0.1 - - [28/Feb/2013:12:00:00 +0900] "GET / HTTP/1.1" 200 777 "-" "Opera/12.0"

I reproduced it without a broker. The script below calls TopicWatcher#consume with a stub client that returns one message.

Results on master (4a408fb):

format event time
apache2 current time
syslog current time
regexp with time_key ts current time
regexp with time_key time 2013-02-28 03:00:00 UTC (correct)
Expected behavior

The event time is the time in the message. For the apache2 example, it is 2013-02-28 03:00:00 UTC.

Your Environment
- Fluentd version: 1.20.1
- TD Agent version: -
- fluent-plugin-kafka version: master (4a408fb)
- ruby-kafka version: 1.5.0
- Ruby version: 4.0.9
- Operating system: Linux
Your Configuration
<source>
  @type kafka
  brokers localhost:9092
  topics test
  time_source record
  format apache2
  time_format %d/%b/%Y:%H:%M:%S %z
</source>
Your Error Log
No error or warning is logged.
Additional context

A possible fix is the same check as #598. Use the parsed time when parsed_time && !record.has_key?(@record_time_key). I can send a PR.

Reproduction script (run with bundle exec ruby -Ilib repro.rb in this repository) 
require 'fluent/test'
require 'fluent/test/driver/input'
require 'fluent/plugin/in_kafka'

Fluent::Test.setup

Msg = Struct.new(:value, :offset, :key, :create_time)

class FakeKafka
  def initialize(msgs)
    @msgs = msgs
  end

  def fetch_messages(**)
    @msgs
  end
end

class FakeRouter
  attr_reader :events

  def initialize
    @events = []
  end

  def emit_stream(tag, es)
    es.each { |t, r| @events << [tag, t, r] }
  end
end

def run_case(name, conf, value)
  d = Fluent::Test::Driver::Input.new(Fluent::KafkaInput).configure(conf)
  i = d.instance
  iv = ->(n) { i.instance_variable_get(n) }
  router = FakeRouter.new
  te = Fluent::KafkaInput::TopicEntry.new('test', 0, 0)
  tw = Fluent::KafkaInput::TopicWatcher.new(
    te, FakeKafka.new([Msg.new(value, 0, nil, Time.at(1_600_000_000))]), 1,
    iv.(:@parser_proc), nil, nil, nil, router, nil,
    iv.(:@time_source), iv.(:@record_time_key), iv.(:@time_parser),
    iv.(:@tag_source), iv.(:@record_tag_key), {}
  )
  tw.consume
  parser = iv.(:@custom_parser)
  puts "== #{name}"
  puts "  parser time_key=#{parser.time_key.inspect} keep_time_key=#{parser.keep_time_key.inspect} record_time_key=#{iv.(:@record_time_key).inspect}"
  router.events.each do |tag, t, r|
    puts "  tag=#{tag} time=#{t.inspect} (#{Time.at(t.to_r).utc}) record=#{r.inspect}"
  end
end

base = %[
  @type kafka
  brokers localhost:9092
  topics test
  time_source record
]

run_case('apache2', base + %[
  format apache2
  time_format %d/%b/%Y:%H:%M:%S %z
], '192.168.0.1 - - [28/Feb/2013:12:00:00 +0900] "GET / HTTP/1.1" 200 777 "-" "Opera/12.0"')

run_case('regexp with time_key ts', base + %[
  format /^(?<ts>[^ ]+) (?<message>.*)$/
  time_key ts
  time_format %Y-%m-%dT%H:%M:%S%z
], '2013-02-28T12:00:00+0900 hello')

run_case('regexp with time_key time (guess matches)', base + %[
  format /^(?<time>[^ ]+) (?<message>.*)$/
  time_format %Y-%m-%dT%H:%M:%S%z
], '2013-02-28T12:00:00+0900 hello')

run_case('syslog', base + %[
  format syslog
  time_format %b %d %H:%M:%S
], 'Feb 28 12:00:00 192.168.0.1 fluentd[11111]: [error] Syslog test')
Dominant language
Ruby
Stars
307
Forks
179
Avg merge
1d 6h
Merged PRs (30d)
22

Getting set up

This project ships no dev container, Dockerfile or contributing guide, so setting up is up to you: start from its README, and see our first-contribution guide for the general steps.

First steps

  1. Read the whole issue, then the project's contributing guide.
  2. Comment on the issue to say you are picking it up — it saves two people doing the same work.
  3. Fork the repository and make your change on a branch.
  4. Open a pull request that references the issue number.

More from fluent/fluent-plugin-kafka

All issues in fluent/fluent-plugin-kafka

Similar issues

More Ruby issues

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.