Publish

4 files changed, 399 insertions, 0 deletions

diff --git a/vespa_feed_perf/src/test/java/com/yahoo/vespa/feed/perf/FeederParamsTest.java b/vespa_feed_perf/src/test/java/com/yahoo/vespa/feed/perf/FeederParamsTest.java
new file mode 100644
index 00000000000..dec31d9e1d9
--- /dev/null
+++ b/vespa_feed_perf/src/test/java/com/yahoo/vespa/feed/perf/FeederParamsTest.java
@@ -0,0 +1,88 @@
+// Copyright 2016 Yahoo Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+package com.yahoo.vespa.feed.perf;
+
+import com.yahoo.messagebus.routing.Route;
+import org.apache.commons.cli.ParseException;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+import java.io.InputStream;
+import java.io.PrintStream;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotSame;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * @author <a href="mailto:simon@yahoo-inc.com">Simon Thoresen Hult</a>
+ */
+public class FeederParamsTest {
+
+    @Test
+    public void requireThatAccessorsWork() {
+        FeederParams params = new FeederParams();
+
+        InputStream stdIn = Mockito.mock(InputStream.class);
+        params.setStdIn(stdIn);
+        assertSame(stdIn, params.getStdIn());
+
+        PrintStream stdErr = Mockito.mock(PrintStream.class);
+        params.setStdErr(stdErr);
+        assertSame(stdErr, params.getStdErr());
+
+        PrintStream stdOut = Mockito.mock(PrintStream.class);
+        params.setStdOut(stdOut);
+        assertSame(stdOut, params.getStdOut());
+
+        Route route = Route.parse("my_route");
+        params.setRoute(route);
+        assertEquals(route, params.getRoute());
+        assertNotSame(route, params.getRoute());
+
+        params.setConfigId("my_config_id");
+        assertEquals("my_config_id", params.getConfigId());
+
+        params.setSerialTransfer(false);
+        assertFalse(params.isSerialTransferEnabled());
+        params.setSerialTransfer(true);
+        assertTrue(params.isSerialTransferEnabled());
+    }
+
+    @Test
+    public void requireThatParamsHaveReasonableDefaults() {
+        FeederParams params = new FeederParams();
+        assertSame(System.in, params.getStdIn());
+        assertSame(System.err, params.getStdErr());
+        assertSame(System.out, params.getStdOut());
+        assertEquals(Route.parse("default"), params.getRoute());
+        assertEquals("client", params.getConfigId());
+        assertFalse(params.isSerialTransferEnabled());
+    }
+
+    @Test
+    public void requireThatSerialTransferOptionIsParsed() throws ParseException {
+        assertTrue(new FeederParams().parseArgs("-s").isSerialTransferEnabled());
+        assertTrue(new FeederParams().parseArgs("foo", "-s").isSerialTransferEnabled());
+        assertTrue(new FeederParams().parseArgs("-s", "foo").isSerialTransferEnabled());
+        assertTrue(new FeederParams().parseArgs("--serial").isSerialTransferEnabled());
+        assertTrue(new FeederParams().parseArgs("foo", "--serial").isSerialTransferEnabled());
+        assertTrue(new FeederParams().parseArgs("--serial", "foo").isSerialTransferEnabled());
+    }
+
+    @Test
+    public void requireThatArgumentsAreParsedAsRoute() throws ParseException {
+        assertEquals(Route.parse("foo bar"), new FeederParams().parseArgs("foo", "bar").getRoute());
+        assertEquals(Route.parse("foo bar"), new FeederParams().parseArgs("-s", "foo", "bar").getRoute());
+        assertEquals(Route.parse("foo bar"), new FeederParams().parseArgs("foo", "-s", "bar").getRoute());
+        assertEquals(Route.parse("foo bar"), new FeederParams().parseArgs("foo", "bar", "-s").getRoute());
+    }
+
+    @Test
+    public void requireThatRouteIsAnOptionalArgument() throws ParseException {
+        assertEquals(Route.parse("default"), new FeederParams().parseArgs().getRoute());
+        assertEquals(Route.parse("default"), new FeederParams().parseArgs("-s").getRoute());
+    }
+
+}
diff --git a/vespa_feed_perf/src/test/java/com/yahoo/vespa/feed/perf/SimpleFeederTest.java b/vespa_feed_perf/src/test/java/com/yahoo/vespa/feed/perf/SimpleFeederTest.java
new file mode 100644
index 00000000000..809277670ad
--- /dev/null
+++ b/vespa_feed_perf/src/test/java/com/yahoo/vespa/feed/perf/SimpleFeederTest.java
@@ -0,0 +1,198 @@
+// Copyright 2016 Yahoo Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+package com.yahoo.vespa.feed.perf;
+
+import com.yahoo.document.serialization.DeserializationException;
+import com.yahoo.documentapi.messagebus.protocol.DocumentMessage;
+import com.yahoo.jrt.ListenFailedException;
+import com.yahoo.messagebus.DynamicThrottlePolicy;
+import com.yahoo.messagebus.EmptyReply;
+import com.yahoo.messagebus.Error;
+import com.yahoo.messagebus.ErrorCode;
+import com.yahoo.messagebus.Message;
+import com.yahoo.messagebus.MessageHandler;
+import com.yahoo.messagebus.Reply;
+import com.yahoo.messagebus.SourceSession;
+import com.yahoo.messagebus.StaticThrottlePolicy;
+import com.yahoo.messagebus.ThrottlePolicy;
+import org.junit.Test;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.PrintStream;
+import java.lang.reflect.Field;
+import java.nio.charset.StandardCharsets;
+import java.util.regex.Pattern;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+/**
+ * @author <a href="mailto:simon@yahoo-inc.com">Simon Thoresen Hult</a>
+ */
+public class SimpleFeederTest {
+
+    private static final String CONFIG_DIR = "target/test-classes/";
+
+    @Test
+    public void requireThatFeederWorks() throws Throwable {
+        assertFeed("<vespafeed>" +
+                   "    <document documenttype='simple' documentid='doc:scheme:0'>" +
+                   "        <my_str>foo</my_str>" +
+                   "    </document>" +
+                   "    <update documenttype='simple' documentid='doc:scheme:1'>" +
+                   "        <assign field='my_str'>bar</assign>" +
+                   "    </update>" +
+                   "    <remove documenttype='simple' documentid='doc:scheme:2'/>" +
+                   "</vespafeed>",
+                   new MessageHandler() {
+
+                       @Override
+                       public void handleMessage(Message msg) {
+                           Reply reply = ((DocumentMessage)msg).createReply();
+                           reply.swapState(msg);
+                           reply.popHandler().handleReply(reply);
+                       }
+                   },
+                   "",
+                   "(.+\n)+" +
+                   "\\s*\\d+,\\s*3,.+\n");
+    }
+
+    @Test
+    public void requireThatParseFailuresThrowInMainThread() throws Throwable {
+        TestDriver driver = new TestDriver(new FeederParams(),
+                                           "<vespafeed>" +
+                                           "    <document documenttype='unknown' documentid='doc:scheme:0'/>" +
+                                           "</vespafeed>",
+                                           null);
+        try {
+            driver.run();
+            fail();
+        } catch (DeserializationException e) {
+            assertEquals("Field 'doc:scheme:0': Must specify an existing document type, not 'unknown' (at line 1, column 76)",
+                         e.getMessage());
+        }
+        assertTrue(driver.close());
+    }
+
+    @Test
+    public void requireThatSyncFailuresThrowInMainThread() throws Throwable {
+        TestDriver driver = new TestDriver(new FeederParams(),
+                                           "<vespafeed>" +
+                                           "    <document documenttype='simple' documentid='doc:scheme:0'/>" +
+                                           "</vespafeed>",
+                                           null);
+        getSourceSession(driver).close();
+        try {
+            driver.run();
+            fail();
+        } catch (IOException e) {
+            assertEquals("[SEND_QUEUE_CLOSED @ localhost]: Source session is closed.", e.getMessage());
+        }
+        assertTrue(driver.close());
+    }
+
+    @Test
+    public void requireThatAsyncFailuresThrowInMainThread() throws Throwable {
+        TestDriver driver = new TestDriver(new FeederParams(),
+                                           "<vespafeed><document documenttype='simple' documentid='doc:scheme:0'/></vespafeed>",
+                                           new MessageHandler() {
+
+                                               @Override
+                                               public void handleMessage(Message msg) {
+                                                   Reply reply = new EmptyReply();
+                                                   reply.swapState(msg);
+                                                   reply.addError(new Error(ErrorCode.APP_FATAL_ERROR + 6, "foo"));
+                                                   reply.addError(new Error(ErrorCode.APP_FATAL_ERROR + 9, "bar"));
+                                                   reply.popHandler().handleReply(reply);
+                                               }
+                                           });
+        try {
+            driver.run();
+            fail();
+        } catch (IOException e) {
+            assertMatches("com.yahoo.documentapi.messagebus.protocol.PutDocumentMessage@.+\n" +
+                          "\\[UNKNOWN\\(250006\\) @ .+\\]: foo\n" +
+                          "\\[UNKNOWN\\(250009\\) @ .+\\]: bar\n",
+                          e.getMessage());
+        }
+        assertTrue(driver.close());
+    }
+
+    @Test
+    public void requireThatDynamicThrottlingIsDefault() throws Exception {
+        TestDriver driver = new TestDriver(new FeederParams(), "", null);
+        assertEquals(DynamicThrottlePolicy.class, getThrottlePolicy(driver).getClass());
+        assertTrue(driver.close());
+    }
+
+    @Test
+    public void requireThatSerialTransferModeConfiguresStaticThrottling() throws Exception {
+        TestDriver driver = new TestDriver(new FeederParams().setSerialTransfer(true), "", null);
+        assertEquals(StaticThrottlePolicy.class, getThrottlePolicy(driver).getClass());
+        assertTrue(driver.close());
+    }
+
+    private static SourceSession getSourceSession(TestDriver driver) {
+        return (SourceSession)getField(driver.feeder, "session");
+    }
+
+    private static ThrottlePolicy getThrottlePolicy(TestDriver driver) {
+        return (ThrottlePolicy)getField(getSourceSession(driver), "throttlePolicy");
+    }
+
+    private static Object getField(Object obj, String fieldName) {
+        try {
+            Field field = obj.getClass().getDeclaredField(fieldName);
+            field.setAccessible(true);
+            return field.get(obj);
+        } catch (IllegalAccessException | NoSuchFieldException e) {
+            throw new AssertionError(e);
+        }
+    }
+
+    private static void assertFeed(String in, MessageHandler validator, String expectedErr, String expectedOut)
+            throws Throwable {
+        TestDriver driver = new TestDriver(new FeederParams(), in, validator);
+        driver.run();
+        assertMatches(expectedErr, new String(driver.err.toByteArray(), StandardCharsets.UTF_8));
+        assertMatches(expectedOut, new String(driver.out.toByteArray(), StandardCharsets.UTF_8));
+        assertTrue(driver.close());
+    }
+
+    private static void assertMatches(String expected, String actual) {
+        if (!Pattern.matches(expected, actual)) {
+            assertEquals(expected, actual);
+        }
+    }
+
+    private static class TestDriver {
+
+        final ByteArrayOutputStream err = new ByteArrayOutputStream();
+        final ByteArrayOutputStream out = new ByteArrayOutputStream();
+        final SimpleFeeder feeder;
+        final SimpleServer server;
+
+        public TestDriver(FeederParams params, String in, MessageHandler validator)
+                throws IOException, ListenFailedException {
+            server = new SimpleServer(CONFIG_DIR, validator);
+            feeder = new SimpleFeeder(params.setConfigId("dir:" + CONFIG_DIR)
+                                            .setStdErr(new PrintStream(err))
+                                            .setStdIn(new ByteArrayInputStream(in.getBytes(StandardCharsets.UTF_8)))
+                                            .setStdOut(new PrintStream(out)));
+        }
+
+        void run() throws Throwable {
+            feeder.run();
+        }
+
+        boolean close() {
+            feeder.close();
+            server.close();
+            return true;
+        }
+    }
+
+}
diff --git a/vespa_feed_perf/src/test/java/com/yahoo/vespa/feed/perf/SimpleServer.java b/vespa_feed_perf/src/test/java/com/yahoo/vespa/feed/perf/SimpleServer.java
new file mode 100644
index 00000000000..5df725865dd
--- /dev/null
+++ b/vespa_feed_perf/src/test/java/com/yahoo/vespa/feed/perf/SimpleServer.java
@@ -0,0 +1,63 @@
+// Copyright 2016 Yahoo Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+package com.yahoo.vespa.feed.perf;
+
+import com.yahoo.document.DocumentTypeManager;
+import com.yahoo.documentapi.messagebus.protocol.DocumentProtocol;
+import com.yahoo.jrt.ListenFailedException;
+import com.yahoo.jrt.slobrok.server.Slobrok;
+import com.yahoo.messagebus.DestinationSession;
+import com.yahoo.messagebus.DestinationSessionParams;
+import com.yahoo.messagebus.MessageBus;
+import com.yahoo.messagebus.MessageBusParams;
+import com.yahoo.messagebus.MessageHandler;
+import com.yahoo.messagebus.network.Identity;
+import com.yahoo.messagebus.network.rpc.RPCNetwork;
+import com.yahoo.messagebus.network.rpc.RPCNetworkParams;
+
+import java.io.FileWriter;
+import java.io.IOException;
+import java.io.PrintWriter;
+
+/**
+ * @author <a href="mailto:simon@yahoo-inc.com">Simon Thoresen Hult</a>
+ */
+public class SimpleServer {
+
+    private final DocumentTypeManager documentMgr;
+    private final Slobrok slobrok;
+    private final MessageBus mbus;
+    private final DestinationSession session;
+
+    public SimpleServer(String configDir, MessageHandler msgHandler) throws IOException, ListenFailedException {
+        slobrok = new Slobrok();
+        documentMgr = new DocumentTypeManager();
+        documentMgr.configure("dir:" + configDir);
+        mbus = new MessageBus(new RPCNetwork(new RPCNetworkParams()
+                                                     .setSlobrokConfigId(slobrok.configId())
+                                                     .setIdentity(new Identity("server"))),
+                              new MessageBusParams().addProtocol(new DocumentProtocol(documentMgr)));
+        session = mbus.createDestinationSession(new DestinationSessionParams().setMessageHandler(msgHandler));
+
+        PrintWriter writer = new PrintWriter(new FileWriter(configDir + "/messagebus.cfg"));
+        writer.println("routingtable[1]\n" +
+                       "routingtable[0].protocol \"document\"\n" +
+                       "routingtable[0].hop[0]\n" +
+                       "routingtable[0].route[1]\n" +
+                       "routingtable[0].route[0].name \"default\"\n" +
+                       "routingtable[0].route[0].hop[1]\n" +
+                       "routingtable[0].route[0].hop[0] \"" + session.getConnectionSpec() + "\"");
+        writer.close();
+
+        writer = new PrintWriter(new FileWriter(configDir + "/slobroks.cfg"));
+        writer.println(slobrok.configId().substring(4));
+        writer.close();
+    }
+
+    public final void close() {
+        session.destroy();
+        mbus.destroy();
+        documentMgr.shutdown();
+        slobrok.stop();
+    }
+
+}
diff --git a/vespa_feed_perf/src/test/resources/documentmanager.cfg b/vespa_feed_perf/src/test/resources/documentmanager.cfg
new file mode 100644
index 00000000000..ebb6e767bef
--- /dev/null
+++ b/vespa_feed_perf/src/test/resources/documentmanager.cfg
@@ -0,0 +1,50 @@
+enablecompression false
+annotationtype[0]
+datatype[3]
+datatype[0].id -1101456630
+datatype[0].annotationreftype[0]
+datatype[0].arraytype[0]
+datatype[0].documenttype[0]
+datatype[0].maptype[0]
+datatype[0].structtype[1]
+datatype[0].structtype[0].compresslevel 0
+datatype[0].structtype[0].compressminsize 800
+datatype[0].structtype[0].compressthreshold 95
+datatype[0].structtype[0].compresstype NONE
+datatype[0].structtype[0].name "simple.body"
+datatype[0].structtype[0].version 0
+datatype[0].structtype[0].field[0]
+datatype[0].structtype[0].inherits[0]
+datatype[0].weightedsettype[0]
+datatype[1].id -459619403
+datatype[1].annotationreftype[0]
+datatype[1].arraytype[0]
+datatype[1].documenttype[0]
+datatype[1].maptype[0]
+datatype[1].structtype[1]
+datatype[1].structtype[0].compresslevel 0
+datatype[1].structtype[0].compressminsize 800
+datatype[1].structtype[0].compressthreshold 95
+datatype[1].structtype[0].compresstype NONE
+datatype[1].structtype[0].name "simple.header"
+datatype[1].structtype[0].version 0
+datatype[1].structtype[0].field[1]
+datatype[1].structtype[0].field[0].datatype 2
+datatype[1].structtype[0].field[0].name "my_str"
+datatype[1].structtype[0].field[0].id[0]
+datatype[1].structtype[0].inherits[0]
+datatype[1].weightedsettype[0]
+datatype[2].id -1668955062
+datatype[2].annotationreftype[0]
+datatype[2].arraytype[0]
+datatype[2].documenttype[1]
+datatype[2].documenttype[0].bodystruct -1101456630
+datatype[2].documenttype[0].headerstruct -459619403
+datatype[2].documenttype[0].name "simple"
+datatype[2].documenttype[0].version 0
+datatype[2].documenttype[0].inherits[1]
+datatype[2].documenttype[0].inherits[0].name "document"
+datatype[2].documenttype[0].inherits[0].version 0
+datatype[2].maptype[0]
+datatype[2].structtype[0]
+datatype[2].weightedsettype[0]