implement

Signed-off-by: Kuat Yessenov <[email protected]>

docs/root/configuration/advanced/well_known_dynamic_metadata.rst

@@ -31,6 +31,7 @@ The following Envoy filters can be configured to consume dynamic metadata emitte
 * :ref:`External Authorization Filter via the metadata context namespaces
   <envoy_v3_api_field_extensions.filters.http.ext_authz.v3.ExtAuthz.metadata_context_namespaces>`
 * :ref:`RateLimit Filter limit override <config_http_filters_rate_limit_override_dynamic_metadata>`
+* :ref:`Original destination listener filter <config_listener_filters_original_dst>`
 
 .. _shared_dynamic_metadata:
 

docs/root/configuration/advanced/well_known_filter_state.rst

@@ -30,6 +30,12 @@ The following list of filter state objects are consumed by Envoy extensions:
      - | :ref:`Dynamic forward proxy <envoy_v3_api_msg_extensions.clusters.dynamic_forward_proxy.v3.ClusterConfig>`
        | upstream port override on a per-connection basis.
        | Accepts a port number string as a constructor.
+   * - ``envoy.filters.listener.original_dst.local_ip``
+     - | :ref:`Original destination listener filter <config_listener_filters_original_dst>`
+       | destination address selection for internal listeners.
+   * - ``envoy.filters.listener.original_dst.remote_ip``
+     - | :ref:`Original destination listener filter <config_listener_filters_original_dst>`
+       | source address selection for internal listeners.
 
 
 The filter state object fields can be used in the format strings. For example,

docs/root/configuration/listeners/listener_filters/original_dst_filter.rst

@@ -40,5 +40,26 @@ rather than the address at which the listener is listening at. Furthermore, :ref
 destination cluster <arch_overview_service_discovery_types_original_destination>` may be used to
 forward HTTP requests or TCP connections to the restored destination address.
 
+Internal listeners
+------------------
+
+Original destination listener filter reads the dynamic metadata and the filter
+state objects on the user space sockets handled by the :ref:`internal listeners
+<config_internal_listener>`.
+
+Dynamic metadata for the destination address is expected to be placed into the
+key `envoy.filters.listener.original_dst` under the field `local` and should
+contain a string with an IP and a port address.
+
+The filter state objects consumed by this filter are:
+
+* `envoy.filters.listener.original_dst.local_ip` for the destination address
+* `envoy.filters.listener.original_dst.source_ip` for the source address
+
+:ref:`Internal upstream transport <config_internal_upstream_transport>` allows
+passing dynamic metadata from a host to the socket metadata and filter state
+objects that are shared with the upstream connection through a user space
+socket.
+
 * This filter should be configured with the type URL ``type.googleapis.com/envoy.extensions.filters.listener.original_dst.v3.OriginalDst``.
 * :ref:`v3 API reference <envoy_v3_api_msg_extensions.filters.listener.original_dst.v3.OriginalDst>`

docs/root/configuration/other_features/internal_listener.rst

@@ -36,6 +36,8 @@ Third, an upstream cluster must include an endpoint with an internal address ref
             envoy_internal_address:
               server_listener_name: internal_listener
 
+.. _config_internal_upstream_transport:
+
 Internal upstream transport
 ---------------------------
 

source/extensions/filters/listener/original_dst/BUILD

@@ -17,12 +17,17 @@ envoy_cc_library(
     srcs = ["original_dst.cc"],
     hdrs = ["original_dst.h"],
     deps = [
+        "//envoy/common:hashable_interface",
         "//envoy/network:filter_interface",
         "//envoy/network:listen_socket_interface",
+        "//envoy/stream_info:filter_state_interface",
         "//source/common/common:assert_lib",
+        "//source/common/common:hash_lib",
         "//source/common/common:minimal_logger_lib",
+        "//source/common/config:metadata_lib",
         "//source/common/network:upstream_socket_options_filter_state_lib",
         "//source/common/network:utility_lib",
+        "//source/common/singleton:const_singleton",
     ],
 )
 

source/extensions/filters/listener/original_dst/config.cc

@@ -50,7 +50,7 @@ class OriginalDstConfigFactory : public Server::Configuration::NamedListenerFilt
     return std::make_unique<envoy::extensions::filters::listener::original_dst::v3::OriginalDst>();
   }
 
-  std::string name() const override { return "envoy.filters.listener.original_dst"; }
+  std::string name() const override { return FilterNames::get().Name; }
 };
 
 /**

source/extensions/filters/listener/original_dst/original_dst.cc

@@ -3,6 +3,8 @@
 #include "envoy/network/listen_socket.h"
 
 #include "source/common/common/assert.h"
+#include "source/common/common/hash.h"
+#include "source/common/config/metadata.h"
 #include "source/common/network/socket_option_factory.h"
 #include "source/common/network/upstream_socket_options_filter_state.h"
 #include "source/common/network/utility.h"
@@ -12,6 +14,10 @@ namespace Extensions {
 namespace ListenerFilters {
 namespace OriginalDst {
 
+absl::optional<uint64_t> AddressObject::hash() const {
+  return Envoy::HashUtil::xxHash64(address_->asStringView());
+}
+
 Network::Address::InstanceConstSharedPtr OriginalDstFilter::getOriginalDst(Network::Socket& sock) {
   return Network::Utility::getOriginalDst(sock);
 }
@@ -20,7 +26,8 @@ Network::FilterStatus OriginalDstFilter::onAccept(Network::ListenerFilterCallbac
   ENVOY_LOG(trace, "original_dst: new connection accepted");
   Network::ConnectionSocket& socket = cb.socket();
 
-  if (socket.addressType() == Network::Address::Type::Ip) {
+  switch (socket.addressType()) {
+  case Network::Address::Type::Ip: {
     Network::Address::InstanceConstSharedPtr original_local_address = getOriginalDst(socket);
     // A listener that has the use_original_dst flag set to true can still receive
     // connections that are NOT redirected using iptables. If a connection was not redirected,
@@ -68,6 +75,41 @@ Network::FilterStatus OriginalDstFilter::onAccept(Network::ListenerFilterCallbac
       // Restore the local address to the original one.
       socket.connectionInfoProvider().restoreLocalAddress(original_local_address);
     }
+    break;
+  }
+  case Network::Address::Type::EnvoyInternal: {
+    const auto& local_value = Config::Metadata::metadataValue(
+        &cb.dynamicMetadata(), FilterNames::get().Name, FilterNames::get().LocalField);
+    if (local_value.has_string_value()) {
+      const auto local_address = Envoy::Network::Utility::parseInternetAddressAndPortNoThrow(
+          local_value.string_value(), /*v6only=*/false);
+      if (local_address) {
+        ENVOY_LOG_MISC(debug, "original_dst: set destination from metadata to {}",
+                       local_address->asString());
+        socket.connectionInfoProvider().restoreLocalAddress(local_address);
+      } else {
+        ENVOY_LOG_MISC(debug, "original_dst: failed to parse address: {}",
+                       local_address->asString());
+      }
+    } else {
+      const auto* local_object =
+          cb.filterState().getDataReadOnly<AddressObject>(FilterNames::get().LocalFilterStateKey);
+      if (local_object) {
+        ENVOY_LOG_MISC(debug, "original_dst: set destination from filter state to {}",
+                       local_object->address()->asString());
+        socket.connectionInfoProvider().restoreLocalAddress(local_object->address());
+      }
+    }
+    const auto* remote_object =
+        cb.filterState().getDataReadOnly<AddressObject>(FilterNames::get().RemoteFilterStateKey);
+    if (remote_object) {
+      ENVOY_LOG_MISC(debug, "original_dst: set source from filter state to {}",
+                     remote_object->address()->asString());
+      socket.connectionInfoProvider().setRemoteAddress(remote_object->address());
+    }
+  }
+  default:
+    break;
   }
 
   return Network::FilterStatus::Continue;

source/extensions/filters/listener/original_dst/original_dst.h

@@ -1,14 +1,38 @@
 #pragma once
 
+#include "envoy/common/hashable.h"
 #include "envoy/network/filter.h"
+#include "envoy/stream_info/filter_state.h"
 
 #include "source/common/common/logger.h"
+#include "source/common/singleton/const_singleton.h"
 
 namespace Envoy {
 namespace Extensions {
 namespace ListenerFilters {
 namespace OriginalDst {
 
+class FilterNameValues {
+public:
+  const std::string Name = "envoy.filters.listener.original_dst";
+  const std::string LocalField = "local";
+  const std::string LocalFilterStateKey = "envoy.filters.listener.original_dst.local_ip";
+  const std::string RemoteFilterStateKey = "envoy.filters.listener.original_dst.remote_ip";
+};
+
+using FilterNames = ConstSingleton<FilterNameValues>;
+
+class AddressObject : public StreamInfo::FilterState::Object, public Hashable {
+public:
+  AddressObject(const Network::Address::InstanceConstSharedPtr& address) : address_(address) {}
+  absl::optional<std::string> serializeAsString() const override { return address_->asString(); }
+  absl::optional<uint64_t> hash() const override;
+  const Network::Address::InstanceConstSharedPtr& address() const { return address_; }
+
+private:
+  const Network::Address::InstanceConstSharedPtr address_;
+};
+
 /**
  * Implementation of an original destination listener filter.
  */

test/extensions/filters/listener/original_dst/BUILD

@@ -39,3 +39,14 @@ envoy_cc_test(
         "@envoy_api//envoy/extensions/filters/network/http_connection_manager/v3:pkg_cc_proto",
     ],
 )
+
+envoy_cc_test(
+    name = "original_dst_test",
+    srcs = ["original_dst_test.cc"],
+    deps = [
+        "//source/common/network:utility_lib",
+        "//source/extensions/filters/listener/original_dst:original_dst_lib",
+        "//test/mocks/network:network_mocks",
+        "//test/test_common:utility_lib",
+    ],
+)

test/extensions/filters/listener/original_dst/original_dst_test.cc

@@ -0,0 +1,71 @@
+#include "source/common/network/utility.h"
+#include "source/extensions/filters/listener/original_dst/original_dst.h"
+
+#include "test/mocks/network/mocks.h"
+#include "test/test_common/utility.h"
+
+#include "gtest/gtest.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace ListenerFilters {
+namespace OriginalDst {
+namespace {
+
+using testing::Return;
+using testing::ReturnRef;
+
+class OriginalDstTest : public testing::Test {
+public:
+  OriginalDstTest() : filter_(envoy::config::core::v3::TrafficDirection::OUTBOUND) {
+    EXPECT_CALL(cb_, socket()).WillRepeatedly(ReturnRef(socket_));
+    EXPECT_CALL(cb_, dynamicMetadata()).WillRepeatedly(ReturnRef(metadata_));
+    EXPECT_CALL(cb_, filterState()).Times(testing::AtLeast(1));
+  }
+
+  OriginalDstFilter filter_;
+  Network::MockListenerFilterCallbacks cb_;
+  Network::MockConnectionSocket socket_;
+  envoy::config::core::v3::Metadata metadata_;
+};
+
+TEST_F(OriginalDstTest, InternalNone) {
+  EXPECT_CALL(socket_, addressType()).WillRepeatedly(Return(Network::Address::Type::EnvoyInternal));
+  filter_.onAccept(cb_);
+  EXPECT_FALSE(socket_.connectionInfoProvider().localAddressRestored());
+}
+
+TEST_F(OriginalDstTest, InternalDynamicMetadata) {
+  EXPECT_CALL(socket_, addressType()).WillRepeatedly(Return(Network::Address::Type::EnvoyInternal));
+  TestUtility::loadFromYaml(R"EOF(
+    filter_metadata:
+      envoy.filters.listener.original_dst:
+        local: 127.0.0.1:8080
+  )EOF",
+                            metadata_);
+  filter_.onAccept(cb_);
+  EXPECT_TRUE(socket_.connectionInfoProvider().localAddressRestored());
+  EXPECT_EQ("127.0.0.1:8080", socket_.connectionInfoProvider().localAddress()->asString());
+}
+
+TEST_F(OriginalDstTest, InternalFilterState) {
+  EXPECT_CALL(socket_, addressType()).WillRepeatedly(Return(Network::Address::Type::EnvoyInternal));
+  const auto local = Network::Utility::parseInternetAddress("10.20.30.40", 456, false);
+  const auto remote = Network::Utility::parseInternetAddress("127.0.0.1", 8000, false);
+  cb_.filter_state_.setData(
+      "envoy.filters.listener.original_dst.local_ip", std::make_shared<AddressObject>(local),
+      StreamInfo::FilterState::StateType::Mutable, StreamInfo::FilterState::LifeSpan::Connection);
+  cb_.filter_state_.setData(
+      "envoy.filters.listener.original_dst.remote_ip", std::make_shared<AddressObject>(remote),
+      StreamInfo::FilterState::StateType::Mutable, StreamInfo::FilterState::LifeSpan::Connection);
+  filter_.onAccept(cb_);
+  EXPECT_TRUE(socket_.connectionInfoProvider().localAddressRestored());
+  EXPECT_EQ(local->asString(), socket_.connectionInfoProvider().localAddress()->asString());
+  EXPECT_EQ(remote->asString(), socket_.connectionInfoProvider().remoteAddress()->asString());
+}
+
+} // namespace
+} // namespace OriginalDst
+} // namespace ListenerFilters
+} // namespace Extensions
+} // namespace Envoy