impl(bigtable): add options for enabling directpath and directpath metrics

google/cloud/bigtable/data_connection.cc

@@ -18,6 +18,7 @@
 #include "google/cloud/bigtable/internal/data_connection_impl.h"
 #include "google/cloud/bigtable/internal/data_tracing_connection.h"
 #include "google/cloud/bigtable/internal/defaults.h"
+#include "google/cloud/bigtable/internal/grpc_metrics_exporter.h"
 #include "google/cloud/bigtable/internal/mutate_rows_limiter.h"
 #include "google/cloud/bigtable/internal/partial_result_set_source.h"
 #include "google/cloud/bigtable/internal/row_reader_impl.h"
@@ -29,7 +30,12 @@
 #include "google/cloud/grpc_options.h"
 #include "google/cloud/internal/opentelemetry.h"
 #include "google/cloud/internal/unified_grpc_credentials.h"
+#ifdef GOOGLE_CLOUD_CPP_BIGTABLE_WITH_OTEL_METRICS
+#include "google/cloud/monitoring/v3/metric_connection.h"
+#include "google/cloud/internal/random.h"
+#endif  // GOOGLE_CLOUD_CPP_BIGTABLE_WITH_OTEL_METRICS
 #include <memory>
+#include <mutex>
 
 namespace google {
 namespace cloud {
@@ -194,6 +200,41 @@ std::shared_ptr<DataConnection> MakeDataConnection(Options options) {
       background->cq(), options);
   auto limiter =
       bigtable_internal::MakeMutateRowsLimiter(background->cq(), options);
+
+  std::shared_ptr<monitoring_v3::MetricServiceConnection>
+      metric_service_connection;
+  std::unique_ptr<bigtable_internal::OperationContextFactory>
+      operation_context_factory;
+
+#ifdef GOOGLE_CLOUD_CPP_BIGTABLE_WITH_OTEL_METRICS
+  if (options.get<EnableMetricsOption>()) {
+    metric_service_connection = monitoring_v3::MakeMetricServiceConnection(
+        internal::MetricsExporterConnectionOptions(options));
+    auto gen = google::cloud::internal::MakeDefaultPRNG();
+    std::string client_uid = google::cloud::internal::Sample(
+        gen, 16, "abcdefghijklmnopqrstuvwxyz0123456789");
+#ifdef GOOGLE_CLOUD_CPP_BIGTABLE_WITH_GRPC_OTEL_METRICS
+    if (bigtable::internal::IsDirectPath(options) &&
+        options.has<bigtable_internal::InstanceChannelAffinityOption>() &&
+        options.get<experimental::DirectPathMetricsModeOption>() ==
+            experimental::DirectPathMetricsMode::kEnabled) {
+      bigtable_internal::EnableGrpcMetrics(metric_service_connection, options,
+                                           client_uid);
+    }
+#endif  // GOOGLE_CLOUD_CPP_BIGTABLE_WITH_GRPC_OTEL_METRICS
+    operation_context_factory =
+        std::make_unique<bigtable_internal::MetricsOperationContextFactory>(
+            std::move(client_uid), std::move(metric_service_connection),
+            options);
+  } else {
+    operation_context_factory =
+        std::make_unique<bigtable_internal::SimpleOperationContextFactory>();
+  }
+#else
+  operation_context_factory =
+      std::make_unique<bigtable_internal::SimpleOperationContextFactory>();
+#endif  // GOOGLE_CLOUD_CPP_BIGTABLE_WITH_OTEL_METRICS
+
   std::shared_ptr<DataConnection> conn;
 
   if (options.has<bigtable_internal::InstanceChannelAffinityOption>()) {
@@ -212,14 +253,16 @@ std::shared_ptr<DataConnection> MakeDataConnection(Options options) {
         std::move(background),
         std::make_unique<bigtable_internal::StubManager>(
             std::move(affinity_stubs), stub_creation_fn),
-        std::move(limiter), std::move(options));
+        std::move(operation_context_factory), std::move(limiter),
+        std::move(options));
   } else {
     auto stub = bigtable_internal::CreateBigtableStub(
         std::move(auth), background->cq(), options);
     conn = std::make_shared<bigtable_internal::DataConnectionImpl>(
         std::move(background),
         std::make_unique<bigtable_internal::StubManager>(std::move(stub)),
-        std::move(limiter), std::move(options));
+        std::move(operation_context_factory), std::move(limiter),
+        std::move(options));
   }
   if (google::cloud::internal::TracingEnabled(conn->options())) {
     conn = bigtable_internal::MakeDataTracingConnection(std::move(conn));

google/cloud/bigtable/data_connection_test.cc

@@ -109,6 +109,46 @@ TEST(MakeDataConnection, TestingOtelCollectorEnvVar) {
   auto conn = MakeDataConnection(TestOptions().set<EnableMetricsOption>(true));
   EXPECT_NE(conn, nullptr);
 }
+
+TEST(MakeDataConnection, DirectPathMetricsModeOptionDisabled) {
+  testing_util::ScopedEnvironment env("GOOGLE_CLOUD_CPP_TESTING_OTEL_COLLECTOR",
+                                      "1");
+  InstanceResource instance_a{Project("my-project"), "instance-a"};
+  auto conn = MakeDataConnection(
+      {instance_a}, TestOptions()
+                        .set<EnableMetricsOption>(true)
+                        .set<experimental::DirectPathMetricsModeOption>(
+                            experimental::DirectPathMetricsMode::kDisabled));
+  EXPECT_NE(conn, nullptr);
+}
+
+TEST(MakeDataConnection, DirectPathMetricsModeOptionEnabled) {
+  testing_util::ScopedEnvironment env("GOOGLE_CLOUD_CPP_TESTING_OTEL_COLLECTOR",
+                                      "1");
+  InstanceResource instance_a{Project("my-project"), "instance-a"};
+  auto conn = MakeDataConnection(
+      {instance_a}, TestOptions()
+                        .set<EnableMetricsOption>(true)
+                        .set<experimental::DirectPathModeOption>(
+                            experimental::DirectPathMode::kEnabled)
+                        .set<experimental::DirectPathMetricsModeOption>(
+                            experimental::DirectPathMetricsMode::kEnabled));
+  EXPECT_NE(conn, nullptr);
+}
+
+TEST(MakeDataConnection, DirectPathModeOptionDisabledNoGrpcMetrics) {
+  testing_util::ScopedEnvironment env("GOOGLE_CLOUD_CPP_TESTING_OTEL_COLLECTOR",
+                                      "1");
+  InstanceResource instance_a{Project("my-project"), "instance-a"};
+  auto conn = MakeDataConnection(
+      {instance_a}, TestOptions()
+                        .set<EnableMetricsOption>(true)
+                        .set<experimental::DirectPathModeOption>(
+                            experimental::DirectPathMode::kDisabled)
+                        .set<experimental::DirectPathMetricsModeOption>(
+                            experimental::DirectPathMetricsMode::kEnabled));
+  EXPECT_NE(conn, nullptr);
+}
 #endif
 
 }  // namespace

google/cloud/bigtable/internal/bigtable_stub_factory.cc

@@ -64,8 +64,8 @@ std::string CreateFeaturesMetadata(bool is_direct_path) {
   return internal::UrlsafeBase64EncodeWithPadding(proto.SerializeAsString());
 }
 
-std::string FeaturesMetadata() {
-  if (bigtable::internal::IsDirectPath()) {
+std::string const& FeaturesMetadata(Options const& options) {
+  if (bigtable::internal::IsDirectPath(options)) {
     static auto const* const kDirectPathFeatures =
         new std::string(CreateFeaturesMetadata(true));
     return *kDirectPathFeatures;
@@ -84,7 +84,7 @@ std::shared_ptr<BigtableStub> ApplyCommonDecorators(
   stub = std::make_shared<BigtableMetadata>(
       std::move(stub),
       std::multimap<std::string, std::string>{
-          {"bigtable-features", FeaturesMetadata()}},
+          {"bigtable-features", FeaturesMetadata(options)}},
       internal::HandCraftedLibClientHeader());
   if (internal::Contains(options.get<LoggingComponentsOption>(), "rpc")) {
     GCP_LOG(INFO) << "Enabled logging for gRPC calls";

google/cloud/bigtable/internal/data_connection_impl.cc

@@ -18,7 +18,7 @@
 #include "google/cloud/bigtable/internal/async_row_sampler.h"
 #include "google/cloud/bigtable/internal/bulk_mutator.h"
 #include "google/cloud/bigtable/internal/default_row_reader.h"
-#include "google/cloud/bigtable/internal/defaults.h"
+#include "google/cloud/bigtable/internal/grpc_metrics_exporter.h"
 #include "google/cloud/bigtable/internal/logging_result_set_reader.h"
 #include "google/cloud/bigtable/internal/operation_context.h"
 #include "google/cloud/bigtable/internal/partial_result_set_reader.h"
@@ -39,12 +39,8 @@
 #include "google/cloud/internal/async_retry_loop.h"
 #include "google/cloud/internal/getenv.h"
 #include "google/cloud/internal/make_status.h"
-#include "google/cloud/internal/random.h"
 #include "google/cloud/internal/retry_loop.h"
 #include "google/cloud/internal/streaming_read_rpc.h"
-#ifdef GOOGLE_CLOUD_CPP_BIGTABLE_WITH_OTEL_METRICS
-#include "google/cloud/monitoring/v3/metric_connection.h"
-#endif  // GOOGLE_CLOUD_CPP_BIGTABLE_WITH_OTEL_METRICS
 #include "google/cloud/universe_domain_options.h"
 #include <memory>
 #include <string>
@@ -201,23 +197,6 @@ std::string_view InstanceNameFromTableName(std::string_view table_name) {
   if (pos == std::string_view::npos) return {};
   return table_name.substr(0, pos);
 }
-
-#ifdef GOOGLE_CLOUD_CPP_BIGTABLE_WITH_OTEL_METRICS
-Options MetricsExporterConnectionOptions(Options options) {
-  // We start with a copy of the client options to preserve credentials and
-  // universe domain, but we must unset Bigtable-specific endpoints/authorities
-  // to allow default Monitoring defaults.
-  options.unset<EndpointOption>();
-  options.unset<AuthorityOption>();
-  auto collector = internal::GetEnv("GOOGLE_CLOUD_CPP_TESTING_OTEL_COLLECTOR");
-  if (collector.has_value()) {
-    // Override credentials when using the otel_collector test server.
-    options.set<UnifiedCredentialsOption>(MakeInsecureCredentials());
-  }
-  return options;
-}
-#endif  // GOOGLE_CLOUD_CPP_BIGTABLE_WITH_OTEL_METRICS
-
 }  // namespace
 
 bigtable::Row TransformReadModifyWriteRowResponse(
@@ -240,45 +219,6 @@ bigtable::Row TransformReadModifyWriteRowResponse(
   return bigtable::Row(std::move(*row.mutable_key()), std::move(cells));
 }
 
-DataConnectionImpl::DataConnectionImpl(
-    std::unique_ptr<BackgroundThreads> background,
-    std::unique_ptr<StubManager> stub_manager,
-    std::shared_ptr<MutateRowsLimiter> limiter, Options options)
-    : background_(std::move(background)),
-      stub_manager_(std::move(stub_manager)),
-      limiter_(std::move(limiter)),
-      options_(MergeOptions(std::move(options), DataConnection::options())) {
-#ifdef GOOGLE_CLOUD_CPP_BIGTABLE_WITH_OTEL_METRICS
-  if (options_.get<bigtable::EnableMetricsOption>()) {
-    metric_service_connection_ = monitoring_v3::MakeMetricServiceConnection(
-        MetricsExporterConnectionOptions(options_));
-    // The client_uid is eventually used in conjunction with other data labels
-    // to identify metric data points. This pseudorandom string is used to aid
-    // in disambiguation.
-    auto gen = internal::MakeDefaultPRNG();
-    std::string client_uid =
-        internal::Sample(gen, 16, "abcdefghijklmnopqrstuvwxyz0123456789");
-    operation_context_factory_ =
-        std::make_unique<MetricsOperationContextFactory>(
-            std::move(client_uid), metric_service_connection_, options_);
-  } else {
-    operation_context_factory_ =
-        std::make_unique<SimpleOperationContextFactory>();
-  }
-#else
-  operation_context_factory_ =
-      std::make_unique<SimpleOperationContextFactory>();
-#endif  // GOOGLE_CLOUD_CPP_BIGTABLE_WITH_OTEL_METRICS
-}
-
-DataConnectionImpl::DataConnectionImpl(
-    std::unique_ptr<BackgroundThreads> background,
-    std::shared_ptr<BigtableStub> stub,
-    std::shared_ptr<MutateRowsLimiter> limiter, Options options)
-    : DataConnectionImpl(std::move(background),
-                         std::make_unique<StubManager>(std::move(stub)),
-                         std::move(limiter), std::move(options)) {}
-
 DataConnectionImpl::DataConnectionImpl(
     std::unique_ptr<BackgroundThreads> background,
     std::unique_ptr<StubManager> stub_manager,
@@ -290,6 +230,8 @@ DataConnectionImpl::DataConnectionImpl(
       limiter_(std::move(limiter)),
       options_(MergeOptions(std::move(options), DataConnection::options())) {}
 
+DataConnectionImpl::~DataConnectionImpl() = default;
+
 DataConnectionImpl::DataConnectionImpl(
     std::unique_ptr<BackgroundThreads> background,
     std::shared_ptr<BigtableStub> stub,

google/cloud/bigtable/internal/data_connection_impl.h

@@ -52,26 +52,14 @@ bigtable::Row TransformReadModifyWriteRowResponse(
 
 class DataConnectionImpl : public bigtable::DataConnection {
  public:
-  ~DataConnectionImpl() override = default;
+  ~DataConnectionImpl() override;
 
-  DataConnectionImpl(std::unique_ptr<BackgroundThreads> background,
-                     std::unique_ptr<StubManager> stub_manager,
-                     std::shared_ptr<MutateRowsLimiter> limiter,
-                     Options options);
-
-  // This constructor is used for testing.
   DataConnectionImpl(
       std::unique_ptr<BackgroundThreads> background,
       std::unique_ptr<StubManager> stub_manager,
       std::unique_ptr<OperationContextFactory> operation_context_factory,
       std::shared_ptr<MutateRowsLimiter> limiter, Options options);
 
-  DataConnectionImpl(std::unique_ptr<BackgroundThreads> background,
-                     std::shared_ptr<BigtableStub> stub,
-                     std::shared_ptr<MutateRowsLimiter> limiter,
-                     Options options);
-
-  // This constructor is used for testing.
   DataConnectionImpl(
       std::unique_ptr<BackgroundThreads> background,
       std::shared_ptr<BigtableStub> stub,

google/cloud/bigtable/internal/data_connection_impl_test.cc

@@ -16,6 +16,7 @@
 #include "google/cloud/bigtable/data_connection.h"
 #include "google/cloud/bigtable/internal/crc32c.h"
 #include "google/cloud/bigtable/internal/defaults.h"
+#include "google/cloud/bigtable/internal/grpc_metrics_exporter.h"
 #include "google/cloud/bigtable/internal/query_plan.h"
 #ifdef GOOGLE_CLOUD_CPP_BIGTABLE_WITH_OTEL_METRICS
 #include "google/cloud/bigtable/internal/metrics.h"
@@ -267,7 +268,9 @@ std::shared_ptr<DataConnectionImpl> TestConnection(
         std::make_shared<NoopMutateRowsLimiter>()) {
   auto background = internal::MakeBackgroundThreadsFactory()();
   return std::make_shared<DataConnectionImpl>(
-      std::move(background), std::move(stub), std::move(limiter), Options{});
+      std::move(background), std::move(stub),
+      std::make_unique<SimpleOperationContextFactory>(), std::move(limiter),
+      Options{});
 }
 
 std::shared_ptr<DataConnectionImpl> TestConnection(

google/cloud/bigtable/internal/defaults.cc

@@ -126,16 +126,20 @@ int DefaultConnectionPoolSize() {
                     cpu_count * BIGTABLE_CLIENT_DEFAULT_CHANNELS_PER_CPU);
 }
 
-bool IsDirectPath() {
+bool IsDirectPath(Options const& options) {
   auto const direct_path =
-      google::cloud::internal::GetEnv("GOOGLE_CLOUD_ENABLE_DIRECT_PATH")
-          .value_or("");
+      google::cloud::internal::GetEnv("GOOGLE_CLOUD_ENABLE_DIRECT_PATH");
   // Bigtable specific env var for Direct Path support used by all clients.
   auto const cbt_direct_path =
-      google::cloud::internal::GetEnv("CBT_ENABLE_DIRECTPATH").value_or("");
-  return absl::c_any_of(absl::StrSplit(direct_path, ','),
-                        [](absl::string_view v) { return v == "bigtable"; }) ||
-         cbt_direct_path == "true";
+      google::cloud::internal::GetEnv("CBT_ENABLE_DIRECTPATH");
+  if (direct_path.has_value() || cbt_direct_path.has_value()) {
+    return absl::c_any_of(
+               absl::StrSplit(direct_path.value_or(""), ','),
+               [](absl::string_view v) { return v == "bigtable"; }) ||
+           cbt_direct_path.value_or("") == "true";
+  }
+  return options.get<experimental::DirectPathModeOption>() ==
+         experimental::DirectPathMode::kEnabled;
 }
 
 Options HandleUniverseDomain(Options opts) {
@@ -184,7 +188,7 @@ Options DefaultOptions(Options opts) {
   }
 
   // Set the specific data endpoints if Direct Path is enabled.
-  if (IsDirectPath()) {
+  if (IsDirectPath(opts)) {
     opts.set<::google::cloud::bigtable_internal::DataEndpointOption>(
             "google-c2p:///bigtable.googleapis.com")
         .set<AuthorityOption>("bigtable.googleapis.com");
@@ -316,6 +320,20 @@ Options DefaultTableAdminOptions(Options opts) {
       opts.get<::google::cloud::bigtable_internal::AdminEndpointOption>());
 }
 
+#ifdef GOOGLE_CLOUD_CPP_BIGTABLE_WITH_OTEL_METRICS
+Options MetricsExporterConnectionOptions(Options options) {
+  options.unset<EndpointOption>();
+  options.unset<AuthorityOption>();
+  auto collector = google::cloud::internal::GetEnv(
+      "GOOGLE_CLOUD_CPP_TESTING_OTEL_COLLECTOR");
+  if (collector.has_value()) {
+    options.set<google::cloud::UnifiedCredentialsOption>(
+        google::cloud::MakeInsecureCredentials());
+  }
+  return options;
+}
+#endif
+
 }  // namespace internal
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
 }  // namespace bigtable

google/cloud/bigtable/internal/defaults.h

@@ -29,7 +29,7 @@ int DefaultConnectionPoolSize();
 /**
  * Returns true if Direct Path is enabled for Bigtable.
  */
-bool IsDirectPath();
+bool IsDirectPath(Options const& options);
 
 /**
  * Returns an `Options` with the appropriate defaults for Bigtable.
@@ -52,6 +52,10 @@ Options DefaultInstanceAdminOptions(Options opts);
 
 Options DefaultTableAdminOptions(Options opts);
 
+#ifdef GOOGLE_CLOUD_CPP_BIGTABLE_WITH_OTEL_METRICS
+Options MetricsExporterConnectionOptions(Options options);
+#endif
+
 }  // namespace internal
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
 }  // namespace bigtable