def _get_occupied_nodes()

in cli/gke_check.py [0:0]

• 27 lines of code
• 9 McCabe index (conditional complexity)

  def _get_occupied_nodes(self) -> set[str]:
    """Gets all requested nodes that are currently occupied.

    Returns:
      A list of occupied nodes. Optionally, if nodes is provided, only the
      occupied nodes of those provided will be returned.
    """
    occupied_nodes = set()
    try:
      pods = self._v1.list_pod_for_all_namespaces(
          watch=False, field_selector='status.phase=Running'
      )
      for pod in pods.items:
        for container in pod.spec.containers:
          if container.resources.requests:
            requested_gpus = set(
                resource_name
                for resource_name, _ in container.resources.requests.items()
                if resource_name == 'nvidia.com/gpu'
            )
            if requested_gpus:
              occupied_nodes.add(pod.spec.node_name)
    except client.rest.ApiException as e:
      click.echo(
          click.style(
              f'Failed to list nodes in cluster: {e}', fg='red', bold=True
          )
      )

    return (
        occupied_nodes
        if not self.nodes
        else set(self.nodes).intersection(occupied_nodes)
    )