stream_info: Add the concept of "life span" to StreamInfo::FilterState

include/envoy/stream_info/filter_state.h

@@ -15,13 +15,36 @@ namespace Envoy {
 namespace StreamInfo {
 
 /**
- * FilterState represents dynamically generated information regarding a
- * stream (TCP or HTTP level) by various filters in Envoy. FilterState can
- * be write-once or write-many.
+ * FilterState represents dynamically generated information regarding a stream (TCP or HTTP level)
+ * or a connection by various filters in Envoy. FilterState can be write-once or write-many.
  */
 class FilterState {
 public:
   enum class StateType { ReadOnly, Mutable };
+  // Objects stored in the FilterState may have different life span. Life span is what controls
+  // how long an object stored in FilterState lives. Implementation of this interface actually
+  // stores objects in a (reverse) tree manner - multiple FilterStateImpl with shorter life span may
+  // share the same FilterStateImpl as parent, which may recursively share parent with other
+  // FilterStateImpl at the same life span. This interface is supposed to be accessed at the leaf
+  // level (FilterChain) for objects with any desired longer life span.
+  //
+  // - FilterChain has the shortest life span, which is as long as the filter chain lives.
+  //
+  // - DownstreamRequest is longer than FilterChain. When internal redirect is enabled, one
+  //   downstream request may create multiple filter chains. DownstreamRequest allows an object to
+  //   survive across filter chains for bookkeeping needs.
+  //
+  // - DownstreamConnection makes an object survive the entire duration of a downstream connection.
+  //   Any stream within this connection can see the same object.
+  //
+  // Note that order matters in this enum because it's assumed that life span grows as enum number
+  // grows.
+  enum LifeSpan {
+    FilterChain,
+    DownstreamRequest,
+    DownstreamConnection,
+    TopSpan = DownstreamConnection
+  };
 
   class Object {
   public:
@@ -41,16 +64,18 @@ class FilterState {
    * @param data_name the name of the data being set.
    * @param data an owning pointer to the data to be stored.
    * @param state_type indicates whether the object is mutable or not.
+   * @param life_span indicates the life span of the object: bound to the filter chain, a
+   * downstream request, or a downstream connection.
    *
    * Note that it is an error to call setData() twice with the same
    * data_name, if the existing object is immutable. Similarly, it is an
    * error to call setData() with same data_name but different state_types
-   * (mutable and readOnly, or readOnly and mutable). This is to enforce a
-   * single authoritative source for each piece of immutable data stored in
-   * FilterState.
+   * (mutable and readOnly, or readOnly and mutable) or different life_span.
+   * This is to enforce a single authoritative source for each piece of
+   * data stored in FilterState.
    */
-  virtual void setData(absl::string_view data_name, std::unique_ptr<Object>&& data,
-                       StateType state_type) PURE;
+  virtual void setData(absl::string_view data_name, std::shared_ptr<Object> data,
+                       StateType state_type, LifeSpan life_span = LifeSpan::FilterChain) PURE;
 
   /**
    * @param data_name the name of the data being looked up (mutable/readonly).
@@ -102,6 +127,24 @@ class FilterState {
    */
   virtual bool hasDataWithName(absl::string_view data_name) const PURE;
 
+  /**
+   * @param life_span the LifeSpan above which data existence is checked.
+   * @return whether data of any type exist with LifeSpan greater than life_span.
+   */
+  virtual bool hasDataAtOrAboveLifeSpan(LifeSpan life_span) const PURE;
+
+  /**
+   * @return the LifeSpan of objects stored by this instance. Objects with
+   * LifeSpan longer than this are handled recursively.
+   */
+  virtual LifeSpan lifeSpan() const PURE;
+
+  /**
+   * @return the pointer of the parent FilterState that has longer life span. nullptr means this is
+   * either the top LifeSpan or the parent is not yet created.
+   */
+  virtual std::shared_ptr<FilterState> parent() const PURE;
+
 protected:
   virtual const Object* getDataReadOnlyGeneric(absl::string_view data_name) const PURE;
   virtual Object* getDataMutableGeneric(absl::string_view data_name) PURE;

source/common/http/conn_manager_impl.cc

@@ -14,6 +14,7 @@
 #include "envoy/router/router.h"
 #include "envoy/ssl/connection.h"
 #include "envoy/stats/scope.h"
+#include "envoy/stream_info/filter_state.h"
 #include "envoy/tracing/http_tracer.h"
 
 #include "common/buffer/buffer_impl.h"
@@ -485,7 +486,8 @@ ConnectionManagerImpl::ActiveStream::ActiveStream(ConnectionManagerImpl& connect
       stream_id_(connection_manager.random_generator_.random()),
       request_response_timespan_(new Stats::HistogramCompletableTimespanImpl(
           connection_manager_.stats_.named_.downstream_rq_time_, connection_manager_.timeSource())),
-      stream_info_(connection_manager_.codec_->protocol(), connection_manager_.timeSource()),
+      stream_info_(connection_manager_.codec_->protocol(), connection_manager_.timeSource(),
+                   connection_manager.filterState()),
       upstream_options_(std::make_shared<Network::Socket::Options>()) {
   ASSERT(!connection_manager.config_.isRoutable() ||
              ((connection_manager.config_.routeConfigProvider() == nullptr &&
@@ -2244,6 +2246,17 @@ bool ConnectionManagerImpl::ActiveStreamDecoderFilter::recreateStream() {
   parent_.connection_manager_.doEndStream(this->parent_);
 
   StreamDecoder& new_stream = parent_.connection_manager_.newStream(*response_encoder, true);
+  // We don't need to copy over the old parent FilterState from the old StreamInfo if it did not
+  // store any objects with a LifeSpan at or above DownstreamRequest. This is to avoid unnecessary
+  // heap allocation.
+  if (parent_.stream_info_.filter_state_->hasDataAtOrAboveLifeSpan(
+          StreamInfo::FilterState::LifeSpan::DownstreamRequest)) {
+    (*parent_.connection_manager_.streams_.begin())->stream_info_.filter_state_ =
+        std::make_shared<StreamInfo::FilterStateImpl>(
+            parent_.stream_info_.filter_state_->parent(),
+            StreamInfo::FilterState::LifeSpan::FilterChain);
+  }
+
   new_stream.decodeHeaders(std::move(request_headers), true);
   return true;
 }

source/common/http/conn_manager_impl.h

@@ -90,6 +90,9 @@ class ConnectionManagerImpl : Logger::Loggable<Logger::Id::http>,
 
   TimeSource& timeSource() { return time_source_; }
 
+  // Return a reference to the shared_ptr so that it can be lazy created on demand.
+  std::shared_ptr<StreamInfo::FilterState>& filterState() { return filter_state_; }
+
 private:
   struct ActiveStream;
 
@@ -701,6 +704,7 @@ class ConnectionManagerImpl : Logger::Loggable<Logger::Id::http>,
   const Server::OverloadActionState& overload_stop_accepting_requests_ref_;
   const Server::OverloadActionState& overload_disable_keepalive_ref_;
   TimeSource& time_source_;
+  std::shared_ptr<StreamInfo::FilterState> filter_state_;
 };
 
 } // namespace Http

source/common/stream_info/filter_state_impl.cc

@@ -5,8 +5,21 @@
 namespace Envoy {
 namespace StreamInfo {
 
-void FilterStateImpl::setData(absl::string_view data_name, std::unique_ptr<Object>&& data,
-                              FilterState::StateType state_type) {
+void FilterStateImpl::setData(absl::string_view data_name, std::shared_ptr<Object> data,
+                              FilterState::StateType state_type, FilterState::LifeSpan life_span) {
+  if (life_span > life_span_) {
+    if (hasDataWithNameInternally(data_name)) {
+      throw EnvoyException(
+          "FilterState::setData<T> called twice with conflicting life_span on the same data_name.");
+    }
+    maybeCreateParent(ParentAccessMode::ReadWrite);
+    parent_->setData(data_name, data, state_type, life_span);
+    return;
+  }
+  if (parent_ && parent_->hasDataWithName(data_name)) {
+    throw EnvoyException(
+        "FilterState::setData<T> called twice with conflicting life_span on the same data_name.");
+  }
   const auto& it = data_storage_.find(data_name);
   if (it != data_storage_.end()) {
     // We have another object with same data_name. Check for mutability
@@ -23,20 +36,23 @@ void FilterStateImpl::setData(absl::string_view data_name, std::unique_ptr<Objec
   }
 
   std::unique_ptr<FilterStateImpl::FilterObject> filter_object(new FilterStateImpl::FilterObject());
-  filter_object->data_ = std::move(data);
+  filter_object->data_ = data;
   filter_object->state_type_ = state_type;
   data_storage_[data_name] = std::move(filter_object);
 }
 
 bool FilterStateImpl::hasDataWithName(absl::string_view data_name) const {
-  return data_storage_.count(data_name) > 0;
+  return hasDataWithNameInternally(data_name) || (parent_ && parent_->hasDataWithName(data_name));
 }
 
 const FilterState::Object*
 FilterStateImpl::getDataReadOnlyGeneric(absl::string_view data_name) const {
   const auto& it = data_storage_.find(data_name);
 
   if (it == data_storage_.end()) {
+    if (parent_) {
+      return &(parent_->getDataReadOnly<FilterState::Object>(data_name));
+    }
     throw EnvoyException("FilterState::getDataReadOnly<T> called for unknown data name.");
   }
 
@@ -48,6 +64,9 @@ FilterState::Object* FilterStateImpl::getDataMutableGeneric(absl::string_view da
   const auto& it = data_storage_.find(data_name);
 
   if (it == data_storage_.end()) {
+    if (parent_) {
+      return &(parent_->getDataMutable<FilterState::Object>(data_name));
+    }
     throw EnvoyException("FilterState::getDataMutable<T> called for unknown data name.");
   }
 
@@ -60,5 +79,54 @@ FilterState::Object* FilterStateImpl::getDataMutableGeneric(absl::string_view da
   return current->data_.get();
 }
 
+bool FilterStateImpl::hasDataAtOrAboveLifeSpan(FilterState::LifeSpan life_span) const {
+  if (life_span > life_span_) {
+    return parent_ && parent_->hasDataAtOrAboveLifeSpan(life_span);
+  }
+  return !data_storage_.empty() || (parent_ && parent_->hasDataAtOrAboveLifeSpan(life_span));
+}
+
+bool FilterStateImpl::hasDataWithNameInternally(absl::string_view data_name) const {
+  return data_storage_.count(data_name) > 0;
+}
+
+void FilterStateImpl::maybeCreateParent(ParentAccessMode parent_access_mode) {
+  if (parent_ != nullptr) {
+    return;
+  }
+  if (life_span_ >= FilterState::LifeSpan::TopSpan) {
+    return;
+  }
+  if (absl::holds_alternative<std::shared_ptr<FilterState>>(ancestor_)) {
+    std::shared_ptr<FilterState> ancestor = absl::get<std::shared_ptr<FilterState>>(ancestor_);
+    if (ancestor == nullptr || ancestor->lifeSpan() != life_span_ + 1) {
+      parent_ = std::make_shared<FilterStateImpl>(ancestor, FilterState::LifeSpan(life_span_ + 1));
+    } else {
+      parent_ = ancestor;
+    }
+    return;
+  }
+
+  auto lazy_create_ancestor = absl::get<LazyCreateAncestor>(ancestor_);
+  // If we're only going to read data from our parent, we don't need to create lazy ancestor,
+  // because they're empty anyways.
+  if (parent_access_mode == ParentAccessMode::ReadOnly && lazy_create_ancestor.first == nullptr) {
+    return;
+  }
+
+  // Lazy ancestor is not our immediate parent.
+  if (lazy_create_ancestor.second != life_span_ + 1) {
+    parent_ = std::make_shared<FilterStateImpl>(lazy_create_ancestor,
+                                                FilterState::LifeSpan(life_span_ + 1));
+    return;
+  }
+  // Lazy parent is our immediate parent.
+  if (lazy_create_ancestor.first == nullptr) {
+    lazy_create_ancestor.first =
+        std::make_shared<FilterStateImpl>(FilterState::LifeSpan(life_span_ + 1));
+  }
+  parent_ = lazy_create_ancestor.first;
+}
+
 } // namespace StreamInfo
 } // namespace Envoy

source/common/stream_info/filter_state_impl.h

@@ -1,6 +1,7 @@
 #pragma once
 
 #include <memory>
+#include <utility>
 #include <vector>
 
 #include "envoy/stream_info/filter_state.h"
@@ -13,19 +14,57 @@ namespace StreamInfo {
 
 class FilterStateImpl : public FilterState {
 public:
+  FilterStateImpl(FilterState::LifeSpan life_span) : life_span_(life_span) {
+    maybeCreateParent(ParentAccessMode::ReadOnly);
+  }
+
+  /**
+   * @param ancestor a std::shared_ptr storing an already created ancestor.
+   * @param life_span the life span this is handling.
+   */
+  FilterStateImpl(std::shared_ptr<FilterState> ancestor, FilterState::LifeSpan life_span)
+      : ancestor_(ancestor), life_span_(life_span) {
+    maybeCreateParent(ParentAccessMode::ReadOnly);
+  }
+
+  using LazyCreateAncestor = std::pair<std::shared_ptr<FilterState>&, FilterState::LifeSpan>;
+  /**
+   * @param ancestor a std::pair storing an ancestor, that can be passed in as a way to lazy
+   * initialize a FilterState that's owned by an object with bigger scope than this. This is to
+   * avoid creating a FilterState that's empty in most cases.
+   * @param life_span the life span this is handling.
+   */
+  FilterStateImpl(LazyCreateAncestor lazy_create_ancestor, FilterState::LifeSpan life_span)
+      : ancestor_(lazy_create_ancestor), life_span_(life_span) {
+    maybeCreateParent(ParentAccessMode::ReadOnly);
+  }
+
   // FilterState
-  void setData(absl::string_view data_name, std::unique_ptr<Object>&& data,
-               FilterState::StateType state_type) override;
+  void setData(absl::string_view data_name, std::shared_ptr<Object> data,
+               FilterState::StateType state_type,
+               FilterState::LifeSpan life_span = FilterState::LifeSpan::FilterChain) override;
   bool hasDataWithName(absl::string_view) const override;
   const Object* getDataReadOnlyGeneric(absl::string_view data_name) const override;
   Object* getDataMutableGeneric(absl::string_view data_name) override;
+  bool hasDataAtOrAboveLifeSpan(FilterState::LifeSpan life_span) const override;
+
+  FilterState::LifeSpan lifeSpan() const override { return life_span_; }
+  std::shared_ptr<FilterState> parent() const override { return parent_; }
 
 private:
+  // This only checks the local data_storage_ for data_name existence.
+  bool hasDataWithNameInternally(absl::string_view data_name) const;
+  enum class ParentAccessMode { ReadOnly, ReadWrite };
+  void maybeCreateParent(ParentAccessMode parent_access_mode);
+
   struct FilterObject {
-    std::unique_ptr<Object> data_;
+    std::shared_ptr<Object> data_;
     FilterState::StateType state_type_;
   };
 
+  absl::variant<std::shared_ptr<FilterState>, LazyCreateAncestor> ancestor_;
+  std::shared_ptr<FilterState> parent_;
+  const FilterState::LifeSpan life_span_;
   absl::flat_hash_map<std::string, std::unique_ptr<FilterObject>> data_storage_;
 };
 

source/common/stream_info/stream_info_impl.h

@@ -15,13 +15,24 @@ namespace Envoy {
 namespace StreamInfo {
 
 struct StreamInfoImpl : public StreamInfo {
-  explicit StreamInfoImpl(TimeSource& time_source)
+  StreamInfoImpl(TimeSource& time_source)
       : time_source_(time_source), start_time_(time_source.systemTime()),
-        start_time_monotonic_(time_source.monotonicTime()) {}
+        start_time_monotonic_(time_source.monotonicTime()),
+        filter_state_(std::make_shared<FilterStateImpl>(FilterState::LifeSpan::FilterChain)) {}
 
-  StreamInfoImpl(Http::Protocol protocol, TimeSource& time_source) : StreamInfoImpl(time_source) {
-    protocol_ = protocol;
-  }
+  StreamInfoImpl(Http::Protocol protocol, TimeSource& time_source)
+      : time_source_(time_source), start_time_(time_source.systemTime()),
+        start_time_monotonic_(time_source.monotonicTime()), protocol_(protocol),
+        filter_state_(std::make_shared<FilterStateImpl>(FilterState::LifeSpan::FilterChain)) {}
+
+  StreamInfoImpl(Http::Protocol protocol, TimeSource& time_source,
+                 std::shared_ptr<FilterState>& parent_filter_state)
+      : time_source_(time_source), start_time_(time_source.systemTime()),
+        start_time_monotonic_(time_source.monotonicTime()), protocol_(protocol),
+        filter_state_(std::make_shared<FilterStateImpl>(
+            FilterStateImpl::LazyCreateAncestor(parent_filter_state,
+                                                FilterState::LifeSpan::DownstreamConnection),
+            FilterState::LifeSpan::FilterChain)) {}
 
   SystemTime startTime() const override { return start_time_; }
 
@@ -204,8 +215,8 @@ struct StreamInfoImpl : public StreamInfo {
     (*metadata_.mutable_filter_metadata())[name].MergeFrom(value);
   };
 
-  FilterState& filterState() override { return filter_state_; }
-  const FilterState& filterState() const override { return filter_state_; }
+  FilterState& filterState() override { return *filter_state_; }
+  const FilterState& filterState() const override { return *filter_state_; }
 
   void setRequestedServerName(absl::string_view requested_server_name) override {
     requested_server_name_ = std::string(requested_server_name);
@@ -249,7 +260,7 @@ struct StreamInfoImpl : public StreamInfo {
   bool health_check_request_{};
   const Router::RouteEntry* route_entry_{};
   envoy::api::v2::core::Metadata metadata_{};
-  FilterStateImpl filter_state_{};
+  std::shared_ptr<FilterStateImpl> filter_state_;
   std::string route_name_;
 
 private:

source/extensions/filters/http/grpc_stats/grpc_stats_filter.cc

@@ -81,7 +81,8 @@ class GrpcStatsFilter : public Http::PassThroughFilter {
       filter_object_ = state.get();
       decoder_callbacks_->streamInfo().filterState().setData(
           HttpFilterNames::get().GrpcStats, std::move(state),
-          StreamInfo::FilterState::StateType::Mutable);
+          StreamInfo::FilterState::StateType::Mutable,
+          StreamInfo::FilterState::LifeSpan::FilterChain);
     }
     filter_object_->request_message_count = request_counter_.frameCount();
     filter_object_->response_message_count = response_counter_.frameCount();

source/extensions/filters/network/sni_cluster/sni_cluster.cc

@@ -21,7 +21,8 @@ Network::FilterStatus SniClusterFilter::onNewConnection() {
     read_callbacks_->connection().streamInfo().filterState().setData(
         TcpProxy::PerConnectionCluster::key(),
         std::make_unique<TcpProxy::PerConnectionCluster>(sni),
-        StreamInfo::FilterState::StateType::Mutable);
+        StreamInfo::FilterState::StateType::Mutable,
+        StreamInfo::FilterState::LifeSpan::DownstreamConnection);
   }
 
   return Network::FilterStatus::Continue;