Support Sentinel with TLS

sazzad16 · sazzad16 · commit 77736ad50451 · 2021-03-07T16:21:40.000+06:00
diff --git a/Makefile b/Makefile
@@ -254,6 +254,12 @@ pid = /tmp/stunnel.pid
 [redis]
 accept = 127.0.0.1:6390
 connect = 127.0.0.1:6379
+[redis_3]
+accept = 127.0.0.1:16381
+connect = 127.0.0.1:6381
+[redis_4]
+accept = 127.0.0.1:16382
+connect = 127.0.0.1:6382
 [redis_cluster_1]
 accept = 127.0.0.1:8379
 connect = 127.0.0.1:7379
@@ -269,6 +275,18 @@ connect = 127.0.0.1:7382
 [redis_cluster_5]
 accept = 127.0.0.1:8383
 connect = 127.0.0.1:7383
+[redis_sentinel_1]
+accept = 127.0.0.1:36379
+connect = 127.0.0.1:26379
+[redis_sentinel_2]
+accept = 127.0.0.1:36380
+connect = 127.0.0.1:26380
+[redis_sentinel_3]
+accept = 127.0.0.1:36381
+connect = 127.0.0.1:26381
+[redis_sentinel_4]
+accept = 127.0.0.1:36382
+connect = 127.0.0.1:26382
 endef
 
 export REDIS1_CONF
diff --git a/src/main/java/redis/clients/jedis/JedisSentinelPool.java b/src/main/java/redis/clients/jedis/JedisSentinelPool.java
@@ -1,10 +1,12 @@
 package redis.clients.jedis;
 
 import java.util.Arrays;
-import java.util.HashSet;
+import java.util.Collection;
+import java.util.LinkedList;
 import java.util.List;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.stream.Collectors;
 
 import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
 import org.slf4j.Logger;
@@ -15,30 +17,15 @@
 
 public class JedisSentinelPool extends JedisPoolAbstract {
 
-  /**
-   * @deprecated This will be private in future.
-   */
-  @Deprecated
-  protected static Logger log = LoggerFactory.getLogger(JedisSentinelPool.class);
+  private static final Logger LOG = LoggerFactory.getLogger(JedisSentinelPool.class);
 
-  protected final GenericObjectPoolConfig poolConfig;
+  private final GenericObjectPoolConfig poolConfig;
 
-  protected final int connectionTimeout;
-  protected final int soTimeout;
-  protected final int infiniteSoTimeout;
+  private final JedisClientConfig masterClientConfig;
 
-  protected final String user;
-  protected final String password;
-  protected final int database;
-  protected final String clientName;
+  private final JedisClientConfig sentinelClientConfig;
 
-  protected int sentinelConnectionTimeout;
-  protected int sentinelSoTimeout;
-  protected String sentinelUser;
-  protected String sentinelPassword;
-  protected String sentinelClientName;
-
-  protected final Set<MasterListener> masterListeners = new HashSet<>();
+  private final Collection<MasterListener> masterListeners = new LinkedList<>();
 
   private volatile JedisFactory factory;
   private volatile HostAndPort currentHostMaster;
@@ -160,25 +147,32 @@ public JedisSentinelPool(String masterName, Set<String> sentinels,
       final String user, final String password, final int database, final String clientName,
       final int sentinelConnectionTimeout, final int sentinelSoTimeout, final String sentinelUser,
       final String sentinelPassword, final String sentinelClientName) {
+    this(masterName, parseHostAndPorts(sentinels), poolConfig,
+        DefaultJedisClientConfig.builder().withConnectionTimeout(connectionTimeout)
+            .withSoTimeout(soTimeout).withInfiniteSoTimeout(infiniteSoTimeout).withUser(user)
+            .withPassword(password).withDatabse(database).withClientName(clientName).build(),
+        DefaultJedisClientConfig.builder().withConnectionTimeout(sentinelConnectionTimeout)
+            .withSoTimeout(sentinelSoTimeout).withUser(sentinelUser).withPassword(sentinelPassword)
+            .withClientName(sentinelClientName).build()
+    );
+  }
+
+  public JedisSentinelPool(String masterName, Set<HostAndPort> sentinels,
+      final GenericObjectPoolConfig poolConfig, final JedisClientConfig masteClientConfig,
+      final JedisClientConfig sentinelClientConfig) {
 
     this.poolConfig = poolConfig;
-    this.connectionTimeout = connectionTimeout;
-    this.soTimeout = soTimeout;
-    this.infiniteSoTimeout = infiniteSoTimeout;
-    this.user = user;
-    this.password = password;
-    this.database = database;
-    this.clientName = clientName;
-    this.sentinelConnectionTimeout = sentinelConnectionTimeout;
-    this.sentinelSoTimeout = sentinelSoTimeout;
-    this.sentinelUser = sentinelUser;
-    this.sentinelPassword = sentinelPassword;
-    this.sentinelClientName = sentinelClientName;
+    this.masterClientConfig = masteClientConfig;
+    this.sentinelClientConfig = sentinelClientConfig;
 
     HostAndPort master = initSentinels(sentinels, masterName);
     initPool(master);
   }
 
+  private static Set<HostAndPort> parseHostAndPorts(Set<String> strings) {
+    return strings.parallelStream().map(str -> HostAndPort.parseString(str)).collect(Collectors.toSet());
+  }
+
   @Override
   public void destroy() {
     for (MasterListener m : masterListeners) {
@@ -192,13 +186,12 @@ public HostAndPort getCurrentHostMaster() {
     return currentHostMaster;
   }
 
-  private void initPool(HostAndPort master) {
-    synchronized(initPoolLock){
+  private void initPool(final HostAndPort master) {
+    synchronized (initPoolLock) {
       if (!master.equals(currentHostMaster)) {
         currentHostMaster = master;
         if (factory == null) {
-          factory = new JedisFactory(master.getHost(), master.getPort(), connectionTimeout,
-              soTimeout, infiniteSoTimeout, user, password, database, clientName);
+          factory = new JedisFactory(master, masterClientConfig);
           initPool(poolConfig, factory);
         } else {
           factory.setHostAndPort(currentHostMaster);
@@ -207,59 +200,48 @@ private void initPool(HostAndPort master) {
           clearInternalPool();
         }
 
-        log.info("Created JedisPool to master at {}", master);
+        LOG.info("Created JedisSentinelPool to master at {}", master);
       }
     }
   }
 
-  private HostAndPort initSentinels(Set<String> sentinels, final String masterName) {
+  private HostAndPort initSentinels(Set<HostAndPort> sentinels, final String masterName) {
 
     HostAndPort master = null;
     boolean sentinelAvailable = false;
 
-    log.info("Trying to find master from available Sentinels...");
+    LOG.info("Trying to find master from available Sentinels...");
 
-    for (String sentinel : sentinels) {
-      final HostAndPort hap = HostAndPort.parseString(sentinel);
+    for (HostAndPort sentinel : sentinels) {
 
-      log.debug("Connecting to Sentinel {}", hap);
+      LOG.debug("Connecting to Sentinel {}", sentinel);
 
-      
-      try (Jedis jedis = new Jedis(hap.getHost(), hap.getPort(), sentinelConnectionTimeout, sentinelSoTimeout)){
-        if (sentinelUser != null) {
-          jedis.auth(sentinelUser, sentinelPassword);
-        } else if (sentinelPassword != null) {
-          jedis.auth(sentinelPassword);
-        }
-        if (sentinelClientName != null) {
-          jedis.clientSetname(sentinelClientName);
-        }
+      try (Jedis jedis = new Jedis(sentinel, sentinelClientConfig)) {
 
         List<String> masterAddr = jedis.sentinelGetMasterAddrByName(masterName);
 
         // connected to sentinel...
         sentinelAvailable = true;
 
         if (masterAddr == null || masterAddr.size() != 2) {
-          log.warn("Can not get master addr, master name: {}. Sentinel: {}", masterName, hap);
+          LOG.warn("Can not get master addr, master name: {}. Sentinel: {}", masterName, sentinel);
           continue;
         }
 
         master = toHostAndPort(masterAddr);
-        log.debug("Found Redis master at {}", master);
+        LOG.debug("Found Redis master at {}", master);
         break;
       } catch (JedisException e) {
         // resolves #1036, it should handle JedisException there's another chance
         // of raising JedisDataException
-        log.warn(
-          "Cannot get master address from sentinel running @ {}. Reason: {}. Trying next one.", hap, e);
+        LOG.warn(
+          "Cannot get master address from sentinel running @ {}. Reason: {}. Trying next one.", sentinel, e);
       }
     }
 
     if (master == null) {
       if (sentinelAvailable) {
-        // can connect to sentinel, but master name seems to not
-        // monitored
+        // can connect to sentinel, but master name seems to not monitored
         throw new JedisException("Can connect to sentinel, but " + masterName
             + " seems to be not monitored...");
       } else {
@@ -268,11 +250,11 @@ private HostAndPort initSentinels(Set<String> sentinels, final String masterName
       }
     }
 
-    log.info("Redis master running at {}, starting Sentinel listeners...", master);
+    LOG.info("Redis master running at {}, starting Sentinel listeners...", master);
+
+    for (HostAndPort sentinel : sentinels) {
 
-    for (String sentinel : sentinels) {
-      final HostAndPort hap = HostAndPort.parseString(sentinel);
-      MasterListener masterListener = new MasterListener(masterName, hap.getHost(), hap.getPort());
+      MasterListener masterListener = new MasterListener(masterName, sentinel.getHost(), sentinel.getPort());
       // whether MasterListener threads are alive or not, process can be stopped
       masterListener.setDaemon(true);
       masterListeners.add(masterListener);
@@ -317,7 +299,7 @@ public void returnResource(final Jedis resource) {
         returnResourceObject(resource);
       } catch (Exception e) {
         returnBrokenResource(resource);
-        log.debug("Resource is returned to the pool as broken", e);
+        LOG.debug("Resource is returned to the pool as broken", e);
       }
     }
   }
@@ -360,28 +342,21 @@ public void run() {
             break;
           }
           
-          j = new Jedis(host, port, sentinelConnectionTimeout, sentinelSoTimeout);
-          if (sentinelUser != null) {
-            j.auth(sentinelUser, sentinelPassword);
-          } else if (sentinelPassword != null) {
-            j.auth(sentinelPassword);
-          }
-          if (sentinelClientName != null) {
-            j.clientSetname(sentinelClientName);
-          }
+          final HostAndPort hostPort = new HostAndPort(host, port);
+          j = new Jedis(hostPort, sentinelClientConfig);
 
           // code for active refresh
           List<String> masterAddr = j.sentinelGetMasterAddrByName(masterName);
           if (masterAddr == null || masterAddr.size() != 2) {
-            log.warn("Can not get master addr, master name: {}. Sentinel: {}:{}.", masterName, host, port);
+            LOG.warn("Can not get master addr, master name: {}. Sentinel: {}.", masterName, hostPort);
           } else {
             initPool(toHostAndPort(masterAddr));
           }
 
           j.subscribe(new JedisPubSub() {
             @Override
             public void onMessage(String channel, String message) {
-              log.debug("Sentinel {}:{} published: {}.", host, port, message);
+              LOG.debug("Sentinel {} published: {}.", hostPort, message);
 
               String[] switchMasterMsg = message.split(" ");
 
@@ -390,31 +365,30 @@ public void onMessage(String channel, String message) {
                 if (masterName.equals(switchMasterMsg[0])) {
                   initPool(toHostAndPort(Arrays.asList(switchMasterMsg[3], switchMasterMsg[4])));
                 } else {
-                  log.debug(
+                  LOG.debug(
                     "Ignoring message on +switch-master for master name {}, our master name is {}",
                     switchMasterMsg[0], masterName);
                 }
 
               } else {
-                log.error(
-                  "Invalid message received on Sentinel {}:{} on channel +switch-master: {}", host,
-                  port, message);
+                LOG.error("Invalid message received on Sentinel {} on channel +switch-master: {}",
+                    hostPort, message);
               }
             }
           }, "+switch-master");
 
         } catch (JedisException e) {
 
           if (running.get()) {
-            log.error("Lost connection to Sentinel at {}:{}. Sleeping 5000ms and retrying.", host,
+            LOG.error("Lost connection to Sentinel at {}:{}. Sleeping 5000ms and retrying.", host,
               port, e);
             try {
               Thread.sleep(subscribeRetryWaitTimeMillis);
             } catch (InterruptedException e1) {
-              log.error("Sleep interrupted: ", e1);
+              LOG.error("Sleep interrupted: ", e1);
             }
           } else {
-            log.debug("Unsubscribing from Sentinel at {}:{}", host, port);
+            LOG.debug("Unsubscribing from Sentinel at {}:{}", host, port);
           }
         } finally {
           if (j != null) {
@@ -426,14 +400,14 @@ public void onMessage(String channel, String message) {
 
     public void shutdown() {
       try {
-        log.debug("Shutting down listener on {}:{}", host, port);
+        LOG.debug("Shutting down listener on {}:{}", host, port);
         running.set(false);
         // This isn't good, the Jedis object is not thread safe
         if (j != null) {
-          j.disconnect();
+          j.close();
         }
       } catch (Exception e) {
-        log.error("Caught exception while shutting down: ", e);
+        LOG.error("Caught exception while shutting down: ", e);
       }
     }
   }
diff --git a/src/test/java/redis/clients/jedis/tests/SSLJedisSentinelPoolTest.java b/src/test/java/redis/clients/jedis/tests/SSLJedisSentinelPoolTest.java
