def can_schedule()

in gke-topology-scheduler/schedule-daemon.py [0:0]

• 11 lines of code
• 6 McCabe index (conditional complexity)

def can_schedule(node: dict[str, Any], pod: dict[str, Any]) -> bool:
  """Checks if a given pod can be scheduled on a given node.

  The decision is based on resource node availability and pod requriements. The
  node_selector is also checked, if the pod has it.

  Args:
    node: The node to check if the pod can be scheduled on.
    pod: The pod to check if it can be scheduled on the node.

  Returns:
    True if the pod can be scheduled on the node, False otherwise.
  """
  node_selector = pod.get('node_selector', {})
  node_labels = node.get('node_labels', {})

  for key, value in node_selector.items():
    if key not in node_labels or node_labels[key] != value:
      return False

  return (
      node['cpu'] >= pod['cpu']
      and node['memory'] >= pod['memory']
      and node['gpu'] >= pod['gpu']
  )