Add better support for file rotation (logstash-plugins#192)

* Add better support for file rotation

* Move common_restat for watched and active to one iteration
to better handle when a file is rotated that has never been active.
Change the way @previous_inode_size is cached, it needs to be set only
when an inode change is detected.
Change discover associate behaviour if the matched watched_file is open.
Add spec for when a file is rotated that has never been active.
Alter specs because we can rely on ordering with "file_sort_by"

* Ensure single file `path` option is ok, reverts earlier change in this PR

* Improve and expand our use of FFI on Windows

* jnr and ffi interop, yay

* put self back in

* use fieldId label

* handle rotations better.
especially in the case where the rotated files are discoverable - in
this case we were not the moving the state and the sincedb record
correctly.
Next - abstract the *nix and Windows stat calls and structure into
two classes with the same API.

* fixes for rebase from master

* Abstract Stat part 1

* Abstract Stat part 2

* Finally have (all) the kinks worked out

* Try to fix travis failures.

* Try to fix travis failures 2

* Try fix travis 3

* Try fix travis 4

* Try fix travis 5

* Try fix travis 6

* Remove io based stat reliance. travis jruby 1.7.27 should pass now.

Move all tail processing state iteration into its own method
Clean up rspec sequencing usage.
Remove all Mutex use in favour of AtomicBoolean, AtomicArray, AtomicHash
Add wait for completely_stopped before removing sincedb in file_tail spec

* Some windows fixes

* more windows fixes

* windows changes 2

* rename rspec run tag in ci/build.sh

* move one trace logging line

* add first run discovery methods

* fix regression on files seen after inital run, travis 2 use docker.

* add execute permissions

* fix path ordering travis failures

* fix jar loading so it works for tests and when installed in LS

* reorder the jar require statements

Fixes logstash-plugins#198

Author: Guy Boertje

.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/build.sh

@@ -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/setup.sh

@@ -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/Dockerfile

@@ -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-compose.yml

@@ -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/docker-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

ci/unit/docker-setup.sh

@@ -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

ci/unit/run.sh

@@ -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/bootstrap.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)