feat(storage): adding logging decorator for object_descriptor_connection

google/cloud/storage/async/client.cc

@@ -14,10 +14,12 @@
 
 #include "google/cloud/storage/async/client.h"
 #include "google/cloud/storage/internal/async/connection_impl.h"
+#include "google/cloud/storage/internal/async/connection_logging.h"
 #include "google/cloud/storage/internal/async/connection_tracing.h"
 #include "google/cloud/storage/internal/async/default_options.h"
 #include "google/cloud/storage/internal/grpc/stub.h"
 #include "google/cloud/grpc_options.h"
+#include "google/cloud/internal/algorithm.h"
 #include <memory>
 #include <string>
 #include <utility>
@@ -36,6 +38,12 @@ AsyncClient::AsyncClient(Options options) {
   connection_ = storage_internal::MakeTracingAsyncConnection(
       storage_internal::MakeAsyncConnection(background_->cq(),
                                             std::move(options)));
+  auto const components = connection_->options().get<LoggingComponentsOption>();
+  if (std::find(components.begin(), components.end(), "rpc") !=
+      components.end()) {
+    connection_ =
+        storage_internal::MakeLoggingAsyncConnection(std::move(connection_));
+  }
 }
 
 AsyncClient::AsyncClient(std::shared_ptr<AsyncConnection> connection)

google/cloud/storage/async/client_test.cc

@@ -13,12 +13,15 @@
 // limitations under the License.
 
 #include "google/cloud/storage/async/client.h"
+#include "google/cloud/storage/internal/async/connection_logging.h"
 #include "google/cloud/storage/mocks/mock_async_connection.h"
 #include "google/cloud/storage/mocks/mock_async_object_descriptor_connection.h"
 #include "google/cloud/storage/mocks/mock_async_reader_connection.h"
 #include "google/cloud/storage/mocks/mock_async_rewriter_connection.h"
 #include "google/cloud/storage/mocks/mock_async_writer_connection.h"
+#include "google/cloud/common_options.h"
 #include "google/cloud/testing_util/is_proto_equal.h"
+#include "google/cloud/testing_util/scoped_log.h"
 #include "google/cloud/testing_util/status_matchers.h"
 #include <google/protobuf/text_format.h>
 #include <gmock/gmock.h>
@@ -42,7 +45,10 @@ using ::google::cloud::testing_util::IsOk;
 using ::google::cloud::testing_util::IsOkAndHolds;
 using ::google::cloud::testing_util::IsProtoEqual;
 using ::google::protobuf::TextFormat;
+using ::testing::Contains;
 using ::testing::ElementsAre;
+using ::testing::HasSubstr;
+using ::testing::IsEmpty;
 using ::testing::Optional;
 using ::testing::ResultOf;
 using ::testing::Return;
@@ -1397,6 +1403,81 @@ TEST(AsyncClient, ResumeRewrite2) {
   EXPECT_THAT(rt->first, MatchRewriteResponse());
 }
 
+TEST(AsyncClient, LoggingEnabled) {
+  auto mock = std::make_shared<MockAsyncConnection>();
+  EXPECT_CALL(*mock, options)
+      .WillRepeatedly(Return(Options{}.set<LoggingComponentsOption>({"rpc"})));
+  EXPECT_CALL(*mock, ReadObject)
+      .WillOnce([](AsyncConnection::ReadObjectParams const& p) {
+        EXPECT_THAT(p.options.get<LoggingComponentsOption>(), Contains("rpc"));
+        auto reader = std::make_unique<MockAsyncReaderConnection>();
+        EXPECT_CALL(*reader, Read).WillOnce([] {
+          return make_ready_future(
+              AsyncReaderConnection::ReadResponse{Status{}});
+        });
+        return make_ready_future(make_status_or(
+            std::unique_ptr<AsyncReaderConnection>(std::move(reader))));
+      });
+
+  google::cloud::testing_util::ScopedLog log;
+
+  auto client = AsyncClient(storage_internal::MakeLoggingAsyncConnection(mock));
+  auto rt = client.ReadObject(BucketName("test-bucket"), "test-object").get();
+  ASSERT_STATUS_OK(rt);
+  AsyncReader r;
+  AsyncToken t;
+  std::tie(r, t) = *std::move(rt);
+  ASSERT_TRUE(t.valid());
+
+  auto pt = r.Read(std::move(t)).get();
+  AsyncReaderConnection::ReadResponse p;
+  AsyncToken t2;
+  ASSERT_STATUS_OK(pt);
+  std::tie(p, t2) = *std::move(pt);
+  EXPECT_FALSE(t2.valid());
+  EXPECT_THAT(
+      p, VariantWith<ReadPayload>(ResultOf(
+             "empty response", [](auto const& p) { return p.size(); }, 0)));
+
+  auto const log_lines = log.ExtractLines();
+  EXPECT_THAT(log_lines, Contains(HasSubstr("ReadObject")));
+  EXPECT_THAT(log_lines, Contains(HasSubstr("ReaderConnectionLogging::Read")));
+}
+
+TEST(AsyncClient, LoggingDisabled) {
+  auto mock = std::make_shared<MockAsyncConnection>();
+  EXPECT_CALL(*mock, options).WillRepeatedly(Return(Options{}));
+  EXPECT_CALL(*mock, ReadObject).WillOnce([](auto const&) {
+    auto reader = std::make_unique<MockAsyncReaderConnection>();
+    EXPECT_CALL(*reader, Read).WillOnce([] {
+      return make_ready_future(AsyncReaderConnection::ReadResponse{Status{}});
+    });
+    return make_ready_future(make_status_or(
+        std::unique_ptr<AsyncReaderConnection>(std::move(reader))));
+  });
+
+  google::cloud::testing_util::ScopedLog log;
+  auto client = AsyncClient(mock);
+  auto rt = client.ReadObject(BucketName("test-bucket"), "test-object").get();
+  ASSERT_STATUS_OK(rt);
+  AsyncReader r;
+  AsyncToken t;
+  std::tie(r, t) = *std::move(rt);
+  ASSERT_TRUE(t.valid());
+
+  auto pt = r.Read(std::move(t)).get();
+  AsyncReaderConnection::ReadResponse p;
+  AsyncToken t2;
+  ASSERT_STATUS_OK(pt);
+  std::tie(p, t2) = *std::move(pt);
+  EXPECT_FALSE(t2.valid());
+  EXPECT_THAT(
+      p, VariantWith<ReadPayload>(ResultOf(
+             "empty response", [](auto const& p) { return p.size(); }, 0)));
+
+  EXPECT_THAT(log.ExtractLines(), IsEmpty());
+}
+
 }  // namespace
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
 }  // namespace storage_experimental

google/cloud/storage/google_cloud_cpp_storage_grpc.bzl

@@ -39,10 +39,12 @@ google_cloud_cpp_storage_grpc_hdrs = [
     "grpc_plugin.h",
     "internal/async/connection_fwd.h",
     "internal/async/connection_impl.h",
+    "internal/async/connection_logging.h",
     "internal/async/connection_tracing.h",
     "internal/async/default_options.h",
     "internal/async/handle_redirect_error.h",
     "internal/async/insert_object.h",
+    "internal/async/object_descriptor_connection_logging.h",
     "internal/async/object_descriptor_connection_tracing.h",
     "internal/async/object_descriptor_impl.h",
     "internal/async/object_descriptor_reader.h",
@@ -55,6 +57,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_logging.h",
     "internal/async/reader_connection_resume.h",
     "internal/async/reader_connection_tracing.h",
     "internal/async/rewriter_connection_impl.h",
@@ -116,10 +119,12 @@ google_cloud_cpp_storage_grpc_srcs = [
     "async/writer.cc",
     "grpc_plugin.cc",
     "internal/async/connection_impl.cc",
+    "internal/async/connection_logging.cc",
     "internal/async/connection_tracing.cc",
     "internal/async/default_options.cc",
     "internal/async/handle_redirect_error.cc",
     "internal/async/insert_object.cc",
+    "internal/async/object_descriptor_connection_logging.cc",
     "internal/async/object_descriptor_connection_tracing.cc",
     "internal/async/object_descriptor_impl.cc",
     "internal/async/object_descriptor_reader.cc",
@@ -130,6 +135,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_logging.cc",
     "internal/async/reader_connection_resume.cc",
     "internal/async/reader_connection_tracing.cc",
     "internal/async/rewriter_connection_impl.cc",

google/cloud/storage/google_cloud_cpp_storage_grpc.cmake

@@ -102,6 +102,8 @@ add_library(
     internal/async/connection_fwd.h
     internal/async/connection_impl.cc
     internal/async/connection_impl.h
+    internal/async/connection_logging.cc
+    internal/async/connection_logging.h
     internal/async/connection_tracing.cc
     internal/async/connection_tracing.h
     internal/async/default_options.cc
@@ -110,6 +112,8 @@ add_library(
     internal/async/handle_redirect_error.h
     internal/async/insert_object.cc
     internal/async/insert_object.h
+    internal/async/object_descriptor_connection_logging.cc
+    internal/async/object_descriptor_connection_logging.h
     internal/async/object_descriptor_connection_tracing.cc
     internal/async/object_descriptor_connection_tracing.h
     internal/async/object_descriptor_impl.cc
@@ -132,6 +136,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_logging.cc
+    internal/async/reader_connection_logging.h
     internal/async/reader_connection_resume.cc
     internal/async/reader_connection_resume.h
     internal/async/reader_connection_tracing.cc
@@ -438,10 +444,12 @@ set(storage_client_grpc_unit_tests
     internal/async/connection_impl_test.cc
     internal/async/connection_impl_upload_hash_test.cc
     internal/async/connection_impl_upload_test.cc
+    internal/async/connection_logging_test.cc
     internal/async/connection_tracing_test.cc
     internal/async/default_options_test.cc
     internal/async/handle_redirect_error_test.cc
     internal/async/insert_object_test.cc
+    internal/async/object_descriptor_connection_logging_test.cc
     internal/async/object_descriptor_connection_tracing_test.cc
     internal/async/object_descriptor_impl_test.cc
     internal/async/object_descriptor_reader_test.cc
@@ -453,6 +461,7 @@ set(storage_client_grpc_unit_tests
     internal/async/read_range_test.cc
     internal/async/reader_connection_factory_test.cc
     internal/async/reader_connection_impl_test.cc
+    internal/async/reader_connection_logging_test.cc
     internal/async/reader_connection_resume_test.cc
     internal/async/reader_connection_tracing_test.cc
     internal/async/rewriter_connection_impl_test.cc

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

@@ -0,0 +1,156 @@
+// Copyright 2025 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/connection_logging.h"
+#include "google/cloud/storage/internal/async/object_descriptor_connection_logging.h"
+#include "google/cloud/storage/internal/async/reader_connection_logging.h"
+#include "google/cloud/storage/options.h"
+#include "google/cloud/grpc_options.h"
+#include "google/cloud/log.h"
+#include "google/cloud/tracing_options.h"
+#include <algorithm>
+#include <memory>
+#include <string>
+#include <utility>
+
+namespace google {
+namespace cloud {
+namespace storage_internal {
+GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
+namespace {
+
+using ::google::cloud::storage_experimental::AsyncConnection;
+using ::google::cloud::storage_experimental::AsyncReaderConnection;
+using ::google::cloud::storage_experimental::AsyncRewriterConnection;
+using ::google::cloud::storage_experimental::AsyncWriterConnection;
+using ::google::cloud::storage_experimental::ObjectDescriptorConnection;
+
+class AsyncConnectionLogging : public AsyncConnection {
+ public:
+  explicit AsyncConnectionLogging(std::shared_ptr<AsyncConnection> child)
+      : child_(std::move(child)) {}
+
+  Options options() const override { return child_->options(); }
+
+  future<StatusOr<google::storage::v2::Object>> InsertObject(
+      InsertObjectParams p) override {
+    return child_->InsertObject(std::move(p));
+  }
+
+  future<StatusOr<std::shared_ptr<ObjectDescriptorConnection>>> Open(
+      OpenParams p) override {
+    return child_->Open(std::move(p));
+  }
+
+  future<StatusOr<std::unique_ptr<AsyncReaderConnection>>> ReadObject(
+      ReadObjectParams p) override {
+    GCP_LOG(INFO) << "ReadObject("
+                  << "bucket=" << p.request.bucket()
+                  << ", object=" << p.request.object() << ")";
+    auto options = p.options;
+    auto fut = child_->ReadObject(std::move(p));
+    return fut.then(
+        [options](auto f) -> StatusOr<std::unique_ptr<AsyncReaderConnection>> {
+          auto r = f.get();
+          if (!r) {
+            GCP_LOG(ERROR) << "ReadObject failed: " << r.status();
+            return std::move(r).status();
+          }
+          GCP_LOG(INFO) << "ReadObject succeeded";
+          return MakeLoggingReaderConnection(options, *std::move(r));
+        });
+  }
+
+  future<StatusOr<storage_experimental::ReadPayload>> ReadObjectRange(
+      ReadObjectParams p) override {
+    GCP_LOG(INFO) << "ReadObjectRange("
+                  << "bucket=" << p.request.bucket()
+                  << ", object=" << p.request.object() << ")";
+    auto fut = child_->ReadObjectRange(std::move(p));
+    return fut.then([](auto f) {
+      auto result = f.get();
+      if (!result.ok()) {
+        GCP_LOG(ERROR) << "ReadObjectRange failed: " << result.status();
+      } else {
+        GCP_LOG(INFO) << "ReadObjectRange succeeded";
+      }
+      return result;
+    });
+  }
+
+  future<StatusOr<std::unique_ptr<AsyncWriterConnection>>>
+  StartAppendableObjectUpload(AppendableUploadParams p) override {
+    return child_->StartAppendableObjectUpload(std::move(p));
+  }
+
+  future<StatusOr<std::unique_ptr<AsyncWriterConnection>>>
+  ResumeAppendableObjectUpload(AppendableUploadParams p) override {
+    return child_->ResumeAppendableObjectUpload(std::move(p));
+  }
+
+  future<StatusOr<std::unique_ptr<AsyncWriterConnection>>>
+  StartUnbufferedUpload(UploadParams p) override {
+    return child_->StartUnbufferedUpload(std::move(p));
+  }
+
+  future<StatusOr<std::unique_ptr<AsyncWriterConnection>>> StartBufferedUpload(
+      UploadParams p) override {
+    return child_->StartBufferedUpload(std::move(p));
+  }
+
+  future<StatusOr<std::unique_ptr<AsyncWriterConnection>>>
+  ResumeUnbufferedUpload(ResumeUploadParams p) override {
+    return child_->ResumeUnbufferedUpload(std::move(p));
+  }
+
+  future<StatusOr<std::unique_ptr<AsyncWriterConnection>>> ResumeBufferedUpload(
+      ResumeUploadParams p) override {
+    return child_->ResumeBufferedUpload(std::move(p));
+  }
+
+  future<StatusOr<google::storage::v2::Object>> ComposeObject(
+      ComposeObjectParams p) override {
+    return child_->ComposeObject(std::move(p));
+  }
+
+  future<Status> DeleteObject(DeleteObjectParams p) override {
+    return child_->DeleteObject(std::move(p));
+  }
+
+  std::shared_ptr<AsyncRewriterConnection> RewriteObject(
+      RewriteObjectParams p) override {
+    return child_->RewriteObject(std::move(p));
+  }
+
+ private:
+  std::shared_ptr<AsyncConnection> child_;
+};
+
+}  // namespace
+
+std::shared_ptr<storage_experimental::AsyncConnection>
+MakeLoggingAsyncConnection(
+    std::shared_ptr<storage_experimental::AsyncConnection> impl) {
+  auto const components = impl->options().get<LoggingComponentsOption>();
+  if (std::find(components.begin(), components.end(), "rpc") ==
+      components.end()) {
+    return impl;
+  }
+  return std::make_shared<AsyncConnectionLogging>(std::move(impl));
+}
+
+GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
+}  // namespace storage_internal
+}  // namespace cloud
+}  // namespace google