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

fix CSV parser corrupts duration if duration is last element in line

Browse Source
This commit is contained in:
Andreas Huber
2019-11-14 18:40:14 +01:00
parent d9b2327f35
commit 10a7710940
3 changed files with 160 additions and 116 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -240,7 +240,7 @@ public class TcpIngestor implements Ingestor, AutoCloseable, DisposableBean {
} else if (key == keyTimestamp) { } else if (key == keyTimestamp) {
epochMilli = dateParser.parseAsEpochMilli(line, lastSeparatorPosition + 1); epochMilli = dateParser.parseAsEpochMilli(line, lastSeparatorPosition + 1);
} else if (key == keyDuration) { } else if (key == keyDuration) {
duration = parseLong(line, lastSeparatorPosition + 1); duration = parseLong(line, lastSeparatorPosition + 1, separatorPosition);
} else if (lastSeparatorPosition + 1 < separatorPosition) { // value is not empty } else if (lastSeparatorPosition + 1 < separatorPosition) { // value is not empty
final int value = Tags.STRING_COMPRESSOR.put(line, lastSeparatorPosition + 1, final int value = Tags.STRING_COMPRESSOR.put(line, lastSeparatorPosition + 1,
separatorPosition); separatorPosition);
@@ -258,7 +258,7 @@ public class TcpIngestor implements Ingestor, AutoCloseable, DisposableBean {
return null; return null;
} }
private static long parseLong(final byte[] bytes, final int start) { private static long parseLong(final byte[] bytes, final int start, int endExclusive) {
long result = 0; long result = 0;
int i = start; int i = start;
int c = bytes[i]; int c = bytes[i];
@@ -267,7 +267,7 @@ public class TcpIngestor implements Ingestor, AutoCloseable, DisposableBean {
sign = -1; sign = -1;
i++; i++;
} }
while ((c = bytes[i]) >= 48 && c <= 57) { while (i < endExclusive && (c = bytes[i]) >= 48 && c <= 57) {
result = result * 10 + (c - 48); result = result * 10 + (c - 48);
i++; i++;
} }

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

@@ -28,147 +28,154 @@ import org.slf4j.LoggerFactory;
import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.databind.ObjectMapper;
public class PdbTestUtil { public class PdbTestUtil {
private static final Logger LOGGER = LoggerFactory.getLogger(PdbTestUtil.class); private static final Logger LOGGER = LoggerFactory.getLogger(PdbTestUtil.class);
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)
throws IOException, InterruptedException { throws IOException, InterruptedException {
switch (format) { switch (format) {
case "csv": case "csv":
sendAsCsv(entries); sendAsCsv(entries);
break; break;
case "json": case "json":
sendAsJson(entries); sendAsJson(entries);
break; break;
default: default:
throw new IllegalStateException("unhandled format: " + format); throw new IllegalStateException("unhandled format: " + format);
} }
} }
@SafeVarargs @SafeVarargs
public static final void sendAsCsv(final Map<String, Object>... entries) throws IOException, InterruptedException { public static final void sendAsCsv(final Map<String, Object>... entries) throws IOException, InterruptedException {
sendAsCsv(Arrays.asList(entries)); sendAsCsv(Arrays.asList(entries));
} }
public static final void sendAsCsv(final Collection<Map<String, Object>> entries) public static final void sendAsCsv(final Collection<Map<String, Object>> entries)
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());
final StringBuilder csv = new StringBuilder(); sendAsCsv(keys, entries);
}
csv.append(String.join(",", keys)); public static final void sendAsCsv(Collection<String> keys, final Collection<Map<String, Object>> entries)
csv.append("\n"); throws IOException, InterruptedException {
for (final Map<String, Object> entry : entries) {
final List<String> line = new ArrayList<>();
for (final String key : keys) {
final String value = String.valueOf(entry.getOrDefault(key, ""));
line.add(value);
}
csv.append(String.join(",", line));
csv.append("\n");
}
System.out.println("sending: " + csv);
send(csv.toString());
}
@SafeVarargs final StringBuilder csv = new StringBuilder();
public static final void sendAsJson(final Map<String, Object>... entries) throws IOException, InterruptedException {
sendAsJson(Arrays.asList(entries)); csv.append(String.join(",", keys));
} csv.append("\n");
public static final void sendAsJson(final Collection<Map<String, Object>> entries) for (final Map<String, Object> entry : entries) {
throws IOException, InterruptedException { final List<String> line = new ArrayList<>();
final LinkedBlockingDeque<Map<String, Object>> queue = new LinkedBlockingDeque<>(entries); for (final String key : keys) {
queue.put(POISON); final String value = String.valueOf(entry.getOrDefault(key, ""));
sendAsJson(queue); line.add(value);
} }
csv.append(String.join(",", line));
csv.append("\n");
}
System.out.println("sending: " + csv);
send(csv.toString());
}
public static final void sendAsJson(final BlockingQueue<Map<String, Object>> aEntriesSupplier) throws IOException { @SafeVarargs
public static final void sendAsJson(final Map<String, Object>... entries) throws IOException, InterruptedException {
final ObjectMapper mapper = new ObjectMapper(); sendAsJson(Arrays.asList(entries));
final SocketChannel channel = connect(); }
Map<String, Object> entry; public static final void sendAsJson(final Collection<Map<String, Object>> entries)
while ((entry = aEntriesSupplier.poll()) != POISON) { throws IOException, InterruptedException {
final LinkedBlockingDeque<Map<String, Object>> queue = new LinkedBlockingDeque<>(entries);
queue.put(POISON);
sendAsJson(queue);
}
final StringBuilder streamData = new StringBuilder(); public static final void sendAsJson(final BlockingQueue<Map<String, Object>> aEntriesSupplier) throws IOException {
streamData.append(mapper.writeValueAsString(entry));
streamData.append("\n");
final ByteBuffer src = ByteBuffer.wrap(streamData.toString().getBytes(StandardCharsets.UTF_8)); final ObjectMapper mapper = new ObjectMapper();
channel.write(src); final SocketChannel channel = connect();
}
try { Map<String, Object> entry;
// ugly workaround: the channel was closed too early and not all while ((entry = aEntriesSupplier.poll()) != POISON) {
// 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 StringBuilder streamData = new StringBuilder();
streamData.append(mapper.writeValueAsString(entry));
streamData.append("\n");
final SocketChannel channel = connect(); final ByteBuffer src = ByteBuffer.wrap(streamData.toString().getBytes(StandardCharsets.UTF_8));
channel.write(src);
}
final StringBuilder streamData = new StringBuilder(); try {
streamData.append(data); // 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");
}
final ByteBuffer src = ByteBuffer.wrap(streamData.toString().getBytes(StandardCharsets.UTF_8)); public static final void send(final String data) throws IOException {
channel.write(src);
try { final SocketChannel channel = connect();
// 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 void send(final Path file) throws IOException { final StringBuilder streamData = new StringBuilder();
final SocketChannel outputChannel = connect(); streamData.append(data);
try (final FileChannel inputChannel = FileChannel.open(file, StandardOpenOption.READ)) { final ByteBuffer src = ByteBuffer.wrap(streamData.toString().getBytes(StandardCharsets.UTF_8));
inputChannel.transferTo(0, Long.MAX_VALUE, outputChannel); channel.write(src);
}
try { try {
// ugly workaround: the channel was closed too early and not all // ugly workaround: the channel was closed too early and not all
// data was received // data was received
TimeUnit.MILLISECONDS.sleep(10); TimeUnit.MILLISECONDS.sleep(10);
} catch (final InterruptedException e) { } catch (final InterruptedException e) {
throw new IllegalStateException(e); throw new IllegalStateException(e);
} }
outputChannel.close(); channel.close();
LOGGER.trace("closed sender connection"); LOGGER.trace("closed sender connection");
} }
private static SocketChannel connect() throws IOException { public static void send(final Path file) throws IOException {
final SocketChannel outputChannel = connect();
SocketChannel result = null; try (final FileChannel inputChannel = FileChannel.open(file, StandardOpenOption.READ)) {
inputChannel.transferTo(0, Long.MAX_VALUE, outputChannel);
}
while (true) { try {
try { // ugly workaround: the channel was closed too early and not all
result = SocketChannel.open(); // data was received
result.configureBlocking(true); TimeUnit.MILLISECONDS.sleep(10);
result.connect(new InetSocketAddress("127.0.0.1", TcpIngestor.PORT)); } catch (final InterruptedException e) {
break; throw new IllegalStateException(e);
} catch (final ConnectException e) { }
// server socket not yet ready, it should be ready any time soon outputChannel.close();
} LOGGER.trace("closed sender connection");
} }
return result; 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;
}
} }

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

@@ -281,4 +281,41 @@ public class TcpIngestorTest {
"the ignored field is not returned"); "the ignored field is not returned");
} }
} }
public void testCsvIngestorHandlesDurationAtEnd() throws Exception {
String host = "someHost";
long value1 = 222;
long value2= 1;
try (TcpIngestor ingestor = new TcpIngestor(dataDirectory)) {
ingestor.start();
final Map<String, Object> entry1 = new HashMap<>();
entry1.put("@timestamp",
Instant.ofEpochMilli(1).atOffset(ZoneOffset.UTC).format(DateTimeFormatter.ISO_ZONED_DATE_TIME));
entry1.put("host", host);
entry1.put("duration", value1);
final Map<String, Object> entry2 = new HashMap<>();
entry2.put("@timestamp",
Instant.ofEpochMilli(2).atOffset(ZoneOffset.UTC).format(DateTimeFormatter.ISO_ZONED_DATE_TIME));
entry2.put("host", host);
entry2.put("duration", value2);
PdbTestUtil.sendAsCsv(List.of("@timestamp","host","duration"), List.of(entry1, entry2));
} catch (final Exception e) {
LOGGER.error("", e);
throw e;
}
try (PerformanceDb db = new PerformanceDb(dataDirectory)) {
final LongList result = db.get(new Query("host=" + host, DateTimeRange.max())).singleGroup().flatMap();
Assert.assertEquals(result.size(), 4);
Assert.assertEquals(result.get(1), value1);
Assert.assertEquals(result.get(3), value2);
}
}
} }