merge: Add provisioning tool for experiments (#104)

This pull request implements a new tool to help provision and manage the 
experimental environment.

## Implementation Notes :hammer_and_pick:

* Add service registry for cloud services 
* Add provisioning tool for experiments
* Add provisioning step for workflow service 
* Add provisioners for FaaS service 
* Use experiment base for Capelin experiments
* Use experiment base for web runner
* Integrate compute workload classes
* Remove Topology interface 

## Breaking API Changes :warning:

* Removal of the `opendc-compute-workload`, `opendc-faas-workload`, 
   and `opendc-workflow-workload` modules. These are now located 
   inside `opendc-experiments`
* Removal of `ComputeServiceHelper`. Use `Provisioner` to provision 
   a `ComputeService`.

13 files changed, 293 insertions, 243 deletions

diff --git a/opendc-experiments/opendc-experiments-capelin/src/jmh/kotlin/org/opendc/experiments/capelin/CapelinBenchmarks.kt b/opendc-experiments/opendc-experiments-capelin/src/jmh/kotlin/org/opendc/experiments/capelin/CapelinBenchmarks.kt
index c09ce96a..f021e223 100644
--- a/opendc-experiments/opendc-experiments-capelin/src/jmh/kotlin/org/opendc/experiments/capelin/CapelinBenchmarks.kt
+++ b/opendc-experiments/opendc-experiments-capelin/src/jmh/kotlin/org/opendc/experiments/capelin/CapelinBenchmarks.kt
@@ -22,15 +22,16 @@
 
 package org.opendc.experiments.capelin
 
+import org.opendc.compute.service.ComputeService
 import org.opendc.compute.service.scheduler.FilterScheduler
 import org.opendc.compute.service.scheduler.filters.ComputeFilter
 import org.opendc.compute.service.scheduler.filters.RamFilter
 import org.opendc.compute.service.scheduler.filters.VCpuFilter
 import org.opendc.compute.service.scheduler.weights.CoreRamWeigher
-import org.opendc.compute.workload.*
-import org.opendc.compute.workload.topology.Topology
-import org.opendc.compute.workload.topology.apply
 import org.opendc.experiments.capelin.topology.clusterTopology
+import org.opendc.experiments.compute.*
+import org.opendc.experiments.compute.topology.HostSpec
+import org.opendc.experiments.provisioner.Provisioner
 import org.opendc.simulator.core.runBlockingSimulation
 import org.openjdk.jmh.annotations.*
 import java.io.File
@@ -46,7 +47,7 @@ import java.util.concurrent.TimeUnit
 @Measurement(iterations = 5, time = 5, timeUnit = TimeUnit.SECONDS)
 class CapelinBenchmarks {
     private lateinit var vms: List<VirtualMachine>
-    private lateinit var topology: Topology
+    private lateinit var topology: List<HostSpec>
 
     @Param("true", "false")
     private var isOptimized: Boolean = false
@@ -54,29 +55,27 @@ class CapelinBenchmarks {
     @Setup
     fun setUp() {
         val loader = ComputeWorkloadLoader(File("src/test/resources/trace"))
-        val source = trace("bitbrains-small")
         vms = trace("bitbrains-small").resolve(loader, Random(1L))
         topology = checkNotNull(object {}.javaClass.getResourceAsStream("/topology.txt")).use { clusterTopology(it) }
     }
 
     @Benchmark
     fun benchmarkCapelin() = runBlockingSimulation {
-        val computeScheduler = FilterScheduler(
-            filters = listOf(ComputeFilter(), VCpuFilter(16.0), RamFilter(1.0)),
-            weighers = listOf(CoreRamWeigher(multiplier = 1.0))
-        )
-        val runner = ComputeServiceHelper(
-            coroutineContext,
-            clock,
-            computeScheduler,
-            seed = 0L
-        )
+        val serviceDomain = "compute.opendc.org"
 
-        try {
-            runner.apply(topology, isOptimized)
-            runner.run(vms, interference = true)
-        } finally {
-            runner.close()
+        Provisioner(coroutineContext, clock, seed = 0).use { provisioner ->
+            val computeScheduler = FilterScheduler(
+                filters = listOf(ComputeFilter(), VCpuFilter(16.0), RamFilter(1.0)),
+                weighers = listOf(CoreRamWeigher(multiplier = 1.0))
+            )
+
+            provisioner.runSteps(
+                setupComputeService(serviceDomain, { computeScheduler }),
+                setupHosts(serviceDomain, topology, optimize = isOptimized)
+            )
+
+            val service = provisioner.registry.resolve(serviceDomain, ComputeService::class.java)!!
+            service.replay(clock, vms, 0L, interference = true)
         }
     }
 }
diff --git a/opendc-experiments/opendc-experiments-capelin/src/main/kotlin/org/opendc/experiments/capelin/CapelinRunner.kt b/opendc-experiments/opendc-experiments-capelin/src/main/kotlin/org/opendc/experiments/capelin/CapelinRunner.kt
index dbb5ced3..f1214b08 100644
--- a/opendc-experiments/opendc-experiments-capelin/src/main/kotlin/org/opendc/experiments/capelin/CapelinRunner.kt
+++ b/opendc-experiments/opendc-experiments-capelin/src/main/kotlin/org/opendc/experiments/capelin/CapelinRunner.kt
@@ -22,15 +22,12 @@
 
 package org.opendc.experiments.capelin
 
-import org.opendc.compute.workload.ComputeServiceHelper
-import org.opendc.compute.workload.ComputeWorkloadLoader
-import org.opendc.compute.workload.createComputeScheduler
-import org.opendc.compute.workload.export.parquet.ParquetComputeMonitor
-import org.opendc.compute.workload.grid5000
-import org.opendc.compute.workload.telemetry.ComputeMetricReader
-import org.opendc.compute.workload.topology.apply
+import org.opendc.compute.service.ComputeService
 import org.opendc.experiments.capelin.model.Scenario
 import org.opendc.experiments.capelin.topology.clusterTopology
+import org.opendc.experiments.compute.*
+import org.opendc.experiments.compute.export.parquet.ParquetComputeMonitor
+import org.opendc.experiments.provisioner.Provisioner
 import org.opendc.simulator.core.runBlockingSimulation
 import java.io.File
 import java.time.Duration
@@ -58,54 +55,41 @@ public class CapelinRunner(
      * Run a single [scenario] with the specified seed.
      */
     fun runScenario(scenario: Scenario, seed: Long) = runBlockingSimulation {
-        val seeder = Random(seed)
-
-        val operationalPhenomena = scenario.operationalPhenomena
-        val computeScheduler = createComputeScheduler(scenario.allocationPolicy, seeder)
-        val failureModel =
-            if (operationalPhenomena.failureFrequency > 0)
-                grid5000(Duration.ofSeconds((operationalPhenomena.failureFrequency * 60).roundToLong()))
-            else
-                null
-        val vms = scenario.workload.source.resolve(workloadLoader, seeder)
-        val runner = ComputeServiceHelper(
-            coroutineContext,
-            clock,
-            computeScheduler,
-            seed,
-        )
-
+        val serviceDomain = "compute.opendc.org"
         val topology = clusterTopology(File(envPath, "${scenario.topology.name}.txt"))
 
-        val partitions = scenario.partitions + ("seed" to seed.toString())
-        val partition = partitions.map { (k, v) -> "$k=$v" }.joinToString("/")
-        val exporter = if (outputPath != null) {
-            ComputeMetricReader(
-                this,
-                clock,
-                runner.service,
-                ParquetComputeMonitor(
-                    outputPath,
-                    partition,
-                    bufferSize = 4096
-                ),
-                exportInterval = Duration.ofMinutes(5)
+        Provisioner(coroutineContext, clock, seed).use { provisioner ->
+            provisioner.runSteps(
+                setupComputeService(serviceDomain, { createComputeScheduler(scenario.allocationPolicy, Random(it.seeder.nextLong())) }),
+                setupHosts(serviceDomain, topology, optimize = true)
             )
-        } else {
-            null
-        }
 
-        try {
-            // Instantiate the desired topology
-            runner.apply(topology, optimize = true)
+            if (outputPath != null) {
+                val partitions = scenario.partitions + ("seed" to seed.toString())
+                val partition = partitions.map { (k, v) -> "$k=$v" }.joinToString("/")
+
+                provisioner.runStep(
+                    registerComputeMonitor(
+                        serviceDomain,
+                        ParquetComputeMonitor(
+                            outputPath,
+                            partition,
+                            bufferSize = 4096
+                        )
+                    )
+                )
+            }
 
-            // Run the workload trace
-            runner.run(vms, failureModel = failureModel, interference = operationalPhenomena.hasInterference)
+            val service = provisioner.registry.resolve(serviceDomain, ComputeService::class.java)!!
+            val vms = scenario.workload.source.resolve(workloadLoader, Random(seed))
+            val operationalPhenomena = scenario.operationalPhenomena
+            val failureModel =
+                if (operationalPhenomena.failureFrequency > 0)
+                    grid5000(Duration.ofSeconds((operationalPhenomena.failureFrequency * 60).roundToLong()))
+                else
+                    null
 
-            // Stop the metric collection
-            exporter?.close()
-        } finally {
-            runner.close()
+            service.replay(clock, vms, seed, failureModel = failureModel, interference = operationalPhenomena.hasInterference)
         }
     }
 }
diff --git a/opendc-experiments/opendc-experiments-capelin/src/main/kotlin/org/opendc/experiments/capelin/model/Topology.kt b/opendc-experiments/opendc-experiments-capelin/src/main/kotlin/org/opendc/experiments/capelin/model/Topology.kt
index fe16a294..c90194ce 100644
--- a/opendc-experiments/opendc-experiments-capelin/src/main/kotlin/org/opendc/experiments/capelin/model/Topology.kt
+++ b/opendc-experiments/opendc-experiments-capelin/src/main/kotlin/org/opendc/experiments/capelin/model/Topology.kt
@@ -23,6 +23,6 @@
 package org.opendc.experiments.capelin.model
 
 /**
- * The topology topology on which we test the workload.
+ * The topology on which we simulate the workload.
  */
 public data class Topology(val name: String)
diff --git a/opendc-experiments/opendc-experiments-capelin/src/main/kotlin/org/opendc/experiments/capelin/model/Workload.kt b/opendc-experiments/opendc-experiments-capelin/src/main/kotlin/org/opendc/experiments/capelin/model/Workload.kt
index a2e71243..ed2588f0 100644
--- a/opendc-experiments/opendc-experiments-capelin/src/main/kotlin/org/opendc/experiments/capelin/model/Workload.kt
+++ b/opendc-experiments/opendc-experiments-capelin/src/main/kotlin/org/opendc/experiments/capelin/model/Workload.kt
@@ -22,7 +22,7 @@
 
 package org.opendc.experiments.capelin.model
 
-import org.opendc.compute.workload.ComputeWorkload
+import org.opendc.experiments.compute.ComputeWorkload
 
 /**
  * A single workload originating from a trace.
diff --git a/opendc-experiments/opendc-experiments-capelin/src/main/kotlin/org/opendc/experiments/capelin/portfolio/CompositeWorkloadPortfolio.kt b/opendc-experiments/opendc-experiments-capelin/src/main/kotlin/org/opendc/experiments/capelin/portfolio/CompositeWorkloadPortfolio.kt
index 68eb15b3..80b8859c 100644
--- a/opendc-experiments/opendc-experiments-capelin/src/main/kotlin/org/opendc/experiments/capelin/portfolio/CompositeWorkloadPortfolio.kt
+++ b/opendc-experiments/opendc-experiments-capelin/src/main/kotlin/org/opendc/experiments/capelin/portfolio/CompositeWorkloadPortfolio.kt
@@ -22,12 +22,12 @@
 
 package org.opendc.experiments.capelin.portfolio
 
-import org.opendc.compute.workload.composite
-import org.opendc.compute.workload.trace
 import org.opendc.experiments.capelin.model.OperationalPhenomena
 import org.opendc.experiments.capelin.model.Scenario
 import org.opendc.experiments.capelin.model.Topology
 import org.opendc.experiments.capelin.model.Workload
+import org.opendc.experiments.compute.composite
+import org.opendc.experiments.compute.trace
 
 /**
  * A [Portfolio] that explores the effect of a composite workload.
diff --git a/opendc-experiments/opendc-experiments-capelin/src/main/kotlin/org/opendc/experiments/capelin/portfolio/HorVerPortfolio.kt b/opendc-experiments/opendc-experiments-capelin/src/main/kotlin/org/opendc/experiments/capelin/portfolio/HorVerPortfolio.kt
index 0d7f3072..f3c002ac 100644
--- a/opendc-experiments/opendc-experiments-capelin/src/main/kotlin/org/opendc/experiments/capelin/portfolio/HorVerPortfolio.kt
+++ b/opendc-experiments/opendc-experiments-capelin/src/main/kotlin/org/opendc/experiments/capelin/portfolio/HorVerPortfolio.kt
@@ -22,12 +22,12 @@
 
 package org.opendc.experiments.capelin.portfolio
 
-import org.opendc.compute.workload.sampleByLoad
-import org.opendc.compute.workload.trace
 import org.opendc.experiments.capelin.model.OperationalPhenomena
 import org.opendc.experiments.capelin.model.Scenario
 import org.opendc.experiments.capelin.model.Topology
 import org.opendc.experiments.capelin.model.Workload
+import org.opendc.experiments.compute.sampleByLoad
+import org.opendc.experiments.compute.trace
 
 /**
  * A [Portfolio] that explores the difference between horizontal and vertical scaling.
diff --git a/opendc-experiments/opendc-experiments-capelin/src/main/kotlin/org/opendc/experiments/capelin/portfolio/MoreHpcPortfolio.kt b/opendc-experiments/opendc-experiments-capelin/src/main/kotlin/org/opendc/experiments/capelin/portfolio/MoreHpcPortfolio.kt
index 6afffc09..22f9f3ac 100644
--- a/opendc-experiments/opendc-experiments-capelin/src/main/kotlin/org/opendc/experiments/capelin/portfolio/MoreHpcPortfolio.kt
+++ b/opendc-experiments/opendc-experiments-capelin/src/main/kotlin/org/opendc/experiments/capelin/portfolio/MoreHpcPortfolio.kt
@@ -22,13 +22,13 @@
 
 package org.opendc.experiments.capelin.portfolio
 
-import org.opendc.compute.workload.sampleByHpc
-import org.opendc.compute.workload.sampleByHpcLoad
-import org.opendc.compute.workload.trace
 import org.opendc.experiments.capelin.model.OperationalPhenomena
 import org.opendc.experiments.capelin.model.Scenario
 import org.opendc.experiments.capelin.model.Topology
 import org.opendc.experiments.capelin.model.Workload
+import org.opendc.experiments.compute.sampleByHpc
+import org.opendc.experiments.compute.sampleByHpcLoad
+import org.opendc.experiments.compute.trace
 
 /**
  * A [Portfolio] to explore the effect of HPC workloads.
diff --git a/opendc-experiments/opendc-experiments-capelin/src/main/kotlin/org/opendc/experiments/capelin/portfolio/MoreVelocityPortfolio.kt b/opendc-experiments/opendc-experiments-capelin/src/main/kotlin/org/opendc/experiments/capelin/portfolio/MoreVelocityPortfolio.kt
index 92bf80b3..e63a5807 100644
--- a/opendc-experiments/opendc-experiments-capelin/src/main/kotlin/org/opendc/experiments/capelin/portfolio/MoreVelocityPortfolio.kt
+++ b/opendc-experiments/opendc-experiments-capelin/src/main/kotlin/org/opendc/experiments/capelin/portfolio/MoreVelocityPortfolio.kt
@@ -22,12 +22,12 @@
 
 package org.opendc.experiments.capelin.portfolio
 
-import org.opendc.compute.workload.sampleByLoad
-import org.opendc.compute.workload.trace
 import org.opendc.experiments.capelin.model.OperationalPhenomena
 import org.opendc.experiments.capelin.model.Scenario
 import org.opendc.experiments.capelin.model.Topology
 import org.opendc.experiments.capelin.model.Workload
+import org.opendc.experiments.compute.sampleByLoad
+import org.opendc.experiments.compute.trace
 
 /**
  * A [Portfolio] that explores the effect of adding more velocity to a cluster (e.g., faster machines).
diff --git a/opendc-experiments/opendc-experiments-capelin/src/main/kotlin/org/opendc/experiments/capelin/portfolio/OperationalPhenomenaPortfolio.kt b/opendc-experiments/opendc-experiments-capelin/src/main/kotlin/org/opendc/experiments/capelin/portfolio/OperationalPhenomenaPortfolio.kt
index f9a9d681..12570108 100644
--- a/opendc-experiments/opendc-experiments-capelin/src/main/kotlin/org/opendc/experiments/capelin/portfolio/OperationalPhenomenaPortfolio.kt
+++ b/opendc-experiments/opendc-experiments-capelin/src/main/kotlin/org/opendc/experiments/capelin/portfolio/OperationalPhenomenaPortfolio.kt
@@ -22,12 +22,12 @@
 
 package org.opendc.experiments.capelin.portfolio
 
-import org.opendc.compute.workload.sampleByLoad
-import org.opendc.compute.workload.trace
 import org.opendc.experiments.capelin.model.OperationalPhenomena
 import org.opendc.experiments.capelin.model.Scenario
 import org.opendc.experiments.capelin.model.Topology
 import org.opendc.experiments.capelin.model.Workload
+import org.opendc.experiments.compute.sampleByLoad
+import org.opendc.experiments.compute.trace
 
 /**
  * A [Portfolio] that explores the effect of operational phenomena on metrics.
diff --git a/opendc-experiments/opendc-experiments-capelin/src/main/kotlin/org/opendc/experiments/capelin/portfolio/TestPortfolio.kt b/opendc-experiments/opendc-experiments-capelin/src/main/kotlin/org/opendc/experiments/capelin/portfolio/TestPortfolio.kt
index 944e9f43..6f126b87 100644
--- a/opendc-experiments/opendc-experiments-capelin/src/main/kotlin/org/opendc/experiments/capelin/portfolio/TestPortfolio.kt
+++ b/opendc-experiments/opendc-experiments-capelin/src/main/kotlin/org/opendc/experiments/capelin/portfolio/TestPortfolio.kt
@@ -22,11 +22,11 @@
 
 package org.opendc.experiments.capelin.portfolio
 
-import org.opendc.compute.workload.trace
 import org.opendc.experiments.capelin.model.OperationalPhenomena
 import org.opendc.experiments.capelin.model.Scenario
 import org.opendc.experiments.capelin.model.Topology
 import org.opendc.experiments.capelin.model.Workload
+import org.opendc.experiments.compute.trace
 
 /**
  * A [Portfolio] to perform a simple test run.
diff --git a/opendc-experiments/opendc-experiments-capelin/src/main/kotlin/org/opendc/experiments/capelin/topology/TopologyFactories.kt b/opendc-experiments/opendc-experiments-capelin/src/main/kotlin/org/opendc/experiments/capelin/topology/TopologyFactories.kt
index 5ab4261a..054adfcd 100644
--- a/opendc-experiments/opendc-experiments-capelin/src/main/kotlin/org/opendc/experiments/capelin/topology/TopologyFactories.kt
+++ b/opendc-experiments/opendc-experiments-capelin/src/main/kotlin/org/opendc/experiments/capelin/topology/TopologyFactories.kt
@@ -23,8 +23,7 @@
 @file:JvmName("TopologyFactories")
 package org.opendc.experiments.capelin.topology
 
-import org.opendc.compute.workload.topology.HostSpec
-import org.opendc.compute.workload.topology.Topology
+import org.opendc.experiments.compute.topology.HostSpec
 import org.opendc.simulator.compute.model.MachineModel
 import org.opendc.simulator.compute.model.MemoryUnit
 import org.opendc.simulator.compute.model.ProcessingNode
@@ -43,61 +42,55 @@ import kotlin.math.roundToLong
 private val reader = ClusterSpecReader()
 
 /**
- * Construct a [Topology] from the specified [file].
+ * Construct a topology from the specified [file].
  */
 fun clusterTopology(
     file: File,
     powerModel: PowerModel = LinearPowerModel(350.0, idlePower = 200.0),
     random: Random = Random(0)
-): Topology = clusterTopology(reader.read(file), powerModel, random)
+): List<HostSpec> {
+    return clusterTopology(reader.read(file), powerModel, random)
+}
 
 /**
- * Construct a [Topology] from the specified [input].
+ * Construct a topology from the specified [input].
  */
 fun clusterTopology(
     input: InputStream,
     powerModel: PowerModel = LinearPowerModel(350.0, idlePower = 200.0),
     random: Random = Random(0)
-): Topology = clusterTopology(reader.read(input), powerModel, random)
+): List<HostSpec> {
+    return clusterTopology(reader.read(input), powerModel, random)
+}
 
 /**
- * Construct a [Topology] from the given list of [clusters].
+ * Construct a topology from the given list of [clusters].
  */
-fun clusterTopology(
-    clusters: List<ClusterSpec>,
-    powerModel: PowerModel,
-    random: Random = Random(0)
-): Topology {
-    return object : Topology {
-        override fun resolve(): List<HostSpec> {
-            val hosts = mutableListOf<HostSpec>()
-            for (cluster in clusters) {
-                val cpuSpeed = cluster.cpuSpeed
-                val memoryPerHost = cluster.memCapacityPerHost.roundToLong()
-
-                val unknownProcessingNode = ProcessingNode("unknown", "unknown", "unknown", cluster.cpuCountPerHost)
-                val unknownMemoryUnit = MemoryUnit("unknown", "unknown", -1.0, memoryPerHost)
-                val machineModel = MachineModel(
-                    List(cluster.cpuCountPerHost) { coreId -> ProcessingUnit(unknownProcessingNode, coreId, cpuSpeed) },
-                    listOf(unknownMemoryUnit)
-                )
-
-                repeat(cluster.hostCount) {
-                    val spec = HostSpec(
-                        UUID(random.nextLong(), it.toLong()),
-                        "node-${cluster.name}-$it",
-                        mapOf("cluster" to cluster.id),
-                        machineModel,
-                        SimplePowerDriver(powerModel)
-                    )
+fun clusterTopology(clusters: List<ClusterSpec>, powerModel: PowerModel, random: Random = Random(0)): List<HostSpec> {
+    return clusters.flatMap { it.toHostSpecs(random, powerModel) }
+}
 
-                    hosts += spec
-                }
-            }
+/**
+ * Helper method to convert a [ClusterSpec] into a list of [HostSpec]s.
+ */
+private fun ClusterSpec.toHostSpecs(random: Random, powerModel: PowerModel): List<HostSpec> {
+    val cpuSpeed = cpuSpeed
+    val memoryPerHost = memCapacityPerHost.roundToLong()
 
-            return hosts
-        }
+    val unknownProcessingNode = ProcessingNode("unknown", "unknown", "unknown", cpuCountPerHost)
+    val unknownMemoryUnit = MemoryUnit("unknown", "unknown", -1.0, memoryPerHost)
+    val machineModel = MachineModel(
+        List(cpuCountPerHost) { coreId -> ProcessingUnit(unknownProcessingNode, coreId, cpuSpeed) },
+        listOf(unknownMemoryUnit)
+    )
 
-        override fun toString(): String = "ClusterSpecTopology"
+    return List(hostCount) {
+        HostSpec(
+            UUID(random.nextLong(), it.toLong()),
+            "node-$name-$it",
+            mapOf("cluster" to id),
+            machineModel,
+            SimplePowerDriver(powerModel)
+        )
     }
 }
diff --git a/opendc-experiments/opendc-experiments-capelin/src/test/kotlin/org/opendc/experiments/capelin/CapelinIntegrationTest.kt b/opendc-experiments/opendc-experiments-capelin/src/test/kotlin/org/opendc/experiments/capelin/CapelinIntegrationTest.kt
index eae3c993..9be2d522 100644
--- a/opendc-experiments/opendc-experiments-capelin/src/test/kotlin/org/opendc/experiments/capelin/CapelinIntegrationTest.kt
+++ b/opendc-experiments/opendc-experiments-capelin/src/test/kotlin/org/opendc/experiments/capelin/CapelinIntegrationTest.kt
@@ -26,18 +26,19 @@ import org.junit.jupiter.api.Assertions.assertEquals
 import org.junit.jupiter.api.BeforeEach
 import org.junit.jupiter.api.Test
 import org.junit.jupiter.api.assertAll
+import org.opendc.compute.service.ComputeService
 import org.opendc.compute.service.scheduler.FilterScheduler
 import org.opendc.compute.service.scheduler.filters.ComputeFilter
 import org.opendc.compute.service.scheduler.filters.RamFilter
 import org.opendc.compute.service.scheduler.filters.VCpuFilter
 import org.opendc.compute.service.scheduler.weights.CoreRamWeigher
-import org.opendc.compute.workload.*
-import org.opendc.compute.workload.telemetry.ComputeMetricReader
-import org.opendc.compute.workload.telemetry.ComputeMonitor
-import org.opendc.compute.workload.telemetry.table.HostTableReader
-import org.opendc.compute.workload.topology.Topology
-import org.opendc.compute.workload.topology.apply
 import org.opendc.experiments.capelin.topology.clusterTopology
+import org.opendc.experiments.compute.*
+import org.opendc.experiments.compute.telemetry.ComputeMonitor
+import org.opendc.experiments.compute.telemetry.table.HostTableReader
+import org.opendc.experiments.compute.telemetry.table.ServiceTableReader
+import org.opendc.experiments.compute.topology.HostSpec
+import org.opendc.experiments.provisioner.Provisioner
 import org.opendc.simulator.core.runBlockingSimulation
 import java.io.File
 import java.time.Duration
@@ -82,45 +83,41 @@ class CapelinIntegrationTest {
     fun testLarge() = runBlockingSimulation {
         val seed = 0L
         val workload = createTestWorkload(1.0, seed)
-        val runner = ComputeServiceHelper(
-            coroutineContext,
-            clock,
-            computeScheduler,
-            seed,
-        )
         val topology = createTopology()
-        val reader = ComputeMetricReader(this, clock, runner.service, monitor)
-
-        try {
-            runner.apply(topology)
-            runner.run(workload)
+        val monitor = monitor
 
-            val serviceMetrics = runner.service.getSchedulerStats()
-            println(
-                "Scheduler " +
-                    "Success=${serviceMetrics.attemptsSuccess} " +
-                    "Failure=${serviceMetrics.attemptsFailure} " +
-                    "Error=${serviceMetrics.attemptsError} " +
-                    "Pending=${serviceMetrics.serversPending} " +
-                    "Active=${serviceMetrics.serversActive}"
+        Provisioner(coroutineContext, clock, seed).use { provisioner ->
+            provisioner.runSteps(
+                setupComputeService(serviceDomain = "compute.opendc.org", { computeScheduler }),
+                registerComputeMonitor(serviceDomain = "compute.opendc.org", monitor),
+                setupHosts(serviceDomain = "compute.opendc.org", topology),
             )
 
-            // Note that these values have been verified beforehand
-            assertAll(
-                { assertEquals(50, serviceMetrics.attemptsSuccess, "The scheduler should schedule 50 VMs") },
-                { assertEquals(0, serviceMetrics.serversActive, "All VMs should finish after a run") },
-                { assertEquals(0, serviceMetrics.attemptsFailure, "No VM should be unscheduled") },
-                { assertEquals(0, serviceMetrics.serversPending, "No VM should not be in the queue") },
-                { assertEquals(223393683, this@CapelinIntegrationTest.monitor.idleTime) { "Incorrect idle time" } },
-                { assertEquals(66977508, this@CapelinIntegrationTest.monitor.activeTime) { "Incorrect active time" } },
-                { assertEquals(3160381, this@CapelinIntegrationTest.monitor.stealTime) { "Incorrect steal time" } },
-                { assertEquals(0, this@CapelinIntegrationTest.monitor.lostTime) { "Incorrect lost time" } },
-                { assertEquals(5.840939264814157E9, this@CapelinIntegrationTest.monitor.energyUsage, 0.01) { "Incorrect power draw" } },
-            )
-        } finally {
-            runner.close()
-            reader.close()
+            val service = provisioner.registry.resolve("compute.opendc.org", ComputeService::class.java)!!
+            service.replay(clock, workload, seed)
         }
+
+        println(
+            "Scheduler " +
+                "Success=${monitor.attemptsSuccess} " +
+                "Failure=${monitor.attemptsFailure} " +
+                "Error=${monitor.attemptsError} " +
+                "Pending=${monitor.serversPending} " +
+                "Active=${monitor.serversActive}"
+        )
+
+        // Note that these values have been verified beforehand
+        assertAll(
+            { assertEquals(50, monitor.attemptsSuccess, "The scheduler should schedule 50 VMs") },
+            { assertEquals(0, monitor.serversActive, "All VMs should finish after a run") },
+            { assertEquals(0, monitor.attemptsFailure, "No VM should be unscheduled") },
+            { assertEquals(0, monitor.serversPending, "No VM should not be in the queue") },
+            { assertEquals(223393683, monitor.idleTime) { "Incorrect idle time" } },
+            { assertEquals(66977508, monitor.activeTime) { "Incorrect active time" } },
+            { assertEquals(3160381, monitor.stealTime) { "Incorrect steal time" } },
+            { assertEquals(0, monitor.lostTime) { "Incorrect lost time" } },
+            { assertEquals(5.840939264814157E9, monitor.energyUsage, 0.01) { "Incorrect power draw" } },
+        )
     }
 
     /**
@@ -130,40 +127,36 @@ class CapelinIntegrationTest {
     fun testSmall() = runBlockingSimulation {
         val seed = 1L
         val workload = createTestWorkload(0.25, seed)
-        val runner = ComputeServiceHelper(
-            coroutineContext,
-            clock,
-            computeScheduler,
-            seed,
-        )
         val topology = createTopology("single")
-        val reader = ComputeMetricReader(this, clock, runner.service, monitor)
-
-        try {
-            runner.apply(topology)
-            runner.run(workload)
+        val monitor = monitor
 
-            val serviceMetrics = runner.service.getSchedulerStats()
-            println(
-                "Scheduler " +
-                    "Success=${serviceMetrics.attemptsSuccess} " +
-                    "Failure=${serviceMetrics.attemptsFailure} " +
-                    "Error=${serviceMetrics.attemptsError} " +
-                    "Pending=${serviceMetrics.serversPending} " +
-                    "Active=${serviceMetrics.serversActive}"
+        Provisioner(coroutineContext, clock, seed).use { provisioner ->
+            provisioner.runSteps(
+                setupComputeService(serviceDomain = "compute.opendc.org", { computeScheduler }),
+                registerComputeMonitor(serviceDomain = "compute.opendc.org", monitor),
+                setupHosts(serviceDomain = "compute.opendc.org", topology),
             )
-        } finally {
-            runner.close()
-            reader.close()
+
+            val service = provisioner.registry.resolve("compute.opendc.org", ComputeService::class.java)!!
+            service.replay(clock, workload, seed)
         }
 
+        println(
+            "Scheduler " +
+                "Success=${monitor.attemptsSuccess} " +
+                "Failure=${monitor.attemptsFailure} " +
+                "Error=${monitor.attemptsError} " +
+                "Pending=${monitor.serversPending} " +
+                "Active=${monitor.serversActive}"
+        )
+
         // Note that these values have been verified beforehand
         assertAll(
-            { assertEquals(10999592, this@CapelinIntegrationTest.monitor.idleTime) { "Idle time incorrect" } },
-            { assertEquals(9741207, this@CapelinIntegrationTest.monitor.activeTime) { "Active time incorrect" } },
-            { assertEquals(0, this@CapelinIntegrationTest.monitor.stealTime) { "Steal time incorrect" } },
-            { assertEquals(0, this@CapelinIntegrationTest.monitor.lostTime) { "Lost time incorrect" } },
-            { assertEquals(7.011676470304312E8, this@CapelinIntegrationTest.monitor.energyUsage, 0.01) { "Incorrect power draw" } }
+            { assertEquals(10999592, monitor.idleTime) { "Idle time incorrect" } },
+            { assertEquals(9741207, monitor.activeTime) { "Active time incorrect" } },
+            { assertEquals(0, monitor.stealTime) { "Steal time incorrect" } },
+            { assertEquals(0, monitor.lostTime) { "Lost time incorrect" } },
+            { assertEquals(7.011676470304312E8, monitor.energyUsage, 0.01) { "Incorrect power draw" } }
         )
     }
 
@@ -174,40 +167,34 @@ class CapelinIntegrationTest {
     fun testInterference() = runBlockingSimulation {
         val seed = 0L
         val workload = createTestWorkload(1.0, seed)
-
-        val simulator = ComputeServiceHelper(
-            coroutineContext,
-            clock,
-            computeScheduler,
-            seed
-        )
         val topology = createTopology("single")
-        val reader = ComputeMetricReader(this, clock, simulator.service, monitor)
 
-        try {
-            simulator.apply(topology)
-            simulator.run(workload, interference = true)
-
-            val serviceMetrics = simulator.service.getSchedulerStats()
-            println(
-                "Scheduler " +
-                    "Success=${serviceMetrics.attemptsSuccess} " +
-                    "Failure=${serviceMetrics.attemptsFailure} " +
-                    "Error=${serviceMetrics.attemptsError} " +
-                    "Pending=${serviceMetrics.serversPending} " +
-                    "Active=${serviceMetrics.serversActive}"
+        Provisioner(coroutineContext, clock, seed).use { provisioner ->
+            provisioner.runSteps(
+                setupComputeService(serviceDomain = "compute.opendc.org", { computeScheduler }),
+                registerComputeMonitor(serviceDomain = "compute.opendc.org", monitor),
+                setupHosts(serviceDomain = "compute.opendc.org", topology),
             )
-        } finally {
-            simulator.close()
-            reader.close()
+
+            val service = provisioner.registry.resolve("compute.opendc.org", ComputeService::class.java)!!
+            service.replay(clock, workload, seed, interference = true)
         }
 
+        println(
+            "Scheduler " +
+                "Success=${monitor.attemptsSuccess} " +
+                "Failure=${monitor.attemptsFailure} " +
+                "Error=${monitor.attemptsError} " +
+                "Pending=${monitor.serversPending} " +
+                "Active=${monitor.serversActive}"
+        )
+
         // Note that these values have been verified beforehand
         assertAll(
-            { assertEquals(6028050, this@CapelinIntegrationTest.monitor.idleTime) { "Idle time incorrect" } },
-            { assertEquals(14712749, this@CapelinIntegrationTest.monitor.activeTime) { "Active time incorrect" } },
-            { assertEquals(12532907, this@CapelinIntegrationTest.monitor.stealTime) { "Steal time incorrect" } },
-            { assertEquals(485510, this@CapelinIntegrationTest.monitor.lostTime) { "Lost time incorrect" } }
+            { assertEquals(6028050, monitor.idleTime) { "Idle time incorrect" } },
+            { assertEquals(14712749, monitor.activeTime) { "Active time incorrect" } },
+            { assertEquals(12532907, monitor.stealTime) { "Steal time incorrect" } },
+            { assertEquals(470593, monitor.lostTime) { "Lost time incorrect" } }
         )
     }
 
@@ -217,41 +204,28 @@ class CapelinIntegrationTest {
     @Test
     fun testFailures() = runBlockingSimulation {
         val seed = 0L
-        val simulator = ComputeServiceHelper(
-            coroutineContext,
-            clock,
-            computeScheduler,
-            seed
-        )
         val topology = createTopology("single")
         val workload = createTestWorkload(0.25, seed)
-        val reader = ComputeMetricReader(this, clock, simulator.service, monitor)
+        val monitor = monitor
 
-        try {
-            simulator.apply(topology)
-            simulator.run(workload, failureModel = grid5000(Duration.ofDays(7)))
-
-            val serviceMetrics = simulator.service.getSchedulerStats()
-            println(
-                "Scheduler " +
-                    "Success=${serviceMetrics.attemptsSuccess} " +
-                    "Failure=${serviceMetrics.attemptsFailure} " +
-                    "Error=${serviceMetrics.attemptsError} " +
-                    "Pending=${serviceMetrics.serversPending} " +
-                    "Active=${serviceMetrics.serversActive}"
+        Provisioner(coroutineContext, clock, seed).use { provisioner ->
+            provisioner.runSteps(
+                setupComputeService(serviceDomain = "compute.opendc.org", { computeScheduler }),
+                registerComputeMonitor(serviceDomain = "compute.opendc.org", monitor),
+                setupHosts(serviceDomain = "compute.opendc.org", topology),
             )
-        } finally {
-            simulator.close()
-            reader.close()
+
+            val service = provisioner.registry.resolve("compute.opendc.org", ComputeService::class.java)!!
+            service.replay(clock, workload, seed, failureModel = grid5000(Duration.ofDays(7)))
         }
 
         // Note that these values have been verified beforehand
         assertAll(
-            { assertEquals(10982026, monitor.idleTime) { "Idle time incorrect" } },
-            { assertEquals(9740058, monitor.activeTime) { "Active time incorrect" } },
+            { assertEquals(10085158, monitor.idleTime) { "Idle time incorrect" } },
+            { assertEquals(8539158, monitor.activeTime) { "Active time incorrect" } },
             { assertEquals(0, monitor.stealTime) { "Steal time incorrect" } },
             { assertEquals(0, monitor.lostTime) { "Lost time incorrect" } },
-            { assertEquals(2590260605, monitor.uptime) { "Uptime incorrect" } },
+            { assertEquals(2328039558, monitor.uptime) { "Uptime incorrect" } },
         )
     }
 
@@ -266,12 +240,26 @@ class CapelinIntegrationTest {
     /**
      * Obtain the topology factory for the test.
      */
-    private fun createTopology(name: String = "topology"): Topology {
+    private fun createTopology(name: String = "topology"): List<HostSpec> {
         val stream = checkNotNull(object {}.javaClass.getResourceAsStream("/env/$name.txt"))
         return stream.use { clusterTopology(stream) }
     }
 
     class TestComputeMonitor : ComputeMonitor {
+        var attemptsSuccess = 0
+        var attemptsFailure = 0
+        var attemptsError = 0
+        var serversPending = 0
+        var serversActive = 0
+
+        override fun record(reader: ServiceTableReader) {
+            attemptsSuccess = reader.attemptsSuccess
+            attemptsFailure = reader.attemptsFailure
+            attemptsError = reader.attemptsError
+            serversPending = reader.serversPending
+            serversActive = reader.serversActive
+        }
+
         var idleTime = 0L
         var activeTime = 0L
         var stealTime = 0L
diff --git a/opendc-experiments/opendc-experiments-capelin/src/test/kotlin/org/opendc/experiments/capelin/CapelinRunnerTest.kt b/opendc-experiments/opendc-experiments-capelin/src/test/kotlin/org/opendc/experiments/capelin/CapelinRunnerTest.kt
new file mode 100644
index 00000000..2aeb9ff9
--- /dev/null
+++ b/opendc-experiments/opendc-experiments-capelin/src/test/kotlin/org/opendc/experiments/capelin/CapelinRunnerTest.kt
@@ -0,0 +1,86 @@
+/*
+ * Copyright (c) 2022 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.experiments.capelin
+
+import org.junit.jupiter.api.Test
+import org.junit.jupiter.api.assertDoesNotThrow
+import org.opendc.experiments.capelin.model.OperationalPhenomena
+import org.opendc.experiments.capelin.model.Scenario
+import org.opendc.experiments.capelin.model.Topology
+import org.opendc.experiments.capelin.model.Workload
+import org.opendc.experiments.compute.trace
+import java.io.File
+import java.nio.file.Files
+
+/**
+ * Test suite for [CapelinRunner].
+ */
+class CapelinRunnerTest {
+    /**
+     * The path to the environments.
+     */
+    private val envPath = File("src/test/resources/env")
+
+    /**
+     * The path to the traces.
+     */
+    private val tracePath = File("src/test/resources/trace")
+
+    /**
+     * Smoke test with output.
+     */
+    @Test
+    fun testSmoke() {
+        val outputPath = Files.createTempDirectory("output").toFile()
+
+        try {
+            val runner = CapelinRunner(envPath, tracePath, outputPath)
+            val scenario = Scenario(
+                Topology("topology"),
+                Workload("bitbrains-small", trace("bitbrains-small")),
+                OperationalPhenomena(failureFrequency = 24.0 * 7, hasInterference = true),
+                "active-servers"
+            )
+
+            assertDoesNotThrow { runner.runScenario(scenario, seed = 0L) }
+        } finally {
+            outputPath.delete()
+        }
+    }
+
+    /**
+     * Smoke test without output.
+     */
+    @Test
+    fun testSmokeNoOutput() {
+        val runner = CapelinRunner(envPath, tracePath, null)
+        val scenario = Scenario(
+            Topology("topology"),
+            Workload("bitbrains-small", trace("bitbrains-small")),
+            OperationalPhenomena(failureFrequency = 24.0 * 7, hasInterference = true),
+            "active-servers"
+        )
+
+        assertDoesNotThrow { runner.runScenario(scenario, seed = 0L) }
+    }
+}