diff --git a/google/cloud/storage/google_cloud_cpp_storage_grpc.bzl b/google/cloud/storage/google_cloud_cpp_storage_grpc.bzl index ad210011ea06f..13841ceacfb4d 100644 --- a/google/cloud/storage/google_cloud_cpp_storage_grpc.bzl +++ b/google/cloud/storage/google_cloud_cpp_storage_grpc.bzl @@ -50,6 +50,7 @@ google_cloud_cpp_storage_grpc_hdrs = [ "internal/async/object_descriptor_reader.h", "internal/async/object_descriptor_reader_tracing.h", "internal/async/open_object.h", + "internal/async/open_object_telemetry.h", "internal/async/open_stream.h", "internal/async/partial_upload.h", "internal/async/read_payload_fwd.h", @@ -128,6 +129,7 @@ google_cloud_cpp_storage_grpc_srcs = [ "internal/async/object_descriptor_reader.cc", "internal/async/object_descriptor_reader_tracing.cc", "internal/async/open_object.cc", + "internal/async/open_object_telemetry.cc", "internal/async/open_stream.cc", "internal/async/partial_upload.cc", "internal/async/read_range.cc", diff --git a/google/cloud/storage/google_cloud_cpp_storage_grpc.cmake b/google/cloud/storage/google_cloud_cpp_storage_grpc.cmake index 952d6dbef46d2..4e194e80a4345 100644 --- a/google/cloud/storage/google_cloud_cpp_storage_grpc.cmake +++ b/google/cloud/storage/google_cloud_cpp_storage_grpc.cmake @@ -123,6 +123,8 @@ add_library( internal/async/object_descriptor_reader_tracing.h internal/async/open_object.cc internal/async/open_object.h + internal/async/open_object_telemetry.cc + internal/async/open_object_telemetry.h internal/async/open_stream.cc internal/async/open_stream.h internal/async/partial_upload.cc @@ -243,18 +245,25 @@ target_include_directories( target_compile_options(google_cloud_cpp_storage_grpc PUBLIC ${GOOGLE_CLOUD_CPP_EXCEPTIONS_FLAG}) target_compile_definitions(google_cloud_cpp_storage_grpc - PUBLIC GOOGLE_CLOUD_CPP_STORAGE_HAVE_GRPC) + PUBLIC GOOGLE_CLOUD_CPP_STORAGE_HAVE_GRPC + GOOGLE_CLOUD_CPP_HAVE_OPENTELEMETRY) if (GOOGLE_CLOUD_CPP_ENABLE_CTYPE_CORD_WORKAROUND) target_compile_definitions( google_cloud_cpp_storage_grpc PRIVATE GOOGLE_CLOUD_CPP_ENABLE_CTYPE_CORD_WORKAROUND) endif () +if (GOOGLE_CLOUD_CPP_HAVE_OPENTELEMETRY) + target_compile_definitions( + google_cloud_cpp_storage_grpc + PRIVATE GOOGLE_CLOUD_CPP_HAVE_OPENTELEMETRY) +endif () if ((TARGET gRPC::grpcpp_otel_plugin) AND (TARGET google-cloud-cpp::opentelemetry) AND (TARGET opentelemetry-cpp::metrics)) target_compile_definitions( google_cloud_cpp_storage_grpc - PRIVATE GOOGLE_CLOUD_CPP_STORAGE_WITH_OTEL_METRICS) + PRIVATE GOOGLE_CLOUD_CPP_STORAGE_WITH_OTEL_METRICS + GOOGLE_CLOUD_CPP_HAVE_OPENTELEMETRY) target_link_libraries( google_cloud_cpp_storage_grpc PUBLIC google-cloud-cpp::opentelemetry gRPC::grpcpp_otel_plugin diff --git a/google/cloud/storage/internal/async/open_object.cc b/google/cloud/storage/internal/async/open_object.cc index 1f3b87d805e3f..f1368cc41e83c 100644 --- a/google/cloud/storage/internal/async/open_object.cc +++ b/google/cloud/storage/internal/async/open_object.cc @@ -15,6 +15,11 @@ #include "google/cloud/storage/internal/async/open_object.h" #include "google/cloud/internal/make_status.h" #include "absl/strings/str_cat.h" +#if defined(GOOGLE_CLOUD_CPP_STORAGE_WITH_OTEL_METRICS) || \ + defined(GOOGLE_CLOUD_CPP_HAVE_OPENTELEMETRY) +#include "google/cloud/internal/opentelemetry.h" +#include +#endif #include namespace google { @@ -41,6 +46,10 @@ OpenObject::OpenObject(storage_internal::StorageStub& stub, CompletionQueue& cq, initial_request_(std::move(request)) {} future> OpenObject::Call() { + metrics_.RecordCall(); +#ifdef GOOGLE_CLOUD_CPP_HAVE_OPENTELEMETRY + span_ = opentelemetry::trace::Tracer::GetCurrentSpan(); +#endif auto future = promise_.get_future(); rpc_->Start().then([w = WeakFromThis()](auto f) { if (auto self = w.lock()) self->OnStart(f.get()); @@ -63,6 +72,7 @@ std::unique_ptr OpenObject::CreateRpc( } void OpenObject::OnStart(bool ok) { + metrics_.RecordStart(); if (!ok) return DoFinish(); rpc_->Write(initial_request_).then([w = WeakFromThis()](auto f) { if (auto self = w.lock()) self->OnWrite(f.get()); @@ -70,6 +80,7 @@ void OpenObject::OnStart(bool ok) { } void OpenObject::OnWrite(bool ok) { + metrics_.RecordWrite(); if (!ok) return DoFinish(); rpc_->Read().then([w = WeakFromThis()](auto f) { if (auto self = w.lock()) self->OnRead(f.get()); @@ -78,6 +89,12 @@ void OpenObject::OnWrite(bool ok) { void OpenObject::OnRead( std::optional response) { +#ifdef GOOGLE_CLOUD_CPP_HAVE_OPENTELEMETRY + metrics_.RecordRead(initial_request_.read_object_spec().bucket(), span_); +#else + metrics_.RecordRead(initial_request_.read_object_spec().bucket()); +#endif + if (!response) return DoFinish(); promise_.set_value(OpenStreamResult{std::move(rpc_), std::move(*response)}); } diff --git a/google/cloud/storage/internal/async/open_object.h b/google/cloud/storage/internal/async/open_object.h index 65f01dba4024c..a5002b4b11079 100644 --- a/google/cloud/storage/internal/async/open_object.h +++ b/google/cloud/storage/internal/async/open_object.h @@ -15,6 +15,7 @@ #ifndef GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_STORAGE_INTERNAL_ASYNC_OPEN_OBJECT_H #define GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_STORAGE_INTERNAL_ASYNC_OPEN_OBJECT_H +#include "google/cloud/storage/internal/async/open_object_telemetry.h" #include "google/cloud/storage/internal/async/open_stream.h" #include "google/cloud/storage/internal/storage_stub.h" #include "google/cloud/completion_queue.h" @@ -25,6 +26,7 @@ #include "google/cloud/version.h" #include "google/storage/v2/storage.pb.h" #include +#include #include #include @@ -107,6 +109,10 @@ class OpenObject : public std::enable_shared_from_this { std::shared_ptr rpc_; promise> promise_; google::storage::v2::BidiReadObjectRequest initial_request_; +#ifdef GOOGLE_CLOUD_CPP_HAVE_OPENTELEMETRY + opentelemetry::nostd::shared_ptr span_; +#endif + OpenObjectTelemetry metrics_; }; GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END diff --git a/google/cloud/storage/internal/async/open_object_telemetry.cc b/google/cloud/storage/internal/async/open_object_telemetry.cc new file mode 100644 index 0000000000000..5656a328a81ab --- /dev/null +++ b/google/cloud/storage/internal/async/open_object_telemetry.cc @@ -0,0 +1,140 @@ +// 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/open_object_telemetry.h" + +#ifdef GOOGLE_CLOUD_CPP_STORAGE_WITH_OTEL_METRICS +#include +#endif + +namespace google { +namespace cloud { +namespace storage_internal { +GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN + +#ifdef GOOGLE_CLOUD_CPP_STORAGE_WITH_OTEL_METRICS +namespace { +struct StreamOpenMetrics { + opentelemetry::nostd::shared_ptr> + network_handshake; + opentelemetry::nostd::shared_ptr> + server_metadata_latency; + opentelemetry::nostd::shared_ptr> + stream_open_latency; + + static StreamOpenMetrics const& Instance() { + static auto const metrics = [] { + auto meter = + opentelemetry::metrics::Provider::GetMeterProvider()->GetMeter( + "storage", "v1"); + return StreamOpenMetrics{ + meter->CreateDoubleHistogram("gl-cpp.latency.network_handshake", + "Network Handshake", "us"), + meter->CreateDoubleHistogram("gl-cpp.latency.server_metadata", + "Server Metadata Latency", "us"), + meter->CreateDoubleHistogram("gl-cpp.latency.stream_open", + "End-to-End Stream Open", "us")}; + }(); + return metrics; + } +}; +} // namespace +#endif + +void OpenObjectTelemetry::RecordCall() { +#if defined(GOOGLE_CLOUD_CPP_STORAGE_WITH_OTEL_METRICS) || \ + defined(GOOGLE_CLOUD_CPP_HAVE_OPENTELEMETRY) + t0_ = std::chrono::steady_clock::now(); +#endif +} + +void OpenObjectTelemetry::RecordStart() { +#if defined(GOOGLE_CLOUD_CPP_STORAGE_WITH_OTEL_METRICS) || \ + defined(GOOGLE_CLOUD_CPP_HAVE_OPENTELEMETRY) + t1_ = std::chrono::steady_clock::now(); +#endif +} + +void OpenObjectTelemetry::RecordWrite() { +#if defined(GOOGLE_CLOUD_CPP_STORAGE_WITH_OTEL_METRICS) || \ + defined(GOOGLE_CLOUD_CPP_HAVE_OPENTELEMETRY) + t2_ = std::chrono::steady_clock::now(); +#endif +} + +#ifdef GOOGLE_CLOUD_CPP_STORAGE_WITH_OTEL_METRICS +void OpenObjectTelemetry::RecordMetrics( + std::string const& bucket, std::chrono::steady_clock::time_point t3) { + auto const& metrics = StreamOpenMetrics::Instance(); + auto p1 = static_cast( + std::chrono::duration_cast(t1_ - t0_).count()); + auto p2 = static_cast( + std::chrono::duration_cast(t3 - t2_).count()); + auto p3 = static_cast( + std::chrono::duration_cast(t3 - t0_).count()); + + if (metrics.network_handshake) + metrics.network_handshake->Record(p1, {{"gcp.storage.bucket", bucket}}, + opentelemetry::context::Context{}); + if (metrics.server_metadata_latency) + metrics.server_metadata_latency->Record(p2, + {{"gcp.storage.bucket", bucket}}, + opentelemetry::context::Context{}); + if (metrics.stream_open_latency) + metrics.stream_open_latency->Record(p3, {{"gcp.storage.bucket", bucket}}, + opentelemetry::context::Context{}); +} +#endif + +#ifdef GOOGLE_CLOUD_CPP_HAVE_OPENTELEMETRY +void OpenObjectTelemetry::RecordRead( + std::string const& bucket, + opentelemetry::nostd::shared_ptr const& span) { + auto t3 = std::chrono::steady_clock::now(); +#ifdef GOOGLE_CLOUD_CPP_STORAGE_WITH_OTEL_METRICS + RecordMetrics(bucket, t3); +#else + (void)bucket; +#endif + + if (span && span->GetContext().IsValid()) { + auto p1 = static_cast( + std::chrono::duration_cast(t1_ - t0_) + .count()); + auto p2 = static_cast( + std::chrono::duration_cast(t3 - t2_) + .count()); + auto p3 = static_cast( + std::chrono::duration_cast(t3 - t0_) + .count()); + span->AddEvent("gl-cpp.stream_open.latency", + {{"gl-cpp.latency.network_handshake", p1}, + {"gl-cpp.latency.server_metadata", p2}, + {"gl-cpp.latency.stream_open", p3}}); + } +} +#else +void OpenObjectTelemetry::RecordRead(std::string const& bucket) { +#ifdef GOOGLE_CLOUD_CPP_STORAGE_WITH_OTEL_METRICS + RecordMetrics(bucket, std::chrono::steady_clock::now()); +#else + (void)bucket; +#endif +} +#endif + +GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END +} // namespace storage_internal +} // namespace cloud +} // namespace google diff --git a/google/cloud/storage/internal/async/open_object_telemetry.h b/google/cloud/storage/internal/async/open_object_telemetry.h new file mode 100644 index 0000000000000..943c1843d1bfd --- /dev/null +++ b/google/cloud/storage/internal/async/open_object_telemetry.h @@ -0,0 +1,64 @@ +// 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. + +#ifndef GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_STORAGE_INTERNAL_ASYNC_OPEN_OBJECT_TELEMETRY_H +#define GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_STORAGE_INTERNAL_ASYNC_OPEN_OBJECT_TELEMETRY_H + +#include "google/cloud/version.h" +#include +#include + +#ifdef GOOGLE_CLOUD_CPP_HAVE_OPENTELEMETRY +#include +#endif + +namespace google { +namespace cloud { +namespace storage_internal { +GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN + +class OpenObjectTelemetry { + public: + void RecordCall(); + void RecordStart(); + void RecordWrite(); + +#ifdef GOOGLE_CLOUD_CPP_HAVE_OPENTELEMETRY + void RecordRead( + std::string const& bucket, + opentelemetry::nostd::shared_ptr const& span); +#else + void RecordRead(std::string const& bucket); +#endif + + private: +#ifdef GOOGLE_CLOUD_CPP_STORAGE_WITH_OTEL_METRICS + void RecordMetrics(std::string const& bucket, + std::chrono::steady_clock::time_point t3); +#endif + +#if defined(GOOGLE_CLOUD_CPP_STORAGE_WITH_OTEL_METRICS) || \ + defined(GOOGLE_CLOUD_CPP_HAVE_OPENTELEMETRY) + std::chrono::steady_clock::time_point t0_; + std::chrono::steady_clock::time_point t1_; + std::chrono::steady_clock::time_point t2_; +#endif +}; + +GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END +} // namespace storage_internal +} // namespace cloud +} // namespace google + +#endif // GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_STORAGE_INTERNAL_ASYNC_OPEN_OBJECT_TELEMETRY_H diff --git a/google/cloud/storage/internal/async/open_object_test.cc b/google/cloud/storage/internal/async/open_object_test.cc index 265cfdfd5ad9e..81959a62dbe1d 100644 --- a/google/cloud/storage/internal/async/open_object_test.cc +++ b/google/cloud/storage/internal/async/open_object_test.cc @@ -24,6 +24,12 @@ #include #include +#ifdef GOOGLE_CLOUD_CPP_HAVE_OPENTELEMETRY +#include "google/cloud/internal/opentelemetry.h" +#include "google/cloud/opentelemetry_options.h" +#include "google/cloud/testing_util/opentelemetry_matchers.h" +#endif + namespace google { namespace cloud { namespace storage_internal { @@ -43,6 +49,17 @@ using ::testing::Field; using ::testing::NotNull; using ::testing::Pair; +#ifdef GOOGLE_CLOUD_CPP_HAVE_OPENTELEMETRY +using ::google::cloud::testing_util::EventNamed; +using ::google::cloud::testing_util::InstallSpanCatcher; +using ::google::cloud::testing_util::OTelAttribute; +using ::google::cloud::testing_util::SpanEventAttributesAre; +using ::google::cloud::testing_util::SpanHasEvents; +using ::google::cloud::testing_util::SpanNamed; +using ::testing::_; +using ::testing::ElementsAre; +#endif + using MockStream = google::cloud::mocks::MockAsyncStreamingReadWriteRpc< google::storage::v2::BidiReadObjectRequest, google::storage::v2::BidiReadObjectResponse>; @@ -155,6 +172,97 @@ TEST(OpenImpl, Basic) { ASSERT_THAT(response, IsOkAndHolds(expected_result())); } +#ifdef GOOGLE_CLOUD_CPP_HAVE_OPENTELEMETRY +TEST(OpenImpl, StreamOpenLatencySpanEvent) { + auto span_catcher = InstallSpanCatcher(); + auto options = internal::MakeImmutableOptions( + Options{}.set(true)); + internal::OptionsSpan options_span(options); + + auto span = internal::MakeSpan("test-open-span"); + opentelemetry::trace::Scope scope(span); + + auto constexpr kText = R"pb( + bucket: "projects/_/buckets/test-bucket" + object: "test-object" + generation: 42 + )pb"; + auto constexpr kReadResponse = R"pb( + metadata { + bucket: "projects/_/buckets/test-bucket" + name: "test-object" + generation: 42 + } + read_handle { handle: "handle-123" } + )pb"; + + auto request = google::storage::v2::BidiReadObjectRequest{}; + ASSERT_TRUE( + TextFormat::ParseFromString(kText, request.mutable_read_object_spec())); + auto expected_response = google::storage::v2::BidiReadObjectResponse{}; + ASSERT_TRUE(TextFormat::ParseFromString(kReadResponse, &expected_response)); + + AsyncSequencer sequencer; + MockStorageStub mock; + EXPECT_CALL(mock, AsyncBidiReadObject) + .WillOnce([&](CompletionQueue const&, + std::shared_ptr const&, + google::cloud::internal::ImmutableOptions const&) { + auto stream = std::make_unique(); + EXPECT_CALL(*stream, Start).WillOnce([&sequencer]() { + return sequencer.PushBack("Start").then( + [](auto f) { return f.get(); }); + }); + EXPECT_CALL(*stream, Write) + .WillOnce( + [&sequencer](google::storage::v2::BidiReadObjectRequest const&, + grpc::WriteOptions) { + return sequencer.PushBack("Write").then( + [](auto f) { return f.get(); }); + }); + EXPECT_CALL(*stream, Read).WillOnce([&sequencer, expected_response]() { + return sequencer.PushBack("Read").then([expected_response](auto) { + return std::make_optional(expected_response); + }); + }); + return std::unique_ptr(std::move(stream)); + }); + + CompletionQueue cq; + auto coro = std::make_shared( + mock, cq, std::make_shared(), + internal::MakeImmutableOptions({}), request); + auto pending = coro->Call(); + auto next = sequencer.PopFrontWithName(); + EXPECT_EQ(next.second, "Start"); + next.first.set_value(true); + next = sequencer.PopFrontWithName(); + EXPECT_EQ(next.second, "Write"); + next.first.set_value(true); + next = sequencer.PopFrontWithName(); + EXPECT_EQ(next.second, "Read"); + next.first.set_value(true); + + auto response = pending.get(); + EXPECT_THAT(response, StatusIs(StatusCode::kOk)); + internal::EndSpan(*span, Status{}); + + auto spans = span_catcher->GetSpans(); + using EventMatcher = + testing::Matcher; + EXPECT_THAT( + spans, + ElementsAre(AllOf( + SpanNamed("test-open-span"), + SpanHasEvents(EventMatcher(AllOf( + EventNamed("gl-cpp.stream_open.latency"), + SpanEventAttributesAre( + OTelAttribute("gl-cpp.latency.network_handshake", _), + OTelAttribute("gl-cpp.latency.server_metadata", _), + OTelAttribute("gl-cpp.latency.stream_open", _)))))))); +} +#endif + TEST(OpenImpl, BasicReadHandle) { auto constexpr kText = R"pb( bucket: "projects/_/buckets/test-bucket"