feat(faas): Add helper tools for FaaS simulations

This change adds a new module, opendc-faas-workload that contains
helper code for constructing simulations of FaaS-based workloads
using OpenDC. In addition, we add an integration test that demonstrates
the capabilities of the helper tool and the FaaS platform of OpenDC.

5 files changed, 398 insertions, 0 deletions

diff --git a/opendc-faas/opendc-faas-workload/src/main/kotlin/org/opendc/faas/workload/FaaSServiceHelper.kt b/opendc-faas/opendc-faas-workload/src/main/kotlin/org/opendc/faas/workload/FaaSServiceHelper.kt
new file mode 100644
index 00000000..ede6ac54
--- /dev/null
+++ b/opendc-faas/opendc-faas-workload/src/main/kotlin/org/opendc/faas/workload/FaaSServiceHelper.kt
@@ -0,0 +1,153 @@
+/*
+ * 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.faas.workload
+
+import kotlinx.coroutines.*
+import mu.KotlinLogging
+import org.opendc.faas.api.FaaSFunction
+import org.opendc.faas.service.FaaSService
+import org.opendc.faas.service.FunctionObject
+import org.opendc.faas.service.autoscaler.FunctionTerminationPolicy
+import org.opendc.faas.service.deployer.FunctionDeployer
+import org.opendc.faas.service.deployer.FunctionInstance
+import org.opendc.faas.service.deployer.FunctionInstanceListener
+import org.opendc.faas.service.router.RoutingPolicy
+import org.opendc.faas.simulator.SimFunctionDeployer
+import org.opendc.faas.simulator.delay.ColdStartModel
+import org.opendc.faas.simulator.delay.StochasticDelayInjector
+import org.opendc.faas.simulator.delay.ZeroDelayInjector
+import org.opendc.simulator.compute.model.MachineModel
+import java.time.Clock
+import java.util.*
+import kotlin.coroutines.CoroutineContext
+import kotlin.math.max
+
+/**
+ * Helper class to simulate FaaS-based workloads in OpenDC.
+ *
+ * @param context A [CoroutineContext] to run the simulation in.
+ * @param clock A [Clock] instance tracking simulation time.
+ * @param machineModel The [MachineModel] that models the physical machine on which the functions run.
+ * @param routingPolicy The routing policy to use.
+ * @param terminationPolicy The function termination policy to use.
+ * @param coldStartModel The cold start models to test.
+ * @param seed The seed of the simulation.
+ */
+public class FaaSServiceHelper(
+    private val context: CoroutineContext,
+    private val clock: Clock,
+    private val machineModel: MachineModel,
+    private val routingPolicy: RoutingPolicy,
+    private val terminationPolicy: FunctionTerminationPolicy,
+    private val coldStartModel: ColdStartModel? = null,
+) : AutoCloseable {
+    /**
+     * The scope of this helper.
+     */
+    private val scope = CoroutineScope(context + Job())
+
+    /**
+     * The logger for this class.
+     */
+    private val logger = KotlinLogging.logger {}
+
+    /**
+     * The simulated function deployer.
+     */
+    private val deployer = object : FunctionDeployer {
+        override fun deploy(function: FunctionObject, listener: FunctionInstanceListener): FunctionInstance {
+            val deployer = checkNotNull(_deployer)
+            return deployer.deploy(function, listener)
+        }
+    }
+    private var _deployer: SimFunctionDeployer? = null
+
+    /**
+     * The [FaaSService] created by the helper.
+     */
+    public val service: FaaSService = FaaSService(
+        context,
+        clock,
+        deployer,
+        routingPolicy,
+        terminationPolicy
+    )
+
+    /**
+     * Run a simulation of the [FaaSService] by replaying the workload trace given by [trace].
+     *
+     * @param trace The trace to simulate.
+     * @param seed The seed for the simulation.
+     * @param functions The functions that have been created by the runner.
+     */
+    public suspend fun run(trace: List<FunctionTrace>, seed: Long = 0, functions: MutableList<FaaSFunction>? = null) {
+        // Set up the simulated deployer
+        val delayInjector = if (coldStartModel != null)
+            StochasticDelayInjector(coldStartModel, Random(seed))
+        else
+            ZeroDelayInjector
+        val traceById = trace.associateBy { it.id }
+        _deployer = SimFunctionDeployer(clock, scope, machineModel, delayInjector) {
+            FunctionTraceWorkload(traceById.getValue(it.name))
+        }
+
+        val client = service.newClient()
+        try {
+            coroutineScope {
+                for (entry in trace) {
+                    launch {
+                        val function = client.newFunction(entry.id, entry.maxMemory.toLong())
+                        functions?.add(function)
+
+                        var offset = Long.MIN_VALUE
+
+                        for (sample in entry.samples) {
+                            if (sample.invocations == 0) {
+                                continue
+                            }
+
+                            if (offset < 0) {
+                                offset = sample.timestamp - clock.millis()
+                            }
+
+                            delay(max(0, (sample.timestamp - offset) - clock.millis()))
+
+                            logger.info { "Invoking function ${entry.id} ${sample.invocations} times [${sample.timestamp}]" }
+
+                            repeat(sample.invocations) {
+                                function.invoke()
+                            }
+                        }
+                    }
+                }
+            }
+        } finally {
+            client.close()
+        }
+    }
+
+    override fun close() {
+        service.close()
+        scope.cancel()
+    }
+}
diff --git a/opendc-faas/opendc-faas-workload/src/main/kotlin/org/opendc/faas/workload/FunctionSample.kt b/opendc-faas/opendc-faas-workload/src/main/kotlin/org/opendc/faas/workload/FunctionSample.kt
new file mode 100644
index 00000000..418f895d
--- /dev/null
+++ b/opendc-faas/opendc-faas-workload/src/main/kotlin/org/opendc/faas/workload/FunctionSample.kt
@@ -0,0 +1,44 @@
+/*
+ * 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.faas.workload
+
+/**
+ * A sample of a single function.
+ *
+ * @param timestamp The timestamp of the function.
+ * @param duration The average execution time of the function.
+ * @param invocations The number of invocations.
+ * @param provisionedCpu The provisioned CPU for this function in MHz.
+ * @param provisionedMem The amount of memory provisioned for this function in MB.
+ * @param cpuUsage The actual CPU usage in MHz.
+ * @param memUsage The actual memory usage in MB.
+ */
+public data class FunctionSample(
+    val timestamp: Long,
+    val duration: Long,
+    val invocations: Int,
+    val provisionedCpu: Int,
+    val provisionedMem: Int,
+    val cpuUsage: Double,
+    val memUsage: Double
+)
diff --git a/opendc-faas/opendc-faas-workload/src/main/kotlin/org/opendc/faas/workload/FunctionTrace.kt b/opendc-faas/opendc-faas-workload/src/main/kotlin/org/opendc/faas/workload/FunctionTrace.kt
new file mode 100644
index 00000000..712267e5
--- /dev/null
+++ b/opendc-faas/opendc-faas-workload/src/main/kotlin/org/opendc/faas/workload/FunctionTrace.kt
@@ -0,0 +1,28 @@
+/*
+ * 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.faas.workload
+
+/**
+ * A trace for a single function
+ */
+public data class FunctionTrace(val id: String, val maxMemory: Int, val samples: List<FunctionSample>)
diff --git a/opendc-faas/opendc-faas-workload/src/main/kotlin/org/opendc/faas/workload/FunctionTraceWorkload.kt b/opendc-faas/opendc-faas-workload/src/main/kotlin/org/opendc/faas/workload/FunctionTraceWorkload.kt
new file mode 100644
index 00000000..cdb800c3
--- /dev/null
+++ b/opendc-faas/opendc-faas-workload/src/main/kotlin/org/opendc/faas/workload/FunctionTraceWorkload.kt
@@ -0,0 +1,37 @@
+/*
+ * 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.faas.workload
+
+import org.opendc.faas.simulator.workload.SimFaaSWorkload
+import org.opendc.simulator.compute.workload.SimTrace
+import org.opendc.simulator.compute.workload.SimTraceFragment
+import org.opendc.simulator.compute.workload.SimTraceWorkload
+import org.opendc.simulator.compute.workload.SimWorkload
+
+/**
+ * A [SimFaaSWorkload] for a [FunctionTrace].
+ */
+public class FunctionTraceWorkload(trace: FunctionTrace) :
+    SimFaaSWorkload, SimWorkload by SimTraceWorkload(SimTrace.ofFragments(trace.samples.map { SimTraceFragment(it.timestamp, it.duration, it.cpuUsage, 1) })) {
+    override suspend fun invoke() {}
+}
diff --git a/opendc-faas/opendc-faas-workload/src/main/kotlin/org/opendc/faas/workload/ServerlessTraceReader.kt b/opendc-faas/opendc-faas-workload/src/main/kotlin/org/opendc/faas/workload/ServerlessTraceReader.kt
new file mode 100644
index 00000000..3694cf30
--- /dev/null
+++ b/opendc-faas/opendc-faas-workload/src/main/kotlin/org/opendc/faas/workload/ServerlessTraceReader.kt
@@ -0,0 +1,136 @@
+/*
+ * 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.faas.workload
+
+import com.fasterxml.jackson.core.JsonToken
+import com.fasterxml.jackson.dataformat.csv.CsvFactory
+import com.fasterxml.jackson.dataformat.csv.CsvParser
+import com.fasterxml.jackson.dataformat.csv.CsvSchema
+import mu.KotlinLogging
+import java.io.File
+import kotlin.math.max
+
+/**
+ * A trace reader for the serverless workload trace used in the OpenDC Serverless thesis.
+ */
+public class ServerlessTraceReader {
+    /**
+     * The logger for this trace reader instance.
+     */
+    private val logger = KotlinLogging.logger {}
+
+    /**
+     * The [CsvFactory] used to create the parser.
+     */
+    private val factory = CsvFactory()
+        .enable(CsvParser.Feature.ALLOW_COMMENTS)
+        .enable(CsvParser.Feature.TRIM_SPACES)
+
+    /**
+     * Parse the traces at the specified [path].
+     */
+    public fun parse(path: File): List<FunctionTrace> {
+        return if (path.isFile) {
+            listOf(parseSingle(path))
+        } else {
+            path.walk()
+                .filterNot { it.isDirectory }
+                .sorted()
+                .map { file ->
+                    logger.info { "Parsing $file" }
+                    parseSingle(file)
+                }
+                .toList()
+        }
+    }
+
+    /**
+     * Parse a single trace.
+     */
+    private fun parseSingle(path: File): FunctionTrace {
+        val samples = mutableListOf<FunctionSample>()
+
+        val parser = factory.createParser(path)
+        parser.schema = schema
+
+        var id = ""
+        var timestamp = 0L
+        var invocations = 0
+        var execTime = 0L
+        var provisionedCpu = 0
+        var provisionedMem = 0
+        var cpuUsage = 0.0
+        var memUsage = 0.0
+        var maxMemory = 0
+
+        while (!parser.isClosed) {
+            val token = parser.nextValue()
+            if (token == JsonToken.END_OBJECT) {
+                maxMemory = max(maxMemory, provisionedMem)
+                samples.add(FunctionSample(timestamp, execTime, invocations, provisionedCpu, provisionedMem, cpuUsage, memUsage))
+
+                id = ""
+                timestamp = 0
+                invocations = 0
+                execTime = 0
+                provisionedCpu = 0
+                provisionedMem = 0
+                cpuUsage = 0.0
+                memUsage = 0.0
+
+                continue
+            }
+
+            when (parser.currentName) {
+                "Timestamp [ms]" -> timestamp = parser.valueAsLong
+                "Invocations" -> invocations = parser.valueAsInt
+                "Avg Exec time per Invocation" -> execTime = parser.valueAsLong
+                "Provisioned CPU [Mhz]" -> provisionedCpu = parser.valueAsInt
+                "Provisioned Memory [mb]" -> provisionedMem = parser.valueAsInt
+                "Avg cpu usage per Invocation [Mhz]" -> cpuUsage = parser.valueAsDouble
+                "Avg mem usage per Invocation [mb]" -> memUsage = parser.valueAsDouble
+                "name" -> id = parser.text
+            }
+        }
+
+        return FunctionTrace(id, maxMemory, samples)
+    }
+
+    private companion object {
+        /**
+         * The [CsvSchema] that is used to parse the trace.
+         */
+        val schema = CsvSchema.builder()
+            .addColumn("Timestamp [ms]", CsvSchema.ColumnType.NUMBER)
+            .addColumn("Invocations", CsvSchema.ColumnType.NUMBER)
+            .addColumn("Avg Exec time per Invocation", CsvSchema.ColumnType.NUMBER)
+            .addColumn("Provisioned CPU [Mhz]", CsvSchema.ColumnType.NUMBER)
+            .addColumn("Provisioned Memory [mb]", CsvSchema.ColumnType.NUMBER)
+            .addColumn("Avg cpu usage per Invocation [Mhz]", CsvSchema.ColumnType.NUMBER)
+            .addColumn("Avg mem usage per Invocation [mb]", CsvSchema.ColumnType.NUMBER)
+            .addColumn("name", CsvSchema.ColumnType.STRING)
+            .setAllowComments(true)
+            .setUseHeader(true)
+            .build()
+    }
+}