rik/jlibredis
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

make sure that JedisSentinelPool use timeout, password and db

Browse Source
This commit is contained in:
Jonathan Leibiusky
2013-09-15 15:33:22 -03:00
parent 4d9090827f
commit ad58f816b6
3 changed files with 332 additions and 263 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
Makefile
View File

@@ -21,6 +21,7 @@ endef
define REDIS3_CONF define REDIS3_CONF
daemonize yes daemonize yes
port 6381 port 6381
requirepass foobared
pidfile /tmp/redis3.pid pidfile /tmp/redis3.pid
logfile /tmp/redis3.log logfile /tmp/redis3.log
save "" save ""
@@ -30,6 +31,8 @@ endef
define REDIS4_CONF define REDIS4_CONF
daemonize yes daemonize yes
port 6382 port 6382
requirepass foobared
masterauth foobared
pidfile /tmp/redis4.pid pidfile /tmp/redis4.pid
logfile /tmp/redis4.log logfile /tmp/redis4.log
save "" save ""
@@ -39,7 +42,7 @@ endef
define REDIS_SENTINEL1 define REDIS_SENTINEL1
port 26379 port 26379
daemonize yes daemonize yes
sentinel monitor mymaster 127.0.0.1 6379 1 sentinel monitor mymaster 127.0.0.1 6381 2
sentinel auth-pass mymaster foobared sentinel auth-pass mymaster foobared
sentinel down-after-milliseconds mymaster 3000 sentinel down-after-milliseconds mymaster 3000
sentinel failover-timeout mymaster 900000 sentinel failover-timeout mymaster 900000
@@ -53,6 +56,7 @@ define REDIS_SENTINEL2
port 26380 port 26380
daemonize yes daemonize yes
sentinel monitor mymaster 127.0.0.1 6381 2 sentinel monitor mymaster 127.0.0.1 6381 2
sentinel auth-pass mymaster foobared
sentinel down-after-milliseconds mymaster 3000 sentinel down-after-milliseconds mymaster 3000
sentinel can-failover mymaster yes sentinel can-failover mymaster yes
sentinel parallel-syncs mymaster 1 sentinel parallel-syncs mymaster 1
@@ -65,6 +69,7 @@ define REDIS_SENTINEL3
port 26381 port 26381
daemonize yes daemonize yes
sentinel monitor mymaster 127.0.0.1 6381 2 sentinel monitor mymaster 127.0.0.1 6381 2
sentinel auth-pass mymaster foobared
sentinel down-after-milliseconds mymaster 3000 sentinel down-after-milliseconds mymaster 3000
sentinel can-failover mymaster yes sentinel can-failover mymaster yes
sentinel parallel-syncs mymaster 1 sentinel parallel-syncs mymaster 1

489
src/main/java/redis/clients/jedis/JedisSentinelPool.java
View File

@@ -14,246 +14,299 @@ import redis.clients.util.Pool;
public class JedisSentinelPool extends Pool<Jedis> { public class JedisSentinelPool extends Pool<Jedis> {
protected Config poolConfig; protected Config poolConfig;
protected int timeout = Protocol.DEFAULT_TIMEOUT; protected int timeout = Protocol.DEFAULT_TIMEOUT;
protected String password; protected String password;
protected int database = Protocol.DEFAULT_DATABASE; protected int database = Protocol.DEFAULT_DATABASE;
protected Set<MasterListener> masterListeners = new HashSet<MasterListener>(); protected Set<MasterListener> masterListeners = new HashSet<MasterListener>();
protected Logger log = Logger.getLogger(getClass().getName()); protected Logger log = Logger.getLogger(getClass().getName());
public JedisSentinelPool(String masterName, Set<String> sentinels, final Config poolConfig) { public JedisSentinelPool(String masterName, Set<String> sentinels,
this(masterName, sentinels, poolConfig, Protocol.DEFAULT_TIMEOUT, null, Protocol.DEFAULT_DATABASE); final Config poolConfig) {
this(masterName, sentinels, poolConfig, Protocol.DEFAULT_TIMEOUT, null,
Protocol.DEFAULT_DATABASE);
} }
public JedisSentinelPool(String masterName, Set<String> sentinels) { public JedisSentinelPool(String masterName, Set<String> sentinels) {
this(masterName, sentinels, new Config(), Protocol.DEFAULT_TIMEOUT, null, Protocol.DEFAULT_DATABASE); this(masterName, sentinels, new Config(), Protocol.DEFAULT_TIMEOUT,
null, Protocol.DEFAULT_DATABASE);
} }
public JedisSentinelPool(String masterName, Set<String> sentinels, String password) { public JedisSentinelPool(String masterName, Set<String> sentinels,
this(masterName, sentinels, new Config(), Protocol.DEFAULT_TIMEOUT, password); String password) {
} this(masterName, sentinels, new Config(), Protocol.DEFAULT_TIMEOUT,
password);
public JedisSentinelPool(String masterName, Set<String> sentinels, final Config poolConfig, int timeout, final String password) {
this(masterName, sentinels, poolConfig, timeout, password, Protocol.DEFAULT_DATABASE);
}
public JedisSentinelPool(String masterName, Set<String> sentinels, final Config poolConfig, final int timeout) {
this(masterName, sentinels, poolConfig, timeout, null, Protocol.DEFAULT_DATABASE);
}
public JedisSentinelPool(String masterName, Set<String> sentinels, final Config poolConfig, final String password) {
this(masterName, sentinels, poolConfig, Protocol.DEFAULT_TIMEOUT, password);
} }
public JedisSentinelPool(String masterName, Set<String> sentinels, final Config poolConfig, int timeout, final String password, public JedisSentinelPool(String masterName, Set<String> sentinels,
final int database) { final Config poolConfig, int timeout, final String password) {
this.poolConfig = poolConfig; this(masterName, sentinels, poolConfig, timeout, password,
this.timeout = timeout; Protocol.DEFAULT_DATABASE);
this.password = password; }
this.database = database;
HostAndPort master = initSentinels(sentinels, masterName); public JedisSentinelPool(String masterName, Set<String> sentinels,
initPool(master); final Config poolConfig, final int timeout) {
this(masterName, sentinels, poolConfig, timeout, null,
Protocol.DEFAULT_DATABASE);
}
public JedisSentinelPool(String masterName, Set<String> sentinels,
final Config poolConfig, final String password) {
this(masterName, sentinels, poolConfig, Protocol.DEFAULT_TIMEOUT,
password);
}
public JedisSentinelPool(String masterName, Set<String> sentinels,
final Config poolConfig, int timeout, final String password,
final int database) {
this.poolConfig = poolConfig;
this.timeout = timeout;
this.password = password;
this.database = database;
HostAndPort master = initSentinels(sentinels, masterName);
initPool(master);
} }
public void returnBrokenResource(final BinaryJedis resource) { public void returnBrokenResource(final BinaryJedis resource) {
returnBrokenResourceObject(resource); returnBrokenResourceObject(resource);
} }
public void returnResource(final BinaryJedis resource) { public void returnResource(final BinaryJedis resource) {
returnResourceObject(resource); returnResourceObject(resource);
} }
private class HostAndPort {
String host;
int port;
@Override private class HostAndPort {
public boolean equals(Object obj) { String host;
if (obj instanceof HostAndPort) { int port;
HostAndPort hp = (HostAndPort) obj;
return port == hp.port && host.equals(hp.host);
}
return false;
}
@Override
public String toString() {
return host + ":" + port;
}
}
private volatile HostAndPort currentHostMaster;
public void destroy() { @Override
for (MasterListener m : masterListeners) { public boolean equals(Object obj) {
m.shutdown(); if (obj instanceof HostAndPort) {
} HostAndPort hp = (HostAndPort) obj;
return port == hp.port && host.equals(hp.host);
super.destroy(); }
} return false;
}
public HostAndPort getCurrentHostMaster() { @Override
return currentHostMaster; public String toString() {
} return host + ":" + port;
}
}
private void initPool(HostAndPort master) { private volatile HostAndPort currentHostMaster;
if (!master.equals(currentHostMaster)) {
currentHostMaster = master;
log.info("Created JedisPool to master at " + master);
initPool(poolConfig, new JedisFactory(master.host, master.port, timeout, password, database));
}
}
private HostAndPort initSentinels(Set<String> sentinels, final String masterName) { public void destroy() {
for (MasterListener m : masterListeners) {
HostAndPort master = null; m.shutdown();
boolean running = true; }
outer: while (running) { super.destroy();
}
log.info("Trying to find master from available Sentinels...");
public HostAndPort getCurrentHostMaster() {
for (String sentinel : sentinels) { return currentHostMaster;
}
final HostAndPort hap = toHostAndPort(Arrays.asList(sentinel.split(":")));
private void initPool(HostAndPort master) {
log.fine("Connecting to Sentinel " + hap); if (!master.equals(currentHostMaster)) {
currentHostMaster = master;
try { log.info("Created JedisPool to master at " + master);
Jedis jedis = new Jedis(hap.host, hap.port); initPool(poolConfig, new JedisFactory(master.host, master.port,
timeout, password, database));
if (master == null) { }
master = toHostAndPort(jedis.sentinelGetMasterAddrByName(masterName)); }
log.fine("Found Redis master at " + master);
jedis.disconnect(); private HostAndPort initSentinels(Set<String> sentinels,
break outer; final String masterName) {
}
} catch (JedisConnectionException e) { HostAndPort master = null;
log.warning("Cannot connect to sentinel running @ " + hap + ". Trying next one."); boolean running = true;
}
} outer: while (running) {
try { log.info("Trying to find master from available Sentinels...");
log.severe("All sentinels down, cannot determine where is " + masterName + " master is running... sleeping 1000ms.");
Thread.sleep(1000); for (String sentinel : sentinels) {
} catch (InterruptedException e) {
e.printStackTrace(); final HostAndPort hap = toHostAndPort(Arrays.asList(sentinel
} .split(":")));
}
log.fine("Connecting to Sentinel " + hap);
log.info("Redis master running at " + master + ", starting Sentinel listeners...");
try {
for (String sentinel : sentinels) { Jedis jedis = new Jedis(hap.host, hap.port);
final HostAndPort hap = toHostAndPort(Arrays.asList(sentinel.split(":")));
MasterListener masterListener = new MasterListener(masterName, hap.host, hap.port); if (master == null) {
masterListeners.add(masterListener); master = toHostAndPort(jedis
masterListener.start(); .sentinelGetMasterAddrByName(masterName));
} log.fine("Found Redis master at " + master);
jedis.disconnect();
return master; break outer;
} }
} catch (JedisConnectionException e) {
private HostAndPort toHostAndPort(List<String> getMasterAddrByNameResult) { log.warning("Cannot connect to sentinel running @ " + hap
final HostAndPort hap = new HostAndPort(); + ". Trying next one.");
hap.host = getMasterAddrByNameResult.get(0); }
hap.port = Integer.parseInt(getMasterAddrByNameResult.get(1)); }
return hap;
} try {
log.severe("All sentinels down, cannot determine where is "
protected class JedisPubSubAdapter extends JedisPubSub { + masterName + " master is running... sleeping 1000ms.");
@Override Thread.sleep(1000);
public void onMessage(String channel, String message) {} } catch (InterruptedException e) {
@Override e.printStackTrace();
public void onPMessage(String pattern, String channel, String message) {} }
@Override }
public void onPSubscribe(String pattern, int subscribedChannels) {}
@Override log.info("Redis master running at " + master
public void onPUnsubscribe(String pattern, int subscribedChannels) {} + ", starting Sentinel listeners...");
@Override
public void onSubscribe(String channel, int subscribedChannels) {} for (String sentinel : sentinels) {
@Override final HostAndPort hap = toHostAndPort(Arrays.asList(sentinel
public void onUnsubscribe(String channel, int subscribedChannels) {} .split(":")));
} MasterListener masterListener = new MasterListener(masterName,
hap.host, hap.port);
protected class MasterListener extends Thread { masterListeners.add(masterListener);
masterListener.start();
protected String masterName; }
protected String host;
protected int port; return master;
protected long subscribeRetryWaitTimeMillis = 5000; }
protected Jedis j;
protected AtomicBoolean running = new AtomicBoolean(false); private HostAndPort toHostAndPort(List<String> getMasterAddrByNameResult) {
final HostAndPort hap = new HostAndPort();
protected MasterListener() {} hap.host = getMasterAddrByNameResult.get(0);
hap.port = Integer.parseInt(getMasterAddrByNameResult.get(1));
public MasterListener(String masterName, String host, int port) { return hap;
this.masterName = masterName; }
this.host = host;
this.port = port; protected class JedisPubSubAdapter extends JedisPubSub {
} @Override
public void onMessage(String channel, String message) {
public MasterListener(String masterName, String host, int port, long subscribeRetryWaitTimeMillis) { }
this(masterName, host, port);
this.subscribeRetryWaitTimeMillis = subscribeRetryWaitTimeMillis; @Override
} public void onPMessage(String pattern, String channel, String message) {
}
public void run() {
@Override
running.set(true); public void onPSubscribe(String pattern, int subscribedChannels) {
}
while (running.get()) {
@Override
j = new Jedis(host, port); public void onPUnsubscribe(String pattern, int subscribedChannels) {
}
try {
j.subscribe(new JedisPubSubAdapter() { @Override
@Override public void onSubscribe(String channel, int subscribedChannels) {
public void onMessage(String channel, String message) { }
log.fine("Sentinel " + host + ":" + port + " published: " + message + "."); @Override
public void onUnsubscribe(String channel, int subscribedChannels) {
String[] switchMasterMsg = message.split(" "); }
}
if (switchMasterMsg.length > 3) {
protected class MasterListener extends Thread {
if (masterName.equals(switchMasterMsg[0])) {
initPool(toHostAndPort(Arrays.asList(switchMasterMsg[3], switchMasterMsg[4]))); protected String masterName;
} else { protected String host;
log.fine("Ignoring message on +switch-master for master name " + switchMasterMsg[0] + ", our master name is " + masterName); protected int port;
} protected long subscribeRetryWaitTimeMillis = 5000;
protected Jedis j;
} else { protected AtomicBoolean running = new AtomicBoolean(false);
log.severe("Invalid message received on Sentinel " + host + ":" + port + " on channel +switch-master: " + message);
} protected MasterListener() {
} }
}, "+switch-master");
public MasterListener(String masterName, String host, int port) {
} catch (JedisConnectionException e) { this.masterName = masterName;
this.host = host;
if (running.get()) { this.port = port;
log.severe("Lost connection to Sentinel at " + host + ":" + port + ". Sleeping 5000ms and retrying."); }
try { Thread.sleep(subscribeRetryWaitTimeMillis); } catch (InterruptedException e1) { e1.printStackTrace(); }
} else { public MasterListener(String masterName, String host, int port,
log.fine("Unsubscribing from Sentinel at " + host + ":" + port); long subscribeRetryWaitTimeMillis) {
} this(masterName, host, port);
} this.subscribeRetryWaitTimeMillis = subscribeRetryWaitTimeMillis;
} }
}
public void run() {
public void shutdown() {
try { running.set(true);
log.fine("Shutting down listener on " + host + ":" + port);
running.set(false); while (running.get()) {
// This isn't good, the Jedis object is not thread safe
j.disconnect(); j = new Jedis(host, port);
} catch (Exception e) {
log.severe("Caught exception while shutting down: " + e.getMessage()); try {
} j.subscribe(new JedisPubSubAdapter() {
} @Override
} public void onMessage(String channel, String message) {
log.fine("Sentinel " + host + ":" + port
+ " published: " + message + ".");
String[] switchMasterMsg = message.split(" ");
if (switchMasterMsg.length > 3) {
if (masterName.equals(switchMasterMsg[0])) {
initPool(toHostAndPort(Arrays.asList(
switchMasterMsg[3],
switchMasterMsg[4])));
} else {
log.fine("Ignoring message on +switch-master for master name "
+ switchMasterMsg[0]
+ ", our master name is "
+ masterName);
}
} else {
log.severe("Invalid message received on Sentinel "
+ host
+ ":"
+ port
+ " on channel +switch-master: "
+ message);
}
}
}, "+switch-master");
} catch (JedisConnectionException e) {
if (running.get()) {
log.severe("Lost connection to Sentinel at " + host
+ ":" + port
+ ". Sleeping 5000ms and retrying.");
try {
Thread.sleep(subscribeRetryWaitTimeMillis);
} catch (InterruptedException e1) {
e1.printStackTrace();
}
} else {
log.fine("Unsubscribing from Sentinel at " + host + ":"
+ port);
}
}
}
}
public void shutdown() {
try {
log.fine("Shutting down listener on " + host + ":" + port);
running.set(false);
// This isn't good, the Jedis object is not thread safe
j.disconnect();
} catch (Exception e) {
log.severe("Caught exception while shutting down: "
+ e.getMessage());
}
}
}
} }

99
src/test/java/redis/clients/jedis/tests/JedisSentinelPoolTest.java
View File

@@ -3,6 +3,7 @@ package redis.clients.jedis.tests;
import java.util.HashSet; import java.util.HashSet;
import java.util.Set; import java.util.Set;
import org.apache.commons.pool.impl.GenericObjectPool.Config;
import org.junit.Before; import org.junit.Before;
import org.junit.Test; import org.junit.Test;
@@ -12,53 +13,63 @@ import redis.clients.jedis.JedisSentinelPool;
import redis.clients.jedis.tests.HostAndPortUtil.HostAndPort; import redis.clients.jedis.tests.HostAndPortUtil.HostAndPort;
public class JedisSentinelPoolTest extends JedisTestBase { public class JedisSentinelPoolTest extends JedisTestBase {
protected static HostAndPort master = HostAndPortUtil.getRedisServers().get(2);
protected static HostAndPort slave1 = HostAndPortUtil.getRedisServers().get(3);
protected static HostAndPort sentinel1 = HostAndPortUtil.getSentinelServers().get(1);
protected static HostAndPort sentinel2 = HostAndPortUtil.getSentinelServers().get(2);
protected static Jedis masterJedis; protected static HostAndPort master = HostAndPortUtil.getRedisServers()
protected static Jedis slaveJedis1; .get(2);
protected static Jedis sentinelJedis1; protected static HostAndPort slave1 = HostAndPortUtil.getRedisServers()
.get(3);
protected Set<String> sentinels = new HashSet<String>(); protected static HostAndPort sentinel1 = HostAndPortUtil
.getSentinelServers().get(1);
@Before protected static HostAndPort sentinel2 = HostAndPortUtil
public void setUp() throws Exception { .getSentinelServers().get(2);
// set up master and slaves protected static Jedis masterJedis;
masterJedis = new Jedis(master.host, master.port); protected static Jedis slaveJedis1;
masterJedis.slaveofNoOne();
protected Set<String> sentinels = new HashSet<String>();
slaveJedis1 = new Jedis(slave1.host, slave1.port);
slaveJedis1.slaveof(master.host, master.port); @Before
public void setUp() throws Exception {
sentinelJedis1 = new Jedis(sentinel1.host, sentinel1.port);
sentinels.add(sentinel1.toString()); // set up master and slaves
sentinels.add(sentinel2.toString()); masterJedis = new Jedis(master.host, master.port);
masterJedis.auth("foobared");
// FIXME: The following allows the master/slave relationship to masterJedis.slaveofNoOne();
// be established. We can do this more elegantly.
Thread.sleep(10000); slaveJedis1 = new Jedis(slave1.host, slave1.port);
slaveJedis1.auth("foobared");
slaveJedis1.slaveof(master.host, master.port);
sentinels.add(sentinel1.toString());
sentinels.add(sentinel2.toString());
// FIXME: The following allows the master/slave relationship to
// be established. We can do this more elegantly.
Thread.sleep(10000);
} }
@Test @Test
public void segfaultMaster() throws InterruptedException { public void segfaultMaster() throws InterruptedException {
JedisSentinelPool pool = new JedisSentinelPool("mymaster", sentinels); JedisSentinelPool pool = new JedisSentinelPool("mymaster", sentinels,
new Config(), 1000, "foobared", 2);
Jedis jedis = pool.getResource();
assertEquals("PONG", jedis.ping()); Jedis jedis = pool.getResource();
assertEquals("PONG", jedis.ping());
try { masterJedis.debug(DebugParams.SEGFAULT()); } catch (Exception e) {}
try {
// wait for the sentinel to promote a master masterJedis.debug(DebugParams.SEGFAULT());
// FIXME: we can query the sentinel and sleep } catch (Exception e) {
// right until the master is promoted }
Thread.sleep(35000);
// wait for the sentinel to promote a master
jedis = pool.getResource(); // FIXME: we can query the sentinel and sleep
assertEquals("PONG", jedis.ping()); // right until the master is promoted
Thread.sleep(35000);
jedis = pool.getResource();
assertEquals("PONG", jedis.ping());
assertEquals("foobared", jedis.configGet("requirepass").get(1));
assertEquals(2, jedis.getDB().intValue());
} }
} }