[GcpObservability C++] De-experimentalize API (#32715)

This PR aims to de-experimentalize the APIs for GCP Observability. 

We would have ideally wanted public feedback before declaring the APIs
stable, but we need stable APIs for GA.

Changes made after API review with @markdroth @veblush, @ctiller and the
entire Core/C++ team -
* The old experimental APIs `grpc::experimental::GcpObservabilityInit`
and `grpc::experimental::GcpObservabilityClose` are now deprecated and
will be deleted after v.1.55 release.
* The new API gets rid of the Close method and follows the RAII idiom
with a single `grpc::GcpObservability::Init()` call that returns an
`GcpObservability` object, the lifetime of which controls when
observability data is flushed.
* The `GcpObservability` class could in the future add more methods. For
example, a debug method that shows the current configuration.
* Document that GcpObservability initialization and flushing (on
`GcpObservability` destruction) are blocking calls.
* Document that gRPC is still usable if GcpObservability initialization
failed. (Added a test to prove the same).
* Since we don't have a good way to flush stats and tracing with
OpenCensus, the examples required users to sleep for 25 seconds. This
sleep is now part of `GcpObservability` destruction.

Additional Implementation details -
* `GcpObservability::Init` is now marked with `GRPC_MUST_USE_RESULT` to
make sure that the results are used. We ideally want users to store it,
but this is better than nothing.
* Added a note on GCP Observability lifetime guarantees.

<!--

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

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

-->

examples/cpp/gcp_observability/helloworld/greeter_client.cc

@@ -88,11 +88,11 @@ int main(int argc, char** argv) {
   // Turn on GCP Observability for the whole binary. Based on the configuration,
   // this will emit observability data (stats, tracing and logging) to GCP
   // backends. Note that this should be done before any other gRPC operation.
-  auto status = grpc::experimental::GcpObservabilityInit();
-  if (!status.ok()) {
-    std::cerr << "GcpObservabilityInit() failed: " << status.ToString()
-              << std::endl;
-    return static_cast<int>(status.code());
+  auto observability = grpc::GcpObservability::Init();
+  if (!observability.ok()) {
+    std::cerr << "GcpObservability::Init() failed: "
+              << observability.status().ToString() << std::endl;
+    return static_cast<int>(observability.status().code());
   }
   std::cout << "Initialized GCP Observability" << std::endl;
   // We indicate that the channel isn't authenticated (use of
@@ -102,15 +102,7 @@ int main(int argc, char** argv) {
   std::string user("world");
   std::string reply = greeter.SayHello(user);
   std::cout << "Greeter received: " << reply << std::endl;
-  // Flush out any pending Observability data
-  std::cout << "Closing GCP Observability" << std::endl;
-  grpc::experimental::GcpObservabilityClose();
-  std::cout << "Sleeping for 25 seconds to make sure Observability stats and "
-               "tracing are flushed. Don't shut off server either."
-            << std::endl;
-  // Currently, GcpObservabilityClose() only supports flushing logs. Stats and
-  // tracing get automatically flushed at a regular interval, so sleep for an
-  // interval to make sure that those are flushed too.
-  std::this_thread::sleep_for(std::chrono::seconds(25));
+  // 'observability' object going out of scope will flush observability data.
+  std::cout << "Closing and flushing GCP Observability data" << std::endl;
   return 0;
 }

examples/cpp/gcp_observability/helloworld/greeter_server.cc

@@ -99,24 +99,15 @@ int main(int argc, char** argv) {
   // Turn on GCP Observability for the whole binary. Based on the configuration,
   // this will emit observability data (stats, tracing and logging) to GCP
   // backends. Note that this should be done before any other gRPC operation.
-  auto status = grpc::experimental::GcpObservabilityInit();
-  if (!status.ok()) {
-    std::cerr << "GcpObservabilityInit() failed: " << status.ToString()
-              << std::endl;
-    return static_cast<int>(status.code());
+  auto observability = grpc::GcpObservability::Init();
+  if (!observability.ok()) {
+    std::cerr << "GcpObservability::Init() failed: "
+              << observability.status().ToString() << std::endl;
+    return static_cast<int>(observability.status().code());
   }
   std::cout << "Initialized GCP Observability" << std::endl;
   RunServer(absl::GetFlag(FLAGS_port));
-  // Flush out any pending Observability data
-  std::cout << "Closing GCP Observability" << std::endl;
-  grpc::experimental::GcpObservabilityClose();
-  std::cout << "Sleeping for 25 seconds to make sure Observability stats and "
-               "tracing are flushed.(Another Ctrl+C will immediately exit the "
-               "program.)"
-            << std::endl;
-  // Currently, GcpObservabilityClose() only supports flushing logs. Stats and
-  // tracing get automatically flushed at a regular interval, so sleep for an
-  // interval to make sure that those are flushed too.
-  std::this_thread::sleep_for(std::chrono::seconds(25));
+  // 'observability' object going out of scope will flush observability data.
+  std::cout << "Closing and flushing GCP Observability data" << std::endl;
   return 0;
 }

include/grpcpp/ext/gcp_observability.h

@@ -17,25 +17,97 @@
 #ifndef GRPCPP_EXT_GCP_OBSERVABILITY_H
 #define GRPCPP_EXT_GCP_OBSERVABILITY_H
 
+#include <grpc/support/port_platform.h>
+
 #include "absl/status/status.h"
+#include "absl/status/statusor.h"
+
+#include <grpcpp/impl/grpc_library.h>
 
 namespace grpc {
-namespace experimental {
 
-// Initialize GCP Observability for gRPC.
-// This should be called before any other gRPC operations like creating a
-// channel, server, credentials etc.
-// The most common usage would call this at the top (or near the top) in main().
-// As an implementation detail, this properly initializes the OpenCensus stats
-// and tracing plugin, so applications do not need to perform any additional
-// gRPC C++ OpenCensus setup/registration to get GCP Observability for gRPC.
-absl::Status GcpObservabilityInit();
+// GcpObservability objects follow the RAII idiom and help manage the lifetime
+// of gRPC Observability data exporting to GCP. `GcpObservability::Init()`
+// should be invoked instead to return an `GcpObservability` instance.
+// Observability data is flushed at regular intervals, and also when this
+// instance goes out of scope and its destructor is invoked.
+class GcpObservability {
+ public:
+  // Initialize GCP Observability for gRPC.
+  // This should be called before any other gRPC operations like creating a
+  // channel, server, credentials etc.
+  // The return value helps determine whether observability was
+  // successfully enabled or not. On success, an object of class `Observability`
+  // is returned. When this object goes out of scope, GCP Observability stats,
+  // tracing and logging data is flushed. On failure, the status message can be
+  // used to determine the cause of failure. It is up to the applications to
+  // either crash on failure, or continue without GCP observability being
+  // enabled. The status codes do not have any special meaning at present, and
+  // users should not make any assumptions based on the status code, other than
+  // a non-OK status code meaning that observability initialization failed.
+  //
+  // The expected usage is to call this at the top (or near the top) in
+  // main(), and let it go out of scope after all RPCs and activities that we
+  // want to observe are done. Please look at
+  // https://github.com/grpc/grpc/blob/master/examples/cpp/gcp_observability/helloworld/greeter_client.cc
+  // and
+  // https://github.com/grpc/grpc/blob/master/examples/cpp/gcp_observability/helloworld/greeter_server.cc
+  // for sample usage.
+  //
+  // It is possible for an initialized GcpObservability object to go out of
+  // scope while RPCs and other gRPC operations are still ongoing. In this case,
+  // GCP Observability tries to flush all observability data collected till that
+  // point.
+  //
+  // Note that this is a blocking call which properly sets up gRPC Observability
+  // to work with GCP and might take a few seconds to return.  Similarly, the
+  // destruction of a non-moved-from `Observability` object is also blocking
+  // since it flushes the observability data to GCP.
+  //
+  // As an implementation detail, this properly initializes the OpenCensus stats
+  // and tracing plugin, so applications do not need to perform any additional
+  // gRPC C++ OpenCensus setup/registration to get GCP Observability for gRPC.
+  static absl::StatusOr<GcpObservability> Init() GRPC_MUST_USE_RESULT;
 
-// Gracefully shuts down GCP Observability.
-// Note that graceful shutdown for stats and tracing is not yet supported.
-void GcpObservabilityClose();
+  GcpObservability() = default;
+  // Move constructor and Move-assignment operator.
+  // The moved-from object will no longer be valid and will not cause GCP
+  // Observability stats, tracing and logging data to flush.
+  GcpObservability(GcpObservability&& other) noexcept;
+  GcpObservability& operator=(GcpObservability&& other) noexcept;
+
+  // Delete copy and copy-assignment operator
+  GcpObservability(const GcpObservability&) = delete;
+  GcpObservability& operator=(const GcpObservability&) = delete;
 
+ private:
+  // Helper class that aids in implementing GCP Observability.
+  // Inheriting from GrpcLibrary makes sure that gRPC is initialized and remains
+  // initialized for the lifetime of GCP Observability. In the future, when gRPC
+  // initialization goes away, we might still want to keep gRPC Event Engine
+  // initialized, just in case, we need to perform some IO operations during
+  // observability close.
+  // Note that the lifetime guarantees are only one way, i.e., GcpObservability
+  // object guarantees that gRPC will not shutdown while the object is still in
+  // scope, but the other way around does not hold true. Even though that is not
+  // the expected usage, GCP Observability can shutdown before gRPC shuts down.
+  // It follows that gRPC should not hold any callbacks from GcpObservability. A
+  // change in this restriction should go through a design review.
+  class GcpObservabilityImpl : private internal::GrpcLibrary {
+   public:
+    ~GcpObservabilityImpl() override;
+  };
+  std::unique_ptr<GcpObservabilityImpl> impl_;
+};
+
+namespace experimental {
+// TODO(yashykt): Delete this after the 1.55 release.
+GRPC_DEPRECATED("Use grpc::GcpObservability::Init() instead.")
+absl::Status GcpObservabilityInit();
+GRPC_DEPRECATED("Use grpc::GcpObservability::Init() instead.")
+void GcpObservabilityClose();
 }  // namespace experimental
+
 }  // namespace grpc
 
 #endif  // GRPCPP_EXT_GCP_OBSERVABILITY_H

src/cpp/ext/gcp/BUILD

@@ -40,6 +40,7 @@ grpc_cc_library(
         "absl/status",
         "absl/status:statusor",
         "absl/strings",
+        "absl/time",
         "absl/types:optional",
         "google/api:monitored_resource_cc_proto",
         "googleapis_monitoring_grpc_service",

src/cpp/ext/gcp/environment_autodetect.h

@@ -34,13 +34,10 @@
 
 namespace grpc {
 
-namespace experimental {
-// Forward declaration for GcpObservabilityInit
-absl::Status GcpObservabilityInit();
-}  // namespace experimental
-
 namespace internal {
 
+absl::Status GcpObservabilityInit();
+
 class EnvironmentAutoDetect {
  public:
   struct ResourceType {
@@ -64,7 +61,7 @@ class EnvironmentAutoDetect {
   }
 
  private:
-  friend absl::Status grpc::experimental::GcpObservabilityInit();
+  friend absl::Status grpc::internal::GcpObservabilityInit();
 
   // GcpObservabilityInit() is responsible for setting up the singleton with the
   // project_id.

src/cpp/ext/gcp/observability.cc

@@ -28,6 +28,8 @@
 #include "absl/status/status.h"
 #include "absl/status/statusor.h"
 #include "absl/strings/str_cat.h"
+#include "absl/time/clock.h"
+#include "absl/time/time.h"
 #include "absl/types/optional.h"
 #include "google/api/monitored_resource.pb.h"
 #include "google/devtools/cloudtrace/v2/tracing.grpc.pb.h"
@@ -45,6 +47,7 @@
 #include <grpcpp/support/channel_arguments.h>
 
 #include "src/core/ext/filters/logging/logging_filter.h"
+#include "src/core/lib/gprpp/crash.h"
 #include "src/core/lib/gprpp/notification.h"
 #include "src/cpp/client/client_stats_interceptor.h"
 #include "src/cpp/ext/filters/census/client_filter.h"
@@ -55,12 +58,14 @@
 #include "src/cpp/ext/gcp/observability_logging_sink.h"
 
 namespace grpc {
-namespace experimental {
 
+namespace internal {
 namespace {
 
 grpc::internal::ObservabilityLoggingSink* g_logging_sink = nullptr;
 
+bool g_gcp_observability_initialized = false;
+
 // TODO(yashykt): These constants are currently derived from the example at
 // https://cloud.google.com/traffic-director/docs/observability-proxyless#c++.
 // We might want these to be configurable.
@@ -74,18 +79,20 @@ constexpr char kGoogleStackdriverStatsAddress[] = "monitoring.googleapis.com";
 
 void RegisterOpenCensusViewsForGcpObservability() {
   // Register client default views for GCP observability
-  ClientStartedRpcs().RegisterForExport();
-  ClientCompletedRpcs().RegisterForExport();
-  ClientRoundtripLatency().RegisterForExport();
+  experimental::ClientStartedRpcs().RegisterForExport();
+  experimental::ClientCompletedRpcs().RegisterForExport();
+  experimental::ClientRoundtripLatency().RegisterForExport();
   internal::ClientApiLatency().RegisterForExport();
-  ClientSentCompressedMessageBytesPerRpc().RegisterForExport();
-  ClientReceivedCompressedMessageBytesPerRpc().RegisterForExport();
+  experimental::ClientSentCompressedMessageBytesPerRpc().RegisterForExport();
+  experimental::ClientReceivedCompressedMessageBytesPerRpc()
+      .RegisterForExport();
   // Register server default views for GCP observability
-  ServerStartedRpcs().RegisterForExport();
-  ServerCompletedRpcs().RegisterForExport();
-  ServerSentCompressedMessageBytesPerRpc().RegisterForExport();
-  ServerReceivedCompressedMessageBytesPerRpc().RegisterForExport();
-  ServerServerLatency().RegisterForExport();
+  experimental::ServerStartedRpcs().RegisterForExport();
+  experimental::ServerCompletedRpcs().RegisterForExport();
+  experimental::ServerSentCompressedMessageBytesPerRpc().RegisterForExport();
+  experimental::ServerReceivedCompressedMessageBytesPerRpc()
+      .RegisterForExport();
+  experimental::ServerServerLatency().RegisterForExport();
 }
 
 }  // namespace
@@ -100,6 +107,10 @@ absl::Status GcpObservabilityInit() {
       !config->cloud_logging.has_value()) {
     return absl::OkStatus();
   }
+  if (g_gcp_observability_initialized) {
+    grpc_core::Crash("GCP Observability for gRPC was already initialized.");
+  }
+  g_gcp_observability_initialized = true;
   grpc::internal::EnvironmentAutoDetect::Create(config->project_id);
   if (!config->cloud_trace.has_value()) {
     // Disable OpenCensus tracing
@@ -206,7 +217,55 @@ void GcpObservabilityClose() {
   if (g_logging_sink != nullptr) {
     g_logging_sink->FlushAndClose();
   }
+  // Currently, GcpObservabilityClose() only supports flushing logs. Stats and
+  // tracing get automatically flushed at a regular interval, so sleep for an
+  // interval to make sure that those are flushed too.
+  absl::SleepFor(absl::Seconds(25));
 }
 
+}  // namespace internal
+
+namespace experimental {
+
+absl::Status GcpObservabilityInit() {
+  return grpc::internal::GcpObservabilityInit();
+}
+
+void GcpObservabilityClose() { return grpc::internal::GcpObservabilityClose(); }
+
 }  // namespace experimental
+
+//
+// GcpObservability
+//
+
+absl::StatusOr<GcpObservability> GcpObservability::Init() {
+  absl::Status status = grpc::internal::GcpObservabilityInit();
+  if (!status.ok()) {
+    return status;
+  }
+  GcpObservability obj;
+  obj.impl_ = std::make_unique<GcpObservabilityImpl>();
+  return obj;
+}
+
+GcpObservability::GcpObservability(GcpObservability&& other) noexcept
+    : impl_(std::move(other.impl_)) {}
+
+GcpObservability& GcpObservability::operator=(
+    GcpObservability&& other) noexcept {
+  if (this != &other) {
+    impl_ = std::move(other.impl_);
+  }
+  return *this;
+}
+
+//
+// GcpObservability::GcpObservabilityImpl
+//
+
+GcpObservability::GcpObservabilityImpl::~GcpObservabilityImpl() {
+  grpc::internal::GcpObservabilityClose();
+}
+
 }  // namespace grpc

test/cpp/ext/gcp/BUILD

@@ -30,6 +30,8 @@ grpc_cc_test(
     uses_polling = False,
     deps = [
         "//:grpcpp_gcp_observability",
+        "//src/proto/grpc/testing:echo_proto",
+        "//test/cpp/end2end:test_service_impl",
         "//test/cpp/util:test_util",
     ],
 )

test/cpp/ext/gcp/observability_test.cc

@@ -17,25 +17,64 @@
 #include "gmock/gmock.h"
 #include "gtest/gtest.h"
 
+#include <grpc++/grpc++.h>
 #include <grpcpp/ext/gcp_observability.h>
 
 #include "src/core/lib/config/core_configuration.h"
+#include "src/proto/grpc/testing/echo.grpc.pb.h"
+#include "src/proto/grpc/testing/echo_messages.pb.h"
+#include "test/core/util/port.h"
 #include "test/core/util/test_config.h"
+#include "test/cpp/end2end/test_service_impl.h"
 
+namespace grpc {
+namespace testing {
 namespace {
 
-TEST(GcpObservabilityTest, RegistrationTest) {
-  auto status = grpc::experimental::GcpObservabilityInit();
-  EXPECT_EQ(status,
+TEST(GcpObservabilityTest, Basic) {
+  auto observability = grpc::GcpObservability::Init();
+  EXPECT_EQ(observability.status(),
             absl::FailedPreconditionError(
                 "Environment variables GRPC_GCP_OBSERVABILITY_CONFIG_FILE or "
                 "GRPC_GCP_OBSERVABILITY_CONFIG "
                 "not defined"));
-
   grpc_core::CoreConfiguration::Reset();
 }
 
+TEST(GcpObservabilityTest, ContinuesWorkingAfterFailure) {
+  auto observability = grpc::GcpObservability::Init();
+  EXPECT_FALSE(observability.ok());
+
+  // Set up a synchronous server on a different thread to avoid the asynch
+  // interface.
+  grpc::ServerBuilder builder;
+  TestServiceImpl service;
+  int port = grpc_pick_unused_port_or_die();
+  auto server_address = absl::StrCat("localhost:", port);
+  // Use IPv4 here because it's less flaky than IPv6 ("[::]:0") on Travis.
+  builder.AddListeningPort(server_address, grpc::InsecureServerCredentials(),
+                           &port);
+  builder.RegisterService(&service);
+  auto server = builder.BuildAndStart();
+  ASSERT_NE(nullptr, server);
+  auto server_thread = std::thread([&]() { server->Wait(); });
+  // Send a single RPC to make sure that things work.
+  auto stub = EchoTestService::NewStub(
+      grpc::CreateChannel(server_address, grpc::InsecureChannelCredentials()));
+  EchoRequest request;
+  request.set_message("foo");
+  EchoResponse response;
+  grpc::ClientContext context;
+  grpc::Status status = stub->Echo(&context, request, &response);
+  EXPECT_TRUE(status.ok());
+  EXPECT_EQ(response.message(), "foo");
+  server->Shutdown();
+  server_thread.join();
+}
+
 }  // namespace
+}  // namespace testing
+}  // namespace grpc
 
 int main(int argc, char** argv) {
   grpc::testing::TestEnvironment env(&argc, argv);