Skip to content

feat(storage): add resource span attributes for ACO ( App Centric Observability ) for async client #16151

New issue

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

Sign up for GitHub

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

Already on GitHub? Sign in to your account

Open
bajajneha27 wants to merge 20 commits into
base: main
Choose a base branch
from
+419 −27
Open

feat(storage): add resource span attributes for ACO ( App Centric Observability ) for async client #16151

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
20 commits
Select commit Hold shift + click to select a range
cf0f095
feat(storage): add resource span attributes ACO ( App Centric Observa…
bajajneha27 May 26, 2026
b9f230f
Adding LRU cache and span attributes to other bucket and object opera…
bajajneha27 May 27, 2026
4457605
fix ci failures
bajajneha27 Jun 2, 2026
bc689c9
fix ci failures
bajajneha27 Jun 2, 2026
7e66128
fix ci failures
bajajneha27 Jun 3, 2026
a21862f
fix ci failures
bajajneha27 Jun 3, 2026
fb735a9
fix ci failures
bajajneha27 Jun 8, 2026
872b588
fix ci failures
bajajneha27 Jun 8, 2026
5b48aef
move BucketMetadataCache to separate file
bajajneha27 Jun 9, 2026
ea26fad
more refactoring
bajajneha27 Jun 9, 2026
11ec323
more refactoring
bajajneha27 Jun 9, 2026
2fcc7d8
feat(storage): add resource span attributes for ACO ( App Centric Obs…
bajajneha27 Jun 10, 2026
e9dbaa3
fix ci failure
bajajneha27 Jun 11, 2026
e2c9822
Refactoring
bajajneha27 Jun 11, 2026
7a2cfe2
sync changes
bajajneha27 Jul 16, 2026
b67e245
refactor and remove unnecessary code
bajajneha27 Jul 16, 2026
f05f9bb
fix ci failure
bajajneha27 Jul 17, 2026
c7fa23a
rebase with main branch
bajajneha27 Jul 21, 2026
b9bd172
fix ci failure
bajajneha27 Jul 21, 2026
f18bad4
Add tests
bajajneha27 Jul 21, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
176 changes: 157 additions & 19 deletions google/cloud/storage/internal/async/connection_tracing.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,12 +18,21 @@
#include "google/cloud/storage/internal/async/reader_connection_tracing.h"
#include "google/cloud/storage/internal/async/rewriter_connection_tracing.h"
#include "google/cloud/storage/internal/async/writer_connection_tracing.h"
#include "google/cloud/storage/internal/bucket_metadata_cache.h"
#include "google/cloud/storage/options.h"
#include "google/cloud/internal/opentelemetry.h"
#include "google/cloud/version.h"
#include "google/storage/v2/storage.pb.h"
#include <algorithm>
#include <chrono>
#include <future>
#include <memory>
#include <mutex>
#include <vector>

namespace google {
namespace cloud {

namespace storage_internal {
GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN

Expand All @@ -33,28 +42,32 @@ class AsyncConnectionTracing : public storage::AsyncConnection {
public:
explicit AsyncConnectionTracing(
std::shared_ptr<storage::AsyncConnection> impl)
: impl_(std::move(impl)) {}
: impl_(std::move(impl)),
cache_(std::make_shared<BucketMetadataCache>()) {}

Options options() const override { return impl_->options(); }
~AsyncConnectionTracing() override = default;

future<StatusOr<google::storage::v2::Bucket>> GetBucket(
GetBucketParams p) override {
auto span = internal::MakeSpan("storage::AsyncConnection::GetBucket");
internal::OTelScope scope(span);
return internal::EndSpan(std::move(span), impl_->GetBucket(std::move(p)));
}
Options options() const override { return impl_->options(); }

future<StatusOr<google::storage::v2::Object>> InsertObject(
InsertObjectParams p) override {
auto span = internal::MakeSpan("storage::AsyncConnection::InsertObject");
EnrichSpan(*span, p.options,
p.request.write_object_spec().resource().bucket());
internal::OTelScope scope(span);
return internal::EndSpan(std::move(span),
impl_->InsertObject(std::move(p)));
return impl_->InsertObject(std::move(p))
.then([oc = opentelemetry::context::RuntimeContext::GetCurrent(),

@kalragauri kalragauri Jul 23, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Do we need these lines? The EndSpan template overload that takes a future parameter already handles this context detachment automatically under the hood. Same comment for ComposeObject, and DeleteObject methods.

span = std::move(span)](auto f) {
auto result = f.get();
internal::DetachOTelContext(oc);
return internal::EndSpan(*span, std::move(result));
});
}

future<StatusOr<std::shared_ptr<storage::ObjectDescriptorConnection>>> Open(
OpenParams p) override {
auto span = internal::MakeSpan("storage::AsyncConnection::Open");
EnrichSpan(*span, p.options, p.read_spec.bucket());
internal::OTelScope scope(span);
return impl_->Open(std::move(p))
.then([oc = opentelemetry::context::RuntimeContext::GetCurrent(),
Expand All @@ -74,13 +87,16 @@ class AsyncConnectionTracing : public storage::AsyncConnection {
future<StatusOr<std::unique_ptr<storage::AsyncReaderConnection>>> ReadObject(
ReadObjectParams p) override {
auto span = internal::MakeSpan("storage::AsyncConnection::ReadObject");
EnrichSpan(*span, p.options, p.request.bucket());
internal::OTelScope scope(span);
auto wrap = [oc = opentelemetry::context::RuntimeContext::GetCurrent(),
span = std::move(span)](auto f)
-> StatusOr<std::unique_ptr<storage::AsyncReaderConnection>> {
auto reader = f.get();
internal::DetachOTelContext(oc);
if (!reader) return internal::EndSpan(*span, std::move(reader).status());
if (!reader) {
return internal::EndSpan(*span, std::move(reader).status());
}
return MakeTracingReaderConnection(std::move(span), *std::move(reader));
};
return impl_->ReadObject(std::move(p)).then(std::move(wrap));
Expand All @@ -89,6 +105,7 @@ class AsyncConnectionTracing : public storage::AsyncConnection {
future<StatusOr<storage::ReadPayload>> ReadObjectRange(
ReadObjectParams p) override {
auto span = internal::MakeSpan("storage::AsyncConnection::ReadObjectRange");
EnrichSpan(*span, p.options, p.request.bucket());
internal::OTelScope scope(span);
return impl_->ReadObjectRange(std::move(p))
.then([oc = opentelemetry::context::RuntimeContext::GetCurrent(),
Expand All @@ -103,14 +120,18 @@ class AsyncConnectionTracing : public storage::AsyncConnection {
StartAppendableObjectUpload(AppendableUploadParams p) override {
auto span = internal::MakeSpan(
"storage::AsyncConnection::StartAppendableObjectUpload");
EnrichSpan(*span, p.options,
p.request.write_object_spec().resource().bucket());
internal::OTelScope scope(span);
return impl_->StartAppendableObjectUpload(std::move(p))
.then([oc = opentelemetry::context::RuntimeContext::GetCurrent(),
span = std::move(span)](auto f)
-> StatusOr<std::unique_ptr<storage::AsyncWriterConnection>> {
auto w = f.get();
internal::DetachOTelContext(oc);
if (!w) return internal::EndSpan(*span, std::move(w).status());
if (!w) {
return internal::EndSpan(*span, std::move(w).status());
}
return MakeTracingWriterConnection(span, *std::move(w));
});
}
Expand All @@ -119,14 +140,18 @@ class AsyncConnectionTracing : public storage::AsyncConnection {
ResumeAppendableObjectUpload(AppendableUploadParams p) override {
auto span = internal::MakeSpan(
"storage::AsyncConnection::ResumeAppendableObjectUpload");
EnrichSpan(*span, p.options,
p.request.write_object_spec().resource().bucket());
internal::OTelScope scope(span);
return impl_->ResumeAppendableObjectUpload(std::move(p))
.then([oc = opentelemetry::context::RuntimeContext::GetCurrent(),
span = std::move(span)](auto f)
-> StatusOr<std::unique_ptr<storage::AsyncWriterConnection>> {
auto w = f.get();
internal::DetachOTelContext(oc);
if (!w) return internal::EndSpan(*span, std::move(w).status());
if (!w) {
return internal::EndSpan(*span, std::move(w).status());
}
return MakeTracingWriterConnection(span, *std::move(w));
});
}
Expand All @@ -135,14 +160,18 @@ class AsyncConnectionTracing : public storage::AsyncConnection {
StartUnbufferedUpload(UploadParams p) override {
auto span =
internal::MakeSpan("storage::AsyncConnection::StartUnbufferedUpload");
EnrichSpan(*span, p.options,
p.request.write_object_spec().resource().bucket());
internal::OTelScope scope(span);
return impl_->StartUnbufferedUpload(std::move(p))
.then([oc = opentelemetry::context::RuntimeContext::GetCurrent(),
span = std::move(span)](auto f)
-> StatusOr<std::unique_ptr<storage::AsyncWriterConnection>> {
auto w = f.get();
internal::DetachOTelContext(oc);
if (!w) return internal::EndSpan(*span, std::move(w).status());
if (!w) {
return internal::EndSpan(*span, std::move(w).status());
}
return MakeTracingWriterConnection(span, *std::move(w));
});
}
Expand All @@ -151,14 +180,18 @@ class AsyncConnectionTracing : public storage::AsyncConnection {
StartBufferedUpload(UploadParams p) override {
auto span =
internal::MakeSpan("storage::AsyncConnection::StartBufferedUpload");
EnrichSpan(*span, p.options,
p.request.write_object_spec().resource().bucket());
internal::OTelScope scope(span);
return impl_->StartBufferedUpload(std::move(p))
.then([oc = opentelemetry::context::RuntimeContext::GetCurrent(),
span = std::move(span)](auto f)
-> StatusOr<std::unique_ptr<storage::AsyncWriterConnection>> {
auto w = f.get();
internal::DetachOTelContext(oc);
if (!w) return internal::EndSpan(*span, std::move(w).status());
if (!w) {
return internal::EndSpan(*span, std::move(w).status());
}
return MakeTracingWriterConnection(span, *std::move(w));
});
}
Expand Down Expand Up @@ -198,16 +231,28 @@ class AsyncConnectionTracing : public storage::AsyncConnection {
future<StatusOr<google::storage::v2::Object>> ComposeObject(
ComposeObjectParams p) override {
auto span = internal::MakeSpan("storage::AsyncConnection::ComposeObject");
EnrichSpan(*span, p.options, p.request.destination().bucket());
internal::OTelScope scope(span);
return internal::EndSpan(std::move(span),
impl_->ComposeObject(std::move(p)));
return impl_->ComposeObject(std::move(p))
.then([oc = opentelemetry::context::RuntimeContext::GetCurrent(),
span = std::move(span)](auto f) {
auto result = f.get();
internal::DetachOTelContext(oc);
return internal::EndSpan(*span, std::move(result));
});
}

future<Status> DeleteObject(DeleteObjectParams p) override {
auto span = internal::MakeSpan("storage::AsyncConnection::DeleteObject");
EnrichSpan(*span, p.options, p.request.bucket());
internal::OTelScope scope(span);
return internal::EndSpan(std::move(span),
impl_->DeleteObject(std::move(p)));
return impl_->DeleteObject(std::move(p))
.then([oc = opentelemetry::context::RuntimeContext::GetCurrent(),
span = std::move(span)](auto f) {
auto result = f.get();
internal::DetachOTelContext(oc);
return internal::EndSpan(*span, std::move(result));
});
}

std::shared_ptr<storage::AsyncRewriterConnection> RewriteObject(
Expand All @@ -217,8 +262,101 @@ class AsyncConnectionTracing : public storage::AsyncConnection {
impl_->RewriteObject(std::move(p)), enabled);
}

future<StatusOr<google::storage::v2::Bucket>> GetBucket(
GetBucketParams p) override {
auto span = internal::MakeSpan("storage::AsyncConnection::GetBucket");
internal::OTelScope scope(span);
auto const bucket_name = p.request.name();
auto const options = p.options;
return impl_->GetBucket(std::move(p))
.then([oc = opentelemetry::context::RuntimeContext::GetCurrent(),
span = std::move(span), this, bucket_name,
options](future<StatusOr<google::storage::v2::Bucket>> f)
-> StatusOr<google::storage::v2::Bucket> {
auto result = f.get();
internal::DetachOTelContext(oc);
if (result.ok()) {

@kalragauri kalragauri Jul 23, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If the AsyncConnectionTracing object is destroyed before the future resolves, invoking cache().MaybeInvalidate(...) or calling EnrichSpan member method inside the callback will result in a use-after-free crash.

EnrichSpan(*span, options, *result, bucket_name);
} else {
cache().MaybeInvalidate(result, bucket_name);
}
return internal::EndSpan(*span, std::move(result));
});
}

private:
BucketMetadataCache& cache() const { return *cache_; }

void MaybeTriggerBackgroundFetch(Options const& options,
std::string const& bucket_name) {
if (!cache().StartFetch(bucket_name)) {
return;
}

auto guard = ScopedFetch(cache_, bucket_name);
google::storage::v2::GetBucketRequest request;
auto const normalized_bucket_name =
BucketMetadataCache::NormalizeBucketName(bucket_name);
request.set_name("projects/_/buckets/" + normalized_bucket_name);
GetBucketParams params{std::move(request), options};

impl_->GetBucket(std::move(params))
.then([cache = cache_, bucket_name, guard = std::move(guard)](
future<StatusOr<google::storage::v2::Bucket>> f) {
auto metadata = f.get();
if (metadata.ok()) {
BucketCacheEntry entry = BucketCacheEntry::FromLocation(
metadata->project() + "/buckets/" +
BucketMetadataCache::NormalizeBucketName(bucket_name),
metadata->location(), metadata->location_type());
cache->Put(bucket_name, std::move(entry));
} else if (metadata.status().code() ==
StatusCode::kPermissionDenied) {
BucketCacheEntry entry{
"projects/_/buckets/" +
BucketMetadataCache::NormalizeBucketName(bucket_name),
"global"};
cache->Put(bucket_name, std::move(entry));
}
});
}

static void EnrichSpan(opentelemetry::trace::Span& span,
BucketCacheEntry const& entry) {
span.SetAttribute("gcp.resource.destination.id", entry.id);
span.SetAttribute("gcp.resource.destination.location", entry.location);
}

void EnrichSpan(opentelemetry::trace::Span& span, Options const& options,
google::storage::v2::Bucket const& bucket,
std::string const& bucket_name) {
auto const enabled = options.get<
google::cloud::storage_experimental::OTelSpanEnrichmentOption>();
if (!enabled) return;
auto entry = BucketCacheEntry::FromLocation(
bucket.project() + "/buckets/" +
BucketMetadataCache::NormalizeBucketName(bucket_name),
bucket.location(), bucket.location_type());
EnrichSpan(span, entry);
cache().Put(bucket_name, std::move(entry));
}

void EnrichSpan(opentelemetry::trace::Span& span, Options const& options,
std::string const& bucket_name) {
if (bucket_name.empty()) return;
auto const enabled = options.get<
google::cloud::storage_experimental::OTelSpanEnrichmentOption>();
if (!enabled) return;
auto entry = cache().Get(bucket_name);
if (entry.has_value()) {
EnrichSpan(span, *entry);
} else {
MaybeTriggerBackgroundFetch(options, bucket_name);
}
}

std::shared_ptr<storage::AsyncConnection> impl_;
std::shared_ptr<BucketMetadataCache> cache_;
};

} // namespace
Expand Down
Loading
Loading