Collect and Export metric data before PeriodicMetricReader shutdown. (#…

…1860)
open-telemetry · Jan 4, 2023 · d557191 · d557191
1 parent 61bc860
commit d557191
Show file tree

Hide file tree

Showing 3 changed files with 44 additions and 29 deletions.
diff --git a/sdk/include/opentelemetry/sdk/metrics/export/periodic_exporting_metric_reader.h b/sdk/include/opentelemetry/sdk/metrics/export/periodic_exporting_metric_reader.h
@@ -57,6 +57,7 @@ class PeriodicExportingMetricReader : public MetricReader
   std::chrono::milliseconds export_timeout_millis_;
 
   void DoBackgroundWork();
+  bool CollectAndExportOnce();
 
   /* The background worker thread */
   std::thread worker_thread_;

diff --git a/sdk/src/metrics/export/periodic_exporting_metric_reader.cc b/sdk/src/metrics/export/periodic_exporting_metric_reader.cc
@@ -51,40 +51,52 @@ void PeriodicExportingMetricReader::DoBackgroundWork()
   std::unique_lock<std::mutex> lk(cv_m_);
   do
   {
-    if (IsShutdown())
+    auto start  = std::chrono::steady_clock::now();
+    auto status = CollectAndExportOnce();
+    if (!status)
     {
-      break;
+      OTEL_INTERNAL_LOG_ERROR("[Periodic Exporting Metric Reader]  Collect-Export Cycle Failure.")
     }
-    std::atomic<bool> cancel_export_for_timeout{false};
-    auto start          = std::chrono::steady_clock::now();
-    auto future_receive = std::async(std::launch::async, [this, &cancel_export_for_timeout] {
-      Collect([this, &cancel_export_for_timeout](ResourceMetrics &metric_data) {
-        if (cancel_export_for_timeout)
-        {
-          OTEL_INTERNAL_LOG_ERROR(
-              "[Periodic Exporting Metric Reader] Collect took longer configured time: "
-              << export_timeout_millis_.count() << " ms, and timed out");
-          return false;
-        }
-        this->exporter_->Export(metric_data);
-        return true;
-      });
-    });
-    std::future_status status;
-    do
-    {
-      status = future_receive.wait_for(std::chrono::milliseconds(export_timeout_millis_));
-      if (status == std::future_status::timeout)
-      {
-        cancel_export_for_timeout = true;
-        break;
-      }
-    } while (status != std::future_status::ready);
     auto end            = std::chrono::steady_clock::now();
     auto export_time_ms = std::chrono::duration_cast<std::chrono::milliseconds>(end - start);
     auto remaining_wait_interval_ms = export_interval_millis_ - export_time_ms;
     cv_.wait_for(lk, remaining_wait_interval_ms);
-  } while (true);
+  } while (IsShutdown() != true);
+  // One last Collect and Export before shutdown
+  auto status = CollectAndExportOnce();
+  if (!status)
+  {
+    OTEL_INTERNAL_LOG_ERROR("[Periodic Exporting Metric Reader]  Collect-Export Cycle Failure.")
+  }
+}
+
+bool PeriodicExportingMetricReader::CollectAndExportOnce()
+{
+  std::atomic<bool> cancel_export_for_timeout{false};
+  auto future_receive = std::async(std::launch::async, [this, &cancel_export_for_timeout] {
+    Collect([this, &cancel_export_for_timeout](ResourceMetrics &metric_data) {
+      if (cancel_export_for_timeout)
+      {
+        OTEL_INTERNAL_LOG_ERROR(
+            "[Periodic Exporting Metric Reader] Collect took longer configured time: "
+            << export_timeout_millis_.count() << " ms, and timed out");
+        return false;
+      }
+      this->exporter_->Export(metric_data);
+      return true;
+    });
+  });
+  std::future_status status;
+  do
+  {
+    status = future_receive.wait_for(std::chrono::milliseconds(export_timeout_millis_));
+    if (status == std::future_status::timeout)
+    {
+      cancel_export_for_timeout = true;
+      break;
+    }
+  } while (status != std::future_status::ready);
+  return true;
 }
 
 bool PeriodicExportingMetricReader::OnForceFlush(std::chrono::microseconds timeout) noexcept

diff --git a/sdk/src/metrics/metric_reader.cc b/sdk/src/metrics/metric_reader.cc
@@ -28,10 +28,12 @@ bool MetricReader::Collect(
     OTEL_INTERNAL_LOG_WARN(
         "MetricReader::Collect Cannot invoke Collect(). No MetricProducer registered for "
         "collection!")
+    return false;
   }
   if (IsShutdown())
   {
-    OTEL_INTERNAL_LOG_WARN("MetricReader::Collect Cannot invoke Collect(). Shutdown in progress!");
+    // Continue with warning, and let pull and push MetricReader state machine handle this.
+    OTEL_INTERNAL_LOG_WARN("MetricReader::Collect invoked while Shutdown in progress!");
   }
 
   return metric_producer_->Collect(callback);