merge: Simplify usage of ComputeMetricExporter

This pull request addresses some issues with the current implementation of 
the `ComputeMetricExporter` class. 
In particular, the construction of `ComputeMetricExporter` does not require a `Clock` anymore.

- Ensure shutdown of exporter is called 
- Do not require clock for ComputeMetricExporter
- Do not recover guests in non-error state
- Write null values explicitly in Parquet exporter
- Report cause of compute exporter failure

**Breaking API Changes**

- `ComputeMetricExporter` is now an abstract class that can be extended to collect metrics
- `ParquetComputeMonitor` has been renamed to `ParquetComputeMetricExporter` and extends `ComputeMetricExporter`

6 files changed, 40 insertions, 36 deletions

diff --git a/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/SimHost.kt b/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/SimHost.kt
index ff55c585..fdb3f1dc 100644
--- a/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/SimHost.kt
+++ b/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/SimHost.kt
@@ -262,7 +262,7 @@ public class SimHost(
     }
 
     override suspend fun delete(server: Server) {
-        val guest = guests.remove(server) ?: return
+        val guest = guests[server] ?: return
         guest.terminate()
     }
 
@@ -296,7 +296,7 @@ public class SimHost(
         _bootTime = clock.millis()
 
         for (guest in guests.values) {
-            guest.start()
+            guest.recover()
         }
     }
 
diff --git a/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/internal/Guest.kt b/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/internal/Guest.kt
index 90562e2f..7f33154a 100644
--- a/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/internal/Guest.kt
+++ b/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/internal/Guest.kt
@@ -96,8 +96,8 @@ internal class Guest(
             }
             ServerState.RUNNING -> return
             ServerState.DELETED -> {
-                logger.warn { "User tried to start terminated server" }
-                throw IllegalArgumentException("Server is terminated")
+                logger.warn { "User tried to start deleted server" }
+                throw IllegalArgumentException("Server is deleted")
             }
             else -> assert(false) { "Invalid state transition" }
         }
@@ -144,6 +144,17 @@ internal class Guest(
     }
 
     /**
+     * Recover the guest if it is in an error state.
+     */
+    suspend fun recover() {
+        if (state != ServerState.ERROR) {
+            return
+        }
+
+        doStart()
+    }
+
+    /**
      * The [Job] representing the current active virtual machine instance or `null` if no virtual machine is active.
      */
     private var job: Job? = null
diff --git a/opendc-compute/opendc-compute-simulator/src/test/kotlin/org/opendc/compute/simulator/SimHostTest.kt b/opendc-compute/opendc-compute-simulator/src/test/kotlin/org/opendc/compute/simulator/SimHostTest.kt
index 9c879e5e..e75c31a0 100644
--- a/opendc-compute/opendc-compute-simulator/src/test/kotlin/org/opendc/compute/simulator/SimHostTest.kt
+++ b/opendc-compute/opendc-compute-simulator/src/test/kotlin/org/opendc/compute/simulator/SimHostTest.kt
@@ -43,7 +43,6 @@ import org.opendc.simulator.compute.workload.SimTraceWorkload
 import org.opendc.simulator.core.runBlockingSimulation
 import org.opendc.simulator.resources.SimResourceInterpreter
 import org.opendc.telemetry.compute.ComputeMetricExporter
-import org.opendc.telemetry.compute.ComputeMonitor
 import org.opendc.telemetry.compute.HOST_ID
 import org.opendc.telemetry.compute.table.HostData
 import org.opendc.telemetry.compute.table.ServerData
@@ -138,16 +137,13 @@ internal class SimHostTest {
         // Setup metric reader
         val reader = CoroutineMetricReader(
             this, listOf(meterProvider as MetricProducer),
-            ComputeMetricExporter(
-                clock,
-                object : ComputeMonitor {
-                    override fun record(data: HostData) {
-                        activeTime += data.cpuActiveTime
-                        idleTime += data.cpuIdleTime
-                        stealTime += data.cpuStealTime
-                    }
+            object : ComputeMetricExporter() {
+                override fun record(data: HostData) {
+                    activeTime += data.cpuActiveTime
+                    idleTime += data.cpuIdleTime
+                    stealTime += data.cpuStealTime
                 }
-            ),
+            },
             exportInterval = Duration.ofSeconds(duration)
         )
 
@@ -237,22 +233,19 @@ internal class SimHostTest {
         // Setup metric reader
         val reader = CoroutineMetricReader(
             this, listOf(meterProvider as MetricProducer),
-            ComputeMetricExporter(
-                clock,
-                object : ComputeMonitor {
-                    override fun record(data: HostData) {
-                        activeTime += data.cpuActiveTime
-                        idleTime += data.cpuIdleTime
-                        uptime += data.uptime
-                        downtime += data.downtime
-                    }
+            object : ComputeMetricExporter() {
+                override fun record(data: HostData) {
+                    activeTime += data.cpuActiveTime
+                    idleTime += data.cpuIdleTime
+                    uptime += data.uptime
+                    downtime += data.downtime
+                }
 
-                    override fun record(data: ServerData) {
-                        guestUptime += data.uptime
-                        guestDowntime += data.downtime
-                    }
+                override fun record(data: ServerData) {
+                    guestUptime += data.uptime
+                    guestDowntime += data.downtime
                 }
-            ),
+            },
             exportInterval = Duration.ofSeconds(duration)
         )
 
diff --git a/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/export/parquet/ParquetExportMonitor.kt b/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/export/parquet/ParquetComputeMetricExporter.kt
index f41a2241..ad182d67 100644
--- a/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/export/parquet/ParquetExportMonitor.kt
+++ b/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/export/parquet/ParquetComputeMetricExporter.kt
@@ -22,6 +22,8 @@
 
 package org.opendc.compute.workload.export.parquet
 
+import io.opentelemetry.sdk.common.CompletableResultCode
+import org.opendc.telemetry.compute.ComputeMetricExporter
 import org.opendc.telemetry.compute.ComputeMonitor
 import org.opendc.telemetry.compute.table.HostData
 import org.opendc.telemetry.compute.table.ServerData
@@ -31,7 +33,7 @@ import java.io.File
 /**
  * A [ComputeMonitor] that logs the events to a Parquet file.
  */
-public class ParquetExportMonitor(base: File, partition: String, bufferSize: Int) : ComputeMonitor, AutoCloseable {
+public class ParquetComputeMetricExporter(base: File, partition: String, bufferSize: Int) : ComputeMetricExporter() {
     private val serverWriter = ParquetServerDataWriter(
         File(base, "server/$partition/data.parquet").also { it.parentFile.mkdirs() },
         bufferSize
@@ -59,9 +61,11 @@ public class ParquetExportMonitor(base: File, partition: String, bufferSize: Int
         serviceWriter.write(data)
     }
 
-    override fun close() {
+    override fun shutdown(): CompletableResultCode {
         hostWriter.close()
         serviceWriter.close()
         serverWriter.close()
+
+        return CompletableResultCode.ofSuccess()
     }
 }
diff --git a/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/export/parquet/ParquetHostDataWriter.kt b/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/export/parquet/ParquetHostDataWriter.kt
index 37066a0d..98a0739e 100644
--- a/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/export/parquet/ParquetHostDataWriter.kt
+++ b/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/export/parquet/ParquetHostDataWriter.kt
@@ -54,9 +54,7 @@ public class ParquetHostDataWriter(path: File, bufferSize: Int) :
         builder["uptime"] = data.uptime
         builder["downtime"] = data.downtime
         val bootTime = data.bootTime
-        if (bootTime != null) {
-            builder["boot_time"] = bootTime.toEpochMilli()
-        }
+        builder["boot_time"] = bootTime?.toEpochMilli()
 
         builder["cpu_count"] = data.host.cpuCount
         builder["cpu_limit"] = data.cpuLimit
diff --git a/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/export/parquet/ParquetServerDataWriter.kt b/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/export/parquet/ParquetServerDataWriter.kt
index bea23d32..0d11ec23 100644
--- a/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/export/parquet/ParquetServerDataWriter.kt
+++ b/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/export/parquet/ParquetServerDataWriter.kt
@@ -56,9 +56,7 @@ public class ParquetServerDataWriter(path: File, bufferSize: Int) :
         builder["uptime"] = data.uptime
         builder["downtime"] = data.downtime
         val bootTime = data.bootTime
-        if (bootTime != null) {
-            builder["boot_time"] = bootTime.toEpochMilli()
-        }
+        builder["boot_time"] = bootTime?.toEpochMilli()
         builder["scheduling_latency"] = data.schedulingLatency
 
         builder["cpu_count"] = data.server.cpuCount