code: def get_last_logs(self):
      if not self.job_id:
          raise RuntimeError(
              "Job has not been started, therefore there are no logs to obtain."
          )
  
      if self._last_logs:
          return self._last_logs
  
      # Skip loading logs when the job ran for too long and collected too much logs.
      if self._duration is not None and self._duration > 4 * 3_600:
          return None
  
      def _get_logs():
          job_driver_log, ray_error_log = self._get_ray_logs()
          assert job_driver_log or ray_error_log, "No logs fetched"
          if job_driver_log:
              return job_driver_log
          else:
              return ray_error_log
  
      ret = exponential_backoff_retry(
          _get_logs,
          retry_exceptions=Exception,
          initial_retry_delay_s=30,
          max_retries=3,
      )
      if ret and not self.in_progress:
          self._last_logs = ret
      return ret
  
imports: import os
  import time
  import subprocess
  import tempfile
  from collections import deque
  from contextlib import contextmanager
  from typing import Any, Dict, Optional, Tuple, List
  from anyscale.sdk.anyscale_client.models import CreateProductionJob, HaJobStates
  from ray_release.anyscale_util import LAST_LOGS_LENGTH, get_cluster_name
  from ray_release.cluster_manager.cluster_manager import ClusterManager
  from ray_release.exception import CommandTimeout, JobStartupTimeout, JobStartupFailed
  from ray_release.logger import logger
  from ray_release.signal_handling import register_handler, unregister_handler
  from ray_release.util import ANYSCALE_HOST, ERROR_LOG_PATTERNS, exponential_backoff_retry, anyscale_job_url, format_link
label_desc: Use image node_id 17 for calling the AnyscaleJobManager obj's underlying member method code with example usage: obj.get_last_logs() and returns: ret, self, None, job_driver_log, ray_error_log