Class: Fluent::TailInput::TailWatcher

Inherits:

Object

• Object
• Fluent::TailInput::TailWatcher

Defined in:

lib/fluent/plugin/in_tail.rb

Defined Under Namespace

Classes: IOHandler, NullIOHandler, RotateHandler, RotationRequest, StatWatcher, TimerWatcher

Constant Summary

MAX_LINES_AT_ONCE =

1000

Instance Attribute Summary

• #log ⇒ Object

  We use accessor approach to assign each logger, not passing log object at initialization, because several plugins depend on these internal classes.

Instance Method Summary

• #attach(loop) ⇒ Object
• #close ⇒ Object
• #detach ⇒ Object
• #initialize(path, rotate_wait, pe, &receive_lines) ⇒ TailWatcher constructor

  A new instance of TailWatcher.

• #on_notify ⇒ Object
• #on_rotate(io) ⇒ Object

Constructor Details

#initialize(path, rotate_wait, pe, &receive_lines) ⇒ TailWatcher

Returns a new instance of TailWatcher.

# File 'lib/fluent/plugin/in_tail.rb', line 920

def initialize(path, rotate_wait, pe, &receive_lines)
  @path = path
  @rotate_wait = rotate_wait
  @pe = pe || MemoryPositionEntry.new
  @receive_lines = receive_lines

  @rotate_queue = []

  @timer_trigger = TimerWatcher.new(1, true, &method(:on_notify))
  @stat_trigger = StatWatcher.new(path, &method(:on_notify))

  @rotate_handler = RotateHandler.new(path, &method(:on_rotate))
  @io_handler = nil
  @log = $log
end

Instance Attribute Details

#log ⇒ Object

We use accessor approach to assign each logger, not passing log object at initialization, because several plugins depend on these internal classes. This approach avoids breaking plugins with new log_level option.

# File 'lib/fluent/plugin/in_tail.rb', line 939

def log
  @log
end

Instance Method Details

#attach(loop) ⇒ Object

# File 'lib/fluent/plugin/in_tail.rb', line 948

def attach(loop)
  @timer_trigger.attach(loop)
  @stat_trigger.attach(loop)
  on_notify
end

#close ⇒ Object

# File 'lib/fluent/plugin/in_tail.rb', line 959

def close
  @rotate_queue.reject! {|req|
    req.io.close
    true
  }
  detach
end

#detach ⇒ Object

# File 'lib/fluent/plugin/in_tail.rb', line 954

def detach
  @timer_trigger.detach if @timer_trigger.attached?
  @stat_trigger.detach if @stat_trigger.attached?
end

#on_notify ⇒ Object

# File 'lib/fluent/plugin/in_tail.rb', line 967

def on_notify
  @rotate_handler.on_notify
  return unless @io_handler
  @io_handler.on_notify

  # proceeds rotate queue
  return if @rotate_queue.empty?
  @rotate_queue.first.tick

  while @rotate_queue.first.ready?
    if io = @rotate_queue.first.io
      stat = io.stat
      inode = stat.ino
      if inode == @pe.read_inode
        # rotated file has the same inode number with the last file.
        # assuming following situation:
        #   a) file was once renamed and backed, or
        #   b) symlink or hardlink to the same file is recreated
        # in either case, seek to the saved position
        pos = @pe.read_pos
      else
        pos = io.pos
      end
      @pe.update(inode, pos)
      io_handler = IOHandler.new(io, @pe, log, &@receive_lines)
    else
      io_handler = NullIOHandler.new
    end
    @io_handler.close
    @io_handler = io_handler
    @rotate_queue.shift
    break if @rotate_queue.empty?
  end
end

#on_rotate(io) ⇒ Object

# File 'lib/fluent/plugin/in_tail.rb', line 1002

def on_rotate(io)
  if @io_handler == nil
    if io
      # first time
      stat = io.stat
      fsize = stat.size
      inode = stat.ino

      last_inode = @pe.read_inode
      if inode == last_inode
        # seek to the saved position
        pos = @pe.read_pos
      elsif last_inode != 0
        # this is FilePositionEntry and fluentd once started.
        # read data from the head of the rotated file.
        # logs never duplicate because this file is a rotated new file.
        pos = 0
        @pe.update(inode, pos)
      else
        # this is MemoryPositionEntry or this is the first time fluentd started.
        # seek to the end of the any files.
        # logs may duplicate without this seek because it's not sure the file is
        # existent file or rotated new file.
        pos = fsize
        @pe.update(inode, pos)
      end
      io.seek(pos)

      @io_handler = IOHandler.new(io, @pe, log, &@receive_lines)
    else
      @io_handler = NullIOHandler.new
    end

  else
    if io && @rotate_queue.find {|req| req.io == io }
      return
    end
    last_io = @rotate_queue.empty? ? @io_handler.io : @rotate_queue.last.io
    if last_io == nil
      log.info "detected rotation of #{@path}"
      # rotate imeediately if previous file is nil
      wait = 0
    else
      log.info "detected rotation of #{@path}; waiting #{@rotate_wait} seconds"
      wait = @rotate_wait
      wait -= @rotate_queue.first.wait unless @rotate_queue.empty?
    end
    @rotate_queue << RotationRequest.new(io, wait)
  end
end