Allow enabling logging in core worker with empty log_dir (#8529)

python/ray/_raylet.pyx

@@ -667,6 +667,7 @@ cdef class CoreWorker:
         options.raylet_socket = raylet_socket.encode("ascii")
         options.job_id = job_id.native()
         options.gcs_options = gcs_options.native()[0]
+        options.enable_logging = True
         options.log_dir = log_dir.encode("utf-8")
         options.install_failure_signal_handler = True
         options.node_ip_address = node_ip_address.encode("utf-8")

python/ray/includes/libcoreworker.pxd

@@ -193,6 +193,7 @@ cdef extern from "ray/core_worker/core_worker.h" nogil:
         c_string raylet_socket
         CJobID job_id
         CGcsClientOptions gcs_options
+        c_bool enable_logging
         c_string log_dir
         c_bool install_failure_signal_handler
         c_string node_ip_address

src/ray/core_worker/core_worker.cc

@@ -91,9 +91,7 @@ CoreWorkerProcess::CoreWorkerProcess(const CoreWorkerOptions &options)
           options.worker_type == WorkerType::DRIVER
               ? ComputeDriverIdFromJob(options_.job_id)
               : (options_.num_workers == 1 ? WorkerID::FromRandom() : WorkerID::Nil())) {
-  // Initialize logging if log_dir is passed. Otherwise, it must be initialized
-  // and cleaned up by the caller.
-  if (options_.log_dir != "") {
+  if (options_.enable_logging) {
     std::stringstream app_name;
     app_name << LanguageString(options_.language) << "-core-"
              << WorkerTypeString(options_.worker_type);
@@ -104,6 +102,11 @@ CoreWorkerProcess::CoreWorkerProcess(const CoreWorkerOptions &options)
     if (options_.install_failure_signal_handler) {
       RayLog::InstallFailureSignalHandler();
     }
+  } else {
+    RAY_CHECK(options_.log_dir.empty())
+        << "log_dir must be empty because ray log is disabled.";
+    RAY_CHECK(!options_.install_failure_signal_handler)
+        << "install_failure_signal_handler must be false because ray log is disabled.";
   }
 
   RAY_CHECK(options_.num_workers > 0);
@@ -136,7 +139,7 @@ CoreWorkerProcess::~CoreWorkerProcess() {
     absl::ReaderMutexLock lock(&worker_map_mutex_);
     RAY_CHECK(workers_.empty());
   }
-  if (options_.log_dir != "") {
+  if (options_.enable_logging) {
     RayLog::ShutDownRayLog();
   }
 }
@@ -1873,7 +1876,7 @@ void CoreWorker::HandleCancelTask(const rpc::CancelTaskRequest &request,
   if (success && request.force_kill()) {
     RAY_LOG(INFO) << "Force killing a worker running " << main_thread_task_id_;
     RAY_IGNORE_EXPR(local_raylet_client_->Disconnect());
-    if (options_.log_dir != "") {
+    if (options_.enable_logging) {
       RayLog::ShutDownRayLog();
     }
     // NOTE(hchen): Use `_Exit()` to force-exit this process without doing cleanup.
@@ -1912,7 +1915,7 @@ void CoreWorker::HandleKillActor(const rpc::KillActorRequest &request,
              "please create the Java actor with some dynamic options to make it being "
              "hosted in a dedicated worker process.";
     }
-    if (options_.log_dir != "") {
+    if (options_.enable_logging) {
       RayLog::ShutDownRayLog();
     }
     // NOTE(hchen): Use `_Exit()` to force-exit this process without doing cleanup.

src/ray/core_worker/core_worker.h

@@ -76,6 +76,9 @@ struct CoreWorkerOptions {
   JobID job_id;
   /// Options for the GCS client.
   gcs::GcsClientOptions gcs_options;
+  /// Initialize logging if true. Otherwise, it must be initialized and cleaned up by the
+  /// caller.
+  bool enable_logging;
   /// Directory to write logs to. If this is empty, logs won't be written to a file.
   std::string log_dir;
   /// If false, will not call `RayLog::InstallFailureSignalHandler()`.

src/ray/core_worker/lib/java/io_ray_runtime_RayNativeRuntime.cc

@@ -109,6 +109,7 @@ JNIEXPORT void JNICALL Java_io_ray_runtime_RayNativeRuntime_nativeInitialize(
       JavaStringToNativeString(env, rayletSocket),  // raylet_socket
       JavaByteArrayToId<ray::JobID>(env, jobId),    // job_id
       ToGcsClientOptions(env, gcsClientOptions),    // gcs_options
+      true,                                         // enable_logging
       JavaStringToNativeString(env, logDir),        // log_dir
       // TODO (kfstorm): JVM would crash if install_failure_signal_handler was set to true
       false,                                         // install_failure_signal_handler

src/ray/core_worker/test/core_worker_test.cc

@@ -264,6 +264,7 @@ class CoreWorkerTest : public RedisServiceManagerForTest {
           raylet_socket_names_[0],        // raylet_socket
           NextJobId(),                    // job_id
           gcs_options_,                   // gcs_options
+          true,                           // enable_logging
           "",                             // log_dir
           true,                           // install_failure_signal_handler
           "127.0.0.1",                    // node_ip_address

src/ray/core_worker/test/mock_worker.cc

@@ -41,6 +41,7 @@ class MockWorker {
         raylet_socket,       // raylet_socket
         JobID::FromInt(1),   // job_id
         gcs_options,         // gcs_options
+        true,                // enable_logging
         "",                  // log_dir
         true,                // install_failure_signal_handler
         "127.0.0.1",         // node_ip_address

src/ray/util/logging.cc

@@ -209,9 +209,7 @@ void RayLog::UninstallSignalAction() {
 void RayLog::ShutDownRayLog() {
 #ifdef RAY_USE_GLOG
   UninstallSignalAction();
-  if (!log_dir_.empty()) {
-    google::ShutdownGoogleLogging();
-  }
+  google::ShutdownGoogleLogging();
 #endif
 }
 

streaming/src/test/mock_actor.cc

@@ -300,6 +300,7 @@ class StreamingWorker {
         raylet_socket,       // raylet_socket
         JobID::FromInt(1),   // job_id
         gcs_options,         // gcs_options
+        true,                // enable_logging
         "",                  // log_dir
         true,                // install_failure_signal_handler
         "127.0.0.1",         // node_ip_address

streaming/src/test/queue_tests_base.h

@@ -315,6 +315,7 @@ class StreamingQueueTestBase : public ::testing::TestWithParam<uint64_t> {
         raylet_socket_names_[0],        // raylet_socket
         NextJobId(),                    // job_id
         gcs_options_,                   // gcs_options
+        true,                           // enable_logging
         "",                             // log_dir
         true,                           // install_failure_signal_handler
         "127.0.0.1",                    // node_ip_address