handle corrupt json

Entries must be separated by a newline. This allows
us to handle corrupt json entries, because we know
that entries only start at a line beginning.

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<Void> {
 
-		private final ObjectMapper objectMapper = new ObjectMapper();
-
 		private final TypeReference<Map<String, Object>> typeReferenceForMap = new TypeReference<Map<String, Object>>() {
 		};
 
@@ -76,19 +74,19 @@ 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<Object> 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")
+					try {
-					final Map<String, Object> object = (Map<String, Object>) iterator.next();
+						final Map<String, Object> object = objectReader.readValue(line);
 
 						final Optional<Entry> entry = createEntry(object);
 						final long end = System.nanoTime();
@@ -105,12 +103,12 @@ public class TcpIngestor implements Ingestor, AutoCloseable, DisposableBean {
 							LOGGER.debug("adding entry to queue: {}", entry);
 							queue.put(entry.get());
 						}
-
+					} catch (JsonParseException e) {
+						LOGGER.info("json parse error in line '" + line + "'", e);
+					}
 				}
 				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<Entry> createEntry(final Map<String, Object> map) {
 			try {
 
-				if (map.containsKey("duration")
+				if (map.containsKey("duration") && map.containsKey("@timestamp")) {
-						&& map.containsKey("@timestamp")) {
 					final OffsetDateTime date = getDate(map);
 					final long duration = (int) map.get("duration");
 

pdb-ui/src/test/java/org/lucares/performance/db/ingestor/PdbTestUtil.java

@@ -59,6 +59,27 @@ public class PdbTestUtil {
 		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;

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<String, Object> 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<String, Object> 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)) {