code: def get_owner_info(node_ids):
      node_addrs = {
          n["NodeID"]: (n["NodeManagerAddress"], n["NodeManagerPort"])
          for n in ray.nodes()
      }
      # Force a global gc to clean up the object store.
      ray._private.internal_api.global_gc()
      owner_stats = {n: 0 for n in node_ids}
      primary_copy_stats = {n: 0 for n in node_ids}
  
      for node_id in node_ids:
          node_stats = ray._private.internal_api.node_stats(
              node_addrs[node_id][0], node_addrs[node_id][1], False
          )
          num_owned_objects = sum(
              [
                  stats.num_owned_objects
                  for stats in node_stats.core_workers_stats
              ]
          )
          num_owned_actors = sum(
              [
                  stats.num_owned_actors
                  for stats in node_stats.core_workers_stats
              ]
          )
          owner_stats[node_id] = (num_owned_objects, num_owned_actors)
          primary_copy_stats[
              node_id
          ] = node_stats.store_stats.num_object_store_primary_copies
  
      print(owner_stats)
      print(node_ids)
      owner_stats = [
          owner_stats.get(node_id, 0) for node_id in node_ids
      ]
      primary_copy_stats = [
          primary_copy_stats.get(node_id, 0) for node_id in node_ids
      ]
      print("owner_stats", owner_stats)
      print("primary_copy_stats", primary_copy_stats)
  
      return owner_stats, primary_copy_stats
  
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 2 for calling a global function with example usage: get_owner_info(node_ids) and returns: owner_stats,  primary_copy_stats