WIP: config: Add FilterChain Discovery Service.

api/envoy/api/v2/BUILD

@@ -97,6 +97,24 @@ api_go_grpc_library(
     ],
 )
 
+api_proto_library_internal(
+    name = "fcds",
+    srcs = ["fcds.proto"],
+    has_services = 1,
+    visibility = [":friends"],
+    deps = [
+        ":discovery",
+    ],
+)
+
+api_go_grpc_library(
+    name = "fcds",
+    proto = ":fcds",
+    deps = [
+        ":discovery_go_proto",
+    ],
+)
+
 api_proto_library_internal(
     name = "lds",
     srcs = ["lds.proto"],
@@ -106,6 +124,7 @@ api_proto_library_internal(
         ":discovery",
         "//envoy/api/v2/core:address",
         "//envoy/api/v2/core:base",
+        "//envoy/api/v2/core:config_source",
         "//envoy/api/v2/listener",
     ],
 )
@@ -117,6 +136,7 @@ api_go_grpc_library(
         ":discovery_go_proto",
         "//envoy/api/v2/core:address_go_proto",
         "//envoy/api/v2/core:base_go_proto",
+        "//envoy/api/v2/core:config_source_go_proto",
         "//envoy/api/v2/listener:listener_go_proto",
     ],
 )

api/envoy/api/v2/fcds.proto

@@ -0,0 +1,36 @@
+syntax = "proto3";
+
+package envoy.api.v2;
+
+option java_generic_services = true;
+
+import "envoy/api/v2/discovery.proto";
+
+import "google/api/annotations.proto";
+import "google/protobuf/wrappers.proto";
+
+import "validate/validate.proto";
+import "gogoproto/gogo.proto";
+
+option (gogoproto.equal_all) = true;
+
+// [#protodoc-title: Filter Chain Discovery Service]
+
+// The resource_names field in DiscoveryRequest specifies a listener.
+// The resources field in DiscoveryResponse is values of type listener.FilterChain.
+service FilterChainDiscoveryService {
+  rpc StreamFilterChains(stream DiscoveryRequest) returns (stream DiscoveryResponse) {
+  }
+
+  rpc IncrementalFilterChains(stream IncrementalDiscoveryRequest)
+      returns (stream IncrementalDiscoveryResponse) {
+  }
+
+  rpc FetchFilterChains(DiscoveryRequest) returns (DiscoveryResponse) {
+    option (google.api.http) = {
+      post: "/v2/discovery:filterchains"
+      body: "*"
+    };
+  }
+}
+

api/envoy/api/v2/lds.proto

@@ -6,6 +6,7 @@ option java_generic_services = true;
 
 import "envoy/api/v2/core/address.proto";
 import "envoy/api/v2/core/base.proto";
+import "envoy/api/v2/core/config_source.proto";
 import "envoy/api/v2/discovery.proto";
 import "envoy/api/v2/listener/listener.proto";
 
@@ -37,7 +38,7 @@ service ListenerDiscoveryService {
   }
 }
 
-// [#comment:next free field: 16]
+// [#comment:next free field: 17]
 message Listener {
   // The unique name by which this listener is known. If no name is provided,
   // Envoy will allocate an internal UUID for the listener. If the listener is to be dynamically
@@ -57,10 +58,21 @@ message Listener {
   // :ref:`FilterChainMatch <envoy_api_msg_listener.FilterChainMatch>` criteria is used on a
   // connection.
   //
+  // Precisely one of filter_chains and fdcs_config must be set.
+  //
   // Example using SNI for filter chain selection can be found in the
   // :ref:`FAQ entry <faq_how_to_setup_sni>`.
-  repeated listener.FilterChain filter_chains = 3
-      [(validate.rules).repeated .min_items = 1, (gogoproto.nullable) = false];
+  repeated listener.FilterChain filter_chains = 3 [(gogoproto.nullable) = false];
+
+  message FcdsConfig {
+    core.ConfigSource config = 1 [(validate.rules).message.required = true];
+
+    // Optional alternative to the listener name to present to FCDS.
+    string filter_chain_name = 2;
+  }
+
+  // Precisely one of filter_chains and fdcs_config must be set.
+  FcdsConfig fcds_config = 17;
 
   // If a connection is redirected using *iptables*, the port on which the proxy
   // receives it might be different from the original destination address. When this flag is set to

source/server/BUILD

@@ -228,6 +228,7 @@ envoy_cc_library(
         "//include/envoy/server:worker_interface",
         "//source/common/api:os_sys_calls_lib",
         "//source/common/common:empty_string",
+        "//source/common/config:subscription_factory_lib",
         "//source/common/config:utility_lib",
         "//source/common/network:cidr_range_lib",
         "//source/common/network:lc_trie_lib",

source/server/listener_manager_impl.cc

@@ -135,6 +135,25 @@ ListenerImpl::ListenerImpl(const envoy::api::v2::Listener& config, const std::st
       config_(config), version_info_(version_info),
       listener_filters_timeout_(
           PROTOBUF_GET_MS_OR_DEFAULT(config, listener_filters_timeout, 15000)) {
+  if (config.has_fcds_config()) {
+    if (!config.filter_chains().empty()) {
+      throw EnvoyException(fmt::format(
+          "Invalid configuration of listener {}: both fcds_config and filter_chains are set.",
+          config.name()));
+    }
+
+    fcds_subscription_ = Config::SubscriptionFactory::subscriptionFromConfigSource<
+        envoy::api::v2::listener::FilterChain>(
+        config.fcds_config().config(), parent.server_.localInfo(), parent.server_.dispatcher(),
+        parent.server_.clusterManager(), parent.server_.random(), *listener_scope_,
+        []() -> Config::Subscription<envoy::api::v2::listener::FilterChain>* {
+          throw EnvoyException("FilterChainDiscoveryService does not support rest_legacy");
+        },
+        "envoy.api.v2.FilterChainDiscoveryService.FetchFilterChains",
+        "envoy.api.v2.FilterChainDiscoveryService.StreamFilterChains");
+
+    initManager().registerTarget(*this);
+  }
   if (config.has_transparent()) {
     addListenSocketOptions(Network::SocketOptionFactory::buildIpTransparentOptions());
   }
@@ -167,7 +186,8 @@ ListenerImpl::ListenerImpl(const envoy::api::v2::Listener& config, const std::st
   // TODO(jrajahalme): This is the last listener filter on purpose. When filter chain matching
   //                   is implemented, this needs to be run after the filter chain has been
   //                   selected.
-  if (PROTOBUF_GET_WRAPPED_OR_DEFAULT(config.filter_chains()[0], use_proxy_proto, false)) {
+  if (!config.filter_chains().empty() &&
+      PROTOBUF_GET_WRAPPED_OR_DEFAULT(config.filter_chains()[0], use_proxy_proto, false)) {
     auto& factory =
         Config::Utility::getAndCheckFactory<Configuration::NamedListenerFilterConfigFactory>(
             Extensions::ListenerFilters::ListenerFilterNames::get().ProxyProtocol);
@@ -592,6 +612,32 @@ void ListenerImpl::initialize() {
   }
 }
 
+void ListenerImpl::initialize(std::function<void()> callback) {
+  fcds_initialized_cb_ = callback;
+
+  const auto& alternate_name = config_.fcds_config().filter_chain_name();
+  const std::string& resource = alternate_name.empty() ? config_.name() : alternate_name;
+  fcds_subscription_->start({resource}, *this);
+}
+
+void ListenerImpl::onConfigUpdate(const ResourceVector& resources,
+                                  const std::string& version_info) {
+  ENVOY_LOG(warn, "onConfigUpdate {} {}", resources.size(), version_info);
+  fcds_initialized_cb_();
+  fcds_initialized_cb_ = nullptr;
+}
+
+void ListenerImpl::onConfigUpdateFailed(const EnvoyException* e) {
+  ENVOY_LOG(warn, "onConfigUpdateFailed {}", e->what());
+  fcds_initialized_cb_();
+  fcds_initialized_cb_ = nullptr;
+}
+
+std::string ListenerImpl::resourceName(const ProtobufWkt::Any&) {
+  // TODO
+  return "";
+}
+
 Init::Manager& ListenerImpl::initManager() {
   // See initialize() for why we choose different init managers to return.
   if (workers_started_) {

source/server/listener_manager_impl.h

@@ -12,6 +12,7 @@
 #include "envoy/stats/scope.h"
 
 #include "common/common/logger.h"
+#include "common/config/subscription_factory.h"
 #include "common/network/cidr_range.h"
 #include "common/network/lc_trie.h"
 
@@ -188,6 +189,8 @@ class ListenerManagerImpl : public ListenerManager, Logger::Loggable<Logger::Id:
  * Maps proto config to runtime config for a listener with a network filter chain.
  */
 class ListenerImpl : public Network::ListenerConfig,
+                     public Init::Target,
+                     public Config::SubscriptionCallbacks<envoy::api::v2::listener::FilterChain>,
                      public Configuration::ListenerFactoryContext,
                      public Network::DrainDecision,
                      public Network::FilterChainManager,
@@ -257,6 +260,14 @@ class ListenerImpl : public Network::ListenerConfig,
   const std::string& name() const override { return name_; }
   bool reverseWriteFilterOrder() const override { return reverse_write_filter_order_; }
 
+  // Init::Target
+  void initialize(std::function<void()> callback) override;
+
+  // Config::SubscriptionCallbacks
+  void onConfigUpdate(const ResourceVector& resources, const std::string& version_info) override;
+  void onConfigUpdateFailed(const EnvoyException* e) override;
+  std::string resourceName(const ProtobufWkt::Any& resource) override;
+
   // Server::Configuration::ListenerFactoryContext
   AccessLog::AccessLogManager& accessLogManager() override {
     return parent_.server_.accessLogManager();
@@ -407,6 +418,8 @@ class ListenerImpl : public Network::ListenerConfig,
   const std::string version_info_;
   Network::Socket::OptionsSharedPtr listen_socket_options_;
   const std::chrono::milliseconds listener_filters_timeout_;
+  std::unique_ptr<Config::Subscription<envoy::api::v2::listener::FilterChain>> fcds_subscription_;
+  std::function<void()> fcds_initialized_cb_;
 };
 
 class FilterChainImpl : public Network::FilterChain {

test/integration/BUILD

@@ -75,6 +75,18 @@ envoy_cc_test(
     ],
 )
 
+envoy_cc_test(
+    name = "fcds_integration_test",
+    srcs = ["fcds_integration_test.cc"],
+    deps = [
+        ":http_integration_lib",
+        "//source/common/upstream:load_balancer_lib",
+        "//test/config:utility_lib",
+        "//test/test_common:network_utility_lib",
+        "@envoy_api//envoy/api/v2:fcds_cc",
+    ],
+)
+
 # TODO(mattklein123): This test uses extensions mixed in, so we just register all extensions.
 # This will go away when we delete v1 configuration.
 envoy_cc_test(

test/integration/fcds_integration_test.cc

@@ -0,0 +1,35 @@
+#include "test/config/utility.h"
+#include "test/integration/http_integration.h"
+#include "test/test_common/network_utility.h"
+
+#include "gtest/gtest.h"
+
+namespace Envoy {
+namespace {
+
+// Integration test for EDS features. EDS is consumed via filesystem
+// subscription.
+class FcdsIntegrationTest : public HttpIntegrationTest,
+                            public testing::TestWithParam<Network::Address::IpVersion> {
+public:
+  FcdsIntegrationTest()
+      : HttpIntegrationTest(Http::CodecClient::Type::HTTP1, GetParam(), realTime()) {}
+
+  void initialize() override {
+    // setUpstreamCount(4);
+    config_helper_.addConfigModifier([this](envoy::config::bootstrap::v2::Bootstrap& bootstrap) {
+      auto* listener_0 = bootstrap.mutable_static_resources()->mutable_listeners(0);
+      listener_0->mutable_filter_chains()->Clear();
+      listener_0->mutable_fcds_config()->mutable_config()->set_path("/");
+    });
+    HttpIntegrationTest::initialize();
+  }
+};
+
+INSTANTIATE_TEST_CASE_P(IpVersions, FcdsIntegrationTest,
+                        testing::ValuesIn(TestEnvironment::getIpVersionsForTest()));
+
+TEST_P(FcdsIntegrationTest, test) { initialize(); }
+
+} // namespace
+} // namespace Envoy