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