def run_rule()

in gcpdiag/lint/composer/warn_2022_002_fluentd_pod_crashloop.py [0:0]

• 33 lines of code
• 17 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

  envs = envs_by_project[context.project_id]
  name_to_env = {env.name: env for env in envs}

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

  stuck_in_crashloop_envs = []

  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'] != 'INFO' or \
         POD_NAME != get_path(log_entry,
                     ('labels', 'k8s-pod/name'), default='') or \
         MATCH_STR not in log_entry.get('textPayload', '') or \
         MATCH_STR2 not in log_entry.get('textPayload', ''):
        continue

      # region-name-suffix-gke format (us-east4-composer-stg-v2-a106130c-gke)
      cluster_name = get_path(log_entry, ('resource', 'labels', 'cluster_name'),
                              default='')

      env_name = '-'.join(cluster_name.split('-')[2:-2])
      if env_name and env_name not in stuck_in_crashloop_envs:
        stuck_in_crashloop_envs.append(env_name)

  for env_name in stuck_in_crashloop_envs:
    report.add_failed(name_to_env[env_name],
                      'has fluentd pods stuck in the crashloop')

  for env_name in [
      env_name for env_name in name_to_env
      if env_name not in stuck_in_crashloop_envs
  ]:
    report.add_ok(name_to_env[env_name])