async client: set xff and charge internal stats

source/common/filter/auth/client_ssl.cc

@@ -13,13 +13,12 @@ namespace Auth {
 namespace ClientSsl {
 
 Config::Config(const Json::Object& config, ThreadLocal::Instance& tls, Upstream::ClusterManager& cm,
-               Event::Dispatcher& dispatcher, Stats::Store& stats_store, Runtime::Loader& runtime,
-               const std::string& local_address)
+               Event::Dispatcher& dispatcher, Stats::Store& stats_store, Runtime::Loader& runtime)
     : tls_(tls), tls_slot_(tls.allocateSlot()), cm_(cm),
       auth_api_cluster_(config.getString("auth_api_cluster")),
       interval_timer_(dispatcher.createTimer([this]() -> void { refreshPrincipals(); })),
       ip_white_list_(config), stats_(generateStats(stats_store, config.getString("stat_prefix"))),
-      runtime_(runtime), local_address_(local_address) {
+      runtime_(runtime) {
 
   if (!cm_.get(auth_api_cluster_)) {
     throw EnvoyException(
@@ -88,7 +87,6 @@ void Config::refreshPrincipals() {
   message->headers().addViaMoveValue(Http::Headers::get().Method, "GET");
   message->headers().addViaMoveValue(Http::Headers::get().Path, "/v1/certs/list/approved");
   message->headers().addViaCopy(Http::Headers::get().Host, auth_api_cluster_);
-  message->headers().addViaCopy(Http::Headers::get().ForwardedFor, local_address_);
   cm_.httpAsyncClientForCluster(auth_api_cluster_)
       .send(std::move(message), *this, Optional<std::chrono::milliseconds>());
 }

source/common/filter/auth/client_ssl.h

@@ -66,8 +66,7 @@ typedef std::shared_ptr<AllowedPrincipals> AllowedPrincipalsPtr;
 class Config : public Http::AsyncClient::Callbacks {
 public:
   Config(const Json::Object& config, ThreadLocal::Instance& tls, Upstream::ClusterManager& cm,
-         Event::Dispatcher& dispatcher, Stats::Store& stats_store, Runtime::Loader& runtime,
-         const std::string& local_address);
+         Event::Dispatcher& dispatcher, Stats::Store& stats_store, Runtime::Loader& runtime);
 
   const AllowedPrincipals& allowedPrincipals();
   const Network::IpWhiteList& ipWhiteList() { return ip_white_list_; }
@@ -91,7 +90,6 @@ class Config : public Http::AsyncClient::Callbacks {
   Network::IpWhiteList ip_white_list_;
   GlobalStats stats_;
   Runtime::Loader& runtime_;
-  const std::string local_address_;
 };
 
 typedef std::shared_ptr<Config> ConfigPtr;

source/common/http/async_client_impl.cc

@@ -1,4 +1,5 @@
 #include "async_client_impl.h"
+#include "headers.h"
 
 namespace Http {
 
@@ -10,10 +11,11 @@ AsyncClientImpl::AsyncClientImpl(const Upstream::Cluster& cluster, Stats::Store&
                                  Event::Dispatcher& dispatcher, const std::string& local_zone_name,
                                  Upstream::ClusterManager& cm, Runtime::Loader& runtime,
                                  Runtime::RandomGenerator& random,
-                                 Router::ShadowWriterPtr&& shadow_writer)
+                                 Router::ShadowWriterPtr&& shadow_writer,
+                                 const std::string& local_address)
     : cluster_(cluster), config_("http.async-client.", local_zone_name, stats_store, cm, runtime,
                                  random, std::move(shadow_writer)),
-      dispatcher_(dispatcher) {}
+      dispatcher_(dispatcher), local_address_(local_address) {}
 
 AsyncClientImpl::~AsyncClientImpl() { ASSERT(active_requests_.empty()); }
 
@@ -39,6 +41,8 @@ AsyncRequestImpl::AsyncRequestImpl(MessagePtr&& request, AsyncClientImpl& parent
       request_info_(EMPTY_STRING), route_(parent_.cluster_.name(), timeout) {
 
   router_.setDecoderFilterCallbacks(*this);
+  request_->headers().addViaMoveValue(Headers::get().EnvoyInternalRequest, "true");
+  request_->headers().addViaCopy(Headers::get().ForwardedFor, parent_.local_address_);
   router_.decodeHeaders(request_->headers(), !request_->body());
   if (!complete_ && request_->body()) {
     router_.decodeData(*request_->body(), true);

source/common/http/async_client_impl.h

@@ -24,7 +24,8 @@ class AsyncClientImpl final : public AsyncClient {
   AsyncClientImpl(const Upstream::Cluster& cluster, Stats::Store& stats_store,
                   Event::Dispatcher& dispatcher, const std::string& local_zone_name,
                   Upstream::ClusterManager& cm, Runtime::Loader& runtime,
-                  Runtime::RandomGenerator& random, Router::ShadowWriterPtr&& shadow_writer);
+                  Runtime::RandomGenerator& random, Router::ShadowWriterPtr&& shadow_writer,
+                  const std::string& local_address);
   ~AsyncClientImpl();
 
   // Http::AsyncClient
@@ -36,6 +37,7 @@ class AsyncClientImpl final : public AsyncClient {
   Router::FilterConfig config_;
   Event::Dispatcher& dispatcher_;
   std::list<std::unique_ptr<AsyncRequestImpl>> active_requests_;
+  const std::string local_address_;
 
   friend class AsyncRequestImpl;
 };

source/common/upstream/cluster_manager_impl.cc

@@ -20,7 +20,8 @@ ClusterManagerImpl::ClusterManagerImpl(const Json::Object& config, Stats::Store&
                                        Network::DnsResolver& dns_resolver,
                                        Ssl::ContextManager& ssl_context_manager,
                                        Runtime::Loader& runtime, Runtime::RandomGenerator& random,
-                                       const std::string& local_zone_name)
+                                       const std::string& local_zone_name,
+                                       const std::string& local_address)
     : runtime_(runtime), tls_(tls), stats_(stats), thread_local_slot_(tls.allocateSlot()) {
 
   std::vector<Json::Object> clusters = config.getObjectArray("clusters");
@@ -43,11 +44,11 @@ ClusterManagerImpl::ClusterManagerImpl(const Json::Object& config, Stats::Store&
   }
 
   tls.set(thread_local_slot_,
-          [this, &stats, &runtime, &random, local_zone_name](Event::Dispatcher& dispatcher)
-              -> ThreadLocal::ThreadLocalObjectPtr {
-                return ThreadLocal::ThreadLocalObjectPtr{new ThreadLocalClusterManagerImpl(
-                    *this, dispatcher, runtime, random, local_zone_name)};
-              });
+          [this, &stats, &runtime, &random, local_zone_name, local_address](
+              Event::Dispatcher& dispatcher) -> ThreadLocal::ThreadLocalObjectPtr {
+            return ThreadLocal::ThreadLocalObjectPtr{new ThreadLocalClusterManagerImpl(
+                *this, dispatcher, runtime, random, local_zone_name, local_address)};
+          });
 
   // To avoid threading issues, for those clusters that start with hosts already in them (like
   // the static cluster), we need to post an update onto each thread to notify them of the update.
@@ -204,11 +205,13 @@ Http::AsyncClient& ClusterManagerImpl::httpAsyncClientForCluster(const std::stri
 
 ClusterManagerImpl::ThreadLocalClusterManagerImpl::ThreadLocalClusterManagerImpl(
     ClusterManagerImpl& parent, Event::Dispatcher& dispatcher, Runtime::Loader& runtime,
-    Runtime::RandomGenerator& random, const std::string& local_zone_name)
+    Runtime::RandomGenerator& random, const std::string& local_zone_name,
+    const std::string& local_address)
     : parent_(parent), dispatcher_(dispatcher) {
   for (auto& cluster : parent.primary_clusters_) {
-    thread_local_clusters_[cluster.first].reset(new ClusterEntry(
-        *this, *cluster.second, runtime, random, parent.stats_, dispatcher, local_zone_name));
+    thread_local_clusters_[cluster.first].reset(new ClusterEntry(*this, *cluster.second, runtime,
+                                                                 random, parent.stats_, dispatcher,
+                                                                 local_zone_name, local_address));
   }
 
   for (auto& cluster : thread_local_clusters_) {
@@ -275,11 +278,11 @@ void ClusterManagerImpl::ThreadLocalClusterManagerImpl::shutdown() {
 ClusterManagerImpl::ThreadLocalClusterManagerImpl::ClusterEntry::ClusterEntry(
     ThreadLocalClusterManagerImpl& parent, const Cluster& cluster, Runtime::Loader& runtime,
     Runtime::RandomGenerator& random, Stats::Store& stats_store, Event::Dispatcher& dispatcher,
-    const std::string& local_zone_name)
+    const std::string& local_zone_name, const std::string& local_address)
     : parent_(parent), primary_cluster_(cluster),
-      http_async_client_(cluster, stats_store, dispatcher, local_zone_name, parent.parent_, runtime,
-                         random,
-                         Router::ShadowWriterPtr{new Router::ShadowWriterImpl(parent.parent_)}) {
+      http_async_client_(
+          cluster, stats_store, dispatcher, local_zone_name, parent.parent_, runtime, random,
+          Router::ShadowWriterPtr{new Router::ShadowWriterImpl(parent.parent_)}, local_address) {
 
   switch (cluster.lbType()) {
   case LoadBalancerType::LeastRequest: {

source/common/upstream/cluster_manager_impl.h

@@ -21,7 +21,7 @@ class ClusterManagerImpl : public ClusterManager {
   ClusterManagerImpl(const Json::Object& config, Stats::Store& stats, ThreadLocal::Instance& tls,
                      Network::DnsResolver& dns_resolver, Ssl::ContextManager& ssl_context_manager,
                      Runtime::Loader& runtime, Runtime::RandomGenerator& random,
-                     const std::string& local_zone_name);
+                     const std::string& local_zone_name, const std::string& local_address);
 
   // Upstream::ClusterManager
   void setInitializedCb(std::function<void()> callback) override {
@@ -74,7 +74,7 @@ class ClusterManagerImpl : public ClusterManager {
       ClusterEntry(ThreadLocalClusterManagerImpl& parent, const Cluster& cluster,
                    Runtime::Loader& runtime, Runtime::RandomGenerator& random,
                    Stats::Store& stats_store, Event::Dispatcher& dispatcher,
-                   const std::string& local_zone_name);
+                   const std::string& local_zone_name, const std::string& local_address);
 
       Http::ConnectionPool::Instance* connPool(ResourcePriority priority);
 
@@ -89,7 +89,8 @@ class ClusterManagerImpl : public ClusterManager {
 
     ThreadLocalClusterManagerImpl(ClusterManagerImpl& parent, Event::Dispatcher& dispatcher,
                                   Runtime::Loader& runtime, Runtime::RandomGenerator& random,
-                                  const std::string& local_zone_name);
+                                  const std::string& local_zone_name,
+                                  const std::string& local_address);
     void drainConnPools(HostPtr old_host, ConnPoolsContainer& container);
     static void updateClusterMembership(const std::string& name, ConstHostVectorPtr hosts,
                                         ConstHostVectorPtr healthy_hosts,

source/server/config/network/client_ssl_auth.cc

@@ -22,7 +22,7 @@ class ClientSslAuthConfigFactory : public NetworkFilterConfigFactory {
 
     Filter::Auth::ClientSsl::ConfigPtr config(new Filter::Auth::ClientSsl::Config(
         json_config, server.threadLocal(), server.clusterManager(), server.dispatcher(),
-        server.stats(), server.runtime(), server.getLocalAddress()));
+        server.stats(), server.runtime()));
     return [config](Network::Connection& connection) -> void {
       connection.addReadFilter(
           Network::ReadFilterPtr{new Filter::Auth::ClientSsl::Instance(config)});

source/server/configuration_impl.cc

@@ -21,7 +21,7 @@ void MainImpl::initialize(const std::string& file_path) {
   cluster_manager_.reset(new Upstream::ProdClusterManagerImpl(
       loader.getObject("cluster_manager"), server_.stats(), server_.threadLocal(),
       server_.dnsResolver(), server_.sslContextManager(), server_.runtime(), server_.random(),
-      server_.options().serviceZone()));
+      server_.options().serviceZone(), server_.getLocalAddress()));
 
   std::vector<Json::Object> listeners = loader.getObjectArray("listeners");
   log().notice("loading {} listener(s)", listeners.size());

test/common/filter/auth/client_ssl_test.cc

@@ -42,7 +42,7 @@ class ClientSslAuthFilterTest : public testing::Test {
     Json::StringLoader loader(json);
     EXPECT_CALL(cm_, get("vpn"));
     setupRequest();
-    config_.reset(new Config(loader, tls_, cm_, dispatcher_, stats_store_, runtime_, "127.0.0.1"));
+    config_.reset(new Config(loader, tls_, cm_, dispatcher_, stats_store_, runtime_));
 
     createAuthFilter();
   }
@@ -56,9 +56,8 @@ class ClientSslAuthFilterTest : public testing::Test {
     EXPECT_CALL(cm_, httpAsyncClientForCluster("vpn")).WillOnce(ReturnRef(cm_.async_client_));
     EXPECT_CALL(cm_.async_client_, send_(_, _, _))
         .WillOnce(
-            Invoke([this](Http::MessagePtr& request, Http::AsyncClient::Callbacks& callbacks,
+            Invoke([this](Http::MessagePtr&, Http::AsyncClient::Callbacks& callbacks,
                           Optional<std::chrono::milliseconds>) -> Http::AsyncClient::Request* {
-              EXPECT_EQ("127.0.0.1", request->headers().get("x-forwarded-for"));
               callbacks_ = &callbacks;
               return &request_;
             }));
@@ -88,8 +87,7 @@ TEST_F(ClientSslAuthFilterTest, NoCluster) {
 
   Json::StringLoader loader(json);
   EXPECT_CALL(cm_, get("bad_cluster")).WillOnce(Return(nullptr));
-  EXPECT_THROW(new Config(loader, tls_, cm_, dispatcher_, stats_store_, runtime_, "127.0.0.1"),
-               EnvoyException);
+  EXPECT_THROW(new Config(loader, tls_, cm_, dispatcher_, stats_store_, runtime_), EnvoyException);
 }
 
 TEST_F(ClientSslAuthFilterTest, Basic) {