merge: Expose metrics directly to user (#80)

This pull request adds the ability to access the metrics of resources modeled
by the OpenDC Compute, Workflow, FaaS, and TensorFlow services directly from
their corresponding interfaces. Previously, users would have to interact with
OpenTelemetry to obtain these values, which is complex and provides
significant overhead.

With this pull request, users can access the metrics of all cloud resources
modeled by OpenDC via methods such as `getSchedulerStats()`, etc.

** Breaking Changes **

- `ComputeService.hostCount` removed in favour of `ComputeService.hosts.size`

1 files changed, 7 insertions, 32 deletions

diff --git a/opendc-workflow/opendc-workflow-service/src/test/kotlin/org/opendc/workflow/service/WorkflowServiceTest.kt b/opendc-workflow/opendc-workflow-service/src/test/kotlin/org/opendc/workflow/service/WorkflowServiceTest.kt
index 1fd332b9..d5f06587 100644
--- a/opendc-workflow/opendc-workflow-service/src/test/kotlin/org/opendc/workflow/service/WorkflowServiceTest.kt
+++ b/opendc-workflow/opendc-workflow-service/src/test/kotlin/org/opendc/workflow/service/WorkflowServiceTest.kt
@@ -22,7 +22,6 @@
 
 package org.opendc.workflow.service
 
-import io.opentelemetry.sdk.metrics.export.MetricProducer
 import org.junit.jupiter.api.Assertions.assertEquals
 import org.junit.jupiter.api.DisplayName
 import org.junit.jupiter.api.Test
@@ -66,7 +65,6 @@ internal class WorkflowServiceTest {
     @Test
     fun testTrace() = runBlockingSimulation {
         // Configure the ComputeService that is responsible for mapping virtual machines onto physical hosts
-        val HOST_COUNT = 4
         val computeScheduler = FilterScheduler(
             filters = listOf(ComputeFilter(), VCpuFilter(1.0), RamFilter(1.0)),
             weighers = listOf(VCpuWeigher(1.0, multiplier = 1.0))
@@ -74,7 +72,8 @@ internal class WorkflowServiceTest {
 
         val computeHelper = ComputeServiceHelper(coroutineContext, clock, NoopTelemetryManager(), computeScheduler, schedulingQuantum = Duration.ofSeconds(1))
 
-        repeat(HOST_COUNT) { computeHelper.registerHost(createHostSpec(it)) }
+        val hostCount = 4
+        repeat(hostCount) { computeHelper.registerHost(createHostSpec(it)) }
 
         // Configure the WorkflowService that is responsible for scheduling the workflow tasks onto machines
         val workflowScheduler = WorkflowSchedulerSpec(
@@ -98,13 +97,13 @@ internal class WorkflowServiceTest {
             computeHelper.close()
         }
 
-        val metrics = collectMetrics(workflowHelper.metricProducer)
+        val metrics = workflowHelper.service.getSchedulerStats()
 
         assertAll(
-            { assertEquals(758, metrics.jobsSubmitted, "No jobs submitted") },
-            { assertEquals(0, metrics.jobsActive, "Not all submitted jobs started") },
-            { assertEquals(metrics.jobsSubmitted, metrics.jobsFinished, "Not all started jobs finished") },
-            { assertEquals(0, metrics.tasksActive, "Not all started tasks finished") },
+            { assertEquals(758, metrics.workflowsSubmitted, "No jobs submitted") },
+            { assertEquals(0, metrics.workflowsRunning, "Not all submitted jobs started") },
+            { assertEquals(metrics.workflowsSubmitted, metrics.workflowsFinished, "Not all started jobs finished") },
+            { assertEquals(0, metrics.tasksRunning, "Not all started tasks finished") },
             { assertEquals(metrics.tasksSubmitted, metrics.tasksFinished, "Not all started tasks finished") },
             { assertEquals(32649883L, clock.millis()) { "Total duration incorrect" } }
         )
@@ -130,28 +129,4 @@ internal class WorkflowServiceTest {
             SimSpaceSharedHypervisorProvider()
         )
     }
-
-    class WorkflowMetrics {
-        var jobsSubmitted = 0L
-        var jobsActive = 0L
-        var jobsFinished = 0L
-        var tasksSubmitted = 0L
-        var tasksActive = 0L
-        var tasksFinished = 0L
-    }
-
-    /**
-     * Collect the metrics of the workflow service.
-     */
-    private fun collectMetrics(metricProducer: MetricProducer): WorkflowMetrics {
-        val metrics = metricProducer.collectAllMetrics().associateBy { it.name }
-        val res = WorkflowMetrics()
-        res.jobsSubmitted = metrics["jobs.submitted"]?.longSumData?.points?.last()?.value ?: 0
-        res.jobsActive = metrics["jobs.active"]?.longSumData?.points?.last()?.value ?: 0
-        res.jobsFinished = metrics["jobs.finished"]?.longSumData?.points?.last()?.value ?: 0
-        res.tasksSubmitted = metrics["tasks.submitted"]?.longSumData?.points?.last()?.value ?: 0
-        res.tasksActive = metrics["tasks.active"]?.longSumData?.points?.last()?.value ?: 0
-        res.tasksFinished = metrics["tasks.finished"]?.longSumData?.points?.last()?.value ?: 0
-        return res
-    }
 }