Input
# File lib/fluent/plugin/in_tail.rb, line 36 def configure(conf) super @paths = @path.split(',').map {|path| path.strip } if @paths.empty? raise ConfigError, "tail: 'path' parameter is required on tail input" end if @pos_file @pf_file = File.open(@pos_file, File::RDWR|File::CREAT, DEFAULT_FILE_PERMISSION) @pf_file.sync = true @pf = PositionFile.parse(@pf_file) else $log.warn "'pos_file PATH' parameter is not set to a 'tail' source." $log.warn "this parameter is highly recommended to save the position to resume tailing." end configure_parser(conf) end
# File lib/fluent/plugin/in_tail.rb, line 56 def configure_parser(conf) @parser = TextParser.new @parser.configure(conf) end
# File lib/fluent/plugin/in_tail.rb, line 113 def parse_line(line) return @parser.parse(line) end
# File lib/fluent/plugin/in_tail.rb, line 89 def receive_lines(lines) es = MultiEventStream.new lines.each {|line| begin line.chomp! # remove \n time, record = parse_line(line) if time && record es.add(time, record) end rescue $log.warn line.dump, :error=>$!.to_s $log.debug_backtrace end } unless es.empty? begin Engine.emit_stream(@tag, es) rescue # ignore errors. Engine shows logs and backtraces. end end end
# File lib/fluent/plugin/in_tail.rb, line 82 def run @loop.run rescue $log.error "unexpected error", :error=>$!.to_s $log.error_backtrace end
# File lib/fluent/plugin/in_tail.rb, line 73 def shutdown @tails.each {|tail| tail.close } @loop.stop @thread.join @pf_file.close if @pf_file end
# File lib/fluent/plugin/in_tail.rb, line 61 def start @loop = Coolio::Loop.new @tails = @paths.map {|path| pe = @pf ? @pf[path] : MemoryPositionEntry.new TailWatcher.new(path, @rotate_wait, pe, &method(:receive_lines)) } @tails.each {|tail| tail.attach(@loop) } @thread = Thread.new(&method(:run)) end
Generated with the Darkfish Rdoc Generator 2.