impl(bigtable): client schema metric updates from testing (#16385)

Author: scotthart

google/cloud/bigtable/internal/client_schema_metrics.cc

@@ -18,6 +18,8 @@
 #include "google/cloud/bigtable/internal/data_connection_impl.h"
 #include "google/cloud/bigtable/options.h"
 #include "google/cloud/bigtable/version.h"
+#include "absl/strings/str_join.h"
+#include "absl/strings/str_split.h"
 #include <opentelemetry/metrics/meter.h>
 #include <opentelemetry/semconv/incubating/cloud_attributes.h>
 #include <opentelemetry/semconv/incubating/faas_attributes.h>
@@ -66,8 +68,8 @@ LabelMap BaseLabels(ClientResourceLabels const& r) {
   return {{"project_id", r.project_id},   {"instance", r.instance},
           {"app_profile", r.app_profile}, {"client_name", r.client_name},
           {"client_uid", r.client_uid},   {"client_project", r.client_project},
-          {"location", r.location},       {"cloud_platform", r.cloud_platform},
-          {"host_id", r.host_id},         {"hostname", r.hostname}};
+          {"region", r.region},           {"cloud_platform", r.cloud_platform},
+          {"host_id", r.host_id},         {"host_name", r.host_name}};
 }
 }  // namespace
 
@@ -154,23 +156,35 @@ ClientResourceLabels MakeClientResourceLabels(
     client_project = project_id;
   }
 
+  std::string region = by_name(sc::cloud::kCloudRegion);
+  if (region.empty()) {
+    std::string zone = by_name(sc::cloud::kCloudAvailabilityZone);
+    std::vector<std::string_view> parts = absl::StrSplit(zone, '-');
+    if (parts.size() >= 3 && parts.back().size() == 1) {
+      parts.pop_back();
+      region = absl::StrJoin(parts, "-");
+    } else {
+      region = std::move(zone);
+    }
+  }
+  if (region.empty()) {
+    region = "global";
+  }
+
   ClientResourceLabels labels;
   labels.project_id = std::move(project_id);
   labels.instance = std::move(instance);
   labels.app_profile = std::move(app_profile);
   labels.client_name = "cpp.Bigtable/" + bigtable::version_string();
   labels.client_uid = client_uid;
   labels.client_project = std::move(client_project);
-  labels.location = by_name(sc::cloud::kCloudAvailabilityZone);
-  if (labels.location.empty()) {
-    labels.location = by_name(sc::cloud::kCloudRegion, "global");
-  }
+  labels.region = std::move(region);
   labels.cloud_platform = by_name(sc::cloud::kCloudPlatform, "unknown");
   labels.host_id = by_name("faas.id");
   if (labels.host_id.empty()) {
     labels.host_id = by_name(sc::host::kHostId, "unknown");
   }
-  labels.hostname = by_name(sc::host::kHostName);
+  labels.host_name = by_name(sc::host::kHostName);
   return labels;
 }
 
@@ -193,9 +207,16 @@ void OutstandingRpcs::StubSelection(
     StubSelectionParams const& p) {
   ClientOutstandingRpcLabels data_labels{p.transport_type,
                                          p.channel_pool_lb_policy, p.streaming};
-  outstanding_rpcs_->Record(static_cast<double>(p.outstanding_rpcs),
-                            IntoLabelMap(resource_labels_, data_labels, {}),
-                            context);
+  // `channel_pool_lb_policy` is filtered out because the Cloud Monitoring
+  // metric descriptor for
+  // `bigtable.googleapis.com/internal/client/connection_pool/outstanding_rpcs`
+  // does not recognize this label.
+  static auto const* const kFilteredDataLabels =
+      new std::set<std::string>{"channel_pool_lb_policy"};
+  outstanding_rpcs_->Record(
+      static_cast<double>(p.outstanding_rpcs),
+      IntoLabelMap(resource_labels_, data_labels, *kFilteredDataLabels),
+      context);
 }
 
 std::unique_ptr<ClientSchemaMetric> OutstandingRpcs::clone(
@@ -205,41 +226,92 @@ std::unique_ptr<ClientSchemaMetric> OutstandingRpcs::clone(
   return m;
 }
 
+#if OPENTELEMETRY_ABI_VERSION_NO >= 2
 DirectAccessCompatibility::DirectAccessCompatibility(
     std::string const& instrumentation_scope,
     opentelemetry::nostd::shared_ptr<
-        opentelemetry::metrics::MeterProvider> const& provider)
-    : gauge_(
-#if OPENTELEMETRY_ABI_VERSION_NO >= 2
-          provider
-              ->GetMeter(instrumentation_scope,
-                         kMeterInstrumentationScopeVersion)
-              ->CreateInt64Gauge(
-                  "direct_access/compatible",
-                  "Compatibility check result for Bigtable DirectPath.", "1")
-#else
-          provider
-              ->GetMeter(instrumentation_scope,
-                         kMeterInstrumentationScopeVersion)
-              ->CreateDoubleHistogram(
-                  "direct_access/compatible",
-                  "Compatibility check result for Bigtable DirectPath.", "1")
-#endif
-      ) {
-}
+        opentelemetry::metrics::MeterProvider> const& provider,
+    ClientResourceLabels resource_labels)
+    : resource_labels_(std::move(resource_labels)),
+      gauge_(provider
+                 ->GetMeter(instrumentation_scope,
+                            kMeterInstrumentationScopeVersion)
+                 ->CreateInt64Gauge(
+                     "direct_access/compatible",
+                     "Compatibility check result for Bigtable DirectPath.",
+                     "1")) {}
 
 void DirectAccessCompatibility::Record(
     opentelemetry::context::Context const& context, std::int64_t value,
     DirectAccessCompatibilityLabels const& data_labels) {
-#if OPENTELEMETRY_ABI_VERSION_NO >= 2
-  gauge_->Record(value, IntoLabelMap(resource_labels_, data_labels, {}),
-                 context);
+  if (gauge_ == nullptr) return;
+  LabelMap const labels = IntoLabelMap(resource_labels_, data_labels, {});
+  gauge_->Record(value, labels, context);
+}
 #else
-  gauge_->Record(static_cast<double>(value),
-                 IntoLabelMap(resource_labels_, data_labels, {}), context);
-#endif
+void DirectAccessCompatibility::ObserveCallback(
+    opentelemetry::metrics::ObserverResult observer_result, void* state_ptr) {
+  auto* state = static_cast<State*>(state_ptr);
+  if (state == nullptr) return;
+  std::lock_guard<std::mutex> lk(state->mu);
+  if (!state->has_value) return;
+
+  if (opentelemetry::nostd::holds_alternative<opentelemetry::nostd::shared_ptr<
+          opentelemetry::metrics::ObserverResultT<std::int64_t> > >(
+          observer_result)) {
+    auto observer = opentelemetry::nostd::get<opentelemetry::nostd::shared_ptr<
+        opentelemetry::metrics::ObserverResultT<std::int64_t> > >(
+        observer_result);
+    observer->Observe(state->value, state->labels);
+  }
+}
+
+DirectAccessCompatibility::DirectAccessCompatibility(
+    std::string const& instrumentation_scope,
+    opentelemetry::nostd::shared_ptr<
+        opentelemetry::metrics::MeterProvider> const& provider,
+    ClientResourceLabels resource_labels)
+    : resource_labels_(std::move(resource_labels)),
+      state_(std::make_shared<State>()),
+      gauge_(provider
+                 ->GetMeter(instrumentation_scope,
+                            kMeterInstrumentationScopeVersion)
+                 ->CreateInt64ObservableGauge(
+                     "direct_access/compatible",
+                     "Compatibility check result for Bigtable DirectPath.",
+                     "1")) {
+  if (gauge_ != nullptr) {
+    gauge_->AddCallback(ObserveCallback, state_.get());
+  }
+}
+
+DirectAccessCompatibility::~DirectAccessCompatibility() {
+  if (gauge_ != nullptr) {
+    gauge_->RemoveCallback(ObserveCallback, state_.get());
+  }
+}
+
+DirectAccessCompatibility::DirectAccessCompatibility(
+    DirectAccessCompatibility const& other)
+    : resource_labels_(other.resource_labels_),
+      state_(std::make_shared<State>()),
+      gauge_(other.gauge_) {
+  if (gauge_ != nullptr) {
+    gauge_->AddCallback(ObserveCallback, state_.get());
+  }
 }
 
+void DirectAccessCompatibility::Record(
+    opentelemetry::context::Context const&, std::int64_t value,
+    DirectAccessCompatibilityLabels const& data_labels) {
+  LabelMap labels = IntoLabelMap(resource_labels_, data_labels, {});
+  std::lock_guard<std::mutex> lk(state_->mu);
+  state_->value = value;
+  state_->labels = std::move(labels);
+  state_->has_value = true;
+}
+#endif
+
 std::unique_ptr<ClientSchemaMetric> DirectAccessCompatibility::clone(
     ClientResourceLabels const& resource_labels) const {
   auto m = std::make_unique<DirectAccessCompatibility>(*this);

google/cloud/bigtable/internal/client_schema_metrics.h

@@ -19,10 +19,13 @@
 
 #include "google/cloud/bigtable/internal/metrics.h"
 #include "google/cloud/options.h"
+#include <opentelemetry/metrics/async_instruments.h>
 #include <opentelemetry/metrics/meter_provider.h>
+#include <opentelemetry/metrics/observer_result.h>
 #include <opentelemetry/metrics/sync_instruments.h>
 #include <opentelemetry/sdk/resource/resource.h>
 #include <memory>
+#include <mutex>
 #include <set>
 #include <string>
 #include <unordered_map>
@@ -39,10 +42,10 @@ struct ClientResourceLabels {
   std::string client_name;
   std::string client_uid;
   std::string client_project;
-  std::string location;
+  std::string region;
   std::string cloud_platform;
   std::string host_id;
-  std::string hostname;
+  std::string host_name;
 };
 
 struct ClientOutstandingRpcLabels {
@@ -95,10 +98,29 @@ class OutstandingRpcs : public ClientSchemaMetric {
 
 class DirectAccessCompatibility : public ClientSchemaMetric {
  public:
+#if OPENTELEMETRY_ABI_VERSION_NO < 2
+  struct State {
+    std::mutex mu;
+    bool has_value = false;
+    std::int64_t value = 0;
+    LabelMap labels;
+  };
+#endif
+
   DirectAccessCompatibility(
       std::string const& instrumentation_scope,
       opentelemetry::nostd::shared_ptr<
-          opentelemetry::metrics::MeterProvider> const& provider);
+          opentelemetry::metrics::MeterProvider> const& provider,
+      ClientResourceLabels resource_labels);
+#if OPENTELEMETRY_ABI_VERSION_NO < 2
+  ~DirectAccessCompatibility() override;
+
+  DirectAccessCompatibility(DirectAccessCompatibility const& other);
+  DirectAccessCompatibility& operator=(DirectAccessCompatibility const&) =
+      delete;
+  DirectAccessCompatibility(DirectAccessCompatibility&&) = delete;
+  DirectAccessCompatibility& operator=(DirectAccessCompatibility&&) = delete;
+#endif
 
   void Record(opentelemetry::context::Context const& context,
               std::int64_t value,
@@ -108,12 +130,17 @@ class DirectAccessCompatibility : public ClientSchemaMetric {
       ClientResourceLabels const& resource_labels) const override;
 
  private:
-  ClientResourceLabels resource_labels_;
 #if OPENTELEMETRY_ABI_VERSION_NO >= 2
+  ClientResourceLabels resource_labels_;
   opentelemetry::nostd::shared_ptr<opentelemetry::metrics::Gauge<std::int64_t>>
       gauge_;
 #else
-  opentelemetry::nostd::shared_ptr<opentelemetry::metrics::Histogram<double>>
+  static void ObserveCallback(
+      opentelemetry::metrics::ObserverResult observer_result, void* state_ptr);
+
+  ClientResourceLabels resource_labels_;
+  std::shared_ptr<State> state_;
+  opentelemetry::nostd::shared_ptr<opentelemetry::metrics::ObservableInstrument>
       gauge_;
 #endif
 };