Bunch of improvements

Author: Juuso Mäyränen

lib/logstash/inputs/redis.rb

@@ -3,6 +3,7 @@
 require "logstash/inputs/base"
 require "logstash/inputs/threadable"
 require 'redis'
+require 'concurrent/atomics'
 
 # This input will read events from a Redis instance; it supports both Redis channels and lists.
 # The list command (BLPOP) used by Logstash is supported in Redis v1.3.1+, and
@@ -229,16 +230,27 @@ def list_pattern_stop
 
     @redis.quit rescue nil
     @redis = nil
+    @worker_threads.each do |key, thread|
+      if thread.alive?
+        thread.join(5)
+        @logger.warn("Thread #{key} joined")
+      end
+    end
   end
 
-  def launch_worker(key)
+  def launch_worker(output_queue, key)
      Thread.new do
       redis = new_redis_instance
       begin
         (0..MAX_ITEMS_PER_WORKER).each do
+          if stop?
+            @logger.warn("Breaking from thread #{key} as it was requested to stop")
+            break
+          end
           value = redis.lpop(key)
           break if value.nil?
-          @events << value
+          queue_event(value, output_queue)
+          @items_processed.increment
         end
       ensure
         redis.quit rescue nil
@@ -255,45 +267,40 @@ def clear_finished_workers
     finished_threads.each { |key| @worker_threads.delete(key) }
   end
 
-  def ensure_workers
+  def ensure_workers(output_queue)
     free_slots = @worker_thread_count - @worker_threads.length
     return if free_slots == 0
     keys = @redis.keys(@key)
     keys.shuffle
     keys.each do |key|
       next if @worker_threads.has_key?(key)
-      @worker_threads[key] = launch_worker(key)
+      @worker_threads[key] = launch_worker(output_queue, key)
       free_slots -= 1
       break if free_slots == 0
     end
   end
 
-  def init_list_pattern_runner
-    @events ||= Queue.new
+  def init_list_pattern_runner(output_queue)
     @worker_threads ||= {}
     @redis ||= connect
+    @items_processed ||= Concurrent::AtomicFixnum.new
     clear_finished_workers
-    ensure_workers
+    ensure_workers(output_queue)
   end
 
   # private
   def list_pattern_runner(output_queue)
     items = 0
     while !stop?
       begin
-        init_list_pattern_runner
-        item = @events.pop(true)
-        next unless item
-        queue_event(item, output_queue)
-        items += 1
-      rescue ThreadError => e
-        @logger.warn("Items processed: #{items}")
+        init_list_pattern_runner(output_queue)
         sleep(0.1)
       rescue ::Redis::BaseError => e
         @logger.warn("Redis connection problem", :exception => e)
         @redis = nil
       end
     end
+    @logger.warn("Total items processed: #{@items_processed}")
   end
 
   def list_batch_listener(redis, output_queue)