Support batches in list pattern

Author: Juuso Mäyränen

lib/logstash/inputs/redis.rb

@@ -122,8 +122,6 @@ def register
       @stop_method = method(:subscribe_stop)
     end
 
-    @list_method = batched? ? method(:list_batch_listener) : method(:list_single_listener)
-
     @identity = "#{@redis_url} #{@data_type}:#{@key}"
     @logger.info("Registering Redis", :identity => @identity)
   end # def register
@@ -147,7 +145,7 @@ def batched?
 
   # private
   def is_list_type?
-    @data_type == 'list'
+    @data_type == 'list' || @data_type == 'pattern_list'
   end
 
   # private
@@ -235,6 +233,7 @@ def list_stop
 
   # private
   def list_runner(output_queue)
+    @list_method = batched? ? method(:list_batch_listener) : method(:list_single_listener)
     while !stop?
       begin
         @redis ||= connect
@@ -264,28 +263,53 @@ def list_pattern_stop
     reset_threadpool
   end
 
+  # private
+  def list_pattern_process_item(redis, output_queue, key)
+    if stop?
+      @logger.debug("Breaking from thread #{key} as it was requested to stop")
+      return false
+    end
+    value = redis.lpop(key)
+    return false if value.nil?
+    queue_event(value, output_queue)
+    true
+  end
+
+  # private
+  def list_pattern_single_processor(redis, output_queue, key)
+    (0...@pattern_list_max_items).each do
+      break unless list_pattern_process_item(redis, output_queue, key)
+    end
+  end
+
+  # private
+  def list_pattern_batch_processor(redis, output_queue, key)
+    items_left = @pattern_list_max_items
+    while items_left > 0
+      limit = [items_left, @batch_count].min
+      processed = process_batch(redis, output_queue, key, limit, 0)
+      @logger.warn("Got batch size #{processed} for #{key}")
+      if processed.zero? || processed < limit
+        return
+      end
+      items_left -= processed
+    end
+  end
+
   # private
   def worker_consume(output_queue, key)
     @logger.debug("Launched worker for #{key}")
     begin
       redis ||= connect
-      (0...@pattern_list_max_items).each do
-        if stop?
-          @logger.debug("Breaking from thread #{key} as it was requested to stop")
-          break
-        end
-        value = redis.lpop(key)
-        break if value.nil?
-        queue_event(value, output_queue)
-      end
+      @list_pattern_processor.call(redis, output_queue, key)
     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
-    @logger.debug("Exit worker for #{key}")
+    @logger.warn("Exit worker for #{key}")
   end
 
   # private
@@ -319,23 +343,22 @@ def ensure_workers(output_queue)
 
   # private
   def list_pattern_runner(output_queue)
+    @list_pattern_processor = batched? ? method(:list_pattern_batch_processor) : method(:list_pattern_single_processor)
     while !stop?
       init_threadpool if @threadpool.nil?
       ensure_workers(output_queue)
       sleep(@pattern_list_threadpool_sleep)
     end
   end
 
-  def list_batch_listener(redis, output_queue)
+  def process_batch(redis, output_queue, key, batch_size, sleep_time)
     begin
-      results = redis.evalsha(@redis_script_sha, [@key], [@batch_count-1])
+      results = redis.evalsha(@redis_script_sha, [key], [batch_size-1])
       results.each do |item|
         queue_event(item, output_queue)
       end
-
-      if results.size.zero?
-        sleep BATCH_EMPTY_SLEEP
-      end
+      sleep sleep_time if results.size.zero? && sleep_time > 0
+      results.size
 
       # Below is a commented-out implementation of 'batch fetch'
       # using pipelined LPOP calls. This in practice has been observed to
@@ -364,6 +387,10 @@ def list_batch_listener(redis, output_queue)
     end
   end
 
+  def list_batch_listener(redis, output_queue)
+    process_batch(redis, output_queue, @key, @batch_count, BATCH_EMPTY_SLEEP)
+  end
+
   def list_single_listener(redis, output_queue)
     item = redis.blpop(@key, 0, :timeout => 1)
     return unless item # from timeout or other conditions