diff --git a/ci/cloudbuild/builds/observability.sh b/ci/cloudbuild/builds/observability.sh index 22403fa9c27a1..9abbd5f73ccf9 100755 --- a/ci/cloudbuild/builds/observability.sh +++ b/ci/cloudbuild/builds/observability.sh @@ -159,10 +159,12 @@ chmod +x /tmp/observability_integration_test-dynamic-pool TEST_EXIT_CODE=0 echo "Running observability_integration_test-default..." +GOOGLE_CLOUD_CPP_BIGTABLE_TESTING_CHANNEL_POOL=static \ /tmp/observability_integration_test-default \ --gtest_output=xml:/tmp/test-default.xml > /tmp/test-default.log 2>&1 || TEST_EXIT_CODE=\$? echo "Running observability_integration_test-dynamic-pool..." +GOOGLE_CLOUD_CPP_BIGTABLE_TESTING_CHANNEL_POOL=dynamic \ /tmp/observability_integration_test-dynamic-pool \ --gtest_output=xml:/tmp/test-dynamic-pool.xml > /tmp/test-dynamic-pool.log 2>&1 || TEST_EXIT_CODE=\$? diff --git a/generator/generator_config.textproto b/generator/generator_config.textproto index 84f63c752cc38..e817bbec734de 100644 --- a/generator/generator_config.textproto +++ b/generator/generator_config.textproto @@ -575,7 +575,7 @@ service { omit_client: true omit_connection: true omit_stub_factory: true - generate_round_robin_decorator: true + generate_round_robin_decorator: false experimental_bigtable_operation_context: true omitted_rpcs: [ "GenerateInitialChangeStreamPartitions", diff --git a/google/cloud/bigtable/internal/async_row_reader_test.cc b/google/cloud/bigtable/internal/async_row_reader_test.cc index 7cbac64a29d73..955ee49843903 100644 --- a/google/cloud/bigtable/internal/async_row_reader_test.cc +++ b/google/cloud/bigtable/internal/async_row_reader_test.cc @@ -13,6 +13,7 @@ // limitations under the License. #include "google/cloud/bigtable/internal/async_row_reader.h" +#include "google/cloud/bigtable/internal/operation_context.h" #include "google/cloud/bigtable/row_reader.h" #include "google/cloud/bigtable/testing/mock_bigtable_stub.h" #include "google/cloud/internal/async_streaming_read_rpc_impl.h" diff --git a/google/cloud/bigtable/internal/async_row_sampler_test.cc b/google/cloud/bigtable/internal/async_row_sampler_test.cc index ee7ce168fd0f4..4004ebbfe895f 100644 --- a/google/cloud/bigtable/internal/async_row_sampler_test.cc +++ b/google/cloud/bigtable/internal/async_row_sampler_test.cc @@ -13,6 +13,7 @@ // limitations under the License. #include "google/cloud/bigtable/internal/async_row_sampler.h" +#include "google/cloud/bigtable/internal/operation_context.h" #include "google/cloud/bigtable/testing/mock_bigtable_stub.h" #include "google/cloud/grpc_options.h" #include "google/cloud/internal/async_streaming_read_rpc_impl.h" diff --git a/google/cloud/bigtable/internal/bigtable_random_two_least_used_decorator.cc b/google/cloud/bigtable/internal/bigtable_random_two_least_used_decorator.cc index 605c9bcb01b40..47bf9c4d691d3 100644 --- a/google/cloud/bigtable/internal/bigtable_random_two_least_used_decorator.cc +++ b/google/cloud/bigtable/internal/bigtable_random_two_least_used_decorator.cc @@ -13,9 +13,12 @@ // limitations under the License. #include "google/cloud/bigtable/internal/bigtable_random_two_least_used_decorator.h" +#include "google/cloud/bigtable/internal/metrics.h" +#include "google/cloud/bigtable/internal/operation_context.h" #include "google/cloud/async_streaming_read_write_rpc.h" #include "google/cloud/internal/async_streaming_read_rpc.h" #include "google/cloud/internal/streaming_read_rpc.h" +#include "google/cloud/status_or.h" #include #include #include @@ -108,11 +111,29 @@ class AsyncStreamingReadWriteRpcTracking template Response UnaryHelper(std::shared_ptr>& pool, + OperationContext& oc, std::function fn) { - auto child = pool->GetChannelRandomTwoLeastUsed(); - auto stub = child->AcquireStub(); + auto selection = pool->GetChannelRandomTwoLeastUsed(); + oc.StubSelection(StubSelectionParams{ + selection.outstanding_rpcs, ChannelPoolLbPolicy::kRandomTwoLeastUsed, + pool->transport_type(), RpcType::kUnary}); + auto stub = selection.channel->AcquireStub(); auto result = fn(*stub); - child->ReleaseStub(); + selection.channel->ReleaseStub(); + return result; +} + +template +Response AsyncHelper(std::shared_ptr>& pool, + std::shared_ptr operation_context, + std::function fn) { + auto selection = pool->GetChannelRandomTwoLeastUsed(); + operation_context->StubSelection(StubSelectionParams{ + selection.outstanding_rpcs, ChannelPoolLbPolicy::kRandomTwoLeastUsed, + pool->transport_type(), RpcType::kUnary}); + auto stub = selection.channel->AcquireStub(); + auto result = fn(*stub); + selection.channel->ReleaseStub(); return result; } @@ -120,13 +141,17 @@ template std::unique_ptr> StreamingHelper( std::shared_ptr>& pool, + std::shared_ptr operation_context, std::function>(BigtableStub&)> fn) { - auto child = pool->GetChannelRandomTwoLeastUsed(); - auto stub = child->AcquireStub(); + auto selection = pool->GetChannelRandomTwoLeastUsed(); + operation_context->StubSelection(StubSelectionParams{ + selection.outstanding_rpcs, ChannelPoolLbPolicy::kRandomTwoLeastUsed, + pool->transport_type(), RpcType::kStreaming}); + auto stub = selection.channel->AcquireStub(); auto result = fn(*stub); - auto release_fn = [weak = child->MakeWeak()] { + auto release_fn = [weak = selection.channel->MakeWeak()] { auto child = weak.lock(); if (child) child->ReleaseStub(); }; @@ -138,14 +163,18 @@ template std::unique_ptr> AsyncStreamingHelper( std::shared_ptr>& pool, + std::shared_ptr operation_context, std::function>( BigtableStub&)> fn) { - auto child = pool->GetChannelRandomTwoLeastUsed(); - auto stub = child->AcquireStub(); + auto selection = pool->GetChannelRandomTwoLeastUsed(); + operation_context->StubSelection(StubSelectionParams{ + selection.outstanding_rpcs, ChannelPoolLbPolicy::kRandomTwoLeastUsed, + pool->transport_type(), RpcType::kStreaming}); + auto stub = selection.channel->AcquireStub(); auto result = fn(*stub); - auto release_fn = [weak = child->MakeWeak()] { + auto release_fn = [weak = selection.channel->MakeWeak()] { auto child = weak.lock(); if (child) child->ReleaseStub(); }; @@ -157,13 +186,17 @@ template std::unique_ptr> AsyncStreamingHelper( std::shared_ptr>& pool, + std::shared_ptr operation_context, std::function>(BigtableStub&)> fn) { - auto child = pool->GetChannelRandomTwoLeastUsed(); - auto stub = child->AcquireStub(); + auto selection = pool->GetChannelRandomTwoLeastUsed(); + operation_context->StubSelection(StubSelectionParams{ + selection.outstanding_rpcs, ChannelPoolLbPolicy::kRandomTwoLeastUsed, + pool->transport_type(), RpcType::kStreaming}); + auto stub = selection.channel->AcquireStub(); auto result = fn(*stub); - auto release_fn = [weak = child->MakeWeak()] { + auto release_fn = [weak = selection.channel->MakeWeak()] { auto child = weak.lock(); if (child) child->ReleaseStub(); }; @@ -181,9 +214,9 @@ BigtableRandomTwoLeastUsed::ReadRows( google::bigtable::v2::ReadRowsRequest const& request, std::shared_ptr operation_context) { return StreamingHelper( - pool_, [&, context = std::move(context), - operation_context = - std::move(operation_context)](BigtableStub& stub) mutable { + pool_, operation_context, + [&, context = std::move(context), + operation_context](BigtableStub& stub) mutable { return stub.ReadRows(std::move(context), options, request, std::move(operation_context)); }); @@ -196,9 +229,9 @@ BigtableRandomTwoLeastUsed::SampleRowKeys( google::bigtable::v2::SampleRowKeysRequest const& request, std::shared_ptr operation_context) { return StreamingHelper( - pool_, [&, context = std::move(context), - operation_context = - std::move(operation_context)](BigtableStub& stub) mutable { + pool_, operation_context, + [&, context = std::move(context), + operation_context](BigtableStub& stub) mutable { return stub.SampleRowKeys(std::move(context), options, request, std::move(operation_context)); }); @@ -210,7 +243,7 @@ BigtableRandomTwoLeastUsed::MutateRow( google::bigtable::v2::MutateRowRequest const& request, OperationContext& operation_context) { return UnaryHelper>( - pool_, [&](BigtableStub& stub) { + pool_, operation_context, [&](BigtableStub& stub) { return stub.MutateRow(context, options, request, operation_context); }); } @@ -222,9 +255,9 @@ BigtableRandomTwoLeastUsed::MutateRows( google::bigtable::v2::MutateRowsRequest const& request, std::shared_ptr operation_context) { return StreamingHelper( - pool_, [&, context = std::move(context), - operation_context = - std::move(operation_context)](BigtableStub& stub) mutable { + pool_, operation_context, + [&, context = std::move(context), + operation_context](BigtableStub& stub) mutable { return stub.MutateRows(std::move(context), options, request, std::move(operation_context)); }); @@ -236,7 +269,7 @@ BigtableRandomTwoLeastUsed::CheckAndMutateRow( google::bigtable::v2::CheckAndMutateRowRequest const& request, OperationContext& operation_context) { return UnaryHelper>( - pool_, [&](BigtableStub& stub) { + pool_, operation_context, [&](BigtableStub& stub) { return stub.CheckAndMutateRow(context, options, request, operation_context); }); @@ -248,7 +281,7 @@ BigtableRandomTwoLeastUsed::PingAndWarm( google::bigtable::v2::PingAndWarmRequest const& request, OperationContext& operation_context) { return UnaryHelper>( - pool_, [&](BigtableStub& stub) { + pool_, operation_context, [&](BigtableStub& stub) { return stub.PingAndWarm(context, options, request, operation_context); }); } @@ -260,7 +293,7 @@ BigtableRandomTwoLeastUsed::ReadModifyWriteRow( OperationContext& operation_context) { return UnaryHelper< StatusOr>( - pool_, [&](BigtableStub& stub) { + pool_, operation_context, [&](BigtableStub& stub) { return stub.ReadModifyWriteRow(context, options, request, operation_context); }); @@ -272,7 +305,7 @@ BigtableRandomTwoLeastUsed::PrepareQuery( google::bigtable::v2::PrepareQueryRequest const& request, OperationContext& operation_context) { return UnaryHelper>( - pool_, [&](BigtableStub& stub) { + pool_, operation_context, [&](BigtableStub& stub) { return stub.PrepareQuery(context, options, request, operation_context); }); } @@ -284,9 +317,9 @@ BigtableRandomTwoLeastUsed::ExecuteQuery( google::bigtable::v2::ExecuteQueryRequest const& request, std::shared_ptr operation_context) { return StreamingHelper( - pool_, [&, context = std::move(context), - operation_context = - std::move(operation_context)](BigtableStub& stub) mutable { + pool_, operation_context, + [&, context = std::move(context), + operation_context](BigtableStub& stub) mutable { return stub.ExecuteQuery(std::move(context), options, request, std::move(operation_context)); }); @@ -301,9 +334,9 @@ BigtableRandomTwoLeastUsed::AsyncReadRows( google::bigtable::v2::ReadRowsRequest const& request, std::shared_ptr operation_context) { return AsyncStreamingHelper( - pool_, [&, context = std::move(context), options = std::move(options), - operation_context = - std::move(operation_context)](BigtableStub& stub) mutable { + pool_, operation_context, + [&, context = std::move(context), options = std::move(options), + operation_context](BigtableStub& stub) mutable { return stub.AsyncReadRows(cq, std::move(context), std::move(options), request, std::move(operation_context)); }); @@ -318,9 +351,9 @@ BigtableRandomTwoLeastUsed::AsyncSampleRowKeys( google::bigtable::v2::SampleRowKeysRequest const& request, std::shared_ptr operation_context) { return AsyncStreamingHelper( - pool_, [&, context = std::move(context), options = std::move(options), - operation_context = - std::move(operation_context)](BigtableStub& stub) mutable { + pool_, operation_context, + [&, context = std::move(context), options = std::move(options), + operation_context](BigtableStub& stub) mutable { return stub.AsyncSampleRowKeys(cq, std::move(context), std::move(options), request, std::move(operation_context)); @@ -334,10 +367,10 @@ BigtableRandomTwoLeastUsed::AsyncMutateRow( google::cloud::internal::ImmutableOptions options, google::bigtable::v2::MutateRowRequest const& request, std::shared_ptr operation_context) { - return UnaryHelper>>( - pool_, [&, context = std::move(context), options = std::move(options), - operation_context = - std::move(operation_context)](BigtableStub& stub) mutable { + return AsyncHelper>>( + pool_, operation_context, + [&, context = std::move(context), options = std::move(options), + operation_context](BigtableStub& stub) mutable { return stub.AsyncMutateRow(cq, std::move(context), std::move(options), request, std::move(operation_context)); }); @@ -352,9 +385,9 @@ BigtableRandomTwoLeastUsed::AsyncMutateRows( google::bigtable::v2::MutateRowsRequest const& request, std::shared_ptr operation_context) { return AsyncStreamingHelper( - pool_, [&, context = std::move(context), options = std::move(options), - operation_context = - std::move(operation_context)](BigtableStub& stub) mutable { + pool_, operation_context, + [&, context = std::move(context), options = std::move(options), + operation_context](BigtableStub& stub) mutable { return stub.AsyncMutateRows(cq, std::move(context), std::move(options), request, std::move(operation_context)); }); @@ -367,11 +400,11 @@ BigtableRandomTwoLeastUsed::AsyncCheckAndMutateRow( google::cloud::internal::ImmutableOptions options, google::bigtable::v2::CheckAndMutateRowRequest const& request, std::shared_ptr operation_context) { - return UnaryHelper< + return AsyncHelper< future>>( - pool_, [&, context = std::move(context), options = std::move(options), - operation_context = - std::move(operation_context)](BigtableStub& stub) mutable { + pool_, operation_context, + [&, context = std::move(context), options = std::move(options), + operation_context](BigtableStub& stub) mutable { return stub.AsyncCheckAndMutateRow(cq, std::move(context), std::move(options), request, std::move(operation_context)); @@ -385,11 +418,11 @@ BigtableRandomTwoLeastUsed::AsyncPingAndWarm( google::cloud::internal::ImmutableOptions options, google::bigtable::v2::PingAndWarmRequest const& request, std::shared_ptr operation_context) { - return UnaryHelper< + return AsyncHelper< future>>( - pool_, [&, context = std::move(context), options = std::move(options), - operation_context = - std::move(operation_context)](BigtableStub& stub) mutable { + pool_, operation_context, + [&, context = std::move(context), options = std::move(options), + operation_context](BigtableStub& stub) mutable { return stub.AsyncPingAndWarm(cq, std::move(context), std::move(options), request, std::move(operation_context)); }); @@ -402,11 +435,11 @@ BigtableRandomTwoLeastUsed::AsyncReadModifyWriteRow( google::cloud::internal::ImmutableOptions options, google::bigtable::v2::ReadModifyWriteRowRequest const& request, std::shared_ptr operation_context) { - return UnaryHelper< + return AsyncHelper< future>>( - pool_, [&, context = std::move(context), options = std::move(options), - operation_context = - std::move(operation_context)](BigtableStub& stub) mutable { + pool_, operation_context, + [&, context = std::move(context), options = std::move(options), + operation_context](BigtableStub& stub) mutable { return stub.AsyncReadModifyWriteRow(cq, std::move(context), std::move(options), request, std::move(operation_context)); @@ -420,11 +453,11 @@ BigtableRandomTwoLeastUsed::AsyncPrepareQuery( google::cloud::internal::ImmutableOptions options, google::bigtable::v2::PrepareQueryRequest const& request, std::shared_ptr operation_context) { - return UnaryHelper< + return AsyncHelper< future>>( - pool_, [&, context = std::move(context), options = std::move(options), - operation_context = - std::move(operation_context)](BigtableStub& stub) mutable { + pool_, operation_context, + [&, context = std::move(context), options = std::move(options), + operation_context](BigtableStub& stub) mutable { return stub.AsyncPrepareQuery(cq, std::move(context), std::move(options), request, std::move(operation_context)); @@ -437,7 +470,7 @@ BigtableRandomTwoLeastUsed::GetClientConfiguration( google::bigtable::v2::GetClientConfigurationRequest const& request, OperationContext& operation_context) { return UnaryHelper>( - pool_, [&](BigtableStub& stub) { + pool_, operation_context, [&](BigtableStub& stub) { return stub.GetClientConfiguration(context, options, request, operation_context); }); @@ -453,9 +486,9 @@ BigtableRandomTwoLeastUsed::AsyncOpenTable( std::shared_ptr operation_context) { return AsyncStreamingHelper( - pool_, [&, context = std::move(context), options = std::move(options), - operation_context = - std::move(operation_context)](BigtableStub& stub) mutable { + pool_, operation_context, + [&, context = std::move(context), options = std::move(options), + operation_context](BigtableStub& stub) mutable { return stub.AsyncOpenTable(cq, std::move(context), std::move(options), std::move(operation_context)); }); @@ -471,9 +504,9 @@ BigtableRandomTwoLeastUsed::AsyncOpenAuthorizedView( std::shared_ptr operation_context) { return AsyncStreamingHelper( - pool_, [&, context = std::move(context), options = std::move(options), - operation_context = - std::move(operation_context)](BigtableStub& stub) mutable { + pool_, operation_context, + [&, context = std::move(context), options = std::move(options), + operation_context](BigtableStub& stub) mutable { return stub.AsyncOpenAuthorizedView(cq, std::move(context), std::move(options), std::move(operation_context)); @@ -490,9 +523,9 @@ BigtableRandomTwoLeastUsed::AsyncOpenMaterializedView( std::shared_ptr operation_context) { return AsyncStreamingHelper( - pool_, [&, context = std::move(context), options = std::move(options), - operation_context = - std::move(operation_context)](BigtableStub& stub) mutable { + pool_, operation_context, + [&, context = std::move(context), options = std::move(options), + operation_context](BigtableStub& stub) mutable { return stub.AsyncOpenMaterializedView(cq, std::move(context), std::move(options), std::move(operation_context)); diff --git a/google/cloud/bigtable/internal/bigtable_round_robin_decorator.cc b/google/cloud/bigtable/internal/bigtable_round_robin_decorator.cc index 5df35deb3f3fd..a0af88795e411 100644 --- a/google/cloud/bigtable/internal/bigtable_round_robin_decorator.cc +++ b/google/cloud/bigtable/internal/bigtable_round_robin_decorator.cc @@ -12,11 +12,9 @@ // See the License for the specific language governing permissions and // limitations under the License. -// Generated by the Codegen C++ plugin. -// If you make any local changes, they will be lost. -// source: google/bigtable/v2/bigtable.proto - #include "google/cloud/bigtable/internal/bigtable_round_robin_decorator.h" +#include "google/cloud/bigtable/internal/defaults.h" +#include "google/cloud/bigtable/internal/operation_context.h" #include #include #include diff --git a/google/cloud/bigtable/internal/bigtable_round_robin_decorator.h b/google/cloud/bigtable/internal/bigtable_round_robin_decorator.h index 95bf02e7c06d5..8bcd676650367 100644 --- a/google/cloud/bigtable/internal/bigtable_round_robin_decorator.h +++ b/google/cloud/bigtable/internal/bigtable_round_robin_decorator.h @@ -12,10 +12,6 @@ // See the License for the specific language governing permissions and // limitations under the License. -// Generated by the Codegen C++ plugin. -// If you make any local changes, they will be lost. -// source: google/bigtable/v2/bigtable.proto - #ifndef GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_BIGTABLE_INTERNAL_BIGTABLE_ROUND_ROBIN_DECORATOR_H #define GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_BIGTABLE_INTERNAL_BIGTABLE_ROUND_ROBIN_DECORATOR_H diff --git a/google/cloud/bigtable/internal/bigtable_stub_factory.cc b/google/cloud/bigtable/internal/bigtable_stub_factory.cc index d2b3e45c22083..8bea357c31eb7 100644 --- a/google/cloud/bigtable/internal/bigtable_stub_factory.cc +++ b/google/cloud/bigtable/internal/bigtable_stub_factory.cc @@ -178,7 +178,10 @@ std::shared_ptr 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 CreateDecoratedStubs( diff --git a/google/cloud/bigtable/internal/bigtable_stub_factory_test.cc b/google/cloud/bigtable/internal/bigtable_stub_factory_test.cc index c5be858f15f93..4b2c45e33304e 100644 --- a/google/cloud/bigtable/internal/bigtable_stub_factory_test.cc +++ b/google/cloud/bigtable/internal/bigtable_stub_factory_test.cc @@ -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"); @@ -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" diff --git a/google/cloud/bigtable/internal/connection_refresh_state_test.cc b/google/cloud/bigtable/internal/connection_refresh_state_test.cc index 8aa93a0cf0257..d22ca33c909c8 100644 --- a/google/cloud/bigtable/internal/connection_refresh_state_test.cc +++ b/google/cloud/bigtable/internal/connection_refresh_state_test.cc @@ -156,7 +156,7 @@ TEST_F(ScheduleStubRefreshTest, RefreshedUsingAsyncPingAndWarm) { [&](CompletionQueue&, std::shared_ptr const&, internal::ImmutableOptions const&, google::bigtable::v2::PingAndWarmRequest const& request, - auto const&) + std::shared_ptr) -> future> { EXPECT_THAT(request.name(), Eq(instance_name)); return p2.get_future(); diff --git a/google/cloud/bigtable/internal/data_connection_impl.cc b/google/cloud/bigtable/internal/data_connection_impl.cc index bc1252f1681fe..9f03e8bb44f96 100644 --- a/google/cloud/bigtable/internal/data_connection_impl.cc +++ b/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/grpc_metrics_exporter.h" +#include "google/cloud/bigtable/internal/defaults.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,8 +39,12 @@ #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 #include @@ -197,6 +201,7 @@ std::string_view InstanceNameFromTableName(std::string_view table_name) { if (pos == std::string_view::npos) return {}; return table_name.substr(0, pos); } + } // namespace bigtable::Row TransformReadModifyWriteRowResponse( diff --git a/google/cloud/bigtable/internal/data_connection_impl_test.cc b/google/cloud/bigtable/internal/data_connection_impl_test.cc index e5413b611e16f..03f4b1d605115 100644 --- a/google/cloud/bigtable/internal/data_connection_impl_test.cc +++ b/google/cloud/bigtable/internal/data_connection_impl_test.cc @@ -16,7 +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/operation_context.h" #include "google/cloud/bigtable/internal/query_plan.h" #ifdef GOOGLE_CLOUD_CPP_BIGTABLE_WITH_OTEL_METRICS #include "google/cloud/bigtable/internal/metrics.h" @@ -512,7 +512,8 @@ TEST_F(DataConnectionTest, ApplySuccess) { auto mock = std::make_shared(); EXPECT_CALL(*mock, MutateRow) .WillOnce([](grpc::ClientContext&, Options const&, - v2::MutateRowRequest const& request, auto const&) { + v2::MutateRowRequest const& request, + bigtable_internal::OperationContext&) { EXPECT_EQ(kAppProfile, request.app_profile_id()); EXPECT_EQ(kTableName, request.table_name()); EXPECT_EQ("row", request.row_key()); @@ -544,7 +545,8 @@ TEST_F(DataConnectionTest, ApplyPermanentFailure) { auto mock = std::make_shared(); EXPECT_CALL(*mock, MutateRow) .WillOnce([](grpc::ClientContext&, Options const&, - v2::MutateRowRequest const& request, auto const&) { + v2::MutateRowRequest const& request, + bigtable_internal::OperationContext&) { EXPECT_EQ(kAppProfile, request.app_profile_id()); EXPECT_EQ(kTableName, request.table_name()); EXPECT_EQ("row", request.row_key()); @@ -576,14 +578,16 @@ TEST_F(DataConnectionTest, ApplyRetryThenSuccess) { auto mock = std::make_shared(); EXPECT_CALL(*mock, MutateRow) .WillOnce([](grpc::ClientContext&, Options const&, - v2::MutateRowRequest const& request, auto const&) { + v2::MutateRowRequest const& request, + bigtable_internal::OperationContext&) { EXPECT_EQ(kAppProfile, request.app_profile_id()); EXPECT_EQ(kTableName, request.table_name()); EXPECT_EQ("row", request.row_key()); return TransientError(); }) .WillOnce([](grpc::ClientContext&, Options const&, - v2::MutateRowRequest const& request, auto const&) { + v2::MutateRowRequest const& request, + bigtable_internal::OperationContext&) { EXPECT_EQ(kAppProfile, request.app_profile_id()); EXPECT_EQ(kTableName, request.table_name()); EXPECT_EQ("row", request.row_key()); @@ -616,7 +620,8 @@ TEST_F(DataConnectionTest, ApplyRetryExhausted) { EXPECT_CALL(*mock, MutateRow) .Times(kNumRetries + 1) .WillRepeatedly([](grpc::ClientContext&, Options const&, - v2::MutateRowRequest const& request, auto const&) { + v2::MutateRowRequest const& request, + bigtable_internal::OperationContext&) { EXPECT_EQ(kAppProfile, request.app_profile_id()); EXPECT_EQ(kTableName, request.table_name()); EXPECT_EQ("row", request.row_key()); @@ -641,7 +646,8 @@ TEST_F(DataConnectionTest, ApplyRetryIdempotency) { auto mock = std::make_shared(); EXPECT_CALL(*mock, MutateRow) .WillOnce([](grpc::ClientContext&, Options const&, - v2::MutateRowRequest const& request, auto const&) { + v2::MutateRowRequest const& request, + bigtable_internal::OperationContext&) { EXPECT_EQ(kAppProfile, request.app_profile_id()); EXPECT_EQ(kTableName, request.table_name()); EXPECT_EQ("row", request.row_key()); @@ -682,14 +688,16 @@ TEST_F(DataConnectionTest, ApplyBigtableCookie) { auto mock = std::make_shared(); EXPECT_CALL(*mock, MutateRow) .WillOnce([this](grpc::ClientContext& context, Options const&, - v2::MutateRowRequest const&, auto const&) { + v2::MutateRowRequest const&, + bigtable_internal::OperationContext&) { // Return a bigtable cookie in the first request. metadata_fixture_.SetServerMetadata( context, {{}, {{"x-goog-cbt-cookie-routing", "routing"}}}); return TransientError(); }) .WillOnce([this](grpc::ClientContext& context, Options const&, - v2::MutateRowRequest const&, auto const&) { + v2::MutateRowRequest const&, + bigtable_internal::OperationContext&) { // Verify that the next request includes the bigtable cookie from // above. auto headers = metadata_fixture_.GetMetadata(context); @@ -732,7 +740,8 @@ TEST_F(DataConnectionTest, AsyncApplySuccess) { auto mock = std::make_shared(); EXPECT_CALL(*mock, AsyncMutateRow) .WillOnce([](google::cloud::CompletionQueue&, auto, auto, - v2::MutateRowRequest const& request, auto const&) { + v2::MutateRowRequest const& request, + std::shared_ptr) { EXPECT_EQ(kAppProfile, request.app_profile_id()); EXPECT_EQ(kTableName, request.table_name()); EXPECT_EQ("row", request.row_key()); @@ -749,7 +758,8 @@ TEST_F(DataConnectionTest, AsyncApplyPermanentFailure) { auto mock = std::make_shared(); EXPECT_CALL(*mock, AsyncMutateRow) .WillOnce([](google::cloud::CompletionQueue&, auto, auto, - v2::MutateRowRequest const& request, auto const&) { + v2::MutateRowRequest const& request, + std::shared_ptr) { EXPECT_EQ(kAppProfile, request.app_profile_id()); EXPECT_EQ(kTableName, request.table_name()); EXPECT_EQ("row", request.row_key()); @@ -783,7 +793,8 @@ TEST_F(DataConnectionTest, AsyncApplyRetryExhausted) { EXPECT_CALL(*mock, AsyncMutateRow) .Times(kNumRetries + 1) .WillRepeatedly([](google::cloud::CompletionQueue&, auto, auto, - v2::MutateRowRequest const& request, auto const&) { + v2::MutateRowRequest const& request, + std::shared_ptr) { EXPECT_EQ(kAppProfile, request.app_profile_id()); EXPECT_EQ(kTableName, request.table_name()); EXPECT_EQ("row", request.row_key()); @@ -809,7 +820,8 @@ TEST_F(DataConnectionTest, AsyncApplyRetryIdempotency) { auto mock = std::make_shared(); EXPECT_CALL(*mock, AsyncMutateRow) .WillOnce([](google::cloud::CompletionQueue&, auto, auto, - v2::MutateRowRequest const& request, auto const&) { + v2::MutateRowRequest const& request, + std::shared_ptr) { EXPECT_EQ(kAppProfile, request.app_profile_id()); EXPECT_EQ(kTableName, request.table_name()); EXPECT_EQ("row", request.row_key()); @@ -837,7 +849,8 @@ TEST_F(DataConnectionTest, AsyncApplyBigtableCookie) { EXPECT_CALL(*mock, AsyncMutateRow) .WillOnce([this](CompletionQueue&, std::shared_ptr const& context, - auto, v2::MutateRowRequest const&, auto const&) { + auto, v2::MutateRowRequest const&, + std::shared_ptr) { // Return a bigtable cookie in the first request. metadata_fixture_.SetServerMetadata( *context, {{}, {{"x-goog-cbt-cookie-routing", "routing"}}}); @@ -846,7 +859,8 @@ TEST_F(DataConnectionTest, AsyncApplyBigtableCookie) { }) .WillOnce([this](CompletionQueue&, std::shared_ptr const& context, - auto, v2::MutateRowRequest const&, auto const&) { + auto, v2::MutateRowRequest const&, + std::shared_ptr) { // Verify that the next request includes the bigtable cookie from above. auto headers = metadata_fixture_.GetMetadata(*context); EXPECT_THAT(headers, @@ -915,7 +929,7 @@ TEST_F(DataConnectionTest, BulkApplySuccess) { EXPECT_CALL(*mock, MutateRows) .WillOnce([](auto, auto const&, google::bigtable::v2::MutateRowsRequest const& request, - auto const&) { + std::shared_ptr) { EXPECT_EQ(kAppProfile, request.app_profile_id()); EXPECT_EQ(kTableName, request.table_name()); EXPECT_THAT(request.entries(), ElementsAre(Entry("r0"), Entry("r1"))); @@ -964,7 +978,7 @@ TEST_F(DataConnectionTest, BulkApplyRetryMutationPolicy) { EXPECT_CALL(*mock, MutateRows) .WillOnce([](auto, auto const&, google::bigtable::v2::MutateRowsRequest const& request, - auto const&) { + std::shared_ptr) { EXPECT_EQ(kAppProfile, request.app_profile_id()); EXPECT_EQ(kTableName, request.table_name()); auto stream = std::make_unique(); @@ -982,7 +996,7 @@ TEST_F(DataConnectionTest, BulkApplyRetryMutationPolicy) { }) .WillOnce([](auto, auto const&, google::bigtable::v2::MutateRowsRequest const& request, - auto const&) { + std::shared_ptr) { EXPECT_EQ(kAppProfile, request.app_profile_id()); EXPECT_EQ(kTableName, request.table_name()); EXPECT_THAT(request.entries(), @@ -1026,7 +1040,7 @@ TEST_F(DataConnectionTest, BulkApplyIncompleteStreamRetried) { EXPECT_CALL(*mock, MutateRows) .WillOnce([](auto, auto const&, google::bigtable::v2::MutateRowsRequest const& request, - auto const&) { + std::shared_ptr) { EXPECT_EQ(kAppProfile, request.app_profile_id()); EXPECT_EQ(kTableName, request.table_name()); auto stream = std::make_unique(); @@ -1040,7 +1054,7 @@ TEST_F(DataConnectionTest, BulkApplyIncompleteStreamRetried) { }) .WillOnce([](auto, auto const&, google::bigtable::v2::MutateRowsRequest const& request, - auto const&) { + std::shared_ptr) { EXPECT_EQ(kAppProfile, request.app_profile_id()); EXPECT_EQ(kTableName, request.table_name()); EXPECT_THAT(request.entries(), ElementsAre(Entry("forgotten"))); @@ -1083,7 +1097,7 @@ TEST_F(DataConnectionTest, BulkApplyStreamRetryExhausted) { .Times(kNumRetries + 1) .WillRepeatedly([](auto, auto const&, google::bigtable::v2::MutateRowsRequest const& request, - auto const&) { + std::shared_ptr) { EXPECT_EQ(kAppProfile, request.app_profile_id()); EXPECT_EQ(kTableName, request.table_name()); auto stream = std::make_unique(); @@ -1128,7 +1142,7 @@ TEST_F(DataConnectionTest, BulkApplyStreamPermanentError) { EXPECT_CALL(*mock, MutateRows) .WillOnce([](auto, auto const&, google::bigtable::v2::MutateRowsRequest const& request, - auto const&) { + std::shared_ptr) { EXPECT_EQ(kAppProfile, request.app_profile_id()); EXPECT_EQ(kTableName, request.table_name()); auto stream = std::make_unique(); @@ -1150,7 +1164,7 @@ TEST_F(DataConnectionTest, BulkApplyNoSleepIfNoPendingMutations) { EXPECT_CALL(*mock, MutateRows) .WillOnce([](auto, auto const&, google::bigtable::v2::MutateRowsRequest const& request, - auto const&) { + std::shared_ptr) { EXPECT_EQ(kAppProfile, request.app_profile_id()); EXPECT_EQ(kTableName, request.table_name()); auto stream = std::make_unique(); @@ -1184,11 +1198,10 @@ TEST_F(DataConnectionTest, BulkApplyRetriesOkStreamWithFailedMutations) { .Times(kNumRetries + 1) .WillRepeatedly([](auto, auto const&, google::bigtable::v2::MutateRowsRequest const& request, - auto const&) { + std::shared_ptr) { EXPECT_EQ(kAppProfile, request.app_profile_id()); EXPECT_EQ(kTableName, request.table_name()); auto stream = std::make_unique(); - // The overall stream succeeds, but it contains failed mutations. // Our retry and backoff policies should take effect. EXPECT_CALL(*stream, Read) .WillOnce([](google::bigtable::v2::MutateRowsResponse* r) { @@ -1218,25 +1231,25 @@ TEST_F(DataConnectionTest, BulkApplyRetryInfoHeeded) { auto mock = std::make_shared(); EXPECT_CALL(*mock, MutateRows) - .WillOnce( - [](auto, auto const&, v2::MutateRowsRequest const&, auto const&) { - auto status = PermanentError(); - internal::SetRetryInfo(status, internal::RetryInfo{ms(0)}); - auto stream = std::make_unique(); - EXPECT_CALL(*stream, Read).WillOnce(Return(status)); - return stream; - }) - .WillOnce( - [](auto, auto const&, v2::MutateRowsRequest const&, auto const&) { - auto stream = std::make_unique(); - EXPECT_CALL(*stream, Read) - .WillOnce([](google::bigtable::v2::MutateRowsResponse* r) { - *r = MakeBulkApplyResponse({{0, grpc::StatusCode::OK}}); - return std::nullopt; - }) - .WillOnce(Return(Status())); - return stream; - }); + .WillOnce([](auto, auto const&, v2::MutateRowsRequest const&, + std::shared_ptr) { + auto status = PermanentError(); + internal::SetRetryInfo(status, internal::RetryInfo{ms(0)}); + auto stream = std::make_unique(); + EXPECT_CALL(*stream, Read).WillOnce(Return(status)); + return stream; + }) + .WillOnce([](auto, auto const&, v2::MutateRowsRequest const&, + std::shared_ptr) { + auto stream = std::make_unique(); + EXPECT_CALL(*stream, Read) + .WillOnce([](google::bigtable::v2::MutateRowsResponse* r) { + *r = MakeBulkApplyResponse({{0, grpc::StatusCode::OK}}); + return std::nullopt; + }) + .WillOnce(Return(Status())); + return stream; + }); auto conn = TestConnection(std::move(mock)); internal::OptionsSpan span( @@ -1251,14 +1264,14 @@ TEST_F(DataConnectionTest, BulkApplyRetryInfoIgnored) { auto mock = std::make_shared(); EXPECT_CALL(*mock, MutateRows) - .WillOnce( - [](auto, auto const&, v2::MutateRowsRequest const&, auto const&) { - auto status = PermanentError(); - internal::SetRetryInfo(status, internal::RetryInfo{ms(0)}); - auto stream = std::make_unique(); - EXPECT_CALL(*stream, Read).WillOnce(Return(status)); - return stream; - }); + .WillOnce([](auto, auto const&, v2::MutateRowsRequest const&, + std::shared_ptr) { + auto status = PermanentError(); + internal::SetRetryInfo(status, internal::RetryInfo{ms(0)}); + auto stream = std::make_unique(); + EXPECT_CALL(*stream, Read).WillOnce(Return(status)); + return stream; + }); auto conn = TestConnection(std::move(mock)); internal::OptionsSpan span( @@ -1303,7 +1316,8 @@ TEST_F(DataConnectionTest, AsyncBulkApply) { .WillOnce(Return(ByMove(make_ready_future()))); EXPECT_CALL(*mock_stub, AsyncMutateRows) .WillOnce([](CompletionQueue const&, auto, auto, - v2::MutateRowsRequest const& request, auto const&) { + v2::MutateRowsRequest const& request, + std::shared_ptr) { EXPECT_EQ(kAppProfile, request.app_profile_id()); EXPECT_EQ(kTableName, request.table_name()); using ErrorStream = @@ -1325,7 +1339,7 @@ TEST_F(DataConnectionTest, ReadRows) { EXPECT_CALL(*mock, ReadRows) .WillOnce([](auto, auto const&, google::bigtable::v2::ReadRowsRequest const& request, - auto const&) { + std::shared_ptr) { EXPECT_EQ(kAppProfile, request.app_profile_id()); EXPECT_EQ(kTableName, request.table_name()); EXPECT_EQ(42, request.rows_limit()); @@ -1349,7 +1363,7 @@ TEST_F(DataConnectionTest, ReadRowsReverseScan) { EXPECT_CALL(*mock, ReadRows) .WillOnce([](auto, auto const&, google::bigtable::v2::ReadRowsRequest const& request, - auto const&) { + std::shared_ptr) { EXPECT_TRUE(request.reversed()); auto stream = std::make_unique(); @@ -1370,7 +1384,7 @@ TEST_F(DataConnectionTest, ReadRowsFull) { EXPECT_CALL(*mock, ReadRows) .WillOnce([](auto, auto const&, google::bigtable::v2::ReadRowsRequest const& request, - auto const&) { + std::shared_ptr) { EXPECT_EQ(kAppProfile, request.app_profile_id()); EXPECT_EQ(kTableName, request.table_name()); EXPECT_EQ(42, request.rows_limit()); @@ -1394,7 +1408,8 @@ TEST_F(DataConnectionTest, ReadRowsRetryInfoHeeded) { auto mock = std::make_shared(); EXPECT_CALL(*mock, ReadRows) .WillOnce([](auto, auto const&, - google::bigtable::v2::ReadRowsRequest const&, auto const&) { + google::bigtable::v2::ReadRowsRequest const&, + std::shared_ptr) { auto status = PermanentError(); internal::SetRetryInfo(status, internal::RetryInfo{ms(0)}); auto stream = std::make_unique(); @@ -1402,7 +1417,8 @@ TEST_F(DataConnectionTest, ReadRowsRetryInfoHeeded) { return stream; }) .WillOnce([](auto, auto const&, - google::bigtable::v2::ReadRowsRequest const&, auto const&) { + google::bigtable::v2::ReadRowsRequest const&, + std::shared_ptr) { auto stream = std::make_unique(); EXPECT_CALL(*stream, Read).WillOnce(Return(Status())); return stream; @@ -1420,7 +1436,8 @@ TEST_F(DataConnectionTest, ReadRowsRetryInfoIgnored) { auto mock = std::make_shared(); EXPECT_CALL(*mock, ReadRows) .WillOnce([](auto, auto const&, - google::bigtable::v2::ReadRowsRequest const&, auto const&) { + google::bigtable::v2::ReadRowsRequest const&, + std::shared_ptr) { auto status = PermanentError(); internal::SetRetryInfo(status, internal::RetryInfo{ms(0)}); auto stream = std::make_unique(); @@ -1458,7 +1475,7 @@ TEST_F(DataConnectionTest, ReadRowEmpty) { EXPECT_CALL(*mock, ReadRows) .WillOnce([](auto, auto const&, google::bigtable::v2::ReadRowsRequest const& request, - auto const&) { + std::shared_ptr) { EXPECT_EQ(kAppProfile, request.app_profile_id()); EXPECT_EQ(kTableName, request.table_name()); EXPECT_EQ(1, request.rows_limit()); @@ -1497,7 +1514,7 @@ TEST_F(DataConnectionTest, ReadRowSuccess) { EXPECT_CALL(*mock, ReadRows) .WillOnce([](auto, auto const&, google::bigtable::v2::ReadRowsRequest const& request, - auto const&) { + std::shared_ptr) { EXPECT_EQ(kAppProfile, request.app_profile_id()); EXPECT_EQ(kTableName, request.table_name()); EXPECT_EQ(1, request.rows_limit()); @@ -1548,7 +1565,7 @@ TEST_F(DataConnectionTest, ReadRowFailure) { EXPECT_CALL(*mock, ReadRows) .WillOnce([](auto, auto const&, google::bigtable::v2::ReadRowsRequest const& request, - auto const&) { + std::shared_ptr) { EXPECT_EQ(kAppProfile, request.app_profile_id()); EXPECT_EQ(kTableName, request.table_name()); EXPECT_EQ(1, request.rows_limit()); @@ -1598,7 +1615,8 @@ TEST_F(DataConnectionTest, CheckAndMutateRowSuccess) { auto mock = std::make_shared(); EXPECT_CALL(*mock, CheckAndMutateRow) .WillOnce([&](grpc::ClientContext&, Options const&, - v2::CheckAndMutateRowRequest const& request, auto const&) { + v2::CheckAndMutateRowRequest const& request, + bigtable_internal::OperationContext&) { EXPECT_EQ(kAppProfile, request.app_profile_id()); EXPECT_EQ(kTableName, request.table_name()); EXPECT_EQ("row", request.row_key()); @@ -1613,7 +1631,8 @@ TEST_F(DataConnectionTest, CheckAndMutateRowSuccess) { return resp; }) .WillOnce([&](grpc::ClientContext&, Options const&, - v2::CheckAndMutateRowRequest const& request, auto const&) { + v2::CheckAndMutateRowRequest const& request, + bigtable_internal::OperationContext&) { EXPECT_EQ(kAppProfile, request.app_profile_id()); EXPECT_EQ(kTableName, request.table_name()); EXPECT_EQ("row", request.row_key()); @@ -1650,7 +1669,8 @@ TEST_F(DataConnectionTest, CheckAndMutateRowIdempotency) { auto mock = std::make_shared(); EXPECT_CALL(*mock, CheckAndMutateRow) .WillOnce([&](grpc::ClientContext&, Options const&, - v2::CheckAndMutateRowRequest const& request, auto const&) { + v2::CheckAndMutateRowRequest const& request, + bigtable_internal::OperationContext&) { EXPECT_EQ(kAppProfile, request.app_profile_id()); EXPECT_EQ(kTableName, request.table_name()); EXPECT_EQ("row", request.row_key()); @@ -1703,7 +1723,8 @@ TEST_F(DataConnectionTest, CheckAndMutateRowPermanentError) { auto mock = std::make_shared(); EXPECT_CALL(*mock, CheckAndMutateRow) .WillOnce([&](grpc::ClientContext&, Options const&, - v2::CheckAndMutateRowRequest const& request, auto const&) { + v2::CheckAndMutateRowRequest const& request, + bigtable_internal::OperationContext&) { EXPECT_EQ(kAppProfile, request.app_profile_id()); EXPECT_EQ(kTableName, request.table_name()); EXPECT_EQ("row", request.row_key()); @@ -1748,7 +1769,7 @@ TEST_F(DataConnectionTest, CheckAndMutateRowRetryExhausted) { .Times(kNumRetries + 1) .WillRepeatedly([&](grpc::ClientContext&, Options const&, v2::CheckAndMutateRowRequest const& request, - auto const&) { + bigtable_internal::OperationContext&) { EXPECT_EQ(kAppProfile, request.app_profile_id()); EXPECT_EQ(kTableName, request.table_name()); EXPECT_EQ("row", request.row_key()); @@ -1787,14 +1808,16 @@ TEST_F(DataConnectionTest, CheckAndMutateRowBigtableCookie) { auto mock = std::make_shared(); EXPECT_CALL(*mock, CheckAndMutateRow) .WillOnce([this](grpc::ClientContext& context, Options const&, - v2::CheckAndMutateRowRequest const&, auto const&) { + v2::CheckAndMutateRowRequest const&, + bigtable_internal::OperationContext&) { // Return a bigtable cookie in the first request. metadata_fixture_.SetServerMetadata( context, {{}, {{"x-goog-cbt-cookie-routing", "routing"}}}); return TransientError(); }) .WillOnce([this](grpc::ClientContext& context, Options const&, - v2::CheckAndMutateRowRequest const&, auto const&) { + v2::CheckAndMutateRowRequest const&, + bigtable_internal::OperationContext&) { // Verify that the next request includes the bigtable cookie from above. auto headers = metadata_fixture_.GetMetadata(context); EXPECT_THAT(headers, @@ -1853,7 +1876,8 @@ TEST_F(DataConnectionTest, AsyncCheckAndMutateRowSuccess) { auto mock = std::make_shared(); EXPECT_CALL(*mock, AsyncCheckAndMutateRow) .WillOnce([&](google::cloud::CompletionQueue&, auto, auto, - v2::CheckAndMutateRowRequest const& request, auto const&) { + v2::CheckAndMutateRowRequest const& request, + std::shared_ptr) { EXPECT_EQ(kAppProfile, request.app_profile_id()); EXPECT_EQ(kTableName, request.table_name()); EXPECT_EQ("row", request.row_key()); @@ -1868,7 +1892,8 @@ TEST_F(DataConnectionTest, AsyncCheckAndMutateRowSuccess) { return make_ready_future(make_status_or(resp)); }) .WillOnce([&](google::cloud::CompletionQueue&, auto, auto, - v2::CheckAndMutateRowRequest const& request, auto const&) { + v2::CheckAndMutateRowRequest const& request, + std::shared_ptr) { EXPECT_EQ(kAppProfile, request.app_profile_id()); EXPECT_EQ(kTableName, request.table_name()); EXPECT_EQ("row", request.row_key()); @@ -1907,7 +1932,8 @@ TEST_F(DataConnectionTest, AsyncCheckAndMutateRowIdempotency) { auto mock = std::make_shared(); EXPECT_CALL(*mock, AsyncCheckAndMutateRow) .WillOnce([&](google::cloud::CompletionQueue&, auto, auto, - v2::CheckAndMutateRowRequest const& request, auto const&) { + v2::CheckAndMutateRowRequest const& request, + std::shared_ptr) { EXPECT_EQ(kAppProfile, request.app_profile_id()); EXPECT_EQ(kTableName, request.table_name()); EXPECT_EQ("row", request.row_key()); @@ -1961,7 +1987,8 @@ TEST_F(DataConnectionTest, AsyncCheckAndMutateRowPermanentError) { auto mock = std::make_shared(); EXPECT_CALL(*mock, AsyncCheckAndMutateRow) .WillOnce([&](google::cloud::CompletionQueue&, auto, auto, - v2::CheckAndMutateRowRequest const& request, auto const&) { + v2::CheckAndMutateRowRequest const& request, + std::shared_ptr) { EXPECT_EQ(kAppProfile, request.app_profile_id()); EXPECT_EQ(kTableName, request.table_name()); EXPECT_EQ("row", request.row_key()); @@ -2005,20 +2032,21 @@ TEST_F(DataConnectionTest, AsyncCheckAndMutateRowRetryExhausted) { auto mock = std::make_shared(); EXPECT_CALL(*mock, AsyncCheckAndMutateRow) .Times(kNumRetries + 1) - .WillRepeatedly([&](google::cloud::CompletionQueue&, auto, auto, - v2::CheckAndMutateRowRequest const& request, - auto const&) { - EXPECT_EQ(kAppProfile, request.app_profile_id()); - EXPECT_EQ(kTableName, request.table_name()); - EXPECT_EQ("row", request.row_key()); - EXPECT_THAT(request.predicate_filter(), IsTestFilter()); - EXPECT_THAT(request.true_mutations(), - ElementsAre(MatchMutation(t1), MatchMutation(t2))); - EXPECT_THAT(request.false_mutations(), - ElementsAre(MatchMutation(f1), MatchMutation(f2))); - return make_ready_future>( - TransientError()); - }); + .WillRepeatedly( + [&](google::cloud::CompletionQueue&, auto, auto, + v2::CheckAndMutateRowRequest const& request, + std::shared_ptr) { + EXPECT_EQ(kAppProfile, request.app_profile_id()); + EXPECT_EQ(kTableName, request.table_name()); + EXPECT_EQ("row", request.row_key()); + EXPECT_THAT(request.predicate_filter(), IsTestFilter()); + EXPECT_THAT(request.true_mutations(), + ElementsAre(MatchMutation(t1), MatchMutation(t2))); + EXPECT_THAT(request.false_mutations(), + ElementsAre(MatchMutation(f1), MatchMutation(f2))); + return make_ready_future>( + TransientError()); + }); auto mock_b = std::make_unique(); EXPECT_CALL(*mock_b, clone).WillOnce([]() { @@ -2048,7 +2076,8 @@ TEST_F(DataConnectionTest, AsyncCheckAndMutateRowBigtableCookie) { EXPECT_CALL(*mock, AsyncCheckAndMutateRow) .WillOnce([this](CompletionQueue&, std::shared_ptr const& context, - auto, v2::CheckAndMutateRowRequest const&, auto const&) { + auto, v2::CheckAndMutateRowRequest const&, + std::shared_ptr) { // Return a bigtable cookie in the first request. metadata_fixture_.SetServerMetadata( *context, {{}, {{"x-goog-cbt-cookie-routing", "routing"}}}); @@ -2057,7 +2086,8 @@ TEST_F(DataConnectionTest, AsyncCheckAndMutateRowBigtableCookie) { }) .WillOnce([this](CompletionQueue&, std::shared_ptr const& context, - auto, v2::CheckAndMutateRowRequest const&, auto const&) { + auto, v2::CheckAndMutateRowRequest const&, + std::shared_ptr) { // Verify that the next request includes the bigtable cookie from above. auto headers = metadata_fixture_.GetMetadata(*context); EXPECT_THAT(headers, @@ -2103,7 +2133,8 @@ TEST_F(DataConnectionTest, SampleRowsSuccess) { auto mock = std::make_shared(); EXPECT_CALL(*mock, SampleRowKeys) .WillOnce([this](auto client_context, auto const&, - v2::SampleRowKeysRequest const& request, auto const&) { + v2::SampleRowKeysRequest const& request, + std::shared_ptr) { metadata_fixture_.SetServerMetadata(*client_context, {}); EXPECT_EQ(kAppProfile, request.app_profile_id()); EXPECT_EQ(kTableName, request.table_name()); @@ -2153,7 +2184,7 @@ TEST_F(DataConnectionTest, SampleRowsRetryResetsSamples) { auto mock = std::make_shared(); EXPECT_CALL(*mock, SampleRowKeys) .WillOnce([](auto, auto const&, v2::SampleRowKeysRequest const& request, - auto const&) { + std::shared_ptr) { EXPECT_EQ(kAppProfile, request.app_profile_id()); EXPECT_EQ(kTableName, request.table_name()); auto stream = std::make_unique(); @@ -2166,7 +2197,7 @@ TEST_F(DataConnectionTest, SampleRowsRetryResetsSamples) { return stream; }) .WillOnce([](auto, auto const&, v2::SampleRowKeysRequest const& request, - auto const&) { + std::shared_ptr) { EXPECT_EQ(kAppProfile, request.app_profile_id()); EXPECT_EQ(kTableName, request.table_name()); auto stream = std::make_unique(); @@ -2207,16 +2238,17 @@ TEST_F(DataConnectionTest, SampleRowsRetryExhausted) { auto mock = std::make_shared(); EXPECT_CALL(*mock, SampleRowKeys) .Times(kNumRetries + 1) - .WillRepeatedly([this](auto context, auto const&, - v2::SampleRowKeysRequest const& request, - auto const&) { - metadata_fixture_.SetServerMetadata(*context, {}); - EXPECT_EQ(kAppProfile, request.app_profile_id()); - EXPECT_EQ(kTableName, request.table_name()); - auto stream = std::make_unique(); - EXPECT_CALL(*stream, Read).WillOnce(Return(TransientError())); - return stream; - }); + .WillRepeatedly( + [this](auto context, auto const&, + v2::SampleRowKeysRequest const& request, + std::shared_ptr) { + metadata_fixture_.SetServerMetadata(*context, {}); + EXPECT_EQ(kAppProfile, request.app_profile_id()); + EXPECT_EQ(kTableName, request.table_name()); + auto stream = std::make_unique(); + EXPECT_CALL(*stream, Read).WillOnce(Return(TransientError())); + return stream; + }); auto mock_b = std::make_unique(); EXPECT_CALL(*mock_b, clone).WillOnce([]() { @@ -2255,7 +2287,8 @@ TEST_F(DataConnectionTest, SampleRowsPermanentError) { auto mock = std::make_shared(); EXPECT_CALL(*mock, SampleRowKeys) .WillOnce([this](auto client_context, auto const&, - v2::SampleRowKeysRequest const& request, auto const&) { + v2::SampleRowKeysRequest const& request, + std::shared_ptr) { metadata_fixture_.SetServerMetadata(*client_context, {}); EXPECT_EQ(kAppProfile, request.app_profile_id()); EXPECT_EQ(kTableName, request.table_name()); @@ -2278,7 +2311,8 @@ TEST_F(DataConnectionTest, SampleRowsBigtableCookie) { auto mock = std::make_shared(); EXPECT_CALL(*mock, SampleRowKeys) .WillOnce([this](auto context, auto const&, - v2::SampleRowKeysRequest const&, auto const&) { + v2::SampleRowKeysRequest const&, + std::shared_ptr) { // Return a bigtable cookie in the first request. metadata_fixture_.SetServerMetadata( *context, {{}, {{"x-goog-cbt-cookie-routing", "routing"}}}); @@ -2287,7 +2321,8 @@ TEST_F(DataConnectionTest, SampleRowsBigtableCookie) { return stream; }) .WillOnce([this](auto context, auto const&, - v2::SampleRowKeysRequest const&, auto const&) { + v2::SampleRowKeysRequest const&, + std::shared_ptr) { // Verify that the next request includes the bigtable cookie from // above. auto headers = metadata_fixture_.GetMetadata(*context); @@ -2316,20 +2351,20 @@ TEST_F(DataConnectionTest, SampleRowsBigtableCookie) { TEST_F(DataConnectionTest, SampleRowsRetryInfoHeeded) { auto mock = std::make_shared(); EXPECT_CALL(*mock, SampleRowKeys) - .WillOnce( - [](auto, auto const&, v2::SampleRowKeysRequest const&, auto const&) { - auto status = PermanentError(); - internal::SetRetryInfo(status, internal::RetryInfo{ms(0)}); - auto stream = std::make_unique(); - EXPECT_CALL(*stream, Read).WillOnce(Return(status)); - return stream; - }) - .WillOnce( - [](auto, auto const&, v2::SampleRowKeysRequest const&, auto const&) { - auto stream = std::make_unique(); - EXPECT_CALL(*stream, Read).WillOnce(Return(Status())); - return stream; - }); + .WillOnce([](auto, auto const&, v2::SampleRowKeysRequest const&, + std::shared_ptr) { + auto status = PermanentError(); + internal::SetRetryInfo(status, internal::RetryInfo{ms(0)}); + auto stream = std::make_unique(); + EXPECT_CALL(*stream, Read).WillOnce(Return(status)); + return stream; + }) + .WillOnce([](auto, auto const&, v2::SampleRowKeysRequest const&, + std::shared_ptr) { + auto stream = std::make_unique(); + EXPECT_CALL(*stream, Read).WillOnce(Return(Status())); + return stream; + }); auto conn = TestConnection(std::move(mock)); internal::OptionsSpan span( @@ -2341,14 +2376,14 @@ TEST_F(DataConnectionTest, SampleRowsRetryInfoHeeded) { TEST_F(DataConnectionTest, SampleRowsRetryInfoIgnored) { auto mock = std::make_shared(); EXPECT_CALL(*mock, SampleRowKeys) - .WillOnce( - [](auto, auto const&, v2::SampleRowKeysRequest const&, auto const&) { - auto status = PermanentError(); - internal::SetRetryInfo(status, internal::RetryInfo{ms(0)}); - auto stream = std::make_unique(); - EXPECT_CALL(*stream, Read).WillOnce(Return(status)); - return stream; - }); + .WillOnce([](auto, auto const&, v2::SampleRowKeysRequest const&, + std::shared_ptr) { + auto status = PermanentError(); + internal::SetRetryInfo(status, internal::RetryInfo{ms(0)}); + auto stream = std::make_unique(); + EXPECT_CALL(*stream, Read).WillOnce(Return(status)); + return stream; + }); auto conn = TestConnection(std::move(mock)); internal::OptionsSpan span( @@ -2363,7 +2398,8 @@ TEST_F(DataConnectionTest, AsyncSampleRows) { auto mock = std::make_shared(); EXPECT_CALL(*mock, AsyncSampleRowKeys) .WillOnce([](CompletionQueue const&, auto, auto, - v2::SampleRowKeysRequest const& request, auto const&) { + v2::SampleRowKeysRequest const& request, + std::shared_ptr) { EXPECT_EQ(kAppProfile, request.app_profile_id()); EXPECT_EQ(kTableName, request.table_name()); using ErrorStream = @@ -2411,7 +2447,7 @@ TEST_F(DataConnectionTest, ReadModifyWriteRowSuccess) { EXPECT_CALL(*mock, ReadModifyWriteRow) .WillOnce([&response](grpc::ClientContext&, Options const&, v2::ReadModifyWriteRowRequest const& request, - auto const&) { + bigtable_internal::OperationContext&) { EXPECT_EQ(kAppProfile, request.app_profile_id()); EXPECT_EQ(kTableName, request.table_name()); EXPECT_EQ("row", request.row_key()); @@ -2452,7 +2488,8 @@ TEST_F(DataConnectionTest, ReadModifyWriteRowPermanentError) { auto mock = std::make_shared(); EXPECT_CALL(*mock, ReadModifyWriteRow) .WillOnce([](grpc::ClientContext&, Options const&, - v2::ReadModifyWriteRowRequest const& request, auto const&) { + v2::ReadModifyWriteRowRequest const& request, + bigtable_internal::OperationContext&) { EXPECT_EQ(kAppProfile, request.app_profile_id()); EXPECT_EQ(kTableName, request.table_name()); EXPECT_EQ("row", request.row_key()); @@ -2489,7 +2526,8 @@ TEST_F(DataConnectionTest, ReadModifyWriteRowTransientErrorNotRetried) { auto mock = std::make_shared(); EXPECT_CALL(*mock, ReadModifyWriteRow) .WillOnce([](grpc::ClientContext&, Options const&, - v2::ReadModifyWriteRowRequest const& request, auto const&) { + v2::ReadModifyWriteRowRequest const& request, + bigtable_internal::OperationContext&) { EXPECT_EQ(kAppProfile, request.app_profile_id()); EXPECT_EQ(kTableName, request.table_name()); EXPECT_EQ("row", request.row_key()); @@ -2549,7 +2587,8 @@ TEST_F(DataConnectionTest, AsyncReadModifyWriteRowSuccess) { EXPECT_CALL(*mock, AsyncReadModifyWriteRow) .WillOnce([&response, this]( google::cloud::CompletionQueue&, auto client_context, auto, - v2::ReadModifyWriteRowRequest const& request, auto const&) { + v2::ReadModifyWriteRowRequest const& request, + std::shared_ptr) { metadata_fixture_.SetServerMetadata(*client_context, {}); EXPECT_EQ(kAppProfile, request.app_profile_id()); EXPECT_EQ(kTableName, request.table_name()); @@ -2596,7 +2635,7 @@ TEST_F(DataConnectionTest, AsyncReadModifyWriteRowPermanentError) { EXPECT_CALL(*mock, AsyncReadModifyWriteRow) .WillOnce([this](google::cloud::CompletionQueue&, auto client_context, auto, v2::ReadModifyWriteRowRequest const& request, - auto const&) { + std::shared_ptr) { metadata_fixture_.SetServerMetadata(*client_context, {}); EXPECT_EQ(kAppProfile, request.app_profile_id()); EXPECT_EQ(kTableName, request.table_name()); @@ -2625,7 +2664,7 @@ TEST_F(DataConnectionTest, AsyncReadModifyWriteRowTransientErrorNotRetried) { EXPECT_CALL(*mock, AsyncReadModifyWriteRow) .WillOnce([this](google::cloud::CompletionQueue&, auto client_context, auto, v2::ReadModifyWriteRowRequest const& request, - auto const&) { + std::shared_ptr) { metadata_fixture_.SetServerMetadata(*client_context, {}); EXPECT_EQ(kAppProfile, request.app_profile_id()); EXPECT_EQ(kTableName, request.table_name()); @@ -2664,7 +2703,8 @@ TEST_F(DataConnectionTest, AsyncReadRows) { auto mock = std::make_shared(); EXPECT_CALL(*mock, AsyncReadRows) .WillOnce([](CompletionQueue const&, auto, auto, - v2::ReadRowsRequest const& request, auto const&) { + v2::ReadRowsRequest const& request, + std::shared_ptr) { EXPECT_EQ(kAppProfile, request.app_profile_id()); EXPECT_EQ(kTableName, request.table_name()); EXPECT_EQ(42, request.rows_limit()); @@ -2694,7 +2734,8 @@ TEST_F(DataConnectionTest, AsyncReadRowsReverseScan) { auto mock = std::make_shared(); EXPECT_CALL(*mock, AsyncReadRows) .WillOnce([](CompletionQueue const&, auto, auto, - v2::ReadRowsRequest const& request, auto const&) { + v2::ReadRowsRequest const& request, + std::shared_ptr) { EXPECT_TRUE(request.reversed()); using ErrorStream = internal::AsyncStreamingReadRpcError; @@ -2735,7 +2776,8 @@ TEST_F(DataConnectionTest, AsyncReadRowEmpty) { auto mock = std::make_shared(); EXPECT_CALL(*mock, AsyncReadRows) .WillOnce([](CompletionQueue const&, auto, auto, - v2::ReadRowsRequest const& request, auto const&) { + v2::ReadRowsRequest const& request, + std::shared_ptr) { EXPECT_EQ(kAppProfile, request.app_profile_id()); EXPECT_EQ(kTableName, request.table_name()); EXPECT_EQ(1, request.rows_limit()); @@ -2781,7 +2823,8 @@ TEST_F(DataConnectionTest, AsyncReadRowSuccess) { auto mock = std::make_shared(); EXPECT_CALL(*mock, AsyncReadRows) .WillOnce([](CompletionQueue const&, auto, auto, - v2::ReadRowsRequest const& request, auto const&) { + v2::ReadRowsRequest const& request, + std::shared_ptr) { EXPECT_EQ(kAppProfile, request.app_profile_id()); EXPECT_EQ(kTableName, request.table_name()); EXPECT_EQ(1, request.rows_limit()); @@ -2838,7 +2881,8 @@ TEST_F(DataConnectionTest, AsyncReadRowFailure) { auto mock = std::make_shared(); EXPECT_CALL(*mock, AsyncReadRows) .WillOnce([](CompletionQueue const&, auto, auto, - v2::ReadRowsRequest const& request, auto const&) { + v2::ReadRowsRequest const& request, + std::shared_ptr) { EXPECT_EQ(kAppProfile, request.app_profile_id()); EXPECT_EQ(kTableName, request.table_name()); EXPECT_EQ(1, request.rows_limit()); @@ -2892,7 +2936,8 @@ TEST_F(DataConnectionTest, PrepareQuerySuccess) { kResultMetadataText, response.mutable_metadata())); EXPECT_CALL(*mock, PrepareQuery) .WillOnce([&](grpc::ClientContext&, Options const&, - v2::PrepareQueryRequest const& request, auto const&) { + v2::PrepareQueryRequest const& request, + bigtable_internal::OperationContext&) { EXPECT_EQ(kAppProfile, request.app_profile_id()); EXPECT_EQ("projects/the-project/instances/the-instance", request.instance_name()); @@ -2941,7 +2986,9 @@ TEST_F(DataConnectionTest, PrepareQueryPermanentError) { EXPECT_CALL(*mock, PrepareQuery) .WillOnce([](grpc::ClientContext&, Options const&, v2::PrepareQueryRequest const&, - auto const&) { return PermanentError(); }); + bigtable_internal::OperationContext&) { + return PermanentError(); + }); auto conn = TestConnection(std::move(mock), std::move(factory)); internal::OptionsSpan span(CallOptions()); @@ -2985,7 +3032,8 @@ TEST_F(DataConnectionTest, AsyncPrepareQuerySuccess) { kResultMetadataText, response.mutable_metadata())); EXPECT_CALL(*mock, AsyncPrepareQuery) .WillOnce([&](CompletionQueue const&, auto, auto, - v2::PrepareQueryRequest const& request, auto const&) { + v2::PrepareQueryRequest const& request, + std::shared_ptr) { EXPECT_EQ(kAppProfile, request.app_profile_id()); EXPECT_EQ("projects/the-project/instances/the-instance", request.instance_name()); @@ -3032,7 +3080,7 @@ TEST_F(DataConnectionTest, AsyncPrepareQueryPermanentError) { auto mock = std::make_shared(); EXPECT_CALL(*mock, AsyncPrepareQuery) .WillOnce([](CompletionQueue&, auto, auto, v2::PrepareQueryRequest const&, - auto const&) { + std::shared_ptr) { return make_ready_future>( PermanentError()); }); @@ -3097,21 +3145,21 @@ TEST_F(DataConnectionTest, ExecuteQuerySuccessWithTransientErrors) { EXPECT_CALL(*mock, ExecuteQuery) .WillOnce([&](auto, auto const&, google::bigtable::v2::ExecuteQueryRequest const&, - auto const&) { + std::shared_ptr) { auto error_stream = std::make_unique(); EXPECT_CALL(*error_stream, Read).WillOnce(Return(TransientError())); return error_stream; }) .WillOnce([&](auto, auto const&, google::bigtable::v2::ExecuteQueryRequest const&, - auto const&) { + std::shared_ptr) { auto error_stream = std::make_unique(); EXPECT_CALL(*error_stream, Read).WillOnce(Return(TransientError())); return error_stream; }) .WillOnce([&](auto, auto const&, google::bigtable::v2::ExecuteQueryRequest const& request, - auto const&) { + std::shared_ptr) { EXPECT_EQ(request.app_profile_id(), kAppProfile); EXPECT_EQ(request.instance_name(), "projects/test-project/instances/test-instance"); @@ -3199,7 +3247,8 @@ TEST_F(DataConnectionTest, ExecuteQueryFailure) { std::move(refresh_fn)); EXPECT_CALL(*mock, ExecuteQuery) - .WillOnce([&](auto, auto const&, auto const&, auto const&) { + .WillOnce([&](auto, auto const&, auto const&, + std::shared_ptr) { auto stream = std::make_unique(); EXPECT_CALL(*stream, Read).WillOnce(Return(PermanentError())); return stream; @@ -3254,11 +3303,13 @@ TEST_F(DataConnectionTest, ExecuteQueryOperationRetryExhausted) { EXPECT_CALL(*mock, ExecuteQuery) .Times(3) - .WillRepeatedly([&](auto, auto const&, auto const&, auto const&) { - auto stream = std::make_unique(); - EXPECT_CALL(*stream, Read).WillOnce(Return(TransientError())); - return stream; - }); + .WillRepeatedly( + [&](auto, auto const&, auto const&, + std::shared_ptr) { + auto stream = std::make_unique(); + EXPECT_CALL(*stream, Read).WillOnce(Return(TransientError())); + return stream; + }); auto conn = TestConnection(std::move(mock), std::move(factory)); internal::OptionsSpan span(CallOptions()); @@ -3354,7 +3405,7 @@ TEST_F(DataConnectionTest, ExecuteQuerySuccessWithQueryPlanRefresh) { EXPECT_CALL(*mock, ExecuteQuery) .WillOnce([&](auto, auto const&, google::bigtable::v2::ExecuteQueryRequest const& request, - auto const&) { + std::shared_ptr) { EXPECT_EQ(request.prepared_query(), "test-pq-id-initial"); auto error_stream = std::make_unique(); EXPECT_CALL(*error_stream, Read).WillOnce(Return(QueryPlanError())); @@ -3362,7 +3413,7 @@ TEST_F(DataConnectionTest, ExecuteQuerySuccessWithQueryPlanRefresh) { }) .WillOnce([&](auto, auto const&, google::bigtable::v2::ExecuteQueryRequest const& request, - auto const&) { + std::shared_ptr) { EXPECT_EQ(request.app_profile_id(), kAppProfile); EXPECT_EQ(request.instance_name(), "projects/test-project/instances/test-instance"); @@ -3518,22 +3569,25 @@ TEST_F(DataConnectionTest, PrepareAndExecuteQuerySuccessWithQueryPlanRefresh) { EXPECT_CALL(*mock, PrepareQuery) .WillOnce([&](grpc::ClientContext&, Options const&, PrepareQueryRequest const&, - auto const&) { return initial_pq_response; }); + bigtable_internal::OperationContext&) { + return initial_pq_response; + }); EXPECT_CALL(*mock, AsyncPrepareQuery) .WillOnce([&](CompletionQueue const&, auto, auto, - v2::PrepareQueryRequest const&, auto const&) { + v2::PrepareQueryRequest const&, + std::shared_ptr) { return make_ready_future(make_status_or(refresh_pq_response)); }); EXPECT_CALL(*mock, ExecuteQuery) .WillOnce([&](auto, auto const&, ExecuteQueryRequest const& request, - auto const&) { + std::shared_ptr) { EXPECT_EQ(request.prepared_query(), "test-pq-id-initial"); auto error_stream = std::make_unique(); EXPECT_CALL(*error_stream, Read).WillOnce(Return(QueryPlanError())); return error_stream; }) .WillOnce([&](auto, auto const&, ExecuteQueryRequest const& request, - auto const&) { + std::shared_ptr) { EXPECT_EQ(request.app_profile_id(), kAppProfile); EXPECT_EQ(request.instance_name(), "projects/test-project/instances/test-instance"); @@ -3687,23 +3741,25 @@ TEST_F(DataConnectionTest, auto mock = std::make_shared(); EXPECT_CALL(*mock, AsyncPrepareQuery) .WillOnce([&](CompletionQueue const&, auto, auto, - PrepareQueryRequest const&, auto const&) { + PrepareQueryRequest const&, + std::shared_ptr) { return make_ready_future(make_status_or(initial_pq_response)); }) .WillOnce([&](CompletionQueue const&, auto, auto, - PrepareQueryRequest const&, auto const&) { + PrepareQueryRequest const&, + std::shared_ptr) { return make_ready_future(make_status_or(refresh_pq_response)); }); EXPECT_CALL(*mock, ExecuteQuery) .WillOnce([&](auto, auto const&, ExecuteQueryRequest const& request, - auto const&) { + std::shared_ptr) { EXPECT_EQ(request.prepared_query(), "test-pq-id-initial"); auto error_stream = std::make_unique(); EXPECT_CALL(*error_stream, Read).WillOnce(Return(QueryPlanError())); return error_stream; }) .WillOnce([&](auto, auto const&, ExecuteQueryRequest const& request, - auto const&) { + std::shared_ptr) { EXPECT_EQ(request.app_profile_id(), kAppProfile); EXPECT_EQ(request.instance_name(), "projects/test-project/instances/test-instance"); @@ -3796,7 +3852,7 @@ TEST_F(DataConnectionTest, ExecuteQueryFailureWithSchemaChange) { EXPECT_CALL(*mock, ExecuteQuery) .WillOnce([&](auto, auto const&, google::bigtable::v2::ExecuteQueryRequest const& request, - auto const&) { + std::shared_ptr) { EXPECT_EQ(request.app_profile_id(), kAppProfile); EXPECT_EQ(request.instance_name(), "projects/test-project/instances/test-instance"); @@ -3860,9 +3916,10 @@ TEST_F(DataConnectionTest, PrepareQueryFailsOnInvalidType) { std::chrono::system_clock::now() + std::chrono::seconds(3600)); EXPECT_CALL(*mock, PrepareQuery) - .WillOnce([&](grpc::ClientContext&, Options const&, - v2::PrepareQueryRequest const&, - auto const&) { return pq_response; }); + .WillOnce( + [&](grpc::ClientContext&, Options const&, + v2::PrepareQueryRequest const&, + bigtable_internal::OperationContext&) { return pq_response; }); auto conn = TestConnection(std::move(mock)); internal::OptionsSpan span(CallOptions()); @@ -3906,7 +3963,8 @@ TEST_F(DataConnectionTest, AsyncPrepareQueryFailsOnInvalidType) { EXPECT_CALL(*mock, AsyncPrepareQuery) .WillOnce([&](CompletionQueue const&, auto, auto, - v2::PrepareQueryRequest const&, auto const&) { + v2::PrepareQueryRequest const&, + std::shared_ptr) { return make_ready_future(make_status_or(pq_response)); }); diff --git a/google/cloud/bigtable/internal/default_row_reader_test.cc b/google/cloud/bigtable/internal/default_row_reader_test.cc index e6aa0dff3f426..6d42a78a4077d 100644 --- a/google/cloud/bigtable/internal/default_row_reader_test.cc +++ b/google/cloud/bigtable/internal/default_row_reader_test.cc @@ -13,6 +13,7 @@ // limitations under the License. #include "google/cloud/bigtable/internal/default_row_reader.h" +#include "google/cloud/bigtable/internal/operation_context.h" #include "google/cloud/bigtable/row_reader.h" #include "google/cloud/bigtable/testing/mock_bigtable_stub.h" #include "google/cloud/bigtable/testing/mock_policies.h" diff --git a/google/cloud/bigtable/internal/dynamic_channel_pool.h b/google/cloud/bigtable/internal/dynamic_channel_pool.h index 7bec7a78bf8a7..951ecc597cebe 100644 --- a/google/cloud/bigtable/internal/dynamic_channel_pool.h +++ b/google/cloud/bigtable/internal/dynamic_channel_pool.h @@ -18,6 +18,8 @@ #include "google/cloud/bigtable/instance_resource.h" #include "google/cloud/bigtable/internal/channel_usage.h" #include "google/cloud/bigtable/internal/connection_refresh_state.h" +#include "google/cloud/bigtable/internal/defaults.h" +#include "google/cloud/bigtable/internal/metrics.h" #include "google/cloud/bigtable/internal/stub_manager.h" #include "google/cloud/bigtable/options.h" #include "google/cloud/completion_queue.h" @@ -42,6 +44,12 @@ GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN // remove ChannelUsage objects per the configuration present in the // DynamicChannelPoolSizingPolicyOption. // +template +struct SelectedChannel { + std::shared_ptr> channel; + int outstanding_rpcs; +}; + template class DynamicChannelPool : public std::enable_shared_from_this> { @@ -58,11 +66,12 @@ class DynamicChannelPool std::vector>> initial_channels, std::shared_ptr refresh_state, StubFactoryFn stub_factory_fn, - bigtable::experimental::DynamicChannelPoolSizingPolicy sizing_policy) { + bigtable::experimental::DynamicChannelPoolSizingPolicy sizing_policy, + TransportType transport_type = TransportType::kCloudPath) { auto pool = std::shared_ptr(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; } @@ -97,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 @@ -111,7 +122,7 @@ class DynamicChannelPool // // If there are no healthy channels in channels_, create a new channel and // use that one. Also call ScheduleAddChannels to replenish channels_. - std::shared_ptr> GetChannelRandomTwoLeastUsed() { + SelectedChannel GetChannelRandomTwoLeastUsed() { std::scoped_lock lk(mu_); CheckPoolChannelHealth(lk); @@ -136,12 +147,13 @@ class DynamicChannelPool // This is the most common case so we try it first. if (d.channel_1_rpcs.ok() && d.channel_2_rpcs.ok()) { - return *d.channel_1_rpcs < *d.channel_2_rpcs ? *d.channel_1_iter - : *d.channel_2_iter; + return *d.channel_1_rpcs < *d.channel_2_rpcs + ? SelectedChannel{*d.channel_1_iter, *d.channel_1_rpcs} + : SelectedChannel{*d.channel_2_iter, *d.channel_2_rpcs}; } if (d.iterators.size() == 1 && d.channel_1_rpcs.ok()) { // Pool contains exactly 1 good channel. - return *d.channel_1_iter; + return SelectedChannel{*d.channel_1_iter, *d.channel_1_rpcs}; } if (d.iterators.empty()) { // Pool is empty, create a channel immediately and return it. While the @@ -150,7 +162,8 @@ class DynamicChannelPool channels_.push_back(stub_factory_fn_(next_channel_id_++, instance_name_, StubManager::Priming::kNoPriming) .value()); - return channels_.front(); + auto sor = channels_.front()->instant_outstanding_rpcs(); + return SelectedChannel{channels_.front(), sor ? *sor : 0}; } return HandleBadChannels(lk, d); } @@ -163,14 +176,16 @@ class DynamicChannelPool std::vector>> initial_wrapped_channels, std::shared_ptr refresh_state, StubFactoryFn stub_factory_fn, - bigtable::experimental::DynamicChannelPoolSizingPolicy sizing_policy) + bigtable::experimental::DynamicChannelPoolSizingPolicy sizing_policy, + TransportType transport_type = TransportType::kCloudPath) : 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(channels_.size())) { + next_channel_id_(static_cast(channels_.size())), + transport_type_(transport_type) { std::scoped_lock lk(mu_); SetSizeDecreaseCooldownTimer(lk); } @@ -204,8 +219,8 @@ class DynamicChannelPool // We have one or more bad channels. Spending time finding a good channel // will be cheaper than trying to use a bad channel in the long run. - std::shared_ptr> HandleBadChannels( - std::scoped_lock const& lk, ChannelSelectionData& d) { + SelectedChannel HandleBadChannels(std::scoped_lock const& lk, + ChannelSelectionData& d) { std::vector bad_channel_iters; if (d.shuffle_iter != d.iterators.end()) ++d.shuffle_iter; ChannelSelectionData::FindGoodChannel(d.iterators, d.channel_1_iter, @@ -216,14 +231,22 @@ class DynamicChannelPool bad_channel_iters); std::shared_ptr> channel; + int outstanding_rpcs = 0; if (d.channel_1_rpcs.ok() || d.channel_2_rpcs.ok()) { if (d.channel_1_rpcs.ok() && d.channel_2_rpcs.ok()) { - channel = *d.channel_1_rpcs < *d.channel_2_rpcs ? *d.channel_1_iter - : *d.channel_2_iter; + if (*d.channel_1_rpcs < *d.channel_2_rpcs) { + channel = *d.channel_1_iter; + outstanding_rpcs = *d.channel_1_rpcs; + } else { + channel = *d.channel_2_iter; + outstanding_rpcs = *d.channel_2_rpcs; + } } else if (d.channel_1_rpcs.ok()) { channel = *d.channel_1_iter; + outstanding_rpcs = *d.channel_1_rpcs; } else if (d.channel_2_rpcs.ok()) { channel = *d.channel_2_iter; + outstanding_rpcs = *d.channel_2_rpcs; } // Wait until we no longer need valid iterators to call EvictBadChannels. EvictBadChannels(lk, bad_channel_iters); @@ -240,9 +263,11 @@ class DynamicChannelPool .value()); std::swap(channels_.front(), channels_.back()); channel = channels_.front(); + auto sor = channel->instant_outstanding_rpcs(); + outstanding_rpcs = sor ? *sor : 0; } ScheduleRemoveChannels(lk); - return channel; + return SelectedChannel{std::move(channel), outstanding_rpcs}; } // Determines the number of channels to add and reserves the channel ids to @@ -433,6 +458,7 @@ class DynamicChannelPool future> pool_size_decrease_cooldown_timer_; std::uint32_t next_channel_id_; + TransportType const transport_type_ = TransportType::kCloudPath; }; GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END diff --git a/google/cloud/bigtable/internal/dynamic_channel_pool_test.cc b/google/cloud/bigtable/internal/dynamic_channel_pool_test.cc index 06bd3dd93337b..e3f3be2b56bfe 100644 --- a/google/cloud/bigtable/internal/dynamic_channel_pool_test.cc +++ b/google/cloud/bigtable/internal/dynamic_channel_pool_test.cc @@ -1,4 +1,3 @@ -#include "google/cloud/bigtable/internal/operation_context.h" // Copyright 2026 Google LLC // // Licensed under the Apache License, Version 2.0 (the "License"); @@ -14,6 +13,7 @@ // limitations under the License. #include "google/cloud/bigtable/internal/dynamic_channel_pool.h" +#include "google/cloud/bigtable/internal/operation_context.h" #include "google/cloud/bigtable/testing/mock_bigtable_stub.h" #include "google/cloud/internal/make_status.h" #include "google/cloud/testing_util/fake_completion_queue_impl.h" @@ -36,7 +36,7 @@ class DynamicChannelPoolTestWrapper { using ChannelSelectionData = DynamicChannelPool::ChannelSelectionData; - std::shared_ptr> HandleBadChannels( + SelectedChannel HandleBadChannels( std::scoped_lock const& lk, DynamicChannelPool::ChannelSelectionData& d) { return pool_->HandleBadChannels(lk, d); @@ -153,14 +153,14 @@ TEST_F(DynamicChannelPoolTest, SelectLeastUsedFromTwoChannels) { EXPECT_CALL(*mock_stub_0, CheckAndMutateRow) .WillOnce([](grpc::ClientContext&, Options const&, google::bigtable::v2::CheckAndMutateRowRequest const&, - auto&) { + OperationContext&) { google::bigtable::v2::CheckAndMutateRowResponse response; response.set_predicate_matched(true); return response; }); auto mock_stub_1 = std::make_shared(); EXPECT_CALL(*mock_stub_1, CheckAndMutateRow).Times(0); - int initial_rpc_count = 0; + int initial_rpc_count = 5; channels.push_back(std::make_shared>( std::move(mock_stub_0), initial_rpc_count++)); channels.push_back(std::make_shared>( @@ -181,11 +181,14 @@ TEST_F(DynamicChannelPoolTest, SelectLeastUsedFromTwoChannels) { auto pool = DynamicChannelPool::Create( instance_name, CompletionQueue(mock_cq_impl_), channels, refresh_state, stub_factory_fn, sizing_policy); - auto selected_stub = pool->GetChannelRandomTwoLeastUsed(); + auto selected = pool->GetChannelRandomTwoLeastUsed(); + EXPECT_THAT(selected.outstanding_rpcs, Eq(5)); grpc::ClientContext context; - OperationContext op_ctx; - auto response = - selected_stub->AcquireStub()->CheckAndMutateRow(context, {}, {}, op_ctx); + OperationContext default_oc; + auto response = selected.channel->AcquireStub()->CheckAndMutateRow( + context, {}, {}, default_oc); + EXPECT_THAT(selected.channel->instant_outstanding_rpcs(), + testing_util::IsOkAndHolds(Eq(6))); ASSERT_STATUS_OK(response); EXPECT_TRUE(response->predicate_matched()); @@ -219,7 +222,7 @@ TEST_F(DynamicChannelPoolTest, OneInitialChannel) { EXPECT_CALL(*mock_stub_0, CheckAndMutateRow) .WillOnce([](grpc::ClientContext&, Options const&, google::bigtable::v2::CheckAndMutateRowRequest const&, - auto&) { + OperationContext&) { google::bigtable::v2::CheckAndMutateRowResponse response; response.set_predicate_matched(true); return response; @@ -247,11 +250,12 @@ TEST_F(DynamicChannelPoolTest, OneInitialChannel) { stub_factory_fn.AsStdFunction(), sizing_policy); EXPECT_THAT(pool->size(), Eq(1)); - auto selected_stub = pool->GetChannelRandomTwoLeastUsed(); + auto selected = pool->GetChannelRandomTwoLeastUsed(); + EXPECT_THAT(selected.outstanding_rpcs, Eq(0)); grpc::ClientContext context; - OperationContext op_ctx; - auto response = selected_stub->AcquireStub()->CheckAndMutateRow(context, {}, - {}, op_ctx); + OperationContext default_oc; + auto response = selected.channel->AcquireStub()->CheckAndMutateRow( + context, {}, {}, default_oc); ASSERT_STATUS_OK(response); EXPECT_TRUE(response->predicate_matched()); } @@ -276,7 +280,7 @@ TEST_F(DynamicChannelPoolTest, EmptyInitialPool) { EXPECT_CALL(*mock_stub, CheckAndMutateRow) .WillOnce([](grpc::ClientContext&, Options const&, google::bigtable::v2::CheckAndMutateRowRequest const&, - auto&) { + OperationContext&) { google::bigtable::v2::CheckAndMutateRowResponse response; response.set_predicate_matched(true); return response; @@ -310,11 +314,12 @@ TEST_F(DynamicChannelPoolTest, EmptyInitialPool) { EXPECT_THAT(*pool, ::testing::IsEmpty()); - auto selected_stub = pool->GetChannelRandomTwoLeastUsed(); + auto selected = pool->GetChannelRandomTwoLeastUsed(); + EXPECT_THAT(selected.outstanding_rpcs, Eq(0)); grpc::ClientContext context; - OperationContext op_ctx; - auto response = selected_stub->AcquireStub()->CheckAndMutateRow(context, {}, - {}, op_ctx); + OperationContext default_oc; + auto response = selected.channel->AcquireStub()->CheckAndMutateRow( + context, {}, {}, default_oc); ASSERT_STATUS_OK(response); EXPECT_TRUE(response->predicate_matched()); @@ -751,7 +756,7 @@ TEST_F(DynamicChannelPoolTest, HandleBadChannelsTwoChannelsOneBad) { EXPECT_CALL(*mock_stub, CheckAndMutateRow) .WillOnce([](grpc::ClientContext&, Options const&, google::bigtable::v2::CheckAndMutateRowRequest const&, - auto&) { + OperationContext&) { google::bigtable::v2::CheckAndMutateRowResponse response; response.set_predicate_matched(true); return response; @@ -778,7 +783,7 @@ TEST_F(DynamicChannelPoolTest, HandleBadChannelsTwoChannelsOneBad) { DynamicChannelPoolTestWrapper wrapper(pool); auto draining_channels = wrapper.SetDrainingChannels({}); - std::shared_ptr> selected_stub; + SelectedChannel selected_stub; { auto lock = wrapper.CreateLock(); selected_stub = wrapper.HandleBadChannels(lock, data); @@ -786,9 +791,9 @@ TEST_F(DynamicChannelPoolTest, HandleBadChannelsTwoChannelsOneBad) { EXPECT_THAT(draining_channels, IsEmpty()); grpc::ClientContext context; - OperationContext op_ctx; - auto response = - selected_stub->AcquireStub()->CheckAndMutateRow(context, {}, {}, op_ctx); + OperationContext default_oc; + auto response = selected_stub.channel->AcquireStub()->CheckAndMutateRow( + context, {}, {}, default_oc); ASSERT_STATUS_OK(response); EXPECT_TRUE(response->predicate_matched()); EXPECT_THAT(pool->size(), Eq(1)); @@ -832,7 +837,7 @@ TEST_F(DynamicChannelPoolTest, HandleBadChannelsTwoChannelsOtherOneBad) { EXPECT_CALL(*mock_stub, CheckAndMutateRow) .WillOnce([](grpc::ClientContext&, Options const&, google::bigtable::v2::CheckAndMutateRowRequest const&, - auto&) { + OperationContext&) { google::bigtable::v2::CheckAndMutateRowResponse response; response.set_predicate_matched(true); return response; @@ -861,7 +866,7 @@ TEST_F(DynamicChannelPoolTest, HandleBadChannelsTwoChannelsOtherOneBad) { DynamicChannelPoolTestWrapper wrapper(pool); auto draining_channels = wrapper.SetDrainingChannels({}); - std::shared_ptr> selected_stub; + SelectedChannel selected_stub; { auto lock = wrapper.CreateLock(); selected_stub = wrapper.HandleBadChannels(lock, data); @@ -869,9 +874,9 @@ TEST_F(DynamicChannelPoolTest, HandleBadChannelsTwoChannelsOtherOneBad) { EXPECT_THAT(draining_channels, IsEmpty()); grpc::ClientContext context; - OperationContext op_ctx; - auto response = - selected_stub->AcquireStub()->CheckAndMutateRow(context, {}, {}, op_ctx); + OperationContext default_oc; + auto response = selected_stub.channel->AcquireStub()->CheckAndMutateRow( + context, {}, {}, default_oc); ASSERT_STATUS_OK(response); EXPECT_TRUE(response->predicate_matched()); EXPECT_THAT(pool->size(), Eq(1)); @@ -915,7 +920,7 @@ TEST_F(DynamicChannelPoolTest, HandleBadChannelsThreeChannelsOneBad) { EXPECT_CALL(*mock_stub_0, CheckAndMutateRow) .WillOnce([](grpc::ClientContext&, Options const&, google::bigtable::v2::CheckAndMutateRowRequest const&, - auto&) { + OperationContext&) { google::bigtable::v2::CheckAndMutateRowResponse response; response.set_predicate_matched(true); return response; @@ -950,7 +955,7 @@ TEST_F(DynamicChannelPoolTest, HandleBadChannelsThreeChannelsOneBad) { DynamicChannelPoolTestWrapper wrapper(pool); auto draining_channels = wrapper.SetDrainingChannels({}); - std::shared_ptr> selected_stub; + SelectedChannel selected_stub; { auto lock = wrapper.CreateLock(); selected_stub = wrapper.HandleBadChannels(lock, data); @@ -958,9 +963,9 @@ TEST_F(DynamicChannelPoolTest, HandleBadChannelsThreeChannelsOneBad) { EXPECT_THAT(draining_channels, IsEmpty()); grpc::ClientContext context; - OperationContext op_ctx; - auto response = - selected_stub->AcquireStub()->CheckAndMutateRow(context, {}, {}, op_ctx); + OperationContext default_oc; + auto response = selected_stub.channel->AcquireStub()->CheckAndMutateRow( + context, {}, {}, default_oc); ASSERT_STATUS_OK(response); EXPECT_TRUE(response->predicate_matched()); EXPECT_THAT(pool->size(), Eq(2)); @@ -1004,7 +1009,7 @@ TEST_F(DynamicChannelPoolTest, HandleBadChannelsAllChannelsBad) { EXPECT_CALL(*mock_stub, CheckAndMutateRow) .WillOnce([](grpc::ClientContext&, Options const&, google::bigtable::v2::CheckAndMutateRowRequest const&, - auto&) { + OperationContext&) { google::bigtable::v2::CheckAndMutateRowResponse response; response.set_predicate_matched(true); return response; @@ -1049,7 +1054,7 @@ TEST_F(DynamicChannelPoolTest, HandleBadChannelsAllChannelsBad) { DynamicChannelPoolTestWrapper wrapper(pool); auto draining_channels = wrapper.SetDrainingChannels({}); - std::shared_ptr> selected_stub; + SelectedChannel selected_stub; { auto lock = wrapper.CreateLock(); selected_stub = wrapper.HandleBadChannels(lock, data); @@ -1057,9 +1062,9 @@ TEST_F(DynamicChannelPoolTest, HandleBadChannelsAllChannelsBad) { EXPECT_THAT(draining_channels, IsEmpty()); grpc::ClientContext context; - OperationContext op_ctx; - auto response = - selected_stub->AcquireStub()->CheckAndMutateRow(context, {}, {}, op_ctx); + OperationContext default_oc; + auto response = selected_stub.channel->AcquireStub()->CheckAndMutateRow( + context, {}, {}, default_oc); ASSERT_STATUS_OK(response); EXPECT_TRUE(response->predicate_matched()); EXPECT_THAT(pool->size(), Eq(1)); diff --git a/google/cloud/bigtable/internal/grpc_metrics_exporter.cc b/google/cloud/bigtable/internal/grpc_metrics_exporter.cc index 214f3da6e6211..9f1723a8906b5 100644 --- a/google/cloud/bigtable/internal/grpc_metrics_exporter.cc +++ b/google/cloud/bigtable/internal/grpc_metrics_exporter.cc @@ -25,9 +25,7 @@ #include "google/cloud/log.h" #include "absl/strings/match.h" #include "absl/strings/str_cat.h" -#include "absl/strings/str_join.h" -#include "absl/strings/str_replace.h" -#include "absl/strings/str_split.h" +#include "absl/strings/string_view.h" #include #include #include @@ -54,7 +52,6 @@ #include #include #include -#include #include #include #endif // GOOGLE_CLOUD_CPP_BIGTABLE_WITH_GRPC_OTEL_METRICS @@ -200,6 +197,53 @@ std::shared_ptr MakeGrpcMeterProvider( std::move(provider)); } +ClientResourceLabels MakeClientResourceLabels( + std::string project_id, std::string instance, std::string app_profile, + Options const& options, std::string const& client_uid, + opentelemetry::sdk::resource::Resource const& detected_resource) { + namespace sc = ::opentelemetry::semconv; + auto const& detected_attributes = detected_resource.GetAttributes(); + auto by_name = [&](std::string const& name, std::string default_value = {}) { + auto const l = detected_attributes.find(name); + if (l == detected_attributes.end() || + !opentelemetry::nostd::holds_alternative(l->second)) { + return default_value; + } + return opentelemetry::nostd::get(l->second); + }; + + if (project_id.empty() && + options.has()) { + auto const& instances = + options.get(); + if (!instances.empty()) { + project_id = instances[0].project_id(); + } + } + if (project_id.empty()) { + project_id = by_name(sc::cloud::kCloudAccountId); + } + + auto client_project = by_name(sc::cloud::kCloudAccountId); + if (client_project.empty()) { + client_project = project_id; + } + + 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, + by_name(sc::cloud::kCloudRegion, "global")); + labels.cloud_platform = by_name(sc::cloud::kCloudPlatform, "unknown"); + labels.host_id = by_name("faas.id", by_name(sc::host::kHostId, "unknown")); + labels.hostname = by_name(sc::host::kHostName); + return labels; +} + MonitoredResourceResult MakeMonitoredResource( opentelemetry::sdk::metrics::PointDataAttributes const& pda, opentelemetry::sdk::resource::Resource const& detected_resource, @@ -286,7 +330,7 @@ MonitoredResourceResult MakeMonitoredResource( labels["host_name"] = hostname; } - return MonitoredResourceResult{std::move(project_id), std::move(resource)}; + return MonitoredResourceResult{project_id, std::move(resource)}; } void EnableGrpcMetrics( @@ -307,42 +351,14 @@ void EnableGrpcMetrics( std::move(res.resource)); }; - std::set excluded_labels{ - "project_id", - "instance", - "app_profile", - "client_project", - "cloud_platform", - "region", - "client_region", - "host_id", - "host_name", - "client_name", - "uuid", - "service_name", - "service_namespace", - "service_instance_id", - "service.name", - "service.namespace", - "service.instance.id", - }; + std::set excluded_labels{"project_id", "instance"}; auto resource_filter_fn = [excluded_labels = std::move(excluded_labels)](std::string const& key) { return internal::Contains(excluded_labels, key); }; - auto constexpr kBigtableMetricNamePath = - "bigtable.googleapis.com/internal/client/"; - auto exporter_options = options; - // Internal metrics must use ServiceTimeSeries to avoid permission issues. - exporter_options.set(true); - exporter_options.set([=](auto const& name) { - return absl::StrCat(kBigtableMetricNamePath, - absl::StrReplaceAll(name, {{".", "/"}})); - }); - auto exporter = otel_internal::MakeMonitoringExporter( - dynamic_resource_fn, resource_filter_fn, conn, exporter_options); + dynamic_resource_fn, resource_filter_fn, conn, options); auto reader_options = opentelemetry::sdk::metrics::PeriodicExportingMetricReaderOptions{}; @@ -357,26 +373,18 @@ void EnableGrpcMetrics( auto provider = MakeGrpcMeterProvider(std::move(exporter), std::move(reader_options)); - auto const metrics = std::vector{ - std::string_view{"grpc.client.attempt.duration"}, - std::string_view{"grpc.lb.rls.default_target_picks"}, - std::string_view{"grpc.lb.rls.target_picks"}, - std::string_view{"grpc.lb.rls.failed_picks"}, - std::string_view{"grpc.xds_client.server_failure"}, - std::string_view{"grpc.xds_client.resource_updates_invalid"}, - std::string_view{"grpc.subchannel.disconnections"}, - std::string_view{"grpc.subchannel.connection_attempts_succeeded"}, - std::string_view{"grpc.subchannel.connection_attempts_failed"}, - std::string_view{"grpc.subchannel.open_connections"}, - }; - - auto const disable_metrics = std::vector{ - std::string_view{ - "grpc.client.attempt.sent_total_compressed_message_size"}, - std::string_view{ - "grpc.client.attempt.rcvd_total_compressed_message_size"}, + auto const metrics = std::vector{ + absl::string_view{"grpc.client.attempt.duration"}, + absl::string_view{"grpc.lb.rls.default_target_picks"}, + absl::string_view{"grpc.lb.rls.target_picks"}, + absl::string_view{"grpc.lb.rls.failed_picks"}, + absl::string_view{"grpc.xds_client.server_failure"}, + absl::string_view{"grpc.xds_client.resource_updates_invalid"}, + absl::string_view{"grpc.subchannel.disconnections"}, + absl::string_view{"grpc.subchannel.connection_attempts_succeeded"}, + absl::string_view{"grpc.subchannel.connection_attempts_failed"}, + absl::string_view{"grpc.subchannel.open_connections"}, }; - auto scope_filter = [authority = std::move(authority)]( grpc::OpenTelemetryPluginBuilder::ChannelScope const& scope) { @@ -390,8 +398,7 @@ void EnableGrpcMetrics( grpc::OpenTelemetryPluginBuilder() .SetMeterProvider(provider) .EnableMetrics(metrics) - .DisableMetrics(disable_metrics) - .SetGenericMethodAttributeFilter([](std::string_view target) { + .SetGenericMethodAttributeFilter([](absl::string_view target) { return absl::StartsWith(target, "google.bigtable.v2"); }) .SetChannelScopeFilter(std::move(scope_filter)) diff --git a/google/cloud/bigtable/internal/grpc_metrics_exporter.h b/google/cloud/bigtable/internal/grpc_metrics_exporter.h index 8ab268b7c2fb1..14662cdde3c01 100644 --- a/google/cloud/bigtable/internal/grpc_metrics_exporter.h +++ b/google/cloud/bigtable/internal/grpc_metrics_exporter.h @@ -23,6 +23,7 @@ #include #ifdef GOOGLE_CLOUD_CPP_BIGTABLE_WITH_GRPC_OTEL_METRICS +#include "google/cloud/bigtable/internal/metrics.h" #include "google/cloud/monitoring/v3/metric_connection.h" #include "google/api/monitored_resource.pb.h" #include @@ -75,6 +76,11 @@ struct MonitoredResourceResult { google::api::MonitoredResource resource; }; +ClientResourceLabels MakeClientResourceLabels( + std::string project_id, std::string instance, std::string app_profile, + Options const& options, std::string const& client_uid, + opentelemetry::sdk::resource::Resource const& detected_resource); + MonitoredResourceResult MakeMonitoredResource( opentelemetry::sdk::metrics::PointDataAttributes const& pda, opentelemetry::sdk::resource::Resource const& detected_resource, diff --git a/google/cloud/bigtable/internal/grpc_metrics_exporter_test.cc b/google/cloud/bigtable/internal/grpc_metrics_exporter_test.cc index b4452d66c751e..d87c6ffac0e41 100644 --- a/google/cloud/bigtable/internal/grpc_metrics_exporter_test.cc +++ b/google/cloud/bigtable/internal/grpc_metrics_exporter_test.cc @@ -213,6 +213,28 @@ TEST(GrpcMetricsExporterTest, MakeMonitoredResource) { EXPECT_THAT(labels.at("host_id"), Eq("unknown")); } +TEST(GrpcMetricsExporterTest, MakeClientResourceLabels) { + Options options; + options.set("test-app-profile"); + std::string const client_uid = "test-client-uid"; + + auto labels = MakeClientResourceLabels( + "test-project", "test-instance", "test-app-profile", options, client_uid, + opentelemetry::sdk::resource::Resource::Create({})); + + EXPECT_THAT(labels.project_id, Eq("test-project")); + EXPECT_THAT(labels.instance, Eq("test-instance")); + EXPECT_THAT(labels.app_profile, Eq("test-app-profile")); + EXPECT_THAT(labels.client_name, + Eq("cpp.Bigtable/" + bigtable::version_string())); + EXPECT_THAT(labels.client_uid, Eq("test-client-uid")); + EXPECT_THAT(labels.client_project, Eq("test-project")); + EXPECT_THAT(labels.location, Eq("global")); + EXPECT_THAT(labels.cloud_platform, Eq("unknown")); + EXPECT_THAT(labels.host_id, Eq("unknown")); + EXPECT_THAT(labels.hostname, IsEmpty()); +} + TEST(GrpcMetricsExporterTest, MakeMonitoredResourceMissingAttributes) { Options options; std::string const client_uid = "test-client-uid"; diff --git a/google/cloud/bigtable/internal/metrics.cc b/google/cloud/bigtable/internal/metrics.cc index 54faaeccab986..a1a15010e7d0d 100644 --- a/google/cloud/bigtable/internal/metrics.cc +++ b/google/cloud/bigtable/internal/metrics.cc @@ -89,6 +89,76 @@ LabelMap IntoLabelMap(TableResourceLabels const& r, TableDataLabels const& d, return labels; } +std::string ToString(ChannelPoolLbPolicy policy) { + switch (policy) { + case ChannelPoolLbPolicy::kRoundRobin: + return "ROUND_ROBIN"; + case ChannelPoolLbPolicy::kRandomTwoLeastUsed: + return "RANDOM_TWO_LEAST_USED"; + } + return "ROUND_ROBIN"; +} + +std::string ToString(TransportType type) { + switch (type) { + case TransportType::kCloudPath: + return "CloudPath"; + case TransportType::kDirectPath: + return "DirectPath"; + } + return "CloudPath"; +} + +std::string ToString(RpcType streaming) { + switch (streaming) { + case RpcType::kUnary: + return "false"; + case RpcType::kStreaming: + return "true"; + } + return "false"; +} + +LabelMap IntoLabelMap(ClientResourceLabels const& r, + ClientOutstandingRpcLabels const& d, + std::set const& filtered_data_labels) { + LabelMap labels = { + {"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}}; + + std::map data = {{ + {"transport_type", ToString(d.transport_type)}, + {"channel_pool_lb_policy", ToString(d.channel_pool_lb_policy)}, + {"streaming", ToString(d.streaming)}, + }}; + + if (filtered_data_labels.empty()) { + labels.insert(data.begin(), data.end()); + return labels; + } + + struct Compare { + bool operator()(std::pair const& a, + std::string const& b) { + return a.first < b; + } + + bool operator()(std::string const& a, + std::pair const& b) { + return a < b.first; + } + }; + + std::set_difference(data.begin(), data.end(), filtered_data_labels.begin(), + filtered_data_labels.end(), + std::inserter(labels, labels.begin()), Compare()); + + return labels; +} + bool HasServerTiming(grpc::ClientContext const& client_context) { auto const& initial_metadata = client_context.GetServerInitialMetadata(); auto it = initial_metadata.find("server-timing"); @@ -545,6 +615,37 @@ std::unique_ptr ApplicationBlockingLatency::clone( return m; } +OutstandingRpcs::OutstandingRpcs( + std::string const& instrumentation_scope, + opentelemetry::nostd::shared_ptr< + opentelemetry::metrics::MeterProvider> const& provider) + : outstanding_rpcs_( + provider + ->GetMeter(instrumentation_scope, + kMeterInstrumentationScopeVersion) + ->CreateDoubleHistogram( + "connection_pool/outstanding_rpcs", + "Instantaneous count of outstanding RPCs on the selected " + "channel.", + "1")) {} + +void OutstandingRpcs::StubSelection( + opentelemetry::context::Context const& context, + StubSelectionParams const& p) { + ClientOutstandingRpcLabels data_labels{p.transport_type, + p.channel_pool_lb_policy, p.streaming}; + outstanding_rpcs_->Record(static_cast(p.outstanding_rpcs), + IntoLabelMap(resource_labels_, data_labels), + context); +} + +std::unique_ptr OutstandingRpcs::clone( + ClientResourceLabels const& resource_labels) const { + auto m = std::make_unique(*this); + m->resource_labels_ = resource_labels; + return m; +} + GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END } // namespace bigtable_internal } // namespace cloud diff --git a/google/cloud/bigtable/internal/metrics.h b/google/cloud/bigtable/internal/metrics.h index 6a4a08999ecdf..81d5266c2da8c 100644 --- a/google/cloud/bigtable/internal/metrics.h +++ b/google/cloud/bigtable/internal/metrics.h @@ -44,6 +44,25 @@ struct TableResourceLabels { std::string zone; }; +enum class ChannelPoolLbPolicy { + kRoundRobin, + kRandomTwoLeastUsed, +}; + +enum class TransportType { + kCloudPath, + kDirectPath, +}; + +enum class RpcType { + kUnary, + kStreaming, +}; + +std::string ToString(ChannelPoolLbPolicy policy); +std::string ToString(TransportType type); +std::string ToString(RpcType streaming); + struct TableDataLabels { std::string method; std::string streaming; @@ -53,6 +72,32 @@ struct TableDataLabels { std::string status; }; +struct ClientResourceLabels { + std::string project_id; + std::string instance; + std::string app_profile; + std::string client_name; + std::string client_uid; + std::string client_project; + std::string location; + std::string cloud_platform; + std::string host_id; + std::string hostname; +}; + +struct ClientOutstandingRpcLabels { + TransportType transport_type; + ChannelPoolLbPolicy channel_pool_lb_policy; + RpcType streaming; +}; + +struct StubSelectionParams { + std::int64_t outstanding_rpcs; + ChannelPoolLbPolicy channel_pool_lb_policy; + TransportType transport_type; + RpcType streaming; +}; + // Labels populated from the peer info metadata. struct PeerInfoLabels { std::string transport_type; @@ -67,6 +112,10 @@ LabelMap IntoLabelMap( std::set const& filtered_data_labels = {}, std::optional const& peer_info_labels = std::nullopt); +LabelMap IntoLabelMap(ClientResourceLabels const& r, + ClientOutstandingRpcLabels const& d, + std::set const& filtered_data_labels = {}); + bool HasServerTiming(grpc::ClientContext const& client_context); bool IsConnectivityError(google::cloud::Status const& status, grpc::ClientContext const& client_context); @@ -109,6 +158,8 @@ class Metric { using LatencyDuration = std::chrono::duration; virtual ~Metric() = 0; + virtual void StubSelection(opentelemetry::context::Context const&, + StubSelectionParams const&) {} virtual void PreCall(opentelemetry::context::Context const&, PreCallParams const&) {} virtual void PostCall(opentelemetry::context::Context const&, @@ -120,7 +171,12 @@ class Metric { virtual void ElementDelivery(opentelemetry::context::Context const&, ElementDeliveryParams const&) {} virtual std::unique_ptr clone(TableResourceLabels const&, - TableDataLabels const&) const = 0; + TableDataLabels const&) const { + return nullptr; + } + virtual std::unique_ptr clone(ClientResourceLabels const&) const { + return nullptr; + } }; class OperationLatency : public Metric { @@ -320,6 +376,22 @@ class ConnectivityErrorCount : public Metric { opentelemetry::metrics::Counter> connectivity_error_count_; }; + +class OutstandingRpcs : public Metric { + public: + OutstandingRpcs(std::string const& instrumentation_scope, + opentelemetry::nostd::shared_ptr< + opentelemetry::metrics::MeterProvider> const& provider); + void StubSelection(opentelemetry::context::Context const&, + StubSelectionParams const& p) override; + std::unique_ptr clone( + ClientResourceLabels const& resource_labels) const override; + + private: + ClientResourceLabels resource_labels_; + opentelemetry::nostd::shared_ptr> + outstanding_rpcs_; +}; GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END } // namespace bigtable_internal } // namespace cloud diff --git a/google/cloud/bigtable/internal/metrics_test.cc b/google/cloud/bigtable/internal/metrics_test.cc index 71c18cb3290ba..1de431637d1df 100644 --- a/google/cloud/bigtable/internal/metrics_test.cc +++ b/google/cloud/bigtable/internal/metrics_test.cc @@ -2480,6 +2480,95 @@ TEST(ApplicationBlockingLatency, StreamingData) { clone->PostCall(otel_context, client_context, {clock->Now(), Status{}}); clone->OnDone(otel_context, {clock->Now(), Status{}}); } + +TEST(MetricsTest, IntoLabelMapClient) { + ClientResourceLabels resource{"p-1", "i-1", "app-1", "client-1", + "uid-1", "cp-1", "loc-1", "cloud-1", + "host-1", "hostname-1"}; + ClientOutstandingRpcLabels data{TransportType::kDirectPath, + ChannelPoolLbPolicy::kRandomTwoLeastUsed, + RpcType::kStreaming}; + auto actual = IntoLabelMap(resource, data); + EXPECT_THAT(actual, + UnorderedElementsAre( + Pair("project_id", "p-1"), Pair("instance", "i-1"), + Pair("app_profile", "app-1"), Pair("client_name", "client-1"), + Pair("client_uid", "uid-1"), Pair("client_project", "cp-1"), + Pair("location", "loc-1"), Pair("cloud_platform", "cloud-1"), + Pair("host_id", "host-1"), Pair("hostname", "hostname-1"), + Pair("transport_type", "DirectPath"), + Pair("channel_pool_lb_policy", "RANDOM_TWO_LEAST_USED"), + Pair("streaming", "true"))); +} + +TEST(MetricsTest, OutstandingRpcsMetric) { + auto mock_histogram = std::make_unique>(); + EXPECT_CALL( + *mock_histogram, + Record(A(), A(), + A())) + .WillOnce([](double value, + opentelemetry::common::KeyValueIterable const& attrs, + opentelemetry::context::Context const&) { + EXPECT_THAT(value, Eq(42.0)); + EXPECT_THAT( + MakeAttributesMap(attrs), + UnorderedElementsAre( + Pair("project_id", "my-project"), + Pair("instance", "my-instance"), + Pair("app_profile", "my-app-profile"), + Pair("client_name", "my-client-name"), + Pair("client_uid", "my-uid"), + Pair("client_project", "my-client-project"), + Pair("location", "us-east1"), Pair("cloud_platform", "gcp"), + Pair("host_id", "my-host"), Pair("hostname", "my-hostname"), + Pair("transport_type", "DirectPath"), + Pair("channel_pool_lb_policy", "RANDOM_TWO_LEAST_USED"), + Pair("streaming", "false"))); + }); + + opentelemetry::nostd::shared_ptr mock_meter = + std::make_shared(); + EXPECT_CALL(*mock_meter, CreateDoubleHistogram) + .WillOnce([mock = std::move(mock_histogram)]( + opentelemetry::nostd::string_view name, + opentelemetry::nostd::string_view, + opentelemetry::nostd::string_view) mutable { + EXPECT_THAT(name, Eq("connection_pool/outstanding_rpcs")); + return std::move(mock); + }); + + opentelemetry::nostd::shared_ptr mock_provider = + std::make_shared(); + EXPECT_CALL(*mock_provider, GetMeter) +#if OPENTELEMETRY_ABI_VERSION_NO >= 2 + .WillOnce([&](opentelemetry::nostd::string_view scope, + opentelemetry::nostd::string_view scope_version, + opentelemetry::nostd::string_view, + opentelemetry::common::KeyValueIterable const*) mutable { +#else + .WillOnce([&](opentelemetry::nostd::string_view scope, + opentelemetry::nostd::string_view scope_version, + opentelemetry::nostd::string_view) mutable { +#endif + EXPECT_THAT(scope, Eq("my-instrument-scope")); + EXPECT_THAT(scope_version, Eq("v1")); + return mock_meter; + }); + + OutstandingRpcs outstanding_rpcs("my-instrument-scope", mock_provider); + ClientResourceLabels resource_labels{ + "my-project", "my-instance", "my-app-profile", "my-client-name", + "my-uid", "my-client-project", "us-east1", "gcp", + "my-host", "my-hostname"}; + auto clone = outstanding_rpcs.clone(resource_labels); + + auto otel_context = opentelemetry::context::RuntimeContext::GetCurrent(); + clone->StubSelection( + otel_context, + StubSelectionParams{42, ChannelPoolLbPolicy::kRandomTwoLeastUsed, + TransportType::kDirectPath, RpcType::kUnary}); +} } // namespace GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END } // namespace bigtable_internal diff --git a/google/cloud/bigtable/internal/operation_context.cc b/google/cloud/bigtable/internal/operation_context.cc index cb6c31cb229c0..39c8cdfcb8848 100644 --- a/google/cloud/bigtable/internal/operation_context.cc +++ b/google/cloud/bigtable/internal/operation_context.cc @@ -30,11 +30,18 @@ namespace { std::vector> CloneMetrics( TableResourceLabels const& resource_labels, TableDataLabels const& data_labels, + ClientResourceLabels const& client_resource_labels, std::vector> const& metrics) { std::vector> v; v.reserve(metrics.size()); for (auto const& m : metrics) { - v.emplace_back(m->clone(resource_labels, data_labels)); + auto clone = m->clone(resource_labels, data_labels); + if (!clone) { + clone = m->clone(client_resource_labels); + } + if (clone) { + v.emplace_back(std::move(clone)); + } } return v; } @@ -59,9 +66,26 @@ OperationContext::OperationContext( TableDataLabels const& data_labels, std::vector> const& metrics, std::shared_ptr clock) - : cloned_metrics_(CloneMetrics(resource_labels, data_labels, metrics)), + : OperationContext(resource_labels, data_labels, ClientResourceLabels{}, + metrics, std::move(clock)) {} + +OperationContext::OperationContext( + TableResourceLabels const& resource_labels, + TableDataLabels const& data_labels, + ClientResourceLabels const& client_resource_labels, + std::vector> const& metrics, + std::shared_ptr clock) + : cloned_metrics_(CloneMetrics(resource_labels, data_labels, + client_resource_labels, metrics)), clock_(std::move(clock)) {} +void OperationContext::StubSelection(StubSelectionParams const& params) { + auto otel_context = opentelemetry::context::RuntimeContext::GetCurrent(); + for (auto& m : cloned_metrics_) { + m->StubSelection(otel_context, params); + } +} + void OperationContext::PreCall(grpc::ClientContext& client_context) { auto otel_context = opentelemetry::context::RuntimeContext::GetCurrent(); auto attempt_start = clock_->Now(); diff --git a/google/cloud/bigtable/internal/operation_context.h b/google/cloud/bigtable/internal/operation_context.h index 8956771ff6736..65e078e5700cd 100644 --- a/google/cloud/bigtable/internal/operation_context.h +++ b/google/cloud/bigtable/internal/operation_context.h @@ -30,6 +30,8 @@ namespace cloud { namespace bigtable_internal { GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN +struct ClientResourceLabels; +struct StubSelectionParams; struct TableDataLabels; struct TableResourceLabels; class Metric; @@ -68,7 +70,14 @@ class OperationContext { TableDataLabels const& data_labels, std::vector> const& metrics, std::shared_ptr clock); + OperationContext(TableResourceLabels const& resource_labels, + TableDataLabels const& data_labels, + ClientResourceLabels const& client_resource_labels, + std::vector> const& metrics, + std::shared_ptr clock); + // Called when a stub is selected from a channel pool. + void StubSelection(StubSelectionParams const& params); // Called before each RPC attempt. void PreCall(grpc::ClientContext& client_context); // Called after receiving RPC attempt response. diff --git a/google/cloud/bigtable/internal/operation_context_factory.cc b/google/cloud/bigtable/internal/operation_context_factory.cc index f85f6673e1960..282ec4987508b 100644 --- a/google/cloud/bigtable/internal/operation_context_factory.cc +++ b/google/cloud/bigtable/internal/operation_context_factory.cc @@ -149,6 +149,17 @@ std::shared_ptr SimpleOperationContextFactory::ExecuteQuery( #ifdef GOOGLE_CLOUD_CPP_BIGTABLE_WITH_OTEL_METRICS +namespace { +ClientResourceLabels MakeClientLabels( + ClientResourceLabels base, TableResourceLabels const& resource_labels, + std::string const& app_profile) { + base.project_id = resource_labels.project_id; + base.instance = resource_labels.instance; + base.app_profile = app_profile; + return base; +} +} // namespace + MetricsOperationContextFactory::MetricsOperationContextFactory( std::string client_uid, std::shared_ptr conn, @@ -212,25 +223,56 @@ void MetricsOperationContextFactory::InitializeProvider( auto dynamic_resource_fn = [=](opentelemetry::sdk::metrics::PointDataAttributes const& pda) { + auto const& attributes = pda.attributes.GetAttributes(); + auto get_attr = [&](std::string const& key) { + auto it = attributes.find(key); + if (it == attributes.end() || + !opentelemetry::nostd::holds_alternative( + it->second)) { + return std::string{}; + } + return opentelemetry::nostd::get(it->second); + }; + + if (attributes.find(kTableLabel) != attributes.end()) { + google::api::MonitoredResource resource; + resource.set_type(kResourceType); + auto& labels = *resource.mutable_labels(); + labels[kProjectLabel] = get_attr(kProjectLabel); + labels[kInstanceLabel] = get_attr(kInstanceLabel); + labels[kTableLabel] = get_attr(kTableLabel); + labels[kClusterLabel] = get_attr(kClusterLabel); + labels[kZoneLabel] = get_attr(kZoneLabel); + return std::make_pair(labels[kProjectLabel], resource); + } + google::api::MonitoredResource resource; - resource.set_type(kResourceType); + resource.set_type("bigtable.googleapis.com/Client"); auto& labels = *resource.mutable_labels(); - auto const& attributes = pda.attributes.GetAttributes(); - labels[kProjectLabel] = - std::get(attributes.find(kProjectLabel)->second); - labels[kInstanceLabel] = - std::get(attributes.find(kInstanceLabel)->second); - labels[kTableLabel] = - std::get(attributes.find(kTableLabel)->second); - labels[kClusterLabel] = - std::get(attributes.find(kClusterLabel)->second); - labels[kZoneLabel] = - std::get(attributes.find(kZoneLabel)->second); - return std::make_pair(labels[kProjectLabel], resource); + labels["project_id"] = get_attr("project_id"); + labels["instance"] = get_attr("instance"); + labels["app_profile"] = get_attr("app_profile"); + labels["client_name"] = get_attr("client_name"); + labels["uuid"] = get_attr("client_uid"); + auto client_project = get_attr("client_project"); + if (!client_project.empty()) { + labels["client_project"] = std::move(client_project); + } + labels["location"] = get_attr("location"); + labels["cloud_platform"] = get_attr("cloud_platform"); + labels["host_id"] = get_attr("host_id"); + auto hostname = get_attr("hostname"); + if (!hostname.empty()) { + labels["hostname"] = std::move(hostname); + } + return std::make_pair(labels["project_id"], resource); }; - std::set s{kProjectLabel, kInstanceLabel, kTableLabel, - kClusterLabel, kZoneLabel}; + std::set s{kProjectLabel, kInstanceLabel, kTableLabel, + kClusterLabel, kZoneLabel, "app_profile", + "client_name", "client_uid", "uuid", + "client_project", "location", "cloud_platform", + "host_id", "hostname"}; auto resource_filter_fn = [resource_labels = std::move(s)](std::string const& key) { return internal::Contains(resource_labels, key); @@ -289,6 +331,7 @@ std::shared_ptr MetricsOperationContextFactory::ReadRow( std::make_shared(kRpc, provider_)); v.emplace_back(std::make_shared(kRpc, provider_)); v.emplace_back(std::make_shared(kRpc, provider_)); + v.emplace_back(std::make_shared(kRpc, provider_)); swap(read_row_metrics_.metrics, v); }); @@ -300,8 +343,10 @@ std::shared_ptr MetricsOperationContextFactory::ReadRow( app_profile, "" /*=status*/}; - return std::make_shared(resource_labels, data_labels, - read_row_metrics_.metrics, clock_); + return std::make_shared( + resource_labels, data_labels, + MakeClientLabels(client_resource_labels_, resource_labels, app_profile), + read_row_metrics_.metrics, clock_); } std::shared_ptr MetricsOperationContextFactory::ReadRows( @@ -318,6 +363,7 @@ std::shared_ptr MetricsOperationContextFactory::ReadRows( std::make_shared(kRpc, provider_)); v.emplace_back(std::make_shared(kRpc, provider_)); v.emplace_back(std::make_shared(kRpc, provider_)); + v.emplace_back(std::make_shared(kRpc, provider_)); swap(read_rows_metrics_.metrics, v); }); @@ -329,8 +375,10 @@ std::shared_ptr MetricsOperationContextFactory::ReadRows( app_profile, "" /*=status*/}; - return std::make_shared(resource_labels, data_labels, - read_rows_metrics_.metrics, clock_); + return std::make_shared( + resource_labels, data_labels, + MakeClientLabels(client_resource_labels_, resource_labels, app_profile), + read_rows_metrics_.metrics, clock_); } std::shared_ptr MetricsOperationContextFactory::MutateRow( @@ -346,6 +394,7 @@ std::shared_ptr MetricsOperationContextFactory::MutateRow( std::make_shared(kRpc, provider_)); v.emplace_back(std::make_shared(kRpc, provider_)); v.emplace_back(std::make_shared(kRpc, provider_)); + v.emplace_back(std::make_shared(kRpc, provider_)); swap(mutate_row_metrics_.metrics, v); }); @@ -358,7 +407,9 @@ std::shared_ptr MetricsOperationContextFactory::MutateRow( "" /*=status*/}; return std::make_shared( - resource_labels, data_labels, mutate_row_metrics_.metrics, clock_); + resource_labels, data_labels, + MakeClientLabels(client_resource_labels_, resource_labels, app_profile), + mutate_row_metrics_.metrics, clock_); } std::shared_ptr MetricsOperationContextFactory::MutateRows( @@ -374,6 +425,7 @@ std::shared_ptr MetricsOperationContextFactory::MutateRows( std::make_shared(kRpc, provider_)); v.emplace_back(std::make_shared(kRpc, provider_)); v.emplace_back(std::make_shared(kRpc, provider_)); + v.emplace_back(std::make_shared(kRpc, provider_)); swap(mutate_rows_metrics_.metrics, v); }); @@ -386,7 +438,9 @@ std::shared_ptr MetricsOperationContextFactory::MutateRows( "" /*=status*/}; return std::make_shared( - resource_labels, data_labels, mutate_rows_metrics_.metrics, clock_); + resource_labels, data_labels, + MakeClientLabels(client_resource_labels_, resource_labels, app_profile), + mutate_rows_metrics_.metrics, clock_); } std::shared_ptr @@ -403,6 +457,7 @@ MetricsOperationContextFactory::CheckAndMutateRow( std::make_shared(kRpc, provider_)); v.emplace_back(std::make_shared(kRpc, provider_)); v.emplace_back(std::make_shared(kRpc, provider_)); + v.emplace_back(std::make_shared(kRpc, provider_)); swap(check_and_mutate_row_metrics_.metrics, v); }); @@ -415,8 +470,9 @@ MetricsOperationContextFactory::CheckAndMutateRow( "" /*=status*/}; return std::make_shared( - resource_labels, data_labels, check_and_mutate_row_metrics_.metrics, - clock_); + resource_labels, data_labels, + MakeClientLabels(client_resource_labels_, resource_labels, app_profile), + check_and_mutate_row_metrics_.metrics, clock_); } std::shared_ptr MetricsOperationContextFactory::SampleRowKeys( @@ -432,6 +488,7 @@ std::shared_ptr MetricsOperationContextFactory::SampleRowKeys( std::make_shared(kRpc, provider_)); v.emplace_back(std::make_shared(kRpc, provider_)); v.emplace_back(std::make_shared(kRpc, provider_)); + v.emplace_back(std::make_shared(kRpc, provider_)); swap(sample_row_keys_metrics_.metrics, v); }); @@ -444,7 +501,9 @@ std::shared_ptr MetricsOperationContextFactory::SampleRowKeys( "" /*=status*/}; return std::make_shared( - resource_labels, data_labels, sample_row_keys_metrics_.metrics, clock_); + resource_labels, data_labels, + MakeClientLabels(client_resource_labels_, resource_labels, app_profile), + sample_row_keys_metrics_.metrics, clock_); } std::shared_ptr @@ -461,6 +520,7 @@ MetricsOperationContextFactory::ReadModifyWriteRow( std::make_shared(kRpc, provider_)); v.emplace_back(std::make_shared(kRpc, provider_)); v.emplace_back(std::make_shared(kRpc, provider_)); + v.emplace_back(std::make_shared(kRpc, provider_)); swap(read_modify_write_row_metrics_.metrics, v); }); @@ -473,8 +533,9 @@ MetricsOperationContextFactory::ReadModifyWriteRow( "" /*=status*/}; return std::make_shared( - resource_labels, data_labels, read_modify_write_row_metrics_.metrics, - clock_); + resource_labels, data_labels, + MakeClientLabels(client_resource_labels_, resource_labels, app_profile), + read_modify_write_row_metrics_.metrics, clock_); } std::shared_ptr MetricsOperationContextFactory::PrepareQuery( @@ -488,6 +549,7 @@ std::shared_ptr MetricsOperationContextFactory::PrepareQuery( v.emplace_back(std::make_shared(kRpc, provider_)); v.emplace_back(std::make_shared(kRpc, provider_)); v.emplace_back(std::make_shared(kRpc, provider_)); + v.emplace_back(std::make_shared(kRpc, provider_)); swap(prepare_query_metrics_.metrics, v); }); @@ -500,7 +562,9 @@ std::shared_ptr MetricsOperationContextFactory::PrepareQuery( "" /*=status*/}; return std::make_shared( - resource_labels, data_labels, prepare_query_metrics_.metrics, clock_); + resource_labels, data_labels, + MakeClientLabels(client_resource_labels_, resource_labels, app_profile), + prepare_query_metrics_.metrics, clock_); } std::shared_ptr MetricsOperationContextFactory::ExecuteQuery( @@ -517,6 +581,7 @@ std::shared_ptr MetricsOperationContextFactory::ExecuteQuery( std::make_shared(kRpc, provider_)); v.emplace_back(std::make_shared(kRpc, provider_)); v.emplace_back(std::make_shared(kRpc, provider_)); + v.emplace_back(std::make_shared(kRpc, provider_)); swap(execute_query_metrics_.metrics, v); }); @@ -529,7 +594,9 @@ std::shared_ptr MetricsOperationContextFactory::ExecuteQuery( "" /*=status*/}; return std::make_shared( - resource_labels, data_labels, execute_query_metrics_.metrics, clock_); + resource_labels, data_labels, + MakeClientLabels(client_resource_labels_, resource_labels, app_profile), + execute_query_metrics_.metrics, clock_); } #endif // GOOGLE_CLOUD_CPP_BIGTABLE_WITH_OTEL_METRICS diff --git a/google/cloud/bigtable/internal/operation_context_factory.h b/google/cloud/bigtable/internal/operation_context_factory.h index c8d77b4f7c6b0..a57ab04f4c2c4 100644 --- a/google/cloud/bigtable/internal/operation_context_factory.h +++ b/google/cloud/bigtable/internal/operation_context_factory.h @@ -18,6 +18,7 @@ #include "google/cloud/bigtable/internal/operation_context.h" #include "google/cloud/bigtable/version.h" #ifdef GOOGLE_CLOUD_CPP_BIGTABLE_WITH_OTEL_METRICS +#include "google/cloud/bigtable/internal/metrics.h" #include "google/cloud/monitoring/v3/metric_connection.h" #include "absl/base/call_once.h" #include @@ -140,6 +141,7 @@ class MetricsOperationContextFactory : public OperationContextFactory { std::string client_uid_; std::shared_ptr clock_; std::shared_ptr provider_; + ClientResourceLabels client_resource_labels_; // These vectors are initialized exactly once and the initialization is // delayed until the first time the corresponding method is called. diff --git a/google/cloud/bigtable/internal/operation_context_factory_test.cc b/google/cloud/bigtable/internal/operation_context_factory_test.cc index d00ecf8240045..31e73b0d997f1 100644 --- a/google/cloud/bigtable/internal/operation_context_factory_test.cc +++ b/google/cloud/bigtable/internal/operation_context_factory_test.cc @@ -26,6 +26,7 @@ namespace { using ::testing::Eq; using ::testing::IsEmpty; +using ::testing::NotNull; class MockMetric : public Metric { public: @@ -208,6 +209,20 @@ TEST(MetricsOperationContextFactoryTest, ExecuteQuery) { factory.ExecuteQuery(instance_full_name, app_profile); } +TEST(MetricsOperationContextFactoryTest, IncludesOutstandingRpcs) { + std::string app_profile = "my-app-profile"; + std::string table_full_name = + "projects/my-project/instances/my-instance/tables/my-table"; + MetricsOperationContextFactory factory( + "test-uid", + std::shared_ptr(nullptr)); + auto operation_context = factory.ReadRow(table_full_name, app_profile); + EXPECT_THAT(operation_context, NotNull()); + operation_context->StubSelection( + StubSelectionParams{10, ChannelPoolLbPolicy::kRandomTwoLeastUsed, + TransportType::kDirectPath, RpcType::kUnary}); +} + } // namespace GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END } // namespace bigtable_internal diff --git a/google/cloud/bigtable/internal/stub_manager_test.cc b/google/cloud/bigtable/internal/stub_manager_test.cc index d8b2e7cdc750b..299862b1d9f51 100644 --- a/google/cloud/bigtable/internal/stub_manager_test.cc +++ b/google/cloud/bigtable/internal/stub_manager_test.cc @@ -1,4 +1,3 @@ -#include "google/cloud/bigtable/internal/operation_context.h" // Copyright 2026 Google LLC // // Licensed under the Apache License, Version 2.0 (the "License"); @@ -13,8 +12,9 @@ // See the License for the specific language governing permissions and // limitations under the License. -#include "google/cloud/bigtable/instance_resource.h" #include "google/cloud/bigtable/internal/stub_manager.h" +#include "google/cloud/bigtable/instance_resource.h" +#include "google/cloud/bigtable/internal/operation_context.h" #include "google/cloud/bigtable/table_resource.h" #include "google/cloud/bigtable/testing/mock_bigtable_stub.h" #include "google/cloud/testing_util/scoped_log.h" @@ -43,7 +43,7 @@ TEST(StubManagerTest, NoAffinity) { EXPECT_CALL(*mock, MutateRow) .WillOnce([&](grpc::ClientContext&, Options const&, google::bigtable::v2::MutateRowRequest const& request, - auto const&) { + OperationContext&) { EXPECT_THAT(request.table_name(), Eq(expected_table_name)); return google::bigtable::v2::MutateRowResponse{}; }); @@ -54,8 +54,8 @@ TEST(StubManagerTest, NoAffinity) { grpc::ClientContext context; google::bigtable::v2::MutateRowRequest request; request.set_table_name(expected_table_name); - OperationContext op_ctx; - auto result = stub->MutateRow(context, {}, request, op_ctx); + OperationContext default_oc; + auto result = stub->MutateRow(context, {}, request, default_oc); EXPECT_THAT(result, IsOk()); } @@ -69,7 +69,7 @@ TEST(StubManagerTest, AffinityToExistingInstance) { .WillOnce([instance_name = instance.FullName()]( grpc::ClientContext&, Options const&, google::bigtable::v2::MutateRowRequest const& request, - auto const&) { + OperationContext&) { EXPECT_THAT(request.table_name(), StartsWith(instance_name)); return google::bigtable::v2::MutateRowResponse{}; }); @@ -89,8 +89,8 @@ TEST(StubManagerTest, AffinityToExistingInstance) { grpc::ClientContext context; google::bigtable::v2::MutateRowRequest request; request.set_table_name(expected_table_name); - OperationContext op_ctx; - auto result = stub->MutateRow(context, {}, request, op_ctx); + OperationContext default_oc; + auto result = stub->MutateRow(context, {}, request, default_oc); EXPECT_THAT(result, IsOk()); } @@ -111,7 +111,7 @@ TEST(StubManagerTest, AffinityToMissingInstance) { .WillOnce([instance_name = std::string{instance_name}]( grpc::ClientContext&, Options const&, google::bigtable::v2::MutateRowRequest const& request, - auto const&) { + OperationContext&) { EXPECT_THAT(request.table_name(), StartsWith(instance_name)); return google::bigtable::v2::MutateRowResponse{}; }); @@ -127,8 +127,8 @@ TEST(StubManagerTest, AffinityToMissingInstance) { grpc::ClientContext context; google::bigtable::v2::MutateRowRequest request; request.set_table_name(expected_table_name); - OperationContext op_ctx; - auto result = stub->MutateRow(context, {}, request, op_ctx); + OperationContext default_oc; + auto result = stub->MutateRow(context, {}, request, default_oc); EXPECT_THAT(result, IsOk()); EXPECT_THAT(log.ExtractLines(), Contains(HasSubstr( diff --git a/google/cloud/bigtable/tests/CMakeLists.txt b/google/cloud/bigtable/tests/CMakeLists.txt index 103b258e94141..bab997045a665 100644 --- a/google/cloud/bigtable/tests/CMakeLists.txt +++ b/google/cloud/bigtable/tests/CMakeLists.txt @@ -33,11 +33,6 @@ include(CreateBazelConfig) export_list_to_bazel("bigtable_client_integration_tests.bzl" "bigtable_client_integration_tests" YEAR "2018") -if (NOT TARGET otel_collector) - add_subdirectory("${PROJECT_SOURCE_DIR}/ci/otel_collector" - "${CMAKE_BINARY_DIR}/ci/otel_collector") -endif () - foreach (fname ${bigtable_client_integration_tests}) google_cloud_cpp_add_executable(target "bigtable" "${fname}") target_link_libraries( @@ -55,7 +50,7 @@ foreach (fname ${bigtable_client_integration_tests}) gRPC::grpc++ gRPC::grpc protobuf::libprotobuf) - if (TARGET otel_collector) + if ("${fname}" STREQUAL "observability_integration_test.cc") target_link_libraries(${target} PRIVATE otel_collector) endif () add_test(NAME ${target} COMMAND ${target}) diff --git a/google/cloud/bigtable/tests/observability_integration_test.cc b/google/cloud/bigtable/tests/observability_integration_test.cc index b60341363c1ff..db01ffc45337a 100644 --- a/google/cloud/bigtable/tests/observability_integration_test.cc +++ b/google/cloud/bigtable/tests/observability_integration_test.cc @@ -12,74 +12,77 @@ // See the License for the specific language governing permissions and // limitations under the License. -#ifndef _WIN32 - +#include "google/cloud/internal/disable_deprecation_warnings.inc" #include "google/cloud/bigtable/options.h" #include "google/cloud/bigtable/testing/table_integration_test.h" #include "google/cloud/credentials.h" -#include "google/cloud/grpc_options.h" #include "google/cloud/internal/getenv.h" #include "google/cloud/testing_util/scoped_environment.h" #include "google/cloud/testing_util/status_matchers.h" -#include "absl/strings/match.h" #include "absl/strings/str_cat.h" #include "absl/strings/str_split.h" #include "ci/otel_collector/otel_collector.h" -#include #include -#include -#include #include #include -#include -#include -#include -#include namespace google { namespace cloud { namespace bigtable { -namespace testing { +GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN namespace { -bool IsDirectPathReachable() { - int s = socket(AF_INET6, SOCK_STREAM, 0); - if (s < 0) return false; - int flags = fcntl(s, F_GETFL, 0); - (void)fcntl(s, F_SETFL, flags | O_NONBLOCK); - sockaddr_in6 addr{}; - addr.sin6_family = AF_INET6; - addr.sin6_port = htons(443); - inet_pton(AF_INET6, "2607:f8b0:4001:c2f::5f", &addr.sin6_addr); - int res = connect(s, reinterpret_cast(&addr), sizeof(addr)); - if (res < 0 && errno == EINPROGRESS) { - pollfd pfd{s, POLLOUT, 0}; - res = poll(&pfd, 1, 1000); - if (res > 0) { - int err = 0; - socklen_t len = sizeof(err); - (void)getsockopt(s, SOL_SOCKET, SO_ERROR, &err, &len); - res = (err == 0) ? 0 : -1; - } else { - res = -1; - } - } - close(s); - return res == 0; -} - using ::google::cloud::bigtable::testing::TableTestEnvironment; using ::google::cloud::testing_util::ScopedEnvironment; +using ::testing::AllOf; using ::testing::Contains; +using ::testing::Each; using ::testing::Eq; +using ::testing::ExplainMatchResult; +using ::testing::HasSubstr; using ::testing::IsEmpty; using ::testing::Not; +using ::testing::Property; using ::testing::StartsWith; +MATCHER_P(MetricType, matcher, "") { + return ExplainMatchResult(matcher, arg.metric().type(), result_listener); +} + +MATCHER_P(ResourceType, matcher, "") { + return ExplainMatchResult(matcher, arg.resource().type(), result_listener); +} + +MATCHER_P2(HasMetricLabel, key, val_matcher, "") { + auto const& labels = arg.metric().labels(); + auto it = labels.find(key); + if (it == labels.end()) { + *result_listener << "no metric label '" << key << "'"; + return false; + } + return ExplainMatchResult(val_matcher, it->second, result_listener); +} + +MATCHER_P2(HasResourceLabel, key, val_matcher, "") { + auto const& labels = arg.resource().labels(); + auto it = labels.find(key); + if (it == labels.end()) { + *result_listener << "no resource label '" << key << "'"; + return false; + } + return ExplainMatchResult(val_matcher, it->second, result_listener); +} + +MATCHER_P(HasTimeSeries, ts_matcher, "") { + return ExplainMatchResult(Contains(ts_matcher), arg.time_series(), + result_listener); +} + class ObservabilityIntegrationTest : public ::google::cloud::bigtable::testing::TableIntegrationTest { protected: - static void SetUpTestSuite() { + void SetUp() override { + TableIntegrationTest::SetUp(); int port = 0; grpc::ServerBuilder builder; builder.AddListeningPort("localhost:0", grpc::InsecureServerCredentials(), @@ -93,42 +96,21 @@ class ObservabilityIntegrationTest &collector_service_)); server_ = builder.BuildAndStart(); server_address_ = absl::StrCat("localhost:", port); - env_endpoint_ = std::make_unique( - "GOOGLE_CLOUD_CPP_METRIC_SERVICE_ENDPOINT", server_address_); - env_otel_ = std::make_unique( - "GOOGLE_CLOUD_CPP_TESTING_OTEL_COLLECTOR", "1"); } - static void TearDownTestSuite() { - env_endpoint_.reset(); - env_otel_.reset(); + void TearDown() override { if (server_) { - server_->Shutdown(std::chrono::system_clock::now() + - std::chrono::seconds(1)); + server_->Shutdown(); server_->Wait(); } + TableIntegrationTest::TearDown(); } - void SetUp() override { - TableIntegrationTest::SetUp(); - data_connection_.reset(); - collector_service_.Clear(); - } - - static google::cloud::testing_util::OtelCollectorServer collector_service_; - static std::unique_ptr server_; - static std::string server_address_; - static std::unique_ptr env_endpoint_; - static std::unique_ptr env_otel_; + google::cloud::testing_util::OtelCollectorServer collector_service_; + std::unique_ptr server_; + std::string server_address_; }; -google::cloud::testing_util::OtelCollectorServer - ObservabilityIntegrationTest::collector_service_; -std::unique_ptr ObservabilityIntegrationTest::server_; -std::string ObservabilityIntegrationTest::server_address_; -std::unique_ptr ObservabilityIntegrationTest::env_endpoint_; -std::unique_ptr ObservabilityIntegrationTest::env_otel_; - /// Use Table::Apply() to insert a single row. void Apply(Table& table, std::string const& row_key, std::vector const& cells) { @@ -144,52 +126,36 @@ void Apply(Table& table, std::string const& row_key, ASSERT_STATUS_OK(status); } -void VerifyResourceLabels(google::monitoring::v3::TimeSeries const& ts, - std::string const& expected_project_id, - std::string const& expected_instance_id, - std::string const& expected_table_id) { - auto const& labels = ts.resource().labels(); - auto project_it = labels.find("project_id"); - if (project_it != labels.end()) { - EXPECT_THAT(project_it->second, Eq(expected_project_id)); - } - auto instance_it = labels.find("instance"); - if (instance_it != labels.end()) { - EXPECT_THAT(instance_it->second, Eq(expected_instance_id)); - } - auto table_it = labels.find("table"); - if (table_it != labels.end()) { - EXPECT_THAT(table_it->second, Eq(expected_table_id)); - } - auto zone_it = labels.find("zone"); - if (zone_it != labels.end() && !TableTestEnvironment::zone_a().empty()) { - std::vector parts = - absl::StrSplit(TableTestEnvironment::zone_a(), '-'); - auto prefix = parts.size() >= 2 ? absl::StrCat(parts[0], "-", parts[1]) - : TableTestEnvironment::zone_a(); - EXPECT_THAT(zone_it->second, StartsWith(prefix)); - } -} - TEST_F(ObservabilityIntegrationTest, VerifyOperationAndAttemptMetrics) { if (UsingCloudBigtableEmulator()) { GTEST_SKIP() << "Metrics export integration test runs against production"; } + // Redirect Cloud Monitoring metric export to local otel_collector + ScopedEnvironment env("GOOGLE_CLOUD_CPP_METRIC_SERVICE_ENDPOINT", + server_address_); + ScopedEnvironment env_otel("GOOGLE_CLOUD_CPP_TESTING_OTEL_COLLECTOR", "1"); + // Set MetricsPeriodOption to 5s (minimum allowed by DefaultOptions; smaller // periods reset to 60s) - auto options = Options{} - .set(true) - .set(std::chrono::seconds(5)) - .set(std::chrono::hours(1)) - .set(std::chrono::hours(1)); + auto options = + Options{}.set(true).set( + std::chrono::seconds(5)); auto const table_id = TableTestEnvironment::table_id(); + bool const is_dynamic = google::cloud::internal::GetEnv( + "GOOGLE_CLOUD_CPP_BIGTABLE_TESTING_CHANNEL_POOL") + .value_or("") == "dynamic"; // Add scoped connection to ensure metrics are flushed on destruction. { - auto conn = MakeDataConnection( - {InstanceResource(Project(project_id()), instance_id())}, options); + std::shared_ptr conn; + if (is_dynamic) { + conn = MakeDataConnection( + {InstanceResource(Project(project_id()), instance_id())}, options); + } else { + conn = MakeDataConnection(options); + } auto table = Table(std::move(conn), TableResource(project_id(), instance_id(), table_id)); @@ -199,7 +165,7 @@ TEST_F(ObservabilityIntegrationTest, VerifyOperationAndAttemptMetrics) { // Perform mutations and read calls Apply(table, row_key, expected); - auto actual = ReadRows(table, Filter::PassAllFilter()); + auto actual = ReadRows(table, Filter::RowKeysRegex(row_key)); CheckEqualUnordered(expected, actual); // Wait for the periodic 5-second exporter background thread to flush @@ -208,169 +174,79 @@ TEST_F(ObservabilityIntegrationTest, VerifyOperationAndAttemptMetrics) { } auto recorded = collector_service_.recorded_metrics(); - ASSERT_FALSE(recorded.empty()); - - bool found_operation_latencies = false; - bool found_attempt_latencies = false; - - for (auto const& req : recorded) { - EXPECT_THAT(req.name(), Eq(absl::StrCat("projects/", project_id()))); - - for (auto const& ts : req.time_series()) { - auto const& metric_type = ts.metric().type(); - // Skip gRPC metrics (which now start with - // "bigtable.googleapis.com/internal/client/grpc/") or non-Bigtable client - // metrics. This test only validates custom Bigtable client metrics - // ("bigtable_client_raw"). - if (ts.resource().type() != "bigtable_client_raw" || - !absl::StartsWith(metric_type, - "bigtable.googleapis.com/internal/client/")) { - continue; - } - - if (absl::StrContains(metric_type, "operation_latencies")) { - found_operation_latencies = true; - } - if (absl::StrContains(metric_type, "attempt_latencies")) { - found_attempt_latencies = true; - } - - VerifyResourceLabels(ts, project_id(), instance_id(), table_id); - } - } - - EXPECT_TRUE(found_operation_latencies); - EXPECT_TRUE(found_attempt_latencies); -} - -bool VerifyDirectPathGrpcResourceLabels( - google::monitoring::v3::TimeSeries const& ts, - absl::optional const& expected_location, - absl::optional const& expected_cloud_platform, - std::string const& expected_client_project, - absl::optional const& expected_hostname) { - if (ts.resource().type() != "bigtable_client") { - return false; - } + ASSERT_THAT(recorded, Not(IsEmpty())); + EXPECT_THAT( + recorded, + Each(Property(&google::monitoring::v3::CreateTimeSeriesRequest::name, + Eq(absl::StrCat("projects/", project_id()))))); - auto const& labels = ts.resource().labels(); - auto region_it = labels.find("region"); - auto platform_it = labels.find("cloud_platform"); - auto host_id_it = labels.find("host_id"); - auto client_project_it = labels.find("client_project"); + auto has_resource_labels = AllOf(HasResourceLabel("project_id", project_id()), + HasResourceLabel("instance", instance_id()), + HasResourceLabel("table", table_id)); - if (region_it == labels.end() || platform_it == labels.end() || - host_id_it == labels.end() || client_project_it == labels.end()) { - return false; - } - - EXPECT_THAT(region_it->second, Not(IsEmpty())); - if (expected_location.has_value() && !expected_location->empty()) { + if (!TableTestEnvironment::zone_a().empty()) { std::vector parts = - absl::StrSplit(*expected_location, '-'); - auto region_prefix = parts.size() >= 2 - ? absl::StrCat(parts[0], "-", parts[1]) - : *expected_location; - EXPECT_THAT(region_it->second, StartsWith(region_prefix)); - } - - EXPECT_THAT(platform_it->second, Not(IsEmpty())); - if (expected_cloud_platform.has_value() && - !expected_cloud_platform->empty()) { - EXPECT_THAT(platform_it->second, Eq(*expected_cloud_platform)); - } - - EXPECT_THAT(host_id_it->second, Not(IsEmpty())); - - EXPECT_THAT(client_project_it->second, Not(IsEmpty())); - if (!expected_client_project.empty()) { - EXPECT_THAT(client_project_it->second, Eq(expected_client_project)); - } - - if (expected_hostname.has_value() && !expected_hostname->empty()) { - auto hostname_it = labels.find("host_name"); - if (hostname_it != labels.end()) { - EXPECT_THAT(hostname_it->second, Eq(*expected_hostname)); - } - } - - return true; -} - -std::set ProcessRecordedGrpcMetrics( - std::vector const& + absl::StrSplit(TableTestEnvironment::zone_a(), '-'); + auto prefix = parts.size() >= 2 ? absl::StrCat(parts[0], "-", parts[1]) + : TableTestEnvironment::zone_a(); + EXPECT_THAT(recorded, Contains(HasTimeSeries(AllOf( + MetricType(HasSubstr("operation_latencies")), + has_resource_labels, + HasResourceLabel("zone", StartsWith(prefix)))))); + EXPECT_THAT(recorded, Contains(HasTimeSeries(AllOf( + MetricType(HasSubstr("attempt_latencies")), + has_resource_labels, + HasResourceLabel("zone", StartsWith(prefix)))))); + } else { + EXPECT_THAT(recorded, Contains(HasTimeSeries(AllOf( + MetricType(HasSubstr("operation_latencies")), + has_resource_labels)))); + EXPECT_THAT( recorded, - std::string const& project_id, - absl::optional const& expected_location, - absl::optional const& expected_cloud_platform, - std::string const& expected_client_project, - absl::optional const& expected_hostname, - bool& verified_resource_labels) { - std::set grpc_metric_types; - for (auto const& req : recorded) { - EXPECT_THAT(req.name(), Eq(absl::StrCat("projects/", project_id))); - - for (auto const& ts : req.time_series()) { - auto const& metric_type = ts.metric().type(); - if (absl::StartsWith(metric_type, - "bigtable.googleapis.com/internal/client/grpc/")) { - grpc_metric_types.insert(metric_type); - if (VerifyDirectPathGrpcResourceLabels( - ts, expected_location, expected_cloud_platform, - expected_client_project, expected_hostname)) { - verified_resource_labels = true; - } - } - } + Contains(HasTimeSeries(AllOf(MetricType(HasSubstr("attempt_latencies")), + has_resource_labels)))); } - return grpc_metric_types; } -TEST_F(ObservabilityIntegrationTest, VerifyDirectPathGrpcMetrics) { +TEST_F(ObservabilityIntegrationTest, VerifyOutstandingRpcsMetric) { if (UsingCloudBigtableEmulator()) { GTEST_SKIP() << "Metrics export integration test runs against production"; } - auto disable_direct_path = - google::cloud::internal::GetEnv("GOOGLE_CLOUD_DISABLE_DIRECT_PATH") - .value_or(""); - if (disable_direct_path == "true" || !IsDirectPathReachable()) { - GTEST_SKIP() << "DirectPath is disabled or network is unreachable in this " - "test environment"; - } + // Redirect Cloud Monitoring metric export to local otel_collector + ScopedEnvironment env("GOOGLE_CLOUD_CPP_METRIC_SERVICE_ENDPOINT", + server_address_); + ScopedEnvironment env_otel("GOOGLE_CLOUD_CPP_TESTING_OTEL_COLLECTOR", "1"); // Set MetricsPeriodOption to 5s (minimum allowed by DefaultOptions; smaller // periods reset to 60s) - auto options = Options{} - .set(true) - .set( - experimental::DirectPathMode::kEnabled) - .set(std::chrono::seconds(5)) - .set(std::chrono::hours(1)) - .set(std::chrono::hours(1)) - .set({ - {"grpc.client_idle_timeout_ms", "1000"}, - {"grpc.max_reconnect_backoff_ms", "1000"}, - }); - - auto const& table_id = TableTestEnvironment::table_id(); + auto options = + Options{}.set(true).set( + std::chrono::seconds(5)); - // Add scoped connection to ensure metrics are flushed on destruction. + auto const table_id = TableTestEnvironment::table_id(); + bool const is_dynamic = google::cloud::internal::GetEnv( + "GOOGLE_CLOUD_CPP_BIGTABLE_TESTING_CHANNEL_POOL") + .value_or("") == "dynamic"; + if (!is_dynamic) { + GTEST_SKIP() + << "OutstandingRpcs metric is only supported for dynamic channel pools"; + } + + collector_service_.Clear(); { auto conn = MakeDataConnection( {InstanceResource(Project(project_id()), instance_id())}, options); auto table = Table(std::move(conn), TableResource(project_id(), instance_id(), table_id)); - std::string const row_key = "observability-directpath-row-1"; - std::vector expected{ - {row_key, "family4", "c0", 1000, "v1000"}, - {row_key, "family4", "c1", 2000, "v2000"}, - }; + std::string const row_key = "observability-rpc-RANDOM_TWO_LEAST_USED"; + std::vector expected{{row_key, "family4", "c0", 1000, "v1000"}, + {row_key, "family4", "c1", 2000, "v2000"}}; - // Perform mutations and read calls over DirectPath + // Perform mutations and read calls Apply(table, row_key, expected); - auto actual = ReadRows(table, Filter::PassAllFilter()); + auto actual = ReadRows(table, Filter::RowKeysRegex(row_key)); CheckEqualUnordered(expected, actual); // Wait for the periodic 5-second exporter background thread to flush @@ -379,45 +255,26 @@ TEST_F(ObservabilityIntegrationTest, VerifyDirectPathGrpcMetrics) { } auto recorded = collector_service_.recorded_metrics(); - ASSERT_FALSE(recorded.empty()); - - bool verified_resource_labels = false; - - auto expected_client_project = - google::cloud::internal::GetEnv( - "GOOGLE_CLOUD_CPP_TEST_EXPECTED_CLIENT_PROJECT") - .value_or(project_id()); - auto expected_location = google::cloud::internal::GetEnv( - "GOOGLE_CLOUD_CPP_TEST_EXPECTED_LOCATION"); - auto expected_cloud_platform = google::cloud::internal::GetEnv( - "GOOGLE_CLOUD_CPP_TEST_EXPECTED_CLOUD_PLATFORM"); - auto expected_hostname = google::cloud::internal::GetEnv( - "GOOGLE_CLOUD_CPP_TEST_EXPECTED_HOSTNAME"); - - auto grpc_metric_types = ProcessRecordedGrpcMetrics( - recorded, project_id(), expected_location, expected_cloud_platform, - expected_client_project, expected_hostname, verified_resource_labels); - - EXPECT_TRUE(verified_resource_labels); - - // Verify that specific gRPC client metrics configured in GrpcMetricsExporter - // are present. OpenTelemetry metric names are formatted with the - // "bigtable.googleapis.com/internal/client/" prefix and slashes ("/") in - // place of dots ("."). - // - // Note: Event-driven and failure-driven metrics configured in - // GrpcMetricsExporter (such as grpc.lb.rls.*, grpc.xds_client.*, and - // grpc.subchannel.* disconnections/failures) are only exported when those - // specific events or errors occur during the export window. Therefore, only - // RPC attempt duration metric is guaranteed to produce time series during a - // healthy test run. - EXPECT_THAT(grpc_metric_types, - Contains("bigtable.googleapis.com/internal/client/" - "grpc/client/attempt/duration")); + ASSERT_THAT(recorded, Not(IsEmpty())); + EXPECT_THAT( + recorded, + Each(Property(&google::monitoring::v3::CreateTimeSeriesRequest::name, + Eq(absl::StrCat("projects/", project_id()))))); + + EXPECT_THAT( + recorded, + Contains(HasTimeSeries(AllOf( + MetricType(HasSubstr("connection_pool/outstanding_rpcs")), + ResourceType("bigtable.googleapis.com/Client"), + HasResourceLabel("project_id", project_id()), + HasResourceLabel("instance", instance_id()), + HasMetricLabel("channel_pool_lb_policy", "RANDOM_TWO_LEAST_USED"), + HasMetricLabel("transport_type", Not(IsEmpty())), + HasMetricLabel("streaming", Not(IsEmpty())))))); } } // namespace -} // namespace testing +GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END } // namespace bigtable } // namespace cloud } // namespace google @@ -428,5 +285,3 @@ int main(int argc, char* argv[]) { new ::google::cloud::bigtable::testing::TableTestEnvironment); return RUN_ALL_TESTS(); } - -#endif // _WIN32 diff --git a/protos/google/cloud/opentelemetry/CMakeLists.txt b/protos/google/cloud/opentelemetry/CMakeLists.txt index 9de6f89d0cb2e..95d90e2a47ff5 100644 --- a/protos/google/cloud/opentelemetry/CMakeLists.txt +++ b/protos/google/cloud/opentelemetry/CMakeLists.txt @@ -19,3 +19,9 @@ endif () if (IS_DIRECTORY "${CMAKE_CURRENT_SOURCE_DIR}/testing") add_subdirectory(testing) endif () + +if (IS_DIRECTORY "${PROJECT_SOURCE_DIR}/ci/otel_collector" + AND EXISTS "${PROJECT_SOURCE_DIR}/ci/otel_collector/CMakeLists.txt") + add_subdirectory("${PROJECT_SOURCE_DIR}/ci/otel_collector" + "${CMAKE_CURRENT_BINARY_DIR}/ci_otel_collector") +endif ()