stream_info: adding more upstream timing metrics.

docs/root/version_history/current.rst

@@ -61,6 +61,7 @@ New Features
 * dns_resolver: added :ref:`AppleDnsResolverConfig<envoy_v3_api_msg_extensions.network.dns_resolver.apple.v3.AppleDnsResolverConfig>` to support apple DNS resolver as an extension.
 * ext_authz: added :ref:`query_parameters_to_set <envoy_v3_api_field_service.auth.v3.OkHttpResponse.query_parameters_to_set>` and :ref:`query_parameters_to_remove <envoy_v3_api_field_service.auth.v3.OkHttpResponse.query_parameters_to_remove>` for adding and removing query string parameters when using a gRPC authorization server.
 * http: added support for :ref:`retriable health check status codes <envoy_v3_api_field_config.core.v3.HealthCheck.HttpHealthCheck.retriable_statuses>`.
+* http: added timing information about upstream connection and encryption establishment to stream info. These can currently be accessed via custom access loggers.
 * listener: added API for extensions to access :ref:`typed_filter_metadata <envoy_v3_api_field_config.core.v3.Metadata.typed_filter_metadata>` configured in the listener's :ref:`metadata <envoy_v3_api_field_config.listener.v3.Listener.metadata>` field.
 * listener: added support for :ref:`MPTCP <envoy_v3_api_field_config.listener.v3.Listener.enable_mptcp>` (multipath TCP).
 * oauth filter: added :ref:`cookie_names <envoy_v3_api_field_extensions.filters.http.oauth2.v3.OAuth2Credentials.cookie_names>` to allow overriding (default) cookie names (``BearerToken``, ``OauthHMAC``, and ``OauthExpires``) set by the filter.

envoy/stream_info/stream_info.h

@@ -232,10 +232,27 @@ struct UpstreamTiming {
     last_upstream_rx_byte_received_ = time_source.monotonicTime();
   }
 
+  void onUpstreamConnectStart(TimeSource& time_source) {
+    ASSERT(!upstream_connect_start_);
+    upstream_connect_start_ = time_source.monotonicTime();
+  }
+
+  void onUpstreamConnectComplete(TimeSource& time_source) {
+    upstream_connect_complete_ = time_source.monotonicTime();
+  }
+
+  void onUpstreamHandshakeComplete(TimeSource& time_source) {
+    upstream_handshake_complete_ = time_source.monotonicTime();
+  }
+
   absl::optional<MonotonicTime> first_upstream_tx_byte_sent_;
   absl::optional<MonotonicTime> last_upstream_tx_byte_sent_;
   absl::optional<MonotonicTime> first_upstream_rx_byte_received_;
   absl::optional<MonotonicTime> last_upstream_rx_byte_received_;
+
+  absl::optional<MonotonicTime> upstream_connect_start_;
+  absl::optional<MonotonicTime> upstream_connect_complete_;
+  absl::optional<MonotonicTime> upstream_handshake_complete_;
 };
 
 class DownstreamTiming {
@@ -405,6 +422,14 @@ class StreamInfo {
    */
   virtual void setUpstreamTiming(const UpstreamTiming& upstream_timing) PURE;
 
+  /**
+   * Returns the upstream timing information for this stream.
+   * It is not expected that the fields in upstreamTiming() will be set until
+   * the upstream request is complete.
+   */
+  virtual UpstreamTiming& upstreamTiming() PURE;
+  virtual const UpstreamTiming& upstreamTiming() const 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/network/connection_impl.cc

@@ -671,6 +671,7 @@ void ConnectionImpl::onWriteReady() {
     if (error == 0) {
       ENVOY_CONN_LOG(debug, "connected", *this);
       connecting_ = false;
+      onConnected();
       transport_socket_->onConnected();
       // It's possible that we closed during the connect callback.
       if (state() != State::Open) {
@@ -845,7 +846,7 @@ ClientConnectionImpl::ClientConnectionImpl(
     const Network::ConnectionSocket::OptionsSharedPtr& options)
     : ConnectionImpl(dispatcher, std::move(socket), std::move(transport_socket), stream_info_,
                      false),
-      stream_info_(dispatcher.timeSource(), socket_->connectionInfoProviderSharedPtr()) {
+      stream_info_(dispatcher_.timeSource(), socket_->connectionInfoProviderSharedPtr()) {
 
   // There are no meaningful socket options or source address semantics for
   // non-IP sockets, so skip.
@@ -890,6 +891,7 @@ void ClientConnectionImpl::connect() {
                  socket_->connectionInfoProvider().remoteAddress()->asString());
   const Api::SysCallIntResult result =
       socket_->connect(socket_->connectionInfoProvider().remoteAddress());
+  stream_info_.upstreamTiming().onUpstreamConnectStart(dispatcher_.timeSource());
   if (result.return_value_ == 0) {
     // write will become ready.
     ASSERT(connecting_);
@@ -918,5 +920,10 @@ void ClientConnectionImpl::connect() {
   }
 }
 
+void ClientConnectionImpl::onConnected() {
+  stream_info_.upstreamTiming().onUpstreamConnectComplete(dispatcher_.timeSource());
+  ConnectionImpl::onConnected();
+}
+
 } // namespace Network
 } // namespace Envoy

source/common/network/connection_impl.h

@@ -149,6 +149,10 @@ class ConnectionImpl : public ConnectionImplBase, public TransportSocketCallback
   void onWriteBufferLowWatermark();
   void onWriteBufferHighWatermark();
 
+  // This is called when the underlying socket is connected, not when the
+  // connected event is raised.
+  virtual void onConnected() {}
+
   TransportSocketPtr transport_socket_;
   ConnectionSocketPtr socket_;
   StreamInfo::StreamInfo& stream_info_;
@@ -254,6 +258,8 @@ class ClientConnectionImpl : public ConnectionImpl, virtual public ClientConnect
   void connect() override;
 
 private:
+  void onConnected() override;
+
   StreamInfo::StreamInfoImpl stream_info_;
 };
 

source/common/quic/envoy_quic_client_session.cc

@@ -31,6 +31,7 @@ EnvoyQuicClientSession::~EnvoyQuicClientSession() {
 absl::string_view EnvoyQuicClientSession::requestedServerName() const { return server_id().host(); }
 
 void EnvoyQuicClientSession::connect() {
+  streamInfo().upstreamTiming().onUpstreamConnectStart(dispatcher_.timeSource());
   dynamic_cast<EnvoyQuicClientConnection*>(network_connection_)
       ->setUpConnectionSocket(
           *static_cast<EnvoyQuicClientConnection*>(connection())->connectionSocket(), *this);
@@ -132,6 +133,9 @@ void EnvoyQuicClientSession::OnTlsHandshakeComplete() {
   // before use. This may result in OnCanCreateNewOutgoingStream with zero
   // available streams.
   OnCanCreateNewOutgoingStream(false);
+  streamInfo().upstreamTiming().onUpstreamConnectComplete(dispatcher_.timeSource());
+  streamInfo().upstreamTiming().onUpstreamHandshakeComplete(dispatcher_.timeSource());
+
   raiseConnectionEvent(Network::ConnectionEvent::Connected);
 }
 

source/common/router/upstream_request.cc

@@ -427,6 +427,11 @@ void UpstreamRequest::onPoolReady(
     stream_info_.protocol(protocol.value());
   }
 
+  upstream_timing_.upstream_connect_start_ = info.upstreamTiming().upstream_connect_start_;
+  upstream_timing_.upstream_connect_complete_ = info.upstreamTiming().upstream_connect_complete_;
+  upstream_timing_.upstream_handshake_complete_ =
+      info.upstreamTiming().upstream_handshake_complete_;
+
   stream_info_.setUpstreamFilterState(std::make_shared<StreamInfo::FilterStateImpl>(
       info.filterState().parent()->parent(), StreamInfo::FilterState::LifeSpan::Request));
   parent_.callbacks()->streamInfo().setUpstreamFilterState(

source/common/stream_info/stream_info_impl.h

@@ -86,6 +86,9 @@ struct StreamInfoImpl : public StreamInfo {
     upstream_timing_ = upstream_timing;
   }
 
+  UpstreamTiming& upstreamTiming() override { return upstream_timing_; }
+  const UpstreamTiming& upstreamTiming() const override { return upstream_timing_; }
+
   absl::optional<std::chrono::nanoseconds> firstUpstreamTxByteSent() const override {
     return duration(upstream_timing_.first_upstream_tx_byte_sent_);
   }

source/extensions/transport_sockets/tls/ssl_socket.cc

@@ -43,6 +43,7 @@ class NotReadySslSocket : public Network::TransportSocket {
   Ssl::ConnectionInfoConstSharedPtr ssl() const override { return nullptr; }
   bool startSecureTransport() override { return false; }
 };
+
 } // namespace
 
 SslSocket::SslSocket(Envoy::Ssl::ContextSharedPtr ctx, InitialState state,
@@ -179,6 +180,8 @@ Network::Connection& SslSocket::connection() const { return callbacks_->connecti
 
 void SslSocket::onSuccess(SSL* ssl) {
   ctx_->logHandshake(ssl);
+  callbacks_->connection().streamInfo().upstreamTiming().onUpstreamHandshakeComplete(
+      callbacks_->connection().dispatcher().timeSource());
   callbacks_->raiseEvent(Network::ConnectionEvent::Connected);
 }
 

test/common/stream_info/test_util.h

@@ -137,6 +137,10 @@ class TestStreamInfo : public StreamInfo::StreamInfo {
   void setUpstreamTiming(const Envoy::StreamInfo::UpstreamTiming& upstream_timing) override {
     upstream_timing_ = upstream_timing;
   }
+  Envoy::StreamInfo::UpstreamTiming& upstreamTiming() override { return upstream_timing_; }
+  const Envoy::StreamInfo::UpstreamTiming& upstreamTiming() const override {
+    return upstream_timing_;
+  }
 
   absl::optional<std::chrono::nanoseconds> requestComplete() const override {
     return duration(end_time_);

test/extensions/transport_sockets/tls/ssl_socket_test.cc

@@ -363,7 +363,7 @@ void testUtil(const TestUtilOptions& options) {
       client_ssl_socket_factory.createTransportSocket(nullptr), nullptr);
   Network::ConnectionPtr server_connection;
   Network::MockConnectionCallbacks server_connection_callbacks;
-  StreamInfo::MockStreamInfo stream_info;
+  NiceMock<StreamInfo::MockStreamInfo> stream_info;
   EXPECT_CALL(callbacks, onAccept_(_))
       .WillOnce(Invoke([&](Network::ConnectionSocketPtr& socket) -> void {
         server_connection = dispatcher->createServerConnection(
@@ -708,7 +708,7 @@ void testUtilV2(const TestUtilOptionsV2& options) {
 
   Network::ConnectionPtr server_connection;
   Network::MockConnectionCallbacks server_connection_callbacks;
-  StreamInfo::MockStreamInfo stream_info;
+  NiceMock<StreamInfo::MockStreamInfo> stream_info;
   EXPECT_CALL(callbacks, onAccept_(_))
       .WillOnce(Invoke([&](Network::ConnectionSocketPtr& socket) -> void {
         std::string sni = options.transportSocketOptions() != nullptr &&

test/integration/filters/stream_info_to_headers_filter.cc

@@ -24,8 +24,31 @@ class StreamInfoToHeadersFilter : public Http::PassThroughFilter {
       headers.addCopy(Http::LowerCaseString("alpn"),
                       decoder_callbacks_->streamInfo().upstreamSslConnection()->alpn());
     }
+
     return Http::FilterHeadersStatus::Continue;
   }
+  Http::FilterTrailersStatus encodeTrailers(Http::ResponseTrailerMap& trailers) override {
+    StreamInfo::UpstreamTiming& upstream_timing = decoder_callbacks_->streamInfo().upstreamTiming();
+    // Upstream metrics aren't available until the response is complete.
+    if (upstream_timing.upstream_connect_start_.has_value()) {
+      trailers.addCopy(
+          Http::LowerCaseString("upstream_connect_start"),
+          absl::StrCat(upstream_timing.upstream_connect_start_.value().time_since_epoch().count()));
+    }
+    if (upstream_timing.upstream_connect_complete_.has_value()) {
+      trailers.addCopy(
+          Http::LowerCaseString("upstream_connect_complete"),
+          absl::StrCat(
+              upstream_timing.upstream_connect_complete_.value().time_since_epoch().count()));
+    }
+    if (upstream_timing.upstream_handshake_complete_.has_value()) {
+      trailers.addCopy(
+          Http::LowerCaseString("upstream_handshake_complete"),
+          absl::StrCat(
+              upstream_timing.upstream_handshake_complete_.value().time_since_epoch().count()));
+    }
+    return Http::FilterTrailersStatus::Continue;
+  }
 };
 
 constexpr char StreamInfoToHeadersFilter::name[];

test/integration/multiplexed_upstream_integration_test.cc

@@ -152,6 +152,12 @@ void MultiplexedUpstreamIntegrationTest::bidirectionalStreaming(uint32_t bytes)
   ASSERT_FALSE(response->headers().get(Http::LowerCaseString("alpn")).empty());
   ASSERT_EQ(response->headers().get(Http::LowerCaseString("alpn"))[0]->value().getStringView(),
             expected_alpn);
+
+  ASSERT_FALSE(response->trailers()->get(Http::LowerCaseString("upstream_connect_start")).empty());
+  ASSERT_FALSE(
+      response->trailers()->get(Http::LowerCaseString("upstream_connect_complete")).empty());
+  ASSERT_FALSE(
+      response->trailers()->get(Http::LowerCaseString("upstream_handshake_complete")).empty());
 }
 
 TEST_P(MultiplexedUpstreamIntegrationTest, BidirectionalStreaming) { bidirectionalStreaming(1024); }

test/mocks/stream_info/mocks.cc

@@ -141,6 +141,8 @@ MockStreamInfo::MockStreamInfo()
       .WillByDefault(Invoke([this](const BytesMeterSharedPtr& downstream_bytes_meter) {
         downstream_bytes_meter_ = downstream_bytes_meter;
       }));
+  ON_CALL(*this, upstreamTiming()).WillByDefault(ReturnRef(upstream_timing_));
+  ON_CALL(Const(*this), upstreamTiming()).WillByDefault(ReturnRef(upstream_timing_));
 }
 
 MockStreamInfo::~MockStreamInfo() = default;

test/mocks/stream_info/mocks.h

@@ -31,6 +31,8 @@ class MockStreamInfo : public StreamInfo {
   MOCK_METHOD(MonotonicTime, startTimeMonotonic, (), (const));
   MOCK_METHOD(absl::optional<std::chrono::nanoseconds>, lastDownstreamRxByteReceived, (), (const));
   MOCK_METHOD(void, setUpstreamTiming, (const UpstreamTiming&));
+  MOCK_METHOD(UpstreamTiming&, upstreamTiming, ());
+  MOCK_METHOD(const UpstreamTiming&, upstreamTiming, (), (const));
   MOCK_METHOD(absl::optional<std::chrono::nanoseconds>, firstUpstreamTxByteSent, (), (const));
   MOCK_METHOD(void, onFirstUpstreamTxByteSent, ());
   MOCK_METHOD(absl::optional<std::chrono::nanoseconds>, lastUpstreamTxByteSent, (), (const));
@@ -122,6 +124,7 @@ class MockStreamInfo : public StreamInfo {
   absl::optional<uint32_t> response_code_;
   absl::optional<std::string> response_code_details_;
   absl::optional<std::string> connection_termination_details_;
+  UpstreamTiming upstream_timing_;
   uint64_t response_flags_{};
   envoy::config::core::v3::Metadata metadata_;
   FilterStateSharedPtr upstream_filter_state_;