code: def test_node_start_end_time(ray_start_cluster):
      cluster = ray_start_cluster
      # head
      cluster.add_node(num_cpus=0)
      nodes = list_nodes(detail=True)
      head_node_id = nodes[0]["node_id"]
  
      worker_node = cluster.add_node(num_cpus=0)
      nodes = list_nodes(detail=True)
      worker_node_data = list(
          filter(
              lambda x: x["node_id"] != head_node_id
              and x["state"] == "ALIVE",
              nodes,
          )
      )[0]
      assert worker_node_data["start_time_ms"] > 0
      assert worker_node_data["end_time_ms"] == 0
  
      # Test expected exit.
      cluster.remove_node(worker_node, allow_graceful=True)
      nodes = list_nodes(detail=True)
      worker_node_data = list(
          filter(
              lambda x: x["node_id"] != head_node_id
              and x["state"] == "DEAD",
              nodes,
          )
      )[0]
      assert worker_node_data["start_time_ms"] > 0
      assert worker_node_data["end_time_ms"] > 0
  
      # Test unexpected exit.
      worker_node = cluster.add_node(num_cpus=0)
      nodes = list_nodes(detail=True)
      worker_node_data = list(
          filter(
              lambda x: x["node_id"] != head_node_id
              and x["state"] == "ALIVE",
              nodes,
          )
      )[0]
      assert worker_node_data["start_time_ms"] > 0
      assert worker_node_data["end_time_ms"] == 0
  
      cluster.remove_node(worker_node, allow_graceful=False)
      nodes = list_nodes(detail=True)
      worker_node_data = list(
          filter(
              lambda x: x["node_id"] != head_node_id
              and x["state"] == "DEAD",
              nodes,
          )
      )[0]
      assert worker_node_data["start_time_ms"] > 0
      assert worker_node_data["end_time_ms"] > 0
  
file_path: ray/python/ray/tests/test_exit_observability.py
imports: import asyncio
  import os
  import signal
  import sys
  import pytest
  from ray._private.state_api_test_utils import verify_failed_task
  import ray
  from ray._private.test_utils import run_string_as_driver, wait_for_condition
  from ray.util.state import list_workers, list_nodes, list_tasks
  from ray.util.scheduling_strategies import PlacementGroupSchedulingStrategy
label_desc: Use image node_id 6 for calling a global function with example usage: test_node_start_end_time(ray_start_cluster) without return types