Adding EDS caching - entire change - Do not submit

Signed-off-by: Adi Suissa-Peleg <adip@google.com>

envoy/config/BUILD

@@ -139,3 +139,11 @@ envoy_cc_library(
         "@com_google_googleapis//google/rpc:status_cc_proto",
     ],
 )
+
+envoy_cc_library(
+    name = "eds_resources_cache_interface",
+    hdrs = ["eds_resources_cache.h"],
+    deps = [
+        "@envoy_api//envoy/config/endpoint/v3:pkg_cc_proto",
+    ],
+)

envoy/config/eds_resources_cache.h

@@ -0,0 +1,109 @@
+#pragma once
+
+#include "envoy/common/optref.h"
+#include "envoy/common/pure.h"
+#include "envoy/config/endpoint/v3/endpoint.pb.h"
+
+#include "absl/strings/string_view.h"
+
+namespace Envoy {
+namespace Config {
+
+// An interface for cached resource removed callback.
+class EdsResourceRemovalCallback {
+public:
+  virtual ~EdsResourceRemovalCallback() = default;
+
+  // Invoked when a cached resource is removed from the cache.
+  virtual void onCachedResourceRemoved(absl::string_view resource_name) PURE;
+};
+
+// Represents an xDS resources cache for EDS resources, and currently supports
+// a single config-source (ADS). The motivation is ithat clusters that are
+// updated (not added) during a CDS response will be able to use the current EDS
+// configuration, thus avoiding an additional EDS response.
+// However, using the cached EDS config is not always desired, for example when
+// the cluster changes from non-TLS to TLS (see discussion in #5168). Thus, this
+// cache allows storing EDS resources, and using them only when the EDS
+// subscription decides to.
+//
+// This cache will be instantiated once in the cluster-manager, and passed to
+// the xDS components that handle subscription registrations (currently only gRPC-based
+// subscriptions are supported), and to the EDS subscriptions.
+//
+// Resources lifetime in the cache is determined by the gRPC mux that adds/updates a
+// resource when it receives its contents, and removes a resource when there is
+// no longer interest in that resource.
+// An EDS subscription may fetch a resource from the cache, and optionally
+// install a callback to be triggered if the resource is removes from the cache.
+// In addition, a resource in the cache may have an expiration timer if
+// "endpoint_stale_after" (TTL) is set for that resource. Once the timer
+// expires, the callbacks will be triggered to remove the resource.
+class EdsResourcesCache {
+public:
+  virtual ~EdsResourcesCache() = default;
+
+  /**
+   * Adds or updates a given resource name with its resource.
+   * Any watcher that was previously assigned to the resource will be removed
+   * without any notification.
+   * @param resource_name the name of the resource to add/update.
+   * @param resource the contents of the resource.
+   */
+  virtual void
+  setResource(absl::string_view resource_name,
+              const envoy::config::endpoint::v3::ClusterLoadAssignment& resource) PURE;
+
+  /**
+   * Removes a resource from the resource cache given the resource name.
+   * The watchers for the resource will be notified that the resource is
+   * removed.
+   * @param resource_name the name of the resource that will be removed from
+   *        the cache.
+   */
+  virtual void removeResource(absl::string_view resource_name) PURE;
+
+  /**
+   * Retrieves a resource from the cache, and adds the given callback (if any)
+   * to the resource's removal list. if the resource is removed, all callbacks
+   * for that resource will be invoked.
+   * @param resource_name the name of the resource to fetch.
+   * @param removal_cb an optional callback that will be invoked if the resource is removed
+   *        in the future. Note that setting the resource will also remove the callback.
+   * @return A reference to the cluster load assignment resource, or nullopt if the
+   *         resource doesn't exist.
+   */
+  virtual OptRef<const envoy::config::endpoint::v3::ClusterLoadAssignment>
+  getResource(absl::string_view resource_name, EdsResourceRemovalCallback* removal_cb) PURE;
+
+  /**
+   * Removes a callback for a given resource name (if it was previously added).
+   * @param resource_name the name of the resource for which the callback should be removed.
+   * @param removal_cb a pointer to the callback that needs to be removed.
+   */
+  virtual void removeCallback(absl::string_view resource_name,
+                              EdsResourceRemovalCallback* removal_cb) PURE;
+
+  /**
+   * Returns the number of items in the cache. Only used in tests.
+   * @return the number of items in the cache.
+   */
+  virtual uint32_t cacheSizeForTest() const PURE;
+
+  /**
+   * Sets an expiry timer for the given resource_name after the given ms milliseconds.
+   * Once the timer expires, the callbacks for that resource (if any) will be
+  */
+  virtual void setExpiryTimer(absl::string_view resource_name, std::chrono::milliseconds ms) PURE;
+
+  /**
+   * Disables the expiration timer for the given resource_name.
+  */
+  virtual void disableExpiryTimer(absl::string_view resource_name) PURE;
+};
+
+using EdsResourcesCachePtr = std::unique_ptr<EdsResourcesCache>;
+using EdsResourcesCacheOptRef = OptRef<EdsResourcesCache>;
+
+} // namespace Config
+} // namespace Envoy

envoy/config/grpc_mux.h

@@ -4,6 +4,7 @@
 
 #include "envoy/common/exception.h"
 #include "envoy/common/pure.h"
+#include "envoy/config/eds_resources_cache.h"
 #include "envoy/config/subscription.h"
 #include "envoy/stats/stats_macros.h"
 
@@ -105,6 +106,11 @@ class GrpcMux {
 
   virtual void requestOnDemandUpdate(const std::string& type_url,
                                      const absl::flat_hash_set<std::string>& for_update) PURE;
+
+  /**
+   * Returns an EdsResourcesCache for this GrpcMux.
+   */
+  virtual Config::EdsResourcesCacheOptRef edsResourcesCache() PURE;
 };
 
 using GrpcMuxPtr = std::unique_ptr<GrpcMux>;

envoy/config/subscription_factory.h

@@ -121,7 +121,8 @@ class MuxFactory : public Config::UntypedFactory {
          const LocalInfo::LocalInfo& local_info,
          std::unique_ptr<CustomConfigValidators>&& config_validators,
          BackOffStrategyPtr&& backoff_strategy, OptRef<XdsConfigTracker> xds_config_tracker,
-         OptRef<XdsResourcesDelegate> xds_resources_delegate) PURE;
+         OptRef<XdsResourcesDelegate> xds_resources_delegate,
+         bool use_eds_resources_cache) PURE;
 };
 
 } // namespace Config

envoy/upstream/BUILD

@@ -21,6 +21,7 @@ envoy_cc_library(
         ":upstream_interface",
         "//envoy/access_log:access_log_interface",
         "//envoy/common:random_generator_interface",
+        "//envoy/config:eds_resources_cache_interface",
         "//envoy/config:grpc_mux_interface",
         "//envoy/config:subscription_factory_interface",
         "//envoy/grpc:async_client_manager_interface",

envoy/upstream/cluster_manager.h

@@ -13,6 +13,7 @@
 #include "envoy/config/core/v3/address.pb.h"
 #include "envoy/config/core/v3/config_source.pb.h"
 #include "envoy/config/core/v3/protocol.pb.h"
+#include "envoy/config/eds_resources_cache.h"
 #include "envoy/config/grpc_mux.h"
 #include "envoy/config/subscription_factory.h"
 #include "envoy/grpc/async_client_manager.h"
@@ -444,6 +445,11 @@ class ClusterManager {
   virtual std::shared_ptr<const envoy::config::cluster::v3::Cluster::CommonLbConfig>
   getCommonLbConfigPtr(
       const envoy::config::cluster::v3::Cluster::CommonLbConfig& common_lb_config) PURE;
+
+  /**
+   * Returns an EdsResourcesCache that is unique for the cluster-manager.
+   */
+  virtual Config::EdsResourcesCacheOptRef edsResourcesCache() PURE;
 };
 
 using ClusterManagerPtr = std::unique_ptr<ClusterManager>;

envoy/upstream/upstream.h

@@ -591,6 +591,7 @@ class PrioritySet {
 #define ALL_CLUSTER_CONFIG_UPDATE_STATS(COUNTER, GAUGE, HISTOGRAM, TEXT_READOUT, STATNAME)         \
   COUNTER(assignment_stale)                                                                        \
   COUNTER(assignment_timeout_received)                                                             \
+  COUNTER(assignment_use_cached)                                                                   \
   COUNTER(update_attempt)                                                                          \
   COUNTER(update_empty)                                                                            \
   COUNTER(update_failure)                                                                          \

source/common/config/null_grpc_mux_impl.h

@@ -27,6 +27,10 @@ class NullGrpcMuxImpl : public GrpcMux,
     ENVOY_BUG(false, "unexpected request for on demand update");
   }
 
+  EdsResourcesCacheOptRef edsResourcesCache() override {
+    return absl::nullopt;
+  }
+
   void onWriteable() override {}
   void onStreamEstablished() override {}
   void onEstablishmentFailure() override {}

source/common/config/resource_name.h

@@ -1,7 +1,6 @@
 #pragma once
 
 #include <string>
-#include <vector>
 
 namespace Envoy {
 namespace Config {

source/common/runtime/runtime_features.cc

@@ -86,6 +86,7 @@ RUNTIME_GUARD(envoy_restart_features_explicit_wildcard_resource);
 RUNTIME_GUARD(envoy_restart_features_remove_runtime_singleton);
 RUNTIME_GUARD(envoy_restart_features_udp_read_normalize_addresses);
 RUNTIME_GUARD(envoy_restart_features_use_apple_api_for_dns_lookups);
+RUNTIME_GUARD(envoy_restart_features_use_eds_cache_for_ads);
 
 // Begin false flags. These should come with a TODO to flip true.
 // Sentinel and test flag.

source/common/upstream/cluster_manager_impl.cc

@@ -413,7 +413,8 @@ ClusterManagerImpl::ClusterManagerImpl(
               ->createUncachedRawAsyncClient(),
           main_thread_dispatcher, random_, *stats_.rootScope(), dyn_resources.ads_config(),
           local_info, std::move(custom_config_validators), std::move(backoff_strategy),
-          makeOptRefFromPtr(xds_config_tracker_.get()), {});
+          makeOptRefFromPtr(xds_config_tracker_.get()), {},
+          Runtime::runtimeFeatureEnabled("envoy.restart_features.use_eds_cache_for_ads"));
     } else {
       Config::Utility::checkTransportVersion(dyn_resources.ads_config());
       const std::string target_xds_authority =
@@ -436,7 +437,8 @@ ClusterManagerImpl::ClusterManagerImpl(
               ->createUncachedRawAsyncClient(),
           main_thread_dispatcher, random_, *stats_.rootScope(), dyn_resources.ads_config(),
           local_info, std::move(custom_config_validators), std::move(backoff_strategy),
-          makeOptRefFromPtr(xds_config_tracker_.get()), xds_delegate_opt_ref);
+          makeOptRefFromPtr(xds_config_tracker_.get()), xds_delegate_opt_ref,
+          Runtime::runtimeFeatureEnabled("envoy.restart_features.use_eds_cache_for_ads"));
     }
   } else {
     ads_mux_ = std::make_unique<Config::NullGrpcMuxImpl>();
@@ -1365,6 +1367,14 @@ void ClusterManagerImpl::notifyClusterDiscoveryStatus(absl::string_view name,
       });
 }
 
+Config::EdsResourcesCacheOptRef ClusterManagerImpl::edsResourcesCache() {
+  // EDS caching is only supported for ADS.
+  if (ads_mux_) {
+    return ads_mux_->edsResourcesCache();
+  }
+  return {};
+}
+
 ClusterDiscoveryManager
 ClusterManagerImpl::createAndSwapClusterDiscoveryManager(std::string thread_name) {
   ThreadLocalClusterManagerImpl& cluster_manager = *tls_;

source/common/upstream/cluster_manager_impl.h

@@ -11,11 +11,13 @@
 
 #include "envoy/api/api.h"
 #include "envoy/common/callback.h"
+#include "envoy/common/optref.h"
 #include "envoy/common/random_generator.h"
 #include "envoy/config/bootstrap/v3/bootstrap.pb.h"
 #include "envoy/config/cluster/v3/cluster.pb.h"
 #include "envoy/config/core/v3/address.pb.h"
 #include "envoy/config/core/v3/config_source.pb.h"
+#include "envoy/config/eds_resources_cache.h"
 #include "envoy/config/xds_resources_delegate.h"
 #include "envoy/http/codes.h"
 #include "envoy/local_info/local_info.h"
@@ -362,6 +364,8 @@ class ClusterManagerImpl : public ClusterManager,
     return common_lb_config_pool_->getObject(common_lb_config);
   }
 
+  Config::EdsResourcesCacheOptRef edsResourcesCache() override;
+
 protected:
   virtual void postThreadLocalRemoveHosts(const Cluster& cluster, const HostVector& hosts_removed);