code: def test_node_object_metrics(ray_start_cluster):
      NUM_NODES = 3
      cluster = ray_start_cluster
      for i in range(NUM_NODES):
          cluster.add_node(True, resources={f"node_{i}": 1})
          if i == 0:
              ray.init(address=cluster.address)
      node_ids = []
  
      for i in range(NUM_NODES):
  
          @ray.remote(resources={f"node_{i}": 1})
          def get_node_id():
              return ray.get_runtime_context().get_node_id()
  
          node_ids.append(ray.get(get_node_id.remote()))
  
      # Object store stats
      # x is owned by node_0
      # x is stored at node_0
      x = ray.put([1])  # noqa: F841
      wait_for_condition(
          lambda: get_owner_info(node_ids)
          == ([(1, 0), (0, 0), (0, 0)], [1, 0, 0])
      )
  
      # Test nested with put
      @ray.remote(resources={"node_1": 1})
      def big_obj():
          # b is owned by node_1
          # b is stored at node_1
          b = ray.put([1] * 1024 * 1024 * 10)
          return b
  
      # Object store stats
      # big_obj is owned by node_0
      # big_obj is stored in memory (no primary copy)
      big_obj_ref = big_obj.remote()  # noqa: F841
      wait_for_condition(
          lambda: get_owner_info(node_ids)
          == ([(2, 0), (1, 0), (0, 0)], [1, 1, 0])
      )
  
      # Test nested with task (small output)
      @ray.remote(resources={"node_1": 1})
      def nest_task(s):
          @ray.remote(resources={"node_2": 1})
          def task():
              return [1] * s
  
          # t is owned by node_1
          # if s is small,
          # then it's is stored in memory of node_1 (no primary copy)
          # else it's stored in object store of node_1
          t = task.remote()
          return t
  
      # nest_ref is owned by node_0
      # nest_ref is stored in memory (no primary copy)
      nest_ref = nest_task.remote(1)  # noqa: F841
      wait_for_condition(
          lambda: get_owner_info(node_ids)
          == ([(3, 0), (2, 0), (0, 0)], [1, 1, 0])
      )
  
      big_nest = nest_task.remote(1024 * 1024 * 10)  # noqa: F841
  
      wait_for_condition(
          lambda: get_owner_info(node_ids)
          == ([(4, 0), (3, 0), (0, 0)], [1, 1, 1])
      )
  
      # Test with assigned owned
      @ray.remote(resources={"node_2": 0.5}, num_cpus=0)
      class A:
          def ready(self):
              return
  
          def gen(self):
              return ray.put(10)
  
      # actor is owned by node_0
      # actor is not an object, so no object store copies
      actor = A.remote()  # noqa: F841
      ray.get(actor.ready.remote())
      # o is owned by actor (node_2)
      # o is stored in object store of node_0
      o = ray.put(1, _owner=actor)  # noqa: F841
      wait_for_condition(
          lambda: get_owner_info(node_ids)
          == ([(4, 1), (3, 0), (1, 0)], [2, 1, 1])
      )
  
      # Test with detached owned
      # detached actor is owned by GCS. So it's not counted in the owner stats
      detached_actor = A.options(lifetime="detached", name="A").remote()
      ray.get(detached_actor.ready.remote())
      for i in range(3):
          assert get_owner_info(node_ids) == (
              [(4, 1), (3, 0), (1, 0)],
              [2, 1, 1],
          )
          import time
  
          time.sleep(1)
      # gen_obj is owned by node_0
      # the inner object is owned by A (node_2)
      # the inner object is stored in object store of node_2
      gen_obj = detached_actor.gen.remote()  # noqa: F841
      wait_for_condition(
          lambda: get_owner_info(node_ids)
          == ([(5, 1), (3, 0), (2, 0)], [2, 1, 2])
      )
  
file_path: ray/python/ray/tests/test_metrics.py
imports: import os
  import platform
  import psutil
  import pytest
  import requests
  import ray
  from ray._private.test_utils import wait_for_condition, wait_until_succeeded_without_exception, get_node_stats
  from ray.core.generated import common_pb2
label_desc: Use image node_id 3 for calling a global function with example usage: test_node_object_metrics(ray_start_cluster) and returns: b, t, ray, unknown, ray