From acd4b43537ca2176c31317d3eaec93174644b388 Mon Sep 17 00:00:00 2001 From: Gustavo Date: Wed, 8 Jul 2026 15:37:13 +0000 Subject: [PATCH 1/5] adds peer metadata exchange through thread local storage Signed-off-by: Gustavo --- contrib/istio/filters/common/source/BUILD | 17 ++ .../common/source/peer_metadata_registry.cc | 77 ++++++++ .../common/source/peer_metadata_registry.h | 58 ++++++ .../network/peer_metadata/source/BUILD | 3 + .../peer_metadata/source/peer_metadata.cc | 176 +++++++----------- .../peer_metadata/source/peer_metadata.h | 49 ++--- .../peer_metadata/test/peer_metadata_test.cc | 4 +- 7 files changed, 243 insertions(+), 141 deletions(-) create mode 100644 contrib/istio/filters/common/source/peer_metadata_registry.cc create mode 100644 contrib/istio/filters/common/source/peer_metadata_registry.h diff --git a/contrib/istio/filters/common/source/BUILD b/contrib/istio/filters/common/source/BUILD index b40a722553a25..7ad13e3d5d662 100644 --- a/contrib/istio/filters/common/source/BUILD +++ b/contrib/istio/filters/common/source/BUILD @@ -58,3 +58,20 @@ envoy_cc_library( "@envoy_api//contrib/envoy/extensions/filters/common/workload_discovery/v3:pkg_cc_proto", ], ) + +envoy_cc_library( + name = "peer_metadata_registry_lib", + srcs = ["peer_metadata_registry.cc"], + hdrs = ["peer_metadata_registry.h"], + visibility = ["//visibility:public"], + deps = [ + "//envoy/registry", + "//envoy/server:bootstrap_extension_config_interface", + "//envoy/server:factory_context_interface", + "//envoy/singleton:instance_interface", + "//envoy/singleton:manager_interface", + "//envoy/thread_local:thread_local_interface", + "//source/common/protobuf", + "@com_google_absl//absl/container:flat_hash_map", + ], +) diff --git a/contrib/istio/filters/common/source/peer_metadata_registry.cc b/contrib/istio/filters/common/source/peer_metadata_registry.cc new file mode 100644 index 0000000000000..120ed9d1121c4 --- /dev/null +++ b/contrib/istio/filters/common/source/peer_metadata_registry.cc @@ -0,0 +1,77 @@ +#include "contrib/istio/filters/common/source/peer_metadata_registry.h" + +#include "envoy/registry/registry.h" +#include "envoy/server/bootstrap_extension_config.h" +#include "envoy/singleton/instance.h" +#include "envoy/singleton/manager.h" +#include "envoy/thread_local/thread_local.h" +#include "envoy/thread_local/thread_local_object.h" + +#include "source/common/protobuf/protobuf.h" + +#include "absl/container/flat_hash_map.h" + +namespace Envoy { +namespace Extensions { +namespace Filters { +namespace Common { +namespace PeerMetadataShared { + +namespace { + +struct ThreadLocalRegistry : public ThreadLocal::ThreadLocalObject { + absl::flat_hash_map values_; +}; + +class PeerMetadataRegistryImpl : public PeerMetadataRegistry, public Singleton::Instance { +public: + explicit PeerMetadataRegistryImpl(ThreadLocal::SlotAllocator& tls) + : tls_(ThreadLocal::TypedSlot::makeUnique(tls)) { + tls_->set([](Event::Dispatcher&) { return std::make_shared(); }); + } + + void setValue(uint64_t key, const std::string& value) override { + if (auto tls = tls_->get(); tls.has_value()) { + tls->values_[key] = value; + } + } + + absl::optional getValue(uint64_t key) const override { + if (auto tls = tls_->get(); tls.has_value()) { + const auto it = tls->values_.find(key); + if (it != tls->values_.end()) { + return it->second; + } + } + return absl::nullopt; + } + + void removeValue(uint64_t key) override { + if (auto tls = tls_->get(); tls.has_value()) { + tls->values_.erase(key); + } + } + +private: + ThreadLocal::TypedSlotPtr tls_; +}; + +} // namespace + +SINGLETON_MANAGER_REGISTRATION(peer_metadata_registry); + +PeerMetadataRegistrySharedPtr +getRegistry(Server::Configuration::ServerFactoryContext& context) { + return context.singletonManager().getTyped( + SINGLETON_MANAGER_REGISTERED_NAME(peer_metadata_registry), + [&context] { + return std::make_shared(context.threadLocal()); + }, + /*pin=*/true); +} + +} // namespace PeerMetadataShared +} // namespace Common +} // namespace Filters +} // namespace Extensions +} // namespace Envoy diff --git a/contrib/istio/filters/common/source/peer_metadata_registry.h b/contrib/istio/filters/common/source/peer_metadata_registry.h new file mode 100644 index 0000000000000..a8e7ef46eedbc --- /dev/null +++ b/contrib/istio/filters/common/source/peer_metadata_registry.h @@ -0,0 +1,58 @@ +#pragma once + +#include +#include +#include + +#include "envoy/server/factory_context.h" + +#include "absl/strings/string_view.h" +#include "absl/types/optional.h" + +namespace Envoy { +namespace Extensions { +namespace Filters { +namespace Common { +namespace PeerMetadataShared { + +// Process-wide singleton that exposes a per-worker-thread key/value store used +// to hand peer metadata information from the peer_metadata network filter on an +// internal listener (writer) to the peer_metadata upstream network filter on a +// cluster (reader). +// +// The backing storage is a ThreadLocal::TypedSlot, so each worker thread owns +// its own map and reads/writes are lock-free. This only works as a hand-off +// when the writing and reading filters run on the *same* worker thread (which +// is the case for internal listener connections), and when both sides agree on +// the key (the originating upstream connection ID). +class PeerMetadataRegistry { +public: + virtual ~PeerMetadataRegistry() = default; + + // Store `value` under `key` in the current worker thread's store. + virtual void setValue(uint64_t key, const std::string& value) PURE; + + // Look up `key` in the current worker thread's store. Returns absl::nullopt + // if no value was stored (on this thread) under `key`. + virtual absl::optional getValue(uint64_t key) const PURE; + + // Remove the entry for `key` from the current worker thread's store. + virtual void removeValue(uint64_t key) PURE; +}; + +using PeerMetadataRegistrySharedPtr = std::shared_ptr; + +// Returns the process-wide PeerMetadataRegistry singleton, creating it if +// necessary. The registry is pinned in the singleton manager. +PeerMetadataRegistrySharedPtr +getRegistry(Server::Configuration::ServerFactoryContext& context); + +// Filter state key under which the originating upstream connection ID is stored. +constexpr absl::string_view ConnectionIdFilterStateKey = + "envoy.peer_metadata.upstream_connection_id"; + +} // namespace PeerMetadataShared +} // namespace Common +} // namespace Filters +} // namespace Extensions +} // namespace Envoy diff --git a/contrib/istio/filters/network/peer_metadata/source/BUILD b/contrib/istio/filters/network/peer_metadata/source/BUILD index 0328e316815a8..3aaf3b69646dd 100644 --- a/contrib/istio/filters/network/peer_metadata/source/BUILD +++ b/contrib/istio/filters/network/peer_metadata/source/BUILD @@ -14,14 +14,17 @@ envoy_cc_contrib_extension( hdrs = ["peer_metadata.h"], deps = [ "//contrib/istio/filters/common/source:metadata_object_lib", + "//contrib/istio/filters/common/source:peer_metadata_registry_lib", "//envoy/local_info:local_info_interface", "//envoy/network:address_interface", "//envoy/network:filter_interface", "//envoy/server:filter_config_interface", + "//envoy/stream_info:uint64_accessor_interface", "//source/common/common:minimal_logger_lib", "//source/common/router:string_accessor_lib", "//source/common/singleton:const_singleton", "//source/common/stream_info:bool_accessor_lib", + "//source/common/stream_info:uint64_accessor_lib", "//source/common/tcp_proxy", "//source/extensions/filters/common/expr:cel_state_lib", "//source/extensions/filters/network/common:factory_base_lib", diff --git a/contrib/istio/filters/network/peer_metadata/source/peer_metadata.cc b/contrib/istio/filters/network/peer_metadata/source/peer_metadata.cc index 781e6b39d25b4..98fc7ac838fad 100644 --- a/contrib/istio/filters/network/peer_metadata/source/peer_metadata.cc +++ b/contrib/istio/filters/network/peer_metadata/source/peer_metadata.cc @@ -4,8 +4,11 @@ #include "source/common/router/string_accessor_impl.h" #include "source/common/stream_info/bool_accessor_impl.h" +#include "source/common/stream_info/uint64_accessor_impl.h" #include "source/common/tcp_proxy/tcp_proxy.h" +#include "envoy/stream_info/uint64_accessor.h" + namespace Envoy { namespace Extensions { namespace NetworkFilters { @@ -47,10 +50,9 @@ bool discoveryDisabled(const ::envoy::config::core::v3::Metadata& metadata) { } // namespace -const uint32_t PeerMetadataHeader::magic_number = 0xabcd1234; - -Filter::Filter(const Config& config, const LocalInfo::LocalInfo& local_info) - : config_(config), baggage_(baggageValue(local_info)) {} +Filter::Filter(const Config& config, const LocalInfo::LocalInfo& local_info, + Filters::Common::PeerMetadataShared::PeerMetadataRegistrySharedPtr registry) + : config_(config), baggage_(baggageValue(local_info)), registry_(std::move(registry)) {} Network::FilterStatus Filter::onData(Buffer::Instance&, bool) { return Network::FilterStatus::Continue; @@ -58,6 +60,13 @@ Network::FilterStatus Filter::onData(Buffer::Instance&, bool) { Network::FilterStatus Filter::onNewConnection() { ENVOY_LOG(trace, "New connection from downstream"); + // Store this connection's ID in filter state so the listener-side filter + // can use it as a key in the thread-local registry. + read_callbacks_->connection().streamInfo().filterState()->setData( + Filters::Common::PeerMetadataShared::ConnectionIdFilterStateKey, + std::make_shared(read_callbacks_->connection().id()), + StreamInfo::FilterState::LifeSpan::Connection, + StreamInfo::StreamSharingMayImpactPooling::None); populateBaggage(); if (disableDiscovery()) { state_ = PeerMetadataState::PassThrough; @@ -80,9 +89,7 @@ Network::FilterStatus Filter::onWrite(Buffer::Instance& buffer, bool) { // no peer metadata available, we can give up waiting for it. std::optional peer_metadata = discoverPeerMetadata(); if (peer_metadata) { - propagatePeerMetadata(*peer_metadata); - } else { - propagateNoPeerMetadata(); + storeInRegistry(*peer_metadata); } state_ = PeerMetadataState::PassThrough; break; @@ -168,55 +175,42 @@ std::optional Filter::discoverPeerMetadata() { return wrapped; } -void Filter::propagatePeerMetadata(const Envoy::Protobuf::Any& peer_metadata) { - ENVOY_LOG(trace, "Sending peer metadata downstream with the data stream"); - ASSERT(write_callbacks_); - - if (state_ != PeerMetadataState::WaitingForData) { - // It's only safe and correct to send the peer metadata downstream with - // the data if we haven't done that already, otherwise the downstream - // could be very confused by the data they received. - ENVOY_LOG(trace, "Filter has already sent the peer metadata downstream"); +void Filter::storeInRegistry(const Envoy::Protobuf::Any& peer_metadata) { + if (registry_ == nullptr || read_callbacks_ == nullptr) { return; } - - std::string data = peer_metadata.SerializeAsString(); - PeerMetadataHeader header{PeerMetadataHeader::magic_number, static_cast(data.size())}; - - Buffer::OwnedImpl buffer{ - absl::string_view(reinterpret_cast(&header), sizeof(header))}; - buffer.add(data); - write_callbacks_->injectWriteDataToFilterChain(buffer, false); -} - -void Filter::propagateNoPeerMetadata() { - ENVOY_LOG(trace, "Sending no peer metadata downstream with the data"); - ASSERT(write_callbacks_); - - PeerMetadataHeader header{PeerMetadataHeader::magic_number, 0}; - Buffer::OwnedImpl buffer{ - absl::string_view(reinterpret_cast(&header), sizeof(header))}; - write_callbacks_->injectWriteDataToFilterChain(buffer, false); + // Read the upstream connection ID that was passed through PassthroughState. + const auto* connection_id = + read_callbacks_->connection().streamInfo().filterState()->getDataReadOnly< + StreamInfo::UInt64Accessor>( + Filters::Common::PeerMetadataShared::ConnectionIdFilterStateKey); + if (connection_id == nullptr) { + ENVOY_LOG(trace, "No upstream connection ID in filter state, cannot store in registry"); + return; + } + std::string serialized = peer_metadata.SerializeAsString(); + registry_->setValue(connection_id->value(), serialized); + ENVOY_LOG(trace, "Stored peer metadata in registry for connection ID {}", + connection_id->value()); } -UpstreamFilter::UpstreamFilter() = default; - -Network::FilterStatus UpstreamFilter::onData(Buffer::Instance& buffer, bool end_stream) { - ENVOY_LOG(trace, "Read {} bytes from the upstream connection", buffer.length()); +UpstreamFilter::UpstreamFilter( + Filters::Common::PeerMetadataShared::PeerMetadataRegistrySharedPtr registry) + : registry_(std::move(registry)) {} +Network::FilterStatus UpstreamFilter::onData(Buffer::Instance&, bool) { switch (state_) { case PeerMetadataState::WaitingForData: if (disableDiscovery()) { state_ = PeerMetadataState::PassThrough; break; } - if (consumePeerMetadata(buffer, end_stream)) { + if (tryRegistryLookup()) { state_ = PeerMetadataState::PassThrough; } else { - // If we got here it means that we are waiting for more data to arrive. - // NOTE: if error happened, we will not get here, consumePeerMetadata - // will just return true and we will enter PassThrough state. - return Network::FilterStatus::StopIteration; + ENVOY_LOG(trace, "No peer metadata available in registry"); + populateNoPeerMetadata(); + state_ = PeerMetadataState::PassThrough; } break; default: @@ -277,79 +271,42 @@ bool UpstreamFilter::disableDiscovery() const { return false; } -bool UpstreamFilter::consumePeerMetadata(Buffer::Instance& buffer, bool end_stream) { - ENVOY_LOG(trace, "Trying to consume peer metadata from the data stream"); - using namespace ::Istio::Common; - - ASSERT(callbacks_); - - if (state_ != PeerMetadataState::WaitingForData) { - ENVOY_LOG(trace, "The filter already consumed peer metadata from the data stream"); - return true; - } - - if (buffer.length() < sizeof(PeerMetadataHeader)) { - if (end_stream) { - ENVOY_LOG(trace, "Not enough data in the data stream for peer metadata header and no more " - "data is coming"); - populateNoPeerMetadata(); - return true; - } - ENVOY_LOG(trace, - "Not enough data in the data stream for peer metadata header, waiting for more data"); +bool UpstreamFilter::tryRegistryLookup() { + if (registry_ == nullptr || callbacks_ == nullptr) { return false; } - - PeerMetadataHeader header; - buffer.copyOut(0, sizeof(PeerMetadataHeader), &header); - - if (header.magic != PeerMetadataHeader::magic_number) { - ENVOY_LOG(trace, "Magic number in the peer metadata header didn't match expected value"); - populateNoPeerMetadata(); - return true; - } - - if (header.data_size == 0) { - ENVOY_LOG(trace, "Peer metadata is empty"); - populateNoPeerMetadata(); - buffer.drain(sizeof(PeerMetadataHeader)); - return true; - } - - const size_t peer_metadata_size = sizeof(PeerMetadataHeader) + header.data_size; - - if (buffer.length() < peer_metadata_size) { - if (end_stream) { - ENVOY_LOG(trace, - "Not enough data in the data stream for peer metadata and no more data is coming"); - populateNoPeerMetadata(); - return true; - } - ENVOY_LOG(trace, "Not enough data in the data stream for peer metadata, waiting for more data"); + // Use this connection's ID as the registry key — the listener-side Filter + // stored peer metadata under the same key after receiving it via + // PassthroughState. + uint64_t conn_id = callbacks_->connection().id(); + auto value = registry_->getValue(conn_id); + if (!value.has_value()) { + ENVOY_LOG(trace, "No peer metadata in registry for connection ID {}", conn_id); return false; } - absl::string_view data{static_cast(buffer.linearize(peer_metadata_size)), - peer_metadata_size}; - data = data.substr(sizeof(PeerMetadataHeader)); + ENVOY_LOG(trace, "Found peer metadata in registry for connection ID {}", conn_id); + // Remove the entry to avoid leaking memory. + registry_->removeValue(conn_id); + Envoy::Protobuf::Any any; - if (!any.ParseFromArray(data.data(), data.size())) { - ENVOY_LOG(trace, "Failed to parse peer metadata proto from the data stream"); + if (!any.ParseFromString(*value)) { + ENVOY_LOG(trace, "Failed to parse peer metadata from registry"); populateNoPeerMetadata(); return true; } Envoy::Protobuf::Struct peer_metadata; if (!any.UnpackTo(&peer_metadata)) { - ENVOY_LOG(trace, "Failed to unpack peer metadata struct"); + ENVOY_LOG(trace, "Failed to unpack peer metadata struct from registry"); populateNoPeerMetadata(); return true; } - std::unique_ptr workload = convertStructToWorkloadMetadata(peer_metadata); + std::unique_ptr<::Istio::Common::WorkloadMetadataObject> workload = + ::Istio::Common::convertStructToWorkloadMetadata(peer_metadata); populatePeerMetadata(*workload); - buffer.drain(peer_metadata_size); - ENVOY_LOG(trace, "Successfully consumed peer metadata from the data stream"); + ENVOY_LOG(trace, "Successfully retrieved peer metadata from registry"); return true; } @@ -396,15 +353,22 @@ ConfigFactory::ConfigFactory() absl::StatusOr ConfigFactory::createFilterFactoryFromProtoTyped(const Config& config, Server::Configuration::FactoryContext& context) { - return [config, &context](Network::FilterManager& filter_manager) -> void { + auto registry = + Filters::Common::PeerMetadataShared::getRegistry(context.serverFactoryContext()); + return [config, &context, + registry = std::move(registry)](Network::FilterManager& filter_manager) -> void { const auto& local_info = context.serverFactoryContext().localInfo(); - filter_manager.addFilter(std::make_shared(config, local_info)); + filter_manager.addFilter(std::make_shared(config, local_info, registry)); }; } Network::FilterFactoryCb UpstreamConfigFactory::createFilterFactoryFromProto( - const Protobuf::Message& config, Server::Configuration::UpstreamFactoryContext&) { - return createFilterFactory(dynamic_cast(config)); + const Protobuf::Message&, Server::Configuration::UpstreamFactoryContext& context) { + auto registry = + Filters::Common::PeerMetadataShared::getRegistry(context.serverFactoryContext()); + return [registry = std::move(registry)](Network::FilterManager& filter_manager) -> void { + filter_manager.addReadFilter(std::make_shared(registry)); + }; } ProtobufTypes::MessagePtr UpstreamConfigFactory::createEmptyConfigProto() { @@ -423,12 +387,6 @@ bool UpstreamConfigFactory::isTerminalFilterByProto(const Protobuf::Message&, return true; } -Network::FilterFactoryCb UpstreamConfigFactory::createFilterFactory(const UpstreamConfig&) { - return [](Network::FilterManager& filter_manager) -> void { - filter_manager.addReadFilter(std::make_shared()); - }; -} - namespace { REGISTER_FACTORY(ConfigFactory, Server::Configuration::NamedNetworkFilterConfigFactory); diff --git a/contrib/istio/filters/network/peer_metadata/source/peer_metadata.h b/contrib/istio/filters/network/peer_metadata/source/peer_metadata.h index 6c136e8caec31..f461b54870db2 100644 --- a/contrib/istio/filters/network/peer_metadata/source/peer_metadata.h +++ b/contrib/istio/filters/network/peer_metadata/source/peer_metadata.h @@ -1,5 +1,6 @@ #pragma once +#include #include #include @@ -12,6 +13,8 @@ #include "source/extensions/filters/common/expr/cel_state.h" #include "source/extensions/filters/network/common/factory_base.h" +#include "contrib/istio/filters/common/source/peer_metadata_registry.h" + #include "contrib/envoy/extensions/filters/network/peer_metadata/v3/peer_metadata.pb.h" #include "contrib/envoy/extensions/filters/network/peer_metadata/v3/peer_metadata.pb.validate.h" #include "contrib/istio/filters/common/source/metadata_object.h" @@ -111,23 +114,19 @@ enum class PeerMetadataState { PassThrough, }; -PACKED_STRUCT(struct PeerMetadataHeader { - uint32_t magic; - static const uint32_t magic_number; - uint32_t data_size; -}); - /** * This is a regular network filter that will be installed in the * connect_originate or inner_connect_originate filter chains. It will take * baggage header information from filter state (we expect TCP Proxy to * populate it), collect other details that are missing from the baggage, i.e. - * the upstream peer principle, encode those details into a sequence of bytes - * and will inject it dowstream. + * the upstream peer identity, and store the discovered peer metadata in a + * thread-local registry keyed by the upstream connection ID, allowing the + * upstream peer_metadata filter on the cluster side to retrieve it directly. */ class Filter : public Network::Filter, Logger::Loggable { public: - Filter(const Config& config, const LocalInfo::LocalInfo& local_info); + Filter(const Config& config, const LocalInfo::LocalInfo& local_info, + Filters::Common::PeerMetadataShared::PeerMetadataRegistrySharedPtr registry); // Network::ReadFilter Network::FilterStatus onNewConnection() override; @@ -142,35 +141,27 @@ class Filter : public Network::Filter, Logger::Loggable { void populateBaggage(); bool disableDiscovery() const; std::optional discoverPeerMetadata(); - void propagatePeerMetadata(const Envoy::Protobuf::Any& peer_metadata); - void propagateNoPeerMetadata(); + void storeInRegistry(const Envoy::Protobuf::Any& peer_metadata); PeerMetadataState state_ = PeerMetadataState::WaitingForData; Network::WriteFilterCallbacks* write_callbacks_{}; Network::ReadFilterCallbacks* read_callbacks_{}; Config config_; std::string baggage_; + Filters::Common::PeerMetadataShared::PeerMetadataRegistrySharedPtr registry_; }; /** * This is an upstream network filter complementing the filter above. It will - * be installed in all the service clusters that may use HBONE (or double - * HBONE) to communicate with the upstream peers and it will parse and remove - * the data injected by the filter above. The parsed peer metadata details will - * be saved in the filter state. - * - * NOTE: This filter has built-in safety checks that would prevent it from - * trying to interpret the actual connection data as peer metadata injected - * by the filter above. However, those checks are rather shallow and rely on a - * bunch of implicit assumptions (i.e., the magic number does not match - * accidentally, the upstream host actually sends back some data that we can - * check, etc). What I'm trying to say is that in correct setup we don't need - * to rely on those checks for correctness and if it's not the case, then we - * definitely have a bug. + * be installed in all the service clusters that communicate with upstream peers + * via internal listeners. It reads peer metadata from the thread-local registry + * (stored by the listener-side Filter keyed by this connection's ID) and + * populates filter state for use by telemetry filters. */ class UpstreamFilter : public Network::ReadFilter, Logger::Loggable { public: - UpstreamFilter(); + explicit UpstreamFilter( + Filters::Common::PeerMetadataShared::PeerMetadataRegistrySharedPtr registry); // Network::ReadFilter Network::FilterStatus onData(Buffer::Instance& buffer, bool end_stream) override; @@ -179,7 +170,7 @@ class UpstreamFilter : public Network::ReadFilter, Logger::Loggable(config, local_info_); + filter_ = std::make_unique(config, local_info_, nullptr); filter_->initializeReadFilterCallbacks(read_filter_callbacks_); filter_->initializeWriteFilterCallbacks(write_filter_callbacks_); } @@ -336,7 +336,7 @@ class PeerMetadataUpstreamFilterTest : public ::testing::Test { host_metadata_ = std::make_shared<::envoy::config::core::v3::Metadata>(); - filter_ = std::make_unique(); + filter_ = std::make_unique(nullptr); filter_->initializeReadFilterCallbacks(callbacks_); } From 29dbf7e7329bb0da43d4b5ec4306a2a7b559c0db Mon Sep 17 00:00:00 2001 From: Gustavo Date: Wed, 8 Jul 2026 16:46:28 +0000 Subject: [PATCH 2/5] reverts peer_metadata_test Signed-off-by: Gustavo --- .../filters/network/peer_metadata/test/peer_metadata_test.cc | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/contrib/istio/filters/network/peer_metadata/test/peer_metadata_test.cc b/contrib/istio/filters/network/peer_metadata/test/peer_metadata_test.cc index e971a5d380467..93c7360dfeb58 100644 --- a/contrib/istio/filters/network/peer_metadata/test/peer_metadata_test.cc +++ b/contrib/istio/filters/network/peer_metadata/test/peer_metadata_test.cc @@ -124,7 +124,7 @@ class PeerMetadataFilterTest : public ::testing::Test { ON_CALL(Const(write_filter_callbacks_.connection_), streamInfo()) .WillByDefault(ReturnRef(stream_info_)); - filter_ = std::make_unique(config, local_info_, nullptr); + filter_ = std::make_unique(config, local_info_); filter_->initializeReadFilterCallbacks(read_filter_callbacks_); filter_->initializeWriteFilterCallbacks(write_filter_callbacks_); } @@ -336,7 +336,7 @@ class PeerMetadataUpstreamFilterTest : public ::testing::Test { host_metadata_ = std::make_shared<::envoy::config::core::v3::Metadata>(); - filter_ = std::make_unique(nullptr); + filter_ = std::make_unique(); filter_->initializeReadFilterCallbacks(callbacks_); } From dbb16cfb75196d79861ca8225ad138d29c09478e Mon Sep 17 00:00:00 2001 From: Gustavo Date: Wed, 15 Jul 2026 14:25:29 +0000 Subject: [PATCH 3/5] using filter md from set_filter_state filter Signed-off-by: Gustavo --- .../common/source/peer_metadata_registry.cc | 10 ++--- .../common/source/peer_metadata_registry.h | 13 ++---- .../peer_metadata/source/peer_metadata.cc | 40 +++++++++---------- 3 files changed, 27 insertions(+), 36 deletions(-) diff --git a/contrib/istio/filters/common/source/peer_metadata_registry.cc b/contrib/istio/filters/common/source/peer_metadata_registry.cc index 120ed9d1121c4..659898362908a 100644 --- a/contrib/istio/filters/common/source/peer_metadata_registry.cc +++ b/contrib/istio/filters/common/source/peer_metadata_registry.cc @@ -20,7 +20,7 @@ namespace PeerMetadataShared { namespace { struct ThreadLocalRegistry : public ThreadLocal::ThreadLocalObject { - absl::flat_hash_map values_; + absl::flat_hash_map values_; }; class PeerMetadataRegistryImpl : public PeerMetadataRegistry, public Singleton::Instance { @@ -30,13 +30,13 @@ class PeerMetadataRegistryImpl : public PeerMetadataRegistry, public Singleton:: tls_->set([](Event::Dispatcher&) { return std::make_shared(); }); } - void setValue(uint64_t key, const std::string& value) override { + void setValue(absl::string_view key, const std::string& value) override { if (auto tls = tls_->get(); tls.has_value()) { - tls->values_[key] = value; + tls->values_[std::string(key)] = value; } } - absl::optional getValue(uint64_t key) const override { + absl::optional getValue(absl::string_view key) const override { if (auto tls = tls_->get(); tls.has_value()) { const auto it = tls->values_.find(key); if (it != tls->values_.end()) { @@ -46,7 +46,7 @@ class PeerMetadataRegistryImpl : public PeerMetadataRegistry, public Singleton:: return absl::nullopt; } - void removeValue(uint64_t key) override { + void removeValue(absl::string_view key) override { if (auto tls = tls_->get(); tls.has_value()) { tls->values_.erase(key); } diff --git a/contrib/istio/filters/common/source/peer_metadata_registry.h b/contrib/istio/filters/common/source/peer_metadata_registry.h index a8e7ef46eedbc..fff1862122f22 100644 --- a/contrib/istio/filters/common/source/peer_metadata_registry.h +++ b/contrib/istio/filters/common/source/peer_metadata_registry.h @@ -29,15 +29,11 @@ class PeerMetadataRegistry { public: virtual ~PeerMetadataRegistry() = default; - // Store `value` under `key` in the current worker thread's store. - virtual void setValue(uint64_t key, const std::string& value) PURE; + virtual void setValue(absl::string_view key, const std::string& value) PURE; - // Look up `key` in the current worker thread's store. Returns absl::nullopt - // if no value was stored (on this thread) under `key`. - virtual absl::optional getValue(uint64_t key) const PURE; + virtual absl::optional getValue(absl::string_view key) const PURE; - // Remove the entry for `key` from the current worker thread's store. - virtual void removeValue(uint64_t key) PURE; + virtual void removeValue(absl::string_view key) PURE; }; using PeerMetadataRegistrySharedPtr = std::shared_ptr; @@ -47,9 +43,8 @@ using PeerMetadataRegistrySharedPtr = std::shared_ptr; PeerMetadataRegistrySharedPtr getRegistry(Server::Configuration::ServerFactoryContext& context); -// Filter state key under which the originating upstream connection ID is stored. constexpr absl::string_view ConnectionIdFilterStateKey = - "envoy.peer_metadata.upstream_connection_id"; + "envoy.peer_metadata.downstream_connection_id"; } // namespace PeerMetadataShared } // namespace Common diff --git a/contrib/istio/filters/network/peer_metadata/source/peer_metadata.cc b/contrib/istio/filters/network/peer_metadata/source/peer_metadata.cc index 98fc7ac838fad..d771a86f5c25f 100644 --- a/contrib/istio/filters/network/peer_metadata/source/peer_metadata.cc +++ b/contrib/istio/filters/network/peer_metadata/source/peer_metadata.cc @@ -60,13 +60,6 @@ Network::FilterStatus Filter::onData(Buffer::Instance&, bool) { Network::FilterStatus Filter::onNewConnection() { ENVOY_LOG(trace, "New connection from downstream"); - // Store this connection's ID in filter state so the listener-side filter - // can use it as a key in the thread-local registry. - read_callbacks_->connection().streamInfo().filterState()->setData( - Filters::Common::PeerMetadataShared::ConnectionIdFilterStateKey, - std::make_shared(read_callbacks_->connection().id()), - StreamInfo::FilterState::LifeSpan::Connection, - StreamInfo::StreamSharingMayImpactPooling::None); populateBaggage(); if (disableDiscovery()) { state_ = PeerMetadataState::PassThrough; @@ -177,21 +170,19 @@ std::optional Filter::discoverPeerMetadata() { void Filter::storeInRegistry(const Envoy::Protobuf::Any& peer_metadata) { if (registry_ == nullptr || read_callbacks_ == nullptr) { + ENVOY_LOG(error, "No instance for registry or callbacks"); return; } - // Read the upstream connection ID that was passed through PassthroughState. const auto* connection_id = read_callbacks_->connection().streamInfo().filterState()->getDataReadOnly< - StreamInfo::UInt64Accessor>( + Router::StringAccessor>( Filters::Common::PeerMetadataShared::ConnectionIdFilterStateKey); if (connection_id == nullptr) { - ENVOY_LOG(trace, "No upstream connection ID in filter state, cannot store in registry"); + ENVOY_LOG(debug, "No upstream connection ID in filter state, cannot store in TLS registry"); return; } std::string serialized = peer_metadata.SerializeAsString(); - registry_->setValue(connection_id->value(), serialized); - ENVOY_LOG(trace, "Stored peer metadata in registry for connection ID {}", - connection_id->value()); + registry_->setValue(connection_id->asString(), serialized); } UpstreamFilter::UpstreamFilter( @@ -275,30 +266,35 @@ bool UpstreamFilter::tryRegistryLookup() { if (registry_ == nullptr || callbacks_ == nullptr) { return false; } - // Use this connection's ID as the registry key — the listener-side Filter - // stored peer metadata under the same key after receiving it via - // PassthroughState. - uint64_t conn_id = callbacks_->connection().id(); + // UpstreamFilter reads the downstream connection ID shared via transitive + // filter state and use it as the registry key, the listener-side Filter + // stores peer metadata under the same key. + const auto* connection_id = + callbacks_->connection().streamInfo().filterState()->getDataReadOnly( + Filters::Common::PeerMetadataShared::ConnectionIdFilterStateKey); + if (connection_id == nullptr) { + ENVOY_LOG(debug, "No downstream connection ID in filter state"); + return false; + } + absl::string_view conn_id = connection_id->asString(); auto value = registry_->getValue(conn_id); if (!value.has_value()) { - ENVOY_LOG(trace, "No peer metadata in registry for connection ID {}", conn_id); + ENVOY_LOG(debug, "No peer metadata in registry for connection ID {}", conn_id); return false; } - ENVOY_LOG(trace, "Found peer metadata in registry for connection ID {}", conn_id); - // Remove the entry to avoid leaking memory. registry_->removeValue(conn_id); Envoy::Protobuf::Any any; if (!any.ParseFromString(*value)) { - ENVOY_LOG(trace, "Failed to parse peer metadata from registry"); + ENVOY_LOG(error, "Failed to parse peer metadata from registry"); populateNoPeerMetadata(); return true; } Envoy::Protobuf::Struct peer_metadata; if (!any.UnpackTo(&peer_metadata)) { - ENVOY_LOG(trace, "Failed to unpack peer metadata struct from registry"); + ENVOY_LOG(error, "Failed to unpack peer metadata struct from registry"); populateNoPeerMetadata(); return true; } From c98d7d42c228dab9256ff0fa92ddc05ef80c3328 Mon Sep 17 00:00:00 2001 From: Gustavo Date: Wed, 15 Jul 2026 15:42:09 +0000 Subject: [PATCH 4/5] format fix Signed-off-by: Gustavo --- .../common/source/peer_metadata_registry.cc | 11 ++++------ .../common/source/peer_metadata_registry.h | 6 ++--- .../peer_metadata/source/peer_metadata.cc | 22 +++++++++---------- .../peer_metadata/source/peer_metadata.h | 3 +-- 4 files changed, 18 insertions(+), 24 deletions(-) diff --git a/contrib/istio/filters/common/source/peer_metadata_registry.cc b/contrib/istio/filters/common/source/peer_metadata_registry.cc index 659898362908a..77790b4b1f36b 100644 --- a/contrib/istio/filters/common/source/peer_metadata_registry.cc +++ b/contrib/istio/filters/common/source/peer_metadata_registry.cc @@ -36,14 +36,14 @@ class PeerMetadataRegistryImpl : public PeerMetadataRegistry, public Singleton:: } } - absl::optional getValue(absl::string_view key) const override { + std::optional getValue(absl::string_view key) const override { if (auto tls = tls_->get(); tls.has_value()) { const auto it = tls->values_.find(key); if (it != tls->values_.end()) { return it->second; } } - return absl::nullopt; + return std::nullopt; } void removeValue(absl::string_view key) override { @@ -60,13 +60,10 @@ class PeerMetadataRegistryImpl : public PeerMetadataRegistry, public Singleton:: SINGLETON_MANAGER_REGISTRATION(peer_metadata_registry); -PeerMetadataRegistrySharedPtr -getRegistry(Server::Configuration::ServerFactoryContext& context) { +PeerMetadataRegistrySharedPtr getRegistry(Server::Configuration::ServerFactoryContext& context) { return context.singletonManager().getTyped( SINGLETON_MANAGER_REGISTERED_NAME(peer_metadata_registry), - [&context] { - return std::make_shared(context.threadLocal()); - }, + [&context] { return std::make_shared(context.threadLocal()); }, /*pin=*/true); } diff --git a/contrib/istio/filters/common/source/peer_metadata_registry.h b/contrib/istio/filters/common/source/peer_metadata_registry.h index fff1862122f22..534cd2944920c 100644 --- a/contrib/istio/filters/common/source/peer_metadata_registry.h +++ b/contrib/istio/filters/common/source/peer_metadata_registry.h @@ -7,7 +7,6 @@ #include "envoy/server/factory_context.h" #include "absl/strings/string_view.h" -#include "absl/types/optional.h" namespace Envoy { namespace Extensions { @@ -31,7 +30,7 @@ class PeerMetadataRegistry { virtual void setValue(absl::string_view key, const std::string& value) PURE; - virtual absl::optional getValue(absl::string_view key) const PURE; + virtual std::optional getValue(absl::string_view key) const PURE; virtual void removeValue(absl::string_view key) PURE; }; @@ -40,8 +39,7 @@ using PeerMetadataRegistrySharedPtr = std::shared_ptr; // Returns the process-wide PeerMetadataRegistry singleton, creating it if // necessary. The registry is pinned in the singleton manager. -PeerMetadataRegistrySharedPtr -getRegistry(Server::Configuration::ServerFactoryContext& context); +PeerMetadataRegistrySharedPtr getRegistry(Server::Configuration::ServerFactoryContext& context); constexpr absl::string_view ConnectionIdFilterStateKey = "envoy.peer_metadata.downstream_connection_id"; diff --git a/contrib/istio/filters/network/peer_metadata/source/peer_metadata.cc b/contrib/istio/filters/network/peer_metadata/source/peer_metadata.cc index d771a86f5c25f..a46d3cfe2a1fd 100644 --- a/contrib/istio/filters/network/peer_metadata/source/peer_metadata.cc +++ b/contrib/istio/filters/network/peer_metadata/source/peer_metadata.cc @@ -2,13 +2,13 @@ #include +#include "envoy/stream_info/uint64_accessor.h" + #include "source/common/router/string_accessor_impl.h" #include "source/common/stream_info/bool_accessor_impl.h" #include "source/common/stream_info/uint64_accessor_impl.h" #include "source/common/tcp_proxy/tcp_proxy.h" -#include "envoy/stream_info/uint64_accessor.h" - namespace Envoy { namespace Extensions { namespace NetworkFilters { @@ -174,9 +174,11 @@ void Filter::storeInRegistry(const Envoy::Protobuf::Any& peer_metadata) { return; } const auto* connection_id = - read_callbacks_->connection().streamInfo().filterState()->getDataReadOnly< - Router::StringAccessor>( - Filters::Common::PeerMetadataShared::ConnectionIdFilterStateKey); + read_callbacks_->connection() + .streamInfo() + .filterState() + ->getDataReadOnly( + Filters::Common::PeerMetadataShared::ConnectionIdFilterStateKey); if (connection_id == nullptr) { ENVOY_LOG(debug, "No upstream connection ID in filter state, cannot store in TLS registry"); return; @@ -266,8 +268,8 @@ bool UpstreamFilter::tryRegistryLookup() { if (registry_ == nullptr || callbacks_ == nullptr) { return false; } - // UpstreamFilter reads the downstream connection ID shared via transitive - // filter state and use it as the registry key, the listener-side Filter + // UpstreamFilter reads the downstream connection ID shared via transitive + // filter state and use it as the registry key, the listener-side Filter // stores peer metadata under the same key. const auto* connection_id = callbacks_->connection().streamInfo().filterState()->getDataReadOnly( @@ -349,8 +351,7 @@ ConfigFactory::ConfigFactory() absl::StatusOr ConfigFactory::createFilterFactoryFromProtoTyped(const Config& config, Server::Configuration::FactoryContext& context) { - auto registry = - Filters::Common::PeerMetadataShared::getRegistry(context.serverFactoryContext()); + auto registry = Filters::Common::PeerMetadataShared::getRegistry(context.serverFactoryContext()); return [config, &context, registry = std::move(registry)](Network::FilterManager& filter_manager) -> void { const auto& local_info = context.serverFactoryContext().localInfo(); @@ -360,8 +361,7 @@ ConfigFactory::createFilterFactoryFromProtoTyped(const Config& config, Network::FilterFactoryCb UpstreamConfigFactory::createFilterFactoryFromProto( const Protobuf::Message&, Server::Configuration::UpstreamFactoryContext& context) { - auto registry = - Filters::Common::PeerMetadataShared::getRegistry(context.serverFactoryContext()); + auto registry = Filters::Common::PeerMetadataShared::getRegistry(context.serverFactoryContext()); return [registry = std::move(registry)](Network::FilterManager& filter_manager) -> void { filter_manager.addReadFilter(std::make_shared(registry)); }; diff --git a/contrib/istio/filters/network/peer_metadata/source/peer_metadata.h b/contrib/istio/filters/network/peer_metadata/source/peer_metadata.h index f461b54870db2..233b357dd838a 100644 --- a/contrib/istio/filters/network/peer_metadata/source/peer_metadata.h +++ b/contrib/istio/filters/network/peer_metadata/source/peer_metadata.h @@ -13,11 +13,10 @@ #include "source/extensions/filters/common/expr/cel_state.h" #include "source/extensions/filters/network/common/factory_base.h" -#include "contrib/istio/filters/common/source/peer_metadata_registry.h" - #include "contrib/envoy/extensions/filters/network/peer_metadata/v3/peer_metadata.pb.h" #include "contrib/envoy/extensions/filters/network/peer_metadata/v3/peer_metadata.pb.validate.h" #include "contrib/istio/filters/common/source/metadata_object.h" +#include "contrib/istio/filters/common/source/peer_metadata_registry.h" /** * PeerMetadata network and upstream network filters are used in one of ambient From de2c3e53f3197295db5d5dfdfcc1be40b2afbef3 Mon Sep 17 00:00:00 2001 From: Gustavo Date: Thu, 16 Jul 2026 11:38:59 +0000 Subject: [PATCH 5/5] unit tests Signed-off-by: Gustavo --- .../filters/network/peer_metadata/test/BUILD | 2 + .../peer_metadata/test/peer_metadata_test.cc | 226 +++++++++--------- 2 files changed, 118 insertions(+), 110 deletions(-) diff --git a/contrib/istio/filters/network/peer_metadata/test/BUILD b/contrib/istio/filters/network/peer_metadata/test/BUILD index 800f5590be5cc..a6a5161ed69ac 100644 --- a/contrib/istio/filters/network/peer_metadata/test/BUILD +++ b/contrib/istio/filters/network/peer_metadata/test/BUILD @@ -12,8 +12,10 @@ envoy_cc_test( name = "peer_metadata_test", srcs = ["peer_metadata_test.cc"], deps = [ + "//contrib/istio/filters/common/source:peer_metadata_registry_lib", "//contrib/istio/filters/network/peer_metadata/source:config", "//envoy/router:string_accessor_interface", + "//source/common/router:string_accessor_lib", "//test/mocks/local_info:local_info_mocks", "//test/mocks/network:network_mocks", "//test/mocks/ssl:ssl_mocks", diff --git a/contrib/istio/filters/network/peer_metadata/test/peer_metadata_test.cc b/contrib/istio/filters/network/peer_metadata/test/peer_metadata_test.cc index 93c7360dfeb58..d6b6e2a550c6b 100644 --- a/contrib/istio/filters/network/peer_metadata/test/peer_metadata_test.cc +++ b/contrib/istio/filters/network/peer_metadata/test/peer_metadata_test.cc @@ -12,6 +12,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ +#include #include #include #include @@ -21,6 +22,7 @@ #include "source/common/http/utility.h" #include "source/common/network/address_impl.h" +#include "source/common/router/string_accessor_impl.h" #include "source/common/tcp_proxy/tcp_proxy.h" #include "test/mocks/local_info/mocks.h" @@ -29,6 +31,7 @@ #include "test/mocks/upstream/host.h" #include "absl/strings/str_cat.h" +#include "contrib/istio/filters/common/source/peer_metadata_registry.h" #include "contrib/istio/filters/network/peer_metadata/source/peer_metadata.h" #include "gmock/gmock.h" #include "gtest/gtest.h" @@ -44,26 +47,64 @@ constexpr absl::string_view defaultIdentity = "spiffe://cluster.local/ns/default constexpr absl::string_view defaultBaggage = "k8s.deployment.name=server,service.name=server-service,service.version=v2,k8s.namespace.name=" "default,k8s.cluster.name=cluster"; +constexpr absl::string_view defaultConnectionId = "1234"; using ::testing::Const; -using ::testing::Invoke; using ::testing::NiceMock; using ::testing::Return; using ::testing::ReturnRef; -bool parsePeerMetadataHeader(const std::string& data, PeerMetadataHeader& header) { - if (data.size() < sizeof(header)) { - return false; +// In-memory PeerMetadataRegistry used by tests to stand in for the thread-local +// registry that hands peer metadata from the listener-side filter to the +// upstream (cluster-side) filter. +class TestPeerMetadataRegistry : public Filters::Common::PeerMetadataShared::PeerMetadataRegistry { +public: + void setValue(absl::string_view key, const std::string& value) override { + store_[std::string(key)] = value; } - std::memcpy(&header, data.data(), sizeof(header)); - return true; + + std::optional getValue(absl::string_view key) const override { + const auto it = store_.find(std::string(key)); + if (it == store_.end()) { + return std::nullopt; + } + return it->second; + } + + void removeValue(absl::string_view key) override { store_.erase(std::string(key)); } + +private: + std::map store_; +}; + +// Serializes peer metadata the same way the listener-side filter stores it in +// the registry: a google.protobuf.Struct packed into an Any. +std::string encodePeerMetadata(absl::string_view baggage, absl::string_view identity) { + std::unique_ptr metadata = + Istio::Common::convertBaggageToWorkloadMetadata(baggage, identity); + Protobuf::Struct data = ::Istio::Common::convertWorkloadMetadataToStruct(*metadata); + Protobuf::Any wrapped; + std::ignore = wrapped.PackFrom(data); + return wrapped.SerializeAsString(); } -bool parsePeerMetadata(const std::string& data, PeerMetadataHeader& header, Protobuf::Any& any) { - if (!parsePeerMetadataHeader(data, header)) { - return false; +// Decodes a value stored in the registry back into a WorkloadMetadataObject. +std::optional +decodePeerMetadata(const std::string& serialized) { + Protobuf::Any any; + if (!any.ParseFromString(serialized)) { + return std::nullopt; + } + Protobuf::Struct data; + if (!any.UnpackTo(&data)) { + return std::nullopt; } - return any.ParseFromArray(data.data() + sizeof(header), data.size() - sizeof(header)); + std::unique_ptr metadata = + Istio::Common::convertStructToWorkloadMetadata(data); + if (!metadata) { + return std::nullopt; + } + return *metadata; } Config configWithBaggageKey(absl::string_view baggage_key) { @@ -124,17 +165,26 @@ class PeerMetadataFilterTest : public ::testing::Test { ON_CALL(Const(write_filter_callbacks_.connection_), streamInfo()) .WillByDefault(ReturnRef(stream_info_)); - filter_ = std::make_unique(config, local_info_); + filter_ = std::make_unique(config, local_info_, registry_); filter_->initializeReadFilterCallbacks(read_filter_callbacks_); filter_->initializeWriteFilterCallbacks(write_filter_callbacks_); } + void setConnectionId(absl::string_view id) { + stream_info_.filterState()->setData( + Filters::Common::PeerMetadataShared::ConnectionIdFilterStateKey, + std::make_shared(id), + StreamInfo::FilterState::LifeSpan::Connection); + } + ::envoy::config::core::v3::Node node_metadata_; NiceMock local_info_; NiceMock stream_info_; NiceMock read_filter_callbacks_; NiceMock write_filter_callbacks_; std::vector sans_; + std::shared_ptr registry_ = + std::make_shared(); std::unique_ptr filter_; }; @@ -174,30 +224,18 @@ TEST_F(PeerMetadataFilterTest, TestPeerMetadataDiscoveredAndPropagated) { std::vector sans{identity}; populateUpstreamSans(sans); - std::string injected; - EXPECT_CALL(write_filter_callbacks_, injectWriteDataToFilterChain(_, false)) - .WillRepeatedly(Invoke([&injected](Buffer::Instance& buffer, bool) { - injected = std::string(static_cast(buffer.linearize(buffer.length())), - buffer.length()); - })); + setConnectionId(defaultConnectionId); EXPECT_EQ(filter_->onNewConnection(), Network::FilterStatus::Continue); Buffer::OwnedImpl data; EXPECT_EQ(filter_->onWrite(data, /*end_stream*/ false), Network::FilterStatus::Continue); - EXPECT_FALSE(injected.empty()); - - PeerMetadataHeader header; - Protobuf::Any any; - EXPECT_TRUE(parsePeerMetadata(injected, header, any)); - EXPECT_EQ(header.magic, PeerMetadataHeader::magic_number); - EXPECT_GT(header.data_size, 0); - Protobuf::Struct metadata; - EXPECT_TRUE(any.UnpackTo(&metadata)); + const auto stored = registry_->getValue(defaultConnectionId); + ASSERT_TRUE(stored.has_value()); - std::unique_ptr workload = - Istio::Common::convertStructToWorkloadMetadata(metadata); + const auto workload = decodePeerMetadata(*stored); + ASSERT_TRUE(workload.has_value()); EXPECT_EQ(workload->baggage(), baggage); EXPECT_EQ(workload->identity(), identity); } @@ -211,23 +249,14 @@ TEST_F(PeerMetadataFilterTest, TestNoFiltersStateFromTcpProxy) { std::vector sans{identity}; populateUpstreamSans(sans); - std::string injected; - EXPECT_CALL(write_filter_callbacks_, injectWriteDataToFilterChain(_, false)) - .WillRepeatedly(Invoke([&injected](Buffer::Instance& buffer, bool) { - injected = std::string(static_cast(buffer.linearize(buffer.length())), - buffer.length()); - })); + setConnectionId(defaultConnectionId); EXPECT_EQ(filter_->onNewConnection(), Network::FilterStatus::Continue); Buffer::OwnedImpl data; EXPECT_EQ(filter_->onWrite(data, /*end_stream*/ false), Network::FilterStatus::Continue); - EXPECT_FALSE(injected.empty()); - PeerMetadataHeader header; - EXPECT_TRUE(parsePeerMetadataHeader(injected, header)); - EXPECT_EQ(header.magic, PeerMetadataHeader::magic_number); - EXPECT_EQ(header.data_size, 0); + EXPECT_FALSE(registry_->getValue(defaultConnectionId).has_value()); } TEST_F(PeerMetadataFilterTest, TestNoBaggageHeaderFromTcpProxy) { @@ -243,23 +272,14 @@ TEST_F(PeerMetadataFilterTest, TestNoBaggageHeaderFromTcpProxy) { std::vector sans{identity}; populateUpstreamSans(sans); - std::string injected; - EXPECT_CALL(write_filter_callbacks_, injectWriteDataToFilterChain(_, false)) - .WillRepeatedly(Invoke([&injected](Buffer::Instance& buffer, bool) { - injected = std::string(static_cast(buffer.linearize(buffer.length())), - buffer.length()); - })); + setConnectionId(defaultConnectionId); EXPECT_EQ(filter_->onNewConnection(), Network::FilterStatus::Continue); Buffer::OwnedImpl data; EXPECT_EQ(filter_->onWrite(data, /*end_stream*/ false), Network::FilterStatus::Continue); - EXPECT_FALSE(injected.empty()); - PeerMetadataHeader header; - EXPECT_TRUE(parsePeerMetadataHeader(injected, header)); - EXPECT_EQ(header.magic, PeerMetadataHeader::magic_number); - EXPECT_EQ(header.data_size, 0); + EXPECT_FALSE(registry_->getValue(defaultConnectionId).has_value()); } TEST_F(PeerMetadataFilterTest, TestDisablePeerDiscovery) { @@ -275,21 +295,15 @@ TEST_F(PeerMetadataFilterTest, TestDisablePeerDiscovery) { std::vector sans{identity}; populateUpstreamSans(sans); + setConnectionId(defaultConnectionId); disablePeerDiscovery(); - std::string injected; - ON_CALL(write_filter_callbacks_, injectWriteDataToFilterChain(_, false)) - .WillByDefault(Invoke([&injected](Buffer::Instance& buffer, bool) { - injected = std::string(static_cast(buffer.linearize(buffer.length())), - buffer.length()); - })); - EXPECT_CALL(write_filter_callbacks_, injectWriteDataToFilterChain(_, false)).Times(0); - EXPECT_EQ(filter_->onNewConnection(), Network::FilterStatus::Continue); Buffer::OwnedImpl data; EXPECT_EQ(filter_->onWrite(data, /*end_stream*/ false), Network::FilterStatus::Continue); - EXPECT_TRUE(injected.empty()); + + EXPECT_FALSE(registry_->getValue(defaultConnectionId).has_value()); } std::shared_ptr> @@ -308,26 +322,6 @@ std::shared_ptr> makeIpv4Host(absl::stri return host; } -std::string encodePeerMetadataHeaderOnly(const PeerMetadataHeader& header) { - return std::string(absl::string_view(reinterpret_cast(&header), sizeof(header))); -} - -std::string encodeMetadataOnly(absl::string_view baggage, absl::string_view identity) { - std::unique_ptr metadata = - Istio::Common::convertBaggageToWorkloadMetadata(baggage, identity); - Protobuf::Struct data = ::Istio::Common::convertWorkloadMetadataToStruct(*metadata); - Protobuf::Any wrapped; - std::ignore = wrapped.PackFrom(data); - return wrapped.SerializeAsString(); -} - -std::string encodePeerMetadata(absl::string_view baggage, absl::string_view identity) { - std::string metadata = encodeMetadataOnly(baggage, identity); - PeerMetadataHeader header{PeerMetadataHeader::magic_number, - static_cast(metadata.size())}; - return encodePeerMetadataHeaderOnly(header) + metadata; -} - class PeerMetadataUpstreamFilterTest : public ::testing::Test { public: void initialize() { @@ -336,10 +330,17 @@ class PeerMetadataUpstreamFilterTest : public ::testing::Test { host_metadata_ = std::make_shared<::envoy::config::core::v3::Metadata>(); - filter_ = std::make_unique(); + filter_ = std::make_unique(registry_); filter_->initializeReadFilterCallbacks(callbacks_); } + void setConnectionId(absl::string_view id) { + stream_info_.filterState()->setData( + Filters::Common::PeerMetadataShared::ConnectionIdFilterStateKey, + std::make_shared(id), + StreamInfo::FilterState::LifeSpan::Connection); + } + ::envoy::config::core::v3::Metadata& hostMetadata() { return *host_metadata_; } ::envoy::config::core::v3::Metadata& clusterMetadata() { @@ -379,22 +380,29 @@ class PeerMetadataUpstreamFilterTest : public ::testing::Test { NiceMock callbacks_; std::shared_ptr> upstream_host_; std::shared_ptr<::envoy::config::core::v3::Metadata> host_metadata_; + std::shared_ptr registry_ = + std::make_shared(); std::unique_ptr filter_; }; TEST_F(PeerMetadataUpstreamFilterTest, TestPeerMetadataConsumedAndPropagated) { initialize(); setUpstreamHost(makeInternalListenerHost("connect_originate")); + setConnectionId(defaultConnectionId); + registry_->setValue(defaultConnectionId, encodePeerMetadata(defaultBaggage, defaultIdentity)); EXPECT_EQ(filter_->onNewConnection(), Network::FilterStatus::Continue); + const std::string payload{"application data"}; Buffer::OwnedImpl data; - data.add(encodePeerMetadata(defaultBaggage, defaultIdentity)); + data.add(payload); EXPECT_EQ(filter_->onData(data, /*end_stream*/ false), Network::FilterStatus::Continue); - EXPECT_EQ(data.length(), 0); + // The registry entry is consumed once read. + EXPECT_FALSE(registry_->getValue(defaultConnectionId).has_value()); + const auto peer_info = peerInfoFromFilterState(); - EXPECT_TRUE(peer_info.has_value()); + ASSERT_TRUE(peer_info.has_value()); EXPECT_EQ(peer_info->identity(), defaultIdentity); EXPECT_EQ(peer_info->baggage(), defaultBaggage); } @@ -402,14 +410,18 @@ TEST_F(PeerMetadataUpstreamFilterTest, TestPeerMetadataConsumedAndPropagated) { TEST_F(PeerMetadataUpstreamFilterTest, TestPeerMetadataNotConsumedForUnknownInternalListeners) { initialize(); setUpstreamHost(makeInternalListenerHost("not_connect_originate")); + setConnectionId(defaultConnectionId); + registry_->setValue(defaultConnectionId, encodePeerMetadata(defaultBaggage, defaultIdentity)); EXPECT_EQ(filter_->onNewConnection(), Network::FilterStatus::Continue); - std::string peer_metadata = encodePeerMetadata(defaultBaggage, defaultIdentity); + const std::string payload{"application data"}; Buffer::OwnedImpl data; - data.add(peer_metadata); + data.add(payload); EXPECT_EQ(filter_->onData(data, /*end_stream*/ false), Network::FilterStatus::Continue); - EXPECT_EQ(data.length(), peer_metadata.size()); + + // Internal listener is unknown, so the registry entry is left untouched. + EXPECT_TRUE(registry_->getValue(defaultConnectionId).has_value()); const auto peer_info = peerInfoFromFilterState(); EXPECT_FALSE(peer_info.has_value()); } @@ -417,14 +429,18 @@ TEST_F(PeerMetadataUpstreamFilterTest, TestPeerMetadataNotConsumedForUnknownInte TEST_F(PeerMetadataUpstreamFilterTest, TestPeerMetadataNotConsumedForExternalHosts) { initialize(); setUpstreamHost(makeIpv4Host("192.168.0.1")); + setConnectionId(defaultConnectionId); + registry_->setValue(defaultConnectionId, encodePeerMetadata(defaultBaggage, defaultIdentity)); EXPECT_EQ(filter_->onNewConnection(), Network::FilterStatus::Continue); - std::string peer_metadata = encodePeerMetadata(defaultBaggage, defaultIdentity); + const std::string payload{"application data"}; Buffer::OwnedImpl data; - data.add(peer_metadata); + data.add(payload); EXPECT_EQ(filter_->onData(data, /*end_stream*/ false), Network::FilterStatus::Continue); - EXPECT_EQ(data.length(), peer_metadata.size()); + + EXPECT_EQ(data.length(), payload.size()); + EXPECT_TRUE(registry_->getValue(defaultConnectionId).has_value()); const auto peer_info = peerInfoFromFilterState(); EXPECT_FALSE(peer_info.has_value()); } @@ -438,14 +454,17 @@ TEST_F(PeerMetadataUpstreamFilterTest, TestPeerMetadataNotConsumedWhenDisabledVi (*fields)[FilterNames::get().DisableDiscoveryField].set_bool_value(true); (*hostMetadata().mutable_filter_metadata())[FilterNames::get().Name].MergeFrom(metadata); + setConnectionId(defaultConnectionId); + registry_->setValue(defaultConnectionId, encodePeerMetadata(defaultBaggage, defaultIdentity)); + EXPECT_EQ(filter_->onNewConnection(), Network::FilterStatus::Continue); - std::string peer_metadata = encodePeerMetadata(defaultBaggage, defaultIdentity); + const std::string payload{"application data"}; Buffer::OwnedImpl data; - data.add(peer_metadata); + data.add(payload); EXPECT_EQ(filter_->onData(data, /*end_stream*/ false), Network::FilterStatus::Continue); - EXPECT_EQ(data.length(), peer_metadata.size()); + EXPECT_TRUE(registry_->getValue(defaultConnectionId).has_value()); const auto peer_info = peerInfoFromFilterState(); EXPECT_FALSE(peer_info.has_value()); } @@ -459,30 +478,17 @@ TEST_F(PeerMetadataUpstreamFilterTest, TestPeerMetadataNotConsumedWhenDisabledVi (*fields2)[FilterNames::get().DisableDiscoveryField].set_bool_value(true); (*clusterMetadata().mutable_filter_metadata())[FilterNames::get().Name].MergeFrom(metadata2); - EXPECT_EQ(filter_->onNewConnection(), Network::FilterStatus::Continue); - - std::string peer_metadata = encodePeerMetadata(defaultBaggage, defaultIdentity); - Buffer::OwnedImpl data; - data.add(peer_metadata); - EXPECT_EQ(filter_->onData(data, /*end_stream*/ false), Network::FilterStatus::Continue); - - EXPECT_EQ(data.length(), peer_metadata.size()); - const auto peer_info = peerInfoFromFilterState(); - EXPECT_FALSE(peer_info.has_value()); -} - -TEST_F(PeerMetadataUpstreamFilterTest, TestPeerMetadataNotConsumedWhenMalformed) { - initialize(); - setUpstreamHost(makeInternalListenerHost("connect_originate")); + setConnectionId(defaultConnectionId); + registry_->setValue(defaultConnectionId, encodePeerMetadata(defaultBaggage, defaultIdentity)); EXPECT_EQ(filter_->onNewConnection(), Network::FilterStatus::Continue); - std::string malformed{"well hello there, my friend!"}; + const std::string payload{"application data"}; Buffer::OwnedImpl data; - data.add(malformed); + data.add(payload); EXPECT_EQ(filter_->onData(data, /*end_stream*/ false), Network::FilterStatus::Continue); - EXPECT_EQ(data.length(), malformed.size()); + EXPECT_TRUE(registry_->getValue(defaultConnectionId).has_value()); const auto peer_info = peerInfoFromFilterState(); EXPECT_FALSE(peer_info.has_value()); }