You are viewing a plain text version of this content. The canonical link for it is here.
Posted to oak-commits@jackrabbit.apache.org by al...@apache.org on 2014/10/17 14:36:14 UTC
svn commit: r1632554 [2/2] - in /jackrabbit/oak/branches/1.0: ./ oak-core/
oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/segment/ oak-doc/
oak-doc/src/site/ oak-doc/src/site/markdown/coldstandby/ oak-run/
oak-run/src/main/java/org/apache/jac...
Modified: jackrabbit/oak/branches/1.0/oak-tarmk-failover/src/main/java/org/apache/jackrabbit/oak/plugins/segment/failover/server/FailoverServer.java
URL: http://svn.apache.org/viewvc/jackrabbit/oak/branches/1.0/oak-tarmk-failover/src/main/java/org/apache/jackrabbit/oak/plugins/segment/failover/server/FailoverServer.java?rev=1632554&r1=1606087&r2=1632554&view=diff
==============================================================================
--- jackrabbit/oak/branches/1.0/oak-tarmk-failover/src/main/java/org/apache/jackrabbit/oak/plugins/segment/failover/server/FailoverServer.java (original)
+++ jackrabbit/oak/branches/1.0/oak-tarmk-failover/src/main/java/org/apache/jackrabbit/oak/plugins/segment/failover/server/FailoverServer.java Fri Oct 17 12:36:13 2014
@@ -19,6 +19,7 @@
package org.apache.jackrabbit.oak.plugins.segment.failover.server;
import io.netty.bootstrap.ServerBootstrap;
+import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
@@ -27,30 +28,84 @@ import io.netty.channel.nio.NioEventLoop
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.LineBasedFrameDecoder;
+import io.netty.handler.codec.compression.SnappyFramedEncoder;
import io.netty.handler.codec.string.StringDecoder;
+import io.netty.handler.ssl.SslContext;
+import io.netty.handler.ssl.util.SelfSignedCertificate;
import io.netty.util.CharsetUtil;
import java.io.Closeable;
+import java.lang.management.ManagementFactory;
+import java.security.cert.CertificateException;
import java.util.concurrent.TimeUnit;
+import io.netty.util.concurrent.Future;
import org.apache.jackrabbit.oak.plugins.segment.SegmentStore;
+import org.apache.jackrabbit.oak.plugins.segment.failover.CommunicationObserver;
+import org.apache.jackrabbit.oak.plugins.segment.failover.jmx.FailoverStatusMBean;
import org.apache.jackrabbit.oak.plugins.segment.failover.codec.RecordIdEncoder;
import org.apache.jackrabbit.oak.plugins.segment.failover.codec.SegmentEncoder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
-public class FailoverServer implements Closeable {
+import javax.management.MBeanServer;
+import javax.management.ObjectName;
+import javax.management.StandardMBean;
+import javax.net.ssl.SSLException;
+
+public class FailoverServer implements FailoverStatusMBean, Closeable {
+
+ private static final Logger log = LoggerFactory
+ .getLogger(FailoverServer.class);
private final int port;
private final EventLoopGroup bossGroup;
private final EventLoopGroup workerGroup;
private final ServerBootstrap b;
+ private final CommunicationObserver observer;
+ private final FailoverServerHandler handler;
+ private SslContext sslContext;
+ private ChannelFuture channelFuture;
+ private boolean running;
+
+ public FailoverServer(int port, final SegmentStore store)
+ throws CertificateException, SSLException {
+ this(port, store, null, false);
+ }
+
+ public FailoverServer(int port, final SegmentStore store, boolean secure)
+ throws CertificateException, SSLException {
+ this(port, store, null, secure);
+ }
+
+ public FailoverServer(int port, final SegmentStore store, String[] allowedClientIPRanges)
+ throws CertificateException, SSLException {
+ this(port, store, allowedClientIPRanges, false);
+ }
- public FailoverServer(int port, final SegmentStore store) {
+ public FailoverServer(int port, final SegmentStore store, String[] allowedClientIPRanges, boolean secure)
+ throws CertificateException, SSLException {
this.port = port;
+ if (secure) {
+ SelfSignedCertificate ssc = new SelfSignedCertificate();
+ sslContext = SslContext.newServerContext(ssc.certificate(), ssc.privateKey());
+ }
+
+ observer = new CommunicationObserver("master");
+ handler = new FailoverServerHandler(store, observer, allowedClientIPRanges);
bossGroup = new NioEventLoopGroup(1);
workerGroup = new NioEventLoopGroup();
+ final MBeanServer jmxServer = ManagementFactory.getPlatformMBeanServer();
+ try {
+ jmxServer.registerMBean(new StandardMBean(this, FailoverStatusMBean.class), new ObjectName(this.getMBeanName()));
+ }
+ catch (Exception e) {
+ log.error("can't register failover status mbean", e);
+ }
+
b = new ServerBootstrap();
b.group(bossGroup, workerGroup);
b.channel(NioServerSocketChannel.class);
@@ -65,34 +120,110 @@ public class FailoverServer implements C
@Override
public void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline p = ch.pipeline();
+ if (sslContext != null) {
+ p.addLast(sslContext.newHandler(ch.alloc()));
+ }
p.addLast(new LineBasedFrameDecoder(8192));
p.addLast(new StringDecoder(CharsetUtil.UTF_8));
+ p.addLast(new SnappyFramedEncoder());
p.addLast(new RecordIdEncoder());
p.addLast(new SegmentEncoder());
- p.addLast(new FailoverServerHandler(store));
+ p.addLast(handler);
}
});
}
- public void start() {
- try {
- b.bind("127.0.0.1", port).sync().channel().closeFuture().sync();
- } catch (InterruptedException e) {
- close();
- }
+ public String getMBeanName() {
+ return FailoverStatusMBean.JMX_NAME + ",id=" + this.port;
}
- @Override
public void close() {
+ stop();
+ handler.state = STATUS_CLOSING;
+ observer.unregister();
+ final MBeanServer jmxServer = ManagementFactory.getPlatformMBeanServer();
+ try {
+ jmxServer.unregisterMBean(new ObjectName(this.getMBeanName()));
+ }
+ catch (Exception e) {
+ log.error("can unregister failover status mbean", e);
+ }
if (bossGroup != null && !bossGroup.isShuttingDown()) {
- bossGroup.shutdownGracefully(1, 2, TimeUnit.SECONDS)
- .syncUninterruptibly();
+ bossGroup.shutdownGracefully(1, 2, TimeUnit.SECONDS).syncUninterruptibly();
}
if (workerGroup != null && !workerGroup.isShuttingDown()) {
- workerGroup.shutdownGracefully(1, 2, TimeUnit.SECONDS)
- .syncUninterruptibly();
- ;
+ workerGroup.shutdownGracefully(1, 2, TimeUnit.SECONDS).syncUninterruptibly();
}
+ handler.state = STATUS_CLOSED;
}
+ private void start(boolean wait) {
+ if (running) return;
+
+ running = true;
+ this.handler.state = STATUS_STARTING;
+
+ final Thread close = new Thread() {
+ @Override
+ public void run() {
+ try {
+ running = true;
+ channelFuture.sync().channel().closeFuture().sync();
+ } catch (InterruptedException e) {
+ FailoverServer.this.stop();
+ }
+ }
+ };
+ Future<?> startup = bossGroup.submit(new Runnable() {
+ @Override
+ public void run() {
+ //netty 4.0.20 has a race condition issue with
+ //asynchronous channel registration. As a workaround
+ //we bind asynchronously from the boss event group to make
+ //the channel registration synchronous.
+ //Note that now this method will return immediately.
+ channelFuture = b.bind(port);
+ close.start();
+ }
+ });
+ if (!startup.awaitUninterruptibly(10000)) {
+ log.error("FailoverServer failed to start within 10 seconds and will be canceled");
+ startup.cancel(true);
+ } else if (wait) {
+ try {
+ close.join();
+ } catch (InterruptedException ignored) {}
+ }
+ }
+
+ public void startAndWait() {
+ start(true);
+ }
+
+ @Override
+ public void start() {
+ start(false);
+ }
+
+ @Override
+ public String getMode() {
+ return "master";
+ }
+
+ @Override
+ public boolean isRunning() { return running; }
+
+ @Override
+ public void stop() {
+ if (running) {
+ running = false;
+ this.handler.state = STATUS_STOPPED;
+ channelFuture.channel().disconnect();
+ }
+ }
+
+ @Override
+ public String getStatus() {
+ return handler == null ? STATUS_INITIALIZING : handler.state;
+ }
}
Modified: jackrabbit/oak/branches/1.0/oak-tarmk-failover/src/main/java/org/apache/jackrabbit/oak/plugins/segment/failover/server/FailoverServerHandler.java
URL: http://svn.apache.org/viewvc/jackrabbit/oak/branches/1.0/oak-tarmk-failover/src/main/java/org/apache/jackrabbit/oak/plugins/segment/failover/server/FailoverServerHandler.java?rev=1632554&r1=1606087&r2=1632554&view=diff
==============================================================================
--- jackrabbit/oak/branches/1.0/oak-tarmk-failover/src/main/java/org/apache/jackrabbit/oak/plugins/segment/failover/server/FailoverServerHandler.java (original)
+++ jackrabbit/oak/branches/1.0/oak-tarmk-failover/src/main/java/org/apache/jackrabbit/oak/plugins/segment/failover/server/FailoverServerHandler.java Fri Oct 17 12:36:13 2014
@@ -18,6 +18,9 @@
*/
package org.apache.jackrabbit.oak.plugins.segment.failover.server;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.UnknownHostException;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
@@ -30,6 +33,7 @@ import org.apache.jackrabbit.oak.plugins
import org.apache.jackrabbit.oak.plugins.segment.Segment;
import org.apache.jackrabbit.oak.plugins.segment.SegmentId;
import org.apache.jackrabbit.oak.plugins.segment.SegmentStore;
+import org.apache.jackrabbit.oak.plugins.segment.failover.CommunicationObserver;
import org.apache.jackrabbit.oak.plugins.segment.failover.codec.Messages;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -41,9 +45,14 @@ public class FailoverServerHandler exten
.getLogger(FailoverServerHandler.class);
private final SegmentStore store;
+ private final CommunicationObserver observer;
+ private final String[] allowedIPRanges;
+ public String state;
- public FailoverServerHandler(SegmentStore store) {
+ public FailoverServerHandler(SegmentStore store, CommunicationObserver observer, String[] allowedIPRanges) {
this.store = store;
+ this.observer = observer;
+ this.allowedIPRanges = allowedIPRanges;
}
private RecordId headId() {
@@ -53,49 +62,120 @@ public class FailoverServerHandler exten
return null;
}
- @Override
- public void channelRead0(ChannelHandlerContext ctx, String request)
- throws Exception {
- if (Messages.GET_HEAD.equalsIgnoreCase(request)) {
- RecordId r = headId();
- if (r != null) {
- ctx.writeAndFlush(r);
- } else {
- ctx.writeAndFlush(Unpooled.EMPTY_BUFFER);
+ private static long ipToLong(InetAddress ip) {
+ byte[] octets = ip.getAddress();
+ long result = 0;
+ for (byte octet : octets) {
+ result <<= 8;
+ result |= octet & 0xff;
+ }
+ return result;
+ }
+
+ private boolean clientAllowed(InetSocketAddress client) {
+ if (this.allowedIPRanges != null && this.allowedIPRanges.length > 0) {
+ for (String s : this.allowedIPRanges) {
+ try {
+ if (ipToLong(InetAddress.getByName(s)) == ipToLong(client.getAddress())) {
+ return true;
+ }
+ }
+ catch (UnknownHostException ignored) { /* it's an ip range */ }
+ int i = s.indexOf('-');
+ if (i > 0) {
+ try {
+ long startIPRange = ipToLong(InetAddress.getByName(s.substring(0, i).trim()));
+ long endIPRange = ipToLong(InetAddress.getByName(s.substring(i + 1).trim()));
+ long ipl = ipToLong(client.getAddress());
+ if (startIPRange <= ipl && ipl <= endIPRange) return true;
+ }
+ catch (Exception e) {
+ log.warn("invalid IP-range format: " + s);
+ }
+ }
}
+ return false;
+ }
+ return true;
+ }
- } else if (request.startsWith(Messages.GET_SEGMENT)) {
- String sid = request.substring(Messages.GET_SEGMENT.length());
- log.debug("request segment id {}", sid);
- UUID uuid = UUID.fromString(sid);
+ @Override
+ public void channelRegistered(io.netty.channel.ChannelHandlerContext ctx) throws java.lang.Exception {
+ state = "channel registered";
+ super.channelRegistered(ctx);
+ }
- Segment s = null;
+ @Override
+ public void channelActive(io.netty.channel.ChannelHandlerContext ctx) throws java.lang.Exception {
+ state = "channel active";
+ super.channelActive(ctx);
+ }
- for (int i = 0; i < 3; i++) {
- try {
- s = store.readSegment(new SegmentId(store.getTracker(),
- uuid.getMostSignificantBits(), uuid
- .getLeastSignificantBits()));
- } catch (IllegalStateException e) {
- // segment not found
- log.warn(e.getMessage());
+ @Override
+ public void channelInactive(io.netty.channel.ChannelHandlerContext ctx) throws java.lang.Exception {
+ state = "channel inactive";
+ super.channelInactive(ctx);
+ }
+
+ @Override
+ public void channelUnregistered(io.netty.channel.ChannelHandlerContext ctx) throws java.lang.Exception {
+ state = "channel unregistered";
+ super.channelUnregistered(ctx);
+ }
+
+ @Override
+ public void channelRead0(ChannelHandlerContext ctx, String payload)
+ throws Exception {
+ state = "got message";
+
+ String request = Messages.extractMessageFrom(payload);
+ InetSocketAddress client = (InetSocketAddress)ctx.channel().remoteAddress();
+
+ if (!clientAllowed(client)) {
+ log.warn("Got request from client " + client + " which is not in the allowed ip ranges! Request will be ignored.");
+ }
+ else {
+ String clientID = Messages.extractClientFrom(payload);
+ observer.gotMessageFrom(clientID, request, client);
+ if (Messages.GET_HEAD.equalsIgnoreCase(request)) {
+ RecordId r = headId();
+ if (r != null) {
+ ctx.writeAndFlush(r);
+ log.debug("returning from head request");
+ return;
}
- if (s != null) {
- break;
- } else {
- TimeUnit.MILLISECONDS.sleep(500);
+ } else if (request.startsWith(Messages.GET_SEGMENT)) {
+ String sid = request.substring(Messages.GET_SEGMENT.length());
+ log.debug("request segment id {}", sid);
+ UUID uuid = UUID.fromString(sid);
+
+ Segment s = null;
+
+ for (int i = 0; i < 10; i++) {
+ try {
+ s = store.readSegment(new SegmentId(store.getTracker(),
+ uuid.getMostSignificantBits(), uuid
+ .getLeastSignificantBits()));
+ } catch (IllegalStateException e) {
+ // segment not found
+ log.info("waiting for segment. Got exception: " + e.getMessage());
+ TimeUnit.MILLISECONDS.sleep(1000);
+ }
+ if (s != null) break;
}
- }
- if (s != null) {
- ctx.writeAndFlush(s);
+ if (s != null) {
+ log.info("sending segment" + sid + " to " + client);
+ ctx.writeAndFlush(s);
+ observer.didSendSegmentBytes(clientID, s.size());
+ log.debug("master returns from segment request");
+ return;
+ }
} else {
- ctx.writeAndFlush(Unpooled.EMPTY_BUFFER);
+ log.warn("Unknown request {}, ignoring.", request);
}
- } else {
- log.warn("Unknown request {}, ignoring.", request);
- ctx.writeAndFlush(Unpooled.EMPTY_BUFFER);
}
+ ctx.writeAndFlush(Unpooled.EMPTY_BUFFER);
}
@Override
@@ -105,7 +185,8 @@ public class FailoverServerHandler exten
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
+ state = "exception occurred: " + cause.getMessage();
log.error(cause.getMessage(), cause);
- ctx.close();
+ ctx.fireExceptionCaught(cause);
}
}
Modified: jackrabbit/oak/branches/1.0/oak-tarmk-failover/src/main/java/org/apache/jackrabbit/oak/plugins/segment/failover/store/FailoverStore.java
URL: http://svn.apache.org/viewvc/jackrabbit/oak/branches/1.0/oak-tarmk-failover/src/main/java/org/apache/jackrabbit/oak/plugins/segment/failover/store/FailoverStore.java?rev=1632554&r1=1606087&r2=1632554&view=diff
==============================================================================
--- jackrabbit/oak/branches/1.0/oak-tarmk-failover/src/main/java/org/apache/jackrabbit/oak/plugins/segment/failover/store/FailoverStore.java (original)
+++ jackrabbit/oak/branches/1.0/oak-tarmk-failover/src/main/java/org/apache/jackrabbit/oak/plugins/segment/failover/store/FailoverStore.java Fri Oct 17 12:36:13 2014
@@ -30,8 +30,11 @@ import org.apache.jackrabbit.oak.plugins
import org.apache.jackrabbit.oak.plugins.segment.SegmentStore;
import org.apache.jackrabbit.oak.plugins.segment.SegmentTracker;
import org.apache.jackrabbit.oak.spi.blob.BlobStore;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
public class FailoverStore implements SegmentStore {
+ private static final Logger log = LoggerFactory.getLogger(FailoverStore.class);
private final SegmentTracker tracker = new SegmentTracker(this);
@@ -65,7 +68,7 @@ public class FailoverStore implements Se
@Override
public Segment readSegment(SegmentId sid) {
-
+ log.info("shall read segment " + sid);
Deque<SegmentId> ids = new ArrayDeque<SegmentId>();
ids.offer(sid);
int err = 0;
@@ -74,8 +77,10 @@ public class FailoverStore implements Se
while (!ids.isEmpty()) {
SegmentId id = ids.remove();
if (!seen.contains(id) && !delegate.containsSegment(id)) {
- Segment s = loader.readSegment(id);
+ log.debug("trying to read segment " + id);
+ Segment s = loader.readSegment(id.toString());
if (s != null) {
+ log.info("got segment " + id + " with size " + s.size());
ByteArrayOutputStream bout = new ByteArrayOutputStream(
s.size());
if (id.isDataSegmentId()) {
@@ -92,6 +97,7 @@ public class FailoverStore implements Se
ids.removeAll(seen);
err = 0;
} else {
+ log.error("could NOT read segment " + id);
if (loader.isClosed() || err == 4) {
loader.close();
throw new IllegalStateException(
@@ -105,6 +111,7 @@ public class FailoverStore implements Se
}
}
+ log.info("calling delegate to return segment " + sid);
return delegate.readSegment(sid);
}
Modified: jackrabbit/oak/branches/1.0/oak-tarmk-failover/src/main/java/org/apache/jackrabbit/oak/plugins/segment/failover/store/FailoverStoreService.java
URL: http://svn.apache.org/viewvc/jackrabbit/oak/branches/1.0/oak-tarmk-failover/src/main/java/org/apache/jackrabbit/oak/plugins/segment/failover/store/FailoverStoreService.java?rev=1632554&r1=1606087&r2=1632554&view=diff
==============================================================================
--- jackrabbit/oak/branches/1.0/oak-tarmk-failover/src/main/java/org/apache/jackrabbit/oak/plugins/segment/failover/store/FailoverStoreService.java (original)
+++ jackrabbit/oak/branches/1.0/oak-tarmk-failover/src/main/java/org/apache/jackrabbit/oak/plugins/segment/failover/store/FailoverStoreService.java Fri Oct 17 12:36:13 2014
@@ -16,15 +16,17 @@
*/
package org.apache.jackrabbit.oak.plugins.segment.failover.store;
-import static java.lang.Integer.parseInt;
import static java.lang.String.valueOf;
import static org.apache.felix.scr.annotations.ReferencePolicy.STATIC;
import static org.apache.felix.scr.annotations.ReferencePolicyOption.GREEDY;
import java.io.IOException;
+import java.security.cert.CertificateException;
import java.util.Dictionary;
import java.util.Hashtable;
+import javax.net.ssl.SSLException;
+
import org.apache.felix.scr.annotations.Activate;
import org.apache.felix.scr.annotations.Component;
import org.apache.felix.scr.annotations.ConfigurationPolicy;
@@ -32,56 +34,70 @@ import org.apache.felix.scr.annotations.
import org.apache.felix.scr.annotations.Property;
import org.apache.felix.scr.annotations.PropertyOption;
import org.apache.felix.scr.annotations.Reference;
-import org.apache.jackrabbit.oak.plugins.segment.SegmentNodeStoreService;
+import org.apache.jackrabbit.oak.commons.PropertiesUtil;
+import org.apache.jackrabbit.oak.plugins.segment.SegmentStoreProvider;
import org.apache.jackrabbit.oak.plugins.segment.SegmentStore;
import org.apache.jackrabbit.oak.plugins.segment.failover.client.FailoverClient;
import org.apache.jackrabbit.oak.plugins.segment.failover.server.FailoverServer;
-import org.apache.jackrabbit.oak.spi.state.NodeStore;
import org.osgi.framework.ServiceRegistration;
import org.osgi.service.component.ComponentContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-@Component(policy = ConfigurationPolicy.REQUIRE)
+
+@Property(name = "org.apache.sling.installer.configuration.persist", label="Persist configuration", description = "Must be always disabled to avoid storing the configuration in the repository", boolValue = false)
+@Component(metatype = true, policy = ConfigurationPolicy.REQUIRE)
public class FailoverStoreService {
private final Logger log = LoggerFactory.getLogger(getClass());
- @Property(label = "Mode", description = "TarMK Failover mode (master or slave)", options = {
- @PropertyOption(name = "master", value = "master"),
- @PropertyOption(name = "slave", value = "slave") }, value = "master")
+ private static final String MODE_MASTER = "master";
+ private static final String MODE_SLAVE = "slave";
+
+ public static final String MODE_DEFAULT = MODE_MASTER;
+ @Property(options = {
+ @PropertyOption(name = MODE_MASTER, value = MODE_MASTER),
+ @PropertyOption(name = MODE_SLAVE, value = MODE_SLAVE) },
+ value = MODE_DEFAULT)
public static final String MODE = "mode";
- @Property(label = "Port", description = "TarMK Failover port", intValue = 8023)
+ public static final int PORT_DEFAULT = 8023;
+ @Property(intValue = PORT_DEFAULT)
public static final String PORT = "port";
- @Property(label = "Master Host", description = "TarMK Failover master host (enabled for slave mode only)", value = "127.0.0.1")
+ public static final String MASTER_HOST_DEFAULT = "127.0.0.1";
+ @Property(value = MASTER_HOST_DEFAULT)
public static final String MASTER_HOST = "master.host";
- @Property(label = "Sync interval (seconds)", description = "TarMK Failover sync interval (seconds)", intValue = 5)
+ public static final int INTERVAL_DEFAULT = 5;
+ @Property(intValue = INTERVAL_DEFAULT)
public static final String INTERVAL = "interval";
- private static String MODE_MASTER = "master";
-
- private static String MODE_SLAVE = "slave";
+ public static final String[] ALLOWED_CLIENT_IP_RANGES_DEFAULT = new String[] {};
+ @Property(cardinality = Integer.MAX_VALUE)
+ public static final String ALLOWED_CLIENT_IP_RANGES = "master.allowed-client-ip-ranges";
+
+ public static final boolean SECURE_DEFAULT = false;
+ @Property(boolValue = SECURE_DEFAULT)
+ public static final String SECURE = "secure";
@Reference(policy = STATIC, policyOption = GREEDY)
- private NodeStore store = null;
+ private SegmentStoreProvider storeProvider = null;
private SegmentStore segmentStore;
private FailoverServer master = null;
private FailoverClient sync = null;
+
private ServiceRegistration syncReg = null;
@Activate
- private void activate(ComponentContext context) throws IOException {
- if (store instanceof SegmentNodeStoreService) {
- segmentStore = ((SegmentNodeStoreService) store).getSegmentStore();
+ private void activate(ComponentContext context) throws IOException, CertificateException {
+ if (storeProvider != null) {
+ segmentStore = storeProvider.getSegmentStore();
} else {
throw new IllegalArgumentException(
- "Unexpected NodeStore impl, expecting SegmentNodeStoreService, got "
- + store.getClass());
+ "Missing SegmentStoreProvider service");
}
String mode = valueOf(context.getProperties().get(MODE));
if (MODE_MASTER.equals(mode)) {
@@ -108,21 +124,24 @@ public class FailoverStoreService {
}
}
- private void bootstrapMaster(ComponentContext context) {
+ private void bootstrapMaster(ComponentContext context) throws CertificateException, SSLException {
Dictionary<?, ?> props = context.getProperties();
- int port = parseInt(valueOf(props.get(PORT)));
- master = new FailoverServer(port, segmentStore);
+ int port = PropertiesUtil.toInteger(props.get(PORT), PORT_DEFAULT);
+ String[] ranges = PropertiesUtil.toStringArray(props.get(ALLOWED_CLIENT_IP_RANGES), ALLOWED_CLIENT_IP_RANGES_DEFAULT);
+ boolean secure = PropertiesUtil.toBoolean(props.get(SECURE), SECURE_DEFAULT);
+ master = new FailoverServer(port, segmentStore, ranges, secure);
master.start();
- log.info("started failover master on port {}.", port);
+ log.info("started failover master on port {} with allowed ip ranges {}.", port, ranges);
}
- private void bootstrapSlave(ComponentContext context) {
+ private void bootstrapSlave(ComponentContext context) throws SSLException {
Dictionary<?, ?> props = context.getProperties();
- int port = parseInt(valueOf(props.get(PORT)));
- long interval = parseInt(valueOf(props.get(INTERVAL)));
- String host = valueOf(context.getProperties().get(MASTER_HOST));
+ int port = PropertiesUtil.toInteger(props.get(PORT), PORT_DEFAULT);
+ long interval = PropertiesUtil.toInteger(props.get(INTERVAL), INTERVAL_DEFAULT);
+ String host = PropertiesUtil.toString(props.get(MASTER_HOST), MASTER_HOST_DEFAULT);
+ boolean secure = PropertiesUtil.toBoolean(props.get(SECURE), SECURE_DEFAULT);
- sync = new FailoverClient(host, port, segmentStore);
+ sync = new FailoverClient(host, port, segmentStore, secure);
Dictionary<Object, Object> dictionary = new Hashtable<Object, Object>();
dictionary.put("scheduler.period", interval);
dictionary.put("scheduler.concurrent", false);
Modified: jackrabbit/oak/branches/1.0/oak-tarmk-failover/src/main/java/org/apache/jackrabbit/oak/plugins/segment/failover/store/RemoteSegmentLoader.java
URL: http://svn.apache.org/viewvc/jackrabbit/oak/branches/1.0/oak-tarmk-failover/src/main/java/org/apache/jackrabbit/oak/plugins/segment/failover/store/RemoteSegmentLoader.java?rev=1632554&r1=1606087&r2=1632554&view=diff
==============================================================================
--- jackrabbit/oak/branches/1.0/oak-tarmk-failover/src/main/java/org/apache/jackrabbit/oak/plugins/segment/failover/store/RemoteSegmentLoader.java (original)
+++ jackrabbit/oak/branches/1.0/oak-tarmk-failover/src/main/java/org/apache/jackrabbit/oak/plugins/segment/failover/store/RemoteSegmentLoader.java Fri Oct 17 12:36:13 2014
@@ -17,11 +17,10 @@
package org.apache.jackrabbit.oak.plugins.segment.failover.store;
import org.apache.jackrabbit.oak.plugins.segment.Segment;
-import org.apache.jackrabbit.oak.plugins.segment.SegmentId;
public interface RemoteSegmentLoader {
- Segment readSegment(SegmentId id);
+ Segment readSegment(String id);
void close();
Copied: jackrabbit/oak/branches/1.0/oak-tarmk-failover/src/test/java/org/apache/jackrabbit/oak/plugins/segment/NetworkErrorProxy.java (from r1622479, jackrabbit/oak/trunk/oak-tarmk-failover/src/test/java/org/apache/jackrabbit/oak/plugins/segment/NetworkErrorProxy.java)
URL: http://svn.apache.org/viewvc/jackrabbit/oak/branches/1.0/oak-tarmk-failover/src/test/java/org/apache/jackrabbit/oak/plugins/segment/NetworkErrorProxy.java?p2=jackrabbit/oak/branches/1.0/oak-tarmk-failover/src/test/java/org/apache/jackrabbit/oak/plugins/segment/NetworkErrorProxy.java&p1=jackrabbit/oak/trunk/oak-tarmk-failover/src/test/java/org/apache/jackrabbit/oak/plugins/segment/NetworkErrorProxy.java&r1=1622479&r2=1632554&rev=1632554&view=diff
==============================================================================
--- jackrabbit/oak/trunk/oak-tarmk-failover/src/test/java/org/apache/jackrabbit/oak/plugins/segment/NetworkErrorProxy.java (original)
+++ jackrabbit/oak/branches/1.0/oak-tarmk-failover/src/test/java/org/apache/jackrabbit/oak/plugins/segment/NetworkErrorProxy.java Fri Oct 17 12:36:13 2014
@@ -1,3 +1,21 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
package org.apache.jackrabbit.oak.plugins.segment;
import io.netty.bootstrap.Bootstrap;
@@ -40,6 +58,10 @@ public class NetworkErrorProxy {
this.fh.skipBytes = n;
}
+ public void flipByte(int pos) {
+ this.fh.flipPosition = pos;
+ }
+
public void run() throws Exception {
try {
ServerBootstrap b = new ServerBootstrap();
@@ -81,11 +103,13 @@ class ForwardHandler extends ChannelInbo
public long transferredBytes;
public int skipPosition;
public int skipBytes;
+ public int flipPosition;
private ChannelFuture remote;
public ForwardHandler(String host, int port) {
this.targetHost = host;
this.targetPort = port;
+ this.flipPosition = -1;
}
@Override
@@ -99,7 +123,14 @@ class ForwardHandler extends ChannelInbo
cb.handler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
- ch.pipeline().addFirst(new SwallowingHandler(c, ForwardHandler.this.skipPosition, ForwardHandler.this.skipBytes));
+ SendBackHandler sbh = new SendBackHandler(c);
+ if (ForwardHandler.this.flipPosition >= 0) {
+ sbh = new BitFlipHandler(c, ForwardHandler.this.flipPosition);
+ }
+ else if (ForwardHandler.this.skipBytes > 0) {
+ sbh = new SwallowingHandler(c, ForwardHandler.this.skipPosition, ForwardHandler.this.skipBytes);
+ }
+ ch.pipeline().addFirst(sbh);
}
});
remote = cb.connect(this.targetHost, this.targetPort).sync();
@@ -229,3 +260,32 @@ class SwallowingHandler extends SendBack
}
+class BitFlipHandler extends SendBackHandler {
+ private static final Logger log = LoggerFactory
+ .getLogger(BitFlipHandler.class);
+
+ private int startingPos;
+
+ public BitFlipHandler(ChannelHandlerContext ctx, int pos) {
+ super(ctx);
+ this.startingPos = pos;
+ }
+
+ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
+ if (msg instanceof ByteBuf) {
+ ByteBuf bb = (ByteBuf)msg;
+ log.debug("FlipHandler. Got Buffer size: " + bb.readableBytes());
+ if (this.startingPos >= 0) {
+ if (this.transferredBytes + bb.readableBytes() >= this.startingPos) {
+ int i = this.startingPos - (int)this.transferredBytes;
+ log.info("FlipHandler flips byte at offset " + (this.transferredBytes + i));
+ byte b = (byte) (bb.getByte(i) ^ 0x01);
+ bb.setByte(i, b);
+ this.startingPos = -1;
+ }
+ }
+ }
+ super.channelRead(ctx, msg);
+ }
+
+}
Modified: jackrabbit/oak/branches/1.0/oak-tarmk-failover/src/test/java/org/apache/jackrabbit/oak/plugins/segment/SegmentTestUtils.java
URL: http://svn.apache.org/viewvc/jackrabbit/oak/branches/1.0/oak-tarmk-failover/src/test/java/org/apache/jackrabbit/oak/plugins/segment/SegmentTestUtils.java?rev=1632554&r1=1606087&r2=1632554&view=diff
==============================================================================
--- jackrabbit/oak/branches/1.0/oak-tarmk-failover/src/test/java/org/apache/jackrabbit/oak/plugins/segment/SegmentTestUtils.java (original)
+++ jackrabbit/oak/branches/1.0/oak-tarmk-failover/src/test/java/org/apache/jackrabbit/oak/plugins/segment/SegmentTestUtils.java Fri Oct 17 12:36:13 2014
@@ -18,9 +18,20 @@
*/
package org.apache.jackrabbit.oak.plugins.segment;
+import org.apache.jackrabbit.oak.api.CommitFailedException;
+import org.apache.jackrabbit.oak.plugins.segment.file.FileStore;
+import org.apache.jackrabbit.oak.spi.commit.CommitInfo;
+import org.apache.jackrabbit.oak.spi.commit.EmptyHook;
+import org.apache.jackrabbit.oak.spi.state.NodeBuilder;
+import org.apache.jackrabbit.oak.spi.state.NodeStore;
+
+import static java.io.File.createTempFile;
import static org.apache.jackrabbit.oak.plugins.segment.Segment.MAX_SEGMENT_SIZE;
import static org.apache.jackrabbit.oak.plugins.segment.Segment.RECORD_ALIGN_BITS;
+import static org.junit.Assert.assertEquals;
+import java.io.File;
+import java.io.IOException;
import java.util.Random;
public class SegmentTestUtils {
@@ -38,4 +49,30 @@ public class SegmentTestUtils {
RecordId r = new RecordId(id, newValidOffset(random));
return r;
}
+
+ public static void assertEqualStores(File d1, File d2) throws IOException {
+ FileStore f1 = new FileStore(d1, 1, false);
+ FileStore f2 = new FileStore(d2, 1, false);
+ try {
+ assertEquals(f1.getHead(), f2.getHead());
+ } finally {
+ f1.close();
+ f2.close();
+ }
+ }
+
+ public static void addTestContent(NodeStore store, String child)
+ throws CommitFailedException {
+ NodeBuilder builder = store.getRoot().builder();
+ builder.child(child).setProperty("ts", System.currentTimeMillis());
+ store.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY);
+ }
+
+ public static File createTmpTargetDir(String name) throws IOException {
+ File f = createTempFile(name, "dir", new File("target"));
+ f.delete();
+ f.mkdir();
+ return f;
+ }
+
}
Copied: jackrabbit/oak/branches/1.0/oak-tarmk-failover/src/test/java/org/apache/jackrabbit/oak/plugins/segment/failover/BrokenNetworkTest.java (from r1622479, jackrabbit/oak/trunk/oak-tarmk-failover/src/test/java/org/apache/jackrabbit/oak/plugins/segment/failover/BrokenNetworkTest.java)
URL: http://svn.apache.org/viewvc/jackrabbit/oak/branches/1.0/oak-tarmk-failover/src/test/java/org/apache/jackrabbit/oak/plugins/segment/failover/BrokenNetworkTest.java?p2=jackrabbit/oak/branches/1.0/oak-tarmk-failover/src/test/java/org/apache/jackrabbit/oak/plugins/segment/failover/BrokenNetworkTest.java&p1=jackrabbit/oak/trunk/oak-tarmk-failover/src/test/java/org/apache/jackrabbit/oak/plugins/segment/failover/BrokenNetworkTest.java&r1=1622479&r2=1632554&rev=1632554&view=diff
==============================================================================
--- jackrabbit/oak/trunk/oak-tarmk-failover/src/test/java/org/apache/jackrabbit/oak/plugins/segment/failover/BrokenNetworkTest.java (original)
+++ jackrabbit/oak/branches/1.0/oak-tarmk-failover/src/test/java/org/apache/jackrabbit/oak/plugins/segment/failover/BrokenNetworkTest.java Fri Oct 17 12:36:13 2014
@@ -75,6 +75,36 @@ public class BrokenNetworkTest extends T
useProxy(true, 400, 10, true);
}
+ @Test
+ public void testProxyFlippedStartByte() throws Exception {
+ useProxy(false, 0, 0, 0, false);
+ }
+
+ @Test
+ public void testProxyFlippedStartByteSSL() throws Exception {
+ useProxy(true, 0, 0, 0, false);
+ }
+
+ @Test
+ public void testProxyFlippedIntermediateByte() throws Exception {
+ useProxy(false, 0, 0, 150, false);
+ }
+
+ @Test
+ public void testProxyFlippedIntermediateByteSSL() throws Exception {
+ useProxy(true, 0, 0, 560, false);
+ }
+
+ @Test
+ public void testProxyFlippedEndByte() throws Exception {
+ useProxy(false, 0, 0, 220, false);
+ }
+
+ @Test
+ public void testProxyFlippedEndByteSSL() throws Exception {
+ useProxy(true, 0, 0, 575, false);
+ }
+
// private helper
private void useProxy(boolean ssl) throws Exception {
@@ -82,8 +112,13 @@ public class BrokenNetworkTest extends T
}
private void useProxy(boolean ssl, int skipPosition, int skipBytes, boolean intermediateChange) throws Exception {
+ useProxy(ssl, skipPosition, skipBytes, -1, intermediateChange);
+ }
+
+ private void useProxy(boolean ssl, int skipPosition, int skipBytes, int flipPosition, boolean intermediateChange) throws Exception {
NetworkErrorProxy p = new NetworkErrorProxy(PROXY_PORT, LOCALHOST, port);
p.skipBytes(skipPosition, skipBytes);
+ p.flipByte(flipPosition);
p.run();
NodeStore store = new SegmentNodeStore(storeS);
@@ -96,7 +131,7 @@ public class BrokenNetworkTest extends T
cl.run();
try {
- if (skipBytes > 0) {
+ if (skipBytes > 0 || flipPosition >= 0) {
assertFalse("stores are not expected to be equal", storeS.getHead().equals(storeC.getHead()));
assertEquals(storeC2.getHead(), storeC.getHead());
Copied: jackrabbit/oak/branches/1.0/oak-tarmk-failover/src/test/java/org/apache/jackrabbit/oak/plugins/segment/failover/BulkTest.java (from r1622479, jackrabbit/oak/trunk/oak-tarmk-failover/src/test/java/org/apache/jackrabbit/oak/plugins/segment/failover/BulkTest.java)
URL: http://svn.apache.org/viewvc/jackrabbit/oak/branches/1.0/oak-tarmk-failover/src/test/java/org/apache/jackrabbit/oak/plugins/segment/failover/BulkTest.java?p2=jackrabbit/oak/branches/1.0/oak-tarmk-failover/src/test/java/org/apache/jackrabbit/oak/plugins/segment/failover/BulkTest.java&p1=jackrabbit/oak/trunk/oak-tarmk-failover/src/test/java/org/apache/jackrabbit/oak/plugins/segment/failover/BulkTest.java&r1=1622479&r2=1632554&rev=1632554&view=diff
==============================================================================
--- jackrabbit/oak/trunk/oak-tarmk-failover/src/test/java/org/apache/jackrabbit/oak/plugins/segment/failover/BulkTest.java (original)
+++ jackrabbit/oak/branches/1.0/oak-tarmk-failover/src/test/java/org/apache/jackrabbit/oak/plugins/segment/failover/BulkTest.java Fri Oct 17 12:36:13 2014
@@ -18,7 +18,6 @@
*/
package org.apache.jackrabbit.oak.plugins.segment.failover;
-import junit.framework.Assert;
import org.apache.jackrabbit.oak.plugins.segment.SegmentNodeStore;
import org.apache.jackrabbit.oak.plugins.segment.failover.client.FailoverClient;
import org.apache.jackrabbit.oak.plugins.segment.failover.jmx.FailoverStatusMBean;
@@ -79,19 +78,10 @@ public class BulkTest extends TestBase {
}
@Test
- public void test1MillionNodesNoChecksum() throws Exception {
- test(1000000, 87, 87, 22700000, 22800000, false, false);
- }
-
- @Test
public void test1MillionNodesUsingSSL() throws Exception {
- test(1000000, 87, 87, 22700000, 22800000, true, true);
+ test(1000000, 87, 87, 22700000, 22800000, true);
}
- @Test
- public void test1MillionNodesUsingSSLNoChecksum() throws Exception {
- test(1000000, 87, 87, 22700000, 22800000, true, false);
- }
/*
@Test
public void test10MillionNodes() throws Exception {
@@ -102,11 +92,11 @@ public class BulkTest extends TestBase {
// private helper
private void test(int number, int minExpectedSegments, int maxExpectedSegments, long minExpectedBytes, long maxExpectedBytes) throws Exception {
- test(number, minExpectedSegments, maxExpectedSegments, minExpectedBytes, maxExpectedBytes, false, true);
+ test(number, minExpectedSegments, maxExpectedSegments, minExpectedBytes, maxExpectedBytes, false);
}
private void test(int number, int minExpectedSegments, int maxExpectedSegments, long minExpectedBytes, long maxExpectedBytes,
- boolean useSSL, boolean useChecksum) throws Exception {
+ boolean useSSL) throws Exception {
NodeStore store = new SegmentNodeStore(storeS);
NodeBuilder rootbuilder = store.getRoot().builder();
NodeBuilder b = rootbuilder.child("store");
@@ -119,11 +109,11 @@ public class BulkTest extends TestBase {
store.merge(rootbuilder, EmptyHook.INSTANCE, CommitInfo.EMPTY);
storeS.flush();
- final FailoverServer server = new FailoverServer(port, storeS, useSSL, useChecksum);
+ final FailoverServer server = new FailoverServer(port, storeS, useSSL);
server.start();
System.setProperty(FailoverClient.CLIENT_ID_PROPERTY_NAME, "Bar");
- FailoverClient cl = new FailoverClient("127.0.0.1", port, storeC, useSSL, useChecksum);
+ FailoverClient cl = new FailoverClient("127.0.0.1", port, storeC, useSSL);
final MBeanServer jmxServer = ManagementFactory.getPlatformMBeanServer();
ObjectName status = new ObjectName(FailoverStatusMBean.JMX_NAME + ",id=*");
Copied: jackrabbit/oak/branches/1.0/oak-tarmk-failover/src/test/java/org/apache/jackrabbit/oak/plugins/segment/failover/FailoverIPRangeTest.java (from r1620634, jackrabbit/oak/trunk/oak-tarmk-failover/src/test/java/org/apache/jackrabbit/oak/plugins/segment/failover/FailoverIPRangeTest.java)
URL: http://svn.apache.org/viewvc/jackrabbit/oak/branches/1.0/oak-tarmk-failover/src/test/java/org/apache/jackrabbit/oak/plugins/segment/failover/FailoverIPRangeTest.java?p2=jackrabbit/oak/branches/1.0/oak-tarmk-failover/src/test/java/org/apache/jackrabbit/oak/plugins/segment/failover/FailoverIPRangeTest.java&p1=jackrabbit/oak/trunk/oak-tarmk-failover/src/test/java/org/apache/jackrabbit/oak/plugins/segment/failover/FailoverIPRangeTest.java&r1=1620634&r2=1632554&rev=1632554&view=diff
==============================================================================
--- jackrabbit/oak/trunk/oak-tarmk-failover/src/test/java/org/apache/jackrabbit/oak/plugins/segment/failover/FailoverIPRangeTest.java (original)
+++ jackrabbit/oak/branches/1.0/oak-tarmk-failover/src/test/java/org/apache/jackrabbit/oak/plugins/segment/failover/FailoverIPRangeTest.java Fri Oct 17 12:36:13 2014
@@ -48,18 +48,32 @@ public class FailoverIPRangeTest extends
public void testFailoverAllClients() throws Exception {
createTestWithConfig(null, true);
}
- @Test
+ @Test
public void testFailoverLocalClient() throws Exception {
createTestWithConfig(new String[]{"127.0.0.1"}, true);
}
@Test
+ public void testFailoverLocalClientUseIPv6() throws Exception {
+ if (!noDualStackSupport) {
+ createTestWithConfig("::1", new String[]{"::1"}, true);
+ }
+ }
+
+ @Test
public void testFailoverWrongClient() throws Exception {
createTestWithConfig(new String[]{"127.0.0.2"}, false);
}
@Test
+ public void testFailoverWrongClientIPv6() throws Exception {
+ if (!noDualStackSupport) {
+ createTestWithConfig(new String[]{"::2"}, false);
+ }
+ }
+
+ @Test
public void testFailoverLocalhost() throws Exception {
createTestWithConfig(new String[]{"localhost"}, true);
}
@@ -95,17 +109,50 @@ public class FailoverIPRangeTest extends
}
@Test
+ public void testFailoverCorrectListIPv6() throws Exception {
+ if (!noDualStackSupport) {
+ createTestWithConfig(new String[]{"foobar", "122-126", "::1", "126.0.0.1", "127.0.0.0-127.255.255.255"}, true);
+ }
+ }
+
+ @Test
public void testFailoverWrongList() throws Exception {
- createTestWithConfig(new String[]{"foobar","126.0.0.1", "128.0.0.1-255.255.255.255", "128.0.0.0-127.255.255.255"}, false);
+ createTestWithConfig(new String[]{"foobar", "126.0.0.1", "::2", "128.0.0.1-255.255.255.255", "128.0.0.0-127.255.255.255"}, false);
+ }
+
+ @Test
+ public void testFailoverCorrectListUseIPv6() throws Exception {
+ if (!noDualStackSupport) {
+ createTestWithConfig("::1", new String[]{"foobar","127-128", "0:0:0:0:0:0:0:1", "126.0.0.1", "127.0.0.0-127.255.255.255"}, true);
+ }
+ }
+
+ @Test
+ public void testFailoverCorrectListIPv6UseIPv6() throws Exception {
+ if (!noDualStackSupport) {
+ createTestWithConfig("::1", new String[]{"foobar", "122-126", "::1", "126.0.0.1", "127.0.0.0-127.255.255.255"}, true);
+ }
+ }
+
+ @Test
+ public void testFailoverWrongListUseIPv6() throws Exception {
+ if (!noDualStackSupport) {
+ createTestWithConfig("::1", new String[]{"foobar", "126.0.0.1", "::2", "128.0.0.1-255.255.255.255", "128.0.0.0-127.255.255.255"}, false);
+ }
}
private void createTestWithConfig(String[] ipRanges, boolean expectedToWork) throws Exception {
+ createTestWithConfig("127.0.0.1", ipRanges, expectedToWork);
+ }
+
+ private void createTestWithConfig(String host, String[] ipRanges, boolean expectedToWork) throws Exception {
NodeStore store = new SegmentNodeStore(storeS);
final FailoverServer server = new FailoverServer(port, storeS, ipRanges);
server.start();
addTestContent(store, "server");
+ storeS.flush(); // this speeds up the test a little bit...
- FailoverClient cl = new FailoverClient("127.0.0.1", port, storeC);
+ FailoverClient cl = new FailoverClient(host, port, storeC);
cl.run();
try {
Copied: jackrabbit/oak/branches/1.0/oak-tarmk-failover/src/test/java/org/apache/jackrabbit/oak/plugins/segment/failover/FailoverMultipleClientsTest.java (from r1620634, jackrabbit/oak/trunk/oak-tarmk-failover/src/test/java/org/apache/jackrabbit/oak/plugins/segment/failover/FailoverMultipleClientsTest.java)
URL: http://svn.apache.org/viewvc/jackrabbit/oak/branches/1.0/oak-tarmk-failover/src/test/java/org/apache/jackrabbit/oak/plugins/segment/failover/FailoverMultipleClientsTest.java?p2=jackrabbit/oak/branches/1.0/oak-tarmk-failover/src/test/java/org/apache/jackrabbit/oak/plugins/segment/failover/FailoverMultipleClientsTest.java&p1=jackrabbit/oak/trunk/oak-tarmk-failover/src/test/java/org/apache/jackrabbit/oak/plugins/segment/failover/FailoverMultipleClientsTest.java&r1=1620634&r2=1632554&rev=1632554&view=diff
==============================================================================
--- jackrabbit/oak/trunk/oak-tarmk-failover/src/test/java/org/apache/jackrabbit/oak/plugins/segment/failover/FailoverMultipleClientsTest.java (original)
+++ jackrabbit/oak/branches/1.0/oak-tarmk-failover/src/test/java/org/apache/jackrabbit/oak/plugins/segment/failover/FailoverMultipleClientsTest.java Fri Oct 17 12:36:13 2014
@@ -25,7 +25,6 @@ import org.apache.jackrabbit.oak.plugins
import org.apache.jackrabbit.oak.spi.state.NodeStore;
import org.junit.After;
-import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
@@ -50,6 +49,7 @@ public class FailoverMultipleClientsTest
final FailoverServer server = new FailoverServer(port, storeS);
server.start();
SegmentTestUtils.addTestContent(store, "server");
+ storeS.flush(); // this speeds up the test a little bit...
FailoverClient cl1 = new FailoverClient("127.0.0.1", port, storeC);
FailoverClient cl2 = new FailoverClient("127.0.0.1", port, storeC2);
@@ -70,7 +70,7 @@ public class FailoverMultipleClientsTest
cl2.run();
assertEquals(storeS.getHead(), storeC2.getHead());
- Assert.assertFalse("first client updated in stopped state!", storeS.getHead().equals(storeC.getHead()));
+ assertFalse("first client updated in stopped state!", storeS.getHead().equals(storeC.getHead()));
cl1.start();
assertEquals(storeS.getHead(), storeC.getHead());
Modified: jackrabbit/oak/branches/1.0/oak-tarmk-failover/src/test/java/org/apache/jackrabbit/oak/plugins/segment/failover/FailoverTest.java
URL: http://svn.apache.org/viewvc/jackrabbit/oak/branches/1.0/oak-tarmk-failover/src/test/java/org/apache/jackrabbit/oak/plugins/segment/failover/FailoverTest.java?rev=1632554&r1=1606087&r2=1632554&view=diff
==============================================================================
--- jackrabbit/oak/branches/1.0/oak-tarmk-failover/src/test/java/org/apache/jackrabbit/oak/plugins/segment/failover/FailoverTest.java (original)
+++ jackrabbit/oak/branches/1.0/oak-tarmk-failover/src/test/java/org/apache/jackrabbit/oak/plugins/segment/failover/FailoverTest.java Fri Oct 17 12:36:13 2014
@@ -1,66 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
package org.apache.jackrabbit.oak.plugins.segment.failover;
-import static java.io.File.createTempFile;
+import static org.apache.jackrabbit.oak.plugins.segment.SegmentTestUtils.addTestContent;
+import static org.apache.jackrabbit.oak.plugins.segment.SegmentTestUtils.createTmpTargetDir;
import static org.junit.Assert.assertEquals;
import java.io.File;
-import java.io.IOException;
-import org.apache.commons.io.FileUtils;
-import org.apache.jackrabbit.oak.api.CommitFailedException;
import org.apache.jackrabbit.oak.plugins.segment.SegmentNodeStore;
import org.apache.jackrabbit.oak.plugins.segment.failover.client.FailoverClient;
import org.apache.jackrabbit.oak.plugins.segment.failover.server.FailoverServer;
import org.apache.jackrabbit.oak.plugins.segment.file.FileStore;
-import org.apache.jackrabbit.oak.spi.commit.CommitInfo;
-import org.apache.jackrabbit.oak.spi.commit.EmptyHook;
-import org.apache.jackrabbit.oak.spi.state.NodeBuilder;
import org.apache.jackrabbit.oak.spi.state.NodeStore;
+
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class FailoverTest {
-
- private int port = Integer.valueOf(System.getProperty(
- "failover.server.port", "52808"));
-
- private File directoryS;
- private FileStore storeS;
-
- private File directoryC;
-
- private FileStore storeC;
+public class FailoverTest extends TestBase {
@Before
- public void setUp() throws IOException {
- File target = new File("target");
-
- // server
- directoryS = createTempFile("FailoverServerTest", "dir", target);
- directoryS.delete();
- directoryS.mkdir();
- storeS = new FileStore(directoryS, 1, false);
-
- // client
- directoryC = createTempFile("FailoverClientTest", "dir", target);
- directoryC.delete();
- directoryC.mkdir();
- storeC = new FileStore(directoryC, 1, false);
+ public void setUp() throws Exception {
+ setUpServerAndClient();
}
@After
public void after() {
- storeS.close();
- storeC.close();
- try {
- FileUtils.deleteDirectory(directoryS);
- FileUtils.deleteDirectory(directoryC);
- } catch (IOException e) {
- }
+ closeServerAndClient();
}
@Test
@@ -68,13 +51,8 @@ public class FailoverTest {
NodeStore store = new SegmentNodeStore(storeS);
final FailoverServer server = new FailoverServer(port, storeS);
- Thread s = new Thread() {
- public void run() {
- server.start();
- }
- };
- s.start();
- addTestContent(store);
+ server.start();
+ addTestContent(store, "server");
FailoverClient cl = new FailoverClient("127.0.0.1", port, storeC);
cl.run();
@@ -88,26 +66,8 @@ public class FailoverTest {
}
- private static void addTestContent(NodeStore store)
- throws CommitFailedException {
- NodeBuilder builder = store.getRoot().builder();
- builder.child("server").setProperty("ts", System.currentTimeMillis());
- store.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY);
- }
-
- static void assertEqualStores(File d1, File d2) throws IOException {
- FileStore f1 = new FileStore(d1, 1, false);
- FileStore f2 = new FileStore(d2, 1, false);
- try {
- assertEquals(f1.getHead(), f2.getHead());
- } finally {
- f1.close();
- f2.close();
- }
- }
-
public static void main(String[] args) throws Exception {
- File d = createTempFile("FailoverLiveTest", "dir", new File("target"));
+ File d = createTmpTargetDir("FailoverLiveTest");
d.delete();
d.mkdir();
FileStore s = new FileStore(d, 256, false);
Copied: jackrabbit/oak/branches/1.0/oak-tarmk-failover/src/test/java/org/apache/jackrabbit/oak/plugins/segment/failover/MBeanTest.java (from r1620634, jackrabbit/oak/trunk/oak-tarmk-failover/src/test/java/org/apache/jackrabbit/oak/plugins/segment/failover/MBeanTest.java)
URL: http://svn.apache.org/viewvc/jackrabbit/oak/branches/1.0/oak-tarmk-failover/src/test/java/org/apache/jackrabbit/oak/plugins/segment/failover/MBeanTest.java?p2=jackrabbit/oak/branches/1.0/oak-tarmk-failover/src/test/java/org/apache/jackrabbit/oak/plugins/segment/failover/MBeanTest.java&p1=jackrabbit/oak/trunk/oak-tarmk-failover/src/test/java/org/apache/jackrabbit/oak/plugins/segment/failover/MBeanTest.java&r1=1620634&r2=1632554&rev=1632554&view=diff
==============================================================================
--- jackrabbit/oak/trunk/oak-tarmk-failover/src/test/java/org/apache/jackrabbit/oak/plugins/segment/failover/MBeanTest.java (original)
+++ jackrabbit/oak/branches/1.0/oak-tarmk-failover/src/test/java/org/apache/jackrabbit/oak/plugins/segment/failover/MBeanTest.java Fri Oct 17 12:36:13 2014
@@ -24,6 +24,7 @@ import org.apache.jackrabbit.oak.plugins
import org.junit.After;
import org.junit.Before;
+import org.junit.Ignore;
import org.junit.Test;
import javax.management.MBeanServer;
@@ -31,9 +32,7 @@ import javax.management.ObjectName;
import java.lang.management.ManagementFactory;
import java.util.Set;
-import static junit.framework.Assert.assertEquals;
-import static junit.framework.Assert.assertTrue;
-import static junit.framework.Assert.fail;
+import static junit.framework.Assert.*;
public class MBeanTest extends TestBase {
@@ -63,7 +62,7 @@ public class MBeanTest extends TestBase
assertEquals("master", jmxServer.getAttribute(status, "Mode"));
String m = jmxServer.getAttribute(status, "Status").toString();
- if (!m.equals(FailoverStatusMBean.STATUS_STARTING) && !m.equals("Channel unregistered"))
+ if (!m.equals(FailoverStatusMBean.STATUS_STARTING) && !m.equals("channel unregistered"))
fail("unexpected Status" + m);
assertEquals(FailoverStatusMBean.STATUS_STARTING, jmxServer.getAttribute(status, "Status"));
@@ -99,6 +98,9 @@ public class MBeanTest extends TestBase
String m = jmxServer.getAttribute(status, "Mode").toString();
if (!m.startsWith("client: ")) fail("unexpected mode " + m);
+ assertEquals("1", jmxServer.getAttribute(status, "FailedRequests").toString());
+ assertEquals("-1", jmxServer.getAttribute(status, "SecondsSinceLastSuccess").toString());
+
assertEquals(FailoverStatusMBean.STATUS_STOPPED, jmxServer.getAttribute(status, "Status"));
assertEquals(false, jmxServer.getAttribute(status, "Running"));
@@ -126,6 +128,12 @@ public class MBeanTest extends TestBase
try {
assertTrue(jmxServer.isRegistered(status));
assertEquals("client: Foo", jmxServer.getAttribute(status, "Mode"));
+
+ assertEquals("1", jmxServer.getAttribute(status, "FailedRequests").toString());
+ assertEquals("-1", jmxServer.getAttribute(status, "SecondsSinceLastSuccess").toString());
+
+ assertEquals("1", jmxServer.invoke(status, "calcFailedRequests", null, null).toString());
+ assertEquals("-1", jmxServer.invoke(status, "calcSecondsSinceLastSuccess", null, null).toString());
} finally {
client.close();
}
@@ -134,6 +142,7 @@ public class MBeanTest extends TestBase
}
@Test
+ @Ignore("OAK-2086")
public void testClientAndServerEmptyConfig() throws Exception {
final FailoverServer server = new FailoverServer(this.port, this.storeS);
server.start();
@@ -150,8 +159,15 @@ public class MBeanTest extends TestBase
Set<ObjectName> instances = jmxServer.queryNames(status, null);
assertEquals(3, instances.size());
+ ObjectName connectionStatus = null;
+ for (ObjectName s : instances) {
+ if (!s.equals(clientStatus) && !s.equals(serverStatus)) connectionStatus = s;
+ }
+ assertNotNull(connectionStatus);
+
assertTrue(jmxServer.isRegistered(clientStatus));
assertTrue(jmxServer.isRegistered(serverStatus));
+ assertTrue(jmxServer.isRegistered(connectionStatus));
String m = jmxServer.getAttribute(clientStatus, "Mode").toString();
if (!m.startsWith("client: ")) fail("unexpected mode " + m);
@@ -161,11 +177,26 @@ public class MBeanTest extends TestBase
assertEquals(true, jmxServer.getAttribute(serverStatus, "Running"));
assertEquals(true, jmxServer.getAttribute(clientStatus, "Running"));
+ assertEquals("0", jmxServer.getAttribute(clientStatus, "FailedRequests").toString());
+ assertEquals("0", jmxServer.getAttribute(clientStatus, "SecondsSinceLastSuccess").toString());
+ assertEquals("0", jmxServer.invoke(clientStatus, "calcFailedRequests", null, null).toString());
+ assertEquals("0", jmxServer.invoke(clientStatus, "calcSecondsSinceLastSuccess", null, null).toString());
+
+ Thread.sleep(1000);
+
+ assertEquals("0", jmxServer.getAttribute(clientStatus, "FailedRequests").toString());
+ assertEquals("1", jmxServer.getAttribute(clientStatus, "SecondsSinceLastSuccess").toString());
+ assertEquals("0", jmxServer.invoke(clientStatus, "calcFailedRequests", null, null).toString());
+ assertEquals("1", jmxServer.invoke(clientStatus, "calcSecondsSinceLastSuccess", null, null).toString());
+
+ assertEquals(new Long(2), jmxServer.getAttribute(connectionStatus, "TransferredSegments"));
+ assertEquals(new Long(128), jmxServer.getAttribute(connectionStatus, "TransferredSegmentBytes"));
+
// stop the master
jmxServer.invoke(serverStatus, "stop", null, null);
assertEquals(false, jmxServer.getAttribute(serverStatus, "Running"));
m = jmxServer.getAttribute(serverStatus, "Status").toString();
- if (!m.equals(FailoverStatusMBean.STATUS_STOPPED) && !m.equals("Channel unregistered"))
+ if (!m.equals(FailoverStatusMBean.STATUS_STOPPED) && !m.equals("channel unregistered"))
fail("unexpected Status" + m);
// restart the master
@@ -173,7 +204,7 @@ public class MBeanTest extends TestBase
assertEquals(true, jmxServer.getAttribute(serverStatus, "Running"));
assertEquals(true, jmxServer.getAttribute(clientStatus, "Running"));
m = jmxServer.getAttribute(serverStatus, "Status").toString();
- if (!m.equals(FailoverStatusMBean.STATUS_STARTING) && !m.equals("Channel unregistered"))
+ if (!m.equals(FailoverStatusMBean.STATUS_STARTING) && !m.equals("channel unregistered"))
fail("unexpected Status" + m);
// stop the slave
Copied: jackrabbit/oak/branches/1.0/oak-tarmk-failover/src/test/java/org/apache/jackrabbit/oak/plugins/segment/failover/RecoverTest.java (from r1620634, jackrabbit/oak/trunk/oak-tarmk-failover/src/test/java/org/apache/jackrabbit/oak/plugins/segment/failover/RecoverTest.java)
URL: http://svn.apache.org/viewvc/jackrabbit/oak/branches/1.0/oak-tarmk-failover/src/test/java/org/apache/jackrabbit/oak/plugins/segment/failover/RecoverTest.java?p2=jackrabbit/oak/branches/1.0/oak-tarmk-failover/src/test/java/org/apache/jackrabbit/oak/plugins/segment/failover/RecoverTest.java&p1=jackrabbit/oak/trunk/oak-tarmk-failover/src/test/java/org/apache/jackrabbit/oak/plugins/segment/failover/RecoverTest.java&r1=1620634&r2=1632554&rev=1632554&view=diff
==============================================================================
--- jackrabbit/oak/trunk/oak-tarmk-failover/src/test/java/org/apache/jackrabbit/oak/plugins/segment/failover/RecoverTest.java (original)
+++ jackrabbit/oak/branches/1.0/oak-tarmk-failover/src/test/java/org/apache/jackrabbit/oak/plugins/segment/failover/RecoverTest.java Fri Oct 17 12:36:13 2014
@@ -70,4 +70,28 @@ public class RecoverTest extends TestBas
}
}
+
+ @Test
+ public void testLocalChanges() throws Exception {
+
+ NodeStore store = new SegmentNodeStore(storeC);
+ addTestContent(store, "client");
+
+ final FailoverServer server = new FailoverServer(port, storeS);
+ server.start();
+ store = new SegmentNodeStore(storeS);
+ addTestContent(store, "server");
+ storeS.flush();
+
+ FailoverClient cl = new FailoverClient("127.0.0.1", port, storeC);
+ try {
+ assertFalse("stores are not expected to be equal", storeS.getHead().equals(storeC.getHead()));
+ cl.run();
+ assertEquals(storeS.getHead(), storeC.getHead());
+ } finally {
+ server.close();
+ cl.close();
+ }
+
+ }
}
Copied: jackrabbit/oak/branches/1.0/oak-tarmk-failover/src/test/java/org/apache/jackrabbit/oak/plugins/segment/failover/TestBase.java (from r1620634, jackrabbit/oak/trunk/oak-tarmk-failover/src/test/java/org/apache/jackrabbit/oak/plugins/segment/failover/TestBase.java)
URL: http://svn.apache.org/viewvc/jackrabbit/oak/branches/1.0/oak-tarmk-failover/src/test/java/org/apache/jackrabbit/oak/plugins/segment/failover/TestBase.java?p2=jackrabbit/oak/branches/1.0/oak-tarmk-failover/src/test/java/org/apache/jackrabbit/oak/plugins/segment/failover/TestBase.java&p1=jackrabbit/oak/trunk/oak-tarmk-failover/src/test/java/org/apache/jackrabbit/oak/plugins/segment/failover/TestBase.java&r1=1620634&r2=1632554&rev=1632554&view=diff
==============================================================================
--- jackrabbit/oak/trunk/oak-tarmk-failover/src/test/java/org/apache/jackrabbit/oak/plugins/segment/failover/TestBase.java (original)
+++ jackrabbit/oak/branches/1.0/oak-tarmk-failover/src/test/java/org/apache/jackrabbit/oak/plugins/segment/failover/TestBase.java Fri Oct 17 12:36:13 2014
@@ -19,6 +19,7 @@
package org.apache.jackrabbit.oak.plugins.segment.failover;
import org.apache.commons.io.FileUtils;
+import org.apache.commons.lang3.SystemUtils;
import org.apache.jackrabbit.oak.plugins.segment.file.FileStore;
import java.io.File;
@@ -28,6 +29,7 @@ import static org.apache.jackrabbit.oak.
public class TestBase {
int port = Integer.valueOf(System.getProperty("failover.server.port", "52808"));
+ final static String LOCALHOST = "127.0.0.1";
File directoryS;
FileStore storeS;
@@ -38,6 +40,12 @@ public class TestBase {
File directoryC2;
FileStore storeC2;
+ /*
+ Java 6 on Windows doesn't support dual IP stacks, so we will skip our IPv6
+ tests.
+ */
+ protected final boolean noDualStackSupport = SystemUtils.IS_OS_WINDOWS && SystemUtils.IS_JAVA_1_6;
+
public void setUpServerAndClient() throws IOException {
// server
directoryS = createTmpTargetDir("FailoverServerTest");
Modified: jackrabbit/oak/branches/1.0/pom.xml
URL: http://svn.apache.org/viewvc/jackrabbit/oak/branches/1.0/pom.xml?rev=1632554&r1=1632553&r2=1632554&view=diff
==============================================================================
--- jackrabbit/oak/branches/1.0/pom.xml (original)
+++ jackrabbit/oak/branches/1.0/pom.xml Fri Oct 17 12:36:13 2014
@@ -54,6 +54,7 @@
<module>oak-run</module>
<module>oak-it</module>
<module>oak-pojosr</module>
+ <module>oak-tarmk-failover</module>
</modules>
<scm>