Skip to content

Support Sentinel with TLS (II) #2403

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 9 commits into from
Mar 31, 2021
Merged

Support Sentinel with TLS (II) #2403

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
6cfa72f
Support Sentinel with TLS
sazzad16 Mar 7, 2021
95f68a7
Merge branch 'master' into sentinel-tls-3
gkorland Mar 10, 2021
a30e163
change config variables to protected
sazzad16 Mar 10, 2021
0c7bfea
use method reference and avoid parallel stream
sazzad16 Mar 24, 2021
3a7feac
Merge remote-tracking branch 'redis/master' into sentinel-tls-3
sazzad16 Mar 24, 2021
8f79673
remove more deprecated variables
sazzad16 Mar 25, 2021
73f5979
Merge branch 'master' into sentinel-tls-3
sazzad16 Mar 25, 2021
71d9ac2
Merge branch 'master' into sentinel-tls-3
sazzad16 Mar 29, 2021
f6aa3f5
Merge branch 'master' into sentinel-tls-3
sazzad16 Mar 31, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 18 additions & 0 deletions Makefile
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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
Expand Down
142 changes: 58 additions & 84 deletions src/main/java/redis/clients/jedis/JedisSentinelPool.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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;
Expand All @@ -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<Jedis> poolConfig;

protected final int connectionTimeout;
protected final int soTimeout;
protected final int infiniteSoTimeout;
protected final JedisClientConfig masterClientConfig;

protected final String user;
protected final String password;
protected final int database;
protected final String clientName;
protected 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<>();
protected final Collection<MasterListener> masterListeners = new LinkedList<>();

private volatile JedisFactory factory;
private volatile HostAndPort currentHostMaster;
Expand Down Expand Up @@ -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().withConnectionTimeoutMillis(connectionTimeout)
.withSoTimeoutMillis(soTimeout).withInfiniteSoTimeoutMillis(infiniteSoTimeout).withUser(user)
.withPassword(password).withDatabse(database).withClientName(clientName).build(),
DefaultJedisClientConfig.builder().withConnectionTimeoutMillis(sentinelConnectionTimeout)
.withSoTimeoutMillis(sentinelSoTimeout).withUser(sentinelUser).withPassword(sentinelPassword)
.withClientName(sentinelClientName).build()
);
}

public JedisSentinelPool(String masterName, Set<HostAndPort> sentinels,
final GenericObjectPoolConfig<Jedis> 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) {
Expand All @@ -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);
Expand All @@ -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 {
Expand All @@ -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);
Expand Down Expand Up @@ -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);
}
}
}
Expand Down Expand Up @@ -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(" ");

Expand All @@ -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) {
Expand All @@ -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);
}
}
}
Expand Down
Loading