xds: implement extension config discovery for HCM

include/envoy/router/BUILD

@@ -36,6 +36,17 @@ envoy_cc_library(
     ],
 )
 
+envoy_cc_library(
+    name = "filter_config_provider_interface",
+    hdrs = ["filter_config_provider.h"],
+    deps = [
+        "//include/envoy/http:filter_interface",
+        "//include/envoy/init:manager_interface",
+        "//include/envoy/server:filter_config_interface",
+        "@envoy_api//envoy/config/core/v3:pkg_cc_proto",
+    ],
+)
+
 envoy_cc_library(
     name = "route_config_update_info_interface",
     hdrs = ["route_config_update_receiver.h"],

include/envoy/router/filter_config_provider.h

@@ -0,0 +1,86 @@
+#pragma once
+
+#include "envoy/config/core/v3/config_source.pb.h"
+#include "envoy/config/core/v3/extension.pb.h"
+#include "envoy/http/filter.h"
+#include "envoy/init/manager.h"
+#include "envoy/server/filter_config.h"
+
+namespace Envoy {
+namespace Router {
+
+/**
+ * A provider for constant filter configurations.
+ */
+class FilterConfigProvider {
+public:
+  virtual ~FilterConfigProvider() = default;
+
+  /**
+   * Get the filter config resource name.
+   **/
+  virtual const std::string& name() PURE;
+
+  /**
+   * @return Http::FilterFactoryCb a filter config to be instantiated on the subsequent streams.
+   * Note that if the provider has not yet performed an initial configuration load and no default is
+   * provided, no information will be returned.
+   */
+  virtual absl::optional<Http::FilterFactoryCb> config() PURE;
+
+  /**
+   * Validate if the route configuration can be applied in the context of the filter manager.
+   * @param Server::Configuration::NamedHttpFilterConfigFactory a filter factory to validate in the
+   * context of the filter manager filter chains.
+   */
+  virtual void validateConfig(Server::Configuration::NamedHttpFilterConfigFactory& factory) PURE;
+
+  /**
+   * Update the provider about the configuration changes.
+   * @param config is a filter factory to be used on the subsequent streams.
+   * @param version_info is the version of the new filter configuration.
+   */
+  virtual void onConfigUpdate(Http::FilterFactoryCb config, const std::string& version_info) PURE;
+};
+
+using FilterConfigProviderPtr = std::unique_ptr<FilterConfigProvider>;
+
+/**
+ * The FilterConfigProviderManager exposes the ability to get a FilterConfigProvider. This interface
+ * is exposed to the Server's FactoryContext in order to allow stream creators to get
+ * FilterConfigProviders for filters on the stream.
+ */
+class FilterConfigProviderManager {
+public:
+  virtual ~FilterConfigProviderManager() = default;
+
+  /**
+   * Get a FilterConfigProviderPtr for a filter config. The config providers may share
+   * the underlying subscriptions to the filter config discovery service.
+   * @param config_source supplies the configuration source for the filter configs.
+   * @param filter_config_name the filter config resource name.
+   * @param require_terminal enforces that the filter config must be for a terminal filter
+   * @param factory_context is the context to use for the filter config provider.
+   * @param stat_prefix supplies the stat_prefix to use for the provider stats.
+   * @param apply_without_warming initializes immediately with the default config and starts the
+   * subscription.
+   */
+  virtual FilterConfigProviderPtr
+  createDynamicFilterConfigProvider(const envoy::config::core::v3::ConfigSource& config_source,
+                                    const std::string& filter_config_name, bool require_terminal,
+                                    Server::Configuration::FactoryContext& factory_context,
+                                    const std::string& stat_prefix,
+                                    bool apply_without_warming) PURE;
+
+  /**
+   * Get a FilterConfigProviderPtr for a statically inlined filter config.
+   * @param config is a fully resolved filter instantiation factory.
+   * @param filter_config_name the filter config resource name.
+   */
+  virtual FilterConfigProviderPtr
+  createStaticFilterConfigProvider(const Http::FilterFactoryCb& config,
+                                   const std::string& filter_config_name) PURE;
+};
+
+} // namespace Router
+} // namespace Envoy

include/envoy/stream_info/stream_info.h

@@ -76,8 +76,10 @@ enum ResponseFlag {
   UpstreamMaxStreamDurationReached = 0x80000,
   // True if the response was served from an Envoy cache filter.
   ResponseFromCacheFilter = 0x100000,
+  // Filter config was not received within the permitted warming deadline.
+  NoFilterConfigFound = 0x200000,
   // ATTENTION: MAKE SURE THIS REMAINS EQUAL TO THE LAST FLAG.
-  LastFlag = ResponseFromCacheFilter
+  LastFlag = NoFilterConfigFound
 };
 
 /**

source/common/config/utility.h

@@ -234,27 +234,36 @@ class Utility {
    */
   template <class Factory, class ProtoMessage>
   static Factory& getAndCheckFactory(const ProtoMessage& message) {
-    const ProtobufWkt::Any& typed_config = message.typed_config();
+    Factory* factory = Utility::getFactoryByType<Factory>(message.typed_config());
+    if (factory != nullptr) {
+      return *factory;
+    }
+
+    return Utility::getAndCheckFactoryByName<Factory>(message.name());
+  }
+
+  /**
+   * Get a Factory from the registry by type URL.
+   * @param typed_config for the extension config.
+   */
+  template <class Factory> static Factory* getFactoryByType(const ProtobufWkt::Any& typed_config) {
     static const std::string& typed_struct_type =
         udpa::type::v1::TypedStruct::default_instance().GetDescriptor()->full_name();
 
-    if (!typed_config.type_url().empty()) {
-      // Unpack methods will only use the fully qualified type name after the last '/'.
-      // https://github.com/protocolbuffers/protobuf/blob/3.6.x/src/google/protobuf/any.proto#L87
-      auto type = std::string(TypeUtil::typeUrlToDescriptorFullName(typed_config.type_url()));
-      if (type == typed_struct_type) {
-        udpa::type::v1::TypedStruct typed_struct;
-        MessageUtil::unpackTo(typed_config, typed_struct);
-        // Not handling nested structs or typed structs in typed structs
-        type = std::string(TypeUtil::typeUrlToDescriptorFullName(typed_struct.type_url()));
-      }
-      Factory* factory = Registry::FactoryRegistry<Factory>::getFactoryByType(type);
-      if (factory != nullptr) {
-        return *factory;
-      }
+    if (typed_config.type_url().empty()) {
+      return nullptr;
     }
 
-    return Utility::getAndCheckFactoryByName<Factory>(message.name());
+    // Unpack methods will only use the fully qualified type name after the last '/'.
+    // https://github.com/protocolbuffers/protobuf/blob/3.6.x/src/google/protobuf/any.proto#L87
+    auto type = std::string(TypeUtil::typeUrlToDescriptorFullName(typed_config.type_url()));
+    if (type == typed_struct_type) {
+      udpa::type::v1::TypedStruct typed_struct;
+      MessageUtil::unpackTo(typed_config, typed_struct);
+      // Not handling nested structs or typed structs in typed structs
+      type = std::string(TypeUtil::typeUrlToDescriptorFullName(typed_struct.type_url()));
+    }
+    return Registry::FactoryRegistry<Factory>::getFactoryByType(type);
   }
 
   /**

source/common/router/BUILD

@@ -185,6 +185,29 @@ envoy_cc_library(
     ],
 )
 
+envoy_cc_library(
+    name = "filter_config_discovery_lib",
+    srcs = ["filter_config_discovery_impl.cc"],
+    hdrs = ["filter_config_discovery_impl.h"],
+    deps = [
+        ":config_lib",
+        "//include/envoy/config:subscription_interface",
+        "//include/envoy/router:filter_config_provider_interface",
+        "//include/envoy/server:filter_config_interface",
+        "//include/envoy/singleton:instance_interface",
+        "//include/envoy/thread_local:thread_local_interface",
+        "//source/common/config:subscription_base_interface",
+        "//source/common/config:subscription_factory_lib",
+        "//source/common/config:utility_lib",
+        "//source/common/grpc:common_lib",
+        "//source/common/init:manager_lib",
+        "//source/common/init:target_lib",
+        "//source/common/init:watcher_lib",
+        "//source/common/protobuf:utility_lib",
+        "@envoy_api//envoy/config/core/v3:pkg_cc_proto",
+    ],
+)
+
 envoy_cc_library(
     name = "scoped_config_lib",
     srcs = ["scoped_config_impl.cc"],

source/common/router/filter_config_discovery_impl.cc

@@ -0,0 +1,197 @@
+#include "common/router/filter_config_discovery_impl.h"
+
+#include "envoy/config/core/v3/extension.pb.validate.h"
+#include "envoy/server/filter_config.h"
+
+#include "common/config/utility.h"
+#include "common/grpc/common.h"
+#include "common/protobuf/utility.h"
+
+namespace Envoy {
+namespace Router {
+
+DynamicFilterConfigProviderImpl::DynamicFilterConfigProviderImpl(
+    FilterConfigSubscriptionSharedPtr&& subscription, bool require_terminal,
+    Server::Configuration::FactoryContext& factory_context)
+    : subscription_(std::move(subscription)), require_terminal_(require_terminal),
+      tls_(factory_context.threadLocal().allocateSlot()),
+      init_target_("DynamicFilterConfigProviderImpl", [this]() {
+        subscription_->start();
+        init_target_.ready();
+      }) {
+  subscription_->filter_config_providers_.insert(this);
+  tls_->set([](Event::Dispatcher&) -> ThreadLocal::ThreadLocalObjectSharedPtr {
+    return std::make_shared<ThreadLocalConfig>();
+  });
+}
+
+DynamicFilterConfigProviderImpl::~DynamicFilterConfigProviderImpl() {
+  subscription_->filter_config_providers_.erase(this);
+}
+
+const std::string& DynamicFilterConfigProviderImpl::name() { return subscription_->name(); }
+
+absl::optional<Http::FilterFactoryCb> DynamicFilterConfigProviderImpl::config() {
+  return tls_->getTyped<ThreadLocalConfig>().config_;
+}
+
+void DynamicFilterConfigProviderImpl::validateConfig(
+    Server::Configuration::NamedHttpFilterConfigFactory& factory) {
+  if (factory.isTerminalFilter() && !require_terminal_) {
+    throw EnvoyException(
+        fmt::format("Error: filter config {} must not be the last in the filter chain", name()));
+  } else if (!factory.isTerminalFilter() && require_terminal_) {
+    throw EnvoyException(
+        fmt::format("Error: filter config {} must be the last in the filter chain", name()));
+  }
+}
+
+void DynamicFilterConfigProviderImpl::onConfigUpdate(Http::FilterFactoryCb config,
+                                                     const std::string&) {
+  tls_->runOnAllThreads([config](ThreadLocal::ThreadLocalObjectSharedPtr previous)
+                            -> ThreadLocal::ThreadLocalObjectSharedPtr {
+    auto prev_config = std::dynamic_pointer_cast<ThreadLocalConfig>(previous);
+    prev_config->config_ = config;
+    return previous;
+  });
+}
+
+FilterConfigSubscription::FilterConfigSubscription(
+    const envoy::config::core::v3::ConfigSource& config_source,
+    const std::string& filter_config_name, Server::Configuration::FactoryContext& factory_context,
+    const std::string& stat_prefix, FilterConfigProviderManagerImpl& filter_config_provider_manager,
+    const std::string& subscription_id)
+    : Envoy::Config::SubscriptionBase<envoy::config::core::v3::TypedExtensionConfig>(
+          envoy::config::core::v3::ApiVersion::V3,
+          factory_context.messageValidationContext().dynamicValidationVisitor(), "name"),
+      filter_config_name_(filter_config_name), factory_context_(factory_context),
+      validator_(factory_context.messageValidationContext().dynamicValidationVisitor()),
+      parent_init_target_(fmt::format("FilterConfigSubscription init {}", filter_config_name_),
+                          [this]() { local_init_manager_.initialize(local_init_watcher_); }),
+      local_init_watcher_(
+          fmt::format("FilterConfigSubscription local-init-watcher {}", filter_config_name_),
+          [this]() { parent_init_target_.ready(); }),
+      local_init_target_(
+          fmt::format("FilterConfigSubscription local-init-target {}", filter_config_name_),
+          [this]() { start(); }),
+      local_init_manager_(
+          fmt::format("FilterConfigSubscription local-init-manager {}", filter_config_name_)),
+      scope_(factory_context.scope().createScope(stat_prefix + "filter_config_discovery." +
+                                                 filter_config_name_ + ".")),
+      stat_prefix_(stat_prefix), filter_config_provider_manager_(filter_config_provider_manager),
+      subscription_id_(subscription_id) {
+  const auto resource_name = getResourceName();
+  subscription_ =
+      factory_context.clusterManager().subscriptionFactory().subscriptionFromConfigSource(
+          config_source, Grpc::Common::typeUrl(resource_name), *scope_, *this, resource_decoder_);
+  local_init_manager_.add(local_init_target_);
+}
+
+void FilterConfigSubscription::start() {
+  if (!started_) {
+    started_ = true;
+    subscription_->start({filter_config_name_});
+  }
+}
+
+void FilterConfigSubscription::onConfigUpdate(
+    const std::vector<Envoy::Config::DecodedResourceRef>& resources,
+    const std::string& version_info) {
+  // Make sure to make progress in case the control plane is temporarily inconsistent.
+  local_init_target_.ready();
+
+  if (resources.size() != 1) {
+    throw EnvoyException(fmt::format(
+        "Unexpected number of resources in FilterConfigDS response: {}", resources.size()));
+  }
+  const auto& filter_config = dynamic_cast<const envoy::config::core::v3::TypedExtensionConfig&>(
+      resources[0].get().resource());
+  if (filter_config.name() != filter_config_name_) {
+    throw EnvoyException(fmt::format("Unexpected resource name in FilterConfigDS response: {}",
+                                     filter_config.name()));
+  }
+  auto& factory = Envoy::Config::Utility::getAndCheckFactory<
+      Server::Configuration::NamedHttpFilterConfigFactory>(filter_config);
+  // Ensure that the factory is valid in the filter chain context.
+  for (auto* provider : filter_config_providers_) {
+    provider->validateConfig(factory);
+  }
+  ProtobufTypes::MessagePtr message = Envoy::Config::Utility::translateAnyToFactoryConfig(
+      filter_config.typed_config(), validator_, factory);
+  Http::FilterFactoryCb factory_callback =
+      factory.createFilterFactoryFromProto(*message, stat_prefix_, factory_context_);
+  ENVOY_LOG(debug, "Updating filter config {}", filter_config_name_);
+  for (auto* provider : filter_config_providers_) {
+    provider->onConfigUpdate(factory_callback, version_info);
+  }
+}
+
+void FilterConfigSubscription::onConfigUpdate(
+    const std::vector<Envoy::Config::DecodedResourceRef>& added_resources,
+    const Protobuf::RepeatedPtrField<std::string>& removed_resources, const std::string&) {
+  if (!removed_resources.empty()) {
+    ENVOY_LOG(error,
+              "Server sent a delta FilterConfigDS update attempting to remove a resource (name: "
+              "{}). Ignoring.",
+              removed_resources[0]);
+  }
+  if (!added_resources.empty()) {
+    onConfigUpdate(added_resources, added_resources[0].get().version());
+  }
+}
+
+void FilterConfigSubscription::onConfigUpdateFailed(Envoy::Config::ConfigUpdateFailureReason reason,
+                                                    const EnvoyException*) {
+  ENVOY_LOG(debug, "Updating filter config {} failed due to {}", filter_config_name_, reason);
+  // Make sure to make progress in case the control plane is temporarily failing.
+  local_init_target_.ready();
+}
+
+FilterConfigSubscription::~FilterConfigSubscription() {
+  // If we get destroyed during initialization, make sure we signal that we "initialized".
+  local_init_target_.ready();
+  // Remove the subscription from the provider manager.
+  filter_config_provider_manager_.subscriptions_.erase(subscription_id_);
+}
+
+std::shared_ptr<FilterConfigSubscription> FilterConfigProviderManagerImpl::getSubscription(
+    const envoy::config::core::v3::ConfigSource& config_source, const std::string& name,
+    Server::Configuration::FactoryContext& factory_context, const std::string& stat_prefix) {
+  // FilterConfigSubscriptions are unique based on their config source and filter config name
+  // combination.
+  const std::string subscription_id = absl::StrCat(MessageUtil::hash(config_source), ".", name);
+  auto it = subscriptions_.find(subscription_id);
+  if (it == subscriptions_.end()) {
+    auto subscription = std::make_shared<FilterConfigSubscription>(
+        config_source, name, factory_context, stat_prefix, *this, subscription_id);
+    subscriptions_.insert({subscription_id, std::weak_ptr<FilterConfigSubscription>(subscription)});
+    return subscription;
+  } else {
+    auto existing = it->second.lock();
+    RELEASE_ASSERT(existing != nullptr,
+                   absl::StrCat("Cannot find subscribed filter config resource ", name));
+    return existing;
+  }
+}
+
+FilterConfigProviderPtr FilterConfigProviderManagerImpl::createDynamicFilterConfigProvider(
+    const envoy::config::core::v3::ConfigSource& config_source,
+    const std::string& filter_config_name, bool require_terminal,
+    Server::Configuration::FactoryContext& factory_context, const std::string& stat_prefix,
+    bool apply_without_warming) {
+  auto subscription =
+      getSubscription(config_source, filter_config_name, factory_context, stat_prefix);
+  if (!apply_without_warming) {
+    factory_context.initManager().add(subscription->initTarget());
+  }
+  auto provider = std::make_unique<DynamicFilterConfigProviderImpl>(
+      std::move(subscription), require_terminal, factory_context);
+  // Ensure the subscription starts if it has not already.
+  if (apply_without_warming) {
+    factory_context.initManager().add(provider->init_target_);
+  }
+  return provider;
+}
+
+} // namespace Router
+} // namespace Envoy