From 0bbb0adb97ba4783bbd0073f845781725e6212e8 Mon Sep 17 00:00:00 2001
From: Fabian Mastenbroek <mail.fabianm@gmail.com>
Date: Thu, 25 Mar 2021 10:23:47 +0100
Subject: compute: Add test suite for ComputeService

This change adds a test suite for the OpenDC compute service.
---
 .../compute/service/internal/ClientFlavor.kt       |   6 +
 .../opendc/compute/service/internal/ClientImage.kt |   6 +
 .../compute/service/internal/ClientServer.kt       |   6 +
 .../compute/service/internal/ComputeServiceImpl.kt | 274 +++++++++++----------
 .../opendc/compute/service/internal/HostView.kt    |   2 +
 .../compute/service/internal/InternalFlavor.kt     |   4 +-
 .../compute/service/internal/InternalImage.kt      |   4 +-
 .../compute/service/internal/InternalServer.kt     |  46 +++-
 .../service/scheduler/ReplayAllocationPolicy.kt    |   7 +-
 9 files changed, 202 insertions(+), 153 deletions(-)

(limited to 'simulator/opendc-compute/opendc-compute-service/src/main')

diff --git a/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/internal/ClientFlavor.kt b/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/internal/ClientFlavor.kt
index 29f10e27..4a8d3046 100644
--- a/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/internal/ClientFlavor.kt
+++ b/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/internal/ClientFlavor.kt
@@ -59,4 +59,10 @@ internal class ClientFlavor(private val delegate: Flavor) : Flavor {
         labels = delegate.labels
         meta = delegate.meta
     }
+
+    override fun equals(other: Any?): Boolean = other is Flavor && other.uid == uid
+
+    override fun hashCode(): Int = uid.hashCode()
+
+    override fun toString(): String = "Flavor[uid=$uid,name=$name]"
 }
diff --git a/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/internal/ClientImage.kt b/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/internal/ClientImage.kt
index 6c5b2ab0..e0b5c171 100644
--- a/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/internal/ClientImage.kt
+++ b/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/internal/ClientImage.kt
@@ -52,4 +52,10 @@ internal class ClientImage(private val delegate: Image) : Image {
         labels = delegate.labels
         meta = delegate.meta
     }
+
+    override fun equals(other: Any?): Boolean = other is Image && other.uid == uid
+
+    override fun hashCode(): Int = uid.hashCode()
+
+    override fun toString(): String = "Image[uid=$uid,name=$name]"
 }
diff --git a/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/internal/ClientServer.kt b/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/internal/ClientServer.kt
index ae4cee3b..f2929bf3 100644
--- a/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/internal/ClientServer.kt
+++ b/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/internal/ClientServer.kt
@@ -104,4 +104,10 @@ internal class ClientServer(private val delegate: Server) : Server, ServerWatche
             watcher.onStateChanged(this, newState)
         }
     }
+
+    override fun equals(other: Any?): Boolean = other is Server && other.uid == uid
+
+    override fun hashCode(): Int = uid.hashCode()
+
+    override fun toString(): String = "Server[uid=$uid,name=$name,state=$state]"
 }
diff --git a/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/internal/ComputeServiceImpl.kt b/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/internal/ComputeServiceImpl.kt
index aa7e0aa1..62808b4d 100644
--- a/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/internal/ComputeServiceImpl.kt
+++ b/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/internal/ComputeServiceImpl.kt
@@ -57,7 +57,7 @@ public class ComputeServiceImpl(
     /**
      * The [CoroutineScope] of the service bounded by the lifecycle of the service.
      */
-    private val scope = CoroutineScope(context)
+    private val scope = CoroutineScope(context + Job())
 
     /**
      * The logger instance of this server.
@@ -133,130 +133,133 @@ public class ComputeServiceImpl(
     override val hostCount: Int
         get() = hostToView.size
 
-    override fun newClient(): ComputeClient = object : ComputeClient {
-        private var isClosed: Boolean = false
+    override fun newClient(): ComputeClient {
+        check(scope.isActive) { "Service is already closed" }
+        return object : ComputeClient {
+            private var isClosed: Boolean = false
 
-        override suspend fun queryFlavors(): List<Flavor> {
-            check(!isClosed) { "Client is already closed" }
+            override suspend fun queryFlavors(): List<Flavor> {
+                check(!isClosed) { "Client is already closed" }
 
-            return flavors.values.map { ClientFlavor(it) }
-        }
+                return flavors.values.map { ClientFlavor(it) }
+            }
 
-        override suspend fun findFlavor(id: UUID): Flavor? {
-            check(!isClosed) { "Client is already closed" }
+            override suspend fun findFlavor(id: UUID): Flavor? {
+                check(!isClosed) { "Client is already closed" }
 
-            return flavors[id]?.let { ClientFlavor(it) }
-        }
+                return flavors[id]?.let { ClientFlavor(it) }
+            }
 
-        override suspend fun newFlavor(
-            name: String,
-            cpuCount: Int,
-            memorySize: Long,
-            labels: Map<String, String>,
-            meta: Map<String, Any>
-        ): Flavor {
-            check(!isClosed) { "Client is already closed" }
-
-            val uid = UUID(clock.millis(), random.nextLong())
-            val flavor = InternalFlavor(
-                this@ComputeServiceImpl,
-                uid,
-                name,
-                cpuCount,
-                memorySize,
-                labels,
-                meta
-            )
+            override suspend fun newFlavor(
+                name: String,
+                cpuCount: Int,
+                memorySize: Long,
+                labels: Map<String, String>,
+                meta: Map<String, Any>
+            ): Flavor {
+                check(!isClosed) { "Client is already closed" }
+
+                val uid = UUID(clock.millis(), random.nextLong())
+                val flavor = InternalFlavor(
+                    this@ComputeServiceImpl,
+                    uid,
+                    name,
+                    cpuCount,
+                    memorySize,
+                    labels,
+                    meta
+                )
 
-            flavors[uid] = flavor
+                flavors[uid] = flavor
 
-            return ClientFlavor(flavor)
-        }
+                return ClientFlavor(flavor)
+            }
 
-        override suspend fun queryImages(): List<Image> {
-            check(!isClosed) { "Client is already closed" }
+            override suspend fun queryImages(): List<Image> {
+                check(!isClosed) { "Client is already closed" }
 
-            return images.values.map { ClientImage(it) }
-        }
+                return images.values.map { ClientImage(it) }
+            }
 
-        override suspend fun findImage(id: UUID): Image? {
-            check(!isClosed) { "Client is already closed" }
+            override suspend fun findImage(id: UUID): Image? {
+                check(!isClosed) { "Client is already closed" }
 
-            return images[id]?.let { ClientImage(it) }
-        }
+                return images[id]?.let { ClientImage(it) }
+            }
 
-        override suspend fun newImage(name: String, labels: Map<String, String>, meta: Map<String, Any>): Image {
-            check(!isClosed) { "Client is already closed" }
+            override suspend fun newImage(name: String, labels: Map<String, String>, meta: Map<String, Any>): Image {
+                check(!isClosed) { "Client is already closed" }
 
-            val uid = UUID(clock.millis(), random.nextLong())
-            val image = InternalImage(this@ComputeServiceImpl, uid, name, labels, meta)
+                val uid = UUID(clock.millis(), random.nextLong())
+                val image = InternalImage(this@ComputeServiceImpl, uid, name, labels, meta)
 
-            images[uid] = image
+                images[uid] = image
 
-            return ClientImage(image)
-        }
+                return ClientImage(image)
+            }
 
-        override suspend fun newServer(
-            name: String,
-            image: Image,
-            flavor: Flavor,
-            labels: Map<String, String>,
-            meta: Map<String, Any>,
-            start: Boolean
-        ): Server {
-            check(!isClosed) { "Client is closed" }
-            tracer.commit(VmSubmissionEvent(name, image, flavor))
+            override suspend fun newServer(
+                name: String,
+                image: Image,
+                flavor: Flavor,
+                labels: Map<String, String>,
+                meta: Map<String, Any>,
+                start: Boolean
+            ): Server {
+                check(!isClosed) { "Client is closed" }
+                tracer.commit(VmSubmissionEvent(name, image, flavor))
 
-            _events.emit(
-                ComputeServiceEvent.MetricsAvailable(
+                _events.emit(
+                    ComputeServiceEvent.MetricsAvailable(
+                        this@ComputeServiceImpl,
+                        hostCount,
+                        availableHosts.size,
+                        ++submittedVms,
+                        runningVms,
+                        finishedVms,
+                        ++queuedVms,
+                        unscheduledVms
+                    )
+                )
+
+                val uid = UUID(clock.millis(), random.nextLong())
+                val server = InternalServer(
                     this@ComputeServiceImpl,
-                    hostCount,
-                    availableHosts.size,
-                    ++submittedVms,
-                    runningVms,
-                    finishedVms,
-                    ++queuedVms,
-                    unscheduledVms
+                    uid,
+                    name,
+                    requireNotNull(flavors[flavor.uid]) { "Unknown flavor" },
+                    requireNotNull(images[image.uid]) { "Unknown image" },
+                    labels.toMutableMap(),
+                    meta.toMutableMap()
                 )
-            )
 
-            val uid = UUID(clock.millis(), random.nextLong())
-            val server = InternalServer(
-                this@ComputeServiceImpl,
-                uid,
-                name,
-                flavor,
-                image,
-                labels.toMutableMap(),
-                meta.toMutableMap()
-            )
+                servers[uid] = server
 
-            servers[uid] = server
+                if (start) {
+                    server.start()
+                }
 
-            if (start) {
-                server.start()
+                return ClientServer(server)
             }
 
-            return ClientServer(server)
-        }
+            override suspend fun findServer(id: UUID): Server? {
+                check(!isClosed) { "Client is already closed" }
 
-        override suspend fun findServer(id: UUID): Server? {
-            check(!isClosed) { "Client is already closed" }
+                return servers[id]?.let { ClientServer(it) }
+            }
 
-            return servers[id]?.let { ClientServer(it) }
-        }
+            override suspend fun queryServers(): List<Server> {
+                check(!isClosed) { "Client is already closed" }
 
-        override suspend fun queryServers(): List<Server> {
-            check(!isClosed) { "Client is already closed" }
+                return servers.values.map { ClientServer(it) }
+            }
 
-            return servers.values.map { ClientServer(it) }
-        }
+            override fun close() {
+                isClosed = true
+            }
 
-        override fun close() {
-            isClosed = true
+            override fun toString(): String = "ComputeClient"
         }
-
-        override fun toString(): String = "ComputeClient"
     }
 
     override fun addHost(host: Host) {
@@ -285,23 +288,25 @@ public class ComputeServiceImpl(
         scope.cancel()
     }
 
-    internal fun schedule(server: InternalServer) {
+    internal fun schedule(server: InternalServer): SchedulingRequest {
         logger.debug { "Enqueueing server ${server.uid} to be assigned to host." }
 
-        queue.add(SchedulingRequest(server))
+        val request = SchedulingRequest(server)
+        queue.add(request)
         requestSchedulingCycle()
+        return request
     }
 
     internal fun delete(flavor: InternalFlavor) {
-        checkNotNull(flavors.remove(flavor.uid)) { "Flavor was not known" }
+        flavors.remove(flavor.uid)
     }
 
     internal fun delete(image: InternalImage) {
-        checkNotNull(images.remove(image.uid)) { "Image was not known" }
+        images.remove(image.uid)
     }
 
     internal fun delete(server: InternalServer) {
-        checkNotNull(servers.remove(server.uid)) { "Server was not known" }
+        servers.remove(server.uid)
     }
 
     /**
@@ -338,7 +343,7 @@ public class ComputeServiceImpl(
             val server = request.server
             val hv = allocationLogic.select(availableHosts, request.server)
             if (hv == null || !hv.host.canFit(server)) {
-                logger.trace { "Server $server selected for scheduling but no capacity available for it." }
+                logger.trace { "Server $server selected for scheduling but no capacity available for it at the moment" }
 
                 if (server.flavor.memorySize > maxMemory || server.flavor.cpuCount > maxCores) {
                     tracer.commit(VmSubmissionInvalidEvent(server.name))
@@ -360,6 +365,8 @@ public class ComputeServiceImpl(
                     queue.poll()
 
                     logger.warn("Failed to spawn $server: does not fit [${clock.millis()}]")
+
+                    server.state = ServerState.ERROR
                     continue
                 } else {
                     break
@@ -372,42 +379,39 @@ public class ComputeServiceImpl(
             queue.poll()
 
             logger.info { "Assigned server $server to host $host." }
-            try {
-                // Speculatively update the hypervisor view information to prevent other images in the queue from
-                // deciding on stale values.
-                hv.numberOfActiveServers++
-                hv.provisionedCores += server.flavor.cpuCount
-                hv.availableMemory -= server.flavor.memorySize // XXX Temporary hack
-
-                scope.launch {
-                    try {
-                        server.assignHost(host)
-                        host.spawn(server)
-                        activeServers[server] = host
-
-                        tracer.commit(VmScheduledEvent(server.name))
-                        _events.emit(
-                            ComputeServiceEvent.MetricsAvailable(
-                                this@ComputeServiceImpl,
-                                hostCount,
-                                availableHosts.size,
-                                submittedVms,
-                                ++runningVms,
-                                finishedVms,
-                                --queuedVms,
-                                unscheduledVms
-                            )
+
+            // Speculatively update the hypervisor view information to prevent other images in the queue from
+            // deciding on stale values.
+            hv.numberOfActiveServers++
+            hv.provisionedCores += server.flavor.cpuCount
+            hv.availableMemory -= server.flavor.memorySize // XXX Temporary hack
+
+            scope.launch {
+                try {
+                    server.host = host
+                    host.spawn(server)
+                    activeServers[server] = host
+
+                    tracer.commit(VmScheduledEvent(server.name))
+                    _events.emit(
+                        ComputeServiceEvent.MetricsAvailable(
+                            this@ComputeServiceImpl,
+                            hostCount,
+                            availableHosts.size,
+                            submittedVms,
+                            ++runningVms,
+                            finishedVms,
+                            --queuedVms,
+                            unscheduledVms
                         )
-                    } catch (e: Throwable) {
-                        logger.error("Failed to deploy VM", e)
+                    )
+                } catch (e: Throwable) {
+                    logger.error("Failed to deploy VM", e)
 
-                        hv.numberOfActiveServers--
-                        hv.provisionedCores -= server.flavor.cpuCount
-                        hv.availableMemory += server.flavor.memorySize
-                    }
+                    hv.numberOfActiveServers--
+                    hv.provisionedCores -= server.flavor.cpuCount
+                    hv.availableMemory += server.flavor.memorySize
                 }
-            } catch (e: Exception) {
-                logger.warn(e) { "Failed to assign server $server to $host. " }
             }
         }
     }
@@ -415,7 +419,7 @@ public class ComputeServiceImpl(
     /**
      * A request to schedule an [InternalServer] onto one of the [Host]s.
      */
-    private data class SchedulingRequest(val server: InternalServer) {
+    internal data class SchedulingRequest(val server: InternalServer) {
         /**
          * A flag to indicate that the request is cancelled.
          */
diff --git a/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/internal/HostView.kt b/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/internal/HostView.kt
index 1bdfdf1a..5793541f 100644
--- a/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/internal/HostView.kt
+++ b/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/internal/HostView.kt
@@ -32,4 +32,6 @@ public class HostView(public val host: Host) {
     public var numberOfActiveServers: Int = 0
     public var availableMemory: Long = host.model.memorySize
     public var provisionedCores: Int = 0
+
+    override fun toString(): String = "HostView[host=$host]"
 }
diff --git a/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/internal/InternalFlavor.kt b/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/internal/InternalFlavor.kt
index 95e280df..b8fb6279 100644
--- a/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/internal/InternalFlavor.kt
+++ b/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/internal/InternalFlavor.kt
@@ -58,7 +58,9 @@ internal class InternalFlavor(
         service.delete(this)
     }
 
-    override fun equals(other: Any?): Boolean = other is InternalFlavor && uid == other.uid
+    override fun equals(other: Any?): Boolean = other is Flavor && uid == other.uid
 
     override fun hashCode(): Int = uid.hashCode()
+
+    override fun toString(): String = "Flavor[uid=$uid,name=$name]"
 }
diff --git a/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/internal/InternalImage.kt b/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/internal/InternalImage.kt
index 86f2f6b9..d9ed5896 100644
--- a/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/internal/InternalImage.kt
+++ b/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/internal/InternalImage.kt
@@ -48,7 +48,9 @@ internal class InternalImage(
         service.delete(this)
     }
 
-    override fun equals(other: Any?): Boolean = other is InternalImage && uid == other.uid
+    override fun equals(other: Any?): Boolean = other is Image && uid == other.uid
 
     override fun hashCode(): Int = uid.hashCode()
+
+    override fun toString(): String = "Image[uid=$uid,name=$name]"
 }
diff --git a/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/internal/InternalServer.kt b/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/internal/InternalServer.kt
index ff7c1d15..d9d0f3fc 100644
--- a/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/internal/InternalServer.kt
+++ b/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/internal/InternalServer.kt
@@ -34,8 +34,8 @@ internal class InternalServer(
     private val service: ComputeServiceImpl,
     override val uid: UUID,
     override val name: String,
-    override val flavor: Flavor,
-    override val image: Image,
+    override val flavor: InternalFlavor,
+    override val image: InternalImage,
     override val labels: MutableMap<String, String>,
     override val meta: MutableMap<String, Any>
 ) : Server {
@@ -54,6 +54,11 @@ internal class InternalServer(
      */
     internal var host: Host? = null
 
+    /**
+     * The current scheduling request.
+     */
+    private var request: ComputeServiceImpl.SchedulingRequest? = null
+
     override suspend fun start() {
         when (state) {
             ServerState.RUNNING -> {
@@ -66,35 +71,43 @@ internal class InternalServer(
             }
             ServerState.DELETED -> {
                 logger.warn { "User tried to start terminated server" }
-                throw IllegalArgumentException("Server is terminated")
+                throw IllegalStateException("Server is terminated")
             }
             else -> {
                 logger.info { "User requested to start server $uid" }
                 state = ServerState.PROVISIONING
-                service.schedule(this)
+                assert(request == null) { "Scheduling request already active" }
+                request = service.schedule(this)
             }
         }
     }
 
     override suspend fun stop() {
         when (state) {
-            ServerState.PROVISIONING -> {} // TODO Find way to interrupt these
+            ServerState.PROVISIONING -> {
+                cancelProvisioningRequest()
+                state = ServerState.TERMINATED
+            }
             ServerState.RUNNING, ServerState.ERROR -> {
                 val host = checkNotNull(host) { "Server not running" }
                 host.stop(this)
             }
-            ServerState.TERMINATED -> {} // No work needed
-            ServerState.DELETED -> throw IllegalStateException("Server is terminated")
+            ServerState.TERMINATED, ServerState.DELETED -> {} // No work needed
         }
     }
 
     override suspend fun delete() {
         when (state) {
-            ServerState.PROVISIONING -> {} // TODO Find way to interrupt these
-            ServerState.RUNNING -> {
+            ServerState.PROVISIONING, ServerState.TERMINATED -> {
+                cancelProvisioningRequest()
+                service.delete(this)
+                state = ServerState.DELETED
+            }
+            ServerState.RUNNING, ServerState.ERROR -> {
                 val host = checkNotNull(host) { "Server not running" }
                 host.delete(this)
                 service.delete(this)
+                state = ServerState.DELETED
             }
             else -> {} // No work needed
         }
@@ -121,11 +134,20 @@ internal class InternalServer(
             field = value
         }
 
-    internal fun assignHost(host: Host) {
-        this.host = host
+    /**
+     * Cancel the provisioning request if active.
+     */
+    private fun cancelProvisioningRequest() {
+        val request = request
+        if (request != null) {
+            this.request = null
+            request.isCancelled = true
+        }
     }
 
-    override fun equals(other: Any?): Boolean = other is InternalServer && uid == other.uid
+    override fun equals(other: Any?): Boolean = other is Server && uid == other.uid
 
     override fun hashCode(): Int = uid.hashCode()
+
+    override fun toString(): String = "Server[uid=$uid,state=$state]"
 }
diff --git a/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/scheduler/ReplayAllocationPolicy.kt b/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/scheduler/ReplayAllocationPolicy.kt
index ed1dc662..2c953f8b 100644
--- a/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/scheduler/ReplayAllocationPolicy.kt
+++ b/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/scheduler/ReplayAllocationPolicy.kt
@@ -20,14 +20,11 @@
  * SOFTWARE.
  */
 
-package org.opendc.compute.simulator.allocation
+package org.opendc.compute.service.scheduler
 
 import mu.KotlinLogging
 import org.opendc.compute.api.Server
 import org.opendc.compute.service.internal.HostView
-import org.opendc.compute.service.scheduler.AllocationPolicy
-
-private val logger = KotlinLogging.logger {}
 
 /**
  * Policy replaying VM-cluster assignment.
@@ -36,6 +33,8 @@ private val logger = KotlinLogging.logger {}
  * assigned the VM image.
  */
 public class ReplayAllocationPolicy(private val vmPlacements: Map<String, String>) : AllocationPolicy {
+    private val logger = KotlinLogging.logger {}
+
     override fun invoke(): AllocationPolicy.Logic = object : AllocationPolicy.Logic {
         override fun select(
             hypervisors: Set<HostView>,
-- 
cgit v1.2.3


From 0c5f0a07f87f85119be83b923ffe60b3863ebb9d Mon Sep 17 00:00:00 2001
From: Fabian Mastenbroek <mail.fabianm@gmail.com>
Date: Thu, 25 Mar 2021 14:33:56 +0100
Subject: compute: Hide internals of compute service implementation

This change changes the compute service and users of the compute service
to not rely on the internals of `ComputeServiceImpl` and instead use its
public API.
---
 .../opendc/compute/service/internal/ComputeServiceImpl.kt    | 12 ++++++------
 1 file changed, 6 insertions(+), 6 deletions(-)

(limited to 'simulator/opendc-compute/opendc-compute-service/src/main')

diff --git a/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/internal/ComputeServiceImpl.kt b/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/internal/ComputeServiceImpl.kt
index 62808b4d..18255922 100644
--- a/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/internal/ComputeServiceImpl.kt
+++ b/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/internal/ComputeServiceImpl.kt
@@ -47,7 +47,7 @@ import kotlin.math.max
  * @param context The [CoroutineContext] to use.
  * @param clock The clock instance to keep track of time.
  */
-public class ComputeServiceImpl(
+internal class ComputeServiceImpl(
     private val context: CoroutineContext,
     private val clock: Clock,
     private val tracer: EventTracer,
@@ -104,11 +104,11 @@ public class ComputeServiceImpl(
      */
     private val servers = mutableMapOf<UUID, InternalServer>()
 
-    public var submittedVms: Int = 0
-    public var queuedVms: Int = 0
-    public var runningVms: Int = 0
-    public var finishedVms: Int = 0
-    public var unscheduledVms: Int = 0
+    private var submittedVms: Int = 0
+    private var queuedVms: Int = 0
+    private var runningVms: Int = 0
+    private var finishedVms: Int = 0
+    private var unscheduledVms: Int = 0
 
     private var maxCores = 0
     private var maxMemory = 0L
-- 
cgit v1.2.3


From 0d66ef47d6e1ec0861b4939800c5070f96600ca0 Mon Sep 17 00:00:00 2001
From: Fabian Mastenbroek <mail.fabianm@gmail.com>
Date: Thu, 25 Mar 2021 14:49:07 +0100
Subject: compute: Remove dependency on event tracer

This change removes the dependency on the event tracer in
`opendc-trace`, since we are in the process of migrating OpenDC to
use OpenTelemetry for distributed tracing and metrics.
---
 .../org/opendc/compute/service/ComputeService.kt   |  5 +---
 .../service/events/HypervisorAvailableEvent.kt     | 31 ---------------------
 .../service/events/HypervisorUnavailableEvent.kt   | 31 ---------------------
 .../compute/service/events/VmScheduledEvent.kt     | 30 --------------------
 .../compute/service/events/VmStoppedEvent.kt       | 30 --------------------
 .../compute/service/events/VmSubmissionEvent.kt    | 32 ----------------------
 .../service/events/VmSubmissionInvalidEvent.kt     | 30 --------------------
 .../compute/service/internal/ComputeServiceImpl.kt | 14 ----------
 8 files changed, 1 insertion(+), 202 deletions(-)
 delete mode 100644 simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/events/HypervisorAvailableEvent.kt
 delete mode 100644 simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/events/HypervisorUnavailableEvent.kt
 delete mode 100644 simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/events/VmScheduledEvent.kt
 delete mode 100644 simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/events/VmStoppedEvent.kt
 delete mode 100644 simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/events/VmSubmissionEvent.kt
 delete mode 100644 simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/events/VmSubmissionInvalidEvent.kt

(limited to 'simulator/opendc-compute/opendc-compute-service/src/main')

diff --git a/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/ComputeService.kt b/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/ComputeService.kt
index 593e4b56..28cef83a 100644
--- a/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/ComputeService.kt
+++ b/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/ComputeService.kt
@@ -27,7 +27,6 @@ import org.opendc.compute.api.ComputeClient
 import org.opendc.compute.service.driver.Host
 import org.opendc.compute.service.internal.ComputeServiceImpl
 import org.opendc.compute.service.scheduler.AllocationPolicy
-import org.opendc.trace.core.EventTracer
 import java.time.Clock
 import kotlin.coroutines.CoroutineContext
 
@@ -76,17 +75,15 @@ public interface ComputeService : AutoCloseable {
          *
          * @param context The [CoroutineContext] to use in the service.
          * @param clock The clock instance to use.
-         * @param tracer The event tracer to use.
          * @param allocationPolicy The allocation policy to use.
          */
         public operator fun invoke(
             context: CoroutineContext,
             clock: Clock,
-            tracer: EventTracer,
             allocationPolicy: AllocationPolicy,
             schedulingQuantum: Long = 300000,
         ): ComputeService {
-            return ComputeServiceImpl(context, clock, tracer, allocationPolicy, schedulingQuantum)
+            return ComputeServiceImpl(context, clock, allocationPolicy, schedulingQuantum)
         }
     }
 }
diff --git a/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/events/HypervisorAvailableEvent.kt b/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/events/HypervisorAvailableEvent.kt
deleted file mode 100644
index a7974062..00000000
--- a/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/events/HypervisorAvailableEvent.kt
+++ /dev/null
@@ -1,31 +0,0 @@
-/*
- * 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 org.opendc.compute.service.events
-
-import org.opendc.trace.core.Event
-import java.util.*
-
-/**
- * This event is emitted when a hypervisor has become available.
- */
-public class HypervisorAvailableEvent(public val uid: UUID) : Event()
diff --git a/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/events/HypervisorUnavailableEvent.kt b/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/events/HypervisorUnavailableEvent.kt
deleted file mode 100644
index 75bb09ed..00000000
--- a/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/events/HypervisorUnavailableEvent.kt
+++ /dev/null
@@ -1,31 +0,0 @@
-/*
- * 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 org.opendc.compute.service.events
-
-import org.opendc.trace.core.Event
-import java.util.*
-
-/**
- * This event is emitted when a hypervisor has become unavailable.
- */
-public class HypervisorUnavailableEvent(public val uid: UUID) : Event()
diff --git a/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/events/VmScheduledEvent.kt b/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/events/VmScheduledEvent.kt
deleted file mode 100644
index f59c74b7..00000000
--- a/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/events/VmScheduledEvent.kt
+++ /dev/null
@@ -1,30 +0,0 @@
-/*
- * 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 org.opendc.compute.service.events
-
-import org.opendc.trace.core.Event
-
-/**
- * This event is emitted when a virtual machine has successfully been scheduled on a hypervisor.
- */
-public class VmScheduledEvent(public val name: String) : Event()
diff --git a/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/events/VmStoppedEvent.kt b/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/events/VmStoppedEvent.kt
deleted file mode 100644
index eaf0736b..00000000
--- a/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/events/VmStoppedEvent.kt
+++ /dev/null
@@ -1,30 +0,0 @@
-/*
- * 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 org.opendc.compute.service.events
-
-import org.opendc.trace.core.Event
-
-/**
- * This event is emitted when a virtual machine has stopped running.
- */
-public class VmStoppedEvent(public val name: String) : Event()
diff --git a/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/events/VmSubmissionEvent.kt b/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/events/VmSubmissionEvent.kt
deleted file mode 100644
index fa0a8a13..00000000
--- a/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/events/VmSubmissionEvent.kt
+++ /dev/null
@@ -1,32 +0,0 @@
-/*
- * 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 org.opendc.compute.service.events
-
-import org.opendc.compute.api.Flavor
-import org.opendc.compute.api.Image
-import org.opendc.trace.core.Event
-
-/**
- * This event is emitted when a virtual machine is submitted to the provisioning service.
- */
-public class VmSubmissionEvent(public val name: String, public val image: Image, public val flavor: Flavor) : Event()
diff --git a/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/events/VmSubmissionInvalidEvent.kt b/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/events/VmSubmissionInvalidEvent.kt
deleted file mode 100644
index 52b91616..00000000
--- a/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/events/VmSubmissionInvalidEvent.kt
+++ /dev/null
@@ -1,30 +0,0 @@
-/*
- * 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 org.opendc.compute.service.events
-
-import org.opendc.trace.core.Event
-
-/**
- * An event that is emitted when the submission is deemed to be invalid.
- */
-public class VmSubmissionInvalidEvent(public val name: String) : Event()
diff --git a/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/internal/ComputeServiceImpl.kt b/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/internal/ComputeServiceImpl.kt
index 18255922..f9bd7fbc 100644
--- a/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/internal/ComputeServiceImpl.kt
+++ b/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/internal/ComputeServiceImpl.kt
@@ -31,9 +31,7 @@ import org.opendc.compute.service.ComputeServiceEvent
 import org.opendc.compute.service.driver.Host
 import org.opendc.compute.service.driver.HostListener
 import org.opendc.compute.service.driver.HostState
-import org.opendc.compute.service.events.*
 import org.opendc.compute.service.scheduler.AllocationPolicy
-import org.opendc.trace.core.EventTracer
 import org.opendc.utils.TimerScheduler
 import org.opendc.utils.flow.EventFlow
 import java.time.Clock
@@ -50,7 +48,6 @@ import kotlin.math.max
 internal class ComputeServiceImpl(
     private val context: CoroutineContext,
     private val clock: Clock,
-    private val tracer: EventTracer,
     private val allocationPolicy: AllocationPolicy,
     private val schedulingQuantum: Long
 ) : ComputeService, HostListener {
@@ -207,8 +204,6 @@ internal class ComputeServiceImpl(
                 start: Boolean
             ): Server {
                 check(!isClosed) { "Client is closed" }
-                tracer.commit(VmSubmissionEvent(name, image, flavor))
-
                 _events.emit(
                     ComputeServiceEvent.MetricsAvailable(
                         this@ComputeServiceImpl,
@@ -346,8 +341,6 @@ internal class ComputeServiceImpl(
                 logger.trace { "Server $server selected for scheduling but no capacity available for it at the moment" }
 
                 if (server.flavor.memorySize > maxMemory || server.flavor.cpuCount > maxCores) {
-                    tracer.commit(VmSubmissionInvalidEvent(server.name))
-
                     _events.emit(
                         ComputeServiceEvent.MetricsAvailable(
                             this@ComputeServiceImpl,
@@ -392,7 +385,6 @@ internal class ComputeServiceImpl(
                     host.spawn(server)
                     activeServers[server] = host
 
-                    tracer.commit(VmScheduledEvent(server.name))
                     _events.emit(
                         ComputeServiceEvent.MetricsAvailable(
                             this@ComputeServiceImpl,
@@ -437,8 +429,6 @@ internal class ComputeServiceImpl(
                     availableHosts += hv
                 }
 
-                tracer.commit(HypervisorAvailableEvent(host.uid))
-
                 _events.emit(
                     ComputeServiceEvent.MetricsAvailable(
                         this@ComputeServiceImpl,
@@ -461,8 +451,6 @@ internal class ComputeServiceImpl(
                 val hv = hostToView[host] ?: return
                 availableHosts -= hv
 
-                tracer.commit(HypervisorUnavailableEvent(hv.uid))
-
                 _events.emit(
                     ComputeServiceEvent.MetricsAvailable(
                         this@ComputeServiceImpl,
@@ -495,8 +483,6 @@ internal class ComputeServiceImpl(
         if (newState == ServerState.TERMINATED || newState == ServerState.DELETED) {
             logger.info { "[${clock.millis()}] Server ${server.uid} ${server.name} ${server.flavor} finished." }
 
-            tracer.commit(VmStoppedEvent(server.name))
-
             _events.emit(
                 ComputeServiceEvent.MetricsAvailable(
                     this@ComputeServiceImpl,
-- 
cgit v1.2.3


From 608ff59b2d7e8ce696fe6f7271d80b5efc9c4b87 Mon Sep 17 00:00:00 2001
From: Fabian Mastenbroek <mail.fabianm@gmail.com>
Date: Thu, 25 Mar 2021 21:50:45 +0100
Subject: compute: Integrate OpenTelemetry Metrics in OpenDC Compute

This change integrates the OpenTelemetry Metrics API in the OpenDC
Compute Service implementation. This replaces the old infrastructure for
gathering metrics.
---
 .../org/opendc/compute/service/ComputeService.kt   |   9 +-
 .../opendc/compute/service/ComputeServiceEvent.kt  |  47 ------
 .../compute/service/internal/ComputeServiceImpl.kt | 174 ++++++++++-----------
 3 files changed, 85 insertions(+), 145 deletions(-)
 delete mode 100644 simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/ComputeServiceEvent.kt

(limited to 'simulator/opendc-compute/opendc-compute-service/src/main')

diff --git a/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/ComputeService.kt b/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/ComputeService.kt
index 28cef83a..4bc0ba78 100644
--- a/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/ComputeService.kt
+++ b/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/ComputeService.kt
@@ -22,6 +22,7 @@
 
 package org.opendc.compute.service
 
+import io.opentelemetry.api.metrics.Meter
 import kotlinx.coroutines.flow.Flow
 import org.opendc.compute.api.ComputeClient
 import org.opendc.compute.service.driver.Host
@@ -34,11 +35,6 @@ import kotlin.coroutines.CoroutineContext
  * The [ComputeService] hosts the API implementation of the OpenDC Compute service.
  */
 public interface ComputeService : AutoCloseable {
-    /**
-     * The events emitted by the service.
-     */
-    public val events: Flow<ComputeServiceEvent>
-
     /**
      * The hosts that are used by the compute service.
      */
@@ -80,10 +76,11 @@ public interface ComputeService : AutoCloseable {
         public operator fun invoke(
             context: CoroutineContext,
             clock: Clock,
+            meter: Meter,
             allocationPolicy: AllocationPolicy,
             schedulingQuantum: Long = 300000,
         ): ComputeService {
-            return ComputeServiceImpl(context, clock, allocationPolicy, schedulingQuantum)
+            return ComputeServiceImpl(context, clock, meter, allocationPolicy, schedulingQuantum)
         }
     }
 }
diff --git a/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/ComputeServiceEvent.kt b/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/ComputeServiceEvent.kt
deleted file mode 100644
index 193008a7..00000000
--- a/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/ComputeServiceEvent.kt
+++ /dev/null
@@ -1,47 +0,0 @@
-/*
- * 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.service
-
-/**
- * An event that is emitted by the [ComputeService].
- */
-public sealed class ComputeServiceEvent {
-    /**
-     * The service that has emitted the event.
-     */
-    public abstract val provisioner: ComputeService
-
-    /**
-     * An event emitted for writing metrics.
-     */
-    public data class MetricsAvailable(
-        override val provisioner: ComputeService,
-        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
-    ) : ComputeServiceEvent()
-}
diff --git a/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/internal/ComputeServiceImpl.kt b/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/internal/ComputeServiceImpl.kt
index f9bd7fbc..26a34ad9 100644
--- a/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/internal/ComputeServiceImpl.kt
+++ b/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/internal/ComputeServiceImpl.kt
@@ -22,18 +22,16 @@
 
 package org.opendc.compute.service.internal
 
+import io.opentelemetry.api.metrics.Meter
 import kotlinx.coroutines.*
-import kotlinx.coroutines.flow.Flow
 import mu.KotlinLogging
 import org.opendc.compute.api.*
 import org.opendc.compute.service.ComputeService
-import org.opendc.compute.service.ComputeServiceEvent
 import org.opendc.compute.service.driver.Host
 import org.opendc.compute.service.driver.HostListener
 import org.opendc.compute.service.driver.HostState
 import org.opendc.compute.service.scheduler.AllocationPolicy
 import org.opendc.utils.TimerScheduler
-import org.opendc.utils.flow.EventFlow
 import java.time.Clock
 import java.util.*
 import kotlin.coroutines.CoroutineContext
@@ -48,6 +46,7 @@ import kotlin.math.max
 internal class ComputeServiceImpl(
     private val context: CoroutineContext,
     private val clock: Clock,
+    private val meter: Meter,
     private val allocationPolicy: AllocationPolicy,
     private val schedulingQuantum: Long
 ) : ComputeService, HostListener {
@@ -101,24 +100,70 @@ internal class ComputeServiceImpl(
      */
     private val servers = mutableMapOf<UUID, InternalServer>()
 
-    private var submittedVms: Int = 0
-    private var queuedVms: Int = 0
-    private var runningVms: Int = 0
-    private var finishedVms: Int = 0
-    private var unscheduledVms: Int = 0
-
     private var maxCores = 0
     private var maxMemory = 0L
 
+    /**
+     * The number of servers that have been submitted to the service for provisioning.
+     */
+    private val _submittedServers = meter.longCounterBuilder("servers.submitted")
+        .setDescription("Number of start requests")
+        .setUnit("1")
+        .build()
+
+    /**
+     * The number of servers that failed to be scheduled.
+     */
+    private val _unscheduledServers = meter.longCounterBuilder("servers.unscheduled")
+        .setDescription("Number of unscheduled servers")
+        .setUnit("1")
+        .build()
+
+    /**
+     * The number of servers that are waiting to be provisioned.
+     */
+    private val _waitingServers = meter.longUpDownCounterBuilder("servers.waiting")
+        .setDescription("Number of servers waiting to be provisioned")
+        .setUnit("1")
+        .build()
+
+    /**
+     * The number of servers that are waiting to be provisioned.
+     */
+    private val _runningServers = meter.longUpDownCounterBuilder("servers.active")
+        .setDescription("Number of servers currently running")
+        .setUnit("1")
+        .build()
+
+    /**
+     * The number of servers that have finished running.
+     */
+    private val _finishedServers = meter.longCounterBuilder("servers.finished")
+        .setDescription("Number of servers that finished running")
+        .setUnit("1")
+        .build()
+
+    /**
+     * The number of hosts registered at the compute service.
+     */
+    private val _hostCount = meter.longUpDownCounterBuilder("hosts.total")
+        .setDescription("Number of hosts")
+        .setUnit("1")
+        .build()
+
+    /**
+     * The number of available hosts registered at the compute service.
+     */
+    private val _availableHostCount = meter.longUpDownCounterBuilder("hosts.available")
+        .setDescription("Number of available hosts")
+        .setUnit("1")
+        .build()
+
     /**
      * The allocation logic to use.
      */
     private val allocationLogic = allocationPolicy()
 
-    override val events: Flow<ComputeServiceEvent>
-        get() = _events
-    private val _events = EventFlow<ComputeServiceEvent>()
-
     /**
      * The [TimerScheduler] to use for scheduling the scheduler cycles.
      */
@@ -204,18 +249,6 @@ internal class ComputeServiceImpl(
                 start: Boolean
             ): Server {
                 check(!isClosed) { "Client is closed" }
-                _events.emit(
-                    ComputeServiceEvent.MetricsAvailable(
-                        this@ComputeServiceImpl,
-                        hostCount,
-                        availableHosts.size,
-                        ++submittedVms,
-                        runningVms,
-                        finishedVms,
-                        ++queuedVms,
-                        unscheduledVms
-                    )
-                )
 
                 val uid = UUID(clock.millis(), random.nextLong())
                 val server = InternalServer(
@@ -269,14 +302,23 @@ internal class ComputeServiceImpl(
         hostToView[host] = hv
 
         if (host.state == HostState.UP) {
+            _availableHostCount.add(1)
             availableHosts += hv
         }
 
+        _hostCount.add(1)
         host.addListener(this)
     }
 
     override fun removeHost(host: Host) {
-        host.removeListener(this)
+        val view = hostToView.remove(host)
+        if (view != null) {
+            if (availableHosts.remove(view)) {
+                _availableHostCount.add(-1)
+            }
+            host.removeListener(this)
+            _hostCount.add(-1)
+        }
     }
 
     override fun close() {
@@ -288,6 +330,8 @@ internal class ComputeServiceImpl(
 
         val request = SchedulingRequest(server)
         queue.add(request)
+        _submittedServers.add(1)
+        _waitingServers.add(1)
         requestSchedulingCycle()
         return request
     }
@@ -332,6 +376,7 @@ internal class ComputeServiceImpl(
 
             if (request.isCancelled) {
                 queue.poll()
+                _waitingServers.add(-1)
                 continue
             }
 
@@ -341,21 +386,10 @@ internal class ComputeServiceImpl(
                 logger.trace { "Server $server selected for scheduling but no capacity available for it at the moment" }
 
                 if (server.flavor.memorySize > maxMemory || server.flavor.cpuCount > maxCores) {
-                    _events.emit(
-                        ComputeServiceEvent.MetricsAvailable(
-                            this@ComputeServiceImpl,
-                            hostCount,
-                            availableHosts.size,
-                            submittedVms,
-                            runningVms,
-                            finishedVms,
-                            --queuedVms,
-                            ++unscheduledVms
-                        )
-                    )
-
                     // Remove the incoming image
                     queue.poll()
+                    _waitingServers.add(-1)
+                    _unscheduledServers.add(1)
 
                     logger.warn("Failed to spawn $server: does not fit [${clock.millis()}]")
 
@@ -370,6 +404,7 @@ internal class ComputeServiceImpl(
 
             // Remove request from queue
             queue.poll()
+            _waitingServers.add(-1)
 
             logger.info { "Assigned server $server to host $host." }
 
@@ -384,19 +419,6 @@ internal class ComputeServiceImpl(
                     server.host = host
                     host.spawn(server)
                     activeServers[server] = host
-
-                    _events.emit(
-                        ComputeServiceEvent.MetricsAvailable(
-                            this@ComputeServiceImpl,
-                            hostCount,
-                            availableHosts.size,
-                            submittedVms,
-                            ++runningVms,
-                            finishedVms,
-                            --queuedVms,
-                            unscheduledVms
-                        )
-                    )
                 } catch (e: Throwable) {
                     logger.error("Failed to deploy VM", e)
 
@@ -427,21 +449,9 @@ internal class ComputeServiceImpl(
                 if (hv != null) {
                     // Corner case for when the hypervisor already exists
                     availableHosts += hv
+                    _availableHostCount.add(1)
                 }
 
-                _events.emit(
-                    ComputeServiceEvent.MetricsAvailable(
-                        this@ComputeServiceImpl,
-                        hostCount,
-                        availableHosts.size,
-                        submittedVms,
-                        runningVms,
-                        finishedVms,
-                        queuedVms,
-                        unscheduledVms
-                    )
-                )
-
                 // Re-schedule on the new machine
                 requestSchedulingCycle()
             }
@@ -450,19 +460,7 @@ internal class ComputeServiceImpl(
 
                 val hv = hostToView[host] ?: return
                 availableHosts -= hv
-
-                _events.emit(
-                    ComputeServiceEvent.MetricsAvailable(
-                        this@ComputeServiceImpl,
-                        hostCount,
-                        availableHosts.size,
-                        submittedVms,
-                        runningVms,
-                        finishedVms,
-                        queuedVms,
-                        unscheduledVms
-                    )
-                )
+                _availableHostCount.add(-1)
 
                 requestSchedulingCycle()
             }
@@ -480,23 +478,15 @@ internal class ComputeServiceImpl(
 
         server.state = newState
 
-        if (newState == ServerState.TERMINATED || newState == ServerState.DELETED) {
+        if (newState == ServerState.RUNNING) {
+            _runningServers.add(1)
+        } else if (newState == ServerState.TERMINATED || newState == ServerState.DELETED) {
             logger.info { "[${clock.millis()}] Server ${server.uid} ${server.name} ${server.flavor} finished." }
 
-            _events.emit(
-                ComputeServiceEvent.MetricsAvailable(
-                    this@ComputeServiceImpl,
-                    hostCount,
-                    availableHosts.size,
-                    submittedVms,
-                    --runningVms,
-                    ++finishedVms,
-                    queuedVms,
-                    unscheduledVms
-                )
-            )
-
             activeServers -= server
+            _runningServers.add(-1)
+            _finishedServers.add(1)
+
             val hv = hostToView[host]
             if (hv != null) {
                 hv.provisionedCores -= server.flavor.cpuCount
-- 
cgit v1.2.3


From be95e0b10b648e256c4eb8c2190ad6ed61f8afdd Mon Sep 17 00:00:00 2001
From: Fabian Mastenbroek <mail.fabianm@gmail.com>
Date: Fri, 26 Mar 2021 15:36:44 +0100
Subject: workflow: Remove event tracer from workflow service

This change removes the event tracer from the OpenDC Workflow service as
we start migrating to the industry standard OpenTelemetry.
---
 .../src/main/kotlin/org/opendc/compute/service/ComputeService.kt         | 1 -
 1 file changed, 1 deletion(-)

(limited to 'simulator/opendc-compute/opendc-compute-service/src/main')

diff --git a/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/ComputeService.kt b/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/ComputeService.kt
index 4bc0ba78..98566da3 100644
--- a/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/ComputeService.kt
+++ b/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/ComputeService.kt
@@ -23,7 +23,6 @@
 package org.opendc.compute.service
 
 import io.opentelemetry.api.metrics.Meter
-import kotlinx.coroutines.flow.Flow
 import org.opendc.compute.api.ComputeClient
 import org.opendc.compute.service.driver.Host
 import org.opendc.compute.service.internal.ComputeServiceImpl
-- 
cgit v1.2.3


From 5b0eaf76ec00192c755b268b7655f6463c5bc62f Mon Sep 17 00:00:00 2001
From: Fabian Mastenbroek <mail.fabianm@gmail.com>
Date: Sat, 27 Mar 2021 12:33:13 +0100
Subject: compute: Migrate compute service simulator to OpenTelemetry

This change updates the compute service simulator to use OpenTelemetry
for reporting metrics of the (simulated) hosts as opposed to using
custom event flows.

This approach is more generic, flexible and possibly offers better
performance as we can collect metrics of all services in a single sweep,
as opposed to listening to several services and each invoking the
handlers.
---
 .../org/opendc/compute/service/driver/Host.kt      |  6 --
 .../org/opendc/compute/service/driver/HostEvent.kt | 72 ----------------------
 2 files changed, 78 deletions(-)
 delete mode 100644 simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/driver/HostEvent.kt

(limited to 'simulator/opendc-compute/opendc-compute-service/src/main')

diff --git a/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/driver/Host.kt b/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/driver/Host.kt
index c3c39572..bed15dfd 100644
--- a/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/driver/Host.kt
+++ b/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/driver/Host.kt
@@ -22,7 +22,6 @@
 
 package org.opendc.compute.service.driver
 
-import kotlinx.coroutines.flow.Flow
 import org.opendc.compute.api.Server
 import java.util.*
 
@@ -55,11 +54,6 @@ public interface Host {
      */
     public val meta: Map<String, Any>
 
-    /**
-     * The events emitted by the driver.
-     */
-    public val events: Flow<HostEvent>
-
     /**
      * Determine whether the specified [instance][server] can still fit on this host.
      */
diff --git a/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/driver/HostEvent.kt b/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/driver/HostEvent.kt
deleted file mode 100644
index 97350679..00000000
--- a/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/driver/HostEvent.kt
+++ /dev/null
@@ -1,72 +0,0 @@
-/*
- * 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.service.driver
-
-/**
- * An event that is emitted by a [Host].
- */
-public sealed class HostEvent {
-    /**
-     * The driver that emitted the event.
-     */
-    public abstract val driver: Host
-
-    /**
-     * This event is emitted when the number of active servers on the server managed by this driver is updated.
-     *
-     * @property driver The driver that emitted the event.
-     * @property numberOfActiveServers The number of active servers.
-     * @property availableMemory The available memory, in MB.
-     */
-    public data class VmsUpdated(
-        override val driver: Host,
-        public val numberOfActiveServers: Int,
-        public val availableMemory: Long
-    ) : HostEvent()
-
-    /**
-     * This event is emitted when a slice is finished.
-     *
-     * @property driver The driver that emitted the event.
-     * @property requestedBurst The total requested CPU time (can be above capacity).
-     * @property grantedBurst The actual total granted capacity, which might be lower than the requested burst due to
-     * the hypervisor being interrupted during a slice.
-     * @property overcommissionedBurst The CPU time that the hypervisor could not grant to the virtual machine since
-     * it did not have the capacity.
-     * @property interferedBurst The sum of CPU time that virtual machines could not utilize due to performance
-     * interference.
-     * @property cpuUsage CPU use in megahertz.
-     * @property cpuDemand CPU demand in megahertz.
-     * @property numberOfDeployedImages The number of images deployed on this hypervisor.
-     */
-    public data class SliceFinished(
-        override val driver: Host,
-        public val requestedBurst: Long,
-        public val grantedBurst: Long,
-        public val overcommissionedBurst: Long,
-        public val interferedBurst: Long,
-        public val cpuUsage: Double,
-        public val cpuDemand: Double,
-        public val numberOfDeployedImages: Int,
-    ) : HostEvent()
-}
-- 
cgit v1.2.3