diff --git a/pdb-ui/src/main/java/org/lucares/pdbui/TcpIngestor.java b/pdb-ui/src/main/java/org/lucares/pdbui/TcpIngestor.java index 895b75f..dbef96e 100644 --- a/pdb-ui/src/main/java/org/lucares/pdbui/TcpIngestor.java +++ b/pdb-ui/src/main/java/org/lucares/pdbui/TcpIngestor.java @@ -34,8 +34,8 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.scheduling.annotation.Async; import org.springframework.stereotype.Component; +import com.fasterxml.jackson.core.JsonParseException; import com.fasterxml.jackson.core.type.TypeReference; -import com.fasterxml.jackson.databind.MappingIterator; import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.databind.ObjectReader; @@ -56,8 +56,6 @@ public class TcpIngestor implements Ingestor, AutoCloseable, DisposableBean { public final static class Handler implements Callable { - private final ObjectMapper objectMapper = new ObjectMapper(); - private final TypeReference> typeReferenceForMap = new TypeReference>() { }; @@ -76,41 +74,41 @@ public class TcpIngestor implements Ingestor, AutoCloseable, DisposableBean { LOGGER.debug("opening streams to client"); try (PrintWriter out = new PrintWriter(clientSocket.getOutputStream(), true); BufferedReader in = new BufferedReader(new InputStreamReader(clientSocket.getInputStream())); - ) { + final ObjectMapper objectMapper = new ObjectMapper(); final ObjectReader objectReader = objectMapper.readerFor(typeReferenceForMap); - final MappingIterator iterator = objectReader.readValues(in); double duration = 0.0; int count = 0; LOGGER.debug("reading from stream"); - while (iterator.hasNext()) { + String line; + while ((line = in.readLine()) != null) { final long start = System.nanoTime(); - @SuppressWarnings("unchecked") - final Map object = (Map) iterator.next(); + try { + final Map object = objectReader.readValue(line); - final Optional entry = createEntry(object); - final long end = System.nanoTime(); - duration += (end - start) / 1_000_000.0; + final Optional entry = createEntry(object); + final long end = System.nanoTime(); + duration += (end - start) / 1_000_000.0; - count++; - if (count == 100000) { - METRICS_LOGGER.debug("reading {} took {} ms", count, duration); - duration = 0.0; - count = 0; + count++; + if (count == 100000) { + METRICS_LOGGER.debug("reading {} took {} ms", count, duration); + duration = 0.0; + count = 0; + } + + if (entry.isPresent()) { + LOGGER.debug("adding entry to queue: {}", entry); + queue.put(entry.get()); + } + } catch (JsonParseException e) { + LOGGER.info("json parse error in line '" + line + "'", e); } - - if (entry.isPresent()) { - LOGGER.debug("adding entry to queue: {}", entry); - queue.put(entry.get()); - } - } LOGGER.debug("connection closed: " + clientAddress); - } - catch (Exception e) - { + } catch (Exception e) { LOGGER.warn("Stream handling failed", e); throw e; } @@ -121,8 +119,7 @@ public class TcpIngestor implements Ingestor, AutoCloseable, DisposableBean { public Optional createEntry(final Map map) { try { - if (map.containsKey("duration") - && map.containsKey("@timestamp")) { + if (map.containsKey("duration") && map.containsKey("@timestamp")) { final OffsetDateTime date = getDate(map); final long duration = (int) map.get("duration"); diff --git a/pdb-ui/src/test/java/org/lucares/performance/db/ingestor/PdbTestUtil.java b/pdb-ui/src/test/java/org/lucares/performance/db/ingestor/PdbTestUtil.java index d4a30db..ade94c0 100644 --- a/pdb-ui/src/test/java/org/lucares/performance/db/ingestor/PdbTestUtil.java +++ b/pdb-ui/src/test/java/org/lucares/performance/db/ingestor/PdbTestUtil.java @@ -1,80 +1,101 @@ -package org.lucares.performance.db.ingestor; - -import java.io.IOException; -import java.net.ConnectException; -import java.net.InetSocketAddress; -import java.nio.ByteBuffer; -import java.nio.channels.SocketChannel; -import java.nio.charset.StandardCharsets; -import java.util.Arrays; -import java.util.HashMap; -import java.util.Map; -import java.util.concurrent.BlockingQueue; -import java.util.concurrent.LinkedBlockingDeque; -import java.util.concurrent.TimeUnit; - -import org.lucares.pdbui.TcpIngestor; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import com.fasterxml.jackson.databind.ObjectMapper; - -public class PdbTestUtil { - private static final Logger LOGGER = LoggerFactory.getLogger(PdbTestUtil.class); - - private static final Map POISON = new HashMap<>(); - - @SafeVarargs - public static final void send(final Map... entries) throws IOException, InterruptedException { - - final LinkedBlockingDeque> queue = new LinkedBlockingDeque<>(Arrays.asList(entries)); - queue.put(POISON); - send(queue); - } - - public static final void send(final BlockingQueue> aEntriesSupplier) throws IOException { - - final ObjectMapper mapper = new ObjectMapper(); - final SocketChannel channel = connect(); - - Map entry; - while ((entry = aEntriesSupplier.poll()) != POISON) { - - final StringBuilder streamData = new StringBuilder(); - streamData.append(mapper.writeValueAsString(entry)); - streamData.append("\n"); - - final ByteBuffer src = ByteBuffer.wrap(streamData.toString().getBytes(StandardCharsets.UTF_8)); - channel.write(src); - } - - try { - // ugly workaround: the channel was closed too early and not all - // data was received - TimeUnit.MILLISECONDS.sleep(10); - } catch (final InterruptedException e) { - throw new IllegalStateException(e); - } - channel.close(); - LOGGER.trace("closed sender connection"); - } - - private static SocketChannel connect() throws IOException { - - SocketChannel result = null; - - while (true) { - try { - result = SocketChannel.open(); - result.configureBlocking(true); - result.connect(new InetSocketAddress("127.0.0.1", TcpIngestor.PORT)); - break; - } catch (final ConnectException e) { - // server socket not yet ready, it should be ready any time soon - } - } - - return result; - } - -} +package org.lucares.performance.db.ingestor; + +import java.io.IOException; +import java.net.ConnectException; +import java.net.InetSocketAddress; +import java.nio.ByteBuffer; +import java.nio.channels.SocketChannel; +import java.nio.charset.StandardCharsets; +import java.util.Arrays; +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.LinkedBlockingDeque; +import java.util.concurrent.TimeUnit; + +import org.lucares.pdbui.TcpIngestor; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.fasterxml.jackson.databind.ObjectMapper; + +public class PdbTestUtil { + private static final Logger LOGGER = LoggerFactory.getLogger(PdbTestUtil.class); + + private static final Map POISON = new HashMap<>(); + + @SafeVarargs + public static final void send(final Map... entries) throws IOException, InterruptedException { + + final LinkedBlockingDeque> queue = new LinkedBlockingDeque<>(Arrays.asList(entries)); + queue.put(POISON); + send(queue); + } + + public static final void send(final BlockingQueue> aEntriesSupplier) throws IOException { + + final ObjectMapper mapper = new ObjectMapper(); + final SocketChannel channel = connect(); + + Map entry; + while ((entry = aEntriesSupplier.poll()) != POISON) { + + final StringBuilder streamData = new StringBuilder(); + streamData.append(mapper.writeValueAsString(entry)); + streamData.append("\n"); + + final ByteBuffer src = ByteBuffer.wrap(streamData.toString().getBytes(StandardCharsets.UTF_8)); + channel.write(src); + } + + try { + // ugly workaround: the channel was closed too early and not all + // data was received + TimeUnit.MILLISECONDS.sleep(10); + } catch (final InterruptedException e) { + throw new IllegalStateException(e); + } + channel.close(); + LOGGER.trace("closed sender connection"); + } + + public static final void send(final String data) throws IOException { + + final SocketChannel channel = connect(); + + final StringBuilder streamData = new StringBuilder(); + streamData.append(data); + + final ByteBuffer src = ByteBuffer.wrap(streamData.toString().getBytes(StandardCharsets.UTF_8)); + channel.write(src); + + try { + // ugly workaround: the channel was closed too early and not all + // data was received + TimeUnit.MILLISECONDS.sleep(10); + } catch (final InterruptedException e) { + throw new IllegalStateException(e); + } + channel.close(); + LOGGER.trace("closed sender connection"); + } + + private static SocketChannel connect() throws IOException { + + SocketChannel result = null; + + while (true) { + try { + result = SocketChannel.open(); + result.configureBlocking(true); + result.connect(new InetSocketAddress("127.0.0.1", TcpIngestor.PORT)); + break; + } catch (final ConnectException e) { + // server socket not yet ready, it should be ready any time soon + } + } + + return result; + } + +} diff --git a/pdb-ui/src/test/java/org/lucares/performance/db/ingestor/TcpIngestorTest.java b/pdb-ui/src/test/java/org/lucares/performance/db/ingestor/TcpIngestorTest.java index 93a9bda..28d22f3 100644 --- a/pdb-ui/src/test/java/org/lucares/performance/db/ingestor/TcpIngestorTest.java +++ b/pdb-ui/src/test/java/org/lucares/performance/db/ingestor/TcpIngestorTest.java @@ -23,6 +23,8 @@ import org.testng.annotations.AfterMethod; import org.testng.annotations.BeforeMethod; import org.testng.annotations.Test; +import com.fasterxml.jackson.databind.ObjectMapper; + @Test public class TcpIngestorTest { @@ -89,20 +91,28 @@ public class TcpIngestorTest { try (TcpIngestor tcpIngestor = new TcpIngestor(dataDirectory)) { tcpIngestor.start(); - // this entry will be skipped, because the date is invalid + // skipped, because the date is invalid final Map entryA = new HashMap<>(); entryA.put("duration", 1); entryA.put("@timestamp", invalidDate.format(DateTimeFormatter.ISO_ZONED_DATE_TIME)); entryA.put("host", host); entryA.put("tags", Collections.emptyList()); + // skipped, because it is not valid json + String corrupEntry = "{\"corrupt..."; + + // valid entry final Map entryB = new HashMap<>(); entryB.put("duration", 2); entryB.put("@timestamp", dateB.format(DateTimeFormatter.ISO_ZONED_DATE_TIME)); entryB.put("host", host); entryB.put("tags", Collections.emptyList()); - PdbTestUtil.send(entryA, entryB); + final ObjectMapper objectMapper = new ObjectMapper(); + final String data = objectMapper.writeValueAsString(entryA)+"\n"+corrupEntry+"\n"+objectMapper.writeValueAsString(entryB)+"\n"; + + + PdbTestUtil.send(data); } try (PerformanceDb db = new PerformanceDb(dataDirectory)) {