[CSM O11Y] Fix issue when CSM optional labels are present in server metrics (#35633)

<!--

If you know who should review your pull request, please assign it to that
person, otherwise the pull request would get assigned randomly.

If your pull request is for a specific language, please add the appropriate
lang label.

-->

Closes #35633

COPYBARA_INTEGRATE_REVIEW=https://github.com/grpc/grpc/pull/35633 from yijiem:labels-injector-patch 6876243943
PiperOrigin-RevId: 600931754

src/cpp/ext/csm/metadata_exchange.cc

@@ -462,12 +462,17 @@ void ServiceMeshLabelsInjector::AddLabels(
 }
 
 bool ServiceMeshLabelsInjector::AddOptionalLabels(
+    bool is_client,
     absl::Span<const std::shared_ptr<std::map<std::string, std::string>>>
         optional_labels_span,
     opentelemetry::nostd::function_ref<
         bool(opentelemetry::nostd::string_view,
              opentelemetry::common::AttributeValue)>
         callback) const {
+  if (!is_client) {
+    // Currently the CSM optional labels are only set on client.
+    return true;
+  }
   // According to the CSM Observability Metric spec, if the control plane fails
   // to provide these labels, the client will set their values to "unknown".
   // These default values are set below.
@@ -493,9 +498,10 @@ bool ServiceMeshLabelsInjector::AddOptionalLabels(
 }
 
 size_t ServiceMeshLabelsInjector::GetOptionalLabelsSize(
+    bool is_client,
     absl::Span<const std::shared_ptr<std::map<std::string, std::string>>>)
     const {
-  return 2;
+  return is_client ? 2 : 0;
 }
 
 }  // namespace internal

src/cpp/ext/csm/metadata_exchange.h

@@ -52,6 +52,7 @@ class ServiceMeshLabelsInjector : public LabelsInjector {
 
   // Add optional labels to the traced calls.
   bool AddOptionalLabels(
+      bool is_client,
       absl::Span<const std::shared_ptr<std::map<std::string, std::string>>>
           optional_labels_span,
       opentelemetry::nostd::function_ref<
@@ -61,6 +62,7 @@ class ServiceMeshLabelsInjector : public LabelsInjector {
 
   // Gets the size of the actual optional labels.
   size_t GetOptionalLabelsSize(
+      bool is_client,
       absl::Span<const std::shared_ptr<std::map<std::string, std::string>>>
           optional_labels_span) const override;
 

src/cpp/ext/otel/key_value_iterable.h

@@ -56,13 +56,15 @@ class KeyValueIterable : public opentelemetry::common::KeyValueIterable {
           additional_labels,
       const ActivePluginOptionsView* active_plugin_options_view,
       absl::Span<const std::shared_ptr<std::map<std::string, std::string>>>
-          optional_labels_span)
+          optional_labels_span,
+      bool is_client)
       : injected_labels_iterable_(injected_labels_iterable),
         injected_labels_from_plugin_options_(
             injected_labels_from_plugin_options),
         additional_labels_(additional_labels),
         active_plugin_options_view_(active_plugin_options_view),
-        optional_labels_(optional_labels_span) {}
+        optional_labels_(optional_labels_span),
+        is_client_(is_client) {}
 
   bool ForEachKeyValue(opentelemetry::nostd::function_ref<
                        bool(opentelemetry::nostd::string_view,
@@ -79,7 +81,7 @@ class KeyValueIterable : public opentelemetry::common::KeyValueIterable {
     }
     if (OpenTelemetryPluginState().labels_injector != nullptr &&
         !OpenTelemetryPluginState().labels_injector->AddOptionalLabels(
-            optional_labels_, callback)) {
+            is_client_, optional_labels_, callback)) {
       return false;
     }
     if (active_plugin_options_view_ != nullptr &&
@@ -88,7 +90,7 @@ class KeyValueIterable : public opentelemetry::common::KeyValueIterable {
                 const InternalOpenTelemetryPluginOption& plugin_option,
                 size_t /*index*/) {
               return plugin_option.labels_injector()->AddOptionalLabels(
-                  optional_labels_, callback);
+                  is_client_, optional_labels_, callback);
             })) {
       return false;
     }
@@ -126,14 +128,14 @@ class KeyValueIterable : public opentelemetry::common::KeyValueIterable {
     size += additional_labels_.size();
     if (OpenTelemetryPluginState().labels_injector != nullptr) {
       size += OpenTelemetryPluginState().labels_injector->GetOptionalLabelsSize(
-          optional_labels_);
+          is_client_, optional_labels_);
     }
     if (active_plugin_options_view_ != nullptr) {
       active_plugin_options_view_->ForEach(
           [&size, this](const InternalOpenTelemetryPluginOption& plugin_option,
                         size_t /*index*/) {
             size += plugin_option.labels_injector()->GetOptionalLabelsSize(
-                optional_labels_);
+                is_client_, optional_labels_);
             return true;
           });
     }
@@ -149,6 +151,7 @@ class KeyValueIterable : public opentelemetry::common::KeyValueIterable {
   const ActivePluginOptionsView* active_plugin_options_view_;
   absl::Span<const std::shared_ptr<std::map<std::string, std::string>>>
       optional_labels_;
+  bool is_client_;
 };
 
 }  // namespace internal

src/cpp/ext/otel/otel_client_filter.cc

@@ -131,10 +131,10 @@ OpenTelemetryCallTracer::OpenTelemetryCallAttemptTracer::
     // We might not have all the injected labels that we want at this point, so
     // avoid recording a subset of injected labels here.
     OpenTelemetryPluginState().client.attempt.started->Add(
-        1, KeyValueIterable(/*injected_labels_iterable=*/nullptr, {},
-                            additional_labels,
-                            /*active_plugin_options_view=*/nullptr,
-                            /*optional_labels_span=*/{}));
+        1, KeyValueIterable(
+               /*injected_labels_iterable=*/nullptr, {}, additional_labels,
+               /*active_plugin_options_view=*/nullptr,
+               /*optional_labels_span=*/{}, /*is_client=*/true));
   }
 }
 
@@ -213,7 +213,7 @@ void OpenTelemetryCallTracer::OpenTelemetryCallAttemptTracer::
   KeyValueIterable labels(
       injected_labels_.get(), injected_labels_from_plugin_options_,
       additional_labels, &parent_->parent_->active_plugin_options_view(),
-      optional_labels_array_);
+      optional_labels_array_, /*is_client=*/true);
   if (OpenTelemetryPluginState().client.attempt.duration != nullptr) {
     OpenTelemetryPluginState().client.attempt.duration->Record(
         absl::ToDoubleSeconds(absl::Now() - start_time_), labels,

src/cpp/ext/otel/otel_plugin.h

@@ -84,6 +84,7 @@ class LabelsInjector {
   // corresponds to the CallAttemptTracer::OptionalLabelComponent enum. Returns
   // false when callback returns false.
   virtual bool AddOptionalLabels(
+      bool is_client,
       absl::Span<const std::shared_ptr<std::map<std::string, std::string>>>
           optional_labels_span,
       opentelemetry::nostd::function_ref<
@@ -94,6 +95,7 @@ class LabelsInjector {
   // Gets the actual size of the optional labels that the Plugin is going to
   // produce through the AddOptionalLabels method.
   virtual size_t GetOptionalLabelsSize(
+      bool is_client,
       absl::Span<const std::shared_ptr<std::map<std::string, std::string>>>
           optional_labels_span) const = 0;
 };

src/cpp/ext/otel/otel_server_call_tracer.cc

@@ -200,7 +200,8 @@ void OpenTelemetryServerCallTracer::RecordReceivedInitialMetadata(
     OpenTelemetryPluginState().server.call.started->Add(
         1, KeyValueIterable(/*injected_labels_iterable=*/nullptr, {},
                             additional_labels,
-                            /*active_plugin_options_view=*/nullptr, {}));
+                            /*active_plugin_options_view=*/nullptr, {},
+                            /*is_client=*/false));
   }
 }
 
@@ -222,7 +223,8 @@ void OpenTelemetryServerCallTracer::RecordEnd(
   KeyValueIterable labels(
       injected_labels_.get(), injected_labels_from_plugin_options_,
       additional_labels,
-      /*active_plugin_options_view=*/nullptr, /*optional_labels_span=*/{});
+      /*active_plugin_options_view=*/nullptr, /*optional_labels_span=*/{},
+      /*is_client=*/false);
   if (OpenTelemetryPluginState().server.call.duration != nullptr) {
     OpenTelemetryPluginState().server.call.duration->Record(
         absl::ToDoubleSeconds(elapsed_time_), labels,

test/cpp/ext/csm/metadata_exchange_test.cc

@@ -171,7 +171,7 @@ class MetadataExchangeTest
       const std::map<std::string,
                      opentelemetry::sdk::common::OwnedAttributeValue>&
           attributes,
-      bool verify_client_only_attributes = true) {
+      bool is_client) {
     EXPECT_EQ(
         absl::get<std::string>(attributes.at("csm.workload_canonical_service")),
         "canonical_service");
@@ -179,12 +179,18 @@ class MetadataExchangeTest
     EXPECT_EQ(absl::get<std::string>(
                   attributes.at("csm.remote_workload_canonical_service")),
               "canonical_service");
-    if (verify_client_only_attributes) {
+    if (is_client) {
       EXPECT_EQ(absl::get<std::string>(attributes.at("csm.service_name")),
                 "unknown");
       EXPECT_EQ(
           absl::get<std::string>(attributes.at("csm.service_namespace_name")),
           "unknown");
+    } else {
+      // The CSM optional labels should not be present in server metrics.
+      EXPECT_THAT(attributes, ::testing::Not(::testing::Contains(
+                                  ::testing::Key("csm.service_name"))));
+      EXPECT_THAT(attributes, ::testing::Not(::testing::Contains(::testing::Key(
+                                  "csm.service_namespace_name"))));
     }
     switch (GetParam().type()) {
       case TestScenario::ResourceType::kGke:
@@ -286,7 +292,7 @@ TEST_P(MetadataExchangeTest, ClientAttemptDuration) {
   EXPECT_EQ(absl::get<std::string>(attributes.at("grpc.target")),
             canonical_server_address_);
   EXPECT_EQ(absl::get<std::string>(attributes.at("grpc.status")), "OK");
-  VerifyServiceMeshAttributes(attributes);
+  VerifyServiceMeshAttributes(attributes, /*is_client=*/true);
 }
 
 // Verify that grpc.server.call.started does not get service mesh attributes
@@ -331,8 +337,7 @@ TEST_P(MetadataExchangeTest, ServerCallDuration) {
   const auto& attributes = data[kMetricName][0].attributes.GetAttributes();
   EXPECT_EQ(absl::get<std::string>(attributes.at("grpc.method")), kMethodName);
   EXPECT_EQ(absl::get<std::string>(attributes.at("grpc.status")), "OK");
-  VerifyServiceMeshAttributes(attributes,
-                              /*verify_client_only_attributes=*/false);
+  VerifyServiceMeshAttributes(attributes, /*is_client=*/false);
 }
 
 // Test that the server records unknown when the client does not send metadata

test/cpp/ext/otel/otel_plugin_test.cc

@@ -754,6 +754,7 @@ class CustomLabelInjector : public grpc::internal::LabelsInjector {
       const override {}
 
   bool AddOptionalLabels(
+      bool /*is_client*/,
       absl::Span<const std::shared_ptr<std::map<std::string, std::string>>>
       /*optional_labels_span*/,
       opentelemetry::nostd::function_ref<
@@ -764,6 +765,7 @@ class CustomLabelInjector : public grpc::internal::LabelsInjector {
   }
 
   size_t GetOptionalLabelsSize(
+      bool /*is_client*/,
       absl::Span<const std::shared_ptr<std::map<std::string, std::string>>>
       /*optional_labels_span*/) const override {
     return 0;