gRPC: Add support to control envoy generated headers (#34179)

This PR provides gRPC client level control over Envoy generated headers. It currently controls x-envoy-internal and x-forwarded-for (can be expanded if needed)

If false, header will be added. But it can be overridden by setting setSendInternal or setSendXff to false in
Http::AsyncClient::StreamOptions, as per stream control.
If true, header will be removed and can not be overridden by per stream option.
This logic is designed in this way because:

Preserve backwards compatible behavior:
Both headers are still sent by default
If any existing users remove them with StreamOptions, headers are still removed
Still provide the per stream override control
Override here implicitly means setting to false as their default value in AsyncClient::StreamOptions is true.
Thus, per stream override is still available, just in one-way direction: disable on per stream basis
The only thing is that now user can not set StreamOptions to true if they are disabled in config. But it should be fine because:
For existing user, no one should set them to true in StreamOptions as they are already default to true.
For future user, per stream control can still be achieved as stated above.

Signed-off-by: Tianyu Xia <[email protected]>

api/envoy/config/core/v3/grpc_service.proto

@@ -29,6 +29,7 @@ option (udpa.annotations.file_status).package_version_status = ACTIVE;
 message GrpcService {
   option (udpa.annotations.versioning).previous_message_type = "envoy.api.v2.core.GrpcService";
 
+  // [#next-free-field: 6]
   message EnvoyGrpc {
     option (udpa.annotations.versioning).previous_message_type =
         "envoy.api.v2.core.GrpcService.EnvoyGrpc";
@@ -55,6 +56,12 @@ message GrpcService {
     // This limit is applied to individual messages in the streaming response and not the total size of streaming response.
     // Defaults to 0, which means unlimited.
     google.protobuf.UInt32Value max_receive_message_length = 4;
+
+    // This provides gRPC client level control over envoy generated headers.
+    // If false, the header will be sent but it can be overridden by per stream option.
+    // If true, the header will be removed and can not be overridden by per stream option.
+    // Default to false.
+    bool skip_envoy_headers = 5;
   }
 
   // [#next-free-field: 9]

source/common/grpc/async_client_impl.cc

@@ -19,7 +19,8 @@ AsyncClientImpl::AsyncClientImpl(Upstream::ClusterManager& cm,
                                  TimeSource& time_source)
     : max_recv_message_length_(
           PROTOBUF_GET_WRAPPED_OR_DEFAULT(config.envoy_grpc(), max_receive_message_length, 0)),
-      cm_(cm), remote_cluster_name_(config.envoy_grpc().cluster_name()),
+      skip_envoy_headers_(config.envoy_grpc().skip_envoy_headers()), cm_(cm),
+      remote_cluster_name_(config.envoy_grpc().cluster_name()),
       host_name_(config.envoy_grpc().authority()), time_source_(time_source),
       metadata_parser_(THROW_OR_RETURN_VALUE(
           Router::HeaderParser::configure(
@@ -85,6 +86,16 @@ AsyncStreamImpl::AsyncStreamImpl(AsyncClientImpl& parent, absl::string_view serv
   if (!options.retry_policy.has_value() && parent_.retryPolicy().has_value()) {
     options_.setRetryPolicy(*parent_.retryPolicy());
   }
+
+  // Apply parent `skip_envoy_headers_` setting from configuration, if no per-stream
+  // override. (i.e., no override of default stream option from true to false)
+  if (options.send_internal) {
+    options_.setSendInternal(!parent_.skip_envoy_headers_);
+  }
+  if (options.send_xff) {
+    options_.setSendXff(!parent_.skip_envoy_headers_);
+  }
+
   // Configure the maximum frame length
   decoder_.setMaxFrameLength(parent_.max_recv_message_length_);
 

source/common/grpc/async_client_impl.h

@@ -44,6 +44,7 @@ class AsyncClientImpl final : public RawAsyncClient {
 
 private:
   const uint32_t max_recv_message_length_;
+  const bool skip_envoy_headers_;
   Upstream::ClusterManager& cm_;
   const std::string remote_cluster_name_;
   // The host header value in the http transport.

test/common/grpc/grpc_client_integration_test.cc

@@ -64,6 +64,37 @@ TEST_P(GrpcClientIntegrationTest, BasicStream) {
   dispatcher_helper_.runDispatcher();
 }
 
+// A simple request-reply stream, "x-envoy-internal" and `x-forward-for` headers
+// are removed due to grpc service configuration.
+TEST_P(GrpcClientIntegrationTest, BasicStreamRemoveInternalHeaders) {
+  // "x-envoy-internal" and `x-forward-for` headers are only available on Envoy gRPC path.
+  SKIP_IF_GRPC_CLIENT(ClientType::GoogleGrpc);
+  skip_envoy_headers_ = true;
+  initialize();
+  auto stream = createStream(empty_metadata_);
+  stream->sendRequest();
+  stream->sendServerInitialMetadata(empty_metadata_);
+  stream->sendReply();
+  stream->sendServerTrailers(Status::WellKnownGrpcStatus::Ok, "", empty_metadata_);
+  dispatcher_helper_.runDispatcher();
+}
+
+// A simple request-reply stream, "x-envoy-internal" and `x-forward-for` headers
+// are removed due to per stream options which overrides the gRPC service configuration.
+TEST_P(GrpcClientIntegrationTest, BasicStreamRemoveInternalHeadersWithStreamOption) {
+  // "x-envoy-internal" and `x-forward-for` headers are only available on Envoy gRPC path.
+  SKIP_IF_GRPC_CLIENT(ClientType::GoogleGrpc);
+  send_internal_header_stream_option_ = false;
+  send_xff_header_stream_option_ = false;
+  initialize();
+  auto stream = createStream(empty_metadata_);
+  stream->sendRequest();
+  stream->sendServerInitialMetadata(empty_metadata_);
+  stream->sendReply();
+  stream->sendServerTrailers(Status::WellKnownGrpcStatus::Ok, "", empty_metadata_);
+  dispatcher_helper_.runDispatcher();
+}
+
 // Validate that a simple request-reply stream works with bytes metering in Envoy gRPC.
 TEST_P(GrpcClientIntegrationTest, BasicStreamWithBytesMeter) {
   // Currently, only Envoy gRPC's bytes metering is based on the bytes meter in Envoy.

test/common/grpc/grpc_client_integration_test_harness.h

@@ -357,6 +357,8 @@ template <class TimeSystemVariant> class GrpcClientIntegrationTestBase {
           envoy_grpc_max_recv_msg_length);
     }
 
+    config.mutable_envoy_grpc()->set_skip_envoy_headers(skip_envoy_headers_);
+
     fillServiceWideInitialMetadata(config);
     return std::make_unique<AsyncClientImpl>(cm_, config, dispatcher_->timeSource());
   }
@@ -394,6 +396,22 @@ template <class TimeSystemVariant> class GrpcClientIntegrationTestBase {
     EXPECT_EQ("/helloworld.Greeter/SayHello", stream_headers_->get_(":path"));
     EXPECT_EQ("application/grpc", stream_headers_->get_("content-type"));
     EXPECT_EQ("trailers", stream_headers_->get_("te"));
+
+    // "x-envoy-internal" and `x-forward-for` headers are only available in envoy gRPC path.
+    // They will be removed when either envoy gRPC config or stream option is false.
+    if (getClientType() == ClientType::EnvoyGrpc) {
+      if (!skip_envoy_headers_ && send_internal_header_stream_option_) {
+        EXPECT_FALSE(stream_headers_->get_("x-envoy-internal").empty());
+      } else {
+        EXPECT_TRUE(stream_headers_->get_("x-envoy-internal").empty());
+      }
+      if (!skip_envoy_headers_ && send_xff_header_stream_option_) {
+        EXPECT_FALSE(stream_headers_->get_("x-forwarded-for").empty());
+      } else {
+        EXPECT_TRUE(stream_headers_->get_("x-forwarded-for").empty());
+      }
+    }
+
     for (const auto& value : initial_metadata) {
       EXPECT_EQ(value.second, stream_headers_->get_(value.first));
     }
@@ -471,6 +489,8 @@ template <class TimeSystemVariant> class GrpcClientIntegrationTestBase {
     envoy::config::core::v3::Metadata m;
     (*m.mutable_filter_metadata())["com.foo.bar"] = {};
     options.setMetadata(m);
+    options.setSendInternal(send_internal_header_stream_option_);
+    options.setSendXff(send_xff_header_stream_option_);
     stream->grpc_stream_ = grpc_client_->start(*method_descriptor_, *stream, options);
     EXPECT_NE(stream->grpc_stream_, nullptr);
 
@@ -533,6 +553,9 @@ template <class TimeSystemVariant> class GrpcClientIntegrationTestBase {
   Router::MockShadowWriter* mock_shadow_writer_ = new Router::MockShadowWriter();
   Router::ShadowWriterPtr shadow_writer_ptr_{mock_shadow_writer_};
   Network::ClientConnectionPtr client_connection_;
+  bool skip_envoy_headers_{false};
+  bool send_internal_header_stream_option_{true};
+  bool send_xff_header_stream_option_{true};
 };
 
 // The integration test for Envoy gRPC and Google gRPC. It uses `TestRealTimeSystem`.