code: def _create_node(self, node_config, tags, count):
      created_nodes_dict = {}
  
      tags = to_aws_format(tags)
      conf = node_config.copy()
  
      tag_pairs = [
          {
              "Key": TAG_RAY_CLUSTER_NAME,
              "Value": self.cluster_name,
          }
      ]
      for k, v in tags.items():
          tag_pairs.append(
              {
                  "Key": k,
                  "Value": v,
              }
          )
      if CloudwatchHelper.cloudwatch_config_exists(
          self.provider_config, "agent"
      ):
          cwa_installed = self._check_ami_cwa_installation(node_config)
          if cwa_installed:
              tag_pairs.extend(
                  [
                      {
                          "Key": CLOUDWATCH_AGENT_INSTALLED_TAG,
                          "Value": "True",
                      }
                  ]
              )
      tag_specs = [
          {
              "ResourceType": "instance",
              "Tags": tag_pairs,
          }
      ]
      user_tag_specs = conf.get("TagSpecifications", [])
      AWSNodeProvider._merge_tag_specs(tag_specs, user_tag_specs)
  
      # SubnetIds is not a real config key: we must resolve to a
      # single SubnetId before invoking the AWS API.
      subnet_ids = conf.pop("SubnetIds")
  
      # update config with min/max node counts and tag specs
      conf.update(
          {
              "MinCount": 1,
              "MaxCount": count,
              "TagSpecifications": tag_specs,
          }
      )
  
      # Try to always launch in the first listed subnet.
      subnet_idx = 0
      cli_logger_tags = {}
      # NOTE: This ensures that we try ALL availability zones before
      # throwing an error.
      max_tries = max(BOTO_CREATE_MAX_RETRIES, len(subnet_ids))
      for attempt in range(1, max_tries + 1):
          try:
              if "NetworkInterfaces" in conf:
                  net_ifs = conf["NetworkInterfaces"]
                  # remove security group IDs previously copied from network
                  # interfaces (create_instances call fails otherwise)
                  conf.pop("SecurityGroupIds", None)
                  cli_logger_tags["network_interfaces"] = str(net_ifs)
              else:
                  subnet_id = subnet_ids[subnet_idx % len(subnet_ids)]
                  conf["SubnetId"] = subnet_id
                  cli_logger_tags["subnet_id"] = subnet_id
  
              created = self.ec2_fail_fast.create_instances(**conf)
              created_nodes_dict = {n.id: n for n in created}
  
              # todo: timed?
              # todo: handle plurality?
              with cli_logger.group(
                  "Launched {} nodes", count, _tags=cli_logger_tags
              ):
                  for instance in created:
                      # NOTE(maximsmol): This is needed for mocking
                      # boto3 for tests. This is likely a bug in moto
                      # but AWS docs don't seem to say.
                      # You can patch moto/ec2/responses/instances.py
                      # to fix this (add <stateReason> to EC2_RUN_INSTANCES)
  
                      # The correct value is technically
                      # {"code": "0", "Message": "pending"}
                      state_reason = instance.state_reason or {
                          "Message": "pending"
                      }
  
                      cli_logger.print(
                          "Launched instance {}",
                          instance.instance_id,
                          _tags=dict(
                              state=instance.state["Name"],
                              info=state_reason["Message"],
                          ),
                      )
              break
          except botocore.exceptions.ClientError as exc:
              # Launch failure may be due to instance type availability in
              # the given AZ
              subnet_idx += 1
              if attempt == max_tries:
                  try:
                      exc = NodeLaunchException(
                          category=exc.response["Error"]["Code"],
                          description=exc.response["Error"]["Message"],
                          src_exc_info=sys.exc_info(),
                      )
                  except Exception:
                      # In theory, all ClientError's we expect to get should
                      # have these fields, but just in case we can't parse
                      # it, it's fine, just throw the original error.
                      logger.warning("Couldn't parse exception.", exc)
                      pass
                  cli_logger.abort(
                      "Failed to launch instances. Max attempts exceeded.",
                      exc=exc,
                  )
              else:
                  cli_logger.warning(
                      "create_instances: Attempt failed with {}, retrying.",
                      exc,
                  )
  
      return created_nodes_dict
  
imports: import copy
  import logging
  import sys
  import threading
  import time
  from collections import OrderedDict, defaultdict
  from typing import Any, Dict, List
  import botocore
  from boto3.resources.base import ServiceResource
  import ray
  import ray._private.ray_constants
  from ray.autoscaler._private.aws.cloudwatch.cloudwatch_helper import CLOUDWATCH_AGENT_INSTALLED_AMI_TAG, CLOUDWATCH_AGENT_INSTALLED_TAG, CloudwatchHelper
  from ray.autoscaler._private.aws.config import bootstrap_aws
  from ray.autoscaler._private.aws.utils import boto_exception_handler, client_cache, resource_cache
  from ray.autoscaler._private.cli_logger import cf, cli_logger
  from ray.autoscaler._private.constants import BOTO_CREATE_MAX_RETRIES, BOTO_MAX_RETRIES
  from ray.autoscaler._private.log_timer import LogTimer
  from ray.autoscaler.node_launch_exception import NodeLaunchException
  from ray.autoscaler.node_provider import NodeProvider
  from ray.autoscaler.tags import TAG_RAY_CLUSTER_NAME, TAG_RAY_LAUNCH_CONFIG, TAG_RAY_NODE_KIND, TAG_RAY_NODE_NAME, TAG_RAY_USER_NODE_TYPE
label_desc: Use image node_id 13 for calling the AWSNodeProvider obj's underlying member method code with example usage: obj._create_node(node_config, tags, count) and returns: created_nodes_dict