Skip to content

impl(bigtable): enable outstanding_rpcs metric and StubSelection event #16362

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
scotthart merged 3 commits into from
Aug 20, 2026
+560 −219
Merged

impl(bigtable): enable outstanding_rpcs metric and StubSelection event #16362

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
bddd3fe
impl(bigtable): enable outstanding_rpcs metric and StubSelection event
scotthart Aug 20, 2026
2322efe
standardize on using constants in InitializeProvider
scotthart Aug 20, 2026
853510b
add unit tests for better coverage
scotthart Aug 20, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
151 changes: 95 additions & 56 deletions google/cloud/bigtable/internal/bigtable_random_two_least_used_decorator.cc
View file
Open in desktop

Large diffs are not rendered by default.

3 changes: 2 additions & 1 deletion google/cloud/bigtable/internal/bigtable_random_two_least_used_decorator_test.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,7 +50,8 @@ class BigtableRandomTwoLeastUsedTest : public ::testing::Test {

pool_ = DynamicChannelPool<BigtableStub>::Create(
instance_name, cq_, channels, refresh_state,
stub_factory_fn_.AsStdFunction(), sizing_policy);
stub_factory_fn_.AsStdFunction(), sizing_policy,
TransportType::kCloudPath);
}

~BigtableRandomTwoLeastUsedTest() override {
Expand Down
5 changes: 4 additions & 1 deletion google/cloud/bigtable/internal/bigtable_stub_factory.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -178,7 +178,10 @@ std::shared_ptr<BigtableStub> CreateBigtableStubRandomTwoLeastUsed(
std::move(children), std::move(refresh_state),
std::move(refreshing_channel_stub_factory),
options.get<
bigtable::experimental::DynamicChannelPoolSizingPolicyOption>()));
bigtable::experimental::DynamicChannelPoolSizingPolicyOption>(),
bigtable::internal::IsDirectPath(options)
? TransportType::kDirectPath
: TransportType::kCloudPath));
}

std::shared_ptr<BigtableStub> CreateDecoratedStubs(
Expand Down
4 changes: 2 additions & 2 deletions google/cloud/bigtable/internal/bigtable_stub_factory_test.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,3 @@
#include "google/cloud/bigtable/internal/operation_context.h"
// Copyright 2022 Google LLC
//
// Licensed under the Apache License, Version 2.0 (the "License");
Expand All @@ -13,9 +12,10 @@
// See the License for the specific language governing permissions and
// limitations under the License.

#include "google/cloud/bigtable/internal/bigtable_random_two_least_used_decorator.h"
#include "google/cloud/bigtable/internal/bigtable_stub_factory.h"
#include "google/cloud/bigtable/internal/bigtable_random_two_least_used_decorator.h"
#include "google/cloud/bigtable/internal/dynamic_channel_pool.h"
#include "google/cloud/bigtable/internal/operation_context.h"
#include "google/cloud/bigtable/options.h"
#include "google/cloud/bigtable/testing/mock_bigtable_stub.h"
#include "google/cloud/common_options.h"
Expand Down
14 changes: 10 additions & 4 deletions google/cloud/bigtable/internal/dynamic_channel_pool.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -66,11 +66,12 @@ class DynamicChannelPool
std::vector<std::shared_ptr<ChannelUsage<T>>> initial_channels,
std::shared_ptr<ConnectionRefreshState> refresh_state,
StubFactoryFn stub_factory_fn,
bigtable::experimental::DynamicChannelPoolSizingPolicy sizing_policy) {
bigtable::experimental::DynamicChannelPoolSizingPolicy sizing_policy,
TransportType transport_type) {
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),
std::move(sizing_policy)));
std::move(sizing_policy), transport_type));
return pool;
}

Expand Down Expand Up @@ -105,6 +106,8 @@ class DynamicChannelPool
return sizing_policy_;
}

TransportType transport_type() const { return transport_type_; }

// Calls CheckPoolChannelHealth before picking a channel.
//
// Pick two random channels from channels_ and return the channel with the
Expand Down Expand Up @@ -173,14 +176,16 @@ class DynamicChannelPool
std::vector<std::shared_ptr<ChannelUsage<T>>> initial_wrapped_channels,
std::shared_ptr<ConnectionRefreshState> refresh_state,
StubFactoryFn stub_factory_fn,
bigtable::experimental::DynamicChannelPoolSizingPolicy sizing_policy)
bigtable::experimental::DynamicChannelPoolSizingPolicy sizing_policy,
TransportType transport_type)
: instance_name_(std::move(instance_name)),
cq_(std::move(cq)),
refresh_state_(std::move(refresh_state)),
stub_factory_fn_(std::move(stub_factory_fn)),
channels_(std::move(initial_wrapped_channels)),
sizing_policy_(std::move(sizing_policy)),
next_channel_id_(static_cast<std::uint32_t>(channels_.size())) {
next_channel_id_(static_cast<std::uint32_t>(channels_.size())),
transport_type_(transport_type) {
std::scoped_lock lk(mu_);
SetSizeDecreaseCooldownTimer(lk);
}
Expand Down Expand Up @@ -453,6 +458,7 @@ class DynamicChannelPool
future<StatusOr<std::chrono::system_clock::time_point>>
pool_size_decrease_cooldown_timer_;
std::uint32_t next_channel_id_;
TransportType const transport_type_ = TransportType::kCloudPath;
};

GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
Expand Down
88 changes: 70 additions & 18 deletions google/cloud/bigtable/internal/dynamic_channel_pool_test.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -180,7 +180,7 @@ TEST_F(DynamicChannelPoolTest, SelectLeastUsedFromTwoChannels) {
sizing_policy.minimum_channel_pool_size = 2;
auto pool = DynamicChannelPool<BigtableStub>::Create(
instance_name, CompletionQueue(mock_cq_impl_), channels, refresh_state,
stub_factory_fn, sizing_policy);
stub_factory_fn, sizing_policy, TransportType::kCloudPath);
auto selected = pool->GetChannelRandomTwoLeastUsed();
EXPECT_THAT(selected.outstanding_rpcs, Eq(5));
grpc::ClientContext context;
Expand Down Expand Up @@ -247,7 +247,8 @@ TEST_F(DynamicChannelPoolTest, OneInitialChannel) {
sizing_policy.minimum_channel_pool_size = 1;
auto pool = DynamicChannelPool<BigtableStub>::Create(
instance_name, CompletionQueue(mock_cq_impl_), channels, refresh_state,
stub_factory_fn.AsStdFunction(), sizing_policy);
stub_factory_fn.AsStdFunction(), sizing_policy,
TransportType::kCloudPath);
EXPECT_THAT(pool->size(), Eq(1));

auto selected = pool->GetChannelRandomTwoLeastUsed();
Expand Down Expand Up @@ -310,7 +311,8 @@ TEST_F(DynamicChannelPoolTest, EmptyInitialPool) {
sizing_policy.minimum_channel_pool_size = 0;
auto pool = DynamicChannelPool<BigtableStub>::Create(
instance_name, CompletionQueue(mock_cq_impl_), channels, refresh_state,
stub_factory_fn.AsStdFunction(), sizing_policy);
stub_factory_fn.AsStdFunction(), sizing_policy,
TransportType::kCloudPath);

EXPECT_THAT(*pool, ::testing::IsEmpty());

Expand Down Expand Up @@ -356,7 +358,8 @@ TEST_F(DynamicChannelPoolTest, ScheduleAddChannelsPoolUndersized) {

auto pool = DynamicChannelPool<BigtableStub>::Create(
instance_name, CompletionQueue(mock_cq_impl_), channels, refresh_state,
stub_factory_fn.AsStdFunction(), sizing_policy);
stub_factory_fn.AsStdFunction(), sizing_policy,
TransportType::kCloudPath);
DynamicChannelPoolTestWrapper wrapper(pool);

{
Expand Down Expand Up @@ -416,7 +419,8 @@ TEST_F(DynamicChannelPoolTest, ScheduleAddChannelsPoolAtMax) {

auto pool = DynamicChannelPool<BigtableStub>::Create(
instance_name, CompletionQueue(mock_cq_impl_), channels, refresh_state,
stub_factory_fn.AsStdFunction(), sizing_policy);
stub_factory_fn.AsStdFunction(), sizing_policy,
TransportType::kCloudPath);
DynamicChannelPoolTestWrapper wrapper(pool);

EXPECT_CALL(*mock_cq_impl_, RunAsync).Times(1);
Expand Down Expand Up @@ -463,7 +467,8 @@ TEST_F(DynamicChannelPoolTest, ScheduleAddChannelsPoolBelowMax) {

auto pool = DynamicChannelPool<BigtableStub>::Create(
instance_name, CompletionQueue(mock_cq_impl_), channels, refresh_state,
stub_factory_fn.AsStdFunction(), sizing_policy);
stub_factory_fn.AsStdFunction(), sizing_policy,
TransportType::kCloudPath);
DynamicChannelPoolTestWrapper wrapper(pool);

EXPECT_CALL(*mock_cq_impl_, RunAsync).Times(1);
Expand Down Expand Up @@ -520,7 +525,8 @@ TEST_F(DynamicChannelPoolTest, AddChannels) {

auto pool = DynamicChannelPool<BigtableStub>::Create(
instance_name, CompletionQueue(mock_cq_impl_), channels, refresh_state,
stub_factory_fn.AsStdFunction(), sizing_policy);
stub_factory_fn.AsStdFunction(), sizing_policy,
TransportType::kCloudPath);
DynamicChannelPoolTestWrapper wrapper(pool);
std::vector<int> new_channel_ids = {0, 1};
wrapper.set_num_pending_channels(new_channel_ids.size());
Expand Down Expand Up @@ -557,7 +563,8 @@ TEST_F(DynamicChannelPoolTest, ScheduleRemoveChannelsAlreadyPending) {

auto pool = DynamicChannelPool<BigtableStub>::Create(
instance_name, CompletionQueue(mock_cq_impl_), channels, refresh_state,
stub_factory_fn.AsStdFunction(), sizing_policy);
stub_factory_fn.AsStdFunction(), sizing_policy,
TransportType::kCloudPath);
DynamicChannelPoolTestWrapper wrapper(pool);

promise<void> p;
Expand Down Expand Up @@ -597,7 +604,8 @@ TEST_F(DynamicChannelPoolTest, ScheduleRemoveChannelsNotAlreadyPending) {

auto pool = DynamicChannelPool<BigtableStub>::Create(
instance_name, CompletionQueue(mock_cq_impl_), channels, refresh_state,
stub_factory_fn.AsStdFunction(), sizing_policy);
stub_factory_fn.AsStdFunction(), sizing_policy,
TransportType::kCloudPath);
DynamicChannelPoolTestWrapper wrapper(pool);

EXPECT_CALL(*mock_cq_impl_, MakeRelativeTimer)
Expand Down Expand Up @@ -643,7 +651,8 @@ TEST_F(DynamicChannelPoolTest, RemoveChannelsLoneChannelDrained) {

auto pool = DynamicChannelPool<BigtableStub>::Create(
instance_name, CompletionQueue(mock_cq_impl_), channels, refresh_state,
stub_factory_fn.AsStdFunction(), sizing_policy);
stub_factory_fn.AsStdFunction(), sizing_policy,
TransportType::kCloudPath);
DynamicChannelPoolTestWrapper wrapper(pool);

std::vector<std::shared_ptr<ChannelUsage<BigtableStub>>> draining_channels;
Expand Down Expand Up @@ -684,7 +693,8 @@ TEST_F(DynamicChannelPoolTest, RemoveChannelsSomeChannelsDrained) {

auto pool = DynamicChannelPool<BigtableStub>::Create(
instance_name, CompletionQueue(mock_cq_impl_), channels, refresh_state,
stub_factory_fn.AsStdFunction(), sizing_policy);
stub_factory_fn.AsStdFunction(), sizing_policy,
TransportType::kCloudPath);
DynamicChannelPoolTestWrapper wrapper(pool);

std::vector<std::shared_ptr<ChannelUsage<BigtableStub>>> draining_channels;
Expand Down Expand Up @@ -779,7 +789,8 @@ TEST_F(DynamicChannelPoolTest, HandleBadChannelsTwoChannelsOneBad) {
sizing_policy.minimum_channel_pool_size = 2;
auto pool = DynamicChannelPool<BigtableStub>::Create(
instance_name, CompletionQueue(mock_cq_impl_), channels, refresh_state,
stub_factory_fn.AsStdFunction(), sizing_policy);
stub_factory_fn.AsStdFunction(), sizing_policy,
TransportType::kCloudPath);
DynamicChannelPoolTestWrapper wrapper(pool);
auto draining_channels = wrapper.SetDrainingChannels({});

Expand Down Expand Up @@ -862,7 +873,8 @@ TEST_F(DynamicChannelPoolTest, HandleBadChannelsTwoChannelsOtherOneBad) {
sizing_policy.minimum_channel_pool_size = 2;
auto pool = DynamicChannelPool<BigtableStub>::Create(
instance_name, CompletionQueue(mock_cq_impl_), channels, refresh_state,
stub_factory_fn.AsStdFunction(), sizing_policy);
stub_factory_fn.AsStdFunction(), sizing_policy,
TransportType::kCloudPath);
DynamicChannelPoolTestWrapper wrapper(pool);
auto draining_channels = wrapper.SetDrainingChannels({});

Expand Down Expand Up @@ -951,7 +963,8 @@ TEST_F(DynamicChannelPoolTest, HandleBadChannelsThreeChannelsOneBad) {
sizing_policy.minimum_channel_pool_size = 2;
auto pool = DynamicChannelPool<BigtableStub>::Create(
instance_name, CompletionQueue(mock_cq_impl_), channels, refresh_state,
stub_factory_fn.AsStdFunction(), sizing_policy);
stub_factory_fn.AsStdFunction(), sizing_policy,
TransportType::kCloudPath);
DynamicChannelPoolTestWrapper wrapper(pool);
auto draining_channels = wrapper.SetDrainingChannels({});

Expand Down Expand Up @@ -1050,7 +1063,8 @@ TEST_F(DynamicChannelPoolTest, HandleBadChannelsAllChannelsBad) {
sizing_policy.minimum_channel_pool_size = 2;
auto pool = DynamicChannelPool<BigtableStub>::Create(
instance_name, CompletionQueue(mock_cq_impl_), channels, refresh_state,
stub_factory_fn.AsStdFunction(), sizing_policy);
stub_factory_fn.AsStdFunction(), sizing_policy,
TransportType::kCloudPath);
DynamicChannelPoolTestWrapper wrapper(pool);
auto draining_channels = wrapper.SetDrainingChannels({});

Expand Down Expand Up @@ -1112,7 +1126,8 @@ TEST_F(DynamicChannelPoolTest, CheckChannelPoolHealthNeedsIncrease) {
sizing_policy.maximum_channel_pool_size = 1;
auto pool = DynamicChannelPool<BigtableStub>::Create(
instance_name, CompletionQueue(mock_cq_impl_), channels, refresh_state,
stub_factory_fn.AsStdFunction(), sizing_policy);
stub_factory_fn.AsStdFunction(), sizing_policy,
TransportType::kCloudPath);
DynamicChannelPoolTestWrapper wrapper(pool);

// ScheduleAddChannels will NOT be called as the pool has max channels.
Expand All @@ -1126,7 +1141,8 @@ TEST_F(DynamicChannelPoolTest, CheckChannelPoolHealthNeedsIncrease) {
sizing_policy.maximum_channel_pool_size = 10;
auto pool = DynamicChannelPool<BigtableStub>::Create(
instance_name, CompletionQueue(mock_cq_impl_), channels, refresh_state,
stub_factory_fn.AsStdFunction(), sizing_policy);
stub_factory_fn.AsStdFunction(), sizing_policy,
TransportType::kCloudPath);
DynamicChannelPoolTestWrapper wrapper(pool);

// ScheduleAddChannels will be called.
Expand Down Expand Up @@ -1192,7 +1208,8 @@ TEST_F(DynamicChannelPoolTest, CheckChannelPoolHealthNeedsDecrease) {
sizing_policy.minimum_average_outstanding_rpcs_per_channel = 5;
auto pool = DynamicChannelPool<BigtableStub>::Create(
instance_name, CompletionQueue(mock_cq_impl_), channels, refresh_state,
stub_factory_fn.AsStdFunction(), sizing_policy);
stub_factory_fn.AsStdFunction(), sizing_policy,
TransportType::kCloudPath);
DynamicChannelPoolTestWrapper wrapper(pool);

// ScheduleAddChannels will NOT be called.
Expand All @@ -1207,6 +1224,41 @@ TEST_F(DynamicChannelPoolTest, CheckChannelPoolHealthNeedsDecrease) {
EXPECT_THAT(pool->size(), Eq(3));
}

TEST_F(DynamicChannelPoolTest, TransportType) {
auto instance_name =
bigtable::InstanceResource(Project("my-project"), "my-instance")
.FullName();
auto refresh_state = std::make_shared<ConnectionRefreshState>(
fake_cq_impl_, std::chrono::milliseconds(1),
std::chrono::milliseconds(10));
std::vector<std::shared_ptr<ChannelUsage<BigtableStub>>> channels;
DynamicChannelPoolSizingPolicy sizing_policy;
MockFunction<StatusOr<std::shared_ptr<ChannelUsage<BigtableStub>>>(
std::uint32_t, std::string const&, StubManager::Priming)>
stub_factory_fn;

EXPECT_CALL(*mock_cq_impl_, MakeRelativeTimer)
.WillRepeatedly([&](std::chrono::nanoseconds ns) {
EXPECT_THAT(ns.count(),
Eq(std::chrono::nanoseconds(
sizing_policy.pool_size_decrease_cooldown_interval)
.count()));
return make_ready_future(StatusOr(std::chrono::system_clock::now()));
});

auto pool_cloud = DynamicChannelPool<BigtableStub>::Create(
instance_name, CompletionQueue(mock_cq_impl_), channels, refresh_state,
stub_factory_fn.AsStdFunction(), sizing_policy,
TransportType::kCloudPath);
EXPECT_THAT(pool_cloud->transport_type(), Eq(TransportType::kCloudPath));

auto pool_direct = DynamicChannelPool<BigtableStub>::Create(
instance_name, CompletionQueue(mock_cq_impl_), channels, refresh_state,
stub_factory_fn.AsStdFunction(), sizing_policy,
TransportType::kDirectPath);
EXPECT_THAT(pool_direct->transport_type(), Eq(TransportType::kDirectPath));
}

} // namespace
GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
} // namespace bigtable_internal
Expand Down
Loading
Loading