extension: add test case

source/extensions/upstreams/http/http/BUILD

@@ -33,6 +33,7 @@ envoy_cc_library(
     hdrs = [
         "upstream_request.h",
     ],
+    visibility = ["//visibility:public"],
     deps = [
         "//include/envoy/http:codes_interface",
         "//include/envoy/http:conn_pool_interface",

source/extensions/upstreams/http/http/upstream_request.h

@@ -45,7 +45,7 @@ class HttpConnPool : public Router::GenericConnPool, public Envoy::Http::Connect
 
   bool valid() { return conn_pool_ != nullptr; }
 
-private:
+protected:
   // Points to the actual connection pool to create streams from.
   Envoy::Http::ConnectionPool::Instance* conn_pool_{};
   Envoy::Http::ConnectionPool::Cancellable* conn_pool_stream_handle_{};

source/extensions/upstreams/http/tcp/BUILD

@@ -33,6 +33,7 @@ envoy_cc_library(
     hdrs = [
         "upstream_request.h",
     ],
+    visibility = ["//visibility:public"],
     deps = [
         "//include/envoy/http:codes_interface",
         "//include/envoy/http:filter_interface",

test/integration/BUILD

@@ -1591,3 +1591,19 @@ envoy_cc_test(
         "@envoy_api//envoy/config/core/v3:pkg_cc_proto",
     ],
 )
+
+envoy_cc_test(
+    name = "cluster_upstream_extension_integration_test",
+    srcs = [
+        "cluster_upstream_extension_integration_test.cc",
+    ],
+    deps = [
+        ":http_integration_lib",
+        "//source/common/config:api_version_lib",
+        "//source/common/protobuf",
+        "//test/integration/upstreams:per_host_upstream_config",
+        "//test/test_common:utility_lib",
+        "@envoy_api//envoy/config/bootstrap/v3:pkg_cc_proto",
+        "@envoy_api//envoy/extensions/filters/network/http_connection_manager/v3:pkg_cc_proto",
+    ],
+)

test/integration/cluster_upstream_extension_integration_test.cc

@@ -0,0 +1,92 @@
+#include "envoy/config/bootstrap/v3/bootstrap.pb.h"
+#include "envoy/extensions/filters/network/http_connection_manager/v3/http_connection_manager.pb.h"
+#include "envoy/registry/registry.h"
+#include "envoy/router/router.h"
+
+#include "common/buffer/buffer_impl.h"
+
+#include "test/integration/fake_upstream.h"
+#include "test/integration/http_integration.h"
+#include "test/integration/upstreams/per_host_upstream_config.h"
+#include "test/test_common/registry.h"
+
+#include "gtest/gtest.h"
+
+namespace Envoy {
+
+namespace {
+class ClusterUpstreamExtensionIntegrationTest
+    : public testing::TestWithParam<Network::Address::IpVersion>,
+      public HttpIntegrationTest {
+public:
+  ClusterUpstreamExtensionIntegrationTest()
+      : HttpIntegrationTest(Http::CodecClient::Type::HTTP1, GetParam()) {}
+
+  void populateMetadataTestData(envoy::config::core::v3::Metadata& metadata,
+                                const std::string& key1, const std::string& key2,
+                                const std::string& value) {
+
+    ProtobufWkt::Struct struct_obj;
+    (*struct_obj.mutable_fields())[key2] = ValueUtil::stringValue(value);
+    (*metadata.mutable_filter_metadata())[key1] = struct_obj;
+  }
+
+  void initialize() override {
+    config_helper_.addConfigModifier([this](envoy::config::bootstrap::v3::Bootstrap& bootstrap) {
+      auto* cluster = bootstrap.mutable_static_resources()->mutable_clusters(0);
+      cluster->mutable_upstream_config()->set_name("envoy.filters.connection_pools.http.per_host");
+      cluster->mutable_upstream_config()->mutable_typed_config();
+      populateMetadataTestData(*cluster->mutable_metadata(), "foo", "bar", "cluster-value");
+      populateMetadataTestData(*cluster->mutable_load_assignment()
+                                    ->mutable_endpoints(0)
+                                    ->mutable_lb_endpoints(0)
+                                    ->mutable_metadata(),
+                               "foo", "bar", "host-value");
+    });
+    HttpIntegrationTest::initialize();
+  }
+  PerHostGenericConnPoolFactory per_host_upstream_factory_;
+};
+
+INSTANTIATE_TEST_SUITE_P(IpVersions, ClusterUpstreamExtensionIntegrationTest,
+                         testing::ValuesIn(TestEnvironment::getIpVersionsForTest()),
+                         TestUtility::ipTestParamsToString);
+
+// This test verifies that cluster upstream extensions can fulfill the requirement that they rewrite
+// http headers after cluster and host is selected. See
+// https://github.com/envoyproxy/envoy/issues/12236 This test case should be rewritten once upstream
+// http filters(https://github.com/envoyproxy/envoy/issues/10455) is landed.
+TEST_P(ClusterUpstreamExtensionIntegrationTest,
+       VerifyRequestHeadersAreRewrittenByClusterAndHostMetadata) {
+  initialize();
+  Registry::InjectFactory<Router::GenericConnPoolFactory> registration(per_host_upstream_factory_);
+
+  codec_client_ = makeHttpConnection(lookupPort("http"));
+  auto response = sendRequestAndWaitForResponse(
+      default_request_headers_, 0, Http::TestResponseHeaderMapImpl{{":status", "200"}}, 0);
+  EXPECT_TRUE(upstream_request_->complete());
+
+  {
+    const auto header_values = upstream_request_->headers().get(Http::LowerCaseString("X-foo"));
+    ASSERT_EQ(1, header_values.size());
+    EXPECT_EQ("foo-common", header_values[0]->value().getStringView());
+  }
+  {
+    const auto cluster_header_values =
+        upstream_request_->headers().get(Http::LowerCaseString("X-cluster-foo"));
+    ASSERT_EQ(1, cluster_header_values.size());
+    EXPECT_EQ("cluster-value", cluster_header_values[0]->value().getStringView());
+  }
+  {
+    const auto host_header_values =
+        upstream_request_->headers().get(Http::LowerCaseString("X-host-foo"));
+    ASSERT_EQ(1, host_header_values.size());
+    EXPECT_EQ("host-value", host_header_values[0]->value().getStringView());
+  }
+
+  response->waitForEndStream();
+  ASSERT_TRUE(response->complete());
+  EXPECT_EQ("200", response->headers().getStatusValue());
+}
+} // namespace
+} // namespace Envoy

test/integration/upstreams/BUILD

@@ -0,0 +1,30 @@
+load(
+    "//bazel:envoy_build_system.bzl",
+    "envoy_cc_library",
+    "envoy_package",
+)
+
+licenses(["notice"])  # Apache 2
+
+envoy_package()
+
+envoy_cc_library(
+    name = "per_host_upstream_config",
+    srcs = [
+        "per_host_upstream_config.h",
+    ],
+    deps = [
+        "//include/envoy/http:codes_interface",
+        "//include/envoy/http:conn_pool_interface",
+        "//include/envoy/http:filter_interface",
+        "//include/envoy/upstream:cluster_manager_interface",
+        "//include/envoy/upstream:upstream_interface",
+        "//source/common/common:assert_lib",
+        "//source/common/http:header_map_lib",
+        "//source/common/http:headers_lib",
+        "//source/common/router:router_lib",
+        "//source/common/upstream:load_balancer_lib",
+        "//source/extensions/upstreams/http/http:upstream_request_lib",
+        "//source/extensions/upstreams/http/tcp:upstream_request_lib",
+    ],
+)

test/integration/upstreams/per_host_upstream_config.h

@@ -0,0 +1,122 @@
+#pragma once
+
+#include "envoy/http/conn_pool.h"
+#include "envoy/http/metadata_interface.h"
+#include "envoy/http/protocol.h"
+#include "envoy/router/router.h"
+#include "envoy/upstream/cluster_manager.h"
+#include "envoy/upstream/host_description.h"
+
+#include "common/http/header_map_impl.h"
+#include "common/router/router.h"
+#include "common/router/upstream_request.h"
+
+#include "extensions/upstreams/http/http/upstream_request.h"
+
+namespace Envoy {
+
+namespace {
+void addHeader(Envoy::Http::RequestHeaderMap& header_map, absl::string_view header_name,
+               const envoy::config::core::v3::Metadata& metadata, absl::string_view meta_key_0,
+               absl::string_view meta_key_1) {
+  if (auto filter_metadata = metadata.filter_metadata().find(meta_key_0);
+      filter_metadata != metadata.filter_metadata().end()) {
+    const ProtobufWkt::Struct& data_struct = filter_metadata->second;
+    const auto& fields = data_struct.fields();
+    if (auto iter = fields.find(meta_key_1); iter != fields.end()) {
+      if (iter->second.kind_case() == ProtobufWkt::Value::kStringValue) {
+        header_map.setCopy(Envoy::Http::LowerCaseString(std::string(header_name)),
+                           iter->second.string_value());
+      }
+    }
+  }
+}
+} // namespace
+
+class PerHostHttpUpstream : public Router::GenericUpstream {
+public:
+  PerHostHttpUpstream(Router::UpstreamToDownstream& upstream_request,
+                      Envoy::Http::RequestEncoder* encoder,
+                      Upstream::HostDescriptionConstSharedPtr host)
+      : sub_upstream_(upstream_request, encoder), host_(host) {}
+
+  // GenericUpstream
+  void encodeData(Buffer::Instance& data, bool end_stream) override {
+    sub_upstream_.encodeData(data, end_stream);
+  }
+
+  void encodeMetadata(const Envoy::Http::MetadataMapVector& metadata_map_vector) override {
+    sub_upstream_.encodeMetadata(metadata_map_vector);
+  }
+
+  Http::Status encodeHeaders(const Envoy::Http::RequestHeaderMap& headers,
+                             bool end_stream) override {
+    auto dup = Envoy::Http::RequestHeaderMapImpl::create();
+    Envoy::Http::HeaderMapImpl::copyFrom(*dup, headers);
+    dup->setCopy(Envoy::Http::LowerCaseString("X-foo"), "foo-common");
+    addHeader(*dup, "X-cluster-foo", host_->cluster().metadata(), "foo", "bar");
+    if (host_->metadata() != nullptr) {
+      addHeader(*dup, "X-host-foo", *host_->metadata(), "foo", "bar");
+    }
+    return sub_upstream_.encodeHeaders(*dup, end_stream);
+  }
+
+  void encodeTrailers(const Envoy::Http::RequestTrailerMap& trailers) override {
+    sub_upstream_.encodeTrailers(trailers);
+  }
+
+  void readDisable(bool disable) override { sub_upstream_.readDisable(disable); }
+
+  void resetStream() override { sub_upstream_.resetStream(); }
+
+private:
+  Extensions::Upstreams::Http::Http::HttpUpstream sub_upstream_;
+  Upstream::HostDescriptionConstSharedPtr host_;
+};
+
+class PerHostHttpConnPool : public Extensions::Upstreams::Http::Http::HttpConnPool {
+public:
+  PerHostHttpConnPool(Upstream::ClusterManager& cm, bool is_connect,
+                      const Router::RouteEntry& route_entry,
+                      absl::optional<Envoy::Http::Protocol> downstream_protocol,
+                      Upstream::LoadBalancerContext* ctx)
+      : HttpConnPool(cm, is_connect, route_entry, downstream_protocol, ctx) {}
+
+  void onPoolReady(Envoy::Http::RequestEncoder& callbacks_encoder,
+                   Upstream::HostDescriptionConstSharedPtr host,
+                   const StreamInfo::StreamInfo& info) override {
+    conn_pool_stream_handle_ = nullptr;
+    auto upstream = std::make_unique<PerHostHttpUpstream>(callbacks_->upstreamToDownstream(),
+                                                          &callbacks_encoder, host);
+    callbacks_->onPoolReady(std::move(upstream), host,
+                            callbacks_encoder.getStream().connectionLocalAddress(), info);
+  }
+};
+
+/**
+ * Config registration for the HttpConnPool. @see Router::GenericConnPoolFactory
+ */
+class PerHostGenericConnPoolFactory : public Router::GenericConnPoolFactory {
+public:
+  std::string name() const override { return "envoy.filters.connection_pools.http.per_host"; }
+  std::string category() const override { return "envoy.upstreams"; }
+  Router::GenericConnPoolPtr
+  createGenericConnPool(Upstream::ClusterManager& cm, bool is_connect,
+                        const Router::RouteEntry& route_entry,
+                        absl::optional<Envoy::Http::Protocol> downstream_protocol,
+                        Upstream::LoadBalancerContext* ctx) const override {
+    if (is_connect) {
+      // This example factory doesn't support terminating CONNECT stream.
+      return nullptr;
+    }
+    auto upstream_http_conn_pool = std::make_unique<PerHostHttpConnPool>(
+        cm, is_connect, route_entry, downstream_protocol, ctx);
+    return (upstream_http_conn_pool->valid() ? std::move(upstream_http_conn_pool) : nullptr);
+  }
+
+  ProtobufTypes::MessagePtr createEmptyConfigProto() override {
+    return std::make_unique<ProtobufWkt::Struct>();
+  }
+};
+
+} // namespace Envoy