compute: Implement filter scheduler

This pull request implements the filter scheduler modeled after the scheduler 
from [OpenStack](https://docs.openstack.org/nova/latest/user/filter-scheduler.html). 
The scheduler is functionally equivalent to the old allocation policies, but is more 
flexible and allows policies to be combined.

* A new interface, `ComputeScheduler` is introduced, which is used by the 
  `ComputeServiceImpl` to pick hosts to schedule on.
* `FilterScheduler` is implemented, which works by filtering and weighing the available hosts.

**Breaking API Changes**

* Removal of the `AllocationPolicy` interface and its implementations. 
   Users should migrate to the filter scheduler which offers the same functionality and more.

2 files changed, 11 insertions, 222 deletions

diff --git a/simulator/opendc-compute/opendc-compute-service/src/test/kotlin/org/opendc/compute/service/ComputeServiceTest.kt b/simulator/opendc-compute/opendc-compute-service/src/test/kotlin/org/opendc/compute/service/ComputeServiceTest.kt
index 45a306aa..c6e24346 100644
--- a/simulator/opendc-compute/opendc-compute-service/src/test/kotlin/org/opendc/compute/service/ComputeServiceTest.kt
+++ b/simulator/opendc-compute/opendc-compute-service/src/test/kotlin/org/opendc/compute/service/ComputeServiceTest.kt
@@ -39,7 +39,10 @@ import org.opendc.compute.service.driver.Host
 import org.opendc.compute.service.driver.HostListener
 import org.opendc.compute.service.driver.HostModel
 import org.opendc.compute.service.driver.HostState
-import org.opendc.compute.service.scheduler.AvailableMemoryAllocationPolicy
+import org.opendc.compute.service.scheduler.FilterScheduler
+import org.opendc.compute.service.scheduler.filters.ComputeCapabilitiesFilter
+import org.opendc.compute.service.scheduler.filters.ComputeFilter
+import org.opendc.compute.service.scheduler.weights.MemoryWeigher
 import org.opendc.simulator.utils.DelayControllerClockAdapter
 import java.util.*
 
@@ -55,9 +58,12 @@ internal class ComputeServiceTest {
     fun setUp() {
         scope = TestCoroutineScope()
         val clock = DelayControllerClockAdapter(scope)
-        val policy = AvailableMemoryAllocationPolicy()
+        val computeScheduler = FilterScheduler(
+            filters = listOf(ComputeFilter(), ComputeCapabilitiesFilter()),
+            weighers = listOf(MemoryWeigher() to -1.0)
+        )
         val meter = MeterProvider.noop().get("opendc-compute")
-        service = ComputeService(scope.coroutineContext, clock, meter, policy)
+        service = ComputeService(scope.coroutineContext, clock, meter, computeScheduler)
     }
 
     @AfterEach
@@ -257,6 +263,7 @@ internal class ComputeServiceTest {
         server.start()
         delay(5 * 60 * 1000)
 
+        every { host.state } returns HostState.UP
         listeners.forEach { it.onStateChanged(host, HostState.UP) }
 
         delay(5 * 60 * 1000)
@@ -286,6 +293,7 @@ internal class ComputeServiceTest {
 
         delay(5 * 60 * 1000)
 
+        every { host.state } returns HostState.DOWN
         listeners.forEach { it.onStateChanged(host, HostState.DOWN) }
 
         server.start()
diff --git a/simulator/opendc-compute/opendc-compute-service/src/test/kotlin/org/opendc/compute/service/scheduler/AllocationPolicyTest.kt b/simulator/opendc-compute/opendc-compute-service/src/test/kotlin/org/opendc/compute/service/scheduler/AllocationPolicyTest.kt
deleted file mode 100644
index db377914..00000000
--- a/simulator/opendc-compute/opendc-compute-service/src/test/kotlin/org/opendc/compute/service/scheduler/AllocationPolicyTest.kt
+++ /dev/null
@@ -1,219 +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.scheduler
-
-import io.mockk.every
-import io.mockk.mockk
-import org.junit.jupiter.api.Assertions.assertEquals
-import org.junit.jupiter.params.ParameterizedTest
-import org.junit.jupiter.params.provider.Arguments
-import org.junit.jupiter.params.provider.MethodSource
-import org.opendc.compute.api.Server
-import org.opendc.compute.service.internal.HostView
-import java.util.*
-import java.util.stream.Stream
-import kotlin.random.Random
-
-/**
- * Test suite for the [AllocationPolicy] interface.
- */
-internal class AllocationPolicyTest {
-    @ParameterizedTest
-    @MethodSource("activeServersArgs")
-    fun testActiveServersPolicy(
-        reversed: Boolean,
-        hosts: Set<HostView>,
-        server: Server,
-        expectedHost: HostView?
-    ) {
-        val policy = NumberOfActiveServersAllocationPolicy(reversed)
-        assertEquals(expectedHost, policy.invoke().select(hosts, server))
-    }
-
-    @ParameterizedTest
-    @MethodSource("availableMemoryArgs")
-    fun testAvailableMemoryPolicy(
-        reversed: Boolean,
-        hosts: Set<HostView>,
-        server: Server,
-        expectedHost: HostView?
-    ) {
-        val policy = AvailableMemoryAllocationPolicy(reversed)
-        assertEquals(expectedHost, policy.invoke().select(hosts, server))
-    }
-
-    @ParameterizedTest
-    @MethodSource("availableCoreMemoryArgs")
-    fun testAvailableCoreMemoryPolicy(
-        reversed: Boolean,
-        hosts: Set<HostView>,
-        server: Server,
-        expectedHost: HostView?
-    ) {
-        val policy = AvailableMemoryAllocationPolicy(reversed)
-        assertEquals(expectedHost, policy.invoke().select(hosts, server))
-    }
-
-    @ParameterizedTest
-    @MethodSource("provisionedCoresArgs")
-    fun testProvisionedPolicy(
-        reversed: Boolean,
-        hosts: Set<HostView>,
-        server: Server,
-        expectedHost: HostView?
-    ) {
-        val policy = ProvisionedCoresAllocationPolicy(reversed)
-        assertEquals(expectedHost, policy.invoke().select(hosts, server))
-    }
-
-    @Suppress("unused")
-    private companion object {
-        /**
-         * Test arguments for the [NumberOfActiveServersAllocationPolicy].
-         */
-        @JvmStatic
-        fun activeServersArgs(): Stream<Arguments> {
-            val random = Random(1)
-            val hosts = List(4) { i ->
-                val view = mockk<HostView>()
-                every { view.host.uid } returns UUID(0, i.toLong())
-                every { view.host.model.cpuCount } returns random.nextInt(1, 16)
-                every { view.host.model.memorySize } returns random.nextLong(1024, 1024 * 1024)
-                every { view.availableMemory } returns random.nextLong(0, view.host.model.memorySize)
-                every { view.numberOfActiveServers } returns random.nextInt(0, 6)
-                every { view.provisionedCores } returns random.nextInt(0, view.host.model.cpuCount)
-                every { view.toString() } returns "HostView[$i,numberOfActiveServers=${view.numberOfActiveServers}]"
-                view
-            }
-
-            val servers = List(2) {
-                val server = mockk<Server>()
-                every { server.flavor.cpuCount } returns random.nextInt(1, 8)
-                every { server.flavor.memorySize } returns random.nextLong(1024, 1024 * 512)
-                server
-            }
-
-            return Stream.of(
-                Arguments.of(false, hosts.toSet(), servers[0], hosts[2]),
-                Arguments.of(false, hosts.toSet(), servers[1], hosts[1]),
-                Arguments.of(true, hosts.toSet(), servers[1], hosts[0]),
-            )
-        }
-
-        /**
-         * Test arguments for the [AvailableCoreMemoryAllocationPolicy].
-         */
-        @JvmStatic
-        fun availableCoreMemoryArgs(): Stream<Arguments> {
-            val random = Random(1)
-            val hosts = List(4) { i ->
-                val view = mockk<HostView>()
-                every { view.host.uid } returns UUID(0, i.toLong())
-                every { view.host.model.cpuCount } returns random.nextInt(1, 16)
-                every { view.host.model.memorySize } returns random.nextLong(1024, 1024 * 1024)
-                every { view.availableMemory } returns random.nextLong(0, view.host.model.memorySize)
-                every { view.numberOfActiveServers } returns random.nextInt(0, 6)
-                every { view.provisionedCores } returns random.nextInt(0, view.host.model.cpuCount)
-                every { view.toString() } returns "HostView[$i,availableMemory=${view.availableMemory}]"
-                view
-            }
-
-            val servers = List(2) {
-                val server = mockk<Server>()
-                every { server.flavor.cpuCount } returns random.nextInt(1, 8)
-                every { server.flavor.memorySize } returns random.nextLong(1024, 1024 * 512)
-                server
-            }
-
-            return Stream.of(
-                Arguments.of(false, hosts.toSet(), servers[0], hosts[2]),
-                Arguments.of(false, hosts.toSet(), servers[1], hosts[2]),
-                Arguments.of(true, hosts.toSet(), servers[1], hosts[1]),
-            )
-        }
-
-        /**
-         * Test arguments for the [AvailableMemoryAllocationPolicy].
-         */
-        @JvmStatic
-        fun availableMemoryArgs(): Stream<Arguments> {
-            val random = Random(1)
-            val hosts = List(4) { i ->
-                val view = mockk<HostView>()
-                every { view.host.uid } returns UUID(0, i.toLong())
-                every { view.host.model.cpuCount } returns random.nextInt(1, 16)
-                every { view.host.model.memorySize } returns random.nextLong(1024, 1024 * 1024)
-                every { view.availableMemory } returns random.nextLong(0, view.host.model.memorySize)
-                every { view.numberOfActiveServers } returns random.nextInt(0, 6)
-                every { view.provisionedCores } returns random.nextInt(0, view.host.model.cpuCount)
-                every { view.toString() } returns "HostView[$i,availableMemory=${view.availableMemory}]"
-                view
-            }
-
-            val servers = List(2) {
-                val server = mockk<Server>()
-                every { server.flavor.cpuCount } returns random.nextInt(1, 8)
-                every { server.flavor.memorySize } returns random.nextLong(1024, 1024 * 512)
-                server
-            }
-
-            return Stream.of(
-                Arguments.of(false, hosts.toSet(), servers[0], hosts[2]),
-                Arguments.of(false, hosts.toSet(), servers[1], hosts[2]),
-                Arguments.of(true, hosts.toSet(), servers[1], hosts[1]),
-            )
-        }
-
-        /**
-         * Test arguments for the [ProvisionedCoresAllocationPolicy].
-         */
-        @JvmStatic
-        fun provisionedCoresArgs(): Stream<Arguments> {
-            val random = Random(1)
-            val hosts = List(4) { i ->
-                val view = mockk<HostView>()
-                every { view.host.uid } returns UUID(0, i.toLong())
-                every { view.host.model.cpuCount } returns random.nextInt(1, 16)
-                every { view.host.model.memorySize } returns random.nextLong(1024, 1024 * 1024)
-                every { view.availableMemory } returns random.nextLong(0, view.host.model.memorySize)
-                every { view.numberOfActiveServers } returns random.nextInt(0, 6)
-                every { view.provisionedCores } returns random.nextInt(0, view.host.model.cpuCount)
-                every { view.toString() } returns "HostView[$i,provisionedCores=${view.provisionedCores}]"
-                view
-            }
-
-            val servers = List(2) {
-                val server = mockk<Server>()
-                every { server.flavor.cpuCount } returns random.nextInt(1, 8)
-                every { server.flavor.memorySize } returns random.nextLong(1024, 1024 * 512)
-                server
-            }
-
-            return Stream.of(
-                Arguments.of(false, hosts.toSet(), servers[0], hosts[2]),
-                Arguments.of(false, hosts.toSet(), servers[1], hosts[0]),
-                Arguments.of(true, hosts.toSet(), servers[1], hosts[0]),
-            )
-        }
-    }
-}