code: def test_log_get(ray_start_cluster):
      cluster = ray_start_cluster
      cluster.add_node(num_cpus=0)
      ray.init(address=cluster.address)
      head_node = list_nodes()[0]
      cluster.add_node(num_cpus=1)
  
      @ray.remote(num_cpus=1)
      class Actor:
          def print(self, i):
              for _ in range(i):
                  print("1")
  
          def getpid(self):
              import os
  
              return os.getpid()
  
      """
      Test filename match
      """
  
      def verify():
          # By default, node id should be configured to the head node.
          for log in get_log(
              node_id=head_node["node_id"],
              filename="raylet.out",
              tail=10,
          ):
              # + 1 since the last line is just empty.
              assert len(log.split("\n")) == 11
          return True
  
      wait_for_condition(verify)
  
      """
      Test worker pid / IP match
      """
      a = Actor.remote()
      pid = ray.get(a.getpid.remote())
      ray.get(a.print.remote(20))
  
      def verify():
          # By default, node id should be configured to the head node.
          for log in get_log(
              node_ip=head_node["node_ip"], pid=pid, tail=10
          ):
              # + 1 since the last line is just empty.
              assert len(log.split("\n")) == 11
          return True
  
      wait_for_condition(verify)
  
      """
      Test actor logs.
      """
      actor_id = a._actor_id.hex()
  
      def verify():
          # By default, node id should be configured to the head node.
          for log in get_log(actor_id=actor_id, tail=10):
              # + 1 since the last line is just empty.
              assert len(log.split("\n")) == 11
          return True
  
      wait_for_condition(verify)
  
      del a
      """
      Test log suffix selection for worker/actor
      """
      ACTOR_LOG_LINE = "{dest}:test actor log"
  
      @ray.remote
      class Actor:
          def __init__(self):
              import sys
  
              print(ACTOR_LOG_LINE.format(dest="out"))
              print(ACTOR_LOG_LINE.format(dest="err"), file=sys.stderr)
  
      actor = Actor.remote()
      actor_id = actor._actor_id.hex()
  
      WORKER_LOG_LINE = "{dest}:test worker log"
  
      @ray.remote
      def worker_func():
          import os
          import sys
  
          print(WORKER_LOG_LINE.format(dest="out"))
          print(WORKER_LOG_LINE.format(dest="err"), file=sys.stderr)
          return os.getpid()
  
      pid = ray.get(worker_func.remote())
  
      def verify():
          # Test actors
          lines = get_log(actor_id=actor_id, suffix="err")
          assert ACTOR_LOG_LINE.format(dest="err") in "".join(lines)
  
          lines = get_log(actor_id=actor_id, suffix="out")
          assert ACTOR_LOG_LINE.format(dest="out") in "".join(lines)
  
          # Default to out
          lines = get_log(actor_id=actor_id)
          assert ACTOR_LOG_LINE.format(dest="out") in "".join(lines)
  
          # Test workers
          lines = get_log(
              node_ip=head_node["node_ip"], pid=pid, suffix="err"
          )
          assert WORKER_LOG_LINE.format(dest="err") in "".join(lines)
  
          lines = get_log(
              node_ip=head_node["node_ip"], pid=pid, suffix="out"
          )
          assert WORKER_LOG_LINE.format(dest="out") in "".join(lines)
  
          lines = get_log(node_ip=head_node["node_ip"], pid=pid)
          assert WORKER_LOG_LINE.format(dest="out") in "".join(lines)
  
          return True
  
      wait_for_condition(verify)
  
      def verify():
          runner = CliRunner()
          result = runner.invoke(
              logs_state_cli_group,
              ["actor", "--id", actor_id],
          )
          assert result.exit_code == 0, result.exception
          assert ACTOR_LOG_LINE.format(dest="out") in result.output
  
          result = runner.invoke(
              logs_state_cli_group,
              [
                  "actor",
                  "--id",
                  actor_id,
                  "--err",
              ],
          )
          assert result.exit_code == 0, result.exception
          assert ACTOR_LOG_LINE.format(dest="err") in result.output
          return True
  
      wait_for_condition(verify)
      ##############################
      # Test binary files and encodings.
      ##############################
      # Write a binary file to ray log directory.
      log_dir = (
          ray._private.worker.global_worker.node.get_logs_dir_path()
      )
      file = "test.bin"
      binary_file = os.path.join(log_dir, file)
      with open(binary_file, "wb") as f:
          data = bytearray(i for i in range(256))
          f.write(data)
  
      # Get the log
      def verify():
          for read in get_log(
              node_ip=head_node["node_ip"], filename=file, encoding=None
          ):
              assert read == data
  
          # Default utf-8
          for read in get_log(
              node_ip=head_node["node_ip"],
              filename=file,
              errors="replace",
          ):
              assert read == data.decode(
                  encoding="utf-8", errors="replace"
              )
  
          for read in get_log(
              node_ip=head_node["node_ip"],
              filename=file,
              encoding="iso-8859-1",
              errors="replace",
          ):
              assert read == data.decode(
                  encoding="iso-8859-1", errors="replace"
              )
  
          return True
  
      wait_for_condition(verify)
  
      # Test running task logs
      @ray.remote
      def sleep_task(out_msg):
          print(out_msg, end="", file=sys.stdout)
          import time
  
          time.sleep(10)
  
      expected_out = "This is a test log from stdout\n"
      task = sleep_task.remote(expected_out)
  
      def verify():
          lines = get_log(task_id=task.task_id().hex())
          assert expected_out == "".join(lines)
  
          return True
  
      wait_for_condition(verify)
  
imports: import json
  import os
  import sys
  import asyncio
  from typing import List
  import urllib
  from unittest.mock import MagicMock, AsyncMock
  import pytest
  from ray.util.state.state_cli import logs_state_cli_group
  from ray.util.state import list_jobs
  import requests
  from click.testing import CliRunner
  import grpc
  from pathlib import Path
  import ray
  from ray._private.test_utils import format_web_url, wait_for_condition, wait_until_server_available
  from ray._private.ray_constants import LOG_PREFIX_TASK_ATTEMPT_START, LOG_PREFIX_TASK_ATTEMPT_END
  from ray._raylet import ActorID, NodeID, TaskID, WorkerID
  from ray.core.generated.common_pb2 import Address
  from ray.core.generated.gcs_service_pb2 import GetTaskEventsReply
  from ray.core.generated.reporter_pb2 import ListLogsReply, StreamLogReply
  from ray.core.generated.gcs_pb2 import ActorTableData, TaskEvents, TaskStateUpdate, TaskLogInfo
  from ray.dashboard.modules.actor.actor_head import actor_table_data_to_dict
  from ray.dashboard.modules.log.log_agent import find_offset_of_content_in_file, find_end_offset_file, find_end_offset_next_n_lines_from_offset, find_start_offset_last_n_lines_from_offset, LogAgentV1Grpc
  from ray.dashboard.modules.log.log_agent import _stream_log_in_chunk
  from ray.dashboard.modules.log.log_manager import LogsManager
  from ray.dashboard.tests.conftest import *
  from ray.util.state import get_log, list_logs, list_nodes, list_workers
  from ray.util.state.common import GetLogOptions
  from ray.util.state.exception import DataSourceUnavailable, RayStateApiException
  from ray.util.state.state_manager import StateDataSourceClient
label_desc: Use image node_id 18 for calling a global function with example usage: test_log_get(ray_start_cluster) and returns: True, True, True, os, True, True, True, True, os