Revert "Revert "[core] add runtime env info to task spec debug string (…

…ray-project#20631)" (ray-project#20668)"

This reverts commit 720bca8.

src/ray/common/task/task_spec.cc

@@ -387,6 +387,23 @@ std::string TaskSpecification::DebugString() const {
            << "}";
   }
 
+  // Print runtime env.
+  if (HasRuntimeEnv()) {
+    const auto &runtime_env_info = RuntimeEnvInfo();
+    stream << ", serialized_runtime_env=" << SerializedRuntimeEnv();
+    const auto &uris = runtime_env_info.uris();
+    if (uris.size() > 0) {
+      stream << ", runtime_env_uris=";
+      for (const auto &uri : uris) {
+        stream << uri << ":";
+      }
+      // Erase the last ":"
+      stream.seekp(-1, std::ios_base::end);
+    }
+    stream << ", runtime_env_eager_install="
+           << runtime_env_info.runtime_env_eager_install();
+  }
+
   return stream.str();
 }
 

src/ray/core_worker/core_worker.cc

@@ -1407,7 +1407,7 @@ std::vector<rpc::ObjectReference> CoreWorker::SubmitTask(
                       task_options.serialized_runtime_env, task_options.runtime_env_uris);
   builder.SetNormalTaskSpec(max_retries, retry_exceptions);
   TaskSpecification task_spec = builder.Build();
-  RAY_LOG(DEBUG) << "Submit task " << task_spec.DebugString();
+  RAY_LOG(DEBUG) << "Submitting normal task " << task_spec.DebugString();
   std::vector<rpc::ObjectReference> returned_refs;
   if (options_.is_local_mode) {
     returned_refs = ExecuteTaskLocalMode(task_spec);
@@ -1493,6 +1493,7 @@ Status CoreWorker::CreateActor(const RayFunction &function,
       << "Actor " << actor_id << " already exists";
   *return_actor_id = actor_id;
   TaskSpecification task_spec = builder.Build();
+  RAY_LOG(DEBUG) << "Submitting actor creation task " << task_spec.DebugString();
   if (options_.is_local_mode) {
     // TODO(suquark): Should we consider namespace in local mode? Currently
     // it looks like two actors with two different namespaces become the
@@ -1671,6 +1672,7 @@ std::vector<rpc::ObjectReference> CoreWorker::SubmitActorTask(
 
   // Submit task.
   TaskSpecification task_spec = builder.Build();
+  RAY_LOG(DEBUG) << "Submitting actor task " << task_spec.DebugString();
   std::vector<rpc::ObjectReference> returned_refs;
   if (options_.is_local_mode) {
     returned_refs = ExecuteTaskLocalMode(task_spec, actor_id);