Add sentinel and cluster support

.travis.yml

@@ -10,4 +10,57 @@ addons:
 
 before_install:
   - sudo service redis-server stop
-  - sudo service redis-server start --bind 0.0.0.0
+
+  # --- Standalone (port 6379) ---
+  - redis-server --port 6379 --daemonize yes --logfile /tmp/redis-standalone.log
+
+  # --- Sentinel Setup (ports 7000–7002, sentinels 26379–26381) ---
+  - mkdir -p /tmp/redis-sentinel/{7000,7001,7002}
+
+  # Primary
+  - redis-server --port 7000 --daemonize yes --logfile /tmp/redis-sentinel/7000/redis.log --dir /tmp/redis-sentinel/7000
+
+  # Replicas
+  - redis-server --port 7001 --daemonize yes --replicaof 127.0.0.1 7000 --logfile /tmp/redis-sentinel/7001/redis.log --dir /tmp/redis-sentinel/7001
+  - redis-server --port 7002 --daemonize yes --replicaof 127.0.0.1 7000 --logfile /tmp/redis-sentinel/7002/redis.log --dir /tmp/redis-sentinel/7002
+
+  # Sentinel config files
+  - |
+    for port in 26379 26380 26381; do
+      {
+        echo "port ${port}"
+        echo "sentinel monitor mymaster 127.0.0.1 7000 2"
+        echo "sentinel down-after-milliseconds mymaster 5000"
+        echo "sentinel failover-timeout mymaster 10000"
+        echo "sentinel parallel-syncs mymaster 1"
+        echo "daemonize yes"
+        echo "logfile /tmp/sentinel-${port}.log"
+      } > /tmp/sentinel-${port}.conf
+      redis-server /tmp/sentinel-${port}.conf --sentinel
+    done
+
+  # --- Cluster Setup (ports 7003–7008) ---
+  - mkdir -p /tmp/redis-cluster/{7003,7004,7005,7006,7007,7008}
+
+  - |
+    for port in 7003 7004 7005 7006 7007 7008; do
+      {
+        echo "port ${port}"
+        echo "cluster-enabled yes"
+        echo "cluster-config-file /tmp/redis-cluster/${port}/nodes.conf"
+        echo "cluster-node-timeout 5000"
+        echo "appendonly yes"
+        echo "daemonize yes"
+        echo "logfile /tmp/redis-cluster/${port}/redis.log"
+        echo "dir /tmp/redis-cluster/${port}"
+      } > /tmp/redis-cluster/${port}/redis.conf
+      redis-server /tmp/redis-cluster/${port}/redis.conf
+    done
+
+  # Bootstrap the cluster (3 masters + 3 replicas)
+  - sleep 1
+  - |
+    echo "yes" | redis-cli --cluster create \
+      127.0.0.1:7003 127.0.0.1:7004 127.0.0.1:7005 \
+      127.0.0.1:7006 127.0.0.1:7007 127.0.0.1:7008 \
+      --cluster-replicas 1

CHANGELOG.md

@@ -1,3 +1,7 @@
+## 3.8.0
+  - Add Redis sentinel and cluster support [#101](https://github.com/logstash-plugins/logstash-input-redis/pull/101/changes)
+  - Updated redis client dependency to ~> 5
+
 ## 3.7.1
   - Add documentation to "threads" option [#95](https://github.com/logstash-plugins/logstash-input-redis/pull/95)
 

docs/index.asciidoc

@@ -43,6 +43,7 @@ This plugin supports the following configuration options plus the <<plugins-{typ
 |=======================================================================
 |Setting |Input type|Required
 | <<plugins-{type}s-{plugin}-batch_count>> |<<number,number>>|No
+| <<plugins-{type}s-{plugin}-cluster_hosts>> |<<array,array>>|No
 | <<plugins-{type}s-{plugin}-command_map>> |<<hash,hash>>|No
 | <<plugins-{type}s-{plugin}-data_type>> |<<string,string>>, one of `["list", "channel", "pattern_channel"]`|Yes
 | <<plugins-{type}s-{plugin}-db>> |<<number,number>>|No
@@ -51,6 +52,8 @@ This plugin supports the following configuration options plus the <<plugins-{typ
 | <<plugins-{type}s-{plugin}-key>> |<<string,string>>|Yes
 | <<plugins-{type}s-{plugin}-password>> |<<password,password>>|No
 | <<plugins-{type}s-{plugin}-port>> |<<number,number>>|No
+| <<plugins-{type}s-{plugin}-sentinel_hosts>> |<<array,array>>|No
+| <<plugins-{type}s-{plugin}-sentinel_master_name>> |<<string,string>>|No
 | <<plugins-{type}s-{plugin}-ssl>> |<<boolean,boolean>>|No
 | <<plugins-{type}s-{plugin}-threads>> |<<number,number>>|No
 | <<plugins-{type}s-{plugin}-timeout>> |<<number,number>>|No
@@ -69,6 +72,15 @@ input plugins.
 
 The number of events to return from Redis using EVAL.
 
+[id="plugins-{type}s-{plugin}-cluster_hosts"]
+===== `cluster_hosts`
+
+  * Value type is <<array,array>>
+  * There is no default value for this setting.
+  * Cluster will override Host or Sentinel configuration if both specified.
+
+The connection URLs of your Redis Cluster nodes, e.g. `["redis://127.0.0.1:7000", "redis://127.0.0.1:7001"]`.
+
 [id="plugins-{type}s-{plugin}-command_map"]
 ===== `command_map`
 
@@ -112,7 +124,7 @@ The hostname of your Redis server.
 
   * Value type is <<string,string>>
   * There is no default value for this setting.
-  * Path will override Host configuration if both specified.
+  * Path will override Host, Sentinel or Cluster configuration if both specified.
 
 The unix socket path of your Redis server.
 
@@ -141,6 +153,23 @@ Password to authenticate with. There is no authentication by default.
 
 The port to connect on.
 
+[id="plugins-{type}s-{plugin}-sentinel_hosts"]
+===== `sentinel_hosts`
+
+  * Value type is <<array,array>>
+  * There is no default value for this setting.
+  * Sentinel will override Host configuration if both specified.
+
+The hostnames and ports of your Redis Sentinel servers, e.g. `["127.0.0.1:26379", "127.0.0.1:26380"]`.
+
+[id="plugins-{type}s-{plugin}-sentinel_master_name"]
+===== `sentinel_master_name`
+
+  * Value type is <<string,string>>
+  * Default value is `"mymaster"`
+
+The name of the Redis Sentinel master group to connect to. Only used when `sentinel_hosts` is set.
+
 [id="plugins-{type}s-{plugin}-ssl"]
 ===== `ssl` 
 

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.
@@ -27,13 +28,22 @@ module LogStash module Inputs class Redis < LogStash::Inputs::Threadable
   # The hostname of your Redis server.
   config :host, :validate => :string, :default => "127.0.0.1"
 
+  # The hostnames and ports of your sentinel servers.
+  config :sentinel_hosts, :validate => :array
+
+  # The connection URLs of your cluster servers.
+  config :cluster_hosts, :validate => :array
+
   # The port to connect on.
   config :port, :validate => :number, :default => 6379
 
+  # The name of the sentinel master to connect to.
+  config :sentinel_master_name, :validate => :string, :default => "mymaster"
+
   # SSL
   config :ssl, :validate => :boolean, :default => false
 
-  # The unix socket path to connect on. Will override host and port if defined.
+  # The unix socket path to connect on. Will override host and port and sentinel_hosts and cluster_hosts if defined.
   # There is no unix socket path by default.
   config :path, :validate => :string
 
@@ -63,7 +73,15 @@ module LogStash module Inputs class Redis < LogStash::Inputs::Threadable
   public
 
   def register
-    @redis_url = @path.nil? ? "redis://#{@password}@#{@host}:#{@port}/#{@db}" : "#{@password}@#{@path}/#{@db}"
+    if !@path.nil?
+      @redis_url = "#{@password}@#{@path}/#{@db}"
+    elsif !@cluster_hosts.nil?
+      @redis_url = "#{@password}@#{@cluster_hosts.map { |h| "#{h}" }.join(',')}/#{@db}"
+    elsif !@sentinel_hosts.nil?
+      @redis_url = "redis://#{@password}@#{@sentinel_master_name}(#{@sentinel_hosts.map { |h| "#{h}" }.join(',')})/#{@db}"
+    else
+      @redis_url = "redis://#{@password}@#{@host}:#{@port}/#{@db}"
+    end
 
     # just switch on data_type once
     if @data_type == 'list' || @data_type == 'dummy'
@@ -115,29 +133,42 @@ def redis_params
     }
 
     if @path.nil?
-      params[:host] = @host
-      params[:port] = @port
+      if !@cluster_hosts.nil?
+        params = {
+          :nodes => cluster_hosts,
+        }
+      elsif !@sentinel_hosts.nil?
+        hosts = @sentinel_hosts.map do |sentinel_host|
+          host, port = sentinel_host.split(':', 2)
+          { host: host, port: port ? port.to_i : 26379 }
+        end
+        params = {
+          :name => @sentinel_master_name,
+          :sentinels => hosts,
+          :role => :master
+        }
+      else
+        params = {
+          :host => @host,
+          :port => @port
+        }
+      end
     else
-      @logger.warn("Parameter 'path' is set, ignoring parameters: 'host' and 'port'")
+      @logger.warn("Parameter 'path' is set, ignoring parameters: 'host', 'port', 'sentinel_hosts' and 'cluster_hosts'")
       params[:path] = @path
     end
 
     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
@@ -231,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