merge: Add tests for interference and failures in Capelin

This pull request updates the Capelin experiments to test for interference and failure scenarios. 
This allows us to track regressions in these subsystems more easily.

* Clean up Bitbrains trace reader to enable re-use 
* Keep trace order after sampling
* Update Bitbrains trace tests
* Add support for reporting interfered work
* Add support for SimHost failure
* Add tests for interference and failures

4 files changed, 132 insertions, 6 deletions

diff --git a/opendc-experiments/opendc-experiments-capelin/src/test/kotlin/org/opendc/experiments/capelin/CapelinIntegrationTest.kt b/opendc-experiments/opendc-experiments-capelin/src/test/kotlin/org/opendc/experiments/capelin/CapelinIntegrationTest.kt
index 393fb88d..a3300b71 100644
--- a/opendc-experiments/opendc-experiments-capelin/src/test/kotlin/org/opendc/experiments/capelin/CapelinIntegrationTest.kt
+++ b/opendc-experiments/opendc-experiments-capelin/src/test/kotlin/org/opendc/experiments/capelin/CapelinIntegrationTest.kt
@@ -39,12 +39,15 @@ import org.opendc.experiments.capelin.env.ClusterEnvironmentReader
 import org.opendc.experiments.capelin.model.Workload
 import org.opendc.experiments.capelin.monitor.ExperimentMonitor
 import org.opendc.experiments.capelin.trace.ParquetTraceReader
+import org.opendc.experiments.capelin.trace.PerformanceInterferenceReader
 import org.opendc.experiments.capelin.trace.RawParquetTraceReader
 import org.opendc.format.environment.EnvironmentReader
 import org.opendc.format.trace.TraceReader
+import org.opendc.simulator.compute.kernel.interference.VmInterferenceModel
 import org.opendc.simulator.compute.workload.SimWorkload
 import org.opendc.simulator.core.runBlockingSimulation
 import java.io.File
+import java.util.*
 
 /**
  * An integration test suite for the SC20 experiments.
@@ -63,6 +66,9 @@ class CapelinIntegrationTest {
         monitor = TestExperimentReporter()
     }
 
+    /**
+     * Test a large simulation setup.
+     */
     @Test
     fun testLarge() = runBlockingSimulation {
         val failures = false
@@ -114,13 +120,16 @@ class CapelinIntegrationTest {
             { assertEquals(0, monitorResults.runningVms, "All VMs should finish after a run") },
             { assertEquals(0, monitorResults.unscheduledVms, "No VM should not be unscheduled") },
             { assertEquals(0, monitorResults.queuedVms, "No VM should not be in the queue") },
-            { assertEquals(155252275351, monitor.totalRequestedBurst) { "Incorrect requested burst" } },
-            { assertEquals(155086837645, monitor.totalGrantedBurst) { "Incorrect granted burst" } },
-            { assertEquals(725049, monitor.totalOvercommissionedBurst) { "Incorrect overcommitted burst" } },
+            { assertEquals(219751355711, monitor.totalRequestedBurst) { "Incorrect requested burst" } },
+            { assertEquals(206351165081, monitor.totalGrantedBurst) { "Incorrect granted burst" } },
+            { assertEquals(1148906334, monitor.totalOvercommissionedBurst) { "Incorrect overcommitted burst" } },
             { assertEquals(0, monitor.totalInterferedBurst) { "Incorrect interfered burst" } }
         )
     }
 
+    /**
+     * Test a small simulation setup.
+     */
     @Test
     fun testSmall() = runBlockingSimulation {
         val seed = 1
@@ -151,9 +160,105 @@ class CapelinIntegrationTest {
 
         // Note that these values have been verified beforehand
         assertAll(
-            { assertEquals(29454904468, monitor.totalRequestedBurst) { "Total requested work incorrect" } },
-            { assertEquals(29355293349, monitor.totalGrantedBurst) { "Total granted work incorrect" } },
-            { assertEquals(0, monitor.totalOvercommissionedBurst) { "Total overcommitted work incorrect" } },
+            { assertEquals(37954956986, monitor.totalRequestedBurst) { "Total requested work incorrect" } },
+            { assertEquals(34840774250, monitor.totalGrantedBurst) { "Total granted work incorrect" } },
+            { assertEquals(971076806, monitor.totalOvercommissionedBurst) { "Total overcommitted work incorrect" } },
+            { assertEquals(0, monitor.totalInterferedBurst) { "Total interfered work incorrect" } }
+        )
+    }
+
+    /**
+     * Test a small simulation setup with interference.
+     */
+    @Test
+    fun testInterference() = runBlockingSimulation {
+        val seed = 1
+        val chan = Channel<Unit>(Channel.CONFLATED)
+        val allocationPolicy = FilterScheduler(
+            filters = listOf(ComputeFilter(), VCpuFilter(16.0), RamFilter(1.0)),
+            weighers = listOf(CoreRamWeigher(multiplier = 1.0))
+        )
+        val traceReader = createTestTraceReader(0.25, seed)
+        val environmentReader = createTestEnvironmentReader("single")
+
+        val perfInterferenceInput = checkNotNull(CapelinIntegrationTest::class.java.getResourceAsStream("/bitbrains-perf-interference.json"))
+        val performanceInterferenceModel =
+            PerformanceInterferenceReader(perfInterferenceInput).use { VmInterferenceModel(it.read(), Random(seed.toLong())) }
+
+        val meterProvider = createMeterProvider(clock)
+
+        withComputeService(clock, meterProvider, environmentReader, allocationPolicy, performanceInterferenceModel) { scheduler ->
+            withMonitor(monitor, clock, meterProvider as MetricProducer, scheduler) {
+                processTrace(
+                    clock,
+                    traceReader,
+                    scheduler,
+                    chan,
+                    monitor
+                )
+            }
+        }
+
+        val metrics = collectMetrics(meterProvider as MetricProducer)
+        println("Finish SUBMIT=${metrics.submittedVms} FAIL=${metrics.unscheduledVms} QUEUE=${metrics.queuedVms} RUNNING=${metrics.runningVms}")
+
+        // Note that these values have been verified beforehand
+        assertAll(
+            { assertEquals(37954956986, monitor.totalRequestedBurst) { "Total requested work incorrect" } },
+            { assertEquals(34840774250, monitor.totalGrantedBurst) { "Total granted work incorrect" } },
+            { assertEquals(971076806, monitor.totalOvercommissionedBurst) { "Total overcommitted work incorrect" } },
+            { assertEquals(13885404, monitor.totalInterferedBurst) { "Total interfered work incorrect" } }
+        )
+    }
+
+    /**
+     * Test a small simulation setup with failures.
+     */
+    @Test
+    fun testFailures() = runBlockingSimulation {
+        val seed = 1
+        val chan = Channel<Unit>(Channel.CONFLATED)
+        val allocationPolicy = FilterScheduler(
+            filters = listOf(ComputeFilter(), VCpuFilter(16.0), RamFilter(1.0)),
+            weighers = listOf(CoreRamWeigher(multiplier = 1.0))
+        )
+        val traceReader = createTestTraceReader(0.25, seed)
+        val environmentReader = createTestEnvironmentReader("single")
+
+        val meterProvider = createMeterProvider(clock)
+
+        withComputeService(clock, meterProvider, environmentReader, allocationPolicy) { scheduler ->
+            val failureDomain =
+                createFailureDomain(
+                    this,
+                    clock,
+                    seed,
+                    24.0 * 7,
+                    scheduler,
+                    chan
+                )
+
+            withMonitor(monitor, clock, meterProvider as MetricProducer, scheduler) {
+                processTrace(
+                    clock,
+                    traceReader,
+                    scheduler,
+                    chan,
+                    monitor
+                )
+            }
+
+            failureDomain.cancel()
+        }
+
+        val metrics = collectMetrics(meterProvider as MetricProducer)
+        println("Finish SUBMIT=${metrics.submittedVms} FAIL=${metrics.unscheduledVms} QUEUE=${metrics.queuedVms} RUNNING=${metrics.runningVms}")
+
+        // Note that these values have been verified beforehand
+        assertAll(
+            { assertEquals(25336984869, monitor.totalRequestedBurst) { "Total requested work incorrect" } },
+            { assertEquals(23668547517, monitor.totalGrantedBurst) { "Total granted work incorrect" } },
+            { assertEquals(368151656, monitor.totalOvercommissionedBurst) { "Total overcommitted work incorrect" } },
             { assertEquals(0, monitor.totalInterferedBurst) { "Total interfered work incorrect" } }
         )
     }
diff --git a/opendc-experiments/opendc-experiments-capelin/src/test/resources/bitbrains-perf-interference.json b/opendc-experiments/opendc-experiments-capelin/src/test/resources/bitbrains-perf-interference.json
new file mode 100644
index 00000000..51fc6366
--- /dev/null
+++ b/opendc-experiments/opendc-experiments-capelin/src/test/resources/bitbrains-perf-interference.json
@@ -0,0 +1,21 @@
+[
+    {
+        "vms": [
+            "141",
+            "379",
+            "851",
+            "116"
+        ],
+        "minServerLoad": 0.0,
+        "performanceScore": 0.8830158730158756
+    },
+    {
+        "vms": [
+            "205",
+            "116",
+            "463"
+        ],
+        "minServerLoad": 0.0,
+        "performanceScore": 0.7133055555552751
+    }
+]
diff --git a/opendc-experiments/opendc-experiments-capelin/src/test/resources/trace/meta.parquet b/opendc-experiments/opendc-experiments-capelin/src/test/resources/trace/meta.parquet
index ce7a812c..ee76d38f 100644
--- a/opendc-experiments/opendc-experiments-capelin/src/test/resources/trace/meta.parquet
+++ b/opendc-experiments/opendc-experiments-capelin/src/test/resources/trace/meta.parquet
diff --git a/opendc-experiments/opendc-experiments-capelin/src/test/resources/trace/trace.parquet b/opendc-experiments/opendc-experiments-capelin/src/test/resources/trace/trace.parquet
index 1d7ce882..9b1cde13 100644
--- a/opendc-experiments/opendc-experiments-capelin/src/test/resources/trace/trace.parquet
+++ b/opendc-experiments/opendc-experiments-capelin/src/test/resources/trace/trace.parquet