code: def get_log(
      address: Optional[str] = None,
      node_id: Optional[str] = None,
      node_ip: Optional[str] = None,
      filename: Optional[str] = None,
      actor_id: Optional[str] = None,
      task_id: Optional[str] = None,
      pid: Optional[int] = None,
      follow: bool = False,
      tail: int = -1,
      timeout: int = DEFAULT_RPC_TIMEOUT,
      suffix: str = "out",
      encoding: Optional[str] = "utf-8",
      errors: Optional[str] = "strict",
      submission_id: Optional[str] = None,
      attempt_number: int = 0,
      _interval: Optional[float] = None,
  ) -> Generator[str, None, None]:
      """Retrieve log file based on file name or some entities ids (pid, actor id, task id).
  
      Examples:
          .. testcode::
              :hide:
  
              import ray
              import time
  
              ray.shutdown()
              ray.init()
  
              # Wait for the node to be registered to the dashboard
              time.sleep(5)
  
          .. testcode::
  
              import ray
              from ray.util.state import get_log
  
              # Node id could be retrieved from list_nodes() or ray.nodes()
              node_id = ray.nodes()[0]["NodeID"]
              filename = "raylet.out"
              for l in get_log(filename=filename, node_id=node_id):
                 print(l)
  
          .. testoutput::
              :options: +MOCK
  
              [2023-05-19 12:35:18,347 I 4259 68399276] (raylet) io_service_pool.cc:35: IOServicePool is running with 1 io_service.
              [2023-05-19 12:35:18,348 I 4259 68399276] (raylet) store_runner.cc:32: Allowing the Plasma store to use up to 2.14748GB of memory.
              [2023-05-19 12:35:18,348 I 4259 68399276] (raylet) store_runner.cc:48: Starting object store with directory /tmp, fallback /tmp/ray, and huge page support disabled
  
      Args:
          address: Ray bootstrap address, could be `auto`, `localhost:6379`.
              If not specified, it will be retrieved from the initialized ray cluster.
          node_id: Id of the node containing the logs .
          node_ip: Ip of the node containing the logs. (At least one of the node_id and
              node_ip have to be supplied when identifying a node).
          filename: Name of the file (relative to the ray log directory) to be retrieved.
          actor_id: Id of the actor if getting logs from an actor.
          task_id: Id of the task if getting logs from a non concurrent actor.
              For concurrent actor, please query the log with actor_id.
          pid: PID of the worker if getting logs generated by a worker. When querying
              with pid, either node_id or node_ip must be supplied.
          follow: When set to True, logs will be streamed and followed.
          tail: Number of lines to get from the end of the log file. Set to -1 for getting
              the entire log.
          timeout: Max timeout for requests made when getting the logs.
          suffix: The suffix of the log file if query by id of tasks/workers/actors. Default to "out".
          encoding: The encoding used to decode the content of the log file. Default is
              "utf-8". Use None to get binary data directly.
          errors: The error handling scheme to use for decoding errors. Default is
              "strict". See https://docs.python.org/3/library/codecs.html#error-handlers
          submission_id: Job submission ID if getting log from a submission job.
          attempt_number: The attempt number of the task if getting logs generated by a task.
          _interval: The interval in secs to print new logs when `follow=True`.
  
      Return:
          A Generator of log line, None for SendType and ReturnType.
  
      Raises:
          Exceptions: :class:`RayStateApiException <ray.util.state.exception.RayStateApiException>` if the CLI
              failed to query the data.
      """  # noqa: E501
  
      api_server_url = ray_address_to_api_server_url(address)
      media_type = "stream" if follow else "file"
  
      options = GetLogOptions(
          node_id=node_id,
          node_ip=node_ip,
          filename=filename,
          actor_id=actor_id,
          task_id=task_id,
          pid=pid,
          lines=tail,
          interval=_interval,
          media_type=media_type,
          timeout=timeout,
          suffix=suffix,
          submission_id=submission_id,
          attempt_number=attempt_number,
      )
      options_dict = {}
      for field in fields(options):
          option_val = getattr(options, field.name)
          if option_val is not None:
              options_dict[field.name] = option_val
  
      with requests.get(
          f"{api_server_url}/api/v0/logs/{media_type}?"
          f"{urllib.parse.urlencode(options_dict)}",
          stream=True,
      ) as r:
          if r.status_code != 200:
              raise RayStateApiException(r.text)
          for bytes in r.iter_content(chunk_size=None):
              bytes = bytearray(bytes)
              # First byte 1 means success.
              if bytes.startswith(b"1"):
                  bytes.pop(0)
                  logs = bytes
                  if encoding is not None:
                      logs = bytes.decode(
                          encoding=encoding, errors=errors
                      )
              else:
                  assert bytes.startswith(b"0")
                  error_msg = bytes.decode("utf-8")
                  raise RayStateApiException(error_msg)
              yield logs
  
imports: import logging
  import threading
  import urllib
  import warnings
  from contextlib import contextmanager
  from dataclasses import fields
  from typing import Any, Dict, Generator, List, Optional, Tuple, Union
  import requests
  from ray.dashboard.modules.dashboard_sdk import SubmissionClient
  from ray.dashboard.utils import get_address_for_submission_client, ray_address_to_api_server_url
  from ray.util.annotations import DeveloperAPI
  from ray.util.state.common import DEFAULT_LIMIT, DEFAULT_RPC_TIMEOUT, ActorState, ClusterEventState, GetApiOptions, GetLogOptions, JobState, ListApiOptions, NodeState, ObjectState, PlacementGroupState, PredicateType, RuntimeEnvState, StateResource, SummaryApiOptions, SummaryResource, SupportedFilterType, TaskState, WorkerState, dict_to_state
  from ray.util.state.exception import RayStateApiException, ServerUnavailable
label_desc: Use image node_id 18 for calling a global function with example usage: get_log(address, node_id, node_ip, filename, actor_id, task_id, pid, follow, tail, timeout, suffix, encoding, errors, submission_id, attempt_number, _interval) without return types