code: def test_large_log_file(log_file_size_byte: int):
      if log_file_size_byte == 0:
          logger.info("Skipping test with 0 log file size")
          return
  
      import sys
      import string
      import random
      import hashlib
  
      @ray.remote
      class LogActor:
          def write_log(self, log_file_size_byte: int):
              ctx = hashlib.sha1()
              job_id = ray.get_runtime_context().get_job_id()
              prefix = f"{LOG_PREFIX_JOB_ID}{job_id}\n{LOG_PREFIX_ACTOR_NAME}LogActor\n"
              ctx.update(prefix.encode())
              while log_file_size_byte > 0:
                  n = min(log_file_size_byte, 4 * MiB)
                  chunk = "".join(
                      random.choices(string.ascii_letters, k=n)
                  )
                  sys.stdout.writelines([chunk])
                  ctx.update(chunk.encode())
                  log_file_size_byte -= n
  
              sys.stdout.flush()
              return (
                  ctx.hexdigest(),
                  ray.get_runtime_context().get_node_id(),
              )
  
      actor = LogActor.remote()
  
      task = actor.write_log.remote(
          log_file_size_byte=log_file_size_byte
      )
      expected_hash, node_id = ray.get(task)
      assert (
          expected_hash is not None
      ), "Empty checksum from the log actor"
      assert node_id is not None, "Empty node id from the log actor"
  
      # Retrieve the log and compare the checksum
      ctx = hashlib.sha1()
  
      time_taken = 0
      t_start = time.perf_counter()
      for s in get_log(actor_id=actor._actor_id.hex(), tail=1000000000):
          t_end = time.perf_counter()
          time_taken += t_end - t_start
          # Not including this time
          ctx.update(s.encode())
          # Only time the iterator's performance
          t_start = time.perf_counter()
  
      assert expected_hash == ctx.hexdigest(), "Mismatch log file"
  
      metric = StateAPIMetric(time_taken, log_file_size_byte)
      GLOBAL_STATE_STATS.calls["get_log"].append(metric)
  
imports: import click
  import json
  import ray
  from ray._private.ray_constants import LOG_PREFIX_ACTOR_NAME, LOG_PREFIX_JOB_ID
  from ray._private.state_api_test_utils import STATE_LIST_LIMIT, StateAPIMetric, aggregate_perf_results, invoke_state_api, GLOBAL_STATE_STATS
  import ray._private.test_utils
  import tqdm
  import time
  import os
  from ray.util.placement_group import placement_group, remove_placement_group
  from ray.util.scheduling_strategies import PlacementGroupSchedulingStrategy
  from ray.util.state import get_log, list_actors, list_objects, list_tasks
  import logging
label_desc: Use image node_id 5 for calling a global function with example usage: test_large_log_file(log_file_size_byte) and returns: ctx,  ray