Almost complete rewrite of cache filter

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

@@ -20,7 +20,7 @@ option (udpa.annotations.file_status).package_version_status = ACTIVE;
 // [#protodoc-title: HTTP Cache Filter]
 
 // [#extension: envoy.filters.http.cache]
-// [#next-free-field: 7]
+// [#next-free-field: 8]
 message CacheConfig {
   option (udpa.annotations.versioning).previous_message_type =
       "envoy.config.filter.http.cache.v2alpha.CacheConfig";
@@ -93,4 +93,16 @@ message CacheConfig {
   // causes the cache to validate with its upstream even if the lookup is a hit. Setting this
   // to true will ignore these headers.
   bool ignore_request_cache_control_header = 6;
+
+  // If this is set, requests sent upstream to populate the cache will go to the
+  // specified cluster rather than the cluster selected by the vhost and route.
+  //
+  // If you have actions to be taken by the router filter - either
+  // ``upstream_http_filters`` or one of the ``RouteConfiguration`` actions such as
+  // ``response_headers_to_add`` - then the cache's side-channel going directly to the
+  // routed cluster will bypass these actions. You can set ``override_upstream_cluster``
+  // to an internal listener which duplicates the relevant ``RouteConfiguration``, to
+  // replicate the desired behavior on the side-channel upstream request issued by the
+  // cache.
+  string override_upstream_cluster = 7;
 }

source/extensions/common/dynamic_forward_proxy/dns_cache_manager_impl.cc

@@ -29,7 +29,7 @@ absl::StatusOr<DnsCacheSharedPtr> DnsCacheManagerImpl::getCache(
   auto cache_or_status = DnsCacheImpl::createDnsCacheImpl(context_, config);
   RETURN_IF_NOT_OK_REF(cache_or_status.status());
   DnsCacheSharedPtr new_cache = std::move(cache_or_status.value());
-  caches_.emplace(config.name(), ActiveCache{config, new_cache});
+  caches_.emplace(config.name(), CacheSessions{config, new_cache});
   return new_cache;
 }
 

source/extensions/common/dynamic_forward_proxy/dns_cache_manager_impl.h

@@ -23,17 +23,18 @@ class DnsCacheManagerImpl : public DnsCacheManager, public Singleton::Instance {
   DnsCacheSharedPtr lookUpCacheByName(absl::string_view cache_name) override;
 
 private:
-  struct ActiveCache {
-    ActiveCache(const envoy::extensions::common::dynamic_forward_proxy::v3::DnsCacheConfig& config,
-                DnsCacheSharedPtr cache)
+  struct CacheSessions {
+    CacheSessions(
+        const envoy::extensions::common::dynamic_forward_proxy::v3::DnsCacheConfig& config,
+        DnsCacheSharedPtr cache)
         : config_(config), cache_(cache) {}
 
     const envoy::extensions::common::dynamic_forward_proxy::v3::DnsCacheConfig config_;
     DnsCacheSharedPtr cache_;
   };
 
   Server::GenericFactoryContextImpl context_;
-  absl::flat_hash_map<std::string, ActiveCache> caches_;
+  absl::flat_hash_map<std::string, CacheSessions> caches_;
 };
 
 class DnsCacheManagerFactoryImpl : public DnsCacheManagerFactory {

source/extensions/filters/http/cache/BUILD

@@ -12,25 +12,87 @@ licenses(["notice"])  # Apache 2
 
 envoy_extension_package()
 
+envoy_cc_library(
+    name = "http_source_interface",
+    hdrs = ["http_source.h"],
+    deps = [
+        "//envoy/buffer:buffer_interface",
+        "//envoy/http:header_map_interface",
+        "//source/extensions/filters/http/cache:range_utils_lib",
+        "@com_google_absl//absl/functional:any_invocable",
+    ],
+)
+
+envoy_cc_library(
+    name = "upstream_request_lib",
+    srcs = ["upstream_request_impl.cc"],
+    hdrs = [
+        "upstream_request.h",
+        "upstream_request_impl.h",
+    ],
+    deps = [
+        ":http_source_interface",
+        ":range_utils_lib",
+        ":stats",
+        "//source/common/buffer:watermark_buffer_lib",
+        "//source/common/common:cancel_wrapper_lib",
+        "//source/common/common:logger_lib",
+        "@com_google_absl//absl/types:variant",
+    ],
+)
+
+envoy_cc_library(
+    name = "cache_sessions_lib",
+    srcs = [
+        "cache_sessions.cc",
+    ],
+    hdrs = [
+        "cache_sessions.h",
+    ],
+    deps = [
+        ":http_cache_lib",
+        ":stats",
+        ":upstream_request_lib",
+        "//source/common/http:utility_lib",
+    ],
+)
+
+envoy_cc_library(
+    name = "cache_sessions_impl_lib",
+    srcs = [
+        "cache_sessions_impl.cc",
+    ],
+    hdrs = [
+        "cache_sessions_impl.h",
+    ],
+    deps = [
+        ":cache_sessions_lib",
+        ":cacheability_utils_lib",
+        ":upstream_request_lib",
+        "//source/common/common:cancel_wrapper_lib",
+    ],
+)
+
 envoy_cc_library(
     name = "cache_filter_lib",
     srcs = [
         "cache_filter.cc",
-        "upstream_request.cc",
     ],
     hdrs = [
         "cache_filter.h",
-        "filter_state.h",
-        "upstream_request.h",
     ],
     deps = [
         ":cache_custom_headers",
         ":cache_entry_utils_lib",
-        ":cache_filter_logging_info_lib",
         ":cache_headers_utils_lib",
-        ":cache_insert_queue_lib",
+        ":cache_sessions_impl_lib",
+        ":cache_sessions_lib",
         ":cacheability_utils_lib",
         ":http_cache_lib",
+        ":stats",
+        ":upstream_request_lib",
+        "//source/common/buffer:buffer_lib",
+        "//source/common/common:cancel_wrapper_lib",
         "//source/common/common:enum_to_int",
         "//source/common/common:logger_lib",
         "//source/common/common:macros",
@@ -65,24 +127,13 @@ envoy_cc_library(
     ],
 )
 
-envoy_cc_library(
-    name = "cache_insert_queue_lib",
-    srcs = ["cache_insert_queue.cc"],
-    hdrs = ["cache_insert_queue.h"],
-    deps = [
-        ":http_cache_lib",
-        "//source/common/buffer:buffer_lib",
-    ],
-)
-
 envoy_cc_library(
     name = "cache_policy_lib",
     hdrs = ["cache_policy.h"],
     deps = [
         ":cache_headers_utils_lib",
         ":http_cache_lib",
         "//source/common/http:header_map_lib",
-        "//source/common/stream_info:filter_state_lib",
     ],
 )
 
@@ -91,6 +142,15 @@ envoy_proto_library(
     srcs = ["key.proto"],
 )
 
+envoy_cc_library(
+    name = "cache_progress_receiver_interface",
+    hdrs = ["cache_progress_receiver.h"],
+    deps = [
+        "//envoy/http:header_map_interface",
+        "//source/extensions/filters/http/cache:range_utils_lib",
+    ],
+)
+
 envoy_cc_library(
     name = "http_cache_lib",
     srcs = ["http_cache.cc"],
@@ -99,6 +159,8 @@ envoy_cc_library(
         ":cache_custom_headers",
         ":cache_entry_utils_lib",
         ":cache_headers_utils_lib",
+        ":cache_progress_receiver_interface",
+        ":http_source_interface",
         ":key_cc_proto",
         ":range_utils_lib",
         "//envoy/buffer:buffer_interface",
@@ -137,6 +199,7 @@ envoy_cc_library(
     hdrs = ["cache_headers_utils.h"],
     deps = [
         ":cache_custom_headers",
+        ":key_cc_proto",
         "//envoy/common:time_interface",
         "//envoy/http:header_map_interface",
         "//source/common/common:matchers_lib",
@@ -159,12 +222,12 @@ envoy_cc_library(
     ],
 )
 
-envoy_cc_library(
-    name = "cache_filter_logging_info_lib",
-    srcs = ["cache_filter_logging_info.cc"],
-    hdrs = ["cache_filter_logging_info.h"],
+envoy_cc_extension(
+    name = "stats",
+    srcs = ["stats.cc"],
+    hdrs = ["stats.h"],
     deps = [
-        "//source/common/stream_info:filter_state_lib",
+        ":cache_entry_utils_lib",
     ],
 )
 
@@ -174,6 +237,8 @@ envoy_cc_extension(
     hdrs = ["config.h"],
     deps = [
         ":cache_filter_lib",
+        ":cache_sessions_lib",
+        ":stats",
         "//source/extensions/filters/http/common:factory_base_lib",
         "@envoy_api//envoy/extensions/filters/http/cache/v3:pkg_cc_proto",
     ],

source/extensions/filters/http/cache/cache_entry_utils.cc

@@ -9,16 +9,26 @@ namespace Cache {
 
 absl::string_view cacheEntryStatusString(CacheEntryStatus s) {
   switch (s) {
-  case CacheEntryStatus::Ok:
-    return "Ok";
-  case CacheEntryStatus::Unusable:
-    return "Unusable";
-  case CacheEntryStatus::RequiresValidation:
-    return "RequiresValidation";
+  case CacheEntryStatus::Hit:
+    return "Hit";
+  case CacheEntryStatus::Miss:
+    return "Miss";
+  case CacheEntryStatus::Follower:
+    return "Follower";
+  case CacheEntryStatus::Uncacheable:
+    return "Uncacheable";
+  case CacheEntryStatus::Validated:
+    return "Validated";
+  case CacheEntryStatus::ValidatedFree:
+    return "ValidatedFree";
+  case CacheEntryStatus::FailedValidation:
+    return "FailedValidation";
   case CacheEntryStatus::FoundNotModified:
     return "FoundNotModified";
   case CacheEntryStatus::LookupError:
     return "LookupError";
+  case CacheEntryStatus::UpstreamReset:
+    return "UpstreamReset";
   }
   IS_ENVOY_BUG(absl::StrCat("Unexpected CacheEntryStatus: ", s));
   return "UnexpectedCacheEntryStatus";

source/extensions/filters/http/cache/cache_entry_utils.h

@@ -22,23 +22,37 @@ struct ResponseMetadata {
   // calculations at: https://httpwg.org/specs/rfc7234.html#age.calculations
   Envoy::SystemTime response_time_;
 };
-using ResponseMetadataPtr = std::unique_ptr<ResponseMetadata>;
 
 // Whether a given cache entry is good for the current request.
 enum class CacheEntryStatus {
   // This entry is fresh, and an appropriate response to the request.
-  Ok,
-  // No usable entry was found. If this was generated for a cache entry, the
-  // cache should delete that entry.
-  Unusable,
-  // This entry is stale, but appropriate for validating
-  RequiresValidation,
+  Hit,
+  // The request was cacheable and was not already in the cache. This also means
+  // the cache was populated by this request.
+  Miss,
+  // The entry was being inserted when this request was made - it's like a
+  // hit, but streamed from the same request as the original "Miss", so still
+  // potentially subject to upstream reset because the cache entry isn't fully
+  // populated yet.
+  Follower,
+  // The request was not cacheable. All matching requests will go to the
+  // upstream.
+  Uncacheable,
+  // This entry required validation, and validated successfully.
+  Validated,
+  // This entry required validation while another entry was already validating,
+  // so it validated successfully without its own lookup.
+  ValidatedFree,
+  // This entry required validation, and did not validate.
+  FailedValidation,
   // This entry is fresh, and an appropriate basis for a 304 Not Modified
   // response.
   FoundNotModified,
   // The cache lookup failed, e.g. because the cache was unreachable or an RPC
-  // timed out. The caller shouldn't use this lookup's context for an insert.
+  // timed out. Mostly behaves the same as Uncacheable but may retry each time.
   LookupError,
+  // The cache attempted to read from upstream for insert, but upstream reset.
+  UpstreamReset,
 };
 
 absl::string_view cacheEntryStatusString(CacheEntryStatus s);