[Core] Fix the segfault from Opencensus upon shutdown (#36906)

src/ray/core_worker/core_worker_process.cc

@@ -146,7 +146,6 @@ CoreWorkerProcessImpl::CoreWorkerProcessImpl(const CoreWorkerOptions &options)
 
 CoreWorkerProcessImpl::~CoreWorkerProcessImpl() {
   RAY_LOG(INFO) << "Destructing CoreWorkerProcessImpl. pid: " << getpid();
-  RAY_LOG(DEBUG) << "Stats stop in core worker.";
   // Shutdown stats module if worker process exits.
   stats::Shutdown();
   if (options_.enable_logging) {

src/ray/stats/metric_exporter.cc

@@ -111,6 +111,7 @@ OpenCensusProtoExporter::OpenCensusProtoExporter(const int port,
                                                  const std::string address,
                                                  const WorkerID &worker_id)
     : client_call_manager_(io_service), worker_id_(worker_id) {
+  absl::MutexLock l(&mu_);
   client_.reset(new rpc::MetricsAgentClient(address, port, client_call_manager_));
 };
 
@@ -202,15 +203,18 @@ void OpenCensusProtoExporter::ExportViewData(
     }
   }
 
-  client_->ReportOCMetrics(
-      request_proto, [](const Status &status, const rpc::ReportOCMetricsReply &reply) {
-        RAY_UNUSED(reply);
-        if (!status.ok()) {
-          RAY_LOG_EVERY_N(WARNING, 10000)
-              << "Export metrics to agent failed: " << status
-              << ". This won't affect Ray, but you can lose metrics from the cluster.";
-        }
-      });
+  {
+    absl::MutexLock l(&mu_);
+    client_->ReportOCMetrics(
+        request_proto, [](const Status &status, const rpc::ReportOCMetricsReply &reply) {
+          RAY_UNUSED(reply);
+          if (!status.ok()) {
+            RAY_LOG_EVERY_N(WARNING, 10000)
+                << "Export metrics to agent failed: " << status
+                << ". This won't affect Ray, but you can lose metrics from the cluster.";
+          }
+        });
+  }
 }
 
 }  // namespace stats

src/ray/stats/metric_exporter.h

@@ -117,8 +117,10 @@ class OpenCensusProtoExporter final : public opencensus::stats::StatsExporter::H
  private:
   /// Call Manager for gRPC client.
   rpc::ClientCallManager client_call_manager_;
+  /// Lock to protect the client
+  mutable absl::Mutex mu_;
   /// Client to call a metrics agent gRPC server.
-  std::unique_ptr<rpc::MetricsAgentClient> client_;
+  std::unique_ptr<rpc::MetricsAgentClient> client_ GUARDED_BY(&mu_);
   /// The worker ID of the current component.
   WorkerID worker_id_;
 };

src/ray/stats/stats.h

@@ -125,6 +125,7 @@ static inline void Shutdown() {
   metrics_io_service_pool = nullptr;
   exporter = nullptr;
   StatsConfig::instance().SetIsInitialized(false);
+  RAY_LOG(INFO) << "Stats module has shutdown.";
 }
 
 }  // namespace stats