feat(storage): Create OTel tracing decorator for Client::UploadFile() (#15245)

Author: shubham-up-47

google/cloud/storage/client.cc

@@ -130,98 +130,17 @@ bool Client::UseSimpleUpload(std::string const& file_name, std::size_t& size) {
 StatusOr<ObjectMetadata> Client::UploadFileSimple(
     std::string const& file_name, std::size_t file_size,
     internal::InsertObjectMediaRequest request) {
-  auto upload_offset = request.GetOption<UploadFromOffset>().value_or(0);
-  if (file_size < upload_offset) {
-    std::ostringstream os;
-    os << __func__ << "(" << request << ", " << file_name
-       << "): UploadFromOffset (" << upload_offset
-       << ") is bigger than the size of file source (" << file_size << ")";
-    return google::cloud::internal::InvalidArgumentError(std::move(os).str(),
-                                                         GCP_ERROR_INFO());
-  }
-  auto upload_size = (std::min)(
-      request.GetOption<UploadLimit>().value_or(file_size - upload_offset),
-      file_size - upload_offset);
-
-  std::ifstream is(file_name, std::ios::binary);
-  if (!is.is_open()) {
-    std::ostringstream os;
-    os << __func__ << "(" << request << ", " << file_name
-       << "): cannot open upload file source";
-    return google::cloud::internal::NotFoundError(std::move(os).str(),
-                                                  GCP_ERROR_INFO());
-  }
-
-  std::string payload(static_cast<std::size_t>(upload_size), char{});
-  is.seekg(upload_offset, std::ios::beg);
-  // We need to use `&payload[0]` until C++17
-  // NOLINTNEXTLINE(readability-container-data-pointer)
-  is.read(&payload[0], payload.size());
-  if (static_cast<std::size_t>(is.gcount()) < payload.size()) {
-    std::ostringstream os;
-    os << __func__ << "(" << request << ", " << file_name << "): Actual read ("
-       << is.gcount() << ") is smaller than upload_size (" << payload.size()
-       << ")";
-    return google::cloud::internal::InternalError(std::move(os).str(),
-                                                  GCP_ERROR_INFO());
-  }
-  is.close();
-  request.set_payload(payload);
-
+  auto payload = connection_->UploadFileSimple(file_name, file_size, request);
+  if (!payload) return payload.status();
+  request.set_payload(*payload.value());
   return connection_->InsertObjectMedia(request);
 }
 
 StatusOr<ObjectMetadata> Client::UploadFileResumable(
-    std::string const& file_name,
-    google::cloud::storage::internal::ResumableUploadRequest request) {
-  auto upload_offset = request.GetOption<UploadFromOffset>().value_or(0);
-  auto status = google::cloud::internal::status(file_name);
-  if (!is_regular(status)) {
-    GCP_LOG(WARNING) << "Trying to upload " << file_name
-                     << R"""( which is not a regular file.
-This is often a problem because:
-  - Some non-regular files are infinite sources of data, and the load will
-    never complete.
-  - Some non-regular files can only be read once, and UploadFile() may need to
-    read the file more than once to compute the checksum and hashes needed to
-    preserve data integrity.
-
-Consider using UploadLimit option or Client::WriteObject(). You may also need to disable data
-integrity checks using the DisableMD5Hash() and DisableCrc32cChecksum() options.
-)""";
-  } else {
-    std::error_code size_err;
-    auto file_size = google::cloud::internal::file_size(file_name, size_err);
-    if (size_err) {
-      return google::cloud::internal::NotFoundError(size_err.message(),
-                                                    GCP_ERROR_INFO());
-    }
-    if (file_size < upload_offset) {
-      std::ostringstream os;
-      os << __func__ << "(" << request << ", " << file_name
-         << "): UploadFromOffset (" << upload_offset
-         << ") is bigger than the size of file source (" << file_size << ")";
-      return google::cloud::internal::InvalidArgumentError(std::move(os).str(),
-                                                           GCP_ERROR_INFO());
-    }
-
-    auto upload_size = (std::min)(
-        request.GetOption<UploadLimit>().value_or(file_size - upload_offset),
-        file_size - upload_offset);
-    request.set_option(UploadContentLength(upload_size));
-  }
-  std::ifstream source(file_name, std::ios::binary);
-  if (!source.is_open()) {
-    std::ostringstream os;
-    os << __func__ << "(" << request << ", " << file_name
-       << "): cannot open upload file source";
-    return google::cloud::internal::NotFoundError(std::move(os).str(),
-                                                  GCP_ERROR_INFO());
-  }
-  // We set its offset before passing it to `UploadStreamResumable` so we don't
-  // need to compute `UploadFromOffset` again.
-  source.seekg(upload_offset, std::ios::beg);
-  return UploadStreamResumable(source, request);
+    std::string const& file_name, internal::ResumableUploadRequest request) {
+  auto source = connection_->UploadFileResumable(file_name, request);
+  if (!source) return source.status();
+  return UploadStreamResumable(*source.value(), request);
 }
 
 StatusOr<ObjectMetadata> Client::UploadStreamResumable(

google/cloud/storage/client.h

@@ -3530,7 +3530,7 @@ class Client {
       internal::InsertObjectMediaRequest request(bucket_name, object_name,
                                                  std::string{});
       request.set_multiple_options(std::forward<Options>(options)...);
-      return UploadFileSimple(file_name, file_size, request);
+      return UploadFileSimple(file_name, file_size, std::move(request));
     }
     internal::ResumableUploadRequest request(bucket_name, object_name);
     request.set_multiple_options(std::forward<Options>(options)...);

google/cloud/storage/client_object_test.cc

@@ -254,6 +254,54 @@ TEST_F(ObjectTest, UploadFile) {
   EXPECT_EQ(expected, *actual);
 }
 
+TEST_F(ObjectTest, UploadFileSimple) {
+  std::string text = R"""({
+      "name": "test-bucket-name/test-object-name/1"
+})""";
+  ObjectMetadata expected =
+      storage::internal::ObjectMetadataParser::FromString(text).value();
+  std::string const contents = std::string{"some simple contents"};
+
+  EXPECT_CALL(*mock_, InsertObjectMedia)
+      .WillOnce([&](internal::InsertObjectMediaRequest const& request) {
+        EXPECT_EQ("test-bucket-name", request.bucket_name());
+        EXPECT_EQ("test-object-name", request.object_name());
+        EXPECT_EQ(contents, request.payload());
+        return make_status_or(expected);
+      });
+
+  TempFile temp(contents);
+  auto client = ClientForMock();
+  StatusOr<ObjectMetadata> actual =
+      client.UploadFile(temp.name(), "test-bucket-name", "test-object-name");
+  ASSERT_STATUS_OK(actual);
+  EXPECT_EQ(expected, *actual);
+}
+
+TEST_F(ObjectTest, UploadFileResumable) {
+  std::string text = R"""({
+      "name": "test-bucket-name/test-object-name/1"
+})""";
+  ObjectMetadata expected =
+      storage::internal::ObjectMetadataParser::FromString(text).value();
+  std::string const contents = std::string{"some not so simple contents"};
+
+  EXPECT_CALL(*mock_, CreateResumableUpload)
+      .WillOnce(
+          Return(internal::CreateResumableUploadResponse{"test-upload-id"}));
+  EXPECT_CALL(*mock_, UploadChunk)
+      .WillOnce(Return(make_status_or(
+          internal::QueryResumableUploadResponse{absl::nullopt, expected})));
+
+  TempFile temp(contents);
+  auto client = ClientForMock();
+  StatusOr<ObjectMetadata> actual =
+      client.UploadFile(temp.name(), "test-bucket-name", "test-object-name",
+                        UseResumableUploadSession(""));
+  ASSERT_STATUS_OK(actual);
+  EXPECT_EQ(expected, *actual);
+}
+
 TEST_F(ObjectTest, DeleteResumableUpload) {
   EXPECT_CALL(*mock_, DeleteResumableUpload)
       .WillOnce(Return(StatusOr<internal::EmptyResponse>(TransientError())))

google/cloud/storage/google_cloud_cpp_storage.cmake

@@ -462,6 +462,7 @@ if (BUILD_TESTING)
         internal/connection_impl_bucket_acl_test.cc
         internal/connection_impl_bucket_test.cc
         internal/connection_impl_default_object_acl_test.cc
+        internal/connection_impl_file_upload_test.cc
         internal/connection_impl_notifications_test.cc
         internal/connection_impl_object_acl_test.cc
         internal/connection_impl_object_copy_test.cc
@@ -499,6 +500,7 @@ if (BUILD_TESTING)
         internal/service_account_requests_test.cc
         internal/sign_blob_requests_test.cc
         internal/signed_url_requests_test.cc
+        internal/storage_connection_test.cc
         internal/tracing_connection_test.cc
         internal/tracing_object_read_source_test.cc
         internal/tuple_filter_test.cc

google/cloud/storage/internal/connection_impl.cc

@@ -14,10 +14,13 @@
 
 #include "google/cloud/storage/internal/connection_impl.h"
 #include "google/cloud/storage/internal/retry_object_read_source.h"
+#include "google/cloud/internal/filesystem.h"
 #include "google/cloud/internal/opentelemetry.h"
 #include "google/cloud/internal/rest_retry_loop.h"
+#include "google/cloud/log.h"
 #include "absl/strings/match.h"
 #include <chrono>
+#include <fstream>
 #include <functional>
 #include <memory>
 #include <sstream>
@@ -754,6 +757,103 @@ StatusOr<QueryResumableUploadResponse> StorageConnectionImpl::UploadChunk(
   return RetryError(last_status, *retry_policy, __func__);
 }
 
+StatusOr<std::unique_ptr<std::string>> StorageConnectionImpl::UploadFileSimple(
+    std::string const& file_name, std::size_t file_size,
+    InsertObjectMediaRequest& request) {
+  auto upload_offset = request.GetOption<UploadFromOffset>().value_or(0);
+  if (file_size < upload_offset) {
+    std::ostringstream os;
+    os << __func__ << "(" << request << ", " << file_name
+       << "): UploadFromOffset (" << upload_offset
+       << ") is bigger than the size of file source (" << file_size << ")";
+    return google::cloud::internal::InvalidArgumentError(std::move(os).str(),
+                                                         GCP_ERROR_INFO());
+  }
+  auto upload_size = (std::min)(
+      request.GetOption<UploadLimit>().value_or(file_size - upload_offset),
+      file_size - upload_offset);
+
+  std::ifstream is(file_name, std::ios::binary);
+  if (!is.is_open()) {
+    std::ostringstream os;
+    os << __func__ << "(" << request << ", " << file_name
+       << "): cannot open upload file source";
+    return google::cloud::internal::NotFoundError(std::move(os).str(),
+                                                  GCP_ERROR_INFO());
+  }
+
+  auto payload = std::make_unique<std::string>(
+      static_cast<std::size_t>(upload_size), char{});
+  is.seekg(upload_offset, std::ios::beg);
+  // We need to use `&payload[0]` until C++17
+  // NOLINTNEXTLINE(readability-container-data-pointer)
+  is.read(&(*payload)[0], payload->size());
+  if (static_cast<std::size_t>(is.gcount()) < payload->size()) {
+    std::ostringstream os;
+    os << __func__ << "(" << request << ", " << file_name << "): Actual read ("
+       << is.gcount() << ") is smaller than upload_size (" << payload->size()
+       << ")";
+    return google::cloud::internal::InternalError(std::move(os).str(),
+                                                  GCP_ERROR_INFO());
+  }
+  is.close();
+
+  return payload;
+}
+
+StatusOr<std::unique_ptr<std::istream>>
+StorageConnectionImpl::UploadFileResumable(std::string const& file_name,
+                                           ResumableUploadRequest& request) {
+  auto upload_offset = request.GetOption<UploadFromOffset>().value_or(0);
+  auto status = google::cloud::internal::status(file_name);
+  if (!is_regular(status)) {
+    GCP_LOG(WARNING) << "Trying to upload " << file_name
+                     << R"""( which is not a regular file.
+This is often a problem because:
+  - Some non-regular files are infinite sources of data, and the load will
+    never complete.
+  - Some non-regular files can only be read once, and UploadFile() may need to
+    read the file more than once to compute the checksum and hashes needed to
+    preserve data integrity.
+
+Consider using UploadLimit option or Client::WriteObject(). You may also need to disable data
+integrity checks using the DisableMD5Hash() and DisableCrc32cChecksum() options.
+)""";
+  } else {
+    std::error_code size_err;
+    auto file_size = google::cloud::internal::file_size(file_name, size_err);
+    if (size_err) {
+      return google::cloud::internal::NotFoundError(size_err.message(),
+                                                    GCP_ERROR_INFO());
+    }
+    if (file_size < upload_offset) {
+      std::ostringstream os;
+      os << __func__ << "(" << request << ", " << file_name
+         << "): UploadFromOffset (" << upload_offset
+         << ") is bigger than the size of file source (" << file_size << ")";
+      return google::cloud::internal::InvalidArgumentError(std::move(os).str(),
+                                                           GCP_ERROR_INFO());
+    }
+
+    auto upload_size = (std::min)(
+        request.GetOption<UploadLimit>().value_or(file_size - upload_offset),
+        file_size - upload_offset);
+    request.set_option(UploadContentLength(upload_size));
+  }
+  auto source = std::make_unique<std::ifstream>(file_name, std::ios::binary);
+  if (!source->is_open()) {
+    std::ostringstream os;
+    os << __func__ << "(" << request << ", " << file_name
+       << "): cannot open upload file source";
+    return google::cloud::internal::NotFoundError(std::move(os).str(),
+                                                  GCP_ERROR_INFO());
+  }
+  // We set its offset before passing it to `UploadStreamResumable` so we don't
+  // need to compute `UploadFromOffset` again.
+  source->seekg(upload_offset, std::ios::beg);
+  return std::unique_ptr<std::istream>(std::move(source));
+}
+
 StatusOr<ListBucketAclResponse> StorageConnectionImpl::ListBucketAcl(
     ListBucketAclRequest const& request) {
   auto const idempotency = current_idempotency_policy().IsIdempotent(request)

google/cloud/storage/internal/connection_impl.h

@@ -99,6 +99,11 @@ class StorageConnectionImpl
       DeleteResumableUploadRequest const& request) override;
   StatusOr<QueryResumableUploadResponse> UploadChunk(
       UploadChunkRequest const& request) override;
+  StatusOr<std::unique_ptr<std::string>> UploadFileSimple(
+      std::string const& file_name, std::size_t file_size,
+      InsertObjectMediaRequest& request) override;
+  StatusOr<std::unique_ptr<std::istream>> UploadFileResumable(
+      std::string const& file_name, ResumableUploadRequest& request) override;
 
   StatusOr<ListBucketAclResponse> ListBucketAcl(
       ListBucketAclRequest const& request) override;