Merge branch 'feat/database-metrics' into '2.x'

Implement experiment runner in Kotlin

Closes #63

See merge request opendc/opendc-simulator!66

5 files changed, 163 insertions, 33 deletions

diff --git a/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/core/workload/PerformanceInterferenceModel.kt b/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/core/workload/PerformanceInterferenceModel.kt
index 45024a49..f458877b 100644
--- a/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/core/workload/PerformanceInterferenceModel.kt
+++ b/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/core/workload/PerformanceInterferenceModel.kt
@@ -40,30 +40,12 @@ const val IMAGE_PERF_INTERFERENCE_MODEL = "image:performance-interference"
  * @param items The [PerformanceInterferenceModelItem]s that make up this model.
  */
 class PerformanceInterferenceModel(
-    items: Set<PerformanceInterferenceModelItem>,
+    val items: SortedSet<PerformanceInterferenceModelItem>,
     val random: Random = Random(0)
 ) {
     private var intersectingItems: List<PerformanceInterferenceModelItem> = emptyList()
-    private val comparator = Comparator<PerformanceInterferenceModelItem> { lhs, rhs ->
-        var cmp = lhs.performanceScore.compareTo(rhs.performanceScore)
-        if (cmp != 0) {
-            return@Comparator cmp
-        }
-
-        cmp = lhs.minServerLoad.compareTo(rhs.minServerLoad)
-        if (cmp != 0) {
-            return@Comparator cmp
-        }
-
-        lhs.hashCode().compareTo(rhs.hashCode())
-    }
-    val items = TreeSet(comparator)
     private val colocatedWorkloads = TreeSet<String>()
 
-    init {
-        this.items.addAll(items)
-    }
-
     fun vmStarted(server: Server) {
         colocatedWorkloads.add(server.image.name)
         intersectingItems = items.filter { item -> doesMatch(item) }
@@ -113,7 +95,7 @@ data class PerformanceInterferenceModelItem(
     val workloadNames: SortedSet<String>,
     val minServerLoad: Double,
     val performanceScore: Double
-) {
+) : Comparable<PerformanceInterferenceModelItem> {
     override fun equals(other: Any?): Boolean {
         if (this === other) return true
         if (javaClass != other?.javaClass) return false
@@ -126,4 +108,18 @@ data class PerformanceInterferenceModelItem(
     }
 
     override fun hashCode(): Int = workloadNames.hashCode()
+
+    override fun compareTo(other: PerformanceInterferenceModelItem): Int {
+        var cmp = performanceScore.compareTo(other.performanceScore)
+        if (cmp != 0) {
+            return cmp
+        }
+
+        cmp = minServerLoad.compareTo(other.minServerLoad)
+        if (cmp != 0) {
+            return cmp
+        }
+
+        return hashCode().compareTo(other.hashCode())
+    }
 }
diff --git a/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/driver/SimpleVirtDriver.kt b/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/driver/SimpleVirtDriver.kt
index 53fa463b..ce814dd8 100644
--- a/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/driver/SimpleVirtDriver.kt
+++ b/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/driver/SimpleVirtDriver.kt
@@ -339,7 +339,7 @@ class SimpleVirtDriver(
                     min(totalRequestedSubBurst, totalGrantedBurst), // We can run more than requested due to timing
                     totalOvercommissionedBurst,
                     totalInterferedBurst, // Might be smaller than zero due to FP rounding errors,
-                    totalAllocatedUsage,
+                    min(totalAllocatedUsage, totalRequestedUsage), // The allocated usage might be slightly higher due to FP rounding
                     totalRequestedUsage,
                     vmCount, // Some VMs might already have finished, so keep initial VM count
                     server
diff --git a/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/service/SimpleVirtProvisioningService.kt b/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/service/SimpleVirtProvisioningService.kt
index 3603ae69..c3d9c745 100644
--- a/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/service/SimpleVirtProvisioningService.kt
+++ b/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/service/SimpleVirtProvisioningService.kt
@@ -1,6 +1,7 @@
 package com.atlarge.opendc.compute.virt.service
 
 import com.atlarge.odcsim.SimulationContext
+import com.atlarge.odcsim.flow.EventFlow
 import com.atlarge.odcsim.simulationContext
 import com.atlarge.opendc.compute.core.Flavor
 import com.atlarge.opendc.compute.core.Server
@@ -19,6 +20,7 @@ import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.launchIn
 import kotlinx.coroutines.flow.onEach
 import kotlinx.coroutines.launch
@@ -57,11 +59,11 @@ class SimpleVirtProvisioningService(
      */
     private val activeImages: MutableSet<ImageView> = mutableSetOf()
 
-    public var submittedVms = 0L
-    public var queuedVms = 0L
-    public var runningVms = 0L
-    public var finishedVms = 0L
-    public var unscheduledVms = 0L
+    public var submittedVms = 0
+    public var queuedVms = 0
+    public var runningVms = 0
+    public var finishedVms = 0
+    public var unscheduledVms = 0
 
     private var maxCores = 0
     private var maxMemory = 0L
@@ -71,6 +73,13 @@ class SimpleVirtProvisioningService(
      */
     private val allocationLogic = allocationPolicy()
 
+    /**
+     * The [EventFlow] to emit the events.
+     */
+    internal val eventFlow = EventFlow<VirtProvisioningEvent>()
+
+    override val events: Flow<VirtProvisioningEvent> = eventFlow
+
     init {
         launch {
             val provisionedNodes = provisioningService.nodes()
@@ -96,8 +105,17 @@ class SimpleVirtProvisioningService(
         image: Image,
         flavor: Flavor
     ): Server = withContext(coroutineContext) {
-        submittedVms++
-        queuedVms++
+        eventFlow.emit(VirtProvisioningEvent.MetricsAvailable(
+            this@SimpleVirtProvisioningService,
+            hypervisors.size,
+            availableHypervisors.size,
+            ++submittedVms,
+            runningVms,
+            finishedVms,
+            ++queuedVms,
+            unscheduledVms
+        ))
+
         suspendCancellableCoroutine<Server> { cont ->
             val vmInstance = ImageView(name, image, flavor, cont)
             incomingImages += vmInstance
@@ -141,7 +159,17 @@ class SimpleVirtProvisioningService(
 
             if (selectedHv == null) {
                 if (requiredMemory > maxMemory || imageInstance.flavor.cpuCount > maxCores) {
-                    unscheduledVms++
+                    eventFlow.emit(VirtProvisioningEvent.MetricsAvailable(
+                        this@SimpleVirtProvisioningService,
+                        hypervisors.size,
+                        availableHypervisors.size,
+                        submittedVms,
+                        runningVms,
+                        finishedVms,
+                        queuedVms,
+                        ++unscheduledVms
+                    ))
+
                     incomingImages -= imageInstance
 
                     logger.warn("Failed to spawn ${imageInstance.image}: does not fit [${clock.millis()}]")
@@ -168,8 +196,17 @@ class SimpleVirtProvisioningService(
                 )
                 imageInstance.server = server
                 imageInstance.continuation.resume(server)
-                queuedVms--
-                runningVms++
+
+                eventFlow.emit(VirtProvisioningEvent.MetricsAvailable(
+                    this@SimpleVirtProvisioningService,
+                    hypervisors.size,
+                    availableHypervisors.size,
+                    submittedVms,
+                    ++runningVms,
+                    finishedVms,
+                    --queuedVms,
+                    unscheduledVms
+                ))
                 activeImages += imageInstance
 
                 server.events
@@ -178,8 +215,17 @@ class SimpleVirtProvisioningService(
                             is ServerEvent.StateChanged -> {
                                 if (event.server.state == ServerState.SHUTOFF) {
                                     logger.info { "Server ${event.server.uid} ${event.server.name} ${event.server.flavor} finished." }
-                                    runningVms--
-                                    finishedVms++
+
+                                    eventFlow.emit(VirtProvisioningEvent.MetricsAvailable(
+                                        this@SimpleVirtProvisioningService,
+                                        hypervisors.size,
+                                        availableHypervisors.size,
+                                        submittedVms,
+                                        --runningVms,
+                                        ++finishedVms,
+                                        queuedVms,
+                                        unscheduledVms
+                                    ))
 
                                     activeImages -= imageInstance
                                     selectedHv.provisionedCores -= server.flavor.cpuCount
@@ -223,11 +269,33 @@ class SimpleVirtProvisioningService(
                     maxMemory = max(maxMemory, server.flavor.memorySize)
                     hypervisors[server] = hv
                 }
+
+                eventFlow.emit(VirtProvisioningEvent.MetricsAvailable(
+                    this@SimpleVirtProvisioningService,
+                    hypervisors.size,
+                    availableHypervisors.size,
+                    submittedVms,
+                    runningVms,
+                    finishedVms,
+                    queuedVms,
+                    unscheduledVms
+                ))
             }
             ServerState.SHUTOFF, ServerState.ERROR -> {
                 val hv = hypervisors[server] ?: return
                 availableHypervisors -= hv
 
+                eventFlow.emit(VirtProvisioningEvent.MetricsAvailable(
+                    this@SimpleVirtProvisioningService,
+                    hypervisors.size,
+                    availableHypervisors.size,
+                    submittedVms,
+                    runningVms,
+                    finishedVms,
+                    queuedVms,
+                    unscheduledVms
+                ))
+
                 if (incomingImages.isNotEmpty()) {
                     requestCycle()
                 }
@@ -242,6 +310,17 @@ class SimpleVirtProvisioningService(
             hv.driver = server.services[VirtDriver]
             availableHypervisors += hv
 
+            eventFlow.emit(VirtProvisioningEvent.MetricsAvailable(
+                this@SimpleVirtProvisioningService,
+                hypervisors.size,
+                availableHypervisors.size,
+                submittedVms,
+                runningVms,
+                finishedVms,
+                queuedVms,
+                unscheduledVms
+            ))
+
             hv.driver.events
                 .onEach { event ->
                     if (event is HypervisorEvent.VmsUpdated) {
diff --git a/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/service/VirtProvisioningEvent.kt b/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/service/VirtProvisioningEvent.kt
new file mode 100644
index 00000000..c3fb99f9
--- /dev/null
+++ b/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/service/VirtProvisioningEvent.kt
@@ -0,0 +1,49 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2020 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 com.atlarge.opendc.compute.virt.service
+
+/**
+ * An event that is emitted by the [VirtProvisioningService].
+ */
+public sealed class VirtProvisioningEvent {
+    /**
+     * The service that has emitted the event.
+     */
+    public abstract val provisioner: VirtProvisioningService
+
+    /**
+     * An event emitted for writing metrics.
+     */
+    data class MetricsAvailable(
+        override val provisioner: VirtProvisioningService,
+        public val totalHostCount: Int,
+        public val availableHostCount: Int,
+        public val totalVmCount: Int,
+        public val activeVmCount: Int,
+        public val inactiveVmCount: Int,
+        public val waitingVmCount: Int,
+        public val failedVmCount: Int
+    ) : VirtProvisioningEvent()
+}
diff --git a/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/service/VirtProvisioningService.kt b/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/service/VirtProvisioningService.kt
index 2ad7df84..c4cbd711 100644
--- a/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/service/VirtProvisioningService.kt
+++ b/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/service/VirtProvisioningService.kt
@@ -5,6 +5,7 @@ import com.atlarge.opendc.compute.core.Server
 import com.atlarge.opendc.compute.core.image.Image
 import com.atlarge.opendc.compute.virt.driver.VirtDriver
 import com.atlarge.opendc.compute.virt.service.allocation.AllocationPolicy
+import kotlinx.coroutines.flow.Flow
 
 /**
  * A service for VM provisioning on a cloud.
@@ -16,6 +17,11 @@ interface VirtProvisioningService {
     val allocationPolicy: AllocationPolicy
 
     /**
+     * The events emitted by the service.
+     */
+    public val events: Flow<VirtProvisioningEvent>
+
+    /**
      * Obtain the active hypervisors for this provisioner.
      */
     public suspend fun drivers(): Set<VirtDriver>