* Main export loop for exporting OTel metrics to the configured backend. * * This method runs in a loop, waiting for new metrics to be available for export. In case of export failures, * it retries the export as per OTel spec[^1] with exponential backoff until the export succeeds or the exporter * is stopped. After a successful export, it clears the exported metrics from @c m_Request to make r
| 299 | * [^1]: https://opentelemetry.io/docs/specs/otlp/#retryable-response-codes |
| 300 | */ |
| 301 | void OTel::ExportLoop(boost::asio::yield_context& yc) |
| 302 | { |
| 303 | Defer cleanup{[this] { |
| 304 | m_Request.reset(); |
| 305 | m_ExportAsioCV.NotifyAll(); |
| 306 | ResetExporting(true /* notify all */); |
| 307 | }}; |
| 308 | |
| 309 | namespace ch = std::chrono; |
| 310 | |
| 311 | while (true) { |
| 312 | // Wait for a new export request to be available. If the exporter is stopped while waiting, |
| 313 | // we will be notified without a new request, so we also check the stopped state here to |
| 314 | // avoid waiting indefinitely in that case. |
| 315 | while (!m_Request && !m_Stopped) { |
| 316 | m_ExportAsioCV.Wait(yc); |
| 317 | } |
| 318 | |
| 319 | if (m_Stopped) { |
| 320 | break; |
| 321 | } |
| 322 | |
| 323 | if (!m_Stream) { |
| 324 | Connect(yc); |
| 325 | } |
| 326 | |
| 327 | for (uint64_t attempt = 1; m_Stream && !m_Stopped; ++attempt) { |
| 328 | try { |
| 329 | ExportImpl(yc); |
| 330 | m_Request.reset(); |
| 331 | m_ExportAsioCV.NotifyAll(); |
| 332 | ResetExporting(false /* notify one */); |
| 333 | break; |
| 334 | } catch (const RetryableExportError& ex) { |
| 335 | ch::milliseconds retryAfter; |
| 336 | if (auto throttle = ex.Throttle(); throttle > 0ms) { |
| 337 | retryAfter = throttle; |
| 338 | } else { |
| 339 | retryAfter = Backoff(attempt); |
| 340 | } |
| 341 | |
| 342 | Log(LogWarning, "OTelExporter") |
| 343 | << "Failed to export metrics to OpenTelemetry backend (attempt #" << attempt << "). Retrying in " |
| 344 | << retryAfter.count() << "ms."; |
| 345 | |
| 346 | boost::system::error_code ec; |
| 347 | m_RetryExportAndConnTimer.expires_after(retryAfter); |
| 348 | m_RetryExportAndConnTimer.async_wait(yc[ec]); |
| 349 | } catch (const std::exception& ex) { |
| 350 | LogSeverity severity = LogCritical; |
| 351 | const auto* ser{dynamic_cast<const boost::system::system_error*>(&ex)}; |
| 352 | // Since we don't have a proper connection health check mechanism, we assume that certain errors |
| 353 | // indicate a broken connection and force a reconnect in those cases. For the `end_of_stream` case, |
| 354 | // we downgrade the log severity to debug level since this is a normal occurrence when using an OTEL |
| 355 | // collector compatible backend that don't honor keep-alive connections (e.g., OpenSearch Data Prepper). |
| 356 | if (m_Stopped || (ser && (ser->code() == http::error::end_of_stream || ser->code() == boost::asio::error::broken_pipe))) { |
| 357 | severity = LogDebug; |
| 358 | } |