merge: Standardize simulator metrics

This pull request standardizes the metrics emitted by the simulator based on OpenTelemetry conventions. 
From now on, all metrics exposed by the simulator are exported through OpenTelemetry 
following the recommended practices for naming, collection, etc.

**Implementation Notes**

- Improve ParquetDataWriter implementation 
- Simplify CoroutineMetricReader
- Create separate MeterProvider per service/host
- Standardize compute scheduler metrics 
- Standardize SimHost metrics 
- Use logical types for Parquet output columns 

**External Dependencies**

- Update to OpenTelemetry 1.6.0

**Breaking API Changes**

- Instead of supplying a `Meter` instances, key classes are now responsible for constructing 
   a `Meter` instance from the supplied `MeterProvider`.
- Export format has been changed to suit the outputted metrics
- Energy experiments shell has been removed

1 files changed, 63 insertions, 71 deletions

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 31215e9a..9c879e5e 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
@@ -23,11 +23,9 @@
 package org.opendc.compute.simulator
 
 import io.opentelemetry.api.metrics.MeterProvider
-import io.opentelemetry.sdk.common.CompletableResultCode
 import io.opentelemetry.sdk.metrics.SdkMeterProvider
-import io.opentelemetry.sdk.metrics.data.MetricData
-import io.opentelemetry.sdk.metrics.export.MetricExporter
 import io.opentelemetry.sdk.metrics.export.MetricProducer
+import io.opentelemetry.sdk.resources.Resource
 import kotlinx.coroutines.*
 import org.junit.jupiter.api.Assertions.assertEquals
 import org.junit.jupiter.api.BeforeEach
@@ -44,15 +42,20 @@ import org.opendc.simulator.compute.model.ProcessingUnit
 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
 import org.opendc.telemetry.sdk.metrics.export.CoroutineMetricReader
 import org.opendc.telemetry.sdk.toOtelClock
+import java.time.Duration
 import java.util.*
 import kotlin.coroutines.resume
 
 /**
  * Basic test-suite for the hypervisor.
  */
-@OptIn(ExperimentalCoroutinesApi::class)
 internal class SimHostTest {
     private lateinit var machineModel: MachineModel
 
@@ -71,24 +74,29 @@ internal class SimHostTest {
      */
     @Test
     fun testOvercommitted() = runBlockingSimulation {
-        var requestedWork = 0L
-        var grantedWork = 0L
-        var overcommittedWork = 0L
+        var idleTime = 0L
+        var activeTime = 0L
+        var stealTime = 0L
 
+        val hostId = UUID.randomUUID()
+        val hostResource = Resource.builder()
+            .put(HOST_ID, hostId.toString())
+            .build()
         val meterProvider: MeterProvider = SdkMeterProvider
             .builder()
+            .setResource(hostResource)
             .setClock(clock.toOtelClock())
             .build()
 
         val interpreter = SimResourceInterpreter(coroutineContext, clock)
         val virtDriver = SimHost(
-            uid = UUID.randomUUID(),
+            uid = hostId,
             name = "test",
             model = machineModel,
             meta = emptyMap(),
             coroutineContext,
             interpreter,
-            meterProvider.get("opendc-compute-simulator"),
+            meterProvider,
             SimFairShareHypervisorProvider()
         )
         val duration = 5 * 60L
@@ -130,26 +138,17 @@ internal class SimHostTest {
         // Setup metric reader
         val reader = CoroutineMetricReader(
             this, listOf(meterProvider as MetricProducer),
-            object : MetricExporter {
-                override fun export(metrics: Collection<MetricData>): CompletableResultCode {
-                    val metricsByName = metrics.associateBy { it.name }
-                    metricsByName["cpu.work.total"]?.let {
-                        requestedWork = it.doubleSumData.points.sumOf { point -> point.value }.toLong()
-                    }
-                    metricsByName["cpu.work.granted"]?.let {
-                        grantedWork = it.doubleSumData.points.sumOf { point -> point.value }.toLong()
+            ComputeMetricExporter(
+                clock,
+                object : ComputeMonitor {
+                    override fun record(data: HostData) {
+                        activeTime += data.cpuActiveTime
+                        idleTime += data.cpuIdleTime
+                        stealTime += data.cpuStealTime
                     }
-                    metricsByName["cpu.work.overcommit"]?.let {
-                        overcommittedWork = it.doubleSumData.points.sumOf { point -> point.value }.toLong()
-                    }
-                    return CompletableResultCode.ofSuccess()
                 }
-
-                override fun flush(): CompletableResultCode = CompletableResultCode.ofSuccess()
-
-                override fun shutdown(): CompletableResultCode = CompletableResultCode.ofSuccess()
-            },
-            exportInterval = duration * 1000L
+            ),
+            exportInterval = Duration.ofSeconds(duration)
         )
 
         coroutineScope {
@@ -175,9 +174,9 @@ internal class SimHostTest {
         reader.close()
 
         assertAll(
-            { assertEquals(4147200, requestedWork, "Requested work does not match") },
-            { assertEquals(2107200, grantedWork, "Granted work does not match") },
-            { assertEquals(2040000, overcommittedWork, "Overcommitted work does not match") },
+            { assertEquals(659, activeTime, "Active time does not match") },
+            { assertEquals(2342, idleTime, "Idle time does not match") },
+            { assertEquals(638, stealTime, "Steal time does not match") },
             { assertEquals(1500001, clock.millis()) }
         )
     }
@@ -187,27 +186,32 @@ internal class SimHostTest {
      */
     @Test
     fun testFailure() = runBlockingSimulation {
-        var requestedWork = 0L
-        var grantedWork = 0L
-        var totalTime = 0L
-        var downTime = 0L
-        var guestTotalTime = 0L
-        var guestDownTime = 0L
-
+        var activeTime = 0L
+        var idleTime = 0L
+        var uptime = 0L
+        var downtime = 0L
+        var guestUptime = 0L
+        var guestDowntime = 0L
+
+        val hostId = UUID.randomUUID()
+        val hostResource = Resource.builder()
+            .put(HOST_ID, hostId.toString())
+            .build()
         val meterProvider: MeterProvider = SdkMeterProvider
             .builder()
+            .setResource(hostResource)
             .setClock(clock.toOtelClock())
             .build()
 
         val interpreter = SimResourceInterpreter(coroutineContext, clock)
         val host = SimHost(
-            uid = UUID.randomUUID(),
+            uid = hostId,
             name = "test",
             model = machineModel,
             meta = emptyMap(),
             coroutineContext,
             interpreter,
-            meterProvider.get("opendc-compute-simulator"),
+            meterProvider,
             SimFairShareHypervisorProvider()
         )
         val duration = 5 * 60L
@@ -233,35 +237,23 @@ internal class SimHostTest {
         // Setup metric reader
         val reader = CoroutineMetricReader(
             this, listOf(meterProvider as MetricProducer),
-            object : MetricExporter {
-                override fun export(metrics: Collection<MetricData>): CompletableResultCode {
-                    val metricsByName = metrics.associateBy { it.name }
-                    metricsByName["cpu.work.total"]?.let {
-                        requestedWork = it.doubleSumData.points.sumOf { point -> point.value }.toLong()
+            ComputeMetricExporter(
+                clock,
+                object : ComputeMonitor {
+                    override fun record(data: HostData) {
+                        activeTime += data.cpuActiveTime
+                        idleTime += data.cpuIdleTime
+                        uptime += data.uptime
+                        downtime += data.downtime
                     }
-                    metricsByName["cpu.work.granted"]?.let {
-                        grantedWork = it.doubleSumData.points.sumOf { point -> point.value }.toLong()
-                    }
-                    metricsByName["host.time.total"]?.let {
-                        totalTime = it.longSumData.points.first().value
-                    }
-                    metricsByName["host.time.down"]?.let {
-                        downTime = it.longSumData.points.first().value
-                    }
-                    metricsByName["guest.time.total"]?.let {
-                        guestTotalTime = it.longSumData.points.first().value
-                    }
-                    metricsByName["guest.time.error"]?.let {
-                        guestDownTime = it.longSumData.points.first().value
+
+                    override fun record(data: ServerData) {
+                        guestUptime += data.uptime
+                        guestDowntime += data.downtime
                     }
-                    return CompletableResultCode.ofSuccess()
                 }
-
-                override fun flush(): CompletableResultCode = CompletableResultCode.ofSuccess()
-
-                override fun shutdown(): CompletableResultCode = CompletableResultCode.ofSuccess()
-            },
-            exportInterval = duration * 1000L
+            ),
+            exportInterval = Duration.ofSeconds(duration)
         )
 
         coroutineScope {
@@ -289,12 +281,12 @@ internal class SimHostTest {
         reader.close()
 
         assertAll(
-            { assertEquals(2226039, requestedWork, "Total time does not match") },
-            { assertEquals(1086039, grantedWork, "Down time does not match") },
-            { assertEquals(1200001, totalTime, "Total time does not match") },
-            { assertEquals(1200001, guestTotalTime, "Guest total time does not match") },
-            { assertEquals(5000, downTime, "Down time does not match") },
-            { assertEquals(5000, guestDownTime, "Guest down time does not match") },
+            { assertEquals(2661, idleTime, "Idle time does not match") },
+            { assertEquals(339, activeTime, "Active time does not match") },
+            { assertEquals(1195001, uptime, "Uptime does not match") },
+            { assertEquals(5000, downtime, "Downtime does not match") },
+            { assertEquals(1195000, guestUptime, "Guest uptime does not match") },
+            { assertEquals(5000, guestDowntime, "Guest downtime does not match") },
         )
     }