Merge remote-tracking branch 'origin/master' into list-pattern-input

Author: Juuso Mäyränen

.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/[email protected]
+- logstash-plugins/.ci:travis/[email protected]
+
+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,10 @@
+## 3.7.0
+  - Fix: better (Redis) exception handling [#89](https://github.com/logstash-plugins/logstash-input-redis/pull/89)
+  - 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)
+
 ## 3.6.0
   - Remove ruby pipeline dependency. Starting from Logstash 8, Ruby execution engine is not available. All pipelines should use Java pipeline [#84](https://github.com/logstash-plugins/logstash-input-redis/pull/84)
 

README.md

@@ -1,7 +1,7 @@
 # Logstash Plugin
 
 Travis Build
-[![Travis Build Status](https://travis-ci.org/logstash-plugins/logstash-input-redis.svg)](https://travis-ci.org/logstash-plugins/logstash-input-redis)
+[![Travis Build Status](https://travis-ci.com/logstash-plugins/logstash-input-redis.svg)](https://travis-ci.com/logstash-plugins/logstash-input-redis)
 
 This is a plugin for [Logstash](https://github.com/elastic/logstash).
 

lib/logstash/inputs/redis.rb

@@ -77,22 +77,6 @@ module LogStash module Inputs class Redis < LogStash::Inputs::Threadable
   config :pattern_list_threadpool_sleep, :validate => :number, :default => 0.2
 
   public
-  # public API
-  # use to store a proc that can provide a Redis instance or mock
-  def add_external_redis_builder(builder) #callable
-    @redis_builder = builder
-    self
-  end
-
-  # use to apply an instance directly and bypass the builder
-  def use_redis(instance)
-    @redis = instance
-    self
-  end
-
-  def new_redis_instance
-    @redis_builder.call
-  end
 
   def init_threadpool
     @threadpool ||= Concurrent::ThreadPoolExecutor.new(
@@ -106,8 +90,6 @@ def init_threadpool
   def register
     @redis_url = @path.nil? ? "redis://#{@password}@#{@host}:#{@port}/#{@db}" : "#{@password}@#{@path}/#{@db}"
 
-    @redis_builder ||= method(:internal_redis_builder)
-
     # just switch on data_type once
     if @data_type == 'list' || @data_type == 'dummy'
       @run_method = method(:list_runner)
@@ -151,30 +133,25 @@ 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
 
-  # private
-  def internal_redis_builder
+  def new_redis_instance
     ::Redis.new(redis_params)
   end
 
@@ -183,14 +160,12 @@ def connect
     redis = new_redis_instance
 
     # register any renamed Redis commands
-    if @command_map.any?
-      client_command_map = redis.client.command_map
-      @command_map.each do |name, renamed|
-        client_command_map[name.to_sym] = renamed.to_sym
-      end
+    @command_map.each do |name, renamed|
+      redis._client.command_map[name.to_sym] = renamed.to_sym
     end
 
     load_batch_script(redis) if batched? && is_list_type?
+
     redis
   end # def connect
 
@@ -221,9 +196,12 @@ def queue_event(msg, output_queue, channel=nil)
 
   # private
   def reset_redis
-    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 # does client.disconnect internally
+    # check if input retried while executing
+    list_stop unless redis.equal? @redis
     @redis = nil
   end
 
@@ -239,13 +217,9 @@ def list_runner(output_queue)
       begin
         @redis ||= connect
         @list_method.call(@redis, output_queue)
-      rescue ::Redis::BaseError => e
-        @logger.warn("Redis connection problem", :exception => e)
-        # 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
+        log_error(e)
+        retry if reset_for_error_retry(e)
       end
     end
   end
@@ -400,18 +374,19 @@ 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.client.is_a?(::Redis::SubscribedClient)
+    redis = @redis # might change during method invocation
+    return if redis.nil? || !redis.connected?
+
+    if redis.subscribed?
       if @data_type == 'pattern_channel'
-        @redis.client.punsubscribe
+        redis.punsubscribe
       else
-        @redis.client.unsubscribe
+        redis.unsubscribe
       end
-    else
-      @redis.client.disconnect
     end
+    redis.close rescue nil # does client.disconnect
+    # check if input retried while executing
+    subscribe_stop unless redis.equal? @redis
     @redis = nil
   end
 
@@ -420,15 +395,43 @@ 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
+      log_error(e)
+      retry if reset_for_error_retry(e)
+    end
+  end
+
+  def log_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")
+    else
+      info[:backtrace] ||= e.backtrace
+      @logger.error("Unexpected error", info)
     end
   end
 
+  # @return [true] if operation is fine to retry
+  def reset_for_error_retry(e)
+    return if e.is_a?(::LogStash::ShutdownSignal)
+
+    # Reset the redis variable to trigger reconnect
+    @redis = nil
+
+    Stud.stoppable_sleep(1) { stop? }
+    !stop? # retry if not stop-ing
+  end
+
   # private
   def channel_runner(output_queue)
     redis_runner do
@@ -476,6 +479,4 @@ def pattern_channel_listener(output_queue)
     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.0'
+  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"
@@ -23,7 +23,7 @@ Gem::Specification.new do |s|
   s.add_runtime_dependency "logstash-core-plugin-api", ">= 1.60", "<= 2.99"
 
   s.add_runtime_dependency 'logstash-codec-json'
-  s.add_runtime_dependency 'redis', '~> 4'
+  s.add_runtime_dependency 'redis', '>= 4.0.1', '< 5'
 
   s.add_development_dependency 'logstash-devutils'
 end