Skip to content

channel pool metric prototype #16324

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

Draft
scotthart wants to merge 1 commit into
base: main
Choose a base branch
from
Draft

channel pool metric prototype #16324

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
2 changes: 1 addition & 1 deletion google/cloud/bigtable/internal/async_bulk_apply.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -83,7 +83,7 @@ void AsyncBulkApplier::MakeRequest() {
auto self = this->shared_from_this();
PerformAsyncStreamingRead(
stub_->AsyncMutateRows(cq_, client_context_, options_,
state_.BeforeStart()),
state_.BeforeStart(), *operation_context_),
[self](google::bigtable::v2::MutateRowsResponse r) {
self->OnRead(std::move(r));
return make_ready_future(self->keep_reading_.load());
Expand Down
73 changes: 47 additions & 26 deletions google/cloud/bigtable/internal/async_bulk_apply_test.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -163,7 +163,8 @@ class MockMetric : public Metric {
ElementDeliveryParams const&),
(override));
MOCK_METHOD(std::unique_ptr<Metric>, clone,
(ResourceLabels resource_labels, DataLabels data_labels),
(TableResourceLabels const& resource_labels,
TableDataLabels const& data_labels),
(const, override));
};

Expand All @@ -172,7 +173,8 @@ class CloningMetric : public Metric {
public:
explicit CloningMetric(std::unique_ptr<MockMetric> metric)
: metric_(std::move(metric)) {}
std::unique_ptr<Metric> clone(ResourceLabels, DataLabels) const override {
std::unique_ptr<Metric> clone(TableResourceLabels const&,
TableDataLabels const&) const override {
return std::move(metric_);
}

Expand Down Expand Up @@ -207,7 +209,8 @@ TEST_F(AsyncBulkApplyTest, Success) {
auto mock = std::make_shared<MockBigtableStub>();
EXPECT_CALL(*mock, AsyncMutateRows)
.WillOnce([this](CompletionQueue const&, auto client_context, auto,
v2::MutateRowsRequest const& request) {
v2::MutateRowsRequest const& request,
bigtable_internal::OperationContext&) {
metadata_fixture_.SetServerMetadata(*client_context, {});
EXPECT_EQ(kAppProfile, request.app_profile_id());
EXPECT_EQ(kTableName, request.table_name());
Expand Down Expand Up @@ -281,7 +284,8 @@ TEST_F(AsyncBulkApplyTest, PartialStreamIsRetried) {
auto mock = std::make_shared<MockBigtableStub>();
EXPECT_CALL(*mock, AsyncMutateRows)
.WillOnce([this](CompletionQueue const&, auto context, auto,
v2::MutateRowsRequest const& request) {
v2::MutateRowsRequest const& request,
bigtable_internal::OperationContext&) {
metadata_fixture_.SetServerMetadata(*context, {});
EXPECT_EQ(kAppProfile, request.app_profile_id());
EXPECT_EQ(kTableName, request.table_name());
Expand All @@ -306,7 +310,8 @@ TEST_F(AsyncBulkApplyTest, PartialStreamIsRetried) {
return stream;
})
.WillOnce([this](CompletionQueue const&, auto client_context, auto,
v2::MutateRowsRequest const& request) {
v2::MutateRowsRequest const& request,
bigtable_internal::OperationContext&) {
metadata_fixture_.SetServerMetadata(*client_context, {});
EXPECT_EQ(kAppProfile, request.app_profile_id());
EXPECT_EQ(kTableName, request.table_name());
Expand Down Expand Up @@ -383,7 +388,8 @@ TEST_F(AsyncBulkApplyTest, IdempotentMutationPolicy) {
auto mock = std::make_shared<MockBigtableStub>();
EXPECT_CALL(*mock, AsyncMutateRows)
.WillOnce([this](CompletionQueue const&, auto client_context, auto,
v2::MutateRowsRequest const& request) {
v2::MutateRowsRequest const& request,
bigtable_internal::OperationContext&) {
metadata_fixture_.SetServerMetadata(*client_context, {});
EXPECT_EQ(kAppProfile, request.app_profile_id());
EXPECT_EQ(kTableName, request.table_name());
Expand All @@ -408,7 +414,8 @@ TEST_F(AsyncBulkApplyTest, IdempotentMutationPolicy) {
return stream;
})
.WillOnce([this](CompletionQueue const&, auto client_context, auto,
v2::MutateRowsRequest const& request) {
v2::MutateRowsRequest const& request,
bigtable_internal::OperationContext&) {
metadata_fixture_.SetServerMetadata(*client_context, {});
EXPECT_EQ(kAppProfile, request.app_profile_id());
EXPECT_EQ(kTableName, request.table_name());
Expand Down Expand Up @@ -464,7 +471,8 @@ TEST_F(AsyncBulkApplyTest, TooManyStreamFailures) {
EXPECT_CALL(*mock, AsyncMutateRows)
.Times(kNumRetries + 1)
.WillRepeatedly([this](CompletionQueue const&, auto context, auto,
v2::MutateRowsRequest const& request) {
v2::MutateRowsRequest const& request,
bigtable_internal::OperationContext&) {
metadata_fixture_.SetServerMetadata(*context, {});
EXPECT_EQ(kAppProfile, request.app_profile_id());
EXPECT_EQ(kTableName, request.table_name());
Expand Down Expand Up @@ -530,7 +538,8 @@ TEST_F(AsyncBulkApplyTest, RetryInfoHeeded) {
auto mock = std::make_shared<MockBigtableStub>();
EXPECT_CALL(*mock, AsyncMutateRows)
.WillOnce([this](CompletionQueue const&, auto context, auto,
v2::MutateRowsRequest const&) {
v2::MutateRowsRequest const&,
bigtable_internal::OperationContext&) {
metadata_fixture_.SetServerMetadata(*context, {});
auto stream = std::make_unique<MockAsyncMutateRowsStream>();
EXPECT_CALL(*stream, Start).WillOnce([] {
Expand All @@ -544,7 +553,8 @@ TEST_F(AsyncBulkApplyTest, RetryInfoHeeded) {
return stream;
})
.WillOnce([this](CompletionQueue const&, auto context, auto,
v2::MutateRowsRequest const&) {
v2::MutateRowsRequest const&,
bigtable_internal::OperationContext&) {
metadata_fixture_.SetServerMetadata(*context, {});
auto stream = std::make_unique<MockAsyncMutateRowsStream>();
EXPECT_CALL(*stream, Start).WillOnce([] {
Expand Down Expand Up @@ -603,7 +613,8 @@ TEST_F(AsyncBulkApplyTest, RetryInfoIgnored) {
auto mock = std::make_shared<MockBigtableStub>();
EXPECT_CALL(*mock, AsyncMutateRows)
.WillOnce([this](CompletionQueue const&, auto context, auto,
v2::MutateRowsRequest const&) {
v2::MutateRowsRequest const&,
bigtable_internal::OperationContext&) {
metadata_fixture_.SetServerMetadata(*context, {});
auto stream = std::make_unique<MockAsyncMutateRowsStream>();
EXPECT_CALL(*stream, Start).WillOnce([] {
Expand Down Expand Up @@ -640,7 +651,8 @@ TEST_F(AsyncBulkApplyTest, TimerError) {
auto mock = std::make_shared<MockBigtableStub>();
EXPECT_CALL(*mock, AsyncMutateRows)
.WillOnce([this](CompletionQueue const&, auto context, auto,
v2::MutateRowsRequest const& request) {
v2::MutateRowsRequest const& request,
bigtable_internal::OperationContext&) {
metadata_fixture_.SetServerMetadata(*context, {});
EXPECT_EQ(kAppProfile, request.app_profile_id());
EXPECT_EQ(kTableName, request.table_name());
Expand Down Expand Up @@ -705,7 +717,8 @@ TEST_F(AsyncBulkApplyTest, CancelAfterSuccess) {
auto mock = std::make_shared<MockBigtableStub>();
EXPECT_CALL(*mock, AsyncMutateRows)
.WillOnce([&p, this](CompletionQueue const&, auto client_context, auto,
v2::MutateRowsRequest const& request) {
v2::MutateRowsRequest const& request,
bigtable_internal::OperationContext&) {
metadata_fixture_.SetServerMetadata(*client_context, {});
EXPECT_EQ(kAppProfile, request.app_profile_id());
EXPECT_EQ(kTableName, request.table_name());
Expand Down Expand Up @@ -781,7 +794,8 @@ TEST_F(AsyncBulkApplyTest, CancelMidStream) {
auto mock = std::make_shared<MockBigtableStub>();
EXPECT_CALL(*mock, AsyncMutateRows)
.WillOnce([&p, this](CompletionQueue const&, auto client_context, auto,
v2::MutateRowsRequest const& request) {
v2::MutateRowsRequest const& request,
bigtable_internal::OperationContext&) {
metadata_fixture_.SetServerMetadata(*client_context, {});
EXPECT_EQ(kAppProfile, request.app_profile_id());
EXPECT_EQ(kTableName, request.table_name());
Expand Down Expand Up @@ -849,7 +863,8 @@ TEST_F(AsyncBulkApplyTest, CurrentOptionsContinuedOnRetries) {
EXPECT_CALL(*mock, AsyncMutateRows)
.Times(2)
.WillRepeatedly([this](CompletionQueue const&, auto context, auto,
v2::MutateRowsRequest const&) {
v2::MutateRowsRequest const&,
bigtable_internal::OperationContext&) {
EXPECT_EQ(5, internal::CurrentOptions().get<TestOption>());
metadata_fixture_.SetServerMetadata(*context, {});
auto stream = std::make_unique<MockAsyncMutateRowsStream>();
Expand Down Expand Up @@ -902,7 +917,8 @@ TEST_F(AsyncBulkApplyTest, RetriesOkStreamWithFailedMutations) {
EXPECT_CALL(*mock, AsyncMutateRows)
.Times(kNumRetries + 1)
.WillRepeatedly([this](CompletionQueue const&, auto context, auto,
v2::MutateRowsRequest const& request) {
v2::MutateRowsRequest const& request,
bigtable_internal::OperationContext&) {
metadata_fixture_.SetServerMetadata(*context, {});
EXPECT_EQ(kAppProfile, request.app_profile_id());
EXPECT_EQ(kTableName, request.table_name());
Expand Down Expand Up @@ -974,7 +990,8 @@ TEST_F(AsyncBulkApplyTest, Throttling) {
});
EXPECT_CALL(*mock, AsyncMutateRows)
.WillOnce([&limiter, this](CompletionQueue const&, auto client_context,
auto, v2::MutateRowsRequest const&) {
auto, v2::MutateRowsRequest const&,
bigtable_internal::OperationContext&) {
metadata_fixture_.SetServerMetadata(*client_context, {});
::testing::InSequence seq2;
auto stream = std::make_unique<MockAsyncMutateRowsStream>();
Expand Down Expand Up @@ -1011,7 +1028,8 @@ TEST_F(AsyncBulkApplyTest, ThrottlingBeforeEachRetry) {
EXPECT_CALL(*mock, AsyncMutateRows)
.Times(kNumRetries + 1)
.WillRepeatedly([this](CompletionQueue const&, auto context, auto,
v2::MutateRowsRequest const&) {
v2::MutateRowsRequest const&,
bigtable_internal::OperationContext&) {
metadata_fixture_.SetServerMetadata(*context, {});
auto stream = std::make_unique<MockAsyncMutateRowsStream>();
EXPECT_CALL(*stream, Start).WillOnce([] {
Expand Down Expand Up @@ -1065,7 +1083,8 @@ TEST_F(AsyncBulkApplyTest, BigtableCookie) {
auto mock = std::make_shared<MockBigtableStub>();
EXPECT_CALL(*mock, AsyncMutateRows)
.WillOnce([this](CompletionQueue const&, auto context, auto,
v2::MutateRowsRequest const&) {
v2::MutateRowsRequest const&,
bigtable_internal::OperationContext&) {
// Return a bigtable cookie in the first request.
metadata_fixture_.SetServerMetadata(
*context, {{}, {{"x-goog-cbt-cookie-routing", "routing"}}});
Expand All @@ -1079,7 +1098,8 @@ TEST_F(AsyncBulkApplyTest, BigtableCookie) {
return stream;
})
.WillOnce([this](CompletionQueue const&, auto context, auto,
v2::MutateRowsRequest const&) {
v2::MutateRowsRequest const&,
bigtable_internal::OperationContext&) {
// Verify that the next request includes the bigtable cookie from above.
auto headers = metadata_fixture_.GetMetadata(*context);
EXPECT_THAT(headers,
Expand Down Expand Up @@ -1128,7 +1148,7 @@ TEST_F(AsyncBulkApplyTest, TracedBackoff) {
auto mock = std::make_shared<MockBigtableStub>();
EXPECT_CALL(*mock, AsyncMutateRows)
.Times(kNumRetries + 1)
.WillRepeatedly([this](auto&, auto context, auto, auto const&) {
.WillRepeatedly([this](auto&, auto context, auto, auto const&, auto&) {
metadata_fixture_.SetServerMetadata(*context, {});
return std::make_unique<ErrorStream>(TransientError());
});
Expand Down Expand Up @@ -1160,11 +1180,12 @@ TEST_F(AsyncBulkApplyTest, CallSpanActiveThroughout) {
auto mock = std::make_shared<MockBigtableStub>();
EXPECT_CALL(*mock, AsyncMutateRows)
.Times(kNumRetries + 1)
.WillRepeatedly([this, span](auto&, auto context, auto, auto const&) {
metadata_fixture_.SetServerMetadata(*context, {});
EXPECT_THAT(span, IsActive());
return std::make_unique<ErrorStream>(TransientError());
});
.WillRepeatedly(
[this, span](auto&, auto context, auto, auto const&, auto&) {
metadata_fixture_.SetServerMetadata(*context, {});
EXPECT_THAT(span, IsActive());
return std::make_unique<ErrorStream>(TransientError());
});

internal::AutomaticallyCreatedBackgroundThreads background;
auto retry = DataLimitedErrorCountRetryPolicy(kNumRetries).clone();
Expand Down
3 changes: 2 additions & 1 deletion google/cloud/bigtable/internal/async_row_reader.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,7 +51,8 @@ void AsyncRowReader::MakeRequest() {

auto self = this->shared_from_this();
PerformAsyncStreamingRead(
stub_->AsyncReadRows(cq_, client_context_, options_, request),
stub_->AsyncReadRows(cq_, client_context_, options_, request,
*operation_context_),
[self](v2::ReadRowsResponse r) {
return self->OnDataReceived(std::move(r));
},
Expand Down
Loading
Loading