use an agent-id rather than the process PID

dashboard/agent.py

@@ -54,6 +54,7 @@ def __init__(
         raylet_name=None,
         logging_params=None,
         disable_metrics_collection: bool = False,
+        agent_id: int = os.getpid(),
     ):
         """Initialize the DashboardAgent object."""
         # Public attributes are accessible for all agent modules.
@@ -76,6 +77,7 @@ def __init__(
         self.logging_params = logging_params
         self.node_id = os.environ["RAY_NODE_ID"]
         self.metrics_collection_disabled = disable_metrics_collection
+        self.agent_id = agent_id
         # TODO(edoakes): RAY_RAYLET_PID isn't properly set on Windows. This is
         # only used for fate-sharing with the raylet and we need a different
         # fate-sharing mechanism for Windows anyways.
@@ -203,7 +205,7 @@ async def _check_parent():
 
         await raylet_stub.RegisterAgent(
             agent_manager_pb2.RegisterAgentRequest(
-                agent_pid=os.getpid(),
+                agent_pid=self.agent_id,
                 agent_port=self.grpc_port,
                 agent_ip_address=self.ip,
             )
@@ -354,6 +356,13 @@ async def _check_parent():
         action="store_true",
         help=("If this arg is set, metrics report won't be enabled from the agent."),
     )
+    parser.add_argument(
+        "--agent-id",
+        required=False,
+        type=int,
+        default=os.getpid(),
+        help="ID to report register with raylet, default is {}.".format(os.getpid()),
+    )
 
     args = parser.parse_args()
     try:
@@ -383,6 +392,7 @@ async def _check_parent():
             raylet_name=args.raylet_name,
             logging_params=logging_params,
             disable_metrics_collection=args.disable_metrics_collection,
+            agent_id=args.agent_id,
         )
         if os.environ.get("_RAY_AGENT_FAILING"):
             raise Exception("Failure injection failure.")

src/ray/raylet/agent_manager.cc

@@ -56,58 +56,65 @@ void AgentManager::StartAgent() {
     return;
   }
 
-  if (RAY_LOG_ENABLED(DEBUG)) {
-    std::stringstream stream;
-    stream << "Starting agent process with command:";
-    for (const auto &arg : options_.agent_commands) {
-      stream << " " << arg;
-    }
-    RAY_LOG(DEBUG) << stream.str();
-  }
-
-  // Launch the process to create the agent.
-  std::error_code ec;
+  // Create a random agent_id to pass to the child process
+  int agent_id = rand();
+  const std::string agent_id_str = std::to_string(agent_id);
   std::vector<const char *> argv;
   for (const std::string &arg : options_.agent_commands) {
     argv.push_back(arg.c_str());
   }
+  argv.push_back("--agent-id");
+  argv.push_back(agent_id_str.c_str());
+
   // Disable metrics report if needed.
   if (!RayConfig::instance().enable_metrics_collection()) {
     argv.push_back("--disable-metrics-collection");
   }
   argv.push_back(NULL);
+
+  if (RAY_LOG_ENABLED(DEBUG)) {
+    std::stringstream stream;
+    stream << "Starting agent process with command:";
+    for (const auto &arg : argv) {
+      stream << " " << arg;
+    }
+    RAY_LOG(DEBUG) << stream.str();
+  }
+
   // Set node id to agent.
   ProcessEnvironment env;
   env.insert({"RAY_NODE_ID", options_.node_id.Hex()});
   env.insert({"RAY_RAYLET_PID", std::to_string(getpid())});
+
+  // Launch the process to create the agent.
+  std::error_code ec;
   Process child(argv.data(), nullptr, ec, false, env);
   if (!child.IsValid() || ec) {
     // The worker failed to start. This is a fatal error.
     RAY_LOG(FATAL) << "Failed to start agent with return value " << ec << ": "
                    << ec.message();
   }
 
-  std::thread monitor_thread([this, child]() mutable {
+  std::thread monitor_thread([this, child, agent_id]() mutable {
     SetThreadName("agent.monitor");
-    RAY_LOG(INFO) << "Monitor agent process with pid " << child.GetId()
-                  << ", register timeout "
+    RAY_LOG(INFO) << "Monitor agent process with id " << agent_id << ", register timeout "
                   << RayConfig::instance().agent_register_timeout_ms() << "ms.";
     auto timer = delay_executor_(
-        [this, child]() mutable {
-          if (agent_pid_ != child.GetId()) {
-            RAY_LOG(WARNING) << "Agent process with pid " << child.GetId()
-                             << " has not registered. ip " << agent_ip_address_
-                             << ", pid " << agent_pid_;
+        [this, child, agent_id]() mutable {
+          if (agent_pid_ != agent_id) {
+            RAY_LOG(WARNING) << "Agent process with id " << agent_id
+                             << " has not registered. ip " << agent_ip_address_ << ", id "
+                             << agent_pid_;
             child.Kill();
           }
         },
         RayConfig::instance().agent_register_timeout_ms());
 
     int exit_code = child.Wait();
     timer->cancel();
-    RAY_LOG(WARNING) << "Agent process with pid " << child.GetId()
-                     << " exit, return value " << exit_code << ". ip "
-                     << agent_ip_address_ << ". pid " << agent_pid_;
+    RAY_LOG(WARNING) << "Agent process with id " << agent_id << " exit, return value "
+                     << exit_code << ". ip " << agent_ip_address_ << ". pid "
+                     << agent_pid_;
     RAY_LOG(ERROR)
         << "The raylet exited immediately because the Ray agent failed. "
            "The raylet fate shares with the agent. This can happen because the "