gcpdiag/runbook/gce/guestos_bootup.py [170:222]:
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
      else:
        op.add_failed(vm,
                      reason=op.prep_msg(op.FAILURE_REASON,
                                         full_resource_path=vm.full_path,
                                         status=vm.status),
                      remediation=op.prep_msg(op.FAILURE_REMEDIATION,
                                              full_resource_path=vm.full_path,
                                              status=vm.status))

    # file sanity checks
    if op.get(flags.SERIAL_CONSOLE_FILE):
      for file in op.get(flags.SERIAL_CONSOLE_FILE).split(','):
        try:
          with open(file, 'rb') as f:
            results = mimetypes.guess_type(file)[0]
            if results and not results.startswith('text/'):
              # Peek at content for further clues
              content_start = f.read(1024)  # Read a small chunk
              # Check for gzip and xz magic number (first two bytes)
              if content_start.startswith(
                  b'\x1f\x8b') or content_start.startswith(b'\xfd'):
                op.add_skipped(
                    vm,
                    reason=('File {} appears to be compressed, not plain text.'
                           ).format(file))
              else:
                # If not gzip or tar, try simple text encoding detection (UTF-8, etc.)
                try:
                  content_start.decode()
                except UnicodeDecodeError:
                  op.add_skipped(
                      vm,
                      reason=('File {} does not appear to be plain text.'
                             ).format(file))

        except FileNotFoundError:
          op.add_skipped(
              vm,
              reason=('The file {} does not exists. Please verify if '
                      'you have provided the correct absolute file path'
                     ).format(file))


class CloudInitChecks(runbook.CompositeStep):
  """Cloud init related checks"""

  def execute(self):
    """Cloud init related checks"""
    ubuntu_licenses = gce.get_gce_public_licences('ubuntu-os-cloud')
    ubuntu_pro_licenses = gce.get_gce_public_licences('ubuntu-os-pro-cloud')
    licenses = ubuntu_licenses + ubuntu_pro_licenses
    vm = gce.get_instance(project_id=op.get(flags.PROJECT_ID),
                          zone=op.get(flags.ZONE),
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -



gcpdiag/runbook/gce/serial_log_analyzer.py [260:312]:
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
      else:
        op.add_failed(vm,
                      reason=op.prep_msg(op.FAILURE_REASON,
                                         full_resource_path=vm.full_path,
                                         status=vm.status),
                      remediation=op.prep_msg(op.FAILURE_REMEDIATION,
                                              full_resource_path=vm.full_path,
                                              status=vm.status))

    # file sanity checks
    if op.get(flags.SERIAL_CONSOLE_FILE):
      for file in op.get(flags.SERIAL_CONSOLE_FILE).split(','):
        try:
          with open(file, 'rb') as f:
            results = mimetypes.guess_type(file)[0]
            if results and not results.startswith('text/'):
              # Peek at content for further clues
              content_start = f.read(1024)  # Read a small chunk
              # Check for gzip and xz magic number (first two bytes)
              if content_start.startswith(
                  b'\x1f\x8b') or content_start.startswith(b'\xfd'):
                op.add_skipped(
                    vm,
                    reason=('File {} appears to be compressed, not plain text.'
                           ).format(file))
              else:
                # If not gzip or tar, try simple text encoding detection (UTF-8, etc.)
                try:
                  content_start.decode()
                except UnicodeDecodeError:
                  op.add_skipped(
                      vm,
                      reason=('File {} does not appear to be plain text.'
                             ).format(file))

        except FileNotFoundError:
          op.add_skipped(
              vm,
              reason=('The file {} does not exists. Please verify if '
                      'you have provided the correct absolute file path'
                     ).format(file))


class CloudInitChecks(runbook.CompositeStep):
  """Cloud init related checks"""

  def execute(self):
    """Cloud init related checks"""
    ubuntu_licenses = gce.get_gce_public_licences('ubuntu-os-cloud')
    ubuntu_pro_licenses = gce.get_gce_public_licences('ubuntu-os-pro-cloud')
    licenses = ubuntu_licenses + ubuntu_pro_licenses
    vm = gce.get_instance(project_id=op.get(flags.PROJECT_ID),
                          zone=op.get(flags.ZONE),
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -