Generate raylet ID from node ip address (IPv4).

src/ray/common/id.cc

@@ -266,4 +266,40 @@ ID_OSTREAM_OPERATOR(ActorID);
 ID_OSTREAM_OPERATOR(TaskID);
 ID_OSTREAM_OPERATOR(ObjectID);
 ID_OSTREAM_OPERATOR(PlacementGroupID);
+
+NodeID GenerateNodeIdFromIpAddress(const std::string &ip_address) {
+  // check IP version 4 address.
+  boost::system::error_code err;
+  boost::asio::ip::address_v4 ip_v4 = boost::asio::ip::make_address_v4(ip_address, err);
+  if (err) {
+    RAY_LOG(WARNING) << "Inavlid IP address: " << ip_address
+                     << ", error message: " << err.message()
+                     << ". Falling back to a random NodeID.";
+    return NodeID::FromRandom();
+  }
+
+  // Format the IP version 4 address to a 12-digits hex string, and add 1 for the
+  // null terminator.
+  char ip_buffer[13];
+  int split_bit = 8;
+  int mask = (1 << split_bit) - 1;
+  unsigned int uint_ip_v4 = ip_v4.to_uint();
+  sprintf(ip_buffer, "%03d%03d%03d%03d", (uint_ip_v4 >> 3 * split_bit) & mask,
+          (uint_ip_v4 >> 2 * split_bit) & mask, (uint_ip_v4 >> 1 * split_bit) & mask,
+          (uint_ip_v4 >> 0 * split_bit) & mask);
+
+  // Convert hex-format `ip_buffer` to binary.
+  std::string ip_prefix(6, 0);
+  for (unsigned int index = 0; index < ip_prefix.size(); index++) {
+    int number = ((ip_buffer[index * 2] - '0') << 4) + (ip_buffer[index * 2 + 1] - '0');
+    ip_prefix[index] = static_cast<uint8_t>(number);
+  }
+
+  // Filling the remaining bits with random data.
+  std::string random_suffix(NodeID::Size() - ip_prefix.size(), 0);
+  FillRandom(&random_suffix);
+
+  return NodeID::FromBinary(ip_prefix + random_suffix);
+}
+
 }  // namespace ray

src/ray/common/id.h

@@ -16,6 +16,8 @@
 
 #include <inttypes.h>
 #include <limits.h>
+#include <boost/asio/ip/address_v4.hpp>
+#include <boost/system/error_code.hpp>
 
 #include <chrono>
 #include <cstring>
@@ -496,6 +498,18 @@ std::string BaseID<T>::Hex() const {
   return result;
 }
 
+/// Generate a NodeID from an IP address (IPv4).
+/// The IP address will be filled to the first 12 characters of the generated NodeID's
+/// hexadecimal representation, and the remaining will be padded with random binary data.
+/// For example, if the input IP address is `1.10.0.122`, the output will be
+/// `0010100001226f7a6e09a0f51136c6243b502bcd2fac0cf21e05e874`.
+///
+/// \param ip_address The IP address.
+///
+/// \return NodeID, in its hexadecimal representation, the first 12 digits represent the
+/// IP address.
+NodeID GenerateNodeIdFromIpAddress(const std::string &ip_address);
+
 }  // namespace ray
 
 namespace std {

src/ray/common/id_test.cc

@@ -14,8 +14,8 @@
 
 #include "gtest/gtest.h"
 #include "ray/common/common_protocol.h"
+#include "ray/common/network_util.h"
 #include "ray/common/task/task_spec.h"
-
 namespace ray {
 
 void TestFromIndexObjectId(const TaskID &task_id, int64_t index) {
@@ -99,6 +99,35 @@ TEST(HashTest, TestNilHash) {
   ASSERT_NE(id1.Hash(), id2.Hash());
 }
 
+TEST(PlacementGroupIDTest, TestPlacementGroup) {
+  {
+    // test from binary
+    PlacementGroupID placement_group_id_1 = PlacementGroupID::Of(JobID::FromInt(1));
+    const auto placement_group_id_1_binary = placement_group_id_1.Binary();
+    const auto placement_group_id_2 =
+        PlacementGroupID::FromBinary(placement_group_id_1_binary);
+    ASSERT_EQ(placement_group_id_1, placement_group_id_2);
+    const auto placement_group_id_1_hex = placement_group_id_1.Hex();
+    const auto placement_group_id_3 = PlacementGroupID::FromHex(placement_group_id_1_hex);
+    ASSERT_EQ(placement_group_id_1, placement_group_id_3);
+  }
+
+  {
+    // test get job id
+    auto job_id = JobID::FromInt(1);
+    const PlacementGroupID placement_group_id = PlacementGroupID::Of(job_id);
+    ASSERT_EQ(job_id, placement_group_id.JobId());
+  }
+}
+
+TEST(NodeIDTest, TestGenerateNodeIdFromIpAddress) {
+  {
+    std::string ip = std::string("1.10.0.122");
+    const NodeID node_id = GenerateNodeIdFromIpAddress(ip);
+    ASSERT_EQ("001010000122", node_id.Hex().substr(0, 12));
+  }
+}
+
 }  // namespace ray
 
 int main(int argc, char **argv) {

src/ray/raylet/raylet.cc

@@ -61,7 +61,7 @@ Raylet::Raylet(instrumented_io_context &main_service, const std::string &socket_
                const ObjectManagerConfig &object_manager_config,
                std::shared_ptr<gcs::GcsClient> gcs_client, int metrics_export_port)
     : main_service_(main_service),
-      self_node_id_(NodeID::FromRandom()),
+      self_node_id_(GenerateNodeIdFromIpAddress(node_ip_address)),
       gcs_client_(gcs_client),
       object_directory_(
           RayConfig::instance().ownership_based_object_directory_enabled()