feat(storage): add debugging headers to ObjectWriteStream

Make it easier to troubleshoot problems with resumable uploads (aka
`Client::WriteObject()`) by capturing the headers of the last request.
We do not want to capture the headers for all the requests because there
can be an unbounded number of requests, and therefore all the headers
can consume arbitrary amounts of memory.

google/cloud/storage/examples/storage_client_mock_samples.cc

@@ -82,9 +82,9 @@ TEST(StorageMockingSamples, MockWriteObject) {
   EXPECT_CALL(*mock, CreateResumableUpload)
       .WillOnce(Return(CreateResumableUploadResponse{"test-only-upload-id"}));
   EXPECT_CALL(*mock, UploadChunk)
-      .WillOnce(Return(
-          QueryResumableUploadResponse{/*committed_size=*/absl::nullopt,
-                                       /*object_metadata=*/expected_metadata}));
+      .WillOnce(Return(QueryResumableUploadResponse{
+          /*.committed_size=*/absl::nullopt,
+          /*.object_metadata=*/expected_metadata}));
 
   auto stream = client.WriteObject("mock-bucket-name", "mock-object-name");
   stream << "Hello World!";

google/cloud/storage/internal/curl_request.cc

@@ -132,6 +132,7 @@ StatusOr<HttpResponse> CurlRequest::MakeRequestImpl() {
   if (logging_enabled_) handle_.FlushDebug(__func__);
   auto code = handle_.GetResponseCode();
   if (!code.ok()) return std::move(code).status();
+  received_headers_.emplace(":curl-peer", handle_.GetPeer());
   return HttpResponse{code.value(), std::move(response_payload_),
                       std::move(received_headers_)};
 }

google/cloud/storage/internal/grpc_client.cc

@@ -149,7 +149,8 @@ StatusOr<QueryResumableUploadResponse> CloseWriteObjectStream(
   watchdog.cancel();
   if (watchdog.get()) return TimeoutError(timeout, "Close()");
   if (!response) return std::move(response).status();
-  return GrpcObjectRequestParser::FromProto(*std::move(response), options);
+  return GrpcObjectRequestParser::FromProto(*std::move(response), options,
+                                            writer->GetRequestMetadata());
 }
 
 }  // namespace

google/cloud/storage/internal/grpc_object_request_parser.cc

@@ -464,7 +464,8 @@ GrpcObjectRequestParser::ToProto(InsertObjectMediaRequest const& request) {
 }
 
 QueryResumableUploadResponse GrpcObjectRequestParser::FromProto(
-    google::storage::v2::WriteObjectResponse const& p, Options const& options) {
+    google::storage::v2::WriteObjectResponse const& p, Options const& options,
+    google::cloud::internal::StreamingRpcMetadata metadata) {
   QueryResumableUploadResponse response;
   if (p.has_persisted_size()) {
     response.committed_size = static_cast<std::uint64_t>(p.persisted_size());
@@ -473,6 +474,7 @@ QueryResumableUploadResponse GrpcObjectRequestParser::FromProto(
     response.payload =
         GrpcObjectMetadataParser::FromProto(p.resource(), options);
   }
+  response.request_metadata = std::move(metadata);
   return response;
 }
 

google/cloud/storage/internal/grpc_object_request_parser.h

@@ -17,6 +17,7 @@
 
 #include "google/cloud/storage/internal/raw_client.h"
 #include "google/cloud/storage/version.h"
+#include "google/cloud/internal/grpc_request_metadata.h"
 #include <google/storage/v2/storage.pb.h>
 
 namespace google {
@@ -47,8 +48,8 @@ struct GrpcObjectRequestParser {
   static StatusOr<google::storage::v2::WriteObjectRequest> ToProto(
       InsertObjectMediaRequest const& request);
   static QueryResumableUploadResponse FromProto(
-      google::storage::v2::WriteObjectResponse const& p,
-      Options const& options);
+      google::storage::v2::WriteObjectResponse const& p, Options const& options,
+      google::cloud::internal::StreamingRpcMetadata metadata);
 
   static google::storage::v2::ListObjectsRequest ToProto(
       ListObjectsRequest const& request);

google/cloud/storage/internal/grpc_object_request_parser_test.cc

@@ -34,6 +34,7 @@ using ::google::cloud::testing_util::IsProtoEqual;
 using ::google::cloud::testing_util::StatusIs;
 using ::google::protobuf::TextFormat;
 using ::testing::ElementsAre;
+using ::testing::Pair;
 using ::testing::UnorderedElementsAre;
 
 // Use gsutil to obtain the CRC32C checksum (in base64):
@@ -637,7 +638,7 @@ TEST(GrpcObjectRequestParser, WriteObjectResponseSimple) {
       )pb",
       &input));
 
-  auto const actual = GrpcObjectRequestParser::FromProto(input, Options{});
+  auto const actual = GrpcObjectRequestParser::FromProto(input, Options{}, {});
   EXPECT_EQ(actual.committed_size.value_or(0), 123456);
   EXPECT_FALSE(actual.payload.has_value());
 }
@@ -653,12 +654,16 @@ TEST(GrpcObjectRequestParser, WriteObjectResponseWithResource) {
         })pb",
       &input));
 
-  auto const actual = GrpcObjectRequestParser::FromProto(input, Options{});
+  auto const actual = GrpcObjectRequestParser::FromProto(
+      input, Options{}, {{"header", "value"}, {"other-header", "other-value"}});
   EXPECT_FALSE(actual.committed_size.has_value());
   ASSERT_TRUE(actual.payload.has_value());
   EXPECT_EQ(actual.payload->name(), "test-object-name");
   EXPECT_EQ(actual.payload->bucket(), "test-bucket-name");
   EXPECT_EQ(actual.payload->size(), 123456);
+  EXPECT_THAT(actual.request_metadata,
+              UnorderedElementsAre(Pair("header", "value"),
+                                   Pair("other-header", "other-value")));
 }
 
 TEST(GrpcObjectRequestParser, ListObjectsRequestAllFields) {

google/cloud/storage/internal/object_requests.cc

@@ -505,6 +505,7 @@ StatusOr<std::uint64_t> ParseRangeHeader(std::string const& range) {
 StatusOr<QueryResumableUploadResponse>
 QueryResumableUploadResponse::FromHttpResponse(HttpResponse response) {
   QueryResumableUploadResponse result;
+  result.request_metadata = std::move(response.headers);
   auto done = response.status_code == HttpStatusCode::kOk ||
               response.status_code == HttpStatusCode::kCreated;
 
@@ -515,8 +516,8 @@ QueryResumableUploadResponse::FromHttpResponse(HttpResponse response) {
     if (!contents) return std::move(contents).status();
     result.payload = *std::move(contents);
   }
-  auto r = response.headers.find("range");
-  if (r == response.headers.end()) return result;
+  auto r = result.request_metadata.find("range");
+  if (r == result.request_metadata.end()) return result;
 
   auto last_committed_byte = ParseRangeHeader(r->second);
   if (!last_committed_byte) return std::move(last_committed_byte).status();

google/cloud/storage/internal/object_requests.h

@@ -28,6 +28,7 @@
 #include "google/cloud/storage/well_known_parameters.h"
 #include "absl/types/optional.h"
 #include "absl/types/span.h"
+#include <map>
 #include <numeric>
 #include <string>
 #include <vector>
@@ -501,9 +502,22 @@ StatusOr<std::uint64_t> ParseRangeHeader(std::string const& range);
 struct QueryResumableUploadResponse {
   static StatusOr<QueryResumableUploadResponse> FromHttpResponse(
       HttpResponse response);
+  QueryResumableUploadResponse() = default;
+  QueryResumableUploadResponse(
+      absl::optional<std::uint64_t> cs,
+      absl::optional<google::cloud::storage::ObjectMetadata> p)
+      : committed_size(std::move(cs)), payload(std::move(p)) {}
+  QueryResumableUploadResponse(
+      absl::optional<std::uint64_t> cs,
+      absl::optional<google::cloud::storage::ObjectMetadata> p,
+      std::multimap<std::string, std::string> rm)
+      : committed_size(std::move(cs)),
+        payload(std::move(p)),
+        request_metadata(std::move(rm)) {}
 
   absl::optional<std::uint64_t> committed_size;
   absl::optional<google::cloud::storage::ObjectMetadata> payload;
+  std::multimap<std::string, std::string> request_metadata;
 };
 
 bool operator==(QueryResumableUploadResponse const& lhs,

google/cloud/storage/internal/object_requests_test.cc

@@ -32,6 +32,8 @@ using ::google::cloud::testing_util::StatusIs;
 using ::testing::ElementsAre;
 using ::testing::HasSubstr;
 using ::testing::Not;
+using ::testing::Pair;
+using ::testing::UnorderedElementsAre;
 
 TEST(ObjectRequestsTest, ParseFailure) {
   auto actual = internal::ObjectMetadataParser::FromString("{123");
@@ -1006,6 +1008,10 @@ TEST(QueryResumableUploadResponseTest, Base) {
   ASSERT_TRUE(actual.payload.has_value());
   EXPECT_EQ("test-object-name", actual.payload->name());
   EXPECT_EQ(2000, actual.committed_size.value_or(0));
+  EXPECT_THAT(actual.request_metadata,
+              UnorderedElementsAre(Pair("ignored-header", "value"),
+                                   Pair("location", "location-value"),
+                                   Pair("range", "bytes=0-1999")));
 
   std::ostringstream os;
   os << actual;

google/cloud/storage/internal/object_write_streambuf.cc

@@ -69,7 +69,7 @@ void ObjectWriteStreambuf::AutoFlushFinal() {
 StatusOr<QueryResumableUploadResponse> ObjectWriteStreambuf::Close() {
   FlushFinal();
   if (!last_status_.ok()) return last_status_;
-  return QueryResumableUploadResponse{committed_size_, metadata_};
+  return QueryResumableUploadResponse{committed_size_, metadata_, headers_};
 }
 
 bool ObjectWriteStreambuf::IsOpen() const {
@@ -159,6 +159,7 @@ void ObjectWriteStreambuf::FlushFinal() {
   } else {
     committed_size_ = response->committed_size.value_or(0);
     metadata_ = std::move(response->payload);
+    headers_ = std::move(response->request_metadata);
   }
 
   // Reset the iostream put area with valid pointers, but empty.

google/cloud/storage/internal/object_write_streambuf.h

@@ -111,6 +111,7 @@ class ObjectWriteStreambuf : public std::basic_streambuf<char> {
   std::string upload_id_;
   std::uint64_t committed_size_ = 0;
   absl::optional<ObjectMetadata> metadata_;
+  std::multimap<std::string, std::string> headers_;
 
   std::vector<char> current_ios_buffer_;
   std::size_t max_buffer_size_;

google/cloud/storage/internal/retry_client_test.cc

@@ -59,7 +59,7 @@ TEST(RetryClientTest, NonIdempotentErrorHandling) {
   EXPECT_CALL(*mock, DeleteObject)
       .WillOnce(Return(StatusOr<EmptyResponse>(TransientError())));
 
-  // Use a delete operation because this is idempotent only if the it has
+  // Use a delete operation because this is idempotent only if it has
   // the IfGenerationMatch() and/or Generation() option set.
   StatusOr<EmptyResponse> result =
       client->DeleteObject(DeleteObjectRequest("test-bucket", "test-object"));
@@ -511,7 +511,7 @@ TEST(RetryClientTest, UploadChunkMissingRangeHeaderInUpload) {
   // committed size.
   EXPECT_CALL(*mock, UploadChunk)
       .WillOnce(
-          Return(QueryResumableUploadResponse{/*committed_size=*/absl::nullopt,
+          Return(QueryResumableUploadResponse{/*.committed_size=*/absl::nullopt,
                                               /*.payload=*/absl::nullopt}));
   // This should trigger a QueryResumableUpload(), simulate a good response.
   EXPECT_CALL(*mock, QueryResumableUpload)
@@ -552,7 +552,7 @@ TEST(RetryClientTest, UploadChunkMissingRangeHeaderInQueryResumableUpload) {
   // what bytes got uploaded.
   EXPECT_CALL(*mock, UploadChunk)
       .WillOnce(
-          Return(QueryResumableUploadResponse{/*committed_size=*/absl::nullopt,
+          Return(QueryResumableUploadResponse{/*.committed_size=*/absl::nullopt,
                                               /*.payload=*/absl::nullopt}));
   // This should trigger a `QueryResumableUpload()`, which should also have its
   // Range header missing indicating no bytes were uploaded.
@@ -563,7 +563,7 @@ TEST(RetryClientTest, UploadChunkMissingRangeHeaderInQueryResumableUpload) {
   // This should trigger a second upload, which we will let succeed.
   EXPECT_CALL(*mock, UploadChunk)
       .WillOnce(
-          Return(QueryResumableUploadResponse{/*committed_size=*/quantum,
+          Return(QueryResumableUploadResponse{/*.committed_size=*/quantum,
                                               /*.payload=*/absl::nullopt}));
 
   auto response = client->UploadChunk(

google/cloud/storage/object_write_stream.cc

@@ -90,7 +90,7 @@ void ObjectWriteStream::CloseBuf() {
     setstate(std::ios_base::badbit);
     return;
   }
-  headers_ = {};
+  headers_ = std::move(response->request_metadata);
   if (response->payload.has_value()) {
     metadata_ = *std::move(response->payload);
   }