diff --git a/extensions/common/context.cc b/extensions/common/context.cc
index 57fb170c648..626da780235 100644
--- a/extensions/common/context.cc
+++ b/extensions/common/context.cc
@@ -313,7 +313,7 @@ bool extractPeerMetadataFromUpstreamMetadata(
   std::vector<absl::string_view> parts = absl::StrSplit(endpoint_labels, ';');
   // workload label should semicolon separated four parts string:
   // workload_name;namespace;canonical_service;canonical_revision;cluster_id.
-  if (parts.size() < 4) {
+  if (parts.size() < 5) {
     return false;
   }
 
diff --git a/extensions/common/metadata_object.cc b/extensions/common/metadata_object.cc
index 1c81d9c32d4..639efb7dddb 100644
--- a/extensions/common/metadata_object.cc
+++ b/extensions/common/metadata_object.cc
@@ -14,6 +14,7 @@
 
 #include "extensions/common/metadata_object.h"
 
+#include "absl/strings/str_join.h"
 #include "flatbuffers/flatbuffers.h"
 #include "source/common/common/hash.h"
 
@@ -126,12 +127,48 @@ std::string WorkloadMetadataObject::baggage() const {
     default:
       break;
   }
-  return absl::StrCat("k8s.cluster.name=", cluster_name_,
-                      ",k8s.namespace.name=", namespace_name_, ",k8s.",
-                      workload_type, ".name=", workload_name_,
-                      ",service.name=", canonical_name_,
-                      ",service.version=", canonical_revision_,
-                      ",app.name=", app_name_, ",app.version=", app_version_);
+  std::vector<absl::string_view> parts;
+  parts.push_back("k8s.");
+  parts.push_back(workload_type);
+  parts.push_back(".name=");
+  parts.push_back(workload_name_);
+  if (!cluster_name_.empty()) {
+    parts.push_back(",");
+    parts.push_back(ClusterNameToken);
+    parts.push_back("=");
+    parts.push_back(cluster_name_);
+  }
+  if (!namespace_name_.empty()) {
+    parts.push_back(",");
+    parts.push_back(NamespaceNameToken);
+    parts.push_back("=");
+    parts.push_back(namespace_name_);
+  }
+  if (!canonical_name_.empty()) {
+    parts.push_back(",");
+    parts.push_back(ServiceNameToken);
+    parts.push_back("=");
+    parts.push_back(canonical_name_);
+  }
+  if (!canonical_revision_.empty()) {
+    parts.push_back(",");
+    parts.push_back(ServiceVersionToken);
+    parts.push_back("=");
+    parts.push_back(canonical_revision_);
+  }
+  if (!app_name_.empty()) {
+    parts.push_back(",");
+    parts.push_back(AppNameToken);
+    parts.push_back("=");
+    parts.push_back(app_name_);
+  }
+  if (!app_version_.empty()) {
+    parts.push_back(",");
+    parts.push_back(AppVersionToken);
+    parts.push_back("=");
+    parts.push_back(app_version_);
+  }
+  return absl::StrJoin(parts, "");
 }
 
 absl::optional<uint64_t> WorkloadMetadataObject::hash() const {
@@ -152,7 +189,7 @@ std::string_view toStdStringView(absl::string_view view) {
 }  // namespace
 
 flatbuffers::DetachedBuffer convertWorkloadMetadataToFlatNode(
-    const Istio::Common::WorkloadMetadataObject& obj) {
+    const WorkloadMetadataObject& obj) {
   flatbuffers::FlatBufferBuilder fbb;
 
   flatbuffers::Offset<flatbuffers::String> name, cluster, namespace_,
@@ -165,22 +202,22 @@ flatbuffers::DetachedBuffer convertWorkloadMetadataToFlatNode(
   workload_name = fbb.CreateString(toStdStringView(obj.workload_name_));
 
   switch (obj.workload_type_) {
-    case Istio::Common::WorkloadType::Deployment:
+    case WorkloadType::Deployment:
       owner = fbb.CreateString(absl::StrCat(OwnerPrefix, obj.namespace_name_,
                                             "/", DeploymentSuffix, "s/",
                                             obj.workload_name_));
       break;
-    case Istio::Common::WorkloadType::Job:
+    case WorkloadType::Job:
       owner =
           fbb.CreateString(absl::StrCat(OwnerPrefix, obj.namespace_name_, "/",
                                         JobSuffix, "s/", obj.workload_name_));
       break;
-    case Istio::Common::WorkloadType::CronJob:
+    case WorkloadType::CronJob:
       owner = fbb.CreateString(absl::StrCat(OwnerPrefix, obj.namespace_name_,
                                             "/", CronJobSuffix, "s/",
                                             obj.workload_name_));
       break;
-    case Istio::Common::WorkloadType::Pod:
+    case WorkloadType::Pod:
       owner =
           fbb.CreateString(absl::StrCat(OwnerPrefix, obj.namespace_name_, "/",
                                         PodSuffix, "s/", obj.workload_name_));
@@ -213,7 +250,7 @@ flatbuffers::DetachedBuffer convertWorkloadMetadataToFlatNode(
   return fbb.Release();
 }
 
-Istio::Common::WorkloadMetadataObject convertFlatNodeToWorkloadMetadata(
+WorkloadMetadataObject convertFlatNodeToWorkloadMetadata(
     const Wasm::Common::FlatNode& node) {
   const absl::string_view instance = toAbslStringView(node.name());
   const absl::string_view cluster = toAbslStringView(node.cluster_id());
@@ -221,54 +258,74 @@ Istio::Common::WorkloadMetadataObject convertFlatNodeToWorkloadMetadata(
   const absl::string_view namespace_name = toAbslStringView(node.namespace_());
   const auto* labels = node.labels();
 
-  const auto* name_iter =
-      labels->LookupByKey("service.istio.io/canonical-name");
-  const auto* name = name_iter ? name_iter->value() : nullptr;
-  const absl::string_view canonical_name = toAbslStringView(name);
+  absl::string_view canonical_name;
+  absl::string_view canonical_revision;
+  absl::string_view app_name;
+  absl::string_view app_version;
+  if (labels) {
+    const auto* name_iter =
+        labels->LookupByKey("service.istio.io/canonical-name");
+    const auto* name = name_iter ? name_iter->value() : nullptr;
+    canonical_name = toAbslStringView(name);
 
-  const auto* revision_iter =
-      labels->LookupByKey("service.istio.io/canonical-revision");
-  const auto* revision = revision_iter ? revision_iter->value() : nullptr;
-  const absl::string_view canonical_revision = toAbslStringView(revision);
+    const auto* revision_iter =
+        labels->LookupByKey("service.istio.io/canonical-revision");
+    const auto* revision = revision_iter ? revision_iter->value() : nullptr;
+    canonical_revision = toAbslStringView(revision);
 
-  const auto* app_iter = labels->LookupByKey("app");
-  const auto* app = app_iter ? app_iter->value() : nullptr;
-  const absl::string_view app_name = toAbslStringView(app);
+    const auto* app_iter = labels->LookupByKey("app");
+    const auto* app = app_iter ? app_iter->value() : nullptr;
+    app_name = toAbslStringView(app);
 
-  const auto* version_iter = labels->LookupByKey("version");
-  const auto* version = version_iter ? version_iter->value() : nullptr;
-  const absl::string_view app_version = toAbslStringView(version);
+    const auto* version_iter = labels->LookupByKey("version");
+    const auto* version = version_iter ? version_iter->value() : nullptr;
+    app_version = toAbslStringView(version);
+  }
 
-  Istio::Common::WorkloadType workload_type = Istio::Common::WorkloadType::Pod;
+  WorkloadType workload_type = WorkloadType::Pod;
   // Strip "s/workload_name" and check for workload type.
   absl::string_view owner = toAbslStringView(node.owner());
-  owner.remove_suffix(workload.size() + 2);
-  size_t last = owner.rfind('/');
-  if (last != absl::string_view::npos) {
-    const auto it = ALL_WORKLOAD_TOKENS.find(owner.substr(last + 1));
-    if (it != ALL_WORKLOAD_TOKENS.end()) {
-      switch (it->second) {
-        case WorkloadType::Deployment:
-          workload_type = Istio::Common::WorkloadType::Deployment;
-          break;
-        case WorkloadType::CronJob:
-          workload_type = Istio::Common::WorkloadType::CronJob;
-          break;
-        case WorkloadType::Job:
-          workload_type = Istio::Common::WorkloadType::Job;
-          break;
-        case WorkloadType::Pod:
-          workload_type = Istio::Common::WorkloadType::Pod;
-          break;
-        default:
-          break;
+  if (owner.size() > workload.size() + 2) {
+    owner.remove_suffix(workload.size() + 2);
+    size_t last = owner.rfind('/');
+    if (last != absl::string_view::npos) {
+      const auto it = ALL_WORKLOAD_TOKENS.find(owner.substr(last + 1));
+      if (it != ALL_WORKLOAD_TOKENS.end()) {
+        switch (it->second) {
+          case WorkloadType::Deployment:
+            workload_type = WorkloadType::Deployment;
+            break;
+          case WorkloadType::CronJob:
+            workload_type = WorkloadType::CronJob;
+            break;
+          case WorkloadType::Job:
+            workload_type = WorkloadType::Job;
+            break;
+          case WorkloadType::Pod:
+            workload_type = WorkloadType::Pod;
+            break;
+          default:
+            break;
+        }
       }
     }
   }
 
-  return Istio::Common::WorkloadMetadataObject(
-      instance, cluster, namespace_name, workload, canonical_name,
-      canonical_revision, app_name, app_version, workload_type);
+  return WorkloadMetadataObject(instance, cluster, namespace_name, workload,
+                                canonical_name, canonical_revision, app_name,
+                                app_version, workload_type);
+}
+
+absl::optional<WorkloadMetadataObject> convertEndpointMetadata(
+    const std::string& endpoint_encoding) {
+  std::vector<absl::string_view> parts = absl::StrSplit(endpoint_encoding, ';');
+  if (parts.size() < 5) {
+    return {};
+  }
+  // TODO: we cannot determine workload type from the encoding.
+  return absl::make_optional<WorkloadMetadataObject>(
+      "", parts[4], parts[1], parts[0], parts[2], parts[3], "", "",
+      WorkloadType::Pod);
 }
 
 }  // namespace Common
diff --git a/extensions/common/metadata_object.h b/extensions/common/metadata_object.h
index 4e9c7f23b78..efefe2a3c16 100644
--- a/extensions/common/metadata_object.h
+++ b/extensions/common/metadata_object.h
@@ -111,11 +111,19 @@ struct WorkloadMetadataObject : public Envoy::StreamInfo::FilterState::Object,
 
 // Convert metadata object to flatbuffer.
 flatbuffers::DetachedBuffer convertWorkloadMetadataToFlatNode(
-    const Istio::Common::WorkloadMetadataObject& obj);
+    const WorkloadMetadataObject& obj);
 
 // Convert flatbuffer to metadata object.
-Istio::Common::WorkloadMetadataObject convertFlatNodeToWorkloadMetadata(
+WorkloadMetadataObject convertFlatNodeToWorkloadMetadata(
     const Wasm::Common::FlatNode& node);
 
+// Convert endpoint metadata string to a metadata object.
+// Telemetry metadata is compressed into a semicolon separated string:
+// workload-name;namespace;canonical-service-name;canonical-service-revision;cluster-id.
+// Telemetry metadata is stored as a string under "istio", "workload" field
+// path.
+absl::optional<WorkloadMetadataObject> convertEndpointMetadata(
+    const std::string& endpoint_encoding);
+
 }  // namespace Common
 }  // namespace Istio
diff --git a/extensions/common/metadata_object_test.cc b/extensions/common/metadata_object_test.cc
index e0f64e578c0..86044673e16 100644
--- a/extensions/common/metadata_object_test.cc
+++ b/extensions/common/metadata_object_test.cc
@@ -51,26 +51,26 @@ TEST(WorkloadMetadataObjectTest, Baggage) {
                              WorkloadType::Job);
 
   EXPECT_EQ(deploy.baggage(),
-            absl::StrCat("k8s.cluster.name=my-cluster,",
-                         "k8s.namespace.name=default,k8s.deployment.name=foo,",
+            absl::StrCat("k8s.deployment.name=foo,k8s.cluster.name=my-cluster,",
+                         "k8s.namespace.name=default,",
                          "service.name=foo-service,service.version=v1alpha3,",
                          "app.name=foo-app,app.version=v1"));
 
   EXPECT_EQ(pod.baggage(),
-            absl::StrCat("k8s.cluster.name=my-cluster,",
-                         "k8s.namespace.name=default,k8s.pod.name=foo,",
+            absl::StrCat("k8s.pod.name=foo,k8s.cluster.name=my-cluster,",
+                         "k8s.namespace.name=default,",
                          "service.name=foo-service,service.version=v1alpha3,",
                          "app.name=foo-app,app.version=v1"));
 
   EXPECT_EQ(cronjob.baggage(),
-            absl::StrCat("k8s.cluster.name=my-cluster,",
-                         "k8s.namespace.name=default,k8s.cronjob.name=foo,"
+            absl::StrCat("k8s.cronjob.name=foo,k8s.cluster.name=my-cluster,",
+                         "k8s.namespace.name=default,"
                          "service.name=foo-service,service.version=v1alpha3,",
                          "app.name=foo-app,app.version=v1"));
 
   EXPECT_EQ(job.baggage(),
-            absl::StrCat("k8s.cluster.name=my-cluster,",
-                         "k8s.namespace.name=default,k8s.job.name=foo,",
+            absl::StrCat("k8s.job.name=foo,k8s.cluster.name=my-cluster,",
+                         "k8s.namespace.name=default,",
                          "service.name=foo-service,service.version=v1alpha3,",
                          "app.name=foo-app,app.version=v1"));
 }
@@ -86,9 +86,9 @@ void checkFlatNodeConversion(const WorkloadMetadataObject& obj) {
 TEST(WorkloadMetadataObjectTest, FromBaggage) {
   {
     auto obj = WorkloadMetadataObject::fromBaggage(
-        absl::StrCat("k8s.cluster.name=my-cluster,k8s.namespace.name=default,",
-                     "k8s.deployment.name=foo,service.name=foo-service,",
-                     "service.version=v1alpha3"));
+        absl::StrCat("k8s.deployment.name=foo,k8s.cluster.name=my-cluster,k8s."
+                     "namespace.name=default,",
+                     "service.name=foo-service,", "service.version=v1alpha3"));
     EXPECT_EQ(obj.canonical_name_, "foo-service");
     EXPECT_EQ(obj.canonical_revision_, "v1alpha3");
     EXPECT_EQ(obj.workload_type_, WorkloadType::Deployment);
@@ -100,9 +100,9 @@ TEST(WorkloadMetadataObjectTest, FromBaggage) {
 
   {
     auto obj = WorkloadMetadataObject::fromBaggage(
-        absl::StrCat("k8s.cluster.name=my-cluster,k8s.namespace.name=test,k8s."
-                     "pod.name=foo-pod-435,service.name=",
-                     "foo-service,service.version=v1beta2"));
+        absl::StrCat("k8s.pod.name=foo-pod-435,k8s.cluster.name=my-cluster,k8s."
+                     "namespace.name=test,"
+                     "service.name=foo-service,service.version=v1beta2"));
 
     EXPECT_EQ(obj.canonical_name_, "foo-service");
     EXPECT_EQ(obj.canonical_revision_, "v1beta2");
@@ -116,9 +116,9 @@ TEST(WorkloadMetadataObjectTest, FromBaggage) {
 
   {
     auto obj = WorkloadMetadataObject::fromBaggage(
-        absl::StrCat("k8s.cluster.name=my-cluster,k8s.namespace.name=test,",
-                     "k8s.job.name=foo-job-435,service.name=foo-service,",
-                     "service.version=v1beta4"));
+        absl::StrCat("k8s.job.name=foo-job-435,k8s.cluster.name=my-cluster,k8s."
+                     "namespace.name=test,",
+                     "service.name=foo-service,", "service.version=v1beta4"));
 
     EXPECT_EQ(obj.canonical_name_, "foo-service");
     EXPECT_EQ(obj.canonical_revision_, "v1beta4");
@@ -132,9 +132,9 @@ TEST(WorkloadMetadataObjectTest, FromBaggage) {
 
   {
     auto obj = WorkloadMetadataObject::fromBaggage(
-        absl::StrCat("k8s.cluster.name=my-cluster,k8s.namespace.name=test,",
-                     "k8s.cronjob.name=foo-cronjob,service.name=foo-service,",
-                     "service.version=v1beta4"));
+        absl::StrCat("k8s.cronjob.name=foo-cronjob,k8s.cluster.name=my-cluster,"
+                     "k8s.namespace.name=test,",
+                     "service.name=foo-service,", "service.version=v1beta4"));
 
     EXPECT_EQ(obj.canonical_name_, "foo-service");
     EXPECT_EQ(obj.canonical_revision_, "v1beta4");
@@ -149,8 +149,8 @@ TEST(WorkloadMetadataObjectTest, FromBaggage) {
 
   {
     auto obj = WorkloadMetadataObject::fromBaggage(absl::StrCat(
-        "k8s.namespace.name=default,",
-        "k8s.deployment.name=foo,service.name=foo-service,",
+        "k8s.deployment.name=foo,k8s.namespace.name=default,",
+        "service.name=foo-service,",
         "service.version=v1alpha3,app.name=foo-app,app.version=v1"));
 
     EXPECT_EQ(obj.canonical_name_, "foo-service");
@@ -165,5 +165,30 @@ TEST(WorkloadMetadataObjectTest, FromBaggage) {
   }
 }
 
+TEST(WorkloadMetadataObjectTest, ConvertFromFlatNode) {
+  flatbuffers::FlatBufferBuilder fbb;
+  Wasm::Common::FlatNodeBuilder builder(fbb);
+  auto data = builder.Finish();
+  fbb.Finish(data);
+  auto buffer = fbb.Release();
+  const auto& node =
+      *flatbuffers::GetRoot<Wasm::Common::FlatNode>(buffer.data());
+  auto obj = convertFlatNodeToWorkloadMetadata(node);
+  EXPECT_EQ(obj.baggage(), "k8s.pod.name=");
+}
+
+TEST(WorkloadMetadataObjectTest, ConvertFromEndpointMetadata) {
+  EXPECT_EQ(absl::nullopt, convertEndpointMetadata(""));
+  EXPECT_EQ(absl::nullopt, convertEndpointMetadata("a;b"));
+  EXPECT_EQ(absl::nullopt, convertEndpointMetadata("a;;;b"));
+  EXPECT_EQ(absl::nullopt, convertEndpointMetadata("a;b;c;d"));
+  auto obj =
+      convertEndpointMetadata("foo-pod;default;foo-service;v1;my-cluster");
+  ASSERT_TRUE(obj.has_value());
+  EXPECT_EQ(obj->baggage(),
+            "k8s.pod.name=foo-pod,k8s.cluster.name=my-cluster,k8s.namespace."
+            "name=default,service.name=foo-service,service.version=v1");
+}
+
 }  // namespace Common
 }  // namespace Istio
diff --git a/source/extensions/common/BUILD b/source/extensions/common/BUILD
index 973a2df447d..f027db3170e 100644
--- a/source/extensions/common/BUILD
+++ b/source/extensions/common/BUILD
@@ -36,7 +36,6 @@ envoy_cc_library(
         ":utils_lib",
         "//src/istio/authn:context_proto_cc_proto",
         "//src/istio/utils:attribute_names_lib",
-        "//src/istio/utils:utils_lib",
         "@envoy//source/exe:envoy_common_lib",
     ],
 )
diff --git a/source/extensions/common/authn.cc b/source/extensions/common/authn.cc
index 0deb61b3ca7..e6223dd8d09 100644
--- a/source/extensions/common/authn.cc
+++ b/source/extensions/common/authn.cc
@@ -17,9 +17,9 @@
 
 #include "source/common/common/base64.h"
 #include "source/extensions/common/filter_names.h"
+#include "source/extensions/common/utils.h"
 #include "src/istio/authn/context.pb.h"
 #include "src/istio/utils/attribute_names.h"
-#include "src/istio/utils/utils.h"
 
 using istio::authn::Result;
 
@@ -49,10 +49,10 @@ void Authentication::SaveAuthAttributesToStruct(
                 result.peer_user());
     setKeyValue(data, istio::utils::AttributeName::kSourcePrincipal,
                 result.peer_user());
-    std::string source_ns("");
-    if (istio::utils::GetSourceNamespace(result.peer_user(), &source_ns)) {
+    auto source_ns = GetNamespace(result.peer_user());
+    if (source_ns) {
       setKeyValue(data, istio::utils::AttributeName::kSourceNamespace,
-                  source_ns);
+                  std::string(source_ns.value()));
     }
   }
   if (result.has_origin()) {
diff --git a/source/extensions/common/utils.cc b/source/extensions/common/utils.cc
index dfaa270dc2f..d1698a6391d 100644
--- a/source/extensions/common/utils.cc
+++ b/source/extensions/common/utils.cc
@@ -34,6 +34,9 @@ const std::string kPerHostMetadataKey("istio");
 // Attribute field for per-host data override
 const std::string kMetadataDestinationUID("uid");
 
+const std::string kNamespaceKey("/ns/");
+const char kDelimiter = '/';
+
 bool hasSPIFFEPrefix(const std::string& san) {
   return absl::StartsWith(san, kSPIFFEPrefix);
 }
@@ -196,5 +199,19 @@ Status ParseJsonMessage(const std::string& json, Message* output) {
   return ::google::protobuf::util::JsonStringToMessage(json, output, options);
 }
 
+absl::optional<absl::string_view> GetNamespace(absl::string_view principal) {
+  // The namespace is a substring in principal with format:
+  // "<DOMAIN>/ns/<NAMESPACE>/sa/<SERVICE-ACCOUNT>". '/' is not allowed to
+  // appear in actual content except as delimiter between tokens.
+  size_t begin = principal.find(kNamespaceKey);
+  if (begin == absl::string_view::npos) {
+    return {};
+  }
+  begin += kNamespaceKey.length();
+  size_t end = principal.find(kDelimiter, begin);
+  size_t len = (end == std::string::npos ? end : end - begin);
+  return {principal.substr(begin, len)};
+}
+
 }  // namespace Utils
 }  // namespace Envoy
diff --git a/source/extensions/common/utils.h b/source/extensions/common/utils.h
index 0f8757ff9c9..bb51404f225 100644
--- a/source/extensions/common/utils.h
+++ b/source/extensions/common/utils.h
@@ -62,5 +62,8 @@ bool GetRequestedServerName(const Network::Connection* connection,
 ::google::protobuf::util::Status ParseJsonMessage(
     const std::string& json, ::google::protobuf::Message* output);
 
+// Get the namespace part of Istio certificate URI.
+absl::optional<absl::string_view> GetNamespace(absl::string_view principal);
+
 }  // namespace Utils
 }  // namespace Envoy
diff --git a/source/extensions/common/utils_test.cc b/source/extensions/common/utils_test.cc
index 22a5cdcc6d4..7cbc70b4668 100644
--- a/source/extensions/common/utils_test.cc
+++ b/source/extensions/common/utils_test.cc
@@ -131,4 +131,46 @@ INSTANTIATE_TEST_SUITE_P(
       return info.param ? "peer" : "local";
     });
 
+class NamespaceTest : public ::testing::Test {
+ protected:
+  void checkFalse(const std::string& principal) {
+    auto out = Envoy::Utils::GetNamespace(principal);
+    EXPECT_FALSE(out.has_value());
+  }
+
+  void checkTrue(const std::string& principal, absl::string_view ns) {
+    auto out = Envoy::Utils::GetNamespace(principal);
+    ASSERT_TRUE(out.has_value());
+    EXPECT_EQ(ns, out.value());
+  }
+};
+
+TEST_F(NamespaceTest, TestGetNamespace) {
+  checkFalse("");
+  checkFalse("cluster.local");
+  checkFalse("cluster.local/");
+  checkFalse("cluster.local/ns");
+  checkFalse("cluster.local/sa/user");
+  checkFalse("cluster.local/sa/user/ns");
+  checkFalse("cluster.local/sa/user_ns/");
+  checkFalse("cluster.local/sa/user_ns/abc/xyz");
+  checkFalse("cluster.local/NS/abc");
+
+  checkTrue("cluster.local/ns/", "");
+  checkTrue("cluster.local/ns//", "");
+  checkTrue("cluster.local/sa/user/ns/", "");
+  checkTrue("cluster.local/ns//sa/user", "");
+  checkTrue("cluster.local/ns//ns/ns", "");
+
+  checkTrue("cluster.local/ns/ns/ns/ns", "ns");
+  checkTrue("cluster.local/ns/abc_ns", "abc_ns");
+  checkTrue("cluster.local/ns/abc_ns/", "abc_ns");
+  checkTrue("cluster.local/ns/abc_ns/sa/user_ns", "abc_ns");
+  checkTrue("cluster.local/ns/abc_ns/sa/user_ns/other/xyz", "abc_ns");
+  checkTrue("cluster.local/sa/user_ns/ns/abc", "abc");
+  checkTrue("cluster.local/sa/user_ns/ns/abc/", "abc");
+  checkTrue("cluster.local/sa/user_ns/ns/abc_ns", "abc_ns");
+  checkTrue("cluster.local/sa/user_ns/ns/abc_ns/", "abc_ns");
+}
+
 }  // namespace
diff --git a/source/extensions/filters/http/istio_stats/BUILD b/source/extensions/filters/http/istio_stats/BUILD
index b8a77abbc98..8ccc81ede83 100644
--- a/source/extensions/filters/http/istio_stats/BUILD
+++ b/source/extensions/filters/http/istio_stats/BUILD
@@ -32,6 +32,9 @@ envoy_cc_extension(
     deps = [
         "//extensions/common:metadata_object_lib",
         "//extensions/stats:config_cc_proto",
+        "//source/extensions/common:utils_lib",
+        "@com_google_cel_cpp//eval/public:builtin_func_registrar",
+        "@com_google_cel_cpp//eval/public:cel_expr_builder_factory",
         "@com_google_cel_cpp//parser",
         "@envoy//envoy/registry",
         "@envoy//envoy/server:filter_config_interface",
diff --git a/source/extensions/filters/http/istio_stats/istio_stats.cc b/source/extensions/filters/http/istio_stats/istio_stats.cc
index b5492fd0697..38d2f60c07a 100644
--- a/source/extensions/filters/http/istio_stats/istio_stats.cc
+++ b/source/extensions/filters/http/istio_stats/istio_stats.cc
@@ -17,12 +17,15 @@
 #include "envoy/registry/registry.h"
 #include "envoy/server/factory_context.h"
 #include "envoy/singleton/manager.h"
+#include "eval/public/builtin_func_registrar.h"
+#include "eval/public/cel_expr_builder_factory.h"
 #include "extensions/common/metadata_object.h"
 #include "parser/parser.h"
 #include "source/common/grpc/common.h"
 #include "source/common/http/header_map_impl.h"
 #include "source/common/http/header_utility.h"
 #include "source/common/stream_info/utility.h"
+#include "source/extensions/common/utils.h"
 #include "source/extensions/filters/common/expr/cel_state.h"
 #include "source/extensions/filters/common/expr/context.h"
 #include "source/extensions/filters/common/expr/evaluator.h"
@@ -267,10 +270,13 @@ using ContextSharedPtr = std::shared_ptr<Context>;
 
 SINGLETON_MANAGER_REGISTRATION(Context)
 
+using google::api::expr::runtime::CelValue;
+
 // Instructions on dropping, creating, and overriding labels.
 // This is not the "hot path" of the metrics system and thus, fairly
 // unoptimized.
-struct MetricOverrides : public Logger::Loggable<Logger::Id::filter> {
+struct MetricOverrides : public Logger::Loggable<Logger::Id::filter>,
+                         public google::api::expr::runtime::BaseActivation {
   MetricOverrides(ContextSharedPtr& context, Stats::SymbolTable& symbol_table)
       : context_(context), pool_(symbol_table) {}
   ContextSharedPtr context_;
@@ -286,42 +292,83 @@ struct MetricOverrides : public Logger::Loggable<Logger::Id::filter> {
   using TagAdditions = std::vector<std::pair<Stats::StatName, uint32_t>>;
   absl::flat_hash_map<Stats::StatName, TagAdditions> tag_additions_;
 
+  const StreamInfo::StreamInfo* info_;
+  absl::optional<CelValue> FindValue(absl::string_view name,
+                                     Protobuf::Arena* arena) const override {
+    if (name == Filters::Common::Expr::Request) {
+      if (info_) {
+        return CelValue::CreateMap(
+            Protobuf::Arena::Create<Filters::Common::Expr::RequestWrapper>(
+                arena, *arena, nullptr, *info_));
+      }
+    } else if (name == Filters::Common::Expr::Connection) {
+      if (info_) {
+        return CelValue::CreateMap(
+            Protobuf::Arena::Create<Filters::Common::Expr::ConnectionWrapper>(
+                arena, *info_));
+      }
+    } else if (name == Filters::Common::Expr::Source) {
+      if (info_) {
+        return CelValue::CreateMap(
+            Protobuf::Arena::Create<Filters::Common::Expr::PeerWrapper>(
+                arena, *info_, false));
+      }
+    } else if (name == Filters::Common::Expr::Destination) {
+      if (info_) {
+        return CelValue::CreateMap(
+            Protobuf::Arena::Create<Filters::Common::Expr::PeerWrapper>(
+                arena, *info_, true));
+      }
+    } else if (name == Filters::Common::Expr::Upstream) {
+      if (info_) {
+        return CelValue::CreateMap(
+            Protobuf::Arena::Create<Filters::Common::Expr::UpstreamWrapper>(
+                arena, *info_));
+      }
+    } else if (name == Filters::Common::Expr::Metadata) {
+      if (info_) {
+        return Filters::Common::Expr::CelProtoWrapper::CreateMessage(
+            &info_->dynamicMetadata(), arena);
+      }
+    } else if (name == Filters::Common::Expr::FilterState) {
+      if (info_) {
+        return Protobuf::Arena::Create<
+                   Filters::Common::Expr::FilterStateWrapper>(
+                   arena, info_->filterState())
+            ->Produce(arena);
+      }
+    } else if (name == "node") {
+      return Filters::Common::Expr::CelProtoWrapper::CreateMessage(
+          &context_->node_, arena);
+    } else if (name == "route_name") {
+      if (info_) {
+        return Filters::Common::Expr::CelValue::CreateString(
+            &info_->getRouteName());
+      }
+    }
+    if (info_) {
+      const auto* obj =
+          info_->filterState()
+              .getDataReadOnly<
+                  Envoy::Extensions::Filters::Common::Expr::CelState>(
+                  absl::StrCat("wasm.", name));
+      if (obj) {
+        return obj->exprValue(arena, false);
+      }
+    }
+    return {};
+  }
+  std::vector<const google::api::expr::runtime::CelFunction*>
+  FindFunctionOverloads(absl::string_view) const override {
+    return {};
+  }
   Stats::StatName evaluate(const StreamInfo::StreamInfo& info, uint32_t id,
                            Stats::StatNameDynamicPool& pool) {
     Protobuf::Arena arena;
-    Filters::Common::Expr::Activation activation;
-    activation.InsertValueProducer(
-        Filters::Common::Expr::Request,
-        std::make_unique<Filters::Common::Expr::RequestWrapper>(arena, nullptr,
-                                                                info));
-    activation.InsertValueProducer(
-        Filters::Common::Expr::Connection,
-        std::make_unique<Filters::Common::Expr::ConnectionWrapper>(info));
-    activation.InsertValueProducer(
-        Filters::Common::Expr::Upstream,
-        std::make_unique<Filters::Common::Expr::UpstreamWrapper>(info));
-    activation.InsertValueProducer(
-        Filters::Common::Expr::Source,
-        std::make_unique<Filters::Common::Expr::PeerWrapper>(info, false));
-    activation.InsertValueProducer(
-        Filters::Common::Expr::Destination,
-        std::make_unique<Filters::Common::Expr::PeerWrapper>(info, true));
-    activation.InsertValueProducer(
-        Filters::Common::Expr::Metadata,
-        std::make_unique<Filters::Common::Expr::MetadataProducer>(
-            info.dynamicMetadata()));
-    activation.InsertValueProducer(
-        Filters::Common::Expr::FilterState,
-        std::make_unique<Filters::Common::Expr::FilterStateWrapper>(
-            info.filterState()));
-    activation.InsertValue(
-        "node", Filters::Common::Expr::CelProtoWrapper::CreateMessage(
-                    &context_->node_, &arena));
-    activation.InsertValue(
-        "route_name",
-        Filters::Common::Expr::CelValue::CreateString(&info.getRouteName()));
-    auto eval_status = compiled_exprs_[id]->Evaluate(activation, &arena);
-    if (!eval_status.ok()) {
+    info_ = &info;
+    auto eval_status = compiled_exprs_[id]->Evaluate(*this, &arena);
+    info_ = nullptr;
+    if (!eval_status.ok() || eval_status.value().IsError()) {
       return context_->unknown_;
     }
     return pool.add(Filters::Common::Expr::print(eval_status.value()));
@@ -367,7 +414,17 @@ struct MetricOverrides : public Logger::Loggable<Logger::Id::filter> {
       return {};
     }
     if (expr_builder_ == nullptr) {
-      expr_builder_ = Filters::Common::Expr::createBuilder(nullptr);
+      google::api::expr::runtime::InterpreterOptions options;
+      expr_builder_ =
+          google::api::expr::runtime::CreateCelExpressionBuilder(options);
+      auto register_status =
+          google::api::expr::runtime::RegisterBuiltinFunctions(
+              expr_builder_->GetRegistry(), options);
+      if (!register_status.ok()) {
+        throw Extensions::Filters::Common::Expr::CelException(
+            absl::StrCat("failed to register built-in functions: ",
+                         register_status.message()));
+      }
     }
     parsed_exprs_.push_back(parse_status.value().expr());
     compiled_exprs_.push_back(
@@ -437,11 +494,18 @@ struct Config : public Logger::Loggable<Logger::Id::filter> {
               metric_overrides_->tag_overrides_[metric][tag_it->second] = {};
             }
         }
-        for (const auto& [tag, expr] : metric.dimensions()) {
+        // Make order of tags deterministic.
+        std::vector<std::string> tags;
+        tags.reserve(metric.dimensions().size());
+        for (const auto& [tag, _] : metric.dimensions()) {
+          tags.push_back(tag);
+        }
+        std::sort(tags.begin(), tags.end());
+        for (const auto& tag : tags) {
+          const std::string& expr = metric.dimensions().at(tag);
           auto id = metric_overrides_->getOrCreateExpression(expr);
           if (!id.has_value()) {
             ENVOY_LOG(info, "Failed to parse expression: {}", expr);
-            continue;
           }
           const auto& tag_it = context_->all_tags_.find(tag);
           if (tag_it == context_->all_tags_.end()) {
@@ -707,77 +771,137 @@ class IstioStatsFilter : public Http::PassThroughFilter,
   // and after initial bytes read/written by MX TCP filter.
   void populatePeerInfo(const StreamInfo::StreamInfo& info,
                         const StreamInfo::FilterState& filter_state) {
+    // Compute peer info with client-side fallbacks.
     absl::optional<Istio::Common::WorkloadMetadataObject> peer;
     const auto* object = peerInfo(config_->reporter(), filter_state);
     if (object) {
       peer.emplace(Istio::Common::convertFlatNodeToWorkloadMetadata(*object));
+    } else if (config_->reporter() == Reporter::ClientSidecar) {
+      auto upstream_info = info.upstreamInfo();
+      auto upstream_host =
+          upstream_info ? upstream_info->upstreamHost() : nullptr;
+      if (upstream_host && upstream_host->metadata()) {
+        const auto& filter_metadata =
+            upstream_host->metadata()->filter_metadata();
+        const auto& it = filter_metadata.find("istio");
+        if (it != filter_metadata.end()) {
+          const auto& workload_it = it->second.fields().find("workload");
+          if (workload_it != it->second.fields().end()) {
+            auto label_obj = Istio::Common::convertEndpointMetadata(
+                workload_it->second.string_value());
+            if (label_obj) {
+              peer.emplace(label_obj.value());
+            }
+          }
+        }
+      }
     }
-    // Compute destination service with fallbacks.
+
+    // Compute destination service with client-side fallbacks.
     absl::string_view service_host;
     absl::string_view service_host_name;
+    if (!config_->disable_host_header_fallback_) {
+      const auto* headers = info.getRequestHeaders();
+      if (headers && headers->Host()) {
+        service_host = headers->Host()->value().getStringView();
+        service_host_name = service_host;
+      }
+    }
     const auto cluster_info = info.upstreamClusterInfo();
     if (cluster_info && cluster_info.value()) {
-      const auto& filter_metadata =
-          cluster_info.value()->metadata().filter_metadata();
-      const auto& it = filter_metadata.find("istio");
-      if (it != filter_metadata.end()) {
-        const auto& services_it = it->second.fields().find("services");
-        if (services_it != it->second.fields().end()) {
-          const auto& services = services_it->second.list_value();
-          if (services.values_size() > 0) {
-            const auto& service = services.values(0).struct_value().fields();
-            const auto& host_it = service.find("host");
-            if (host_it != service.end()) {
-              service_host = host_it->second.string_value();
-              service_host_name =
-                  service_host.substr(0, service_host.find_first_of('.'));
+      const auto& cluster_name = cluster_info.value()->name();
+      if (cluster_name == "BlackholeCluster" ||
+          cluster_name == "PassthroughCluster" ||
+          cluster_name == "InboundPassthroughClusterIpv4" ||
+          cluster_name == "InboundPassthroughClusterIpv6") {
+        service_host_name = cluster_name;
+      } else {
+        const auto& filter_metadata =
+            cluster_info.value()->metadata().filter_metadata();
+        const auto& it = filter_metadata.find("istio");
+        if (it != filter_metadata.end()) {
+          const auto& services_it = it->second.fields().find("services");
+          if (services_it != it->second.fields().end()) {
+            const auto& services = services_it->second.list_value();
+            if (services.values_size() > 0) {
+              const auto& service = services.values(0).struct_value().fields();
+              const auto& host_it = service.find("host");
+              if (host_it != service.end()) {
+                service_host = host_it->second.string_value();
+                service_host_name =
+                    service_host.substr(0, service_host.find_first_of('.'));
+              }
             }
           }
         }
       }
     }
-    if (service_host.empty() && !config_->disable_host_header_fallback_) {
-      const auto* headers = info.getRequestHeaders();
-      if (headers && headers->Host()) {
-        service_host = headers->Host()->value().getStringView();
-        service_host_name = service_host;
+
+    // Implements fallback from using the namespace from SAN if available to
+    // using peer metadata, otherwise.
+    absl::string_view peer_san;
+    const Ssl::ConnectionInfoConstSharedPtr ssl_info =
+        config_->reporter() == Reporter::ServerSidecar
+            ? info.downstreamAddressProvider().sslConnection()
+            : (info.upstreamInfo()
+                   ? info.upstreamInfo()->upstreamSslConnection()
+                   : nullptr);
+    if (ssl_info && !ssl_info->uriSanPeerCertificate().empty()) {
+      peer_san = ssl_info->uriSanPeerCertificate()[0];
+    }
+    absl::string_view peer_namespace;
+    if (!peer_san.empty()) {
+      const auto san_namespace = Utils::GetNamespace(peer_san);
+      if (san_namespace) {
+        peer_namespace = san_namespace.value();
       }
     }
+    if (peer_namespace.empty() && peer) {
+      peer_namespace = peer->namespace_name_;
+    }
+    absl::string_view local_san;
+    if (ssl_info && !ssl_info->uriSanLocalCertificate().empty()) {
+      local_san = ssl_info->uriSanLocalCertificate()[0];
+    }
+
     switch (config_->reporter()) {
       case Reporter::ServerSidecar: {
         tags_.push_back(
-            {context_.source_workload_,
-             peer ? pool_.add(peer->workload_name_) : context_.unknown_});
-        tags_.push_back(
-            {context_.source_canonical_service_,
-             peer ? pool_.add(peer->canonical_name_) : context_.unknown_});
-        tags_.push_back(
-            {context_.source_canonical_revision_,
-             peer ? pool_.add(peer->canonical_revision_) : context_.latest_});
-        tags_.push_back(
-            {context_.source_workload_namespace_,
-             peer ? pool_.add(peer->namespace_name_) : context_.unknown_});
-        const auto ssl_info = info.downstreamAddressProvider().sslConnection();
-        tags_.push_back({context_.source_principal_,
-                         ssl_info && !ssl_info->uriSanPeerCertificate().empty()
-                             ? pool_.add(ssl_info->uriSanPeerCertificate()[0])
+            {context_.source_workload_, peer && !peer->workload_name_.empty()
+                                            ? pool_.add(peer->workload_name_)
+                                            : context_.unknown_});
+        tags_.push_back({context_.source_canonical_service_,
+                         peer && !peer->canonical_name_.empty()
+                             ? pool_.add(peer->canonical_name_)
                              : context_.unknown_});
-        tags_.push_back({context_.source_app_, peer ? pool_.add(peer->app_name_)
-                                                    : context_.unknown_});
+        tags_.push_back({context_.source_canonical_revision_,
+                         peer && !peer->canonical_revision_.empty()
+                             ? pool_.add(peer->canonical_revision_)
+                             : context_.latest_});
+        tags_.push_back({context_.source_workload_namespace_,
+                         !peer_namespace.empty() ? pool_.add(peer_namespace)
+                                                 : context_.unknown_});
+        tags_.push_back({context_.source_principal_, !peer_san.empty()
+                                                         ? pool_.add(peer_san)
+                                                         : context_.unknown_});
+        tags_.push_back({context_.source_app_, peer && !peer->app_name_.empty()
+                                                   ? pool_.add(peer->app_name_)
+                                                   : context_.unknown_});
         tags_.push_back(
-            {context_.source_version_,
-             peer ? pool_.add(peer->app_version_) : context_.unknown_});
+            {context_.source_version_, peer && !peer->app_version_.empty()
+                                           ? pool_.add(peer->app_version_)
+                                           : context_.unknown_});
         tags_.push_back(
-            {context_.source_cluster_,
-             peer ? pool_.add(peer->cluster_name_) : context_.unknown_});
+            {context_.source_cluster_, peer && !peer->cluster_name_.empty()
+                                           ? pool_.add(peer->cluster_name_)
+                                           : context_.unknown_});
         tags_.push_back(
             {context_.destination_workload_, context_.workload_name_});
         tags_.push_back(
             {context_.destination_workload_namespace_, context_.namespace_});
-        tags_.push_back({context_.destination_principal_,
-                         ssl_info && !ssl_info->uriSanLocalCertificate().empty()
-                             ? pool_.add(ssl_info->uriSanLocalCertificate()[0])
-                             : context_.unknown_});
+        tags_.push_back(
+            {context_.destination_principal_,
+             !local_san.empty() ? pool_.add(local_san) : context_.unknown_});
         tags_.push_back({context_.destination_app_, context_.app_name_});
         tags_.push_back({context_.destination_version_, context_.app_version_});
         tags_.push_back({context_.destination_service_,
@@ -806,51 +930,52 @@ class IstioStatsFilter : public Http::PassThroughFilter,
                          context_.canonical_revision_});
         tags_.push_back(
             {context_.source_workload_namespace_, context_.namespace_});
-        const auto upstream_info = info.upstreamInfo();
-        const Ssl::ConnectionInfoConstSharedPtr ssl_info =
-            upstream_info ? upstream_info->upstreamSslConnection() : nullptr;
-        tags_.push_back({context_.source_principal_,
-                         ssl_info && !ssl_info->uriSanLocalCertificate().empty()
-                             ? pool_.add(ssl_info->uriSanLocalCertificate()[0])
-                             : context_.unknown_});
+        tags_.push_back({context_.source_principal_, !local_san.empty()
+                                                         ? pool_.add(local_san)
+                                                         : context_.unknown_});
         tags_.push_back({context_.source_app_, context_.app_name_});
         tags_.push_back({context_.source_version_, context_.app_version_});
         tags_.push_back({context_.source_cluster_, context_.cluster_name_});
-        tags_.push_back(
-            {context_.destination_workload_,
-             peer ? pool_.add(peer->workload_name_) : context_.unknown_});
-        tags_.push_back(
-            {context_.destination_workload_namespace_,
-             peer ? pool_.add(peer->namespace_name_) : context_.unknown_});
-        tags_.push_back({context_.destination_principal_,
-                         ssl_info && !ssl_info->uriSanPeerCertificate().empty()
-                             ? pool_.add(ssl_info->uriSanPeerCertificate()[0])
+        tags_.push_back({context_.destination_workload_,
+                         peer && !peer->workload_name_.empty()
+                             ? pool_.add(peer->workload_name_)
                              : context_.unknown_});
+        tags_.push_back({context_.destination_workload_namespace_,
+                         !peer_namespace.empty() ? pool_.add(peer_namespace)
+                                                 : context_.unknown_});
         tags_.push_back(
-            {context_.destination_app_,
-             peer ? pool_.add(peer->app_name_) : context_.unknown_});
+            {context_.destination_principal_,
+             !peer_san.empty() ? pool_.add(peer_san) : context_.unknown_});
         tags_.push_back(
-            {context_.destination_version_,
-             peer ? pool_.add(peer->app_version_) : context_.unknown_});
+            {context_.destination_app_, peer && !peer->app_name_.empty()
+                                            ? pool_.add(peer->app_name_)
+                                            : context_.unknown_});
+        tags_.push_back(
+            {context_.destination_version_, peer && !peer->app_version_.empty()
+                                                ? pool_.add(peer->app_version_)
+                                                : context_.unknown_});
         tags_.push_back({context_.destination_service_,
                          service_host.empty() ? context_.unknown_
                                               : pool_.add(service_host)});
-        tags_.push_back(
-            {context_.destination_canonical_service_,
-             peer ? pool_.add(peer->canonical_name_) : context_.unknown_});
-        tags_.push_back(
-            {context_.destination_canonical_revision_,
-             peer ? pool_.add(peer->canonical_revision_) : context_.latest_});
+        tags_.push_back({context_.destination_canonical_service_,
+                         peer && !peer->canonical_name_.empty()
+                             ? pool_.add(peer->canonical_name_)
+                             : context_.unknown_});
+        tags_.push_back({context_.destination_canonical_revision_,
+                         peer && !peer->canonical_revision_.empty()
+                             ? pool_.add(peer->canonical_revision_)
+                             : context_.latest_});
         tags_.push_back({context_.destination_service_name_,
                          service_host_name.empty()
                              ? context_.unknown_
                              : pool_.add(service_host_name)});
+        tags_.push_back({context_.destination_service_namespace_,
+                         !peer_namespace.empty() ? pool_.add(peer_namespace)
+                                                 : context_.unknown_});
         tags_.push_back(
-            {context_.destination_service_namespace_,
-             peer ? pool_.add(peer->namespace_name_) : context_.unknown_});
-        tags_.push_back(
-            {context_.destination_cluster_,
-             peer ? pool_.add(peer->cluster_name_) : context_.unknown_});
+            {context_.destination_cluster_, peer && !peer->cluster_name_.empty()
+                                                ? pool_.add(peer->cluster_name_)
+                                                : context_.unknown_});
         break;
       }
       default:
diff --git a/src/envoy/workload_metadata/workload_metadata_test.cc b/src/envoy/workload_metadata/workload_metadata_test.cc
index 0a54f25e8e5..f9a9bcbdcd9 100644
--- a/src/envoy/workload_metadata/workload_metadata_test.cc
+++ b/src/envoy/workload_metadata/workload_metadata_test.cc
@@ -79,9 +79,9 @@ TEST_F(FilterTest, OnAccept) {
   auto found = filter_state->getDataReadOnly<Envoy::Router::StringAccessor>(
       Istio::Common::kSourceMetadataBaggageKey);
   EXPECT_EQ(found->asString(),
-            "k8s.cluster.name=my-cluster,k8s.namespace.name=default,k8s."
-            "deployment.name=foo,service.name=foo-svc,service.version=v2beta1,"
-            "app.name=,app.version=");
+            "k8s.deployment.name=foo,k8s.cluster.name=my-cluster,k8s.namespace."
+            "name=default,"
+            "service.name=foo-svc,service.version=v2beta1");
 
   setAddressToReturn("tcp://192.168.1.1:5555");
   filter_state = std::make_shared<StreamInfo::FilterStateImpl>(
@@ -96,9 +96,9 @@ TEST_F(FilterTest, OnAccept) {
   found = filter_state->getDataReadOnly<Envoy::Router::StringAccessor>(
       Istio::Common::kSourceMetadataBaggageKey);
   EXPECT_EQ(found->asString(),
-            "k8s.cluster.name=my-cluster,k8s.namespace.name=default,k8s."
-            "deployment.name=foo,service.name=foo-svc,service.version=v2beta1,"
-            "app.name=,app.version=");
+            "k8s.deployment.name=foo,k8s.cluster.name=my-cluster,k8s.namespace."
+            "name=default,"
+            "service.name=foo-svc,service.version=v2beta1");
 
   setAddressToReturn("tcp://4.22.1.1:4343");
   EXPECT_CALL(callbacks_, filterState()).Times(0);
diff --git a/src/istio/utils/BUILD b/src/istio/utils/BUILD
index 70e0b0eb146..32f12596145 100644
--- a/src/istio/utils/BUILD
+++ b/src/istio/utils/BUILD
@@ -22,32 +22,6 @@ load(
     "envoy_cc_test",
 )
 
-envoy_cc_library(
-    name = "utils_lib",
-    srcs = [
-        "utils.cc",
-    ],
-    hdrs = [
-        "utils.h",
-    ],
-    repository = "@envoy",
-    visibility = ["//visibility:public"],
-    deps = [
-        ":attribute_names_lib",
-        "//external:protobuf",
-    ],
-)
-
-envoy_cc_test(
-    name = "utils_test",
-    size = "small",
-    srcs = ["utils_test.cc"],
-    repository = "@envoy",
-    deps = [
-        ":utils_lib",
-    ],
-)
-
 envoy_cc_library(
     name = "attribute_names_lib",
     srcs = [
diff --git a/src/istio/utils/utils.cc b/src/istio/utils/utils.cc
deleted file mode 100644
index 0d6151a9ca8..00000000000
--- a/src/istio/utils/utils.cc
+++ /dev/null
@@ -1,49 +0,0 @@
-/* Copyright 2018 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.
- */
-
-#include "src/istio/utils/utils.h"
-
-#include <sstream>
-#include <vector>
-
-namespace istio {
-namespace utils {
-
-namespace {
-const std::string kNamespaceKey("/ns/");
-const char kDelimiter = '/';
-}  // namespace
-
-bool GetSourceNamespace(const std::string& principal,
-                        std::string* source_namespace) {
-  if (source_namespace) {
-    // The namespace is a substring in principal with format:
-    // "<DOMAIN>/ns/<NAMESPACE>/sa/<SERVICE-ACCOUNT>". '/' is not allowed to
-    // appear in actual content except as delimiter between tokens.
-    size_t begin = principal.find(kNamespaceKey);
-    if (begin == std::string::npos) {
-      return false;
-    }
-    begin += kNamespaceKey.length();
-    size_t end = principal.find(kDelimiter, begin);
-    size_t len = (end == std::string::npos ? end : end - begin);
-    *source_namespace = principal.substr(begin, len);
-    return true;
-  }
-  return false;
-}
-
-}  // namespace utils
-}  // namespace istio
diff --git a/src/istio/utils/utils.h b/src/istio/utils/utils.h
deleted file mode 100644
index bec3bdedf1d..00000000000
--- a/src/istio/utils/utils.h
+++ /dev/null
@@ -1,29 +0,0 @@
-/* Copyright 2018 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.
- */
-
-#pragma once
-
-#include <string>
-
-#include "google/protobuf/struct.pb.h"
-
-namespace istio {
-namespace utils {
-
-// Get source.namespace attribute from principal.
-bool GetSourceNamespace(const std::string& principal,
-                        std::string* source_namespace);
-}  // namespace utils
-}  // namespace istio
diff --git a/src/istio/utils/utils_test.cc b/src/istio/utils/utils_test.cc
deleted file mode 100644
index 7133d727075..00000000000
--- a/src/istio/utils/utils_test.cc
+++ /dev/null
@@ -1,69 +0,0 @@
-/* Copyright 2018 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.
- */
-
-#include "src/istio/utils/utils.h"
-
-#include "gtest/gtest.h"
-
-namespace istio {
-namespace utils {
-namespace {
-
-class UtilsTest : public ::testing::Test {
- protected:
-  void checkFalse(const std::string& principal) {
-    std::string output_ns = "none";
-    EXPECT_FALSE(GetSourceNamespace(principal, &output_ns));
-    EXPECT_EQ(output_ns, output_ns);
-  }
-
-  void checkTrue(const std::string& principal, const std::string& ns) {
-    std::string output_ns = "none";
-    EXPECT_TRUE(GetSourceNamespace(principal, &output_ns));
-    EXPECT_EQ(ns, output_ns);
-  }
-};
-
-TEST_F(UtilsTest, TestGetSourceNamespace) {
-  checkFalse("");
-  checkFalse("cluster.local");
-  checkFalse("cluster.local/");
-  checkFalse("cluster.local/ns");
-  checkFalse("cluster.local/sa/user");
-  checkFalse("cluster.local/sa/user/ns");
-  checkFalse("cluster.local/sa/user_ns/");
-  checkFalse("cluster.local/sa/user_ns/abc/xyz");
-  checkFalse("cluster.local/NS/abc");
-
-  checkTrue("cluster.local/ns/", "");
-  checkTrue("cluster.local/ns//", "");
-  checkTrue("cluster.local/sa/user/ns/", "");
-  checkTrue("cluster.local/ns//sa/user", "");
-  checkTrue("cluster.local/ns//ns/ns", "");
-
-  checkTrue("cluster.local/ns/ns/ns/ns", "ns");
-  checkTrue("cluster.local/ns/abc_ns", "abc_ns");
-  checkTrue("cluster.local/ns/abc_ns/", "abc_ns");
-  checkTrue("cluster.local/ns/abc_ns/sa/user_ns", "abc_ns");
-  checkTrue("cluster.local/ns/abc_ns/sa/user_ns/other/xyz", "abc_ns");
-  checkTrue("cluster.local/sa/user_ns/ns/abc", "abc");
-  checkTrue("cluster.local/sa/user_ns/ns/abc/", "abc");
-  checkTrue("cluster.local/sa/user_ns/ns/abc_ns", "abc_ns");
-  checkTrue("cluster.local/sa/user_ns/ns/abc_ns/", "abc_ns");
-}
-
-}  // namespace
-}  // namespace utils
-}  // namespace istio
diff --git a/test/envoye2e/inventory.go b/test/envoye2e/inventory.go
index 35337d24830..74c7dc83cd6 100644
--- a/test/envoye2e/inventory.go
+++ b/test/envoye2e/inventory.go
@@ -69,7 +69,10 @@ func init() {
 			"TestStatsECDS/envoy.wasm.runtime.v8",
 			"TestStatsECDS/envoy.wasm.runtime.v8#01",
 			"TestStatsECDS/#00",
-			"TestStatsEndpointLabels",
+			"TestStatsEndpointLabels/envoy.wasm.runtime.null",
+			"TestStatsEndpointLabels/envoy.wasm.runtime.v8",
+			"TestStatsEndpointLabels/envoy.wasm.runtime.v8#01",
+			"TestStatsEndpointLabels/#00",
 			"TestStatsServerWaypointProxy",
 			"TestStatsGrpc/envoy.wasm.runtime.null",
 			"TestStatsGrpc/envoy.wasm.runtime.v8",
diff --git a/test/envoye2e/stats_plugin/stats_test.go b/test/envoye2e/stats_plugin/stats_test.go
index 4b1d0f5b62b..c34f88d793b 100644
--- a/test/envoye2e/stats_plugin/stats_test.go
+++ b/test/envoye2e/stats_plugin/stats_test.go
@@ -159,10 +159,13 @@ func TestStatsPayload(t *testing.T) {
 		for _, runtime := range Runtimes {
 			t.Run(testCase.Name+"/"+runtime.WasmRuntime, func(t *testing.T) {
 				env.SkipWasm(t, runtime.WasmRuntime)
-				clientStats := testCase.ClientStats
+				clientStats := make(map[string]driver.StatMatcher)
+				for metric, values := range testCase.ClientStats {
+					clientStats[metric] = values
+				}
 				if testCase.Name == "Customized" && runtime.WasmRuntime == "" {
-					// TODO: complete dimensions support in native stats filter
-					clientStats = map[string]driver.StatMatcher{}
+					// TODO: complete adding custom dimensions to native stats
+					delete(clientStats, "istio_custom")
 				}
 				params := driver.NewTestParams(t, map[string]string{
 					"RequestCount":               "10",
@@ -612,42 +615,47 @@ func TestStatsECDS(t *testing.T) {
 }
 
 func TestStatsEndpointLabels(t *testing.T) {
-	params := driver.NewTestParams(t, map[string]string{
-		"RequestCount":               "10",
-		"MetadataExchangeFilterCode": "inline_string: \"envoy.wasm.metadata_exchange\"",
-		"StatsFilterCode":            "inline_string: \"envoy.wasm.stats\"",
-		"WasmRuntime":                "envoy.wasm.runtime.null",
-		"EnableEndpointMetadata":     "true",
-		"StatsConfig":                driver.LoadTestData("testdata/bootstrap/stats.yaml.tmpl"),
-		"StatsFilterClientConfig":    driver.LoadTestJSON("testdata/stats/client_config.yaml"),
-	}, envoye2e.ProxyE2ETests)
-	params.Vars["ClientMetadata"] = params.LoadTestData("testdata/client_node_metadata.json.tmpl")
-	params.Vars["ServerMetadata"] = params.LoadTestData("testdata/server_node_metadata.json.tmpl")
-	params.Vars["ClientHTTPFilters"] = driver.LoadTestData("testdata/filters/stats_outbound.yaml.tmpl")
-	if err := (&driver.Scenario{
-		Steps: []driver.Step{
-			&driver.XDS{},
-			&driver.Update{Node: "client", Version: "0", Listeners: []string{params.LoadTestData("testdata/listener/client.yaml.tmpl")}},
-			&driver.Update{Node: "server", Version: "0", Listeners: []string{params.LoadTestData("testdata/listener/server.yaml.tmpl")}},
-			&driver.Envoy{Bootstrap: params.LoadTestData("testdata/bootstrap/client.yaml.tmpl")},
-			&driver.Envoy{Bootstrap: params.LoadTestData("testdata/bootstrap/server.yaml.tmpl")},
-			&driver.Sleep{Duration: 1 * time.Second},
-			&driver.Repeat{
-				N: 10,
-				Step: &driver.HTTPCall{
-					Port:         params.Ports.ClientPort,
-					ResponseCode: 200,
-				},
-			},
-			&driver.Stats{
-				AdminPort: params.Ports.ClientAdmin,
-				Matchers: map[string]driver.StatMatcher{
-					"istio_requests_total": &driver.ExactStat{Metric: "testdata/metric/client_request_total_endpoint_labels.yaml.tmpl"},
+	for _, runtime := range Runtimes {
+		t.Run(runtime.WasmRuntime, func(t *testing.T) {
+			env.SkipWasm(t, runtime.WasmRuntime)
+			params := driver.NewTestParams(t, map[string]string{
+				"RequestCount":               "10",
+				"MetadataExchangeFilterCode": runtime.MetadataExchangeFilterCode,
+				"StatsFilterCode":            runtime.StatsFilterCode,
+				"WasmRuntime":                runtime.WasmRuntime,
+				"EnableEndpointMetadata":     "true",
+				"StatsConfig":                driver.LoadTestData("testdata/bootstrap/stats.yaml.tmpl"),
+				"StatsFilterClientConfig":    driver.LoadTestJSON("testdata/stats/client_config.yaml"),
+			}, envoye2e.ProxyE2ETests)
+			params.Vars["ClientMetadata"] = params.LoadTestData("testdata/client_node_metadata.json.tmpl")
+			params.Vars["ServerMetadata"] = params.LoadTestData("testdata/server_node_metadata.json.tmpl")
+			params.Vars["ClientHTTPFilters"] = driver.LoadTestData("testdata/filters/stats_outbound.yaml.tmpl")
+			if err := (&driver.Scenario{
+				Steps: []driver.Step{
+					&driver.XDS{},
+					&driver.Update{Node: "client", Version: "0", Listeners: []string{params.LoadTestData("testdata/listener/client.yaml.tmpl")}},
+					&driver.Update{Node: "server", Version: "0", Listeners: []string{params.LoadTestData("testdata/listener/server.yaml.tmpl")}},
+					&driver.Envoy{Bootstrap: params.LoadTestData("testdata/bootstrap/client.yaml.tmpl")},
+					&driver.Envoy{Bootstrap: params.LoadTestData("testdata/bootstrap/server.yaml.tmpl")},
+					&driver.Sleep{Duration: 1 * time.Second},
+					&driver.Repeat{
+						N: 10,
+						Step: &driver.HTTPCall{
+							Port:         params.Ports.ClientPort,
+							ResponseCode: 200,
+						},
+					},
+					&driver.Stats{
+						AdminPort: params.Ports.ClientAdmin,
+						Matchers: map[string]driver.StatMatcher{
+							"istio_requests_total": &driver.ExactStat{Metric: "testdata/metric/client_request_total_endpoint_labels.yaml.tmpl"},
+						},
+					},
 				},
-			},
-		},
-	}).Run(params); err != nil {
-		t.Fatal(err)
+			}).Run(params); err != nil {
+				t.Fatal(err)
+			}
+		})
 	}
 }
 
diff --git a/testdata/metric/client_request_total_customized.yaml.tmpl b/testdata/metric/client_request_total_customized.yaml.tmpl
index 7e71c43c81d..109ccfc7c8a 100644
--- a/testdata/metric/client_request_total_customized.yaml.tmpl
+++ b/testdata/metric/client_request_total_customized.yaml.tmpl
@@ -2,7 +2,7 @@ name: istio_requests_total
 type: COUNTER
 metric:
 - counter:
-    value: {{ .Vars.RequestCount }}0
+    value: {{ .Vars.RequestCount }}
   label:
   - name: reporter
     value: source
diff --git a/testdata/metric/tcp_server_connection_open_without_mx.yaml.tmpl b/testdata/metric/tcp_server_connection_open_without_mx.yaml.tmpl
index 78c6b18921a..5247c8ad583 100644
--- a/testdata/metric/tcp_server_connection_open_without_mx.yaml.tmpl
+++ b/testdata/metric/tcp_server_connection_open_without_mx.yaml.tmpl
@@ -13,7 +13,11 @@ metric:
   - name: source_canonical_revision
     value: latest
   - name: source_workload_namespace
+{{ if ne .Vars.WasmRuntime "" }}
     value: unknown
+{{ else }}
+    value: default
+{{ end }}
   - name: source_principal
     value: spiffe://cluster.local/ns/default/sa/client
   - name: source_app
diff --git a/testdata/stats/client_config_customized.yaml.tmpl b/testdata/stats/client_config_customized.yaml.tmpl
index 748eaf557f5..e92861e621b 100644
--- a/testdata/stats/client_config_customized.yaml.tmpl
+++ b/testdata/stats/client_config_customized.yaml.tmpl
@@ -1,8 +1,5 @@
 field_separator: ";.;"
 definitions:
-- name: requests_total
-  value: "10"
-  type: COUNTER
 - name: custom
   value: "1"
   type: COUNTER