network: support draining connections after triggered DNS refresh

library/common/config/config.cc

@@ -52,6 +52,7 @@ const std::string config_header = R"(
 - &dns_refresh_rate 60s
 - &dns_resolver_name envoy.network.dns_resolver.cares
 - &dns_resolver_config {"@type":"type.googleapis.com/envoy.extensions.network.dns_resolver.cares.v3.CaresDnsResolverConfig"}
+- &enable_drain_post_dns_refresh false
 - &enable_interface_binding false
 - &h2_connection_keepalive_idle_interval 100000s
 - &h2_connection_keepalive_timeout 10s
@@ -288,6 +289,7 @@ const char* config_template = R"(
           - name: envoy.filters.http.network_configuration
             typed_config:
               "@type": type.googleapis.com/envoymobile.extensions.filters.http.network_configuration.NetworkConfiguration
+              enable_drain_post_dns_refresh: *enable_drain_post_dns_refresh
               enable_interface_binding: *enable_interface_binding
           - name: envoy.filters.http.local_error
             typed_config:

library/common/engine.cc

@@ -301,12 +301,6 @@ void Engine::flushStats() {
   server_->flushStats();
 }
 
-void Engine::drainConnections() {
-  ASSERT(dispatcher_->isThreadSafe(),
-         "drainConnections must be called from the dispatcher's context");
-  server_->clusterManager().drainConnections(nullptr);
-}
-
 Upstream::ClusterManager& Engine::getClusterManager() {
   ASSERT(dispatcher_->isThreadSafe(),
          "getClusterManager must be called from the dispatcher's context");

library/common/engine.h

@@ -122,11 +122,6 @@ class Engine : public Logger::Loggable<Logger::Id::main> {
    */
   void flushStats();
 
-  /**
-   * Drain all upstream connections associated with this Engine.
-   */
-  void drainConnections();
-
   /**
    * Get cluster manager from the Engine.
    */

library/common/extensions/filters/http/network_configuration/config.cc

@@ -13,12 +13,13 @@ Http::FilterFactoryCb NetworkConfigurationFilterFactory::createFilterFactoryFrom
     const std::string&, Server::Configuration::FactoryContext& context) {
 
   auto network_configurator = Network::ConfiguratorFactory{context}.get();
+  bool enable_drain_post_dns_refresh = proto_config.enable_drain_post_dns_refresh();
   bool enable_interface_binding = proto_config.enable_interface_binding();
 
-  return [network_configurator,
+  return [network_configurator, enable_drain_post_dns_refresh,
           enable_interface_binding](Http::FilterChainFactoryCallbacks& callbacks) -> void {
     callbacks.addStreamFilter(std::make_shared<NetworkConfigurationFilter>(
-        network_configurator, enable_interface_binding));
+        network_configurator, enable_drain_post_dns_refresh, enable_interface_binding));
   };
 }
 

library/common/extensions/filters/http/network_configuration/filter.cc

@@ -21,6 +21,7 @@ void NetworkConfigurationFilter::setDecoderFilterCallbacks(
 
   auto options = std::make_shared<Network::Socket::Options>();
   network_configurator_->setInterfaceBindingEnabled(enable_interface_binding_);
+  network_configurator_->setDrainPostDnsRefreshEnabled(enable_drain_post_dns_refresh_);
   extra_stream_info_->configuration_key_ = network_configurator_->addUpstreamSocketOptions(options);
   decoder_callbacks_->addUpstreamSocketOptions(options);
 }

library/common/extensions/filters/http/network_configuration/filter.h

@@ -22,9 +22,10 @@ class NetworkConfigurationFilter final : public Http::PassThroughFilter,
                                          public Logger::Loggable<Logger::Id::filter> {
 public:
   NetworkConfigurationFilter(Network::ConfiguratorSharedPtr network_configurator,
-                             bool enable_interface_binding)
+                             bool enable_drain_post_dns_refresh, bool enable_interface_binding)
       : network_configurator_(network_configurator),
         extra_stream_info_(nullptr), // always set in setDecoderFilterCallbacks
+        enable_drain_post_dns_refresh_(enable_drain_post_dns_refresh),
         enable_interface_binding_(enable_interface_binding) {}
 
   // Http::StreamDecoderFilter
@@ -37,6 +38,7 @@ class NetworkConfigurationFilter final : public Http::PassThroughFilter,
 private:
   Network::ConfiguratorSharedPtr network_configurator_;
   StreamInfo::ExtraStreamInfo* extra_stream_info_;
+  bool enable_drain_post_dns_refresh_;
   bool enable_interface_binding_;
 };
 

library/common/extensions/filters/http/network_configuration/filter.proto

@@ -6,4 +6,8 @@ message NetworkConfiguration {
   // If set to true, the filter will permit the NetworkConfigurator to provide upstream
   // socket option that MAY bind a connection to a specific network interface.
   bool enable_interface_binding = 1;
+
+  // If set to true, the filter will permit the NetworkConfigurator to drain connections
+  // when a DNS refresh is externally triggered.
+  bool enable_drain_post_dns_refresh = 2;
 }

library/common/jni/jni_interface.cc

@@ -1016,11 +1016,11 @@ Java_io_envoyproxy_envoymobile_engine_JniLibrary_registerStringAccessor(JNIEnv*
 }
 
 extern "C" JNIEXPORT void JNICALL
-Java_io_envoyproxy_envoymobile_engine_JniLibrary_drainConnections(JNIEnv* env,
-                                                                  jclass, // class
-                                                                  jlong engine) {
-  jni_log("[Envoy]", "drainConnections");
-  drain_connections(engine);
+Java_io_envoyproxy_envoymobile_engine_JniLibrary_resetConnectivityState(JNIEnv* env,
+                                                                        jclass, // class
+                                                                        jlong engine) {
+  jni_log("[Envoy]", "resetConnectivityState");
+  reset_connectivity_state(engine);
 }
 
 extern "C" JNIEXPORT jint JNICALL

library/common/main_interface.cc

@@ -68,7 +68,7 @@ envoy_status_t reset_stream(envoy_engine_t engine, envoy_stream_t stream) {
 envoy_status_t set_preferred_network(envoy_engine_t engine, envoy_network_t network) {
   envoy_netconf_t configuration_key = Envoy::Network::Configurator::setPreferredNetwork(network);
   Envoy::EngineHandle::runOnEngineDispatcher(engine, [configuration_key](auto& engine) -> void {
-    engine.networkConfigurator().refreshDns(configuration_key);
+    engine.networkConfigurator().refreshDns(configuration_key, true);
   });
   // TODO(snowp): Should this return failure ever?
   return ENVOY_SUCCESS;
@@ -188,9 +188,7 @@ envoy_status_t run_engine(envoy_engine_t engine, const char* config, const char*
 
 void terminate_engine(envoy_engine_t engine) { Envoy::EngineHandle::terminateEngine(engine); }
 
-envoy_status_t drain_connections(envoy_engine_t e) {
-  // This will change once multiple engine support is in place.
-  // https://github.com/envoyproxy/envoy-mobile/issues/332
+envoy_status_t reset_connectivity_state(envoy_engine_t e) {
   return Envoy::EngineHandle::runOnEngineDispatcher(
-      e, [](auto& engine) { engine.drainConnections(); });
+      e, [](auto& engine) { engine.networkConfigurator().resetConnectivityState(); });
 }

library/common/main_interface.h

@@ -203,11 +203,11 @@ envoy_status_t run_engine(envoy_engine_t engine, const char* config, const char*
 void terminate_engine(envoy_engine_t engine);
 
 /**
- * Drain all upstream connections associated with an engine
+ * Refresh DNS, and drain connections associated with an engine.
  * @param engine, handle to the engine to drain.
  * @return envoy_status_t, the resulting status of the operation.
  */
-envoy_status_t drain_connections(envoy_engine_t engine);
+envoy_status_t reset_connectivity_state(envoy_engine_t engine);
 
 #ifdef __cplusplus
 } // functions

library/common/network/configurator.cc

@@ -82,6 +82,15 @@ Configurator::NetworkState Configurator::network_state_{1, ENVOY_NET_GENERIC, Ma
 
 envoy_netconf_t Configurator::setPreferredNetwork(envoy_network_t network) {
   Thread::LockGuard lock{network_state_.mutex_};
+
+  // TODO(goaway): Re-enable this guard. There's some concern that this will miss network updates
+  // moving from offline to online states. We should address this then re-enable this guard to
+  // avoid unnecessary cache refresh and connection drain.
+  // if (network_state_.network_ == network) {
+  //  // Provide a non-current key preventing further scheduled effects (e.g. DNS refresh).
+  //  return network_state_.configuration_key_ - 1;
+  //}
+
   ENVOY_LOG_EVENT(debug, "netconf_network_change", std::to_string(network));
 
   network_state_.configuration_key_++;
@@ -167,13 +176,43 @@ void Configurator::reportNetworkUsage(envoy_netconf_t configuration_key, bool ne
 
   // If configuration state changed, refresh dns.
   if (configuration_updated) {
-    refreshDns(configuration_key);
+    refreshDns(configuration_key, false);
+  }
+}
+
+void Configurator::onDnsResolutionComplete(
+    const std::string& host, const Extensions::Common::DynamicForwardProxy::DnsHostInfoSharedPtr&,
+    Network::DnsResolver::ResolutionStatus) {
+  if (enable_drain_post_dns_refresh_ && pending_drain_) {
+    pending_drain_ = false;
+
+    // We ignore whether DNS resolution has succeeded here. If it failed, we may be offline and
+    // should probably drain connections. If it succeeds, we may have new DNS entries and so we
+    // drain connections. It may be possible to refine this logic in the future.
+    // TODO(goaway): check the set of cached hosts from the last triggered DNS refresh for this
+    // host, and if present, remove it and trigger connection drain for this host specifically.
+    ENVOY_LOG_EVENT(debug, "netconf_post_dns_drain_cx", host);
+    cluster_manager_.drainConnections(nullptr);
+  }
+}
+
+void Configurator::setDrainPostDnsRefreshEnabled(bool enabled) {
+  enable_drain_post_dns_refresh_ = enabled;
+  if (!enabled) {
+    pending_drain_ = false;
+  } else if (!dns_callbacks_handle_) {
+    // Register callbacks once, on demand, using the handle as a sentinel. There may not be
+    // a DNS cache during initialization, but if one is available, it should always exist by the
+    // time this function is called from the NetworkConfigurationFilter.
+    if (auto dns_cache = dnsCache()) {
+      dns_callbacks_handle_ = dns_cache->addUpdateCallbacks(*this);
+    }
   }
 }
 
 void Configurator::setInterfaceBindingEnabled(bool enabled) { enable_interface_binding_ = enabled; }
 
-void Configurator::refreshDns(envoy_netconf_t configuration_key) {
+void Configurator::refreshDns(envoy_netconf_t configuration_key, bool drain_connections) {
   {
     Thread::LockGuard lock{network_state_.mutex_};
 
@@ -188,12 +227,33 @@ void Configurator::refreshDns(envoy_netconf_t configuration_key) {
     }
   }
 
-  if (auto dns_cache = dns_cache_manager_->lookUpCacheByName(BaseDnsCache)) {
+  if (auto dns_cache = dnsCache()) {
     ENVOY_LOG_EVENT(debug, "netconf_refresh_dns", std::to_string(configuration_key));
+    pending_drain_ = drain_connections && enable_drain_post_dns_refresh_;
+    // TODO(goaway): capture the list of currently cached hosts here in a set, to be checked
+    // for draining when DNS resolutions occur.
     dns_cache->forceRefreshHosts();
-  } else {
+  }
+}
+
+Extensions::Common::DynamicForwardProxy::DnsCacheSharedPtr Configurator::dnsCache() {
+  auto cache = dns_cache_manager_->lookUpCacheByName(BaseDnsCache);
+  if (!cache) {
     ENVOY_LOG_EVENT(warn, "netconf_dns_cache_missing", BaseDnsCache);
   }
+  return cache;
+}
+
+void Configurator::resetConnectivityState() {
+  envoy_netconf_t configuration_key;
+  {
+    Thread::LockGuard lock{network_state_.mutex_};
+    network_state_.remaining_faults_ = 1;
+    network_state_.socket_mode_ = DefaultPreferredNetworkMode;
+    configuration_key = ++network_state_.configuration_key_;
+  }
+
+  refreshDns(configuration_key, true);
 }
 
 std::vector<InterfacePair> Configurator::enumerateV4Interfaces() {
@@ -347,7 +407,8 @@ ConfiguratorSharedPtr ConfiguratorFactory::get() {
       SINGLETON_MANAGER_REGISTERED_NAME(network_configurator), [&] {
         Extensions::Common::DynamicForwardProxy::DnsCacheManagerFactoryImpl cache_manager_factory{
             context_};
-        return std::make_shared<Configurator>(cache_manager_factory.get());
+        return std::make_shared<Configurator>(context_.clusterManager(),
+                                              cache_manager_factory.get());
       });
 }
 

library/common/network/configurator.h

@@ -5,7 +5,9 @@
 
 #include "envoy/network/socket.h"
 #include "envoy/singleton/manager.h"
+#include "envoy/upstream/cluster_manager.h"
 
+#include "source/extensions/common/dynamic_forward_proxy/dns_cache.h"
 #include "source/extensions/common/dynamic_forward_proxy/dns_cache_impl.h"
 
 #include "library/common/types/c_types.h"
@@ -60,11 +62,34 @@ using InterfacePair = std::pair<const std::string, Address::InstanceConstSharedP
  * Object responsible for tracking network state, especially with respect to multiple interfaces,
  * and providing auxiliary configuration to network connections, in the form of upstream socket
  * options.
+ *
+ * Code is largely structured to be run exclusively on the engine's main thread. However,
+ * setPreferredNetwork is allowed to be called from any thread, and the internal NetworkState that
+ * it modifies owns a mutex used to synchronize all access to that state.
+ * (Note NetworkState was originally designed to fit into an atomic, and could still feasibly be
+ * switched to one.)
+ *
+ * This object is a singleton per-engine. Note that several pieces of functionality assume a DNS
+ * cache adhering to the one set up in base configuration will be present, but will become no-ops
+ * if that cache is missing either due to alternate configurations, or lifecycle-related timing.
+ *
  */
-class Configurator : public Logger::Loggable<Logger::Id::upstream>, public Singleton::Instance {
+class Configurator : public Logger::Loggable<Logger::Id::upstream>,
+                     public Extensions::Common::DynamicForwardProxy::DnsCache::UpdateCallbacks,
+                     public Singleton::Instance {
 public:
-  Configurator(DnsCacheManagerSharedPtr dns_cache_manager)
-      : enable_interface_binding_{false}, dns_cache_manager_(dns_cache_manager) {}
+  Configurator(Upstream::ClusterManager& cluster_manager,
+               DnsCacheManagerSharedPtr dns_cache_manager)
+      : cluster_manager_(cluster_manager), dns_cache_manager_(dns_cache_manager) {}
+
+  // Extensions::Common::DynamicForwardProxy::DnsCache::UpdateCallbacks
+  void onDnsHostAddOrUpdate(
+      const std::string& /*host*/,
+      const Extensions::Common::DynamicForwardProxy::DnsHostInfoSharedPtr&) override {}
+  void onDnsHostRemove(const std::string& /*host*/) override {}
+  void onDnsResolutionComplete(const std::string& /*host*/,
+                               const Extensions::Common::DynamicForwardProxy::DnsHostInfoSharedPtr&,
+                               Network::DnsResolver::ResolutionStatus) override;
 
   /**
    * @returns a list of local network interfaces supporting IPv4.
@@ -108,12 +133,21 @@ class Configurator : public Logger::Loggable<Logger::Id::upstream>, public Singl
   void reportNetworkUsage(envoy_netconf_t configuration_key, bool network_fault);
 
   /**
-   * Sets the current OS default/preferred network class.
+   * Sets the current OS default/preferred network class. Note this function is allowed to be
+   * called from any thread.
    * @param network, the OS-preferred network.
    * @returns configuration key to associate with any related calls.
    */
   static envoy_netconf_t setPreferredNetwork(envoy_network_t network);
 
+  /**
+   * Configure whether connections should be drained after a triggered DNS refresh. Currently this
+   * may happen either due to an external call to refreshConnectivityState or an update to
+   * setPreferredNetwork.
+   * @param enabled, whether to enable connection drain after DNS refresh.
+   */
+  void setDrainPostDnsRefreshEnabled(bool enabled);
+
   /**
    * Sets whether subsequent calls for upstream socket options may leverage options that bind
    * to specific network interfaces.
@@ -124,8 +158,14 @@ class Configurator : public Logger::Loggable<Logger::Id::upstream>, public Singl
   /**
    * Refresh DNS in response to preferred network update. May be no-op.
    * @param configuration_key, key provided by this class representing the current configuration.
+   * @param drain_connections, request that connections be drained after next DNS resolution.
    */
-  void refreshDns(envoy_netconf_t configuration_key);
+  void refreshDns(envoy_netconf_t configuration_key, bool drain_connections);
+
+  /**
+   * Drain all upstream connections associated with this Engine.
+   */
+  void resetConnectivityState();
 
   /**
    * @returns the current socket options that should be used for connections.
@@ -152,7 +192,21 @@ class Configurator : public Logger::Loggable<Logger::Id::upstream>, public Singl
   Socket::OptionsSharedPtr getAlternateInterfaceSocketOptions(envoy_network_t network);
   InterfacePair getActiveAlternateInterface(envoy_network_t network, unsigned short family);
 
-  bool enable_interface_binding_;
+  /**
+   * Returns the default DNS cache set up in base configuration. This cache may be missing either
+   * due to engine lifecycle-related timing or alternate configurations. If it is, operations
+   * that use it should revert to no-ops.
+   *
+   * @returns the default DNS cache set up in base configuration or nullptr.
+   */
+  Extensions::Common::DynamicForwardProxy::DnsCacheSharedPtr dnsCache();
+
+  bool enable_drain_post_dns_refresh_{false};
+  bool enable_interface_binding_{false};
+  bool pending_drain_{false};
+  Extensions::Common::DynamicForwardProxy::DnsCache::AddUpdateCallbacksHandlePtr
+      dns_callbacks_handle_{nullptr};
+  Upstream::ClusterManager& cluster_manager_;
   DnsCacheManagerSharedPtr dns_cache_manager_;
   static NetworkState network_state_;
 };
@@ -164,15 +218,15 @@ using ConfiguratorSharedPtr = std::shared_ptr<Configurator>;
  */
 class ConfiguratorFactory {
 public:
-  ConfiguratorFactory(Server::Configuration::FactoryContextBase& context) : context_(context) {}
+  ConfiguratorFactory(Server::Configuration::CommonFactoryContext& context) : context_(context) {}
 
   /**
    * @returns singleton Configurator instance.
    */
   ConfiguratorSharedPtr get();
 
 private:
-  Server::Configuration::FactoryContextBase& context_;
+  Server::Configuration::CommonFactoryContext& context_;
 };
 
 /**

library/java/io/envoyproxy/envoymobile/engine/AndroidEngineImpl.java

@@ -90,8 +90,8 @@ public int registerStringAccessor(String accessorName, EnvoyStringAccessor acces
   }
 
   @Override
-  public void drainConnections() {
-    envoyEngine.drainConnections();
+  public void resetConnectivityState() {
+    envoyEngine.resetConnectivityState();
   }
 
   @Override