Refactor redis connection handling for pattern_list

Author: Juuso Mäyränen

lib/logstash/inputs/redis.rb

@@ -267,8 +267,8 @@ def list_pattern_stop
   # private
   def worker_consume(output_queue, key)
     @logger.debug("Launched worker for #{key}")
-    redis = new_redis_instance
     begin
+      redis ||= connect
       (0...@pattern_list_max_items).each do
         if stop?
           @logger.debug("Breaking from thread #{key} as it was requested to stop")
@@ -278,6 +278,10 @@ def worker_consume(output_queue, key)
         break if value.nil?
         queue_event(value, output_queue)
       end
+    rescue ::Redis::BaseError => e
+      @logger.warn("Redis connection problem in thread for key #{key}. Sleeping a while before exiting thread.", :exception => e)
+      sleep 1
+      return
     ensure
       redis.quit rescue nil
     end
@@ -304,31 +308,21 @@ def launch_worker(output_queue, key)
   # private
   def ensure_workers(output_queue)
     return unless threadpool_capacity?
-    @redis.keys(@key).shuffle.each do |key|
-      next if @current_workers.include?(key)
-      launch_worker(output_queue, key)
-      break unless threadpool_capacity?
+    redis_runner do
+      @redis.keys(@key).shuffle.each do |key|
+        next if @current_workers.include?(key)
+        launch_worker(output_queue, key)
+        break unless threadpool_capacity?
+      end
     end
   end
 
-  # private
-  def init_list_pattern_runner
-    init_threadpool
-    @redis ||= connect
-  end
-
   # private
   def list_pattern_runner(output_queue)
     while !stop?
-      begin
-        init_list_pattern_runner if @redis.nil?
-        ensure_workers(output_queue)
-        sleep(@pattern_list_threadpool_sleep)
-      rescue ::Redis::BaseError => e
-        @logger.warn("Redis connection problem", :exception => e)
-        @redis = nil
-        sleep 1
-      end
+      init_threadpool if @threadpool.nil?
+      ensure_workers(output_queue)
+      sleep(@pattern_list_threadpool_sleep)
     end
   end