code: def test_worker_start_end_time(shutdown_only):
      ray.init(num_cpus=1)
  
      @ray.remote
      class Worker:
          def ready(self):
              return os.getpid()
  
      # Test normal exit.
      worker = Worker.remote()
      pid = ray.get(worker.ready.remote())
  
      def verify():
          workers = list_workers(
              detail=True, filters=[("pid", "=", pid)]
          )[0]
          print(workers)
          assert workers["start_time_ms"] > 0
          assert workers["end_time_ms"] == 0
          return True
  
      wait_for_condition(verify)
  
      ray.kill(worker)
  
      def verify():
          workers = list_workers(
              detail=True, filters=[("pid", "=", pid)]
          )[0]
          assert workers["start_time_ms"] > 0
          assert workers["end_time_ms"] > 0
          return True
  
      wait_for_condition(verify)
  
      # Test unexpected exit.
      worker = Worker.remote()
      pid = ray.get(worker.ready.remote())
      os.kill(pid, signal.SIGKILL)
  
      def verify():
          workers = list_workers(
              detail=True, filters=[("pid", "=", pid)]
          )[0]
          assert workers["start_time_ms"] > 0
          assert workers["end_time_ms"] > 0
          return True
  
      wait_for_condition(verify)
  
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 5 for calling a global function with example usage: test_worker_start_end_time(shutdown_only) and returns: True, True, True, os