[OTel] Add a PluginOption API (#35434)

Closes #35434

COPYBARA_INTEGRATE_REVIEW=https://github.com/grpc/grpc/pull/35434 from yashykt:OTelPluginOption 1db870bed4
PiperOrigin-RevId: 596966190

include/grpcpp/ext/csm_observability.h

@@ -28,6 +28,8 @@
 #include "absl/strings/string_view.h"
 #include "opentelemetry/sdk/metrics/meter_provider.h"
 
+#include <grpcpp/ext/otel_plugin.h>
+
 namespace grpc {
 
 namespace internal {
@@ -88,6 +90,18 @@ class CsmObservabilityBuilder {
   std::unique_ptr<grpc::internal::OpenTelemetryPluginBuilderImpl> builder_;
 };
 
+class OpenTelemetryPluginOption;
+
+/// Creates an OpenTelemetryPluginOption that would add additional labels on
+/// gRPC metrics to enhance observability for CSM users.
+///
+/// Sample Usage -
+/// OpenTelemetryPluginBuilder()
+///     .SetMeterProvider(provider)
+///     .AddPluginOption(MakeCsmOpenTelemetryPluginOption())
+///     .BuildAndRegisterGlobal();
+std::unique_ptr<OpenTelemetryPluginOption> MakeCsmOpenTelemetryPluginOption();
+
 }  // namespace experimental
 }  // namespace grpc
 

include/grpcpp/ext/otel_plugin.h

@@ -38,6 +38,11 @@ class OpenTelemetryPluginBuilderImpl;
 
 namespace experimental {
 
+class OpenTelemetryPluginOption {
+ public:
+  virtual ~OpenTelemetryPluginOption() = default;
+};
+
 /// The most common way to use this API is -
 ///
 /// OpenTelemetryPluginBuilder().SetMeterProvider(provider).BuildAndRegister();
@@ -76,6 +81,7 @@ class OpenTelemetryPluginBuilder {
           "grpc.server.call.rcvd_total_compressed_message_size";
 
   OpenTelemetryPluginBuilder();
+  ~OpenTelemetryPluginBuilder();
   /// If `SetMeterProvider()` is not called, no metrics are collected.
   OpenTelemetryPluginBuilder& SetMeterProvider(
       std::shared_ptr<opentelemetry::metrics::MeterProvider> meter_provider);
@@ -95,6 +101,12 @@ class OpenTelemetryPluginBuilder {
   OpenTelemetryPluginBuilder& SetGenericMethodAttributeFilter(
       absl::AnyInvocable<bool(absl::string_view /*generic_method*/) const>
           generic_method_attribute_filter);
+  /// Add a plugin option to add to the opentelemetry plugin being built. At
+  /// present, this type is an opaque type. Ownership of \a option is
+  /// transferred when `AddPluginOption` is invoked. A maximum of 64 plugin
+  /// options can be added.
+  OpenTelemetryPluginBuilder& AddPluginOption(
+      std::unique_ptr<OpenTelemetryPluginOption> option);
   /// Registers a global plugin that acts on all channels and servers running on
   /// the process.
   void BuildAndRegisterGlobal();

src/core/lib/channel/call_tracer.h

@@ -175,7 +175,8 @@ class ServerCallTracerFactory {
 
   virtual ~ServerCallTracerFactory() {}
 
-  virtual ServerCallTracer* CreateNewServerCallTracer(Arena* arena) = 0;
+  virtual ServerCallTracer* CreateNewServerCallTracer(
+      Arena* arena, const ChannelArgs& channel_args) = 0;
 
   // Returns true if a server is to be traced, false otherwise.
   virtual bool IsServerTraced(const ChannelArgs& /*args*/) { return true; }

src/core/lib/surface/call.cc

@@ -859,7 +859,7 @@ grpc_error_handle FilterStackCall::Create(grpc_call_create_args* args,
         args->server->server_call_tracer_factory() != nullptr) {
       auto* server_call_tracer =
           args->server->server_call_tracer_factory()->CreateNewServerCallTracer(
-              arena);
+              arena, args->server->channel_args());
       if (server_call_tracer != nullptr) {
         // Note that we are setting both
         // GRPC_CONTEXT_CALL_TRACER_ANNOTATION_INTERFACE and
@@ -3426,7 +3426,7 @@ ServerPromiseBasedCall::ServerPromiseBasedCall(Arena* arena,
   if (args->server->server_call_tracer_factory() != nullptr) {
     auto* server_call_tracer =
         args->server->server_call_tracer_factory()->CreateNewServerCallTracer(
-            arena);
+            arena, args->server->channel_args());
     if (server_call_tracer != nullptr) {
       // Note that we are setting both
       // GRPC_CONTEXT_CALL_TRACER_ANNOTATION_INTERFACE and

src/cpp/ext/csm/csm_observability.cc

@@ -42,6 +42,58 @@
 #include "src/cpp/ext/otel/otel_plugin.h"
 
 namespace grpc {
+
+namespace internal {
+
+bool CsmServerSelector(const grpc_core::ChannelArgs& args) {
+  return args.GetBool(GRPC_ARG_XDS_ENABLED_SERVER).value_or(false);
+}
+
+bool CsmChannelTargetSelector(absl::string_view target) {
+  auto uri = grpc_core::URI::Parse(target);
+  if (!uri.ok()) {
+    gpr_log(GPR_ERROR, "Failed to parse URI: %s", std::string(target).c_str());
+    return false;
+  }
+  // CSM channels should have an "xds" scheme
+  if (uri->scheme() != "xds") {
+    return false;
+  }
+  // If set, the authority should be TD
+  if (!uri->authority().empty() &&
+      uri->authority() != "traffic-director-global.xds.googleapis.com") {
+    return false;
+  }
+  return true;
+}
+
+class CsmOpenTelemetryPluginOption
+    : public grpc::internal::InternalOpenTelemetryPluginOption {
+ public:
+  CsmOpenTelemetryPluginOption()
+      : labels_injector_(std::make_unique<internal::ServiceMeshLabelsInjector>(
+            google::cloud::otel::MakeResourceDetector()
+                ->Detect()
+                .GetAttributes())) {}
+
+  bool IsActiveOnClientChannel(absl::string_view target) const override {
+    return CsmChannelTargetSelector(target);
+  }
+
+  bool IsActiveOnServer(const grpc_core::ChannelArgs& args) const override {
+    return CsmServerSelector(args);
+  }
+
+  const grpc::internal::LabelsInjector* labels_injector() const override {
+    return labels_injector_.get();
+  }
+
+ private:
+  std::unique_ptr<internal::ServiceMeshLabelsInjector> labels_injector_;
+};
+
+}  // namespace internal
+
 namespace experimental {
 
 //
@@ -78,9 +130,7 @@ CsmObservabilityBuilder::SetGenericMethodAttributeFilter(
 }
 
 absl::StatusOr<CsmObservability> CsmObservabilityBuilder::BuildAndRegister() {
-  builder_->SetServerSelector([](const grpc_core::ChannelArgs& args) {
-    return args.GetBool(GRPC_ARG_XDS_ENABLED_SERVER).value_or(false);
-  });
+  builder_->SetServerSelector(internal::CsmServerSelector);
   builder_->SetTargetSelector(internal::CsmChannelTargetSelector);
   builder_->SetLabelsInjector(
       std::make_unique<internal::ServiceMeshLabelsInjector>(
@@ -91,27 +141,10 @@ absl::StatusOr<CsmObservability> CsmObservabilityBuilder::BuildAndRegister() {
   return CsmObservability();
 }
 
-}  // namespace experimental
-
-namespace internal {
-
-bool CsmChannelTargetSelector(absl::string_view target) {
-  auto uri = grpc_core::URI::Parse(target);
-  if (!uri.ok()) {
-    gpr_log(GPR_ERROR, "Failed to parse URI: %s", std::string(target).c_str());
-    return false;
-  }
-  // CSM channels should have an "xds" scheme
-  if (uri->scheme() != "xds") {
-    return false;
-  }
-  // If set, the authority should be TD
-  if (!uri->authority().empty() &&
-      uri->authority() != "traffic-director-global.xds.googleapis.com") {
-    return false;
-  }
-  return true;
+std::unique_ptr<OpenTelemetryPluginOption> MakeCsmOpenTelemetryPluginOption() {
+  return std::make_unique<grpc::internal::CsmOpenTelemetryPluginOption>();
 }
 
-}  // namespace internal
+}  // namespace experimental
+
 }  // namespace grpc

src/cpp/ext/csm/metadata_exchange.cc

@@ -404,7 +404,7 @@ ServiceMeshLabelsInjector::ServiceMeshLabelsInjector(
 }
 
 std::unique_ptr<LabelsIterable> ServiceMeshLabelsInjector::GetLabels(
-    grpc_metadata_batch* incoming_initial_metadata) {
+    grpc_metadata_batch* incoming_initial_metadata) const {
   auto peer_metadata =
       incoming_initial_metadata->Take(grpc_core::XEnvoyPeerMetadata());
   return std::make_unique<MeshLabelsIterable>(
@@ -414,7 +414,7 @@ std::unique_ptr<LabelsIterable> ServiceMeshLabelsInjector::GetLabels(
 
 void ServiceMeshLabelsInjector::AddLabels(
     grpc_metadata_batch* outgoing_initial_metadata,
-    LabelsIterable* labels_from_incoming_metadata) {
+    LabelsIterable* labels_from_incoming_metadata) const {
   // On the server, if the labels from incoming metadata did not have a
   // non-empty base64 encoded "x-envoy-peer-metadata", do not perform metadata
   // exchange.

src/cpp/ext/csm/metadata_exchange.h

@@ -43,12 +43,12 @@ class ServiceMeshLabelsInjector : public LabelsInjector {
   // Read the incoming initial metadata to get the set of labels to be added to
   // metrics.
   std::unique_ptr<LabelsIterable> GetLabels(
-      grpc_metadata_batch* incoming_initial_metadata) override;
+      grpc_metadata_batch* incoming_initial_metadata) const override;
 
   // Modify the outgoing initial metadata with metadata information to be sent
   // to the peer.
   void AddLabels(grpc_metadata_batch* outgoing_initial_metadata,
-                 LabelsIterable* labels_from_incoming_metadata) override;
+                 LabelsIterable* labels_from_incoming_metadata) const override;
 
  private:
   std::vector<std::pair<absl::string_view, std::string>> local_labels_;

src/cpp/ext/filters/census/server_call_tracer.cc

@@ -265,7 +265,7 @@ void OpenCensusServerCallTracer::RecordEnd(
 
 grpc_core::ServerCallTracer*
 OpenCensusServerCallTracerFactory::CreateNewServerCallTracer(
-    grpc_core::Arena* arena) {
+    grpc_core::Arena* arena, const grpc_core::ChannelArgs& /*args*/) {
   return arena->ManagedNew<OpenCensusServerCallTracer>();
 }
 

src/cpp/ext/filters/census/server_call_tracer.h

@@ -31,7 +31,7 @@ class OpenCensusServerCallTracerFactory
     : public grpc_core::ServerCallTracerFactory {
  public:
   grpc_core::ServerCallTracer* CreateNewServerCallTracer(
-      grpc_core::Arena* arena) override;
+      grpc_core::Arena* arena, const grpc_core::ChannelArgs& /*args*/) override;
 };
 
 }  // namespace internal

src/cpp/ext/otel/key_value_iterable.h

@@ -50,9 +50,13 @@ class KeyValueIterable : public opentelemetry::common::KeyValueIterable {
  public:
   explicit KeyValueIterable(
       LabelsIterable* injected_labels_iterable,
+      const std::vector<std::unique_ptr<LabelsIterable>>&
+          injected_labels_from_plugin_options,
       absl::Span<const std::pair<absl::string_view, absl::string_view>>
           additional_labels)
       : injected_labels_iterable_(injected_labels_iterable),
+        injected_labels_from_plugin_options_(
+            injected_labels_from_plugin_options),
         additional_labels_(additional_labels) {}
 
   bool ForEachKeyValue(opentelemetry::nostd::function_ref<
@@ -68,6 +72,18 @@ class KeyValueIterable : public opentelemetry::common::KeyValueIterable {
         }
       }
     }
+    for (const auto& plugin_option_injected_iterable :
+         injected_labels_from_plugin_options_) {
+      if (plugin_option_injected_iterable != nullptr) {
+        plugin_option_injected_iterable->ResetIteratorPosition();
+        while (const auto& pair = plugin_option_injected_iterable->Next()) {
+          if (!callback(AbslStrViewToOpenTelemetryStrView(pair->first),
+                        AbslStrViewToOpenTelemetryStrView(pair->second))) {
+            return false;
+          }
+        }
+      }
+    }
     for (const auto& pair : additional_labels_) {
       if (!callback(AbslStrViewToOpenTelemetryStrView(pair.first),
                     AbslStrViewToOpenTelemetryStrView(pair.second))) {
@@ -78,14 +94,23 @@ class KeyValueIterable : public opentelemetry::common::KeyValueIterable {
   }
 
   size_t size() const noexcept override {
-    return (injected_labels_iterable_ != nullptr
-                ? injected_labels_iterable_->Size()
-                : 0) +
-           additional_labels_.size();
+    size_t size = injected_labels_iterable_ != nullptr
+                      ? injected_labels_iterable_->Size()
+                      : 0;
+    for (const auto& plugin_option_injected_iterable :
+         injected_labels_from_plugin_options_) {
+      if (plugin_option_injected_iterable != nullptr) {
+        size += plugin_option_injected_iterable->Size();
+      }
+    }
+    size += additional_labels_.size();
+    return size;
   }
 
  private:
   LabelsIterable* injected_labels_iterable_;
+  const std::vector<std::unique_ptr<LabelsIterable>>&
+      injected_labels_from_plugin_options_;
   absl::Span<const std::pair<absl::string_view, absl::string_view>>
       additional_labels_;
 };

src/cpp/ext/otel/otel_call_tracer.h

@@ -98,6 +98,8 @@ class OpenTelemetryCallTracer : public grpc_core::ClientCallTracer {
     // Start time (for measuring latency).
     absl::Time start_time_;
     std::unique_ptr<LabelsIterable> injected_labels_;
+    std::vector<std::unique_ptr<LabelsIterable>>
+        injected_labels_from_plugin_options_;
   };
 
   explicit OpenTelemetryCallTracer(OpenTelemetryClientFilter* parent,

src/cpp/ext/otel/otel_client_filter.cc

@@ -73,14 +73,8 @@ const grpc_channel_filter OpenTelemetryClientFilter::kFilter =
 
 absl::StatusOr<OpenTelemetryClientFilter> OpenTelemetryClientFilter::Create(
     const grpc_core::ChannelArgs& args, ChannelFilter::Args /*filter_args*/) {
-  std::string target = args.GetOwnedString(GRPC_ARG_SERVER_URI).value_or("");
-  // Use the original target string only if a filter on the attribute is not
-  // registered or if the filter returns true, otherwise use "other".
-  if (OpenTelemetryPluginState().target_attribute_filter == nullptr ||
-      OpenTelemetryPluginState().target_attribute_filter(target)) {
-    return OpenTelemetryClientFilter(std::move(target));
-  }
-  return OpenTelemetryClientFilter("other");
+  return OpenTelemetryClientFilter(
+      args.GetOwnedString(GRPC_ARG_SERVER_URI).value_or(""));
 }
 
 grpc_core::ArenaPromise<grpc_core::ServerMetadataHandle>
@@ -106,6 +100,19 @@ OpenTelemetryClientFilter::MakeCallPromise(
   return next_promise_factory(std::move(call_args));
 }
 
+OpenTelemetryClientFilter::OpenTelemetryClientFilter(std::string target)
+    : active_plugin_options_view_(
+          ActivePluginOptionsView::MakeForClient(target)) {
+  // Use the original target string only if a filter on the attribute is not
+  // registered or if the filter returns true, otherwise use "other".
+  if (OpenTelemetryPluginState().target_attribute_filter == nullptr ||
+      OpenTelemetryPluginState().target_attribute_filter(target)) {
+    filtered_target_ = std::move(target);
+  } else {
+    filtered_target_ = "other";
+  }
+}
+
 //
 // OpenTelemetryCallTracer::OpenTelemetryCallAttemptTracer
 //
@@ -120,11 +127,11 @@ OpenTelemetryCallTracer::OpenTelemetryCallAttemptTracer::
     std::array<std::pair<absl::string_view, absl::string_view>, 2>
         additional_labels = {
             {{OpenTelemetryMethodKey(), parent_->MethodForStats()},
-             {OpenTelemetryTargetKey(), parent_->parent_->target()}}};
+             {OpenTelemetryTargetKey(), parent_->parent_->filtered_target()}}};
     // 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,
+        1, KeyValueIterable(/*injected_labels_iterable=*/nullptr, {},
                             additional_labels));
   }
 }
@@ -135,6 +142,15 @@ void OpenTelemetryCallTracer::OpenTelemetryCallAttemptTracer::
     injected_labels_ = OpenTelemetryPluginState().labels_injector->GetLabels(
         recv_initial_metadata);
   }
+  parent_->parent_->active_plugin_options_view().ForEach(
+      [&](const InternalOpenTelemetryPluginOption& plugin_option,
+          size_t /*index*/) {
+        auto* labels_injector = plugin_option.labels_injector();
+        if (labels_injector != nullptr) {
+          injected_labels_from_plugin_options_.push_back(
+              labels_injector->GetLabels(recv_initial_metadata));
+        }
+      });
 }
 
 void OpenTelemetryCallTracer::OpenTelemetryCallAttemptTracer::
@@ -143,6 +159,14 @@ void OpenTelemetryCallTracer::OpenTelemetryCallAttemptTracer::
     OpenTelemetryPluginState().labels_injector->AddLabels(send_initial_metadata,
                                                           nullptr);
   }
+  parent_->parent_->active_plugin_options_view().ForEach(
+      [&](const InternalOpenTelemetryPluginOption& plugin_option,
+          size_t /*index*/) {
+        auto* labels_injector = plugin_option.labels_injector();
+        if (labels_injector != nullptr) {
+          labels_injector->AddLabels(send_initial_metadata, nullptr);
+        }
+      });
 }
 
 void OpenTelemetryCallTracer::OpenTelemetryCallAttemptTracer::RecordSendMessage(
@@ -178,11 +202,13 @@ void OpenTelemetryCallTracer::OpenTelemetryCallAttemptTracer::
   std::array<std::pair<absl::string_view, absl::string_view>, 3>
       additional_labels = {
           {{OpenTelemetryMethodKey(), parent_->MethodForStats()},
-           {OpenTelemetryTargetKey(), parent_->parent_->target()},
+           {OpenTelemetryTargetKey(), parent_->parent_->filtered_target()},
            {OpenTelemetryStatusKey(),
             grpc_status_code_to_string(
                 static_cast<grpc_status_code>(status.code()))}}};
-  KeyValueIterable labels(injected_labels_.get(), additional_labels);
+  KeyValueIterable labels(injected_labels_.get(),
+                          injected_labels_from_plugin_options_,
+                          additional_labels);
   if (OpenTelemetryPluginState().client.attempt.duration != nullptr) {
     OpenTelemetryPluginState().client.attempt.duration->Record(
         absl::ToDoubleSeconds(absl::Now() - start_time_), labels,

src/cpp/ext/otel/otel_client_filter.h

@@ -32,6 +32,7 @@
 #include "src/core/lib/channel/promise_based_filter.h"
 #include "src/core/lib/promise/arena_promise.h"
 #include "src/core/lib/transport/transport.h"
+#include "src/cpp/ext/otel/otel_plugin.h"
 
 namespace grpc {
 namespace internal {
@@ -48,13 +49,17 @@ class OpenTelemetryClientFilter : public grpc_core::ChannelFilter {
       grpc_core::CallArgs call_args,
       grpc_core::NextPromiseFactory next_promise_factory) override;
 
-  absl::string_view target() const { return target_; }
+  absl::string_view filtered_target() const { return filtered_target_; }
+
+  const ActivePluginOptionsView& active_plugin_options_view() const {
+    return active_plugin_options_view_;
+  }
 
  private:
-  explicit OpenTelemetryClientFilter(std::string target)
-      : target_(std::move(target)) {}
+  explicit OpenTelemetryClientFilter(std::string target);
 
-  std::string target_;
+  std::string filtered_target_;
+  ActivePluginOptionsView active_plugin_options_view_;
 };
 
 }  // namespace internal

src/cpp/ext/otel/otel_plugin.cc

@@ -89,6 +89,8 @@ absl::flat_hash_set<std::string> BaseMetrics() {
 OpenTelemetryPluginBuilderImpl::OpenTelemetryPluginBuilderImpl()
     : metrics_(BaseMetrics()) {}
 
+OpenTelemetryPluginBuilderImpl::~OpenTelemetryPluginBuilderImpl() = default;
+
 OpenTelemetryPluginBuilderImpl&
 OpenTelemetryPluginBuilderImpl::SetMeterProvider(
     std::shared_ptr<opentelemetry::metrics::MeterProvider> meter_provider) {
@@ -153,6 +155,14 @@ OpenTelemetryPluginBuilderImpl::SetServerSelector(
   return *this;
 }
 
+OpenTelemetryPluginBuilderImpl& OpenTelemetryPluginBuilderImpl::AddPluginOption(
+    std::unique_ptr<InternalOpenTelemetryPluginOption> option) {
+  // We allow a limit of 64 plugin options to be registered at this time.
+  GPR_ASSERT(plugin_options_.size() < 64);
+  plugin_options_.push_back(std::move(option));
+  return *this;
+}
+
 void OpenTelemetryPluginBuilderImpl::BuildAndRegisterGlobal() {
   opentelemetry::nostd::shared_ptr<opentelemetry::metrics::MeterProvider>
       meter_provider = meter_provider_;
@@ -239,6 +249,7 @@ void OpenTelemetryPluginBuilderImpl::BuildAndRegisterGlobal() {
   g_otel_plugin_state_->generic_method_attribute_filter =
       std::move(generic_method_attribute_filter_);
   g_otel_plugin_state_->meter_provider = std::move(meter_provider);
+  g_otel_plugin_state_->plugin_options = std::move(plugin_options_);
   grpc_core::ServerCallTracerFactory::RegisterGlobal(
       new grpc::internal::OpenTelemetryServerCallTracerFactory());
   grpc_core::CoreConfiguration::RegisterBuilder(
@@ -287,6 +298,8 @@ constexpr absl::string_view OpenTelemetryPluginBuilder::
 OpenTelemetryPluginBuilder::OpenTelemetryPluginBuilder()
     : impl_(std::make_unique<internal::OpenTelemetryPluginBuilderImpl>()) {}
 
+OpenTelemetryPluginBuilder::~OpenTelemetryPluginBuilder() = default;
+
 OpenTelemetryPluginBuilder& OpenTelemetryPluginBuilder::SetMeterProvider(
     std::shared_ptr<opentelemetry::metrics::MeterProvider> meter_provider) {
   impl_->SetMeterProvider(std::move(meter_provider));
@@ -310,6 +323,15 @@ OpenTelemetryPluginBuilder::SetGenericMethodAttributeFilter(
   return *this;
 }
 
+OpenTelemetryPluginBuilder& OpenTelemetryPluginBuilder::AddPluginOption(
+    std::unique_ptr<OpenTelemetryPluginOption> option) {
+  impl_->AddPluginOption(
+      std::unique_ptr<grpc::internal::InternalOpenTelemetryPluginOption>(
+          static_cast<grpc::internal::InternalOpenTelemetryPluginOption*>(
+              option.release())));
+  return *this;
+}
+
 void OpenTelemetryPluginBuilder::BuildAndRegisterGlobal() {
   impl_->BuildAndRegisterGlobal();
 }

src/cpp/ext/otel/otel_plugin.h

@@ -24,6 +24,7 @@
 #include <stddef.h>
 #include <stdint.h>
 
+#include <bitset>
 #include <memory>
 #include <string>
 #include <utility>
@@ -36,6 +37,8 @@
 #include "opentelemetry/metrics/sync_instruments.h"
 #include "opentelemetry/nostd/shared_ptr.h"
 
+#include <grpcpp/ext/otel_plugin.h>
+
 #include "src/core/lib/channel/channel_args.h"
 #include "src/core/lib/transport/metadata_batch.h"
 
@@ -67,14 +70,27 @@ class LabelsInjector {
   // Read the incoming initial metadata to get the set of labels to be added to
   // metrics.
   virtual std::unique_ptr<LabelsIterable> GetLabels(
-      grpc_metadata_batch* incoming_initial_metadata) = 0;
+      grpc_metadata_batch* incoming_initial_metadata) const = 0;
 
   // Modify the outgoing initial metadata with metadata information to be sent
   // to the peer. On the server side, \a labels_from_incoming_metadata returned
   // from `GetLabels` should be provided as input here. On the client side, this
   // should be nullptr.
-  virtual void AddLabels(grpc_metadata_batch* outgoing_initial_metadata,
-                         LabelsIterable* labels_from_incoming_metadata) = 0;
+  virtual void AddLabels(
+      grpc_metadata_batch* outgoing_initial_metadata,
+      LabelsIterable* labels_from_incoming_metadata) const = 0;
+};
+
+class InternalOpenTelemetryPluginOption
+    : public grpc::experimental::OpenTelemetryPluginOption {
+ public:
+  ~InternalOpenTelemetryPluginOption() override = default;
+  // Determines whether a plugin option is active on a given channel target
+  virtual bool IsActiveOnClientChannel(absl::string_view target) const = 0;
+  // Determines whether a plugin option is active on a given server
+  virtual bool IsActiveOnServer(const grpc_core::ChannelArgs& args) const = 0;
+  // Returns the LabelsInjector used by this plugin option, nullptr if none.
+  virtual const grpc::internal::LabelsInjector* labels_injector() const = 0;
 };
 
 struct OpenTelemetryPluginState {
@@ -107,6 +123,8 @@ struct OpenTelemetryPluginState {
       generic_method_attribute_filter;
   absl::AnyInvocable<bool(const grpc_core::ChannelArgs& /*args*/) const>
       server_selector;
+  std::vector<std::unique_ptr<InternalOpenTelemetryPluginOption>>
+      plugin_options;
 };
 
 const struct OpenTelemetryPluginState& OpenTelemetryPluginState();
@@ -119,6 +137,7 @@ absl::string_view OpenTelemetryTargetKey();
 class OpenTelemetryPluginBuilderImpl {
  public:
   OpenTelemetryPluginBuilderImpl();
+  ~OpenTelemetryPluginBuilderImpl();
   // If `SetMeterProvider()` is not called, no metrics are collected.
   OpenTelemetryPluginBuilderImpl& SetMeterProvider(
       std::shared_ptr<opentelemetry::metrics::MeterProvider> meter_provider);
@@ -166,6 +185,8 @@ class OpenTelemetryPluginBuilderImpl {
   OpenTelemetryPluginBuilderImpl& SetGenericMethodAttributeFilter(
       absl::AnyInvocable<bool(absl::string_view /*generic_method*/) const>
           generic_method_attribute_filter);
+  OpenTelemetryPluginBuilderImpl& AddPluginOption(
+      std::unique_ptr<InternalOpenTelemetryPluginOption> option);
   void BuildAndRegisterGlobal();
 
  private:
@@ -179,6 +200,52 @@ class OpenTelemetryPluginBuilderImpl {
       generic_method_attribute_filter_;
   absl::AnyInvocable<bool(const grpc_core::ChannelArgs& /*args*/) const>
       server_selector_;
+  std::vector<std::unique_ptr<InternalOpenTelemetryPluginOption>>
+      plugin_options_;
+};
+
+// Creates a convenience wrapper to help iterate over only those plugin options
+// that are active over a given channel/server.
+class ActivePluginOptionsView {
+ public:
+  static ActivePluginOptionsView MakeForClient(absl::string_view target) {
+    return ActivePluginOptionsView(
+        [target](const InternalOpenTelemetryPluginOption& plugin_option) {
+          return plugin_option.IsActiveOnClientChannel(target);
+        });
+  }
+
+  static ActivePluginOptionsView MakeForServer(
+      const grpc_core::ChannelArgs& args) {
+    return ActivePluginOptionsView(
+        [&args](const InternalOpenTelemetryPluginOption& plugin_option) {
+          return plugin_option.IsActiveOnServer(args);
+        });
+  }
+
+  void ForEach(
+      absl::FunctionRef<void(const InternalOpenTelemetryPluginOption&, size_t)>
+          func) const {
+    for (size_t i = 0; i < OpenTelemetryPluginState().plugin_options.size();
+         ++i) {
+      const auto& plugin_option = OpenTelemetryPluginState().plugin_options[i];
+      if (active_mask_[i]) func(*plugin_option, i);
+    }
+  }
+
+ private:
+  explicit ActivePluginOptionsView(
+      absl::FunctionRef<bool(const InternalOpenTelemetryPluginOption&)> func) {
+    for (size_t i = 0; i < OpenTelemetryPluginState().plugin_options.size();
+         ++i) {
+      const auto& plugin_option = OpenTelemetryPluginState().plugin_options[i];
+      if (plugin_option != nullptr && func(*plugin_option)) {
+        active_mask_.set(i);
+      }
+    }
+  }
+
+  std::bitset<64> active_mask_;
 };
 
 }  // namespace internal

src/cpp/ext/otel/otel_server_call_tracer.cc

@@ -56,7 +56,12 @@ namespace {
 
 class OpenTelemetryServerCallTracer : public grpc_core::ServerCallTracer {
  public:
-  OpenTelemetryServerCallTracer() : start_time_(absl::Now()) {}
+  explicit OpenTelemetryServerCallTracer(const grpc_core::ChannelArgs& args)
+      : start_time_(absl::Now()),
+        active_plugin_options_view_(
+            ActivePluginOptionsView::MakeForServer(args)),
+        injected_labels_from_plugin_options_(
+            OpenTelemetryPluginState().plugin_options.size()) {}
 
   std::string TraceId() override {
     // Not implemented
@@ -81,6 +86,16 @@ class OpenTelemetryServerCallTracer : public grpc_core::ServerCallTracer {
       OpenTelemetryPluginState().labels_injector->AddLabels(
           send_initial_metadata, injected_labels_.get());
     }
+    active_plugin_options_view_.ForEach(
+        [&](const InternalOpenTelemetryPluginOption& plugin_option,
+            size_t index) {
+          auto* labels_injector = plugin_option.labels_injector();
+          if (labels_injector != nullptr) {
+            labels_injector->AddLabels(
+                send_initial_metadata,
+                injected_labels_from_plugin_options_[index].get());
+          }
+        });
   }
 
   void RecordSendTrailingMetadata(
@@ -148,6 +163,11 @@ class OpenTelemetryServerCallTracer : public grpc_core::ServerCallTracer {
   grpc_core::Slice path_;
   std::unique_ptr<LabelsIterable> injected_labels_;
   bool registered_method_;
+  ActivePluginOptionsView active_plugin_options_view_;
+  // TODO(yashykt): It's wasteful to do this per call. When we re-haul the stats
+  // infrastructure, this should move to be done per server.
+  std::vector<std::unique_ptr<LabelsIterable>>
+      injected_labels_from_plugin_options_;
 };
 
 void OpenTelemetryServerCallTracer::RecordReceivedInitialMetadata(
@@ -158,6 +178,15 @@ void OpenTelemetryServerCallTracer::RecordReceivedInitialMetadata(
     injected_labels_ = OpenTelemetryPluginState().labels_injector->GetLabels(
         recv_initial_metadata);
   }
+  active_plugin_options_view_.ForEach(
+      [&](const InternalOpenTelemetryPluginOption& plugin_option,
+          size_t index) {
+        auto* labels_injector = plugin_option.labels_injector();
+        if (labels_injector != nullptr) {
+          injected_labels_from_plugin_options_[index] =
+              labels_injector->GetLabels(recv_initial_metadata);
+        }
+      });
   registered_method_ =
       recv_initial_metadata->get(grpc_core::GrpcRegisteredMethod())
           .value_or(nullptr) != nullptr;
@@ -167,7 +196,7 @@ void OpenTelemetryServerCallTracer::RecordReceivedInitialMetadata(
     // We might not have all the injected labels that we want at this point, so
     // avoid recording a subset of injected labels here.
     OpenTelemetryPluginState().server.call.started->Add(
-        1, KeyValueIterable(/*injected_labels_iterable=*/nullptr,
+        1, KeyValueIterable(/*injected_labels_iterable=*/nullptr, {},
                             additional_labels));
   }
 }
@@ -186,7 +215,9 @@ void OpenTelemetryServerCallTracer::RecordEnd(
           {{OpenTelemetryMethodKey(), MethodForStats()},
            {OpenTelemetryStatusKey(),
             grpc_status_code_to_string(final_info->final_status)}}};
-  KeyValueIterable labels(injected_labels_.get(), additional_labels);
+  KeyValueIterable labels(injected_labels_.get(),
+                          injected_labels_from_plugin_options_,
+                          additional_labels);
   if (OpenTelemetryPluginState().server.call.duration != nullptr) {
     OpenTelemetryPluginState().server.call.duration->Record(
         absl::ToDoubleSeconds(elapsed_time_), labels,
@@ -216,8 +247,8 @@ void OpenTelemetryServerCallTracer::RecordEnd(
 
 grpc_core::ServerCallTracer*
 OpenTelemetryServerCallTracerFactory::CreateNewServerCallTracer(
-    grpc_core::Arena* arena) {
-  return arena->ManagedNew<OpenTelemetryServerCallTracer>();
+    grpc_core::Arena* arena, const grpc_core::ChannelArgs& args) {
+  return arena->ManagedNew<OpenTelemetryServerCallTracer>(args);
 }
 
 bool OpenTelemetryServerCallTracerFactory::IsServerTraced(

src/cpp/ext/otel/otel_server_call_tracer.h

@@ -32,7 +32,8 @@ class OpenTelemetryServerCallTracerFactory
     : public grpc_core::ServerCallTracerFactory {
  public:
   grpc_core::ServerCallTracer* CreateNewServerCallTracer(
-      grpc_core::Arena* arena) override;
+      grpc_core::Arena* arena,
+      const grpc_core::ChannelArgs& channel_args) override;
 
   bool IsServerTraced(const grpc_core::ChannelArgs& args) override;
 };

src/python/grpcio_observability/grpc_observability/server_call_tracer.cc

@@ -261,10 +261,11 @@ void PythonOpenCensusServerCallTracer::RecordEnd(
 
 grpc_core::ServerCallTracer*
 PythonOpenCensusServerCallTracerFactory::CreateNewServerCallTracer(
-    grpc_core::Arena* arena) {
+    grpc_core::Arena* arena, const grpc_core::ChannelArgs& channel_args) {
   // We don't use arena here to to ensure that memory is allocated and freed in
   // the same DLL in Windows.
   (void)arena;
+  (void)channel_args;
   return new PythonOpenCensusServerCallTracer();
 }
 

src/python/grpcio_observability/grpc_observability/server_call_tracer.h

@@ -30,7 +30,8 @@ class PythonOpenCensusServerCallTracerFactory
     : public grpc_core::ServerCallTracerFactory {
  public:
   grpc_core::ServerCallTracer* CreateNewServerCallTracer(
-      grpc_core::Arena* arena) override;
+      grpc_core::Arena* arena,
+      const grpc_core::ChannelArgs& channel_args) override;
 };
 
 inline absl::string_view GetMethod(const grpc_core::Slice& path) {

test/core/channel/server_call_tracer_factory_test.cc

@@ -26,7 +26,8 @@ namespace {
 
 class TestServerCallTracerFactory : public ServerCallTracerFactory {
  public:
-  ServerCallTracer* CreateNewServerCallTracer(Arena* /*arena*/) override {
+  ServerCallTracer* CreateNewServerCallTracer(
+      Arena* /*arena*/, const ChannelArgs& /*args*/) override {
     Crash("Not implemented");
   }
 };

test/core/end2end/tests/http2_stats.cc

@@ -200,7 +200,8 @@ grpc_transport_stream_stats FakeServerCallTracer::transport_stream_stats_;
 
 class FakeServerCallTracerFactory : public ServerCallTracerFactory {
  public:
-  ServerCallTracer* CreateNewServerCallTracer(Arena* arena) override {
+  ServerCallTracer* CreateNewServerCallTracer(
+      Arena* arena, const ChannelArgs& /*args*/) override {
     return arena->ManagedNew<FakeServerCallTracer>();
   }
 };

test/cpp/ext/csm/BUILD

@@ -36,6 +36,8 @@ grpc_cc_test(
     ],
     deps = [
         "//:grpc++",
+        "//:grpcpp_csm_observability",
+        "//:grpcpp_otel_plugin",
         "//src/cpp/ext/csm:csm_observability",
         "//src/cpp/ext/otel:otel_plugin",
         "//test/core/util:grpc_test_util",

test/cpp/ext/csm/csm_observability_test.cc

@@ -22,6 +22,7 @@
 #include "gtest/gtest.h"
 
 #include <grpcpp/ext/csm_observability.h>
+#include <grpcpp/ext/otel_plugin.h>
 
 #include "src/core/lib/gprpp/env.h"
 #include "test/core/util/test_config.h"
@@ -62,6 +63,12 @@ TEST(CsmChannelTargetSelectorTest, XdsTargetsWithTDAuthority) {
       "xds://traffic-director-global.xds.googleapis.com/foo"));
 }
 
+TEST(CsmPluginOptionTest, Basic) {
+  experimental::OpenTelemetryPluginBuilder()
+      .AddPluginOption(experimental::MakeCsmOpenTelemetryPluginOption())
+      .BuildAndRegisterGlobal();
+}
+
 }  // namespace
 }  // namespace testing
 }  // namespace grpc

test/cpp/ext/otel/otel_plugin_test.cc

@@ -653,6 +653,282 @@ TEST_F(OpenTelemetryPluginEnd2EndTest,
   EXPECT_EQ(*status_value, "UNIMPLEMENTED");
 }
 
+using OpenTelemetryPluginOptionEnd2EndTest = OpenTelemetryPluginEnd2EndTest;
+
+class SimpleLabelIterable : public grpc::internal::LabelsIterable {
+ public:
+  explicit SimpleLabelIterable(
+      std::pair<absl::string_view, absl::string_view> label)
+      : label_(label) {}
+
+  absl::optional<std::pair<absl::string_view, absl::string_view>> Next()
+      override {
+    if (iterated_) {
+      return absl::nullopt;
+    }
+    iterated_ = true;
+    return label_;
+  }
+
+  size_t Size() const override { return 1; }
+
+  void ResetIteratorPosition() override { iterated_ = false; }
+
+ private:
+  bool iterated_ = false;
+  std::pair<absl::string_view, absl::string_view> label_;
+};
+
+class CustomLabelInjector : public grpc::internal::LabelsInjector {
+ public:
+  explicit CustomLabelInjector(std::pair<std::string, std::string> label)
+      : label_(std::move(label)) {}
+  ~CustomLabelInjector() override {}
+
+  std::unique_ptr<grpc::internal::LabelsIterable> GetLabels(
+      grpc_metadata_batch* /*incoming_initial_metadata*/) const override {
+    return std::make_unique<SimpleLabelIterable>(label_);
+  }
+
+  void AddLabels(
+      grpc_metadata_batch* /*outgoing_initial_metadata*/,
+      grpc::internal::LabelsIterable* /*labels_from_incoming_metadata*/)
+      const override {}
+
+ private:
+  std::pair<std::string, std::string> label_;
+};
+
+class CustomPluginOption
+    : public grpc::internal::InternalOpenTelemetryPluginOption {
+ public:
+  CustomPluginOption(bool enabled_on_client, bool enabled_on_server,
+                     std::pair<std::string, std::string> label)
+      : enabled_on_client_(enabled_on_client),
+        enabled_on_server_(enabled_on_server),
+        label_injector_(
+            std::make_unique<CustomLabelInjector>(std::move(label))) {}
+
+  ~CustomPluginOption() override {}
+
+  bool IsActiveOnClientChannel(absl::string_view /*target*/) const override {
+    return enabled_on_client_;
+  }
+
+  bool IsActiveOnServer(const grpc_core::ChannelArgs& /*args*/) const override {
+    return enabled_on_server_;
+  }
+
+  const grpc::internal::LabelsInjector* labels_injector() const override {
+    return label_injector_.get();
+  }
+
+ private:
+  bool enabled_on_client_;
+  bool enabled_on_server_;
+  std::unique_ptr<CustomLabelInjector> label_injector_;
+};
+
+TEST_F(OpenTelemetryPluginOptionEnd2EndTest, Basic) {
+  std::vector<
+      std::unique_ptr<grpc::internal::InternalOpenTelemetryPluginOption>>
+      plugin_option_list;
+  plugin_option_list.emplace_back(std::make_unique<CustomPluginOption>(
+      /*enabled_on_client*/ true, /*enabled_on_server*/ true,
+      std::make_pair("key", "value")));
+  Init({grpc::experimental::OpenTelemetryPluginBuilder::
+            kClientAttemptDurationInstrumentName,
+        grpc::experimental::OpenTelemetryPluginBuilder::
+            kServerCallDurationInstrumentName},
+       /*resource=*/opentelemetry::sdk::resource::Resource::Create({}),
+       /*labels_injector=*/nullptr,
+       /*test_no_meter_provider=*/false,
+       /*target_selector=*/absl::AnyInvocable<bool(absl::string_view) const>(),
+       /*target_attribute_filter=*/
+       absl::AnyInvocable<bool(absl::string_view) const>(),
+       /*generic_method_attribute_filter=*/
+       absl::AnyInvocable<bool(absl::string_view /*generic_method*/) const>(),
+       /*plugin_options=*/std::move(plugin_option_list));
+  SendRPC();
+  auto data = ReadCurrentMetricsData(
+      [&](const absl::flat_hash_map<
+          std::string,
+          std::vector<opentelemetry::sdk::metrics::PointDataAttributes>>&
+              data) {
+        return !data.contains("grpc.client.attempt.duration") ||
+               !data.contains("grpc.server.call.duration");
+      });
+  // Verify client side metric
+  ASSERT_EQ(data["grpc.client.attempt.duration"].size(), 1);
+  const auto& client_attributes =
+      data["grpc.client.attempt.duration"][0].attributes.GetAttributes();
+  EXPECT_EQ(client_attributes.size(), 4);
+  EXPECT_EQ(absl::get<std::string>(client_attributes.at("key")), "value");
+  // Verify server side metric
+  ASSERT_EQ(data["grpc.server.call.duration"].size(), 1);
+  const auto& server_attributes =
+      data["grpc.server.call.duration"][0].attributes.GetAttributes();
+  EXPECT_EQ(server_attributes.size(), 3);
+  EXPECT_EQ(absl::get<std::string>(server_attributes.at("key")), "value");
+}
+
+TEST_F(OpenTelemetryPluginOptionEnd2EndTest, ClientOnlyPluginOption) {
+  std::vector<
+      std::unique_ptr<grpc::internal::InternalOpenTelemetryPluginOption>>
+      plugin_option_list;
+  plugin_option_list.emplace_back(std::make_unique<CustomPluginOption>(
+      /*enabled_on_client*/ true, /*enabled_on_server*/ false,
+      std::make_pair("key", "value")));
+  Init({grpc::experimental::OpenTelemetryPluginBuilder::
+            kClientAttemptDurationInstrumentName,
+        grpc::experimental::OpenTelemetryPluginBuilder::
+            kServerCallDurationInstrumentName},
+       /*resource=*/opentelemetry::sdk::resource::Resource::Create({}),
+       /*labels_injector=*/nullptr,
+       /*test_no_meter_provider=*/false,
+       /*target_selector=*/absl::AnyInvocable<bool(absl::string_view) const>(),
+       /*target_attribute_filter=*/
+       absl::AnyInvocable<bool(absl::string_view) const>(),
+       /*generic_method_attribute_filter=*/
+       absl::AnyInvocable<bool(absl::string_view /*generic_method*/) const>(),
+       /*plugin_options=*/std::move(plugin_option_list));
+  SendRPC();
+  auto data = ReadCurrentMetricsData(
+      [&](const absl::flat_hash_map<
+          std::string,
+          std::vector<opentelemetry::sdk::metrics::PointDataAttributes>>&
+              data) {
+        return !data.contains("grpc.client.attempt.duration") ||
+               !data.contains("grpc.server.call.duration");
+      });
+  // Verify client side metric
+  ASSERT_EQ(data["grpc.client.attempt.duration"].size(), 1);
+  const auto& client_attributes =
+      data["grpc.client.attempt.duration"][0].attributes.GetAttributes();
+  EXPECT_EQ(client_attributes.size(), 4);
+  EXPECT_EQ(absl::get<std::string>(client_attributes.at("key")), "value");
+  // Verify server side metric
+  ASSERT_EQ(data["grpc.server.call.duration"].size(), 1);
+  const auto& server_attributes =
+      data["grpc.server.call.duration"][0].attributes.GetAttributes();
+  EXPECT_EQ(server_attributes.size(), 2);
+  EXPECT_THAT(server_attributes,
+              ::testing::Not(::testing::Contains(::testing::Key("key"))));
+}
+
+TEST_F(OpenTelemetryPluginOptionEnd2EndTest, ServerOnlyPluginOption) {
+  std::vector<
+      std::unique_ptr<grpc::internal::InternalOpenTelemetryPluginOption>>
+      plugin_option_list;
+  plugin_option_list.emplace_back(std::make_unique<CustomPluginOption>(
+      /*enabled_on_client*/ false, /*enabled_on_server*/ true,
+      std::make_pair("key", "value")));
+  Init({grpc::experimental::OpenTelemetryPluginBuilder::
+            kClientAttemptDurationInstrumentName,
+        grpc::experimental::OpenTelemetryPluginBuilder::
+            kServerCallDurationInstrumentName},
+       /*resource=*/opentelemetry::sdk::resource::Resource::Create({}),
+       /*labels_injector=*/nullptr,
+       /*test_no_meter_provider=*/false,
+       /*target_selector=*/absl::AnyInvocable<bool(absl::string_view) const>(),
+       /*target_attribute_filter=*/
+       absl::AnyInvocable<bool(absl::string_view) const>(),
+       /*generic_method_attribute_filter=*/
+       absl::AnyInvocable<bool(absl::string_view /*generic_method*/) const>(),
+       /*plugin_options=*/std::move(plugin_option_list));
+  SendRPC();
+  auto data = ReadCurrentMetricsData(
+      [&](const absl::flat_hash_map<
+          std::string,
+          std::vector<opentelemetry::sdk::metrics::PointDataAttributes>>&
+              data) {
+        return !data.contains("grpc.client.attempt.duration") ||
+               !data.contains("grpc.server.call.duration");
+      });
+  // Verify client side metric
+  ASSERT_EQ(data["grpc.client.attempt.duration"].size(), 1);
+  const auto& attributes =
+      data["grpc.client.attempt.duration"][0].attributes.GetAttributes();
+  EXPECT_EQ(attributes.size(), 3);
+  EXPECT_THAT(attributes,
+              ::testing::Not(::testing::Contains(::testing::Key("key"))));
+  // Verify server side metric
+  ASSERT_EQ(data["grpc.server.call.duration"].size(), 1);
+  const auto& server_attributes =
+      data["grpc.server.call.duration"][0].attributes.GetAttributes();
+  EXPECT_EQ(server_attributes.size(), 3);
+  EXPECT_EQ(absl::get<std::string>(server_attributes.at("key")), "value");
+}
+
+TEST_F(OpenTelemetryPluginOptionEnd2EndTest,
+       MultipleEnabledAndDisabledPluginOptions) {
+  std::vector<
+      std::unique_ptr<grpc::internal::InternalOpenTelemetryPluginOption>>
+      plugin_option_list;
+  plugin_option_list.reserve(5);
+  plugin_option_list.emplace_back(std::make_unique<CustomPluginOption>(
+      /*enabled_on_client*/ true, /*enabled_on_server*/ true,
+      std::make_pair("key1", "value1")));
+  plugin_option_list.emplace_back(std::make_unique<CustomPluginOption>(
+      /*enabled_on_client*/ true, /*enabled_on_server*/ false,
+      std::make_pair("key2", "value2")));
+  plugin_option_list.emplace_back(std::make_unique<CustomPluginOption>(
+      /*enabled_on_client*/ true, /*enabled_on_server*/ false,
+      std::make_pair("key3", "value3")));
+  plugin_option_list.emplace_back(std::make_unique<CustomPluginOption>(
+      /*enabled_on_client*/ false, /*enabled_on_server*/ true,
+      std::make_pair("key4", "value4")));
+  plugin_option_list.emplace_back(std::make_unique<CustomPluginOption>(
+      /*enabled_on_client*/ false, /*enabled_on_server*/ true,
+      std::make_pair("key5", "value5")));
+  Init({grpc::experimental::OpenTelemetryPluginBuilder::
+            kClientAttemptDurationInstrumentName,
+        grpc::experimental::OpenTelemetryPluginBuilder::
+            kServerCallDurationInstrumentName},
+       /*resource=*/opentelemetry::sdk::resource::Resource::Create({}),
+       /*labels_injector=*/nullptr,
+       /*test_no_meter_provider=*/false,
+       /*target_selector=*/absl::AnyInvocable<bool(absl::string_view) const>(),
+       /*target_attribute_filter=*/
+       absl::AnyInvocable<bool(absl::string_view) const>(),
+       /*generic_method_attribute_filter=*/
+       absl::AnyInvocable<bool(absl::string_view /*generic_method*/) const>(),
+       /*plugin_options=*/std::move(plugin_option_list));
+  SendRPC();
+  auto data = ReadCurrentMetricsData(
+      [&](const absl::flat_hash_map<
+          std::string,
+          std::vector<opentelemetry::sdk::metrics::PointDataAttributes>>&
+              data) {
+        return !data.contains("grpc.client.attempt.duration") ||
+               !data.contains("grpc.server.call.duration");
+      });
+  // Verify client side metric
+  ASSERT_EQ(data["grpc.client.attempt.duration"].size(), 1);
+  const auto& client_attributes =
+      data["grpc.client.attempt.duration"][0].attributes.GetAttributes();
+  EXPECT_EQ(client_attributes.size(), 6);
+  EXPECT_EQ(absl::get<std::string>(client_attributes.at("key1")), "value1");
+  EXPECT_EQ(absl::get<std::string>(client_attributes.at("key2")), "value2");
+  EXPECT_EQ(absl::get<std::string>(client_attributes.at("key3")), "value3");
+  EXPECT_THAT(client_attributes,
+              ::testing::Not(::testing::Contains(::testing::Key("key4"))));
+  EXPECT_THAT(client_attributes,
+              ::testing::Not(::testing::Contains(::testing::Key("key5"))));
+  // Verify server side metric
+  ASSERT_EQ(data["grpc.server.call.duration"].size(), 1);
+  const auto& server_attributes =
+      data["grpc.server.call.duration"][0].attributes.GetAttributes();
+  EXPECT_EQ(server_attributes.size(), 5);
+  EXPECT_EQ(absl::get<std::string>(server_attributes.at("key1")), "value1");
+  EXPECT_THAT(server_attributes,
+              ::testing::Not(::testing::Contains(::testing::Key("key2"))));
+  EXPECT_THAT(server_attributes,
+              ::testing::Not(::testing::Contains(::testing::Key("key3"))));
+  EXPECT_EQ(absl::get<std::string>(server_attributes.at("key4")), "value4");
+  EXPECT_EQ(absl::get<std::string>(server_attributes.at("key5")), "value5");
+}
+
 }  // namespace
 }  // namespace testing
 }  // namespace grpc

test/cpp/ext/otel/otel_test_library.cc

@@ -48,7 +48,10 @@ void OpenTelemetryPluginEnd2EndTest::Init(
     absl::AnyInvocable<bool(absl::string_view /*target*/) const>
         target_attribute_filter,
     absl::AnyInvocable<bool(absl::string_view /*generic_method*/) const>
-        generic_method_attribute_filter) {
+        generic_method_attribute_filter,
+    std::vector<
+        std::unique_ptr<grpc::internal::InternalOpenTelemetryPluginOption>>
+        plugin_options) {
   // We are resetting the MeterProvider and OpenTelemetry plugin at the start
   // of each test to avoid test results from one test carrying over to another
   // test. (Some measurements can get arbitrarily delayed.)
@@ -76,6 +79,9 @@ void OpenTelemetryPluginEnd2EndTest::Init(
   ot_builder.SetTargetAttributeFilter(std::move(target_attribute_filter));
   ot_builder.SetGenericMethodAttributeFilter(
       std::move(generic_method_attribute_filter));
+  for (auto& option : plugin_options) {
+    ot_builder.AddPluginOption(std::move(option));
+  }
   ot_builder.BuildAndRegisterGlobal();
   grpc_init();
   grpc::ServerBuilder builder;

test/cpp/ext/otel/otel_test_library.h

@@ -70,7 +70,10 @@ class OpenTelemetryPluginEnd2EndTest : public ::testing::Test {
               absl::AnyInvocable<bool(absl::string_view) const>(),
       absl::AnyInvocable<bool(absl::string_view /*generic_method*/) const>
           generic_method_attribute_filter = absl::AnyInvocable<
-              bool(absl::string_view /*generic_method*/) const>());
+              bool(absl::string_view /*generic_method*/) const>(),
+      std::vector<
+          std::unique_ptr<grpc::internal::InternalOpenTelemetryPluginOption>>
+          plugin_options = {});
 
   void TearDown() override;
 

test/cpp/interop/observability_client.cc

@@ -30,10 +30,10 @@
 #include <grpcpp/channel.h>
 #include <grpcpp/client_context.h>
 #include <grpcpp/ext/gcp_observability.h>
+#include <grpcpp/ext/otel_plugin.h>
 
 #include "src/core/lib/gpr/string.h"
 #include "src/core/lib/gprpp/crash.h"
-#include "src/cpp/ext/otel/otel_plugin.h"
 #include "test/core/util/test_config.h"
 #include "test/cpp/interop/client_helper.h"
 #include "test/cpp/interop/interop_client.h"
@@ -236,7 +236,7 @@ int main(int argc, char** argv) {
     auto meter_provider =
         std::make_shared<opentelemetry::sdk::metrics::MeterProvider>();
     meter_provider->AddMetricReader(std::move(prometheus_exporter));
-    grpc::internal::OpenTelemetryPluginBuilderImpl otel_builder;
+    grpc::experimental::OpenTelemetryPluginBuilder otel_builder;
     otel_builder.SetMeterProvider(std::move(meter_provider));
     otel_builder.BuildAndRegisterGlobal();
   }