update redis to prevent cross-slot violations in cluster mode

Author: TinOo512

lib/logstash/inputs/redis.rb

@@ -3,6 +3,7 @@
 require "logstash/inputs/base"
 require "logstash/inputs/threadable"
 require 'redis'
+require 'redis-clustering'
 require "stud/interval"
 
 # This input will read events from a Redis instance; it supports both Redis channels and lists.
@@ -137,12 +138,12 @@ def redis_params
     if @path.nil?
       if !@cluster_hosts.nil?
         params = {
-          :cluster => cluster_hosts,
+          :nodes => cluster_hosts,
         }
       elsif !@sentinel_hosts.nil?
         hosts = @sentinel_hosts.map { |sentinel_host| { host: sentinel_host, port: @sentinel_port } }
         params = {
-          :host => @sentinel_master_name,
+          :name => @sentinel_master_name,
           :sentinels => hosts,
           :role => :master
         }
@@ -161,18 +162,13 @@ def redis_params
   end
 
   def new_redis_instance
-    ::Redis.new(redis_params)
+    @cluster_hosts.nil? ? ::Redis.new(redis_params) : ::Redis::Cluster.new(redis_params)
   end
 
   # private
   def connect
     redis = new_redis_instance
 
-    # register any renamed Redis commands
-    @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
@@ -266,7 +262,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, timeout: 1)
     return unless item # from timeout or other conditions
 
     # blpop returns the 'key' read from as well as the item result

logstash-input-redis.gemspec

@@ -23,7 +23,8 @@ 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.0.1', '< 5'
+  s.add_runtime_dependency 'redis', '>= 5.0', '< 6'
+  s.add_runtime_dependency 'redis-clustering', '>= 5.0', '< 6'
 
   s.add_development_dependency 'logstash-devutils'
 end

spec/inputs/redis_spec.rb

@@ -140,35 +140,25 @@ def process(conf, event_count)
       }
     end
 
-    it 'sets the renamed commands in the command map' do
-      allow_any_instance_of( Redis::Client ).to receive(:call) do |_, command|
+    it 'registers and connects without error' do
+      allow_any_instance_of( Redis::Client ).to receive(:call_v) do |_, command|
         expect(command[0]).to eql :script
-        expect(command[1]).to eql 'load'
-      end
+        expect(command[1].to_s.downcase).to eql 'load'
+      end.and_return('a' * 40)
 
       subject.register
-      redis = subject.send :connect
-
-      command_map = redis._client.command_map
-
-      expect(command_map[:blpop]).to eq config['command_map']['blpop'].to_sym
-      expect(command_map[:evalsha]).to eq config['command_map']['evalsha'].to_sym
-      expect(command_map[:lrange]).to eq config['command_map']['lrange'].to_sym
-      expect(command_map[:ltrim]).to eq config['command_map']['ltrim'].to_sym
-      expect(command_map[:script]).to eq config['command_map']['script'].to_sym
-      expect(command_map[:subscribe]).to eq config['command_map']['subscribe'].to_sym
-      expect(command_map[:psubscribe]).to eq config['command_map']['psubscribe'].to_sym
+      expect { subject.send :connect }.not_to raise_error
     end
 
     it 'loads the batch script with the renamed command' do
-      expect_any_instance_of( Redis::Client ).to receive(:call) do |_, command|
+      expect_any_instance_of( Redis::Client ).to receive(:call_v) do |_, command|
         expect(command[0]).to eql :script
-        expect(command[1]).to eql 'load'
+        expect(command[1].to_s.downcase).to eql 'load'
 
         script = command[2]
         expect(script).to include "redis.call('#{config['command_map']['lrange']}', KEYS[1], 0, batchsize)"
         expect(script).to include "redis.call('#{config['command_map']['ltrim']}', KEYS[1], batchsize + 1, -1)"
-      end
+      end.and_return('a' * 40)
 
       subject.register
       subject.send :connect
@@ -199,9 +189,10 @@ def process(conf, event_count)
     end
 
     it 'calling the run method, adds events to the queue' do
-      allow_any_instance_of( Redis::Client ).to receive(:call_with_timeout) do |_, command, timeout, &block|
+      allow_any_instance_of( Redis::Client ).to receive(:blocking_call_v) do |_, timeout, command|
         expect(command[0]).to eql :blpop
-        expect(command[1]).to eql ['foo', 0]
+        expect(command[1]).to eql 'foo'
+        expect(command[2]).to eql ({ timeout: 1 })
       end.and_return ['foo', "{\"foo1\":\"bar\"}"], nil
 
       tt = Thread.new do
@@ -218,7 +209,7 @@ def process(conf, event_count)
 
     it 'keeps running when a connection error occurs' do
       raised = false
-      allow_any_instance_of( Redis::Client ).to receive(:call_with_timeout) do |_, command, timeout, &block|
+      allow_any_instance_of( Redis::Client ).to receive(:blocking_call_v) do |_, timeout, command|
         expect(command[0]).to eql :blpop
         unless raised
           raised = true
@@ -283,8 +274,8 @@ def process(conf, event_count)
       let(:batch_count) { 10 }
 
       it 'calling the run method, adds events to the queue' do
-        allow_any_instance_of( Redis ).to receive(:script)
-        allow_any_instance_of( Redis::Client ).to receive(:call) do |_, command|
+        allow_any_instance_of( Redis ).to receive(:script).and_return('a' * 40)
+        allow_any_instance_of( Redis::Client ).to receive(:call_v) do |_, command|
           expect(command[0]).to eql :evalsha
         end.and_return ['{"a": 1}', '{"b": 2}'], []
 
@@ -306,8 +297,8 @@ def process(conf, event_count)
       let(:rates) { [] }
 
       it 'will throttle the loop' do
-        allow_any_instance_of( Redis ).to receive(:script)
-        allow_any_instance_of( Redis::Client ).to receive(:call) do |_, command|
+        allow_any_instance_of( Redis ).to receive(:script).and_return('a' * 40)
+        allow_any_instance_of( Redis::Client ).to receive(:call_v) do |_, command|
           expect(command[0]).to eql :evalsha
           rates.unshift Time.now.to_f
         end.and_return []