def run_rule()

in gcpdiag/lint/gce/bp_2024_001_legacy_monitoring_agent.py [0:0]
61 lines of code
15 McCabe index (conditional complexity)

def run_rule(context: models.Context, report: lint.LintReportRuleInterface):
  instances = list(gce.get_instances(context).values())
  if not instances:
    report.add_skipped(
        None, f'No VM instances found in project: {context.project_id}.')
    return

  instances_without_osinventory = []
  for i in sorted(instances, key=op.attrgetter('project_id', 'full_path')):
    if i.is_gke_node():
      continue
    inventory = osconfig.get_inventory(context, i.zone, i.name)
    if inventory is None:
      instances_without_osinventory.append(i)
      continue
    legacy_agent_found = False
    for pkg_name in inventory.installed_packages:
      if LEGACY_MONITORING_AGENT_PACKAGE_NAME in pkg_name:
        report.add_failed(
            i,
            '',
            LEGACY_AGENT_DETECTED,
        )
        legacy_agent_found = True
        break
    if not legacy_agent_found:
      report.add_ok(i, LEGACY_AGENT_NOT_DETECTED)

  query = _query_results_project_id[context.project_id]
  try:
    vms_agents = {
        e['labels']['resource.instance_id']: e['labels']['metric.version']
        for e in query.values()
    }
  except KeyError:
    for i in instances_without_osinventory:
      report.add_skipped(
          i,
          UNABLE_TO_DETECT_EXPLANATION,
          UNABLE_TO_DETECT,
      )
    return

  for i in sorted(
      instances_without_osinventory,
      key=op.attrgetter('project_id', 'name'),
  ):
    if i.is_gke_node():
      continue
    if i.id in vms_agents:
      if LEGACY_MONITORING_AGENT_METRICS_LABEL in vms_agents[i.id]:
        report.add_failed(
            i,
            '',
            LEGACY_AGENT_DETECTED,
        )
      else:
        report.add_ok(i, LEGACY_AGENT_NOT_DETECTED)
    else:
      report.add_skipped(
          i,
          UNABLE_TO_DETECT_EXPLANATION,
          UNABLE_TO_DETECT,
      )