andi/perfdb
1
0
Fork 0
Code Issues 1 Pull Requests Packages Projects Releases Activity

make tests more robust

Browse Source 
1. we are now using a random port for listening
2. TcpIngestor.start() waits until the socket is established.
This commit is contained in:
Andreas Huber
2020-10-31 09:46:12 +01:00
parent af4151bc81
commit de6ffe5097
4 changed files with 75 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
pdb-ui/src/main/java/org/lucares/pdbui/TcpIngestor.java
View File

@@ -6,9 +6,11 @@ import java.net.Socket;
import java.net.SocketTimeoutException; import java.net.SocketTimeoutException;
import java.nio.file.Path; import java.nio.file.Path;
import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService; import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors; import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicBoolean;
import javax.annotation.PreDestroy; import javax.annotation.PreDestroy;
@@ -20,7 +22,6 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean; import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component; import org.springframework.stereotype.Component;
@Component @Component
@@ -37,6 +38,8 @@ public class TcpIngestor implements Ingestor, AutoCloseable, DisposableBean {
private final PerformanceDb db; private final PerformanceDb db;
private volatile int port = PORT;
public TcpIngestor(final Path dataDirectory) throws IOException { public TcpIngestor(final Path dataDirectory) throws IOException {
LOGGER.info("opening performance db: " + dataDirectory); LOGGER.info("opening performance db: " + dataDirectory);
db = new PerformanceDb(dataDirectory); db = new PerformanceDb(dataDirectory);
@@ -48,23 +51,42 @@ public class TcpIngestor implements Ingestor, AutoCloseable, DisposableBean {
this.db = db; this.db = db;
} }
public void useRandomPort() {
port = 0;
}
/**
* Returns the port used. If {@link #useRandomPort()} is used then the port may
* be null until the socket is open.
*
* @return the port used
*/
public int getPort() {
return port;
}
public PerformanceDb getDb() { public PerformanceDb getDb() {
return db; return db;
} }
@Async
@Override @Override
public void start() throws Exception { public void start() throws Exception {
final CountDownLatch started = new CountDownLatch(1);
serverThreadPool.submit(() -> listen()); serverThreadPool.submit(() -> listen(started));
final boolean startedSuccessfully = started.await(5, TimeUnit.SECONDS);
if (!startedSuccessfully) {
throw new TimeoutException("failed to start listener");
}
} }
private Void listen() throws IOException { private Void listen(final CountDownLatch started) throws IOException {
Thread.currentThread().setName("socket-listener"); Thread.currentThread().setName("socket-listener");
try (ServerSocket serverSocket = new ServerSocket(PORT);) { try (ServerSocket serverSocket = new ServerSocket(port);) {
LOGGER.info("listening on port " + PORT); port = serverSocket.getLocalPort();
LOGGER.info("listening on port " + serverSocket.getLocalPort());
serverSocket.setSoTimeout((int) TimeUnit.MILLISECONDS.toMillis(2)); serverSocket.setSoTimeout((int) TimeUnit.MILLISECONDS.toMillis(2));
started.countDown(); // notify caller that the socket is now listening
while (acceptNewConnections.get()) { while (acceptNewConnections.get()) {
try { try {

6
pdb-ui/src/test/java/org/lucares/pdbui/PdbControllerTest.java
View File

@@ -113,10 +113,10 @@ public class PdbControllerTest {
final HttpEntity<MultiValueMap<String, Object>> entity = new HttpEntity<MultiValueMap<String, Object>>( final HttpEntity<MultiValueMap<String, Object>> entity = new HttpEntity<MultiValueMap<String, Object>>(
parameters, headers); parameters, headers);
final ResponseEntity<String> response = rest.exchange("/data?waitUntilFinished=true", HttpMethod.POST, entity, final ResponseEntity<String> response = rest.exchange("/api/data?waitUntilFinished=true", HttpMethod.POST,
String.class); entity, String.class);
Assertions.assertEquals(response.getStatusCode(), HttpStatus.CREATED, "response status"); Assertions.assertEquals(HttpStatus.CREATED, response.getStatusCode(), "response status");
} }
} }

47
pdb-ui/src/test/java/org/lucares/pdbui/PdbTestUtil.java
View File

@@ -36,14 +36,14 @@ public class PdbTestUtil {
static final Map<String, Object> POISON = new HashMap<>(); static final Map<String, Object> POISON = new HashMap<>();
public static final void send(final String format, final Collection<Map<String, Object>> entries) public static final void send(final String format, final Collection<Map<String, Object>> entries, final int port)
throws IOException, InterruptedException { throws IOException, InterruptedException {
switch (format) { switch (format) {
case "csv": case "csv":
sendAsCsv(entries); sendAsCsv(entries, port);
break; break;
case "json": case "json":
sendAsJson(entries); sendAsJson(entries, port);
break; break;
default: default:
throw new IllegalStateException("unhandled format: " + format); throw new IllegalStateException("unhandled format: " + format);
@@ -51,20 +51,21 @@ public class PdbTestUtil {
} }
@SafeVarargs @SafeVarargs
public static final void sendAsCsv(final Map<String, Object>... entries) throws IOException, InterruptedException { public static final void sendAsCsv(final int port, final Map<String, Object>... entries)
sendAsCsv(Arrays.asList(entries)); throws IOException, InterruptedException {
sendAsCsv(Arrays.asList(entries), port);
} }
public static final void sendAsCsv(final Collection<Map<String, Object>> entries) public static final void sendAsCsv(final Collection<Map<String, Object>> entries, final int port)
throws IOException, InterruptedException { throws IOException, InterruptedException {
final Set<String> keys = entries.stream().map(Map::keySet).flatMap(Set::stream).collect(Collectors.toSet()); final Set<String> keys = entries.stream().map(Map::keySet).flatMap(Set::stream).collect(Collectors.toSet());
sendAsCsv(keys, entries); sendAsCsv(keys, entries, port);
} }
public static final void sendAsCsv(final Collection<String> keys, final Collection<Map<String, Object>> entries) public static final void sendAsCsv(final Collection<String> keys, final Collection<Map<String, Object>> entries,
throws IOException, InterruptedException { final int port) throws IOException, InterruptedException {
final StringBuilder csv = new StringBuilder(); final StringBuilder csv = new StringBuilder();
@@ -81,26 +82,28 @@ public class PdbTestUtil {
csv.append("\n"); csv.append("\n");
} }
System.out.println("sending: " + csv); System.out.println("sending: " + csv);
send(csv.toString()); send(csv.toString(), port);
} }
@SafeVarargs @SafeVarargs
public static final void sendAsJson(final Map<String, Object>... entries) throws IOException, InterruptedException { public static final void sendAsJson(final int port, final Map<String, Object>... entries)
throws IOException, InterruptedException {
sendAsJson(Arrays.asList(entries)); sendAsJson(Arrays.asList(entries), port);
} }
public static final void sendAsJson(final Collection<Map<String, Object>> entries) public static final void sendAsJson(final Collection<Map<String, Object>> entries, final int port)
throws IOException, InterruptedException { throws IOException, InterruptedException {
final LinkedBlockingDeque<Map<String, Object>> queue = new LinkedBlockingDeque<>(entries); final LinkedBlockingDeque<Map<String, Object>> queue = new LinkedBlockingDeque<>(entries);
queue.put(POISON); queue.put(POISON);
sendAsJson(queue); sendAsJson(queue, port);
} }
public static final void sendAsJson(final BlockingQueue<Map<String, Object>> aEntriesSupplier) throws IOException { public static final void sendAsJson(final BlockingQueue<Map<String, Object>> aEntriesSupplier, final int port)
throws IOException {
final ObjectMapper mapper = new ObjectMapper(); final ObjectMapper mapper = new ObjectMapper();
final SocketChannel channel = connect(); final SocketChannel channel = connect(port);
Map<String, Object> entry; Map<String, Object> entry;
while ((entry = aEntriesSupplier.poll()) != POISON) { while ((entry = aEntriesSupplier.poll()) != POISON) {
@@ -124,9 +127,9 @@ public class PdbTestUtil {
LOGGER.trace("closed sender connection"); LOGGER.trace("closed sender connection");
} }
public static final void send(final String data) throws IOException { public static final void send(final String data, final int port) throws IOException {
final SocketChannel channel = connect(); final SocketChannel channel = connect(port);
final StringBuilder streamData = new StringBuilder(); final StringBuilder streamData = new StringBuilder();
streamData.append(data); streamData.append(data);
@@ -145,8 +148,8 @@ public class PdbTestUtil {
LOGGER.trace("closed sender connection"); LOGGER.trace("closed sender connection");
} }
public static void send(final Path file) throws IOException { public static void send(final Path file, final int port) throws IOException {
final SocketChannel outputChannel = connect(); final SocketChannel outputChannel = connect(port);
try (final FileChannel inputChannel = FileChannel.open(file, StandardOpenOption.READ)) { try (final FileChannel inputChannel = FileChannel.open(file, StandardOpenOption.READ)) {
inputChannel.transferTo(0, Long.MAX_VALUE, outputChannel); inputChannel.transferTo(0, Long.MAX_VALUE, outputChannel);
@@ -163,7 +166,7 @@ public class PdbTestUtil {
LOGGER.trace("closed sender connection"); LOGGER.trace("closed sender connection");
} }
private static SocketChannel connect() throws IOException { private static SocketChannel connect(final int port) throws IOException {
SocketChannel result = null; SocketChannel result = null;
@@ -171,7 +174,7 @@ public class PdbTestUtil {
try { try {
result = SocketChannel.open(); result = SocketChannel.open();
result.configureBlocking(true); result.configureBlocking(true);
result.connect(new InetSocketAddress("127.0.0.1", TcpIngestor.PORT)); result.connect(new InetSocketAddress("127.0.0.1", port));
break; break;
} catch (final ConnectException e) { } catch (final ConnectException e) {
// server socket not yet ready, it should be ready any time soon // server socket not yet ready, it should be ready any time soon

32
pdb-ui/src/test/java/org/lucares/pdbui/TcpIngestorTest.java
View File

@@ -58,7 +58,7 @@ public class TcpIngestorTest {
final String host = "someHost"; final String host = "someHost";
try (TcpIngestor ingestor = new TcpIngestor(dataDirectory)) { try (TcpIngestor ingestor = new TcpIngestor(dataDirectory)) {
ingestor.useRandomPort();
ingestor.start(); ingestor.start();
final Map<String, Object> entryA = new HashMap<>(); final Map<String, Object> entryA = new HashMap<>();
@@ -73,7 +73,7 @@ public class TcpIngestorTest {
entryB.put("host", host); entryB.put("host", host);
entryB.put("tags", Collections.emptyList()); entryB.put("tags", Collections.emptyList());
PdbTestUtil.sendAsJson(entryA, entryB); PdbTestUtil.sendAsJson(ingestor.getPort(), entryA, entryB);
} catch (final Exception e) { } catch (final Exception e) {
LOGGER.error("", e); LOGGER.error("", e);
throw e; throw e;
@@ -103,7 +103,7 @@ public class TcpIngestorTest {
// 1. insert some data // 1. insert some data
try (TcpIngestor ingestor = new TcpIngestor(dataDirectory)) { try (TcpIngestor ingestor = new TcpIngestor(dataDirectory)) {
ingestor.useRandomPort();
ingestor.start(); ingestor.start();
final long deltaEpochMilliB = dateB - dateA; final long deltaEpochMilliB = dateB - dateA;
@@ -115,7 +115,7 @@ public class TcpIngestorTest {
+ deltaEpochMilliB + ",2,1\n" // dates are the delta the the previous date / using tags with id 1 + deltaEpochMilliB + ",2,1\n" // dates are the delta the the previous date / using tags with id 1
+ deltaEpochMilliC + ",3,0"; // dates are the delta the the previous date / using tags with id 0 + deltaEpochMilliC + ",3,0"; // dates are the delta the the previous date / using tags with id 0
PdbTestUtil.send(data); PdbTestUtil.send(data, ingestor.getPort());
} catch (final Exception e) { } catch (final Exception e) {
LOGGER.error("", e); LOGGER.error("", e);
throw e; throw e;
@@ -129,9 +129,10 @@ public class TcpIngestorTest {
// 4. create a new database // 4. create a new database
try (TcpIngestor ingestor = new TcpIngestor(dataDirectory)) { try (TcpIngestor ingestor = new TcpIngestor(dataDirectory)) {
ingestor.useRandomPort();
ingestor.start(); ingestor.start();
for (final Path exportFile : exportFiles) { for (final Path exportFile : exportFiles) {
PdbTestUtil.send(exportFile); PdbTestUtil.send(exportFile, ingestor.getPort());
} }
} }
@@ -159,6 +160,7 @@ public class TcpIngestorTest {
final String host = "someHost"; final String host = "someHost";
try (TcpIngestor tcpIngestor = new TcpIngestor(dataDirectory)) { try (TcpIngestor tcpIngestor = new TcpIngestor(dataDirectory)) {
tcpIngestor.useRandomPort();
tcpIngestor.start(); tcpIngestor.start();
// has a negative epoch time milli and negative value // has a negative epoch time milli and negative value
@@ -186,7 +188,7 @@ public class TcpIngestorTest {
)// )//
+ "\n"; + "\n";
PdbTestUtil.send(data); PdbTestUtil.send(data, tcpIngestor.getPort());
} }
try (PerformanceDb db = new PerformanceDb(dataDirectory)) { try (PerformanceDb db = new PerformanceDb(dataDirectory)) {
@@ -214,7 +216,7 @@ public class TcpIngestorTest {
final LongList expected = new LongList(); final LongList expected = new LongList();
try (TcpIngestor ingestor = new TcpIngestor(dataDirectory)) { try (TcpIngestor ingestor = new TcpIngestor(dataDirectory)) {
ingestor.useRandomPort();
ingestor.start(); ingestor.start();
final LinkedBlockingDeque<Map<String, Object>> queue = new LinkedBlockingDeque<>(); final LinkedBlockingDeque<Map<String, Object>> queue = new LinkedBlockingDeque<>();
@@ -236,7 +238,7 @@ public class TcpIngestorTest {
expected.addAll(timestamp, duration); expected.addAll(timestamp, duration);
} }
PdbTestUtil.send(format, queue); PdbTestUtil.send(format, queue, ingestor.getPort());
} catch (final Exception e) { } catch (final Exception e) {
LOGGER.error("", e); LOGGER.error("", e);
throw e; throw e;
@@ -252,7 +254,7 @@ public class TcpIngestorTest {
public void testCsvIngestorIgnoresColumns() throws Exception { public void testCsvIngestorIgnoresColumns() throws Exception {
try (TcpIngestor ingestor = new TcpIngestor(dataDirectory)) { try (TcpIngestor ingestor = new TcpIngestor(dataDirectory)) {
ingestor.useRandomPort();
ingestor.start(); ingestor.start();
final Map<String, Object> entry = new HashMap<>(); final Map<String, Object> entry = new HashMap<>();
@@ -262,7 +264,7 @@ public class TcpIngestorTest {
entry.put("host", "someHost"); entry.put("host", "someHost");
entry.put(CsvToEntryTransformer.COLUM_IGNORE_PREFIX + "ignored", "ignoredValue"); entry.put(CsvToEntryTransformer.COLUM_IGNORE_PREFIX + "ignored", "ignoredValue");
PdbTestUtil.sendAsCsv(entry); PdbTestUtil.sendAsCsv(ingestor.getPort(), entry);
} catch (final Exception e) { } catch (final Exception e) {
LOGGER.error("", e); LOGGER.error("", e);
throw e; throw e;
@@ -282,7 +284,7 @@ public class TcpIngestorTest {
final long value1 = 222; final long value1 = 222;
final long value2 = 1; final long value2 = 1;
try (TcpIngestor ingestor = new TcpIngestor(dataDirectory)) { try (TcpIngestor ingestor = new TcpIngestor(dataDirectory)) {
ingestor.useRandomPort();
ingestor.start(); ingestor.start();
final Map<String, Object> entry1 = new HashMap<>(); final Map<String, Object> entry1 = new HashMap<>();
@@ -297,7 +299,8 @@ public class TcpIngestorTest {
entry2.put("host", host); entry2.put("host", host);
entry2.put("duration", value2); entry2.put("duration", value2);
PdbTestUtil.sendAsCsv(List.of("@timestamp", "host", "duration"), List.of(entry1, entry2)); PdbTestUtil.sendAsCsv(List.of("@timestamp", "host", "duration"), List.of(entry1, entry2),
ingestor.getPort());
} catch (final Exception e) { } catch (final Exception e) {
LOGGER.error("", e); LOGGER.error("", e);
throw e; throw e;
@@ -323,7 +326,7 @@ public class TcpIngestorTest {
final OffsetDateTime dateDecember = OffsetDateTime.of(2019, 12, 1, 0, 0, 0, 0, ZoneOffset.UTC); final OffsetDateTime dateDecember = OffsetDateTime.of(2019, 12, 1, 0, 0, 0, 0, ZoneOffset.UTC);
try (TcpIngestor ingestor = new TcpIngestor(dataDirectory)) { try (TcpIngestor ingestor = new TcpIngestor(dataDirectory)) {
ingestor.useRandomPort();
ingestor.start(); ingestor.start();
final Map<String, Object> entry1 = new HashMap<>(); final Map<String, Object> entry1 = new HashMap<>();
@@ -336,7 +339,8 @@ public class TcpIngestorTest {
entry2.put("host", host); entry2.put("host", host);
entry2.put("duration", value2); entry2.put("duration", value2);
PdbTestUtil.sendAsCsv(List.of("@timestamp", "host", "duration"), List.of(entry1, entry2)); PdbTestUtil.sendAsCsv(List.of("@timestamp", "host", "duration"), List.of(entry1, entry2),
ingestor.getPort());
} catch (final Exception e) { } catch (final Exception e) {
LOGGER.error("", e); LOGGER.error("", e);
throw e; throw e;