[k2] use metrics descriptor API

runtime-light/components/confdata/metrics.h

@@ -12,44 +12,50 @@
 
 namespace kphp::confdata::metrics {
 
-// Each instance constructs these builders once, after its script allocator is ready.
+// The component constructs these once, at its initialization: the metrics are
+// component-scoped, every instance writes the same pre-bound series.
 struct capacity final {
-  std::array<kphp::diagnostics::metric_sender, 4> m_memory{
-      kphp::diagnostics::metric_sender::metric("k2_kphp_confdata_memory").tag("kind", "limit"),
-      kphp::diagnostics::metric_sender::metric("k2_kphp_confdata_memory").tag("kind", "used"),
-      kphp::diagnostics::metric_sender::metric("k2_kphp_confdata_memory").tag("kind", "real_used"),
-      kphp::diagnostics::metric_sender::metric("k2_kphp_confdata_memory").tag("kind", "oom_threshold"),
+  // The families are declared before the sender arrays: the senders bind their series on initialization.
+  kphp::diagnostics::gauge_metric m_memory_metric{kphp::diagnostics::gauge_metric::metric("k2_kphp_confdata_memory", {"kind"})};
+  kphp::diagnostics::gauge_metric m_pieces_metric{kphp::diagnostics::gauge_metric::metric("k2_kphp_confdata_pieces", {"state"})};
+
+  std::array<kphp::diagnostics::gauge_sender, 4> m_memory{
+      m_memory_metric.series({{"kind", "limit"}}),
+      m_memory_metric.series({{"kind", "used"}}),
+      m_memory_metric.series({{"kind", "real_used"}}),
+      m_memory_metric.series({{"kind", "oom_threshold"}}),
   };
-  std::array<kphp::diagnostics::metric_sender, 2> m_pieces{
-      kphp::diagnostics::metric_sender::metric("k2_kphp_confdata_pieces").tag("state", "current"),
-      kphp::diagnostics::metric_sender::metric("k2_kphp_confdata_pieces").tag("state", "retired"),
+  std::array<kphp::diagnostics::gauge_sender, 2> m_pieces{
+      m_pieces_metric.series({{"state", "current"}}),
+      m_pieces_metric.series({{"state", "retired"}}),
   };
 };
 
 struct events final {
-  std::array<kphp::diagnostics::metric_sender, 2> m_events{
-      kphp::diagnostics::metric_sender::metric("k2_kphp_confdata_events").tag("kind", "update"),
-      kphp::diagnostics::metric_sender::metric("k2_kphp_confdata_events").tag("kind", "delete"),
+  kphp::diagnostics::counter_metric m_events_metric{kphp::diagnostics::counter_metric::metric("k2_kphp_confdata_events", {"kind"})};
+
+  std::array<kphp::diagnostics::counter_sender, 2> m_events{
+      m_events_metric.series({{"kind", "update"}}),
+      m_events_metric.series({{"kind", "delete"}}),
   };
 };
 
 struct update_failures final {
-  std::array<kphp::diagnostics::metric_sender, 5> m_failures{
-      kphp::diagnostics::metric_sender::metric("k2_kphp_confdata_update_fails").tag("error", "transport"),
-      kphp::diagnostics::metric_sender::metric("k2_kphp_confdata_update_fails").tag("error", "old_offset"),
-      kphp::diagnostics::metric_sender::metric("k2_kphp_confdata_update_fails").tag("error", "malformed_response"),
-      kphp::diagnostics::metric_sender::metric("k2_kphp_confdata_update_fails").tag("error", "not_synced"),
-      kphp::diagnostics::metric_sender::metric("k2_kphp_confdata_update_fails").tag("error", "batch_rejected"),
+  kphp::diagnostics::counter_metric m_failures_metric{kphp::diagnostics::counter_metric::metric("k2_kphp_confdata_update_fails", {"error"})};
+  kphp::diagnostics::gauge_metric m_old_offset_metric{kphp::diagnostics::gauge_metric::metric("k2_kphp_confdata_update_fails_old_offset", {})};
+
+  std::array<kphp::diagnostics::counter_sender, 5> m_failures{
+      m_failures_metric.series({{"error", "transport"}}),          m_failures_metric.series({{"error", "old_offset"}}),
+      m_failures_metric.series({{"error", "malformed_response"}}), m_failures_metric.series({{"error", "not_synced"}}),
+      m_failures_metric.series({{"error", "batch_rejected"}}),
   };
   static_assert(std::to_underlying(kphp::confdata::subscribe_error::transport) == 0);
   static_assert(std::to_underlying(kphp::confdata::subscribe_error::old_offset) == 1);
   static_assert(std::to_underlying(kphp::confdata::subscribe_error::malformed_response) == 2);
   static_assert(std::to_underlying(kphp::confdata::subscribe_error::not_synced) == 3);
   static_assert(std::to_underlying(kphp::confdata::subscribe_error::batch_rejected) == 4);
 
-  kphp::diagnostics::metric_sender m_old_offset{
-      kphp::diagnostics::metric_sender::metric("k2_kphp_confdata_update_fails_old_offset"),
-  };
+  kphp::diagnostics::gauge_sender m_old_offset{m_old_offset_metric.series({})};
 };
 
 } // namespace kphp::confdata::metrics

runtime-light/components/confdata/state/component-state.h

@@ -12,6 +12,7 @@
 #include "runtime-common/core/allocator/script-allocator.h"
 #include "runtime-common/core/std/containers.h"
 #include "runtime-light/allocator/allocator-state.h"
+#include "runtime-light/components/confdata/metrics.h"
 #include "runtime-light/k2-platform/k2-api.h"
 #include "runtime-light/stdlib/diagnostics/logs.h"
 
@@ -49,6 +50,12 @@ struct ComponentState final : private vk::not_copyable {
   size_t m_initial_instance_memory_size{DEFAULT_INIT_INSTANCE_ALLOCATOR_SIZE};
   size_t m_min_instance_extra_memory_size{DEFAULT_MIN_INSTANCE_EXTRA_MEMORY_SIZE};
 
+  // The metric families are registered and their series are bound here, at component
+  // initialization: the metrics are component-scoped, all instances write the same series.
+  kphp::confdata::metrics::events m_events_metrics;
+  kphp::confdata::metrics::capacity m_capacity_metrics;
+  kphp::confdata::metrics::update_failures m_update_failure_metrics;
+
   // === METHODS ==================================================================================
   ComponentState() noexcept;
   static auto get() noexcept -> const ComponentState&;

runtime-light/components/confdata/state/instance-state.cpp

@@ -248,21 +248,15 @@ auto InstanceState::metrics_loop() noexcept -> kphp::coro::task<> {
   for (;;) {
     k2::SystemTime now{};
     k2::system_time(std::addressof(now));
-    report_events_metrics(now.since_epoch_ns);
-    report_capacity_metrics(now.since_epoch_ns);
-    report_update_failure_metrics(now.since_epoch_ns);
+    report_events_metrics(now);
+    report_capacity_metrics(now);
+    report_update_failure_metrics(now);
     co_await m_io_scheduler.schedule(CONFDATA_METRICS_INTERVAL);
   }
 }
 
-auto InstanceState::report_events_metrics(uint64_t timestamp) noexcept -> void {
-  const auto send{[timestamp](kphp::diagnostics::metric_sender& sender, uint64_t& counter) noexcept {
-    // FIXME: needs to be fixed in platform
-    // Zero counts currently abort the platform's metrics batch flush.
-    if (counter == 0) {
-      return;
-    }
-
+auto InstanceState::report_events_metrics(k2::SystemTime timestamp) noexcept -> void {
+  const auto send{[timestamp](const kphp::diagnostics::counter_sender& sender, uint64_t& counter) noexcept {
     const auto count{static_cast<uint32_t>(std::min<uint64_t>(counter, std::numeric_limits<uint32_t>::max()))};
 
     std::ignore = sender.send_count(count, timestamp)
@@ -273,21 +267,15 @@ auto InstanceState::report_events_metrics(uint64_t timestamp) noexcept -> void {
                       });
   }};
 
-  send(m_events_metrics.m_events[0], m_update_events_count);
-  send(m_events_metrics.m_events[1], m_delete_events_count);
+  send(m_component_state.m_events_metrics.m_events[0], m_update_events_count);
+  send(m_component_state.m_events_metrics.m_events[1], m_delete_events_count);
 }
 
-auto InstanceState::report_update_failure_metrics(uint64_t timestamp) noexcept -> void {
+auto InstanceState::report_update_failure_metrics(k2::SystemTime timestamp) noexcept -> void {
   for (size_t index{}; index < m_update_failure_counts.size(); ++index) {
     auto& counter{m_update_failure_counts[index]};
-    // FIXME: needs to be fixed in platform
-    // Zero counts currently abort the platform's metrics batch flush.
-    if (counter == 0) {
-      continue;
-    }
-
     const auto count{static_cast<uint32_t>(std::min<uint64_t>(counter, std::numeric_limits<uint32_t>::max()))};
-    std::ignore = m_update_failure_metrics.m_failures[index]
+    std::ignore = m_component_state.m_update_failure_metrics.m_failures[index]
                       .send_count(count, timestamp)
                       .transform([&counter, count] noexcept { counter -= count; })
                       .or_else([](int32_t error) noexcept -> std::expected<void, int32_t> {
@@ -297,7 +285,7 @@ auto InstanceState::report_update_failure_metrics(uint64_t timestamp) noexcept -
   }
 
   std::ignore = m_update_failure_old_offset.transform([this, timestamp](auto old_offset) noexcept {
-    return m_update_failure_metrics.m_old_offset.send_value(static_cast<double>(old_offset), timestamp)
+    return m_component_state.m_update_failure_metrics.m_old_offset.send_value(static_cast<double>(old_offset), timestamp)
         .transform([this] noexcept { m_update_failure_old_offset.reset(); })
         .or_else([](int32_t error) noexcept -> std::expected<void, int32_t> {
           kphp::log::warning("failed to report 'old_offset' confdata update failure: error -> {}", error);
@@ -306,8 +294,8 @@ auto InstanceState::report_update_failure_metrics(uint64_t timestamp) noexcept -
   });
 }
 
-auto InstanceState::report_capacity_metrics(uint64_t timestamp) noexcept -> void {
-  const auto send{[timestamp](kphp::diagnostics::metric_sender& sender, size_t value) noexcept {
+auto InstanceState::report_capacity_metrics(k2::SystemTime timestamp) noexcept -> void {
+  const auto send{[timestamp](const kphp::diagnostics::gauge_sender& sender, size_t value) noexcept {
     std::ignore = sender.send_value(static_cast<double>(value), timestamp).or_else([](int32_t error) noexcept -> std::expected<void, int32_t> {
       kphp::log::warning("failed to report confdata capacity metrics: error -> {}", error);
       return std::unexpected{error};
@@ -317,16 +305,16 @@ auto InstanceState::report_capacity_metrics(uint64_t timestamp) noexcept -> void
   if (!m_confdata_pieces.empty()) {
     auto& current{m_confdata_pieces.back().storage()};
     const auto usage{current.memory_usage()};
-    send(m_capacity_metrics.m_memory[0], usage.m_capacity);
-    send(m_capacity_metrics.m_memory[1], usage.m_allocated);
-    send(m_capacity_metrics.m_memory[2], usage.m_used);
-    send(m_capacity_metrics.m_memory[3], usage.m_oom_threshold);
+    send(m_component_state.m_capacity_metrics.m_memory[0], usage.m_capacity);
+    send(m_component_state.m_capacity_metrics.m_memory[1], usage.m_allocated);
+    send(m_component_state.m_capacity_metrics.m_memory[2], usage.m_used);
+    send(m_component_state.m_capacity_metrics.m_memory[3], usage.m_oom_threshold);
   }
 
   // These are tracked logical extents, including headers, not RSS or proof of K2 allocation release.
   const size_t current_count{m_confdata_pieces.empty() ? 0U : 1U};
-  send(m_capacity_metrics.m_pieces[0], current_count);
-  send(m_capacity_metrics.m_pieces[1], m_confdata_pieces.size() - current_count);
+  send(m_component_state.m_capacity_metrics.m_pieces[0], current_count);
+  send(m_component_state.m_capacity_metrics.m_pieces[1], m_confdata_pieces.size() - current_count);
 }
 
 auto InstanceState::accept_loop() noexcept -> kphp::coro::task<> {

runtime-light/components/confdata/state/instance-state.h

@@ -56,12 +56,8 @@ struct InstanceState final : vk::not_copyable {
 
   uint64_t m_update_events_count{};
   uint64_t m_delete_events_count{};
-  kphp::confdata::metrics::events m_events_metrics;
 
-  kphp::confdata::metrics::capacity m_capacity_metrics;
-
-  kphp::confdata::metrics::update_failures m_update_failure_metrics;
-  std::array<uint64_t, std::tuple_size_v<decltype(m_update_failure_metrics.m_failures)>> m_update_failure_counts{};
+  std::array<uint64_t, std::tuple_size_v<decltype(kphp::confdata::metrics::update_failures::m_failures)>> m_update_failure_counts{};
   std::optional<int64_t> m_update_failure_old_offset; // We only report the last old offset
 
 public:
@@ -82,9 +78,9 @@ struct InstanceState final : vk::not_copyable {
   auto service_loop() noexcept -> kphp::coro::task<>;
   auto metrics_loop() noexcept -> kphp::coro::task<>;
 
-  auto report_events_metrics(uint64_t timestamp) noexcept -> void;
-  auto report_capacity_metrics(uint64_t timestamp) noexcept -> void;
-  auto report_update_failure_metrics(uint64_t timestamp) noexcept -> void;
+  auto report_events_metrics(k2::SystemTime timestamp) noexcept -> void;
+  auto report_capacity_metrics(k2::SystemTime timestamp) noexcept -> void;
+  auto report_update_failure_metrics(k2::SystemTime timestamp) noexcept -> void;
 
   auto release_reader(confdata_piece_list::iterator piece_it, kphp::confdata::storage::sample_id sample_id) noexcept -> void;
   auto serve_reader_lease(kphp::component::stream reader_stream) noexcept -> kphp::coro::task<>;

runtime-light/k2-platform/k2-api.h

@@ -322,8 +322,62 @@ inline std::expected<void, int32_t> madvise(void* addr, size_t length, int32_t a
   return {};
 }
 
-inline std::expected<void, int32_t> write_metrics(std::span<const std::byte> serialized_metric) noexcept {
-  if (auto error_code{k2_write_metrics(serialized_metric.data(), serialized_metric.size())}; error_code != k2::errno_ok) [[unlikely]] {
+using MetricLabelName = MetricLabelName;
+using MetricLabelValue = MetricLabelValue;
+using MetricLabel = MetricLabel;
+
+inline std::expected<k2::descriptor, int32_t> metrics_register_counter(std::string_view name, std::span<const MetricLabelName> label_names) noexcept {
+  k2::descriptor md{k2::INVALID_PLATFORM_DESCRIPTOR};
+  if (auto error_code{k2_metrics_register_counter(&md, name.data(), name.size(), label_names.data(), label_names.size())}; error_code != k2::errno_ok)
+      [[unlikely]] {
+    return std::unexpected{error_code};
+  }
+  return md;
+}
+
+inline std::expected<k2::descriptor, int32_t> metrics_register_gauge(std::string_view name, std::span<const MetricLabelName> label_names) noexcept {
+  k2::descriptor md{k2::INVALID_PLATFORM_DESCRIPTOR};
+  if (auto error_code{k2_metrics_register_gauge(&md, name.data(), name.size(), label_names.data(), label_names.size())}; error_code != k2::errno_ok)
+      [[unlikely]] {
+    return std::unexpected{error_code};
+  }
+  return md;
+}
+
+inline std::expected<k2::descriptor, int32_t> metrics_register_histogram(std::string_view name, std::span<const double> buckets,
+                                                                         std::span<const MetricLabelName> label_names) noexcept {
+  k2::descriptor md{k2::INVALID_PLATFORM_DESCRIPTOR};
+  if (auto error_code{k2_metrics_register_histogram(&md, name.data(), name.size(), buckets.data(), buckets.size(), label_names.data(), label_names.size())};
+      error_code != k2::errno_ok) [[unlikely]] {
+    return std::unexpected{error_code};
+  }
+  return md;
+}
+
+inline std::expected<k2::descriptor, int32_t> metrics_bind(k2::descriptor md, std::span<const MetricLabel> labels) noexcept {
+  k2::descriptor sd{k2::INVALID_PLATFORM_DESCRIPTOR};
+  if (auto error_code{k2_metrics_bind(&sd, md, labels.data(), labels.size())}; error_code != k2::errno_ok) [[unlikely]] {
+    return std::unexpected{error_code};
+  }
+  return sd;
+}
+
+inline std::expected<void, int32_t> metrics_counter_add(k2::descriptor sd, double value, uint64_t timestamp_ns) noexcept {
+  if (auto error_code{k2_metrics_counter_add(sd, value, timestamp_ns)}; error_code != k2::errno_ok) [[unlikely]] {
+    return std::unexpected{error_code};
+  }
+  return {};
+}
+
+inline std::expected<void, int32_t> metrics_gauge_set(k2::descriptor sd, double value, uint64_t timestamp_ns) noexcept {
+  if (auto error_code{k2_metrics_gauge_set(sd, value, timestamp_ns)}; error_code != k2::errno_ok) [[unlikely]] {
+    return std::unexpected{error_code};
+  }
+  return {};
+}
+
+inline std::expected<void, int32_t> metrics_histogram_observe(k2::descriptor sd, double value, uint64_t timestamp_ns) noexcept {
+  if (auto error_code{k2_metrics_histogram_observe(sd, value, timestamp_ns)}; error_code != k2::errno_ok) [[unlikely]] {
     return std::unexpected{error_code};
   }
   return {};