diff --git a/ci/cloudbuild/builds/observability.sh b/ci/cloudbuild/builds/observability.sh
index ac661b3b1b4be..1b9df7602ba5f 100755
--- a/ci/cloudbuild/builds/observability.sh
+++ b/ci/cloudbuild/builds/observability.sh
@@ -26,17 +26,205 @@ export CC=clang
 export CXX=clang++
 
 mapfile -t args < <(bazel::common_args)
-args+=("--//google/cloud/bigtable:enable_metrics=true")
+args+=(
+  "--//google/cloud/bigtable:enable_metrics=true"
+  "--dynamic_mode=off"
+)
+
 mapfile -t integration_args < <(integration::bazel_args)
 
+observability_key_base="observability-key-$(date +"%Y-%m")"
+readonly KEY_DIR="/dev/shm"
+readonly SECRETS_BUCKET="gs://cloud-cpp-testing-resources-secrets"
+gcloud storage cp --quiet "${SECRETS_BUCKET}/${observability_key_base}.json" "${KEY_DIR}/${observability_key_base}.json" >/dev/null 2>&1 || true
+if [[ -r "${KEY_DIR}/${observability_key_base}.json" ]]; then
+  GOOGLE_CLOUD_CPP_TEST_OBSERVABILITY_KEY_FILE_JSON="${KEY_DIR}/${observability_key_base}.json"
+fi
+
+ORIGINAL_ACCOUNT="$(gcloud config get-value account 2>/dev/null || true)"
+
+# Activate dedicated observability service account credentials for GCE VM & GCS management
+KEY_FILE=""
+for candidate in \
+  "${GOOGLE_CLOUD_CPP_TEST_OBSERVABILITY_KEY_FILE_JSON:-}" \
+  "${GOOGLE_APPLICATION_CREDENTIALS:-}"; do
+  if [[ -n "${candidate}" && -r "${candidate}" ]]; then
+    KEY_FILE="${candidate}"
+    break
+  fi
+done
+
+if [[ -n "${KEY_FILE}" ]]; then
+  io::log_h2 "Activating gcloud service account from ${KEY_FILE}"
+  gcloud auth activate-service-account --key-file="${KEY_FILE}" --quiet || true
+else
+  io::log_yellow "No dedicated observability service account keyfile found; using default gcloud auth."
+fi
+
 io::log_h2 "Building OtelCollector targets and Bigtable Observability integration tests"
 io::run bazel build "${args[@]}" \
   //ci/otel_collector:otel_collector_main \
   //google/cloud/bigtable/tests:observability_integration_test-default \
   //google/cloud/bigtable/tests:observability_integration_test-dynamic-pool
 
-io::log_h2 "Running Bigtable Observability Integration Tests against production endpoint"
-io::run bazel test "${args[@]}" "${integration_args[@]}" \
-  --cache_test_results="auto" \
-  -- //google/cloud/bigtable/tests:observability_integration_test-default \
-  //google/cloud/bigtable/tests:observability_integration_test-dynamic-pool
+BAZEL_BIN="$(bazel info "${args[@]}" bazel-bin)"
+TEST_DEFAULT_BIN="${BAZEL_BIN}/google/cloud/bigtable/tests/observability_integration_test-default"
+TEST_DYNAMIC_BIN="${BAZEL_BIN}/google/cloud/bigtable/tests/observability_integration_test-dynamic-pool"
+
+ZONE="us-central1-f"
+PROJECT_ID="${GOOGLE_CLOUD_PROJECT:-cloud-cpp-testing-resources}"
+BIGTABLE_INSTANCE_ID="${GOOGLE_CLOUD_CPP_BIGTABLE_TEST_INSTANCE_ID:-test-instance}"
+BIGTABLE_CLUSTER_ID="${GOOGLE_CLOUD_CPP_BIGTABLE_TEST_CLUSTER_ID:-test-cluster}"
+BIGTABLE_ZONE_A="${GOOGLE_CLOUD_CPP_BIGTABLE_TEST_ZONE_A:-us-central1-f}"
+BIGTABLE_ZONE_B="${GOOGLE_CLOUD_CPP_BIGTABLE_TEST_ZONE_B:-us-central1-b}"
+BUILD_SUFFIX="${BUILD_ID:-local}-${RANDOM}"
+VM_NAME="dp-obs-test-${BUILD_SUFFIX:0:15}"
+GCS_BUCKET="cloud-cpp-testing-resources-observability-logs"
+GCS_PATH="gs://${GCS_BUCKET}/builds/${VM_NAME}"
+VM_CREATED="false"
+
+cleanup() {
+  local exit_code=$?
+  if [[ "${VM_CREATED:-false}" == "true" ]]; then
+    io::log_h2 "Deleting ephemeral DirectPath VM: ${VM_NAME}"
+    gcloud compute instances delete "${VM_NAME}" \
+      --project="${PROJECT_ID}" \
+      --zone="${ZONE}" \
+      --quiet >/dev/null 2>&1 &
+  fi
+
+  if [[ -n "${ORIGINAL_ACCOUNT:-}" ]]; then
+    io::log_h2 "Restoring original gcloud account: ${ORIGINAL_ACCOUNT}"
+    gcloud config set account "${ORIGINAL_ACCOUNT}" --quiet >/dev/null 2>&1 || true
+  fi
+}
+trap cleanup EXIT SIGINT SIGTERM
+
+# Ensure GCS bucket and 90-day lifecycle policy exist
+if ! gcloud storage buckets describe "gs://${GCS_BUCKET}" --project="${PROJECT_ID}" >/dev/null 2>&1; then
+  io::log_h2 "Creating GCS bucket gs://${GCS_BUCKET} with 90-day lifecycle policy"
+  gcloud storage buckets create "gs://${GCS_BUCKET}" \
+    --project="${PROJECT_ID}" \
+    --location="us-central1" \
+    --uniform-bucket-level-access || true
+
+  cat <<'EOF' >/tmp/observability_lifecycle.json
+{
+  "rule": [
+    {
+      "action": {
+        "type": "Delete"
+      },
+      "condition": {
+        "age": 90
+      }
+    }
+  ]
+}
+EOF
+  gcloud storage buckets update "gs://${GCS_BUCKET}" \
+    --lifecycle-file=/tmp/observability_lifecycle.json \
+    --project="${PROJECT_ID}" || true
+fi
+
+io::log_h2 "Uploading test binaries to GCS: ${GCS_PATH}/binaries/"
+gcloud storage cp --quiet "${TEST_DEFAULT_BIN}" "${TEST_DYNAMIC_BIN}" "${GCS_PATH}/binaries/" >/dev/null 2>&1 || true
+
+# Prepare Startup Script to run on the ephemeral GCE VM
+STARTUP_SCRIPT=$(
+  cat <<EOF
+#!/bin/bash
+exec > /tmp/startup.log 2>&1
+set -x
+
+export HOME="/tmp"
+export GOOGLE_CLOUD_PROJECT="${PROJECT_ID}"
+export GOOGLE_CLOUD_CPP_BIGTABLE_TEST_INSTANCE_ID="${BIGTABLE_INSTANCE_ID}"
+export GOOGLE_CLOUD_CPP_BIGTABLE_TEST_CLUSTER_ID="${BIGTABLE_CLUSTER_ID}"
+export GOOGLE_CLOUD_CPP_BIGTABLE_TEST_ZONE_A="${BIGTABLE_ZONE_A}"
+export GOOGLE_CLOUD_CPP_BIGTABLE_TEST_ZONE_B="${BIGTABLE_ZONE_B}"
+export CBT_ENABLE_DIRECTPATH=true
+export GOOGLE_CLOUD_CPP_TEST_EXPECTED_CLIENT_PROJECT="${PROJECT_ID}"
+export GOOGLE_CLOUD_CPP_TEST_EXPECTED_LOCATION="${ZONE}"
+export GOOGLE_CLOUD_CPP_TEST_EXPECTED_CLOUD_PLATFORM="gcp_compute_engine"
+export GOOGLE_CLOUD_CPP_TEST_EXPECTED_HOSTNAME="\$(hostname)"
+
+echo "Downloading test binaries from GCS..."
+gcloud storage cp --quiet "${GCS_PATH}/binaries/*" /tmp/
+chmod +x /tmp/observability_integration_test-default
+chmod +x /tmp/observability_integration_test-dynamic-pool
+
+TEST_EXIT_CODE=0
+
+echo "Running observability_integration_test-default..."
+/tmp/observability_integration_test-default \
+  --gtest_output=xml:/tmp/test-default.xml > /tmp/test-default.log 2>&1 || TEST_EXIT_CODE=\$?
+
+echo "Running observability_integration_test-dynamic-pool..."
+/tmp/observability_integration_test-dynamic-pool \
+  --gtest_output=xml:/tmp/test-dynamic-pool.xml > /tmp/test-dynamic-pool.log 2>&1 || TEST_EXIT_CODE=\$?
+
+echo "\${TEST_EXIT_CODE}" > /tmp/exit_code.txt
+
+echo "Uploading logs and test results back to GCS..."
+gcloud storage cp --quiet /tmp/startup.log /tmp/*.log /tmp/*.xml /tmp/exit_code.txt "${GCS_PATH}/results/"
+EOF
+)
+
+io::log_h2 "Creating ephemeral DirectPath VM instance: ${VM_NAME}"
+if ! gcloud compute instances create "${VM_NAME}" \
+  --project="${PROJECT_ID}" \
+  --zone="${ZONE}" \
+  --machine-type="e2-standard-4" \
+  --subnet="projects/${PROJECT_ID}/regions/us-central1/subnetworks/directpath-subnet-ipv6" \
+  --stack-type="IPV4_IPV6" \
+  --image-family="ubuntu-2404-lts-amd64" \
+  --image-project="ubuntu-os-cloud" \
+  --metadata="startup-script=${STARTUP_SCRIPT}" \
+  --scopes="cloud-platform" \
+  --quiet >/tmp/vm_create.log 2>&1; then
+  io::log_red "Failed to create ephemeral VM instance: ${VM_NAME}"
+  cat /tmp/vm_create.log || true
+  exit 1
+fi
+VM_CREATED="true"
+
+io::log_h2 "Waiting for test execution to complete on VM ${VM_NAME}..."
+TEST_COMPLETED="false"
+for i in {1..90}; do
+  if gcloud storage cp --quiet "${GCS_PATH}/results/exit_code.txt" /tmp/exit_code.txt >/dev/null 2>&1; then
+    TEST_COMPLETED="true"
+    io::log_yellow "Test execution on VM ${VM_NAME} finished."
+    break
+  fi
+  sleep 5
+done
+
+if [[ "${TEST_COMPLETED}" != "true" ]]; then
+  io::log_red "Timed out waiting for test execution on VM ${VM_NAME}."
+  exit 1
+fi
+
+TEST_RESULT_CODE="$(cat /tmp/exit_code.txt 2>/dev/null || echo "1")"
+
+io::log_h2 "Fetching test logs from GCS: ${GCS_PATH}/results/"
+gcloud storage cp --quiet "${GCS_PATH}/results/*.log" /tmp/ 2>/dev/null || true
+
+if [[ "${TEST_RESULT_CODE}" -ne 0 ]]; then
+  for log_file in /tmp/startup.log /tmp/test-default.log /tmp/test-dynamic-pool.log; do
+    if [[ -f "${log_file}" ]]; then
+      io::log_h2 "=== Content of $(basename "${log_file}") ==="
+      cat "${log_file}" || true
+    fi
+  done
+  io::log_red "Observability integration tests failed with exit code: ${TEST_RESULT_CODE}"
+  exit "${TEST_RESULT_CODE}"
+else
+  for log_file in /tmp/test-default.log /tmp/test-dynamic-pool.log; do
+    if [[ -f "${log_file}" ]]; then
+      io::log_h2 "=== Test Summary: $(basename "${log_file}" .log) (Execution: Live Ephemeral VM - Uncached) ==="
+      grep -E '^\[(==========|----------| RUN      |       OK |  PASSED  |  FAILED  |  SKIPPED )\]' "${log_file}" || cat "${log_file}"
+    fi
+  done
+  io::log_green "Observability integration tests PASSED successfully!"
+fi
diff --git a/ci/otel_collector/BUILD.bazel b/ci/otel_collector/BUILD.bazel
index 4f1adaf496ae2..3c90872bda823 100644
--- a/ci/otel_collector/BUILD.bazel
+++ b/ci/otel_collector/BUILD.bazel
@@ -21,6 +21,7 @@ cc_library(
     srcs = ["otel_collector.cc"],
     hdrs = ["otel_collector.h"],
     deps = [
+        "//:common",
         "//protos/google/cloud/opentelemetry/testing:observability_verification_cc_grpc",
         "//protos/google/cloud/opentelemetry/testing:observability_verification_cc_proto",
         "@googleapis//google/monitoring/v3:monitoring_cc_grpc",
diff --git a/ci/otel_collector/CMakeLists.txt b/ci/otel_collector/CMakeLists.txt
index 313ac861ead35..87c9318233ca6 100644
--- a/ci/otel_collector/CMakeLists.txt
+++ b/ci/otel_collector/CMakeLists.txt
@@ -20,10 +20,14 @@ add_library(otel_collector otel_collector.cc otel_collector.h)
 target_link_libraries(
     otel_collector PUBLIC google_cloud_cpp_opentelemetry_testing_protos
                           google-cloud-cpp::monitoring_protos gRPC::grpc++)
-target_include_directories(otel_collector PUBLIC ${PROJECT_SOURCE_DIR})
+add_dependencies(otel_collector google_cloud_cpp_opentelemetry_testing_protos)
+target_include_directories(otel_collector PUBLIC ${PROJECT_SOURCE_DIR}
+                                                 ${PROJECT_BINARY_DIR})
 google_cloud_cpp_add_common_options(otel_collector NO_WARNINGS)
+set_target_properties(otel_collector PROPERTIES CXX_CLANG_TIDY "")
 
 google_cloud_cpp_add_executable(target "otel_collector_main"
                                 "otel_collector_main.cc")
-target_link_libraries(otel_collector_main PRIVATE otel_collector gRPC::grpc++)
-google_cloud_cpp_add_common_options(otel_collector_main NO_WARNINGS)
+target_link_libraries(${target} PRIVATE otel_collector gRPC::grpc++)
+google_cloud_cpp_add_common_options(${target} NO_WARNINGS)
+set_target_properties(${target} PROPERTIES CXX_CLANG_TIDY "")
diff --git a/ci/otel_collector/otel_collector.cc b/ci/otel_collector/otel_collector.cc
index 36251f17e467e..8e0884d13b0d9 100644
--- a/ci/otel_collector/otel_collector.cc
+++ b/ci/otel_collector/otel_collector.cc
@@ -16,6 +16,7 @@
 
 namespace google {
 namespace cloud {
+GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
 namespace testing_util {
 
 grpc::Status OtelCollectorServer::CreateTimeSeries(
@@ -68,5 +69,6 @@ void OtelCollectorServer::Clear() {
 }
 
 }  // namespace testing_util
+GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
 }  // namespace cloud
 }  // namespace google
diff --git a/ci/otel_collector/otel_collector.h b/ci/otel_collector/otel_collector.h
index 145c5b31fdd8a..43de850bd7b18 100644
--- a/ci/otel_collector/otel_collector.h
+++ b/ci/otel_collector/otel_collector.h
@@ -15,6 +15,7 @@
 #ifndef GOOGLE_CLOUD_CPP_CI_OTEL_COLLECTOR_OTEL_COLLECTOR_H
 #define GOOGLE_CLOUD_CPP_CI_OTEL_COLLECTOR_OTEL_COLLECTOR_H
 
+#include "google/cloud/version.h"
 #include "protos/google/cloud/opentelemetry/testing/observability_verification.grpc.pb.h"
 #include <google/monitoring/v3/metric_service.grpc.pb.h>
 #include <grpcpp/grpcpp.h>
@@ -23,6 +24,7 @@
 
 namespace google {
 namespace cloud {
+GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
 namespace testing_util {
 
 /**
@@ -71,6 +73,7 @@ class OtelCollectorServer final
 };
 
 }  // namespace testing_util
+GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
 }  // namespace cloud
 }  // namespace google
 
diff --git a/google/cloud/bigtable/BUILD.bazel b/google/cloud/bigtable/BUILD.bazel
index ba0cf3e9189d5..d0898645e708f 100644
--- a/google/cloud/bigtable/BUILD.bazel
+++ b/google/cloud/bigtable/BUILD.bazel
@@ -43,7 +43,10 @@ cc_library(
     srcs = google_cloud_cpp_bigtable_srcs,
     hdrs = google_cloud_cpp_bigtable_hdrs,
     local_defines = select({
-        ":metrics_enabled": ["GOOGLE_CLOUD_CPP_BIGTABLE_WITH_OTEL_METRICS"],
+        ":metrics_enabled": [
+            "GOOGLE_CLOUD_CPP_BIGTABLE_WITH_OTEL_METRICS",
+            "GOOGLE_CLOUD_CPP_BIGTABLE_WITH_GRPC_OTEL_METRICS",
+        ],
         "//conditions:default": [],
     }),
     visibility = [
@@ -65,6 +68,7 @@ cc_library(
     ] + select({
         ":metrics_enabled": [
             "//:opentelemetry",
+            "@grpc//:grpcpp_otel_plugin",
             "@opentelemetry-cpp//api",
             "@opentelemetry-cpp//sdk/src/metrics",
         ],
@@ -121,6 +125,7 @@ cc_library(
     local_defines = select({
         ":metrics_enabled": [
             "GOOGLE_CLOUD_CPP_BIGTABLE_WITH_OTEL_METRICS",
+            "GOOGLE_CLOUD_CPP_BIGTABLE_WITH_GRPC_OTEL_METRICS",
             "GOOGLE_CLOUD_CPP_BIGTABLE_QUERY_PLAN_REFRESH_ASSERT",
         ],
         "//conditions:default": ["GOOGLE_CLOUD_CPP_BIGTABLE_QUERY_PLAN_REFRESH_ASSERT"],
diff --git a/google/cloud/bigtable/CMakeLists.txt b/google/cloud/bigtable/CMakeLists.txt
index 2be88e34d3e0c..47fc356283896 100644
--- a/google/cloud/bigtable/CMakeLists.txt
+++ b/google/cloud/bigtable/CMakeLists.txt
@@ -190,6 +190,8 @@ add_library(
     internal/endpoint_options.h
     internal/google_bytes_traits.cc
     internal/google_bytes_traits.h
+    internal/grpc_metrics_exporter.cc
+    internal/grpc_metrics_exporter.h
     internal/logging_result_set_reader.cc
     internal/logging_result_set_reader.h
     internal/metrics.cc
@@ -285,7 +287,17 @@ target_compile_definitions(google_cloud_cpp_bigtable
                            PRIVATE GOOGLE_CLOUD_CPP_BIGTABLE_WITH_OTEL_METRICS)
 target_link_libraries(google_cloud_cpp_bigtable
                       PUBLIC google-cloud-cpp::opentelemetry)
-set(EXTRA_MODULES "google_cloud_cpp_opentelemetry" "opentelemetry_metrics")
+if (TARGET gRPC::grpcpp_otel_plugin)
+    target_compile_definitions(
+        google_cloud_cpp_bigtable
+        PRIVATE GOOGLE_CLOUD_CPP_BIGTABLE_WITH_GRPC_OTEL_METRICS)
+    target_link_libraries(google_cloud_cpp_bigtable
+                          PUBLIC gRPC::grpcpp_otel_plugin)
+    set(EXTRA_MODULES "google_cloud_cpp_opentelemetry" "grpcpp_otel_plugin"
+                      "opentelemetry_metrics")
+else ()
+    set(EXTRA_MODULES "google_cloud_cpp_opentelemetry" "opentelemetry_metrics")
+endif ()
 google_cloud_cpp_add_common_options(google_cloud_cpp_bigtable)
 target_include_directories(
     google_cloud_cpp_bigtable
@@ -460,6 +472,7 @@ if (BUILD_TESTING)
         internal/defaults_test.cc
         internal/dynamic_channel_pool_test.cc
         internal/google_bytes_traits_test.cc
+        internal/grpc_metrics_exporter_test.cc
         internal/logging_result_set_reader_test.cc
         internal/metrics_test.cc
         internal/mutate_rows_limiter_test.cc
@@ -528,6 +541,11 @@ if (BUILD_TESTING)
             target_compile_definitions(
                 ${target} PRIVATE GOOGLE_CLOUD_CPP_BIGTABLE_WITH_OTEL_METRICS)
         endif ()
+        if (TARGET gRPC::grpcpp_otel_plugin)
+            target_compile_definitions(
+                ${target}
+                PRIVATE GOOGLE_CLOUD_CPP_BIGTABLE_WITH_GRPC_OTEL_METRICS)
+        endif ()
         google_cloud_cpp_add_common_options(${target})
         add_test(NAME ${target} COMMAND ${target})
     endforeach ()
diff --git a/google/cloud/bigtable/bigtable_client_unit_tests.bzl b/google/cloud/bigtable/bigtable_client_unit_tests.bzl
index 99c7af16cbaba..acb6542878348 100644
--- a/google/cloud/bigtable/bigtable_client_unit_tests.bzl
+++ b/google/cloud/bigtable/bigtable_client_unit_tests.bzl
@@ -55,6 +55,7 @@ bigtable_client_unit_tests = [
     "internal/defaults_test.cc",
     "internal/dynamic_channel_pool_test.cc",
     "internal/google_bytes_traits_test.cc",
+    "internal/grpc_metrics_exporter_test.cc",
     "internal/logging_result_set_reader_test.cc",
     "internal/metrics_test.cc",
     "internal/mutate_rows_limiter_test.cc",
diff --git a/google/cloud/bigtable/data_connection.cc b/google/cloud/bigtable/data_connection.cc
index 6fd19a08706ae..ca7f043935695 100644
--- a/google/cloud/bigtable/data_connection.cc
+++ b/google/cloud/bigtable/data_connection.cc
@@ -18,6 +18,7 @@
 #include "google/cloud/bigtable/internal/data_connection_impl.h"
 #include "google/cloud/bigtable/internal/data_tracing_connection.h"
 #include "google/cloud/bigtable/internal/defaults.h"
+#include "google/cloud/bigtable/internal/grpc_metrics_exporter.h"
 #include "google/cloud/bigtable/internal/mutate_rows_limiter.h"
 #include "google/cloud/bigtable/internal/partial_result_set_source.h"
 #include "google/cloud/bigtable/internal/row_reader_impl.h"
@@ -29,7 +30,12 @@
 #include "google/cloud/grpc_options.h"
 #include "google/cloud/internal/opentelemetry.h"
 #include "google/cloud/internal/unified_grpc_credentials.h"
+#ifdef GOOGLE_CLOUD_CPP_BIGTABLE_WITH_OTEL_METRICS
+#include "google/cloud/monitoring/v3/metric_connection.h"
+#include "google/cloud/internal/random.h"
+#endif  // GOOGLE_CLOUD_CPP_BIGTABLE_WITH_OTEL_METRICS
 #include <memory>
+#include <mutex>
 
 namespace google {
 namespace cloud {
@@ -194,6 +200,39 @@ std::shared_ptr<DataConnection> MakeDataConnection(Options options) {
       background->cq(), options);
   auto limiter =
       bigtable_internal::MakeMutateRowsLimiter(background->cq(), options);
+
+  std::shared_ptr<monitoring_v3::MetricServiceConnection>
+      metric_service_connection;
+  std::shared_ptr<bigtable_internal::GrpcMetricsExporter> grpc_metrics_exporter;
+  std::unique_ptr<bigtable_internal::OperationContextFactory>
+      operation_context_factory;
+
+#ifdef GOOGLE_CLOUD_CPP_BIGTABLE_WITH_OTEL_METRICS
+  if (options.get<EnableMetricsOption>()) {
+    metric_service_connection = monitoring_v3::MakeMetricServiceConnection(
+        internal::MetricsExporterConnectionOptions(options));
+    auto gen = google::cloud::internal::MakeDefaultPRNG();
+    std::string client_uid = google::cloud::internal::Sample(
+        gen, 16, "abcdefghijklmnopqrstuvwxyz0123456789");
+#ifdef GOOGLE_CLOUD_CPP_BIGTABLE_WITH_GRPC_OTEL_METRICS
+    if (options.has<bigtable_internal::InstanceChannelAffinityOption>()) {
+      grpc_metrics_exporter =
+          bigtable_internal::GrpcMetricsExporterRegistry::Singleton()
+              .GetOrCreate(metric_service_connection, options, client_uid);
+    }
+#endif  // GOOGLE_CLOUD_CPP_BIGTABLE_WITH_GRPC_OTEL_METRICS
+    operation_context_factory =
+        std::make_unique<bigtable_internal::MetricsOperationContextFactory>(
+            std::move(client_uid), metric_service_connection, options);
+  } else {
+    operation_context_factory =
+        std::make_unique<bigtable_internal::SimpleOperationContextFactory>();
+  }
+#else
+  operation_context_factory =
+      std::make_unique<bigtable_internal::SimpleOperationContextFactory>();
+#endif  // GOOGLE_CLOUD_CPP_BIGTABLE_WITH_OTEL_METRICS
+
   std::shared_ptr<DataConnection> conn;
 
   if (options.has<bigtable_internal::InstanceChannelAffinityOption>()) {
@@ -212,6 +251,7 @@ std::shared_ptr<DataConnection> MakeDataConnection(Options options) {
         std::move(background),
         std::make_unique<bigtable_internal::StubManager>(
             std::move(affinity_stubs), stub_creation_fn),
+        std::move(operation_context_factory), std::move(grpc_metrics_exporter),
         std::move(limiter), std::move(options));
   } else {
     auto stub = bigtable_internal::CreateBigtableStub(
@@ -219,6 +259,7 @@ std::shared_ptr<DataConnection> MakeDataConnection(Options options) {
     conn = std::make_shared<bigtable_internal::DataConnectionImpl>(
         std::move(background),
         std::make_unique<bigtable_internal::StubManager>(std::move(stub)),
+        std::move(operation_context_factory), std::move(grpc_metrics_exporter),
         std::move(limiter), std::move(options));
   }
   if (google::cloud::internal::TracingEnabled(conn->options())) {
diff --git a/google/cloud/bigtable/google_cloud_cpp_bigtable.bzl b/google/cloud/bigtable/google_cloud_cpp_bigtable.bzl
index 3548270b00fa9..054d50d102fe4 100644
--- a/google/cloud/bigtable/google_cloud_cpp_bigtable.bzl
+++ b/google/cloud/bigtable/google_cloud_cpp_bigtable.bzl
@@ -94,6 +94,7 @@ google_cloud_cpp_bigtable_hdrs = [
     "internal/dynamic_channel_pool.h",
     "internal/endpoint_options.h",
     "internal/google_bytes_traits.h",
+    "internal/grpc_metrics_exporter.h",
     "internal/logging_result_set_reader.h",
     "internal/metrics.h",
     "internal/mutate_rows_limiter.h",
@@ -206,6 +207,7 @@ google_cloud_cpp_bigtable_srcs = [
     "internal/default_row_reader.cc",
     "internal/defaults.cc",
     "internal/google_bytes_traits.cc",
+    "internal/grpc_metrics_exporter.cc",
     "internal/logging_result_set_reader.cc",
     "internal/metrics.cc",
     "internal/mutate_rows_limiter.cc",
diff --git a/google/cloud/bigtable/internal/data_connection_impl.cc b/google/cloud/bigtable/internal/data_connection_impl.cc
index 5a09bf55acaf1..bdbc8ed065422 100644
--- a/google/cloud/bigtable/internal/data_connection_impl.cc
+++ b/google/cloud/bigtable/internal/data_connection_impl.cc
@@ -18,7 +18,7 @@
 #include "google/cloud/bigtable/internal/async_row_sampler.h"
 #include "google/cloud/bigtable/internal/bulk_mutator.h"
 #include "google/cloud/bigtable/internal/default_row_reader.h"
-#include "google/cloud/bigtable/internal/defaults.h"
+#include "google/cloud/bigtable/internal/grpc_metrics_exporter.h"
 #include "google/cloud/bigtable/internal/logging_result_set_reader.h"
 #include "google/cloud/bigtable/internal/operation_context.h"
 #include "google/cloud/bigtable/internal/partial_result_set_reader.h"
@@ -39,12 +39,8 @@
 #include "google/cloud/internal/async_retry_loop.h"
 #include "google/cloud/internal/getenv.h"
 #include "google/cloud/internal/make_status.h"
-#include "google/cloud/internal/random.h"
 #include "google/cloud/internal/retry_loop.h"
 #include "google/cloud/internal/streaming_read_rpc.h"
-#ifdef GOOGLE_CLOUD_CPP_BIGTABLE_WITH_OTEL_METRICS
-#include "google/cloud/monitoring/v3/metric_connection.h"
-#endif  // GOOGLE_CLOUD_CPP_BIGTABLE_WITH_OTEL_METRICS
 #include "google/cloud/universe_domain_options.h"
 #include <memory>
 #include <string>
@@ -201,23 +197,6 @@ std::string_view InstanceNameFromTableName(std::string_view table_name) {
   if (pos == std::string_view::npos) return {};
   return table_name.substr(0, pos);
 }
-
-#ifdef GOOGLE_CLOUD_CPP_BIGTABLE_WITH_OTEL_METRICS
-Options MetricsExporterConnectionOptions(Options options) {
-  // We start with a copy of the client options to preserve credentials and
-  // universe domain, but we must unset Bigtable-specific endpoints/authorities
-  // to allow default Monitoring defaults.
-  options.unset<EndpointOption>();
-  options.unset<AuthorityOption>();
-  auto collector = internal::GetEnv("GOOGLE_CLOUD_CPP_TESTING_OTEL_COLLECTOR");
-  if (collector.has_value()) {
-    // Override credentials when using the otel_collector test server.
-    options.set<UnifiedCredentialsOption>(MakeInsecureCredentials());
-  }
-  return options;
-}
-#endif  // GOOGLE_CLOUD_CPP_BIGTABLE_WITH_OTEL_METRICS
-
 }  // namespace
 
 bigtable::Row TransformReadModifyWriteRowResponse(
@@ -240,65 +219,32 @@ bigtable::Row TransformReadModifyWriteRowResponse(
   return bigtable::Row(std::move(*row.mutable_key()), std::move(cells));
 }
 
-DataConnectionImpl::DataConnectionImpl(
-    std::unique_ptr<BackgroundThreads> background,
-    std::unique_ptr<StubManager> stub_manager,
-    std::shared_ptr<MutateRowsLimiter> limiter, Options options)
-    : background_(std::move(background)),
-      stub_manager_(std::move(stub_manager)),
-      limiter_(std::move(limiter)),
-      options_(MergeOptions(std::move(options), DataConnection::options())) {
-#ifdef GOOGLE_CLOUD_CPP_BIGTABLE_WITH_OTEL_METRICS
-  if (options_.get<bigtable::EnableMetricsOption>()) {
-    metric_service_connection_ = monitoring_v3::MakeMetricServiceConnection(
-        MetricsExporterConnectionOptions(options_));
-    // The client_uid is eventually used in conjunction with other data labels
-    // to identify metric data points. This pseudorandom string is used to aid
-    // in disambiguation.
-    auto gen = internal::MakeDefaultPRNG();
-    std::string client_uid =
-        internal::Sample(gen, 16, "abcdefghijklmnopqrstuvwxyz0123456789");
-    operation_context_factory_ =
-        std::make_unique<MetricsOperationContextFactory>(
-            std::move(client_uid), metric_service_connection_, options_);
-  } else {
-    operation_context_factory_ =
-        std::make_unique<SimpleOperationContextFactory>();
-  }
-#else
-  operation_context_factory_ =
-      std::make_unique<SimpleOperationContextFactory>();
-#endif  // GOOGLE_CLOUD_CPP_BIGTABLE_WITH_OTEL_METRICS
-}
-
-DataConnectionImpl::DataConnectionImpl(
-    std::unique_ptr<BackgroundThreads> background,
-    std::shared_ptr<BigtableStub> stub,
-    std::shared_ptr<MutateRowsLimiter> limiter, Options options)
-    : DataConnectionImpl(std::move(background),
-                         std::make_unique<StubManager>(std::move(stub)),
-                         std::move(limiter), std::move(options)) {}
-
 DataConnectionImpl::DataConnectionImpl(
     std::unique_ptr<BackgroundThreads> background,
     std::unique_ptr<StubManager> stub_manager,
     std::unique_ptr<OperationContextFactory> operation_context_factory,
+    std::shared_ptr<GrpcMetricsExporter> grpc_metrics_exporter,
     std::shared_ptr<MutateRowsLimiter> limiter, Options options)
     : background_(std::move(background)),
       stub_manager_(std::move(stub_manager)),
+      grpc_metrics_exporter_(std::move(grpc_metrics_exporter)),
       operation_context_factory_(std::move(operation_context_factory)),
       limiter_(std::move(limiter)),
       options_(MergeOptions(std::move(options), DataConnection::options())) {}
 
+DataConnectionImpl::~DataConnectionImpl() = default;
+
 DataConnectionImpl::DataConnectionImpl(
     std::unique_ptr<BackgroundThreads> background,
     std::shared_ptr<BigtableStub> stub,
     std::unique_ptr<OperationContextFactory> operation_context_factory,
+    std::shared_ptr<GrpcMetricsExporter> grpc_metrics_exporter,
     std::shared_ptr<MutateRowsLimiter> limiter, Options options)
     : DataConnectionImpl(std::move(background),
                          std::make_unique<StubManager>(std::move(stub)),
                          std::move(operation_context_factory),
-                         std::move(limiter), std::move(options)) {}
+                         std::move(grpc_metrics_exporter), std::move(limiter),
+                         std::move(options)) {}
 
 Status DataConnectionImpl::Apply(std::string const& table_name,
                                  bigtable::SingleRowMutation mut) {
diff --git a/google/cloud/bigtable/internal/data_connection_impl.h b/google/cloud/bigtable/internal/data_connection_impl.h
index a859d5c938ace..791e224ed70ef 100644
--- a/google/cloud/bigtable/internal/data_connection_impl.h
+++ b/google/cloud/bigtable/internal/data_connection_impl.h
@@ -41,6 +41,8 @@ GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
 namespace bigtable_internal {
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
 
+class GrpcMetricsExporter;
+
 // TODO(#16216): Remove this option in favor of addind a member variable to
 // store the instances.
 struct InstanceChannelAffinityOption {
@@ -52,30 +54,20 @@ bigtable::Row TransformReadModifyWriteRowResponse(
 
 class DataConnectionImpl : public bigtable::DataConnection {
  public:
-  ~DataConnectionImpl() override = default;
-
-  DataConnectionImpl(std::unique_ptr<BackgroundThreads> background,
-                     std::unique_ptr<StubManager> stub_manager,
-                     std::shared_ptr<MutateRowsLimiter> limiter,
-                     Options options);
+  ~DataConnectionImpl() override;
 
-  // This constructor is used for testing.
   DataConnectionImpl(
       std::unique_ptr<BackgroundThreads> background,
       std::unique_ptr<StubManager> stub_manager,
       std::unique_ptr<OperationContextFactory> operation_context_factory,
+      std::shared_ptr<GrpcMetricsExporter> grpc_metrics_exporter,
       std::shared_ptr<MutateRowsLimiter> limiter, Options options);
 
-  DataConnectionImpl(std::unique_ptr<BackgroundThreads> background,
-                     std::shared_ptr<BigtableStub> stub,
-                     std::shared_ptr<MutateRowsLimiter> limiter,
-                     Options options);
-
-  // This constructor is used for testing.
   DataConnectionImpl(
       std::unique_ptr<BackgroundThreads> background,
       std::shared_ptr<BigtableStub> stub,
       std::unique_ptr<OperationContextFactory> operation_context_factory,
+      std::shared_ptr<GrpcMetricsExporter> grpc_metrics_exporter,
       std::shared_ptr<MutateRowsLimiter> limiter, Options options);
 
   Options options() override { return options_; }
@@ -149,6 +141,7 @@ class DataConnectionImpl : public bigtable::DataConnection {
   std::unique_ptr<StubManager> stub_manager_;
   std::shared_ptr<::google::cloud::monitoring_v3::MetricServiceConnection>
       metric_service_connection_;
+  std::shared_ptr<GrpcMetricsExporter> grpc_metrics_exporter_;
   std::unique_ptr<OperationContextFactory> operation_context_factory_;
   std::shared_ptr<MutateRowsLimiter> limiter_;
   Options options_;
diff --git a/google/cloud/bigtable/internal/data_connection_impl_test.cc b/google/cloud/bigtable/internal/data_connection_impl_test.cc
index 7874bb8e2e5f0..0903706ea97f1 100644
--- a/google/cloud/bigtable/internal/data_connection_impl_test.cc
+++ b/google/cloud/bigtable/internal/data_connection_impl_test.cc
@@ -16,6 +16,7 @@
 #include "google/cloud/bigtable/data_connection.h"
 #include "google/cloud/bigtable/internal/crc32c.h"
 #include "google/cloud/bigtable/internal/defaults.h"
+#include "google/cloud/bigtable/internal/grpc_metrics_exporter.h"
 #include "google/cloud/bigtable/internal/query_plan.h"
 #ifdef GOOGLE_CLOUD_CPP_BIGTABLE_WITH_OTEL_METRICS
 #include "google/cloud/bigtable/internal/metrics.h"
@@ -267,7 +268,9 @@ std::shared_ptr<DataConnectionImpl> TestConnection(
         std::make_shared<NoopMutateRowsLimiter>()) {
   auto background = internal::MakeBackgroundThreadsFactory()();
   return std::make_shared<DataConnectionImpl>(
-      std::move(background), std::move(stub), std::move(limiter), Options{});
+      std::move(background), std::move(stub),
+      std::make_unique<SimpleOperationContextFactory>(),
+      /*grpc_metrics_exporter=*/nullptr, std::move(limiter), Options{});
 }
 
 std::shared_ptr<DataConnectionImpl> TestConnection(
@@ -278,7 +281,8 @@ std::shared_ptr<DataConnectionImpl> TestConnection(
         std::make_shared<NoopMutateRowsLimiter>()) {
   return std::make_shared<DataConnectionImpl>(
       std::move(background), std::move(stub),
-      std::move(operation_context_factory), std::move(limiter), Options{});
+      std::move(operation_context_factory),
+      /*grpc_metrics_exporter=*/nullptr, std::move(limiter), Options{});
 }
 
 std::shared_ptr<DataConnectionImpl> TestConnection(
@@ -289,7 +293,8 @@ std::shared_ptr<DataConnectionImpl> TestConnection(
   auto background = internal::MakeBackgroundThreadsFactory()();
   return std::make_shared<DataConnectionImpl>(
       std::move(background), std::move(stub),
-      std::move(operation_context_factory), std::move(limiter), Options{});
+      std::move(operation_context_factory),
+      /*grpc_metrics_exporter=*/nullptr, std::move(limiter), Options{});
 }
 
 TEST(TransformReadModifyWriteRowResponse, Basic) {
diff --git a/google/cloud/bigtable/internal/defaults.cc b/google/cloud/bigtable/internal/defaults.cc
index ff71df2383cba..bf8d592ae2b5e 100644
--- a/google/cloud/bigtable/internal/defaults.cc
+++ b/google/cloud/bigtable/internal/defaults.cc
@@ -316,6 +316,20 @@ Options DefaultTableAdminOptions(Options opts) {
       opts.get<::google::cloud::bigtable_internal::AdminEndpointOption>());
 }
 
+#ifdef GOOGLE_CLOUD_CPP_BIGTABLE_WITH_OTEL_METRICS
+Options MetricsExporterConnectionOptions(Options options) {
+  options.unset<EndpointOption>();
+  options.unset<AuthorityOption>();
+  auto collector = google::cloud::internal::GetEnv(
+      "GOOGLE_CLOUD_CPP_TESTING_OTEL_COLLECTOR");
+  if (collector.has_value()) {
+    options.set<google::cloud::UnifiedCredentialsOption>(
+        google::cloud::MakeInsecureCredentials());
+  }
+  return options;
+}
+#endif
+
 }  // namespace internal
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
 }  // namespace bigtable
diff --git a/google/cloud/bigtable/internal/defaults.h b/google/cloud/bigtable/internal/defaults.h
index d535d30a05af8..ed7b389c1fea6 100644
--- a/google/cloud/bigtable/internal/defaults.h
+++ b/google/cloud/bigtable/internal/defaults.h
@@ -52,6 +52,10 @@ Options DefaultInstanceAdminOptions(Options opts);
 
 Options DefaultTableAdminOptions(Options opts);
 
+#ifdef GOOGLE_CLOUD_CPP_BIGTABLE_WITH_OTEL_METRICS
+Options MetricsExporterConnectionOptions(Options options);
+#endif
+
 }  // namespace internal
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
 }  // namespace bigtable
diff --git a/google/cloud/bigtable/internal/grpc_metrics_exporter.cc b/google/cloud/bigtable/internal/grpc_metrics_exporter.cc
new file mode 100644
index 0000000000000..e0e86ee484348
--- /dev/null
+++ b/google/cloud/bigtable/internal/grpc_metrics_exporter.cc
@@ -0,0 +1,377 @@
+// Copyright 2026 Google LLC
+//
+// 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
+//
+//     https://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 "google/cloud/bigtable/internal/grpc_metrics_exporter.h"
+#include "google/cloud/common_options.h"
+#ifdef GOOGLE_CLOUD_CPP_BIGTABLE_WITH_GRPC_OTEL_METRICS
+#include "google/cloud/bigtable/internal/data_connection_impl.h"
+#include "google/cloud/bigtable/options.h"
+#include "google/cloud/bigtable/version.h"
+#include "google/cloud/opentelemetry/internal/monitoring_exporter.h"
+#include "google/cloud/opentelemetry/resource_detector.h"
+#include "google/cloud/grpc_options.h"
+#include "google/cloud/internal/algorithm.h"
+#include "google/cloud/log.h"
+#include "absl/strings/match.h"
+#include "absl/strings/str_cat.h"
+#include "absl/strings/string_view.h"
+#include <grpcpp/ext/otel_plugin.h>
+#include <grpcpp/grpcpp.h>
+#include <opentelemetry/semconv/incubating/cloud_attributes.h>
+#include <opentelemetry/semconv/incubating/faas_attributes.h>
+#include <opentelemetry/semconv/incubating/host_attributes.h>
+#include <opentelemetry/version.h>
+#if OPENTELEMETRY_VERSION_MAJOR > 1 || \
+    (OPENTELEMETRY_VERSION_MAJOR == 1 && OPENTELEMETRY_VERSION_MINOR >= 10)
+#include <opentelemetry/sdk/metrics/export/periodic_exporting_metric_reader_factory.h>
+#include <opentelemetry/sdk/metrics/meter_provider_factory.h>
+#include <opentelemetry/sdk/metrics/view/instrument_selector_factory.h>
+#include <opentelemetry/sdk/metrics/view/meter_selector_factory.h>
+#include <opentelemetry/sdk/metrics/view/view_factory.h>
+#endif  // OPENTELEMETRY_VERSION_MAJOR > 1 || (OPENTELEMETRY_VERSION_MAJOR == 1
+        // && OPENTELEMETRY_VERSION_MINOR >= 10)
+#include <opentelemetry/sdk/metrics/export/periodic_exporting_metric_reader.h>
+#include <opentelemetry/sdk/metrics/meter_provider.h>
+#include <opentelemetry/sdk/metrics/view/instrument_selector.h>
+#include <opentelemetry/sdk/metrics/view/meter_selector.h>
+#include <opentelemetry/sdk/metrics/view/view.h>
+#include <opentelemetry/sdk/resource/resource.h>
+#include <chrono>
+#include <cstdint>
+#include <memory>
+#include <set>
+#include <string>
+#include <utility>
+#include <vector>
+#endif  // GOOGLE_CLOUD_CPP_BIGTABLE_WITH_GRPC_OTEL_METRICS
+
+namespace google {
+namespace cloud {
+namespace bigtable_internal {
+GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
+
+GrpcMetricsExporterRegistry& GrpcMetricsExporterRegistry::Singleton() {
+  static auto* registry = new GrpcMetricsExporterRegistry;
+  return *registry;
+}
+
+std::shared_ptr<GrpcMetricsExporter> GrpcMetricsExporterRegistry::GetOrCreate(
+    std::shared_ptr<monitoring_v3::MetricServiceConnection> const& conn,
+    Options const& options, std::string const& client_uid) {
+  std::unique_lock<std::mutex> lk(mu_);
+  auto authority = options.get<AuthorityOption>();
+  auto it = exporters_.find(authority);
+  if (it != exporters_.end()) {
+    if (auto exporter = it->second.lock()) {
+      return exporter;
+    }
+  }
+  auto exporter =
+      std::make_shared<GrpcMetricsExporter>(conn, options, client_uid);
+  exporters_[std::move(authority)] = exporter;
+  return exporter;
+}
+
+void GrpcMetricsExporterRegistry::Unregister(std::string const& authority) {
+  std::unique_lock<std::mutex> lk(mu_);
+  auto it = exporters_.find(authority);
+  if (it != exporters_.end() && it->second.expired()) {
+    exporters_.erase(it);
+  }
+}
+
+void GrpcMetricsExporterRegistry::Clear() {
+  std::unique_lock<std::mutex> lk(mu_);
+  exporters_.clear();
+}
+
+#ifdef GOOGLE_CLOUD_CPP_BIGTABLE_WITH_GRPC_OTEL_METRICS
+
+std::vector<double> MakeLatencyHistogramBoundaries() {
+  static auto const kBoundaries = [] {
+    using dseconds = std::chrono::duration<double, std::ratio<1>>;
+    std::vector<double> boundaries;
+    auto boundary = std::chrono::milliseconds(0);
+    auto increment = std::chrono::milliseconds(2);
+    for (int i = 0; i != 50; ++i) {
+      boundaries.push_back(
+          std::chrono::duration_cast<dseconds>(boundary).count());
+      boundary += increment;
+    }
+    increment = std::chrono::milliseconds(10);
+    for (int i = 0; i != 150 && boundary <= std::chrono::minutes(5); ++i) {
+      boundaries.push_back(
+          std::chrono::duration_cast<dseconds>(boundary).count());
+      if (i != 0 && i % 10 == 0) increment *= 2;
+      boundary += increment;
+    }
+    return boundaries;
+  }();
+  return kBoundaries;
+}
+
+namespace {
+
+void AddHistogramView(opentelemetry::sdk::metrics::MeterProvider& provider,
+                      std::vector<double> boundaries, std::string const& name,
+                      std::string const& unit) {
+  auto constexpr kGrpcMeterName = "grpc-c++";
+  auto constexpr kGrpcSchema = "";
+
+  auto histogram_aggregation_config = std::make_unique<
+      opentelemetry::sdk::metrics::HistogramAggregationConfig>();
+  histogram_aggregation_config->boundaries_ = std::move(boundaries);
+  auto aggregation_config =
+      std::shared_ptr<opentelemetry::sdk::metrics::AggregationConfig>(
+          std::move(histogram_aggregation_config));
+
+  auto description = absl::StrCat("A view of ", name,
+                                  " with histogram boundaries more appropriate "
+                                  "for Google Cloud Bigtable RPCs");
+
+#if OPENTELEMETRY_VERSION_MAJOR > 1 || \
+    (OPENTELEMETRY_VERSION_MAJOR == 1 && OPENTELEMETRY_VERSION_MINOR >= 23)
+  (void)unit;
+  provider.AddView(
+      opentelemetry::sdk::metrics::InstrumentSelectorFactory::Create(
+          opentelemetry::sdk::metrics::InstrumentType::kHistogram, name, unit),
+      opentelemetry::sdk::metrics::MeterSelectorFactory::Create(
+          kGrpcMeterName, grpc::Version(), kGrpcSchema),
+      opentelemetry::sdk::metrics::ViewFactory::Create(
+          name, std::move(description),
+          opentelemetry::sdk::metrics::AggregationType::kHistogram,
+          std::move(aggregation_config)));
+#elif OPENTELEMETRY_VERSION_MAJOR > 1 || \
+    (OPENTELEMETRY_VERSION_MAJOR == 1 && OPENTELEMETRY_VERSION_MINOR >= 10)
+  provider.AddView(
+      opentelemetry::sdk::metrics::InstrumentSelectorFactory::Create(
+          opentelemetry::sdk::metrics::InstrumentType::kHistogram, name, unit),
+      opentelemetry::sdk::metrics::MeterSelectorFactory::Create(
+          kGrpcMeterName, grpc::Version(), kGrpcSchema),
+      opentelemetry::sdk::metrics::ViewFactory::Create(
+          name, std::move(description), unit,
+          opentelemetry::sdk::metrics::AggregationType::kHistogram,
+          std::move(aggregation_config)));
+#else   // OPENTELEMETRY_VERSION_MAJOR > 1 || (OPENTELEMETRY_VERSION_MAJOR == 1
+        // && OPENTELEMETRY_VERSION_MINOR >= 10)
+  (void)unit;
+  provider.AddView(
+      std::make_unique<opentelemetry::sdk::metrics::InstrumentSelector>(
+          opentelemetry::sdk::metrics::InstrumentType::kHistogram, name),
+      std::make_unique<opentelemetry::sdk::metrics::MeterSelector>(
+          kGrpcMeterName, grpc::Version(), kGrpcSchema),
+      std::make_unique<opentelemetry::sdk::metrics::View>(
+          name, std::move(description),
+          opentelemetry::sdk::metrics::AggregationType::kHistogram,
+          std::move(aggregation_config)));
+#endif  // OPENTELEMETRY_VERSION_MAJOR > 1 || (OPENTELEMETRY_VERSION_MAJOR == 1
+        // && OPENTELEMETRY_VERSION_MINOR >= 23)
+}
+
+}  // namespace
+
+std::shared_ptr<opentelemetry::metrics::MeterProvider> MakeGrpcMeterProvider(
+    std::unique_ptr<opentelemetry::sdk::metrics::PushMetricExporter> exporter,
+    opentelemetry::sdk::metrics::PeriodicExportingMetricReaderOptions
+        reader_options) {
+#if OPENTELEMETRY_VERSION_MAJOR > 1 || \
+    (OPENTELEMETRY_VERSION_MAJOR == 1 && OPENTELEMETRY_VERSION_MINOR >= 10)
+  auto provider = opentelemetry::sdk::metrics::MeterProviderFactory::Create(
+      std::make_unique<opentelemetry::sdk::metrics::ViewRegistry>(),
+      opentelemetry::sdk::resource::Resource::Create({}));
+#else   // OPENTELEMETRY_VERSION_MAJOR > 1 || (OPENTELEMETRY_VERSION_MAJOR == 1
+        // && OPENTELEMETRY_VERSION_MINOR >= 10)
+  auto provider = std::make_unique<opentelemetry::sdk::metrics::MeterProvider>(
+      std::make_unique<opentelemetry::sdk::metrics::ViewRegistry>(),
+      opentelemetry::sdk::resource::Resource::Create({}));
+#endif  // OPENTELEMETRY_VERSION_MAJOR > 1 || (OPENTELEMETRY_VERSION_MAJOR == 1
+        // && OPENTELEMETRY_VERSION_MINOR >= 10)
+  auto* p =
+      static_cast<opentelemetry::sdk::metrics::MeterProvider*>(provider.get());
+  AddHistogramView(*p, MakeLatencyHistogramBoundaries(),
+                   "grpc.client.attempt.duration", "s");
+
+#if OPENTELEMETRY_VERSION_MAJOR > 1 || OPENTELEMETRY_VERSION_MINOR >= 10
+  p->AddMetricReader(
+      opentelemetry::sdk::metrics::PeriodicExportingMetricReaderFactory::Create(
+          std::move(exporter), std::move(reader_options)));
+#else   // OPENTELEMETRY_VERSION_MAJOR > 1 || OPENTELEMETRY_VERSION_MINOR >= 10
+  p->AddMetricReader(
+      std::make_unique<
+          opentelemetry::sdk::metrics::PeriodicExportingMetricReader>(
+          std::move(exporter), std::move(reader_options)));
+#endif  // OPENTELEMETRY_VERSION_MAJOR > 1 || OPENTELEMETRY_VERSION_MINOR >= 10
+
+  return std::shared_ptr<opentelemetry::metrics::MeterProvider>(
+      std::move(provider));
+}
+
+MonitoredResourceResult MakeMonitoredResource(
+    opentelemetry::sdk::metrics::PointDataAttributes const& pda,
+    opentelemetry::sdk::resource::Resource const& detected_resource,
+    Options const& options, std::string const& client_uid) {
+  namespace sc = ::opentelemetry::semconv;
+  google::api::MonitoredResource resource;
+  resource.set_type("bigtable.googleapis.com/Client");
+  auto& labels = *resource.mutable_labels();
+  auto const& attributes = pda.attributes.GetAttributes();
+  auto get_attr = [&](std::string const& key) {
+    auto it = attributes.find(key);
+    if (it == attributes.end()) return std::string{};
+    return opentelemetry::nostd::get<std::string>(it->second);
+  };
+  auto const& detected_attributes = detected_resource.GetAttributes();
+  auto by_name = [&](std::string const& name, std::string default_value = {}) {
+    auto const l = detected_attributes.find(name);
+    if (l == detected_attributes.end()) return default_value;
+    return opentelemetry::nostd::get<std::string>(l->second);
+  };
+
+  auto project_id = get_attr("project_id");
+  // Fall back to resolving the project ID from `InstanceChannelAffinityOption`
+  // if the static `ProjectIdOption` is not set on the client (which is common
+  // in client configurations with instance routing / channel affinity).
+  if (project_id.empty() &&
+      options.has<bigtable_internal::InstanceChannelAffinityOption>()) {
+    auto const& instances =
+        options.get<bigtable_internal::InstanceChannelAffinityOption>();
+    if (!instances.empty()) {
+      project_id = instances[0].project_id();
+    }
+  }
+  if (project_id.empty()) {
+    project_id = by_name(sc::cloud::kCloudAccountId);
+  }
+
+  labels["project_id"] = project_id;
+  labels["instance"] = get_attr("instance");
+  labels["app_profile"] = options.get<bigtable::AppProfileIdOption>();
+  labels["client_name"] = "cpp.Bigtable/" + bigtable::version_string();
+  labels["uuid"] = client_uid;
+
+  auto client_project = by_name(sc::cloud::kCloudAccountId);
+  if (client_project.empty()) {
+    client_project = project_id;
+  }
+  if (!client_project.empty()) {
+    labels["client_project"] = client_project;
+  }
+  labels["location"] = by_name(sc::cloud::kCloudAvailabilityZone,
+                               by_name(sc::cloud::kCloudRegion, "global"));
+  labels["cloud_platform"] = by_name(sc::cloud::kCloudPlatform, "unknown");
+  labels["host_id"] = by_name("faas.id", by_name(sc::host::kHostId, "unknown"));
+  auto hostname = by_name(sc::host::kHostName);
+  if (!hostname.empty()) {
+    labels["hostname"] = hostname;
+  }
+
+  return MonitoredResourceResult{std::move(project_id), std::move(resource)};
+}
+
+GrpcMetricsExporter::GrpcMetricsExporter(
+    std::shared_ptr<monitoring_v3::MetricServiceConnection> const& conn,
+    Options const& options, std::string const& client_uid)
+    : authority_(options.get<AuthorityOption>()) {
+  auto detector = otel::MakeResourceDetector();
+  auto detected_resource = detector->Detect();
+
+  auto dynamic_resource_fn =
+      [options, client_uid, detected_resource = std::move(detected_resource)](
+          opentelemetry::sdk::metrics::PointDataAttributes const& pda) {
+        auto res =
+            MakeMonitoredResource(pda, detected_resource, options, client_uid);
+        return std::make_pair(std::move(res.project_id),
+                              std::move(res.resource));
+      };
+
+  std::set<std::string> excluded_labels{"project_id", "instance"};
+  auto resource_filter_fn =
+      [excluded_labels = std::move(excluded_labels)](std::string const& key) {
+        return internal::Contains(excluded_labels, key);
+      };
+
+  auto exporter = otel_internal::MakeMonitoringExporter(
+      dynamic_resource_fn, resource_filter_fn, conn, options);
+
+  auto reader_options =
+      opentelemetry::sdk::metrics::PeriodicExportingMetricReaderOptions{};
+  // otel::PeriodicExportingMetricReader enforces that export_timeout_millis <
+  // export_interval_millis.
+  reader_options.export_interval_millis =
+      options.get<bigtable::MetricsPeriodOption>();
+  reader_options.export_timeout_millis =
+      (std::min)(std::chrono::milliseconds(std::chrono::seconds(30)),
+                 reader_options.export_interval_millis / 2);
+
+  provider_ =
+      MakeGrpcMeterProvider(std::move(exporter), std::move(reader_options));
+
+  auto const metrics = std::vector<absl::string_view>{
+      absl::string_view{"grpc.client.attempt.duration"},
+      absl::string_view{"grpc.lb.rls.default_target_picks"},
+      absl::string_view{"grpc.lb.rls.target_picks"},
+      absl::string_view{"grpc.lb.rls.failed_picks"},
+      absl::string_view{"grpc.xds_client.server_failure"},
+      absl::string_view{"grpc.xds_client.resource_updates_invalid"},
+      absl::string_view{"grpc.subchannel.disconnections"},
+      absl::string_view{"grpc.subchannel.connection_attempts_succeeded"},
+      absl::string_view{"grpc.subchannel.connection_attempts_failed"},
+      absl::string_view{"grpc.subchannel.open_connections"},
+  };
+  auto scope_filter =
+      [authority = authority_](
+          grpc::OpenTelemetryPluginBuilder::ChannelScope const& scope) {
+        GCP_LOG(INFO) << "GrpcMetricsExporter: scope filter checking target="
+                      << scope.target()
+                      << " default_authority=" << scope.default_authority()
+                      << " vs expected authority=" << authority;
+        return scope.default_authority() == authority;
+      };
+  auto status =
+      grpc::OpenTelemetryPluginBuilder()
+          .SetMeterProvider(provider_)
+          .EnableMetrics(metrics)
+          .SetGenericMethodAttributeFilter([](absl::string_view target) {
+            return absl::StartsWith(target, "google.bigtable.v2");
+          })
+          .SetChannelScopeFilter(std::move(scope_filter))
+          .BuildAndRegisterGlobal();
+  if (!status.ok()) {
+    GCP_LOG(ERROR) << "Cannot register provider status=" << status.ToString();
+  }
+}
+
+GrpcMetricsExporter::~GrpcMetricsExporter() {
+  if (provider_) {
+    auto* p = static_cast<opentelemetry::sdk::metrics::MeterProvider*>(
+        provider_.get());
+    p->Shutdown();
+  }
+  GrpcMetricsExporterRegistry::Singleton().Unregister(authority_);
+}
+
+#else  // GOOGLE_CLOUD_CPP_BIGTABLE_WITH_GRPC_OTEL_METRICS
+
+GrpcMetricsExporter::GrpcMetricsExporter(
+    std::shared_ptr<monitoring_v3::MetricServiceConnection> const&,
+    Options const&, std::string const&) {}
+
+GrpcMetricsExporter::~GrpcMetricsExporter() = default;
+
+#endif  // GOOGLE_CLOUD_CPP_BIGTABLE_WITH_GRPC_OTEL_METRICS
+
+GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
+}  // namespace bigtable_internal
+}  // namespace cloud
+}  // namespace google
diff --git a/google/cloud/bigtable/internal/grpc_metrics_exporter.h b/google/cloud/bigtable/internal/grpc_metrics_exporter.h
new file mode 100644
index 0000000000000..13173207bc855
--- /dev/null
+++ b/google/cloud/bigtable/internal/grpc_metrics_exporter.h
@@ -0,0 +1,140 @@
+// Copyright 2026 Google LLC
+//
+// 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
+//
+//     https://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.
+
+#ifndef GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_BIGTABLE_INTERNAL_GRPC_METRICS_EXPORTER_H
+#define GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_BIGTABLE_INTERNAL_GRPC_METRICS_EXPORTER_H
+
+#include "google/cloud/options.h"
+#include "google/cloud/version.h"
+#include <map>
+#include <memory>
+#include <mutex>
+#include <string>
+
+#ifdef GOOGLE_CLOUD_CPP_BIGTABLE_WITH_GRPC_OTEL_METRICS
+#include "google/cloud/monitoring/v3/metric_connection.h"
+#include "google/api/monitored_resource.pb.h"
+#include <opentelemetry/metrics/meter_provider.h>
+#include <opentelemetry/sdk/metrics/data/metric_data.h>
+#include <opentelemetry/sdk/metrics/export/periodic_exporting_metric_reader.h>
+#include <opentelemetry/sdk/metrics/push_metric_exporter.h>
+#include <opentelemetry/sdk/resource/resource.h>
+#include <vector>
+#endif  // GOOGLE_CLOUD_CPP_BIGTABLE_WITH_GRPC_OTEL_METRICS
+
+namespace google {
+namespace cloud {
+namespace monitoring_v3 {
+GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
+class MetricServiceConnection;
+GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
+}  // namespace monitoring_v3
+
+namespace bigtable_internal {
+GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
+
+class GrpcMetricsExporter;
+
+/**
+ * Thread-safe registry managing the lifetime and deduplication of
+ * `GrpcMetricsExporter` instances.
+ *
+ * gRPC OpenTelemetry plugin registration (`BuildAndRegisterGlobal`) is
+ * process-global per target channel authority. If an exporter's lifetime were
+ * bound to an individual `DataConnectionImpl` without sharing, closing a single
+ * connection would destroy the `MeterProvider` and call `Shutdown()`,
+ * prematurely halting metric export globally for other active connections in
+ * the same process.
+ *
+ * Conversely, keeping static `std::shared_ptr` singletons permanently violates
+ * Google C++ Style Guide rules against static non-trivially destructible
+ * objects and prevents `MeterProvider::Shutdown()` from ever flushing metrics
+ * upon connection shutdown.
+ *
+ * `GrpcMetricsExporterRegistry` resolves both requirements:
+ * 1. **Deduplication / Sharing**: Tracks active `GrpcMetricsExporter` instances
+ *    via `std::weak_ptr` keyed by channel authority. Multiple connection
+ *    instances targeting the same authority share ownership of the same
+ *    exporter `std::shared_ptr`.
+ * 2. **Clean Shutdown**: When all `DataConnectionImpl` instances sharing an
+ *    exporter are closed, the last `std::shared_ptr` is destroyed, invoking
+ *    `~GrpcMetricsExporter()` which cleanly calls `MeterProvider::Shutdown()`
+ *    and unregisters the authority from the registry.
+ */
+class GrpcMetricsExporterRegistry {
+ public:
+  static GrpcMetricsExporterRegistry& Singleton();
+
+  std::shared_ptr<GrpcMetricsExporter> GetOrCreate(
+      std::shared_ptr<monitoring_v3::MetricServiceConnection> const& conn,
+      Options const& options, std::string const& client_uid);
+
+  void Unregister(std::string const& authority);
+
+  void Clear();
+
+ private:
+  GrpcMetricsExporterRegistry() = default;
+
+  std::map<std::string, std::weak_ptr<GrpcMetricsExporter>> exporters_;
+  std::mutex mu_;
+};
+
+#ifdef GOOGLE_CLOUD_CPP_BIGTABLE_WITH_GRPC_OTEL_METRICS
+struct MonitoredResourceResult {
+  std::string project_id;
+  google::api::MonitoredResource resource;
+};
+
+MonitoredResourceResult MakeMonitoredResource(
+    opentelemetry::sdk::metrics::PointDataAttributes const& pda,
+    opentelemetry::sdk::resource::Resource const& detected_resource,
+    Options const& options, std::string const& client_uid);
+
+inline MonitoredResourceResult MakeMonitoredResource(
+    opentelemetry::sdk::metrics::PointDataAttributes const& pda,
+    Options const& options, std::string const& client_uid) {
+  return MakeMonitoredResource(
+      pda, opentelemetry::sdk::resource::Resource::Create({}), options,
+      client_uid);
+}
+
+std::vector<double> MakeLatencyHistogramBoundaries();
+
+std::shared_ptr<opentelemetry::metrics::MeterProvider> MakeGrpcMeterProvider(
+    std::unique_ptr<opentelemetry::sdk::metrics::PushMetricExporter> exporter,
+    opentelemetry::sdk::metrics::PeriodicExportingMetricReaderOptions
+        reader_options);
+#endif  // GOOGLE_CLOUD_CPP_BIGTABLE_WITH_GRPC_OTEL_METRICS
+
+class GrpcMetricsExporter {
+ public:
+  GrpcMetricsExporter(
+      std::shared_ptr<monitoring_v3::MetricServiceConnection> const& conn,
+      Options const& options, std::string const& client_uid);
+  ~GrpcMetricsExporter();
+
+ private:
+#ifdef GOOGLE_CLOUD_CPP_BIGTABLE_WITH_GRPC_OTEL_METRICS
+  std::string authority_;
+  std::shared_ptr<opentelemetry::metrics::MeterProvider> provider_;
+#endif  // GOOGLE_CLOUD_CPP_BIGTABLE_WITH_GRPC_OTEL_METRICS
+};
+
+GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
+}  // namespace bigtable_internal
+}  // namespace cloud
+}  // namespace google
+
+#endif  // GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_BIGTABLE_INTERNAL_GRPC_METRICS_EXPORTER_H
diff --git a/google/cloud/bigtable/internal/grpc_metrics_exporter_test.cc b/google/cloud/bigtable/internal/grpc_metrics_exporter_test.cc
new file mode 100644
index 0000000000000..ce1ba5750a863
--- /dev/null
+++ b/google/cloud/bigtable/internal/grpc_metrics_exporter_test.cc
@@ -0,0 +1,420 @@
+// Copyright 2026 Google LLC
+//
+// 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
+//
+//     https://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.
+
+#ifdef GOOGLE_CLOUD_CPP_BIGTABLE_WITH_GRPC_OTEL_METRICS
+
+#include "google/cloud/bigtable/internal/grpc_metrics_exporter.h"
+#include "google/cloud/bigtable/options.h"
+#include "google/cloud/bigtable/version.h"
+#include "google/cloud/grpc_options.h"
+#include "google/cloud/options.h"
+#include <gmock/gmock.h>
+#include <grpcpp/grpcpp.h>
+#include <opentelemetry/metrics/meter.h>
+#include <opentelemetry/metrics/meter_provider.h>
+#include <opentelemetry/metrics/sync_instruments.h>
+#include <opentelemetry/sdk/metrics/data/metric_data.h>
+#include <opentelemetry/sdk/metrics/data/point_data.h>
+#include <opentelemetry/sdk/metrics/export/metric_producer.h>
+#include <opentelemetry/sdk/metrics/export/periodic_exporting_metric_reader.h>
+#include <opentelemetry/sdk/metrics/push_metric_exporter.h>
+#include <opentelemetry/sdk/resource/resource.h>
+#include <opentelemetry/semconv/incubating/cloud_attributes.h>
+#include <opentelemetry/semconv/incubating/host_attributes.h>
+#include <algorithm>
+#include <atomic>
+#include <chrono>
+#include <memory>
+#include <numeric>
+#include <string>
+#include <thread>
+#include <vector>
+
+namespace google {
+namespace cloud {
+namespace bigtable_internal {
+GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
+namespace {
+
+using ::testing::AllOf;
+using ::testing::AtLeast;
+using ::testing::Contains;
+using ::testing::ElementsAre;
+using ::testing::ElementsAreArray;
+using ::testing::Eq;
+using ::testing::Ge;
+using ::testing::IsEmpty;
+using ::testing::Le;
+using ::testing::Not;
+using ::testing::ResultOf;
+using ::testing::Return;
+using ::testing::SizeIs;
+using ::testing::VariantWith;
+
+class MockPushMetricExporter
+    : public opentelemetry::sdk::metrics::PushMetricExporter {
+ public:
+  // NOLINTBEGIN(bugprone-exception-escape)
+  MOCK_METHOD(opentelemetry::sdk::common::ExportResult, Export,
+              (opentelemetry::sdk::metrics::ResourceMetrics const&),
+              (noexcept, override));
+
+  MOCK_METHOD(opentelemetry::sdk::metrics::AggregationTemporality,
+              GetAggregationTemporality,
+              (opentelemetry::sdk::metrics::InstrumentType),
+              (const, noexcept, override));
+
+  MOCK_METHOD(bool, ForceFlush, (std::chrono::microseconds),
+              (noexcept, override));
+  MOCK_METHOD(bool, Shutdown, (std::chrono::microseconds),
+              (noexcept, override));
+  // NOLINTEND(bugprone-exception-escape)
+};
+
+auto constexpr kDurationMetric = "grpc.client.attempt.duration";
+
+auto MatchesLatencyBoundaries() {
+  return ResultOf(
+      "boundaries are latency boundaries",
+      [](opentelemetry::sdk::metrics::HistogramPointData const& hpd) {
+        return hpd.boundaries_;
+      },
+      ElementsAreArray(MakeLatencyHistogramBoundaries()));
+}
+
+auto ExpectedLatencyHistogram() {
+  return ResultOf(
+      "data is histogram with right boundaries",
+      [](opentelemetry::sdk::metrics::PointDataAttributes const& pda) {
+        return pda.point_data;
+      },
+      VariantWith<opentelemetry::sdk::metrics::HistogramPointData>(
+          MatchesLatencyBoundaries()));
+}
+
+template <typename Matcher>
+auto WithPointData(Matcher&& matcher) {
+  return ResultOf(
+      "instrument descriptor name",
+      [](opentelemetry::sdk::metrics::MetricData const& md) {
+        return md.point_data_attr_;
+      },
+      std::forward<Matcher>(matcher));
+}
+
+template <typename Matcher>
+auto WithMetricsData(Matcher&& matcher) {
+  return ResultOf(
+      "metric_data_",
+      [](opentelemetry::sdk::metrics::ScopeMetrics const& sm) {
+        return sm.metric_data_;
+      },
+      std::forward<Matcher>(matcher));
+}
+
+auto MatchesInstrumentName(std::string name) {
+  return ResultOf(
+      "instrument descriptor name",
+      [](opentelemetry::sdk::metrics::MetricData const& md) {
+        return md.instrument_descriptor.name_;
+      },
+      std::move(name));
+}
+
+auto MetricDataEmpty() {
+  return ResultOf(
+      "scope_metric_data_",
+      [](opentelemetry::sdk::metrics::ResourceMetrics const& data) {
+        return data.scope_metric_data_;
+      },
+      IsEmpty());
+}
+
+auto MetricDataNotEmpty() {
+  return ResultOf(
+      "scope_metric_data_",
+      [](opentelemetry::sdk::metrics::ResourceMetrics const& data) {
+        return data.scope_metric_data_;
+      },
+      Not(IsEmpty()));
+}
+
+auto constexpr kExportInterval = std::chrono::milliseconds(50);
+auto constexpr kExportTimeout = std::chrono::milliseconds(25);
+
+auto TestReaderOptions() {
+  opentelemetry::sdk::metrics::PeriodicExportingMetricReaderOptions
+      reader_options;
+  reader_options.export_interval_millis = kExportInterval;
+  reader_options.export_timeout_millis = kExportTimeout;
+  return reader_options;
+}
+
+class DummyMetricServiceConnection
+    : public monitoring_v3::MetricServiceConnection {
+ public:
+  ~DummyMetricServiceConnection() override = default;
+};
+
+TEST(GrpcMetricsExporterRegistryTest, GetOrCreateAndLifecycle) {
+  auto& registry = GrpcMetricsExporterRegistry::Singleton();
+  registry.Clear();
+
+  auto conn = std::make_shared<DummyMetricServiceConnection>();
+
+  Options options1;
+  options1.set<AuthorityOption>("test-authority-1");
+  Options options2;
+  options2.set<AuthorityOption>("test-authority-2");
+
+  auto exporter1_a = registry.GetOrCreate(conn, options1, "client-1");
+  ASSERT_NE(exporter1_a, nullptr);
+
+  // Second request for same authority returns the existing shared instance
+  auto exporter1_b = registry.GetOrCreate(conn, options1, "client-2");
+  EXPECT_EQ(exporter1_a, exporter1_b);
+
+  // Different authority gets a different instance
+  auto exporter2 = registry.GetOrCreate(conn, options2, "client-3");
+  ASSERT_NE(exporter2, nullptr);
+  EXPECT_NE(exporter1_a, exporter2);
+
+  // When all references to authority 1 are dropped, destructor runs and
+  // authority is unregistered
+  exporter1_a.reset();
+  exporter1_b.reset();
+
+  // A new request for authority 1 creates a new instance
+  auto exporter1_c = registry.GetOrCreate(conn, options1, "client-4");
+  ASSERT_NE(exporter1_c, nullptr);
+  EXPECT_NE(exporter1_c, exporter2);
+}
+
+TEST(GrpcMetricsExporterTest, MakeMonitoredResource) {
+  Options options;
+  options.set<bigtable::AppProfileIdOption>("test-app-profile");
+  std::string const client_uid = "test-client-uid";
+
+  opentelemetry::sdk::metrics::PointDataAttributes pda;
+  pda.attributes = opentelemetry::sdk::metrics::PointAttributes({
+      {"project_id", "test-project"},
+      {"instance", "test-instance"},
+  });
+
+  auto result = MakeMonitoredResource(pda, options, client_uid);
+  EXPECT_THAT(result.project_id, Eq("test-project"));
+
+  auto const& resource = result.resource;
+  EXPECT_THAT(resource.type(), Eq("bigtable.googleapis.com/Client"));
+
+  auto const& labels = resource.labels();
+  EXPECT_THAT(labels.at("project_id"), Eq("test-project"));
+  EXPECT_THAT(labels.at("client_project"), Eq("test-project"));
+  EXPECT_THAT(labels.at("instance"), Eq("test-instance"));
+  EXPECT_THAT(labels.at("app_profile"), Eq("test-app-profile"));
+  EXPECT_THAT(labels.at("client_name"),
+              Eq("cpp.Bigtable/" + bigtable::version_string()));
+  EXPECT_THAT(labels.at("uuid"), Eq("test-client-uid"));
+  EXPECT_THAT(labels.at("location"), Eq("global"));
+  EXPECT_THAT(labels.at("cloud_platform"), Eq("unknown"));
+  EXPECT_THAT(labels.at("host_id"), Eq("unknown"));
+}
+
+TEST(GrpcMetricsExporterTest, MakeMonitoredResourceMissingAttributes) {
+  Options options;
+  std::string const client_uid = "test-client-uid";
+
+  opentelemetry::sdk::metrics::PointDataAttributes pda;
+  pda.attributes = opentelemetry::sdk::metrics::PointAttributes({});
+
+  auto result = MakeMonitoredResource(pda, options, client_uid);
+  EXPECT_THAT(result.project_id, Eq(""));
+
+  auto const& resource = result.resource;
+  EXPECT_THAT(resource.type(), Eq("bigtable.googleapis.com/Client"));
+
+  auto const& labels = resource.labels();
+  EXPECT_THAT(labels.at("project_id"), Eq(""));
+  EXPECT_THAT(labels.at("instance"), Eq(""));
+  EXPECT_THAT(labels.at("app_profile"), Eq(""));
+  EXPECT_THAT(labels.at("client_name"),
+              Eq("cpp.Bigtable/" + bigtable::version_string()));
+  EXPECT_THAT(labels.at("uuid"), Eq("test-client-uid"));
+  EXPECT_THAT(labels.at("location"), Eq("global"));
+  EXPECT_THAT(labels.at("cloud_platform"), Eq("unknown"));
+  EXPECT_THAT(labels.at("host_id"), Eq("unknown"));
+}
+
+TEST(GrpcMetricsExporterTest, MakeMonitoredResourcePartialAttributes) {
+  Options options;
+  std::string const client_uid = "test-client-uid";
+
+  opentelemetry::sdk::metrics::PointDataAttributes pda;
+  pda.attributes = opentelemetry::sdk::metrics::PointAttributes({
+      {"project_id", "test-project"},
+  });
+
+  auto result = MakeMonitoredResource(pda, options, client_uid);
+  EXPECT_THAT(result.project_id, Eq("test-project"));
+
+  auto const& resource = result.resource;
+  EXPECT_THAT(resource.type(), Eq("bigtable.googleapis.com/Client"));
+
+  auto const& labels = resource.labels();
+  EXPECT_THAT(labels.at("project_id"), Eq("test-project"));
+  EXPECT_THAT(labels.at("client_project"), Eq("test-project"));
+  EXPECT_THAT(labels.at("instance"), Eq(""));
+  EXPECT_THAT(labels.at("app_profile"), Eq(""));
+  EXPECT_THAT(labels.at("client_name"),
+              Eq("cpp.Bigtable/" + bigtable::version_string()));
+  EXPECT_THAT(labels.at("uuid"), Eq("test-client-uid"));
+  EXPECT_THAT(labels.at("location"), Eq("global"));
+  EXPECT_THAT(labels.at("cloud_platform"), Eq("unknown"));
+  EXPECT_THAT(labels.at("host_id"), Eq("unknown"));
+}
+
+TEST(GrpcMetricsExporterTest, MakeMonitoredResourceExtraAttributes) {
+  Options options;
+  std::string const client_uid = "test-client-uid";
+
+  opentelemetry::sdk::metrics::PointDataAttributes pda;
+  pda.attributes = opentelemetry::sdk::metrics::PointAttributes({
+      {"project_id", "test-project"},
+      {"instance", "test-instance"},
+      {"grpc.method", "google.bigtable.v2.Bigtable/ReadRows"},
+      {"grpc.status", "OK"},
+  });
+
+  auto result = MakeMonitoredResource(pda, options, client_uid);
+  EXPECT_THAT(result.project_id, Eq("test-project"));
+
+  auto const& resource = result.resource;
+  EXPECT_THAT(resource.type(), Eq("bigtable.googleapis.com/Client"));
+
+  auto const& labels = resource.labels();
+  EXPECT_THAT(labels.size(), Eq(9));
+  EXPECT_THAT(labels.at("project_id"), Eq("test-project"));
+  EXPECT_THAT(labels.at("client_project"), Eq("test-project"));
+  EXPECT_THAT(labels.at("instance"), Eq("test-instance"));
+  EXPECT_THAT(labels.at("app_profile"), Eq(""));
+  EXPECT_THAT(labels.at("client_name"),
+              Eq("cpp.Bigtable/" + bigtable::version_string()));
+  EXPECT_THAT(labels.at("uuid"), Eq("test-client-uid"));
+  EXPECT_THAT(labels.at("location"), Eq("global"));
+  EXPECT_THAT(labels.at("cloud_platform"), Eq("unknown"));
+  EXPECT_THAT(labels.at("host_id"), Eq("unknown"));
+  EXPECT_THAT(labels.find("grpc.method"), Eq(labels.end()));
+  EXPECT_THAT(labels.find("grpc.status"), Eq(labels.end()));
+}
+
+TEST(GrpcMetricsExporterTest, MakeMonitoredResourceWithDetectedResource) {
+  namespace sc = ::opentelemetry::semconv;
+  Options options;
+  options.set<bigtable::AppProfileIdOption>("test-app-profile");
+  std::string const client_uid = "test-client-uid";
+
+  opentelemetry::sdk::metrics::PointDataAttributes pda;
+  pda.attributes = opentelemetry::sdk::metrics::PointAttributes({
+      {"project_id", "test-project"},
+      {"instance", "test-instance"},
+  });
+
+  auto detected_resource = opentelemetry::sdk::resource::Resource::Create({
+      {sc::cloud::kCloudAccountId, "detected-vm-project"},
+      {sc::cloud::kCloudPlatform, "gcp_compute_engine"},
+      {sc::cloud::kCloudAvailabilityZone, "us-central1-a"},
+      {sc::host::kHostId, "123456789"},
+      {sc::host::kHostName, "test-vm-host"},
+  });
+
+  auto result =
+      MakeMonitoredResource(pda, detected_resource, options, client_uid);
+  EXPECT_THAT(result.project_id, Eq("test-project"));
+
+  auto const& resource = result.resource;
+  EXPECT_THAT(resource.type(), Eq("bigtable.googleapis.com/Client"));
+
+  auto const& labels = resource.labels();
+  EXPECT_THAT(labels.at("project_id"), Eq("test-project"));
+  EXPECT_THAT(labels.at("client_project"), Eq("detected-vm-project"));
+  EXPECT_THAT(labels.at("instance"), Eq("test-instance"));
+  EXPECT_THAT(labels.at("app_profile"), Eq("test-app-profile"));
+  EXPECT_THAT(labels.at("client_name"),
+              Eq("cpp.Bigtable/" + bigtable::version_string()));
+  EXPECT_THAT(labels.at("uuid"), Eq("test-client-uid"));
+  EXPECT_THAT(labels.at("location"), Eq("us-central1-a"));
+  EXPECT_THAT(labels.at("cloud_platform"), Eq("gcp_compute_engine"));
+  EXPECT_THAT(labels.at("host_id"), Eq("123456789"));
+  EXPECT_THAT(labels.at("hostname"), Eq("test-vm-host"));
+}
+
+TEST(GrpcMetricsExporterTest, MakeLatencyHistogramBoundaries) {
+  auto const boundaries = MakeLatencyHistogramBoundaries();
+  ASSERT_THAT(boundaries, Not(IsEmpty()));
+  ASSERT_THAT(boundaries, SizeIs(Le(200U)));
+  auto sorted = boundaries;
+  std::sort(sorted.begin(), sorted.end());
+  EXPECT_THAT(boundaries, ElementsAreArray(sorted.begin(), sorted.end()));
+  std::vector<double> diff;
+  std::adjacent_difference(boundaries.begin(), boundaries.end(),
+                           std::back_inserter(diff));
+  ASSERT_THAT(diff, Not(IsEmpty()));
+  EXPECT_THAT(*std::min_element(std::next(diff.begin()), diff.end()),
+              Ge(0.001));
+  ASSERT_THAT(boundaries.back(), Le(300));
+}
+
+TEST(GrpcMetricsExporterTest, ValidateGrpcClientAttemptDuration) {
+  std::atomic<int> export_count{0};
+  auto mock = std::make_unique<MockPushMetricExporter>();
+  EXPECT_CALL(*mock, Shutdown).WillOnce(Return(true));
+  EXPECT_CALL(*mock, GetAggregationTemporality)
+      .WillRepeatedly(Return(
+          opentelemetry::sdk::metrics::AggregationTemporality::kCumulative));
+  EXPECT_CALL(*mock, Export(MetricDataEmpty()))
+      .WillRepeatedly(
+          Return(opentelemetry::sdk::common::ExportResult::kSuccess));
+  EXPECT_CALL(*mock, Export(MetricDataNotEmpty()))
+      .Times(AtLeast(1))
+      .WillRepeatedly(
+          [&export_count](
+              opentelemetry::sdk::metrics::ResourceMetrics const& data) {
+            EXPECT_THAT(
+                data.scope_metric_data_,
+                Contains(WithMetricsData(Contains(AllOf(
+                    MatchesInstrumentName(kDurationMetric),
+                    WithPointData(ElementsAre(ExpectedLatencyHistogram())))))));
+            ++export_count;
+            return opentelemetry::sdk::common::ExportResult::kSuccess;
+          });
+
+  {
+    auto provider = MakeGrpcMeterProvider(std::move(mock), TestReaderOptions());
+    auto meter = provider->GetMeter("grpc-c++", grpc::Version());
+    auto histogram = meter->CreateDoubleHistogram(kDurationMetric,
+                                                  "test-only-description", "s");
+    for (int i = 0; i != 50 && export_count.load() == 0; ++i) {
+      histogram->Record(1.0, opentelemetry::context::Context{});
+      std::this_thread::sleep_for(kExportInterval);
+    }
+  }
+}
+
+}  // namespace
+GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
+}  // namespace bigtable_internal
+}  // namespace cloud
+}  // namespace google
+
+#endif  // GOOGLE_CLOUD_CPP_BIGTABLE_WITH_GRPC_OTEL_METRICS
diff --git a/google/cloud/bigtable/testing/embedded_server_test_fixture.cc b/google/cloud/bigtable/testing/embedded_server_test_fixture.cc
index 3d8967ae186e3..606de417c6a39 100644
--- a/google/cloud/bigtable/testing/embedded_server_test_fixture.cc
+++ b/google/cloud/bigtable/testing/embedded_server_test_fixture.cc
@@ -16,6 +16,7 @@
 #include "google/cloud/bigtable/internal/bigtable_metadata_decorator.h"
 #include "google/cloud/bigtable/internal/bigtable_stub.h"
 #include "google/cloud/bigtable/internal/data_connection_impl.h"
+#include "google/cloud/bigtable/internal/grpc_metrics_exporter.h"
 #include "google/cloud/bigtable/internal/mutate_rows_limiter.h"
 #include "google/cloud/bigtable/options.h"
 #include "google/cloud/bigtable/retry_policy.h"
@@ -82,7 +83,10 @@ void EmbeddedServerTestFixture::SetUp() {
   data_connection_ = std::make_shared<bigtable_internal::DataConnectionImpl>(
       std::make_unique<
           google::cloud::internal::AutomaticallyCreatedBackgroundThreads>(),
-      stub, std::make_shared<bigtable_internal::NoopMutateRowsLimiter>(),
+      stub,
+      std::make_unique<bigtable_internal::SimpleOperationContextFactory>(),
+      /*grpc_metrics_exporter=*/nullptr,
+      std::make_shared<bigtable_internal::NoopMutateRowsLimiter>(),
       std::move(opts));
 
   table_ = std::make_shared<bigtable::Table>(
diff --git a/google/cloud/bigtable/testing/table_integration_test.cc b/google/cloud/bigtable/testing/table_integration_test.cc
index ab39945248a93..7b3315924de10 100644
--- a/google/cloud/bigtable/testing/table_integration_test.cc
+++ b/google/cloud/bigtable/testing/table_integration_test.cc
@@ -124,6 +124,11 @@ void TableAdminTestEnvironment::TearDown() {
 
 void TableIntegrationTest::SetUp() {
   Options options;
+  // Disable metrics for the setup connection used to clean up/create tables.
+  // This prevents premature registration of the process-global gRPC telemetry
+  // plugin with default (60-second) periods, which would otherwise override
+  // the custom period (5 seconds) requested inside the tests.
+  options.set<google::cloud::bigtable::EnableMetricsOption>(false);
   if (google::cloud::internal::GetEnv(
           "GOOGLE_CLOUD_CPP_BIGTABLE_TESTING_CHANNEL_POOL")
           .value_or("") == "dynamic") {
diff --git a/google/cloud/bigtable/tests/CMakeLists.txt b/google/cloud/bigtable/tests/CMakeLists.txt
index ff7c9e9eb8b04..103b258e94141 100644
--- a/google/cloud/bigtable/tests/CMakeLists.txt
+++ b/google/cloud/bigtable/tests/CMakeLists.txt
@@ -23,6 +23,7 @@ set(bigtable_client_integration_tests
     filters_integration_test.cc
     instance_admin_integration_test.cc
     mutations_integration_test.cc
+    observability_integration_test.cc
     table_admin_backup_integration_test.cc
     table_admin_iam_policy_integration_test.cc
     table_admin_integration_test.cc
@@ -32,6 +33,11 @@ include(CreateBazelConfig)
 export_list_to_bazel("bigtable_client_integration_tests.bzl"
                      "bigtable_client_integration_tests" YEAR "2018")
 
+if (NOT TARGET otel_collector)
+    add_subdirectory("${PROJECT_SOURCE_DIR}/ci/otel_collector"
+                     "${CMAKE_BINARY_DIR}/ci/otel_collector")
+endif ()
+
 foreach (fname ${bigtable_client_integration_tests})
     google_cloud_cpp_add_executable(target "bigtable" "${fname}")
     target_link_libraries(
@@ -49,6 +55,9 @@ foreach (fname ${bigtable_client_integration_tests})
                 gRPC::grpc++
                 gRPC::grpc
                 protobuf::libprotobuf)
+    if (TARGET otel_collector)
+        target_link_libraries(${target} PRIVATE otel_collector)
+    endif ()
     add_test(NAME ${target} COMMAND ${target})
     set_tests_properties(
         ${target} PROPERTIES LABELS
diff --git a/google/cloud/bigtable/tests/bigtable_client_integration_tests.bzl b/google/cloud/bigtable/tests/bigtable_client_integration_tests.bzl
index 907e01e0933dc..763abed29adeb 100644
--- a/google/cloud/bigtable/tests/bigtable_client_integration_tests.bzl
+++ b/google/cloud/bigtable/tests/bigtable_client_integration_tests.bzl
@@ -22,6 +22,7 @@ bigtable_client_integration_tests = [
     "filters_integration_test.cc",
     "instance_admin_integration_test.cc",
     "mutations_integration_test.cc",
+    "observability_integration_test.cc",
     "table_admin_backup_integration_test.cc",
     "table_admin_iam_policy_integration_test.cc",
     "table_admin_integration_test.cc",
diff --git a/google/cloud/bigtable/tests/observability_integration_test.cc b/google/cloud/bigtable/tests/observability_integration_test.cc
index c2bc2b4c02214..af57dd80dc495 100644
--- a/google/cloud/bigtable/tests/observability_integration_test.cc
+++ b/google/cloud/bigtable/tests/observability_integration_test.cc
@@ -12,17 +12,26 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+#ifndef _WIN32
+
 #include "google/cloud/bigtable/options.h"
 #include "google/cloud/bigtable/testing/table_integration_test.h"
 #include "google/cloud/credentials.h"
+#include "google/cloud/grpc_options.h"
+#include "google/cloud/internal/getenv.h"
 #include "google/cloud/testing_util/scoped_environment.h"
 #include "google/cloud/testing_util/status_matchers.h"
+#include "absl/strings/match.h"
 #include "absl/strings/str_cat.h"
 #include "absl/strings/str_split.h"
 #include "ci/otel_collector/otel_collector.h"
+#include <arpa/inet.h>
 #include <gmock/gmock.h>
+#include <netinet/in.h>
 #include <chrono>
 #include <thread>
+#include <sys/socket.h>
+#include <unistd.h>
 
 namespace google {
 namespace cloud {
@@ -30,6 +39,20 @@ namespace bigtable {
 namespace testing {
 namespace {
 
+bool IsDirectPathReachable() {
+  int s = socket(AF_INET6, SOCK_STREAM, 0);
+  if (s < 0) return false;
+  sockaddr_in6 addr{};
+  addr.sin6_family = AF_INET6;
+  addr.sin6_port = htons(443);
+  inet_pton(AF_INET6, "2607:f8b0:4001:c2f::5f", &addr.sin6_addr);
+  timeval tv{1, 0};
+  setsockopt(s, SOL_SOCKET, SO_SNDTIMEO, &tv, sizeof(tv));
+  int res = connect(s, reinterpret_cast<sockaddr*>(&addr), sizeof(addr));
+  close(s);
+  return res == 0;
+}
+
 using ::google::cloud::bigtable::testing::TableTestEnvironment;
 using ::google::cloud::testing_util::ScopedEnvironment;
 using ::testing::StartsWith;
@@ -37,8 +60,7 @@ using ::testing::StartsWith;
 class ObservabilityIntegrationTest
     : public ::google::cloud::bigtable::testing::TableIntegrationTest {
  protected:
-  void SetUp() override {
-    TableIntegrationTest::SetUp();
+  static void SetUpTestSuite() {
     int port = 0;
     grpc::ServerBuilder builder;
     builder.AddListeningPort("localhost:0", grpc::InsecureServerCredentials(),
@@ -52,21 +74,42 @@ class ObservabilityIntegrationTest
             &collector_service_));
     server_ = builder.BuildAndStart();
     server_address_ = absl::StrCat("localhost:", port);
+    env_endpoint_ = std::make_unique<ScopedEnvironment>(
+        "GOOGLE_CLOUD_CPP_METRIC_SERVICE_ENDPOINT", server_address_);
+    env_otel_ = std::make_unique<ScopedEnvironment>(
+        "GOOGLE_CLOUD_CPP_TESTING_OTEL_COLLECTOR", "1");
   }
 
-  void TearDown() override {
+  static void TearDownTestSuite() {
+    env_endpoint_.reset();
+    env_otel_.reset();
     if (server_) {
-      server_->Shutdown();
+      server_->Shutdown(std::chrono::system_clock::now() +
+                        std::chrono::seconds(1));
       server_->Wait();
     }
-    TableIntegrationTest::TearDown();
   }
 
-  google::cloud::testing_util::OtelCollectorServer collector_service_;
-  std::unique_ptr<grpc::Server> server_;
-  std::string server_address_;
+  void SetUp() override {
+    TableIntegrationTest::SetUp();
+    data_connection_.reset();
+    collector_service_.Clear();
+  }
+
+  static google::cloud::testing_util::OtelCollectorServer collector_service_;
+  static std::unique_ptr<grpc::Server> server_;
+  static std::string server_address_;
+  static std::unique_ptr<ScopedEnvironment> env_endpoint_;
+  static std::unique_ptr<ScopedEnvironment> env_otel_;
 };
 
+google::cloud::testing_util::OtelCollectorServer
+    ObservabilityIntegrationTest::collector_service_;
+std::unique_ptr<grpc::Server> ObservabilityIntegrationTest::server_;
+std::string ObservabilityIntegrationTest::server_address_;
+std::unique_ptr<ScopedEnvironment> ObservabilityIntegrationTest::env_endpoint_;
+std::unique_ptr<ScopedEnvironment> ObservabilityIntegrationTest::env_otel_;
+
 /// Use Table::Apply() to insert a single row.
 void Apply(Table& table, std::string const& row_key,
            std::vector<Cell> const& cells) {
@@ -82,21 +125,45 @@ void Apply(Table& table, std::string const& row_key,
   ASSERT_STATUS_OK(status);
 }
 
+void VerifyResourceLabels(google::monitoring::v3::TimeSeries const& ts,
+                          std::string const& expected_project_id,
+                          std::string const& expected_instance_id,
+                          std::string const& expected_table_id) {
+  auto const& labels = ts.resource().labels();
+  auto project_it = labels.find("project_id");
+  if (project_it != labels.end()) {
+    EXPECT_EQ(project_it->second, expected_project_id);
+  }
+  auto instance_it = labels.find("instance");
+  if (instance_it != labels.end()) {
+    EXPECT_EQ(instance_it->second, expected_instance_id);
+  }
+  auto table_it = labels.find("table");
+  if (table_it != labels.end()) {
+    EXPECT_EQ(table_it->second, expected_table_id);
+  }
+  auto zone_it = labels.find("zone");
+  if (zone_it != labels.end() && !TableTestEnvironment::zone_a().empty()) {
+    std::vector<absl::string_view> parts =
+        absl::StrSplit(TableTestEnvironment::zone_a(), '-');
+    auto prefix = parts.size() >= 2 ? absl::StrCat(parts[0], "-", parts[1])
+                                    : TableTestEnvironment::zone_a();
+    EXPECT_THAT(zone_it->second, StartsWith(prefix));
+  }
+}
+
 TEST_F(ObservabilityIntegrationTest, VerifyOperationAndAttemptMetrics) {
   if (UsingCloudBigtableEmulator()) {
     GTEST_SKIP() << "Metrics export integration test runs against production";
   }
 
-  // Redirect Cloud Monitoring metric export to local otel_collector
-  ScopedEnvironment env("GOOGLE_CLOUD_CPP_METRIC_SERVICE_ENDPOINT",
-                        server_address_);
-  ScopedEnvironment env_otel("GOOGLE_CLOUD_CPP_TESTING_OTEL_COLLECTOR", "1");
-
   // Set MetricsPeriodOption to 5s (minimum allowed by DefaultOptions; smaller
   // periods reset to 60s)
-  auto options =
-      Options{}.set<EnableMetricsOption>(true).set<MetricsPeriodOption>(
-          std::chrono::seconds(5));
+  auto options = Options{}
+                     .set<EnableMetricsOption>(true)
+                     .set<MetricsPeriodOption>(std::chrono::seconds(5))
+                     .set<MinConnectionRefreshOption>(std::chrono::hours(1))
+                     .set<MaxConnectionRefreshOption>(std::chrono::hours(1));
 
   auto const table_id = TableTestEnvironment::table_id();
 
@@ -132,37 +199,22 @@ TEST_F(ObservabilityIntegrationTest, VerifyOperationAndAttemptMetrics) {
 
     for (auto const& ts : req.time_series()) {
       auto const& metric_type = ts.metric().type();
-      EXPECT_THAT(metric_type,
-                  StartsWith("bigtable.googleapis.com/internal/client/"));
+      // Skip any non-bigtable metrics (e.g. gRPC metrics) that might be
+      // exported since we globally enabled them in the client connection.
+      // This test only validates custom Bigtable client metrics.
+      if (!absl::StartsWith(metric_type,
+                            "bigtable.googleapis.com/internal/client/")) {
+        continue;
+      }
 
-      if (metric_type.find("operation_latencies") != std::string::npos) {
+      if (absl::StrContains(metric_type, "operation_latencies")) {
         found_operation_latencies = true;
       }
-      if (metric_type.find("attempt_latencies") != std::string::npos) {
+      if (absl::StrContains(metric_type, "attempt_latencies")) {
         found_attempt_latencies = true;
       }
 
-      auto const& labels = ts.resource().labels();
-      auto project_it = labels.find("project_id");
-      if (project_it != labels.end()) {
-        EXPECT_EQ(project_it->second, project_id());
-      }
-      auto instance_it = labels.find("instance");
-      if (instance_it != labels.end()) {
-        EXPECT_EQ(instance_it->second, instance_id());
-      }
-      auto table_it = labels.find("table");
-      if (table_it != labels.end()) {
-        EXPECT_EQ(table_it->second, table_id);
-      }
-      auto zone_it = labels.find("zone");
-      if (zone_it != labels.end() && !TableTestEnvironment::zone_a().empty()) {
-        std::vector<absl::string_view> parts =
-            absl::StrSplit(TableTestEnvironment::zone_a(), '-');
-        auto prefix = parts.size() >= 2 ? absl::StrCat(parts[0], "-", parts[1])
-                                        : TableTestEnvironment::zone_a();
-        EXPECT_THAT(zone_it->second, StartsWith(prefix));
-      }
+      VerifyResourceLabels(ts, project_id(), instance_id(), table_id);
     }
   }
 
@@ -170,6 +222,177 @@ TEST_F(ObservabilityIntegrationTest, VerifyOperationAndAttemptMetrics) {
   EXPECT_TRUE(found_attempt_latencies);
 }
 
+bool VerifyDirectPathGrpcResourceLabels(
+    google::monitoring::v3::TimeSeries const& ts,
+    absl::optional<std::string> const& expected_location,
+    absl::optional<std::string> const& expected_cloud_platform,
+    std::string const& expected_client_project,
+    absl::optional<std::string> const& expected_hostname) {
+  auto const& labels = ts.resource().labels();
+  auto location_it = labels.find("location");
+  auto platform_it = labels.find("cloud_platform");
+  auto host_id_it = labels.find("host_id");
+  auto client_project_it = labels.find("client_project");
+
+  if (location_it == labels.end() || platform_it == labels.end() ||
+      host_id_it == labels.end() || client_project_it == labels.end()) {
+    return false;
+  }
+
+  EXPECT_FALSE(location_it->second.empty());
+  if (expected_location.has_value() && !expected_location->empty()) {
+    std::vector<absl::string_view> parts =
+        absl::StrSplit(*expected_location, '-');
+    auto region_prefix = parts.size() >= 2
+                             ? absl::StrCat(parts[0], "-", parts[1])
+                             : *expected_location;
+    EXPECT_THAT(location_it->second, StartsWith(region_prefix));
+  }
+
+  EXPECT_FALSE(platform_it->second.empty());
+  if (expected_cloud_platform.has_value() &&
+      !expected_cloud_platform->empty()) {
+    EXPECT_EQ(platform_it->second, *expected_cloud_platform);
+  }
+
+  EXPECT_FALSE(host_id_it->second.empty());
+
+  EXPECT_FALSE(client_project_it->second.empty());
+  if (!expected_client_project.empty()) {
+    EXPECT_EQ(client_project_it->second, expected_client_project);
+  }
+
+  if (expected_hostname.has_value() && !expected_hostname->empty()) {
+    auto hostname_it = labels.find("hostname");
+    if (hostname_it != labels.end()) {
+      EXPECT_EQ(hostname_it->second, *expected_hostname);
+    }
+  }
+
+  return true;
+}
+
+std::set<std::string> ProcessRecordedGrpcMetrics(
+    std::vector<google::monitoring::v3::CreateTimeSeriesRequest> const&
+        recorded,
+    std::string const& project_id,
+    absl::optional<std::string> const& expected_location,
+    absl::optional<std::string> const& expected_cloud_platform,
+    std::string const& expected_client_project,
+    absl::optional<std::string> const& expected_hostname,
+    bool& verified_resource_labels) {
+  std::set<std::string> grpc_metric_types;
+  for (auto const& req : recorded) {
+    std::cout << "req=" << req.DebugString() << std::endl;
+    EXPECT_EQ(req.name(), absl::StrCat("projects/", project_id));
+
+    for (auto const& ts : req.time_series()) {
+      auto const& metric_type = ts.metric().type();
+      if (absl::StartsWith(metric_type, "workload.googleapis.com/grpc.")) {
+        grpc_metric_types.insert(metric_type);
+        if (VerifyDirectPathGrpcResourceLabels(
+                ts, expected_location, expected_cloud_platform,
+                expected_client_project, expected_hostname)) {
+          verified_resource_labels = true;
+        }
+      }
+    }
+  }
+  return grpc_metric_types;
+}
+
+TEST_F(ObservabilityIntegrationTest, VerifyDirectPathGrpcMetrics) {
+  if (UsingCloudBigtableEmulator()) {
+    GTEST_SKIP() << "Metrics export integration test runs against production";
+  }
+
+  auto disable_direct_path =
+      google::cloud::internal::GetEnv("GOOGLE_CLOUD_DISABLE_DIRECT_PATH")
+          .value_or("");
+  if (disable_direct_path == "true" || !IsDirectPathReachable()) {
+    GTEST_SKIP() << "DirectPath is disabled or network is unreachable in this "
+                    "test environment";
+  }
+
+  ScopedEnvironment env_dp("CBT_ENABLE_DIRECTPATH", "true");
+
+  // Set MetricsPeriodOption to 5s (minimum allowed by DefaultOptions; smaller
+  // periods reset to 60s)
+  auto options = Options{}
+                     .set<EnableMetricsOption>(true)
+                     .set<MetricsPeriodOption>(std::chrono::seconds(5))
+                     .set<MinConnectionRefreshOption>(std::chrono::hours(1))
+                     .set<MaxConnectionRefreshOption>(std::chrono::hours(1))
+                     .set<GrpcChannelArgumentsOption>({
+                         {"grpc.client_idle_timeout_ms", "1000"},
+                         {"grpc.max_reconnect_backoff_ms", "1000"},
+                     });
+
+  auto const& table_id = TableTestEnvironment::table_id();
+
+  // Add scoped connection to ensure metrics are flushed on destruction.
+  {
+    auto conn = MakeDataConnection(
+        {InstanceResource(Project(project_id()), instance_id())}, options);
+    auto table = Table(std::move(conn),
+                       TableResource(project_id(), instance_id(), table_id));
+
+    std::string const row_key = "observability-directpath-row-1";
+    std::vector<Cell> expected{
+        {row_key, "family4", "c0", 1000, "v1000"},
+        {row_key, "family4", "c1", 2000, "v2000"},
+    };
+
+    // Perform mutations and read calls over DirectPath
+    Apply(table, row_key, expected);
+    auto actual = ReadRows(table, Filter::PassAllFilter());
+    CheckEqualUnordered(expected, actual);
+
+    // Wait for the periodic 5-second exporter background thread to flush
+    // metrics while conn is active
+    std::this_thread::sleep_for(std::chrono::seconds(6));
+  }
+
+  auto recorded = collector_service_.recorded_metrics();
+  ASSERT_FALSE(recorded.empty());
+
+  bool verified_resource_labels = false;
+
+  auto expected_client_project =
+      google::cloud::internal::GetEnv(
+          "GOOGLE_CLOUD_CPP_TEST_EXPECTED_CLIENT_PROJECT")
+          .value_or(project_id());
+  auto expected_location = google::cloud::internal::GetEnv(
+      "GOOGLE_CLOUD_CPP_TEST_EXPECTED_LOCATION");
+  auto expected_cloud_platform = google::cloud::internal::GetEnv(
+      "GOOGLE_CLOUD_CPP_TEST_EXPECTED_CLOUD_PLATFORM");
+  auto expected_hostname = google::cloud::internal::GetEnv(
+      "GOOGLE_CLOUD_CPP_TEST_EXPECTED_HOSTNAME");
+
+  auto grpc_metric_types = ProcessRecordedGrpcMetrics(
+      recorded, project_id(), expected_location, expected_cloud_platform,
+      expected_client_project, expected_hostname, verified_resource_labels);
+
+  EXPECT_TRUE(verified_resource_labels);
+
+  // Verify that specific gRPC client metrics configured in GrpcMetricsExporter
+  // are present. OpenTelemetry metric names are exported to Cloud Monitoring
+  // with the "workload.googleapis.com/" prefix.
+  //
+  // Note: Event-driven and failure-driven metrics configured in
+  // GrpcMetricsExporter (such as grpc.lb.rls.*, grpc.xds_client.*, and
+  // grpc.subchannel.* disconnections/failures) are only exported when those
+  // specific events or errors occur during the export window. Therefore, only
+  // RPC attempt metrics (duration, started, message sizes) are guaranteed to
+  // produce time series during a healthy test run.
+  EXPECT_THAT(grpc_metric_types,
+              ::testing::Contains(
+                  "workload.googleapis.com/grpc.client.attempt.duration"));
+  EXPECT_THAT(grpc_metric_types,
+              ::testing::Contains(
+                  "workload.googleapis.com/grpc.client.attempt.started"));
+}
+
 }  // namespace
 }  // namespace testing
 }  // namespace bigtable
@@ -182,3 +405,5 @@ int main(int argc, char* argv[]) {
       new ::google::cloud::bigtable::testing::TableTestEnvironment);
   return RUN_ALL_TESTS();
 }
+
+#endif  // _WIN32