Fix: make sure plugin can be stop-ed

.ci/docker-compose.override.yml

@@ -0,0 +1,5 @@
+version: '3'
+
+services:
+    logstash:
+      network_mode: host

.ci/run.sh

@@ -0,0 +1,10 @@
+#!/bin/bash
+# This is intended to be run inside the docker container as the command of the docker-compose.
+
+env
+
+set -ex
+
+jruby -rbundler/setup -S rspec -fd
+
+jruby -rbundler/setup -S rspec -fd --tag redis

.travis.yml

@@ -1,2 +1,13 @@
 import:
-- logstash-plugins/.ci:travis/travis.yml@1.x
+- logstash-plugins/.ci:travis/travis.yml@1.x
+
+addons:
+  apt:
+    sources:
+      - sourceline: 'ppa:chris-lea/redis-server'
+    packages:
+      - redis-server
+
+before_install:
+  - sudo service redis-server stop
+  - sudo service redis-server start --bind 0.0.0.0

CHANGELOG.md

@@ -1,3 +1,7 @@
+## 3.7.0
+  - Fix: make sure plugin can be stop-ed in case of a channel data_type [#87](https://github.com/logstash-plugins/logstash-input-redis/pull/87)
+  - Test: start running integration specs on CI
+
 ## 3.6.1
   - Fix: resolve crash when commands_map is set [#86](https://github.com/logstash-plugins/logstash-input-redis/pull/86)
 

lib/logstash/inputs/redis.rb

@@ -107,28 +107,26 @@ def is_list_type?
 
   # private
   def redis_params
+    params = {
+        :timeout => @timeout,
+        :db => @db,
+        :password => @password.nil? ? nil : @password.value,
+        :ssl => @ssl
+    }
+
     if @path.nil?
-      connectionParams = {
-        :host => @host,
-        :port => @port
-      }
+      params[:host] = @host
+      params[:port] = @port
     else
       @logger.warn("Parameter 'path' is set, ignoring parameters: 'host' and 'port'")
-      connectionParams = {
-        :path => @path
-      }
+      params[:path] = @path
     end
 
-    baseParams = {
-      :timeout => @timeout,
-      :db => @db,
-      :password => @password.nil? ? nil : @password.value,
-      :ssl => @ssl
-    }
-
-    return connectionParams.merge(baseParams)
+    params
   end
 
+  TIMEOUT = 5 # Redis only supports Integer values
+
   def new_redis_instance
     ::Redis.new(redis_params)
   end
@@ -174,9 +172,12 @@ def queue_event(msg, output_queue, channel=nil)
 
   # private
   def list_stop
-    return if @redis.nil? || !@redis.connected?
+    redis = @redis # might change during method invocation
+    return if redis.nil? || !redis.connected?
 
-    @redis.quit rescue nil
+    redis.quit rescue nil
+    # check if input retried while executing
+    list_stop unless redis.equal? @redis
     @redis = nil
   end
 
@@ -186,15 +187,8 @@ def list_runner(output_queue)
       begin
         @redis ||= connect
         @list_method.call(@redis, output_queue)
-      rescue ::Redis::BaseError => e
-        info = { message: e.message, exception: e.class }
-        info[:backtrace] = e.backtrace if @logger.debug?
-        @logger.warn("Redis connection problem", info)
-        # Reset the redis variable to trigger reconnect
-        @redis = nil
-        # this sleep does not need to be stoppable as its
-        # in a while !stop? loop
-        sleep 1
+      rescue => e
+        retry if handle_error(e)
       end
     end
   end
@@ -238,7 +232,7 @@ def list_batch_listener(redis, output_queue)
   end
 
   def list_single_listener(redis, output_queue)
-    item = redis.blpop(@key, 0, :timeout => 1)
+    item = redis.blpop(@key, 0, :timeout => TIMEOUT)
     return unless item # from timeout or other conditions
 
     # blpop returns the 'key' read from as well as the item result
@@ -248,18 +242,20 @@ def list_single_listener(redis, output_queue)
 
   # private
   def subscribe_stop
-    return if @redis.nil? || !@redis.connected?
-    # if its a SubscribedClient then:
-    # it does not have a disconnect method (yet)
-    if @redis.subscribed?
+    redis = @redis # might change during method invocation
+    return if redis.nil? || !redis.connected?
+
+    if redis.subscribed?
       if @data_type == 'pattern_channel'
-        @redis.punsubscribe
+        redis.punsubscribe
       else
-        @redis.unsubscribe
+        redis.unsubscribe
       end
     else
-      @redis.disconnect!
+      redis.disconnect!
     end
+    # check if input retried while executing
+    subscribe_stop unless redis.equal? @redis
     @redis = nil
   end
 
@@ -268,62 +264,79 @@ def redis_runner
     begin
       @redis ||= connect
       yield
-    rescue ::Redis::BaseError => e
-      @logger.warn("Redis connection problem", :exception => e)
-      # Reset the redis variable to trigger reconnect
-      @redis = nil
-      Stud.stoppable_sleep(1) { stop? }
-      retry if !stop?
+    rescue => e
+      retry if handle_error(e)
+    end
+  end
+
+  def handle_error(e)
+    info = { message: e.message, exception: e.class }
+    info[:backtrace] = e.backtrace if @logger.debug?
+
+    case e
+    when ::Redis::TimeoutError
+      # expected for channels in case no data is available
+      @logger.debug("Redis timeout, retrying", info)
+    when ::Redis::BaseConnectionError, ::Redis::ProtocolError
+      @logger.warn("Redis connection error", info)
+    when ::Redis::BaseError
+      @logger.error("Redis error", info)
+    when ::LogStash::ShutdownSignal
+      @logger.debug("Received shutdown signal")
+      return false # stop retry-ing
+    else
+      info[:backtrace] ||= e.backtrace
+      @logger.error("Unexpected error", info)
     end
+
+    # Reset the redis variable to trigger reconnect
+    @redis = nil
+
+    Stud.stoppable_sleep(1) { stop? }
+    !stop? # return true unless stop?
   end
 
   # private
   def channel_runner(output_queue)
-    redis_runner do
-      channel_listener(output_queue)
-    end
+    redis_runner { channel_listener(output_queue) }
   end
 
   # private
   def channel_listener(output_queue)
-    @redis.subscribe(@key) do |on|
+    @redis.subscribe_with_timeout(TIMEOUT, @key) do |on|
       on.subscribe do |channel, count|
-        @logger.info("Subscribed", :channel => channel, :count => count)
+        @logger.debug("Subscribed", :channel => channel, :count => count)
       end
 
       on.message do |channel, message|
         queue_event(message, output_queue, channel)
       end
 
       on.unsubscribe do |channel, count|
-        @logger.info("Unsubscribed", :channel => channel, :count => count)
+        @logger.debug("Unsubscribed", :channel => channel, :count => count)
       end
     end
   end
 
   def pattern_channel_runner(output_queue)
-    redis_runner do
-      pattern_channel_listener(output_queue)
-    end
+    redis_runner { pattern_channel_listener(output_queue) }
   end
 
   # private
   def pattern_channel_listener(output_queue)
-    @redis.psubscribe @key do |on|
+    @redis.psubscribe_with_timeout(TIMEOUT, @key) do |on|
       on.psubscribe do |channel, count|
-        @logger.info("Subscribed", :channel => channel, :count => count)
+        @logger.debug("Subscribed", :channel => channel, :count => count)
       end
 
       on.pmessage do |pattern, channel, message|
         queue_event(message, output_queue, channel)
       end
 
       on.punsubscribe do |channel, count|
-        @logger.info("Unsubscribed", :channel => channel, :count => count)
+        @logger.debug("Unsubscribed", :channel => channel, :count => count)
       end
     end
   end
 
-# end
-
 end end end # Redis Inputs  LogStash

logstash-input-redis.gemspec

@@ -1,7 +1,7 @@
 Gem::Specification.new do |s|
 
   s.name            = 'logstash-input-redis'
-  s.version         = '3.6.1'
+  s.version         = '3.7.0'
   s.licenses        = ['Apache License (2.0)']
   s.summary         = "Reads events from a Redis instance"
   s.description     = "This gem is a Logstash plugin required to be installed on top of the Logstash core pipeline using $LS_HOME/bin/logstash-plugin install gemname. This gem is not a stand-alone program"