Skip to content

Commit 7453c11

Browse files
bajajneha27bajajneha27
committed
sync changes
1 parent 5e13c77 commit 7453c11

8 files changed

Lines changed: 219 additions & 176 deletions

‎google/cloud/storage/internal/async/connection_tracing.cc‎

Lines changed: 155 additions & 66 deletions
Large diffs are not rendered by default.

‎google/cloud/storage/internal/async/connection_tracing_test.cc‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@
2020
#include "google/cloud/storage/mocks/mock_async_reader_connection.h"
2121
#include "google/cloud/storage/mocks/mock_async_rewriter_connection.h"
2222
#include "google/cloud/storage/mocks/mock_async_writer_connection.h"
23+
#include "google/cloud/storage/options.h"
2324
#include "google/cloud/storage/testing/canonical_errors.h"
2425
#include "google/cloud/opentelemetry_options.h"
2526
#include "google/cloud/testing_util/opentelemetry_matchers.h"
@@ -44,8 +45,10 @@ using ::google::cloud::testing_util::EventNamed;
4445
using ::google::cloud::testing_util::InstallSpanCatcher;
4546
using ::google::cloud::testing_util::IsOk;
4647
using ::google::cloud::testing_util::IsOkAndHolds;
48+
using ::google::cloud::testing_util::OTelAttribute;
4749
using ::google::cloud::testing_util::OTelContextCaptured;
4850
using ::google::cloud::testing_util::PromiseWithOTelContext;
51+
using ::google::cloud::testing_util::SpanHasAttributes;
4952
using ::google::cloud::testing_util::SpanHasEvents;
5053
using ::google::cloud::testing_util::SpanHasInstrumentationScope;
5154
using ::google::cloud::testing_util::SpanKindIsClient;

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

Lines changed: 5 additions & 30 deletions
Original file line numberDiff line numberDiff line change
@@ -14,7 +14,6 @@
1414

1515
#include "google/cloud/storage/internal/bucket_metadata_cache.h"
1616
#include "google/cloud/storage/bucket_metadata.h"
17-
#include "google/cloud/status.h"
1817
#include <mutex>
1918
#include <utility>
2019

@@ -23,11 +22,6 @@ namespace cloud {
2322
namespace storage_internal {
2423
GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
2524

26-
BucketMetadataCache& BucketMetadataCache::Singleton() {
27-
static BucketMetadataCache instance(10000);
28-
return instance;
29-
}
30-
3125
BucketCacheEntry BucketCacheEntry::FromMetadata(
3226
storage::BucketMetadata const& m) {
3327
std::string loc = m.location();
@@ -40,28 +34,12 @@ BucketCacheEntry BucketCacheEntry::FromMetadata(
4034
std::move(loc)};
4135
}
4236

43-
absl::optional<BucketCacheEntry> BucketCacheEntry::Create(
44-
std::string const& bucket_name,
45-
StatusOr<storage::BucketMetadata> const& metadata) {
46-
if (metadata.ok()) {
47-
return FromMetadata(*metadata);
48-
}
49-
if (metadata.status().code() == StatusCode::kPermissionDenied) {
50-
return BucketCacheEntry{
51-
"projects/_/buckets/" +
52-
BucketMetadataCache::NormalizeBucketName(bucket_name),
53-
"global"};
54-
}
55-
return absl::nullopt;
56-
}
57-
5837
void BucketMetadataCache::MoveToFront(std::list<std::string>::iterator it) {
5938
list_.splice(list_.begin(), list_, it);
6039
}
6140

6241
absl::optional<BucketCacheEntry> BucketMetadataCache::Get(
63-
std::string const& raw_bucket_name) {
64-
auto const bucket_name = NormalizeBucketName(raw_bucket_name);
42+
std::string const& bucket_name) {
6543
std::unique_lock<std::mutex> lk(mu_);
6644
auto it = map_.find(bucket_name);
6745
if (it == map_.end()) return absl::nullopt;
@@ -70,7 +48,7 @@ absl::optional<BucketCacheEntry> BucketMetadataCache::Get(
7048
return it->second.first;
7149
}
7250

73-
void BucketMetadataCache::Put(std::string const& raw_bucket_name,
51+
void BucketMetadataCache::Put(std::string const& bucket_name,
7452
BucketCacheEntry entry) {
7553
if (max_size_ == 0) return;
7654
std::unique_lock<std::mutex> lk(mu_);
@@ -91,8 +69,7 @@ void BucketMetadataCache::Put(std::string const& raw_bucket_name,
9169
map_[bucket_name] = {std::move(entry), list_.begin()};
9270
}
9371

94-
void BucketMetadataCache::Invalidate(std::string const& raw_bucket_name) {
95-
auto const bucket_name = NormalizeBucketName(raw_bucket_name);
72+
void BucketMetadataCache::Invalidate(std::string const& bucket_name) {
9673
std::unique_lock<std::mutex> lk(mu_);
9774
auto it = map_.find(bucket_name);
9875
if (it != map_.end()) {
@@ -107,8 +84,7 @@ void BucketMetadataCache::Clear() {
10784
list_.clear();
10885
}
10986

110-
bool BucketMetadataCache::StartFetch(std::string const& raw_bucket_name) {
111-
auto const bucket_name = NormalizeBucketName(raw_bucket_name);
87+
bool BucketMetadataCache::StartFetch(std::string const& bucket_name) {
11288
std::unique_lock<std::mutex> lk(mu_);
11389
if (in_flight_fetch_.find(bucket_name) != in_flight_fetch_.end()) {
11490
return false;
@@ -117,8 +93,7 @@ bool BucketMetadataCache::StartFetch(std::string const& raw_bucket_name) {
11793
return true;
11894
}
11995

120-
void BucketMetadataCache::EndFetch(std::string const& raw_bucket_name) {
121-
auto const bucket_name = NormalizeBucketName(raw_bucket_name);
96+
void BucketMetadataCache::EndFetch(std::string const& bucket_name) {
12297
std::unique_lock<std::mutex> lk(mu_);
12398
in_flight_fetch_.erase(bucket_name);
12499
}

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

Lines changed: 6 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -15,10 +15,7 @@
1515
#ifndef GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_STORAGE_INTERNAL_BUCKET_METADATA_CACHE_H
1616
#define GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_STORAGE_INTERNAL_BUCKET_METADATA_CACHE_H
1717

18-
#include "google/cloud/storage/bucket_metadata.h"
1918
#include "google/cloud/storage/version.h"
20-
#include "google/cloud/status_or.h"
21-
#include "absl/strings/match.h"
2219
#include "absl/types/optional.h"
2320
#include <cstddef>
2421
#include <list>
@@ -31,6 +28,11 @@
3128

3229
namespace google {
3330
namespace cloud {
31+
namespace storage {
32+
GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
33+
class BucketMetadata;
34+
GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
35+
} // namespace storage
3436
namespace storage_internal {
3537
GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
3638

@@ -39,25 +41,12 @@ struct BucketCacheEntry {
3941
std::string location;
4042

4143
static BucketCacheEntry FromMetadata(storage::BucketMetadata const& m);
42-
static absl::optional<BucketCacheEntry> Create(
43-
std::string const& bucket_name,
44-
StatusOr<storage::BucketMetadata> const& metadata);
4544
};
4645

4746
class BucketMetadataCache {
4847
public:
4948
explicit BucketMetadataCache(std::size_t max_size = 10000)
50-
: max_size_(max_size == 0 ? 1 : max_size) {}
51-
52-
static BucketMetadataCache& Singleton();
53-
54-
static std::string NormalizeBucketName(std::string const& bucket) {
55-
auto const prefix = std::string("projects/_/buckets/");
56-
if (absl::StartsWith(bucket, prefix)) {
57-
return bucket.substr(prefix.size());
58-
}
59-
return bucket;
60-
}
49+
: max_size_(max_size) {}
6150

6251
absl::optional<BucketCacheEntry> Get(std::string const& bucket_name);
6352
void Put(std::string const& bucket_name, BucketCacheEntry entry);

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

Lines changed: 30 additions & 34 deletions
Original file line numberDiff line numberDiff line change
@@ -53,30 +53,27 @@ TracingConnection::AsyncRunner const& TracingConnection::runner() {
5353

5454
BucketMetadataCache& TracingConnection::cache() const { return *cache_; }
5555

56-
TracingConnection::~TracingConnection() {
57-
for (auto& f : bg_tasks_) {
58-
if (f.valid()) f.wait();
59-
}
60-
}
56+
TracingConnection::~TracingConnection() = default;
6157

62-
BucketMetadataCache& TracingConnection::cache() {
63-
return BucketMetadataCache::Singleton();
58+
TracingConnection::AsyncRunner const& TracingConnection::runner() {
59+
absl::call_once(once_flag_, [this] {
60+
if (!runner_) {
61+
auto threads =
62+
std::make_shared<google::cloud::rest_internal::
63+
AutomaticallyCreatedRestPureBackgroundThreads>(
64+
1U);
65+
runner_ = [threads](std::function<void()> f) {
66+
threads->cq().RunAsync(std::move(f));
67+
};
68+
}
69+
});
70+
return runner_;
6471
}
6572

66-
void TracingConnection::ResetCacheForTesting() { cache().Clear(); }
73+
BucketMetadataCache& TracingConnection::cache() const { return *cache_; }
6774

6875
Options TracingConnection::options() const { return impl_->options(); }
6976

70-
void TracingConnection::CleanupCompletedTasks() {
71-
std::unique_lock<std::mutex> lk(mu_);
72-
bg_tasks_.erase(std::remove_if(bg_tasks_.begin(), bg_tasks_.end(),
73-
[](std::future<void> const& f) {
74-
return f.wait_for(std::chrono::seconds(0)) ==
75-
std::future_status::ready;
76-
}),
77-
bg_tasks_.end());
78-
}
79-
8077
void TracingConnection::EnrichSpan(opentelemetry::trace::Span& span,
8178
BucketCacheEntry const& entry) {
8279
span.SetAttribute("gcp.resource.destination.id", entry.id);
@@ -85,33 +82,32 @@ void TracingConnection::EnrichSpan(opentelemetry::trace::Span& span,
8582

8683
void TracingConnection::MaybeTriggerBackgroundFetch(
8784
std::string const& bucket_name) {
88-
CleanupCompletedTasks();
89-
9085
if (!cache().StartFetch(bucket_name)) {
9186
return;
9287
}
9388

89+
auto guard = ScopedFetch(cache_, bucket_name);
9490
auto current_options = google::cloud::internal::SaveCurrentOptions();
95-
auto f =
96-
std::async(std::launch::async, [this, bucket_name, current_options]() {
97-
google::cloud::internal::OptionsSpan span(current_options);
98-
storage::internal::GetBucketMetadataRequest request(bucket_name);
99-
auto result = impl_->GetBucketMetadata(request);
100-
101-
auto entry = BucketCacheEntry::Create(bucket_name, result);
102-
if (entry) {
103-
cache().Put(bucket_name, std::move(*entry));
104-
}
105-
106-
cache().EndFetch(bucket_name);
107-
});
91+
runner()([impl = impl_, cache = cache_, bucket_name, current_options,
92+
guard]() {
93+
google::cloud::internal::OptionsSpan span(current_options);
94+
storage::internal::GetBucketMetadataRequest request(bucket_name);
95+
auto result = impl->GetBucketMetadata(request);
10896

109-
bg_tasks_.push_back(std::move(f));
97+
if (result.ok()) {
98+
cache->Put(bucket_name, BucketCacheEntry::FromMetadata(*result));
99+
} else if (result.status().code() == StatusCode::kPermissionDenied) {
100+
cache->Put(bucket_name, {"projects/_/buckets/" + bucket_name, "global"});
101+
}
102+
});
110103
}
111104

112105
void TracingConnection::EnrichSpan(opentelemetry::trace::Span& span,
113106
std::string const& bucket_name) {
114107
if (bucket_name.empty()) return;
108+
auto const enabled =
109+
options().get<storage_experimental::OTelSpanEnrichmentOption>();
110+
if (!enabled) return;
115111
auto entry = cache().Get(bucket_name);
116112
if (entry.has_value()) {
117113
EnrichSpan(span, *entry);

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

Lines changed: 0 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,6 @@
2222
#include "absl/base/call_once.h"
2323
#include <functional>
2424
#include <memory>
25-
#include <mutex>
2625
#include <string>
2726

2827
namespace google {
@@ -37,8 +36,6 @@ class TracingConnection : public storage::internal::StorageConnection {
3736
AsyncRunner runner = {});
3837
~TracingConnection() override;
3938

40-
static void ResetCacheForTesting();
41-
4239
Options options() const override;
4340

4441
StatusOr<storage::internal::ListBucketsResponse> ListBuckets(

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

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -216,7 +216,6 @@ TEST(TracingClientTest, GetBucketMetadataSuccess) {
216216
}
217217

218218
TEST(TracingClientTest, BucketMetadataCacheSuccess) {
219-
TracingConnection::ResetCacheForTesting();
220219
auto span_catcher = InstallSpanCatcher();
221220
auto mock = std::make_shared<MockClient>();
222221

‎google/cloud/storage/tests/object_plenty_clients_serially_integration_test.cc‎

Lines changed: 20 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,6 @@
1717
#include "google/cloud/storage/testing/storage_integration_test.h"
1818
#include "google/cloud/internal/getenv.h"
1919
#include "google/cloud/log.h"
20-
#include "google/cloud/opentelemetry_options.h"
2120
#include "google/cloud/status_or.h"
2221
#include "google/cloud/testing_util/expect_exception.h"
2322
#include "google/cloud/testing_util/status_matchers.h"
@@ -44,18 +43,26 @@ TEST_F(ObjectPlentyClientsSeriallyIntegrationTest, PlentyClientsSerially) {
4443
// own tests.
4544
if (UsingGrpc()) GTEST_SKIP();
4645

47-
auto options =
48-
Options{}.set<google::cloud::OpenTelemetryTracingOption>(false);
46+
// With the advent of Regional Access Boundaries, connecting to non-regional
47+
// endpoints requires background calls to IAM. These background calls use
48+
// additional file descriptors which causes this test to fail when using the
49+
// default endpoint.
50+
auto regional_bucket = google::cloud::internal::GetEnv(
51+
"GOOGLE_CLOUD_CPP_STORAGE_TEST_DESTINATION");
52+
if (!regional_bucket.has_value()) GTEST_SKIP();
53+
bucket_name_ = *regional_bucket;
54+
// The regional_bucket was created in the us-west2 region.
55+
auto options = Options{}.set<RestEndpointOption>(
56+
"https://storage.us-west2.rep.googleapis.com");
57+
58+
auto client = MakeIntegrationTestClient(options);
4959
auto object_name = MakeRandomObjectName();
5060
std::string expected = LoremIpsum();
5161

52-
{
53-
auto client = MakeIntegrationTestClient(options);
54-
StatusOr<ObjectMetadata> meta = client.InsertObject(
55-
bucket_name_, object_name, expected, IfGenerationMatch(0));
56-
ASSERT_STATUS_OK(meta);
57-
ScheduleForDelete(*meta);
58-
}
62+
StatusOr<ObjectMetadata> meta = client.InsertObject(
63+
bucket_name_, object_name, expected, IfGenerationMatch(0));
64+
ASSERT_STATUS_OK(meta);
65+
ScheduleForDelete(*meta);
5966

6067
// Track the number of open files to ensure every client creates the same
6168
// number of file descriptors and none are leaked.
@@ -68,21 +75,6 @@ TEST_F(ObjectPlentyClientsSeriallyIntegrationTest, PlentyClientsSerially) {
6875
EXPECT_EQ(StatusCode::kUnimplemented, num_fds_before_test.status().code());
6976
}
7077
std::size_t delta = 0;
71-
if (track_open_files) {
72-
// Warmup to find the maximum delta, accounting for any asynchronous
73-
// background requests (like IAM AllowedLocations) that might or might
74-
// not have opened their sockets yet when we measure.
75-
for (int i = 0; i != 5; ++i) {
76-
auto read_client = MakeIntegrationTestClient(options);
77-
auto stream = read_client.ReadObject(bucket_name_, object_name);
78-
char c;
79-
stream.read(&c, 1);
80-
auto num_fds_during_test = GetNumOpenFiles();
81-
ASSERT_STATUS_OK(num_fds_during_test);
82-
delta = (std::max)(delta, *num_fds_during_test - *num_fds_before_test);
83-
}
84-
}
85-
8678
for (int i = 0; i != 100; ++i) {
8779
auto read_client = MakeIntegrationTestClient(options);
8880
auto stream = read_client.ReadObject(bucket_name_, object_name);
@@ -91,6 +83,9 @@ TEST_F(ObjectPlentyClientsSeriallyIntegrationTest, PlentyClientsSerially) {
9183
if (track_open_files) {
9284
auto num_fds_during_test = GetNumOpenFiles();
9385
ASSERT_STATUS_OK(num_fds_during_test);
86+
if (delta == 0) {
87+
delta = *num_fds_during_test - *num_fds_before_test;
88+
}
9489
EXPECT_GE(*num_fds_before_test + delta, *num_fds_during_test)
9590
<< "Expect each client to create the same number of file descriptors"
9691
<< ", num_fds_before_test=" << *num_fds_before_test

0 commit comments

Comments
 (0)