def update_node_labels()

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

• 24 lines of code
• 3 McCabe index (conditional complexity)

def update_node_labels(kube: client.CoreV1Api) -> None:
  """Updates Kubernetes node labels based on GCE VM metadata."""
  node_name_url = "http://metadata.google.internal/computeMetadata/v1/instance/name"
  metadata_url = "http://metadata.google.internal/computeMetadata/v1/instance/attributes/physical_host"
  headers = {"Metadata-Flavor": "Google"}

  response = requests.get(node_name_url, headers=headers)

  if response.status_code == 200:
    node_name = response.text
  else:
    print("Node name not found")
    return

  response = requests.get(metadata_url, headers=headers)

  if response.status_code == 200:
    physical_host = response.text
  else:
    print("physical host not found")
    return

  cluster, rack, host = physical_host.split("/")[1:]

  node_labels: Dict[str, str] = {
      "topology.gke.io/cluster": cluster,
      "topology.gke.io/rack": rack,
      "topology.gke.io/host": host,
  }

  kube.patch_node(node_name, {"metadata": {"labels": node_labels}})  # type: ignore
  print(f"Updated labels on node {node_name}: {node_labels}")