meter_context.cc

// Copyright The OpenTelemetry Authors
// SPDX-License-Identifier: Apache-2.0

#ifndef ENABLE_METRICS_PREVIEW
#  include "opentelemetry/sdk/metrics/meter_context.h"
#  include "opentelemetry/sdk/common/global_log_handler.h"
#  include "opentelemetry/sdk/metrics/metric_reader.h"
#  include "opentelemetry/sdk_config.h"
#  include "opentelemetry/version.h"

OPENTELEMETRY_BEGIN_NAMESPACE
namespace sdk
{
namespace metrics
{

MeterContext::MeterContext(std::unique_ptr<ViewRegistry> views,
                           opentelemetry::sdk::resource::Resource resource) noexcept
    : resource_{resource}, views_(std::move(views)), sdk_start_ts_{std::chrono::system_clock::now()}
{}

const resource::Resource &MeterContext::GetResource() const noexcept
{
  return resource_;
}

ViewRegistry *MeterContext::GetViewRegistry() const noexcept
{
  return views_.get();
}

nostd::span<std::shared_ptr<Meter>> MeterContext::GetMeters() noexcept
{
  std::lock_guard<opentelemetry::common::SpinLockMutex> guard(storage_lock_);
  return nostd::span<std::shared_ptr<Meter>>{meters_};
}

nostd::span<std::shared_ptr<CollectorHandle>> MeterContext::GetCollectors() noexcept
{
  return nostd::span<std::shared_ptr<CollectorHandle>>(collectors_);
}

opentelemetry::common::SystemTimestamp MeterContext::GetSDKStartTime() noexcept
{
  return sdk_start_ts_;
}

void MeterContext::AddMetricReader(std::shared_ptr<MetricReader> reader) noexcept
{
  auto collector = std::shared_ptr<MetricCollector>{new MetricCollector(this, reader)};
  collectors_.push_back(collector);
}

void MeterContext::AddView(std::unique_ptr<InstrumentSelector> instrument_selector,
                           std::unique_ptr<MeterSelector> meter_selector,
                           std::unique_ptr<View> view) noexcept
{
  views_->AddView(std::move(instrument_selector), std::move(meter_selector), std::move(view));
}

void MeterContext::AddMeter(std::shared_ptr<Meter> meter)
{
  std::lock_guard<opentelemetry::common::SpinLockMutex> guard(storage_lock_);
  meters_.push_back(meter);
}

bool MeterContext::Shutdown() noexcept
{
  bool result = true;
  // Shutdown only once.
  if (!shutdown_latch_.test_and_set(std::memory_order_acquire))
  {

    for (auto &collector : collectors_)
    {
      bool status = std::static_pointer_cast<MetricCollector>(collector)->Shutdown();
      result      = result && status;
    }
    if (!result)
    {
      OTEL_INTERNAL_LOG_WARN("[MeterContext::Shutdown] Unable to shutdown all metric readers");
    }
  }
  else
  {
    OTEL_INTERNAL_LOG_WARN("[MeterContext::Shutdown] Shutdown can be invoked only once.");
  }
  return result;
}

bool MeterContext::ForceFlush(std::chrono::microseconds timeout) noexcept
{
  bool result = true;
  // Simultaneous flush not allowed.
  const std::lock_guard<opentelemetry::common::SpinLockMutex> locked(forceflush_lock_);
  // Convert to nanos to prevent overflow
  auto timeout_ns = std::chrono::nanoseconds::max();
  if (std::chrono::duration_cast<std::chrono::microseconds>(timeout_ns) > timeout)
  {
    timeout_ns = std::chrono::duration_cast<std::chrono::nanoseconds>(timeout);
  }

  auto current_time = std::chrono::system_clock::now();
  std::chrono::system_clock::time_point expire_time;
  auto overflow_checker = std::chrono::system_clock::time_point::max();

  // check if the expected expire time doesn't overflow.
  if (overflow_checker - current_time > timeout_ns)
  {
    expire_time =
        current_time + std::chrono::duration_cast<std::chrono::system_clock::duration>(timeout_ns);
  }
  else
  {
    // overflow happens, reset expire time to max.
    expire_time = overflow_checker;
  }

  for (auto &collector : collectors_)
  {
    if (!std::static_pointer_cast<MetricCollector>(collector)->ForceFlush(
            std::chrono::duration_cast<std::chrono::microseconds>(timeout_ns)))
    {
      result = false;
    }

    current_time = std::chrono::system_clock::now();

    if (expire_time >= current_time)
    {
      timeout_ns = std::chrono::duration_cast<std::chrono::nanoseconds>(expire_time - current_time);
    }
    else
    {
      timeout_ns = std::chrono::nanoseconds::zero();
    }
  }
  if (!result)
  {
    OTEL_INTERNAL_LOG_WARN("[MeterContext::ForceFlush] Unable to ForceFlush all metric readers");
  }
  return result;
}

}  // namespace metrics
}  // namespace sdk
OPENTELEMETRY_END_NAMESPACE
#endif