feat(storage): Create OTel tracing decorator for storage:: ParallelUploadFile() (#15289)

Author: shubham-up-47

google/cloud/storage/internal/connection_impl.cc

@@ -14,6 +14,7 @@
 
 #include "google/cloud/storage/internal/connection_impl.h"
 #include "google/cloud/storage/internal/retry_object_read_source.h"
+#include "google/cloud/storage/parallel_upload.h"
 #include "google/cloud/internal/filesystem.h"
 #include "google/cloud/internal/opentelemetry.h"
 #include "google/cloud/internal/rest_retry_loop.h"
@@ -854,6 +855,20 @@ integrity checks using the DisableMD5Hash() and DisableCrc32cChecksum() options.
   return std::unique_ptr<std::istream>(std::move(source));
 }
 
+StatusOr<ObjectMetadata> StorageConnectionImpl::ExecuteParallelUploadFile(
+    std::vector<std::thread> threads,
+    std::vector<ParallelUploadFileShard> shards, bool ignore_cleanup_failures) {
+  for (auto& thread : threads) {
+    thread.join();
+  }
+  auto res = shards[0].WaitForCompletion().get();
+  auto cleanup_res = shards[0].EagerCleanup();
+  if (!cleanup_res.ok() && !ignore_cleanup_failures) {
+    return cleanup_res;
+  }
+  return res;
+}
+
 StatusOr<ListBucketAclResponse> StorageConnectionImpl::ListBucketAcl(
     ListBucketAclRequest const& request) {
   auto const idempotency = current_idempotency_policy().IsIdempotent(request)

google/cloud/storage/internal/connection_impl.h

@@ -104,6 +104,10 @@ class StorageConnectionImpl
       InsertObjectMediaRequest& request) override;
   StatusOr<std::unique_ptr<std::istream>> UploadFileResumable(
       std::string const& file_name, ResumableUploadRequest& request) override;
+  StatusOr<ObjectMetadata> ExecuteParallelUploadFile(
+      std::vector<std::thread> threads,
+      std::vector<ParallelUploadFileShard> shards,
+      bool ignore_cleanup_failures) override;
 
   StatusOr<ListBucketAclResponse> ListBucketAcl(
       ListBucketAclRequest const& request) override;

google/cloud/storage/internal/storage_connection.cc

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 #include "google/cloud/storage/internal/storage_connection.h"
+#include "google/cloud/storage/parallel_upload.h"
 #include <utility>
 #include <vector>
 
@@ -44,6 +45,16 @@ StatusOr<CreateOrResumeResponse> CreateOrResume(
                                 std::move(response->payload)};
 }
 
+StatusOr<ObjectMetadata> StorageConnection::ExecuteParallelUploadFile(
+    std::vector<std::thread>,  // NOLINT(performance-unnecessary-value-param)
+    std::vector<
+        ParallelUploadFileShard>,  // NOLINT(performance-unnecessary-value-param)
+    bool) {
+  return Status(
+      StatusCode::kUnimplemented,
+      "ExecuteParallelUploadFile() is not implemented by this Object");
+}
+
 }  // namespace internal
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
 }  // namespace storage

google/cloud/storage/internal/storage_connection.h

@@ -38,13 +38,15 @@
 #include "google/cloud/status_or.h"
 #include <memory>
 #include <string>
+#include <thread>
 #include <vector>
 
 namespace google {
 namespace cloud {
 namespace storage {
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
 namespace internal {
+class ParallelUploadFileShard;
 class ObjectReadStreambuf;
 
 /**
@@ -118,6 +120,8 @@ class StorageConnection {
       std::string const&, ResumableUploadRequest&) {
     return Status(StatusCode::kUnimplemented, "unimplemented");
   }
+  virtual StatusOr<ObjectMetadata> ExecuteParallelUploadFile(
+      std::vector<std::thread>, std::vector<ParallelUploadFileShard>, bool);
   ///@}
 
   ///@{

google/cloud/storage/internal/storage_connection_test.cc

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 #include "google/cloud/storage/internal/storage_connection.h"
+#include "google/cloud/storage/parallel_upload.h"
 #include "google/cloud/testing_util/status_matchers.h"
 #include <gmock/gmock.h>
 
@@ -160,6 +161,16 @@ TEST(StorageConnectionTest, UploadFileResumableUnimplemented) {
   EXPECT_THAT(response, StatusIs(StatusCode::kUnimplemented));
 }
 
+TEST(StorageConnectionTest, ExecuteParallelUploadFileUnimplemented) {
+  TestStorageConnection connection;
+  std::vector<std::thread> threads;
+  std::vector<ParallelUploadFileShard> shards;
+  bool ignore_cleanup_failures = false;
+  auto response = connection.ExecuteParallelUploadFile(
+      std::move(threads), std::move(shards), ignore_cleanup_failures);
+  EXPECT_THAT(response, StatusIs(StatusCode::kUnimplemented));
+}
+
 }  // namespace
 }  // namespace internal
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END

google/cloud/storage/internal/tracing_connection.cc

@@ -14,6 +14,7 @@
 
 #include "google/cloud/storage/internal/tracing_connection.h"
 #include "google/cloud/storage/internal/tracing_object_read_source.h"
+#include "google/cloud/storage/parallel_upload.h"
 #include "google/cloud/internal/opentelemetry.h"
 #include <memory>
 #include <string>
@@ -255,6 +256,18 @@ StatusOr<std::unique_ptr<std::istream>> TracingConnection::UploadFileResumable(
                            impl_->UploadFileResumable(file_name, request));
 }
 
+StatusOr<storage::ObjectMetadata> TracingConnection::ExecuteParallelUploadFile(
+    std::vector<std::thread> threads,
+    std::vector<storage::internal::ParallelUploadFileShard> shards,
+    bool ignore_cleanup_failures) {
+  auto span = internal::MakeSpan(
+      "storage::ParallelUploadFile/ExecuteParallelUploadFile");
+  auto scope = opentelemetry::trace::Scope(span);
+  return internal::EndSpan(*span, impl_->ExecuteParallelUploadFile(
+                                      std::move(threads), std::move(shards),
+                                      ignore_cleanup_failures));
+}
+
 StatusOr<storage::internal::ListBucketAclResponse>
 TracingConnection::ListBucketAcl(
     storage::internal::ListBucketAclRequest const& request) {

google/cloud/storage/internal/tracing_connection.h

@@ -16,6 +16,7 @@
 #define GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_STORAGE_INTERNAL_TRACING_CONNECTION_H
 
 #include "google/cloud/storage/internal/storage_connection.h"
+#include "google/cloud/storage/parallel_upload.h"
 #include "google/cloud/storage/version.h"
 #include <memory>
 #include <string>
@@ -104,6 +105,10 @@ class TracingConnection : public storage::internal::StorageConnection {
   StatusOr<std::unique_ptr<std::istream>> UploadFileResumable(
       std::string const& file_name,
       storage::internal::ResumableUploadRequest& request) override;
+  StatusOr<storage::ObjectMetadata> ExecuteParallelUploadFile(
+      std::vector<std::thread> threads,
+      std::vector<storage::internal::ParallelUploadFileShard> shards,
+      bool ignore_cleanup_failures) override;
 
   StatusOr<storage::internal::ListBucketAclResponse> ListBucketAcl(
       storage::internal::ListBucketAclRequest const& request) override;

google/cloud/storage/internal/tracing_connection_test.cc

@@ -746,6 +746,41 @@ TEST(TracingClientTest, UploadFileResumable) {
                       "gl-cpp.status_code", code_str)))));
 }
 
+TEST(TracingClientTest, ExecuteParallelUploadFile) {
+  auto span_catcher = InstallSpanCatcher();
+  auto mock = std::make_shared<MockClient>();
+  EXPECT_CALL(*mock, ExecuteParallelUploadFile)
+      .WillOnce(
+          [](std::vector<
+                 std::thread>,  // NOLINT(performance-unnecessary-value-param)
+             std::vector<
+                 storage::internal::
+                     ParallelUploadFileShard>,  // NOLINT(performance-unnecessary-value-param)
+             bool) {
+            EXPECT_TRUE(ThereIsAnActiveSpan());
+            return PermanentError();
+          });
+  auto under_test = TracingConnection(mock);
+  std::vector<std::thread> threads;
+  std::vector<storage::internal::ParallelUploadFileShard> shards;
+  bool ignore_cleanup_failures = false;
+  auto actual = under_test.ExecuteParallelUploadFile(
+      std::move(threads), std::move(shards), ignore_cleanup_failures);
+
+  auto const code = PermanentError().code();
+  auto const code_str = StatusCodeToString(code);
+  auto const msg = PermanentError().message();
+  EXPECT_THAT(actual, StatusIs(code));
+  EXPECT_THAT(
+      span_catcher->GetSpans(),
+      ElementsAre(AllOf(
+          SpanNamed("storage::ParallelUploadFile/ExecuteParallelUploadFile"),
+          SpanHasInstrumentationScope(), SpanKindIsClient(),
+          SpanWithStatus(opentelemetry::trace::StatusCode::kError, msg),
+          SpanHasAttributes(
+              OTelAttribute<std::string>("gl-cpp.status_code", code_str)))));
+}
+
 TEST(TracingClientTest, ListBucketAcl) {
   auto span_catcher = InstallSpanCatcher();
   auto mock = std::make_shared<MockClient>();

google/cloud/storage/parallel_upload.h

@@ -1190,6 +1190,7 @@ StatusOr<ObjectMetadata> ParallelUploadFile(
       internal::IsOptionSupportedWithParallelUpload<Options...>::value,
       "Provided Option not found in ParallelUploadFileSupportedOptions.");
 
+  auto connection = internal::ClientImplDetails::GetConnection(client);
   auto shards = internal::CreateParallelUploadShards::Create(
       std::move(client), std::move(file_name), std::move(bucket_name),
       std::move(object_name), std::move(prefix),
@@ -1207,15 +1208,8 @@ StatusOr<ObjectMetadata> ParallelUploadFile(
       shard.Upload();
     });
   }
-  for (auto& thread : threads) {
-    thread.join();
-  }
-  auto res = (*shards)[0].WaitForCompletion().get();
-  auto cleanup_res = (*shards)[0].EagerCleanup();
-  if (!cleanup_res.ok() && !ignore_cleanup_failures) {
-    return cleanup_res;
-  }
-  return res;
+  return connection->ExecuteParallelUploadFile(
+      std::move(threads), std::move(*shards), ignore_cleanup_failures);
 }
 
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END

google/cloud/storage/testing/mock_client.h

@@ -17,6 +17,7 @@
 
 #include "google/cloud/storage/client.h"
 #include "google/cloud/storage/internal/storage_connection.h"
+#include "google/cloud/storage/parallel_upload.h"
 #include <gmock/gmock.h>
 #include <memory>
 #include <string>
@@ -108,6 +109,10 @@ class MockClient : public google::cloud::storage::internal::StorageConnection {
   MOCK_METHOD(StatusOr<std::unique_ptr<std::istream>>, UploadFileResumable,
               (std::string const&, storage::internal::ResumableUploadRequest&),
               (override));
+  MOCK_METHOD(StatusOr<ObjectMetadata>, ExecuteParallelUploadFile,
+              (std::vector<std::thread>,
+               std::vector<storage::internal::ParallelUploadFileShard>, bool),
+              (override));
 
   MOCK_METHOD(StatusOr<internal::ListBucketAclResponse>, ListBucketAcl,
               (internal::ListBucketAclRequest const&), (override));