merge: Integrate v2.1 progress into public repository

This pull request integrates the changes planned for the v2.1 release of
OpenDC into the public Github repository in order to sync the progress
of both repositories.

2 files changed, 274 insertions, 45 deletions

diff --git a/opendc-compute/opendc-compute-simulator/src/test/kotlin/org/opendc/compute/simulator/SimHostTest.kt b/opendc-compute/opendc-compute-simulator/src/test/kotlin/org/opendc/compute/simulator/SimHostTest.kt
index 5594fd59..a0ff9228 100644
--- a/opendc-compute/opendc-compute-simulator/src/test/kotlin/org/opendc/compute/simulator/SimHostTest.kt
+++ b/opendc-compute/opendc-compute-simulator/src/test/kotlin/org/opendc/compute/simulator/SimHostTest.kt
@@ -23,47 +23,48 @@
 package org.opendc.compute.simulator
 
 import io.opentelemetry.api.metrics.MeterProvider
-import io.opentelemetry.sdk.common.CompletableResultCode
 import io.opentelemetry.sdk.metrics.SdkMeterProvider
-import io.opentelemetry.sdk.metrics.data.MetricData
-import io.opentelemetry.sdk.metrics.export.MetricExporter
 import io.opentelemetry.sdk.metrics.export.MetricProducer
+import io.opentelemetry.sdk.resources.Resource
 import kotlinx.coroutines.*
 import org.junit.jupiter.api.Assertions.assertEquals
 import org.junit.jupiter.api.BeforeEach
 import org.junit.jupiter.api.Test
 import org.junit.jupiter.api.assertAll
-import org.opendc.compute.api.Flavor
-import org.opendc.compute.api.Image
-import org.opendc.compute.api.Server
-import org.opendc.compute.api.ServerState
-import org.opendc.compute.api.ServerWatcher
+import org.opendc.compute.api.*
 import org.opendc.compute.service.driver.Host
 import org.opendc.compute.service.driver.HostListener
-import org.opendc.simulator.compute.SimFairShareHypervisorProvider
-import org.opendc.simulator.compute.SimMachineModel
+import org.opendc.simulator.compute.kernel.SimFairShareHypervisorProvider
+import org.opendc.simulator.compute.model.MachineModel
 import org.opendc.simulator.compute.model.MemoryUnit
 import org.opendc.simulator.compute.model.ProcessingNode
 import org.opendc.simulator.compute.model.ProcessingUnit
+import org.opendc.simulator.compute.workload.SimTrace
+import org.opendc.simulator.compute.workload.SimTraceFragment
 import org.opendc.simulator.compute.workload.SimTraceWorkload
 import org.opendc.simulator.core.runBlockingSimulation
+import org.opendc.simulator.flow.FlowEngine
+import org.opendc.telemetry.compute.ComputeMetricExporter
+import org.opendc.telemetry.compute.HOST_ID
+import org.opendc.telemetry.compute.table.HostData
+import org.opendc.telemetry.compute.table.ServerData
 import org.opendc.telemetry.sdk.metrics.export.CoroutineMetricReader
 import org.opendc.telemetry.sdk.toOtelClock
-import java.util.UUID
+import java.time.Duration
+import java.util.*
 import kotlin.coroutines.resume
 
 /**
  * Basic test-suite for the hypervisor.
  */
-@OptIn(ExperimentalCoroutinesApi::class)
 internal class SimHostTest {
-    private lateinit var machineModel: SimMachineModel
+    private lateinit var machineModel: MachineModel
 
     @BeforeEach
     fun setUp() {
         val cpuNode = ProcessingNode("Intel", "Xeon", "amd64", 2)
 
-        machineModel = SimMachineModel(
+        machineModel = MachineModel(
             cpus = List(cpuNode.coreCount) { ProcessingUnit(cpuNode, it, 3200.0) },
             memory = List(4) { MemoryUnit("Crucial", "MTA18ASF4G72AZ-3G2B1", 3200.0, 32_000) }
         )
@@ -74,16 +75,31 @@ internal class SimHostTest {
      */
     @Test
     fun testOvercommitted() = runBlockingSimulation {
-        var requestedWork = 0L
-        var grantedWork = 0L
-        var overcommittedWork = 0L
+        var idleTime = 0L
+        var activeTime = 0L
+        var stealTime = 0L
 
+        val hostId = UUID.randomUUID()
+        val hostResource = Resource.builder()
+            .put(HOST_ID, hostId.toString())
+            .build()
         val meterProvider: MeterProvider = SdkMeterProvider
             .builder()
+            .setResource(hostResource)
             .setClock(clock.toOtelClock())
             .build()
 
-        val virtDriver = SimHost(UUID.randomUUID(), "test", machineModel, emptyMap(), coroutineContext, clock, meterProvider.get("opendc-compute-simulator"), SimFairShareHypervisorProvider())
+        val engine = FlowEngine(coroutineContext, clock)
+        val virtDriver = SimHost(
+            uid = hostId,
+            name = "test",
+            model = machineModel,
+            meta = emptyMap(),
+            coroutineContext,
+            engine,
+            meterProvider,
+            SimFairShareHypervisorProvider()
+        )
         val duration = 5 * 60L
         val vmImageA = MockImage(
             UUID.randomUUID(),
@@ -91,12 +107,13 @@ internal class SimHostTest {
             emptyMap(),
             mapOf(
                 "workload" to SimTraceWorkload(
-                    sequenceOf(
-                        SimTraceWorkload.Fragment(duration * 1000, 2 * 28.0, 2),
-                        SimTraceWorkload.Fragment(duration * 1000, 2 * 3500.0, 2),
-                        SimTraceWorkload.Fragment(duration * 1000, 0.0, 2),
-                        SimTraceWorkload.Fragment(duration * 1000, 2 * 183.0, 2)
+                    SimTrace.ofFragments(
+                        SimTraceFragment(0, duration * 1000, 2 * 28.0, 2),
+                        SimTraceFragment(duration * 1000, duration * 1000, 2 * 3500.0, 2),
+                        SimTraceFragment(duration * 2000, duration * 1000, 0.0, 2),
+                        SimTraceFragment(duration * 3000, duration * 1000, 2 * 183.0, 2)
                     ),
+                    offset = 1
                 )
             )
         )
@@ -106,12 +123,13 @@ internal class SimHostTest {
             emptyMap(),
             mapOf(
                 "workload" to SimTraceWorkload(
-                    sequenceOf(
-                        SimTraceWorkload.Fragment(duration * 1000, 2 * 28.0, 2),
-                        SimTraceWorkload.Fragment(duration * 1000, 2 * 3100.0, 2),
-                        SimTraceWorkload.Fragment(duration * 1000, 0.0, 2),
-                        SimTraceWorkload.Fragment(duration * 1000, 2 * 73.0, 2)
-                    )
+                    SimTrace.ofFragments(
+                        SimTraceFragment(0, duration * 1000, 2 * 28.0, 2),
+                        SimTraceFragment(duration * 1000, duration * 1000, 2 * 3100.0, 2),
+                        SimTraceFragment(duration * 2000, duration * 1000, 0.0, 2),
+                        SimTraceFragment(duration * 3000, duration * 1000, 2 * 73.0, 2)
+                    ),
+                    offset = 1
                 )
             )
         )
@@ -121,20 +139,14 @@ internal class SimHostTest {
         // Setup metric reader
         val reader = CoroutineMetricReader(
             this, listOf(meterProvider as MetricProducer),
-            object : MetricExporter {
-                override fun export(metrics: Collection<MetricData>): CompletableResultCode {
-                    val metricsByName = metrics.associateBy { it.name }
-                    requestedWork += metricsByName.getValue("cpu.work.total").doubleSummaryData.points.first().sum.toLong()
-                    grantedWork += metricsByName.getValue("cpu.work.granted").doubleSummaryData.points.first().sum.toLong()
-                    overcommittedWork += metricsByName.getValue("cpu.work.overcommit").doubleSummaryData.points.first().sum.toLong()
-                    return CompletableResultCode.ofSuccess()
+            object : ComputeMetricExporter() {
+                override fun record(data: HostData) {
+                    activeTime += data.cpuActiveTime
+                    idleTime += data.cpuIdleTime
+                    stealTime += data.cpuStealTime
                 }
-
-                override fun flush(): CompletableResultCode = CompletableResultCode.ofSuccess()
-
-                override fun shutdown(): CompletableResultCode = CompletableResultCode.ofSuccess()
             },
-            exportInterval = duration * 1000
+            exportInterval = Duration.ofSeconds(duration)
         )
 
         coroutineScope {
@@ -155,18 +167,124 @@ internal class SimHostTest {
         }
 
         // Ensure last cycle is collected
-        delay(1000 * duration)
+        delay(1000L * duration)
         virtDriver.close()
         reader.close()
 
         assertAll(
-            { assertEquals(4197600, requestedWork, "Requested work does not match") },
-            { assertEquals(2157600, grantedWork, "Granted work does not match") },
-            { assertEquals(2040000, overcommittedWork, "Overcommitted work does not match") },
+            { assertEquals(658, activeTime, "Active time does not match") },
+            { assertEquals(1741, idleTime, "Idle time does not match") },
+            { assertEquals(637, stealTime, "Steal time does not match") },
             { assertEquals(1500001, clock.millis()) }
         )
     }
 
+    /**
+     * Test failure of the host.
+     */
+    @Test
+    fun testFailure() = runBlockingSimulation {
+        var activeTime = 0L
+        var idleTime = 0L
+        var uptime = 0L
+        var downtime = 0L
+        var guestUptime = 0L
+        var guestDowntime = 0L
+
+        val hostId = UUID.randomUUID()
+        val hostResource = Resource.builder()
+            .put(HOST_ID, hostId.toString())
+            .build()
+        val meterProvider: MeterProvider = SdkMeterProvider
+            .builder()
+            .setResource(hostResource)
+            .setClock(clock.toOtelClock())
+            .build()
+
+        val engine = FlowEngine(coroutineContext, clock)
+        val host = SimHost(
+            uid = hostId,
+            name = "test",
+            model = machineModel,
+            meta = emptyMap(),
+            coroutineContext,
+            engine,
+            meterProvider,
+            SimFairShareHypervisorProvider()
+        )
+        val duration = 5 * 60L
+        val image = MockImage(
+            UUID.randomUUID(),
+            "<unnamed>",
+            emptyMap(),
+            mapOf(
+                "workload" to SimTraceWorkload(
+                    SimTrace.ofFragments(
+                        SimTraceFragment(0, duration * 1000, 2 * 28.0, 2),
+                        SimTraceFragment(duration * 1000L, duration * 1000, 2 * 3500.0, 2),
+                        SimTraceFragment(duration * 2000L, duration * 1000, 0.0, 2),
+                        SimTraceFragment(duration * 3000L, duration * 1000, 2 * 183.0, 2)
+                    ),
+                    offset = 1
+                )
+            )
+        )
+        val flavor = MockFlavor(2, 0)
+        val server = MockServer(UUID.randomUUID(), "a", flavor, image)
+
+        // Setup metric reader
+        val reader = CoroutineMetricReader(
+            this, listOf(meterProvider as MetricProducer),
+            object : ComputeMetricExporter() {
+                override fun record(data: HostData) {
+                    activeTime += data.cpuActiveTime
+                    idleTime += data.cpuIdleTime
+                    uptime += data.uptime
+                    downtime += data.downtime
+                }
+
+                override fun record(data: ServerData) {
+                    guestUptime += data.uptime
+                    guestDowntime += data.downtime
+                }
+            },
+            exportInterval = Duration.ofSeconds(duration)
+        )
+
+        coroutineScope {
+            host.spawn(server)
+            delay(5000L)
+            host.fail()
+            delay(duration * 1000)
+            host.recover()
+
+            suspendCancellableCoroutine<Unit> { cont ->
+                host.addListener(object : HostListener {
+                    override fun onStateChanged(host: Host, server: Server, newState: ServerState) {
+                        if (newState == ServerState.TERMINATED) {
+                            cont.resume(Unit)
+                        }
+                    }
+                })
+            }
+        }
+
+        host.close()
+        // Ensure last cycle is collected
+        delay(1000L * duration)
+
+        reader.close()
+
+        assertAll(
+            { assertEquals(1175, idleTime, "Idle time does not match") },
+            { assertEquals(624, activeTime, "Active time does not match") },
+            { assertEquals(900001, uptime, "Uptime does not match") },
+            { assertEquals(300000, downtime, "Downtime does not match") },
+            { assertEquals(900000, guestUptime, "Guest uptime does not match") },
+            { assertEquals(300000, guestDowntime, "Guest downtime does not match") },
+        )
+    }
+
     private class MockFlavor(
         override val cpuCount: Int,
         override val memorySize: Long
diff --git a/opendc-compute/opendc-compute-simulator/src/test/kotlin/org/opendc/compute/simulator/failure/HostFaultInjectorTest.kt b/opendc-compute/opendc-compute-simulator/src/test/kotlin/org/opendc/compute/simulator/failure/HostFaultInjectorTest.kt
new file mode 100644
index 00000000..f240a25f
--- /dev/null
+++ b/opendc-compute/opendc-compute-simulator/src/test/kotlin/org/opendc/compute/simulator/failure/HostFaultInjectorTest.kt
@@ -0,0 +1,111 @@
+/*
+ * Copyright (c) 2021 AtLarge Research
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package org.opendc.compute.simulator.failure
+
+import io.mockk.coVerify
+import io.mockk.mockk
+import kotlinx.coroutines.delay
+import org.apache.commons.math3.distribution.LogNormalDistribution
+import org.apache.commons.math3.random.Well19937c
+import org.junit.jupiter.api.Test
+import org.opendc.compute.simulator.SimHost
+import org.opendc.simulator.core.runBlockingSimulation
+import java.time.Clock
+import java.time.Duration
+import kotlin.coroutines.CoroutineContext
+import kotlin.math.ln
+
+/**
+ * Test suite for [HostFaultInjector] class.
+ */
+internal class HostFaultInjectorTest {
+    /**
+     * Simple test case to test that nothing happens when the injector is not started.
+     */
+    @Test
+    fun testInjectorNotStarted() = runBlockingSimulation {
+        val host = mockk<SimHost>(relaxUnitFun = true)
+
+        val injector = createSimpleInjector(coroutineContext, clock, setOf(host))
+
+        coVerify(exactly = 0) { host.fail() }
+        coVerify(exactly = 0) { host.recover() }
+
+        injector.close()
+    }
+
+    /**
+     * Simple test case to test a start stop fault where the machine is stopped and started after some time.
+     */
+    @Test
+    fun testInjectorStopsMachine() = runBlockingSimulation {
+        val host = mockk<SimHost>(relaxUnitFun = true)
+
+        val injector = createSimpleInjector(coroutineContext, clock, setOf(host))
+
+        injector.start()
+
+        delay(Duration.ofDays(55).toMillis())
+
+        injector.close()
+
+        coVerify(exactly = 1) { host.fail() }
+        coVerify(exactly = 1) { host.recover() }
+    }
+
+    /**
+     * Simple test case to test a start stop fault where multiple machines are stopped.
+     */
+    @Test
+    fun testInjectorStopsMultipleMachines() = runBlockingSimulation {
+        val hosts = listOf<SimHost>(
+            mockk(relaxUnitFun = true),
+            mockk(relaxUnitFun = true)
+        )
+
+        val injector = createSimpleInjector(coroutineContext, clock, hosts.toSet())
+
+        injector.start()
+
+        delay(Duration.ofDays(55).toMillis())
+
+        injector.close()
+
+        coVerify(exactly = 1) { hosts[0].fail() }
+        coVerify(exactly = 1) { hosts[1].fail() }
+        coVerify(exactly = 1) { hosts[0].recover() }
+        coVerify(exactly = 1) { hosts[1].recover() }
+    }
+
+    /**
+     * Create a simple start stop fault injector.
+     */
+    private fun createSimpleInjector(context: CoroutineContext, clock: Clock, hosts: Set<SimHost>): HostFaultInjector {
+        val rng = Well19937c(0)
+        val iat = LogNormalDistribution(rng, ln(24 * 7.0), 1.03)
+        val selector = StochasticVictimSelector(LogNormalDistribution(rng, 1.88, 1.25))
+        val fault = StartStopHostFault(LogNormalDistribution(rng, 8.89, 2.71))
+
+        return HostFaultInjector(context, clock, hosts, iat, selector, fault)
+    }
+}