dfp: add support for DFP honoring dynamic_host filter state when configured (#39605)

Signed-off-by: Rohit Agrawal <[email protected]>
Co-authored-by: yanavlasov <[email protected]>

Author: agrawroh

api/envoy/extensions/filters/http/dynamic_forward_proxy/v3/dynamic_forward_proxy.proto

@@ -41,6 +41,17 @@ message FilterConfig {
   // ``envoy.stream.upstream_address`` (See
   // :repo:`upstream_address.h<source/common/stream_info/upstream_address.h>`).
   bool save_upstream_address = 2;
+
+  // When this flag is set, the filter will check for the ``envoy.upstream.dynamic_host``
+  // and/or ``envoy.upstream.dynamic_port`` filter state values before using the HTTP
+  // Host header for DNS resolution. This provides consistency with the
+  // :ref:`SNI dynamic forward proxy <envoy_v3_api_msg_extensions.filters.network.sni_dynamic_forward_proxy.v3.FilterConfig>` and
+  // :ref:`UDP dynamic forward proxy <envoy_v3_api_msg_extensions.filters.udp.udp_proxy.session.dynamic_forward_proxy.v3.FilterConfig>`
+  // filters behavior when enabled.
+  //
+  // If the flag is not set (default), the filter will use the HTTP Host header
+  // for DNS resolution, maintaining backward compatibility.
+  bool allow_dynamic_host_from_filter_state = 4;
 }
 
 // Per route Configuration for the dynamic forward proxy HTTP filter.

changelogs/current.yaml

@@ -363,5 +363,12 @@ new_features:
     Added envoy_v3_api_field_extensions.upstreams.http.v3.Http3ProtocolOptions.disable_connection_flow_control_for_streams, an experimental
     option for disabling connection level flow control for streams. This is useful in situations where the streams share the same
     connection but originate from different end-clients, so that each stream can make progress independently at non-front-line proxies.
+- area: dfp
+  change: |
+    Added :ref:`allow_dynamic_host_from_filter_state
+    <envoy_v3_api_field_extensions.filters.http.dynamic_forward_proxy.v3.FilterConfig.allow_dynamic_host_from_filter_state>`
+    flag to HTTP Dynamic Forward Proxy filter. When enabled, the filter will check for ``envoy.upstream.dynamic_host`` and
+    ``envoy.upstream.dynamic_port`` filter state values before using the HTTP Host header, providing consistency with SNI
+    and UDP DFP filters. When disabled (default), maintains backward compatibility by using the HTTP Host header directly.
 
 deprecated:

docs/root/configuration/http/http_filters/dynamic_forward_proxy_filter.rst

@@ -75,6 +75,59 @@ To use :ref:`AppleDnsResolverConfig<envoy_v3_api_msg_extensions.network.dns_reso
 .. literalinclude:: _include/dns-cache-circuit-breaker-apple.yaml
     :language: yaml
 
+Dynamic Host Resolution via Filter State
+----------------------------------------
+
+The dynamic forward proxy filter supports dynamic host and port resolution through filter state when the
+:ref:`allow_dynamic_host_from_filter_state <envoy_v3_api_field_extensions.filters.http.dynamic_forward_proxy.v3.FilterConfig.allow_dynamic_host_from_filter_state>`
+option is enabled. When this feature is enabled, the filter will check for the following filter state values
+before falling back to the HTTP Host header:
+
+* ``envoy.upstream.dynamic_host``: Specifies the target host for DNS resolution and connection.
+* ``envoy.upstream.dynamic_port``: Specifies the target port for connection.
+
+When the ``allow_dynamic_host_from_filter_state`` flag is enabled, the HTTP Dynamic Forward Proxy
+filter will check for filter state values, providing consistency with the SNI and UDP Dynamic Forward
+Proxy filters and allowing the same filter state mechanism to work across all proxy types.
+
+Host Resolution Priority
+^^^^^^^^^^^^^^^^^^^^^^^^
+
+The filter resolves the target host and port using the following priority order:
+
+When ``allow_dynamic_host_from_filter_state`` is enabled:
+
+1. **Filter State Values:** ``envoy.upstream.dynamic_host`` and ``envoy.upstream.dynamic_port`` from the stream's filter state.
+2. **Host Rewrite Configuration:** Values specified in the route's or virtual host's ``typed_per_filter_config``.
+3. **HTTP Host Header:** The host and port from the incoming HTTP request's Host/Authority header.
+
+When ``allow_dynamic_host_from_filter_state`` is disabled (default):
+
+1. **Host Rewrite Configuration:** Values specified in the route's or virtual host's ``typed_per_filter_config``.
+2. **HTTP Host Header:** The host and port from the incoming HTTP request's Host/Authority header.
+
+Filter State Usage Example
+^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+The filter state values can be set by other filters in the filter chain before the dynamic forward proxy
+filter processes the request. For example, using a Set Filter State HTTP filter:
+
+.. code-block:: yaml
+
+  http_filters:
+  - name: envoy.filters.http.set_filter_state
+    typed_config:
+      "@type": type.googleapis.com/envoy.extensions.filters.http.set_filter_state.v3.Config
+      on_request_headers:
+      - object_key: "envoy.upstream.dynamic_host"
+        format_string:
+          text_format_source:
+            inline_string: "example.com"
+      - object_key: "envoy.upstream.dynamic_port"
+        format_string:
+          text_format_source:
+            inline_string: "443"
+
 Statistics
 ----------
 

source/extensions/clusters/dynamic_forward_proxy/cluster.cc

@@ -366,36 +366,56 @@ Cluster::LoadBalancer::chooseHost(Upstream::LoadBalancerContext* context) {
     return {nullptr};
   }
 
+  // For host lookup, we need to make sure to match the host of any DNS cache
+  // insert. Two code points currently do DNS cache insert: the http DFP filter,
+  // which inserts for HTTP traffic, and sets port based on the cluster's
+  // security level, and the SNI DFP network filter which sets port based on
+  // stream metadata, or configuration (which is then added as stream metadata).
+  const bool is_secure = cluster_.info()
+                             ->transportSocketMatcher()
+                             .resolve(nullptr, nullptr)
+                             .factory_.implementsSecureTransport();
+  const uint32_t default_port = is_secure ? 443 : 80;
+
+  const auto* stream_info = context->requestStreamInfo();
   const Router::StringAccessor* dynamic_host_filter_state = nullptr;
-  if (context->requestStreamInfo()) {
-    dynamic_host_filter_state =
-        context->requestStreamInfo()->filterState()->getDataReadOnly<Router::StringAccessor>(
-            DynamicHostFilterStateKey);
+  if (stream_info) {
+    dynamic_host_filter_state = stream_info->filterState().getDataReadOnly<Router::StringAccessor>(
+        DynamicHostFilterStateKey);
   }
 
   absl::string_view raw_host;
+  uint32_t port = default_port;
+
   if (dynamic_host_filter_state) {
+    // Use dynamic host from filter state if available.
     raw_host = dynamic_host_filter_state->asString();
+
+    // Try to get port from filter state first.
+    const StreamInfo::UInt32Accessor* dynamic_port_filter_state =
+        stream_info->filterState().getDataReadOnly<StreamInfo::UInt32Accessor>(
+            DynamicPortFilterStateKey);
+
+    if (dynamic_port_filter_state != nullptr && dynamic_port_filter_state->value() > 0 &&
+        dynamic_port_filter_state->value() <= 65535) {
+      // Use dynamic port from filter state if available.
+      port = dynamic_port_filter_state->value();
+    }
+    // If no dynamic port is in filter state, we just use the default_port.
   } else if (context->downstreamHeaders()) {
     raw_host = context->downstreamHeaders()->getHostValue();
+    // When no filter state is used, we let ``normalizeHostForDfp()`` handle the port parsing.
   } else if (context->downstreamConnection()) {
     raw_host = context->downstreamConnection()->requestedServerName();
   }
 
-  // For host lookup, we need to make sure to match the host of any DNS cache
-  // insert. Two code points currently do DNS cache insert: the http DFP filter,
-  // which inserts for HTTP traffic, and sets port based on the cluster's
-  // security level, and the SNI DFP network filter which sets port based on
-  // stream metadata, or configuration (which is then added as stream metadata).
-  const bool is_secure = cluster_.info()
-                             ->transportSocketMatcher()
-                             .resolve(nullptr, nullptr)
-                             .factory_.implementsSecureTransport();
-  uint32_t port = is_secure ? 443 : 80;
-  if (context->requestStreamInfo()) {
+  // We always check for dynamic port from filter state, even if the host is not from filter state.
+  // This is to maintain the backward compatibility with the existing SNI filter behavior.
+  if (stream_info && !dynamic_host_filter_state) {
     const StreamInfo::UInt32Accessor* dynamic_port_filter_state =
-        context->requestStreamInfo()->filterState()->getDataReadOnly<StreamInfo::UInt32Accessor>(
+        stream_info->filterState().getDataReadOnly<StreamInfo::UInt32Accessor>(
             DynamicPortFilterStateKey);
+
     if (dynamic_port_filter_state != nullptr && dynamic_port_filter_state->value() > 0 &&
         dynamic_port_filter_state->value() <= 65535) {
       port = dynamic_port_filter_state->value();
@@ -406,6 +426,7 @@ Cluster::LoadBalancer::chooseHost(Upstream::LoadBalancerContext* context) {
     ENVOY_LOG(debug, "host empty");
     return {nullptr, "empty_host_header"};
   }
+
   std::string hostname =
       Common::DynamicForwardProxy::DnsHostInfo::normalizeHostForDfp(raw_host, port);
 
@@ -422,7 +443,7 @@ Cluster::LoadBalancer::chooseHost(Upstream::LoadBalancerContext* context) {
     return {host};
   }
 
-  // If the host is not found, the DFP cluster cluster can now do asynchronous lookup.
+  // If the host is not found, the DFP cluster can now do asynchronous lookup.
   Upstream::ResourceAutoIncDecPtr handle = cluster_.dns_cache_->canCreateDnsRequest();
 
   // Return an immediate failure if there's too many requests already.

source/extensions/filters/http/dynamic_forward_proxy/proxy_filter.cc

@@ -4,6 +4,8 @@
 #include "envoy/config/core/v3/base.pb.h"
 #include "envoy/extensions/clusters/dynamic_forward_proxy/v3/cluster.pb.h"
 #include "envoy/extensions/filters/http/dynamic_forward_proxy/v3/dynamic_forward_proxy.pb.h"
+#include "envoy/router/string_accessor.h"
+#include "envoy/stream_info/uint32_accessor.h"
 
 #include "source/common/http/utility.h"
 #include "source/common/network/filter_state_proxy_info.h"
@@ -24,7 +26,33 @@ void latchTime(Http::StreamDecoderFilterCallbacks* decoder_callbacks, absl::stri
   downstream_timing.setValue(key, decoder_callbacks->dispatcher().timeSource().monotonicTime());
 }
 
+// Helper function to apply filter state overrides to host and port.
+// Conditionally checks filter state based on the allow_dynamic_host_from_filter_state flag.
+void applyFilterStateOverrides(absl::string_view& host, uint32_t& port,
+                               Http::StreamDecoderFilterCallbacks* decoder_callbacks,
+                               bool allow_dynamic_host_from_filter_state) {
+  if (!allow_dynamic_host_from_filter_state) {
+    return;
+  }
+
+  const Router::StringAccessor* dynamic_host_filter_state =
+      decoder_callbacks->streamInfo().filterState()->getDataReadOnly<Router::StringAccessor>(
+          "envoy.upstream.dynamic_host");
+  if (dynamic_host_filter_state) {
+    host = dynamic_host_filter_state->asString();
+  }
+
+  const StreamInfo::UInt32Accessor* dynamic_port_filter_state =
+      decoder_callbacks->streamInfo().filterState()->getDataReadOnly<StreamInfo::UInt32Accessor>(
+          "envoy.upstream.dynamic_port");
+  if (dynamic_port_filter_state != nullptr && dynamic_port_filter_state->value() > 0 &&
+      dynamic_port_filter_state->value() <= 65535) {
+    port = dynamic_port_filter_state->value();
+  }
+}
+
 } // namespace
+
 struct ResponseStringValues {
   const std::string DnsCacheOverflow = "DNS cache overflow";
   const std::string PendingRequestOverflow = "Dynamic forward proxy pending request overflow";
@@ -64,7 +92,8 @@ ProxyFilterConfig::ProxyFilterConfig(
       tls_slot_(context.serverFactoryContext().threadLocal()),
       cluster_init_timeout_(PROTOBUF_GET_MS_OR_DEFAULT(proto_config.sub_cluster_config(),
                                                        cluster_init_timeout, 5000)),
-      save_upstream_address_(proto_config.save_upstream_address()) {
+      save_upstream_address_(proto_config.save_upstream_address()),
+      allow_dynamic_host_from_filter_state_(proto_config.allow_dynamic_host_from_filter_state()) {
   tls_slot_.set(
       [&](Event::Dispatcher&) { return std::make_shared<ThreadLocalClusterInfo>(*this); });
 }
@@ -115,16 +144,15 @@ LoadClusterEntryHandlePtr ProxyFilterConfig::addDynamicCluster(
 
 ProxyFilterConfig::ThreadLocalClusterInfo::~ThreadLocalClusterInfo() {
   for (const auto& it : pending_clusters_) {
-    for (auto cluster : it.second) {
+    for (const auto cluster : it.second) {
       cluster->cancel();
     }
   }
 }
 void ProxyFilterConfig::ThreadLocalClusterInfo::onClusterAddOrUpdate(
     absl::string_view cluster_name, Upstream::ThreadLocalClusterCommand&) {
   ENVOY_LOG(debug, "thread local cluster {} added or updated", cluster_name);
-  auto it = pending_clusters_.find(cluster_name);
-  if (it != pending_clusters_.end()) {
+  if (const auto it = pending_clusters_.find(cluster_name); it != pending_clusters_.end()) {
     for (auto* cluster : it->second) {
       auto& callbacks = cluster->callbacks_;
       cluster->cancel();
@@ -170,8 +198,8 @@ bool ProxyFilter::isProxying() {
 
 Http::FilterHeadersStatus ProxyFilter::decodeHeaders(Http::RequestHeaderMap& headers, bool) {
   Router::RouteConstSharedPtr route = decoder_callbacks_->route();
-  const Router::RouteEntry* route_entry;
-  if (!route || !(route_entry = route->routeEntry())) {
+  const Router::RouteEntry* route_entry = route ? route->routeEntry() : nullptr;
+  if (!route_entry) {
     return Http::FilterHeadersStatus::Continue;
   }
 
@@ -288,20 +316,32 @@ Http::FilterHeadersStatus ProxyFilter::decodeHeaders(Http::RequestHeaderMap& hea
       return Http::FilterHeadersStatus::StopIteration;
     }
   }
-  auto result = config_->cache().loadDnsCacheEntryWithForceRefresh(
-      headers.Host()->value().getStringView(), default_port, is_proxying, force_cache_refresh,
-      *this);
-  cache_load_handle_ = std::move(result.handle_);
+
+  // Get host value from the request headers.
+  const auto host_attributes =
+      Http::Utility::parseAuthority(headers.Host()->value().getStringView());
+  absl::string_view host = host_attributes.host_;
+  uint16_t port = host_attributes.port_.value_or(default_port);
+
+  // Apply filter state overrides for host and port.
+  uint32_t port_u32 = port;
+  applyFilterStateOverrides(host, port_u32, decoder_callbacks_,
+                            config_->allowDynamicHostFromFilterState());
+  port = port_u32;
+
+  auto [status_, handle_, host_info_] = config_->cache().loadDnsCacheEntryWithForceRefresh(
+      host, port, is_proxying, force_cache_refresh, *this);
+  cache_load_handle_ = std::move(handle_);
   if (cache_load_handle_ == nullptr) {
     circuit_breaker_.reset();
   }
 
-  switch (result.status_) {
+  switch (status_) {
   case LoadDnsCacheEntryStatus::InCache: {
     ASSERT(cache_load_handle_ == nullptr);
     ENVOY_STREAM_LOG(debug, "DNS cache entry already loaded, continuing", *decoder_callbacks_);
 
-    auto const& host = result.host_info_;
+    auto const& host = host_info_;
     latchTime(decoder_callbacks_, DNS_END);
     if (is_proxying) {
       ENVOY_BUG(host.has_value(), "Proxying request but no host entry in DNS cache.");
@@ -330,18 +370,28 @@ Http::FilterHeadersStatus ProxyFilter::decodeHeaders(Http::RequestHeaderMap& hea
   PANIC_DUE_TO_CORRUPT_ENUM;
 }
 
-Http::FilterHeadersStatus ProxyFilter::loadDynamicCluster(
-    Extensions::Common::DynamicForwardProxy::DfpClusterSharedPtr cluster,
-    Http::RequestHeaderMap& headers, uint16_t default_port) {
+Http::FilterHeadersStatus
+ProxyFilter::loadDynamicCluster(const Common::DynamicForwardProxy::DfpClusterSharedPtr& cluster,
+                                const Http::RequestHeaderMap& headers, uint16_t default_port) {
+
+  // Parse host and port from headers.
   const auto host_attributes = Http::Utility::parseAuthority(headers.getHostValue());
   auto host = std::string(host_attributes.host_);
   auto port = host_attributes.port_.value_or(default_port);
 
+  // Apply filter state overrides using the helper function.
+  absl::string_view host_view = host; // Create string_view for the helper.
+  uint32_t port_u32 = port;
+  applyFilterStateOverrides(host_view, port_u32, decoder_callbacks_,
+                            config_->allowDynamicHostFromFilterState());
+  host = std::string(host_view); // Convert back to string.
+  port = port_u32;
+
   latchTime(decoder_callbacks_, DNS_START);
 
   // cluster name is prefix + host + port
-  auto cluster_name = "DFPCluster:" + host + ":" + std::to_string(port);
-  Upstream::ThreadLocalCluster* local_cluster =
+  const auto cluster_name = "DFPCluster:" + host + ":" + std::to_string(port);
+  const Upstream::ThreadLocalCluster* local_cluster =
       config_->clusterManager().getThreadLocalCluster(cluster_name);
   if (local_cluster && cluster->touch(cluster_name)) {
     ENVOY_STREAM_LOG(debug, "using the thread local cluster after touch success",
@@ -352,7 +402,7 @@ Http::FilterHeadersStatus ProxyFilter::loadDynamicCluster(
 
   // Still need to add dynamic cluster again even the thread local cluster exists while touch
   // failed, that means the cluster is removed in main thread due to ttl reached.
-  // Otherwise, we may not be able to get the thread local cluster in router.
+  // Otherwise, we may not be able to get the thread local cluster in the router.
 
   // Create a new cluster & register a callback to tls
   cluster_load_handle_ = config_->addDynamicCluster(cluster, cluster_name, host, port, *this);