From 5fa0cf915ecf643e94a0de972125e8f862308f80 Mon Sep 17 00:00:00 2001
From: Fabian Mastenbroek <mail.fabianm@gmail.com>
Date: Wed, 22 Sep 2021 11:22:00 +0200
Subject: fix(telemetry): Ensure shutdown of exporter is called

This change updates the CoroutineMetricReader to ensure that the
exporter is shutdown when the metric reader fails or is shutdown.
---
 .../sdk/metrics/export/CoroutineMetricReader.kt    | 28 +++++++++++-----------
 1 file changed, 14 insertions(+), 14 deletions(-)

(limited to 'opendc-telemetry')

diff --git a/opendc-telemetry/opendc-telemetry-sdk/src/main/kotlin/org/opendc/telemetry/sdk/metrics/export/CoroutineMetricReader.kt b/opendc-telemetry/opendc-telemetry-sdk/src/main/kotlin/org/opendc/telemetry/sdk/metrics/export/CoroutineMetricReader.kt
index 07f0ff7f..1de235e7 100644
--- a/opendc-telemetry/opendc-telemetry-sdk/src/main/kotlin/org/opendc/telemetry/sdk/metrics/export/CoroutineMetricReader.kt
+++ b/opendc-telemetry/opendc-telemetry-sdk/src/main/kotlin/org/opendc/telemetry/sdk/metrics/export/CoroutineMetricReader.kt
@@ -22,7 +22,6 @@
 
 package org.opendc.telemetry.sdk.metrics.export
 
-import io.opentelemetry.sdk.metrics.data.MetricData
 import io.opentelemetry.sdk.metrics.export.MetricExporter
 import io.opentelemetry.sdk.metrics.export.MetricProducer
 import kotlinx.coroutines.*
@@ -54,24 +53,25 @@ public class CoroutineMetricReader(
     private val job = scope.launch {
         val intervalMs = exportInterval.toMillis()
 
-        while (isActive) {
-            delay(intervalMs)
+        try {
+            while (isActive) {
+                delay(intervalMs)
 
-            val metrics = mutableListOf<MetricData>()
-            for (producer in producers) {
-                metrics.addAll(producer.collectAllMetrics())
-            }
+                val metrics = producers.flatMap(MetricProducer::collectAllMetrics)
 
-            try {
-                val result = exporter.export(metrics)
-                result.whenComplete {
-                    if (!result.isSuccess) {
-                        logger.trace { "Exporter failed" }
+                try {
+                    val result = exporter.export(metrics)
+                    result.whenComplete {
+                        if (!result.isSuccess) {
+                            logger.warn { "Exporter failed" }
+                        }
                     }
+                } catch (cause: Throwable) {
+                    logger.warn(cause) { "Exporter threw an Exception" }
                 }
-            } catch (cause: Throwable) {
-                logger.warn(cause) { "Exporter threw an Exception" }
             }
+        } finally {
+            exporter.shutdown()
         }
     }
 
-- 
cgit v1.2.3