Add better support for file rotation

.travis.yml

@@ -1,18 +1,16 @@
 ---
-sudo: false
-language: ruby
-cache: bundler
+sudo: required
+services: docker
+addons:
+  apt:
+    packages:
+      - docker-ce
 matrix:
   include:
-  - rvm: jruby-9.1.13.0
-    env: LOGSTASH_BRANCH=master
-  - rvm: jruby-9.1.13.0
-    env: LOGSTASH_BRANCH=6.x
-  - rvm: jruby-9.1.13.0
-    env: LOGSTASH_BRANCH=6.0
-  - rvm: jruby-1.7.27
-    env: LOGSTASH_BRANCH=5.6
+  - env: ELASTIC_STACK_VERSION=5.6.10
+  - env: ELASTIC_STACK_VERSION=6.3.0
+  - env: ELASTIC_STACK_VERSION=6.4.0-SNAPSHOT
+  - env: ELASTIC_STACK_VERSION=7.0.0-alpha1-SNAPSHOT
   fast_finish: true
-install: true
-script: ci/build.sh
-jdk: oraclejdk8
+install: ci/unit/docker-setup.sh
+script: ci/unit/docker-run.sh

JAR_VERSION

@@ -1 +1 @@
-1.0.0
+1.0.1

README.md

@@ -2,9 +2,6 @@
 Travis Build
 [![Travis Build Status](https://travis-ci.org/logstash-plugins/logstash-input-file.svg)](https://travis-ci.org/logstash-plugins/logstash-input-file)
 
-Jenkins Build
-[![Travis Build Status](https://travis-ci.org/logstash-plugins/logstash-input-file.svg)](https://travis-ci.org/logstash-plugins/logstash-input-file)
-
 This is a plugin for [Logstash](https://github.com/elastic/logstash).
 
 It is fully free and fully open source. The license is Apache 2.0, meaning you are pretty much free to use it however you want in whatever way.

ci/unit/Dockerfile

@@ -0,0 +1,11 @@
+ARG ELASTIC_STACK_VERSION
+FROM docker.elastic.co/logstash/logstash:$ELASTIC_STACK_VERSION
+WORKDIR /usr/share/logstash/logstash-core
+RUN cp versions-gem-copy.yml ../logstash-core-plugin-api/versions-gem-copy.yml
+COPY --chown=logstash:logstash . /usr/share/plugins/this
+WORKDIR /usr/share/plugins/this
+ENV PATH=/usr/share/logstash/vendor/jruby/bin:${PATH}
+ENV LOGSTASH_SOURCE 1
+RUN jruby -S gem install bundler
+RUN jruby -S bundle install --jobs=3 --retry=3
+RUN jruby -S bundle exec rake vendor

ci/unit/docker-compose.yml

@@ -0,0 +1,17 @@
+version: '3'
+
+# run tests: docker-compose -f ci/unit/docker-compose.yml up --build --force-recreate
+# only set up: docker-compose -f ci/unit/docker-compose.yml up --build --no-start --force-recreate
+# start manually: docker-compose -f ci/unit/docker-compose.yml run logstash
+services:
+  logstash:
+    build:
+      context: ../../
+      dockerfile: ci/unit/Dockerfile
+      args:
+        - ELASTIC_STACK_VERSION=$ELASTIC_STACK_VERSION
+    command: /usr/share/plugins/this/ci/unit/run.sh
+    environment:
+      LS_JAVA_OPTS: "-Xmx256m -Xms256m"
+      OSS: "true"
+    tty: true

ci/unit/docker-run.sh

@@ -0,0 +1,7 @@
+#!/bin/bash
+
+# This is intended to be run the plugin's root directory. `ci/unit/docker-test.sh`
+# Ensure you have Docker installed locally and set the ELASTIC_STACK_VERSION environment variable.
+set -e
+
+docker-compose -f ci/unit/docker-compose.yml run logstash

ci/unit/docker-setup.sh

@@ -0,0 +1,31 @@
+#!/bin/bash
+
+# This is intended to be run the plugin's root directory. `ci/unit/docker-test.sh`
+# Ensure you have Docker installed locally and set the ELASTIC_STACK_VERSION environment variable.
+set -e
+
+if [ "$ELASTIC_STACK_VERSION" ]; then
+    echo "Testing against version: $ELASTIC_STACK_VERSION"
+
+    if [[ "$ELASTIC_STACK_VERSION" = *"-SNAPSHOT" ]]; then
+        cd /tmp
+        wget https://snapshots.elastic.co/docker/logstash-"$ELASTIC_STACK_VERSION".tar.gz
+        tar xfvz logstash-"$ELASTIC_STACK_VERSION".tar.gz  repositories
+        echo "Loading docker image: "
+        cat repositories
+        docker load < logstash-"$ELASTIC_STACK_VERSION".tar.gz
+        rm logstash-"$ELASTIC_STACK_VERSION".tar.gz
+        cd -
+    fi
+
+    if [ -f Gemfile.lock ]; then
+        rm Gemfile.lock
+    fi
+
+    docker-compose -f ci/unit/docker-compose.yml down
+    docker-compose -f ci/unit/docker-compose.yml up --no-start --build --force-recreate logstash
+else
+    echo "Please set the ELASTIC_STACK_VERSION environment variable"
+    echo "For example: export ELASTIC_STACK_VERSION=6.2.4"
+    exit 1
+fi

ci/unit/run.sh

@@ -0,0 +1,6 @@
+#!/bin/bash
+
+# This is intended to be run inside the docker container as the command of the docker-compose.
+set -ex
+
+bundle exec rspec -fd --pattern spec/**/*_spec.rb,spec/**/*_specs.rb

lib/filewatch/bootstrap.rb

@@ -1,7 +1,5 @@
 # encoding: utf-8
-require "rbconfig"
 require "pathname"
-# require "logstash/environment"
 
 ## Common setup
 #  all the required constants and files
@@ -13,36 +11,26 @@ module FileWatch
   # this is used in the read loop e.g.
   # @opts[:file_chunk_count].times do
   # where file_chunk_count defaults to this constant
-  FIXNUM_MAX = (2**(0.size * 8 - 2) - 1)
+  MAX_ITERATIONS = (2**(0.size * 8 - 2) - 2) / 32768
 
   require_relative "helper"
 
-  module WindowsInode
-    def prepare_inode(path, stat)
-      fileId = Winhelper.GetWindowsUniqueFileIdentifier(path)
-      [fileId, 0, 0] # dev_* doesn't make sense on Windows
-    end
-  end
-
-  module UnixInode
-    def prepare_inode(path, stat)
-      [stat.ino.to_s, stat.dev_major, stat.dev_minor]
-    end
-  end
-
-  jar_version = Pathname.new(__FILE__).dirname.join("../../JAR_VERSION").realpath.read.strip
-
+  gem_root_dir = Pathname.new(__FILE__).dirname.join("../../").realpath
+  jar_version = gem_root_dir.join("JAR_VERSION").read.strip
+  fullpath = gem_root_dir.join("lib/jars/filewatch-#{jar_version}.jar").expand_path.to_path
   require "java"
-  require_relative "../../lib/jars/filewatch-#{jar_version}.jar"
+  require fullpath
   require "jruby_file_watch"
 
   if LogStash::Environment.windows?
     require_relative "winhelper"
+    require_relative "stat/windows_path"
+    PathStatClass = Stat::WindowsPath
     FileOpener = FileExt
-    InodeMixin = WindowsInode
   else
+    require_relative "stat/generic"
+    PathStatClass = Stat::Generic
     FileOpener = ::File
-    InodeMixin = UnixInode
   end
 
   # Structs can be used as hash keys because they compare by value

lib/filewatch/discoverer.rb

@@ -10,8 +10,8 @@ class Discoverer
     include LogStash::Util::Loggable
 
     def initialize(watched_files_collection, sincedb_collection, settings)
-      @watching = []
-      @exclude = []
+      @watching = Concurrent::Array.new
+      @exclude = Concurrent::Array.new
       @watched_files_collection = watched_files_collection
       @sincedb_collection = sincedb_collection
       @settings = settings
@@ -21,13 +21,13 @@ def initialize(watched_files_collection, sincedb_collection, settings)
     def add_path(path)
       return if @watching.member?(path)
       @watching << path
-      discover_files(path)
+      discover_files_new_path(path)
       self
     end
 
     def discover
       @watching.each do |path|
-        discover_files(path)
+        discover_files_ongoing(path)
       end
     end
 
@@ -37,7 +37,7 @@ def can_exclude?(watched_file, new_discovery)
       @exclude.each do |pattern|
         if watched_file.pathname.fnmatch?(pattern)
           if new_discovery
-            logger.debug("Discoverer can_exclude?: #{watched_file.path}: skipping " +
+            logger.trace("Discoverer can_exclude?: #{watched_file.path}: skipping " +
               "because it matches exclude #{pattern}")
           end
           watched_file.unwatch
@@ -47,45 +47,52 @@ def can_exclude?(watched_file, new_discovery)
       false
     end
 
-    def discover_files(path)
-      globbed = Dir.glob(path)
-      globbed = [path] if globbed.empty?
-      logger.debug("Discoverer found files, count: #{globbed.size}")
-      globbed.each do |file|
-        logger.debug("Discoverer found file, path: #{file}")
+    def discover_files_new_path(path)
+      discover_any_files(path, false)
+    end
+
+    def discover_files_ongoing(path)
+      discover_any_files(path, true)
+    end
+
+    def discover_any_files(path, ongoing)
+      fileset = Dir.glob(path).select{|f| File.file?(f) && !File.symlink?(f)}
+      logger.trace("discover_files",  "count" => fileset.size)
+      fileset.each do |file|
         pathname = Pathname.new(file)
-        next unless pathname.file?
-        next if pathname.symlink?
         new_discovery = false
         watched_file = @watched_files_collection.watched_file_by_path(file)
         if watched_file.nil?
-          logger.debug("Discoverer discover_files: #{path}: new: #{file} (exclude is #{@exclude.inspect})")
           new_discovery = true
-          watched_file = WatchedFile.new(pathname, pathname.stat, @settings)
+          watched_file = WatchedFile.new(pathname, PathStatClass.new(pathname), @settings)
         end
         # if it already unwatched or its excluded then we can skip
         next if watched_file.unwatched? || can_exclude?(watched_file, new_discovery)
 
+        logger.trace("discover_files handling:", "new discovery"=> new_discovery, "watched_file details" => watched_file.details)
+
         if new_discovery
-          if watched_file.file_ignorable?
-            logger.debug("Discoverer discover_files: #{file}: skipping because it was last modified more than #{@settings.ignore_older} seconds ago")
-            # on discovery ignorable watched_files are put into the ignored state and that
-            # updates the size from the internal stat
-            # so the existing contents are not read.
-            # because, normally, a newly discovered file will
-            # have a watched_file size of zero
-            # they are still added to the collection so we know they are there for the next periodic discovery
-            watched_file.ignore
-          end
-          # now add the discovered file to the watched_files collection and adjust the sincedb collections
-          @watched_files_collection.add(watched_file)
+          watched_file.initial_completed if ongoing
           # initially when the sincedb collection is filled with records from the persistence file
           # each value is not associated with a watched file
           # a sincedb_value can be:
           #   unassociated
           #   associated with this watched_file
           #   associated with a different watched_file
-          @sincedb_collection.associate(watched_file)
+          if @sincedb_collection.associate(watched_file)
+            if watched_file.file_ignorable?
+              logger.trace("Discoverer discover_files: #{file}: skipping because it was last modified more than #{@settings.ignore_older} seconds ago")
+              # on discovery ignorable watched_files are put into the ignored state and that
+              # updates the size from the internal stat
+              # so the existing contents are not read.
+              # because, normally, a newly discovered file will
+              # have a watched_file size of zero
+              # they are still added to the collection so we know they are there for the next periodic discovery
+              watched_file.ignore_as_unread
+            end
+            # now add the discovered file to the watched_files collection and adjust the sincedb collections
+            @watched_files_collection.add(watched_file)
+          end
         end
         # at this point the watched file is created, is in the db but not yet opened or being processed
       end

lib/filewatch/observing_base.rb

@@ -44,7 +44,8 @@ def initialize(opts={})
         :exclude => [],
         :start_new_files_at => :end,
         :delimiter => "\n",
-        :file_chunk_count => FIXNUM_MAX,
+        :file_chunk_count => MAX_ITERATIONS,
+        :file_chunk_size => FILE_READ_SIZE,
         :file_sort_by => "last_modified",
         :file_sort_direction => "asc",
       }.merge(opts)