code: def test_log_redirect_to_stderr(shutdown_only):
      log_components = {
          ray_constants.PROCESS_TYPE_DASHBOARD: "Dashboard head grpc address",
          ray_constants.PROCESS_TYPE_DASHBOARD_AGENT: "",
          ray_constants.PROCESS_TYPE_GCS_SERVER: "Loading job table data",
          # No log monitor output if all components are writing to stderr.
          ray_constants.PROCESS_TYPE_LOG_MONITOR: "",
          ray_constants.PROCESS_TYPE_MONITOR: "Starting monitor using ray installation",
          ray_constants.PROCESS_TYPE_PYTHON_CORE_WORKER: "worker server started",
          ray_constants.PROCESS_TYPE_PYTHON_CORE_WORKER_DRIVER: "driver server started",
          # TODO(Clark): Add coverage for Ray Client.
          # ray_constants.PROCESS_TYPE_RAY_CLIENT_SERVER: "Starting Ray Client server",
          ray_constants.PROCESS_TYPE_RAY_CLIENT_SERVER: "",
          ray_constants.PROCESS_TYPE_RAYLET: "Starting object store with directory",
          # No reaper process run (kernel fate-sharing).
          ray_constants.PROCESS_TYPE_REAPER: "",
          # No reporter process run.
          ray_constants.PROCESS_TYPE_REPORTER: "",
          # No web UI process run.
          ray_constants.PROCESS_TYPE_WEB_UI: "",
          # Unused.
          ray_constants.PROCESS_TYPE_WORKER: "",
      }
  
      script = """
  import os
  from pathlib import Path
  
  import ray
  
  os.environ["RAY_LOG_TO_STDERR"] = "1"
  ray.init()
  
  session_dir = ray._private.worker.global_worker.node.address_info["session_dir"]
  session_path = Path(session_dir)
  log_dir_path = session_path / "logs"
  
  # Run the basic workload.
  @ray.remote
  def f():
      for i in range(10):
          print(f"test {{i}}")
  
  ray.get(f.remote())
  
  log_component_names = {}
  
  # Confirm that no log files are created for any of the components.
  paths = list(path.stem for path in log_dir_path.iterdir())
  assert set(log_component_names).isdisjoint(set(paths)), paths
  """.format(
          str(list(log_components.keys()))
      )
      stderr = run_string_as_driver(script)
  
      # Make sure that the expected startup log records for each of the
      # components appears in the stderr stream.
      for component, canonical_record in log_components.items():
          if not canonical_record:
              # Process not run or doesn't generate logs; skip.
              continue
          assert canonical_record in stderr, stderr
          if component == ray_constants.PROCESS_TYPE_REDIS_SERVER:
              # Redis doesn't expose hooks for custom log formats, so we aren't able to
              # inject the Redis server component name into the log records.
              continue
          # NOTE: We do a prefix match instead of including the enclosing right
          # parentheses since some components, like the core driver and worker, add a
          # unique ID suffix.
          assert f"({component}" in stderr, stderr
  
imports: import io
  import os
  import re
  import subprocess
  import sys
  import tempfile
  import time
  import logging
  from collections import Counter, defaultdict
  from contextlib import redirect_stderr, redirect_stdout
  from pathlib import Path
  from unittest.mock import Mock, MagicMock
  import pytest
  import ray
  from ray._private import ray_constants
  from ray._private.log_monitor import LOG_NAME_UPDATE_INTERVAL_S, RAY_LOG_MONITOR_MANY_FILES_THRESHOLD, LogFileInfo, LogMonitor, is_proc_alive
  from ray._private.test_utils import get_log_batch, get_log_message, get_log_data, init_log_pubsub, run_string_as_driver, wait_for_condition
  from ray.cross_language import java_actor_class
  from ray.autoscaler._private.cli_logger import cli_logger
label_desc: Use image node_id 11 for calling a global function with example usage: test_log_redirect_to_stderr(shutdown_only) without return types