code: def test_spilling(object_spilling_config, shutdown_only):
      """Test metrics with object spilling occurred"""
  
      object_spilling_config, _ = object_spilling_config
      delta = 5
      info = ray.init(
          num_cpus=1,
          object_store_memory=100 * MiB + delta * MiB,
          _system_config={
              **_SYSTEM_CONFIG,
              **{"object_spilling_config": object_spilling_config},
          },
      )
  
      # Create and use 100MiB data, which should fit in memory
      objs1 = [
          ray.put(np.zeros(50 * MiB, dtype=np.uint8)) for _ in range(2)
      ]
  
      expected = {
          "MMAP_SHM": 100 * MiB,
          "MMAP_DISK": 0,
          "SPILLED": 0,
          "WORKER_HEAP": 0,
      }
  
      wait_for_condition(
          # 1KiB for metadata difference
          lambda: approx_eq_dict_in(
              objects_by_loc(info), expected, 1 * KiB
          ),
          timeout=20,
          retry_interval_ms=500,
      )
  
      # Create additional 100MiB, so that it needs to be triggered
      objs2 = [
          ray.put(np.zeros(50 * MiB, dtype=np.uint8)) for _ in range(2)
      ]
  
      expected = {
          "WORKER_HEAP": 0,
          "MMAP_SHM": 100 * MiB,
          "MMAP_DISK": 0,
          "SPILLED": 100 * MiB,
      }
      wait_for_condition(
          # 1KiB for metadata difference
          lambda: approx_eq_dict_in(
              objects_by_loc(info), expected, 1 * KiB
          ),
          timeout=20,
          retry_interval_ms=500,
      )
  
      # Delete spilled objects
      del objs1
      expected = {
          "MMAP_SHM": 100 * MiB,
          "MMAP_DISK": 0,
          "SPILLED": 0,
          "WORKER_HEAP": 0,
      }
      wait_for_condition(
          # 1KiB for metadata difference
          lambda: approx_eq_dict_in(
              objects_by_loc(info), expected, 1 * KiB
          ),
          timeout=20,
          retry_interval_ms=500,
      )
  
      # Delete all
      del objs2
      expected = {
          "MMAP_SHM": 0,
          "MMAP_DISK": 0,
          "SPILLED": 0,
          "WORKER_HEAP": 0,
      }
      wait_for_condition(
          # 1KiB for metadata difference
          lambda: approx_eq_dict_in(
              objects_by_loc(info), expected, 1 * KiB
          ),
          timeout=20,
          retry_interval_ms=500,
      )
  
file_path: ray/python/ray/tests/test_object_store_metrics.py
imports: from collections import defaultdict
  import pytest
  from typing import Dict
  import numpy
  import sys
  import requests
  import ray
  from ray._private.test_utils import raw_metrics, wait_for_condition
  from ray._private.worker import RayContext
label_desc: Use image node_id 6 for calling a global function with example usage: test_spilling(object_spilling_config, shutdown_only) without return types