feat(storage): add stream open latency metrics and trace annotations

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_metrics.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_metrics.cc",
     "internal/async/open_stream.cc",
     "internal/async/partial_upload.cc",
     "internal/async/read_range.cc",

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_metrics.cc
+    internal/async/open_object_metrics.h
     internal/async/open_stream.cc
     internal/async/open_stream.h
     internal/async/partial_upload.cc

google/cloud/storage/internal/async/open_object.cc

@@ -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 {
@@ -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
+  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,13 +71,15 @@ std::unique_ptr<OpenStream::StreamingRpc> 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());
   });
 }
 
 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 +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)});
 }

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_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"
@@ -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>
 
@@ -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_;
+#endif
+  OpenObjectMetrics metrics_;
 };
 
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END

google/cloud/storage/internal/async/open_object_metrics.cc

@@ -0,0 +1,136 @@
+// Copyright 2024 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_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
+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
+}
+
+#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}});
+  }
+}
+#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

google/cloud/storage/internal/async/open_object_metrics.h

@@ -0,0 +1,64 @@
+// Copyright 2024 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_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
+};
+
+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