MX support additional labels

extensions/common/metadata_object.cc

@@ -62,11 +62,45 @@ absl::optional<absl::string_view> toSuffix(WorkloadType workload_type) {
 
 Envoy::ProtobufTypes::MessagePtr WorkloadMetadataObject::serializeAsProto() const {
   auto message = std::make_unique<Envoy::ProtobufWkt::Struct>();
-  auto& fields = *message->mutable_fields();
-  const auto parts = serializeAsPairs();
-  for (const auto& p : parts) {
-    fields[std::string(p.first)] = Envoy::ValueUtil::stringValue(std::string(p.second));
+  const auto suffix = toSuffix(workload_type_);
+  if (suffix) {
+    (*message->mutable_fields())[WorkloadTypeToken].set_string_value(*suffix);
+  }
+  if (!workload_name_.empty()) {
+    (*message->mutable_fields())[WorkloadNameToken].set_string_value(workload_name_);
+  }
+  if (!cluster_name_.empty()) {
+    (*message->mutable_fields())[InstanceNameToken].set_string_value(instance_name_);
+  }
+  if (!cluster_name_.empty()) {
+    (*message->mutable_fields())[ClusterNameToken].set_string_value(cluster_name_);
+  }
+  if (!namespace_name_.empty()) {
+    (*message->mutable_fields())[NamespaceNameToken].set_string_value(namespace_name_);
+  }
+  if (!canonical_name_.empty()) {
+    (*message->mutable_fields())[ServiceNameToken].set_string_value(canonical_name_);
+  }
+  if (!canonical_revision_.empty()) {
+    (*message->mutable_fields())[ServiceVersionToken].set_string_value(canonical_revision_);
+  }
+  if (!app_name_.empty()) {
+    (*message->mutable_fields())[AppNameToken].set_string_value(app_name_);
   }
+  if (!app_version_.empty()) {
+    (*message->mutable_fields())[AppVersionToken].set_string_value(app_version_);
+  }
+  if (!identity_.empty()) {
+    (*message->mutable_fields())[IdentityToken].set_string_value(identity_);
+  }
+
+  if (!labels_.empty()) {
+    auto* labels = (*message->mutable_fields())[LabelsToken].mutable_struct_value();
+    for (const auto& l : labels_) {
+      (*labels->mutable_fields())[std::string(l.first)].set_string_value(std::string(l.second));
+    }
+  }
+
   return message;
 }
 
@@ -101,6 +135,11 @@ WorkloadMetadataObject::serializeAsPairs() const {
   if (!app_version_.empty()) {
     parts.push_back({AppVersionToken, app_version_});
   }
+  if (!labels_.empty()) {
+    for (const auto& l : labels_) {
+      parts.push_back({absl::StrCat("labels[]", l.first), absl::string_view(l.second)});
+    }
+  }
   return parts;
 }
 
@@ -168,6 +207,11 @@ google::protobuf::Struct convertWorkloadMetadataToStruct(const WorkloadMetadataO
   if (!obj.app_version_.empty()) {
     (*labels->mutable_fields())[AppVersionLabel].set_string_value(obj.app_version_);
   }
+  if (!obj.getLabels().empty()) {
+    for (const auto& lbl : obj.getLabels()) {
+      (*labels->mutable_fields())[std::string(lbl.first)].set_string_value(std::string(lbl.second));
+    }
+  }
   if (const auto owner = obj.owner(); owner.has_value()) {
     (*metadata.mutable_fields())[OwnerMetadataField].set_string_value(*owner);
   }
@@ -177,8 +221,15 @@ google::protobuf::Struct convertWorkloadMetadataToStruct(const WorkloadMetadataO
 // Convert struct to a metadata object.
 std::unique_ptr<WorkloadMetadataObject>
 convertStructToWorkloadMetadata(const google::protobuf::Struct& metadata) {
+  return convertStructToWorkloadMetadata(metadata, {});
+}
+
+std::unique_ptr<WorkloadMetadataObject>
+convertStructToWorkloadMetadata(const google::protobuf::Struct& metadata,
+                                const absl::flat_hash_set<std::string>& additional_labels) {
   absl::string_view instance, namespace_name, owner, workload, cluster, canonical_name,
       canonical_revision, app_name, app_version;
+  std::vector<std::pair<std::string, std::string>> labels;
   for (const auto& it : metadata.fields()) {
     if (it.first == InstanceMetadataField) {
       instance = it.second.string_value();
@@ -200,13 +251,19 @@ convertStructToWorkloadMetadata(const google::protobuf::Struct& metadata) {
           app_name = labels_it.second.string_value();
         } else if (labels_it.first == AppVersionLabel) {
           app_version = labels_it.second.string_value();
+        } else if (!additional_labels.empty() &&
+                   additional_labels.contains(std::string(labels_it.first))) {
+          labels.push_back(
+              {std::string(labels_it.first), std::string(labels_it.second.string_value())});
         }
       }
     }
   }
-  return std::make_unique<WorkloadMetadataObject>(instance, cluster, namespace_name, workload,
-                                                  canonical_name, canonical_revision, app_name,
-                                                  app_version, parseOwner(owner, workload), "");
+  auto obj = std::make_unique<WorkloadMetadataObject>(instance, cluster, namespace_name, workload,
+                                                      canonical_name, canonical_revision, app_name,
+                                                      app_version, parseOwner(owner, workload), "");
+  obj->setLabels(labels);
+  return obj;
 }
 
 absl::optional<WorkloadMetadataObject>

extensions/common/metadata_object.h

@@ -19,7 +19,6 @@
 
 #include "source/common/protobuf/protobuf.h"
 
-#include "absl/strings/str_split.h"
 #include "absl/types/optional.h"
 
 #include "google/protobuf/struct.pb.h"
@@ -77,6 +76,8 @@ constexpr absl::string_view AppVersionToken = "version";
 constexpr absl::string_view WorkloadNameToken = "workload";
 constexpr absl::string_view WorkloadTypeToken = "type";
 constexpr absl::string_view InstanceNameToken = "name";
+constexpr absl::string_view LabelsToken = "labels";
+constexpr absl::string_view IdentityToken = "identity";
 
 constexpr absl::string_view InstanceMetadataField = "NAME";
 constexpr absl::string_view NamespaceMetadataField = "NAMESPACE";
@@ -107,6 +108,8 @@ class WorkloadMetadataObject : public Envoy::StreamInfo::FilterState::Object,
   bool hasFieldSupport() const override { return true; }
   using Envoy::StreamInfo::FilterState::Object::FieldType;
   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_; }
 
   const std::string instance_name_;
   const std::string cluster_name_;
@@ -118,6 +121,7 @@ class WorkloadMetadataObject : public Envoy::StreamInfo::FilterState::Object,
   const std::string app_version_;
   const WorkloadType workload_type_;
   const std::string identity_;
+  std::vector<std::pair<std::string, std::string>> labels_;
 };
 
 // Parse string workload type.
@@ -133,6 +137,10 @@ google::protobuf::Struct convertWorkloadMetadataToStruct(const WorkloadMetadataO
 std::unique_ptr<WorkloadMetadataObject>
 convertStructToWorkloadMetadata(const google::protobuf::Struct& metadata);
 
+std::unique_ptr<WorkloadMetadataObject>
+convertStructToWorkloadMetadata(const google::protobuf::Struct& metadata,
+                                const absl::flat_hash_set<std::string>& additional_labels);
+
 // 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.

extensions/common/metadata_object_test.cc

@@ -65,6 +65,20 @@ void checkStructConversion(const Envoy::StreamInfo::FilterState::Object& data) {
   EXPECT_EQ(obj2->hash(), obj.hash());
 }
 
+TEST(WorkloadMetadataObjectTest, ConversionWithLabels) {
+  WorkloadMetadataObject deploy("pod-foo-1234", "my-cluster", "default", "foo", "foo-service",
+                                "v1alpha3", "", "", WorkloadType::Deployment, "");
+  deploy.setLabels({{"label1", "value1"}, {"label2", "value2"}});
+  auto pb = convertWorkloadMetadataToStruct(deploy);
+  auto obj1 = convertStructToWorkloadMetadata(pb, {"label1", "label2"});
+  EXPECT_EQ(obj1->getLabels().size(), 2);
+  auto obj2 = convertStructToWorkloadMetadata(pb, {"label1"});
+  EXPECT_EQ(obj2->getLabels().size(), 1);
+  absl::flat_hash_set<std::string> empty;
+  auto obj3 = convertStructToWorkloadMetadata(pb, empty);
+  EXPECT_EQ(obj3->getLabels().size(), 0);
+}
+
 TEST(WorkloadMetadataObjectTest, Conversion) {
   {
     const auto r = convertBaggageToWorkloadMetadata(

source/extensions/filters/http/istio_stats/istio_stats.cc

@@ -28,6 +28,7 @@
 #include "source/common/network/utility.h"
 #include "source/common/stream_info/utility.h"
 #include "source/extensions/filters/common/expr/context.h"
+#include "source/extensions/filters/common/expr/cel_state.h"
 #include "source/extensions/filters/common/expr/evaluator.h"
 #include "source/extensions/filters/http/common/pass_through_filter.h"
 #include "source/extensions/filters/http/grpc_stats/grpc_stats_filter.h"
@@ -122,13 +123,40 @@ bool peerInfoRead(Reporter reporter, const StreamInfo::FilterState& filter_state
          filter_state.hasDataWithName(Istio::Common::NoPeer);
 }
 
-const Istio::Common::WorkloadMetadataObject* peerInfo(Reporter reporter,
-                                                      const StreamInfo::FilterState& filter_state) {
+std::optional<Istio::Common::WorkloadMetadataObject>
+peerInfo(Reporter reporter, const StreamInfo::FilterState& filter_state) {
   const auto& filter_state_key =
       reporter == Reporter::ServerSidecar || reporter == Reporter::ServerGateway
           ? Istio::Common::DownstreamPeer
           : Istio::Common::UpstreamPeer;
-  return filter_state.getDataReadOnly<Istio::Common::WorkloadMetadataObject>(filter_state_key);
+  // This's a workaround before FilterStateObject support operation like `.labels['role']`.
+  // The workaround is to use CelState to store the peer metadata.
+  // Rebuild the WorkloadMetadataObject from the CelState.
+  const auto* cel_state =
+      filter_state.getDataReadOnly<Envoy::Extensions::Filters::Common::Expr::CelState>(
+          filter_state_key);
+  if (!cel_state) {
+    return {};
+  }
+
+  ProtobufWkt::Struct obj;
+  if (!obj.ParseFromString(absl::string_view(cel_state->value()))) {
+    return {};
+  }
+
+  Istio::Common::WorkloadMetadataObject peer_info(
+      extractString(obj, Istio::Common::InstanceNameToken),
+      extractString(obj, Istio::Common::ClusterNameToken),
+      extractString(obj, Istio::Common::NamespaceNameToken),
+      extractString(obj, Istio::Common::WorkloadNameToken),
+      extractString(obj, Istio::Common::ServiceNameToken),
+      extractString(obj, Istio::Common::ServiceVersionToken),
+      extractString(obj, Istio::Common::AppNameToken),
+      extractString(obj, Istio::Common::AppVersionToken),
+      Istio::Common::fromSuffix(extractString(obj, Istio::Common::WorkloadTypeToken)),
+      extractString(obj, Istio::Common::IdentityToken));
+
+  return peer_info;
 }
 
 // Process-wide context shared with all filter instances.
@@ -313,8 +341,6 @@ 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.
@@ -651,6 +677,13 @@ struct Config : public Logger::Loggable<Logger::Id::filter> {
           Protobuf::Arena arena;
           auto eval_status = compiled_exprs[id].first->Evaluate(*this, &arena);
           if (!eval_status.ok() || eval_status.value().IsError()) {
+            if (!eval_status.ok()) {
+              ENVOY_LOG(debug, "Failed to evaluate metric expression: {}", eval_status.status());
+            }
+            if (eval_status.value().IsError()) {
+              ENVOY_LOG(debug, "Failed to evaluate metric expression: {}",
+                        eval_status.value().ErrorOrDie()->message());
+            }
             expr_values_.push_back(std::make_pair(parent_.context_->unknown_, 0));
           } else {
             const auto string_value = Filters::Common::Expr::print(eval_status.value());
@@ -769,6 +802,7 @@ struct Config : public Logger::Loggable<Logger::Id::filter> {
 using ConfigSharedPtr = std::shared_ptr<Config>;
 
 class IstioStatsFilter : public Http::PassThroughFilter,
+                         public Logger::Loggable<Logger::Id::filter>,
                          public AccessLog::Instance,
                          public Network::ReadFilter,
                          public Network::ConnectionCallbacks {
@@ -895,6 +929,7 @@ class IstioStatsFilter : public Http::PassThroughFilter,
         const auto& info = decoder_callbacks_->streamInfo();
         peer_read_ = peerInfoRead(config_->reporter(), info.filterState());
         if (peer_read_ || end_stream) {
+          ENVOY_LOG(trace, "Populating peer metadata from HTTP MX.");
           populatePeerInfo(info, info.filterState());
         }
         if (is_grpc_ && (peer_read_ || end_stream)) {
@@ -933,6 +968,7 @@ class IstioStatsFilter : public Http::PassThroughFilter,
       peer_read_ = peerInfoRead(config_->reporter(), filter_state);
       // Report connection open once peer info is read or connection is closed.
       if (peer_read_ || end_stream) {
+        ENVOY_LOG(trace, "Populating peer metadata from TCP MX.");
         populatePeerInfo(info, filter_state);
         tags_.push_back({context_.request_protocol_, context_.tcp_});
         populateFlagsAndConnectionSecurity(info);
@@ -977,9 +1013,9 @@ class IstioStatsFilter : public Http::PassThroughFilter,
                         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);
+    auto object = peerInfo(config_->reporter(), filter_state);
     if (object) {
-      peer.emplace(*object);
+      peer.emplace(object.value());
     } else if (config_->reporter() == Reporter::ClientSidecar) {
       if (auto label_obj = extractEndpointMetadata(info); label_obj) {
         peer.emplace(label_obj.value());
@@ -1121,9 +1157,9 @@ class IstioStatsFilter : public Http::PassThroughFilter,
       switch (config_->reporter()) {
       case Reporter::ServerGateway: {
         std::optional<Istio::Common::WorkloadMetadataObject> endpoint_peer;
-        const auto* endpoint_object = peerInfo(Reporter::ClientSidecar, filter_state);
+        auto endpoint_object = peerInfo(Reporter::ClientSidecar, filter_state);
         if (endpoint_object) {
-          endpoint_peer.emplace(*endpoint_object);
+          endpoint_peer.emplace(endpoint_object.value());
         }
         tags_.push_back(
             {context_.destination_workload_,

source/extensions/filters/http/peer_metadata/BUILD

@@ -40,6 +40,7 @@ envoy_cc_library(
         "@envoy//source/common/http:header_utility_lib",
         "@envoy//source/common/http:utility_lib",
         "@envoy//source/common/network:utility_lib",
+        "@envoy//source/extensions/filters/common/expr:cel_state_lib",
         "@envoy//source/extensions/filters/http/common:factory_base_lib",
         "@envoy//source/extensions/filters/http/common:pass_through_filter_lib",
     ],

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

@@ -75,4 +75,8 @@ message Config {
 
   // True to enable sharing with the upstream.
   bool shared_with_upstream = 5;
+
+  // Additional labels to be added to the peer metadata to help your understand the traffic.
+  // e.g. `role`, `location` etc.
+  repeated string additional_labels = 6;
 }