Introduce new filters discovering peer metadata from baggage header

BUILD

@@ -36,6 +36,7 @@ ISTIO_EXTENSIONS = [
     "//source/extensions/filters/http/istio_stats",
     "//source/extensions/filters/http/peer_metadata:filter_lib",
     "//source/extensions/filters/network/metadata_exchange:config_lib",
+    "//source/extensions/filters/network/peer_metadata",
 ]
 
 envoy_cc_binary(

extensions/common/metadata_object.cc

@@ -61,6 +61,34 @@ absl::optional<absl::string_view> toSuffix(WorkloadType workload_type) {
 
 } // namespace
 
+std::string WorkloadMetadataObject::baggage() const {
+  const auto workload_type = toSuffix(workload_type_).value_or(PodSuffix);
+  std::vector<std::string> parts;
+  if (!workload_name_.empty()) {
+    parts.push_back("k8s." + std::string(workload_type) + ".name=" + std::string(workload_name_));
+  }
+  // Map the workload metadata fields to baggage tokens
+  const std::vector<std::pair<absl::string_view, absl::string_view>> field_to_baggage = {
+      {Istio::Common::NamespaceNameToken, "k8s.namespace.name"},
+      {Istio::Common::ClusterNameToken, "k8s.cluster.name"},
+      {Istio::Common::ServiceNameToken, "service.name"},
+      {Istio::Common::ServiceVersionToken, "service.version"},
+      {Istio::Common::AppNameToken, "app.name"},
+      {Istio::Common::AppVersionToken, "app.version"},
+      {Istio::Common::InstanceNameToken, "k8s.instance.name"},
+  };
+
+  for (const auto& [field_name, baggage_key] : field_to_baggage) {
+    const auto field_result = getField(field_name);
+    if (auto field_value = std::get_if<absl::string_view>(&field_result)) {
+      if (!field_value->empty()) {
+        parts.push_back(absl::StrCat(baggage_key, "=", *field_value));
+      }
+    }
+  }
+  return absl::StrJoin(parts, ",");
+}
+
 Envoy::ProtobufTypes::MessagePtr WorkloadMetadataObject::serializeAsProto() const {
   auto message = std::make_unique<Envoy::Protobuf::Struct>();
   const auto suffix = toSuffix(workload_type_);
@@ -321,7 +349,13 @@ WorkloadMetadataObject::getField(absl::string_view field_name) const {
   return {};
 }
 
-std::unique_ptr<WorkloadMetadataObject> convertBaggageToWorkloadMetadata(absl::string_view data) {
+std::unique_ptr<WorkloadMetadataObject>
+convertBaggageToWorkloadMetadata(absl::string_view baggage) {
+  return convertBaggageToWorkloadMetadata(baggage, "");
+}
+
+std::unique_ptr<WorkloadMetadataObject>
+convertBaggageToWorkloadMetadata(absl::string_view data, absl::string_view identity) {
   absl::string_view instance;
   absl::string_view cluster;
   absl::string_view workload;
@@ -369,7 +403,7 @@ std::unique_ptr<WorkloadMetadataObject> convertBaggageToWorkloadMetadata(absl::s
   }
   return std::make_unique<WorkloadMetadataObject>(instance, cluster, namespace_name, workload,
                                                   canonical_name, canonical_revision, app_name,
-                                                  app_version, workload_type, "");
+                                                  app_version, workload_type, identity);
 }
 
 } // namespace Common

extensions/common/metadata_object.h

@@ -110,6 +110,7 @@ class WorkloadMetadataObject : public Envoy::StreamInfo::FilterState::Object,
   FieldType getField(absl::string_view) const override;
   void setLabels(std::vector<std::pair<std::string, std::string>> labels) { labels_ = labels; }
   std::vector<std::pair<std::string, std::string>> getLabels() const { return labels_; }
+  std::string baggage() const;
 
   const std::string instance_name_;
   const std::string cluster_name_;
@@ -152,7 +153,9 @@ convertEndpointMetadata(const std::string& endpoint_encoding);
 std::string serializeToStringDeterministic(const google::protobuf::Struct& metadata);
 
 // Convert from baggage encoding.
-std::unique_ptr<WorkloadMetadataObject> convertBaggageToWorkloadMetadata(absl::string_view data);
+std::unique_ptr<WorkloadMetadataObject> convertBaggageToWorkloadMetadata(absl::string_view baggage);
+std::unique_ptr<WorkloadMetadataObject>
+convertBaggageToWorkloadMetadata(absl::string_view baggage, absl::string_view identity);
 
 } // namespace Common
 } // namespace Istio

source/extensions/filters/http/peer_metadata/config.proto

@@ -43,12 +43,25 @@ message Config {
     bool skip_external_clusters = 1;
   }
 
+  // This method extracts peer metadata from the upstream filter state if it's available.
+  //
+  // Upstream filter state could be populated by multiple means in general, but in practice the intention here is that
+  // upstream PeerMetadata filter will populate the filter state with peer details extracted from the baggage header
+  // sent in response.
+  //
+  // Naturally this metadata discovery method only makes sense for upstream peer metadata discovery.
+  message UpstreamFilterState {
+    // Upstream filter state key that will be used to store peer metadata.
+    string peer_metadata_key = 1;
+  }
+
   // An exhaustive list of the derivation methods.
   message DiscoveryMethod {
     oneof method_specifier {
       Baggage baggage = 1;
       WorkloadDiscovery workload_discovery = 2;
       IstioHeaders istio_headers = 3;
+      UpstreamFilterState upstream_filter_state = 4;
     }
   }
 

source/extensions/filters/http/peer_metadata/filter.cc

@@ -145,6 +145,51 @@ absl::optional<PeerInfo> MXMethod::lookup(absl::string_view id, absl::string_vie
   return *out;
 }
 
+class UpstreamFilterStateMethod : public DiscoveryMethod {
+public:
+  UpstreamFilterStateMethod(
+      const io::istio::http::peer_metadata::Config_UpstreamFilterState& config)
+      : peer_metadata_key_(config.peer_metadata_key()) {}
+  absl::optional<PeerInfo> derivePeerInfo(const StreamInfo::StreamInfo&, Http::HeaderMap&,
+                                          Context&) const override;
+
+private:
+  std::string peer_metadata_key_;
+};
+
+absl::optional<PeerInfo>
+UpstreamFilterStateMethod::derivePeerInfo(const StreamInfo::StreamInfo& info, Http::HeaderMap&,
+                                          Context&) const {
+  const auto upstream = info.upstreamInfo();
+  if (!upstream) {
+    return {};
+  }
+
+  const auto filter_state = upstream->upstreamFilterState();
+  if (!filter_state) {
+    return {};
+  }
+
+  const auto* cel_state =
+      filter_state->getDataReadOnly<Envoy::Extensions::Filters::Common::Expr::CelState>(
+          peer_metadata_key_);
+  if (!cel_state) {
+    return {};
+  }
+
+  google::protobuf::Struct obj;
+  if (!obj.ParseFromString(absl::string_view(cel_state->value()))) {
+    return {};
+  }
+
+  std::unique_ptr<PeerInfo> peer_info = ::Istio::Common::convertStructToWorkloadMetadata(obj);
+  if (!peer_info) {
+    return {};
+  }
+
+  return *peer_info;
+}
+
 MXPropagationMethod::MXPropagationMethod(
     bool downstream, Server::Configuration::ServerFactoryContext& factory_context,
     const absl::flat_hash_set<std::string>& additional_labels,
@@ -211,6 +256,16 @@ std::vector<DiscoveryMethodPtr> FilterConfig::buildDiscoveryMethods(
       methods.push_back(std::make_unique<MXMethod>(downstream, additional_labels,
                                                    factory_context.serverFactoryContext()));
       break;
+    case io::istio::http::peer_metadata::Config::DiscoveryMethod::MethodSpecifierCase::
+        kUpstreamFilterState:
+      if (!downstream) {
+        methods.push_back(
+            std::make_unique<UpstreamFilterStateMethod>(method.upstream_filter_state()));
+      } else {
+        ENVOY_LOG(warn, "UpstreamFilterState peer metadata discovery option is only available for "
+                        "upstream peer discovery");
+      }
+      break;
     default:
       break;
     }

source/extensions/filters/network/peer_metadata/BUILD

@@ -0,0 +1,57 @@
+# Copyright 2026 Istio Authors. All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+##########################################################################
+
+# Ambient Peer Metadata filters
+load(
+    "@envoy//bazel:envoy_build_system.bzl",
+    "envoy_cc_library",
+)
+load(
+    "@envoy//bazel:envoy_library.bzl",
+    "envoy_proto_library",
+)
+
+package(default_visibility = ["//visibility:public"])
+
+licenses(["notice"])
+
+envoy_proto_library(
+    name = "config",
+    srcs = ["config.proto"],
+)
+
+envoy_cc_library(
+    name = "peer_metadata",
+    srcs = [
+        "peer_metadata.cc",
+    ],
+    repository = "@envoy",
+    deps = [
+        ":config_cc_proto",
+        "//extensions/common:metadata_object_lib",
+        "@envoy//envoy/buffer:buffer_interface",
+        "@envoy//envoy/network:address_interface",
+        "@envoy//envoy/network:filter_interface",
+        "@envoy//envoy/server:filter_config_interface",
+        "@envoy//source/common/common:minimal_logger_lib",
+        "@envoy//source/common/router:string_accessor_lib",
+        "@envoy//source/common/singleton:const_singleton",
+        "@envoy//source/common/stream_info:bool_accessor_lib",
+        "@envoy//source/common/tcp_proxy",
+        "@envoy//source/extensions/filters/common/expr:cel_state_lib",
+        "@envoy//source/extensions/filters/network/common:factory_base_lib",
+    ],
+)

source/extensions/filters/network/peer_metadata/config.proto

@@ -0,0 +1,44 @@
+/* Copyright 2026 Istio Authors. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+syntax = "proto3";
+
+package envoy.extensions.network_filters.peer_metadata;
+
+message Config {
+  // What filter state to use to save the baggage value that encodes the proxy
+  // workload.
+  //
+  // The upstream filter that will populate the baggage header in the HBONE
+  // request should be configured to use the same key.
+  //
+  // Why share baggage value via filter state instead of configruing upstream
+  // filter to use the baggage key value directly?
+  //
+  // ztunnel and waypoint have to be aware of the baggage header format,
+  // because they should be able to parse baggage headers to extract the
+  // metadata and report the metrics. However, pilot does not need to be aware
+  // of the baggage encoding yet.
+  //
+  // If instead of using custom filter to generate baggage header value we just
+  // let pilot generate it, it would spread the logic for generating baggage to
+  // the pilot as well. While not a big deal, if there is no clear reason to do
+  // it, let's not duplicate the implementation of baggage logic in pilot and
+  // just re-use the logic we already have in Envoy.
+  string baggage_key = 1;
+}
+
+message UpstreamConfig {
+}