impl(bigtable): make dynamic pool options available and update integration tests

Author: scotthart

google/cloud/bigtable/internal/bigtable_random_two_least_used_decorator_test.cc

@@ -38,7 +38,7 @@ class BigtableRandomTwoLeastUsedTest : public ::testing::Test {
     auto instance_name =
         bigtable::InstanceResource(Project("my-project"), "my-instance")
             .FullName();
-    DynamicChannelPoolSizingPolicy sizing_policy;
+    bigtable::experimental::DynamicChannelPoolSizingPolicy sizing_policy;
     auto refresh_state = std::make_shared<ConnectionRefreshState>(
         fake_cq_impl_, std::chrono::milliseconds(1),
         std::chrono::milliseconds(10));

google/cloud/bigtable/internal/bigtable_stub_factory.h

@@ -26,30 +26,6 @@
 #include <functional>
 #include <memory>
 
-namespace google::cloud::bigtable {
-GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
-
-// TODO(#16035): Move this Option to bigtable/options.h in the experimental
-// namespace when the feature is ready.
-namespace experimental {
-/**
- *  If set, a dynamic channel pool is created for each instance that requests
- *  are destined. Instances specified as part of this Option have dynamic
- *  channel pools created and primed as part of DataConnection construction. If
- *  no Instances are specified, then dynamic channel pool creation is deferred
- *  until the first request sent, increasing time to first byte latency.
- *
- * @note This option must be supplied to `MakeDataConnection()` in order to take
- * effect.
- */
-struct InstanceChannelAffinityOption {
-  using Type = std::vector<bigtable::InstanceResource>;
-};
-}  // namespace experimental
-
-GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
-}  // namespace google::cloud::bigtable
-
 namespace google {
 namespace cloud {
 namespace bigtable_internal {

google/cloud/bigtable/internal/dynamic_channel_pool.h

@@ -34,55 +34,6 @@ namespace cloud {
 namespace bigtable_internal {
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
 
-// TODO(#16035): Move this struct and Option to bigtable/options.h in the
-// experimental namespace when the feature is ready.
-struct DynamicChannelPoolSizingPolicy {
-  // To reduce channel churn, the pool will not add channels more frequently
-  // than this period.
-  std::chrono::milliseconds pool_size_increase_cooldown_interval =
-      std::chrono::seconds(10);
-
-  // Removing unused channels is not as performance critical as adding channels
-  // to handle a surge in RPC calls. Thus, there are separate cooldown settings
-  // for each.
-  std::chrono::milliseconds pool_size_decrease_cooldown_interval =
-      std::chrono::seconds(120);
-
-  struct DiscreteChannels {
-    explicit DiscreteChannels(int number = 0) : number(number) {}
-    int number;
-  };
-  struct PercentageOfPoolSize {
-    explicit PercentageOfPoolSize(double percentage = 0.0)
-        : percentage(percentage) {}
-    double percentage;
-  };
-  absl::variant<DiscreteChannels, PercentageOfPoolSize>
-      channels_to_add_per_resize = DiscreteChannels{1};
-
-  // If the average number of outstanding RPCs is below this threshold,
-  // the pool size will be decreased.
-  int minimum_average_outstanding_rpcs_per_channel = 1;
-  // If the average number of outstanding RPCs is above this threshold,
-  // the pool size will be increased.
-  int maximum_average_outstanding_rpcs_per_channel = 25;
-
-  // When channels are removed from the pool, we have to wait until all
-  // outstanding RPCs on that channel are completed before destroying it.
-  std::chrono::milliseconds remove_channel_polling_interval =
-      std::chrono::seconds(30);
-
-  // Limits how large the pool can grow. Default is twice the minimum_pool_size.
-  std::size_t maximum_channel_pool_size = 0;
-
-  // This is set to the value of GrpcNumChannelsOption.
-  std::size_t minimum_channel_pool_size = 0;
-};
-
-struct DynamicChannelPoolSizingPolicyOption {
-  using Type = DynamicChannelPoolSizingPolicy;
-};
-
 //
 // This class manages a pool of Stubs wrapped in a ChannelUsage object, and
 // selects one for use using a "Random Two Least Used" strategy.
@@ -107,7 +58,8 @@ class DynamicChannelPool
       std::vector<std::shared_ptr<ChannelUsage<T>>> initial_channels,
       std::shared_ptr<ConnectionRefreshState> refresh_state,
       StubFactoryFn stub_factory_fn,
-      DynamicChannelPoolSizingPolicy sizing_policy = {}) {
+      bigtable::experimental::DynamicChannelPoolSizingPolicy sizing_policy =
+          {}) {
     auto pool = std::shared_ptr<DynamicChannelPool>(new DynamicChannelPool(
         std::move(instance_name), std::move(cq), std::move(initial_channels),
         std::move(refresh_state), std::move(stub_factory_fn),
@@ -207,7 +159,7 @@ class DynamicChannelPool
       std::vector<std::shared_ptr<ChannelUsage<T>>> initial_wrapped_channels,
       std::shared_ptr<ConnectionRefreshState> refresh_state,
       StubFactoryFn stub_factory_fn,
-      DynamicChannelPoolSizingPolicy sizing_policy)
+      bigtable::experimental::DynamicChannelPoolSizingPolicy sizing_policy)
       : instance_name_(std::move(instance_name)),
         cq_(std::move(cq)),
         refresh_state_(std::move(refresh_state)),
@@ -293,13 +245,13 @@ class DynamicChannelPool
     std::size_t pool_size;
     explicit ChannelAddVisitor(std::size_t pool_size) : pool_size(pool_size) {}
     std::size_t operator()(
-        typename DynamicChannelPoolSizingPolicy::DiscreteChannels const& c)
-        const {
+        typename bigtable::experimental::DynamicChannelPoolSizingPolicy::
+            DiscreteChannels const& c) const {
       return c.number;
     }
     std::size_t operator()(
-        typename DynamicChannelPoolSizingPolicy::PercentageOfPoolSize const& c)
-        const {
+        typename bigtable::experimental::DynamicChannelPoolSizingPolicy::
+            PercentageOfPoolSize const& c) const {
       return static_cast<std::size_t>(
           std::floor(static_cast<double>(pool_size) * c.percentage));
     }
@@ -487,7 +439,7 @@ class DynamicChannelPool
   StubFactoryFn stub_factory_fn_;
   std::vector<std::shared_ptr<ChannelUsage<T>>> channels_;
   std::size_t num_pending_channels_ = 0;
-  DynamicChannelPoolSizingPolicy sizing_policy_;
+  bigtable::experimental::DynamicChannelPoolSizingPolicy sizing_policy_;
   std::vector<std::shared_ptr<ChannelUsage<T>>> draining_channels_;
   future<void> remove_channel_poll_timer_;
   future<StatusOr<std::chrono::system_clock::time_point>>

google/cloud/bigtable/internal/dynamic_channel_pool_test.cc

@@ -103,6 +103,7 @@ class DynamicChannelPoolTestWrapper {
 
 namespace {
 
+using ::google::cloud::bigtable::experimental::DynamicChannelPoolSizingPolicy;
 using ::google::cloud::bigtable::testing::MockBigtableStub;
 using ::google::cloud::testing_util::FakeCompletionQueueImpl;
 using ::google::cloud::testing_util::MockCompletionQueueImpl;

google/cloud/bigtable/options.h

@@ -39,6 +39,7 @@
  */
 
 #include "google/cloud/bigtable/idempotent_mutation_policy.h"
+#include "google/cloud/bigtable/instance_resource.h"
 #include "google/cloud/bigtable/internal/endpoint_options.h"
 #include "google/cloud/bigtable/retry_policy.h"
 #include "google/cloud/bigtable/rpc_retry_policy.h"
@@ -47,6 +48,7 @@
 #include "google/cloud/options.h"
 #include <chrono>
 #include <string>
+#include <variant>
 
 namespace google {
 namespace cloud {
@@ -173,6 +175,68 @@ struct QueryPlanRefreshFunctionRetryPolicyOption {
   using Type = std::shared_ptr<DataRetryPolicy>;
 };
 
+/**
+ *  If set, a dynamic channel pool is created for each instance that requests
+ *  are destined. Instances specified as part of this Option have dynamic
+ *  channel pools created and primed as part of DataConnection construction. If
+ *  no Instances are specified, then dynamic channel pool creation is deferred
+ *  until the first request sent, increasing time to first byte latency.
+ *
+ * @note This option must be supplied to `MakeDataConnection()` in order to take
+ * effect.
+ */
+struct InstanceChannelAffinityOption {
+  using Type = std::vector<bigtable::InstanceResource>;
+};
+
+/**
+ *  If the `InstanceChannelAffinityOption` is set, then all connections will be
+ *  managed by a Dynamic Channel Pool. The `DynamicChannelPoolSizingPolicy` can
+ *  be provided via the `DynamicChannelPoolSizingPolicyOption` and configures
+ *  the behavior of the `DynamicChannelPool`.
+ */
+struct DynamicChannelPoolSizingPolicy {
+  // Removing unused channels is not as performance critical as adding channels
+  // to handle a surge in RPC calls. Thus, there are separate cooldown settings
+  // for each.
+  std::chrono::milliseconds pool_size_decrease_cooldown_interval =
+      std::chrono::seconds(120);
+
+  struct DiscreteChannels {
+    explicit DiscreteChannels(int number = 0) : number(number) {}
+    int number;
+  };
+  struct PercentageOfPoolSize {
+    explicit PercentageOfPoolSize(double percentage = 0.0)
+        : percentage(percentage) {}
+    double percentage;
+  };
+  std::variant<DiscreteChannels, PercentageOfPoolSize>
+      channels_to_add_per_resize = DiscreteChannels{1};
+
+  // If the average number of outstanding RPCs is below this threshold,
+  // the pool size will be decreased.
+  int minimum_average_outstanding_rpcs_per_channel = 1;
+  // If the average number of outstanding RPCs is above this threshold,
+  // the pool size will be increased.
+  int maximum_average_outstanding_rpcs_per_channel = 25;
+
+  // When channels are removed from the pool, we have to wait until all
+  // outstanding RPCs on that channel are completed before destroying it.
+  std::chrono::milliseconds remove_channel_polling_interval =
+      std::chrono::seconds(30);
+
+  // Limits how large the pool can grow. Default is twice the minimum_pool_size.
+  std::size_t maximum_channel_pool_size = 0;
+
+  // This is set to the value of GrpcNumChannelsOption.
+  std::size_t minimum_channel_pool_size = 0;
+};
+
+struct DynamicChannelPoolSizingPolicyOption {
+  using Type = DynamicChannelPoolSizingPolicy;
+};
+
 }  // namespace experimental
 
 /// The complete list of options accepted by `bigtable::*Client`

google/cloud/bigtable/testing/table_integration_test.cc

@@ -122,7 +122,13 @@ void TableAdminTestEnvironment::TearDown() {
 }
 
 void TableIntegrationTest::SetUp() {
-  data_connection_ = MakeDataConnection();
+  Options options;
+  if (google::cloud::internal::GetEnv(
+          "GOOGLE_CLOUD_CPP_BIGTABLE_TESTING_CHANNEL_POOL")
+          .value_or("") == "dynamic") {
+    options.set<experimental::InstanceChannelAffinityOption>({});
+  }
+  data_connection_ = MakeDataConnection(options);
 
   // In production, we cannot use `DropAllRows()` to cleanup the table because
   // the integration tests sometimes consume all the 'DropRowRangeGroup' quota.

google/cloud/bigtable/tests/BUILD.bazel

@@ -21,12 +21,19 @@ package(default_visibility = ["//visibility:private"])
 
 licenses(["notice"])  # Apache 2.0
 
+VARIATIONS = {
+    "default": {"GOOGLE_CLOUD_CPP_BIGTABLE_TESTING_CHANNEL_POOL": "static"},
+    "dynamic-pool": {"GOOGLE_CLOUD_CPP_BIGTABLE_TESTING_CHANNEL_POOL": "dynamic"},
+}
+
 [cc_test(
-    name = test.replace("/", "_").replace(".cc", ""),
+    name = test.replace("/", "_").replace(".cc", "") + "-" + v_label,
     timeout = "long",
     srcs = [test],
+    env = v_env,
     tags = [
         "integration-test",
+        "integration-test-" + v_label,
     ],
     deps = [
         "//:bigtable",
@@ -36,7 +43,7 @@ licenses(["notice"])  # Apache 2.0
         "//google/cloud/testing_util:google_cloud_cpp_testing_private",
         "@googletest//:gtest_main",
     ],
-) for test in bigtable_client_integration_tests]
+) for test in bigtable_client_integration_tests for v_label, v_env in VARIATIONS.items()]
 
 cc_binary(
     name = "instance_admin_emulator",