[Core] Adding the node id to the base event

doc/source/ray-observability/user-guides/ray-event-export.rst

@@ -81,6 +81,7 @@ An example of a Task Definition Event and a Task Execution Event:
 
             }
         },
+        "nodeId":"ZvxTI6x9dlMFqMlIHErJpg5UEGK1INsKhW2zyg==",
         "message":""
     }
 
@@ -114,6 +115,7 @@ An example of a Task Definition Event and a Task Execution Event:
             "taskAttempt":0,
             "workerPid":0
         },
+        "nodeId":"ZvxTI6x9dlMFqMlIHErJpg5UEGK1INsKhW2zyg==",
         "message":""
     }
 
@@ -147,6 +149,7 @@ For each actor, Ray exports two types of events: Actor Definition Events and Act
             "placementGroupId": "",
             "labelSelector": {}
         },
+        "nodeId": "zpLG7coqThVMl8df9RYHnhK6thhJqrgPodtfjg==",
         "message": ""
     }
 
@@ -169,6 +172,7 @@ For each actor, Ray exports two types of events: Actor Definition Events and Act
                 }
             ]
         },
+        "nodeId": "zpLG7coqThVMl8df9RYHnhK6thhJqrgPodtfjg==",
         "message": ""
     }
 
@@ -200,6 +204,7 @@ For each driver job, Ray exports two types of events: Driver Job Definition Even
                 "metadata": {}
             }
         },
+        "nodeId": "zpLG7coqThVMl8df9RYHnhK6thhJqrgPodtfjg==",
         "message": ""
     }
 
@@ -220,6 +225,7 @@ For each driver job, Ray exports two types of events: Driver Job Definition Even
                 }
             ]
         },
+        "nodeId": "zpLG7coqThVMl8df9RYHnhK6thhJqrgPodtfjg==",
         "message": ""
     }
 
@@ -249,6 +255,7 @@ For each node, Ray exports two types of events: Node Definition Events and Node
             }, 
             "startTimestamp": "2025-10-24T21:19:14.063Z"
         }, 
+        "nodeId": "zpLG7coqThVMl8df9RYHnhK6thhJqrgPodtfjg==",
         "message": ""
     }
 
@@ -271,6 +278,7 @@ For each node, Ray exports two types of events: Node Definition Events and Node
                 }
             ]
         },
+        "nodeId": "zpLG7coqThVMl8df9RYHnhK6thhJqrgPodtfjg==",
         "message": ""
     }
 

python/ray/dashboard/modules/aggregator/tests/test_ray_actor_events.py

@@ -44,6 +44,7 @@ def test_ray_actor_events(ray_start_cluster, httpserver):
         },
     )
     cluster.wait_for_nodes()
+    head_node_id = cluster.head_node.node_id
     all_nodes_ids = [node.node_id for node in cluster.list_all_nodes()]
 
     class A:
@@ -69,9 +70,11 @@ def ping(self):
         base64.b64decode(req_json[0]["actorDefinitionEvent"]["actorId"]).hex()
         == a._actor_id.hex()
     )
+    assert base64.b64decode(req_json[0]["nodeId"]).hex() == head_node_id
     # Verify ActorId and state for ActorLifecycleEvents
     has_alive_state = False
     for actorLifeCycleEvent in req_json[1:]:
+        assert base64.b64decode(actorLifeCycleEvent["nodeId"]).hex() == head_node_id
         assert (
             base64.b64decode(
                 actorLifeCycleEvent["actorLifecycleEvent"]["actorId"]

python/ray/dashboard/modules/aggregator/tests/test_ray_job_events.py

@@ -46,6 +46,8 @@ def f():
     wait_for_condition(lambda: len(httpserver.log) >= 1)
     req, _ = httpserver.log[0]
     req_json = json.loads(req.data)
+    head_node_id = cluster.head_node.node_id
+    assert base64.b64decode(req_json[0]["nodeId"]).hex() == head_node_id
     assert (
         base64.b64decode(req_json[0]["driverJobDefinitionEvent"]["jobId"]).hex()
         == ray.get_runtime_context().get_job_id()

python/ray/dashboard/modules/aggregator/tests/test_ray_node_events.py

@@ -31,6 +31,7 @@ def test_ray_node_events(ray_start_cluster, httpserver):
         },
     )
     cluster.wait_for_nodes()
+    head_node_id = cluster.head_node.node_id
     ray.init(address=cluster.address)
     wait_for_dashboard_agent_available(cluster)
 
@@ -40,10 +41,12 @@ def test_ray_node_events(ray_start_cluster, httpserver):
     req, _ = httpserver.log[0]
     req_json = json.loads(req.data)
     assert len(req_json) == 2
+    assert base64.b64decode(req_json[0]["nodeId"]).hex() == head_node_id
     assert (
         base64.b64decode(req_json[0]["nodeDefinitionEvent"]["nodeId"]).hex()
         == cluster.head_node.node_id
     )
+    assert base64.b64decode(req_json[1]["nodeId"]).hex() == head_node_id
     assert (
         base64.b64decode(req_json[1]["nodeLifecycleEvent"]["nodeId"]).hex()
         == cluster.head_node.node_id

python/ray/tests/test_ray_event_export_task_events.py

@@ -117,17 +117,25 @@ def get_job_id_and_driver_script_task_id_from_events(
     return driver_script_job_id, driver_task_id
 
 
-def check_task_event_base_fields(event: json, preserve_proto_field_name: bool):
+def check_task_event_base_fields(
+    event: json, preserve_proto_field_name: bool, head_node_id: str = None
+):
     assert event["timestamp"] is not None
     assert event["severity"] == "INFO"
     if preserve_proto_field_name:
         assert event["event_id"] is not None
         assert event["source_type"] == "CORE_WORKER"
         assert event["session_name"] is not None
+        if head_node_id is not None:
+            assert "node_id" in event
+            assert base64.b64decode(event["node_id"]).hex() == head_node_id
     else:
         assert event["eventId"] is not None
         assert event["sourceType"] == "CORE_WORKER"
         assert event["sessionName"] is not None
+        if head_node_id is not None:
+            assert "nodeId" in event
+            assert base64.b64decode(event["nodeId"]).hex() == head_node_id
 
 
 def check_task_lifecycle_event_states_and_error_info(
@@ -223,7 +231,12 @@ def run_driver_script_and_wait_for_events(script, httpserver, cluster, validatio
     # issue to track this: https://github.com/ray-project/ray/issues/58007
     assert wait_until_grpc_channel_ready(cluster.gcs_address, node_ids)
     run_string_as_driver_nonblocking(script)
-    wait_for_condition(lambda: get_and_validate_events(httpserver, validation_func))
+    wait_for_condition(
+        lambda: get_and_validate_events(
+            httpserver,
+            lambda events: validation_func(events, cluster.head_node.node_id),
+        )
+    )
 
 
 class TestNormalTaskEvents:
@@ -244,7 +257,7 @@ def normal_task():
 ray.get(normal_task.remote())
     """
 
-        def validate_events(events):
+        def validate_events(events, head_node_id):
             (
                 driver_script_job_id,
                 driver_task_id,
@@ -267,7 +280,9 @@ def validate_events(events):
             for event in events:
                 if preserve_proto_field_name:
                     if event["event_type"] == "TASK_DEFINITION_EVENT":
-                        check_task_event_base_fields(event, preserve_proto_field_name)
+                        check_task_event_base_fields(
+                            event, preserve_proto_field_name, head_node_id
+                        )
 
                         if event["task_definition_event"]["task_type"] == "DRIVER_TASK":
                             if (
@@ -331,7 +346,9 @@ def validate_events(events):
                         assert event["event_type"] == "TASK_LIFECYCLE_EVENT"
                 else:
                     if event["eventType"] == "TASK_DEFINITION_EVENT":
-                        check_task_event_base_fields(event, preserve_proto_field_name)
+                        check_task_event_base_fields(
+                            event, preserve_proto_field_name, head_node_id
+                        )
 
                         if event["taskDefinitionEvent"]["taskType"] == "DRIVER_TASK":
                             if event["taskDefinitionEvent"]["taskId"] != driver_task_id:
@@ -428,7 +445,7 @@ def normal_task():
     pass
         """
 
-        def validate_events(events: json):
+        def validate_events(events: json, head_node_id):
             (
                 driver_script_job_id,
                 driver_task_id,
@@ -443,7 +460,9 @@ def validate_events(events: json):
             for event in events:
                 if preserve_proto_field_name:
                     if event["event_type"] == "TASK_DEFINITION_EVENT":
-                        check_task_event_base_fields(event, preserve_proto_field_name)
+                        check_task_event_base_fields(
+                            event, preserve_proto_field_name, head_node_id
+                        )
 
                         if event["task_definition_event"]["task_type"] == "DRIVER_TASK":
                             if (
@@ -512,7 +531,9 @@ def validate_events(events: json):
                         assert event["event_type"] == "TASK_LIFECYCLE_EVENT"
                 else:
                     if event["eventType"] == "TASK_DEFINITION_EVENT":
-                        check_task_event_base_fields(event, preserve_proto_field_name)
+                        check_task_event_base_fields(
+                            event, preserve_proto_field_name, head_node_id
+                        )
 
                         if event["taskDefinitionEvent"]["taskType"] == "DRIVER_TASK":
                             if event["taskDefinitionEvent"]["taskId"] != driver_task_id:
@@ -651,7 +672,7 @@ def sleep():
     pass
         """
         # Run the driver script and wait for the sleep task to be executing
-        def validate_task_running(events: json):
+        def validate_task_running(events: json, head_node_id):
             # Obtain the task id of the sleep task
             normal_task_id = None
             for event in events:
@@ -706,7 +727,7 @@ def validate_task_running(events: json):
         cluster.remove_node(node)
 
         # Wait and verify the task events
-        def validate_task_killed(events: json):
+        def validate_task_killed(events: json, head_node_id):
             (
                 driver_script_job_id,
                 driver_task_id,
@@ -720,7 +741,9 @@ def validate_task_killed(events: json):
             for event in events:
                 if preserve_proto_field_name:
                     if event["event_type"] == "TASK_DEFINITION_EVENT":
-                        check_task_event_base_fields(event, preserve_proto_field_name)
+                        check_task_event_base_fields(
+                            event, preserve_proto_field_name, head_node_id
+                        )
 
                         if event["task_definition_event"]["task_type"] == "DRIVER_TASK":
                             if (
@@ -784,7 +807,9 @@ def validate_task_killed(events: json):
                         assert event["event_type"] == "TASK_LIFECYCLE_EVENT"
                 else:
                     if event["eventType"] == "TASK_DEFINITION_EVENT":
-                        check_task_event_base_fields(event, preserve_proto_field_name)
+                        check_task_event_base_fields(
+                            event, preserve_proto_field_name, head_node_id
+                        )
 
                         if event["taskDefinitionEvent"]["taskType"] == "DRIVER_TASK":
                             if event["taskDefinitionEvent"]["taskId"] != driver_task_id:
@@ -906,7 +931,7 @@ def task(self, arg):
 ray.get(actor.task.remote(obj))
         """
 
-        def validate_events(events: json):
+        def validate_events(events: json, head_node_id):
             (
                 driver_script_job_id,
                 driver_task_id,
@@ -920,7 +945,9 @@ def validate_events(events: json):
             for event in events:
                 if preserve_proto_field_name:
                     if event["event_type"] == "TASK_DEFINITION_EVENT":
-                        check_task_event_base_fields(event, preserve_proto_field_name)
+                        check_task_event_base_fields(
+                            event, preserve_proto_field_name, head_node_id
+                        )
 
                         if event["task_definition_event"]["task_type"] == "DRIVER_TASK":
                             driver_task_definition_received = True
@@ -1036,7 +1063,9 @@ def validate_events(events: json):
                         assert event["event_type"] == "TASK_LIFECYCLE_EVENT"
                 else:
                     if event["eventType"] == "TASK_DEFINITION_EVENT":
-                        check_task_event_base_fields(event, preserve_proto_field_name)
+                        check_task_event_base_fields(
+                            event, preserve_proto_field_name, head_node_id
+                        )
 
                         if event["taskDefinitionEvent"]["taskType"] == "DRIVER_TASK":
                             driver_task_definition_received = True
@@ -1208,7 +1237,7 @@ def task(self):
 ray.get(actor.task.options().remote())
         """
 
-        def validate_events(events: json):
+        def validate_events(events: json, head_node_id):
             (
                 driver_script_job_id,
                 driver_task_id,
@@ -1222,7 +1251,9 @@ def validate_events(events: json):
             for event in events:
                 if preserve_proto_field_name:
                     if event["event_type"] == "TASK_DEFINITION_EVENT":
-                        check_task_event_base_fields(event, preserve_proto_field_name)
+                        check_task_event_base_fields(
+                            event, preserve_proto_field_name, head_node_id
+                        )
 
                         if event["task_definition_event"]["task_type"] == "DRIVER_TASK":
                             driver_task_definition_received = True
@@ -1336,7 +1367,9 @@ def validate_events(events: json):
                         assert event["event_type"] == "TASK_LIFECYCLE_EVENT"
                 else:
                     if event["eventType"] == "TASK_DEFINITION_EVENT":
-                        check_task_event_base_fields(event, preserve_proto_field_name)
+                        check_task_event_base_fields(
+                            event, preserve_proto_field_name, head_node_id
+                        )
 
                         if event["taskDefinitionEvent"]["taskType"] == "DRIVER_TASK":
                             driver_task_definition_received = True
@@ -1518,7 +1551,7 @@ def task(self):
 ray.kill(actor)
         """
 
-        def validate_events(events: json):
+        def validate_events(events: json, head_node_id):
             (
                 driver_script_job_id,
                 driver_task_id,
@@ -1531,7 +1564,9 @@ def validate_events(events: json):
             for event in events:
                 if preserve_proto_field_name:
                     if event["event_type"] == "TASK_DEFINITION_EVENT":
-                        check_task_event_base_fields(event, preserve_proto_field_name)
+                        check_task_event_base_fields(
+                            event, preserve_proto_field_name, head_node_id
+                        )
 
                         if event["task_definition_event"]["task_type"] == "DRIVER_TASK":
                             driver_task_definition_received = True
@@ -1597,7 +1632,9 @@ def validate_events(events: json):
                         assert event["event_type"] == "TASK_LIFECYCLE_EVENT"
                 else:
                     if event["eventType"] == "TASK_DEFINITION_EVENT":
-                        check_task_event_base_fields(event, preserve_proto_field_name)
+                        check_task_event_base_fields(
+                            event, preserve_proto_field_name, head_node_id
+                        )
 
                         if event["taskDefinitionEvent"]["taskType"] == "DRIVER_TASK":
                             driver_task_definition_received = True

src/ray/core_worker/core_worker.cc

@@ -434,6 +434,7 @@ CoreWorker::CoreWorker(
           /*timestamp=*/absl::GetCurrentTimeNanos(),
           /*is_actor_task_event=*/false,
           options_.session_name,
+          GetCurrentNodeId(),
           std::make_shared<const TaskSpecification>(std::move(spec)));
       task_event_buffer_->AddTaskEvent(std::move(task_event));
     }
@@ -575,7 +576,8 @@ void CoreWorker::Disconnect(
         rpc::TaskStatus::FINISHED,
         /*timestamp=*/absl::GetCurrentTimeNanos(),
         /*is_actor_task_event=*/worker_context_->GetCurrentActorID().IsNil(),
-        options_.session_name);
+        options_.session_name,
+        GetCurrentNodeId());
     task_event_buffer_->AddTaskEvent(std::move(task_event));
   }
 

src/ray/core_worker/core_worker_process.cc

@@ -164,12 +164,6 @@ std::shared_ptr<CoreWorker> CoreWorkerProcessImpl::CreateCoreWorker(
 #endif
   }
 
-  auto task_event_buffer = std::make_unique<worker::TaskEventBufferImpl>(
-      std::make_unique<gcs::GcsClient>(options.gcs_options, options.node_ip_address),
-      std::make_unique<rpc::EventAggregatorClientImpl>(options.metrics_agent_port,
-                                                       *client_call_manager_),
-      options.session_name);
-
   // Start the IO thread first to make sure the checker is working.
   boost::thread::attributes io_thread_attrs;
 #if defined(__APPLE__)
@@ -227,6 +221,13 @@ std::shared_ptr<CoreWorker> CoreWorkerProcessImpl::CreateCoreWorker(
   }
   RAY_CHECK_GE(assigned_port, 0);
 
+  auto task_event_buffer = std::make_unique<worker::TaskEventBufferImpl>(
+      std::make_unique<gcs::GcsClient>(options.gcs_options, options.node_ip_address),
+      std::make_unique<rpc::EventAggregatorClientImpl>(options.metrics_agent_port,
+                                                       *client_call_manager_),
+      options.session_name,
+      local_node_id);
+
   // Initialize raylet client.
   // NOTE(edoakes): the core_worker_server_ must be running before registering with
   // the raylet, as the raylet will start sending some RPC messages immediately.