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,20 @@ load(
 
 envoy_package()
 
+envoy_cc_library(
+    name = "cache_filter_lib",
+    srcs = ["cache_filter.cc"],
+    hdrs = ["cache_filter.h"],
+    deps = [
+        ":http_cache_lib",
+        "//source/common/common:logger_lib",
+        "//source/common/common:macros",
+        "//source/common/http:headers_lib",
+        "//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,192 @@
+#include "extensions/filters/http/cache/cache_filter.h"
+
+#include "common/http/headers.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;
+}
+
+CacheFilter::CacheFilter(const envoy::extensions::filters::http::cache::v3alpha::CacheConfig&,
+                         const std::string&, Stats::Scope&, TimeSource& time_source,
+                         HttpCache& http_cache)
+    : time_source_(time_source), cache_(http_cache) {}
+
+void CacheFilter::onDestroy() {
+  // Clear decoder_callbacks_ so any pending callbacks will see that this filter is no longer
+  // active().
+  decoder_callbacks_ = nullptr;
+  if (lookup_) {
+    lookup_->onDestroy();
+  }
+  if (insert_) {
+    insert_->onDestroy();
+  }
+}
+
+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_);
+
+  ENVOY_STREAM_LOG(debug, "CacheFilter::decodeHeaders starting lookup", *decoder_callbacks_);
+  lookup_->getHeaders([this](LookupResult&& result) { onHeadersAsync(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::onHeaders(LookupResult&& result) {
+  if (!active()) {
+    return;
+  }
+  switch (result.cache_entry_status_) {
+  case CacheEntryStatus::RequiresValidation:
+  case CacheEntryStatus::FoundNotModified:
+  case CacheEntryStatus::UnsatisfiableRange:
+    NOT_IMPLEMENTED_GCOVR_EXCL_LINE; // We don't yet return or support these codes.
+  case CacheEntryStatus::Unusable:
+    decoder_callbacks_->continueDecoding();
+    return;
+  case CacheEntryStatus::Ok:
+    response_has_trailers_ = result.has_trailers_;
+    const bool end_stream = (result.content_length_ == 0 && !response_has_trailers_);
+    // TODO(toddmgreer): Calculate age per https://httpwg.org/specs/rfc7234.html#age.calculations
+    result.headers_->addReferenceKey(Http::Headers::get().Age, 0);
+    decoder_callbacks_->encodeHeaders(std::move(result.headers_), end_stream);
+    if (end_stream) {
+      return;
+    }
+    if (result.content_length_ > 0) {
+      remaining_body_.emplace_back(0, result.content_length_);
+      getBody();
+    } else {
+      lookup_->getTrailers(
+          [this](Http::HeaderMapPtr&& trailers) { onTrailersAsync(std::move(trailers)); });
+    }
+  }
+}
+
+void CacheFilter::onHeadersAsync(LookupResult&& result) {
+  post([this, status = result.cache_entry_status_, headers = result.headers_.release(),
+        response_ranges = std::move(result.response_ranges_),
+        content_length = result.content_length_, has_trailers = result.has_trailers_] {
+    onHeaders(LookupResult{status, absl::WrapUnique(headers), content_length, response_ranges,
+                           has_trailers});
+  });
+}
+
+void CacheFilter::getBody() {
+  ASSERT(!remaining_body_.empty(), "No reason to call getBody when there's no body to get.");
+  lookup_->getBody(remaining_body_[0],
+                   [this](Buffer::InstancePtr&& body) { onBody(std::move(body)); });
+}
+
+void CacheFilter::onBodyAsync(Buffer::InstancePtr&& body) {
+  post([this, body = body.release()] { onBody(absl::WrapUnique(body)); });
+}
+
+// TODO(toddmgreer): Handle downstream backpressure.
+void CacheFilter::onBody(Buffer::InstancePtr&& body) {
+  if (!active()) {
+    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(
+        [this](Http::HeaderMapPtr&& trailers) { onTrailersAsync(std::move(trailers)); });
+  }
+}
+
+void CacheFilter::onTrailers(Http::HeaderMapPtr&& trailers) {
+  if (active()) {
+    decoder_callbacks_->encodeTrailers(std::move(trailers));
+  }
+}
+
+void CacheFilter::onTrailersAsync(Http::HeaderMapPtr&& trailers) {
+  post([this, trailers = trailers.release()] { 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,72 @@
+#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.
+ */
+class CacheFilter : public Http::PassThroughFilter,
+                    public Logger::Loggable<Logger::Id::cache_filter> {
+public:
+  CacheFilter(const envoy::extensions::filters::http::cache::v3alpha::CacheConfig& config,
+              const std::string& stats_prefix, Stats::Scope& scope, TimeSource& time_source,
+              HttpCache& http_cache);
+  // 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:
+  void getBody();
+  void post(std::function<void()> f) const;
+  bool active() const { return decoder_callbacks_; }
+
+  void onHeaders(LookupResult&& result);
+  void onHeadersAsync(LookupResult&& result);
+  void onBody(Buffer::InstancePtr&& body);
+  void onBodyAsync(Buffer::InstancePtr&& body);
+  void onTrailers(Http::HeaderMapPtr&& trailers);
+  void onTrailersAsync(Http::HeaderMapPtr&& trailers);
+
+  // These don't require private access, but are members per envoy convention.
+  static bool isCacheableRequest(Http::HeaderMap& headers);
+  static bool isCacheableResponse(Http::HeaderMap& headers);
+
+  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_;
+};
+using CacheFilterPtr = std::unique_ptr<CacheFilter>;
+
+} // 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) {
@@ -168,7 +168,7 @@ class LookupRequest {
   Key& key() { return key_; }
 
   // Returns the subset of this request's headers that are listed in
-  // envoy::config::filter::http::cache::v3::CacheConfig::allowed_vary_headers. If a cache
+  // envoy::extensions::filters::http::cache::v3alpha::CacheConfig::allowed_vary_headers. If a cache
   // storage implementation forwards lookup requests to a remote cache server that supports *vary*
   // headers, that server may need to see these headers. For local implementations, it may be
   // simpler to instead call makeLookupResult with each potential response.
@@ -231,6 +231,15 @@ class InsertContext {
   // Inserts trailers into the cache.
   virtual void insertTrailers(const Http::HeaderMap& trailers) PURE;
 
+  /**
+   * This routine is called prior to an InsertContext being destroyed. InsertContext is responsible
+   * for making sure that any async events are cleaned up in the context of this routine. This
+   * includes timers, network calls, etc. The reason there is an onDestroy() method vs. doing this
+   * type of cleanup in the destructor is due to the deferred deletion model that Envoy uses to
+   * avoid stack unwind complications. InsertContext must not invoke any callbacks after having
+   * onDestroy() invoked.
+   */
+  virtual void onDestroy() PURE;
   virtual ~InsertContext() = default;
 };
 using InsertContextPtr = std::unique_ptr<InsertContext>;
@@ -240,8 +249,6 @@ using InsertContextPtr = std::unique_ptr<InsertContext>;
 // an in-progress lookup by simply dropping the LookupContextPtr.
 class LookupContext {
 public:
-  virtual ~LookupContext() = default;
-
   // Get the headers from the cache. It is a programming error to call this
   // twice.
   virtual void getHeaders(LookupHeadersCallback&& cb) PURE;
@@ -267,6 +274,17 @@ class LookupContext {
   // Get the trailers from the cache. Only called if LookupResult::has_trailers
   // == true.
   virtual void getTrailers(LookupTrailersCallback&& cb) PURE;
+
+  /**
+   * This routine is called prior to an LookupContext being destroyed. LookupContext is responsible
+   * for making sure that any async events are cleaned up in the context of this routine. This
+   * includes timers, network calls, etc. The reason there is an onDestroy() method vs. doing this
+   * type of cleanup in the destructor is due to the deferred deletion model that Envoy uses to
+   * avoid stack unwind complications. LookupContext must not invoke any callbacks after having
+   * onDestroy() invoked.
+   */
+  virtual void onDestroy() PURE;
+  virtual ~LookupContext() = default;
 };
 using LookupContextPtr = std::unique_ptr<LookupContext>;
 

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

@@ -37,6 +37,7 @@ class SimpleLookupContext : public LookupContext {
   }
 
   const LookupRequest& request() const { return request_; }
+  void onDestroy() override {}
 
 private:
   SimpleHttpCache& cache_;
@@ -74,6 +75,8 @@ class SimpleInsertContext : public InsertContext {
     NOT_IMPLEMENTED_GCOVR_EXCL_LINE; // TODO(toddmgreer): support trailers
   }
 
+  void onDestroy() override {}
+
 private:
   void commit() {
     committed_ = true;

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