refactor(telemetry/compute): Support direct metric access

This change introduces a `ComputeMetricReader` class that can be used as
a replacement for the `CoroutineMetricReader` class when reading metrics
from the Compute service. This implementation operates directly on a
`ComputeService` instance, providing better performance.

3 files changed, 16 insertions, 18 deletions

diff --git a/opendc-web/opendc-web-runner/src/main/kotlin/org/opendc/web/runner/OpenDCRunner.kt b/opendc-web/opendc-web-runner/src/main/kotlin/org/opendc/web/runner/OpenDCRunner.kt
index a150de4e..7c0c43ed 100644
--- a/opendc-web/opendc-web-runner/src/main/kotlin/org/opendc/web/runner/OpenDCRunner.kt
+++ b/opendc-web/opendc-web-runner/src/main/kotlin/org/opendc/web/runner/OpenDCRunner.kt
@@ -23,8 +23,9 @@
 package org.opendc.web.runner
 
 import mu.KotlinLogging
+import org.opendc.compute.api.Server
 import org.opendc.compute.workload.*
-import org.opendc.compute.workload.telemetry.SdkTelemetryManager
+import org.opendc.compute.workload.telemetry.NoopTelemetryManager
 import org.opendc.compute.workload.topology.HostSpec
 import org.opendc.compute.workload.topology.Topology
 import org.opendc.compute.workload.topology.apply
@@ -35,13 +36,12 @@ import org.opendc.simulator.compute.model.ProcessingUnit
 import org.opendc.simulator.compute.power.LinearPowerModel
 import org.opendc.simulator.compute.power.SimplePowerDriver
 import org.opendc.simulator.core.runBlockingSimulation
-import org.opendc.telemetry.compute.collectServiceMetrics
-import org.opendc.telemetry.sdk.metrics.export.CoroutineMetricReader
+import org.opendc.telemetry.compute.ComputeMetricReader
 import org.opendc.web.client.runner.OpenDCRunnerClient
 import org.opendc.web.proto.runner.Job
 import org.opendc.web.proto.runner.Scenario
 import org.opendc.web.runner.internal.JobManager
-import org.opendc.web.runner.internal.WebComputeMetricExporter
+import org.opendc.web.runner.internal.WebComputeMonitor
 import java.io.File
 import java.time.Duration
 import java.util.*
@@ -180,9 +180,9 @@ public class OpenDCRunner(
         private val scenario: Scenario,
         private val repeat: Int,
         private val topology: Topology,
-    ) : RecursiveTask<WebComputeMetricExporter.Results>() {
-        override fun compute(): WebComputeMetricExporter.Results {
-            val exporter = WebComputeMetricExporter()
+    ) : RecursiveTask<WebComputeMonitor.Results>() {
+        override fun compute(): WebComputeMonitor.Results {
+            val monitor = WebComputeMonitor()
 
             // Schedule task that interrupts the simulation if it runs for too long.
             val currentThread = Thread.currentThread()
@@ -206,25 +206,24 @@ public class OpenDCRunner(
                         else
                             null
 
-                    val telemetry = SdkTelemetryManager(clock)
                     val simulator = ComputeServiceHelper(
                         coroutineContext,
                         clock,
-                        telemetry,
+                        NoopTelemetryManager(),
                         computeScheduler,
                         failureModel,
                         interferenceModel.takeIf { phenomena.interference }
                     )
-
-                    telemetry.registerMetricReader(CoroutineMetricReader(this, exporter, exportInterval = Duration.ofHours(1)))
+                    val servers = mutableListOf<Server>()
+                    val reader = ComputeMetricReader(this, clock, simulator.service, servers, monitor)
 
                     try {
                         // Instantiate the topology onto the simulator
                         simulator.apply(topology)
                         // Run workload trace
-                        simulator.run(vms, seeder.nextLong())
+                        simulator.run(vms, seeder.nextLong(), servers)
 
-                        val serviceMetrics = collectServiceMetrics(telemetry.metricProducer)
+                        val serviceMetrics = simulator.service.getSchedulerStats()
                         logger.debug {
                             "Scheduler " +
                                 "Success=${serviceMetrics.attemptsSuccess} " +
@@ -235,14 +234,14 @@ public class OpenDCRunner(
                         }
                     } finally {
                         simulator.close()
-                        telemetry.close()
+                        reader.close()
                     }
                 }
             } finally {
                 interruptTask.cancel(false)
             }
 
-            return exporter.collectResults()
+            return monitor.collectResults()
         }
     }
 
diff --git a/opendc-web/opendc-web-runner/src/main/kotlin/org/opendc/web/runner/internal/JobManager.kt b/opendc-web/opendc-web-runner/src/main/kotlin/org/opendc/web/runner/internal/JobManager.kt
index 8de0cee4..99b8aaf1 100644
--- a/opendc-web/opendc-web-runner/src/main/kotlin/org/opendc/web/runner/internal/JobManager.kt
+++ b/opendc-web/opendc-web-runner/src/main/kotlin/org/opendc/web/runner/internal/JobManager.kt
@@ -62,7 +62,7 @@ internal class JobManager(private val client: OpenDCRunnerClient) {
     /**
      * Persist the specified results.
      */
-    fun finish(id: Long, results: List<WebComputeMetricExporter.Results>) {
+    fun finish(id: Long, results: List<WebComputeMonitor.Results>) {
         client.jobs.update(
             id,
             Job.Update(
diff --git a/opendc-web/opendc-web-runner/src/main/kotlin/org/opendc/web/runner/internal/WebComputeMetricExporter.kt b/opendc-web/opendc-web-runner/src/main/kotlin/org/opendc/web/runner/internal/WebComputeMonitor.kt
index 04437a5f..69350d8c 100644
--- a/opendc-web/opendc-web-runner/src/main/kotlin/org/opendc/web/runner/internal/WebComputeMetricExporter.kt
+++ b/opendc-web/opendc-web-runner/src/main/kotlin/org/opendc/web/runner/internal/WebComputeMonitor.kt
@@ -22,7 +22,6 @@
 
 package org.opendc.web.runner.internal
 
-import org.opendc.telemetry.compute.ComputeMetricExporter
 import org.opendc.telemetry.compute.ComputeMonitor
 import org.opendc.telemetry.compute.table.HostTableReader
 import org.opendc.telemetry.compute.table.ServiceTableReader
@@ -32,7 +31,7 @@ import kotlin.math.roundToLong
 /**
  * A [ComputeMonitor] that tracks the aggregate metrics for each repeat.
  */
-internal class WebComputeMetricExporter : ComputeMetricExporter() {
+internal class WebComputeMonitor : ComputeMonitor {
     override fun record(reader: HostTableReader) {
         val slices = reader.downtime / SLICE_LENGTH