on-demand, upstream: on-demand cluster discovery

api/envoy/extensions/filters/http/on_demand/v3/BUILD

@@ -6,6 +6,7 @@ licenses(["notice"])  # Apache 2
 
 api_proto_package(
     deps = [
+        "//envoy/config/core/v3:pkg",
         "//envoy/config/filter/http/on_demand/v2:pkg",
         "@com_github_cncf_udpa//udpa/annotations:pkg",
     ],

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

@@ -2,6 +2,8 @@ syntax = "proto3";
 
 package envoy.extensions.filters.http.on_demand.v3;
 
+import "envoy/config/core/v3/config_source.proto";
+
 import "udpa/annotations/status.proto";
 import "udpa/annotations/versioning.proto";
 
@@ -10,11 +12,31 @@ option java_outer_classname = "OnDemandProto";
 option java_multiple_files = true;
 option (udpa.annotations.file_status).package_version_status = ACTIVE;
 
-// [#protodoc-title: OnDemand]
-// IP tagging :ref:`configuration overview <config_http_filters_on_demand>`.
+// [#protodoc-title: On Demand Discovery]
+// On Demand Discovery :ref:`configuration overview <config_http_filters_on_demand>`.
 // [#extension: envoy.filters.http.on_demand]
 
+// On Demand Discovery filter config.
 message OnDemand {
   option (udpa.annotations.versioning).previous_message_type =
       "envoy.config.filter.http.on_demand.v2.OnDemand";
+
+  // An optional configuration for on-demand cluster discovery
+  // service. If not specified, the on-demand cluster discovery will
+  // be disabled. When it's specified, the filter will pause a request
+  // to an unknown cluster and will begin a cluster discovery
+  // process. When the discovery is finished (successfully or not),
+  // the request will be resumed for further processing.
+  config.core.v3.ConfigSource odcds_config = 1;
+}
+
+// Per-route configuration for On Demand Discovery.
+message PerRouteConfig {
+  // An optional configuration for on-demand cluster discovery
+  // service. If not specified, the on-demand cluster discovery will
+  // be disabled. When it's specified, the filter will pause a request
+  // to an unknown cluster and will begin a cluster discovery
+  // process. When the discovery is finished (successfully or not),
+  // the request will be resumed for further processing.
+  config.core.v3.ConfigSource odcds_config = 1;
 }

include/envoy/upstream/cluster_manager.h

@@ -39,7 +39,7 @@ namespace Envoy {
 namespace Upstream {
 
 /**
- * ClusterUpdateCallbacks provide a way to exposes Cluster lifecycle events in the
+ * ClusterUpdateCallbacks provide a way to expose Cluster lifecycle events in the
  * ClusterManager.
  */
 class ClusterUpdateCallbacks {
@@ -72,6 +72,39 @@ class ClusterUpdateCallbacksHandle {
 
 using ClusterUpdateCallbacksHandlePtr = std::unique_ptr<ClusterUpdateCallbacksHandle>;
 
+/**
+ * Status enum for the result of an attempted cluster discovery.
+ */
+enum class ClusterDiscoveryStatus {
+  /**
+   * Cluster was not found during the discovery process.
+   */
+  Missing,
+  /**
+   * Cluster found and currently available through ClusterManager.
+   */
+  Available,
+};
+
+/**
+ * ClusterDiscoveryCallback is a callback called at the end of the on-demand cluster discovery
+ * process. The status of the discovery is sent as a parameter.
+ */
+using ClusterDiscoveryCallback = std::function<void(ClusterDiscoveryStatus)>;
+using ClusterDiscoveryCallbackWeakPtr = std::weak_ptr<ClusterDiscoveryCallback>;
+using ClusterDiscoveryCallbackSharedPtr = std::shared_ptr<ClusterDiscoveryCallback>;
+
+/**
+ * ClusterDiscoveryCallbackHandle is a RAII wrapper for a ClusterDiscoveryCallback. Deleting the
+ * ClusterDiscoveryCallbackHandle will remove the callbacks from ClusterManager.
+ */
+class ClusterDiscoveryCallbackHandle {
+public:
+  virtual ~ClusterDiscoveryCallbackHandle() = default;
+};
+
+using ClusterDiscoveryCallbackHandlePtr = std::unique_ptr<ClusterDiscoveryCallbackHandle>;
+
 class ClusterManagerFactory;
 
 // These are per-cluster per-thread, so not "global" stats.
@@ -119,6 +152,9 @@ struct ClusterConnectivityState {
   int64_t connecting_stream_capacity_{};
 };
 
+class OdCdsApi;
+using OdCdsApiSharedPtr = std::shared_ptr<OdCdsApi>;
+
 /**
  * Manages connection pools and load balancing for upstream clusters. The cluster manager is
  * persistent and shared among multiple ongoing requests/connections.
@@ -309,6 +345,24 @@ class ClusterManager {
   virtual const ClusterRequestResponseSizeStatNames&
   clusterRequestResponseSizeStatNames() const PURE;
   virtual const ClusterTimeoutBudgetStatNames& clusterTimeoutBudgetStatNames() const PURE;
+
+  /**
+   * Request an on-demand discovery of a cluster with a passed name. Passed ODCDS may be used to
+   * perform the discovery process in the main thread if there is no discovery going on for this
+   * cluster. The passed callback will be invoked when the cluster is added and warmed up. It is
+   * expected that the callback will be destroyed when it is invoked. To cancel the discovery,
+   * destroy the returned handle and the callback.
+   *
+   * This function is thread-safe.
+   *
+   * @param odcds is a pointer to ODCDS used for discovery. Must not be a nullptr.
+   * @param name is the name of the cluster to be discovered.
+   * @param callback will be called when the discovery is finished.
+   * @return ClusterDiscoveryCallbackHandlePtr the discovery process handle.
+   */
+  virtual ClusterDiscoveryCallbackHandlePtr
+  requestOnDemandClusterDiscovery(OdCdsApiSharedPtr odcds, const std::string& name,
+                                  ClusterDiscoveryCallbackWeakPtr callback) PURE;
 };
 
 using ClusterManagerPtr = std::unique_ptr<ClusterManager>;
@@ -339,6 +393,21 @@ class CdsApi {
 
 using CdsApiPtr = std::unique_ptr<CdsApi>;
 
+/**
+ * Abstract interface for a On-Demand CDS API provider.
+ */
+class OdCdsApi {
+public:
+  virtual ~OdCdsApi() = default;
+
+  /**
+   * File an on-demand request for a cluster.
+   */
+  virtual void updateOnDemand(const std::string& cluster_name) PURE;
+};
+
+using OdCdsApiPtr = std::unique_ptr<OdCdsApi>;
+
 /**
  * Factory for objects needed during cluster manager operation.
  */

source/common/upstream/BUILD

@@ -44,6 +44,25 @@ envoy_cc_library(
     ],
 )
 
+envoy_cc_library(
+    name = "odcds_api_lib",
+    srcs = ["odcds_api_impl.cc"],
+    hdrs = ["odcds_api_impl.h"],
+    deps = [
+        ":cds_api_helper_lib",
+        "//include/envoy/config:subscription_interface",
+        "//include/envoy/protobuf:message_validator_interface",
+        "//include/envoy/stats:stats_interface",
+        "//include/envoy/upstream:cluster_manager_interface",
+        "//source/common/common:minimal_logger_lib",
+        "//source/common/config:subscription_base_interface",
+        "//source/common/grpc:common_lib",
+        "//source/common/protobuf",
+        "@envoy_api//envoy/config/cluster/v3:pkg_cc_proto",
+        "@envoy_api//envoy/config/core/v3:pkg_cc_proto",
+    ],
+)
+
 envoy_cc_library(
     name = "cluster_manager_lib",
     srcs = ["cluster_manager_impl.cc"],
@@ -52,6 +71,7 @@ envoy_cc_library(
         ":cds_api_lib",
         ":load_balancer_lib",
         ":load_stats_reporter_lib",
+        ":odcds_api_lib",
         ":ring_hash_lb_lib",
         ":subset_lb_lib",
         "//include/envoy/api:api_interface",

source/common/upstream/cds_api_helper.h

@@ -37,7 +37,7 @@ class CdsApiHelper : Logger::Loggable<Logger::Id::upstream> {
 
 private:
   ClusterManager& cm_;
-  std::string name_;
+  const std::string name_;
   std::string system_version_info_;
 };