feat(storage): add resource span attributes for ACO ( App Centric Observability )

google/cloud/google_cloud_cpp_common.bzl

@@ -55,6 +55,7 @@ google_cloud_cpp_common_hdrs = [
     "internal/future_fwd.h",
     "internal/future_impl.h",
     "internal/future_then_impl.h",
+    "internal/generic_background_threads_impl.h",
     "internal/getenv.h",
     "internal/group_options.h",
     "internal/invocation_id_generator.h",

google/cloud/google_cloud_cpp_common.cmake

@@ -86,6 +86,7 @@ add_library(
     internal/future_impl.h
     internal/future_then_impl.cc
     internal/future_then_impl.h
+    internal/generic_background_threads_impl.h
     internal/getenv.cc
     internal/getenv.h
     internal/group_options.h

google/cloud/google_cloud_cpp_rest_internal.cmake

@@ -298,6 +298,7 @@ if (BUILD_TESTING)
         internal/rest_lro_helpers_test.cc
         internal/rest_opentelemetry_test.cc
         internal/rest_parse_json_error_test.cc
+        internal/rest_pure_background_threads_impl_test.cc
         internal/rest_request_test.cc
         internal/rest_response_test.cc
         internal/rest_retry_loop_test.cc

google/cloud/google_cloud_cpp_rest_internal_unit_tests.bzl

@@ -61,6 +61,7 @@ google_cloud_cpp_rest_internal_unit_tests = [
     "internal/rest_lro_helpers_test.cc",
     "internal/rest_opentelemetry_test.cc",
     "internal/rest_parse_json_error_test.cc",
+    "internal/rest_pure_background_threads_impl_test.cc",
     "internal/rest_request_test.cc",
     "internal/rest_response_test.cc",
     "internal/rest_retry_loop_test.cc",

google/cloud/internal/background_threads_impl.cc

@@ -21,47 +21,7 @@
 namespace google {
 namespace cloud {
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
-namespace internal {
-
-AutomaticallyCreatedBackgroundThreads::AutomaticallyCreatedBackgroundThreads(
-    std::size_t thread_count)
-    : pool_(thread_count == 0 ? 1 : thread_count) {
-  std::generate_n(pool_.begin(), pool_.size(), [this] {
-    promise<void> started;
-    auto thread = std::thread(
-        [](CompletionQueue cq, promise<void>& started) {
-          started.set_value();
-          cq.Run();
-        },
-        cq_, std::ref(started));
-    started.get_future().wait();
-    return thread;
-  });
-}
-
-AutomaticallyCreatedBackgroundThreads::
-    ~AutomaticallyCreatedBackgroundThreads() {
-  Shutdown();
-}
-
-void AutomaticallyCreatedBackgroundThreads::Shutdown() {
-  cq_.Shutdown();
-  for (auto& t : pool_) {
-#if GOOGLE_CLOUD_CPP_HAVE_EXCEPTIONS
-    try {
-#endif
-      t.join();
-#if GOOGLE_CLOUD_CPP_HAVE_EXCEPTIONS
-    } catch (std::system_error const& e) {
-      GCP_LOG(FATAL) << "AutomaticallyCreatedBackgroundThreads::Shutdown: "
-                     << e.what();
-    }
-#endif
-  }
-  pool_.clear();
-}
-
-}  // namespace internal
+namespace internal {}  // namespace internal
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
 }  // namespace cloud
 }  // namespace google

google/cloud/internal/background_threads_impl.h

@@ -17,6 +17,7 @@
 
 #include "google/cloud/background_threads.h"
 #include "google/cloud/completion_queue.h"
+#include "google/cloud/internal/generic_background_threads_impl.h"
 #include "google/cloud/version.h"
 #include <thread>
 #include <vector>
@@ -40,19 +41,9 @@ class CustomerSuppliedBackgroundThreads : public BackgroundThreads {
 };
 
 /// Create a background thread to perform background operations.
-class AutomaticallyCreatedBackgroundThreads : public BackgroundThreads {
- public:
-  explicit AutomaticallyCreatedBackgroundThreads(std::size_t thread_count = 1U);
-  ~AutomaticallyCreatedBackgroundThreads() override;
-
-  CompletionQueue cq() const override { return cq_; }
-  void Shutdown();
-  std::size_t pool_size() const { return pool_.size(); }
-
- private:
-  CompletionQueue cq_;
-  std::vector<std::thread> pool_;
-};
+using AutomaticallyCreatedBackgroundThreads =
+    AutomaticallyCreatedBackgroundThreadsImpl<CompletionQueue,
+                                              BackgroundThreads>;
 
 }  // namespace internal
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END

google/cloud/internal/generic_background_threads_impl.h

@@ -0,0 +1,91 @@
+// 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_INTERNAL_GENERIC_BACKGROUND_THREADS_IMPL_H
+#define GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_INTERNAL_GENERIC_BACKGROUND_THREADS_IMPL_H
+
+#include "google/cloud/future.h"
+#include "google/cloud/log.h"
+#include "google/cloud/version.h"
+#include <algorithm>
+#include <system_error>
+#include <thread>
+#include <vector>
+
+namespace google {
+namespace cloud {
+GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
+namespace internal {
+
+template <typename QueueType>
+struct DefaultQueueTraits {
+  static QueueType Create() { return QueueType(); }
+  static void Run(QueueType cq, promise<void>& started) {
+    started.set_value();
+    cq.Run();
+  }
+};
+
+template <typename QueueType, typename BaseInterface,
+          typename QueueTraits = DefaultQueueTraits<QueueType>>
+class AutomaticallyCreatedBackgroundThreadsImpl : public BaseInterface {
+ public:
+  explicit AutomaticallyCreatedBackgroundThreadsImpl(
+      std::size_t thread_count = 1U)
+      : cq_(QueueTraits::Create()),
+        pool_(thread_count == 0 ? 1 : thread_count) {
+    std::generate_n(pool_.begin(), pool_.size(), [this] {
+      promise<void> started;
+      auto thread = std::thread(
+          [](QueueType cq, promise<void>& started) {
+            QueueTraits::Run(std::move(cq), started);
+          },
+          cq_, std::ref(started));
+      started.get_future().wait();
+      return thread;
+    });
+  }
+
+  ~AutomaticallyCreatedBackgroundThreadsImpl() override { Shutdown(); }
+
+  QueueType cq() const override { return cq_; }
+  std::size_t pool_size() const { return pool_.size(); }
+
+  void Shutdown() {
+    cq_.Shutdown();
+    for (auto& t : pool_) {
+#if GOOGLE_CLOUD_CPP_HAVE_EXCEPTIONS
+      try {
+#endif
+        if (t.joinable()) t.join();
+#if GOOGLE_CLOUD_CPP_HAVE_EXCEPTIONS
+      } catch (std::system_error const& e) {
+        GCP_LOG(FATAL) << "Shutdown error: " << e.what();
+      }
+#endif
+    }
+    pool_.clear();
+  }
+
+ private:
+  QueueType cq_;
+  std::vector<std::thread> pool_;
+};
+
+}  // namespace internal
+GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
+}  // namespace cloud
+}  // namespace google
+
+#endif  // GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_INTERNAL_GENERIC_BACKGROUND_THREADS_IMPL_H

google/cloud/internal/rest_pure_background_threads_impl.cc

@@ -25,48 +25,6 @@ namespace cloud {
 namespace rest_internal {
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
 
-AutomaticallyCreatedRestPureBackgroundThreads::
-    AutomaticallyCreatedRestPureBackgroundThreads(std::size_t thread_count)
-    : cq_(std::make_shared<RestPureCompletionQueueImpl>()),
-      pool_(thread_count == 0 ? 1 : thread_count) {
-  std::generate_n(pool_.begin(), pool_.size(), [this] {
-    promise<void> started;
-    auto thread = std::thread(
-        [](RestPureCompletionQueue cq, promise<void>& started,
-           internal::CallContext c) {
-          internal::ScopedCallContext scope(std::move(c));
-          started.set_value();
-          cq.Run();
-        },
-        cq_, std::ref(started), internal::CallContext{});
-    started.get_future().wait();
-    return thread;
-  });
-}
-
-AutomaticallyCreatedRestPureBackgroundThreads::
-    ~AutomaticallyCreatedRestPureBackgroundThreads() {
-  Shutdown();
-}
-
-void AutomaticallyCreatedRestPureBackgroundThreads::Shutdown() {
-  cq_.Shutdown();
-  for (auto& t : pool_) {
-#if GOOGLE_CLOUD_CPP_HAVE_EXCEPTIONS
-    try {
-#endif
-      t.join();
-#if GOOGLE_CLOUD_CPP_HAVE_EXCEPTIONS
-    } catch (std::system_error const& e) {
-      GCP_LOG(FATAL)
-          << "AutomaticallyCreatedRestPureBackgroundThreads::Shutdown: "
-          << e.what();
-    }
-#endif
-  }
-  pool_.clear();
-}
-
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
 }  // namespace rest_internal
 }  // namespace cloud

google/cloud/internal/rest_pure_background_threads_impl.h

@@ -15,6 +15,8 @@
 #ifndef GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_INTERNAL_REST_PURE_BACKGROUND_THREADS_IMPL_H
 #define GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_INTERNAL_REST_PURE_BACKGROUND_THREADS_IMPL_H
 
+#include "google/cloud/internal/call_context.h"
+#include "google/cloud/internal/generic_background_threads_impl.h"
 #include "google/cloud/internal/rest_pure_completion_queue_impl.h"
 #include "google/cloud/version.h"
 #include <thread>
@@ -37,23 +39,25 @@ class RestPureBackgroundThreads {
   virtual RestPureCompletionQueue cq() const = 0;
 };
 
-/// Background threads that run on a RestPureCompletionQueue.
-class AutomaticallyCreatedRestPureBackgroundThreads
-    : public RestPureBackgroundThreads {
- public:
-  explicit AutomaticallyCreatedRestPureBackgroundThreads(
-      std::size_t thread_count = 1U);
-  ~AutomaticallyCreatedRestPureBackgroundThreads() override;
-
-  RestPureCompletionQueue cq() const override { return cq_; }
-  void Shutdown();
-  std::size_t pool_size() const { return pool_.size(); }
-
- private:
-  RestPureCompletionQueue cq_;
-  std::vector<std::thread> pool_;
+struct RestPureQueueTraits {
+  static RestPureCompletionQueue Create() {
+    return RestPureCompletionQueue(
+        std::make_shared<RestPureCompletionQueueImpl>());
+  }
+  static void Run(RestPureCompletionQueue cq, promise<void>& started) {
+    internal::CallContext context;
+    internal::ScopedCallContext scope(std::move(context));
+    started.set_value();
+    cq.Run();
+  }
 };
 
+/// Background threads that run on a RestPureCompletionQueue.
+using AutomaticallyCreatedRestPureBackgroundThreads =
+    google::cloud::internal::AutomaticallyCreatedBackgroundThreadsImpl<
+        RestPureCompletionQueue, RestPureBackgroundThreads,
+        RestPureQueueTraits>;
+
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
 }  // namespace rest_internal
 }  // namespace cloud

google/cloud/internal/rest_pure_background_threads_impl_test.cc

@@ -0,0 +1,84 @@
+// 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/internal/rest_pure_background_threads_impl.h"
+#include <gmock/gmock.h>
+#include <set>
+#include <vector>
+
+namespace google {
+namespace cloud {
+namespace rest_internal {
+GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
+namespace {
+
+using ::testing::Contains;
+using ::testing::Not;
+
+/// @test Verify that automatically created completion queues are usable.
+TEST(AutomaticallyCreatedRestPureBackgroundThreads, IsActive) {
+  AutomaticallyCreatedRestPureBackgroundThreads actual;
+  EXPECT_EQ(1, actual.pool_size());
+
+  promise<std::thread::id> bg;
+  actual.cq().RunAsync([&bg] { bg.set_value(std::this_thread::get_id()); });
+  EXPECT_NE(std::this_thread::get_id(), bg.get_future().get());
+}
+
+/// @test Verify that automatically created completion queues are usable.
+TEST(AutomaticallyCreatedRestPureBackgroundThreads, NoEmptyPools) {
+  AutomaticallyCreatedRestPureBackgroundThreads actual(0);
+  EXPECT_EQ(1, actual.pool_size());
+
+  promise<std::thread::id> bg;
+  actual.cq().RunAsync([&bg] { bg.set_value(std::this_thread::get_id()); });
+  EXPECT_NE(std::this_thread::get_id(), bg.get_future().get());
+}
+
+/// @test Verify that automatically created completion queues work.
+TEST(AutomaticallyCreatedRestPureBackgroundThreads, ManyThreads) {
+  auto constexpr kThreadCount = 4;
+  AutomaticallyCreatedRestPureBackgroundThreads actual(kThreadCount);
+  EXPECT_EQ(kThreadCount, actual.pool_size());
+
+  std::vector<promise<std::thread::id>> promises(100 * kThreadCount);
+  for (auto& p : promises) {
+    actual.cq().RunAsync([&p] { p.set_value(std::this_thread::get_id()); });
+  }
+  std::set<std::thread::id> ids;
+  for (auto& p : promises) ids.insert(p.get_future().get());
+  EXPECT_FALSE(ids.empty());
+  EXPECT_GE(kThreadCount, ids.size());
+  EXPECT_THAT(ids, Not(Contains(std::this_thread::get_id())));
+}
+
+/// @test Verify that automatically created completion queues work.
+TEST(AutomaticallyCreatedRestPureBackgroundThreads, ManualShutdown) {
+  auto constexpr kThreadCount = 4;
+  AutomaticallyCreatedRestPureBackgroundThreads actual(kThreadCount);
+  EXPECT_EQ(kThreadCount, actual.pool_size());
+
+  std::vector<promise<void>> promises(2 * kThreadCount);
+  for (auto& p : promises) {
+    actual.cq().RunAsync([&p] { p.set_value(); });
+  }
+  for (auto& p : promises) p.get_future().get();
+  actual.Shutdown();
+}
+
+}  // namespace
+GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
+}  // namespace rest_internal
+}  // namespace cloud
+}  // namespace google

google/cloud/storage/client.cc

@@ -555,7 +555,8 @@ Options DefaultOptions(Options opts) {
                                    STORAGE_CLIENT_DEFAULT_MAXIMUM_BACKOFF_DELAY,
                                    STORAGE_CLIENT_DEFAULT_BACKOFF_SCALING)
               .clone())
-      .set<IdempotencyPolicyOption>(AlwaysRetryIdempotencyPolicy().clone());
+      .set<IdempotencyPolicyOption>(AlwaysRetryIdempotencyPolicy().clone())
+      .set<storage_experimental::OTelSpanEnrichmentOption>(true);
 
   o = google::cloud::internal::MergeOptions(std::move(opts), std::move(o));
   // If the application did not set `DownloadStallTimeoutOption` then use the