diff options
Diffstat (limited to 'simulator/opendc-compute/opendc-compute-service/src')
9 files changed, 611 insertions, 110 deletions
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 2cd91144..c3c39572 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 @@ -51,6 +51,11 @@ public interface Host { public val state: HostState /** + * Meta-data associated with the host. + */ + public val meta: Map<String, Any> + + /** * The events emitted by the driver. */ public val events: Flow<HostEvent> @@ -74,7 +79,7 @@ public interface Host { public operator fun contains(server: Server): Boolean /** - * Stat the server [instance][server] if it is currently not running on this host. + * Start the server [instance][server] if it is currently not running on this host. * * @throws IllegalArgumentException if the server is not present on the host. */ @@ -88,9 +93,9 @@ public interface Host { public suspend fun stop(server: Server) /** - * Terminate the specified [instance][server] on this host and cleanup all resources associated with it. + * Delete the specified [instance][server] on this host and cleanup all resources associated with it. */ - public suspend fun terminate(server: Server) + public suspend fun delete(server: Server) /** * Add a [HostListener] to this host. 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 new file mode 100644 index 00000000..29f10e27 --- /dev/null +++ b/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/internal/ClientFlavor.kt @@ -0,0 +1,62 @@ +/* + * 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.internal + +import org.opendc.compute.api.Flavor +import java.util.UUID + +/** + * A [Flavor] implementation that is passed to clients but delegates its implementation to another class. + */ +internal class ClientFlavor(private val delegate: Flavor) : Flavor { + override val uid: UUID = delegate.uid + + override var name: String = delegate.name + private set + + override var cpuCount: Int = delegate.cpuCount + private set + + override var memorySize: Long = delegate.memorySize + private set + + override var labels: Map<String, String> = delegate.labels.toMap() + private set + + override var meta: Map<String, Any> = delegate.meta.toMap() + private set + + override suspend fun delete() { + delegate.delete() + } + + override suspend fun refresh() { + delegate.refresh() + + name = delegate.name + cpuCount = delegate.cpuCount + memorySize = delegate.memorySize + labels = delegate.labels + meta = delegate.meta + } +} 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 new file mode 100644 index 00000000..6c5b2ab0 --- /dev/null +++ b/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/internal/ClientImage.kt @@ -0,0 +1,55 @@ +/* + * 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.internal + +import org.opendc.compute.api.Image +import java.util.* + +/** + * An [Image] implementation that is passed to clients but delegates its implementation to another class. + */ +internal class ClientImage(private val delegate: Image) : Image { + override val uid: UUID = delegate.uid + + override var name: String = delegate.name + private set + + override var labels: Map<String, String> = delegate.labels.toMap() + private set + + override var meta: Map<String, Any> = delegate.meta.toMap() + private set + + override suspend fun delete() { + delegate.delete() + refresh() + } + + override suspend fun refresh() { + delegate.refresh() + + name = delegate.name + labels = delegate.labels + meta = delegate.meta + } +} 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 f84b7435..ae4cee3b 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 @@ -46,12 +46,30 @@ internal class ClientServer(private val delegate: Server) : Server, ServerWatche override var image: Image = delegate.image private set - override var tags: Map<String, String> = delegate.tags.toMap() + override var labels: Map<String, String> = delegate.labels.toMap() + private set + + override var meta: Map<String, Any> = delegate.meta.toMap() private set override var state: ServerState = delegate.state private set + override suspend fun start() { + delegate.start() + refresh() + } + + override suspend fun stop() { + delegate.stop() + refresh() + } + + override suspend fun delete() { + delegate.delete() + refresh() + } + override fun watch(watcher: ServerWatcher) { if (watchers.isEmpty()) { delegate.watch(this) @@ -69,10 +87,13 @@ internal class ClientServer(private val delegate: Server) : Server, ServerWatche } override suspend fun refresh() { + delegate.refresh() + name = delegate.name flavor = delegate.flavor image = delegate.image - tags = delegate.tags + labels = delegate.labels + meta = delegate.meta state = delegate.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 69d6bb59..2c38f7cb 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,11 +22,8 @@ package org.opendc.compute.service.internal -import kotlinx.coroutines.CoroutineScope -import kotlinx.coroutines.cancel +import kotlinx.coroutines.* import kotlinx.coroutines.flow.Flow -import kotlinx.coroutines.launch -import kotlinx.coroutines.suspendCancellableCoroutine import mu.KotlinLogging import org.opendc.compute.api.* import org.opendc.compute.service.ComputeService @@ -41,9 +38,7 @@ import org.opendc.utils.TimerScheduler import org.opendc.utils.flow.EventFlow import java.time.Clock import java.util.* -import kotlin.coroutines.Continuation import kotlin.coroutines.CoroutineContext -import kotlin.coroutines.resume import kotlin.math.max /** @@ -87,12 +82,27 @@ public class ComputeServiceImpl( /** * The servers that should be launched by the service. */ - private val queue: Deque<LaunchRequest> = ArrayDeque() + private val queue: Deque<SchedulingRequest> = ArrayDeque() /** * The active servers in the system. */ - private val activeServers: MutableSet<Server> = mutableSetOf() + private val activeServers: MutableMap<Server, Host> = mutableMapOf() + + /** + * The registered flavors for this compute service. + */ + internal val flavors = mutableMapOf<UUID, InternalFlavor>() + + /** + * The registered images for this compute service. + */ + internal val images = mutableMapOf<UUID, InternalImage>() + + /** + * The registered servers for this compute service. + */ + private val servers = mutableMapOf<UUID, InternalServer>() public var submittedVms: Int = 0 public var queuedVms: Int = 0 @@ -126,7 +136,74 @@ public class ComputeServiceImpl( override fun newClient(): ComputeClient = object : ComputeClient { private var isClosed: Boolean = false - override suspend fun newServer(name: String, image: Image, flavor: Flavor): Server { + override suspend fun queryFlavors(): List<Flavor> { + check(!isClosed) { "Client is already closed" } + + return flavors.values.map { ClientFlavor(it) } + } + + override suspend fun findFlavor(id: UUID): Flavor? { + check(!isClosed) { "Client is already closed" } + + 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 + ) + + flavors[uid] = flavor + + return ClientFlavor(flavor) + } + + override suspend fun queryImages(): List<Image> { + check(!isClosed) { "Client is already closed" } + + return images.values.map { ClientImage(it) } + } + + override suspend fun findImage(id: UUID): Image? { + check(!isClosed) { "Client is already closed" } + + 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" } + + val uid = UUID(clock.millis(), random.nextLong()) + val image = InternalImage(this@ComputeServiceImpl, uid, name, labels, meta) + + images[uid] = 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)) @@ -143,11 +220,36 @@ public class ComputeServiceImpl( ) ) - return suspendCancellableCoroutine { cont -> - val request = LaunchRequest(createServer(name, image, flavor), cont) - queue += request - requestCycle() + val uid = UUID(clock.millis(), random.nextLong()) + val server = InternalServer( + this@ComputeServiceImpl, + uid, + name, + flavor, + image, + labels.toMutableMap(), + meta.toMutableMap() + ) + + servers[uid] = server + + if (start) { + server.start() } + + return ClientServer(server) + } + + override suspend fun findServer(id: UUID): Server? { + check(!isClosed) { "Client is already closed" } + + return servers[id]?.let { ClientServer(it) } + } + + override suspend fun queryServers(): List<Server> { + check(!isClosed) { "Client is already closed" } + + return servers.values.map { ClientServer(it) } } override fun close() { @@ -183,22 +285,31 @@ public class ComputeServiceImpl( scope.cancel() } - private fun createServer( - name: String, - image: Image, - flavor: Flavor - ): Server { - return ServerImpl( - uid = UUID(random.nextLong(), random.nextLong()), - name = name, - flavor = flavor, - image = image - ) + internal fun schedule(server: InternalServer) { + logger.debug { "Enqueueing server ${server.uid} to be assigned to host." } + + queue.add(SchedulingRequest(server)) + requestSchedulingCycle() + } + + internal fun delete(flavor: InternalFlavor) { + checkNotNull(flavors.remove(flavor.uid)) { "Flavor was not known" } } - private fun requestCycle() { - // Bail out in case we have already requested a new cycle. - if (scheduler.isTimerActive(Unit)) { + internal fun delete(image: InternalImage) { + checkNotNull(images.remove(image.uid)) { "Image was not known" } + } + + internal fun delete(server: InternalServer) { + checkNotNull(servers.remove(server.uid)) { "Server was not known" } + } + + /** + * Indicate that a new scheduling cycle is needed due to a change to the service's state. + */ + private fun requestSchedulingCycle() { + // Bail out in case we have already requested a new cycle or the queue is empty. + if (scheduler.isTimerActive(Unit) || queue.isEmpty()) { return } @@ -208,20 +319,28 @@ public class ComputeServiceImpl( val delay = schedulingQuantum - (clock.millis() % schedulingQuantum) scheduler.startSingleTimer(Unit, delay) { - schedule() + doSchedule() } } - private fun schedule() { + /** + * Run a single scheduling iteration. + */ + private fun doSchedule() { while (queue.isNotEmpty()) { - val (server, cont) = queue.peekFirst() - val requiredMemory = server.flavor.memorySize - val selectedHv = allocationLogic.select(availableHosts, server) + val request = queue.peek() - if (selectedHv == null || !selectedHv.host.canFit(server)) { + if (request.isCancelled) { + queue.poll() + continue + } + + 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." } - if (requiredMemory > maxMemory || server.flavor.cpuCount > maxCores) { + if (server.flavor.memorySize > maxMemory || server.flavor.cpuCount > maxCores) { tracer.commit(VmSubmissionInvalidEvent(server.name)) _events.emit( @@ -247,45 +366,62 @@ public class ComputeServiceImpl( } } - logger.info { "[${clock.millis()}] Spawning $server on ${selectedHv.host.uid} ${selectedHv.host.name} ${selectedHv.host.model}" } - queue.poll() - - // Speculatively update the hypervisor view information to prevent other images in the queue from - // deciding on stale values. - selectedHv.numberOfActiveServers++ - selectedHv.provisionedCores += server.flavor.cpuCount - selectedHv.availableMemory -= requiredMemory // XXX Temporary hack + val host = hv.host - scope.launch { - try { - cont.resume(ClientServer(server)) - selectedHv.host.spawn(server) - activeServers += server + // Remove request from queue + queue.poll() - tracer.commit(VmScheduledEvent(server.name)) - _events.emit( - ComputeServiceEvent.MetricsAvailable( - this@ComputeServiceImpl, - hostCount, - availableHosts.size, - submittedVms, - ++runningVms, - finishedVms, - --queuedVms, - unscheduledVms + 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 + ) ) - ) - } catch (e: Throwable) { - logger.error("Failed to deploy VM", e) + } catch (e: Throwable) { + logger.error("Failed to deploy VM", e) - selectedHv.numberOfActiveServers-- - selectedHv.provisionedCores -= server.flavor.cpuCount - selectedHv.availableMemory += requiredMemory + 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. " } } } } + /** + * A request to schedule an [InternalServer] onto one of the [Host]s. + */ + private data class SchedulingRequest(val server: InternalServer) { + /** + * A flag to indicate that the request is cancelled. + */ + var isCancelled: Boolean = false + } + override fun onStateChanged(host: Host, newState: HostState) { when (newState) { HostState.UP -> { @@ -313,9 +449,7 @@ public class ComputeServiceImpl( ) // Re-schedule on the new machine - if (queue.isNotEmpty()) { - requestCycle() - } + requestSchedulingCycle() } HostState.DOWN -> { logger.debug { "[${clock.millis()}] Host ${host.uid} state changed: $newState" } @@ -338,19 +472,23 @@ public class ComputeServiceImpl( ) ) - if (queue.isNotEmpty()) { - requestCycle() - } + requestSchedulingCycle() } } } override fun onStateChanged(host: Host, server: Server, newState: ServerState) { - val serverImpl = server as ServerImpl - serverImpl.state = newState - serverImpl.watchers.forEach { it.onStateChanged(server, newState) } + require(server is InternalServer) { "Invalid server type passed to service" } - if (newState == ServerState.SHUTOFF) { + if (server.host != host) { + // This can happen when a server is rescheduled and started on another machine, while being deleted from + // the old machine. + return + } + + server.state = newState + + if (newState == ServerState.TERMINATED || newState == ServerState.DELETED) { logger.info { "[${clock.millis()}] Server ${server.uid} ${server.name} ${server.flavor} finished." } tracer.commit(VmStoppedEvent(server.name)) @@ -379,36 +517,7 @@ public class ComputeServiceImpl( } // Try to reschedule if needed - if (queue.isNotEmpty()) { - requestCycle() - } + requestSchedulingCycle() } } - - public data class LaunchRequest(val server: Server, val cont: Continuation<Server>) - - private class ServerImpl( - override val uid: UUID, - override val name: String, - override val flavor: Flavor, - override val image: Image - ) : Server { - val watchers = mutableListOf<ServerWatcher>() - - override fun watch(watcher: ServerWatcher) { - watchers += watcher - } - - override fun unwatch(watcher: ServerWatcher) { - watchers -= watcher - } - - override suspend fun refresh() { - // No-op: this object is the source-of-truth - } - - override val tags: Map<String, String> = emptyMap() - - override var state: ServerState = ServerState.BUILD - } } 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 new file mode 100644 index 00000000..95e280df --- /dev/null +++ b/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/internal/InternalFlavor.kt @@ -0,0 +1,64 @@ +/* + * 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.internal + +import org.opendc.compute.api.Flavor +import java.util.* + +/** + * Internal stateful representation of a [Flavor]. + */ +internal class InternalFlavor( + private val service: ComputeServiceImpl, + override val uid: UUID, + name: String, + cpuCount: Int, + memorySize: Long, + labels: Map<String, String>, + meta: Map<String, Any> +) : Flavor { + override var name: String = name + private set + + override var cpuCount: Int = cpuCount + private set + + override var memorySize: Long = memorySize + private set + + override val labels: MutableMap<String, String> = labels.toMutableMap() + + override val meta: MutableMap<String, Any> = meta.toMutableMap() + + override suspend fun refresh() { + // No-op: this object is the source-of-truth + } + + override suspend fun delete() { + service.delete(this) + } + + override fun equals(other: Any?): Boolean = other is InternalFlavor && uid == other.uid + + override fun hashCode(): Int = uid.hashCode() +} 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 new file mode 100644 index 00000000..86f2f6b9 --- /dev/null +++ b/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/internal/InternalImage.kt @@ -0,0 +1,54 @@ +/* + * 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.internal + +import org.opendc.compute.api.Image +import java.util.* + +/** + * Internal stateful representation of an [Image]. + */ +internal class InternalImage( + private val service: ComputeServiceImpl, + override val uid: UUID, + override val name: String, + labels: Map<String, String>, + meta: Map<String, Any> +) : Image { + + override val labels: MutableMap<String, String> = labels.toMutableMap() + + override val meta: MutableMap<String, Any> = meta.toMutableMap() + + override suspend fun refresh() { + // No-op: this object is the source-of-truth + } + + override suspend fun delete() { + service.delete(this) + } + + override fun equals(other: Any?): Boolean = other is InternalImage && uid == other.uid + + override fun hashCode(): Int = uid.hashCode() +} 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 new file mode 100644 index 00000000..ff7c1d15 --- /dev/null +++ b/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/internal/InternalServer.kt @@ -0,0 +1,131 @@ +/* + * 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.internal + +import mu.KotlinLogging +import org.opendc.compute.api.* +import org.opendc.compute.service.driver.Host +import java.util.UUID + +/** + * Internal implementation of the [Server] interface. + */ +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 labels: MutableMap<String, String>, + override val meta: MutableMap<String, Any> +) : Server { + /** + * The logger instance of this server. + */ + private val logger = KotlinLogging.logger {} + + /** + * The watchers of this server object. + */ + private val watchers = mutableListOf<ServerWatcher>() + + /** + * The [Host] that has been assigned to host the server. + */ + internal var host: Host? = null + + override suspend fun start() { + when (state) { + ServerState.RUNNING -> { + logger.debug { "User tried to start server but server is already running" } + return + } + ServerState.PROVISIONING -> { + logger.debug { "User tried to start server but request is already pending: doing nothing" } + return + } + ServerState.DELETED -> { + logger.warn { "User tried to start terminated server" } + throw IllegalArgumentException("Server is terminated") + } + else -> { + logger.info { "User requested to start server $uid" } + state = ServerState.PROVISIONING + service.schedule(this) + } + } + } + + override suspend fun stop() { + when (state) { + ServerState.PROVISIONING -> {} // TODO Find way to interrupt these + 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") + } + } + + override suspend fun delete() { + when (state) { + ServerState.PROVISIONING -> {} // TODO Find way to interrupt these + ServerState.RUNNING -> { + val host = checkNotNull(host) { "Server not running" } + host.delete(this) + service.delete(this) + } + else -> {} // No work needed + } + } + + override fun watch(watcher: ServerWatcher) { + watchers += watcher + } + + override fun unwatch(watcher: ServerWatcher) { + watchers -= watcher + } + + override suspend fun refresh() { + // No-op: this object is the source-of-truth + } + + override var state: ServerState = ServerState.TERMINATED + set(value) { + if (value != field) { + watchers.forEach { it.onStateChanged(this, value) } + } + + field = value + } + + internal fun assignHost(host: Host) { + this.host = host + } + + override fun equals(other: Any?): Boolean = other is InternalServer && uid == other.uid + + override fun hashCode(): Int = uid.hashCode() +} diff --git a/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/scheduler/RandomAllocationPolicy.kt b/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/scheduler/RandomAllocationPolicy.kt index 3facb182..ac7b351d 100644 --- a/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/scheduler/RandomAllocationPolicy.kt +++ b/simulator/opendc-compute/opendc-compute-service/src/main/kotlin/org/opendc/compute/service/scheduler/RandomAllocationPolicy.kt @@ -38,7 +38,7 @@ public class RandomAllocationPolicy(private val random: Random = Random(0)) : Al ): HostView? { return hypervisors.asIterable() .filter { hv -> - val fitsMemory = hv.availableMemory >= (server.image.tags["required-memory"] as Long) + val fitsMemory = hv.availableMemory >= (server.image.meta["required-memory"] as Long) val fitsCpu = hv.host.model.cpuCount >= server.flavor.cpuCount fitsMemory && fitsCpu } |