def execute()

in gcpdiag/runbook/gke/node_bootstrapping.py [0:0]

• 58 lines of code
• 13 McCabe index (conditional complexity)

  def execute(self):
    """
    Check the provided parameters.
    """
    project = op.get(flags.PROJECT_ID)
    location = op.get(flags.LOCATION)
    node = op.get(flags.NODE)
    name = op.get(flags.NAME)
    project_path = crm.get_project(project)
    start_time = op.get(flags.START_TIME)
    end_time = op.get(flags.END_TIME)

    # check if there are clusters in the project
    if name:
      clusters = gke.get_clusters(
          op.get_new_context(project_id=project, resources=[name]))
      if not clusters:
        op.add_skipped(
            project_path,
            reason=f'No {name} GKE cluster found in project {project}')
        return
    else:
      clusters = gke.get_clusters(
          op.get_new_context(project_id=op.get(flags.PROJECT_ID)))
      if not clusters:
        op.add_skipped(
            project_path,
            reason=('No GKE clusters found in project {}').format(project))
        return

    # check if node exists
    if node:
      node_vm = get_node_instance(project, location, node)

      if node_vm:
        if not node_vm.is_gke_node():
          op.add_skipped(
              project_path,
              reason=
              (f'Instance {node} in location {location} does not appear to be a GKE node'
              ))
          return
        elif not node_vm.is_serial_port_logging_enabled():
          op.add_skipped(
              project_path,
              reason=
              (f'Instance {node} in location {location} does not have Serial Logs enabled, please '
               'enable serial logs for easier troubleshooting.'))
          return

      # fail if Audit Log does not have any log entries for the input provided, meaning there could
      # have been an input error
      filter_str = [
          'log_id("cloudaudit.googleapis.com/activity")',
          f'resource.labels.zone="{location}"',
          f'protoPayload.resourceName:"{node}"'
      ]
      filter_str = '\n'.join(filter_str)

      log_entries = local_realtime_query(filter_str)

      if not log_entries:
        op.add_skipped(
            project_path,
            reason=
            (f'There are no log entries for the provided node {node} and location '
             f'{location} in the provided time range '
             f'{start_time} - {end_time}.\n'
             'Please make sure the node/location pair is correct and it was booted '
             'in the time range provided, then try again this runbook.'))
        return