Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions google/cloud/storage/google_cloud_cpp_storage_grpc.bzl
Original file line number Diff line number Diff line change
Expand Up @@ -58,6 +58,7 @@ google_cloud_cpp_storage_grpc_hdrs = [
"internal/async/read_range.h",
"internal/async/reader_connection_factory.h",
"internal/async/reader_connection_impl.h",
"internal/async/reader_connection_telemetry.h",
"internal/async/reader_connection_resume.h",
"internal/async/reader_connection_tracing.h",
"internal/async/rewriter_connection_impl.h",
Expand Down Expand Up @@ -135,6 +136,7 @@ google_cloud_cpp_storage_grpc_srcs = [
"internal/async/read_range.cc",
"internal/async/reader_connection_factory.cc",
"internal/async/reader_connection_impl.cc",
"internal/async/reader_connection_telemetry.cc",
"internal/async/reader_connection_resume.cc",
"internal/async/reader_connection_tracing.cc",
"internal/async/rewriter_connection_impl.cc",
Expand Down
2 changes: 2 additions & 0 deletions google/cloud/storage/google_cloud_cpp_storage_grpc.cmake
Original file line number Diff line number Diff line change
Expand Up @@ -137,6 +137,8 @@ add_library(
internal/async/reader_connection_factory.h
internal/async/reader_connection_impl.cc
internal/async/reader_connection_impl.h
internal/async/reader_connection_telemetry.cc
internal/async/reader_connection_telemetry.h
internal/async/reader_connection_resume.cc
internal/async/reader_connection_resume.h
internal/async/reader_connection_tracing.cc
Expand Down
32 changes: 16 additions & 16 deletions google/cloud/storage/internal/async/connection_tracing.cc
Original file line number Diff line number Diff line change
Expand Up @@ -56,32 +56,32 @@ class AsyncConnectionTracing : public storage::AsyncConnection {
OpenParams p) override {
auto span = internal::MakeSpan("storage::AsyncConnection::Open");
internal::OTelScope scope(span);
return impl_->Open(std::move(p))
.then([oc = opentelemetry::context::RuntimeContext::GetCurrent(),
span = std::move(span)](auto f)
-> StatusOr<
std::shared_ptr<storage::ObjectDescriptorConnection>> {
auto result = f.get();
internal::DetachOTelContext(oc);
if (!result) {
return internal::EndSpan(*span, std::move(result).status());
}
return MakeTracingObjectDescriptorConnection(std::move(span),
*std::move(result));
});
auto wrap = [oc = opentelemetry::context::RuntimeContext::GetCurrent(),
bucket = p.read_spec.bucket(), span = std::move(span)](auto f)
-> StatusOr<std::shared_ptr<storage::ObjectDescriptorConnection>> {
StatusOr<std::shared_ptr<storage::ObjectDescriptorConnection>> result =
f.get();
internal::DetachOTelContext(oc);
if (!result) return internal::EndSpan(*span, std::move(result).status());
return MakeTracingObjectDescriptorConnection(
std::move(span), *std::move(result), std::move(bucket));
};
return impl_->Open(std::move(p)).then(std::move(wrap));
}

future<StatusOr<std::unique_ptr<storage::AsyncReaderConnection>>> ReadObject(
ReadObjectParams p) override {
auto span = internal::MakeSpan("storage::AsyncConnection::ReadObject");
internal::OTelScope scope(span);
auto wrap = [oc = opentelemetry::context::RuntimeContext::GetCurrent(),
span = std::move(span)](auto f)
bucket = p.request.bucket(), span = std::move(span)](auto f)
-> StatusOr<std::unique_ptr<storage::AsyncReaderConnection>> {
auto reader = f.get();
StatusOr<std::unique_ptr<storage::AsyncReaderConnection>> reader =
f.get();
internal::DetachOTelContext(oc);
if (!reader) return internal::EndSpan(*span, std::move(reader).status());
return MakeTracingReaderConnection(std::move(span), *std::move(reader));
return MakeTracingReaderConnection(std::move(span), *std::move(reader),
std::move(bucket));
};
return impl_->ReadObject(std::move(p)).then(std::move(wrap));
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -34,8 +34,11 @@ class AsyncObjectDescriptorConnectionTracing
public:
explicit AsyncObjectDescriptorConnectionTracing(
opentelemetry::nostd::shared_ptr<opentelemetry::trace::Span> span,
std::shared_ptr<storage::ObjectDescriptorConnection> impl)
: span_(std::move(span)), impl_(std::move(impl)) {}
std::shared_ptr<storage::ObjectDescriptorConnection> impl,
std::string bucket_name)
: span_(std::move(span)),
impl_(std::move(impl)),
bucket_name_(std::move(bucket_name)) {}

~AsyncObjectDescriptorConnectionTracing() override {
internal::EndSpan(*span_);
Expand All @@ -55,7 +58,7 @@ class AsyncObjectDescriptorConnectionTracing
{{sc::thread::kThreadId, internal::CurrentThreadId()},
{"read-start", p.start},
{"read-length", p.length}});
return MakeTracingReaderConnection(span_, std::move(result));
return MakeTracingReaderConnection(span_, std::move(result), bucket_name_);
}

void MakeSubsequentStream() override {
Expand All @@ -65,16 +68,18 @@ class AsyncObjectDescriptorConnectionTracing
private:
opentelemetry::nostd::shared_ptr<opentelemetry::trace::Span> span_;
std::shared_ptr<storage::ObjectDescriptorConnection> impl_;
std::string bucket_name_;
};

} // namespace

std::shared_ptr<storage::ObjectDescriptorConnection>
MakeTracingObjectDescriptorConnection(
opentelemetry::nostd::shared_ptr<opentelemetry::trace::Span> span,
std::shared_ptr<storage::ObjectDescriptorConnection> impl) {
std::shared_ptr<storage::ObjectDescriptorConnection> impl,
std::string bucket_name) {
return std::make_unique<AsyncObjectDescriptorConnectionTracing>(
std::move(span), std::move(impl));
std::move(span), std::move(impl), std::move(bucket_name));
}

GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,8 @@ GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
std::shared_ptr<storage::ObjectDescriptorConnection>
MakeTracingObjectDescriptorConnection(
opentelemetry::nostd::shared_ptr<opentelemetry::trace::Span> span,
std::shared_ptr<storage::ObjectDescriptorConnection> impl);
std::shared_ptr<storage::ObjectDescriptorConnection> impl,
std::string bucket_name);

GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
} // namespace storage_internal
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -79,7 +79,7 @@ TEST(ObjectDescriptorConnectionTracing, Read) {
return std::make_unique<MockAsyncReaderConnection>();
});
auto actual = MakeTracingObjectDescriptorConnection(
internal::MakeSpan("test-span-name"), std::move(mock));
internal::MakeSpan("test-span-name"), std::move(mock), "test-bucket");
auto f1 = actual->Read(ObjectDescriptorConnection::ReadParams{100, 200});

actual.reset();
Expand Down Expand Up @@ -114,7 +114,8 @@ TEST(ObjectDescriptorConnectionTracing, ReadThenRead) {
});

auto connection = MakeTracingObjectDescriptorConnection(
internal::MakeSpan("test-span"), std::move(mock_connection));
internal::MakeSpan("test-span"), std::move(mock_connection),
"test-bucket");

auto reader = connection->Read({});
auto f = reader->Read().then(expect_no_context);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -201,7 +201,8 @@ std::unique_ptr<storage::AsyncReaderConnection> ObjectDescriptorImpl::Read(
return std::unique_ptr<storage::AsyncReaderConnection>(
std::make_unique<ObjectDescriptorReader>(std::move(range)));
}
return MakeTracingObjectDescriptorReader(std::move(range));
return MakeTracingObjectDescriptorReader(std::move(range),
read_object_spec_.bucket());
}

auto it = stream_manager_->GetLeastBusyStream();
Expand All @@ -217,8 +218,8 @@ std::unique_ptr<storage::AsyncReaderConnection> ObjectDescriptorImpl::Read(
return std::unique_ptr<storage::AsyncReaderConnection>(
std::make_unique<ObjectDescriptorReader>(std::move(range)));
}

return MakeTracingObjectDescriptorReader(std::move(range));
return MakeTracingObjectDescriptorReader(std::move(range),
read_object_spec_.bucket());
}

std::shared_ptr<storage::internal::HashFunction>
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -15,9 +15,11 @@
#include "google/cloud/storage/internal/async/object_descriptor_reader_tracing.h"
#include "google/cloud/storage/async/reader_connection.h"
#include "google/cloud/storage/internal/async/object_descriptor_reader.h"
#include "google/cloud/storage/internal/async/reader_connection_telemetry.h"
#include "google/cloud/internal/opentelemetry.h"
#include "google/cloud/version.h"
#include <opentelemetry/semconv/incubating/thread_attributes.h>
#include <chrono>
#include <memory>

namespace google {
Expand All @@ -31,8 +33,11 @@ namespace sc = ::opentelemetry::semconv;

class ObjectDescriptorReaderTracing : public ObjectDescriptorReader {
public:
explicit ObjectDescriptorReaderTracing(std::shared_ptr<ReadRange> impl)
: ObjectDescriptorReader(std::move(impl)) {}
explicit ObjectDescriptorReaderTracing(std::shared_ptr<ReadRange> impl,
std::string bucket_name)
: ObjectDescriptorReader(std::move(impl)),
bucket_name_(
std::make_shared<std::string const>(std::move(bucket_name))) {}

~ObjectDescriptorReaderTracing() override = default;

Expand All @@ -41,18 +46,21 @@ class ObjectDescriptorReaderTracing : public ObjectDescriptorReader {
internal::OTelScope scope(span);
return ObjectDescriptorReader::Read().then(
[span = std::move(span),
oc = opentelemetry::context::RuntimeContext::GetCurrent()](
auto f) -> ReadResponse {
auto result = f.get();
oc = opentelemetry::context::RuntimeContext::GetCurrent(),
bucket_name = bucket_name_,
metrics = metrics_](auto f) -> ReadResponse {
ReadResponse result = f.get();
internal::DetachOTelContext(oc);
if (!absl::holds_alternative<Status>(result)) {
auto const& payload = absl::get<storage::ReadPayload>(result);

if (auto const* payload =
absl::get_if<storage::ReadPayload>(&result)) {
span->AddEvent(
"gl-cpp.read-range",
{{/*sc::kRpcMessageType=*/"rpc.message.type", "RECEIVED"},
{sc::thread::kThreadId, internal::CurrentThreadId()},
{"message.size", static_cast<std::uint32_t>(payload.size())}});
{"message.size",
static_cast<std::uint32_t>(payload->size())}});
metrics.RecordRead(*payload, std::chrono::steady_clock::now(),
*bucket_name, span, "gl-cpp.latency.read-range");
} else {
span->AddEvent(
"gl-cpp.read-range",
Expand All @@ -64,13 +72,19 @@ class ObjectDescriptorReaderTracing : public ObjectDescriptorReader {
return result;
});
}

private:
std::shared_ptr<std::string const> bucket_name_;
ReaderConnectionTelemetry metrics_;
};

} // namespace

std::unique_ptr<storage::AsyncReaderConnection>
MakeTracingObjectDescriptorReader(std::shared_ptr<ReadRange> impl) {
return std::make_unique<ObjectDescriptorReaderTracing>(std::move(impl));
MakeTracingObjectDescriptorReader(std::shared_ptr<ReadRange> impl,
std::string bucket_name) {
return std::make_unique<ObjectDescriptorReaderTracing>(
std::move(impl), std::move(bucket_name));
}

GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,8 @@ namespace storage_internal {
GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN

std::unique_ptr<storage::AsyncReaderConnection>
MakeTracingObjectDescriptorReader(std::shared_ptr<ReadRange> impl);
MakeTracingObjectDescriptorReader(std::shared_ptr<ReadRange> impl,
std::string bucket_name);

GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
} // namespace storage_internal
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -45,7 +45,7 @@ TEST(ObjectDescriptorReaderTracing, Read) {
auto span_catcher = InstallSpanCatcher();

auto impl = std::make_shared<ReadRange>(10000, 30);
auto reader = MakeTracingObjectDescriptorReader(impl);
auto reader = MakeTracingObjectDescriptorReader(impl, "test-bucket");

auto data = google::storage::v2::ObjectRangeData{};
auto constexpr kData0 = R"pb(
Expand Down Expand Up @@ -73,7 +73,7 @@ TEST(ObjectDescriptorReaderTracing, Read) {
TEST(ObjectDescriptorReaderTracing, ReadError) {
auto span_catcher = InstallSpanCatcher();
auto impl = std::make_shared<ReadRange>(10000, 30);
auto reader = MakeTracingObjectDescriptorReader(impl);
auto reader = MakeTracingObjectDescriptorReader(impl, "test-bucket");

impl->OnFinish(PermanentError());

Expand Down
4 changes: 2 additions & 2 deletions google/cloud/storage/internal/async/read_range.h
Original file line number Diff line number Diff line change
Expand Up @@ -87,11 +87,11 @@ class ReadRange {
#if defined(GOOGLE_CLOUD_CPP_STORAGE_WITH_OTEL_METRICS) || \
defined(GOOGLE_CLOUD_CPP_HAVE_OPENTELEMETRY)
void SetT4(std::chrono::steady_clock::time_point t) {
std::unique_lock<std::mutex> lk(self->mu_);
std::unique_lock<std::mutex> lk(mu_);
t4_ = t;
}
void SetT5(std::chrono::steady_clock::time_point t) {
std::unique_lock<std::mutex> lk(self->mu_);
std::unique_lock<std::mutex> lk(mu_);
t5_ = t;
}
#else
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,99 @@
// Copyright 2026 Google LLC
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// https://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

#include "google/cloud/storage/internal/async/reader_connection_telemetry.h"
#include "google/cloud/storage/internal/async/read_payload_impl.h"
#include "google/cloud/internal/opentelemetry.h"
#include "google/cloud/version.h"
#ifdef GOOGLE_CLOUD_CPP_STORAGE_WITH_OTEL_METRICS
#include <opentelemetry/metrics/meter_provider.h>
#include <opentelemetry/metrics/provider.h>
#endif
#include <chrono>

namespace google {
namespace cloud {
namespace storage_internal {
GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN

#ifdef GOOGLE_CLOUD_CPP_STORAGE_WITH_OTEL_METRICS
ReaderConnectionTelemetry::ReaderConnectionTelemetry() {
opentelemetry::nostd::shared_ptr<opentelemetry::metrics::Meter> meter =
opentelemetry::metrics::Provider::GetMeterProvider()->GetMeter(
"google-cloud-cpp", version::version_string());
metrics_ = {
meter->CreateDoubleHistogram("gl-cpp.latency.bidi_read.queue",
"Read Range Queue Latency", "us"),
meter->CreateDoubleHistogram("gl-cpp.latency.bidi_read.network",
"Read Range Network Latency", "us"),
meter->CreateDoubleHistogram("gl-cpp.latency.bidi_read.internal",
"Read Range Internal Overhead", "us"),
};
}

void ReaderConnectionTelemetry::RecordMetrics(std::string const& bucket_name,
double p1, double p2,
double p3) const {
if (metrics_.queue_hist)
metrics_.queue_hist->Record(p1, {{"gcp.storage.bucket", bucket_name}},
opentelemetry::context::Context{});
if (metrics_.network_hist)
metrics_.network_hist->Record(p2, {{"gcp.storage.bucket", bucket_name}},
opentelemetry::context::Context{});
if (metrics_.output_hist)
metrics_.output_hist->Record(p3, {{"gcp.storage.bucket", bucket_name}},
opentelemetry::context::Context{});
}
#else
ReaderConnectionTelemetry::ReaderConnectionTelemetry() = default;
#endif

void ReaderConnectionTelemetry::RecordRead(
storage::ReadPayload const& payload,
std::chrono::steady_clock::time_point t7, std::string const& bucket_name,
opentelemetry::nostd::shared_ptr<opentelemetry::trace::Span> const& span,
std::string_view event_name) const {
std::chrono::steady_clock::time_point t4 = ReadPayloadImpl::GetT4(payload);
std::chrono::steady_clock::time_point t5 = ReadPayloadImpl::GetT5(payload);
std::chrono::steady_clock::time_point t6 = ReadPayloadImpl::GetT6(payload);

if (t4 != std::chrono::steady_clock::time_point{} && t4 <= t5 && t5 <= t6 &&
t6 <= t7) {
double p1 = std::chrono::duration<double, std::micro>(t5 - t4).count();
double p2 = std::chrono::duration<double, std::micro>(t6 - t5).count();
double p3 = std::chrono::duration<double, std::micro>(t7 - t6).count();

#ifdef GOOGLE_CLOUD_CPP_STORAGE_WITH_OTEL_METRICS
RecordMetrics(bucket_name, p1, p2, p3);
#else
(void)bucket_name;
(void)p1;
(void)p2;
(void)p3;
#endif

if (span && span->GetContext().IsValid()) {
span->AddEvent(opentelemetry::nostd::string_view{event_name.data(),
event_name.size()},
{{"gl-cpp.latency.queue", p1},
{"gl-cpp.latency.network", p2},
{"gl-cpp.latency.internal", p3}});
}
}
}

GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
} // namespace storage_internal
} // namespace cloud
} // namespace google
Loading