Class: LogStash::Inputs::File

Inherits:

Base

• Object
• Base
• LogStash::Inputs::File

Defined in:

lib/logstash/inputs/file.rb

Instance Method Summary

• #handle_deletable_path(path) ⇒ Object
• #listener_for(path) ⇒ Object

  def register.

• #log_line_received(path, line) ⇒ Object
• #post_process_this(event) ⇒ Object

  def run.

• #register ⇒ Object
• #run(queue) ⇒ Object
• #start_processing ⇒ Object
• #stop ⇒ Object

Instance Method Details

#handle_deletable_path(path) ⇒ Object

# File 'lib/logstash/inputs/file.rb', line 329

def handle_deletable_path(path)
  return if tail_mode?
  return if @completed_file_handlers.empty?
  @completed_file_handlers.each { |handler| handler.handle(path) }
end

#listener_for(path) ⇒ Object

def register

# File 'lib/logstash/inputs/file.rb', line 302

def listener_for(path)
  # path is the identity
  FileListener.new(path, self)
end

#log_line_received(path, line) ⇒ Object

# File 'lib/logstash/inputs/file.rb', line 335

def log_line_received(path, line)
  return unless @logger.debug?
  @logger.debug("Received line", :path => path, :text => line)
end

#post_process_this(event) ⇒ Object

def run

# File 'lib/logstash/inputs/file.rb', line 322

def post_process_this(event)
  event.set("[@metadata][host]", @host)
  event.set("host", @host) unless event.include?("host")
  decorate(event)
  @queue << event
end

#register ⇒ Object

# File 'lib/logstash/inputs/file.rb', line 225

def register
  require "addressable/uri"
  require "digest/md5"
  @logger.trace("Registering file input", :path => @path)
  @host = Socket.gethostname.force_encoding(Encoding::UTF_8)
  # This check is Logstash 5 specific.  If the class does not exist, and it
  # won't in older versions of Logstash, then we need to set it to nil.
  settings = defined?(LogStash::SETTINGS) ? LogStash::SETTINGS : nil

  @filewatch_config = {
    :exclude => @exclude,
    :stat_interval => @stat_interval,
    :discover_interval => @discover_interval,
    :sincedb_write_interval => @sincedb_write_interval,
    :delimiter => @delimiter,
    :ignore_older => @ignore_older,
    :close_older => @close_older,
    :max_open_files => @max_open_files,
    :sincedb_clean_after => @sincedb_clean_after,
    :file_chunk_count => @file_chunk_count,
    :file_chunk_size => @file_chunk_size,
    :file_sort_by => @file_sort_by,
    :file_sort_direction => @file_sort_direction,
  }

  @completed_file_handlers = []

  @path.each do |path|
    if Pathname.new(path).relative?
      raise ArgumentError.new("File paths must be absolute, relative path specified: #{path}")
    end
  end

  if @sincedb_path.nil?
    base_sincedb_path = build_sincedb_base_from_settings(settings) || build_sincedb_base_from_env
    @sincedb_path = build_random_sincedb_filename(base_sincedb_path)
    @logger.info('No sincedb_path set, generating one based on the "path" setting', :sincedb_path => @sincedb_path.to_s, :path => @path)
  else
    @sincedb_path = Pathname.new(@sincedb_path)
    if @sincedb_path.directory?
      raise ArgumentError.new("The \"sincedb_path\" argument must point to a file, received a directory: \"#{@sincedb_path}\"")
    end
  end

  @filewatch_config[:sincedb_path] = @sincedb_path

  @filewatch_config[:start_new_files_at] = @start_position.to_sym

  if @file_completed_action.include?('log')
    if @file_completed_log_path.nil?
      raise ArgumentError.new('The "file_completed_log_path" setting must be provided when the "file_completed_action" is set to "log" or "log_and_delete"')
    else
      @file_completed_log_path = Pathname.new(@file_completed_log_path)
      unless @file_completed_log_path.exist?
        begin
          FileUtils.touch(@file_completed_log_path)
        rescue
          raise ArgumentError.new("The \"file_completed_log_path\" file can't be created: #{@file_completed_log_path}")
        end
      end
    end
  end

  if tail_mode?
    @watcher_class = FileWatch::ObservingTail
  else
    @watcher_class = FileWatch::ObservingRead
    if @file_completed_action.include?('log')
      @completed_file_handlers << LogCompletedFileHandler.new(@file_completed_log_path)
    end
    if @file_completed_action.include?('delete')
      @completed_file_handlers << DeleteCompletedFileHandler.new
    end
  end
  @codec = LogStash::Codecs::IdentityMapCodec.new(@codec)
end

#run(queue) ⇒ Object

# File 'lib/logstash/inputs/file.rb', line 315

def run(queue)
  start_processing
  @queue = queue
  @watcher.subscribe(self) # halts here until quit is called
  exit_flush
end

#start_processing ⇒ Object

# File 'lib/logstash/inputs/file.rb', line 307

def start_processing
  # if the pipeline restarts this input,
  # make sure previous files are closed
  stop
  @watcher = @watcher_class.new(@filewatch_config)
  @path.each { |path| @watcher.watch_this(path) }
end

#stop ⇒ Object

# File 'lib/logstash/inputs/file.rb', line 340

def stop
  # in filewatch >= 0.6.7, quit will closes and forget all files
  # but it will write their last read positions to since_db
  # beforehand
  if @watcher
    @codec.close
    @watcher.quit
  end
end