def run_rule()

in gcpdiag/lint/dataproc/warn_2022_001_cluster_local_ssd_failed_stop.py [0:0]

• 33 lines of code
• 16 McCabe index (conditional complexity)

def run_rule(context: models.Context, report: lint.LintReportRuleInterface):
  project = crm.get_project(context.project_id)

  # skip entire rule is logging disabled
  if not apis.is_enabled(context.project_id, 'logging'):
    report.add_skipped(project, 'logging api is disabled')
    return

  clusters = dataproc.get_clusters(context)
  name_to_cluster = {cluster.name: cluster for cluster in clusters}

  if not clusters:
    report.add_skipped(project, 'no clusters found')
    return

  failed_to_stop_clusters = []

  if logs_by_project.get(context.project_id) and \
     logs_by_project[context.project_id].entries:
    for log_entry in logs_by_project[context.project_id].entries:
      # Filter out non-relevant log entries.
      if log_entry['severity'] != 'ERROR' or \
         METHOD_NAME not in get_path(log_entry,
                     ('protoPayload', 'methodName'), default='') or \
         MATCH_STR not in get_path(log_entry,
                     ('protoPayload', 'status', 'message'), default=''):
        continue

      cluster_name = get_path(log_entry,
                              ('protoPayload', 'request', 'clusterName'),
                              default='')
      if cluster_name and cluster_name not in failed_to_stop_clusters:
        failed_to_stop_clusters.append(cluster_name)

  for cluster_name in failed_to_stop_clusters:
    report.add_failed(name_to_cluster[cluster_name],
                      'failed to stop due to local SSDs')
  for cluster_name in [
      cluster_name for cluster_name in name_to_cluster
      if cluster_name not in failed_to_stop_clusters
  ]:
    report.add_ok(name_to_cluster[cluster_name])