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 @@ -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",
Expand Down Expand Up @@ -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",
Expand Down
13 changes: 11 additions & 2 deletions google/cloud/storage/google_cloud_cpp_storage_grpc.cmake
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down
17 changes: 17 additions & 0 deletions google/cloud/storage/internal/async/open_object.cc
Original file line number Diff line number Diff line change
Expand Up @@ -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 <opentelemetry/metrics/provider.h>
#endif
#include <utility>

namespace google {
Expand All @@ -41,6 +46,10 @@ OpenObject::OpenObject(storage_internal::StorageStub& stub, CompletionQueue& cq,
initial_request_(std::move(request)) {}

future<StatusOr<OpenStreamResult>> OpenObject::Call() {
metrics_.RecordCall();
#ifdef GOOGLE_CLOUD_CPP_HAVE_OPENTELEMETRY
Comment thread
kalragauri marked this conversation as resolved.
span_ = opentelemetry::trace::Tracer::GetCurrentSpan();
#endif
Comment thread
kalragauri marked this conversation as resolved.
auto future = promise_.get_future();
rpc_->Start().then([w = WeakFromThis()](auto f) {
if (auto self = w.lock()) self->OnStart(f.get());
Expand All @@ -63,13 +72,15 @@ std::unique_ptr<OpenStream::StreamingRpc> OpenObject::CreateRpc(
}

void OpenObject::OnStart(bool ok) {
metrics_.RecordStart();
if (!ok) return DoFinish();
Comment thread
kalragauri marked this conversation as resolved.
rpc_->Write(initial_request_).then([w = WeakFromThis()](auto f) {
if (auto self = w.lock()) self->OnWrite(f.get());
});
}

void OpenObject::OnWrite(bool ok) {
metrics_.RecordWrite();
if (!ok) return DoFinish();
Comment thread
kalragauri marked this conversation as resolved.
rpc_->Read().then([w = WeakFromThis()](auto f) {
if (auto self = w.lock()) self->OnRead(f.get());
Expand All @@ -78,6 +89,12 @@ void OpenObject::OnWrite(bool ok) {

void OpenObject::OnRead(
std::optional<google::storage::v2::BidiReadObjectResponse> 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)});
}
Expand Down
6 changes: 6 additions & 0 deletions google/cloud/storage/internal/async/open_object.h
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand All @@ -25,6 +26,7 @@
#include "google/cloud/version.h"
#include "google/storage/v2/storage.pb.h"
#include <grpcpp/grpcpp.h>
#include <chrono>
#include <memory>
#include <string>

Expand Down Expand Up @@ -107,6 +109,10 @@ class OpenObject : public std::enable_shared_from_this<OpenObject> {
std::shared_ptr<OpenStream> rpc_;
promise<StatusOr<OpenStreamResult>> promise_;
google::storage::v2::BidiReadObjectRequest initial_request_;
#ifdef GOOGLE_CLOUD_CPP_HAVE_OPENTELEMETRY
opentelemetry::nostd::shared_ptr<opentelemetry::trace::Span> span_;
Comment thread
kalragauri marked this conversation as resolved.
#endif
OpenObjectTelemetry metrics_;
};

GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
Expand Down
140 changes: 140 additions & 0 deletions google/cloud/storage/internal/async/open_object_telemetry.cc
Original file line number Diff line number Diff line change
@@ -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 <opentelemetry/metrics/provider.h>
#endif

namespace google {
namespace cloud {
namespace storage_internal {
GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN

#ifdef GOOGLE_CLOUD_CPP_STORAGE_WITH_OTEL_METRICS
Comment thread
kalragauri marked this conversation as resolved.
namespace {
struct StreamOpenMetrics {
opentelemetry::nostd::shared_ptr<opentelemetry::metrics::Histogram<double>>
network_handshake;
opentelemetry::nostd::shared_ptr<opentelemetry::metrics::Histogram<double>>
server_metadata_latency;
opentelemetry::nostd::shared_ptr<opentelemetry::metrics::Histogram<double>>
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<double>(
std::chrono::duration_cast<std::chrono::microseconds>(t1_ - t0_).count());
auto p2 = static_cast<double>(
std::chrono::duration_cast<std::chrono::microseconds>(t3 - t2_).count());
auto p3 = static_cast<double>(
std::chrono::duration_cast<std::chrono::microseconds>(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<opentelemetry::trace::Span> 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<double>(
std::chrono::duration_cast<std::chrono::microseconds>(t1_ - t0_)
.count());
auto p2 = static_cast<double>(
std::chrono::duration_cast<std::chrono::microseconds>(t3 - t2_)
.count());
auto p3 = static_cast<double>(
std::chrono::duration_cast<std::chrono::microseconds>(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}});
}
}
Comment thread
kalragauri marked this conversation as resolved.
#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
64 changes: 64 additions & 0 deletions google/cloud/storage/internal/async/open_object_telemetry.h
Original file line number Diff line number Diff line change
@@ -0,0 +1,64 @@
// Copyright 2026 Google LLC
//
Comment thread
kalragauri marked this conversation as resolved.
// 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 <chrono>
#include <string>

#ifdef GOOGLE_CLOUD_CPP_HAVE_OPENTELEMETRY
#include <opentelemetry/trace/span.h>
#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<opentelemetry::trace::Span> 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
Comment thread
kalragauri marked this conversation as resolved.
};

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
Loading