stream_info: reworking upstream data

envoy/stream_info/stream_info.h

@@ -293,6 +293,88 @@ struct BytesMeter {
 
 using BytesMeterSharedPtr = std::shared_ptr<BytesMeter>;
 
+// TODO(alyssawilk) after landing this, remove all the duplicate getters and
+// setters from StreamInfo.
+class UpstreamInfo {
+ public:
+  virtual ~UpstreamInfo() = default;
+
+  /**
+   * @param connection ID of the upstream connection.
+   */
+  virtual void setUpstreamConnectionId(uint64_t id) PURE;
+
+  /**
+   * @return the ID of the upstream connection, or absl::nullopt if not available.
+   */
+  virtual absl::optional<uint64_t> upstreamConnectionId() const PURE;
+
+  /**
+   * @param connection_info sets the upstream ssl connection.
+   */
+  virtual void
+  setUpstreamSslConnection(const Ssl::ConnectionInfoConstSharedPtr& ssl_connection_info) PURE;
+
+  /**
+   * @return the upstream SSL connection. This will be nullptr if the upstream
+   * connection does not use SSL.
+   */
+  virtual Ssl::ConnectionInfoConstSharedPtr upstreamSslConnection() const PURE;
+
+  /**
+   * Sets the upstream timing information for this stream. This is useful for
+   * when multiple upstream requests are issued and we want to save timing
+   * information for the one that "wins".
+   */
+  virtual void setUpstreamTiming(const UpstreamTiming& upstream_timing) PURE;
+
+  /*
+   * @return the upstream timing for this stream
+   * */
+  virtual const UpstreamTiming& upstreamTiming() PURE;
+
+  /**
+   * @param upstream_local_address sets the local address of the upstream connection. Note that it
+   * can be different than the local address of the downstream connection.
+   */
+  virtual void setUpstreamLocalAddress(
+      const Network::Address::InstanceConstSharedPtr& upstream_local_address) PURE;
+
+  /**
+   * @return the upstream local address.
+   */
+  virtual const Network::Address::InstanceConstSharedPtr& upstreamLocalAddress() const PURE;
+
+  /**
+   * @param failure_reason the upstream transport failure reason.
+   */
+  virtual void setUpstreamTransportFailureReason(absl::string_view failure_reason) PURE;
+
+  /**
+   * @return const std::string& the upstream transport failure reason, e.g. certificate validation
+   *         failed.
+   */
+  virtual const std::string& upstreamTransportFailureReason() const PURE;
+
+  /**
+   * @param host the selected upstream host for the request.
+   */
+  virtual void setUpstreamHost(Upstream::HostDescriptionConstSharedPtr host) PURE;
+
+  /**
+   * @return upstream host description.
+   */
+  virtual Upstream::HostDescriptionConstSharedPtr upstreamHost() const PURE;
+
+  /**
+   * Filter State object to be shared between upstream and downstream filters.
+   * @param pointer to upstream connections filter state.
+   * @return pointer to filter state to be used by upstream connections.
+   */
+  virtual const FilterStateSharedPtr& upstreamFilterState() const PURE;
+  virtual void setUpstreamFilterState(const FilterStateSharedPtr& filter_state) PURE;
+};
+
 /**
  * Additional information about a completed request for logging.
  */
@@ -405,6 +487,9 @@ class StreamInfo {
    */
   virtual void setUpstreamTiming(const UpstreamTiming& upstream_timing) PURE;
 
+  virtual void setUpstreamInfo(std::shared_ptr<UpstreamInfo>) PURE;
+  virtual std::shared_ptr<UpstreamInfo> upstreamInfo() PURE;
+
   /**
    * @return the duration between the first byte of the request was sent upstream and the start of
    * the request. There may be a considerable delta between lastDownstreamByteReceived and this

source/common/stream_info/stream_info_impl.h

@@ -34,6 +34,64 @@ const ReplacementMap& emptySpaceReplacement() {
 
 } // namespace
 
+
+struct UpstreamInfoImpl : public UpstreamInfo {
+  void setUpstreamConnectionId(uint64_t id) override {
+    upstream_connection_id_ = id;
+  }
+
+  absl::optional<uint64_t> upstreamConnectionId() const override {
+    return upstream_connection_id_;
+  }
+
+  void
+  setUpstreamSslConnection(const Ssl::ConnectionInfoConstSharedPtr& ssl_connection_info) override {
+    upstream_ssl_info_ = ssl_connection_info;
+  }
+
+  Ssl::ConnectionInfoConstSharedPtr upstreamSslConnection() const override {
+    return upstream_ssl_info_;
+  }
+  void setUpstreamTiming(const UpstreamTiming& upstream_timing) override {
+    upstream_timing_ = upstream_timing;
+  }
+  const UpstreamTiming& upstreamTiming() override {
+    return upstream_timing_;
+  }
+  const Network::Address::InstanceConstSharedPtr& upstreamLocalAddress() const override {
+    return upstream_local_address_;
+  }
+  void setUpstreamLocalAddress(
+      const Network::Address::InstanceConstSharedPtr& upstream_local_address) override {
+    upstream_local_address_ = upstream_local_address;
+  }
+  void setUpstreamTransportFailureReason(absl::string_view failure_reason) override {
+    upstream_transport_failure_reason_ = std::string(failure_reason);
+  }
+  const std::string& upstreamTransportFailureReason() const override {
+    return upstream_transport_failure_reason_;
+  }
+  void setUpstreamHost(Upstream::HostDescriptionConstSharedPtr host) override {
+    upstream_host_ = host;
+  }
+  const FilterStateSharedPtr& upstreamFilterState() const override {
+    return upstream_filter_state_;
+  }
+  void setUpstreamFilterState(const FilterStateSharedPtr& filter_state) override {
+    upstream_filter_state_ = filter_state;
+  }
+
+  Upstream::HostDescriptionConstSharedPtr upstreamHost() const override { return upstream_host_; }
+
+  Upstream::HostDescriptionConstSharedPtr upstream_host_{};
+  Network::Address::InstanceConstSharedPtr upstream_local_address_;
+  UpstreamTiming upstream_timing_;
+  Ssl::ConnectionInfoConstSharedPtr upstream_ssl_info_;
+  absl::optional<uint64_t> upstream_connection_id_;
+  std::string upstream_transport_failure_reason_;
+  FilterStateSharedPtr upstream_filter_state_;
+};
+
 struct StreamInfoImpl : public StreamInfo {
   StreamInfoImpl(
       TimeSource& time_source,
@@ -71,9 +129,23 @@ struct StreamInfoImpl : public StreamInfo {
                                                                 start_time_monotonic_);
   }
 
-  void setUpstreamConnectionId(uint64_t id) override { upstream_connection_id_ = id; }
+  void maybeCreateUpstreamInfo() {
+    if (!upstream_info_) {
+      upstream_info_ = std::make_shared<UpstreamInfoImpl>();
+    }
+  }
+
+  void setUpstreamConnectionId(uint64_t id) override {
+    maybeCreateUpstreamInfo();
+    upstream_info_->setUpstreamConnectionId(id);
+  }
 
-  absl::optional<uint64_t> upstreamConnectionId() const override { return upstream_connection_id_; }
+  absl::optional<uint64_t> upstreamConnectionId() const override {
+    if (!upstream_info_) {
+      return absl::nullopt;
+    }
+    return upstream_info_->upstreamConnectionId();
+  }
 
   absl::optional<std::chrono::nanoseconds> lastDownstreamRxByteReceived() const override {
     if (!downstream_timing_.has_value()) {
@@ -83,23 +155,44 @@ struct StreamInfoImpl : public StreamInfo {
   }
 
   void setUpstreamTiming(const UpstreamTiming& upstream_timing) override {
-    upstream_timing_ = upstream_timing;
+    maybeCreateUpstreamInfo();
+    upstream_info_->setUpstreamTiming(upstream_timing);
   }
 
+  virtual void setUpstreamInfo(std::shared_ptr<UpstreamInfo> info) override {
+    upstream_info_ = info;
+  }
+  virtual std::shared_ptr<UpstreamInfo> upstreamInfo() override {
+    return upstream_info_;
+  }
+
+
   absl::optional<std::chrono::nanoseconds> firstUpstreamTxByteSent() const override {
-    return duration(upstream_timing_.first_upstream_tx_byte_sent_);
+    if (!upstream_info_) {
+      return absl::nullopt;
+    }
+    return duration(upstream_info_->upstreamTiming().first_upstream_tx_byte_sent_);
   }
 
   absl::optional<std::chrono::nanoseconds> lastUpstreamTxByteSent() const override {
-    return duration(upstream_timing_.last_upstream_tx_byte_sent_);
+    if (!upstream_info_) {
+      return absl::nullopt;
+    }
+    return duration(upstream_info_->upstreamTiming().last_upstream_tx_byte_sent_);
   }
 
   absl::optional<std::chrono::nanoseconds> firstUpstreamRxByteReceived() const override {
-    return duration(upstream_timing_.first_upstream_rx_byte_received_);
+    if (!upstream_info_) {
+      return absl::nullopt;
+    }
+    return duration(upstream_info_->upstreamTiming().first_upstream_rx_byte_received_);
   }
 
   absl::optional<std::chrono::nanoseconds> lastUpstreamRxByteReceived() const override {
-    return duration(upstream_timing_.last_upstream_rx_byte_received_);
+    if (!upstream_info_) {
+      return absl::nullopt;
+    }
+    return duration(upstream_info_->upstreamTiming().last_upstream_rx_byte_received_);
   }
 
   absl::optional<std::chrono::nanoseconds> firstDownstreamTxByteSent() const override {
@@ -177,10 +270,16 @@ struct StreamInfoImpl : public StreamInfo {
   uint64_t responseFlags() const override { return response_flags_; }
 
   void onUpstreamHostSelected(Upstream::HostDescriptionConstSharedPtr host) override {
-    upstream_host_ = host;
+    maybeCreateUpstreamInfo();
+    upstream_info_->setUpstreamHost(host);
   }
 
-  Upstream::HostDescriptionConstSharedPtr upstreamHost() const override { return upstream_host_; }
+  Upstream::HostDescriptionConstSharedPtr upstreamHost() const override {
+    if (!upstream_info_) {
+      return nullptr;
+    }
+    return upstream_info_->upstreamHost();
+  }
 
   void setRouteName(absl::string_view route_name) override {
     route_name_ = std::string(route_name);
@@ -190,11 +289,15 @@ struct StreamInfoImpl : public StreamInfo {
 
   void setUpstreamLocalAddress(
       const Network::Address::InstanceConstSharedPtr& upstream_local_address) override {
-    upstream_local_address_ = upstream_local_address;
+    maybeCreateUpstreamInfo();
+    upstream_info_->setUpstreamLocalAddress(upstream_local_address);
   }
 
   const Network::Address::InstanceConstSharedPtr& upstreamLocalAddress() const override {
-    return upstream_local_address_;
+    if (!upstream_info_) {
+      return legacy_upstream_local_address_;
+    }
+    return upstream_info_->upstreamLocalAddress();
   }
 
   bool healthCheck() const override { return health_check_request_; }
@@ -206,11 +309,12 @@ struct StreamInfoImpl : public StreamInfo {
   }
 
   void setUpstreamSslConnection(const Ssl::ConnectionInfoConstSharedPtr& connection_info) override {
-    upstream_ssl_info_ = connection_info;
+    maybeCreateUpstreamInfo();
+    upstream_info_->setUpstreamSslConnection(connection_info);
   }
 
   Ssl::ConnectionInfoConstSharedPtr upstreamSslConnection() const override {
-    return upstream_ssl_info_;
+    return upstream_info_ ? upstream_info_->upstreamSslConnection() : nullptr;
   }
 
   Router::RouteConstSharedPtr route() const override { return route_; }
@@ -226,18 +330,26 @@ struct StreamInfoImpl : public StreamInfo {
   const FilterState& filterState() const override { return *filter_state_; }
 
   const FilterStateSharedPtr& upstreamFilterState() const override {
-    return upstream_filter_state_;
+    if (!upstream_info_) {
+      return legacy_upstream_filter_state_;
+    }
+    return upstream_info_->upstreamFilterState();
   }
   void setUpstreamFilterState(const FilterStateSharedPtr& filter_state) override {
-    upstream_filter_state_ = filter_state;
+    maybeCreateUpstreamInfo();
+    return upstream_info_->setUpstreamFilterState(filter_state);
   }
 
   void setUpstreamTransportFailureReason(absl::string_view failure_reason) override {
-    upstream_transport_failure_reason_ = std::string(failure_reason);
+    maybeCreateUpstreamInfo();
+    upstream_info_->setUpstreamTransportFailureReason(failure_reason);
   }
 
   const std::string& upstreamTransportFailureReason() const override {
-    return upstream_transport_failure_reason_;
+    if (!upstream_info_) {
+      return legacy_upstream_transport_failure_reason_;
+    }
+    return upstream_info_->upstreamTransportFailureReason();
   }
 
   void setRequestHeaders(const Http::RequestHeaderMap& headers) override {
@@ -259,7 +371,8 @@ struct StreamInfoImpl : public StreamInfo {
 
   void dumpState(std::ostream& os, int indent_level = 0) const {
     const char* spaces = spacesForLevel(indent_level);
-    os << spaces << "StreamInfoImpl " << this << DUMP_OPTIONAL_MEMBER(upstream_connection_id_)
+    // TODO(alyssawilk) dump upstream info.
+    os << spaces << "StreamInfoImpl " << this
        << DUMP_OPTIONAL_MEMBER(protocol_) << DUMP_OPTIONAL_MEMBER(response_code_)
        << DUMP_OPTIONAL_MEMBER(response_code_details_) << DUMP_OPTIONAL_MEMBER(attempt_count_)
        << DUMP_MEMBER(health_check_request_) << DUMP_MEMBER(route_name_) << "\n";
@@ -279,7 +392,6 @@ struct StreamInfoImpl : public StreamInfo {
   }
 
   const std::string& filterChainName() const override { return filter_chain_name_; }
-
   void setAttemptCount(uint32_t attempt_count) override { attempt_count_ = attempt_count; }
 
   absl::optional<uint32_t> attemptCount() const override { return attempt_count_; }
@@ -320,14 +432,12 @@ struct StreamInfoImpl : public StreamInfo {
   absl::optional<std::string> response_code_details_;
   absl::optional<std::string> connection_termination_details_;
   uint64_t response_flags_{};
-  Upstream::HostDescriptionConstSharedPtr upstream_host_{};
   bool health_check_request_{};
   Router::RouteConstSharedPtr route_;
   envoy::config::core::v3::Metadata metadata_{};
   FilterStateSharedPtr filter_state_;
-  FilterStateSharedPtr upstream_filter_state_;
+  FilterStateSharedPtr legacy_upstream_filter_state_;
   std::string route_name_;
-  absl::optional<uint64_t> upstream_connection_id_;
   absl::optional<uint32_t> attempt_count_;
 
 private:
@@ -349,17 +459,17 @@ struct StreamInfoImpl : public StreamInfo {
                                                  : emptyDownstreamAddressProvider()),
         trace_reason_(Tracing::Reason::NotTraceable) {}
 
+
+  std::shared_ptr<UpstreamInfo> upstream_info_;
   uint64_t bytes_received_{};
   uint64_t bytes_sent_{};
-  Network::Address::InstanceConstSharedPtr upstream_local_address_;
+  Network::Address::InstanceConstSharedPtr legacy_upstream_local_address_;
   const Network::ConnectionInfoProviderSharedPtr downstream_connection_info_provider_;
-  Ssl::ConnectionInfoConstSharedPtr upstream_ssl_info_;
   std::string requested_server_name_;
   const Http::RequestHeaderMap* request_headers_{};
   Http::RequestIdStreamInfoProviderSharedPtr request_id_provider_;
   absl::optional<DownstreamTiming> downstream_timing_;
-  UpstreamTiming upstream_timing_;
-  std::string upstream_transport_failure_reason_;
+  std::string legacy_upstream_transport_failure_reason_;
   absl::optional<Upstream::ClusterInfoConstSharedPtr> upstream_cluster_info_;
   std::string filter_chain_name_;
   Tracing::Reason trace_reason_;