code: def real_main():
      (opts, action, cluster_name) = parse_args()
  
      # Input parameter validation
      get_validate_spark_version(
          opts.spark_version, opts.spark_git_repo
      )
  
      if opts.wait is not None:
          # NOTE: DeprecationWarnings are silent in 2.7+ by default.
          #       To show them, run Python with the -Wdefault switch.
          # See: https://docs.python.org/3.5/whatsnew/2.7.html
          warnings.warn(
              "This option is deprecated and has no effect. "
              "spark-ec2 automatically waits as long as necessary for clusters to start up.",
              DeprecationWarning,
          )
  
      if opts.identity_file is not None:
          if not os.path.exists(opts.identity_file):
              print(
                  "ERROR: The identity file '{f}' doesn't exist.".format(
                      f=opts.identity_file
                  ),
                  file=stderr,
              )
              sys.exit(1)
  
          file_mode = os.stat(opts.identity_file).st_mode
          if (
              not (file_mode & S_IRUSR)
              or not oct(file_mode)[-2:] == "00"
          ):
              print(
                  "ERROR: The identity file must be accessible only by you.",
                  file=stderr,
              )
              print(
                  'You can fix this with: chmod 400 "{f}"'.format(
                      f=opts.identity_file
                  ),
                  file=stderr,
              )
              sys.exit(1)
  
      if opts.instance_type not in EC2_INSTANCE_TYPES:
          print(
              "Warning: Unrecognized EC2 instance type for instance-type: {t}".format(
                  t=opts.instance_type
              ),
              file=stderr,
          )
  
      if opts.master_instance_type != "":
          if opts.master_instance_type not in EC2_INSTANCE_TYPES:
              print(
                  "Warning: Unrecognized EC2 instance type for master-instance-type: {t}".format(
                      t=opts.master_instance_type
                  ),
                  file=stderr,
              )
          # Since we try instance types even if we can't resolve them, we check if they resolve first
          # and, if they do, see if they resolve to the same virtualization type.
          if (
              opts.instance_type in EC2_INSTANCE_TYPES
              and opts.master_instance_type in EC2_INSTANCE_TYPES
          ):
              if (
                  EC2_INSTANCE_TYPES[opts.instance_type]
                  != EC2_INSTANCE_TYPES[opts.master_instance_type]
              ):
                  print(
                      "Error: spark-ec2 currently does not support having a master and slaves "
                      "with different AMI virtualization types.",
                      file=stderr,
                  )
                  print(
                      "master instance virtualization type: {t}".format(
                          t=EC2_INSTANCE_TYPES[
                              opts.master_instance_type
                          ]
                      ),
                      file=stderr,
                  )
                  print(
                      "slave instance virtualization type: {t}".format(
                          t=EC2_INSTANCE_TYPES[opts.instance_type]
                      ),
                      file=stderr,
                  )
                  sys.exit(1)
  
      if opts.ebs_vol_num > 8:
          print("ebs-vol-num cannot be greater than 8", file=stderr)
          sys.exit(1)
  
      # Prevent breaking ami_prefix (/, .git and startswith checks)
      # Prevent forks with non spark-ec2 names for now.
      if (
          opts.spark_ec2_git_repo.endswith("/")
          or opts.spark_ec2_git_repo.endswith(".git")
          or not opts.spark_ec2_git_repo.startswith(
              "https://github.com"
          )
          or not opts.spark_ec2_git_repo.endswith("spark-ec2")
      ):
          print(
              "spark-ec2-git-repo must be a github repo and it must not have a trailing / or .git. "
              "Furthermore, we currently only support forks named spark-ec2.",
              file=stderr,
          )
          sys.exit(1)
  
      if not (
          opts.deploy_root_dir is None
          or (
              os.path.isabs(opts.deploy_root_dir)
              and os.path.isdir(opts.deploy_root_dir)
              and os.path.exists(opts.deploy_root_dir)
          )
      ):
          print(
              "--deploy-root-dir must be an absolute path to a directory that exists "
              "on the local file system",
              file=stderr,
          )
          sys.exit(1)
  
      try:
          if opts.profile is None:
              conn = ec2.connect_to_region(opts.region)
          else:
              conn = ec2.connect_to_region(
                  opts.region, profile_name=opts.profile
              )
      except Exception as e:
          print((e), file=stderr)
          sys.exit(1)
  
      # Select an AZ at random if it was not specified.
      if opts.zone == "":
          opts.zone = random.choice(conn.get_all_zones()).name
  
      if action == "launch":
          if opts.slaves <= 0:
              print(
                  "ERROR: You have to start at least 1 slave",
                  file=sys.stderr,
              )
              sys.exit(1)
          if opts.resume:
              (master_nodes, slave_nodes) = get_existing_cluster(
                  conn, opts, cluster_name
              )
          else:
              (master_nodes, slave_nodes) = launch_cluster(
                  conn, opts, cluster_name
              )
          wait_for_cluster_state(
              conn=conn,
              opts=opts,
              cluster_instances=(master_nodes + slave_nodes),
              cluster_state="ssh-ready",
          )
          setup_cluster(conn, master_nodes, slave_nodes, opts, True)
  
      elif action == "destroy":
          (master_nodes, slave_nodes) = get_existing_cluster(
              conn, opts, cluster_name, die_on_error=False
          )
  
          if any(master_nodes + slave_nodes):
              print("The following instances will be terminated:")
              for inst in master_nodes + slave_nodes:
                  print("> %s" % get_dns_name(inst, opts.private_ips))
              print("ALL DATA ON ALL NODES WILL BE LOST!!")
  
          msg = "Are you sure you want to destroy the cluster {c}? (y/N) ".format(
              c=cluster_name
          )
          response = raw_input(msg)
          if response == "y":
              print("Terminating master...")
              for inst in master_nodes:
                  inst.terminate()
              print("Terminating slaves...")
              for inst in slave_nodes:
                  inst.terminate()
  
              # Delete security groups as well
              if opts.delete_groups:
                  group_names = [
                      cluster_name + "-master",
                      cluster_name + "-slaves",
                  ]
                  wait_for_cluster_state(
                      conn=conn,
                      opts=opts,
                      cluster_instances=(master_nodes + slave_nodes),
                      cluster_state="terminated",
                  )
                  print(
                      "Deleting security groups (this will take some time)..."
                  )
                  attempt = 1
                  while attempt <= 3:
                      print("Attempt %d" % attempt)
                      groups = [
                          g
                          for g in conn.get_all_security_groups()
                          if g.name in group_names
                      ]
                      success = True
                      # Delete individual rules in all groups before deleting groups to
                      # remove dependencies between them
                      for group in groups:
                          print(
                              "Deleting rules in security group "
                              + group.name
                          )
                          for rule in group.rules:
                              for grant in rule.grants:
                                  success &= group.revoke(
                                      ip_protocol=rule.ip_protocol,
                                      from_port=rule.from_port,
                                      to_port=rule.to_port,
                                      src_group=grant,
                                  )
  
                      # Sleep for AWS eventual-consistency to catch up, and for instances
                      # to terminate
                      time.sleep(
                          30
                      )  # Yes, it does have to be this long :-(
                      for group in groups:
                          try:
                              # It is needed to use group_id to make it work with VPC
                              conn.delete_security_group(
                                  group_id=group.id
                              )
                              print(
                                  "Deleted security group %s"
                                  % group.name
                              )
                          except boto.exception.EC2ResponseError:
                              success = False
                              print(
                                  "Failed to delete security group %s"
                                  % group.name
                              )
  
                      # Unfortunately, group.revoke() returns True even if a rule was not
                      # deleted, so this needs to be rerun if something fails
                      if success:
                          break
  
                      attempt += 1
  
                  if not success:
                      print(
                          "Failed to delete all security groups after 3 tries."
                      )
                      print("Try re-running in a few minutes.")
  
      elif action == "login":
          (master_nodes, slave_nodes) = get_existing_cluster(
              conn, opts, cluster_name
          )
          if (
              not master_nodes[0].public_dns_name
              and not opts.private_ips
          ):
              print(
                  "Master has no public DNS name.  Maybe you meant to specify --private-ips?"
              )
          else:
              master = get_dns_name(master_nodes[0], opts.private_ips)
              print("Logging into master " + master + "...")
              proxy_opt = []
              if opts.proxy_port is not None:
                  proxy_opt = ["-D", opts.proxy_port]
              subprocess.check_call(
                  ssh_command(opts)
                  + proxy_opt
                  + ["-t", "-t", "%s@%s" % (opts.user, master)]
              )
  
      elif action == "reboot-slaves":
          response = raw_input(
              "Are you sure you want to reboot the cluster "
              + cluster_name
              + " slaves?\n"
              + "Reboot cluster slaves "
              + cluster_name
              + " (y/N): "
          )
          if response == "y":
              (master_nodes, slave_nodes) = get_existing_cluster(
                  conn, opts, cluster_name, die_on_error=False
              )
              print("Rebooting slaves...")
              for inst in slave_nodes:
                  if inst.state not in ["shutting-down", "terminated"]:
                      print("Rebooting " + inst.id)
                      inst.reboot()
  
      elif action == "get-master":
          (master_nodes, slave_nodes) = get_existing_cluster(
              conn, opts, cluster_name
          )
          if (
              not master_nodes[0].public_dns_name
              and not opts.private_ips
          ):
              print(
                  "Master has no public DNS name.  Maybe you meant to specify --private-ips?"
              )
          else:
              print(get_dns_name(master_nodes[0], opts.private_ips))
  
      elif action == "stop":
          response = raw_input(
              "Are you sure you want to stop the cluster "
              + cluster_name
              + "?\nDATA ON EPHEMERAL DISKS WILL BE LOST, "
              + "BUT THE CLUSTER WILL KEEP USING SPACE ON\n"
              + "AMAZON EBS IF IT IS EBS-BACKED!!\n"
              + "All data on spot-instance slaves will be lost.\n"
              + "Stop cluster "
              + cluster_name
              + " (y/N): "
          )
          if response == "y":
              (master_nodes, slave_nodes) = get_existing_cluster(
                  conn, opts, cluster_name, die_on_error=False
              )
              print("Stopping master...")
              for inst in master_nodes:
                  if inst.state not in ["shutting-down", "terminated"]:
                      inst.stop()
              print("Stopping slaves...")
              for inst in slave_nodes:
                  if inst.state not in ["shutting-down", "terminated"]:
                      if inst.spot_instance_request_id:
                          inst.terminate()
                      else:
                          inst.stop()
  
      elif action == "start":
          (master_nodes, slave_nodes) = get_existing_cluster(
              conn, opts, cluster_name
          )
          print("Starting slaves...")
          for inst in slave_nodes:
              if inst.state not in ["shutting-down", "terminated"]:
                  inst.start()
          print("Starting master...")
          for inst in master_nodes:
              if inst.state not in ["shutting-down", "terminated"]:
                  inst.start()
          wait_for_cluster_state(
              conn=conn,
              opts=opts,
              cluster_instances=(master_nodes + slave_nodes),
              cluster_state="ssh-ready",
          )
  
          # Determine types of running instances
          existing_master_type = master_nodes[0].instance_type
          existing_slave_type = slave_nodes[0].instance_type
          # Setting opts.master_instance_type to the empty string indicates we
          # have the same instance type for the master and the slaves
          if existing_master_type == existing_slave_type:
              existing_master_type = ""
          opts.master_instance_type = existing_master_type
          opts.instance_type = existing_slave_type
  
          setup_cluster(conn, master_nodes, slave_nodes, opts, False)
  
      else:
          print("Invalid action: %s" % action, file=stderr)
          sys.exit(1)
  
imports: from __future__ import division, print_function, with_statement
  import codecs
  import hashlib
  import itertools
  import logging
  import os
  import os.path
  import pipes
  import random
  import shutil
  import string
  from stat import S_IRUSR
  import subprocess
  import sys
  import tarfile
  import tempfile
  import textwrap
  import time
  import warnings
  from datetime import datetime
  from optparse import OptionParser
  from sys import stderr
  import boto
  from boto.ec2.blockdevicemapping import BlockDeviceMapping, BlockDeviceType, EBSBlockDeviceType
  from boto import ec2
label_desc: Use image node_id 29 for calling a global function with example usage: real_main() without return types