queryRe
// Query benchmark section (optional)
if (queryResults != null && !queryResults.isEmpty()) {
+ int qReps = queryResults.get(0).repetitionsMs().length;
+ String qRepHeaders = java.util.stream.IntStream.rangeClosed(1, qReps)
+ .mapToObj(i -> "rep" + i + "Ms")
+ .collect(Collectors.joining(","));
pw.println();
pw.println("# QUERY BENCHMARK");
- pw.println("engine,dataset,queryId,queryCategory,medianMs,medianTtfrMs,heapDeltaMb,rowCount,resultHash,status,rep1Ms,rep2Ms,rep3Ms");
+ pw.println("engine,dataset,queryId,queryCategory,medianMs,medianTtfrMs,heapDeltaMb,rowCount,resultHash,status," + qRepHeaders);
for (QueryResult r : queryResults) {
String reps = Arrays.stream(r.repetitionsMs())
.mapToObj(String::valueOf)
diff --git a/src/main/java/fr/inria/corese/benchmark/telemetry/IngestionTelemetrySampler.java b/src/main/java/fr/inria/corese/benchmark/telemetry/IngestionTelemetrySampler.java
new file mode 100644
index 0000000..e958a84
--- /dev/null
+++ b/src/main/java/fr/inria/corese/benchmark/telemetry/IngestionTelemetrySampler.java
@@ -0,0 +1,152 @@
+package fr.inria.corese.benchmark.telemetry;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.LongSupplier;
+
+/**
+ * Periodically samples ingestion telemetry (elapsed time, triples loaded and JVM heap usage)
+ * on a dedicated daemon thread, with minimal overhead on the sampled workload.
+ *
+ * Typical usage:
+ *
{@code
+ * IngestionTelemetrySampler sampler = new IngestionTelemetrySampler(100, () -> tripleCount.get());
+ * sampler.start();
+ * // ... run ingestion ...
+ * List points = sampler.stop(tripleCount.get());
+ * }
+ *
+ * This class is thread-safe: {@link #start()} and {@link #stop(long)} may be called from any
+ * thread, and sampling itself runs on an internal single-threaded scheduler. Resources are
+ * released deterministically through {@link #close()} (also triggered by {@link #stop(long)}),
+ * avoiding thread or memory leaks.
+ */
+public final class IngestionTelemetrySampler implements AutoCloseable {
+
+ private final long samplingPeriodMs;
+ private final LongSupplier triplesSupplier;
+ private final List points = Collections.synchronizedList(new ArrayList<>());
+ private final AtomicBoolean started = new AtomicBoolean(false);
+ private final AtomicBoolean stopped = new AtomicBoolean(false);
+
+ private ScheduledExecutorService scheduler;
+ private volatile long startTimeNanos;
+
+ /**
+ * Creates a sampler.
+ *
+ * @param samplingPeriodMs period, in milliseconds, between two consecutive samples (e.g. 100)
+ * @param triplesSupplier supplier returning the current number of triples loaded; invoked on
+ * the internal scheduler thread, must be cheap and non-blocking
+ */
+ public IngestionTelemetrySampler(long samplingPeriodMs, LongSupplier triplesSupplier) {
+ if (samplingPeriodMs <= 0) {
+ throw new IllegalArgumentException("samplingPeriodMs must be strictly positive");
+ }
+ this.samplingPeriodMs = samplingPeriodMs;
+ this.triplesSupplier = java.util.Objects.requireNonNull(triplesSupplier, "triplesSupplier");
+ }
+
+ /**
+ * Records the initial telemetry point at {@code elapsedMs = 0} and starts the periodic
+ * sampling scheduler. Must be called at most once.
+ */
+ public void start() {
+ if (!started.compareAndSet(false, true)) {
+ throw new IllegalStateException("IngestionTelemetrySampler has already been started");
+ }
+ this.startTimeNanos = System.nanoTime();
+ this.scheduler = Executors.newSingleThreadScheduledExecutor(new DaemonThreadFactory());
+
+ points.add(new TelemetryPoint(0L, triplesSupplier.getAsLong(), currentHeapUsedMb()));
+
+ scheduler.scheduleAtFixedRate(
+ this::sampleOnce,
+ samplingPeriodMs,
+ samplingPeriodMs,
+ TimeUnit.MILLISECONDS);
+ }
+
+ private void sampleOnce() {
+ try {
+ long elapsedMs = elapsedMsSinceStart();
+ points.add(new TelemetryPoint(elapsedMs, triplesSupplier.getAsLong(), currentHeapUsedMb()));
+ } catch (RuntimeException e) {
+ // A failing sample must never kill the scheduler or the ingestion it observes.
+ }
+ }
+
+ /**
+ * Stops the periodic sampling, records a final telemetry point using {@code finalTriples},
+ * releases scheduler resources and returns an immutable, chronologically ordered snapshot
+ * of all recorded points. Safe to call at most once; subsequent calls return the same result.
+ *
+ * @param finalTriples final number of triples loaded, recorded in the last sample
+ * @return the immutable list of telemetry points collected during the sampling session
+ */
+ public List stop(long finalTriples) {
+ if (!started.get()) {
+ throw new IllegalStateException("IngestionTelemetrySampler has not been started");
+ }
+ if (stopped.compareAndSet(false, true)) {
+ long elapsedMs = elapsedMsSinceStart();
+ shutdownScheduler();
+ points.add(new TelemetryPoint(elapsedMs, finalTriples, currentHeapUsedMb()));
+ }
+ synchronized (points) {
+ return List.copyOf(points);
+ }
+ }
+
+ /**
+ * Releases scheduler resources if they have not already been released. Equivalent to calling
+ * {@link #stop(long)} with the last known triple count, except it does not add a final sample
+ * when sampling was never started or was already stopped.
+ */
+ @Override
+ public void close() {
+ shutdownScheduler();
+ }
+
+ private void shutdownScheduler() {
+ ScheduledExecutorService executor = this.scheduler;
+ if (executor == null) {
+ return;
+ }
+ executor.shutdown();
+ try {
+ if (!executor.awaitTermination(2, TimeUnit.SECONDS)) {
+ executor.shutdownNow();
+ }
+ } catch (InterruptedException e) {
+ executor.shutdownNow();
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ private long elapsedMsSinceStart() {
+ return TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startTimeNanos);
+ }
+
+ private static double currentHeapUsedMb() {
+ Runtime runtime = Runtime.getRuntime();
+ long usedBytes = runtime.totalMemory() - runtime.freeMemory();
+ return Math.round(usedBytes / (1024.0 * 1024.0) * 10.0) / 10.0;
+ }
+
+ /** Thread factory producing a single named daemon thread so the JVM can exit cleanly. */
+ private static final class DaemonThreadFactory implements ThreadFactory {
+ @Override
+ public Thread newThread(Runnable r) {
+ Thread thread = new Thread(r, "ingestion-telemetry-sampler");
+ thread.setDaemon(true);
+ return thread;
+ }
+ }
+}
diff --git a/src/main/java/fr/inria/corese/benchmark/telemetry/SystemEnvironmentCollector.java b/src/main/java/fr/inria/corese/benchmark/telemetry/SystemEnvironmentCollector.java
new file mode 100644
index 0000000..cec36f6
--- /dev/null
+++ b/src/main/java/fr/inria/corese/benchmark/telemetry/SystemEnvironmentCollector.java
@@ -0,0 +1,302 @@
+package fr.inria.corese.benchmark.telemetry;
+
+import java.lang.management.ManagementFactory;
+import java.lang.management.RuntimeMXBean;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/**
+ * Collects a snapshot of the host system and JVM environment used to run benchmarks.
+ *
+ * Information is primarily read from Linux pseudo-files ({@code /proc/cpuinfo},
+ * {@code /proc/meminfo}, {@code /proc/sys/vm/swappiness}, {@code /etc/os-release}) and from
+ * {@link ManagementFactory#getRuntimeMXBean()}. Every source is read defensively: if a file is
+ * missing, unreadable, or does not contain the expected field, the collector falls back to a
+ * reasonable value derived from JVM/System properties (or {@code "unknown"}/{@code -1} when no
+ * fallback is available) instead of failing.
+ */
+public final class SystemEnvironmentCollector {
+
+ private static final Path CPUINFO_PATH = Path.of("/proc/cpuinfo");
+ private static final Path MEMINFO_PATH = Path.of("/proc/meminfo");
+ private static final Path SWAPPINESS_PATH = Path.of("/proc/sys/vm/swappiness");
+ private static final Path OS_RELEASE_PATH = Path.of("/etc/os-release");
+ private static final String UNKNOWN = "unknown";
+
+ private SystemEnvironmentCollector() {
+ // Static utility class, not meant to be instantiated.
+ }
+
+ /**
+ * Builds the environment description map.
+ *
+ * @return a mutable, insertion-ordered map containing the keys: {@code cpuModel},
+ * {@code cpuArchitecture}, {@code cpuSockets}, {@code cpuCores}, {@code cpuThreads},
+ * {@code cpuFrequencyMhz}, {@code cpuGovernor}, {@code ramTotalGb}, {@code swapTotalGb},
+ * {@code swappiness}, {@code osDistribution}, {@code osKernel}, {@code jvmVendor},
+ * {@code jvmVersion}, {@code jvmFlags} (as a {@code List})
+ */
+ public static Map collect() {
+ Map env = new LinkedHashMap<>();
+
+ List cpuinfoLines = readAllLinesOrEmpty(CPUINFO_PATH);
+ List meminfoLines = readAllLinesOrEmpty(MEMINFO_PATH);
+
+ env.put("cpuModel", readCpuModel(cpuinfoLines));
+ env.put("cpuArchitecture", readCpuArchitecture());
+ env.put("cpuSockets", readCpuSockets(cpuinfoLines));
+ env.put("cpuCores", readCpuCores(cpuinfoLines));
+ env.put("cpuThreads", readCpuThreads(cpuinfoLines));
+ env.put("cpuFrequencyMhz", readCpuFrequencyMhz(cpuinfoLines));
+ env.put("cpuGovernor", readCpuGovernor());
+ env.put("ramTotalGb", readRamTotalGb(meminfoLines));
+ env.put("swapTotalGb", readSwapTotalGb(meminfoLines));
+ env.put("swappiness", readSwappiness());
+ env.put("osDistribution", readOsDistribution());
+ env.put("osKernel", readOsKernel());
+ env.put("jvmVendor", readJvmVendor());
+ env.put("jvmVersion", readJvmVersion());
+ env.put("jvmFlags", readJvmFlags());
+
+ return env;
+ }
+
+ // ---------------------------------------------------------------------
+ // CPU
+ // ---------------------------------------------------------------------
+
+ private static String readCpuModel(List cpuinfoLines) {
+ for (String line : cpuinfoLines) {
+ if (line.startsWith("model name")) {
+ String value = valueAfterColon(line);
+ if (value != null && !value.isBlank()) {
+ return value.trim();
+ }
+ }
+ }
+ return System.getProperty("os.arch", UNKNOWN) + " CPU";
+ }
+
+ private static String readCpuArchitecture() {
+ return System.getProperty("os.arch", UNKNOWN);
+ }
+
+ private static int readCpuSockets(List cpuinfoLines) {
+ java.util.Set physicalIds = new java.util.HashSet<>();
+ for (String line : cpuinfoLines) {
+ if (line.startsWith("physical id")) {
+ String value = valueAfterColon(line);
+ if (value != null) {
+ physicalIds.add(value.trim());
+ }
+ }
+ }
+ if (!physicalIds.isEmpty()) {
+ return physicalIds.size();
+ }
+ return 1;
+ }
+
+ private static int readCpuCores(List cpuinfoLines) {
+ java.util.Set coreKeys = new java.util.LinkedHashSet<>();
+ String currentPhysicalId = "0";
+ String currentCoreId = null;
+ for (String line : cpuinfoLines) {
+ if (line.startsWith("physical id")) {
+ String value = valueAfterColon(line);
+ currentPhysicalId = (value != null) ? value.trim() : "0";
+ } else if (line.startsWith("core id")) {
+ currentCoreId = valueAfterColon(line);
+ if (currentCoreId != null) {
+ coreKeys.add(currentPhysicalId + ":" + currentCoreId.trim());
+ }
+ }
+ }
+ if (!coreKeys.isEmpty()) {
+ return coreKeys.size();
+ }
+ int threads = readCpuThreads(cpuinfoLines);
+ return threads > 0 ? threads : Runtime.getRuntime().availableProcessors();
+ }
+
+ private static int readCpuThreads(List cpuinfoLines) {
+ int processorCount = 0;
+ for (String line : cpuinfoLines) {
+ if (line.startsWith("processor")) {
+ processorCount++;
+ }
+ }
+ if (processorCount > 0) {
+ return processorCount;
+ }
+ return Runtime.getRuntime().availableProcessors();
+ }
+
+ private static double readCpuFrequencyMhz(List cpuinfoLines) {
+ for (String line : cpuinfoLines) {
+ if (line.startsWith("cpu MHz")) {
+ String value = valueAfterColon(line);
+ if (value != null) {
+ try {
+ return Math.round(Double.parseDouble(value.trim()) * 10.0) / 10.0;
+ } catch (NumberFormatException ignored) {
+ // Fall through to next line / fallback.
+ }
+ }
+ }
+ }
+ return -1.0;
+ }
+
+ private static String readCpuGovernor() {
+ Path governorPath = Path.of("/sys/devices/system/cpu/cpu0/cpufreq/scaling_governor");
+ try {
+ if (Files.isReadable(governorPath)) {
+ String content = Files.readString(governorPath).trim();
+ if (!content.isEmpty()) {
+ return content;
+ }
+ }
+ } catch (java.io.IOException | RuntimeException ignored) {
+ // Fall through to the default below.
+ }
+ return UNKNOWN;
+ }
+
+ // ---------------------------------------------------------------------
+ // Memory
+ // ---------------------------------------------------------------------
+
+ private static double readRamTotalGb(List meminfoLines) {
+ long kb = readMeminfoValueKb(meminfoLines, "MemTotal");
+ if (kb >= 0) {
+ return kbToGb(kb);
+ }
+ // Fallback: JVM-visible max heap is not the total RAM, but is the best signal available
+ // without access to /proc/meminfo; it is clearly a lower bound.
+ return kbToGb(Runtime.getRuntime().maxMemory() / 1024L);
+ }
+
+ private static double readSwapTotalGb(List meminfoLines) {
+ long kb = readMeminfoValueKb(meminfoLines, "SwapTotal");
+ if (kb >= 0) {
+ return kbToGb(kb);
+ }
+ return 0.0;
+ }
+
+ private static long readMeminfoValueKb(List meminfoLines, String key) {
+ Pattern pattern = Pattern.compile("^" + Pattern.quote(key) + "\\s*:\\s*(\\d+)\\s*kB\\s*$");
+ for (String line : meminfoLines) {
+ Matcher matcher = pattern.matcher(line.trim());
+ if (matcher.matches()) {
+ try {
+ return Long.parseLong(matcher.group(1));
+ } catch (NumberFormatException ignored) {
+ return -1L;
+ }
+ }
+ }
+ return -1L;
+ }
+
+ private static double kbToGb(long kb) {
+ return Math.round(kb / (1024.0 * 1024.0) * 100.0) / 100.0;
+ }
+
+ private static int readSwappiness() {
+ try {
+ if (Files.isReadable(SWAPPINESS_PATH)) {
+ String content = Files.readString(SWAPPINESS_PATH).trim();
+ if (!content.isEmpty()) {
+ return Integer.parseInt(content);
+ }
+ }
+ } catch (java.io.IOException | RuntimeException ignored) {
+ // Fall through to the default below.
+ }
+ return -1;
+ }
+
+ // ---------------------------------------------------------------------
+ // Operating system
+ // ---------------------------------------------------------------------
+
+ private static String readOsDistribution() {
+ try {
+ if (Files.isReadable(OS_RELEASE_PATH)) {
+ for (String line : Files.readAllLines(OS_RELEASE_PATH)) {
+ if (line.startsWith("PRETTY_NAME=")) {
+ String value = line.substring("PRETTY_NAME=".length()).trim();
+ return stripSurroundingQuotes(value);
+ }
+ }
+ }
+ } catch (java.io.IOException | RuntimeException ignored) {
+ // Fall through to the fallback below.
+ }
+ return System.getProperty("os.name", UNKNOWN) + " " + System.getProperty("os.version", "");
+ }
+
+ private static String readOsKernel() {
+ return System.getProperty("os.version", UNKNOWN);
+ }
+
+ private static String stripSurroundingQuotes(String value) {
+ if (value.length() >= 2 && value.startsWith("\"") && value.endsWith("\"")) {
+ return value.substring(1, value.length() - 1);
+ }
+ return value;
+ }
+
+ // ---------------------------------------------------------------------
+ // JVM
+ // ---------------------------------------------------------------------
+
+ private static String readJvmVendor() {
+ return System.getProperty("java.vendor", UNKNOWN);
+ }
+
+ private static String readJvmVersion() {
+ return System.getProperty("java.version", UNKNOWN);
+ }
+
+ private static List readJvmFlags() {
+ try {
+ RuntimeMXBean runtimeMXBean = ManagementFactory.getRuntimeMXBean();
+ List arguments = runtimeMXBean.getInputArguments();
+ return (arguments != null) ? List.copyOf(arguments) : List.of();
+ } catch (RuntimeException ignored) {
+ return List.of();
+ }
+ }
+
+ // ---------------------------------------------------------------------
+ // Shared helpers
+ // ---------------------------------------------------------------------
+
+ private static List readAllLinesOrEmpty(Path path) {
+ try {
+ if (Files.isReadable(path)) {
+ return Files.readAllLines(path);
+ }
+ } catch (java.io.IOException | RuntimeException ignored) {
+ // Fall through to the empty list below.
+ }
+ return new ArrayList<>();
+ }
+
+ private static String valueAfterColon(String line) {
+ int colonIndex = line.indexOf(':');
+ if (colonIndex < 0 || colonIndex + 1 >= line.length()) {
+ return null;
+ }
+ return line.substring(colonIndex + 1);
+ }
+}
diff --git a/src/main/java/fr/inria/corese/benchmark/telemetry/TelemetryPoint.java b/src/main/java/fr/inria/corese/benchmark/telemetry/TelemetryPoint.java
new file mode 100644
index 0000000..b6ddae3
--- /dev/null
+++ b/src/main/java/fr/inria/corese/benchmark/telemetry/TelemetryPoint.java
@@ -0,0 +1,11 @@
+package fr.inria.corese.benchmark.telemetry;
+
+/**
+ * Immutable snapshot of ingestion telemetry captured at a given instant.
+ *
+ * @param elapsedMs milliseconds elapsed since the start of the sampled operation
+ * @param triplesLoaded number of triples loaded at the time of the sample
+ * @param heapUsedMb JVM heap used, in megabytes, rounded to one decimal place
+ */
+public record TelemetryPoint(long elapsedMs, long triplesLoaded, double heapUsedMb) {
+}
diff --git a/src/test/java/fr/inria/corese/benchmark/report/BenchmarkReportTest.java b/src/test/java/fr/inria/corese/benchmark/report/BenchmarkReportTest.java
new file mode 100644
index 0000000..e1acfb5
--- /dev/null
+++ b/src/test/java/fr/inria/corese/benchmark/report/BenchmarkReportTest.java
@@ -0,0 +1,156 @@
+package fr.inria.corese.benchmark.report;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import fr.inria.corese.benchmark.metrics.FileResult;
+import fr.inria.corese.benchmark.metrics.QueryResult;
+import fr.inria.corese.benchmark.telemetry.TelemetryPoint;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Collections;
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+class BenchmarkReportTest {
+
+ @Test
+ void reportSerializesJsonV2SchemaProperly(@TempDir Path tempDir) throws Exception {
+ Path jsonOut = tempDir.resolve("json/benchmark-report.json");
+ Files.createDirectories(jsonOut.getParent());
+
+ List timeSeries = List.of(
+ new TelemetryPoint(0L, 0L, 42.5),
+ new TelemetryPoint(100L, 0L, 65.2),
+ new TelemetryPoint(250L, 5000L, 88.0)
+ );
+
+ FileResult fileResult = new FileResult(
+ "corese.next",
+ "bowlogna",
+ "test.ttl",
+ 5000L,
+ 250L,
+ 20000.0,
+ 40.0,
+ 88.0,
+ 48.0,
+ 15.0,
+ new long[]{240L, 250L, 260L},
+ timeSeries
+ );
+
+ QueryResult queryResult = new QueryResult(
+ "corese.next",
+ "bowlogna",
+ "query_1",
+ "AGGREGATION",
+ 12L,
+ new long[]{12L, 13L, 11L},
+ 5L,
+ new long[]{5L, 6L, 5L},
+ 1.2,
+ 42L,
+ "hash123",
+ "PASS"
+ );
+
+ BenchmarkProtocol protocol = new BenchmarkProtocol(1, 3, 3, 45_000L, 600_000L, "median");
+
+ BenchmarkReport.write(
+ List.of(fileResult),
+ List.of(queryResult),
+ List.of("SELECT * WHERE { ?s ?p ?o }"),
+ jsonOut.toString(),
+ protocol
+ );
+
+ assertTrue(Files.exists(jsonOut), "JSON report should be generated");
+
+ ObjectMapper mapper = new ObjectMapper();
+ JsonNode root = mapper.readTree(jsonOut.toFile());
+
+ // Verify metadata block
+ assertTrue(root.has("metadata"), "Report must contain metadata");
+ JsonNode metadata = root.get("metadata");
+ assertEquals("2.0", metadata.get("version").asText());
+ assertTrue(metadata.has("generatedAt"));
+ assertTrue(metadata.has("queries"));
+ assertEquals(1, metadata.get("queries").size());
+ assertEquals("SELECT * WHERE { ?s ?p ?o }", metadata.get("queries").get(0).asText());
+
+ // Verify protocol
+ assertTrue(metadata.has("protocol"));
+ JsonNode protoNode = metadata.get("protocol");
+ assertEquals(1, protoNode.get("warmup").asInt());
+ assertEquals(3, protoNode.get("repetitions").asInt());
+ assertEquals(3, protoNode.get("queryRepetitions").asInt());
+ assertEquals(45_000L, protoNode.get("queryTimeoutMs").asLong());
+ assertEquals(600_000L, protoNode.get("loadTimeoutMs").asLong());
+ assertEquals("median", protoNode.get("metric").asText());
+
+ // Verify environment
+ assertTrue(metadata.has("environment"));
+ JsonNode envNode = metadata.get("environment");
+ assertTrue(envNode.has("cpuModel"));
+ assertTrue(envNode.has("cpuArchitecture"));
+ assertTrue(envNode.has("ramTotalGb"));
+ assertTrue(envNode.has("jvmVersion"));
+
+ // Verify loading runs (both v2 'loadingRuns' and legacy 'runs' are identical)
+ assertTrue(root.has("loadingRuns"), "Must export 'loadingRuns' for v2 dashboard");
+ assertTrue(root.has("runs"), "Must export 'runs' for backward compatibility");
+ assertEquals(root.get("loadingRuns"), root.get("runs"), "loadingRuns and runs must be identical");
+ assertEquals(1, root.get("loadingRuns").size());
+
+ // Verify timeSeries on loadingRun
+ JsonNode run = root.get("loadingRuns").get(0);
+ assertTrue(run.has("timeSeries"), "FileResult must contain timeSeries");
+ JsonNode ts = run.get("timeSeries");
+ assertEquals(3, ts.size());
+ assertEquals(0L, ts.get(0).get("elapsedMs").asLong());
+ assertEquals(42.5, ts.get(0).get("heapUsedMb").asDouble());
+ assertEquals(5000L, ts.get(2).get("triplesLoaded").asLong());
+
+ // Verify queryRuns
+ assertTrue(root.has("queryRuns"));
+ assertEquals(1, root.get("queryRuns").size());
+ assertEquals("query_1", root.get("queryRuns").get(0).get("queryId").asText());
+ assertEquals("PASS", root.get("queryRuns").get(0).get("status").asText());
+ }
+
+ @Test
+ void nullProtocolFallsBackToDefaultProtocol(@TempDir Path tempDir) throws Exception {
+ Path jsonOut = tempDir.resolve("default-report.json");
+ BenchmarkReport.write(Collections.emptyList(), Collections.emptyList(), Collections.emptyList(), jsonOut.toString(), null);
+
+ ObjectMapper mapper = new ObjectMapper();
+ JsonNode root = mapper.readTree(jsonOut.toFile());
+ JsonNode protoNode = root.get("metadata").get("protocol");
+ assertEquals(1, protoNode.get("warmup").asInt());
+ assertEquals(3, protoNode.get("repetitions").asInt());
+ assertEquals(3, protoNode.get("queryRepetitions").asInt());
+ assertEquals(30_000L, protoNode.get("queryTimeoutMs").asLong());
+ assertEquals(600_000L, protoNode.get("loadTimeoutMs").asLong());
+ }
+
+ @Test
+ void protocolFromOptionsAdaptsQuickMode() {
+ BenchmarkProtocol quickProto = BenchmarkProtocol.fromOptions(true, 15);
+ assertEquals(0, quickProto.warmup());
+ assertEquals(1, quickProto.repetitions());
+ assertEquals(1, quickProto.queryRepetitions());
+ assertEquals(15_000L, quickProto.queryTimeoutMs());
+ assertEquals(600_000L, quickProto.loadTimeoutMs());
+ assertEquals("median", quickProto.metric());
+
+ BenchmarkProtocol standardProto = BenchmarkProtocol.fromOptions(false, -1);
+ assertEquals(1, standardProto.warmup());
+ assertEquals(3, standardProto.repetitions());
+ assertEquals(3, standardProto.queryRepetitions());
+ assertEquals(30_000L, standardProto.queryTimeoutMs());
+ }
+}
diff --git a/src/test/java/fr/inria/corese/benchmark/telemetry/IngestionTelemetrySamplerTest.java b/src/test/java/fr/inria/corese/benchmark/telemetry/IngestionTelemetrySamplerTest.java
new file mode 100644
index 0000000..74e5aff
--- /dev/null
+++ b/src/test/java/fr/inria/corese/benchmark/telemetry/IngestionTelemetrySamplerTest.java
@@ -0,0 +1,81 @@
+package fr.inria.corese.benchmark.telemetry;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+import java.util.concurrent.atomic.AtomicLong;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+class IngestionTelemetrySamplerTest {
+
+ @Test
+ void samplerRecordsInitialAndFinalPoints() throws Exception {
+ AtomicLong triples = new AtomicLong(0);
+ try (IngestionTelemetrySampler sampler = new IngestionTelemetrySampler(20, triples::get)) {
+ sampler.start();
+
+ Thread.sleep(60);
+ triples.set(500);
+ Thread.sleep(60);
+
+ List points = sampler.stop(1000);
+
+ assertNotNull(points);
+ assertTrue(points.size() >= 2, "Expected at least start and end points, got: " + points.size());
+
+ TelemetryPoint first = points.getFirst();
+ assertEquals(0L, first.elapsedMs());
+ assertEquals(0L, first.triplesLoaded());
+ assertTrue(first.heapUsedMb() > 0, "Heap usage must be positive");
+
+ TelemetryPoint last = points.getLast();
+ assertTrue(last.elapsedMs() >= 100);
+ assertEquals(1000L, last.triplesLoaded());
+ }
+ }
+
+ @Test
+ void samplerRecordsBufferedEnginesAccuratelyWithoutFabricatingData() throws Exception {
+ // Simulates an engine that buffers everything until final commit (triples supplier returns 0 during load)
+ try (IngestionTelemetrySampler sampler = new IngestionTelemetrySampler(20, () -> 0L)) {
+ sampler.start();
+ Thread.sleep(80);
+ List points = sampler.stop(10_000);
+
+ assertTrue(points.size() >= 3);
+ TelemetryPoint first = points.getFirst();
+ assertEquals(0L, first.triplesLoaded());
+
+ // Intermediate points faithfully record 0 without faking an artificial linear curve
+ for (int i = 1; i < points.size() - 1; i++) {
+ TelemetryPoint p = points.get(i);
+ assertEquals(0L, p.triplesLoaded(), "Intermediate samples must not fabricate unmeasured counts");
+ }
+
+ TelemetryPoint last = points.getLast();
+ assertEquals(10_000L, last.triplesLoaded());
+ }
+ }
+
+ @Test
+ void rejectsInvalidSamplingPeriod() {
+ assertThrows(IllegalArgumentException.class, () -> new IngestionTelemetrySampler(0, () -> 0L));
+ assertThrows(IllegalArgumentException.class, () -> new IngestionTelemetrySampler(-10, () -> 0L));
+ }
+
+ @Test
+ void cannotStartTwice() {
+ try (IngestionTelemetrySampler sampler = new IngestionTelemetrySampler(50, () -> 0L)) {
+ sampler.start();
+ assertThrows(IllegalStateException.class, sampler::start);
+ }
+ }
+
+ @Test
+ void cannotStopBeforeStart() {
+ try (IngestionTelemetrySampler sampler = new IngestionTelemetrySampler(50, () -> 0L)) {
+ assertThrows(IllegalStateException.class, () -> sampler.stop(100));
+ }
+ }
+}
diff --git a/src/test/java/fr/inria/corese/benchmark/telemetry/SystemEnvironmentCollectorTest.java b/src/test/java/fr/inria/corese/benchmark/telemetry/SystemEnvironmentCollectorTest.java
new file mode 100644
index 0000000..c587a8b
--- /dev/null
+++ b/src/test/java/fr/inria/corese/benchmark/telemetry/SystemEnvironmentCollectorTest.java
@@ -0,0 +1,64 @@
+package fr.inria.corese.benchmark.telemetry;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+class SystemEnvironmentCollectorTest {
+
+ @Test
+ void collectsAllRequiredEnvironmentKeys() {
+ Map env = SystemEnvironmentCollector.collect();
+
+ assertNotNull(env);
+
+ // Required keys from schema
+ List expectedKeys = List.of(
+ "cpuModel",
+ "cpuArchitecture",
+ "cpuSockets",
+ "cpuCores",
+ "cpuThreads",
+ "cpuFrequencyMhz",
+ "cpuGovernor",
+ "ramTotalGb",
+ "swapTotalGb",
+ "swappiness",
+ "osDistribution",
+ "osKernel",
+ "jvmVendor",
+ "jvmVersion",
+ "jvmFlags"
+ );
+
+ for (String key : expectedKeys) {
+ assertTrue(env.containsKey(key), "Environment map missing expected key: " + key);
+ assertNotNull(env.get(key), "Value for key " + key + " must not be null");
+ }
+
+ // Validate types
+ assertInstanceOf(String.class, env.get("cpuModel"));
+ assertInstanceOf(String.class, env.get("cpuArchitecture"));
+ assertInstanceOf(Integer.class, env.get("cpuSockets"));
+ assertInstanceOf(Integer.class, env.get("cpuCores"));
+ assertInstanceOf(Integer.class, env.get("cpuThreads"));
+ assertInstanceOf(Double.class, env.get("cpuFrequencyMhz"));
+ assertInstanceOf(String.class, env.get("cpuGovernor"));
+ assertInstanceOf(Double.class, env.get("ramTotalGb"));
+ assertInstanceOf(Double.class, env.get("swapTotalGb"));
+ assertInstanceOf(Integer.class, env.get("swappiness"));
+ assertInstanceOf(String.class, env.get("osDistribution"));
+ assertInstanceOf(String.class, env.get("osKernel"));
+ assertInstanceOf(String.class, env.get("jvmVendor"));
+ assertInstanceOf(String.class, env.get("jvmVersion"));
+ assertInstanceOf(List.class, env.get("jvmFlags"));
+
+ // Basic assertions on host values
+ assertTrue((int) env.get("cpuCores") > 0);
+ assertTrue((int) env.get("cpuThreads") > 0);
+ assertTrue((double) env.get("ramTotalGb") > 0);
+ }
+}