Adds In-Memory Peer Metadata Exchange In Istio PMX Filters

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",
+    ],
+)

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<uint64_t, std::string> values_;
+};
+
+class PeerMetadataRegistryImpl : public PeerMetadataRegistry, public Singleton::Instance {
+public:
+  explicit PeerMetadataRegistryImpl(ThreadLocal::SlotAllocator& tls)
+      : tls_(ThreadLocal::TypedSlot<ThreadLocalRegistry>::makeUnique(tls)) {
+    tls_->set([](Event::Dispatcher&) { return std::make_shared<ThreadLocalRegistry>(); });
+  }
+
+  void setValue(uint64_t key, const std::string& value) override {
+    if (auto tls = tls_->get(); tls.has_value()) {
+      tls->values_[key] = value;
+    }
+  }
+
+  absl::optional<std::string> 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<ThreadLocalRegistry> tls_;
+};
+
+} // namespace
+
+SINGLETON_MANAGER_REGISTRATION(peer_metadata_registry);
+
+PeerMetadataRegistrySharedPtr
+getRegistry(Server::Configuration::ServerFactoryContext& context) {
+  return context.singletonManager().getTyped<PeerMetadataRegistry>(
+      SINGLETON_MANAGER_REGISTERED_NAME(peer_metadata_registry),
+      [&context] {
+        return std::make_shared<PeerMetadataRegistryImpl>(context.threadLocal());
+      },
+      /*pin=*/true);
+}
+
+} // namespace PeerMetadataShared
+} // namespace Common
+} // namespace Filters
+} // namespace Extensions
+} // namespace Envoy

contrib/istio/filters/common/source/peer_metadata_registry.h

@@ -0,0 +1,58 @@
+#pragma once
+
+#include <cstdint>
+#include <memory>
+#include <string>
+
+#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<std::string> 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<PeerMetadataRegistry>;
+
+// 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

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",