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/google/cloud/bigtable/data_connection.cc b/google/cloud/bigtable/data_connection.cc
index 6fd19a08706ae..89a7dc91f0c5e 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,40 @@ 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::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 (bigtable::internal::IsDirectPath(options) &&
+        options.has<bigtable_internal::InstanceChannelAffinityOption>() &&
+        options.get<experimental::DirectPathMetricsModeOption>() ==
+            experimental::DirectPathMetricsMode::kEnabled) {
+      bigtable_internal::EnableGrpcMetrics(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,14 +252,16 @@ 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(limiter), std::move(options));
+        std::move(operation_context_factory), std::move(limiter),
+        std::move(options));
   } else {
     auto stub = bigtable_internal::CreateBigtableStub(
         std::move(auth), background->cq(), options);
     conn = std::make_shared<bigtable_internal::DataConnectionImpl>(
         std::move(background),
         std::make_unique<bigtable_internal::StubManager>(std::move(stub)),
-        std::move(limiter), std::move(options));
+        std::move(operation_context_factory), std::move(limiter),
+        std::move(options));
   }
   if (google::cloud::internal::TracingEnabled(conn->options())) {
     conn = bigtable_internal::MakeDataTracingConnection(std::move(conn));
diff --git a/google/cloud/bigtable/data_connection_test.cc b/google/cloud/bigtable/data_connection_test.cc
index db96239c29d1d..ca4c710426ffc 100644
--- a/google/cloud/bigtable/data_connection_test.cc
+++ b/google/cloud/bigtable/data_connection_test.cc
@@ -109,6 +109,46 @@ TEST(MakeDataConnection, TestingOtelCollectorEnvVar) {
   auto conn = MakeDataConnection(TestOptions().set<EnableMetricsOption>(true));
   EXPECT_NE(conn, nullptr);
 }
+
+TEST(MakeDataConnection, DirectPathMetricsModeOptionDisabled) {
+  testing_util::ScopedEnvironment env("GOOGLE_CLOUD_CPP_TESTING_OTEL_COLLECTOR",
+                                      "1");
+  InstanceResource instance_a{Project("my-project"), "instance-a"};
+  auto conn = MakeDataConnection(
+      {instance_a}, TestOptions()
+                        .set<EnableMetricsOption>(true)
+                        .set<experimental::DirectPathMetricsModeOption>(
+                            experimental::DirectPathMetricsMode::kDisabled));
+  EXPECT_NE(conn, nullptr);
+}
+
+TEST(MakeDataConnection, DirectPathMetricsModeOptionEnabled) {
+  testing_util::ScopedEnvironment env("GOOGLE_CLOUD_CPP_TESTING_OTEL_COLLECTOR",
+                                      "1");
+  InstanceResource instance_a{Project("my-project"), "instance-a"};
+  auto conn = MakeDataConnection(
+      {instance_a}, TestOptions()
+                        .set<EnableMetricsOption>(true)
+                        .set<experimental::DirectPathModeOption>(
+                            experimental::DirectPathMode::kEnabled)
+                        .set<experimental::DirectPathMetricsModeOption>(
+                            experimental::DirectPathMetricsMode::kEnabled));
+  EXPECT_NE(conn, nullptr);
+}
+
+TEST(MakeDataConnection, DirectPathModeOptionDisabledNoGrpcMetrics) {
+  testing_util::ScopedEnvironment env("GOOGLE_CLOUD_CPP_TESTING_OTEL_COLLECTOR",
+                                      "1");
+  InstanceResource instance_a{Project("my-project"), "instance-a"};
+  auto conn = MakeDataConnection(
+      {instance_a}, TestOptions()
+                        .set<EnableMetricsOption>(true)
+                        .set<experimental::DirectPathModeOption>(
+                            experimental::DirectPathMode::kDisabled)
+                        .set<experimental::DirectPathMetricsModeOption>(
+                            experimental::DirectPathMetricsMode::kEnabled));
+  EXPECT_NE(conn, nullptr);
+}
 #endif
 
 }  // namespace
diff --git a/google/cloud/bigtable/internal/bigtable_stub_factory.cc b/google/cloud/bigtable/internal/bigtable_stub_factory.cc
index b9b2830616371..f3dfc55e61c6f 100644
--- a/google/cloud/bigtable/internal/bigtable_stub_factory.cc
+++ b/google/cloud/bigtable/internal/bigtable_stub_factory.cc
@@ -64,8 +64,8 @@ std::string CreateFeaturesMetadata(bool is_direct_path) {
   return internal::UrlsafeBase64EncodeWithPadding(proto.SerializeAsString());
 }
 
-std::string FeaturesMetadata() {
-  if (bigtable::internal::IsDirectPath()) {
+std::string FeaturesMetadata(Options const& options) {
+  if (bigtable::internal::IsDirectPath(options)) {
     static auto const* const kDirectPathFeatures =
         new std::string(CreateFeaturesMetadata(true));
     return *kDirectPathFeatures;
@@ -84,7 +84,7 @@ std::shared_ptr<BigtableStub> ApplyCommonDecorators(
   stub = std::make_shared<BigtableMetadata>(
       std::move(stub),
       std::multimap<std::string, std::string>{
-          {"bigtable-features", FeaturesMetadata()}},
+          {"bigtable-features", FeaturesMetadata(options)}},
       internal::HandCraftedLibClientHeader());
   if (internal::Contains(options.get<LoggingComponentsOption>(), "rpc")) {
     GCP_LOG(INFO) << "Enabled logging for gRPC calls";
diff --git a/google/cloud/bigtable/internal/data_connection_impl.cc b/google/cloud/bigtable/internal/data_connection_impl.cc
index 5a09bf55acaf1..a0da74dd7a9a6 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,45 +219,6 @@ 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,
@@ -290,6 +230,8 @@ DataConnectionImpl::DataConnectionImpl(
       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,
diff --git a/google/cloud/bigtable/internal/data_connection_impl.h b/google/cloud/bigtable/internal/data_connection_impl.h
index a859d5c938ace..82b757582fc1e 100644
--- a/google/cloud/bigtable/internal/data_connection_impl.h
+++ b/google/cloud/bigtable/internal/data_connection_impl.h
@@ -52,26 +52,14 @@ bigtable::Row TransformReadModifyWriteRowResponse(
 
 class DataConnectionImpl : public bigtable::DataConnection {
  public:
-  ~DataConnectionImpl() override = default;
+  ~DataConnectionImpl() override;
 
-  DataConnectionImpl(std::unique_ptr<BackgroundThreads> background,
-                     std::unique_ptr<StubManager> stub_manager,
-                     std::shared_ptr<MutateRowsLimiter> limiter,
-                     Options options);
-
-  // 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<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,
diff --git a/google/cloud/bigtable/internal/data_connection_impl_test.cc b/google/cloud/bigtable/internal/data_connection_impl_test.cc
index 7874bb8e2e5f0..537a29871e968 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>(), std::move(limiter),
+      Options{});
 }
 
 std::shared_ptr<DataConnectionImpl> TestConnection(
diff --git a/google/cloud/bigtable/internal/defaults.cc b/google/cloud/bigtable/internal/defaults.cc
index ff71df2383cba..20b862f92daeb 100644
--- a/google/cloud/bigtable/internal/defaults.cc
+++ b/google/cloud/bigtable/internal/defaults.cc
@@ -126,16 +126,20 @@ int DefaultConnectionPoolSize() {
                     cpu_count * BIGTABLE_CLIENT_DEFAULT_CHANNELS_PER_CPU);
 }
 
-bool IsDirectPath() {
+bool IsDirectPath(Options const& options) {
   auto const direct_path =
-      google::cloud::internal::GetEnv("GOOGLE_CLOUD_ENABLE_DIRECT_PATH")
-          .value_or("");
+      google::cloud::internal::GetEnv("GOOGLE_CLOUD_ENABLE_DIRECT_PATH");
   // Bigtable specific env var for Direct Path support used by all clients.
   auto const cbt_direct_path =
-      google::cloud::internal::GetEnv("CBT_ENABLE_DIRECTPATH").value_or("");
-  return absl::c_any_of(absl::StrSplit(direct_path, ','),
-                        [](absl::string_view v) { return v == "bigtable"; }) ||
-         cbt_direct_path == "true";
+      google::cloud::internal::GetEnv("CBT_ENABLE_DIRECTPATH");
+  if (direct_path.has_value() || cbt_direct_path.has_value()) {
+    return absl::c_any_of(
+               absl::StrSplit(direct_path.value_or(""), ','),
+               [](absl::string_view v) { return v == "bigtable"; }) ||
+           cbt_direct_path.value_or("") == "true";
+  }
+  return options.get<experimental::DirectPathModeOption>() ==
+         experimental::DirectPathMode::kEnabled;
 }
 
 Options HandleUniverseDomain(Options opts) {
@@ -184,7 +188,7 @@ Options DefaultOptions(Options opts) {
   }
 
   // Set the specific data endpoints if Direct Path is enabled.
-  if (IsDirectPath()) {
+  if (IsDirectPath(opts)) {
     opts.set<::google::cloud::bigtable_internal::DataEndpointOption>(
             "google-c2p:///bigtable.googleapis.com")
         .set<AuthorityOption>("bigtable.googleapis.com");
@@ -316,6 +320,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..411dda3dd0b54 100644
--- a/google/cloud/bigtable/internal/defaults.h
+++ b/google/cloud/bigtable/internal/defaults.h
@@ -29,7 +29,7 @@ int DefaultConnectionPoolSize();
 /**
  * Returns true if Direct Path is enabled for Bigtable.
  */
-bool IsDirectPath();
+bool IsDirectPath(Options const& options);
 
 /**
  * Returns an `Options` with the appropriate defaults for Bigtable.
@@ -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/defaults_test.cc b/google/cloud/bigtable/internal/defaults_test.cc
index 732e9fcdb7993..1ef15af5a3e9f 100644
--- a/google/cloud/bigtable/internal/defaults_test.cc
+++ b/google/cloud/bigtable/internal/defaults_test.cc
@@ -579,6 +579,45 @@ TEST(EndpointEnvTest, BigtableDirectPathOverridesUserEndpoints) {
   EXPECT_EQ("bigtable.googleapis.com", opts.get<AuthorityOption>());
 }
 
+TEST(EndpointEnvTest, DirectPathModeOptionEnabled) {
+  ScopedEnvironment emulator("BIGTABLE_EMULATOR_HOST", std::nullopt);
+  ScopedEnvironment direct_path("GOOGLE_CLOUD_ENABLE_DIRECT_PATH",
+                                std::nullopt);
+  ScopedEnvironment cbt_direct_path("CBT_ENABLE_DIRECTPATH", std::nullopt);
+
+  auto opts = Options{}.set<experimental::DirectPathModeOption>(
+      experimental::DirectPathMode::kEnabled);
+  EXPECT_TRUE(IsDirectPath(opts));
+  opts = DefaultOptions(opts);
+  EXPECT_EQ("google-c2p:///bigtable.googleapis.com",
+            opts.get<::google::cloud::bigtable_internal::DataEndpointOption>());
+  EXPECT_EQ("bigtable.googleapis.com", opts.get<AuthorityOption>());
+}
+
+TEST(EndpointEnvTest, DirectPathModeOptionDisabled) {
+  ScopedEnvironment emulator("BIGTABLE_EMULATOR_HOST", std::nullopt);
+  ScopedEnvironment direct_path("GOOGLE_CLOUD_ENABLE_DIRECT_PATH",
+                                std::nullopt);
+  ScopedEnvironment cbt_direct_path("CBT_ENABLE_DIRECTPATH", std::nullopt);
+
+  auto opts = Options{}.set<experimental::DirectPathModeOption>(
+      experimental::DirectPathMode::kDisabled);
+  EXPECT_FALSE(IsDirectPath(opts));
+  auto default_opts = DefaultDataOptions(opts);
+  EXPECT_EQ("bigtable.googleapis.com", default_opts.get<EndpointOption>());
+}
+
+TEST(EndpointEnvTest, DirectPathEnvVarOverridesDirectPathModeOption) {
+  ScopedEnvironment emulator("BIGTABLE_EMULATOR_HOST", std::nullopt);
+  ScopedEnvironment direct_path("GOOGLE_CLOUD_ENABLE_DIRECT_PATH",
+                                std::nullopt);
+  ScopedEnvironment cbt_direct_path("CBT_ENABLE_DIRECTPATH", "false");
+
+  auto opts = Options{}.set<experimental::DirectPathModeOption>(
+      experimental::DirectPathMode::kEnabled);
+  EXPECT_FALSE(IsDirectPath(opts));
+}
+
 TEST(EndpointEnvTest, EmulatorOverridesCloudDirectPath) {
   ScopedEnvironment emulator("BIGTABLE_EMULATOR_HOST", "emulator-host:8000");
   ScopedEnvironment direct_path("GOOGLE_CLOUD_ENABLE_DIRECT_PATH", "bigtable");
diff --git a/google/cloud/bigtable/options.h b/google/cloud/bigtable/options.h
index 92c9b010b1b0c..fcf5bc8762ea8 100644
--- a/google/cloud/bigtable/options.h
+++ b/google/cloud/bigtable/options.h
@@ -213,6 +213,35 @@ struct DynamicChannelPoolSizingPolicyOption {
   using Type = DynamicChannelPoolSizingPolicy;
 };
 
+enum class DirectPathMode {
+  kDisabled,  // Default.
+  kEnabled,
+};
+
+/**
+ * Option to control whether DirectPath should be used for Bigtable
+ * connections.
+ *
+ * By default, DirectPath is disabled.
+ */
+struct DirectPathModeOption {
+  using Type = DirectPathMode;
+};
+
+enum class DirectPathMetricsMode {
+  kEnabled,  // Default.
+  kDisabled,
+};
+
+/**
+ * Option to control whether DirectPath should emit OpenTelemetry metrics.
+ *
+ * By default, DirectPath OpenTelemetry metrics are enabled.
+ */
+struct DirectPathMetricsModeOption {
+  using Type = DirectPathMetricsMode;
+};
+
 }  // namespace experimental
 
 /// The complete list of options accepted by `bigtable::*Client`
diff --git a/google/cloud/bigtable/testing/embedded_server_test_fixture.cc b/google/cloud/bigtable/testing/embedded_server_test_fixture.cc
index 3d8967ae186e3..10c48063e3224 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,9 @@ 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>(),
+      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..b69c77d76c48f 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,174 @@ 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";
+  }
+
+  // Set MetricsPeriodOption to 5s (minimum allowed by DefaultOptions; smaller
+  // periods reset to 60s)
+  auto options = Options{}
+                     .set<EnableMetricsOption>(true)
+                     .set<experimental::DirectPathModeOption>(
+                         experimental::DirectPathMode::kEnabled)
+                     .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 duration metric is guaranteed to produce time series during a
+  // healthy test run.
+  EXPECT_THAT(grpc_metric_types,
+              ::testing::Contains(
+                  "workload.googleapis.com/grpc.client.attempt.duration"));
+}
+
 }  // namespace
 }  // namespace testing
 }  // namespace bigtable
@@ -182,3 +402,5 @@ int main(int argc, char* argv[]) {
       new ::google::cloud::bigtable::testing::TableTestEnvironment);
   return RUN_ALL_TESTS();
 }
+
+#endif  // _WIN32