feat(GCS+gRPC): capture streaming RPCs metadata (#6902)

gRPC returns metadata for all requests, including streaming RPCs. In the
storage service this is very useful for troubleshooting, this change
weaves the metadata (and some additional bits from
`grpc::ClientContext`) through the different layers. The metadata is
exposed as "headers", because that matches the existing APIs (and gRPC
uses HTTP/2 headers under the hood anyway).

Author: coryan

generator/integration_tests/golden/mocks/mock_golden_kitchen_sink_stub.h

@@ -73,6 +73,8 @@ class MockTailLogEntriesStreamingReadRpc
   MOCK_METHOD(void, Cancel, (), (override));
   MOCK_METHOD((absl::variant<Status, ::google::test::admin::database::v1::TailLogEntriesResponse>), Read, (),
   (override));
+  MOCK_METHOD(internal::StreamingRpcMetadata, GetRequestMetadata, (),
+  (const, override));
 };
 
 }  // namespace GOOGLE_CLOUD_CPP_GENERATED_NS

google/cloud/CMakeLists.txt

@@ -427,6 +427,7 @@ if (GOOGLE_CLOUD_CPP_ENABLE_GRPC)
         internal/setup_context.h
         internal/streaming_read_rpc.cc
         internal/streaming_read_rpc.h
+        internal/streaming_read_rpc_logging.cc
         internal/streaming_read_rpc_logging.h
         internal/streaming_write_rpc.cc
         internal/streaming_write_rpc.h

google/cloud/google_cloud_cpp_grpc_utils.bzl

@@ -77,6 +77,7 @@ google_cloud_cpp_grpc_utils_srcs = [
     "internal/minimal_iam_credentials_stub.cc",
     "internal/retry_loop_helpers.cc",
     "internal/streaming_read_rpc.cc",
+    "internal/streaming_read_rpc_logging.cc",
     "internal/streaming_write_rpc.cc",
     "internal/time_utils.cc",
     "internal/unified_grpc_credentials.cc",

google/cloud/internal/resumable_streaming_read_rpc.h

@@ -127,6 +127,10 @@ class ResumableStreamingReadRpc : public StreamingReadRpc<ResponseType> {
     return last_status;
   }
 
+  StreamingRpcMetadata GetRequestMetadata() const override {
+    return impl_ ? impl_->GetRequestMetadata() : StreamingRpcMetadata{};
+  }
+
  private:
   std::unique_ptr<RetryPolicy const> const retry_policy_prototype_;
   std::unique_ptr<BackoffPolicy const> const backoff_policy_prototype_;

google/cloud/internal/resumable_streaming_read_rpc_test.cc

@@ -51,6 +51,7 @@ class MockStreamingReadRpc : public StreamingReadRpc<FakeResponse> {
  public:
   MOCK_METHOD(void, Cancel, (), (override));
   MOCK_METHOD(ReadReturn, Read, (), (override));
+  MOCK_METHOD(StreamingRpcMetadata, GetRequestMetadata, (), (const, override));
 };
 
 struct TestRetryablePolicy {

google/cloud/internal/streaming_read_rpc.cc

@@ -14,11 +14,55 @@
 
 #include "google/cloud/internal/streaming_read_rpc.h"
 #include "google/cloud/log.h"
+#include <grpc/compression.h>
 
 namespace google {
 namespace cloud {
 inline namespace GOOGLE_CLOUD_CPP_NS {
 namespace internal {
+namespace {
+// AFAICT there is no C++ API to get the name, but the C core API is public and
+// documented:
+//    https://grpc.github.io/grpc/core/compression_8h.html
+std::string ToString(grpc_compression_algorithm algo) {
+  char const* name;
+  if (grpc_compression_algorithm_name(algo, &name) == 0) {
+    return "unknown";
+  }
+  return name;
+}
+}  // namespace
+
+StreamingRpcMetadata GetRequestMetadataFromContext(
+    grpc::ClientContext const& context) {
+  StreamingRpcMetadata metadata{
+      // Use invalid header names (starting with ':') to store the
+      // grpc::ClientContext metadata.
+      {":grpc-context-peer", context.peer()},
+      {":grpc-context-compression-algorithm",
+       ToString(context.compression_algorithm())},
+  };
+  auto hint = metadata.end();
+  for (auto const& kv : context.GetServerInitialMetadata()) {
+    // gRPC metadata is stored in `grpc::string_ref`, a type inspired by
+    // `std::string_view`. We need to explicitly convert these to `std::string`.
+    // In addition, we use a prefix to distinguish initial vs. trailing headers.
+    auto key = ":grpc-initial-" + std::string{kv.first.data(), kv.first.size()};
+    auto value = std::string{kv.second.data(), kv.second.size()};
+    hint = std::next(
+        metadata.emplace_hint(hint, std::move(key), std::move(value)));
+  }
+  hint = metadata.end();
+  for (auto const& kv : context.GetServerTrailingMetadata()) {
+    // Same as above, convert `grpc::string_ref` to `std::string`:
+    auto key =
+        ":grpc-trailing-" + std::string{kv.first.data(), kv.first.size()};
+    auto value = std::string{kv.second.data(), kv.second.size()};
+    hint = std::next(
+        metadata.emplace_hint(hint, std::move(key), std::move(value)));
+  }
+  return metadata;
+}
 
 void StreamingReadRpcReportUnhandledError(Status const& status,
                                           char const* tname) {

google/cloud/internal/streaming_read_rpc.h

@@ -21,13 +21,22 @@
 #include "absl/types/variant.h"
 #include <grpcpp/grpcpp.h>
 #include <grpcpp/support/sync_stream.h>
+#include <map>
 #include <memory>
+#include <string>
 
 namespace google {
 namespace cloud {
 inline namespace GOOGLE_CLOUD_CPP_NS {
 namespace internal {
 
+/// A simple representation of request metadata.
+using StreamingRpcMetadata = std::multimap<std::string, std::string>;
+
+/// Return interesting bits of metadata stored in the client context.
+StreamingRpcMetadata GetRequestMetadataFromContext(
+    grpc::ClientContext const& context);
+
 /**
  * Defines the interface for wrappers around gRPC streaming read RPCs.
  *
@@ -50,6 +59,15 @@ class StreamingReadRpc {
 
   /// Return the next element, or the final RPC status.
   virtual absl::variant<Status, ResponseType> Read() = 0;
+
+  /**
+   * Return the request metadata.
+   *
+   * Request metadata is useful for troubleshooting, but may be relatively
+   * expensive to extract.  Library developers should avoid this function in
+   * the critical path.
+   */
+  virtual StreamingRpcMetadata GetRequestMetadata() const = 0;
 };
 
 /// Report the errors in a standalone function to minimize includes
@@ -87,6 +105,11 @@ class StreamingReadRpcImpl : public StreamingReadRpc<ResponseType> {
     return Finish();
   }
 
+  StreamingRpcMetadata GetRequestMetadata() const override {
+    if (!context_) return {};
+    return GetRequestMetadataFromContext(*context_);
+  }
+
  private:
   Status Finish() {
     auto status = MakeStatusFromRpcError(stream_->Finish());
@@ -119,6 +142,8 @@ class StreamingReadRpcError : public StreamingReadRpc<ResponseType> {
 
   absl::variant<Status, ResponseType> Read() override { return status_; }
 
+  StreamingRpcMetadata GetRequestMetadata() const override { return {}; }
+
  private:
   Status status_;
 };

google/cloud/internal/streaming_read_rpc_logging.cc

@@ -0,0 +1,35 @@
+// Copyright 2021 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
+//
+//     http://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/internal/streaming_read_rpc_logging.h"
+#include "google/cloud/internal/absl_str_cat_quiet.h"
+#include "google/cloud/internal/absl_str_join_quiet.h"
+
+namespace google {
+namespace cloud {
+inline namespace GOOGLE_CLOUD_CPP_NS {
+namespace internal {
+
+std::string FormatMetadata(StreamingRpcMetadata const& metadata) {
+  auto formatter = [](std::string* output,
+                      StreamingRpcMetadata::value_type const& p) {
+    *output += absl::StrCat("{", p.first, ": ", p.second, "}");
+  };
+  return absl::StrJoin(metadata.begin(), metadata.end(), ", ", formatter);
+}
+
+}  // namespace internal
+}  // namespace GOOGLE_CLOUD_CPP_NS
+}  // namespace cloud
+}  // namespace google

google/cloud/internal/streaming_read_rpc_logging.h

@@ -32,6 +32,7 @@ namespace google {
 namespace cloud {
 inline namespace GOOGLE_CLOUD_CPP_NS {
 namespace internal {
+std::string FormatMetadata(StreamingRpcMetadata const& metadata);
 
 /**
  * Logging decorator for StreamingReadRpc.
@@ -61,6 +62,12 @@ class StreamingReadRpcLogging : public StreamingReadRpc<ResponseType> {
                    << absl::visit(ResultVisitor(tracing_options_), result);
     return result;
   }
+  StreamingRpcMetadata GetRequestMetadata() const override {
+    auto metadata = reader_->GetRequestMetadata();
+    GCP_LOG(DEBUG) << __func__ << "() >> metadata={" << FormatMetadata(metadata)
+                   << "}";
+    return metadata;
+  }
 
  private:
   class ResultVisitor {

google/cloud/internal/streaming_read_rpc_logging_test.cc

@@ -39,6 +39,7 @@ class MockStreamingReadRpc : public StreamingReadRpc<ResponseType> {
   MOCK_METHOD(void, Cancel, (), (override));
   MOCK_METHOD((absl::variant<Status, google::protobuf::Duration>), Read, (),
               (override));
+  MOCK_METHOD(StreamingRpcMetadata, GetRequestMetadata, (), (const, override));
 };
 
 class StreamingReadRpcLoggingTest : public ::testing::Test {
@@ -100,6 +101,21 @@ TEST_F(StreamingReadRpcLoggingTest, Read) {
   EXPECT_THAT(log_lines, Contains(HasSubstr("Invalid argument.")));
 }
 
+TEST_F(StreamingReadRpcLoggingTest, FormatMetadata) {
+  struct Test {
+    StreamingRpcMetadata metadata;
+    std::string expected;
+  } cases[] = {
+      {{}, ""},
+      {{{"a", "b"}}, "{a: b}"},
+      {{{"a", "b"}, {"k", "v"}}, "{a: b}, {k: v}"},
+  };
+  for (auto const& test : cases) {
+    auto const actual = FormatMetadata(test.metadata);
+    EXPECT_EQ(test.expected, actual);
+  }
+}
+
 }  // namespace
 }  // namespace internal
 }  // namespace GOOGLE_CLOUD_CPP_NS