Skip to content

Commit 8763ff1

Browse files
bajajneha27bajajneha27
committed
Address review comments
1 parent cd223d9 commit 8763ff1

7 files changed

Lines changed: 92 additions & 54 deletions

File tree

‎google/cloud/storage/grpc_plugin.cc‎

Lines changed: 15 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,12 @@
1818
#include "google/cloud/storage/internal/grpc/default_options.h"
1919
#include "google/cloud/storage/internal/grpc/enable_metrics.h"
2020
#include "google/cloud/storage/internal/grpc/stub.h"
21+
#include "google/cloud/storage/internal/tracing_connection.h"
22+
#include "google/cloud/background_threads.h"
23+
#include "google/cloud/grpc_options.h"
2124
#include "google/cloud/internal/getenv.h"
25+
#include "google/cloud/internal/opentelemetry.h"
26+
#include <functional>
2227
#include <memory>
2328
#include <utility>
2429

@@ -31,8 +36,17 @@ google::cloud::storage::Client MakeGrpcClient(Options opts) {
3136
opts = google::cloud::storage_internal::DefaultOptionsGrpc(std::move(opts));
3237
storage_internal::EnableGrpcMetrics(opts);
3338
auto stub = std::make_unique<storage_internal::GrpcStub>(opts);
39+
storage_internal::TracingConnection::AsyncRunner runner;
40+
if (google::cloud::internal::TracingEnabled(opts)) {
41+
auto threads = std::shared_ptr<google::cloud::BackgroundThreads>(
42+
google::cloud::internal::MakeBackgroundThreadsFactory(opts)());
43+
runner = [threads](std::function<void()> f) {
44+
threads->cq().RunAsync(std::move(f));
45+
};
46+
}
3447
return storage::internal::ClientImplDetails::CreateWithoutDecorations(
35-
MakeStorageConnection(std::move(opts), std::move(stub)));
48+
MakeStorageConnection(std::move(opts), std::move(stub),
49+
std::move(runner)));
3650
}
3751

3852
GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END

‎google/cloud/storage/grpc_plugin_test.cc‎

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -14,8 +14,10 @@
1414

1515
#include "google/cloud/storage/grpc_plugin.h"
1616
#include "google/cloud/storage/options.h"
17+
#include "google/cloud/background_threads.h"
1718
#include "google/cloud/common_options.h"
1819
#include "google/cloud/credentials.h"
20+
#include "google/cloud/grpc_options.h"
1921
#include "google/cloud/testing_util/scoped_environment.h"
2022
#include <gmock/gmock.h>
2123

@@ -97,6 +99,23 @@ TEST(GrpcPluginTest, GrpcMetricsExcludedLabelsOptionSingle) {
9799
opts.get<storage_experimental::GrpcMetricsExcludedLabelsOption>());
98100
}
99101

102+
TEST(GrpcPluginTest, GrpcBackgroundThreadsFactoryOptionWithTracing) {
103+
struct Fake : google::cloud::BackgroundThreads {
104+
google::cloud::CompletionQueue cq() const override { return {}; }
105+
};
106+
bool invoked = false;
107+
auto factory = [&invoked] {
108+
invoked = true;
109+
return std::make_unique<Fake>();
110+
};
111+
auto logging = ScopedEnvironment("CLOUD_STORAGE_ENABLE_TRACING", "1");
112+
auto client = MakeGrpcClient(
113+
TestOptions()
114+
.set<GrpcBackgroundThreadsFactoryOption>(factory)
115+
.set<storage_experimental::OTelSpanEnrichmentOption>(true));
116+
EXPECT_TRUE(invoked);
117+
}
118+
100119
} // namespace
101120
GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
102121
} // namespace storage

‎google/cloud/storage/internal/connection_factory.cc‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -34,11 +34,13 @@ std::shared_ptr<storage::internal::StorageConnection> DecorateConnection(
3434
}
3535

3636
std::shared_ptr<storage::internal::StorageConnection> MakeStorageConnection(
37-
Options const& opts, std::unique_ptr<storage_internal::GenericStub> stub) {
37+
Options const& opts, std::unique_ptr<storage_internal::GenericStub> stub,
38+
TracingConnection::AsyncRunner runner) {
3839
std::shared_ptr<storage::internal::StorageConnection> connection =
3940
storage::internal::StorageConnectionImpl::Create(std::move(stub), opts);
4041
if (google::cloud::internal::TracingEnabled(opts)) {
41-
connection = storage_internal::MakeTracingClient(std::move(connection));
42+
connection = storage_internal::MakeTracingClient(std::move(connection),
43+
std::move(runner));
4244
}
4345
return connection;
4446
}

‎google/cloud/storage/internal/connection_factory.h‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@
1616
#define GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_STORAGE_INTERNAL_CONNECTION_FACTORY_H
1717

1818
#include "google/cloud/storage/internal/storage_connection.h"
19+
#include "google/cloud/storage/internal/tracing_connection.h"
1920
#include "google/cloud/options.h"
2021
#include "google/cloud/version.h"
2122
#include <memory>
@@ -46,7 +47,8 @@ std::shared_ptr<storage::internal::StorageConnection> DecorateConnection(
4647
* initialize this stub in the usual `MakeStorageConnection(Options)` function.
4748
*/
4849
std::shared_ptr<storage::internal::StorageConnection> MakeStorageConnection(
49-
Options const& opts, std::unique_ptr<GenericStub> stub);
50+
Options const& opts, std::unique_ptr<GenericStub> stub,
51+
TracingConnection::AsyncRunner runner = {});
5052

5153
/// Creates a fully configured connection for the storage service.
5254
std::shared_ptr<storage::internal::StorageConnection> MakeStorageConnection(

‎google/cloud/storage/internal/tracing_connection.cc‎

Lines changed: 18 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -17,10 +17,7 @@
1717
#include "google/cloud/storage/options.h"
1818
#include "google/cloud/storage/parallel_upload.h"
1919
#include "google/cloud/internal/opentelemetry.h"
20-
#include "google/cloud/options.h"
21-
#if GOOGLE_CLOUD_CPP_STORAGE_HAVE_GRPC
22-
#include "google/cloud/grpc_options.h"
23-
#endif
20+
#include "google/cloud/internal/rest_pure_background_threads_impl.h"
2421
#include <algorithm>
2522
#include <memory>
2623
#include <string>
@@ -32,24 +29,18 @@ namespace cloud {
3229
namespace storage_internal {
3330
GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
3431

35-
namespace {
36-
std::size_t DefaultThreadPoolSize(Options const& options) {
37-
#if GOOGLE_CLOUD_CPP_STORAGE_HAVE_GRPC
38-
auto pool_size = options.get<GrpcBackgroundThreadPoolSizeOption>();
39-
if (pool_size == 0) return 1U;
40-
return pool_size;
41-
#else
42-
(void)options;
43-
return 1U;
44-
#endif
45-
}
46-
} // namespace
47-
48-
TracingConnection::TracingConnection(std::shared_ptr<StorageConnection> impl)
49-
: impl_(std::move(impl)),
50-
background_threads_(
51-
std::make_unique<AutomaticallyCreatedStorageBackgroundThreads>(
52-
DefaultThreadPoolSize(impl_->options()))) {}
32+
TracingConnection::TracingConnection(std::shared_ptr<StorageConnection> impl,
33+
AsyncRunner runner)
34+
: impl_(std::move(impl)), runner_(std::move(runner)) {
35+
if (!runner_) {
36+
auto threads =
37+
std::make_shared<google::cloud::rest_internal::
38+
AutomaticallyCreatedRestPureBackgroundThreads>(1U);
39+
runner_ = [threads](std::function<void()> f) {
40+
threads->cq().RunAsync(std::move(f));
41+
};
42+
}
43+
}
5344

5445
TracingConnection::~TracingConnection() = default;
5546

@@ -75,7 +66,7 @@ void TracingConnection::MaybeTriggerBackgroundFetch(
7566
}
7667

7768
auto current_options = google::cloud::internal::SaveCurrentOptions();
78-
background_threads_->cq().RunAsync([this, bucket_name, current_options]() {
69+
runner_([this, bucket_name, current_options]() {
7970
google::cloud::internal::OptionsSpan span(current_options);
8071
storage::internal::GetBucketMetadataRequest request(bucket_name);
8172
auto result = impl_->GetBucketMetadata(request);
@@ -710,8 +701,10 @@ std::vector<std::string> TracingConnection::InspectStackStructure() const {
710701
}
711702

712703
std::shared_ptr<storage::internal::StorageConnection> MakeTracingClient(
713-
std::shared_ptr<storage::internal::StorageConnection> impl) {
714-
return std::make_shared<TracingConnection>(std::move(impl));
704+
std::shared_ptr<storage::internal::StorageConnection> impl,
705+
TracingConnection::AsyncRunner runner) {
706+
return std::make_shared<TracingConnection>(std::move(impl),
707+
std::move(runner));
715708
}
716709

717710
GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END

‎google/cloud/storage/internal/tracing_connection.h‎

Lines changed: 7 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -19,13 +19,7 @@
1919
#include "google/cloud/storage/internal/storage_connection.h"
2020
#include "google/cloud/storage/parallel_upload.h"
2121
#include "google/cloud/storage/version.h"
22-
#include "google/cloud/internal/generic_background_threads_impl.h"
23-
#if GOOGLE_CLOUD_CPP_STORAGE_HAVE_GRPC
24-
#include "google/cloud/background_threads.h"
25-
#include "google/cloud/completion_queue.h"
26-
#else
27-
#include "google/cloud/internal/rest_pure_background_threads_impl.h"
28-
#endif
22+
#include <functional>
2923
#include <memory>
3024
#include <string>
3125

@@ -36,7 +30,9 @@ GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
3630

3731
class TracingConnection : public storage::internal::StorageConnection {
3832
public:
39-
explicit TracingConnection(std::shared_ptr<StorageConnection> impl);
33+
using AsyncRunner = std::function<void(std::function<void()>)>;
34+
explicit TracingConnection(std::shared_ptr<StorageConnection> impl,
35+
AsyncRunner runner = {});
4036
~TracingConnection() override;
4137

4238
static void ResetCacheForTesting();
@@ -210,27 +206,13 @@ class TracingConnection : public storage::internal::StorageConnection {
210206

211207
static BucketMetadataCache& cache();
212208

213-
#if GOOGLE_CLOUD_CPP_STORAGE_HAVE_GRPC
214-
using StorageBackgroundThreads = google::cloud::BackgroundThreads;
215-
using AutomaticallyCreatedStorageBackgroundThreads =
216-
google::cloud::internal::AutomaticallyCreatedBackgroundThreadsImpl<
217-
google::cloud::CompletionQueue, google::cloud::BackgroundThreads>;
218-
#else
219-
using StorageBackgroundThreads =
220-
google::cloud::rest_internal::RestPureBackgroundThreads;
221-
using AutomaticallyCreatedStorageBackgroundThreads =
222-
google::cloud::internal::AutomaticallyCreatedBackgroundThreadsImpl<
223-
rest_internal::RestPureCompletionQueue,
224-
rest_internal::RestPureBackgroundThreads,
225-
rest_internal::RestPureQueueTraits>;
226-
#endif
227-
228209
std::shared_ptr<StorageConnection> impl_;
229-
std::unique_ptr<StorageBackgroundThreads> background_threads_;
210+
AsyncRunner runner_;
230211
};
231212

232213
std::shared_ptr<storage::internal::StorageConnection> MakeTracingClient(
233-
std::shared_ptr<storage::internal::StorageConnection> impl);
214+
std::shared_ptr<storage::internal::StorageConnection> impl,
215+
TracingConnection::AsyncRunner runner = {});
234216

235217
GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
236218
} // namespace storage_internal

‎google/cloud/storage/internal/tracing_connection_test.cc‎

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -64,6 +64,32 @@ TEST(TracingClientTest, Options) {
6464
EXPECT_EQ(42, options.get<TestOption>());
6565
}
6666

67+
TEST(TracingClientTest, CustomAsyncRunner) {
68+
TracingConnection::ResetCacheForTesting();
69+
bool invoked = false;
70+
auto runner = [&invoked](std::function<void()> f) {
71+
invoked = true;
72+
f();
73+
};
74+
auto mock = std::make_shared<MockClient>();
75+
EXPECT_CALL(*mock, options)
76+
.WillRepeatedly(testing::Return(
77+
Options{}.set<storage_experimental::OTelSpanEnrichmentOption>(true)));
78+
EXPECT_CALL(*mock, GetBucketMetadata).WillOnce([](auto const&) {
79+
storage::BucketMetadata metadata;
80+
metadata.set_name("test-bucket");
81+
return metadata;
82+
});
83+
EXPECT_CALL(*mock, DeleteBucket)
84+
.WillOnce(
85+
testing::Return(make_status_or(storage::internal::EmptyResponse{})));
86+
87+
auto under_test = TracingConnection(mock, runner);
88+
(void)under_test.DeleteBucket(
89+
storage::internal::DeleteBucketRequest("test-bucket"));
90+
EXPECT_TRUE(invoked);
91+
}
92+
6793
TEST(TracingClientTest, ListBuckets) {
6894
auto span_catcher = InstallSpanCatcher();
6995
auto mock = std::make_shared<MockClient>();

0 commit comments

Comments
 (0)