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_metrics.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_metrics.cc",
"internal/async/open_stream.cc",
"internal/async/partial_upload.cc",
"internal/async/read_range.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 @@ -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_metrics.cc
internal/async/open_object_metrics.h
internal/async/open_stream.cc
internal/async/open_stream.h
internal/async/partial_upload.cc
Expand Down
16 changes: 16 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,10 @@
#include "google/cloud/storage/internal/async/open_object.h"
#include "google/cloud/internal/make_status.h"
#include "absl/strings/str_cat.h"
#ifdef GOOGLE_CLOUD_CPP_STORAGE_WITH_OTEL_METRICS
#include "google/cloud/internal/opentelemetry.h"
#include <opentelemetry/metrics/provider.h>
#endif
#include <utility>

namespace google {
Expand All @@ -41,6 +45,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

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.

So, we're mixing tracing changes and metrics changes in one file.
I think these should be segregated in two separate files ( one for metrics and one for tracing ).
That'd totally avoid the if...else blocks.
But we'll need to see how we're going to invoke these classes when both metrics and tracing are enabled.
We can discuss this offline.

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 +71,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 +88,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_metrics.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
OpenObjectMetrics metrics_;
};

GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
Expand Down
136 changes: 136 additions & 0 deletions google/cloud/storage/internal/async/open_object_metrics.cc
Original file line number Diff line number Diff line change
@@ -0,0 +1,136 @@
// Copyright 2024 Google LLC

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.

nit. Copyright year should be 2026

//
// 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_metrics.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

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.

This file should ideally be invoked only when metrics option / environment variable is enabled. So we wouldn't need the if clause.

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 OpenObjectMetrics::RecordCall() {
#if defined(GOOGLE_CLOUD_CPP_STORAGE_WITH_OTEL_METRICS) || \
defined(GOOGLE_CLOUD_CPP_HAVE_OPENTELEMETRY)
t0_ = std::chrono::steady_clock::now();
#endif
}

void OpenObjectMetrics::RecordStart() {
#if defined(GOOGLE_CLOUD_CPP_STORAGE_WITH_OTEL_METRICS) || \
defined(GOOGLE_CLOUD_CPP_HAVE_OPENTELEMETRY)
t1_ = std::chrono::steady_clock::now();
#endif
}

void OpenObjectMetrics::RecordWrite() {
#if defined(GOOGLE_CLOUD_CPP_STORAGE_WITH_OTEL_METRICS) || \
defined(GOOGLE_CLOUD_CPP_HAVE_OPENTELEMETRY)
t2_ = std::chrono::steady_clock::now();
#endif
}
Comment thread
kalragauri marked this conversation as resolved.

#ifdef GOOGLE_CLOUD_CPP_STORAGE_WITH_OTEL_METRICS
void OpenObjectMetrics::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());

metrics.network_handshake->Record(p1, {{"gcp.storage.bucket", bucket}},
opentelemetry::context::Context{});
metrics.server_metadata_latency->Record(p2, {{"gcp.storage.bucket", bucket}},
opentelemetry::context::Context{});
metrics.stream_open_latency->Record(p3, {{"gcp.storage.bucket", bucket}},
opentelemetry::context::Context{});
}
#endif

#ifdef GOOGLE_CLOUD_CPP_HAVE_OPENTELEMETRY
void OpenObjectMetrics::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 OpenObjectMetrics::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_metrics.h
Original file line number Diff line number Diff line change
@@ -0,0 +1,64 @@
// Copyright 2024 Google LLC

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.

nit. Copyright year should be 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.

Can we also add unit tests for metrics?

// 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_METRICS_H
#define GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_STORAGE_INTERNAL_ASYNC_OPEN_OBJECT_METRICS_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 OpenObjectMetrics {
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_METRICS_H