code: def _configure_key_pair(config):
      node_types = config["available_node_types"]
  
      # map from node type key -> source of KeyName field
      key_pair_src_info = {}
      _set_config_info(keypair_src=key_pair_src_info)
  
      if "ssh_private_key" in config["auth"]:
          for node_type_key in node_types:
              # keypairs should be provided in the config
              key_pair_src_info[node_type_key] = "config"
  
          # If the key is not configured via the cloudinit
          # UserData, it should be configured via KeyName or
          # else we will risk starting a node that we cannot
          # SSH into:
  
          for node_type in node_types:
              node_config = node_types[node_type]["node_config"]
              if "UserData" not in node_config:
                  cli_logger.doassert(
                      "KeyName" in node_config,
                      _key_assert_msg(node_type),
                  )
                  assert "KeyName" in node_config
  
          return config
  
      for node_type_key in node_types:
          key_pair_src_info[node_type_key] = "default"
  
      ec2 = _resource("ec2", config)
  
      # Writing the new ssh key to the filesystem fails if the ~/.ssh
      # directory doesn't already exist.
      os.makedirs(os.path.expanduser("~/.ssh"), exist_ok=True)
  
      # Try a few times to get or create a good key pair.
      MAX_NUM_KEYS = 600
      for i in range(MAX_NUM_KEYS):
          key_name = (
              config["provider"].get("key_pair", {}).get("key_name")
          )
  
          key_name, key_path = key_pair(
              i, config["provider"]["region"], key_name
          )
          key = _get_key(key_name, config)
  
          # Found a good key.
          if key and os.path.exists(key_path):
              break
  
          # We can safely create a new key.
          if not key and not os.path.exists(key_path):
              cli_logger.verbose(
                  "Creating new key pair {} for use as the default.",
                  cf.bold(key_name),
              )
              key = ec2.create_key_pair(KeyName=key_name)
  
              # We need to make sure to _create_ the file with the right
              # permissions. In order to do that we need to change the default
              # os.open behavior to include the mode we want.
              with open(
                  key_path, "w", opener=partial(os.open, mode=0o600)
              ) as f:
                  f.write(key.key_material)
              break
  
      if not key:
          cli_logger.abort(
              "No matching local key file for any of the key pairs in this "
              "account with ids from 0..{}. "
              "Consider deleting some unused keys pairs from your account.",
              key_name,
          )
  
      cli_logger.doassert(
          os.path.exists(key_path),
          "Private key file "
          + cf.bold("{}")
          + " not found for "
          + cf.bold("{}"),
          key_path,
          key_name,
      )  # todo: err msg
      assert os.path.exists(
          key_path
      ), "Private key file {} not found for {}".format(
          key_path, key_name
      )
  
      config["auth"]["ssh_private_key"] = key_path
      for node_type in node_types.values():
          node_config = node_type["node_config"]
          node_config["KeyName"] = key_name
  
      return config
  
imports: import copy
  import itertools
  import json
  import logging
  import os
  import time
  from collections import Counter
  from functools import lru_cache, partial
  from typing import Any, Dict, List, Optional, Set, Tuple
  import boto3
  import botocore
  from packaging.version import Version
  from ray.autoscaler._private.aws.cloudwatch.cloudwatch_helper import CloudwatchHelper
  from ray.autoscaler._private.aws.utils import LazyDefaultDict, handle_boto_error, resource_cache
  from ray.autoscaler._private.cli_logger import cf, cli_logger
  from ray.autoscaler._private.event_system import CreateClusterEvent, global_event_system
  from ray.autoscaler._private.providers import _PROVIDER_PRETTY_NAMES
  from ray.autoscaler._private.util import check_legacy_fields
  from ray.autoscaler.tags import NODE_TYPE_LEGACY_HEAD, NODE_TYPE_LEGACY_WORKER
label_desc: Use image node_id 9 for calling a global function with example usage: _configure_key_pair(config) and returns: config, config