def run_health_app()

in src/health_runner/health_runner.py [0:0]

• 29 lines of code
• 5 McCabe index (conditional complexity)

def run_health_app(health_app: str) -> None:
  """Run the health check."""
  health_results = health_results_pb2.HealthResults(
      created_date_time=timestamp_pb2.Timestamp().GetCurrentTime(),
  )
  orchestrator_config = None
  if _HELM_CHART:
    orchestrator_config = checker_common.HelmConfig(
        release_name_base=_HELM_RELEASE_NAME_BASE,
        chart=_HELM_CHART,
        chart_version=_HELM_CHART_VERSION,
        install_flags=_HELM_INSTALL_FLAGS,
    )
  elif _YAML_FILE:
    orchestrator_config = _YAML_FILE

  if health_app == "nccl":
    logging.info("Running NCCL health check via `HEALTH_APP`")
    nccl_result = nccl_runner.run_nccl_healthcheck(
        orchestrator_config=orchestrator_config
    )
    health_results.health_results.append(nccl_result)
  else:
    logging.error("Unsupported health app: %s", health_app)
    return

  print(health_results)

  # If GCS_BUCKET_NAME is set, upload the results to GCS.
  if _GCS_BUCKET_NAME:
    checker_common.upload_results_to_gcs(
        bucket_name=_GCS_BUCKET_NAME,
        health_results=health_results,
    )