Make stuff configurable

Author: Juuso Mäyränen

lib/logstash/inputs/redis.rb

@@ -53,8 +53,10 @@ module LogStash module Inputs class Redis < LogStash::Inputs::Threadable
   config :key, :validate => :string, :required => true
 
   # Specify either list or channel.  If `data_type` is `list`, then we will BLPOP the
-  # key.  If `data_type` is `channel`, then we will SUBSCRIBE to the key.
-  # If `data_type` is `pattern_channel`, then we will PSUBSCRIBE to the key.
+  # key. If `data_type` is `pattern_list`, then we will spawn a number of worker
+  # threads that will LPOP from keys matching that pattern. If `data_type` is
+  # `channel`, then we will SUBSCRIBE to the key. If `data_type` is `pattern_channel`,
+  # then we will PSUBSCRIBE to the key.
   config :data_type, :validate => [ "list", "pattern_list", "channel", "pattern_channel" ], :required => true
 
   # The number of events to return from Redis using EVAL.
@@ -63,8 +65,18 @@ module LogStash module Inputs class Redis < LogStash::Inputs::Threadable
   # Redefined Redis commands to be passed to the Redis client.
   config :command_map, :validate => :hash, :default => {}
 
+  # Maximum number of worker threads to spawn when using `data_type` `pattern_list`.
   config :worker_thread_count, :validate => :number, :default => 20
 
+  # Maximum number of items for a single worker thread to process when `data_type` is `pattern_list`.
+  # After the queue is empty or this number of items have been processed, the thread will exit and a
+  # new one will be started.
+  config :max_items_per_worker, :validate => :number, :default => 1000
+
+  # Time to sleep in main loop after checking if more threads can/need to be spawned.
+  # Applies to `data_type` is `pattern_list`
+  config :threadpool_queue_sleep, :validate => :number, :default => 0.2
+
   public
   # public API
   # use to store a proc that can provide a Redis instance or mock
@@ -252,7 +264,7 @@ def launch_worker(output_queue, key)
     @logger.debug("Launched worker for #{key}")
     redis = new_redis_instance
     begin
-      (0..MAX_ITEMS_PER_WORKER).each do
+      (0..@max_items_per_worker).each do
         if stop?
           @logger.debug("Breaking from thread #{key} as it was requested to stop")
           break
@@ -304,7 +316,7 @@ def list_pattern_runner(output_queue)
       begin
         init_list_pattern_runner if @redis.nil?
         ensure_workers(output_queue)
-        sleep(0.1)
+        sleep(@threadpool_queue_sleep)
       rescue ::Redis::BaseError => e
         @logger.warn("Redis connection problem", :exception => e)
         @redis = nil