From 87fd3d39868a58477f3946dec9107ba58d4a3e93 Mon Sep 17 00:00:00 2001 From: Arnaud Bailly Date: Mon, 23 Jun 2014 16:04:41 +0200 Subject: [PATCH 01/10] introduce REST-based Key-value store over Raft basic PUT interface --- barge-store/pom.xml | 61 ++++++++++++ .../barge/store/OperationsSerializer.java | 97 +++++++++++++++++++ .../robotninjas/barge/store/RaftStore.java | 30 ++++++ .../barge/store/RaftStoreInstance.java | 64 ++++++++++++ .../barge/store/StoreResource.java | 35 +++++++ .../org/robotninjas/barge/store/Write.java | 61 ++++++++++++ .../barge/store/RaftStoreInstanceTest.java | 46 +++++++++ .../barge/store/StoreResourceTest.java | 64 ++++++++++++ pom.xml | 14 +++ 9 files changed, 472 insertions(+) create mode 100644 barge-store/pom.xml create mode 100644 barge-store/src/main/java/org/robotninjas/barge/store/OperationsSerializer.java create mode 100644 barge-store/src/main/java/org/robotninjas/barge/store/RaftStore.java create mode 100644 barge-store/src/main/java/org/robotninjas/barge/store/RaftStoreInstance.java create mode 100644 barge-store/src/main/java/org/robotninjas/barge/store/StoreResource.java create mode 100644 barge-store/src/main/java/org/robotninjas/barge/store/Write.java create mode 100644 barge-store/src/test/java/org/robotninjas/barge/store/RaftStoreInstanceTest.java create mode 100644 barge-store/src/test/java/org/robotninjas/barge/store/StoreResourceTest.java diff --git a/barge-store/pom.xml b/barge-store/pom.xml new file mode 100644 index 0000000..6cd610d --- /dev/null +++ b/barge-store/pom.xml @@ -0,0 +1,61 @@ + + + + barge + org.robotninjas.barge + 0.1.0-alpha2-SNAPSHOT + + 4.0.0 + + barge-store + + A simplistic key-value store backed by Barge + + + + + org.robotninjas.barge + barge-jax-rs + + + + + + org.glassfish.jersey.test-framework + jersey-test-framework-core + test + + + + org.glassfish.jersey.test-framework.providers + jersey-test-framework-provider-inmemory + test + + + + org.assertj + assertj-core + + + + org.assertj + assertj-guava + test + + + + junit + junit + + + + org.mockito + mockito-all + + + + + + \ No newline at end of file diff --git a/barge-store/src/main/java/org/robotninjas/barge/store/OperationsSerializer.java b/barge-store/src/main/java/org/robotninjas/barge/store/OperationsSerializer.java new file mode 100644 index 0000000..3ce5854 --- /dev/null +++ b/barge-store/src/main/java/org/robotninjas/barge/store/OperationsSerializer.java @@ -0,0 +1,97 @@ +package org.robotninjas.barge.store; + +import com.fasterxml.jackson.core.JsonParser; +import com.fasterxml.jackson.core.JsonToken; +import com.fasterxml.jackson.core.Version; +import com.fasterxml.jackson.databind.DeserializationContext; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.module.SimpleModule; +import com.fasterxml.jackson.jaxrs.json.JacksonJaxbJsonProvider; + +import com.google.common.base.Throwables; + +import java.io.*; + + +/** + */ +public class OperationsSerializer { + + public byte[] serialize(Write write) { + ByteArrayOutputStream out = new ByteArrayOutputStream(); + + try { + ObjectOutputStream os = new ObjectOutputStream(out); + os.writeObject(write); + + return out.toByteArray(); + } catch (IOException e) { + throw new OperationsSerializationException(e); + } + } + + /** + * @return a new Object mapper with configured deserializer for barge store writes and reads. + */ + static ObjectMapper objectMapper() { + ObjectMapper mapper = new ObjectMapper(); + SimpleModule testModule = new SimpleModule("BargeStore", new Version(0, 1, 0, null, "org.robotninjas", "barge-store")) + .addDeserializer(Write.class, new WriteDeserializer()); + mapper.registerModule(testModule); + + return mapper; + } + + public static JacksonJaxbJsonProvider jacksonModule() { + JacksonJaxbJsonProvider provider = new JacksonJaxbJsonProvider(); + provider.setMapper(objectMapper()); + + return provider; + } + + public Write deserialize(byte[] bytes) { + ByteArrayInputStream in = new ByteArrayInputStream(bytes); + + try { + return (Write) new ObjectInputStream(in).readObject(); + } catch (IOException e) { + throw Throwables.propagate(e); + } catch (ClassNotFoundException e) { + throw Throwables.propagate(e); + } + } + + + public static class OperationsSerializationException extends RuntimeException { + + public OperationsSerializationException(Throwable throwable) { + super(throwable); + } + + } + + public static class WriteDeserializer extends com.fasterxml.jackson.databind.JsonDeserializer { + + private static final byte[] EMPTY = new byte[0]; + + @Override + public Write deserialize(JsonParser jsonParser, DeserializationContext ctxt) throws IOException { + byte[] value = EMPTY; + String key = ""; + + while (jsonParser.nextToken() != JsonToken.END_OBJECT) { + key = jsonParser.getCurrentName(); + + jsonParser.nextToken(); + + if (key.equals("value")) { + value = jsonParser.getBinaryValue(); + } + } + + jsonParser.close(); + + return new Write(key, value); + } + } +} diff --git a/barge-store/src/main/java/org/robotninjas/barge/store/RaftStore.java b/barge-store/src/main/java/org/robotninjas/barge/store/RaftStore.java new file mode 100644 index 0000000..0a8d8fc --- /dev/null +++ b/barge-store/src/main/java/org/robotninjas/barge/store/RaftStore.java @@ -0,0 +1,30 @@ +package org.robotninjas.barge.store; + +import com.google.common.base.Optional; + + +/** + */ +public interface RaftStore { + + /** + * Writes some key/value pair to this store. + * + *

This operations writes some value under some key in this store. It expects the underlying Raft instance to + * be a leader or to point to a leader, and waits for completion of operation to the cluster of raft instances + * to complete.

+ * + * @param write operation to apply to store. + * @return the old value. + */ + byte[] write(Write write); + + /** + * Reads current value of some key if it exists. + * + * @param key key to read data from. + * @return a byte array containing raw data for this key or {@link com.google.common.base.Optional#absent()} if + * no value is associated with given key in this store. + */ + Optional read(String key); +} diff --git a/barge-store/src/main/java/org/robotninjas/barge/store/RaftStoreInstance.java b/barge-store/src/main/java/org/robotninjas/barge/store/RaftStoreInstance.java new file mode 100644 index 0000000..6717a52 --- /dev/null +++ b/barge-store/src/main/java/org/robotninjas/barge/store/RaftStoreInstance.java @@ -0,0 +1,64 @@ +package org.robotninjas.barge.store; + +import com.google.common.base.Optional; +import static com.google.common.base.Optional.fromNullable; +import static com.google.common.base.Throwables.propagate; +import com.google.common.collect.Maps; + +import org.robotninjas.barge.RaftException; +import org.robotninjas.barge.StateMachine; +import org.robotninjas.barge.state.Raft; + +import java.nio.ByteBuffer; + +import java.util.Map; +import java.util.concurrent.ExecutionException; + +import javax.annotation.Nonnull; + +import javax.inject.Inject; + + +/** + */ +public class RaftStoreInstance implements RaftStore, StateMachine { + + private final Raft raft; + private final OperationsSerializer operationsSerializer = new OperationsSerializer(); + private final Map store = Maps.newConcurrentMap(); + + @Inject + public RaftStoreInstance(Raft raft) { + this.raft = raft; + } + + @Override + public byte[] write(Write write) { + + try { + return (byte[]) raft.commitOperation(operationsSerializer.serialize(write)).get(); + } catch (RaftException e) { + throw propagate(e); + } catch (InterruptedException e) { + throw propagate(e); + } catch (ExecutionException e) { + throw propagate(e); + } + } + + @Override + public Optional read(String key) { + return fromNullable(store.get(key)); + } + + @Override + public Object applyOperation(@Nonnull ByteBuffer entry) { + Write write = operationsSerializer.deserialize(entry.array()); + + store.put(write.getKey(), write.getValue()); + + return write.getValue(); + } + + +} diff --git a/barge-store/src/main/java/org/robotninjas/barge/store/StoreResource.java b/barge-store/src/main/java/org/robotninjas/barge/store/StoreResource.java new file mode 100644 index 0000000..756751a --- /dev/null +++ b/barge-store/src/main/java/org/robotninjas/barge/store/StoreResource.java @@ -0,0 +1,35 @@ +package org.robotninjas.barge.store; + +import javax.inject.Inject; + +import javax.ws.rs.*; +import javax.ws.rs.core.MediaType; +import static javax.ws.rs.core.MediaType.APPLICATION_OCTET_STREAM; +import javax.ws.rs.core.Response; +import static javax.ws.rs.core.Response.created; +import javax.ws.rs.core.UriInfo; + + +@Path("/") +@Produces(MediaType.APPLICATION_JSON) +public class StoreResource { + + private RaftStore raftStore; + + @Inject + private UriInfo uriInfo; + + @Inject + public StoreResource(RaftStore raftStore) { + this.raftStore = raftStore; + } + + @Path("{key:.+}") + @PUT + @Consumes(APPLICATION_OCTET_STREAM) + public Response store(@PathParam("key") String key, byte[] value) { + byte[] bytes = raftStore.write(new Write(key, value)); + + return created(uriInfo.getRequestUri()).type(APPLICATION_OCTET_STREAM).entity(bytes).build(); + } +} diff --git a/barge-store/src/main/java/org/robotninjas/barge/store/Write.java b/barge-store/src/main/java/org/robotninjas/barge/store/Write.java new file mode 100644 index 0000000..444e1c7 --- /dev/null +++ b/barge-store/src/main/java/org/robotninjas/barge/store/Write.java @@ -0,0 +1,61 @@ +package org.robotninjas.barge.store; + +import static com.google.common.base.Objects.toStringHelper; + +import java.io.Serializable; + +import java.util.Arrays; +import java.util.Objects; +import static java.util.Objects.hash; + + +/** + * A write operation to the underlying store. + *

Writes a value for key.

+ */ +public class Write implements Serializable { + + private final String key; + private final byte[] value; + + public Write(String key, byte[] value) { + this.key = key; + this.value = value; + } + + @SuppressWarnings("UnusedDeclaration") + public String getKey() { + return key; + } + + @SuppressWarnings("UnusedDeclaration") + public byte[] getValue() { + return value; + } + + @Override + public String toString() { + return toStringHelper(this).add("key", key).add("value", value).toString(); + } + + @Override + public int hashCode() { + return hash(key, value); + } + + @Override + public boolean equals(Object obj) { + + if (this == obj) { + return true; + } + + if ((obj == null) || (getClass() != obj.getClass())) { + return false; + } + + final Write other = (Write) obj; + + return Objects.equals(this.key, other.key) && Arrays.equals(this.value, other.value); + } +} diff --git a/barge-store/src/test/java/org/robotninjas/barge/store/RaftStoreInstanceTest.java b/barge-store/src/test/java/org/robotninjas/barge/store/RaftStoreInstanceTest.java new file mode 100644 index 0000000..64e8f7c --- /dev/null +++ b/barge-store/src/test/java/org/robotninjas/barge/store/RaftStoreInstanceTest.java @@ -0,0 +1,46 @@ +package org.robotninjas.barge.store; + +import com.google.common.base.Optional; +import com.google.common.util.concurrent.Futures; + +import static org.assertj.core.api.Assertions.assertThat; + +import org.assertj.guava.api.Assertions; + +import org.junit.Test; + +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import org.robotninjas.barge.state.Raft; + +import static java.nio.ByteBuffer.wrap; + + +public class RaftStoreInstanceTest { + + private static final byte[] oldValue = new byte[] { 0x41 }; + private static final Write write = new Write("foo", new byte[] { 0x42 }); + + private final Raft raft = mock(Raft.class); + private final OperationsSerializer serializer = new OperationsSerializer(); + + private final RaftStoreInstance raftStoreInstance = new RaftStoreInstance(raft); + + @Test + public void commitSerializedWriteOperationToRaftThenReturnOldValueWhenWritingKeyValuePair() throws Exception { + when(raft.commitOperation(serializer.serialize(write))).thenReturn(Futures.immediateFuture(oldValue)); + + assertThat(raftStoreInstance.write(write)).isEqualTo(oldValue); + } + + @Test + public void mapsKeyToValueWhenApplyingWriteOperation() throws Exception { + raftStoreInstance.applyOperation(wrap(serializer.serialize(write))); + + Optional read = raftStoreInstance.read("foo"); + + Assertions.assertThat(read).isPresent(); + assertThat(read.get()).isEqualTo(new byte[] { 0x42 }); + } +} diff --git a/barge-store/src/test/java/org/robotninjas/barge/store/StoreResourceTest.java b/barge-store/src/test/java/org/robotninjas/barge/store/StoreResourceTest.java new file mode 100644 index 0000000..f55252e --- /dev/null +++ b/barge-store/src/test/java/org/robotninjas/barge/store/StoreResourceTest.java @@ -0,0 +1,64 @@ +package org.robotninjas.barge.store; + +import com.google.common.collect.Sets; + +import static org.assertj.core.api.Assertions.assertThat; + +import org.eclipse.jetty.http.HttpStatus; + +import org.glassfish.jersey.server.ResourceConfig; +import org.glassfish.jersey.test.JerseyTest; + +import org.junit.Test; + +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import java.util.Set; + +import javax.ws.rs.client.Entity; +import javax.ws.rs.core.Application; +import static javax.ws.rs.core.MediaType.APPLICATION_OCTET_STREAM_TYPE; +import javax.ws.rs.core.Response; + + +/** + */ +public class StoreResourceTest extends JerseyTest { + + private static final byte[] oldValue = { 0x41 }; + private static final byte[] value = { 0x42 }; + + private static final Write write = new Write("foo", value); + + private RaftStore raftStore; + + @Test + public void onPUTReturns201WithLocationAndPreviousValueGivenRaftStoreCompletesSuccessfully() throws Exception { + when(raftStore.write(write)).thenReturn(oldValue); + + Response response = client().target("/foo").request().put(Entity.entity(value, APPLICATION_OCTET_STREAM_TYPE)); + + assertThat(response.getStatus()).isEqualTo(HttpStatus.CREATED_201); + assertThat(response.getHeaderString("Location")).isEqualTo("http://localhost:9998/foo"); + assertThat(response.readEntity(byte[].class)).isEqualTo(oldValue); + } + + + @Override + protected Application configure() { + raftStore = mock(RaftStore.class); + + ResourceConfig resourceConfig = ResourceConfig.forApplication(new Application() { + @Override + public Set getSingletons() { + return Sets.newHashSet((Object) new StoreResource(raftStore)); + } + }); + + resourceConfig.register(OperationsSerializer.jacksonModule()); + + return resourceConfig; + } + +} diff --git a/pom.xml b/pom.xml index be7dc93..2528a2b 100644 --- a/pom.xml +++ b/pom.xml @@ -43,6 +43,7 @@ barge-tools barge-rpc-proto barge-jax-rs + barge-store @@ -65,6 +66,12 @@ ${project.version} + + org.robotninjas.barge + barge-jax-rs + ${project.version} + + org.robotninjas @@ -211,6 +218,13 @@ test + + org.assertj + assertj-guava + 1.2.0 + test + + From baa1df40f8eea7c28a87e00c4f87620e6485bd5e Mon Sep 17 00:00:00 2001 From: Arnaud Bailly Date: Mon, 23 Jun 2014 16:51:19 +0200 Subject: [PATCH 02/10] implement GET on keys --- .../barge/store/RaftStoreInstance.java | 4 +--- .../robotninjas/barge/store/StoreResource.java | 12 ++++++++++++ .../barge/store/RaftStoreInstanceTest.java | 16 ++++++++++++++++ .../barge/store/StoreResourceTest.java | 10 ++++++++++ 4 files changed, 39 insertions(+), 3 deletions(-) diff --git a/barge-store/src/main/java/org/robotninjas/barge/store/RaftStoreInstance.java b/barge-store/src/main/java/org/robotninjas/barge/store/RaftStoreInstance.java index 6717a52..7b1d8c4 100644 --- a/barge-store/src/main/java/org/robotninjas/barge/store/RaftStoreInstance.java +++ b/barge-store/src/main/java/org/robotninjas/barge/store/RaftStoreInstance.java @@ -55,9 +55,7 @@ public Optional read(String key) { public Object applyOperation(@Nonnull ByteBuffer entry) { Write write = operationsSerializer.deserialize(entry.array()); - store.put(write.getKey(), write.getValue()); - - return write.getValue(); + return store.put(write.getKey(), write.getValue()); } diff --git a/barge-store/src/main/java/org/robotninjas/barge/store/StoreResource.java b/barge-store/src/main/java/org/robotninjas/barge/store/StoreResource.java index 756751a..afefb47 100644 --- a/barge-store/src/main/java/org/robotninjas/barge/store/StoreResource.java +++ b/barge-store/src/main/java/org/robotninjas/barge/store/StoreResource.java @@ -1,5 +1,7 @@ package org.robotninjas.barge.store; +import com.google.common.base.Optional; + import javax.inject.Inject; import javax.ws.rs.*; @@ -7,6 +9,7 @@ import static javax.ws.rs.core.MediaType.APPLICATION_OCTET_STREAM; import javax.ws.rs.core.Response; import static javax.ws.rs.core.Response.created; +import static javax.ws.rs.core.Response.ok; import javax.ws.rs.core.UriInfo; @@ -32,4 +35,13 @@ public Response store(@PathParam("key") String key, byte[] value) { return created(uriInfo.getRequestUri()).type(APPLICATION_OCTET_STREAM).entity(bytes).build(); } + + @Path("{key:.+}") + @GET + @Produces(APPLICATION_OCTET_STREAM) + public Response read(@PathParam("key") String key) { + Optional bytes = raftStore.read(key); + + return ok(bytes.get()).type(APPLICATION_OCTET_STREAM).build(); + } } diff --git a/barge-store/src/test/java/org/robotninjas/barge/store/RaftStoreInstanceTest.java b/barge-store/src/test/java/org/robotninjas/barge/store/RaftStoreInstanceTest.java index 64e8f7c..8c85c9e 100644 --- a/barge-store/src/test/java/org/robotninjas/barge/store/RaftStoreInstanceTest.java +++ b/barge-store/src/test/java/org/robotninjas/barge/store/RaftStoreInstanceTest.java @@ -43,4 +43,20 @@ public void mapsKeyToValueWhenApplyingWriteOperation() throws Exception { Assertions.assertThat(read).isPresent(); assertThat(read.get()).isEqualTo(new byte[] { 0x42 }); } + + @Test + public void applyingWriteReturnsOldValueGivenKeyIsAlreadyMapped() throws Exception { + raftStoreInstance.applyOperation(wrap(serializer.serialize(write))); + + byte[] old = (byte[]) raftStoreInstance.applyOperation(wrap(serializer.serialize(new Write("foo", new byte[] { 0x43 })))); + + assertThat(old).isEqualTo(new byte[] { 0x42 }); + } + + @Test + public void applyingWriteReturnsNullValueGivenKeyIsNotMapped() throws Exception { + byte[] old = (byte[]) raftStoreInstance.applyOperation(wrap(serializer.serialize(new Write("foo", new byte[] { 0x43 })))); + + assertThat(old).isNull(); + } } diff --git a/barge-store/src/test/java/org/robotninjas/barge/store/StoreResourceTest.java b/barge-store/src/test/java/org/robotninjas/barge/store/StoreResourceTest.java index f55252e..ee44ce5 100644 --- a/barge-store/src/test/java/org/robotninjas/barge/store/StoreResourceTest.java +++ b/barge-store/src/test/java/org/robotninjas/barge/store/StoreResourceTest.java @@ -1,5 +1,6 @@ package org.robotninjas.barge.store; +import com.google.common.base.Optional; import com.google.common.collect.Sets; import static org.assertj.core.api.Assertions.assertThat; @@ -44,6 +45,15 @@ public void onPUTReturns201WithLocationAndPreviousValueGivenRaftStoreCompletesSu assertThat(response.readEntity(byte[].class)).isEqualTo(oldValue); } + @Test + public void onGETReturns200WithFoundValueGivenStoreContainsRequestedKey() throws Exception { + when(raftStore.read("foo")).thenReturn(Optional.of(value)); + + Response response = client().target("/foo").request().get(Response.class); + + assertThat(response.getStatus()).isEqualTo(HttpStatus.OK_200); + assertThat(response.readEntity(byte[].class)).isEqualTo(value); + } @Override protected Application configure() { From 50851fd58ff32974d78a190aae131545dfd8b7e5 Mon Sep 17 00:00:00 2001 From: Arnaud Bailly Date: Mon, 23 Jun 2014 16:54:09 +0200 Subject: [PATCH 03/10] returns 404 when key is not mapped --- .../java/org/robotninjas/barge/store/StoreResource.java | 6 +++++- .../org/robotninjas/barge/store/StoreResourceTest.java | 9 +++++++++ 2 files changed, 14 insertions(+), 1 deletion(-) diff --git a/barge-store/src/main/java/org/robotninjas/barge/store/StoreResource.java b/barge-store/src/main/java/org/robotninjas/barge/store/StoreResource.java index afefb47..adc81d5 100644 --- a/barge-store/src/main/java/org/robotninjas/barge/store/StoreResource.java +++ b/barge-store/src/main/java/org/robotninjas/barge/store/StoreResource.java @@ -42,6 +42,10 @@ public Response store(@PathParam("key") String key, byte[] value) { public Response read(@PathParam("key") String key) { Optional bytes = raftStore.read(key); - return ok(bytes.get()).type(APPLICATION_OCTET_STREAM).build(); + if (bytes.isPresent()) { + return ok(bytes.get()).type(APPLICATION_OCTET_STREAM).build(); + } + + throw new NotFoundException("key " + key + " is not mapped in this store"); } } diff --git a/barge-store/src/test/java/org/robotninjas/barge/store/StoreResourceTest.java b/barge-store/src/test/java/org/robotninjas/barge/store/StoreResourceTest.java index ee44ce5..cd34b7c 100644 --- a/barge-store/src/test/java/org/robotninjas/barge/store/StoreResourceTest.java +++ b/barge-store/src/test/java/org/robotninjas/barge/store/StoreResourceTest.java @@ -55,6 +55,15 @@ public void onGETReturns200WithFoundValueGivenStoreContainsRequestedKey() throws assertThat(response.readEntity(byte[].class)).isEqualTo(value); } + @Test + public void onGETReturns404GivenStoreDoesNotContainRequestedKey() throws Exception { + when(raftStore.read("foo")).thenReturn(Optional.absent()); + + Response response = client().target("/foo").request().get(Response.class); + + assertThat(response.getStatus()).isEqualTo(HttpStatus.NOT_FOUND_404); + } + @Override protected Application configure() { raftStore = mock(RaftStore.class); From abab6e886c30bb3c24b112a45473f9aa3a6194e0 Mon Sep 17 00:00:00 2001 From: Arnaud Bailly Date: Mon, 23 Jun 2014 17:33:35 +0200 Subject: [PATCH 04/10] extracted builder pattern for Application this should make it easier to inject custom resources, providers and binders into the context in effect the RaftApplication acts as a container --- .../barge/jaxrs/RaftApplication.java | 68 +++++++++++++++---- .../barge/jaxrs/RaftJdkServer.java | 8 ++- .../barge/jaxrs/ws/RaftJettyServer.java | 11 ++- 3 files changed, 72 insertions(+), 15 deletions(-) diff --git a/barge-jax-rs/src/main/java/org/robotninjas/barge/jaxrs/RaftApplication.java b/barge-jax-rs/src/main/java/org/robotninjas/barge/jaxrs/RaftApplication.java index fdf94ab..18edbad 100644 --- a/barge-jax-rs/src/main/java/org/robotninjas/barge/jaxrs/RaftApplication.java +++ b/barge-jax-rs/src/main/java/org/robotninjas/barge/jaxrs/RaftApplication.java @@ -5,28 +5,35 @@ import com.google.common.collect.Lists; import com.google.inject.Guice; import com.google.inject.Injector; + import org.glassfish.hk2.api.Factory; import org.glassfish.hk2.utilities.Binder; import org.glassfish.hk2.utilities.binding.AbstractBinder; + import org.glassfish.jersey.server.ResourceConfig; + import org.robotninjas.barge.ClusterConfig; import org.robotninjas.barge.StateMachine; import org.robotninjas.barge.state.Raft; import org.robotninjas.barge.state.RaftProtocolListener; import org.robotninjas.barge.state.StateTransitionListener; import org.robotninjas.barge.utils.Files; + import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import javax.annotation.Nonnull; -import javax.annotation.Nullable; import java.io.File; import java.io.IOException; + import java.net.URI; + import java.nio.ByteBuffer; import java.util.Collections; import java.util.List; +import javax.annotation.Nonnull; +import javax.annotation.Nullable; + /** */ @@ -125,16 +132,53 @@ public void clean() throws IOException { public void stop() { injector.transform(new Function() { - @Nullable - @Override - public Object apply(@Nullable Injector input) { - Raft instance = null; - if (input != null) { - instance = input.getInstance(Raft.class); - instance.stop(); + @Nullable + @Override + public Object apply(@Nullable Injector input) { + Raft instance = null; + + if (input != null) { + instance = input.getInstance(Raft.class); + instance.stop(); + } + + return instance; } - return instance; - } - }); + }); + } + + public static class Builder { + private int serverIndex; + private URI[] uris; + private File logDir; + private StateTransitionListener[] listeners; + + public Builder setServerIndex(int serverIndex) { + this.serverIndex = serverIndex; + + return this; + } + + public Builder setUris(URI[] uris) { + this.uris = uris; + + return this; + } + + public Builder setLogDir(File logDir) { + this.logDir = logDir; + + return this; + } + + public Builder setListeners(StateTransitionListener... listeners) { + this.listeners = listeners; + + return this; + } + + public RaftApplication build() { + return new RaftApplication(serverIndex, uris, logDir, listeners); + } } } diff --git a/barge-jax-rs/src/main/java/org/robotninjas/barge/jaxrs/RaftJdkServer.java b/barge-jax-rs/src/main/java/org/robotninjas/barge/jaxrs/RaftJdkServer.java index 5572e91..3eda6b0 100644 --- a/barge-jax-rs/src/main/java/org/robotninjas/barge/jaxrs/RaftJdkServer.java +++ b/barge-jax-rs/src/main/java/org/robotninjas/barge/jaxrs/RaftJdkServer.java @@ -16,14 +16,18 @@ package org.robotninjas.barge.jaxrs; import com.google.common.base.Throwables; + import com.sun.net.httpserver.HttpServer; + import org.glassfish.jersey.jdkhttp.JdkHttpServerFactory; -import javax.ws.rs.core.UriBuilder; import java.io.File; import java.io.IOException; + import java.net.URI; +import javax.ws.rs.core.UriBuilder; + /** * A dedicated server for an instance of Raft using JDK's embedded HTTP server. @@ -39,7 +43,7 @@ public class RaftJdkServer implements RaftServer { public RaftJdkServer(int serverIndex, URI[] uris, File logDir) { this.serverIndex = serverIndex; this.uris = uris; - this.application = new RaftApplication(serverIndex, uris, logDir); + this.application = new RaftApplication.Builder().setServerIndex(serverIndex).setUris(uris).setLogDir(logDir).build(); } diff --git a/barge-jax-rs/src/main/java/org/robotninjas/barge/jaxrs/ws/RaftJettyServer.java b/barge-jax-rs/src/main/java/org/robotninjas/barge/jaxrs/ws/RaftJettyServer.java index be701af..52eee14 100644 --- a/barge-jax-rs/src/main/java/org/robotninjas/barge/jaxrs/ws/RaftJettyServer.java +++ b/barge-jax-rs/src/main/java/org/robotninjas/barge/jaxrs/ws/RaftJettyServer.java @@ -1,9 +1,9 @@ package org.robotninjas.barge.jaxrs.ws; import com.google.common.base.Throwables; - import com.google.common.collect.Lists; import com.google.common.io.CharStreams; + import org.eclipse.jetty.server.Server; import org.eclipse.jetty.server.ServerConnector; import org.eclipse.jetty.servlet.ServletContextHandler; @@ -52,6 +52,14 @@ public class RaftJettyServer implements RaftServer { private final RaftApplication raftApplication; private final WsEventListener events; + public RaftJettyServer(int serverIndex, URI[] uris, File logDir) { + server = new Server(); + events = new WsEventListener(); + raftApplication = new RaftApplication.Builder().setServerIndex(serverIndex).setUris(uris).setLogDir(logDir) + .setListeners(events) + .build(); + } + public static void main(String[] args) throws IOException, URISyntaxException { muteJul(); @@ -102,6 +110,7 @@ public static void main(String[] args) throws IOException, URISyntaxException { } private static void waitForInput() throws IOException { + //noinspection ResultOfMethodCallIgnored System.in.read(); } From e8d993f4fa68afd58babde79da47a4e3f0598737 Mon Sep 17 00:00:00 2001 From: Arnaud Bailly Date: Mon, 23 Jun 2014 19:18:08 +0200 Subject: [PATCH 05/10] completed basic barge store * moved main application from barge-jax-rs to barge-store * extracted Leaders class to encapsulate enquiring stabilization of the cluster * moved script packagers (for demo purpose) to barge-store --- barge-jax-rs/pom.xml | 36 ---- .../org/robotninjas/barge/jaxrs/Leaders.java | 69 ++++++ .../barge/jaxrs/RaftApplication.java | 81 ++++--- .../barge/jaxrs/ws/RaftJettyServer.java | 134 ++---------- .../barge/jaxrs/JettyDeploymentTest.java | 4 +- .../robotninjas/barge/jaxrs/ServerTest.java | 34 +-- .../barge/jaxrs/ws/RaftJettyServerTest.java | 9 +- {barge-jax-rs => barge-store}/barge.conf | 0 {barge-jax-rs => barge-store}/init.sh | 0 {barge-jax-rs => barge-store}/pack.sh | 0 barge-store/pom.xml | 38 ++++ {barge-jax-rs => barge-store}/send.sh | 0 .../barge/store/OperationsSerializer.java | 14 +- .../barge/store/RaftStoreInstance.java | 30 +-- .../barge/store/RaftStoreServer.java | 197 ++++++++++++++++++ .../barge/store/StoreResource.java | 2 +- .../barge/store/StoreStateMachine.java | 32 +++ .../barge/store/RaftStoreInstanceTest.java | 32 +-- .../barge/store/StoreResourceTest.java | 8 +- .../barge/store/StoreServerTest.java | 65 ++++++ .../barge/store/StoreStateMachineTest.java | 44 ++++ 21 files changed, 556 insertions(+), 273 deletions(-) create mode 100644 barge-jax-rs/src/main/java/org/robotninjas/barge/jaxrs/Leaders.java rename {barge-jax-rs => barge-store}/barge.conf (100%) rename {barge-jax-rs => barge-store}/init.sh (100%) rename {barge-jax-rs => barge-store}/pack.sh (100%) rename {barge-jax-rs => barge-store}/send.sh (100%) create mode 100644 barge-store/src/main/java/org/robotninjas/barge/store/RaftStoreServer.java create mode 100644 barge-store/src/main/java/org/robotninjas/barge/store/StoreStateMachine.java create mode 100644 barge-store/src/test/java/org/robotninjas/barge/store/StoreServerTest.java create mode 100644 barge-store/src/test/java/org/robotninjas/barge/store/StoreStateMachineTest.java diff --git a/barge-jax-rs/pom.xml b/barge-jax-rs/pom.xml index 00eed4f..f700dbf 100644 --- a/barge-jax-rs/pom.xml +++ b/barge-jax-rs/pom.xml @@ -126,40 +126,4 @@ - - barge-http - - - maven-shade-plugin - 2.1 - - - package - - shade - - - - - *:* - - META-INF/*.SF - META-INF/*.DSA - META-INF/*.RSA - - - - - - - org.robotninjas.barge.jaxrs.ws.RaftJettyServer - - - - - - - - - \ No newline at end of file diff --git a/barge-jax-rs/src/main/java/org/robotninjas/barge/jaxrs/Leaders.java b/barge-jax-rs/src/main/java/org/robotninjas/barge/jaxrs/Leaders.java new file mode 100644 index 0000000..3c7c0e4 --- /dev/null +++ b/barge-jax-rs/src/main/java/org/robotninjas/barge/jaxrs/Leaders.java @@ -0,0 +1,69 @@ +package org.robotninjas.barge.jaxrs; + +import org.robotninjas.barge.state.Raft; +import org.robotninjas.barge.utils.Prober; + +import java.net.URI; + +import java.util.concurrent.Callable; + +import javax.ws.rs.client.Client; + + +/** + * Utility to query for leader in a cluster and wait for cluster to reach a stable state with one leader. + */ +public class Leaders { + + private final URI[] uris; + + public Leaders(URI[] uris) { + this.uris = uris; + } + + public boolean isLeader(Client client, URI uri) { + return client.target(uri).path("/raft/state").request().get(Raft.StateType.class).equals(Raft.StateType.LEADER); + } + + /** + * Retrieves the URI of the current leader in the cluster. + * + * @param client HTTP client to use + * @return the URI of the current leader. + * @throws java.lang.IllegalStateException if there is no current leader in the cluster. + */ + public URI getLeader(Client client) { + + if (isLeader(client, uris[0])) + return uris[0]; + + if (isLeader(client, uris[1])) + return uris[1]; + + if (isLeader(client, uris[2])) + return uris[2]; + + throw new IllegalStateException("expected one server to be a leader"); + } + + /** + * Wait a certain amount of time for a leader to emerge in this cluster. + * @param client the HTTP client to use. + * @param timeoutInMs timeout in milliseconds before giving up. + */ + public void waitForALeader(final Client client, int timeoutInMs) { + new Prober(new Callable() { + @Override + public Boolean call() throws Exception { + + for (URI uri : uris) { + + if (isLeader(client, uri)) + return true; + } + + return false; + } + }).probe(timeoutInMs); + } +} diff --git a/barge-jax-rs/src/main/java/org/robotninjas/barge/jaxrs/RaftApplication.java b/barge-jax-rs/src/main/java/org/robotninjas/barge/jaxrs/RaftApplication.java index 18edbad..c591f22 100644 --- a/barge-jax-rs/src/main/java/org/robotninjas/barge/jaxrs/RaftApplication.java +++ b/barge-jax-rs/src/main/java/org/robotninjas/barge/jaxrs/RaftApplication.java @@ -3,36 +3,31 @@ import com.google.common.base.Function; import com.google.common.base.Optional; import com.google.common.collect.Lists; +import com.google.common.collect.Sets; import com.google.inject.Guice; import com.google.inject.Injector; - import org.glassfish.hk2.api.Factory; import org.glassfish.hk2.utilities.Binder; import org.glassfish.hk2.utilities.binding.AbstractBinder; - import org.glassfish.jersey.server.ResourceConfig; - import org.robotninjas.barge.ClusterConfig; import org.robotninjas.barge.StateMachine; import org.robotninjas.barge.state.Raft; import org.robotninjas.barge.state.RaftProtocolListener; import org.robotninjas.barge.state.StateTransitionListener; import org.robotninjas.barge.utils.Files; - import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import javax.annotation.Nonnull; +import javax.annotation.Nullable; import java.io.File; import java.io.IOException; - import java.net.URI; - import java.nio.ByteBuffer; -import java.util.Collections; +import java.util.Arrays; import java.util.List; - -import javax.annotation.Nonnull; -import javax.annotation.Nullable; +import java.util.Set; /** @@ -44,40 +39,33 @@ public class RaftApplication { private final int serverIndex; private final URI[] uris; private final File logDir; - private final List transitionListeners; private final List protocolListeners; + private final Set> resources = Sets.newHashSet(); + private final Set instances = Sets.newHashSet(); + private final StateMachine stateMachine; private Optional injector = Optional.absent(); - public RaftApplication(int serverIndex, URI[] uris, File logDir, Iterable transitionListener, Iterable protocolListener) { + public RaftApplication(int serverIndex, URI[] uris, File logDir, StateMachine stateMachine, Set> resources, + Set instances, + Iterable transitionListener, Iterable protocolListener) { this.serverIndex = serverIndex; this.uris = uris; this.logDir = logDir; + this.stateMachine = stateMachine; + this.resources.addAll(resources); + this.instances.addAll(instances); this.transitionListeners = Lists.newArrayList(transitionListener); this.protocolListeners = Lists.newArrayList(protocolListener); } - public RaftApplication(int serverIndex, URI[] uris, File logDir) { - this(serverIndex,uris,logDir, Collections.emptyList(),Collections.emptyList()); - } - public ResourceConfig makeResourceConfig() { - ClusterConfig clusterConfig = HttpClusterConfig.from(new HttpReplica(uris[serverIndex]), - remotes()); + ClusterConfig clusterConfig = HttpClusterConfig.from(new HttpReplica(uris[serverIndex]), remotes()); if (!logDir.exists() && !logDir.mkdirs()) logger.warn("failed to create directories for storing logs, bad things will happen"); - StateMachine stateMachine = new StateMachine() { - int i = 0; - - @Override - public Object applyOperation(@Nonnull ByteBuffer entry) { - return i++; - } - }; - final JaxRsRaftModule raftModule = new JaxRsRaftModule(clusterConfig, logDir, stateMachine, 1500, transitionListeners, protocolListeners); injector = Optional.of(Guice.createInjector(raftModule)); @@ -111,6 +99,8 @@ public void dispose(Raft raft) { resourceConfig.register(BargeResource.class); resourceConfig.register(Jackson.customJacksonProvider()); + resourceConfig.registerClasses(resources); + resourceConfig.registerInstances(instances); resourceConfig.register(binder); return resourceConfig; @@ -151,7 +141,18 @@ public static class Builder { private int serverIndex; private URI[] uris; private File logDir; - private StateTransitionListener[] listeners; + private StateTransitionListener[] transitionListeners = new StateTransitionListener[0]; + private Iterable protocolListeners = Lists.newArrayList(); + private Set> resources = Sets.newHashSet(); + private Set instances = Sets.newHashSet(); + private StateMachine stateMachine = new StateMachine() { + int i = 0; + + @Override + public Object applyOperation(@Nonnull ByteBuffer entry) { + return i++; + } + }; public Builder setServerIndex(int serverIndex) { this.serverIndex = serverIndex; @@ -171,14 +172,32 @@ public Builder setLogDir(File logDir) { return this; } - public Builder setListeners(StateTransitionListener... listeners) { - this.listeners = listeners; + public Builder setTransitionListeners(StateTransitionListener... listeners) { + this.transitionListeners = listeners; return this; } public RaftApplication build() { - return new RaftApplication(serverIndex, uris, logDir, listeners); + return new RaftApplication(serverIndex, uris, logDir, stateMachine, resources, instances, Arrays.asList(transitionListeners), protocolListeners); + } + + public Builder register(Class resource) { + this.resources.add(resource); + + return this; + } + + public Builder setStateMachine(StateMachine stateMachine) { + this.stateMachine = stateMachine; + + return this; + } + + public Builder registerInstance(Object o) { + this.instances.add(o); + + return this; } } } diff --git a/barge-jax-rs/src/main/java/org/robotninjas/barge/jaxrs/ws/RaftJettyServer.java b/barge-jax-rs/src/main/java/org/robotninjas/barge/jaxrs/ws/RaftJettyServer.java index 52eee14..c14ba92 100644 --- a/barge-jax-rs/src/main/java/org/robotninjas/barge/jaxrs/ws/RaftJettyServer.java +++ b/barge-jax-rs/src/main/java/org/robotninjas/barge/jaxrs/ws/RaftJettyServer.java @@ -1,31 +1,16 @@ package org.robotninjas.barge.jaxrs.ws; import com.google.common.base.Throwables; -import com.google.common.collect.Lists; -import com.google.common.io.CharStreams; - import org.eclipse.jetty.server.Server; import org.eclipse.jetty.server.ServerConnector; import org.eclipse.jetty.servlet.ServletContextHandler; import org.eclipse.jetty.servlet.ServletHolder; - import org.glassfish.jersey.servlet.ServletContainer; - import org.robotninjas.barge.jaxrs.RaftApplication; import org.robotninjas.barge.jaxrs.RaftServer; -import org.robotninjas.barge.state.RaftProtocolListener; -import org.robotninjas.barge.state.StateTransitionListener; -import org.slf4j.bridge.SLF4JBridgeHandler; -import java.io.File; -import java.io.FileReader; import java.io.IOException; - import java.net.URI; -import java.net.URISyntaxException; -import java.util.Collections; -import java.util.List; -import java.util.logging.Level; /** @@ -37,115 +22,14 @@ */ public class RaftJettyServer implements RaftServer { - private static final String help = "Usage: java -jar barge-http.jar [options] \n" + - "Options:\n" + - " -h : Displays this help message\n" + - " -c : Use given configuration file for cluster configuration\n" + - " This file is a simple property file with indices as keys and URIs as values, eg. like\n\n" + - " 0=http://localhost:1234\n" + - " 1=http://localhost:3456\n" + - " 2=http://localhost:4567\n\n" + - " Default is './barge.conf'\n" + - " : Index of this server in the cluster configuration\n"; - private final Server server; private final RaftApplication raftApplication; private final WsEventListener events; - public RaftJettyServer(int serverIndex, URI[] uris, File logDir) { + public RaftJettyServer(RaftApplication.Builder builder) { server = new Server(); events = new WsEventListener(); - raftApplication = new RaftApplication.Builder().setServerIndex(serverIndex).setUris(uris).setLogDir(logDir) - .setListeners(events) - .build(); - } - - public static void main(String[] args) throws IOException, URISyntaxException { - muteJul(); - - File clusterConfiguration = new File("barge.conf"); - int index = -1; - - for (int i = 0; i < args.length; i++) { - - switch (args[i]) { - - case "-c": - clusterConfiguration = new File(args[++i]); - - break; - - case "-h": - usage(); - System.exit(0); - - default: - - try { - index = Integer.parseInt(args[i].trim()); - } catch (NumberFormatException e) { - usage(); - System.exit(1); - } - - break; - } - } - - if (index == -1) { - usage(); - System.exit(1); - } - - URI[] uris = readConfiguration(clusterConfiguration); - - RaftJettyServer server = new RaftJettyServer(index, uris, new File("log" + index)).start(uris[index].getPort()); - - waitForInput(); - - server.stop(); - - System.out.println("Bye!"); - System.exit(0); - } - - private static void waitForInput() throws IOException { - - //noinspection ResultOfMethodCallIgnored - System.in.read(); - } - - private static void muteJul() { - java.util.logging.Logger.getLogger("").setLevel(Level.ALL); - SLF4JBridgeHandler.removeHandlersForRootLogger(); - SLF4JBridgeHandler.install(); - } - - private static URI[] readConfiguration(File clusterConfiguration) throws IOException, URISyntaxException { - List uris = Lists.newArrayList(); - - int lineNumber = 1; - - for (String line : CharStreams.readLines(new FileReader(clusterConfiguration))) { - String[] pair = line.split("="); - - if (pair.length != 2) - throw new IOException("Invalid cluster configuration at line " + lineNumber); - - uris.add(Integer.parseInt(pair[0].trim()), new URI(pair[1].trim())); - } - - return uris.toArray(new URI[uris.size()]); - } - - private static void usage() { - System.out.println(help); - } - - public RaftJettyServer(int serverIndex, URI[] uris, File logDir) { - server = new Server(); - events = new WsEventListener(); - raftApplication = new RaftApplication(serverIndex, uris, logDir, Collections.singleton(events), Collections.singleton(events)); + raftApplication = builder.setTransitionListeners(events).build(); } public RaftJettyServer start(int port) { @@ -201,4 +85,18 @@ public void clean() { } } + public static class Builder { + private RaftApplication.Builder builder; + + public Builder setApplicationBuilder(RaftApplication.Builder builder) { + this.builder = builder; + + return this; + } + + public RaftJettyServer build() { + return new RaftJettyServer(builder); + } + + } } diff --git a/barge-jax-rs/src/test/java/org/robotninjas/barge/jaxrs/JettyDeploymentTest.java b/barge-jax-rs/src/test/java/org/robotninjas/barge/jaxrs/JettyDeploymentTest.java index e0afa73..17fe66f 100644 --- a/barge-jax-rs/src/test/java/org/robotninjas/barge/jaxrs/JettyDeploymentTest.java +++ b/barge-jax-rs/src/test/java/org/robotninjas/barge/jaxrs/JettyDeploymentTest.java @@ -28,7 +28,9 @@ public class JettyDeploymentTest extends ServerTest { @Override protected RaftJettyServer createServer(int serverIndex, URI[] uris1, File logDir) { - return new RaftJettyServer(serverIndex, uris1, logDir); + return new RaftJettyServer.Builder().setApplicationBuilder(new RaftApplication.Builder().setServerIndex( + serverIndex).setUris(uris1).setLogDir(logDir)) + .build(); } } diff --git a/barge-jax-rs/src/test/java/org/robotninjas/barge/jaxrs/ServerTest.java b/barge-jax-rs/src/test/java/org/robotninjas/barge/jaxrs/ServerTest.java index 5b07ad3..9bd1dc3 100644 --- a/barge-jax-rs/src/test/java/org/robotninjas/barge/jaxrs/ServerTest.java +++ b/barge-jax-rs/src/test/java/org/robotninjas/barge/jaxrs/ServerTest.java @@ -8,14 +8,11 @@ import org.junit.Test; import static org.robotninjas.barge.jaxrs.Logs.uniqueLog; -import org.robotninjas.barge.state.Raft; -import org.robotninjas.barge.utils.Prober; import java.io.File; import java.net.URI; -import java.util.concurrent.Callable; import java.util.logging.Level; import java.util.logging.Logger; @@ -38,6 +35,8 @@ public abstract class ServerTest> { private T httpServer1; private T httpServer2; private T httpServer3; + private Leaders leaders; + @Before public void setUp() throws Exception { @@ -47,6 +46,8 @@ public void setUp() throws Exception { uris[1] = new URI("http://localhost:56790/"); uris[2] = new URI("http://localhost:56791/"); + leaders = new Leaders(uris); + httpServer1 = createServer(0, uris, uniqueLog()).start(56789); httpServer2 = createServer(1, uris, uniqueLog()).start(56790); httpServer3 = createServer(2, uris, uniqueLog()).start(56791); @@ -71,14 +72,9 @@ public void can_commit_data_to_leader_instance() throws Exception { client.target(uris[1]).path("/raft/init").request().post(Entity.json("")); client.target(uris[2]).path("/raft/init").request().post(Entity.json("")); - new Prober(new Callable() { - @Override - public Boolean call() throws Exception { - return isLeader(client, uris[0]) || isLeader(client, uris[1]) || isLeader(client, uris[2]); - } - }).probe(10000); + leaders.waitForALeader(client, 10000); - URI leaderURI = getLeader(client); + URI leaderURI = leaders.getLeader(client); Response result = client.target(leaderURI) .path("/raft/commit") @@ -91,22 +87,4 @@ public Boolean call() throws Exception { protected abstract T createServer(int serverIndex, URI[] uris1, File logDir); - private URI getLeader(Client client) { - - if (isLeader(client, uris[0])) - return uris[0]; - - if (isLeader(client, uris[1])) - return uris[1]; - - if (isLeader(client, uris[2])) - return uris[2]; - - throw new IllegalStateException("expected one server to be a leader"); - } - - private boolean isLeader(Client client, URI uri) { - return client.target(uri).path("/raft/state").request().get(Raft.StateType.class).equals(Raft.StateType.LEADER); - } - } diff --git a/barge-jax-rs/src/test/java/org/robotninjas/barge/jaxrs/ws/RaftJettyServerTest.java b/barge-jax-rs/src/test/java/org/robotninjas/barge/jaxrs/ws/RaftJettyServerTest.java index 82f7f44..854369e 100644 --- a/barge-jax-rs/src/test/java/org/robotninjas/barge/jaxrs/ws/RaftJettyServerTest.java +++ b/barge-jax-rs/src/test/java/org/robotninjas/barge/jaxrs/ws/RaftJettyServerTest.java @@ -12,6 +12,8 @@ import org.junit.Before; import org.junit.Test; import org.robotninjas.barge.jaxrs.Jackson; +import static org.robotninjas.barge.jaxrs.Logs.uniqueLog; +import org.robotninjas.barge.jaxrs.RaftApplication; import org.robotninjas.barge.utils.Prober; import javax.ws.rs.client.Client; @@ -38,8 +40,11 @@ public class RaftJettyServerTest { @Before public void setUp() throws Exception { - this.server = new RaftJettyServer(0, new URI[] { new URI("http://localhost:12345") }, - uniqueLog()).start(0); + final URI[] uris = new URI[] { new URI("http://localhost:12345") }; + this.server = new RaftJettyServer.Builder().setApplicationBuilder(new RaftApplication.Builder().setServerIndex( + 0).setUris( + uris).setLogDir(uniqueLog())).build().start( + 0); this.uri = server.getPort(); wsClient.start(); diff --git a/barge-jax-rs/barge.conf b/barge-store/barge.conf similarity index 100% rename from barge-jax-rs/barge.conf rename to barge-store/barge.conf diff --git a/barge-jax-rs/init.sh b/barge-store/init.sh similarity index 100% rename from barge-jax-rs/init.sh rename to barge-store/init.sh diff --git a/barge-jax-rs/pack.sh b/barge-store/pack.sh similarity index 100% rename from barge-jax-rs/pack.sh rename to barge-store/pack.sh diff --git a/barge-store/pom.xml b/barge-store/pom.xml index 6cd610d..922cdac 100644 --- a/barge-store/pom.xml +++ b/barge-store/pom.xml @@ -58,4 +58,42 @@ + + + barge-http + + + maven-shade-plugin + 2.1 + + + package + + shade + + + + + *:* + + META-INF/*.SF + META-INF/*.DSA + META-INF/*.RSA + + + + + + + org.robotninjas.barge.store.RaftStoreServer + + + + + + + + + + \ No newline at end of file diff --git a/barge-jax-rs/send.sh b/barge-store/send.sh similarity index 100% rename from barge-jax-rs/send.sh rename to barge-store/send.sh diff --git a/barge-store/src/main/java/org/robotninjas/barge/store/OperationsSerializer.java b/barge-store/src/main/java/org/robotninjas/barge/store/OperationsSerializer.java index 3ce5854..938fb22 100644 --- a/barge-store/src/main/java/org/robotninjas/barge/store/OperationsSerializer.java +++ b/barge-store/src/main/java/org/robotninjas/barge/store/OperationsSerializer.java @@ -12,6 +12,8 @@ import java.io.*; +import java.nio.ByteBuffer; + /** */ @@ -54,13 +56,19 @@ public Write deserialize(byte[] bytes) { try { return (Write) new ObjectInputStream(in).readObject(); - } catch (IOException e) { - throw Throwables.propagate(e); - } catch (ClassNotFoundException e) { + } catch (IOException | ClassNotFoundException e) { throw Throwables.propagate(e); } } + public Write deserialize(ByteBuffer entry) { + byte[] bytes = new byte[entry.remaining()]; + + entry.get(bytes); + + return deserialize(bytes); + } + public static class OperationsSerializationException extends RuntimeException { diff --git a/barge-store/src/main/java/org/robotninjas/barge/store/RaftStoreInstance.java b/barge-store/src/main/java/org/robotninjas/barge/store/RaftStoreInstance.java index 7b1d8c4..b44eb23 100644 --- a/barge-store/src/main/java/org/robotninjas/barge/store/RaftStoreInstance.java +++ b/barge-store/src/main/java/org/robotninjas/barge/store/RaftStoreInstance.java @@ -3,33 +3,28 @@ import com.google.common.base.Optional; import static com.google.common.base.Optional.fromNullable; import static com.google.common.base.Throwables.propagate; -import com.google.common.collect.Maps; import org.robotninjas.barge.RaftException; -import org.robotninjas.barge.StateMachine; import org.robotninjas.barge.state.Raft; -import java.nio.ByteBuffer; - -import java.util.Map; import java.util.concurrent.ExecutionException; -import javax.annotation.Nonnull; - import javax.inject.Inject; /** */ -public class RaftStoreInstance implements RaftStore, StateMachine { +public class RaftStoreInstance implements RaftStore { private final Raft raft; private final OperationsSerializer operationsSerializer = new OperationsSerializer(); - private final Map store = Maps.newConcurrentMap(); + + private final StoreStateMachine storeStateMachine; @Inject - public RaftStoreInstance(Raft raft) { + public RaftStoreInstance(Raft raft, StoreStateMachine storeStateMachine) { this.raft = raft; + this.storeStateMachine = storeStateMachine; } @Override @@ -37,25 +32,14 @@ public byte[] write(Write write) { try { return (byte[]) raft.commitOperation(operationsSerializer.serialize(write)).get(); - } catch (RaftException e) { - throw propagate(e); - } catch (InterruptedException e) { - throw propagate(e); - } catch (ExecutionException e) { + } catch (RaftException | InterruptedException | ExecutionException e) { throw propagate(e); } } @Override public Optional read(String key) { - return fromNullable(store.get(key)); - } - - @Override - public Object applyOperation(@Nonnull ByteBuffer entry) { - Write write = operationsSerializer.deserialize(entry.array()); - - return store.put(write.getKey(), write.getValue()); + return fromNullable(storeStateMachine.get(key)); } diff --git a/barge-store/src/main/java/org/robotninjas/barge/store/RaftStoreServer.java b/barge-store/src/main/java/org/robotninjas/barge/store/RaftStoreServer.java new file mode 100644 index 0000000..4106c3e --- /dev/null +++ b/barge-store/src/main/java/org/robotninjas/barge/store/RaftStoreServer.java @@ -0,0 +1,197 @@ +package org.robotninjas.barge.store; + +import com.google.common.collect.Lists; +import com.google.common.io.CharStreams; + +import org.glassfish.hk2.utilities.Binder; +import org.glassfish.hk2.utilities.binding.AbstractBinder; + +import org.robotninjas.barge.jaxrs.Jackson; +import org.robotninjas.barge.jaxrs.RaftApplication; +import org.robotninjas.barge.jaxrs.ws.RaftJettyServer; + +import org.slf4j.bridge.SLF4JBridgeHandler; + +import java.io.File; +import java.io.FileReader; +import java.io.IOException; + +import java.net.URI; +import java.net.URISyntaxException; + +import java.util.List; +import java.util.logging.Level; + +import javax.inject.Singleton; + +import javax.ws.rs.client.Client; +import javax.ws.rs.client.ClientBuilder; +import javax.ws.rs.client.Entity; + + +/** + * A standalone store accepting put/get operations through HTTP requests. + * + *

This is also the main class for the barge-store application. A {@link org.robotninjas.barge.store.RaftStoreServer} instnace + * encapsulates an instance of {@link org.robotninjas.barge.jaxrs.ws.RaftJettyServer} which holds the main Jersey resources + * context. This context is extended with a REST resource {@link org.robotninjas.barge.store.StoreResource} mapped to the URI + * /raft/store/*, a {@link org.robotninjas.barge.store.RaftStore} implementation backed by a {@link org.robotninjas.barge.store.StoreStateMachine} + * which holds the result of Barge's cluster operations.

+ * + *

Within this implementation, writes are always sent to the cluster and reads always done against the current backing + * state machine, implying that only committed read can ever be seen.

+ */ +public class RaftStoreServer { + + private static final String help = "Usage: java -jar barge-store.jar [options] \n" + + "Options:\n" + + " -h : Displays this help message\n" + + " -c : Use given configuration file for cluster configuration\n" + + " This file is a simple property file with indices as keys and URIs as values, eg. like\n\n" + + " 0=http://localhost:1234\n" + + " 1=http://localhost:3456\n" + + " 2=http://localhost:4567\n\n" + + " Default is './barge.conf'\n" + + " : Index of this server in the cluster configuration\n"; + + + private final int serverIndex; + private final URI[] clusterURIs; + private final File logDir; + + private RaftJettyServer server; + + public RaftStoreServer(int serverIndex, URI[] clusterURIs) { + this(serverIndex, clusterURIs, new File("log" + serverIndex)); + } + + public RaftStoreServer(int serverIndex, URI[] clusterURIs, File logDir) { + this.serverIndex = serverIndex; + this.clusterURIs = clusterURIs; + this.logDir = logDir; + } + + + public static void main(String[] args) throws IOException, URISyntaxException { + muteJul(); + + File clusterConfiguration = new File("barge.conf"); + int index = -1; + + for (int i = 0; i < args.length; i++) { + + switch (args[i]) { + + case "-c": + clusterConfiguration = new File(args[++i]); + + break; + + case "-h": + usage(); + System.exit(0); + + default: + + try { + index = Integer.parseInt(args[i].trim()); + } catch (NumberFormatException e) { + usage(); + System.exit(1); + } + + break; + } + } + + if (index == -1) { + usage(); + System.exit(1); + } + + URI[] uris = readConfiguration(clusterConfiguration); + + final File logDir = new File("log" + index); + + RaftStoreServer server = new RaftStoreServer(index, uris, logDir); + + server.start(uris[index].getPort()); + + waitForInput(); + + server.stop(); + + System.out.println("Bye!"); + System.exit(0); + } + + private static void waitForInput() throws IOException { + + //noinspection ResultOfMethodCallIgnored + System.in.read(); + } + + private static void muteJul() { + java.util.logging.Logger.getLogger("").setLevel(Level.ALL); + SLF4JBridgeHandler.removeHandlersForRootLogger(); + SLF4JBridgeHandler.install(); + } + + private static URI[] readConfiguration(File clusterConfiguration) throws IOException, URISyntaxException { + List uris = Lists.newArrayList(); + + int lineNumber = 1; + + for (String line : CharStreams.readLines(new FileReader(clusterConfiguration))) { + String[] pair = line.split("="); + + if (pair.length != 2) + throw new IOException("Invalid cluster configuration at line " + lineNumber); + + uris.add(Integer.parseInt(pair[0].trim()), new URI(pair[1].trim())); + } + + return uris.toArray(new URI[uris.size()]); + } + + private static void usage() { + System.out.println(help); + } + + public void start(int port) { + final StoreStateMachine stateMachine = new StoreStateMachine(); + + // Bridges our store objects to the DI context of the main RaftApplication context and Guice modules + Binder binder = new AbstractBinder() { + @Override + protected void configure() { + bind(stateMachine).to(StoreStateMachine.class); + bind(RaftStoreInstance.class).to(RaftStore.class).in(Singleton.class); + } + }; + + server = + new RaftJettyServer.Builder().setApplicationBuilder(new RaftApplication.Builder() // + .setServerIndex(serverIndex) // + .setUris(clusterURIs) // + .setLogDir(logDir) // + .register(StoreResource.class) // + .registerInstance(binder) // + .setStateMachine(stateMachine)) // + .build(); + + server.start(port); + + // automatically call init on the registered raft instance + init(); + } + + private void init() { + final Client client = ClientBuilder.newBuilder().register(Jackson.customJacksonProvider()).build(); + client.target(clusterURIs[serverIndex]).path("/raft/init").request().post(Entity.json("")); + } + + public void stop() { + server.stop(); + } +} diff --git a/barge-store/src/main/java/org/robotninjas/barge/store/StoreResource.java b/barge-store/src/main/java/org/robotninjas/barge/store/StoreResource.java index adc81d5..c072d60 100644 --- a/barge-store/src/main/java/org/robotninjas/barge/store/StoreResource.java +++ b/barge-store/src/main/java/org/robotninjas/barge/store/StoreResource.java @@ -13,7 +13,7 @@ import javax.ws.rs.core.UriInfo; -@Path("/") +@Path("/store") @Produces(MediaType.APPLICATION_JSON) public class StoreResource { diff --git a/barge-store/src/main/java/org/robotninjas/barge/store/StoreStateMachine.java b/barge-store/src/main/java/org/robotninjas/barge/store/StoreStateMachine.java new file mode 100644 index 0000000..551064c --- /dev/null +++ b/barge-store/src/main/java/org/robotninjas/barge/store/StoreStateMachine.java @@ -0,0 +1,32 @@ +package org.robotninjas.barge.store; + +import com.google.common.collect.Maps; + +import org.robotninjas.barge.StateMachine; + +import java.nio.ByteBuffer; + +import java.util.Map; + +import javax.annotation.Nonnull; + + +/** + */ +public class StoreStateMachine implements StateMachine { + + private final OperationsSerializer operationsSerializer = new OperationsSerializer(); + private final Map store = Maps.newConcurrentMap(); + + @Override + public Object applyOperation(@Nonnull ByteBuffer entry) { + Write write = operationsSerializer.deserialize(entry); + + return store.put(write.getKey(), write.getValue()); + } + + + public byte[] get(String key) { + return store.get(key); + } +} diff --git a/barge-store/src/test/java/org/robotninjas/barge/store/RaftStoreInstanceTest.java b/barge-store/src/test/java/org/robotninjas/barge/store/RaftStoreInstanceTest.java index 8c85c9e..493da35 100644 --- a/barge-store/src/test/java/org/robotninjas/barge/store/RaftStoreInstanceTest.java +++ b/barge-store/src/test/java/org/robotninjas/barge/store/RaftStoreInstanceTest.java @@ -1,6 +1,5 @@ package org.robotninjas.barge.store; -import com.google.common.base.Optional; import com.google.common.util.concurrent.Futures; import static org.assertj.core.api.Assertions.assertThat; @@ -14,8 +13,6 @@ import org.robotninjas.barge.state.Raft; -import static java.nio.ByteBuffer.wrap; - public class RaftStoreInstanceTest { @@ -23,9 +20,11 @@ public class RaftStoreInstanceTest { private static final Write write = new Write("foo", new byte[] { 0x42 }); private final Raft raft = mock(Raft.class); + private final StoreStateMachine storeStateMachine = mock(StoreStateMachine.class); + private final OperationsSerializer serializer = new OperationsSerializer(); - private final RaftStoreInstance raftStoreInstance = new RaftStoreInstance(raft); + private final RaftStoreInstance raftStoreInstance = new RaftStoreInstance(raft, storeStateMachine); @Test public void commitSerializedWriteOperationToRaftThenReturnOldValueWhenWritingKeyValuePair() throws Exception { @@ -35,28 +34,9 @@ public void commitSerializedWriteOperationToRaftThenReturnOldValueWhenWritingKey } @Test - public void mapsKeyToValueWhenApplyingWriteOperation() throws Exception { - raftStoreInstance.applyOperation(wrap(serializer.serialize(write))); - - Optional read = raftStoreInstance.read("foo"); - - Assertions.assertThat(read).isPresent(); - assertThat(read.get()).isEqualTo(new byte[] { 0x42 }); - } - - @Test - public void applyingWriteReturnsOldValueGivenKeyIsAlreadyMapped() throws Exception { - raftStoreInstance.applyOperation(wrap(serializer.serialize(write))); - - byte[] old = (byte[]) raftStoreInstance.applyOperation(wrap(serializer.serialize(new Write("foo", new byte[] { 0x43 })))); - - assertThat(old).isEqualTo(new byte[] { 0x42 }); - } - - @Test - public void applyingWriteReturnsNullValueGivenKeyIsNotMapped() throws Exception { - byte[] old = (byte[]) raftStoreInstance.applyOperation(wrap(serializer.serialize(new Write("foo", new byte[] { 0x43 })))); + public void whenReadingReturnsAbsentOptionalIfKeyIsNotPresentInStore() throws Exception { + when(storeStateMachine.get("foo")).thenReturn(null); - assertThat(old).isNull(); + Assertions.assertThat(raftStoreInstance.read("foo")).isAbsent(); } } diff --git a/barge-store/src/test/java/org/robotninjas/barge/store/StoreResourceTest.java b/barge-store/src/test/java/org/robotninjas/barge/store/StoreResourceTest.java index cd34b7c..c1d1875 100644 --- a/barge-store/src/test/java/org/robotninjas/barge/store/StoreResourceTest.java +++ b/barge-store/src/test/java/org/robotninjas/barge/store/StoreResourceTest.java @@ -38,10 +38,10 @@ public class StoreResourceTest extends JerseyTest { public void onPUTReturns201WithLocationAndPreviousValueGivenRaftStoreCompletesSuccessfully() throws Exception { when(raftStore.write(write)).thenReturn(oldValue); - Response response = client().target("/foo").request().put(Entity.entity(value, APPLICATION_OCTET_STREAM_TYPE)); + Response response = client().target("/store/foo").request().put(Entity.entity(value, APPLICATION_OCTET_STREAM_TYPE)); assertThat(response.getStatus()).isEqualTo(HttpStatus.CREATED_201); - assertThat(response.getHeaderString("Location")).isEqualTo("http://localhost:9998/foo"); + assertThat(response.getHeaderString("Location")).isEqualTo("http://localhost:9998/store/foo"); assertThat(response.readEntity(byte[].class)).isEqualTo(oldValue); } @@ -49,7 +49,7 @@ public void onPUTReturns201WithLocationAndPreviousValueGivenRaftStoreCompletesSu public void onGETReturns200WithFoundValueGivenStoreContainsRequestedKey() throws Exception { when(raftStore.read("foo")).thenReturn(Optional.of(value)); - Response response = client().target("/foo").request().get(Response.class); + Response response = client().target("/store/foo").request().get(Response.class); assertThat(response.getStatus()).isEqualTo(HttpStatus.OK_200); assertThat(response.readEntity(byte[].class)).isEqualTo(value); @@ -59,7 +59,7 @@ public void onGETReturns200WithFoundValueGivenStoreContainsRequestedKey() throws public void onGETReturns404GivenStoreDoesNotContainRequestedKey() throws Exception { when(raftStore.read("foo")).thenReturn(Optional.absent()); - Response response = client().target("/foo").request().get(Response.class); + Response response = client().target("/store/foo").request().get(Response.class); assertThat(response.getStatus()).isEqualTo(HttpStatus.NOT_FOUND_404); } diff --git a/barge-store/src/test/java/org/robotninjas/barge/store/StoreServerTest.java b/barge-store/src/test/java/org/robotninjas/barge/store/StoreServerTest.java new file mode 100644 index 0000000..0da1172 --- /dev/null +++ b/barge-store/src/test/java/org/robotninjas/barge/store/StoreServerTest.java @@ -0,0 +1,65 @@ +package org.robotninjas.barge.store; + +import com.google.common.base.Throwables; + +import static org.assertj.core.api.Assertions.assertThat; + +import org.junit.After; +import org.junit.Before; +import org.junit.Test; + +import org.robotninjas.barge.jaxrs.Leaders; + +import java.net.URI; +import java.net.URISyntaxException; + +import javax.ws.rs.client.Client; +import javax.ws.rs.client.ClientBuilder; +import static javax.ws.rs.client.Entity.entity; +import static javax.ws.rs.core.MediaType.APPLICATION_OCTET_STREAM_TYPE; + + +/** + */ +public class StoreServerTest { + + private RaftStoreServer raftStoreServer; + private URI[] clusterURIs; + + @Before + public void startServer() throws URISyntaxException { + clusterURIs = new URI[] { serverURI() }; + raftStoreServer = new RaftStoreServer(0, clusterURIs); + + raftStoreServer.start(56789); + } + + @After + public void stopServer() { + raftStoreServer.stop(); + } + + @Test + public void canWriteAndReadValuesInAStoreInAClusterOf1() throws Exception { + final Client client = ClientBuilder.newBuilder().register(OperationsSerializer.jacksonModule()).build(); + + new Leaders(clusterURIs).waitForALeader(client, 10000); + + client.target(serverURI()).path("/raft/store/foo").request().put(entity(new byte[] { 0x1 }, APPLICATION_OCTET_STREAM_TYPE)); + + client.target(serverURI()).path("/raft/store/bar").request().put(entity(new byte[] { 0x2 }, APPLICATION_OCTET_STREAM_TYPE)); + + byte[] bytes = client.target(serverURI()).path("/raft/store/foo").request().get(byte[].class); + + assertThat(bytes).isEqualTo(new byte[] { 0x1 }); + } + + private URI serverURI() { + + try { + return new URI("http://localhost:56789/"); + } catch (URISyntaxException e) { + throw Throwables.propagate(e); + } + } +} diff --git a/barge-store/src/test/java/org/robotninjas/barge/store/StoreStateMachineTest.java b/barge-store/src/test/java/org/robotninjas/barge/store/StoreStateMachineTest.java new file mode 100644 index 0000000..1647c49 --- /dev/null +++ b/barge-store/src/test/java/org/robotninjas/barge/store/StoreStateMachineTest.java @@ -0,0 +1,44 @@ +package org.robotninjas.barge.store; + +import static org.assertj.core.api.Assertions.assertThat; + +import org.junit.Test; + +import static java.nio.ByteBuffer.wrap; + + +public class StoreStateMachineTest { + + + private static final Write write = new Write("foo", new byte[] { 0x42 }); + + private final OperationsSerializer serializer = new OperationsSerializer(); + + private final StoreStateMachine raftStoreInstance = new StoreStateMachine(); + + @Test + public void mapsKeyToValueWhenApplyingWriteOperation() throws Exception { + raftStoreInstance.applyOperation(wrap(serializer.serialize(write))); + + byte[] read = raftStoreInstance.get("foo"); + + assertThat(read).isEqualTo(new byte[] { 0x42 }); + } + + @Test + public void applyingWriteReturnsOldValueGivenKeyIsAlreadyMapped() throws Exception { + raftStoreInstance.applyOperation(wrap(serializer.serialize(write))); + + byte[] old = (byte[]) raftStoreInstance.applyOperation(wrap(serializer.serialize(new Write("foo", new byte[] { 0x43 })))); + + assertThat(old).isEqualTo(new byte[] { 0x42 }); + } + + @Test + public void applyingWriteReturnsNullValueGivenKeyIsNotMapped() throws Exception { + byte[] old = (byte[]) raftStoreInstance.applyOperation(wrap(serializer.serialize(new Write("foo", new byte[] { 0x43 })))); + + assertThat(old).isNull(); + } + +} From 445a916b3627d6c9b5d0b42765c998167859d2c2 Mon Sep 17 00:00:00 2001 From: Arnaud Bailly Date: Mon, 23 Jun 2014 21:32:23 +0200 Subject: [PATCH 06/10] fix double ListenableFuture issue when retrieving commit when committing to raft instance, returned Future itself encapsulates a future. We need some flatMap here... --- .../java/org/robotninjas/barge/store/RaftStoreInstance.java | 5 ++++- .../main/java/org/robotninjas/barge/store/StoreResource.java | 2 -- .../org/robotninjas/barge/store/RaftStoreInstanceTest.java | 2 +- .../java/org/robotninjas/barge/store/StoreServerTest.java | 5 ++++- barge-store/src/test/resources/marker | 1 + 5 files changed, 10 insertions(+), 5 deletions(-) create mode 100644 barge-store/src/test/resources/marker diff --git a/barge-store/src/main/java/org/robotninjas/barge/store/RaftStoreInstance.java b/barge-store/src/main/java/org/robotninjas/barge/store/RaftStoreInstance.java index b44eb23..9ae8c18 100644 --- a/barge-store/src/main/java/org/robotninjas/barge/store/RaftStoreInstance.java +++ b/barge-store/src/main/java/org/robotninjas/barge/store/RaftStoreInstance.java @@ -3,6 +3,7 @@ import com.google.common.base.Optional; import static com.google.common.base.Optional.fromNullable; import static com.google.common.base.Throwables.propagate; +import com.google.common.util.concurrent.ListenableFuture; import org.robotninjas.barge.RaftException; import org.robotninjas.barge.state.Raft; @@ -31,7 +32,9 @@ public RaftStoreInstance(Raft raft, StoreStateMachine storeStateMachine) { public byte[] write(Write write) { try { - return (byte[]) raft.commitOperation(operationsSerializer.serialize(write)).get(); + + // TODO this is ugly... + return ((ListenableFuture) raft.commitOperation(operationsSerializer.serialize(write)).get()).get(); } catch (RaftException | InterruptedException | ExecutionException e) { throw propagate(e); } diff --git a/barge-store/src/main/java/org/robotninjas/barge/store/StoreResource.java b/barge-store/src/main/java/org/robotninjas/barge/store/StoreResource.java index c072d60..74bebe7 100644 --- a/barge-store/src/main/java/org/robotninjas/barge/store/StoreResource.java +++ b/barge-store/src/main/java/org/robotninjas/barge/store/StoreResource.java @@ -5,7 +5,6 @@ import javax.inject.Inject; import javax.ws.rs.*; -import javax.ws.rs.core.MediaType; import static javax.ws.rs.core.MediaType.APPLICATION_OCTET_STREAM; import javax.ws.rs.core.Response; import static javax.ws.rs.core.Response.created; @@ -14,7 +13,6 @@ @Path("/store") -@Produces(MediaType.APPLICATION_JSON) public class StoreResource { private RaftStore raftStore; diff --git a/barge-store/src/test/java/org/robotninjas/barge/store/RaftStoreInstanceTest.java b/barge-store/src/test/java/org/robotninjas/barge/store/RaftStoreInstanceTest.java index 493da35..2300a2b 100644 --- a/barge-store/src/test/java/org/robotninjas/barge/store/RaftStoreInstanceTest.java +++ b/barge-store/src/test/java/org/robotninjas/barge/store/RaftStoreInstanceTest.java @@ -28,7 +28,7 @@ public class RaftStoreInstanceTest { @Test public void commitSerializedWriteOperationToRaftThenReturnOldValueWhenWritingKeyValuePair() throws Exception { - when(raft.commitOperation(serializer.serialize(write))).thenReturn(Futures.immediateFuture(oldValue)); + when(raft.commitOperation(serializer.serialize(write))).thenReturn(Futures.immediateFuture(Futures.immediateFuture(oldValue))); assertThat(raftStoreInstance.write(write)).isEqualTo(oldValue); } diff --git a/barge-store/src/test/java/org/robotninjas/barge/store/StoreServerTest.java b/barge-store/src/test/java/org/robotninjas/barge/store/StoreServerTest.java index 0da1172..1a05ae0 100644 --- a/barge-store/src/test/java/org/robotninjas/barge/store/StoreServerTest.java +++ b/barge-store/src/test/java/org/robotninjas/barge/store/StoreServerTest.java @@ -10,6 +10,8 @@ import org.robotninjas.barge.jaxrs.Leaders; +import java.io.File; + import java.net.URI; import java.net.URISyntaxException; @@ -25,11 +27,12 @@ public class StoreServerTest { private RaftStoreServer raftStoreServer; private URI[] clusterURIs; + private File logDir = new File(new File(StoreServerTest.class.getResource("/marker").getFile()).getParentFile(), "log" + System.nanoTime()); @Before public void startServer() throws URISyntaxException { clusterURIs = new URI[] { serverURI() }; - raftStoreServer = new RaftStoreServer(0, clusterURIs); + raftStoreServer = new RaftStoreServer(0, clusterURIs, logDir); raftStoreServer.start(56789); } diff --git a/barge-store/src/test/resources/marker b/barge-store/src/test/resources/marker new file mode 100644 index 0000000..ba69cfc --- /dev/null +++ b/barge-store/src/test/resources/marker @@ -0,0 +1 @@ +# marker file \ No newline at end of file From 527cc5c78d47e3e289d55792b0580e9984fee4fc Mon Sep 17 00:00:00 2001 From: Arnaud Bailly Date: Thu, 10 Jul 2014 18:21:12 +0200 Subject: [PATCH 07/10] added skeletal jepsen test project --- jepsen-tests/.lein-failures | 1 + jepsen-tests/.lein-repl-history | 0 .../jepsen/virtualbox/action_provision | 1 + .../jepsen/virtualbox/action_set_name | 1 + .../.vagrant/machines/jepsen/virtualbox/id | 1 + .../machines/jepsen/virtualbox/index_uuid | 1 + .../machines/jepsen/virtualbox/synced_folders | 1 + jepsen-tests/Vagrantfile | 26 +++++ jepsen-tests/project.clj | 6 ++ jepsen-tests/setup.sh | 35 +++++++ jepsen-tests/src/barge/jepsen/system.clj | 87 ++++++++++++++++ jepsen-tests/test/barge/jepsen/test.clj | 58 +++++++++++ jepsen-tests/todo.org | 98 +++++++++++++++++++ 13 files changed, 316 insertions(+) create mode 100644 jepsen-tests/.lein-failures create mode 100644 jepsen-tests/.lein-repl-history create mode 100644 jepsen-tests/.vagrant/machines/jepsen/virtualbox/action_provision create mode 100644 jepsen-tests/.vagrant/machines/jepsen/virtualbox/action_set_name create mode 100644 jepsen-tests/.vagrant/machines/jepsen/virtualbox/id create mode 100644 jepsen-tests/.vagrant/machines/jepsen/virtualbox/index_uuid create mode 100644 jepsen-tests/.vagrant/machines/jepsen/virtualbox/synced_folders create mode 100644 jepsen-tests/Vagrantfile create mode 100644 jepsen-tests/project.clj create mode 100644 jepsen-tests/setup.sh create mode 100644 jepsen-tests/src/barge/jepsen/system.clj create mode 100644 jepsen-tests/test/barge/jepsen/test.clj create mode 100644 jepsen-tests/todo.org diff --git a/jepsen-tests/.lein-failures b/jepsen-tests/.lein-failures new file mode 100644 index 0000000..ba3884e --- /dev/null +++ b/jepsen-tests/.lein-failures @@ -0,0 +1 @@ +#{barge.jepsen.test} \ No newline at end of file diff --git a/jepsen-tests/.lein-repl-history b/jepsen-tests/.lein-repl-history new file mode 100644 index 0000000..e69de29 diff --git a/jepsen-tests/.vagrant/machines/jepsen/virtualbox/action_provision b/jepsen-tests/.vagrant/machines/jepsen/virtualbox/action_provision new file mode 100644 index 0000000..e421f69 --- /dev/null +++ b/jepsen-tests/.vagrant/machines/jepsen/virtualbox/action_provision @@ -0,0 +1 @@ +1.5:2b928f8c-6664-4c21-bb2c-20be8d47d9e8 \ No newline at end of file diff --git a/jepsen-tests/.vagrant/machines/jepsen/virtualbox/action_set_name b/jepsen-tests/.vagrant/machines/jepsen/virtualbox/action_set_name new file mode 100644 index 0000000..b436678 --- /dev/null +++ b/jepsen-tests/.vagrant/machines/jepsen/virtualbox/action_set_name @@ -0,0 +1 @@ +1405001111 \ No newline at end of file diff --git a/jepsen-tests/.vagrant/machines/jepsen/virtualbox/id b/jepsen-tests/.vagrant/machines/jepsen/virtualbox/id new file mode 100644 index 0000000..036ec97 --- /dev/null +++ b/jepsen-tests/.vagrant/machines/jepsen/virtualbox/id @@ -0,0 +1 @@ +2b928f8c-6664-4c21-bb2c-20be8d47d9e8 \ No newline at end of file diff --git a/jepsen-tests/.vagrant/machines/jepsen/virtualbox/index_uuid b/jepsen-tests/.vagrant/machines/jepsen/virtualbox/index_uuid new file mode 100644 index 0000000..4b0b883 --- /dev/null +++ b/jepsen-tests/.vagrant/machines/jepsen/virtualbox/index_uuid @@ -0,0 +1 @@ +7984abde430c4615827642116c41561e \ No newline at end of file diff --git a/jepsen-tests/.vagrant/machines/jepsen/virtualbox/synced_folders b/jepsen-tests/.vagrant/machines/jepsen/virtualbox/synced_folders new file mode 100644 index 0000000..ed5af14 --- /dev/null +++ b/jepsen-tests/.vagrant/machines/jepsen/virtualbox/synced_folders @@ -0,0 +1 @@ +{"virtualbox":{"/vagrant":{"guestpath":"/vagrant","hostpath":"/Users/arnaud/projects/raft/barge/jepsen-tests"}}} \ No newline at end of file diff --git a/jepsen-tests/Vagrantfile b/jepsen-tests/Vagrantfile new file mode 100644 index 0000000..1b2598c --- /dev/null +++ b/jepsen-tests/Vagrantfile @@ -0,0 +1,26 @@ +# -*- mode: ruby -*- +# vi: set ft=ruby : +# a debian testing box for testing barge's jepsen tests (lots of tests here...) + +VAGRANTFILE_API_VERSION = "2" + +Vagrant.configure(VAGRANTFILE_API_VERSION) do |config| + + config.vm.box = "jessie" + config.vm.box_url = "https://dl.dropboxusercontent.com/u/2060057/package.box" + + proxy = ENV['http_proxy'] + + # machine for running jepsen tests + config.vm.define :jepsen do |mxconsole| + mxconsole.vm.hostname = "barge-jepsen" + + # VM has own network so that we can identify it + config.vm.network "private_network", ip: "192.168.43.2" + + mxconsole.vm.provision :shell, :path => "setup.sh", :args => proxy + end + +end + + diff --git a/jepsen-tests/project.clj b/jepsen-tests/project.clj new file mode 100644 index 0000000..75ac8bd --- /dev/null +++ b/jepsen-tests/project.clj @@ -0,0 +1,6 @@ +(defproject barge-jepsen "0.1.0-alpha2-SNAPSHOT" + :description "Jepsen (Call me Maybe) Tests for Barge's linearizability" + :dependencies [[org.clojure/clojure "1.6.0-beta1"] + [org.clojure/tools.logging "0.2.6"] + [jepsen/jepsen "0.0.3-SNAPSHOT"] + [org.robotninjas.barge/barge-store "0.1.0-alpha2-SNAPSHOT"]]) diff --git a/jepsen-tests/setup.sh b/jepsen-tests/setup.sh new file mode 100644 index 0000000..ee62e98 --- /dev/null +++ b/jepsen-tests/setup.sh @@ -0,0 +1,35 @@ +#!/bin/bash +# configure lxc and boxes for running tests +# need to be run as sudo/root + +PROXY=$1 + +if [ z$PROXY != "z" ]; then + export http_proxy=$PROXY + export https_proxy=$PROXY +fi + +# basic packages +apt-get update +apt-get install -y lxc bridge-utils libvirt-bin debootstrap clusterssh dnsmasq git + +# install java +# from http://www.webupd8.org/2014/03/how-to-install-oracle-java-8-in-debian.html +echo "deb http://ppa.launchpad.net/webupd8team/java/ubuntu trusty main" | tee /etc/apt/sources.list.d/webupd8team-java.list +echo "deb-src http://ppa.launchpad.net/webupd8team/java/ubuntu trusty main" | tee -a /etc/apt/sources.list.d/webupd8team-java.list +apt-key adv --keyserver keyserver.ubuntu.com --recv-keys EEA14886 +apt-get update + +echo oracle-java8-installer shared/accepted-oracle-license-v1-1 select true | /usr/bin/debconf-set-selections +apt-get install -y oracle-java8-installer + +function install_lein() { + lein=~vagrant/bin/lein + mkdir -p $(dirname $lein) + wget -O $lein https://raw.githubusercontent.com/technomancy/leiningen/stable/bin/lein + chmod +x $lein + echo PATH=$(dirname $lein):$PATH >> ~/.bash_profile +} + +install_lein + diff --git a/jepsen-tests/src/barge/jepsen/system.clj b/jepsen-tests/src/barge/jepsen/system.clj new file mode 100644 index 0000000..bedf347 --- /dev/null +++ b/jepsen-tests/src/barge/jepsen/system.clj @@ -0,0 +1,87 @@ +(ns barge.jepsen.system + (:require [clojure.java.io :as io] + [clojure.string :as str] + [clojure.tools.logging :refer [info]] + [jepsen.client :as client] + [jepsen.control :as c] + [jepsen.control.net :as net] + [jepsen.db :as db] + [jepsen.os.debian :as debian] + [jepsen.util :refer [meh timeout]])) + + +;; GAV coordinates and repository where to find barge +(def ^:dynamic *barge-version* "0.1.0-alpha2") +(def ^:dynamic *barge-groupid* "org.robotninjas.barge") +(def ^:dynamic *barge-artifactid* "barge-store") + +(def ^:dynamic *barge-repository* "https://dl.dropboxusercontent.com/u/2060057/maven/") + +(defn barge-jar [] + (str *barge-artifactid* "-" *barge-version* ".jar")) + +(defn barge-uri [] + (str *barge-repository* + (str/replace *barge-groupid* "." "/") "/" + *barge-artifactid* "/" + *barge-version* "/" + (barge-jar))) + +(defn barge-cluster-config + "Output configuration of barge cluster in a format suitable for populating a barge.conf + file, eg: + +``` +n1=http://1.2.3.4:8081 +n2=http://1.2.3.5:8081 +n3=http://1.2.3.6:8081 +``` + + * nodes: A sequence of nodes namespace + * fn-node-uri: Function to retrieve the node's IP given its name + " + [nodes fn-node-ip] + (str/join "\n" + (map (fn [node] + (str node "=http://" (fn-node-ip node) ":56789/")) + nodes + ))) + +(barge-cluster-config ["n1" "n2" "n3"] (fn [_] "http://1.2.3.5:8081") ) + +(def db + (reify db/DB + (setup! [_ test node] + (c/su + (info node "installing barge-store") + (c/cd "/tmp" + (info node "installing java") + (c/exec :apt-get :install :-y :openjdk-7-jre-headless) + + (info node "installing barge " *barge-version*) + (c/exec :wget :-c (barge-uri)) + + (info node "configuring barge") + (c/exec :echo + (barge-cluster-config (:nodes test) #(c/on % (net/local-ip))) + :> "barge.conf") + + (info node "starting barge") + (c/exec :java :-jar (symbol (barge-jar)) node) + + ))) + + (teardown! [_ test node] + (c/su + (info node "removing barge-store") + (c/exec :java :-cp (barge-jar) :stop node) + + (c/exec :rm :-fr (str "log" node)) + )))) + + +(c/with-ssh + {:username "vagrant" + :password "vagrant"} + (c/on "192.168.43.2" + (db/setup! db {:nodes ["192.168.43.2"]} "192.168.43.2"))) diff --git a/jepsen-tests/test/barge/jepsen/test.clj b/jepsen-tests/test/barge/jepsen/test.clj new file mode 100644 index 0000000..92f78e3 --- /dev/null +++ b/jepsen-tests/test/barge/jepsen/test.clj @@ -0,0 +1,58 @@ +(ns barge.jepsen.test + (:use barge.jepsen.system + jepsen.core + jepsen.tests + clojure.test + clojure.pprint) + (:require [clojure.string :as str] + [jepsen.util :as util] + [jepsen.os.debian :as debian] + [jepsen.checker :as checker] + [jepsen.checker.timeline :as timeline] + [jepsen.model :as model] + [jepsen.generator :as gen] + [jepsen.nemesis :as nemesis] + [jepsen.store :as store] + [jepsen.report :as report])) + +(def the-db + (atom nil)) + +(deftest register-test + (let [test (run! + (assoc + noop-test + :name "barge" + :os debian/os + :db (atom-db the-db) + :client (atom-client the-db) + :model (model/set) + :checker (checker/compose {:html timeline/html + :set checker/set}) + :nemesis (nemesis/partitioner nemesis/bridge) + :generator (gen/phases + (->> (range) + (map (fn [x] {:type :invoke + :f :add + :value x})) + gen/seq + (gen/stagger 1/10) + (gen/delay 1) + (gen/nemesis + (gen/seq + (cycle [(gen/sleep 60) + {:type :info :f :start} + (gen/sleep 300) + {:type :info :f :stop}]))) + (gen/time-limit 600)) + (gen/nemesis + (gen/once {:type :info :f :stop})) + (gen/clients + (gen/once {:type :invoke :f :read})))))] + (is (:valid? (:results test))) + (report/linearizability (:linear (:results test))) + (pprint (:results test)))) + + +(gen/ops test [1 2 3 4 5] range) + diff --git a/jepsen-tests/todo.org b/jepsen-tests/todo.org new file mode 100644 index 0000000..59b87dc --- /dev/null +++ b/jepsen-tests/todo.org @@ -0,0 +1,98 @@ +#+TODO: OPTION(o) BREAKDOWN(b) TODO(t) STARTED(s) WAITING(w) | DONE(d) CANCELED(c) +#+TAGS: developing(d) meeting(m) operations (o) planning (p) design(e) + +* TODO run jepsen tests on current barge +** Jepsen Overview + - [[file:~/projects/raft/jepsen/src/jepsen/core.clj][core.clj]] contains the main *run* function which is the entry point for each single test run. It takes care of setting up OS, + DB, launching clients in parallel, runnning generator and perform model checking at end of test + - [[file:~/projects/raft/jepsen/src/jepsen/model.clj][model.clj]] describes knossos model implementations for common case: Sets, CASRegisters. Each model implements single + function *step* that applies some operation on a "model", possibly detecting and generating inconsistencies. For example, + applying a *cas* operation to a CASRegister is consistent iff the old and new values are identical. Simplest model is Mutex + which may acquired/released putting it in a locked/unlocked state. All Models are immutable thus allowing easy generation of + possible worlds from history + - see [[file:~/projects/raft/jepsen/src/jepsen/client.clj][client.clj]] for the protocol to implement per client. Clients are single threaded and send operations to server nodes + - see [[file:~/projects/raft/jepsen/src/jepsen/db.clj][db.clj]] for protocol to implement to setup/teardown each "DB" instance, in our case each barge instance protecting access + to underlying data. + - We need to provide some simple DB implemented over raft, eg. basic K/V store + - [[file:~/projects/raft/jepsen/src/jepsen/generator.clj][generator.clj]] is used to construct the test rig, eg. the generator for events to affect the database, including [[file:~/projects/raft/jepsen/src/jepsen/nemesis.clj][nemesis.clj]] + which encapsulates the logic to create network partitions or outages (drop packets, slow down link...). Implementation of + generators is actually a monad, very reminiscent of what's done in Quickcheck: One can compose generators, introducing + nemesis, delays, mapping functions... + - test results and outputs are persisted to a [[file:~/projects/raft/jepsen/src/jepsen/store.clj][store.clj]], basically a local file containing serializable objects from the test + - [[file:~/projects/raft/jepsen/src/jepsen/checker.clj][checker.clj]] defines the properties we may want to check for a given test, like linearizability which is the main purpose of + knossos library. Some properties we want to test: queue, total queue, set... Checks that the property is verified for a given + history and a given model: The history is what's produced by the (concurrent and generated test) and the model is how we + assume the underlying DB should behave. Properties may be more or less complex to check and for example checking + linearizability is somewhere in the vicinity of factorial in the length of the history... + - [[file:~/projects/raft/jepsen/src/jepsen/os.clj][os.clj]] and [[file:~/projects/raft/jepsen/src/jepsen/os/debian.clj][debian.clj]] provides interface and implementation for setting up OSes to run tests. Test runner uses low-level + commands from [[file:~/projects/raft/jepsen/src/jepsen/control.clj][control.clj]] which provides among other things an evolved *exec* function to run arbitrary processes, a *ssh* + function to run stuff over ssh connection and nice wrappers and macros like su or cd. + - provides system implementations and corresponding tests for the following DBs/Models + - elasticsearch/*set* + - datomic/CAS register + - etcd/CAS client + - rabbitmq/mutex +** Knossos + - provides the conceptual framework to model-check a history against a model, testing for linearizability. + - *linearizability* is a specific level of consistency a concurrent system can exhibit (and guarantee to client). In a + linearizable system all operations which are assumed to take some time, appear to take effect instantly at some point between + their initiation and termination and in an order consistent with the *program order*. For example, given location x having + initial value 0, the concurrent operations of thread A doing a->read/a<-read(0), and thread B doing a->write(1)/a<-write() are + linearizable, as well as a->read()/a<-read(1). + - Of course more insteresting things happen when several concurrent threads write to the same location + - So given a history and a model we want to check if there is a linearization that is *consistent* with a model which + theoretically entails generating all possible interleaving of concurrent operations consistent with program order, and + checking there is at least one such consistent interleaving + - knossos leverages the laziness of clojure +** DONE build/run jepsen +*** DONE update dependencies to latest released versions + - https://github.com/aphyr/knossos -> 0.2 + - clojurewerkz/elastisch -> 2.0.0-rc1 + - test do not compile due to dependency on datomic stuff +** CANCELED setup VM with coreos + - coreOS might not be a good choice in the short term as jepsen implements only debian OS. Need to implement docker-based OS?? +*** DONE upgrade coreos-vagrant +https://github.com/coreos/coreos-vagrant +*** DONE upgrade vagrant -> 1.6.2 + - don't forget to set http_proxy env variables... +*** DONE setup a debian testing (jessie) Box + - installed http://saimei.acc.umu.se/cdimage/jessie_di_alpha_1/amd64/iso-cd/debian-jessie-DI-a1-amd64-netinst.iso in VBox + - used as guideline for configuring box http://blog.codeboutique.com/post/creating-debian-squeeze-box-for-vagrant + - created package.box -> ~700MB +*** DONE configure lxc and test boxes + - based on lxc.md from jepsen sources + - write it as a script (possibly a .clj using what's provided by jepsen...) + - configured all 5 boxes, started them -> works fine +*** DONE install lein and build project + - project can build with lein, still issues with runnning test +*** DONE install iptables on lxc containers +*** DONE understand how to customize test executions for ssh authent on lxc containers +*** DONE fix elastic search test issue +ERROR in (create-test) (rest.clj:64) +Uncaught exception, not in assertion. +expected: nil + actual: java.lang.ClassCastException: java.lang.String cannot be cast to clojurewerkz.elastisch.rest.Connection + at clojurewerkz.elastisch.rest$url_with_path.doInvoke (rest.clj:64) + clojure.lang.RestFn.invoke (RestFn.java:423) + clojurewerkz.elastisch.rest$index_url.invoke (rest.clj:68) + clojurewerkz.elastisch.rest.index$create.doInvoke (index.clj:48) + clojure.lang.RestFn.invoke (RestFn.java:490) + jepsen.system.elasticsearch.CreateSetClient/fn (elasticsearch.clj:118) + jepsen.system.elasticsearch.CreateSetClient.setup_BANG_ (elasticsearch.clj:116) + jepsen.core$run_case_BANG_$fn__5082.invoke (core.clj:232) + clojure.lang.AFn.applyToHelper (AFn.java:154) + clojure.lang.AFn.applyTo (AFn.java:144) + clojure.core$apply.invoke (core.clj:624) + jepsen.core$fcatch$wrapper__5003.doInvoke (core.clj:39) + clojure.lang.RestFn.invoke (RestFn.java:408) + clojure.core$pmap$fn__6328$fn__6329.invoke (core.clj:6463) + clojure.core$binding_conveyor_fn$fn__4145.invoke (core.clj:1910) + clojure.lang.AFn.call (AFn.java:18) + java.util.concurrent.FutureTask.run (FutureTask.java:266) + java.util.concurrent.ThreadPoolExecutor.runWorker (ThreadPoolExecutor.java:1142) + java.util.concurrent.ThreadPoolExecutor$Worker.run (ThreadPoolExecutor.java:617) + java.lang.Thread.run (Thread.java:745) + +** DONE setup 5 docker containers with debian testing +** TODO write test adapter for barge + From 55161f30368e091f31217f4f034ecaeb6cb6d153 Mon Sep 17 00:00:00 2001 From: Arnaud Bailly Date: Thu, 10 Jul 2014 19:21:00 +0200 Subject: [PATCH 08/10] fork server in the background when starting, redirecting outputs to files only way to stop server now is sending kill -TERM! Need to implement proper stop... --- barge-store/pom.xml | 2 +- .../barge/store/RaftStoreServer.java | 17 -------- barge-store/src/main/java/start.java | 39 +++++++++++++++++++ 3 files changed, 40 insertions(+), 18 deletions(-) create mode 100644 barge-store/src/main/java/start.java diff --git a/barge-store/pom.xml b/barge-store/pom.xml index 922cdac..f03c8bf 100644 --- a/barge-store/pom.xml +++ b/barge-store/pom.xml @@ -85,7 +85,7 @@ - org.robotninjas.barge.store.RaftStoreServer + start diff --git a/barge-store/src/main/java/org/robotninjas/barge/store/RaftStoreServer.java b/barge-store/src/main/java/org/robotninjas/barge/store/RaftStoreServer.java index 4106c3e..b19b3d1 100644 --- a/barge-store/src/main/java/org/robotninjas/barge/store/RaftStoreServer.java +++ b/barge-store/src/main/java/org/robotninjas/barge/store/RaftStoreServer.java @@ -61,10 +61,6 @@ public class RaftStoreServer { private RaftJettyServer server; - public RaftStoreServer(int serverIndex, URI[] clusterURIs) { - this(serverIndex, clusterURIs, new File("log" + serverIndex)); - } - public RaftStoreServer(int serverIndex, URI[] clusterURIs, File logDir) { this.serverIndex = serverIndex; this.clusterURIs = clusterURIs; @@ -116,19 +112,6 @@ public static void main(String[] args) throws IOException, URISyntaxException { RaftStoreServer server = new RaftStoreServer(index, uris, logDir); server.start(uris[index].getPort()); - - waitForInput(); - - server.stop(); - - System.out.println("Bye!"); - System.exit(0); - } - - private static void waitForInput() throws IOException { - - //noinspection ResultOfMethodCallIgnored - System.in.read(); } private static void muteJul() { diff --git a/barge-store/src/main/java/start.java b/barge-store/src/main/java/start.java new file mode 100644 index 0000000..1ef419d --- /dev/null +++ b/barge-store/src/main/java/start.java @@ -0,0 +1,39 @@ +import com.google.common.collect.Lists; + +import org.robotninjas.barge.store.RaftStoreServer; + +import java.io.File; +import java.io.IOException; + +import java.util.ArrayList; +import java.util.Collections; + + +/** + * Wrapper over {@link org.robotninjas.barge.store.RaftStoreServer} that forks a new instance in the background. + *

+ * This main class is used to run barge in the background without relying on native mechanism: It forks actual main to another + * process and redirects stdout and stderr to some files. + *

+ */ +public class start { + + private static final String JAVA = System.getProperty("java.home", "java"); + private static final String CLASSPATH = System.getProperty("java.class.path", "barge-store.jar"); + + public static void main(String[] args) throws IOException { + ArrayList arguments = Lists.newArrayList(JAVA + "/bin/java", "-cp", CLASSPATH, RaftStoreServer.class.getName()); + + Collections.addAll(arguments, args); + + long timestamp = System.currentTimeMillis(); + File stdout = new File(".barge." + timestamp + ".out"); + File stderr = new File(".barge." + timestamp + ".err"); + + ProcessBuilder builder = new ProcessBuilder(arguments).redirectError(stderr) + .redirectOutput(stdout) + .redirectInput(ProcessBuilder.Redirect.PIPE); + + builder.start(); + } +} From 0076a9dd1574124717ca5c313aa7dfe9329b4624 Mon Sep 17 00:00:00 2001 From: Arnaud Bailly Date: Thu, 10 Jul 2014 19:38:55 +0200 Subject: [PATCH 09/10] can stop store server instance using REST interface /control/stop --- .../org/robotninjas/barge/store/Control.java | 15 ++++++ .../barge/store/ControlResource.java | 31 ++++++++++++ .../barge/store/RaftStoreServer.java | 4 +- .../barge/store/ControlResourceTest.java | 48 +++++++++++++++++++ 4 files changed, 96 insertions(+), 2 deletions(-) create mode 100644 barge-store/src/main/java/org/robotninjas/barge/store/Control.java create mode 100644 barge-store/src/main/java/org/robotninjas/barge/store/ControlResource.java create mode 100644 barge-store/src/test/java/org/robotninjas/barge/store/ControlResourceTest.java diff --git a/barge-store/src/main/java/org/robotninjas/barge/store/Control.java b/barge-store/src/main/java/org/robotninjas/barge/store/Control.java new file mode 100644 index 0000000..5536a05 --- /dev/null +++ b/barge-store/src/main/java/org/robotninjas/barge/store/Control.java @@ -0,0 +1,15 @@ +package org.robotninjas.barge.store; + +/** + * Thin abstraction over {@link java.lang.System}. + */ +public class Control { + + /** + * Unconditionally shutdown the current JVM with exit code 0. + */ + public void shutdown() { + System.exit(0); + } + +} diff --git a/barge-store/src/main/java/org/robotninjas/barge/store/ControlResource.java b/barge-store/src/main/java/org/robotninjas/barge/store/ControlResource.java new file mode 100644 index 0000000..4ffac0e --- /dev/null +++ b/barge-store/src/main/java/org/robotninjas/barge/store/ControlResource.java @@ -0,0 +1,31 @@ +package org.robotninjas.barge.store; + +import javax.inject.Inject; + +import javax.ws.rs.Consumes; +import javax.ws.rs.POST; +import javax.ws.rs.Path; +import javax.ws.rs.core.MediaType; + + +/** + * Provides REST interface for controlling the store. + */ +@Path("/control") +public class ControlResource { + + private final Control control; + + @Inject + public ControlResource(Control control) { + this.control = control; + } + + + @POST + @Path("/stop") + @Consumes(MediaType.TEXT_PLAIN) + public void stop() { + control.shutdown(); + } +} diff --git a/barge-store/src/main/java/org/robotninjas/barge/store/RaftStoreServer.java b/barge-store/src/main/java/org/robotninjas/barge/store/RaftStoreServer.java index b19b3d1..9244c7a 100644 --- a/barge-store/src/main/java/org/robotninjas/barge/store/RaftStoreServer.java +++ b/barge-store/src/main/java/org/robotninjas/barge/store/RaftStoreServer.java @@ -120,12 +120,12 @@ private static void muteJul() { SLF4JBridgeHandler.install(); } - private static URI[] readConfiguration(File clusterConfiguration) throws IOException, URISyntaxException { + private static URI[] readConfiguration(File clusterConfigurationFile) throws IOException, URISyntaxException { List uris = Lists.newArrayList(); int lineNumber = 1; - for (String line : CharStreams.readLines(new FileReader(clusterConfiguration))) { + for (String line : CharStreams.readLines(new FileReader(clusterConfigurationFile))) { String[] pair = line.split("="); if (pair.length != 2) diff --git a/barge-store/src/test/java/org/robotninjas/barge/store/ControlResourceTest.java b/barge-store/src/test/java/org/robotninjas/barge/store/ControlResourceTest.java new file mode 100644 index 0000000..be7c789 --- /dev/null +++ b/barge-store/src/test/java/org/robotninjas/barge/store/ControlResourceTest.java @@ -0,0 +1,48 @@ +package org.robotninjas.barge.store; + +import com.google.common.collect.Sets; + +import org.glassfish.jersey.server.ResourceConfig; +import org.glassfish.jersey.test.JerseyTest; + +import org.junit.Test; + +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; + +import java.util.Set; + +import javax.ws.rs.client.Entity; +import javax.ws.rs.core.Application; +import javax.ws.rs.core.MediaType; + + +public class ControlResourceTest extends JerseyTest { + + private Control control; + + @Override + protected Application configure() { + control = mock(Control.class); + + ResourceConfig resourceConfig = ResourceConfig.forApplication(new Application() { + @Override + public Set getSingletons() { + return Sets.newHashSet((Object) new ControlResource(control)); + } + }); + + resourceConfig.register(OperationsSerializer.jacksonModule()); + + return resourceConfig; + } + + + @Test + public void onPOSTStopWithEmptyContentRequestsShutdownOfVM() throws Exception { + client().target("/control/stop").request().post(Entity.entity("", MediaType.TEXT_PLAIN)); + + verify(control).shutdown(); + } + +} From 02b5a59049212e3e15c2c660d22fddf7398924ec Mon Sep 17 00:00:00 2001 From: Arnaud Bailly Date: Fri, 11 Jul 2014 08:30:19 +0200 Subject: [PATCH 10/10] extract configuration reader to own object --- .../barge/store/ClusterConfiguration.java | 37 +++++++++++++++++++ .../barge/store/RaftStoreServer.java | 26 +------------ barge-store/src/main/java/stop.java | 13 +++++++ 3 files changed, 52 insertions(+), 24 deletions(-) create mode 100644 barge-store/src/main/java/org/robotninjas/barge/store/ClusterConfiguration.java create mode 100644 barge-store/src/main/java/stop.java diff --git a/barge-store/src/main/java/org/robotninjas/barge/store/ClusterConfiguration.java b/barge-store/src/main/java/org/robotninjas/barge/store/ClusterConfiguration.java new file mode 100644 index 0000000..e352efd --- /dev/null +++ b/barge-store/src/main/java/org/robotninjas/barge/store/ClusterConfiguration.java @@ -0,0 +1,37 @@ +package org.robotninjas.barge.store; + +import com.google.common.collect.Lists; +import com.google.common.io.CharStreams; + +import java.io.File; +import java.io.FileReader; +import java.io.IOException; + +import java.net.URI; +import java.net.URISyntaxException; + +import java.util.List; + + +/** + * Utilities for configuring cluster using some external source of configuration (eg. a file). + */ +public class ClusterConfiguration { + + public URI[] readConfiguration(File clusterConfigurationFile) throws IOException, URISyntaxException { + List uris = Lists.newArrayList(); + + int lineNumber = 1; + + for (String line : CharStreams.readLines(new FileReader(clusterConfigurationFile))) { + String[] pair = line.split("="); + + if (pair.length != 2) + throw new IOException("Invalid cluster configuration at line " + lineNumber); + + uris.add(Integer.parseInt(pair[0].trim()), new URI(pair[1].trim())); + } + + return uris.toArray(new URI[uris.size()]); + } +} diff --git a/barge-store/src/main/java/org/robotninjas/barge/store/RaftStoreServer.java b/barge-store/src/main/java/org/robotninjas/barge/store/RaftStoreServer.java index 9244c7a..fe273fa 100644 --- a/barge-store/src/main/java/org/robotninjas/barge/store/RaftStoreServer.java +++ b/barge-store/src/main/java/org/robotninjas/barge/store/RaftStoreServer.java @@ -1,8 +1,5 @@ package org.robotninjas.barge.store; -import com.google.common.collect.Lists; -import com.google.common.io.CharStreams; - import org.glassfish.hk2.utilities.Binder; import org.glassfish.hk2.utilities.binding.AbstractBinder; @@ -13,13 +10,11 @@ import org.slf4j.bridge.SLF4JBridgeHandler; import java.io.File; -import java.io.FileReader; import java.io.IOException; import java.net.URI; import java.net.URISyntaxException; -import java.util.List; import java.util.logging.Level; import javax.inject.Singleton; @@ -54,11 +49,11 @@ public class RaftStoreServer { " Default is './barge.conf'\n" + " : Index of this server in the cluster configuration\n"; + private static final ClusterConfiguration cluster = new ClusterConfiguration(); private final int serverIndex; private final URI[] clusterURIs; private final File logDir; - private RaftJettyServer server; public RaftStoreServer(int serverIndex, URI[] clusterURIs, File logDir) { @@ -105,7 +100,7 @@ public static void main(String[] args) throws IOException, URISyntaxException { System.exit(1); } - URI[] uris = readConfiguration(clusterConfiguration); + URI[] uris = cluster.readConfiguration(clusterConfiguration); final File logDir = new File("log" + index); @@ -120,23 +115,6 @@ private static void muteJul() { SLF4JBridgeHandler.install(); } - private static URI[] readConfiguration(File clusterConfigurationFile) throws IOException, URISyntaxException { - List uris = Lists.newArrayList(); - - int lineNumber = 1; - - for (String line : CharStreams.readLines(new FileReader(clusterConfigurationFile))) { - String[] pair = line.split("="); - - if (pair.length != 2) - throw new IOException("Invalid cluster configuration at line " + lineNumber); - - uris.add(Integer.parseInt(pair[0].trim()), new URI(pair[1].trim())); - } - - return uris.toArray(new URI[uris.size()]); - } - private static void usage() { System.out.println(help); } diff --git a/barge-store/src/main/java/stop.java b/barge-store/src/main/java/stop.java new file mode 100644 index 0000000..b3272ca --- /dev/null +++ b/barge-store/src/main/java/stop.java @@ -0,0 +1,13 @@ +/** + * Main class to stop a running barge instance from the command-line. + *

+ * Needs a barge.conf file. If given one or more instance names, it will shutdown only those instances, otherwise + * it shuts down all instances fron the configuration file. + *

+ */ +public class stop { + + public static void main(String[] args) { + + } +}