filter: Add CacheFilter: a pluggable HTTP caching filter

source/common/common/logger.h

@@ -27,6 +27,7 @@ namespace Logger {
   FUNCTION(aws)                                                                                    \
   FUNCTION(assert)                                                                                 \
   FUNCTION(backtrace)                                                                              \
+  FUNCTION(cache_filter)                                                                           \
   FUNCTION(client)                                                                                 \
   FUNCTION(config)                                                                                 \
   FUNCTION(connection)                                                                             \

source/common/http/headers.h

@@ -59,6 +59,7 @@ class HeaderValues {
   const LowerCaseString AccessControlExposeHeaders{"access-control-expose-headers"};
   const LowerCaseString AccessControlMaxAge{"access-control-max-age"};
   const LowerCaseString AccessControlAllowCredentials{"access-control-allow-credentials"};
+  const LowerCaseString Age{"age"};
   const LowerCaseString Authorization{"authorization"};
   const LowerCaseString ProxyAuthenticate{"proxy-authenticate"};
   const LowerCaseString ProxyAuthorization{"proxy-authorization"};
@@ -172,6 +173,7 @@ class HeaderValues {
     const std::string NoCache{"no-cache"};
     const std::string NoCacheMaxAge0{"no-cache, max-age=0"};
     const std::string NoTransform{"no-transform"};
+    const std::string Private{"private"};
   } CacheControlValues;
 
   struct {

source/extensions/filters/http/cache/BUILD

@@ -12,6 +12,24 @@ load(
 
 envoy_package()
 
+envoy_cc_library(
+    name = "cache_filter_lib",
+    srcs = ["cache_filter.cc"],
+    hdrs = ["cache_filter.h"],
+    deps = [
+        ":http_cache_lib",
+        "//include/envoy/registry",
+        "//source/common/common:logger_lib",
+        "//source/common/common:macros",
+        "//source/common/config:utility_lib",
+        "//source/common/http:header_map_lib",
+        "//source/common/http:headers_lib",
+        "//source/common/protobuf",
+        "//source/extensions/filters/http/common:pass_through_filter_lib",
+        "@envoy_api//envoy/extensions/filters/http/cache/v3alpha:pkg_cc_proto",
+    ],
+)
+
 envoy_proto_library(
     name = "key",
     srcs = ["key.proto"],

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

@@ -0,0 +1,217 @@
+#include "extensions/filters/http/cache/cache_filter.h"
+
+#include "envoy/registry/registry.h"
+
+#include "common/config/utility.h"
+#include "common/http/headers.h"
+
+#include "absl/memory/memory.h"
+#include "absl/strings/string_view.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace HttpFilters {
+namespace Cache {
+
+bool CacheFilter::isCacheableRequest(Http::HeaderMap& headers) {
+  const Http::HeaderEntry* method = headers.Method();
+  const Http::HeaderEntry* forwarded_proto = headers.ForwardedProto();
+  const Http::HeaderValues& header_values = Http::Headers::get();
+  // TODO(toddmgreer): Also serve HEAD requests from cache.
+  // TODO(toddmgreer): Check all the other cache-related headers.
+  return method && forwarded_proto && headers.Path() && headers.Host() &&
+         (method->value() == header_values.MethodValues.Get) &&
+         (forwarded_proto->value() == header_values.SchemeValues.Http ||
+          forwarded_proto->value() == header_values.SchemeValues.Https);
+}
+
+bool CacheFilter::isCacheableResponse(Http::HeaderMap& headers) {
+  const Http::HeaderEntry* cache_control = headers.CacheControl();
+  // TODO(toddmgreer): fully check for cacheability. See for example
+  // https://github.com/apache/incubator-pagespeed-mod/blob/master/pagespeed/kernel/http/caching_headers.h.
+  if (cache_control) {
+    return !StringUtil::caseFindToken(cache_control->value().getStringView(), ",",
+                                      Http::Headers::get().CacheControlValues.Private);
+  }
+  return false;
+}
+
+HttpCache&
+CacheFilter::getCache(const envoy::extensions::filters::http::cache::v3alpha::CacheConfig& config) {
+  const std::string type{TypeUtil::typeUrlToDescriptorFullName(config.typed_config().type_url())};
+  HttpCacheFactory* const factory =
+      Registry::FactoryRegistry<HttpCacheFactory>::getFactoryByType(type);
+  if (factory == nullptr) {
+    throw EnvoyException(
+        fmt::format("Didn't find a registered implementation for type: '{}'", type));
+  }
+  return factory->getCache(config);
+}
+
+CacheFilter::CacheFilter(
+    const envoy::extensions::filters::http::cache::v3alpha::CacheConfig& config, const std::string&,
+    Stats::Scope&, TimeSource& time_source)
+    : time_source_(time_source), cache_(getCache(config)) {}
+
+void CacheFilter::onDestroy() {
+  lookup_ = nullptr;
+  insert_ = nullptr;
+}
+
+Http::FilterHeadersStatus CacheFilter::decodeHeaders(Http::HeaderMap& headers, bool) {
+  ENVOY_STREAM_LOG(debug, "CacheFilter::decodeHeaders: {}", *decoder_callbacks_, headers);
+  if (!isCacheableRequest(headers)) {
+    ENVOY_STREAM_LOG(debug, "CacheFilter::decodeHeaders ignoring uncacheable request: {}",
+                     *decoder_callbacks_, headers);
+    return Http::FilterHeadersStatus::Continue;
+  }
+  ASSERT(decoder_callbacks_);
+  lookup_ = cache_.makeLookupContext(LookupRequest(headers, time_source_.systemTime()));
+  ASSERT(lookup_);
+
+  CacheFilterSharedPtr self = shared_from_this();
+  ENVOY_STREAM_LOG(debug, "CacheFilter::decodeHeaders starting lookup", *decoder_callbacks_);
+  lookup_->getHeaders([self](LookupResult&& result) { onHeadersAsync(self, std::move(result)); });
+  return Http::FilterHeadersStatus::StopIteration;
+}
+
+Http::FilterHeadersStatus CacheFilter::encodeHeaders(Http::HeaderMap& headers, bool end_stream) {
+  if (lookup_ && isCacheableResponse(headers)) {
+    ENVOY_STREAM_LOG(debug, "CacheFilter::encodeHeaders inserting headers", *encoder_callbacks_);
+    insert_ = cache_.makeInsertContext(std::move(lookup_));
+    insert_->insertHeaders(headers, end_stream);
+  }
+  return Http::FilterHeadersStatus::Continue;
+}
+
+Http::FilterDataStatus CacheFilter::encodeData(Buffer::Instance& data, bool end_stream) {
+  if (insert_) {
+    ENVOY_STREAM_LOG(debug, "CacheFilter::encodeHeaders inserting body", *encoder_callbacks_);
+    // TODO(toddmgreer): Wait for the cache if necessary.
+    insert_->insertBody(
+        data, [](bool) {}, end_stream);
+  }
+  return Http::FilterDataStatus::Continue;
+}
+
+void CacheFilter::onOkHeaders(Http::HeaderMapPtr&& headers,
+                              std::vector<AdjustedByteRange>&& /*response_ranges*/,
+                              uint64_t content_length, bool has_trailers) {
+  if (!lookup_) {
+    return;
+  }
+  response_has_trailers_ = has_trailers;
+  const bool end_stream = (content_length == 0 && !response_has_trailers_);
+  // TODO(toddmgreer): Calculate age per https://httpwg.org/specs/rfc7234.html#age.calculations
+  headers->addReferenceKey(Http::Headers::get().Age, 0);
+  decoder_callbacks_->encodeHeaders(std::move(headers), end_stream);
+  if (end_stream) {
+    return;
+  }
+  if (content_length > 0) {
+    remaining_body_.emplace_back(0, content_length);
+    getBody();
+  } else {
+    lookup_->getTrailers([self = shared_from_this()](Http::HeaderMapPtr&& trailers) {
+      onTrailersAsync(self, std::move(trailers));
+    });
+  }
+}
+
+void CacheFilter::onUnusableHeaders() {
+  if (lookup_) {
+    decoder_callbacks_->continueDecoding();
+  }
+}
+
+void CacheFilter::onHeadersAsync(const CacheFilterSharedPtr& self, LookupResult&& result) {
+  switch (result.cache_entry_status_) {
+  case CacheEntryStatus::RequiresValidation:
+  case CacheEntryStatus::FoundNotModified:
+  case CacheEntryStatus::UnsatisfiableRange:
+    ASSERT(false); // We don't yet return or support these codes.
+    FALLTHRU;
+  case CacheEntryStatus::Unusable: {
+    self->post([self] { self->onUnusableHeaders(); });
+    return;
+  }
+  case CacheEntryStatus::Ok:
+    self->post([self, headers = result.headers_.release(),
+                response_ranges = std::move(result.response_ranges_),
+                content_length = result.content_length_,
+                has_trailers = result.has_trailers_]() mutable {
+      self->onOkHeaders(absl::WrapUnique(headers), std::move(response_ranges), content_length,
+                        has_trailers);
+    });
+  }
+}
+
+void CacheFilter::getBody() {
+  ASSERT(!remaining_body_.empty(), "No reason to call getBody when there's no body to get.");
+  CacheFilterSharedPtr self = shared_from_this();
+  lookup_->getBody(remaining_body_[0],
+                   [self](Buffer::InstancePtr&& body) { self->onBody(std::move(body)); });
+}
+
+void CacheFilter::onBodyAsync(const CacheFilterSharedPtr& self, Buffer::InstancePtr&& body) {
+  self->post([self, body = body.release()] { self->onBody(absl::WrapUnique(body)); });
+}
+
+// TODO(toddmgreer): Handle downstream backpressure.
+void CacheFilter::onBody(Buffer::InstancePtr&& body) {
+  if (!lookup_) {
+    return;
+  }
+  if (remaining_body_.empty()) {
+    ASSERT(false, "CacheFilter doesn't call getBody unless there's more body to get, so this is a "
+                  "bogus callback.");
+    decoder_callbacks_->resetStream();
+    return;
+  }
+
+  if (!body) {
+    ASSERT(false, "Cache said it had a body, but isn't giving it to us.");
+    decoder_callbacks_->resetStream();
+    return;
+  }
+
+  const uint64_t bytes_from_cache = body->length();
+  if (bytes_from_cache < remaining_body_[0].length()) {
+    remaining_body_[0].trimFront(bytes_from_cache);
+  } else if (bytes_from_cache == remaining_body_[0].length()) {
+    remaining_body_.erase(remaining_body_.begin());
+  } else {
+    ASSERT(false, "Received oversized body from cache.");
+    decoder_callbacks_->resetStream();
+    return;
+  }
+
+  const bool end_stream = remaining_body_.empty() && !response_has_trailers_;
+  decoder_callbacks_->encodeData(*body, end_stream);
+  if (!remaining_body_.empty()) {
+    getBody();
+  } else if (response_has_trailers_) {
+    lookup_->getTrailers([self = shared_from_this()](Http::HeaderMapPtr&& trailers) {
+      onTrailersAsync(self, std::move(trailers));
+    });
+  }
+}
+
+void CacheFilter::onTrailers(Http::HeaderMapPtr&& trailers) {
+  if (lookup_) {
+    decoder_callbacks_->encodeTrailers(std::move(trailers));
+  }
+}
+
+void CacheFilter::onTrailersAsync(const CacheFilterSharedPtr& self, Http::HeaderMapPtr&& trailers) {
+  self->post(
+      [self, trailers = trailers.release()] { self->onTrailers(absl::WrapUnique(trailers)); });
+}
+
+void CacheFilter::post(std::function<void()> f) const {
+  decoder_callbacks_->dispatcher().post(std::move(f));
+}
+} // namespace Cache
+} // namespace HttpFilters
+} // namespace Extensions
+} // namespace Envoy

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

@@ -0,0 +1,88 @@
+#pragma once
+
+#include <functional>
+#include <memory>
+#include <string>
+#include <vector>
+
+#include "envoy/extensions/filters/http/cache/v3alpha/cache.pb.h"
+
+#include "common/common/logger.h"
+
+#include "extensions/filters/http/cache/http_cache.h"
+#include "extensions/filters/http/common/pass_through_filter.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace HttpFilters {
+namespace Cache {
+
+/**
+ * A filter that caches responses and attempts to satisfy requests from cache.
+ * It also inherits from std::enable_shared_from_this so it can pass shared_ptrs to async methods,
+ * to ensure that it doesn't get destroyed before they complete.
+ */
+class CacheFilter;
+using CacheFilterSharedPtr = std::shared_ptr<CacheFilter>;
+class CacheFilter : public Http::PassThroughFilter,
+                    public Logger::Loggable<Logger::Id::cache_filter>,
+                    public std::enable_shared_from_this<CacheFilter> {
+public:
+  // Throws EnvoyException if no registered HttpCacheFactory for config.typed_config.
+  static CacheFilterSharedPtr
+  make(const envoy::extensions::filters::http::cache::v3alpha::CacheConfig& config,
+       const std::string& stats_prefix, Stats::Scope& scope, TimeSource& time_source) {
+    // Can't use make_shared due to private constructor.
+    return std::shared_ptr<CacheFilter>(new CacheFilter(config, stats_prefix, scope, time_source));
+  }
+  // Http::StreamFilterBase
+  void onDestroy() override;
+  // Http::StreamDecoderFilter
+  Http::FilterHeadersStatus decodeHeaders(Http::HeaderMap& headers, bool end_stream) override;
+  // Http::StreamEncoderFilter
+  Http::FilterHeadersStatus encodeHeaders(Http::HeaderMap& headers, bool end_stream) override;
+  Http::FilterDataStatus encodeData(Buffer::Instance& buffer, bool end_stream) override;
+
+private:
+  // Throws EnvoyException if no registered HttpCacheFactory for config.typed_config.
+  // Constructor is private to enforce enable_shared_from_this's requirement that this must be owned
+  // by a shared_ptr.
+  CacheFilter(const envoy::extensions::filters::http::cache::v3alpha::CacheConfig& config,
+              const std::string& stats_prefix, Stats::Scope& scope, TimeSource& time_source);
+
+  void getBody();
+  void onOkHeaders(Http::HeaderMapPtr&& headers, std::vector<AdjustedByteRange>&& response_ranges,
+                   uint64_t content_length, bool has_trailers);
+  void onUnusableHeaders();
+  void onBody(Buffer::InstancePtr&& body);
+  void onTrailers(Http::HeaderMapPtr&& trailers);
+  static void onHeadersAsync(const CacheFilterSharedPtr& self, LookupResult&& result);
+  static void onBodyAsync(const CacheFilterSharedPtr& self, Buffer::InstancePtr&& body);
+  static void onTrailersAsync(const CacheFilterSharedPtr& self, Http::HeaderMapPtr&& trailers);
+  void post(std::function<void()> f) const;
+
+  // These don't require private access, but are members per envoy convention.
+  static bool isCacheableRequest(Http::HeaderMap& headers);
+  static bool isCacheableResponse(Http::HeaderMap& headers);
+  static HttpCache&
+  getCache(const envoy::extensions::filters::http::cache::v3alpha::CacheConfig& config);
+
+  TimeSource& time_source_;
+  HttpCache& cache_;
+  LookupContextPtr lookup_;
+  InsertContextPtr insert_;
+
+  // Tracks what body bytes still need to be read from the cache. This is
+  // currently only one Range, but will expand when full range support is added. Initialized by
+  // onOkHeaders.
+  std::vector<AdjustedByteRange> remaining_body_;
+
+  // True if the response has trailers.
+  // TODO(toddmgreer): cache trailers.
+  bool response_has_trailers_;
+};
+
+} // namespace Cache
+} // namespace HttpFilters
+} // namespace Extensions
+} // namespace Envoy

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

@@ -90,7 +90,7 @@ class AdjustedByteRange {
 
 private:
   uint64_t first_;
-  const uint64_t last_;
+  uint64_t last_;
 };
 
 inline bool operator==(const AdjustedByteRange& lhs, const AdjustedByteRange& rhs) {

source/extensions/filters/http/well_known_names.h

@@ -14,6 +14,8 @@ class HttpFilterNameValues {
 public:
   // Buffer filter
   const std::string Buffer = "envoy.buffer";
+  // Cache filter
+  const std::string Cache = "envoy.filters.http.cache";
   // CORS filter
   const std::string Cors = "envoy.cors";
   // CSRF filter

test/extensions/filters/http/cache/BUILD

@@ -31,3 +31,17 @@ envoy_extension_cc_test(
         "//test/test_common:utility_lib",
     ],
 )
+
+envoy_extension_cc_test(
+    name = "cache_filter_test",
+    srcs = ["cache_filter_test.cc"],
+    extension_name = "envoy.filters.http.cache",
+    deps = [
+        "//source/extensions/filters/http/cache:cache_filter_lib",
+        "//source/extensions/filters/http/cache/simple_http_cache:config_cc_proto",
+        "//source/extensions/filters/http/cache/simple_http_cache:simple_http_cache_lib",
+        "//test/mocks/server:server_mocks",
+        "//test/test_common:simulated_time_system_lib",
+        "//test/test_common:utility_lib",
+    ],
+)