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

introduce indexes

Browse Source
This commit is contained in:
Andreas Huber
2021-05-09 10:33:28 +02:00
parent ae545e602c
commit 36ccc57db6
34 changed files with 721 additions and 758 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
pdb-ui/src/test/java/org/lucares/pdbui/CsvToEntryTransformerTest.java
View File

@@ -19,8 +19,9 @@ import org.junit.jupiter.api.Test;
import org.lucares.collections.LongList;
import org.lucares.pdb.api.DateTimeRange;
import org.lucares.pdb.api.Query;
import org.lucares.pdb.datastore.Entries;
import org.lucares.pdb.datastore.PdbIndexId;
import org.lucares.pdbui.CsvReaderSettings.ColumnDefinitions;
import org.lucares.performance.db.Entries;
import org.lucares.performance.db.PerformanceDb;
import org.lucares.utils.file.FileUtils;
@@ -43,9 +44,14 @@ public class CsvToEntryTransformerTest {
final OffsetDateTime dateA = OffsetDateTime.now();
final OffsetDateTime dateB = OffsetDateTime.now();
final String index = "test";
final PdbIndexId indexId = new PdbIndexId(index);
try (final PerformanceDb db = new PerformanceDb(dataDirectory)) {
final String csv = "@timestamp,duration,tag\n"//
db.createIndex(indexId, "test", "");
final String csv = "\u0001" + index + "\n" //
+ "@timestamp,duration,tag\n"//
+ dateA.format(DateTimeFormatter.ISO_ZONED_DATE_TIME) + ",1,tagValue\n"//
+ dateB.format(DateTimeFormatter.ISO_ZONED_DATE_TIME) + ",2,tagValue\n";
@@ -58,7 +64,8 @@ public class CsvToEntryTransformerTest {
}
try (PerformanceDb db = new PerformanceDb(dataDirectory)) {
final LongList result = db.get(new Query("tag=tagValue", DateTimeRange.max())).singleGroup().flatMap();
final LongList result = db.get(new Query("tag=tagValue", DateTimeRange.max(), indexId.getId()))
.singleGroup().flatMap();
Assertions.assertEquals(result.size(), 4);
Assertions.assertEquals(result.get(0), dateA.toInstant().toEpochMilli());
@@ -83,9 +90,13 @@ public class CsvToEntryTransformerTest {
@Test
public void testIgnoreColumns() throws IOException, InterruptedException, TimeoutException {
final String index = "test";
final PdbIndexId indexId = new PdbIndexId(index);
try (final PerformanceDb db = new PerformanceDb(dataDirectory)) {
db.createIndex(indexId, "test", "");
final String csv = "@timestamp,duration,ignoredColumn,-otherIgnoredColumn,tag\n"//
final String csv = "\u0001" + index + "\n"//
+ "@timestamp,duration,ignoredColumn,-otherIgnoredColumn,tag\n"//
+ "2000-01-01T00:00:00.000Z,1,ignoreValue,ignoreValue,tagValue\n"//
+ "2000-01-01T00:00:00.001Z,2,ignoreValue,ignoreValue,tagValue\n";
@@ -100,7 +111,7 @@ public class CsvToEntryTransformerTest {
}
try (PerformanceDb db = new PerformanceDb(dataDirectory)) {
final List<String> availableFields = db.getFields(DateTimeRange.max());
final List<String> availableFields = db.getFields(DateTimeRange.max(), indexId);
Assertions.assertEquals(List.of("tag").toString(), availableFields.toString(),
"the ignored field is not returned");
}

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

@@ -14,6 +14,7 @@ import org.junit.jupiter.api.extension.ExtendWith;
import org.lucares.collections.LongList;
import org.lucares.pdb.api.DateTimeRange;
import org.lucares.pdb.api.Query;
import org.lucares.pdb.datastore.PdbIndexId;
import org.lucares.pdbui.CsvReaderSettings.ColumnDefinitions;
import org.lucares.pdbui.CsvReaderSettings.PostProcessors;
import org.lucares.performance.db.PerformanceDb;
@@ -50,6 +51,9 @@ public class PdbControllerTest {
@Test
public void testUploadCsv() throws InterruptedException {
final PdbIndexId indexId = new PdbIndexId("test");
performanceDb.createIndex(indexId, "test", "");
final String additionalColumn = "additionalColumn";
final String additionalValue = "additionalValue";
final String ignoredColumn = "ignoredColumn";
@@ -59,6 +63,7 @@ public class PdbControllerTest {
final OffsetDateTime dateB = OffsetDateTime.now();
final String csv = "# first line is a comment\n"//
+ "\u0001" + indexId.getId() + "\n"//
+ timeColumn + "," + valueColumn + ",tag," + ignoredColumn + "\n"//
+ dateA.format(DateTimeFormatter.ISO_ZONED_DATE_TIME) + ",1,tagVALUE,ignoredValue\n"//
+ dateB.format(DateTimeFormatter.ISO_ZONED_DATE_TIME) + ",2,TAGvalue,ignoredValue\n";
@@ -70,11 +75,12 @@ public class PdbControllerTest {
settings.putAdditionalTag(additionalColumn, additionalValue);
uploadCsv(settings, csv);
{
final LongList resultTagValue = performanceDb.get(new Query("tag=tagvalue", DateTimeRange.ofDay(dateA)))
.singleGroup().flatMap();
final LongList resultAdditionalValue = performanceDb
.get(new Query(additionalColumn + "=" + additionalValue, DateTimeRange.ofDay(dateA))).singleGroup()
final LongList resultTagValue = performanceDb
.get(new Query("tag=tagvalue", DateTimeRange.ofDay(dateA), indexId.getId())).singleGroup()
.flatMap();
final LongList resultAdditionalValue = performanceDb.get(
new Query(additionalColumn + "=" + additionalValue, DateTimeRange.ofDay(dateA), indexId.getId()))
.singleGroup().flatMap();
System.out.println(PdbTestUtil.timeValueLongListToString(resultTagValue));
Assertions.assertEquals(resultTagValue, resultAdditionalValue,
@@ -90,7 +96,7 @@ public class PdbControllerTest {
Assertions.assertEquals(2, resultTagValue.get(3));
}
{
final List<String> fields = performanceDb.getFields(DateTimeRange.max());
final List<String> fields = performanceDb.getFields(DateTimeRange.max(), indexId);
Assertions.assertTrue(!fields.contains(ignoredColumn), "ignoredColumn not in fields. fields: " + fields);
Assertions.assertTrue(fields.contains(additionalColumn),
additionalColumn + " expected in fields. Fields were: " + fields);

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

@@ -20,55 +20,39 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import org.lucares.collections.LongList;
import org.lucares.pdb.datastore.PdbIndexId;
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);
static final Map<String, Object> POISON = new HashMap<>();
public static final void send(final String format, final Collection<Map<String, Object>> entries, final int port)
throws IOException, InterruptedException {
switch (format) {
case "csv":
sendAsCsv(entries, port);
break;
case "json":
sendAsJson(entries, port);
break;
default:
throw new IllegalStateException("unhandled format: " + format);
}
}
@SafeVarargs
public static final void sendAsCsv(final int port, final Map<String, Object>... entries)
public static final void sendAsCsv(final PdbIndexId indexId, final int port, final Map<String, Object>... entries)
throws IOException, InterruptedException {
sendAsCsv(Arrays.asList(entries), port);
sendAsCsv(indexId, Arrays.asList(entries), port);
}
public static final void sendAsCsv(final Collection<Map<String, Object>> entries, final int port)
throws IOException, InterruptedException {
public static final void sendAsCsv(final PdbIndexId indexId, final Collection<Map<String, Object>> entries,
final int port) throws IOException, InterruptedException {
final Set<String> keys = entries.stream().map(Map::keySet).flatMap(Set::stream).collect(Collectors.toSet());
sendAsCsv(keys, entries, port);
sendAsCsv(indexId, keys, entries, port);
}
public static final void sendAsCsv(final Collection<String> keys, final Collection<Map<String, Object>> entries,
final int port) throws IOException, InterruptedException {
public static final void sendAsCsv(final PdbIndexId indexId, final Collection<String> keys,
final Collection<Map<String, Object>> entries, final int port) throws IOException, InterruptedException {
final StringBuilder csv = new StringBuilder();
csv.append("\u0001" + indexId.getId());
csv.append(String.join(",", keys));
csv.append("\n");
@@ -85,48 +69,6 @@ public class PdbTestUtil {
send(csv.toString(), port);
}
@SafeVarargs
public static final void sendAsJson(final int port, final Map<String, Object>... entries)
throws IOException, InterruptedException {
sendAsJson(Arrays.asList(entries), port);
}
public static final void sendAsJson(final Collection<Map<String, Object>> entries, final int port)
throws IOException, InterruptedException {
final LinkedBlockingDeque<Map<String, Object>> queue = new LinkedBlockingDeque<>(entries);
queue.put(POISON);
sendAsJson(queue, port);
}
public static final void sendAsJson(final BlockingQueue<Map<String, Object>> aEntriesSupplier, final int port)
throws IOException {
final ObjectMapper mapper = new ObjectMapper();
final SocketChannel channel = connect(port);
Map<String, Object> 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, final int port) throws IOException {
final SocketChannel channel = connect(port);

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

@@ -20,13 +20,10 @@ import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
import org.lucares.collections.LongList;
import org.lucares.pdb.api.DateTimeRange;
import org.lucares.pdb.api.Query;
import org.lucares.pdb.datastore.internal.DataStore;
import org.lucares.performance.db.PdbExport;
import org.lucares.pdb.datastore.PdbIndexId;
import org.lucares.performance.db.PerformanceDb;
import org.lucares.utils.file.FileUtils;
import org.slf4j.Logger;
@@ -57,10 +54,13 @@ public class TcpIngestorTest {
final OffsetDateTime dateB = OffsetDateTime.now();
final String host = "someHost";
final PdbIndexId indexId = new PdbIndexId("test");
try (TcpIngestor ingestor = new TcpIngestor(dataDirectory)) {
ingestor.useRandomPort();
ingestor.start();
ingestor.getDb().createIndex(indexId, "test", "");
final Map<String, Object> entryA = new HashMap<>();
entryA.put("duration", 1);
entryA.put("@timestamp", dateA.format(DateTimeFormatter.ISO_ZONED_DATE_TIME));
@@ -73,15 +73,15 @@ public class TcpIngestorTest {
entryB.put("host", host);
entryB.put("tags", Collections.emptyList());
PdbTestUtil.sendAsJson(ingestor.getPort(), entryA, entryB);
PdbTestUtil.sendAsCsv(indexId, ingestor.getPort(), entryA, entryB);
} 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.ofDay(dateA))).singleGroup()
.flatMap();
final LongList result = db.get(new Query("host=" + host, DateTimeRange.ofDay(dateA), indexId.getId()))
.singleGroup().flatMap();
Assertions.assertEquals(4, result.size());
Assertions.assertEquals(dateA.toInstant().toEpochMilli(), result.get(0));
@@ -92,66 +92,6 @@ public class TcpIngestorTest {
}
}
@Test
public void testIngestDataViaTcpStream_CustomFormat() throws Exception {
final long dateA = Instant.now().toEpochMilli();
final long dateB = Instant.now().toEpochMilli() + 1;
final long dateC = Instant.now().toEpochMilli() - 1;
final DateTimeRange dateRange = DateTimeRange.relativeMinutes(1);
final String host = "someHost";
// 1. insert some data
try (TcpIngestor ingestor = new TcpIngestor(dataDirectory)) {
ingestor.useRandomPort();
ingestor.start();
final long deltaEpochMilliB = dateB - dateA;
final long deltaEpochMilliC = dateC - dateB;
final String data = "#$0:host=someHost,pod=somePod\n"//
+ dateA + ",1,0\n"// previous date is 0, therefore the delta is dateA / using tags with id 0
+ "$1:host=someHost,pod=otherPod\n" //
+ 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
PdbTestUtil.send(data, ingestor.getPort());
} catch (final Exception e) {
LOGGER.error("", e);
throw e;
}
// 2. export the data
final List<Path> exportFiles = PdbExport.export(dataDirectory, dataDirectory.resolve("export"));
// 3. delete database
FileUtils.delete(dataDirectory.resolve(DataStore.SUBDIR_STORAGE));
// 4. create a new database
try (TcpIngestor ingestor = new TcpIngestor(dataDirectory)) {
ingestor.useRandomPort();
ingestor.start();
for (final Path exportFile : exportFiles) {
PdbTestUtil.send(exportFile, ingestor.getPort());
}
}
// 5. check that the data is correctly inserted
try (PerformanceDb db = new PerformanceDb(dataDirectory)) {
final LongList result = db.get(new Query("host=" + host, dateRange)).singleGroup().flatMap();
Assertions.assertEquals(6, result.size());
Assertions.assertEquals(dateA, result.get(0));
Assertions.assertEquals(1, result.get(1));
Assertions.assertEquals(dateC, result.get(2));
Assertions.assertEquals(3, result.get(3));
Assertions.assertEquals(dateB, result.get(4));
Assertions.assertEquals(2, result.get(5));
}
}
@Test
public void testIngestionThreadDoesNotDieOnErrors() throws Exception {
final OffsetDateTime dateA = OffsetDateTime.now().minusMinutes(1);
@@ -159,10 +99,13 @@ public class TcpIngestorTest {
final DateTimeRange dateRange = new DateTimeRange(dateA, dateB);
final String host = "someHost";
final PdbIndexId indexId = new PdbIndexId("test");
try (TcpIngestor tcpIngestor = new TcpIngestor(dataDirectory)) {
tcpIngestor.useRandomPort();
tcpIngestor.start();
tcpIngestor.getDb().createIndex(indexId, "test", "");
// has a negative epoch time milli and negative value
final Map<String, Object> entryA = new HashMap<>();
entryA.put("duration", 1);
@@ -192,7 +135,8 @@ public class TcpIngestorTest {
}
try (PerformanceDb db = new PerformanceDb(dataDirectory)) {
final LongList result = db.get(new Query("host=" + host, dateRange)).singleGroup().flatMap();
final LongList result = db.get(new Query("host=" + host, dateRange, indexId.getId())).singleGroup()
.flatMap();
Assertions.assertEquals(4, result.size());
Assertions.assertEquals(dateA.toInstant().truncatedTo(ChronoUnit.MILLIS).toEpochMilli(), result.get(0));
@@ -203,9 +147,7 @@ public class TcpIngestorTest {
}
}
@ParameterizedTest
@ValueSource(strings = { "csv", "json" })
public void testRandomOrder(final String format) throws Exception {
public void testRandomOrder() throws Exception {
final ThreadLocalRandom rnd = ThreadLocalRandom.current();
final String host = "someHost";
@@ -215,10 +157,13 @@ public class TcpIngestorTest {
final LongList expected = new LongList();
final PdbIndexId indexId = new PdbIndexId("test");
try (TcpIngestor ingestor = new TcpIngestor(dataDirectory)) {
ingestor.useRandomPort();
ingestor.start();
ingestor.getDb().createIndex(indexId, "test", "");
final LinkedBlockingDeque<Map<String, Object>> queue = new LinkedBlockingDeque<>();
for (int i = 0; i < 103; i++) // use number of rows that is not a multiple of a page size
@@ -238,14 +183,15 @@ public class TcpIngestorTest {
expected.addAll(timestamp, duration);
}
PdbTestUtil.send(format, queue, ingestor.getPort());
PdbTestUtil.sendAsCsv(indexId, queue, ingestor.getPort());
} catch (final Exception e) {
LOGGER.error("", e);
throw e;
}
try (PerformanceDb db = new PerformanceDb(dataDirectory)) {
final LongList result = db.get(new Query("host=" + host, dateRange)).singleGroup().flatMap();
final LongList result = db.get(new Query("host=" + host, dateRange, indexId.getId())).singleGroup()
.flatMap();
Assertions.assertEquals(LongPair.fromLongList(expected), LongPair.fromLongList(result));
}
}
@@ -253,10 +199,13 @@ public class TcpIngestorTest {
@Test
public void testCsvIngestorIgnoresColumns() throws Exception {
final PdbIndexId indexId = new PdbIndexId("test");
try (TcpIngestor ingestor = new TcpIngestor(dataDirectory)) {
ingestor.useRandomPort();
ingestor.start();
ingestor.getDb().createIndex(indexId, "test", "");
final Map<String, Object> entry = new HashMap<>();
entry.put("@timestamp",
Instant.ofEpochMilli(1).atOffset(ZoneOffset.UTC).format(DateTimeFormatter.ISO_ZONED_DATE_TIME));
@@ -264,14 +213,14 @@ public class TcpIngestorTest {
entry.put("host", "someHost");
entry.put(CsvToEntryTransformer.COLUM_IGNORE_PREFIX + "ignored", "ignoredValue");
PdbTestUtil.sendAsCsv(ingestor.getPort(), entry);
PdbTestUtil.sendAsCsv(indexId, ingestor.getPort(), entry);
} catch (final Exception e) {
LOGGER.error("", e);
throw e;
}
try (PerformanceDb db = new PerformanceDb(dataDirectory)) {
final List<String> availableFields = db.getFields(DateTimeRange.max());
final List<String> availableFields = db.getFields(DateTimeRange.max(), indexId);
Assertions.assertEquals(List.of("host").toString(), availableFields.toString(),
"the ignored field is not returned");
}
@@ -283,10 +232,15 @@ public class TcpIngestorTest {
final String host = "someHost";
final long value1 = 222;
final long value2 = 1;
final PdbIndexId indexId = new PdbIndexId("test");
try (TcpIngestor ingestor = new TcpIngestor(dataDirectory)) {
ingestor.useRandomPort();
ingestor.start();
ingestor.getDb().createIndex(indexId, "test", "");
final Map<String, Object> entry1 = new HashMap<>();
entry1.put("@timestamp",
Instant.ofEpochMilli(1).atOffset(ZoneOffset.UTC).format(DateTimeFormatter.ISO_ZONED_DATE_TIME));
@@ -299,7 +253,7 @@ public class TcpIngestorTest {
entry2.put("host", host);
entry2.put("duration", value2);
PdbTestUtil.sendAsCsv(List.of("@timestamp", "host", "duration"), List.of(entry1, entry2),
PdbTestUtil.sendAsCsv(indexId, List.of("@timestamp", "host", "duration"), List.of(entry1, entry2),
ingestor.getPort());
} catch (final Exception e) {
LOGGER.error("", e);
@@ -307,7 +261,8 @@ public class TcpIngestorTest {
}
try (PerformanceDb db = new PerformanceDb(dataDirectory)) {
final LongList result = db.get(new Query("host=" + host, DateTimeRange.max())).singleGroup().flatMap();
final LongList result = db.get(new Query("host=" + host, DateTimeRange.max(), indexId.getId()))
.singleGroup().flatMap();
Assertions.assertEquals(4, result.size());
Assertions.assertEquals(value1, result.get(1));
@@ -325,10 +280,14 @@ public class TcpIngestorTest {
final OffsetDateTime dateNovember = OffsetDateTime.of(2019, 11, 30, 23, 59, 59, 999, ZoneOffset.UTC);
final OffsetDateTime dateDecember = OffsetDateTime.of(2019, 12, 1, 0, 0, 0, 0, ZoneOffset.UTC);
final PdbIndexId indexId = new PdbIndexId("test");
try (TcpIngestor ingestor = new TcpIngestor(dataDirectory)) {
ingestor.useRandomPort();
ingestor.start();
ingestor.getDb().createIndex(indexId, "test", "");
final Map<String, Object> entry1 = new HashMap<>();
entry1.put("@timestamp", dateNovember.format(DateTimeFormatter.ISO_ZONED_DATE_TIME));
entry1.put("host", host);
@@ -339,7 +298,7 @@ public class TcpIngestorTest {
entry2.put("host", host);
entry2.put("duration", value2);
PdbTestUtil.sendAsCsv(List.of("@timestamp", "host", "duration"), List.of(entry1, entry2),
PdbTestUtil.sendAsCsv(indexId, List.of("@timestamp", "host", "duration"), List.of(entry1, entry2),
ingestor.getPort());
} catch (final Exception e) {
LOGGER.error("", e);
@@ -348,13 +307,15 @@ public class TcpIngestorTest {
try (PerformanceDb db = new PerformanceDb(dataDirectory)) {
final DateTimeRange rangeNovember = new DateTimeRange(dateNovember, dateNovember);
final LongList resultNovember = db.get(new Query("host=" + host, rangeNovember)).singleGroup().flatMap();
final LongList resultNovember = db.get(new Query("host=" + host, rangeNovember, indexId.getId()))
.singleGroup().flatMap();
Assertions.assertEquals(2, resultNovember.size());
Assertions.assertEquals(dateNovember.toInstant().toEpochMilli(), resultNovember.get(0));
Assertions.assertEquals(value1, resultNovember.get(1));
final DateTimeRange rangeDecember = new DateTimeRange(dateDecember, dateDecember);
final LongList resultDecember = db.get(new Query("host=" + host, rangeDecember)).singleGroup().flatMap();
final LongList resultDecember = db.get(new Query("host=" + host, rangeDecember, indexId.getId()))
.singleGroup().flatMap();
Assertions.assertEquals(2, resultDecember.size());
Assertions.assertEquals(dateDecember.toInstant().toEpochMilli(), resultDecember.get(0));
Assertions.assertEquals(value2, resultDecember.get(1));