access log: Allow upstream filter state in upstream_log. (envoyproxy#…

…21344) Currently in the context of an upstream Http request, we cannot add upstream specific filter state data to access log. This change addresses this issue by introducng a new format tag for http router's `upstream_log` component - `UPSTREAM_FILTER_STATE`. This allows upstream compoents (for example, transport socket) to add filter state which then gets propagated to the access log. Signed-off-by: Jojy George Varghese <[email protected]>
AppMonet · Jun 2, 2022 · 416c6f5 · 416c6f5
1 parent 00a6cff
commit 416c6f5
Show file tree

Hide file tree

Showing 25 changed files with 435 additions and 48 deletions.
diff --git a/changelogs/current.yaml b/changelogs/current.yaml
@@ -117,7 +117,7 @@ removed_config_or_runtime:
 new_features:
 - area: access_log
   change: |
-    added new access_log command operators to retrieve upstream connection information change: ``%UPSTREAM_PROTOCOL%``, ``%UPSTREAM_PEER_SUBJECT%``, ``%UPSTREAM_PEER_ISSUER%``, ``%UPSTREAM_TLS_SESSION_ID%``, ``%UPSTREAM_TLS_CIPHER%``, ``%UPSTREAM_TLS_VERSION%``, ``%UPSTREAM_PEER_CERT_V_START%``, ``%UPSTREAM_PEER_CERT_V_END%`` and ``%UPSTREAM_PEER_CERT%``.
+    added new access_log command operators to retrieve upstream connection information change: ``%UPSTREAM_PROTOCOL%``, ``%UPSTREAM_PEER_SUBJECT%``, ``%UPSTREAM_PEER_ISSUER%``, ``%UPSTREAM_TLS_SESSION_ID%``, ``%UPSTREAM_TLS_CIPHER%``, ``%UPSTREAM_TLS_VERSION%``, ``%UPSTREAM_PEER_CERT_V_START%``, ``%UPSTREAM_PEER_CERT_V_END%``, ``%UPSTREAM_PEER_CERT%` and ``%UPSTREAM_FILTER_STATE%``.
 - area: dns_resolver
   change: |
     added :ref:`include_unroutable_families<envoy_v3_api_field_extensions.network.dns_resolver.apple.v3.AppleDnsResolverConfig.include_unroutable_families>` to the Apple DNS resolver.

diff --git a/docs/root/configuration/observability/access_log/usage.rst b/docs/root/configuration/observability/access_log/usage.rst
@@ -759,6 +759,25 @@ The following command operators are supported:
     JSON struct or list is rendered. Structs and lists may be nested. In any event, the maximum
     length is ignored
 
+%UPSTREAM_FILTER_STATE(KEY:F):Z%
+  HTTP
+    Extracts filter state from upstream components like cluster or transport socket extensions.
+
+    :ref:`Filter State <arch_overview_data_sharing_between_filters>` info, where the KEY is required to
+    look up the filter state object. The serialized proto will be logged as JSON string if possible.
+    If the serialized proto is unknown to Envoy it will be logged as protobuf debug string.
+    Z is an optional parameter denoting string truncation up to Z characters long.
+    F is an optional parameter used to indicate which method FilterState uses for serialization.
+    If 'PLAIN' is set, the filter state object will be serialized as an unstructured string.
+    If 'TYPED' is set or no F provided, the filter state object will be serialized as an JSON string.
+
+  TCP/UDP
+    Not implemented.
+
+  .. note::
+
+    This command operator is only available for :ref:`upstream_log <envoy_v3_api_field_extensions.filters.http.router.v3.Router.upstream_log>`
+
 %REQUESTED_SERVER_NAME%
   HTTP/TCP/THRIFT
     String value set on ssl connection socket for Server Name Indication (SNI)

diff --git a/envoy/http/conn_pool.h b/envoy/http/conn_pool.h
@@ -43,7 +43,7 @@ class Callbacks {
    * @param protocol supplies the protocol associated with the stream, or absl::nullopt for raw TCP.
    */
   virtual void onPoolReady(RequestEncoder& encoder, Upstream::HostDescriptionConstSharedPtr host,
-                           const StreamInfo::StreamInfo& info,
+                           StreamInfo::StreamInfo& info,
                            absl::optional<Http::Protocol> protocol) PURE;
 };
 

diff --git a/envoy/router/router.h b/envoy/router/router.h
@@ -1357,7 +1357,7 @@ class GenericConnectionPoolCallbacks {
   virtual void onPoolReady(std::unique_ptr<GenericUpstream>&& upstream,
                            Upstream::HostDescriptionConstSharedPtr host,
                            const Network::Address::InstanceConstSharedPtr& upstream_local_address,
-                           const StreamInfo::StreamInfo& info,
+                           StreamInfo::StreamInfo& info,
                            absl::optional<Http::Protocol> protocol) PURE;
 
   // @return the UpstreamToDownstream interface for this stream.

diff --git a/source/common/formatter/substitution_formatter.cc b/source/common/formatter/substitution_formatter.cc
@@ -429,25 +429,12 @@ SubstitutionFormatParser::getKnownFormatters() {
        {"FILTER_STATE",
         {CommandSyntaxChecker::PARAMS_OPTIONAL | CommandSyntaxChecker::LENGTH_ALLOWED,
          [](const std::string& format, const absl::optional<size_t>& max_length) {
-           std::string key, serialize_type;
-           static constexpr absl::string_view PLAIN_SERIALIZATION{"PLAIN"};
-           static constexpr absl::string_view TYPED_SERIALIZATION{"TYPED"};
-
-           SubstitutionFormatParser::parseSubcommand(format, ':', key, serialize_type);
-           if (key.empty()) {
-             throw EnvoyException("Invalid filter state configuration, key cannot be empty.");
-           }
-
-           if (serialize_type.empty()) {
-             serialize_type = std::string(TYPED_SERIALIZATION);
-           }
-           if (serialize_type != PLAIN_SERIALIZATION && serialize_type != TYPED_SERIALIZATION) {
-             throw EnvoyException("Invalid filter state serialize type, only "
-                                  "support PLAIN/TYPED.");
-           }
-           const bool serialize_as_string = serialize_type == PLAIN_SERIALIZATION;
-
-           return std::make_unique<FilterStateFormatter>(key, max_length, serialize_as_string);
+           return FilterStateFormatter::create(format, max_length, false);
+         }}},
+       {"UPSTREAM_FILTER_STATE",
+        {CommandSyntaxChecker::PARAMS_OPTIONAL | CommandSyntaxChecker::LENGTH_ALLOWED,
+         [](const std::string& format, const absl::optional<size_t>& max_length) {
+           return FilterStateFormatter::create(format, max_length, true);
          }}},
        {"DOWNSTREAM_PEER_CERT_V_START",
         {CommandSyntaxChecker::PARAMS_OPTIONAL,
@@ -1818,15 +1805,55 @@ ClusterMetadataFormatter::ClusterMetadataFormatter(const std::string& filter_nam
                           }
                           return &cluster_info.value()->metadata();
                         }) {}
+
+std::unique_ptr<FilterStateFormatter>
+FilterStateFormatter::create(const std::string& format, const absl::optional<size_t>& max_length,
+                             bool is_upstream) {
+  std::string key, serialize_type;
+  static constexpr absl::string_view PLAIN_SERIALIZATION{"PLAIN"};
+  static constexpr absl::string_view TYPED_SERIALIZATION{"TYPED"};
+
+  SubstitutionFormatParser::parseSubcommand(format, ':', key, serialize_type);
+  if (key.empty()) {
+    throw EnvoyException("Invalid filter state configuration, key cannot be empty.");
+  }
+
+  if (serialize_type.empty()) {
+    serialize_type = std::string(TYPED_SERIALIZATION);
+  }
+  if (serialize_type != PLAIN_SERIALIZATION && serialize_type != TYPED_SERIALIZATION) {
+    throw EnvoyException("Invalid filter state serialize type, only "
+                         "support PLAIN/TYPED.");
+  }
+
+  const bool serialize_as_string = serialize_type == PLAIN_SERIALIZATION;
+
+  return std::make_unique<FilterStateFormatter>(key, max_length, serialize_as_string, is_upstream);
+}
+
 FilterStateFormatter::FilterStateFormatter(const std::string& key,
                                            absl::optional<size_t> max_length,
-                                           bool serialize_as_string)
-    : key_(key), max_length_(max_length), serialize_as_string_(serialize_as_string) {}
+                                           bool serialize_as_string, bool is_upstream)
+    : key_(key), max_length_(max_length), serialize_as_string_(serialize_as_string),
+      is_upstream_(is_upstream) {}
 
 const Envoy::StreamInfo::FilterState::Object*
 FilterStateFormatter::filterState(const StreamInfo::StreamInfo& stream_info) const {
-  const StreamInfo::FilterState& filter_state = stream_info.filterState();
-  return filter_state.getDataReadOnly<StreamInfo::FilterState::Object>(key_);
+  const StreamInfo::FilterState* filter_state = nullptr;
+  if (is_upstream_) {
+    const OptRef<const StreamInfo::UpstreamInfo> upstream_info = stream_info.upstreamInfo();
+    if (upstream_info) {
+      filter_state = upstream_info->upstreamFilterState().get();
+    }
+  } else {
+    filter_state = &stream_info.filterState();
+  }
+
+  if (filter_state) {
+    return filter_state->getDataReadOnly<StreamInfo::FilterState::Object>(key_);
+  }
+
+  return nullptr;
 }
 
 absl::optional<std::string> FilterStateFormatter::format(const Http::RequestHeaderMap&,

diff --git a/source/common/formatter/substitution_formatter.h b/source/common/formatter/substitution_formatter.h
@@ -514,8 +514,11 @@ class ClusterMetadataFormatter : public MetadataFormatter {
  */
 class FilterStateFormatter : public FormatterProvider {
 public:
+  static std::unique_ptr<FilterStateFormatter>
+  create(const std::string& format, const absl::optional<size_t>& max_length, bool is_upstream);
+
   FilterStateFormatter(const std::string& key, absl::optional<size_t> max_length,
-                       bool serialize_as_string);
+                       bool serialize_as_string, bool is_upstream = false);
 
   // FormatterProvider
   absl::optional<std::string> format(const Http::RequestHeaderMap&, const Http::ResponseHeaderMap&,
@@ -533,6 +536,7 @@ class FilterStateFormatter : public FormatterProvider {
   absl::optional<size_t> max_length_;
 
   bool serialize_as_string_;
+  const bool is_upstream_;
 };
 
 /**

diff --git a/source/common/http/codec_client.h b/source/common/http/codec_client.h
@@ -127,7 +127,7 @@ class CodecClient : protected Logger::Loggable<Logger::Id::client>,
   CodecType type() const { return type_; }
 
   // Note this is the L4 stream info, not L7.
-  const StreamInfo::StreamInfo& streamInfo() { return connection_->streamInfo(); }
+  StreamInfo::StreamInfo& streamInfo() { return connection_->streamInfo(); }
 
   /**
    * Connect to the host.

diff --git a/source/common/http/conn_pool_grid.cc b/source/common/http/conn_pool_grid.cc
@@ -125,7 +125,7 @@ ConnectivityGrid::StreamCreationResult ConnectivityGrid::WrapperCallbacks::newSt
 
 void ConnectivityGrid::WrapperCallbacks::onConnectionAttemptReady(
     ConnectionAttemptCallbacks* attempt, RequestEncoder& encoder,
-    Upstream::HostDescriptionConstSharedPtr host, const StreamInfo::StreamInfo& info,
+    Upstream::HostDescriptionConstSharedPtr host, StreamInfo::StreamInfo& info,
     absl::optional<Http::Protocol> protocol) {
   ENVOY_LOG(trace, "{} pool successfully connected to host '{}'.", describePool(attempt->pool()),
             host->hostname());
@@ -160,7 +160,7 @@ void ConnectivityGrid::WrapperCallbacks::maybeMarkHttp3Broken() {
 
 void ConnectivityGrid::WrapperCallbacks::ConnectionAttemptCallbacks::onPoolReady(
     RequestEncoder& encoder, Upstream::HostDescriptionConstSharedPtr host,
-    const StreamInfo::StreamInfo& info, absl::optional<Http::Protocol> protocol) {
+    StreamInfo::StreamInfo& info, absl::optional<Http::Protocol> protocol) {
   cancellable_ = nullptr; // Attempt succeeded and can no longer be cancelled.
   parent_.onConnectionAttemptReady(this, encoder, host, info, protocol);
 }

diff --git a/source/common/http/conn_pool_grid.h b/source/common/http/conn_pool_grid.h
@@ -55,7 +55,7 @@ class ConnectivityGrid : public ConnectionPool::Instance,
                          absl::string_view transport_failure_reason,
                          Upstream::HostDescriptionConstSharedPtr host) override;
       void onPoolReady(RequestEncoder& encoder, Upstream::HostDescriptionConstSharedPtr host,
-                       const StreamInfo::StreamInfo& info,
+                       StreamInfo::StreamInfo& info,
                        absl::optional<Http::Protocol> protocol) override;
 
       ConnectionPool::Instance& pool() { return **pool_it_; }
@@ -93,7 +93,7 @@ class ConnectivityGrid : public ConnectionPool::Instance,
     // Called by a ConnectionAttempt when the underlying pool is ready.
     void onConnectionAttemptReady(ConnectionAttemptCallbacks* attempt, RequestEncoder& encoder,
                                   Upstream::HostDescriptionConstSharedPtr host,
-                                  const StreamInfo::StreamInfo& info,
+                                  StreamInfo::StreamInfo& info,
                                   absl::optional<Http::Protocol> protocol);
 
   private:

diff --git a/source/common/router/upstream_request.cc b/source/common/router/upstream_request.cc
@@ -416,7 +416,7 @@ void UpstreamRequest::onPoolFailure(ConnectionPool::PoolFailureReason reason,
 void UpstreamRequest::onPoolReady(
     std::unique_ptr<GenericUpstream>&& upstream, Upstream::HostDescriptionConstSharedPtr host,
     const Network::Address::InstanceConstSharedPtr& upstream_local_address,
-    const StreamInfo::StreamInfo& info, absl::optional<Http::Protocol> protocol) {
+    StreamInfo::StreamInfo& info, absl::optional<Http::Protocol> protocol) {
   // This may be called under an existing ScopeTrackerScopeState but it will unwind correctly.
   ScopeTrackerScopeState scope(&parent_.callbacks()->scope(), parent_.callbacks()->dispatcher());
   ENVOY_STREAM_LOG(debug, "pool ready", *parent_.callbacks());
@@ -442,16 +442,22 @@ void UpstreamRequest::onPoolReady(
 
   StreamInfo::UpstreamInfo& upstream_info = *stream_info_.upstreamInfo();
   parent_.callbacks()->streamInfo().setUpstreamInfo(stream_info_.upstreamInfo());
-  if (info.upstreamInfo().has_value()) {
-    auto& upstream_timing = info.upstreamInfo().value().get().upstreamTiming();
+  if (info.upstreamInfo()) {
+    auto& upstream_timing = info.upstreamInfo()->upstreamTiming();
     upstreamTiming().upstream_connect_start_ = upstream_timing.upstream_connect_start_;
     upstreamTiming().upstream_connect_complete_ = upstream_timing.upstream_connect_complete_;
     upstreamTiming().upstream_handshake_complete_ = upstream_timing.upstream_handshake_complete_;
-    upstream_info.setUpstreamNumStreams(info.upstreamInfo().value().get().upstreamNumStreams());
+    upstream_info.setUpstreamNumStreams(info.upstreamInfo()->upstreamNumStreams());
   }
 
-  upstream_info.setUpstreamFilterState(std::make_shared<StreamInfo::FilterStateImpl>(
-      info.filterState().parent()->parent(), StreamInfo::FilterState::LifeSpan::Request));
+  // Upstream filters might have already created/set a filter state.
+  const StreamInfo::FilterStateSharedPtr& filter_state = info.filterState();
+  if (!filter_state) {
+    upstream_info.setUpstreamFilterState(
+        std::make_shared<StreamInfo::FilterStateImpl>(StreamInfo::FilterState::LifeSpan::Request));
+  } else {
+    upstream_info.setUpstreamFilterState(filter_state);
+  }
   upstream_info.setUpstreamLocalAddress(upstream_local_address);
   upstream_info.setUpstreamSslConnection(info.downstreamAddressProvider().sslConnection());
 

diff --git a/source/common/router/upstream_request.h b/source/common/router/upstream_request.h
@@ -88,8 +88,7 @@ class UpstreamRequest : public Logger::Loggable<Logger::Id::router>,
   void onPoolReady(std::unique_ptr<GenericUpstream>&& upstream,
                    Upstream::HostDescriptionConstSharedPtr host,
                    const Network::Address::InstanceConstSharedPtr& upstream_local_address,
-                   const StreamInfo::StreamInfo& info,
-                   absl::optional<Http::Protocol> protocol) override;
+                   StreamInfo::StreamInfo& info, absl::optional<Http::Protocol> protocol) override;
   UpstreamToDownstream& upstreamToDownstream() override { return *this; }
 
   void clearRequestEncoder();

diff --git a/source/common/tcp_proxy/upstream.cc b/source/common/tcp_proxy/upstream.cc
@@ -243,7 +243,7 @@ void HttpConnPool::onPoolFailure(ConnectionPool::PoolFailureReason reason,
 
 void HttpConnPool::onPoolReady(Http::RequestEncoder& request_encoder,
                                Upstream::HostDescriptionConstSharedPtr host,
-                               const StreamInfo::StreamInfo& info, absl::optional<Http::Protocol>) {
+                               StreamInfo::StreamInfo& info, absl::optional<Http::Protocol>) {
   upstream_handle_ = nullptr;
   upstream_->setRequestEncoder(request_encoder,
                                host->transportSocketFactory().implementsSecureTransport());

diff --git a/source/common/tcp_proxy/upstream.h b/source/common/tcp_proxy/upstream.h
@@ -60,7 +60,7 @@ class HttpConnPool : public GenericConnPool, public Http::ConnectionPool::Callba
                      absl::string_view transport_failure_reason,
                      Upstream::HostDescriptionConstSharedPtr host) override;
   void onPoolReady(Http::RequestEncoder& request_encoder,
-                   Upstream::HostDescriptionConstSharedPtr host, const StreamInfo::StreamInfo& info,
+                   Upstream::HostDescriptionConstSharedPtr host, StreamInfo::StreamInfo& info,
                    absl::optional<Http::Protocol>) override;
 
   class Callbacks {

diff --git a/source/extensions/transport_sockets/common/BUILD b/source/extensions/transport_sockets/common/BUILD
@@ -12,6 +12,7 @@ envoy_cc_library(
     name = "passthrough_lib",
     srcs = ["passthrough.cc"],
     hdrs = ["passthrough.h"],
+    visibility = ["//visibility:public"],
     deps = [
         "//envoy/network:connection_interface",
         "//envoy/network:transport_socket_interface",

diff --git a/source/extensions/upstreams/http/http/upstream_request.cc b/source/extensions/upstreams/http/http/upstream_request.cc
@@ -57,7 +57,7 @@ void HttpConnPool::onPoolFailure(ConnectionPool::PoolFailureReason reason,
 
 void HttpConnPool::onPoolReady(Envoy::Http::RequestEncoder& request_encoder,
                                Upstream::HostDescriptionConstSharedPtr host,
-                               const StreamInfo::StreamInfo& info,
+                               StreamInfo::StreamInfo& info,
                                absl::optional<Envoy::Http::Protocol> protocol) {
   conn_pool_stream_handle_ = nullptr;
   auto upstream =

diff --git a/source/extensions/upstreams/http/http/upstream_request.h b/source/extensions/upstreams/http/http/upstream_request.h
@@ -40,7 +40,7 @@ class HttpConnPool : public Router::GenericConnPool, public Envoy::Http::Connect
                      absl::string_view transport_failure_reason,
                      Upstream::HostDescriptionConstSharedPtr host) override;
   void onPoolReady(Envoy::Http::RequestEncoder& callbacks_encoder,
-                   Upstream::HostDescriptionConstSharedPtr host, const StreamInfo::StreamInfo& info,
+                   Upstream::HostDescriptionConstSharedPtr host, StreamInfo::StreamInfo& info,
                    absl::optional<Envoy::Http::Protocol> protocol) override;
   Upstream::HostDescriptionConstSharedPtr host() const override {
     return pool_data_.value().host();

diff --git a/test/common/http/common.h b/test/common/http/common.h
@@ -41,7 +41,7 @@ class CodecClientForTest : public Http::CodecClient {
  */
 struct ConnPoolCallbacks : public Http::ConnectionPool::Callbacks {
   void onPoolReady(Http::RequestEncoder& encoder, Upstream::HostDescriptionConstSharedPtr host,
-                   const StreamInfo::StreamInfo&, absl::optional<Http::Protocol>) override {
+                   StreamInfo::StreamInfo&, absl::optional<Http::Protocol>) override {
     outer_encoder_ = &encoder;
     host_ = host;
     pool_ready_.ready();