@SuppressWarnings("Guava")
public class AWSFilters {

  /**
   * A configuration key representing a blacklist.
   */
  public static final String BLACKLIST = "blacklist";

  /**
   * A configuration key representing a whitelist.
   */
  public static final String WHITELIST = "whitelist";

  /**
   * Empty filters.
   */
  public static final AWSFilters EMPTY_FILTERS =
      new AWSFilters(Optional.absent());

  /**
   * Returns AWS filters based on the specified configuration.
   *
   * @param config the underlying configuration
   * @return AWS filters based on the specified configuration
   */
  public static AWSFilters getAWSFilters(Config config) {
    return (config == null) ? EMPTY_FILTERS : new AWSFilters(Optional.of(config));
  }

  /**
   * The underlying configuration.
   */
  private Optional<Config> config;

  /**
   * Creates AWS filters with the specified parameters.
   *
   * @param config the underlying configuration
   */
  public AWSFilters(Optional<Config> config) {
    this.config = Preconditions.checkNotNull(config, "config is null");
  }

  /**
   * Returns the subfilters identified by the specified
   * sequence of keys.
   *
   * @param keys the keys
   * @return the subfilters identified by the specified
   * sequence of keys
   */
  public AWSFilters getSubfilters(String... keys) {
    if (config.isPresent()) {
      Config config = this.config.get();
      for (String key : keys) {
        try {
          config = config.getConfig(HoconConfigUtils.asConfigKey(key));
        } catch (ConfigException.Missing e) {
          return EMPTY_FILTERS;
        }
      }
      return getAWSFilters(config);
    } else {
      return EMPTY_FILTERS;
    }
  }

  /**
   * Returns the filter map for the specified key.
   *
   * @param key the key
   * @return the filter map for the specified key
   */
  public Map<String, String> getFilterMap(String key) {
    Map<String, String> filterMap;
    try {
      filterMap = config.isPresent()
          ? HoconConfigUtils.getStringMap(config.get(), HoconConfigUtils.asConfigKey(key))
          : Collections.emptyMap();
    } catch (Exception e) {
      filterMap = Collections.emptyMap();
    }
    return ImmutableMap.copyOf(filterMap);
  }

  /**
   * Returns whether the specified value is in the specified blacklist.
   *
   * @param key   the key identifying the specific blacklist to check
   * @param value the value to check
   * @return whether the specified value is in the specified blacklist
   */
  public boolean isBlacklisted(String key, String value) {
    return getBlacklistValue(key, value) != null;
  }

  /**
   * If the specified value is in the specified blacklist, returns the associated
   * blacklist value. Otherwise, returns {@code null}.
   *
   * @param key   the key identifying the specific blacklist to check
   * @param value the value to check
   * @return the blacklist value associated with the specified value in the specified blacklist,
   * or {@code null}
   */
  public String getBlacklistValue(String key, String value) {
    Map<String, String> blacklistMap = getSubfilters(key).getFilterMap(AWSFilters.BLACKLIST);
    return blacklistMap.get(value);
  }
} @SuppressWarnings("Guava")
public class AWSTimeouts {

  private final ImmutableMap<String, Long> timeouts;

  /**
   * Creates a new timeouts object from the given configuration. It is
   * expected that every value in the recursive tree of the configuration is
   * numeric. Each (recursive) key in the config serves as the key for a
   * timeout.
   *
   * @param config config holding timeouts
   * @throws IllegalArgumentException if any configuration value is non-numeric
   */
  public AWSTimeouts(Config config) {
    ImmutableMap.Builder<String, Long> b = ImmutableMap.builder();

    if (config != null) {
      for (Map.Entry<String, ConfigValue> e : config.entrySet()) {
        String key = e.getKey();
        ConfigValue value = e.getValue();
        switch (value.valueType()) {
          case NUMBER:
            long num = ((Number) value.unwrapped()).longValue();
            if (num <= 0L) {
              throw new IllegalArgumentException("Timeout " + key + " is negative: " + value);
            }
            b.put(key, num);
            break;
          default:
            throw new IllegalArgumentException("Timeout " + key + " is not a number: " + value);
        }
      }
    }

    timeouts = b.build();
  }

  /**
   * Gets a timeout value.
   *
   * @param key timeout key
   * @return timeout value as an Optional
   */
  public Optional<Long> getTimeout(String key) {
    return Optional.fromNullable(timeouts.get(key));
  }
} public class CustomTagMappings {

  private final ImmutableMap<String, String> customTagNames;

  /**
   * Creates a new custom tag mappings object from the given configuration. The keys of the
   * configuration are well-known tag names, and the values are the custom tag names.
   *
   * @param config config holding custom tags
   * @throws IllegalArgumentException if any configuration value is not a string or is empty
   */
  public CustomTagMappings(Config config) {
    ImmutableMap.Builder<String, String> b = ImmutableMap.builder();

    if (config != null) {
      for (Map.Entry<String, ConfigValue> e : config.entrySet()) {
        String key = e.getKey();
        ConfigValue value = e.getValue();
        switch (value.valueType()) {
          case STRING:
            String customTagName = (String) value.unwrapped();
            if (customTagName.isEmpty()) {
              throw new IllegalArgumentException("Tag mapping " + key + " is empty");
            }
            b.put(key, customTagName);
            break;
          default:
            throw new IllegalArgumentException("Tag mapping " + key + " is not a string: " + value);
        }
      }
    }

    customTagNames = b.build();
  }

  /**
   * Returns the custom tag name for the specified tag key if present, otherwise the key itself.
   *
   * @param tagKey tag key
   * @return the custom tag name for the specified tag key if present, otherwise the key itself
   */
  public String getCustomTagName(String tagKey) {
    return customTagNames.getOrDefault(tagKey, tagKey);
  }

  /**
   * Returns the name of the tag used to record the virtual instance ID as generated by Cloudera
   * Director.
   *
   * @return the name of the tag used to record the virtual instance ID as generated by Cloudera
   * Director
   */
  public String getClouderaDirectorIdTagName() {
    return getCustomTagName(Tags.ResourceTags.CLOUDERA_DIRECTOR_ID.getTagKey());
  }
} public final class Tags {

  /**
   * Director tags for cloud resources.
   */
  public enum ResourceTags {
    /**
     * The name of the resource.
     */
    RESOURCE_NAME("Name"),

    /**
     * The virtual instance ID as generated by Cloudera Altus Director.
     */
    CLOUDERA_DIRECTOR_ID("Cloudera-Director-Id"),

    /**
     * The virtual instance ID as generated by Cloudera Altus Director.
     */
    CLOUDERA_DIRECTOR_TEMPLATE_NAME("Cloudera-Director-Template-Name");

    private final String tagName;

    ResourceTags(String tagName) {
      this.tagName = tagName;
    }

    /**
     * Gets the tag's key.
     * Note that you should always call this method to get the tag's key, and not the
     * {@link ResourceTags#valueOf(String)} method.
     *
     * @return tag's key name
     */
    public String getTagKey() {
      return tagName;
    }
  }

  /**
   * Director tags specifically for cloud instances.
   */
  public enum InstanceTags {
    /**
     * The instance owner.
     */
    OWNER("owner");

    private final String tagName;

    InstanceTags(String tagName) {
      this.tagName = tagName;
    }

    /**
     * Gets the tag's key.
     * Note that you should always call this method to get the tag's key, and not the
     * {@link InstanceTags#valueOf(String)} method.
     *
     * @return tag's key name
     */
    public String getTagKey() {
      return tagName;
    }
  }


  /**
   * Private constructor to prevent instantiation.
   */
  private Tags() {
  }
} public class AWSKMSClientProvider extends AbstractConfiguredOnceClientProvider<AWSKMSClient> {
  private static final Logger LOG = LoggerFactory.getLogger(AWSKMSClientProvider.class);

  /**
   * Creates an Amazon KMS client provider with the specified parameters.
   *
   * @param awsCredentialsProvider the AWS credentials provider
   * @param clientConfiguration    the client configuration
   */
  public AWSKMSClientProvider(
      AWSCredentialsProvider awsCredentialsProvider,
      ClientConfiguration clientConfiguration) {
    super(awsCredentialsProvider, clientConfiguration);
  }

  /**
   * {@inheritDoc}.
   */
  @Override
  protected AWSKMSClient doConfigure(
      Configured configuration,
      PluginExceptionConditionAccumulator accumulator,
      LocalizationContext providerLocalizationContext,
      boolean verify) {

    AWSKMSClient client = new AWSKMSClient(awsCredentialsProvider, clientConfiguration);

    try {
      String regionEndpoint = configuration.getConfigurationValue(KMS_REGION_ENDPOINT, providerLocalizationContext);
      if (regionEndpoint != null) {
        LOG.info("<< Using configured region endpoint for KMS client: {}", regionEndpoint);
      } else {
        String region = configuration.getConfigurationValue(REGION, providerLocalizationContext);
        regionEndpoint = getKMSEndpointForRegion(client, region);
      }
      client.setEndpoint(regionEndpoint);
    } catch (AmazonClientException e) {
      AWSSecurityTokenServiceAsyncClient stsClient = new AWSSTSClientProvider(awsCredentialsProvider,
          clientConfiguration) .doConfigure(configuration, accumulator, providerLocalizationContext, verify);
      throw AWSExceptions.propagate(stsClient, e);
    } catch (IllegalArgumentException e) {
      accumulator.addError(REGION.unwrap().getConfigKey(), e.getMessage());
    }

    return client;
  }

  /**
   * Returns the KMS endpoint URL for the specified region.
   *
   * @param kmsClient  the KMS client
   * @param regionName the desired region
   * @return the endpoint URL for the specified region
   * @throws IllegalArgumentException if the endpoint cannot be determined
   */
  private static String getKMSEndpointForRegion(AWSKMSClient kmsClient, String regionName) {
    requireNonNull(kmsClient, "kmsClient is null");
    requireNonNull(regionName, "regionName is null");

    com.amazonaws.regions.Region region = RegionUtils.getRegion(regionName);

    if (region == null) {
      throw new IllegalArgumentException(String.format("Unable to find the region %s", regionName));
    }

    String serviceName = kmsClient.getServiceName();
    String protocolPrefix = region.hasHttpsEndpoint(serviceName) ? "https://" : "http://";
    return protocolPrefix + region.getServiceEndpoint(serviceName);
  }
} public class AWSSTSClientProvider extends AbstractConfiguredOnceClientProvider<AWSSecurityTokenServiceAsyncClient> {
  private static final Logger LOG = LoggerFactory.getLogger(AWSSTSClientProvider.class);

  /**
   * Creates an Amazon security token client provider with the specified parameters.
   *
   * @param awsCredentialsProvider the AWS credentials provider
   * @param clientConfiguration    the client configuration
   */
  public AWSSTSClientProvider(
      AWSCredentialsProvider awsCredentialsProvider,
      ClientConfiguration clientConfiguration) {
    super(awsCredentialsProvider, clientConfiguration);
  }

  @Override
  protected AWSSecurityTokenServiceAsyncClient doConfigure(Configured configuration, PluginExceptionConditionAccumulator accumulator, LocalizationContext providerLocalizationContext, boolean verify) {

    AWSSecurityTokenServiceAsyncClient client =
        new AWSSecurityTokenServiceAsyncClient(awsCredentialsProvider, clientConfiguration);

    try {
      String regionEndpoint = configuration.getConfigurationValue(STS_REGION_ENDPOINT, providerLocalizationContext);

      // We only configure the region endpoint if it's specifically set, as AWS uses a global endpoint by default.
      // Other regions/endpoints must be explicitly enabled by the user.
      if (regionEndpoint != null) {
        LOG.info("<< Using configured region endpoint for STS client: {}", regionEndpoint);
        client.setEndpoint(regionEndpoint);
      }
    } catch (AmazonClientException e) {
      throw AWSExceptions.propagate(client, e);
    }

    return client;
  }
} public class AmazonAutoScalingClientProvider
    extends AbstractConfiguredOnceClientProvider<AmazonAutoScalingAsyncClient> {
  private static final Logger LOG = LoggerFactory.getLogger(AmazonAutoScalingClientProvider.class);

  /**
   * Creates an Amazon Auto Scaling client provider with the specified parameters.
   *
   * @param awsCredentialsProvider the AWS credentials provider
   * @param clientConfiguration    the client configuration
   */
  public AmazonAutoScalingClientProvider(
      AWSCredentialsProvider awsCredentialsProvider,
      ClientConfiguration clientConfiguration) {
    super(awsCredentialsProvider, clientConfiguration);
  }

  /**
   * {@inheritDoc}.
   */
  @Override
  protected AmazonAutoScalingAsyncClient doConfigure(
      Configured configuration,
      PluginExceptionConditionAccumulator accumulator,
      LocalizationContext providerLocalizationContext,
      boolean verify) {

    AmazonAutoScalingAsyncClient client =
        new AmazonAutoScalingAsyncClient(awsCredentialsProvider, clientConfiguration);

    try {
      String regionEndpoint = configuration.getConfigurationValue(AS_REGION_ENDPOINT, providerLocalizationContext);
      if (regionEndpoint != null) {
        LOG.info("<< Using configured region endpoint for Auto Scaling client: {}", regionEndpoint);
      } else {
        String region = configuration.getConfigurationValue(REGION, providerLocalizationContext);
        regionEndpoint = getASEndpointForRegion(client, region);
      }
      client.setEndpoint(regionEndpoint);
    } catch (AmazonClientException e) {
      AWSSecurityTokenServiceAsyncClient stsClient = new AWSSTSClientProvider(awsCredentialsProvider,
          clientConfiguration) .doConfigure(configuration, accumulator, providerLocalizationContext, verify);
      throw AWSExceptions.propagate(stsClient, e);
    } catch (IllegalArgumentException e) {
      accumulator.addError(REGION.unwrap().getConfigKey(), e.getMessage());
    }

    return client;
  }

  /**
   * Returns the Auto Scaling endpoint URL for the specified region.
   *
   * @param asClient  the Auto Scaling client
   * @param regionName the desired region
   * @return the endpoint URL for the specified region
   * @throws IllegalArgumentException if the endpoint cannot be determined
   */
  private static String getASEndpointForRegion(AmazonAutoScalingAsyncClient asClient, String regionName) {
    requireNonNull(regionName, "regionName is null");

    com.amazonaws.regions.Region region = RegionUtils.getRegion(regionName);

    if (region == null) {
      throw new IllegalArgumentException(String.format("Unable to find the region %s", regionName));
    }

    String serviceName = asClient.getServiceName();
    String protocolPrefix = region.hasHttpsEndpoint(serviceName) ? "https://" : "http://";
    return protocolPrefix + region.getServiceEndpoint(serviceName);
  }
} public class AmazonEC2ClientProvider
    extends AbstractConfiguredOnceClientProvider<AmazonEC2AsyncClient> {
  private static final Logger LOG = LoggerFactory.getLogger(AmazonEC2ClientProvider.class);

  /**
   * Creates an Amazon EC2 client provider with the specified parameters.
   *
   * @param awsCredentialsProvider the AWS credentials provider
   * @param clientConfiguration    the client configuration
   */
  public AmazonEC2ClientProvider(
      AWSCredentialsProvider awsCredentialsProvider,
      ClientConfiguration clientConfiguration) {
    super(awsCredentialsProvider, clientConfiguration);
  }

  /**
   * {@inheritDoc}.
   */
  @Override
  protected AmazonEC2AsyncClient doConfigure(
      Configured configuration,
      PluginExceptionConditionAccumulator accumulator,
      LocalizationContext providerLocalizationContext,
      boolean verify) {

    AmazonEC2AsyncClient client = new AmazonEC2AsyncClient(awsCredentialsProvider, clientConfiguration);

    try {
      String regionEndpoint =
          configuration.getConfigurationValue(REGION_ENDPOINT, providerLocalizationContext);
      if (regionEndpoint != null) {
        LOG.info("<< Using configured region endpoint for EC2 client: {}", regionEndpoint);
      } else {
        String region = configuration.getConfigurationValue(REGION, providerLocalizationContext);
        regionEndpoint = getEndpointForRegion(client, region);
        verify = false;
      }
      client.setEndpoint(regionEndpoint);

      if (verify) {
        // Attempt to use client, to validate credentials and connectivity
        client.describeRegions();
      }
    } catch (AmazonClientException e) {
      AWSSecurityTokenServiceAsyncClient stsClient = new AWSSTSClientProvider(awsCredentialsProvider,
          clientConfiguration) .doConfigure(configuration, accumulator, providerLocalizationContext, verify);
      throw AWSExceptions.propagate(stsClient, e);
    } catch (IllegalArgumentException e) {
      accumulator.addError(REGION.unwrap().getConfigKey(), e.getMessage());
    }

    return client;
  }

  private static String getEndpointForRegion(AmazonEC2Client client, String regionName) {
    requireNonNull(client, "client is null");
    requireNonNull(regionName, "regionName is null");

    LOG.info(">> Describing all regions to find endpoint for '{}'", regionName);

    DescribeRegionsResult result = client.describeRegions();
    List<String> regions = Lists.newArrayListWithExpectedSize(result.getRegions().size());

    for (Region candidate : result.getRegions()) {
      regions.add(candidate.getRegionName());

      if (candidate.getRegionName().equals(regionName)) {
        LOG.info("<< Found endpoint '{}' for region '{}'", candidate.getEndpoint(), regionName);

        return candidate.getEndpoint();
      }
    }

    throw new IllegalArgumentException(String.format("Unable to find an endpoint for region '%s'. "
        + "Choose one of the following regions: %s", regionName, Joiner.on(", ").join(regions)));
  }
} public class AmazonIdentityManagementClientProvider
    extends AbstractConfiguredOnceClientProvider<AmazonIdentityManagementClient> {
  private static final Logger LOG = LoggerFactory.getLogger(AmazonIdentityManagementClientProvider.class);

  /**
   * Creates an Amazon identity management client provider with the specified parameters.
   *
   * @param awsCredentialsProvider the AWS credentials provider
   * @param clientConfiguration    the client configuration
   */
  public AmazonIdentityManagementClientProvider(
      AWSCredentialsProvider awsCredentialsProvider,
      ClientConfiguration clientConfiguration) {
    super(awsCredentialsProvider, clientConfiguration);
  }

  /**
   * {@inheritDoc}.
   */
  @Override
  protected AmazonIdentityManagementClient doConfigure(
      Configured configuration,
      PluginExceptionConditionAccumulator accumulator,
      LocalizationContext providerLocalizationContext,
      boolean verify) {

    AmazonIdentityManagementClient client = new AmazonIdentityManagementClient(
        awsCredentialsProvider, clientConfiguration);

    try {
      String iamEndpoint =
          configuration.getConfigurationValue(IAM_ENDPOINT, providerLocalizationContext);
      if (iamEndpoint != null) {
        LOG.info("<< Using configured IAM endpoint: {}", iamEndpoint);
        client.setEndpoint(iamEndpoint);
      }
      // else use the single default endpoint for all of AWS (outside GovCloud)

      if (verify) {
        // Attempt to use client, to validate credentials and connectivity
        try {
          client.getInstanceProfile(new GetInstanceProfileRequest().withInstanceProfileName("test"));
        } catch (NoSuchEntityException e) {
        /* call succeeded */
        }
      }

    } catch (AmazonClientException e) {
      AWSSecurityTokenServiceAsyncClient stsClient = new AWSSTSClientProvider(awsCredentialsProvider,
          clientConfiguration) .doConfigure(configuration, accumulator, providerLocalizationContext, verify);
      throw AWSExceptions.propagate(stsClient, e);
    } catch (IllegalArgumentException e) {
      accumulator.addError(IAM_ENDPOINT.unwrap().getConfigKey(), e.getMessage());
    }

    return client;
  }

  @Override
  protected boolean isEquals(Configured lhs, Configured rhs, LocalizationContext localizationContext) {
    boolean isEqual = super.isEquals(lhs, rhs, localizationContext);
    isEqual = isEqual || isEqualConfigValue(lhs, rhs, localizationContext, IAM_ENDPOINT);
    return isEqual;
  }

  private boolean isEqualConfigValue(Configured lhs, Configured rhs, LocalizationContext localizationContext,
      ConfigurationPropertyToken configToken) {
    String lhsValue = lhs.getConfigurationValue(configToken, localizationContext);
    String rhsValue = rhs.getConfigurationValue(configToken, localizationContext);
    return Objects.equals(lhsValue, rhsValue);
  }
} @SuppressWarnings({"PMD.TooManyStaticImports", "Guava"})
public class EC2InstanceTemplate extends ComputeInstanceTemplate {

  /**
   * A base 64 encoder for user data.
   */
  private static final BaseEncoding BASE64 = BaseEncoding.base64();

  /**
   * A splitter for comma-separated lists.
   */
  public static final Splitter CSV_SPLITTER = Splitter.on(",").trimResults().omitEmptyStrings();

  /**
   * The list of configuration properties (including inherited properties).
   */
  private static final List<ConfigurationProperty> CONFIGURATION_PROPERTIES =
      ConfigurationPropertiesUtil.merge(
          ComputeInstanceTemplate.getConfigurationProperties(),
          ConfigurationPropertiesUtil.asConfigurationPropertyList(EC2InstanceTemplateConfigurationPropertyToken.values())
      );

  /**
   * Returns the list of configuration properties for creating an EC2 instance template,
   * including inherited properties.
   *
   * @return the list of configuration properties for creating an EC2 instance template,
   * including inherited properties
   */
  public static List<ConfigurationProperty> getConfigurationProperties() {
    return CONFIGURATION_PROPERTIES;
  }

  /**
   * EC2 compute instance configuration properties.
   */
  // Fully qualifying class name due to compiler bug
  public enum EC2InstanceTemplateConfigurationPropertyToken
      implements com.cloudera.director.spi.v2.model.ConfigurationPropertyToken {

    /**
     * Whether instances created from this template are part of an EC2 Auto Scaling group (ASG).
     *
     * @see <a href="http://docs.aws.amazon.com/autoscaling/ec2/userguide/AutoScalingGroup.html">Auto Scaling Groups</a>
     */
    USE_ASG(new SimpleConfigurationPropertyBuilder()
        .configKey(ComputeInstanceTemplateConfigurationPropertyToken.AUTOMATIC.unwrap().getConfigKey())
        .name("Use Auto Scaling Group (ASG)")
        .required(false)
        .widget(ConfigurationProperty.Widget.CHECKBOX)
        .defaultDescription(
            "Whether to use EC2 Auto Scaling groups.<br />" +
                "<a target='_blank' href='http://docs.aws.amazon.com/autoscaling/ec2/userguide/AutoScalingGroup.html'>More Information</a>"
        )
        .build()),

    /**
     * Whether to allow AWS Auto Scaling to automatically perform certain operations on Auto Scaling
     * instances, such as replacing those that fail health checks or rebalancing availability zones.
     *
     * @see <a href="http://docs.aws.amazon.com/autoscaling/ec2/userguide/as-suspend-resume-processes.html">Suspending and Resuming Scaling Processes</a>
     */
    AUTOMATIC_INSTANCE_PROCESSING(new SimpleConfigurationPropertyBuilder()
        .configKey("enableAutomaticInstanceProcesing")
        .name("Enable automatic instance processing")
        .defaultValue("false")
        .type(Property.Type.BOOLEAN)
        .defaultDescription(
            "Whether to allow AWS Auto Scaling to automatically perform certain operations on " +
                "Auto Scaling instances, such as replacing those that fail health checks or " +
                "rebalancing availability zones.<br />" +
                "<a target='_blank' href='http://docs.aws.amazon.com/autoscaling/ec2/userguide/as-suspend-resume-processes.html'>More Information</a>"
        ).widget(ConfigurationProperty.Widget.CHECKBOX)
        .hidden(true)
        .build()),

    /**
     * <p>The availability zone.</p>
     * <p>Multiple availability zones are linked together by high speed low latency connections.
     * Each zone is a distinct failure domain.</p>
     *
     * @see <a href="http://docs.aws.amazon.com/AWSEC2/latest/UserGuide/using-regions-availability-zones.html">Regions and Availability Zones</a>
     */
    AVAILABILITY_ZONE(new SimpleConfigurationPropertyBuilder()
        .configKey("availabilityZone")
        .name("Availability zone")
        .widget(ConfigurationProperty.Widget.OPENLIST)
        .defaultDescription("The availability zone.")
        .hidden(true)
        .build()),

    /**
     * Whether to enable EBS Optimization.
     *
     * @see <a href="http://docs.aws.amazon.com/AWSEC2/latest/UserGuide/EBSOptimized.html">EBS–Optimized Instances</a>
     */
    EBS_OPTIMIZED(new SimpleConfigurationPropertyBuilder()
        .configKey("ebsOptimized")
        .name("EBS Optimized")
        .type(Property.Type.BOOLEAN)
        .defaultValue("false")
        .widget(ConfigurationProperty.Widget.CHECKBOX)
        .defaultDescription(
            "Specify whether to enable EBS Optimized I/O. This optimization isn't available with all instance types. " +
                "Some instance types are EBS Optimized by default regardless of this flag. Additional usage charges " +
                "may apply when using an EBS-optimized instance<br/>" +
                "<a target='_blank' href='http://docs.aws.amazon.com/AWSEC2/latest/UserGuide/EBSOptimized.html'>More Information</a>"
        )
        .build()),

    /**
     * Number of additional EBS volumes to mount.
     */
    EBS_VOLUME_COUNT(new SimpleConfigurationPropertyBuilder()
        .configKey("ebsVolumeCount")
        .name("EBS Volume Count")
        .defaultValue("0")
        .defaultDescription(
            "The number of additional EBS volumes to mount. Cloudera Altus Director will create and attach these volumes to the " +
                "provisioned instance. These added volumes will be deleted when the instance is terminated from Director. <br />" +
                "<a target='_blank' href='http://docs.aws.amazon.com/AWSEC2/latest/UserGuide/EBSVolumes.html'>More Information</a>"
        ).widget(ConfigurationProperty.Widget.NUMBER)
        .type(Property.Type.INTEGER)
        .build()),

    /**
     * The size of the additional EBS volume in GiB.
     */
    EBS_VOLUME_SIZE_GIB(new SimpleConfigurationPropertyBuilder()
        .configKey("ebsVolumeSizeGiB")
        .name("EBS Volume Size (GiB)")
        .defaultValue("500")
        .defaultDescription(
            "The size of the additional EBS volume(s) in GiB."
        ).widget(ConfigurationProperty.Widget.NUMBER)
        .type(Property.Type.INTEGER)
        .build()),

    /**
     * The volume type of the additional EBS volumes.
     */
    EBS_VOLUME_TYPE(new SimpleConfigurationPropertyBuilder()
        .configKey("ebsVolumeType")
        .name("EBS Volume Type")
        .addValidValues("st1", "sc1", "gp2", "io1")
        .defaultValue("st1")
        .defaultDescription(
            "The EBS volume type for the additional EBS volumes. Supported volumes are Throughput Optimized HDD (st1), " +
                "Cold HDD (sc1), General Purpose SSD (gp2) and Provisioned IOPS SSD (io1) <br />" +
                "<a target='_blank' href='http://docs.aws.amazon.com/AWSEC2/latest/UserGuide/EBSVolumeTypes.html'>More Information</a>"
        ).widget(ConfigurationProperty.Widget.OPENLIST)
        .build()),

    /**
     * The number of I/O operations per second (IOPS) when using io1 volume types.
     */
    EBS_IOPS(new SimpleConfigurationPropertyBuilder()
        .configKey("ebsIops")
        .name("EBS IOPS Count")
        .defaultDescription(
            "Only valid for Provisioned IOPS (io1) SSD volumes. The number of I/O operations per second (IOPS) to " +
                "provision for the volume."
        ).widget(ConfigurationProperty.Widget.NUMBER)
        .type(Property.Type.INTEGER)
        .build()),

    /**
     * Configurations for extra EBS volumes to mount, this allows for heterogeneous volumes.
     * Note that this is a temporary solution and this configuration configuration property
     * may be removed in the future.
     */
    SYSTEM_DISKS(new SimpleConfigurationPropertyBuilder()
        .configKey("systemDisks")
        .name("Configurations for extra EBS volumes")
        .widget(ConfigurationProperty.Widget.OPENMULTI)
        .hidden(true)
        .defaultDescription(
            "Configurations for extra EBS volumes"
        )
        .build()),

    /**
     * Whether to enable ebs encryption of the additional EBS volumes.
     *
     * @see <a href="http://docs.aws.amazon.com/AWSEC2/latest/UserGuide/EBSEncryption.html">EBS Encryption</a>
     */
    ENCRYPT_ADDITIONAL_EBS_VOLUMES(new SimpleConfigurationPropertyBuilder()
        .configKey("enableEbsEncryption")
        .name("Enable EBS Encryption")
        .defaultValue("false")
        .type(Property.Type.BOOLEAN)
        .defaultDescription(
            "Whether to enable encryption for the additional EBS volumes. Note that encryption does not apply to the root volume.<br />" +
                "<a target='_blank' href='http://docs.aws.amazon.com/AWSEC2/latest/UserGuide/EBSEncryption.html'>More Information</a>"
        ).widget(ConfigurationProperty.Widget.CHECKBOX)
        .build()),

    /**
     * The CMK to use for encryption of the additional EBS volumes.
     *
     * @see <a href="http://docs.aws.amazon.com/AWSEC2/latest/UserGuide/EBSEncryption.html">EBS Encryption</a>
     */
    EBS_KMS_KEY_ID(new SimpleConfigurationPropertyBuilder()
        .configKey("ebsKmsKeyId")
        .name("EBS KMS Key ID")
        .defaultDescription(
            "The full ARN of the KMS CMK to use when encrypting volumes. If encryption is enabled and this " +
                "is blank, the default CMK will be used. Note that encryption does not apply to the root volume.<br />" +
                "<a target='_blank' href='http://docs.aws.amazon.com/AWSEC2/latest/UserGuide/EBSEncryption.html'>More Information</a>"
        ).build()),

    /**
     * Whether to allocate EBS volumes separately when using non-default KMS Key ID.
     */
    ALLOCATE_EBS_SEPARATELY(new SimpleConfigurationPropertyBuilder()
        .configKey("allocateEbsSeparately")
        .name("Allocate EBS volumes separately")
        .defaultValue("false")
        .type(Property.Type.BOOLEAN)
        .defaultDescription(
            "Whether to allocate EBS volumes separately when using non-default KMS Key ID. This should" +
                "only be enabled on certain regions (like GovCloud) that don't allow specifying EBS KMS" +
                "Key ID as part of the instance launch request."
        ).widget(ConfigurationProperty.Widget.CHECKBOX)
        .hidden(true)
        .build()),

    /**
     * Name of the IAM instance profile.
     *
     * @see <a href="http://docs.aws.amazon.com/AWSEC2/latest/UserGuide/iam-roles-for-amazon-ec2.html">IAM Roles</a>
     */
    IAM_PROFILE_NAME(new SimpleConfigurationPropertyBuilder()
        .configKey("iamProfileName")
        .name("IAM profile name")
        .defaultDescription(
            "The IAM instance profile name.<br />" +
                "<a target='_blank' href='http://docs.aws.amazon.com/AWSEC2/latest/UserGuide/iam-roles-for-amazon-ec2.html'>More Information</a>"
        ).build()),

    /**
     * The image (AMI) ID.
     */
    IMAGE(new SimpleConfigurationPropertyBuilder()
        .configKey(ComputeInstanceTemplateConfigurationPropertyToken.IMAGE.unwrap().getConfigKey())
        .name("Image (AMI) ID")
        .required(true)
        .widget(ConfigurationProperty.Widget.OPENLIST)
        .defaultDescription(
            "Must begin with 'ami-'.<br/><br/>" +
                "To learn more about which AMIs should be used, please see the " +
                "<a target='_blank' href='http://www.cloudera.com/content/cloudera/en/documentation/cloudera-director/latest/topics/director_deployment_requirements.html'>Cloudera Altus Director Deployment Requirements Guide</a> " +
                "about what operating systems when deploying a Cloudera Manager or CDH Cluster.<br/><br/>" +
                "See the AWS documentation on " +
                "<a target='_blank' href='https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/AMIs.html'>Amazon Machine Images</a> " +
                "for more information."
        ).defaultPlaceholder("Enter or select an image")
        .defaultErrorMessage("Image (AMI) ID is mandatory")
        .build()),

    /**
     * Name of the key pair to use for new instances.
     *
     * @see <a href="http://docs.aws.amazon.com/AWSEC2/latest/UserGuide/ec2-key-pairs.html">EC2 Key Pairs</a>
     */
    KEY_NAME(new SimpleConfigurationPropertyBuilder()
        .configKey("keyName")
        .name("Key name")
        .required(false)
        .widget(ConfigurationProperty.Widget.TEXT)
        .defaultDescription(
            "The EC2 key pair ID.<br />" +
                "<a target='_blank' href='http://docs.aws.amazon.com/AWSEC2/latest/UserGuide/ec2-key-pairs.html'>More Information</a>"
        ).hidden(true)
        .build()),

    /**
     * The placement group name for new instances.
     *
     * @see <a href="http://docs.aws.amazon.com/AWSEC2/latest/UserGuide/placement-groups.html">Placement Groups</a>
     */
    PLACEMENT_GROUP(new SimpleConfigurationPropertyBuilder()
        .configKey("placementGroup")
        .name("Placement group")
        .defaultDescription("The placement group name should be defined here. " +
            "This setting allows users to launch the desired instance in a defined placement group. " +
            "Instances using placement groups will deploy in a specific zone within a region which helps network performance. <br/>" +
            "<a target='_blank' href='http://docs.aws.amazon.com/AWSEC2/latest/UserGuide/placement-groups.html'>More information</a>")
        .build()),

    /**
     * Size of the root partition in GBs.
     */
    ROOT_VOLUME_SIZE_GB(new SimpleConfigurationPropertyBuilder()
        .configKey("rootVolumeSizeGB")
        .name("Root volume size (GB)")
        .defaultValue("50")
        .defaultDescription(
            "Specify a size for the root volume (in GB). " +
                "Cloudera Altus Director will automatically expand the filesystem so that you can use all " +
                "the available disk space for your application.<br />" +
                "<a target='_blank' href='http://docs.aws.amazon.com/AWSEC2/latest/UserGuide/storage_expand_partition.html'>More Information</a>"
        ).widget(ConfigurationProperty.Widget.NUMBER)
        .type(Property.Type.INTEGER)
        .build()),

    /**
     * The type of EBS volume to use for the root drive.
     *
     * @see <a href="http://aws.amazon.com/ebs/details/">EBS</a>
     */
    ROOT_VOLUME_TYPE(new SimpleConfigurationPropertyBuilder()
        .configKey("rootVolumeType")
        .name("Root volume type")
        .defaultValue("gp2")
        .widget(ConfigurationProperty.Widget.OPENLIST)
        .defaultDescription(
            "Specify the type of the EBS volume used for the root partition. The choices for AWS are gp2 and standard. Defaults to gp2.<br/>" +
                "<a target='_blank' href='http://docs.aws.amazon.com/AWSEC2/latest/UserGuide/EBSVolumeTypes.html'>More Information</a>"
        ).addValidValues(
            "gp2",
            "standard")
        .build()),

    /**
     * The IDs of the VPC security groups (comma separated).
     *
     * @see <a href="http://docs.aws.amazon.com/AmazonVPC/latest/UserGuide/VPC_SecurityGroups.html">Security Groups</a>
     */
    SECURITY_GROUP_IDS(new SimpleConfigurationPropertyBuilder()
        .configKey("securityGroupsIds")
        .name("Security group IDs")
        .widget(ConfigurationProperty.Widget.OPENMULTI)
        .required(true)
        .defaultDescription(
            "Specify the list of security group IDs. Must begin with 'sg-'.<br />" +
                "<a target='_blank' href='http://docs.aws.amazon.com/AmazonVPC/latest/UserGuide/VPC_SecurityGroups.html'>More Information</a>"
        ).defaultErrorMessage("VPC security group IDs are mandatory")
        .build()),

    /**
     * Whether to use Spot Instances. Default is <code>false</code>.
     *
     * @see <a href="http://aws.amazon.com/ec2/spot/">Spot Instances</a>
     */
    USE_SPOT_INSTANCES(new SimpleConfigurationPropertyBuilder()
        .configKey("useSpotInstances")
        .name("Use Spot Instances")
        .widget(ConfigurationProperty.Widget.CHECKBOX)
        .defaultValue("false")
        .type(Property.Type.BOOLEAN)
        .defaultDescription("Whether to use Spot Instances. " +
            "Since Spot Instances can be terminated unexpectedly if the EC2 Spot price increases, " +
            "they should be used only for workers, and not for nodes that must be reliable, " +
            "such as masters and data nodes.<br />" +
            "<a target='_blank' href='http://aws.amazon.com/ec2/spot/'>More Information</a>")
        .build()),

    /**
     * Spot bid in USD/hr.
     *
     * @deprecated Use {@link #SPOT_PRICE_USD_PER_HR} instead.
     */
    SPOT_BID_USD_PER_HR(new SimpleConfigurationPropertyBuilder()
        .configKey("spotBidUSDPerHr")
        .name("Spot bid (USD/hr)")
        .defaultDescription(
            "Specify a Spot bid (in USD/hr). " +
                "This property is deprecated in favor of spotPriceUSDPerHr, but still works as " +
                "an alternative name for spotPriceUSDPerHr. This property will be removed in a " +
                "future plugin release."
        ).widget(ConfigurationProperty.Widget.NUMBER)
        .type(Property.Type.DOUBLE)
        .hidden(true)
        .build()),

    /**
     * Spot price in USD/hr.
     */
    SPOT_PRICE_USD_PER_HR(new SimpleConfigurationPropertyBuilder()
        .configKey("spotPriceUSDPerHr")
        .name("Spot price (USD/hr)")
        .defaultDescription(
            "Specify a Spot price (in USD/hr). " +
                "The Spot price is the maximum amount you are willing to pay per hour for each Spot Instance. " +
                "When your Spot price exceeds the Spot price set by EC2, then EC2 fulfills your request, " +
                "if capacity is available. Spot Instances will be terminated if the Spot price set by EC2 " +
                "rises above your Spot price."
        ).widget(ConfigurationProperty.Widget.NUMBER)
        .type(Property.Type.DOUBLE)
        .build()),

    /**
     * The required duration for the Spot instances.
     *
     * @see <a href='http://docs.aws.amazon.com/AWSEC2/latest/UserGuide/spot-requests.html#fixed-duration-spot-instances'>Fixed Duration Spot Instances</a>
     */
    BLOCK_DURATION_MINUTES(new SimpleConfigurationPropertyBuilder()
        .configKey("blockDurationMinutes")
        .name("Spot Block Duration (minutes)")
        .defaultDescription(
            "Specify the required duration for the Spot instances (also known as Spot blocks), " +
                "in minutes. This value must be a multiple of 60 (60, 120, 180, 240, 300, or 360). " +
                "Spot block instances will run for the predefined duration - in hourly increments " +
                "up to six hours in length - at a significant discount compared to On-Demand " +
                "pricing.<br />" +
                "<a target='_blank' href='http://docs.aws.amazon.com/AWSEC2/latest/UserGuide/spot-requests.html#fixed-duration-spot-instances'>More Information</a>"
        ).widget(ConfigurationProperty.Widget.LIST)
        .type(Property.Type.INTEGER)
        .addValidValues("60", "120", "180", "240", "300", "360")
        .build()),

    /**
     * The ID of the Amazon VPC subnet.
     *
     * @see <a href="http://docs.aws.amazon.com/AmazonVPC/latest/UserGuide/VPC_Subnets.html">Subnets</a>
     */
    SUBNET_ID(new SimpleConfigurationPropertyBuilder()
        .configKey("subnetId")
        .name("VPC subnet ID")
        .required(true)
        .defaultDescription(
            "The VPC subnet ID.<br />" +
                "<a target='_blank' href='http://docs.aws.amazon.com/AmazonVPC/latest/UserGuide/VPC_Subnets.html'>More Information</a>"
        ).defaultErrorMessage("VPC subnet ID is mandatory")
        .build()),

    /**
     * <p>The tenancy.</p>
     * <p>The tenancy of the instance (if the instance is running in a VPC).
     * An instance with dedicated tenancy runs on single-tenant hardware.</p>
     *
     * @see <a href="http://docs.aws.amazon.com/AWSEC2/latest/UserGuide/how-dedicated-hosts-work.html">Using Dedicated Hosts</a>
     */
    TENANCY(new SimpleConfigurationPropertyBuilder()
        .configKey("tenancy")
        .name("Tenancy")
        .defaultValue("default")
        .widget(ConfigurationProperty.Widget.LIST)
        .defaultDescription(
            "Instance tenancy should be defined here. This setting allows users to launch the desired instance in VPC with a defined tenancy. <br/>" +
                "<a target='_blank' href='http://docs.aws.amazon.com/AmazonVPC/latest/UserGuide/dedicated-instance.html'>More information</a>"
        ).addValidValues(
            "default",
            "dedicated"
        )
        .build()),

    /**
     * The instance type (e.g. t1.micro, m1.medium, etc.).
     */
    TYPE(new SimpleConfigurationPropertyBuilder()
        .configKey(ComputeInstanceTemplateConfigurationPropertyToken.TYPE.unwrap().getConfigKey())
        .name("Instance type")
        .required(true)
        .widget(ConfigurationProperty.Widget.OPENLIST)
        .defaultDescription(
            "Select an instance type as per the needs of the processes to be run on this instance.<br />" +
                "<a target='_blank' href='http://aws.amazon.com/ec2/instance-types/'>More Information</a>"
        ).defaultPlaceholder("Select an instance type")
        .defaultErrorMessage("Instance type is mandatory")
        .addValidValues(
            "t2.micro",
            "t2.small",
            "t2.medium",
            "t2.large",
            "t2.xlarge",
            "t2.2xlarge",
            "t3.micro",
            "t3.small",
            "t3.medium",
            "t3.large",
            "t3.xlarge",
            "t3.2xlarge",
            "m3.medium",
            "m3.large",
            "m3.xlarge",
            "m3.2xlarge",
            "m4.large",
            "m4.xlarge",
            "m4.2xlarge",
            "m4.4xlarge",
            "m4.10xlarge",
            "m4.16xlarge",
            "m5.large",
            "m5.xlarge",
            "m5.2xlarge",
            "m5.4xlarge",
            "m5.12xlarge",
            "m5.24xlarge",
            "c3.large",
            "c3.xlarge",
            "c3.2xlarge",
            "c3.4xlarge",
            "c3.8xlarge",
            "c4.large",
            "c4.xlarge",
            "c4.2xlarge",
            "c4.4xlarge",
            "c4.8xlarge",
            "c5.large",
            "c5.xlarge",
            "c5.2xlarge",
            "c5.4xlarge",
            "c5.9xlarge",
            "c5.18xlarge",
            "g2.2xlarge",
            "g2.8xlarge",
            "g3.4xlarge",
            "g3.8xlarge",
            "g3.16xlarge",
            "p2.xlarge",
            "p2.8xlarge",
            "p2.16xlarge",
            "p3.2xlarge",
            "p3.8xlarge",
            "p3.16xlarge",
            "r3.large",
            "r3.xlarge",
            "r3.2xlarge",
            "r3.4xlarge",
            "r3.8xlarge",
            "r4.large",
            "r4.xlarge",
            "r4.2xlarge",
            "r4.4xlarge",
            "r4.8xlarge",
            "r4.16xlarge",
            "r5.large",
            "r5.xlarge",
            "r5.2xlarge",
            "r5.4xlarge",
            "r5.12xlarge",
            "r5.24xlarge",
            "r5d.large",
            "r5d.xlarge",
            "r5d.2xlarge",
            "r5d.4xlarge",
            "r5d.12xlarge",
            "r5d.24xlarge",
            "z1d.large",
            "z1d.xlarge",
            "z1d.2xlarge",
            "z1d.3xlarge",
            "z1d.6xlarge",
            "z1d.12xlarge",
            "x1.16xlarge",
            "x1.32xlarge",
            "i2.xlarge",
            "i2.2xlarge",
            "i2.4xlarge",
            "i2.8xlarge",
            "i3.large",
            "i3.xlarge",
            "i3.2xlarge",
            "i3.4xlarge",
            "i3.8xlarge",
            "i3.16xlarge",
            "h1.2xlarge",
            "h1.4xlarge",
            "h1.8xlarge",
            "h1.16xlarge",
            "hs1.8xlarge",
            "d2.xlarge",
            "d2.2xlarge",
            "d2.4xlarge",
            "d2.8xlarge",
            "f1.2xlarge",
            "f1.16xlarge")
        .build()),

    /**
     * The user data, base64 encoded. Mutually exclusive with {@link #USER_DATA_UNENCODED}.
     *
     * @see <a href="http://docs.aws.amazon.com/AWSEC2/latest/UserGuide/ec2-instance-metadata.html">Instance Metadata and User Data</a>
     */
    USER_DATA(new SimpleConfigurationPropertyBuilder()
        .configKey("userData")
        .name("User data")
        .required(false)
        .widget(ConfigurationProperty.Widget.TEXT)
        .defaultDescription(
            "The user data, base64 encoded.<br />" +
                "<a target='_blank' href='http://docs.aws.amazon.com/AWSEC2/latest/UserGuide/ec2-instance-metadata.html'>More Information</a>"
        )
        .sensitive(true)
        .build()),

    /**
     * The user data, unencoded. Mutually exclusive with {@link #USER_DATA}.
     *
     * @see <a href="http://docs.aws.amazon.com/AWSEC2/latest/UserGuide/ec2-instance-metadata.html">Instance Metadata and User Data</a>
     */
    USER_DATA_UNENCODED(new SimpleConfigurationPropertyBuilder()
        .configKey("userDataUnencoded")
        .name("User data (unencoded)")
        .required(false)
        .widget(ConfigurationProperty.Widget.TEXTAREA)
        .defaultDescription(
            "The user data, unencoded.<br />" +
                "<a target='_blank' href='http://docs.aws.amazon.com/AWSEC2/latest/UserGuide/ec2-instance-metadata.html'>More Information</a>"
        )
        .sensitive(true)
        .build());

    /**
     * The configuration property.
     */
    private final ConfigurationProperty configurationProperty;

    /**
     * Creates a configuration property token with the specified parameters.
     *
     * @param configurationProperty the configuration property
     */
    EC2InstanceTemplateConfigurationPropertyToken(ConfigurationProperty configurationProperty) {
      this.configurationProperty = configurationProperty;
    }

    @Override
    public ConfigurationProperty unwrap() {
      return configurationProperty;
    }
  }

  /**
   * The instance type.
   */
  private final String type;

  /**
   * The image (AMI) ID.
   */
  private final String image;

  /**
   * The availability zone.
   */
  private final Optional<String> availabilityZone;

  /**
   * The placement group.
   */
  private final Optional<String> placementGroup;

  /**
   * The tenancy.
   */
  private final String tenancy;

  /**
   * The subnet ID.
   */
  private final String subnetId;

  /**
   * The security group IDs.
   */
  private final List<String> securityGroupIds;

  /**
   * The root volume size (in GB).
   */
  private final int rootVolumeSizeGB;

  /**
   * The root volume type.
   */
  private final String rootVolumeType;

  /**
   * Whether to enable EBS Optimization.
   */
  private final boolean ebsOptimized;

  /**
   * Number of additional EBS volumes to mount.
   */
  private final int ebsVolumeCount;

  /**
   * The size of the additional EBS volume in GiB.
   */
  private final int ebsVolumeSizeGiB;

  /**
   * The volume type of the additional EBS volumes.
   */
  private final String ebsVolumeType;

  /**
   * The number of I/O operations per second (IOPS) when using io1 volume types.
   */
  private final Optional<Integer> ebsIops;

  /**
   * Configuration for extra EBS volumes.
   */
  private final List<SystemDisk> systemDisks;

  /**
   * Whether to enable ebs encryption of the additional EBS volumes.
   */
  private final boolean enableEbsEncryption;

  /**
   * The optional CMK to use for encryption of the additional EBS volumes.
   */
  private final Optional<String> ebsKmsKeyId;

  /**
   * Whether to allocate EBS volumes separately when using non-default KMS Key ID.
   */
  private final boolean allocateEbsSeparately;

  /**
   * The optional IAM profile name.
   */
  private final Optional<String> iamProfileName;

  /**
   * The optional EC2 key pair name.
   */
  private final Optional<String> keyName;

  /**
   * Whether to use Spot Instances.
   */
  private final boolean useSpotInstances;

  /**
   * The Spot price, in USD/hr.
   */
  private final Optional<BigDecimal> spotPriceUSDPerHour;

  /**
   * The block duration in minutes for Spot block instances.
   */
  private final Optional<Integer> blockDurationMinutes;

  /**
   * The user data, base64 encoded.
   */
  private final Optional<String> userData;

  /**
   * Whether to allow AWS Auto Scaling to automatically perform certain operations on Auto Scaling
   * instances, such as replacing those that fail health checks or rebalancing availability zones.
   */
  private final boolean enableAutomaticInstanceProcesing;

  /**
   * Creates an EC2 instance template with the specified parameters.
   *
   * @param name                        the name of the template
   * @param configuration               the source of configuration
   * @param tags                        the map of tags to be applied to resources created from the template
   * @param providerLocalizationContext the parent provider localization context
   */
  public EC2InstanceTemplate(String name, Configured configuration, Map<String, String> tags,
      LocalizationContext providerLocalizationContext) {
    super(name, configuration, tags, providerLocalizationContext);
    LocalizationContext localizationContext = getLocalizationContext();
    this.type = getConfigurationValue(TYPE, localizationContext);
    this.image = getConfigurationValue(IMAGE, localizationContext);

    this.subnetId = getConfigurationValue(SUBNET_ID, localizationContext);
    this.securityGroupIds =
        CSV_SPLITTER.splitToList(getConfigurationValue(SECURITY_GROUP_IDS, localizationContext));

    this.availabilityZone =
        Optional.fromNullable(getConfigurationValue(AVAILABILITY_ZONE, localizationContext));
    this.placementGroup =
        Optional.fromNullable(getConfigurationValue(PLACEMENT_GROUP, localizationContext));
    this.tenancy = getConfigurationValue(TENANCY, localizationContext);

    this.rootVolumeSizeGB =
        Integer.parseInt(getConfigurationValue(ROOT_VOLUME_SIZE_GB, localizationContext));
    this.rootVolumeType = getConfigurationValue(ROOT_VOLUME_TYPE, localizationContext);

    this.ebsOptimized = Boolean.parseBoolean(getConfigurationValue(EBS_OPTIMIZED, localizationContext));

    this.ebsVolumeCount = Integer.parseInt(getConfigurationValue(EBS_VOLUME_COUNT, localizationContext));
    this.ebsVolumeSizeGiB = Integer.parseInt(getConfigurationValue(EBS_VOLUME_SIZE_GIB, localizationContext));
    this.ebsVolumeType = getConfigurationValue(EBS_VOLUME_TYPE, localizationContext);

    Optional<String> strEbsIops = Optional.fromNullable(getConfigurationValue(EBS_IOPS, localizationContext));
    this.ebsIops = strEbsIops.isPresent() ?
        Optional.of(Integer.parseInt(strEbsIops.get())) : Optional.absent();

    this.enableEbsEncryption =
        Boolean.parseBoolean(getConfigurationValue(ENCRYPT_ADDITIONAL_EBS_VOLUMES, localizationContext));
    this.ebsKmsKeyId = Optional.fromNullable(getConfigurationValue(EBS_KMS_KEY_ID, localizationContext));

    this.allocateEbsSeparately =
        Boolean.parseBoolean(getConfigurationValue(ALLOCATE_EBS_SEPARATELY, localizationContext));

    this.systemDisks =
        SystemDisk.parse(getConfigurationValue(SYSTEM_DISKS, localizationContext));

    this.iamProfileName =
        Optional.fromNullable(getConfigurationValue(IAM_PROFILE_NAME, localizationContext));
    this.keyName =
        Optional.fromNullable(getConfigurationValue(KEY_NAME, localizationContext));

    boolean useSpotInstances =
        Boolean.parseBoolean(getConfigurationValue(USE_SPOT_INSTANCES, localizationContext));
    this.useSpotInstances = useSpotInstances;
    Optional<String> spotPriceUSDPerHourString =
        Optional.fromNullable(getConfigurationValue(SPOT_PRICE_USD_PER_HR, localizationContext))
            .or(Optional.fromNullable(getConfigurationValue(SPOT_BID_USD_PER_HR, localizationContext)));
    this.spotPriceUSDPerHour = useSpotInstances && spotPriceUSDPerHourString.isPresent()
        ? Optional.of(new BigDecimal(spotPriceUSDPerHourString.get()))
        : Optional.absent();
    String blockDurationMinutesString = Strings.emptyToNull(
        getConfigurationValue(BLOCK_DURATION_MINUTES, localizationContext));
    this.blockDurationMinutes = useSpotInstances && blockDurationMinutesString != null
        ? Optional.of(Integer.parseInt(blockDurationMinutesString))
        : Optional.absent();

    this.userData =
        Optional.fromNullable(getConfigurationValue(USER_DATA, localizationContext))
            .or(Optional.fromNullable(base64Encode(getConfigurationValue(USER_DATA_UNENCODED, localizationContext))));

    this.enableAutomaticInstanceProcesing =
        Boolean.parseBoolean(getConfigurationValue(AUTOMATIC_INSTANCE_PROCESSING, localizationContext));
  }

  private String base64Encode(String s) {
    if (s == null) {
      return null;
    }
    return BASE64.encode(s.getBytes(StandardCharsets.UTF_8));
  }

  /**
   * Returns the instance type.
   *
   * @return the instance type
   */
  public String getType() {
    return type;
  }

  /**
   * Returns the image (AMI) ID.
   *
   * @return the image (AMI) ID
   */
  public String getImage() {
    return image;
  }

  /**
   * Returns the optional availability zone.
   *
   * @return the optional availability zone
   */
  public Optional<String> getAvailabilityZone() {
    return availabilityZone;
  }

  /**
   * Returns the optional placement group.
   *
   * @return the optional placement group
   */
  public Optional<String> getPlacementGroup() {
    return placementGroup;
  }

  /**
   * Returns tenancy.
   *
   * @return tenancy
   */
  public String getTenancy() {
    return tenancy;
  }

  /**
   * Returns the subnet ID.
   *
   * @return the subnet ID
   */
  public String getSubnetId() {
    return subnetId;
  }

  /**
   * Returns the security group IDs.
   *
   * @return the security group IDs
   */
  public List<String> getSecurityGroupIds() {
    return securityGroupIds;
  }

  /**
   * Returns the root volume size (in GB).
   *
   * @return the root volume size (in GB)
   */
  public int getRootVolumeSizeGB() {
    return rootVolumeSizeGB;
  }

  /**
   * Returns the root volume type.
   *
   * @return the root volume type
   */
  public String getRootVolumeType() {
    return rootVolumeType;
  }

  /**
   * Whether to enable EBS Optimization.
   *
   * @return whether to enable EBS Optimization
   */
  public boolean isEbsOptimized() {
    return ebsOptimized;
  }

  /**
   * Returns the EBS volume count.
   *
   * @return the EBS volume count
   */
  public int getEbsVolumeCount() {
    return ebsVolumeCount;
  }

  /**
   * Returns the EBS volume size in GiB.
   *
   * @return the EBS volume size in GiB
   */
  public int getEbsVolumeSizeGiB() {
    return ebsVolumeSizeGiB;
  }

  /**
   * Returns the EBS volume type.
   *
   * @return the EBS volume type.
   */
  public String getEbsVolumeType() {
    return ebsVolumeType;
  }

  /**
   * Returns the number of I/O operations per second (IOPS).
   *
   * @return the number of I/O operations per second (IOPS).
   */
  public Optional<Integer> getEbsIops() {
    return ebsIops;
  }

  /**
   * Returns whether to enable EBS encryption.
   *
   * @return whether to enable EBS encryption.
   */
  public boolean isEnableEbsEncryption() {
    return enableEbsEncryption;
  }

  /**
   * Returns the optional CMK to use for encryption of the additional EBS volumes.
   *
   * @return the optional CMK to use for encryption of the additional EBS volumes.
   */
  public Optional<String> getEbsKmsKeyId() {
    return ebsKmsKeyId;
  }

  /**
   * Returns whether to allocate EBS volumes separately when using non-default KMS Key ID.
   *
   * @return whether to allocate EBS volumes separately when using non-default KMS Key ID
   */
  public boolean isAllocateEbsSeparately() {
    return allocateEbsSeparately;
  }

  /**
   * Returns the system disks.
   *
   * @return the system disk.
   */
  public List<SystemDisk> getSystemDisks() {
    return systemDisks;
  }

  /**
   * Returns the optional IAM profile name.
   *
   * @return the optional IAM profile name
   */
  public Optional<String> getIamProfileName() {
    return iamProfileName;
  }

  /**
   * Returns the optional EC2 key pair name.
   *
   * @return the optional EC2 key pair name
   */
  public Optional<String> getKeyName() {
    return keyName;
  }

  /**
   * Returns whether to use Spot Instances.
   *
   * @return whether to use Spot Instances
   */
  public boolean isUseSpotInstances() {
    return useSpotInstances;
  }

  /**
   * Returns the Spot price, in USD/hr.
   *
   * @return the Spot price, in USD/hr
   */
  public Optional<BigDecimal> getSpotPriceUSDPerHour() {
    return spotPriceUSDPerHour;
  }

  /**
   * Returns the Spot block duration, in minutes.
   *
   * @return the Spot block duration, in minutes
   */
  public Optional<Integer> getBlockDurationMinutes() {
    return blockDurationMinutes;
  }

  /**
   * Returns the user data, base64 encoded.
   *
   * @return the user data, base64 encoded
   */
  public Optional<String> getUserData() {
    return userData;
  }

  /**
   * Returns whether to allow AWS Auto Scaling to automatically perform certain operations on Auto Scaling
   * instances, such as replacing those that fail health checks or rebalancing availability zones.
   *
   * @return whether to allow AWS Auto Scaling to automatically perform certain operations on Auto Scaling
   * instances, such as replacing those that fail health checks or rebalancing availability zones
   */
  public boolean isEnableAutomaticInstanceProcesing() {
    return enableAutomaticInstanceProcesing;
  }
} @SuppressWarnings({"PMD.TooManyStaticImports", "Guava"})
public class EC2Provider extends AbstractComputeProvider<EC2Instance, EC2InstanceTemplate> {

  private static final Logger LOG = LoggerFactory.getLogger(EC2Provider.class);

  private static final Predicate<Entry<String, Instance>> ENTRY_VALUE_NOT_NULL =
      entry -> entry != null && entry.getValue() != null;

  /**
   * The provider configuration properties.
   */
  private static final List<ConfigurationProperty> CONFIGURATION_PROPERTIES =
      ConfigurationPropertiesUtil.asConfigurationPropertyList(
          EC2ProviderConfigurationPropertyToken.values());

  /**
   * EC2 only allows 95 instance status requests per batch.
   */
  private static final int MAX_INSTANCE_STATUS_REQUESTS = 95;

  /**
   * EC2 only allows 200 tag values to be filtered at a time.
   */
  private static final int MAX_TAG_FILTERING_REQUESTS = 200;

  @VisibleForTesting
  static final String DEVICE_TYPE_EBS = "ebs";

  /**
   * The resource provider ID.
   */
  public static final String ID = EC2Provider.class.getCanonicalName();

  /**
   * The resource provider metadata.
   */
  public static final ResourceProviderMetadata METADATA = SimpleResourceProviderMetadata.builder()
      .id(ID)
      .name("EC2 (Elastic Compute Cloud)")
      .description("AWS EC2 compute provider")
      .providerClass(EC2Provider.class)
      .providerConfigurationProperties(CONFIGURATION_PROPERTIES)
      .resourceTemplateConfigurationProperties(EC2InstanceTemplate.getConfigurationProperties())
      .resourceDisplayProperties(EC2Instance.getDisplayProperties())
      .build();

  /**
   * The default wait time to retrieve host key fingerprints.
   */
  private static final long DEFAULT_WAIT_FOR_HOST_KEY_FINGERPRINTS_MS = 6 * 60 * 1000; // 6 min

  /**
   * Instance allocation strategies.
   */
  private enum InstanceAllocationStrategy {
    ON_DEMAND {
      @Override
      InstanceAllocator allocator(EC2Provider ec2Provider,
          EC2InstanceTemplate template, Collection<String> virtualInstanceIds, int minCount) {

        // Tag EBS volumes if they were part of instance launch request
        boolean tagEbsVolumes = (template != null) &&
            (EBSAllocationStrategy.get(template) == EBSAllocationStrategy.AS_INSTANCE_REQUEST);

        return new OnDemandAllocator(ec2Provider.allocationHelper, ec2Provider.client, ec2Provider.stsClient,
            tagEbsVolumes, ec2Provider.useTagOnCreate,
            template, virtualInstanceIds, minCount);
      }
    },

    SPOT {
      @Override
      InstanceAllocator allocator(EC2Provider ec2Provider,
          EC2InstanceTemplate template, Collection<String> virtualInstanceIds, int minCount) {

        // Tag EBS volumes if they were part of instance launch request
        boolean tagEbsVolumes = (template != null) &&
            (EBSAllocationStrategy.get(template) == EBSAllocationStrategy.AS_INSTANCE_REQUEST);

        return new SpotGroupAllocator(
            ec2Provider.allocationHelper, ec2Provider.client, ec2Provider.stsClient,
            tagEbsVolumes, template, virtualInstanceIds, minCount);
      }
    },

    AUTO_SCALING_GROUP {
      @Override
      InstanceAllocator allocator(EC2Provider ec2Provider,
          EC2InstanceTemplate template, Collection<String> instanceIds, int minCount) {

        return new AutoScalingGroupAllocator(ec2Provider.allocationHelper,
            ec2Provider.client, ec2Provider.autoScalingClient, ec2Provider.stsClient,
            template, instanceIds, minCount);
      }
    };

    /**
     * Returns an instance allocator for the specified parameters.
     *
     * @param ec2Provider the EC2 provider
     * @param template    the instance template
     * @param instanceIds the instance IDs
     * @param minCount    the min count
     * @return the instance allocator
     */
    private static InstanceAllocator getInstanceAllocator(EC2Provider ec2Provider,
        EC2InstanceTemplate template, Collection<String> instanceIds, int minCount) {
      InstanceAllocationStrategy allocationStrategy;
      allocationStrategy = getInstanceAllocationStrategy(template);
      return allocationStrategy.allocator(ec2Provider,
          template, instanceIds, minCount);
    }

    /**
     * Returns the instance allocation strategy for the specified template.
     *
     * @param template the instance template
     * @return the corresponding instance allocation strategy
     */
    private static InstanceAllocationStrategy getInstanceAllocationStrategy(
        EC2InstanceTemplate template) {
      EC2Provider.InstanceAllocationStrategy allocationStrategy;
      if (template.isAutomatic()) {
        allocationStrategy = AUTO_SCALING_GROUP;
      } else if (template.isUseSpotInstances()) {
        allocationStrategy = SPOT;
      } else {
        allocationStrategy = ON_DEMAND;
      }
      return allocationStrategy;
    }

    abstract InstanceAllocator allocator(EC2Provider ec2Provider,
        EC2InstanceTemplate template, Collection<String> instanceIds, int minCount);
  }

  private enum EBSAllocationStrategy {
    NO_EBS_VOLUMES,
    AS_INSTANCE_REQUEST,
    AS_SEPARATE_REQUESTS;

    private static EBSAllocationStrategy get(EC2InstanceTemplate template) {
      if (template.getEbsVolumeCount() == 0 && template.getSystemDisks().size() == 0) {
        return NO_EBS_VOLUMES;
      }

      // Ideally we want to request EBS volumes as part of the instance launch request.
      // However due to AWS API limitations, requesting encrypted EBS volumes with a
      // user specified KMS key can not be done as part of instance launch for certain
      // regions (like GovCloud). In this scenario we allow the option to individually
      // create and attach each EBS volume separately after instance launch.

      if (requiresSeparateEbsAllocation(template)) {
        return AS_SEPARATE_REQUESTS;
      } else {
        return AS_INSTANCE_REQUEST;
      }
    }
  }

  private static boolean requiresSeparateEbsAllocation(EC2InstanceTemplate template) {
    return template.isAllocateEbsSeparately() && hasVolumesWithKmsKey(template);
  }

  private static boolean hasVolumesWithKmsKey(EC2InstanceTemplate template) {
    boolean hasVolumesWithKmsKey = (template.getEbsVolumeCount() != 0) ?
        template.getEbsKmsKeyId().isPresent() :
        template.getSystemDisks().get(0).getKmsKeyId() != null;

    for (SystemDisk systemDisk : template.getSystemDisks()) {
      boolean systemDiskUsingKmsKey = systemDisk.getKmsKeyId() != null;
      if (hasVolumesWithKmsKey && !systemDiskUsingKmsKey) {
        throw new IllegalArgumentException(
            "Either all volumes should be encrypted with KMS Key ID or none"
        );
      }
    }
    return hasVolumesWithKmsKey;
  }

  private final AmazonEC2AsyncClient client;
  private final AmazonIdentityManagementClient identityManagementClient;
  private final AWSKMSClient kmsClient;
  private final AWSSecurityTokenServiceAsyncClient stsClient;
  private final AmazonAutoScalingAsyncClient autoScalingClient;

  private final EphemeralDeviceMappings ephemeralDeviceMappings;
  private final EBSDeviceMappings ebsDeviceMappings;
  private final VirtualizationMappings virtualizationMappings;
  private final AWSFilters ec2Filters;
  private final NetworkRules networkRules;

  @VisibleForTesting
  final EC2TagHelper ec2TagHelper;

  @VisibleForTesting
  final AllocationHelperImpl allocationHelper;

  private final boolean associatePublicIpAddresses;
  private final boolean importKeyPairIfMissing;
  private final String keyNamePrefix;

  private final ConfigurationValidator resourceTemplateConfigurationValidator;

  private final ConsoleOutputExtractor consoleOutputExtractor;

  private final AWSTimeouts awsTimeouts;

  private final boolean useTagOnCreate;

  /**
   * Construct a new provider instance and validate all configurations.
   *
   * @param configuration                    the configuration
   * @param ephemeralDeviceMappings          the ephemeral device mappings
   * @param ebsDeviceMappings                the ebs device mappings
   * @param ebsMetadata                      the EBS metadata
   * @param virtualizationMappings           the virtualization mappings
   * @param awsFilters                       the AWS filters
   * @param awsTimeouts                      the AWS timeouts
   * @param customTagMappings                the custom tag mappings
   * @param networkRules                     the network rules
   * @param clientProvider                   the EC2 client provider
   * @param identityManagementClientProvider the IAM client provider
   * @param kmsClientProvider                the KMS client provider
   * @param cloudLocalizationContext         the parent cloud localization context
   * @throws UnrecoverableProviderException if an unrecoverable exception occurs communicating with
   *                                        the provider
   */
  public EC2Provider(
      Configured configuration,
      EphemeralDeviceMappings ephemeralDeviceMappings,
      EBSDeviceMappings ebsDeviceMappings,
      EBSMetadata ebsMetadata,
      VirtualizationMappings virtualizationMappings,
      AWSFilters awsFilters,
      AWSTimeouts awsTimeouts,
      CustomTagMappings customTagMappings,
      NetworkRules networkRules,
      ClientProvider<AmazonEC2AsyncClient> clientProvider,
      ClientProvider<AmazonAutoScalingAsyncClient> autoScalingClientProvider,
      ClientProvider<AmazonIdentityManagementClient> identityManagementClientProvider,
      ClientProvider<AWSKMSClient> kmsClientProvider,
      ClientProvider<AWSSecurityTokenServiceAsyncClient> stsClientProvider,
      boolean useTagOnCreate,
      LocalizationContext cloudLocalizationContext) {

    super(configuration, METADATA, cloudLocalizationContext);
    LocalizationContext localizationContext = getLocalizationContext();
    this.ephemeralDeviceMappings =
        requireNonNull(ephemeralDeviceMappings, "ephemeralDeviceMappings is null");
    this.ebsDeviceMappings =
        requireNonNull(ebsDeviceMappings, "ebsDeviceMappings is null");
    this.virtualizationMappings =
        requireNonNull(virtualizationMappings, "virtualizationMappings is null");
    this.ec2Filters = requireNonNull(awsFilters, "awsFilters").getSubfilters(ID);
    this.ec2TagHelper = new EC2TagHelper(customTagMappings);
    this.networkRules = requireNonNull(networkRules, "networkRules is null");

    PluginExceptionConditionAccumulator accumulator = new PluginExceptionConditionAccumulator();

    this.client = requireNonNull(clientProvider, "clientProvider is null")
        .getClient(configuration, accumulator, localizationContext, false);
    this.autoScalingClient = requireNonNull(
        autoScalingClientProvider, "autoScalingClientProvider is null")
        .getClient(configuration, accumulator, localizationContext, false);
    this.identityManagementClient = requireNonNull(
        identityManagementClientProvider, "identityManagementClientProvider is null")
        .getClient(configuration, accumulator, localizationContext, false);
    this.kmsClient = requireNonNull(kmsClientProvider, "kmsClientProvider is null")
        .getClient(configuration, accumulator, localizationContext, false);
    this.stsClient = requireNonNull(stsClientProvider, "stsClientProvider is null")
        .getClient(configuration, accumulator, localizationContext, false);

    if (accumulator.hasError()) {
      PluginExceptionDetails pluginExceptionDetails =
          new PluginExceptionDetails(accumulator.getConditionsByKey());
      throw new UnrecoverableProviderException("Provider initialization failed",
          pluginExceptionDetails);
    }

    this.associatePublicIpAddresses = Boolean.parseBoolean(
        getConfigurationValue(ASSOCIATE_PUBLIC_IP_ADDRESSES, localizationContext));

    this.importKeyPairIfMissing = Boolean.parseBoolean(
        getConfigurationValue(IMPORT_KEY_PAIR_IF_MISSING, localizationContext));
    this.keyNamePrefix = getConfigurationValue(KEY_NAME_PREFIX, localizationContext);

    this.awsTimeouts = awsTimeouts;

    this.allocationHelper = new AllocationHelperImpl();

    this.resourceTemplateConfigurationValidator =
        new CompositeConfigurationValidator(
            METADATA.getResourceTemplateConfigurationValidator(),
            new EC2InstanceTemplateConfigurationValidator(this, ebsMetadata),
            new EC2NetworkValidator(this)
        );

    this.consoleOutputExtractor = new ConsoleOutputExtractor();

    this.useTagOnCreate = useTagOnCreate;
  }

  /**
   * Returns the AWS EC2 client.
   *
   * @return the AWS EC2 client
   */
  public AmazonEC2Client getClient() {
    return client;
  }

  /**
   * Returns the AWS auto scaling client.
   *
   * @return the AWS auto scaling client
   */
  public AmazonAutoScalingAsyncClient getAutoScalingClient() {
    return autoScalingClient;
  }

  /**
   * Returns the AWS identity management client.
   *
   * @return the AWS identity management client
   */
  @SuppressWarnings("WeakerAccess")
  public AmazonIdentityManagementClient getIdentityManagementClient() {
    return identityManagementClient;
  }

  /**
   * Returns the AWS KMS client.
   *
   * @return the AWS KMS client
   */
  @SuppressWarnings("WeakerAccess")
  public AWSKMSClient getKmsClient() {
    return kmsClient;
  }

  /**
   * Returns the ephemeral device mappings.
   *
   * @return the ephemeral device mappings
   */
  public EphemeralDeviceMappings getEphemeralDeviceMappings() {
    return ephemeralDeviceMappings;
  }

  /**
   * Returns the virtualization mappings.
   *
   * @return the virtualization mappings
   */
  public VirtualizationMappings getVirtualizationMappings() {
    return virtualizationMappings;
  }

  /**
   * Returns the EC2 filters.
   *
   * @return the EC2 filters
   */
  @SuppressWarnings("WeakerAccess")
  public AWSFilters getEC2Filters() {
    return ec2Filters;
  }

  /**
   * Returns the EC2 tag helper.
   *
   * @return the EC2 tag helper
   */
  @SuppressWarnings("WeakerAccess")
  public EC2TagHelper getEC2TagHelper() {
    return ec2TagHelper;
  }

  /**
   * Returns the network rules.
   *
   * @return the network rules
   */
  @SuppressWarnings("WeakerAccess")
  public NetworkRules getNetworkRules() {
    return networkRules;
  }

  @Override
  public ConfigurationValidator getResourceTemplateConfigurationValidator() {
    return resourceTemplateConfigurationValidator;
  }

  @Override
  public Resource.Type getResourceType() {
    return EC2Instance.TYPE;
  }

  @Override
  public EC2InstanceTemplate createResourceTemplate(String name, Configured configuration,
      Map<String, String> tags) {

    LocalizationContext providerLocalizationContext = getLocalizationContext();
    LocalizationContext templateLocalizationContext =
        SimpleResourceTemplate.getTemplateLocalizationContext(providerLocalizationContext);
    configuration = enhanceTemplateConfiguration(name, configuration, templateLocalizationContext);

    return new EC2InstanceTemplate(name, configuration, tags, providerLocalizationContext);
  }

  @Override
  public Collection<EC2Instance> allocate(
      EC2InstanceTemplate template,
      Collection<String> instanceIds,
      int minCount)
      throws InterruptedException {

    // TODO: This should really be taken care of in the SPI validate command.
    // We are punting this for a future change that compels us to release a new version of SPI
    ec2TagHelper.validateTags(template.getTags());

    InstanceAllocator allocator = InstanceAllocationStrategy.getInstanceAllocator(
        this, template, instanceIds, minCount);

    Collection<EC2Instance> allocatedInstances = allocator.allocate();

    if (EBSAllocationStrategy.get(template) == EBSAllocationStrategy.AS_SEPARATE_REQUESTS) {
      if (allocatedInstances.size() > 0) {
        LOG.info(">> Allocating EBS volumes");
        Map<String, String> instanceIdPairs = allocatedInstances.stream().collect(
            Collectors.toMap(AbstractResource::getId, (i) -> i.unwrap().getInstanceId()));

        allocateEbsVolumes(template, instanceIdPairs, minCount);
      } else {
        LOG.info(">> Skipping EBS volume allocation since no instances were allocated");
      }
    }

    return allocatedInstances;
  }

  @Override
  @SuppressWarnings({"PMD.UnusedFormalParameter"})
  public void delete(EC2InstanceTemplate template, Collection<String> instanceIds)
      throws InterruptedException {

    InstanceAllocationStrategy instanceAllocationStrategy =
        (template == null)
            ? InstanceAllocationStrategy.ON_DEMAND
            : InstanceAllocationStrategy.getInstanceAllocationStrategy(template);
    InstanceAllocator allocator =
        instanceAllocationStrategy.allocator(this, template, instanceIds, 0);
    allocator.delete();
  }

  @Override
  public Map<String, Set<String>> getHostKeyFingerprints(EC2InstanceTemplate template,
      Collection<String> instanceIds)
      throws InterruptedException {

    Map<String, String> idsToCheck = isAutomatic(template)
        ? instanceIds.stream().collect(Collectors.toMap((id) -> id, (id) -> id))
        : Maps.newHashMap(getEC2InstanceIdsByVirtualInstanceId(instanceIds));

    Map<String, Set<String>> hostKeyFingerprints = Maps.newHashMapWithExpectedSize(idsToCheck.size());

    LOG.info("Waiting for EC2 console output to display its host key fingerprint for instance(s): {}",
        idsToCheck.keySet());

    Stopwatch stopwatch = Stopwatch.createStarted();
    do {
      Iterator<Entry<String, String>> it = idsToCheck.entrySet().iterator();
      while (it.hasNext()) {
        Entry<String, String> entry = it.next();
        String instanceId = entry.getKey();
        String ec2InstanceId = entry.getValue();

        GetConsoleOutputRequest consoleOutputRequest = new GetConsoleOutputRequest()
            .withInstanceId(ec2InstanceId);
        GetConsoleOutputResult result = client.getConsoleOutput(consoleOutputRequest);

        if (result.getOutput() != null) {
          String consoleOutput = result.getDecodedOutput();
          if (!consoleOutputExtractor.hasHostKeyFingerprintBlock(consoleOutput)) {
            LOG.debug("EC2 Console Output doesn't contain the host key fingerprint yet, retrying ...");
            continue;
          }

          Set<String> instanceHostKeyFingerprints
              = consoleOutputExtractor.getHostKeyFingerprints(result.getDecodedOutput());

          LOG.debug("Host key fingerprints for ID {} are {}", instanceId, instanceHostKeyFingerprints);
          hostKeyFingerprints.put(instanceId, instanceHostKeyFingerprints);
          it.remove();
        } else {
          LOG.debug("EC2 Console Output empty, retrying soon ...");
        }
      }

      if (idsToCheck.isEmpty()) {
        break;
      }

      TimeUnit.SECONDS.sleep(10);
    } while (stopwatch.elapsed(TimeUnit.MILLISECONDS) < DEFAULT_WAIT_FOR_HOST_KEY_FINGERPRINTS_MS);

    if (!idsToCheck.isEmpty()) {
      LOG.warn("Couldn't retrieve SSH host key fingerprints for following {} instance(s): {}",
          idsToCheck.size(), idsToCheck.keySet());
    }

    return hostKeyFingerprints;
  }

  /**
   * Creates and attaches EBS volumes to EC2 instances. This expects that the instances
   * have already been allocated. Instances that could not acquire the correct number of
   * EBS volumes will be terminated along with any leftover volumes. If the minimum
   * number of instances could not acquire EBS volumes, all the specified instances and
   * leftover volumes will be terminated.
   *
   * @param template        the EC2 instance template that contains EBS configurations
   * @param instanceIdPairs map from instance IDs to EC2 instance IDs to attach volumes to
   * @param minCount        the minimum number of instances that need EBS volumes attached
   * @throws InterruptedException if the process is interrupted
   */
  private void allocateEbsVolumes(EC2InstanceTemplate template, Map<String, String> instanceIdPairs,
      int minCount)
      throws InterruptedException {

    boolean success = false;

    Image image = getImage(template.getImage());
    Set<String> existingDeviceNames = getExistingDeviceNames(image.getBlockDeviceMappings());

    EBSAllocator ebsAllocator = new EBSAllocator(this.client, this.stsClient, this.awsTimeouts, ec2TagHelper,
        ebsDeviceMappings, existingDeviceNames, useTagOnCreate);

    List<InstanceEbsVolumes> instanceVolumes = ebsAllocator.createVolumes(template, instanceIdPairs);

    try {
      try {
        instanceVolumes = ebsAllocator.waitUntilVolumesAvailable(instanceVolumes);
        instanceVolumes = ebsAllocator.attachAndOptionallyTagVolumes(template, instanceVolumes);
      } finally {
        // Ensure that delete on termination is set for attached instances no matter what. It's possible
        // for the EBS volume/ attachment to time out but still succeed, which can lead to leaked EBS
        // volumes. This should ensure that the instances are not leaked accidentally.
        instanceVolumes = ebsAllocator.getUpdatedVolumeInfo(instanceVolumes); // Update volume states
        ebsAllocator.addDeleteOnTerminationFlag(instanceVolumes);
      }

      int successfulInstances = getSuccessfulInstanceCount(instanceVolumes);
      LOG.info("{} out of {} instances successfully acquired EBS volumes",
          successfulInstances, instanceIdPairs.size());

      if (successfulInstances < minCount) {
        LOG.warn("A minimum number of {} instances could not acquire EBS volumes, cleaning up by " +
            "deleting all allocated instances and volumes", minCount);
      } else {
        success = true;
      }
    } catch (InterruptedException e) {
      throw e;
    } catch (Exception e) {
      throw new UnrecoverableProviderException("Unexpected problem allocating EBS volumes", e);
    } finally {
      if (!success) {
        deleteAllInstancesAndVolumes(instanceVolumes, template, ebsAllocator);
      } else {
        deleteFailedInstancesAndVolumes(instanceVolumes, template, ebsAllocator);
      }
    }
  }

  /**
   * Terminates failed instances and their associated volumes. In this case a failed
   * instance is any instance that doesn't have all their volumes as ATTACHED.
   */
  private void deleteFailedInstancesAndVolumes(List<InstanceEbsVolumes> instanceEbsVolumesList,
      EC2InstanceTemplate template, EBSAllocator ebsAllocator) throws InterruptedException {

    Set<String> instancesToTerminate = Sets.newHashSet();

    nextInstanceVolume:
    for (InstanceEbsVolumes instanceEbsVolumes : instanceEbsVolumesList) {
      for (VolumeState status : instanceEbsVolumes.getVolumeStates().values()) {
        if (status != VolumeState.InUse) {
          deleteCreatedVolumes(instanceEbsVolumes, ebsAllocator);
          instancesToTerminate.add(instanceEbsVolumes.getInstanceId());
          continue nextInstanceVolume;
        }
      }
    }
    delete(template, instancesToTerminate);
  }


  /**
   * Terminates a list of instances and deletes their associated volumes.
   */
  private void deleteAllInstancesAndVolumes(List<InstanceEbsVolumes> instanceEbsVolumesList,
      EC2InstanceTemplate template, EBSAllocator ebsAllocator) throws InterruptedException {

    Set<String> instancesToTerminate = Sets.newHashSet();

    for (InstanceEbsVolumes instanceEbsVolumes : instanceEbsVolumesList) {
      deleteCreatedVolumes(instanceEbsVolumes, ebsAllocator);
      instancesToTerminate.add(instanceEbsVolumes.getInstanceId());
    }
    delete(template, instancesToTerminate);
  }

  private void deleteCreatedVolumes(InstanceEbsVolumes instanceEbsVolumes, EBSAllocator ebsAllocator)
      throws InterruptedException {
    Map<String, VolumeState> volumesToDelete = Maps.newHashMap();
    for (Entry<String, VolumeState> volumeIdAndStatus
        : instanceEbsVolumes.getVolumeStates().entrySet()) {
      String volumeId = volumeIdAndStatus.getKey();
      VolumeState state = volumeIdAndStatus.getValue();

      if (!volumeId.startsWith(InstanceEbsVolumes.UNCREATED_VOLUME_ID)) {
        volumesToDelete.put(volumeId, state);
      }
    }
    ebsAllocator.deleteVolumes(volumesToDelete);
  }

  /**
   * Get the number of instances that have successfully acquired volumes. Success means
   * that all volumes associated with an instance are in the ATTACHED state.
   */
  private int getSuccessfulInstanceCount(List<InstanceEbsVolumes> instanceEbsVolumesList) {
    int count = 0;
    for (InstanceEbsVolumes instanceEbsVolumes : instanceEbsVolumesList) {
      boolean success = true;
      for (VolumeState state : instanceEbsVolumes.getVolumeStates().values()) {
        if (state != VolumeState.InUse) {
          success = false;
          break;
        }
      }
      if (success) count++;
    }
    return count;
  }

  /**
   * Get EBS volumes attached to the specified virtual instance id.
   *
   * @return list of ebs volumes
   */
  @VisibleForTesting
  List<Volume> getVolumes(String virtualInstanceId) {
    String ec2InstanceId = getOnlyElement(
        getEC2InstanceIdsByVirtualInstanceId(
            Collections.singletonList(virtualInstanceId)
        ).values()
    );

    InstanceAttribute instanceAttribute =
        describeInstanceAttribute(ec2InstanceId, InstanceAttributeName.BlockDeviceMapping);
    List<InstanceBlockDeviceMapping> blockDeviceMappings =
        instanceAttribute.getBlockDeviceMappings();

    List<String> volumeIds = Lists.newArrayList();
    for (InstanceBlockDeviceMapping mapping : blockDeviceMappings) {
      volumeIds.add(mapping.getEbs().getVolumeId());
    }

    DescribeVolumesResult volumeResults = client.describeVolumes(
        new DescribeVolumesRequest().withVolumeIds(volumeIds)
    );

    return volumeResults.getVolumes();
  }

  @Override
  public Collection<EC2Instance> find(final EC2InstanceTemplate template, Collection<String> instanceIds)
      throws InterruptedException {

    if (instanceIds.isEmpty()) {
      InstanceAllocator allocator = InstanceAllocationStrategy.getInstanceAllocator(
          this, template, instanceIds, 0);
      instanceIds = allocator.getInstanceIds();
    }

    LOG.debug("Finding instances {}", instanceIds);

    Collection<EC2Instance> ec2Instances = FluentIterable
        .from(allocationHelper.doFind(template, instanceIds))
        .transform(idToInstance -> {
          requireNonNull(idToInstance, "idToInstance is null");
          Instance instance = idToInstance.getValue();
          fillMissingProperties(instance);
          return new EC2Instance(template, idToInstance.getKey(), instance);
        })
        .toList();

    LOG.debug("Found {} instances for {} instance IDs", ec2Instances.size(), instanceIds.size());
    return ec2Instances;
  }


  private void fillMissingProperties(Instance instance) {
    try {
      InstanceAttribute instanceAttribute = describeInstanceAttribute(
          instance.getInstanceId(), InstanceAttributeName.SriovNetSupport);
      String sriovNetSupport = instanceAttribute.getSriovNetSupport();
      instance.setSriovNetSupport(sriovNetSupport);
    } catch (AmazonServiceException e) {
      // In practice, users may not have appropriate IAM permission for
      // DescribeInstanceAttribute. We need to be more forgiving in those cases,
      // and simply leave a warning in the log here.
      LOG.warn("Could not fill missing properties. Failed to perform " +
          "DescribeInstanceAttribute action.", e);
    }
  }

  /**
   * Retrieves an instance attribute.
   *
   * @param instanceId instance ID
   * @param attribute  name of attribute to describe
   * @return attribute
   */
  @VisibleForTesting
  InstanceAttribute describeInstanceAttribute(String instanceId, InstanceAttributeName attribute) {
    DescribeInstanceAttributeRequest request = new DescribeInstanceAttributeRequest()
        .withInstanceId(instanceId)
        .withAttribute(attribute);

    return client.describeInstanceAttribute(request).getInstanceAttribute();
  }

  @Override
  public Map<String, InstanceState> getInstanceState(EC2InstanceTemplate template,
      Collection<String> instanceIds) {
    IdType idType = getIdType(template);

    Map<String, InstanceState> instanceStateByInstanceId =
        Maps.newHashMapWithExpectedSize(instanceIds.size());

    // Partition full requests into multiple batch requests, AWS limits
    // the total number of instance status requests you can make.
    List<List<String>> partitions =
        Lists.partition(Lists.newArrayList(instanceIds), MAX_INSTANCE_STATUS_REQUESTS);

    for (List<String> partition : partitions) {
      instanceStateByInstanceId.putAll(getBatchInstanceState(partition, idType));
    }

    return instanceStateByInstanceId;
  }

  @Override
  protected Configured enhanceTemplateConfiguration(String name, Configured configuration,
      LocalizationContext templateLocalizationContext) {
    // Add the key name to the configuration if possible.
    String privateKeyString =
        configuration.getConfigurationValue(SSH_JCE_PRIVATE_KEY, templateLocalizationContext);
    if (privateKeyString != null) {
      String publicKeyString =
          configuration.getConfigurationValue(SSH_JCE_PUBLIC_KEY, templateLocalizationContext);
      configuration = addKeyName(configuration, templateLocalizationContext, privateKeyString,
          publicKeyString);
    } else {
      LOG.warn("No private key fingerprint specified for template {}", name);
    }
    return configuration;
  }

  /**
   * Adds the AWS key name corresponding to a private key to the given
   * configuration.
   *
   * @param configuration    the configuration to be enhanced
   * @param privateKeyString private key, in serialized form
   * @return the enhanced configuration
   * @throws IllegalArgumentException if the key could not be deserialized, or if no key known to
   *                                  AWS matches this key's fingerprint
   */
  private Configured addKeyName(Configured configuration,
      LocalizationContext templateLocalizationContext,
      String privateKeyString, String publicKeyString) {
    PrivateKey privateKey;
    PublicKey publicKey;
    try {
      KeySerialization keySerialization = new KeySerialization();
      privateKey = keySerialization.deserializePrivateKey(privateKeyString);
      if (publicKeyString != null) {
        publicKey = keySerialization.deserializePublicKey(publicKeyString);
      } else {
        publicKey = null;
      }
    } catch (IOException e) {
      throw new IllegalArgumentException(
          "Unable to deserialize private key from compute instance template", e);
    }
    String privateKeyFingerprint = getSha1Fingerprint(privateKey);
    String publicKeyFingerprint = getMd5Fingerprint(publicKey);
    String keyName = lookupKeyName(privateKeyFingerprint, publicKeyFingerprint);
    if (keyName == null) {
      if (importKeyPairIfMissing) {
        keyName = keyNamePrefix + publicKeyFingerprint;
        LOG.info("KeyPair not found. Adding public key to EC2 with key name : {}", keyName);
        try {
          //noinspection ConstantConditions
          client.importKeyPair(new ImportKeyPairRequest().withKeyName(keyName).withPublicKeyMaterial(
              BaseEncoding.base64().encode(publicKey.getEncoded())));
        } catch (AmazonEC2Exception e) {
          if (e.getErrorCode().equals("UnauthorizedOperation")) {
            String message = String.format("No private key in EC2 matches the " +
                "fingerprint %s. To auto register keys, " +
                "add ec2:ImportKeyPair permission", privateKeyFingerprint);
            throw new AmazonEC2Exception(message);
          }
          throw e;
        }
      } else {
        throw new IllegalArgumentException("No private key in EC2 matches the fingerprint " +
            privateKeyFingerprint);
      }
    } else {
      LOG.info("Found EC2 key name {} for fingerprint", keyName);
    }
    Map<String, String> configMap =
        Maps.newHashMap(configuration.getConfiguration(templateLocalizationContext));
    configMap.put(KEY_NAME.unwrap().getConfigKey(),
        keyName);
    return new SimpleConfiguration(configMap);
  }

  /**
   * Gets the SHA-1 digest of the private key bits. This is used as a
   * fingerprint by AWS for a key pair generated by AWS.
   *
   * @param privateKey private key
   * @return private key fingerprint, as a lowercase hex string with colons
   * @throws IllegalStateException if the SHA-1 digest algorithm is not
   *                               available
   */
  @VisibleForTesting
  static String getSha1Fingerprint(PrivateKey privateKey) {
    MessageDigest digest;
    try {
      digest = MessageDigest.getInstance("SHA-1");
    } catch (NoSuchAlgorithmException e) {
      throw new IllegalStateException(e);
    }
    return toColonSeparatedHexString(digest, privateKey.getEncoded());
  }

  /**
   * Gets the MD5 digest of the public key bits. This is used as a fingerprint
   * by AWS for an imported key pair.
   *
   * @param publicKey public key
   * @return public key fingerprint, as a lowercase hex string with colons,
   * or null if publicKey is null
   */
  @VisibleForTesting
  static String getMd5Fingerprint(PublicKey publicKey) {
    if (publicKey == null) {
      return null;
    }
    MessageDigest digest;
    try {
      digest = MessageDigest.getInstance("MD5");
    } catch (NoSuchAlgorithmException e) {
      throw new IllegalStateException(e);
    }
    return toColonSeparatedHexString(digest, publicKey.getEncoded());
  }

  private static String toColonSeparatedHexString(MessageDigest digest, byte[] key) {
    byte[] fingerprintBytes = digest.digest(key);
    String binaryHexString =
        DatatypeConverter.printHexBinary(fingerprintBytes).toLowerCase(Locale.US);
    return toColonSeparatedString(binaryHexString);
  }

  @VisibleForTesting
  static String toColonSeparatedString(String string) {
    // Matches every pair of characters (excluding the pair at in the end) and
    // inserts a colon between them
    return string.replaceAll("..(?!$)", "$0:");
  }

  /**
   * Returns the key name corresponding to the specified fingerprints, or {@code null} if it
   * cannot be determined.
   *
   * @param privateKeyFingerprint the private key fingerprint
   * @param publicKeyFingerprint  the public key fingerprint
   * @return the key name corresponding to the specified fingerprints, or {@code null} if it
   * cannot be determined
   */
  private String lookupKeyName(String privateKeyFingerprint, String publicKeyFingerprint) {
    DescribeKeyPairsRequest request = new DescribeKeyPairsRequest()
        .withFilters(new Filter()
            .withName("fingerprint")
            .withValues(privateKeyFingerprint, publicKeyFingerprint)
        );
    DescribeKeyPairsResult result = client.describeKeyPairs(request);
    if (result.getKeyPairs().size() > 0) {
      return result.getKeyPairs().get(0).getKeyName();
    }
    return null;
  }

  /**
   * Returns a map from instance IDs to instance state for the specified batch of instance IDs.
   *
   * @param instanceIds batch of instance IDs
   * @return the map from instance IDs to instance state for the specified batch of instance IDs
   */
  private Map<String, InstanceState> getBatchInstanceState(Collection<String> instanceIds, IdType idType) {
    final Map<String, InstanceState> instanceStateByInstanceId =
        Maps.newHashMapWithExpectedSize(instanceIds.size());

    allocationHelper.forEachInstance(instanceIds, instance -> {
      Preconditions.checkNotNull(instance, "instance is null");
      instanceStateByInstanceId.put(
          getInstanceId(null, instance, idType),
          EC2InstanceState.fromInstanceStateName(
              InstanceStateName.fromValue(instance.getState().getName())));
      return null;
    }, idType);

    return instanceStateByInstanceId;
  }

  private Image getImage(String imageId) {
    DescribeImagesResult result = client.describeImages(
        new DescribeImagesRequest().withImageIds(imageId));
    if (result.getImages().isEmpty()) {
      throw new IllegalArgumentException("The description for image " + imageId +
          " is empty");
    }
    return result.getImages().get(0);
  }

  private Set<String> getExistingDeviceNames(List<BlockDeviceMapping> mappings) {
    Set<String> existingDeviceNames = Sets.newHashSet();
    for (BlockDeviceMapping mapping : mappings) {
      existingDeviceNames.add(mapping.getDeviceName());
    }
    return existingDeviceNames;
  }

  /**
   * Performs a sequence of strict instance ownership checks to avoid any potential harmful
   * accidents.
   *
   * @param instance the instance
   * @param template the template from which the instance was created, or <code>null</code>
   *                 if it is unknown (such as during a delete call)
   * @return the virtual instance ID
   * @throws IllegalStateException if the instance fails an ownership check
   */
  private String checkInstanceIsManagedByDirector(Instance instance, EC2InstanceTemplate template) {
    String virtualInstanceId = allocationHelper.getVirtualInstanceId(instance.getTags(), "instance");
    String instanceIds = instance.getInstanceId() + " / " + virtualInstanceId;
    String instanceKeyName = instance.getKeyName();

    if (template != null) {
      if (!template.getKeyName().equals(Optional.fromNullable(instanceKeyName))) {
        LOG.warn("Found unexpected key name: {} for instance: {}", instanceKeyName, instanceIds);
      }
      String instanceType = instance.getInstanceType();
      if (!template.getType().equals(instanceType)) {
        LOG.warn("Found unexpected type: {} for instance: {}", instanceType, instanceIds);
      }
      String instanceImageId = instance.getImageId();
      if (!template.getImage().equals(instanceImageId)) {
        LOG.warn("Found unexpected image: {} for instance: {}", instanceImageId, instanceIds);
      }
    }
    return virtualInstanceId;
  }

  /**
   * Returns a map from virtual instance ID to corresponding instance ID for the specified
   * virtual instance IDs.
   *
   * @param virtualInstanceIds the virtual instance IDs
   * @return the map from virtual instance ID to corresponding EC2 instance ID
   */
  private BiMap<String, String> getEC2InstanceIdsByVirtualInstanceId(
      Collection<String> virtualInstanceIds) {
    final BiMap<String, String> ec2InstanceIdsByVirtualInstanceId = HashBiMap.create();
    allocationHelper.forEachInstance(virtualInstanceIds, instance -> {
      Preconditions.checkNotNull(instance, "instance is null");
      String virtualInstanceId = checkInstanceIsManagedByDirector(instance, null);
      ec2InstanceIdsByVirtualInstanceId.put(virtualInstanceId, instance.getInstanceId());
      return null;
    }, IdType.VIRTUAL_INSTANCE_ID);
    return ec2InstanceIdsByVirtualInstanceId;
  }

  /**
   * Selects the root device from a list of block device mappings based on the
   * root device name for the mappings' image.
   *
   * @param mappings       list of block device mappings
   * @param rootDeviceName image root device name
   * @return root device mapping, or null if it could not be determined
   */
  @VisibleForTesting
  static BlockDeviceMapping selectRootDevice(List<BlockDeviceMapping> mappings, String rootDeviceName) {
    /*
     * Heuristic to find the root device:
     * - The best match is the EBS device that matches the root device name for the image, but
     *   this may not happen (/dev/sda1 vs. /dev/sda). See:
     *   http://docs.aws.amazon.com/AWSEC2/latest/UserGuide/device_naming.html
     * - If there isn't a best match, then a device whose name is a prefix for the image's root
     *   device name is selected.
     * - If all else fails, it's the first EBS volume in the list.
     */
    BlockDeviceMapping bestMatch = null;
    BlockDeviceMapping firstEbs = null;
    for (BlockDeviceMapping mapping : mappings) {
      if (mapping.getEbs() == null) {
        continue;
      }
      if (firstEbs == null) {
        firstEbs = mapping;
      }
      if (mapping.getDeviceName() == null) {
        continue;
      }
      if (rootDeviceName.equals(mapping.getDeviceName())) {
        return mapping;
      }
      if (rootDeviceName.startsWith(mapping.getDeviceName())) {
        bestMatch = mapping;
      }
    }

    if (bestMatch != null) {
      return bestMatch;
    } else if (firstEbs != null) {
      return firstEbs;
    } else {
      return null;
    }
  }

  /**
   * Returns the ID type for the specified template.
   *
   * @param template the instance template
   * @return the ID type
   */
  private IdType getIdType(EC2InstanceTemplate template) {
    return isAutomatic(template) ? IdType.EC2_INSTANCE_ID : IdType.VIRTUAL_INSTANCE_ID;
  }

  /**
   * Returns whether the specified instance template represents an automatic instance group.
   *
   * @param template the instance template
   * @return whether the specified instance template represents an automatic instance group
   */
  private boolean isAutomatic(EC2InstanceTemplate template) {
    return (template != null) && template.isAutomatic();
  }

  /**
   * Returns the instance ID of the specified type for the specified instance.
   *
   * @param template the instance template
   * @param instance the instance
   * @param idType   the ID type
   * @return the instance ID of the specified type for the specified instance
   */
  private String getInstanceId(EC2InstanceTemplate template, Instance instance, IdType idType) {
    String instanceId;
    switch (idType) {
      case VIRTUAL_INSTANCE_ID:
        instanceId = checkInstanceIsManagedByDirector(instance, template);
        break;
      case EC2_INSTANCE_ID:
        instanceId = instance.getInstanceId();
        break;
      default:
        throw new IllegalStateException("Unknown instance ID type.");
    }
    return instanceId;
  }

  @VisibleForTesting
  class AllocationHelperImpl implements AllocationHelper {

    @Override
    public AWSTimeouts getAWSTimeouts() {
      return awsTimeouts;
    }

    @Override
    public EC2TagHelper getEC2TagHelper() {
      return ec2TagHelper;
    }

    @Override
    public boolean waitUntilInstanceHasStarted(String ec2InstanceId, DateTime timeout) throws InterruptedException, TimeoutException {

      Callable<Boolean> task = () -> {
        DescribeInstanceStatusResult result = client.describeInstanceStatus(
            new DescribeInstanceStatusRequest()
                .withIncludeAllInstances(true)
                .withInstanceIds(ec2InstanceId)
        );
        for (InstanceStatus status : result.getInstanceStatuses()) {
          if (ec2InstanceId.equals(status.getInstanceId())) {
            InstanceStateName currentState = InstanceStateName.fromValue(status.getInstanceState().getName());

            if (currentState.equals(InstanceStateName.Terminated) ||
                currentState.equals(InstanceStateName.ShuttingDown)) {
              if (LOG.isErrorEnabled()) {
                LOG.error("Instance {} has unexpectedly terminated", ec2InstanceId);
              }
              return false;
            } else if (!currentState.equals(InstanceStateName.Pending)) {
              return true;
            }
          }
        }
        AmazonServiceException exception = new AmazonServiceException(INVALID_INSTANCE_ID_NOT_FOUND);
        exception.setErrorCode(INVALID_INSTANCE_ID_NOT_FOUND);
        throw exception;
      };

      try {
        return retryUntil(task, timeout);
      } catch (RetryException e) {
        LOG.info("timeout waiting for instance {} to start", ec2InstanceId);
        throw new TimeoutException(String.format("Timeout waiting for instance %s to start", ec2InstanceId));
      } catch (ExecutionException e) {
        if (AmazonServiceException.class.isInstance(e.getCause())) {
          throw AWSExceptions.propagate(stsClient, (AmazonServiceException) e.getCause());
        }
        throw new UnrecoverableProviderException(e.getCause());
      }
    }

    @Override
    public Collection<EC2Instance> find(EC2InstanceTemplate template,
        Collection<String> instanceIds) throws InterruptedException {
      // Delegate to the provider so we get logging, etc.
      return EC2Provider.this.find(template, instanceIds);
    }

    @Override
    public void forEachInstance(DescribeInstancesResult result,
        Function<Instance, Void> instanceHandler) {
      List<Reservation> reservations;
      while (!(reservations = result.getReservations()).isEmpty()) {
        for (Reservation reservation : reservations) {
          for (Instance instance : reservation.getInstances()) {
            LOG.debug("Calling instance handler with instance {}", instance);
            instanceHandler.apply(instance);
          }
        }

        if (result.getNextToken() != null) {
          result = client.describeInstances(
              new DescribeInstancesRequest().withNextToken(result.getNextToken()));
        } else {
          break;
        }
      }
    }

    @Override
    public String getVirtualInstanceId(List<Tag> tags, String type) {
      String idTagName = ec2TagHelper.getClouderaDirectorIdTagName();
      for (Tag tag : tags) {
        if (tag.getKey().equals(idTagName)) {
          return tag.getValue();
        }
      }

      throw new IllegalStateException(String.format("Any %s managed by " +
          "Cloudera Altus Director should have a %s tag.", type, idTagName));
    }

    @Override
    public List<BlockDeviceMapping> getBlockDeviceMappings(EC2InstanceTemplate template) {
      // Query the AMI about the root device name & mapping information
      Image templateImage = getImage(template.getImage());
      String rootDeviceType = templateImage.getRootDeviceType();
      if (!DEVICE_TYPE_EBS.equals(rootDeviceType)) {
        throw new IllegalArgumentException("The root device for image " + template.getImage() +
            " must be \"" + DEVICE_TYPE_EBS + "\", found: " +
            rootDeviceType);
      }
      List<BlockDeviceMapping> originalMappings = templateImage.getBlockDeviceMappings();
      LOG.info(">> Original image block device mappings: {}", originalMappings);
      if (originalMappings.isEmpty()) {
        throw new IllegalArgumentException("The image " + template.getImage() +
            " has no block device mappings");
      }
      BlockDeviceMapping rootDevice = selectRootDevice(originalMappings,
          templateImage.getRootDeviceName());
      if (rootDevice == null) {
        throw new IllegalArgumentException("Could not determine root device for image " +
            template.getImage() + " based on root device name " +
            templateImage.getRootDeviceName());
      }

      Set<String> existingDeviceNames = getExistingDeviceNames(originalMappings);

      // The encrypted property was added to the block device mapping in version 1.8 of the SDK.
      // It is a Boolean, but defaults to false instead of being unset, so we set it to null here.
      rootDevice.getEbs().setEncrypted(null);
      rootDevice.getEbs().setVolumeSize(template.getRootVolumeSizeGB());
      rootDevice.getEbs().setVolumeType(template.getRootVolumeType());
      rootDevice.getEbs().setDeleteOnTermination(true);

      List<BlockDeviceMapping> deviceMappings = Lists.newArrayList(rootDevice);

      EBSAllocationStrategy ebsAllocationStrategy = EBSAllocationStrategy.get(template);

      switch (ebsAllocationStrategy) {
        case NO_EBS_VOLUMES:
          // The volumes within an instance should be homogeneous. So we only add
          // instance store volumes when additional EBS volumes aren't mounted.
          deviceMappings.addAll(ephemeralDeviceMappings.getBlockDeviceMappings(template.getType(), existingDeviceNames));
          break;
        case AS_INSTANCE_REQUEST:
          LOG.info("EBS volumes will be allocated as part of instance launch request");
          List<BlockDeviceMapping> mappings =
              ebsDeviceMappings.getBlockDeviceMappings(template, existingDeviceNames);
          deviceMappings.addAll(mappings);
          break;
        case AS_SEPARATE_REQUESTS:
          LOG.info("EBS volumes will be separately allocated after instance launch request");
          break;
        default:
          throw new IllegalStateException("Invalid EBS allocation strategy " + ebsAllocationStrategy);
      }

      LOG.info(">> Block device mappings: {}", deviceMappings);
      return deviceMappings;
    }

    @Override
    public InstanceNetworkInterfaceSpecification getInstanceNetworkInterfaceSpecification(
        EC2InstanceTemplate template) {
      InstanceNetworkInterfaceSpecification network = new InstanceNetworkInterfaceSpecification()
          .withDeviceIndex(0)
          .withSubnetId(template.getSubnetId())
          .withGroups(template.getSecurityGroupIds())
          .withDeleteOnTermination(true)
          .withAssociatePublicIpAddress(associatePublicIpAddresses);

      LOG.info(">> Network interface specification: {}", network);
      return network;
    }

    @Override
    public boolean isAssociatePublicIpAddresses() {
      return associatePublicIpAddresses;
    }

    @Override
    public EC2Instance createInstance(EC2InstanceTemplate template, String instanceId, Instance instanceDetails) {
      return new EC2Instance(template, instanceId, instanceDetails);
    }

    @Override
    public Iterable<Entry<String, Instance>> doFind(
        EC2InstanceTemplate template, Iterable<String> instanceIds,
        Predicate<Instance> predicate)
        throws InterruptedException {
      return doFind(instanceIds, template, predicate, Duration.ZERO);
    }

    @SuppressWarnings({"StaticPseudoFunctionalStyleMethod"})
    @VisibleForTesting
    Iterable<Entry<String, Instance>> doFind(
        final Iterable<String> instanceIds,
        final EC2InstanceTemplate template,
        final Predicate<Instance> predicate,
        Duration timeout)
        throws InterruptedException {

      IdType idType = getIdType(template);

      final Map<String, Instance> result = Maps.newHashMap();
      for (String instanceId : instanceIds) {
        result.put(instanceId, null);
      }

      Retryer<Map<String, Instance>> retryer = RetryerBuilder.<Map<String, Instance>>newBuilder()
          .withStopStrategy(StopStrategies.stopAfterDelay(timeout.getMillis(), TimeUnit.MILLISECONDS))
          .withWaitStrategy(WaitStrategies.fixedWait(1000, TimeUnit.MILLISECONDS))
          .retryIfResult(idToInstances -> idToInstances != null &&
              !Iterables.all(idToInstances.entrySet(), ENTRY_VALUE_NOT_NULL))
          .build();

      try {
        retryer.call(() -> {
          Collection<String> ids = FluentIterable.from(result.entrySet())
              .filter(Predicates.not(ENTRY_VALUE_NOT_NULL))
              .transform(Entry::getKey)
              .toList();

          forEachInstance(ids, instance -> {
            if ((instance != null) && predicate.apply(instance)) {
              String instanceId = getInstanceId(template, instance, idType);
              result.put(instanceId, instance);
            }
            return null;
          }, idType);

          return result;
        });
      } catch (ExecutionException e) {
        Throwable cause = e.getCause();
        Throwables.throwIfInstanceOf(cause, InterruptedException.class);
        Throwables.throwIfUnchecked(cause);
        throw new RuntimeException(cause);
      } catch (RetryException e) {
        if (Thread.interrupted()) {
          throw new InterruptedException();
        }
      }

      return Iterables.filter(result.entrySet(), ENTRY_VALUE_NOT_NULL);
    }

    /**
     * Iterates through the instances identified by the specified instance IDs and calls the
     * specified handler on each instance. Instances with the same instance IDs are deduplicated,
     * preferring running instances over terminating or terminated instances.
     *
     * @param instanceIds     the instance IDs
     * @param instanceHandler the instance handler
     * @param idType          the type of ID
     */
    @VisibleForTesting
    void forEachInstance(Collection<String> instanceIds,
        Function<Instance, Void> instanceHandler, IdType idType) {
      List<DescribeInstancesResult> results = Lists.newArrayList();
      String idTagName = ec2TagHelper.getClouderaDirectorIdTagName();
      for (List<String> instanceIdChunk : Iterables.partition(instanceIds, MAX_TAG_FILTERING_REQUESTS)) {
        DescribeInstancesResult result;
        switch (idType) {
          case EC2_INSTANCE_ID:
            try {
              result = client.describeInstances(new DescribeInstancesRequest()
                  .withInstanceIds(instanceIdChunk));
            } catch (AmazonServiceException e) {
              if (!"InvalidInstanceID.Malformed".equals(e.getErrorCode())) {
                throw e;
              }
              continue;
            }
            break;
          case VIRTUAL_INSTANCE_ID:
            result = client.describeInstances(new DescribeInstancesRequest()
                .withFilters(new Filter().withName("tag:" + idTagName)
                    .withValues(instanceIdChunk)));
            break;
          default:
            throw new IllegalStateException("Unknown ID type: " + idType);
        }
        results.add(result);
      }

      // collect Instances, preferring running instances over terminated instances and
      // also ensuring we only process one instance per instance id
      final Map<String, Instance> instanceIdToInstance =
          Maps.newHashMapWithExpectedSize(instanceIds.size());
      for (DescribeInstancesResult result : results) {
        forEachInstance(result, instance -> {
          String instanceId;
          try {
            instanceId = getInstanceId(null, instance, idType);
          } catch (IllegalStateException e) {
            LOG.error("Instance {} is not managed by Director. Skipping", instance.getInstanceId());
            return null;
          }
          Instance oldInstance = instanceIdToInstance.get(instanceId);
          if (oldInstance == null || INSTANCE_IS_TERMINAL.apply(oldInstance)) {
            instanceIdToInstance.put(instanceId, instance);
            if (oldInstance != null) {
              LOG.warn("Retaining new instance {} in preference to terminal instance {}",
                  instance, oldInstance);
            }
          } else {
            if (!INSTANCE_IS_TERMINAL.apply(instance)) {
              LOG.error("Two non-terminal instances with instance id {} exist: {} and {}",
                  instanceId, oldInstance, instance);
            } else {
              LOG.warn("Retaining current non-terminal instance {} in preference to terminal instance {}",
                  oldInstance, instance);
            }
          }
          return null;
        });
      }

      for (Instance instance : instanceIdToInstance.values()) {
        instanceHandler.apply(instance);
      }
    }

    @Override
    public void delete(EC2InstanceTemplate template, Collection<String> instanceIds)
        throws InterruptedException {

      Collection<String> ec2InstanceIds;

      if (isAutomatic(template)) {
        ec2InstanceIds = instanceIds;
      } else {
        ImmutableMap<Entry<String, Instance>, String> virtualInstanceIdToEC2InstanceIds = FluentIterable
            .from(doFind(template, instanceIds))
            .toMap(virtualInstanceIdToInstance -> {
              requireNonNull(virtualInstanceIdToInstance, "virtualInstanceIdToInstance is null");
              return virtualInstanceIdToInstance.getValue().getInstanceId();
            });

        if (virtualInstanceIdToEC2InstanceIds.size() < instanceIds.size()) {
          LOG.info("Unable to terminate instances, unknown {}",
              Sets.difference(Sets.newHashSet(instanceIds), virtualInstanceIdToEC2InstanceIds.keySet()));
        }

        ec2InstanceIds = virtualInstanceIdToEC2InstanceIds.values();
      }

      doDelete(ec2InstanceIds);
    }

    @Override
    public void doDelete(Collection<String> ec2InstanceIds) throws InterruptedException {
      if (ec2InstanceIds.isEmpty()) {
        return;
      }

      try {
        LOG.info(">> Terminating {}", ec2InstanceIds);
        TerminateInstancesRequest request = new TerminateInstancesRequest().withInstanceIds(ec2InstanceIds);
        TerminateInstancesResult result = client.terminateInstances(request);
        LOG.info("<< Result {}", result);

      } catch (AmazonClientException e) {
        throw AWSExceptions.propagate(stsClient, e);
      }
    }
  }
} @RunWith(Parameterized.class)
public class EC2ProviderParameterizedLiveTest {

  private static final Logger LOG = Logger.getLogger(EC2ProviderParameterizedLiveTest.class.getName());

  private static final String VOLUME_NOT_FOUND_ERR_CODE = "InvalidVolume.NotFound";

  private static final Integer DEFAULT_BLOCK_DURATION_MINUTES = 60;

  private static final String TEST_USER_DATA = "Test user data #&*^@#$(*!@";
  private static final String ENCODED_TEST_USER_DATA = Base64Utils.encodeToString(TEST_USER_DATA.getBytes());

  private static String testRegion;
  private static String testSubnet;
  private static String testSecurityGroup;
  private static String testKmsKeyArn;
  private static List<RoleConfiguration> delegatedRoleConfigurations;
  private static String externalAccountRegion;
  private static String externalAccountSubnetId;
  private static String externalAccountSecurityGroup;
  private static String externalAccountKmsKeyArn;

  @BeforeClass
  public static void setup() {
    if (isLive()) {
      LiveTestProperties liveProps = LiveTestProperties.loadLiveTestProperties();

      testRegion = liveProps.getTestRegion();
      testSubnet = liveProps.getTestSubnet();
      testSecurityGroup = liveProps.getTestSecurityGroup();
      testKmsKeyArn = liveProps.getTestKmsKeyArn();

      delegatedRoleConfigurations = liveProps.getDelegatedRoleConfigurations();
      externalAccountRegion = liveProps.getExternalAccountRegion();
      externalAccountSubnetId = liveProps.getExternalAccountSubnetId();
      externalAccountSecurityGroup = liveProps.getExternalAccountSecurityGroup();
      externalAccountKmsKeyArn = liveProps.getExternalAccountKmsKeyArn();
    }
  }

  private enum UserDataParameter {
    UNENCODED,
    ENCODED,
    INVALID;
  }

  @Parameterized.Parameters
  public static Collection<Object[]> data() {
    return Arrays.asList(new Object[][]{
        {false, false, false, false, false, UserDataParameter.UNENCODED}, // on-demand instances
        {false, false, false, false, false, UserDataParameter.INVALID},  // on-demand instances + user data failure
        {false, false, true, false, false, UserDataParameter.ENCODED},  // on-demand instances + ebs volumes + encoded user data
        {false, false, true, true, false, UserDataParameter.UNENCODED},   // on-demand instances + ebs volumes + ebs encryption
        {true, false, false, false, false, UserDataParameter.UNENCODED},  // spot instances
        {true, true, false, false, false, UserDataParameter.UNENCODED},   // spot instances with block duration
        {true, false, true, false, false, UserDataParameter.UNENCODED},   // spot instances + ebs volumes
        {false, false, false, false, true, UserDataParameter.UNENCODED}   // on-demand instances + delegation
    });
  }

  private boolean useSpotInstances;
  private boolean useSpotBlock;
  private boolean useEbsVolumes;
  private boolean testEbsEncryption;
  private boolean testDelegation;
  private UserDataParameter userDataParameter;

  public EC2ProviderParameterizedLiveTest(boolean useSpotInstances, boolean useSpotBlock,
                           boolean useEbsVolumes, boolean testEbsEncryption,
                           boolean testDelegation, UserDataParameter userDataParameter) {
    this.useSpotInstances = useSpotInstances;
    this.useSpotBlock = useSpotBlock;
    this.useEbsVolumes = useEbsVolumes;
    this.testEbsEncryption = testEbsEncryption;
    this.testDelegation = testDelegation;
    this.userDataParameter = userDataParameter;
  }

  private void createEC2Instance(
      AWSCredentialsProvider credentialsProvider,
      String subnetId,
      String securityGroup,
      String kmsKeyArn,
      String region) throws InterruptedException {

    boolean success = true;

    Ami ami;
    String amiId;
    String spotBidUSDPerHour = null;
    String spotString = "";
    Integer blockDurationMinutes = null;

    if (useSpotInstances) {
      ami = CENTOS64_AMI_BY_REGION.get(region);
      amiId = ami.getId();
      spotBidUSDPerHour = "0.50";
      spotString = "Spot ";
      if (useSpotBlock) {
        blockDurationMinutes = DEFAULT_BLOCK_DURATION_MINUTES;
      }
    } else {
      ami = RHEL67_AMI_BY_REGION.get(region);
      amiId = ami.getId();
    }

    String username = System.getProperty("user.name");

    // Configure provider
    LinkedHashMap<String, String> providerConfigMap = new LinkedHashMap<>();
    putConfig(providerConfigMap, REGION, region);

    // Create provider
    AmazonEC2ClientSpyProvider clientSpyProvider =
        new AmazonEC2ClientSpyProvider(credentialsProvider,
            new ClientConfigurationFactory().getConfig());
    EC2Provider ec2Provider = getEc2Provider(new SimpleConfiguration(providerConfigMap),
        credentialsProvider, clientSpyProvider);
    CustomTagMappings customTagMappings = ec2Provider.getEC2TagHelper().getCustomTagMappings();

    // mock eventual consistency when waiting for instances to start
    // NOTE: client spy must be retrieved after getEc2Provider is called
    AmazonEC2AsyncClient ec2AsyncClientSpy = clientSpyProvider.getClientSpy();
    final AmazonServiceException eventualConsistencyException =
        new AmazonServiceException("Eventual consistency exception");
    eventualConsistencyException.setErrorCode(INVALID_INSTANCE_ID_NOT_FOUND);
    Answer<Object> describeInstanceStatusAnswer = new CallsRealMethods() {
      private Map<String, Integer> instancesTimesSeen = Maps.newHashMap();

      @Override
      public Object answer(InvocationOnMock invocationOnMock) throws Throwable {
        // simulate eventual consistency failure the first time we see each instance
        int minTimesSeen = Integer.MAX_VALUE;
        DescribeInstanceStatusRequest request = (DescribeInstanceStatusRequest) invocationOnMock.getArgument(0);
        for (String instanceId : request.getInstanceIds()) {
          Integer timesSeen = instancesTimesSeen.get(instanceId);
          if (timesSeen == null) {
            timesSeen = 0;
          }
          minTimesSeen = Math.min(minTimesSeen, timesSeen);
          instancesTimesSeen.put(instanceId, timesSeen + 1);
        }
        if (minTimesSeen == 0 || minTimesSeen == 2) {
          LOG.info("Simulating eventual consistency exception. Instances seen " + minTimesSeen +
              " times. " + invocationOnMock);
          throw eventualConsistencyException;
        }

        DescribeInstanceStatusResult result =
            (DescribeInstanceStatusResult) super.answer(invocationOnMock);
        if (minTimesSeen == 1) {
          for (com.cloudera.director.aws.shaded.com.amazonaws.services.ec2.model.InstanceStatus
              instanceStatus : result.getInstanceStatuses()) {
            com.cloudera.director.aws.shaded.com.amazonaws.services.ec2.model.InstanceState
                instanceState = instanceStatus.getInstanceState();
            instanceState.setCode(0);
            instanceState.setName(com.cloudera.director.aws.shaded.com.amazonaws.services.ec2.model.InstanceStateName.Pending);
          }
          LOG.info("Instances seen " + minTimesSeen + " time. Modified result = " + result);
        }
        return result;
      }
    };
    doAnswer(describeInstanceStatusAnswer)
        .when(ec2AsyncClientSpy).describeInstanceStatus(any(DescribeInstanceStatusRequest.class));

    // Configure instance template
    TestInstanceTemplate template = new TestInstanceTemplate();
    template.addConfig(IMAGE, amiId);
    template.addConfig(SECURITY_GROUP_IDS, securityGroup);
    template.addConfig(SUBNET_ID, subnetId);
    template.addConfig(TYPE, "m3.medium");
    template.addConfig(USE_SPOT_INSTANCES, String.valueOf(useSpotInstances));
    template.addConfig(SPOT_BID_USD_PER_HR, spotBidUSDPerHour);
    template.addConfig(BLOCK_DURATION_MINUTES, blockDurationMinutes == null ?
        null : blockDurationMinutes.toString());
    switch (userDataParameter) {
      case UNENCODED:
        template.addConfig(USER_DATA_UNENCODED, TEST_USER_DATA);
        break;
      case ENCODED:
        template.addConfig(USER_DATA, ENCODED_TEST_USER_DATA);
        break;
      case INVALID:
        template.addConfig(USER_DATA, TEST_USER_DATA); // should be base64 encoded, ergo invalid
        break;
    }
    template.addTag(EXTRA_TAG, "foo");

    int ebsVolumeCount = 0;
    int ebsVolumeSizeGiB = 500;
    String ebsVolumeType = "st1";
    String enableEbsEncryption = "true";

    Set<String> mountedEbsVolumes = Sets.newHashSet();
    if (useEbsVolumes) {
      ebsVolumeCount = 2;
      template.addConfig(EBS_VOLUME_COUNT, String.valueOf(ebsVolumeCount));
      template.addConfig(EBS_VOLUME_SIZE_GIB, String.valueOf(ebsVolumeSizeGiB));
      template.addConfig(EBS_VOLUME_TYPE, ebsVolumeType);

      if (testEbsEncryption) {
        template.addConfig(ENCRYPT_ADDITIONAL_EBS_VOLUMES, enableEbsEncryption);
        template.addConfig(EBS_KMS_KEY_ID, kmsKeyArn);
      }
    }

    // Create instance template
    EC2InstanceTemplate instanceTemplate = ec2Provider.createResourceTemplate(
        template.getTemplateName(), new SimpleConfiguration(template.getConfigs()), template.getTags());

    // Create instances
    List<String> virtualInstanceIds = Arrays.asList(UUID.randomUUID().toString(),
        UUID.randomUUID().toString());
    int instanceCount = virtualInstanceIds.size();

    LOG.info("Allocating " + instanceCount + " " + spotString + "instances.");

    try {
      ec2Provider.allocate(instanceTemplate, virtualInstanceIds, instanceCount);
    } catch (UnrecoverableProviderException e) {
      if (userDataParameter == UserDataParameter.INVALID) {
        LOG.info("User data failed as expected.");
        return;
      } else {
        fail("Transient exeception while allocating instances.", e);
      }
    }

    if (userDataParameter == UserDataParameter.INVALID) {
      fail("Allocation did not fail when giving invalid user data as expected.");
    }

    Collection<EC2Instance> instances = ec2Provider.find(instanceTemplate, virtualInstanceIds);

    try {
      for (EC2Instance instance : instances) {
        LOG.info("Created " + spotString + "instance: " + instance.getProperties());

        try {
          InstanceAttribute instanceAttribute = ec2Provider.describeInstanceAttribute(
              instance.unwrap().getInstanceId(), InstanceAttributeName.UserData);
          assertThat(TEST_USER_DATA.getBytes()).isEqualTo(
              Base64Utils.decodeFromString(instanceAttribute.getUserData()));
          LOG.info("Verified user data.");
        } catch (AmazonServiceException e) {
          LOG.warning("Cannot verify user data.");
        }
      }

      // Wait for instances to be running
      LOG.info("Waiting for " + spotString + "instances to be running.");
      Map<String, InstanceState> instanceStates = waitUntilRunningOrTerminal(ec2Provider,
          instanceTemplate, virtualInstanceIds);

      for (Map.Entry<String, InstanceState> entry : instanceStates.entrySet()) {
        String virtualInstanceId = entry.getKey();
        InstanceState instanceState = entry.getValue();
        InstanceStatus instanceStatus = instanceState.getInstanceStatus();

        LOG.info(spotString + "Instance: " + virtualInstanceId + " has status: " + instanceStatus);
        if (instanceStatus != InstanceStatus.RUNNING) {
          LOG.severe(spotString + "Instance: " + virtualInstanceId + " has status: " +
              instanceStatus + " instead of RUNNING");
          success = false;
          break;
        }
      }

      if (ami.isSupportsHostKeyFingerprintRetrieval()) {
        checkHostKeyFingerprints(ec2Provider, instanceTemplate, virtualInstanceIds);
      }

      if (useSpotInstances) {
        verifySpotInstance(ec2Provider.getClient(), instances);
      }

      if (useEbsVolumes && success) {
        for (String virtualInstanceId : instanceStates.keySet()) {
          LOG.info("Checking that volumes were properly mounted for instance " + virtualInstanceId);
          List<Volume> volumes = ec2Provider.getVolumes(virtualInstanceId);

          int expectedVolumeCount = ebsVolumeCount + 1; // plus 1 for root volume
          if (volumes.size() != expectedVolumeCount) {
            LOG.severe("Expected " + expectedVolumeCount + " volumes but got " + volumes.size());
            success = false;
            break;
          }

          int rootVolumeCount = 0;
          int mountedVolumeCount = 0;
          for (Volume volume : volumes) {
            String volumeType = volume.getVolumeType();

            if (volumeType.equals(ebsVolumeType)) {
              mountedVolumeCount++;
              mountedEbsVolumes.add(volume.getVolumeId());
              assertTrue(volume.getSize() == ebsVolumeSizeGiB);

              if (testEbsEncryption) {
                assertTrue(volume.getEncrypted());
                assertTrue(volume.getKmsKeyId().equals(kmsKeyArn));
              }

              // EBS volumes should also have the tags from instance template
              Map<String, String> volumeTags = listTagToMap(volume.getTags());
              for (Map.Entry<String, String> entry : template.getTags().entrySet()) {
                assertTrue(volumeTags.containsKey(entry.getKey()));
                assertTrue(volumeTags.containsValue(entry.getValue()));
              }
            } else if (volumeType.equals(ROOT_VOLUME_TYPE.unwrap().getDefaultValue())) {
              rootVolumeCount++;
            } else {
              success = false;
              LOG.severe("Found unexpected volume type " + volumeType);
              break;
            }
          }

          if (rootVolumeCount != 1) {
            LOG.severe("Found more than one volume with root volume type");
            success = false;
            break;
          }

          if (mountedVolumeCount != ebsVolumeCount) {
            LOG.severe("Incorrect number of ebs volumes");
            success = false;
            break;
          }
        }
      }
    } finally {
      // Terminate instances
      ec2Provider.delete(instanceTemplate, virtualInstanceIds);

      // Wait for instances to be deleted
      LOG.info("Waiting for " + spotString + "instances to be deleted.");
      Map<String, InstanceState> instanceStates = waitForInstanceStates(
          ec2Provider,
          instanceTemplate,
          virtualInstanceIds,
          EnumSet.of(InstanceStatus.DELETED, InstanceStatus.UNKNOWN),
          EnumSet.of(InstanceStatus.FAILED, InstanceStatus.STOPPED)
      );

      if (success) {
        for (Map.Entry<String, InstanceState> entry : instanceStates.entrySet()) {
          String virtualInstanceId = entry.getKey();
          InstanceState instanceState = entry.getValue();
          InstanceStatus instanceStatus = instanceState.getInstanceStatus();
          LOG.info(spotString + "Instance: " + virtualInstanceId + " has status: " + instanceStatus);
          if ((instanceStatus != InstanceStatus.DELETED)
              && (instanceStatus != InstanceStatus.UNKNOWN)) {
            success = false;
          }
        }
      }

      if (useEbsVolumes) {
        Set<String> pendingEbsVolumes = Sets.newHashSet(mountedEbsVolumes);

        LOG.info("Verifying that the provisioned EBS volumes are deleted");
        while (true) {
          pendingEbsVolumes = getUndeletedVolumes(ec2Provider.getClient(), pendingEbsVolumes);

          if (pendingEbsVolumes.isEmpty()) break;

          LOG.info(pendingEbsVolumes.size()
              + " EBS volumes still pending to be deleted. Sleeping for 15 seconds...");
          Thread.sleep(15000);
        }
      }
    }

    assertTrue(success);
  }

  private static void checkHostKeyFingerprints(EC2Provider ec2Provider, EC2InstanceTemplate template,
                                               List<String> virtualInstanceIds) throws InterruptedException {
    LOG.info("Checking host key fingerprints");

    Map<String, Set<String>> fingerprintsById = ec2Provider.getHostKeyFingerprints(template, virtualInstanceIds);
    assertEquals(fingerprintsById.size(), virtualInstanceIds.size());

    Set<String> allHostKeyFingerprints = Sets.newHashSet();
    for (Map.Entry<String, Set<String>> entry: fingerprintsById.entrySet()) {
      String vid = entry.getKey();
      Set<String> fingerprints = entry.getValue();

      LOG.info(String.format("VID %s has host key fingerprints %s", vid, fingerprints.toString()));
      assertFalse(fingerprints.isEmpty());

      // verify that fingerprints are unique
      for (String fingerprint : fingerprints) {
        assertTrue(!fingerprint.isEmpty());
        assertTrue(allHostKeyFingerprints.add(fingerprint));
      }
    }
  }

  @Test
  public void testCreateEC2Instance() throws InterruptedException {
    assumeTrue(isLive());
    assumeFalse(testDelegation);

    assumeNotNull(testSubnet);
    assumeNotNull(testSecurityGroup);
    assumeNotNull(testKmsKeyArn);
    assumeNotNull(testRegion);

    // Configure credentials
    Map<String, String> credentialsConfigMap = new LinkedHashMap<>();
    SimpleConfiguration credentialsConfig = new SimpleConfiguration(credentialsConfigMap);
    AWSCredentialsProvider credentialsProvider =
        new AWSCredentialsProviderChainProvider()
            .createCredentials(credentialsConfig, DEFAULT_PLUGIN_LOCALIZATION_CONTEXT);

    createEC2Instance(credentialsProvider,
        testSubnet,
        testSecurityGroup,
        testKmsKeyArn,
        testRegion);
  }

  @Test
  public void testDelegatedRoleAccessTest() throws InterruptedException {
    assumeTrue(isLive());
    assumeTrue(testDelegation);
    assumeFalse(userDataParameter == UserDataParameter.INVALID);

    assumeFalse(delegatedRoleConfigurations.isEmpty());
    assumeNotNull(externalAccountSubnetId);
    assumeNotNull(externalAccountSecurityGroup);
    assumeNotNull(externalAccountKmsKeyArn);
    assumeNotNull(externalAccountRegion);

    LOG.info("Testing delegated role access in external account.");
    // Configure credentials
    RoleConfiguration lastRole = Iterables.getLast(delegatedRoleConfigurations);
    Map<String, String> credentialsConfigMap = new LinkedHashMap<>();
    credentialsConfigMap.put("delegatedRoleArn", lastRole.getRoleArn());
    credentialsConfigMap.put("delegatedRoleExternalId", lastRole.getRoleExternalId());
    SimpleConfiguration credentialsConfig = new SimpleConfiguration(credentialsConfigMap);

    // When creating AWSSecurityTokenServiceClient, it uses the default credential
    // provider chain, DefaultAWSCredentialsProviderChain, which means when the live
    // tests run it needs to get the AWS credentials from environment variables,
    // or Java system properties, or credential profiles file, or ec2 instance profile.
    List<RoleConfiguration> baseRoleConfigurations = Lists.newArrayList(Iterables.limit(
        delegatedRoleConfigurations,
        delegatedRoleConfigurations.size() - 1));
    AWSCredentialsProvider credentialsProvider =
        new AWSCredentialsProviderChainProvider(baseRoleConfigurations)
            .createCredentials(credentialsConfig, DEFAULT_PLUGIN_LOCALIZATION_CONTEXT);

    createEC2Instance(credentialsProvider,
        externalAccountSubnetId,
        externalAccountSecurityGroup,
        externalAccountKmsKeyArn,
        externalAccountRegion);
  }

  private Map<String, String> listTagToMap(List<Tag> tags) {
    Map<String, String> tagMap = Maps.newHashMap();
    for (Tag tag : tags) {
      tagMap.put(tag.getKey(), tag.getValue());
    }
    return tagMap;
  }

  // Return the number of EBS volumes that are not yet deleted.
  // Volumes that can't be found are considered deleted.
  public Set<String> getUndeletedVolumes(AmazonEC2Client client, Collection<String> volumeIds) {
    Set<String> undeleted = Sets.newHashSet();

    for (String id : volumeIds) {
      DescribeVolumesRequest request = new DescribeVolumesRequest().withVolumeIds(id);

      try {
        DescribeVolumesResult volumeResults = client.describeVolumes(request);

        Volume volume = Iterables.getOnlyElement(volumeResults.getVolumes());
        String volumeState = volume.getState();

        if (!volumeState.equals(VolumeState.Deleted.toString())) {
          LOG.info("Volume " + volumeIds + " still has state " + volumeState);
          undeleted.add(volume.getVolumeId());
        }
      } catch (AmazonServiceException ex) {
        if (!ex.getErrorCode().equals(VOLUME_NOT_FOUND_ERR_CODE)) {
          throw ex;
        }
      }
    }
    return undeleted;
  }

  private void verifySpotInstance(AmazonEC2Client client, Collection<EC2Instance> instances) {
    DescribeSpotInstanceRequestsRequest request = new DescribeSpotInstanceRequestsRequest();

    LOG.info("Verifying each instance is spot");
    for (EC2Instance instance : instances) {
      assertTrue(Boolean.parseBoolean(
          instance.getProperties()
              .get(EC2Instance.EC2InstanceDisplayPropertyToken.SPOT.unwrap().getDisplayKey())));

      String spotRequestId = instance.unwrap().getSpotInstanceRequestId();
      request.withSpotInstanceRequestIds(spotRequestId);
    }
    LOG.info("Done verify that each instance is spot");

    if (useSpotBlock && request.getSpotInstanceRequestIds().size() > 0) {
      LOG.info("Verifying blockDurationMinutes");
      DescribeSpotInstanceRequestsResult results = client.describeSpotInstanceRequests(request);
      for (SpotInstanceRequest spotRequest : results.getSpotInstanceRequests()) {
        LOG.info("Spot request = " + spotRequest);
        assertTrue(DEFAULT_BLOCK_DURATION_MINUTES.equals(spotRequest.getBlockDurationMinutes()));
      }
      LOG.info("Done verifying blockDurationMinutes");
    }
  }

  private static class AmazonEC2ClientSpyProvider extends AmazonEC2ClientProvider {
    private AmazonEC2AsyncClient clientSpy = null;

    public AmazonEC2ClientSpyProvider(AWSCredentialsProvider awsCredentialsProvider,
                                      ClientConfiguration clientConfiguration) {
      super(awsCredentialsProvider, clientConfiguration);
    }

    public synchronized AmazonEC2AsyncClient getClientSpy() {
      if (clientSpy == null) {
        throw new IllegalStateException(
            "clientSpy has not be set yet. getClient must be called before getting the clientSpy");
      }
      return clientSpy;
    }

    @Override
    public synchronized AmazonEC2AsyncClient getClient(Configured configuration,
                                                       PluginExceptionConditionAccumulator accumulator,
                                                       LocalizationContext providerLocalizationContext,
                                                       boolean verify) {
      clientSpy = spy(super.getClient(configuration, accumulator, providerLocalizationContext, verify));
      return clientSpy;
    }
  }

} @SuppressWarnings({"PMD.TooManyStaticImports", "Guava"})
public class EphemeralDeviceMappings {

  private static final Logger LOG = LoggerFactory.getLogger(EphemeralDeviceMappings.class);

  public static class EphemeralDeviceMappingsConfigProperties extends DeviceMappingsConfigProperties {

    /**
     * Ephemeral device mappings configuration properties.
     */
    // Fully qualifying class name due to compiler bug
    @SuppressWarnings("PMD.UnnecessaryFullyQualifiedName")
    public enum EphemeralDeviceMappingsConfigurationPropertyToken
        implements com.cloudera.director.spi.v2.model.ConfigurationPropertyToken {
      /**
       * Path for the custom device mappings file. Relative paths are based on
       * the plugin configuration directory.
       */
      CUSTOM_MAPPINGS_PATH(new SimpleConfigurationPropertyBuilder()
          .configKey("customMappingsPath")
          .name("Custom mappings path")
          .defaultDescription("The path for the custom ephemeral device mappings file. Relative " +
              "paths are based on the plugin configuration directory.")
          .build());

      /**
       * The configuration property.
       */
      private final ConfigurationProperty configurationProperty;

      /**
       * Creates a configuration property token with the specified parameters.
       *
       * @param configurationProperty the configuration property
       */
      EphemeralDeviceMappingsConfigurationPropertyToken(
          ConfigurationProperty configurationProperty) {
        this.configurationProperty = configurationProperty;
      }

      @Override
      public ConfigurationProperty unwrap() {
        return configurationProperty;
      }
    }

    public static final String DEFAULT_DEVICE_NAME_PREFIX = "/dev/sd";

    /**
     * Ephemeral drives should be attached in the following range:
     * /dev/(sd|xvd)[b-y].
     */
    public static final char DEFAULT_RANGE_START_FOR_EPHEMERAL_DRIVES = 'b';

    public static final String DEFAULT_CUSTOM_MAPPINGS_PATH =
        "ec2.ephemeraldevicemappings.properties";

    private final File configurationDirectory;

    private String customMappingsPath = DEFAULT_CUSTOM_MAPPINGS_PATH;


    /**
     * Creates ephemeral device mappings config properties with the specified parameters.
     *
     * @param configuration            the configuration
     * @param configurationDirectory   the plugin configuration directory
     * @param cloudLocalizationContext the parent cloud localization context
     */
    public EphemeralDeviceMappingsConfigProperties(Configured configuration,
        File configurationDirectory,
        LocalizationContext cloudLocalizationContext) {
      super(Configurations.EPHEMERAL_DEVICE_MAPPINGS_SECTION, configuration, cloudLocalizationContext);
      this.configurationDirectory = configurationDirectory;
      LocalizationContext localizationContext = new ChildLocalizationContext(
          cloudLocalizationContext, Configurations.EPHEMERAL_DEVICE_MAPPINGS_SECTION);
      setCustomMappingsPath(configuration.getConfigurationValue(CUSTOM_MAPPINGS_PATH,
          localizationContext));
    }

    @Override
    public String getDefaultDeviceNamePrefix() {
      return DEFAULT_DEVICE_NAME_PREFIX;
    }

    @Override
    public char getDefaultRangeStart() {
      return DEFAULT_RANGE_START_FOR_EPHEMERAL_DRIVES;
    }

    public String getCustomMappingsPath() {
      return customMappingsPath;
    }

    File getCustomMappingsFile() {
      return getCustomMappingsFile(customMappingsPath);
    }

    File getCustomMappingsFile(String customMappingsPath) {
      File customMappingsPathFile = new File(customMappingsPath);
      return customMappingsPathFile.isAbsolute() ?
          customMappingsPathFile :
          new File(configurationDirectory, customMappingsPath);
    }

    public void setCustomMappingsPath(String customMappingsPath) {
      if (customMappingsPath != null) {
        LOG.info("Overriding customMappingsPath={} (default {})",
            customMappingsPath, DEFAULT_CUSTOM_MAPPINGS_PATH);
        checkArgument(getCustomMappingsFile(customMappingsPath).exists(),
            "Custom ephemeral device mappings path " +
                customMappingsPath + " does not exist");
        this.customMappingsPath = customMappingsPath;
      }
    }
  }

  public static class EphemeralDeviceMappingsConfig {

    public static final String BUILT_IN_LOCATION =
        "classpath:/com/cloudera/director/aws/ec2/ephemeraldevicemappings.properties";

    protected EphemeralDeviceMappingsConfigProperties
        ephemeralDeviceMappingsConfigProperties;

    public PropertyResolver ephemeralDeviceMappingsResolver() {
      try {
        return PropertyResolvers.newMultiResourcePropertyResolver(
            BUILT_IN_LOCATION,
            "file:" + ephemeralDeviceMappingsConfigProperties.getCustomMappingsFile().getAbsolutePath()
        );
      } catch (IOException e) {
        throw new IllegalArgumentException("Could not load ephemeral device mappings", e);
      }
    }

    /**
     * Creates an ephemeral device mappings config with the specified parameters.
     *
     * @param ephemeralDeviceMappingsConfigProperties the config properties
     */
    public EphemeralDeviceMappingsConfig(
        EphemeralDeviceMappingsConfigProperties ephemeralDeviceMappingsConfigProperties) {
      this.ephemeralDeviceMappingsConfigProperties = ephemeralDeviceMappingsConfigProperties;
    }
  }

  EphemeralDeviceMappingsConfigProperties ephemeralDeviceMappingsConfigProperties;

  PropertyResolver ephemeralDeviceMappingsResolver;

  /**
   * Creates ephemeral device mappings with the specified parameters.
   *
   * @param configuration               the configuration
   * @param configurationDirectory      the plugin configuration directory
   * @param launcherLocalizationContext the parent launcher localization context
   */
  public EphemeralDeviceMappings(Configured configuration, File configurationDirectory,
      LocalizationContext launcherLocalizationContext) {
    this(new EphemeralDeviceMappingsConfigProperties(configuration, configurationDirectory,
        launcherLocalizationContext));
  }

  /**
   * Creates ephemeral device mappings with the specified parameters.
   *
   * @param ephemeralDeviceMappingsConfigProperties the config properties
   */
  public EphemeralDeviceMappings(
      EphemeralDeviceMappingsConfigProperties ephemeralDeviceMappingsConfigProperties) {
    this(ephemeralDeviceMappingsConfigProperties,
        new EphemeralDeviceMappingsConfig(ephemeralDeviceMappingsConfigProperties)
            .ephemeralDeviceMappingsResolver());
  }

  /**
   * Creates ephemeral device mappings with the specified parameters.
   *
   * @param ephemeralDeviceMappingsConfigProperties the config properties
   * @param ephemeralDeviceMappingsResolver         the ephemeral device mappings resolver
   */
  private EphemeralDeviceMappings(
      EphemeralDeviceMappingsConfigProperties ephemeralDeviceMappingsConfigProperties,
      PropertyResolver ephemeralDeviceMappingsResolver) {
    this.ephemeralDeviceMappingsConfigProperties = ephemeralDeviceMappingsConfigProperties;
    this.ephemeralDeviceMappingsResolver = ephemeralDeviceMappingsResolver;
  }

  private final DeviceNameUtils deviceNameUtils = new DeviceNameUtils();

  /**
   * Generates a list of block device mappings for all ephemeral drives for
   * the given instance type.
   *
   * @param instanceType       EC2 instance type
   * @param excludeDeviceNames set of device names that shouldn't be used for the block device mappings
   * @return list of block device mappings
   * @see <a href="http://docs.aws.amazon.com/AWSEC2/latest/UserGuide/block-device-mapping-concepts.html">Block Device Mapping</a>
   */
  public List<BlockDeviceMapping> getBlockDeviceMappings(String instanceType, Set<String> excludeDeviceNames) {
    checkNotNull(instanceType, "instanceType is null");

    Optional<Integer> optCount = getCount(instanceType);
    if (!optCount.isPresent()) {
      LOG.error("Unsupported instance type {}, add its ephemeral instance " +
          "volume count as a custom mapping; assuming zero", instanceType);
      return Collections.emptyList();
    }

    int count = optCount.get();
    if (count == 0) {
      return Collections.emptyList();
    }

    List<String> deviceNames = deviceNameUtils.getDeviceNames(
        ephemeralDeviceMappingsConfigProperties.getDeviceNamePrefix(),
        ephemeralDeviceMappingsConfigProperties.getRangeStart(),
        count, excludeDeviceNames);

    List<BlockDeviceMapping> result = Lists.newArrayListWithExpectedSize(count);
    int index = 0;
    for (String device : deviceNames) {
      result.add(new BlockDeviceMapping()
          .withDeviceName(device)
          .withVirtualName("ephemeral" + index));
      index += 1;
    }

    return result;
  }

  public List<BlockDeviceMapping> getBlockDeviceMappings(String instanceType) {
    return getBlockDeviceMappings(instanceType, Collections.emptySet());
  }

  private Optional<Integer> getCount(String instanceType) {
    return Optional.fromNullable(
        ephemeralDeviceMappingsResolver.getProperty(instanceType, Integer.class)
    );
  }

  /**
   * Gets a test instance of this class that uses only the given mapping.
   *
   * @param counts                      map of instance types to counts
   * @param launcherLocalizationContext the parent launcher localization context
   * @return new mapping object
   */
  public static EphemeralDeviceMappings getTestInstance(Map<String, Integer> counts,
      LocalizationContext launcherLocalizationContext) {
    Map<String, String> propertyMap =
        Maps.transformValues(counts, Functions.toStringFunction());
    PropertyResolver ephemeralDeviceMappingsResolver =
        PropertyResolvers.newMapPropertyResolver(propertyMap);
    File tempDir = Files.createTempDir();
    tempDir.deleteOnExit();
    EphemeralDeviceMappingsConfigProperties ephemeralDeviceMappingsConfigProperties =
        new EphemeralDeviceMappingsConfigProperties(new SimpleConfiguration(),
            tempDir, launcherLocalizationContext);
    return new EphemeralDeviceMappings(ephemeralDeviceMappingsConfigProperties,
        ephemeralDeviceMappingsResolver);
  }
} public class VirtualizationMappings implements Function<String, List<String>> {

  private static final Logger LOG = LoggerFactory.getLogger(VirtualizationMappings.class);

  private static final Splitter SPLITTER = Splitter.on(',').trimResults();
  private static final Joiner JOINER = Joiner.on(',');

  public static class VirtualizationMappingsConfigProperties {

    /**
     * Virtualization mappings configuration properties.
     */
    // Fully qualifying class name due to compiler bug
    public enum VirtualizationMappingsConfigurationPropertyToken
        implements com.cloudera.director.spi.v2.model.ConfigurationPropertyToken {

      /**
       * Path for the custom virtualization mappings file. Relative paths are
       * based on the plugin configuration directory.
       */
      CUSTOM_MAPPINGS_PATH(new SimpleConfigurationPropertyBuilder()
          .configKey("customMappingsPath")
          .name("Custom mappings path")
          .defaultDescription("The path for the custom virtualization mappings file. Relative " +
              "paths are based on the plugin configuration directory.")
          .build());

      /**
       * The configuration property.
       */
      private final ConfigurationProperty configurationProperty;

      /**
       * Creates a configuration property token with the specified parameters.
       *
       * @param configurationProperty the configuration property
       */
      VirtualizationMappingsConfigurationPropertyToken(
          ConfigurationProperty configurationProperty) {
        this.configurationProperty = configurationProperty;
      }

      @Override
      public ConfigurationProperty unwrap() {
        return configurationProperty;
      }
    }

    public static final String DEFAULT_CUSTOM_MAPPINGS_PATH =
        "ec2.virtualizationmappings.properties";

    private final File configurationDirectory;

    private String customMappingsPath = DEFAULT_CUSTOM_MAPPINGS_PATH;

    /**
     * Creates virtualization mappings config properties with the specified parameters.
     *
     * @param configuration            the configuration
     * @param configurationDirectory   the plugin configuration directory
     * @param cloudLocalizationContext the parent cloud localization context
     */
    public VirtualizationMappingsConfigProperties(Configured configuration,
        File configurationDirectory,
        LocalizationContext cloudLocalizationContext) {
      this.configurationDirectory = configurationDirectory;
      LocalizationContext localizationContext = new ChildLocalizationContext(
          cloudLocalizationContext, "virtualizationMappings");
      setCustomMappingsPath(
          configuration.getConfigurationValue(CUSTOM_MAPPINGS_PATH, localizationContext));
    }

    public String getCustomMappingsPath() {
      return customMappingsPath;
    }

    File getCustomMappingsFile() {
      return getCustomMappingsFile(customMappingsPath);
    }

    File getCustomMappingsFile(String customMappingsPath) {
      File customMappingsPathFile = new File(customMappingsPath);
      return customMappingsPathFile.isAbsolute() ?
          customMappingsPathFile :
          new File(configurationDirectory, customMappingsPath);
    }

    public void setCustomMappingsPath(String customMappingsPath) {
      if (customMappingsPath != null) {
        LOG.info("Overriding customMappingsPath=%s (default {})",
            customMappingsPath, DEFAULT_CUSTOM_MAPPINGS_PATH);
        checkArgument(getCustomMappingsFile(customMappingsPath).exists(),
            "Custom virtualization mappings path " +
                customMappingsPath + " does not exist");
        this.customMappingsPath = customMappingsPath;
      }
    }
  }

  public static class VirtualizationMappingsConfig {

    public static final String BUILT_IN_LOCATION =
        "classpath:/com/cloudera/director/aws/ec2/virtualizationmappings.properties";

    protected VirtualizationMappingsConfigProperties
        virtualizationMappingsConfigProperties;

    public PropertyResolver virtualizationMappingsResolver() {
      try {
        return PropertyResolvers.newMultiResourcePropertyResolver(
            BUILT_IN_LOCATION,
            "file:" + virtualizationMappingsConfigProperties.getCustomMappingsFile().getAbsolutePath()
        );
      } catch (IOException e) {
        throw new IllegalArgumentException("Could not load custom virtualization mappings", e);
      }
    }

    /**
     * Creates an virtualization mappings config with the specified parameters.
     *
     * @param virtualizationMappingsConfigProperties the config properties
     */
    public VirtualizationMappingsConfig(
        VirtualizationMappingsConfigProperties virtualizationMappingsConfigProperties) {
      this.virtualizationMappingsConfigProperties = virtualizationMappingsConfigProperties;
    }
  }

  VirtualizationMappingsConfigProperties virtualizationMappingsConfigProperties;

  PropertyResolver virtualizationMappingsResolver;

  /**
   * Creates virtualization mappings with the specified parameters.
   *
   * @param configuration          the configuration
   * @param configurationDirectory the plugin configuration directory
   * @param localizationContext    the localization context
   */
  public VirtualizationMappings(Configured configuration, File configurationDirectory,
      LocalizationContext localizationContext) {
    this(new VirtualizationMappingsConfigProperties(configuration, configurationDirectory,
        localizationContext));
  }

  /**
   * Creates virtualization mappings with the specified parameters.
   *
   * @param virtualizationMappingsConfigProperties the config properties
   */
  public VirtualizationMappings(VirtualizationMappingsConfigProperties virtualizationMappingsConfigProperties) {
    this(virtualizationMappingsConfigProperties, new VirtualizationMappingsConfig(virtualizationMappingsConfigProperties).virtualizationMappingsResolver());
  }

  /**
   * Creates virtualization mappings with the specified parameters.
   *
   * @param virtualizationMappingsConfigProperties the config properties
   * @param virtualizationMappingsResolver         the virtualization mappings resolver
   */
  private VirtualizationMappings(VirtualizationMappingsConfigProperties virtualizationMappingsConfigProperties,
      PropertyResolver virtualizationMappingsResolver) {
    this.virtualizationMappingsConfigProperties = virtualizationMappingsConfigProperties;
    this.virtualizationMappingsResolver = virtualizationMappingsResolver;
  }

  /**
   * Gets the instance types that support the given virtualization type.
   *
   * @param virtualizationType virtualization type
   * @return list of instance types
   * @throws IllegalArgumentException if the virtualization type is unknown
   */
  @Nonnull
  public List<String> apply(String virtualizationType) {
    checkNotNull(virtualizationType, "virtualizationType is null");
    String instanceTypeList =
        virtualizationMappingsResolver.getProperty(virtualizationType);
    if (instanceTypeList == null) {
      throw new IllegalArgumentException("Unknown virtualization type " +
          virtualizationType);
    }
    return SPLITTER.splitToList(instanceTypeList);
  }

  /**
   * Gets a test instance of this class that uses only the given mapping.
   *
   * @param instanceTypes       map of virtualization types to instance types
   * @param localizationContext the localization context
   * @return new mapping object
   */
  public static VirtualizationMappings getTestInstance(
      final Map<String, List<String>> instanceTypes, LocalizationContext localizationContext) {
    Map<String, String> propertyMap =
        Maps.transformValues(instanceTypes, JOINER::join);
    PropertyResolver virtualizationMappingsResolver =
        PropertyResolvers.newMapPropertyResolver(propertyMap);
    File tempDir = Files.createTempDir();
    tempDir.deleteOnExit();
    VirtualizationMappingsConfigProperties virtualizationMappingsConfigProperties =
        new VirtualizationMappingsConfigProperties(new SimpleConfiguration(),
            tempDir, localizationContext);
    return new VirtualizationMappings(virtualizationMappingsConfigProperties,
        virtualizationMappingsResolver);
  }
} public class EBSDeviceMappings {

  private final DeviceNameUtils deviceNameUtils = new DeviceNameUtils();

  static final String DEFAULT_EBS_DEVICE_NAME_PREFIX = "/dev/sd";
  static final char DEFAULT_EBS_DEVICE_NAME_START_CHAR = 'f';

  public static class EbsDeviceMappingsConfigProperties extends DeviceMappingsConfigProperties {

    /**
     * Creates EBS device mappings config properties with the specified parameters.
     *
     * @param configSection            the plugin configuration section
     * @param configuration            the configuration
     * @param cloudLocalizationContext the parent cloud localization context
     */
    public EbsDeviceMappingsConfigProperties(String configSection, Configured configuration,
        LocalizationContext cloudLocalizationContext) {
      super(configSection, configuration, cloudLocalizationContext);
    }

    @Override
    public String getDefaultDeviceNamePrefix() {
      return DEFAULT_EBS_DEVICE_NAME_PREFIX;
    }

    @Override
    public char getDefaultRangeStart() {
      return DEFAULT_EBS_DEVICE_NAME_START_CHAR;
    }
  }

  private final EbsDeviceMappingsConfigProperties ebsDeviceMappingsConfigProperties;

  /**
   * Creates EBS device mappings with the specified configuration.
   *
   * @param configuration            the configuration
   * @param cloudLocalizationContext the parent cloud localization context
   */
  public EBSDeviceMappings(Configured configuration, LocalizationContext cloudLocalizationContext) {
    this.ebsDeviceMappingsConfigProperties = new EbsDeviceMappingsConfigProperties(
        Configurations.EBS_DEVICE_MAPPINGS_SECTION, configuration, cloudLocalizationContext);
  }

  /**
   * Gets a list of EBS block device mappings with the specified parameters.
   *
   * @param template     EC2InstanceTemplate which contains information about
   *                     the EBS volumes requested
   * @param excludeDeviceNames set device names that shouldn't be used for the block device mappings
   * @return set of EBS BlockDeviceMapping
   */
  @SuppressWarnings("Guava")
  public List<BlockDeviceMapping> getBlockDeviceMappings(EC2InstanceTemplate template,
                                                         Set<String> excludeDeviceNames) {
    List<SystemDisk> systemDisks = template.getSystemDisks();

    int templateDiskCount = template.getEbsVolumeCount();
    int systemDiskCount = systemDisks.size();
    int totalVolumeCount = templateDiskCount + systemDiskCount;

    List<String> deviceNames = getDeviceNames(totalVolumeCount, excludeDeviceNames);

    List<BlockDeviceMapping> mappings = Lists.newArrayList();

    for (int i = 0; i < systemDisks.size(); i++) {
      String deviceName = deviceNames.get(i);
      SystemDisk systemDisk = systemDisks.get(i);
      mappings.add(systemDisk.toBlockDeviceMapping(deviceName));
    }

    for (int i = 0; i < templateDiskCount; i++) {
      String deviceName = deviceNames.get(i + systemDiskCount);

      EbsBlockDevice ebs = new EbsBlockDevice()
          .withVolumeType(template.getEbsVolumeType())
          .withVolumeSize(template.getEbsVolumeSizeGiB())
          .withEncrypted(template.isEnableEbsEncryption())
          .withDeleteOnTermination(true);

      Optional<Integer> iops = template.getEbsIops();
      Optional<String> kmsKeyid = template.getEbsKmsKeyId();

      if (iops.isPresent()) {
        ebs = ebs.withIops(iops.get());
      }
      if (kmsKeyid.isPresent()) {
        ebs = ebs.withKmsKeyId(kmsKeyid.get());
      }

      BlockDeviceMapping mapping = new BlockDeviceMapping()
          .withDeviceName(deviceName)
          .withEbs(ebs);

      mappings.add(mapping);
    }

    return mappings;
  }

  List<String> getDeviceNames(int count, Set<String> filter) {
    return deviceNameUtils.getDeviceNames(
        ebsDeviceMappingsConfigProperties.getDeviceNamePrefix(),
        ebsDeviceMappingsConfigProperties.getRangeStart(),
        count, filter
    );
  }


  /**
   * Gets an instance of this class that uses only the given EBS device mapping configuration.
   *
   * @param configuration       EBS device mapping configuration
   * @param localizationContext the localization context
   * @return new EBS metadata object
   */
  public static EBSDeviceMappings getDefaultInstance(
      final Map<String, String> configuration, LocalizationContext localizationContext) {
    return new EBSDeviceMappings(new SimpleConfiguration(configuration), localizationContext);
  }
} @SuppressWarnings("PMD.UnusedPrivateField")
public class EBSMetadata implements Function<String, EBSMetadata.EbsVolumeMetadata> {

  private static final Logger LOG = LoggerFactory.getLogger(EBSMetadata.class);

  /**
   * Contains metadata information for an EBS volume type.
   */
  @SuppressWarnings("Guava")
  public static class EbsVolumeMetadata {
    private final String type;
    private final int minSizeGiB;
    private final int maxSizeGiB;
    private final Optional<Integer> minIops;
    private final Optional<Integer> maxIops;

    public EbsVolumeMetadata(String type, int minSizeGiB, int maxSizeGiB) {
      this.type = Objects.requireNonNull(type, "type is null");
      this.minSizeGiB = minSizeGiB;
      this.maxSizeGiB = maxSizeGiB;
      this.minIops = Optional.absent();
      this.maxIops = Optional.absent();
    }

    public EbsVolumeMetadata(String type, int minSizeGiB, int maxSizeGiB, int minIops, int maxIops) {
      this.type = Objects.requireNonNull(type, "type is null");
      this.minSizeGiB = minSizeGiB;
      this.maxSizeGiB = maxSizeGiB;
      this.minIops = Optional.of(minIops);
      this.maxIops = Optional.of(maxIops);
    }

    public String getType() {
      return type;
    }

    public int getMinSizeGiB() {
      return minSizeGiB;
    }

    public int getMaxSizeGiB() {
      return maxSizeGiB;
    }

    public Optional<Integer> getMinIops() {
      return minIops;
    }

    public Optional<Integer> getMaxIops() {
      return maxIops;
    }

    @SuppressWarnings("SimplifiableIfStatement")
    @Override
    public boolean equals(Object o) {
      if (this == o) return true;
      if (o == null || getClass() != o.getClass()) return false;

      EbsVolumeMetadata that = (EbsVolumeMetadata) o;

      if (minSizeGiB != that.minSizeGiB) return false;
      if (maxSizeGiB != that.maxSizeGiB) return false;
      if (!type.equals(that.type)) return false;
      if (!minIops.equals(that.minIops)) return false;
      return maxIops.equals(that.maxIops);
    }

    @Override
    public int hashCode() {
      int result = type.hashCode();
      result = 31 * result + minSizeGiB;
      result = 31 * result + maxSizeGiB;
      result = 31 * result + minIops.hashCode();
      result = 31 * result + maxIops.hashCode();
      return result;
    }
  }

  public static class EBSMetadataConfigProperties {

    /**
     * EBS metadata configuration properties.
     */
    // Fully qualifying class name due to compiler bug
    public enum EBSMetadataConfigurationPropertyToken
        implements com.cloudera.director.spi.v2.model.ConfigurationPropertyToken {

      /**
       * Path for the custom EBS metadata file. Relative paths are based on the
       * plugin configuration directory.
       */
      CUSTOM_EBS_METADATA_PATH(new SimpleConfigurationPropertyBuilder()
          .configKey("ebsMetadataPath")
          .name("Custom EBS metadata path")
          .defaultDescription("The path for the custom ebs metadata. Relative paths are based " +
              "on the plugin configuration directory.")
          .build());

      /**
       * The configuration property.
       */
      private final ConfigurationProperty configurationProperty;

      /**
       * Creates a configuration property token with the specified parameters.
       *
       * @param configurationProperty the configuration property
       */
      EBSMetadataConfigurationPropertyToken(
          ConfigurationProperty configurationProperty) {
        this.configurationProperty = configurationProperty;
      }

      @Override
      public ConfigurationProperty unwrap() {
        return configurationProperty;
      }
    }

    public static final String DEFAULT_CUSTOM_EBS_METADATA_PATH =
        "ec2.ebs.ebsmetadata.properties";

    private final File configurationDirectory;

    private String ebsMetadataPath = DEFAULT_CUSTOM_EBS_METADATA_PATH;

    /**
     * Creates EBS metadata config properties with the specified parameters.
     *
     * @param configuration            the configuration
     * @param configurationDirectory   the plugin configuration directory
     * @param cloudLocalizationContext the parent cloud localization context
     */
    public EBSMetadataConfigProperties(Configured configuration,
        File configurationDirectory,
        LocalizationContext cloudLocalizationContext) {
      this.configurationDirectory = configurationDirectory;
      LocalizationContext localizationContext = new ChildLocalizationContext(
          cloudLocalizationContext, "ebsMetadata");
      setCustomEbsMetadataPath(
          configuration.getConfigurationValue(EBSMetadataConfigurationPropertyToken.CUSTOM_EBS_METADATA_PATH,
              localizationContext));
    }

    public String getCustomEbsMetadataPath() {
      return ebsMetadataPath;
    }

    File getCustomEbsMetadataFile() {
      return getCustomEbsMetadataFile(ebsMetadataPath);
    }

    File getCustomEbsMetadataFile(String ebsMetadataPath) {
      File ebsMetadataPathFile = new File(ebsMetadataPath);
      return ebsMetadataPathFile.isAbsolute() ?
          ebsMetadataPathFile :
          new File(configurationDirectory, ebsMetadataPath);
    }

    public void setCustomEbsMetadataPath(String ebsMetadataPath) {
      if (ebsMetadataPath != null) {
        LOG.info("Overriding ebsMetadataPath={} (default {})",
            ebsMetadataPath, DEFAULT_CUSTOM_EBS_METADATA_PATH);
        checkArgument(getCustomEbsMetadataFile(ebsMetadataPath).exists(),
            "Custom EBS metadata path " +
                ebsMetadataPath + " does not exist");
        this.ebsMetadataPath = ebsMetadataPath;
      }
    }
  }

  public static class EBSMetadataConfig {

    public static final String BUILT_IN_LOCATION =
        "classpath:/com/cloudera/director/aws/ec2/ebs/ebsmetadata.properties";

    protected EBSMetadataConfigProperties ebsMetadataConfigProperties;

    public PropertyResolver ebsMetadataResolver() {
      try {
        return PropertyResolvers.newMultiResourcePropertyResolver(
            BUILT_IN_LOCATION,
            "file:" + ebsMetadataConfigProperties.getCustomEbsMetadataFile().getAbsolutePath()
        );
      } catch (IOException e) {
        throw new IllegalArgumentException("Could not load custom ebs metadata", e);
      }
    }

    /**
     * Creates EBS metadata config with the specified parameters.
     *
     * @param ebsMetadataConfigProperties the config properties
     */
    public EBSMetadataConfig(EBSMetadataConfigProperties ebsMetadataConfigProperties) {
      this.ebsMetadataConfigProperties = ebsMetadataConfigProperties;
    }
  }

  private final EBSMetadataConfigProperties ebsMetadataConfigProperties;

  private final PropertyResolver ebsMetadataResolver;

  /**
   * Creates EBS metadata with the specified parameters.
   *
   * @param configuration          the configuration
   * @param configurationDirectory the plugin configuration directory
   * @param localizationContext    the localization context
   */
  public EBSMetadata(Configured configuration, File configurationDirectory,
      LocalizationContext localizationContext) {
    this(new EBSMetadataConfigProperties(configuration, configurationDirectory,
        localizationContext));
  }

  /**
   * Creates EBS metadata with the specified parameters.
   *
   * @param ebsMetadataConfigProperties the config properties
   */
  public EBSMetadata(EBSMetadataConfigProperties ebsMetadataConfigProperties) {
    this(ebsMetadataConfigProperties, new EBSMetadataConfig(ebsMetadataConfigProperties).ebsMetadataResolver());
  }

  /**
   * Creates EBS metadata with the specified parameters.
   *
   * @param ebsMetadataConfigProperties the config properties
   * @param ebsMetadataResolver         the ebs metadata resolver
   */
  private EBSMetadata(EBSMetadataConfigProperties ebsMetadataConfigProperties,
      PropertyResolver ebsMetadataResolver) {
    this.ebsMetadataConfigProperties = ebsMetadataConfigProperties;
    this.ebsMetadataResolver = ebsMetadataResolver;
  }

  /**
   * Gets the metadata for an EBS volume type. A custom definition overrides a
   * built-in definition.
   *
   * @param volumeType ebs volume type
   * @return metadata associated with the volume type
   * @throws NullPointerException  if no metadata could be found for a volume type
   * @throws IllegalStateException if the metadata for a volume type has invalid format
   */
  @Override
  public EbsVolumeMetadata apply(String volumeType) {
    String strCapacityRange = ebsMetadataResolver.getProperty(volumeType);
    Objects.requireNonNull(strCapacityRange, String.format("Could not get metadata for volume type %s", volumeType));

    Range capacityRange = Range.resolveRange(volumeType, strCapacityRange);
    int minSizeGiB = capacityRange.getMin();
    int maxSizeGiB = capacityRange.getMax();

    if (volumeType.equals("io1")) {
      String key = "io1-iops";
      String strIopsMetadata = ebsMetadataResolver.getProperty(key);
      Objects.requireNonNull(strIopsMetadata, String.format("Could not get metadata for %s", key));

      Range iopsRange = Range.resolveRange(key, strIopsMetadata);
      return new EbsVolumeMetadata(volumeType, minSizeGiB, maxSizeGiB, iopsRange.getMin(), iopsRange.getMax());
    }

    return new EbsVolumeMetadata(volumeType, minSizeGiB, maxSizeGiB);
  }

  private static class Range {
    private int min;
    private int max;

    private Range(int min, int max) {
      this.min = min;
      this.max = max;
    }

    public int getMin() {
      return min;
    }

    public int getMax() {
      return max;
    }

    static Range resolveRange(String key, String rangeStr) {

      // range string is expected to be in the form "minValue-maxValue"

      String[] parts = rangeStr.trim().split("-");
      checkState(parts.length == 2, String.format("invalid format for %s=%s, expecting " +
          "exactly 2 parts for the value", key, rangeStr));

      int min, max;
      try {
        min = Integer.parseInt(parts[0]);
      } catch (NumberFormatException ex) {
        throw new IllegalStateException(
            String.format("invalid format for %s=%s, %s should be an integer", key, rangeStr, parts[0])
        );
      }

      try {
        max = Integer.parseInt(parts[1]);
      } catch (NumberFormatException ex) {
        throw new IllegalStateException(
            String.format("invalid format for %s=%s, %s should be an integer", key, rangeStr, parts[1])
        );
      }

      checkState(min > 0, String.format("invalid format for %s=%s, %d should " +
          "be a positive integer", key, rangeStr, min));

      checkState(max > 0, String.format("invalid format for %s=%s, %d should " +
          "be a positive integer", key, rangeStr, max));

      checkState(max >= min, String.format("invalid format for %s=%s, maximum size %d should " +
          "be greater than or equal to minimum size %d", key, rangeStr, max, min));

      return new Range(min, max);
    }
  }


  /**
   * Gets an instance of this class that uses only the given ebs metadata.
   *
   * @param metadata            map of ebs volume type to it's associated metadata
   * @param localizationContext the localization context
   * @return new ebs metadata object
   */
  public static EBSMetadata getDefaultInstance(
      final Map<String, String> metadata, LocalizationContext localizationContext) {
    PropertyResolver ebsMetadataResolver =
        PropertyResolvers.newMapPropertyResolver(metadata);
    File tempDir = Files.createTempDir();
    tempDir.deleteOnExit();
    EBSMetadataConfigProperties ebsMetadataConfigProperties =
        new EBSMetadataConfigProperties(new SimpleConfiguration(), tempDir, localizationContext);
    return new EBSMetadata(ebsMetadataConfigProperties, ebsMetadataResolver);
  }
} public final class NetworkRules {

  private static final Logger LOG = LoggerFactory.getLogger(NetworkRules.class);

  /**
   * The root key for network rules config.
   */
  private static final String ROOT_KEY = NetworkRules.class.getCanonicalName();

  /**
   * The table holding all network rules.
   */
  private final Table<Direction, AccessType, Iterable<NetworkRule>> rules;

  /**
   * A constant represents an empty set of network rules.
   */
  public static final NetworkRules EMPTY_RULES
      = new NetworkRules(ImmutableTable.of());

  /**
   * Internal constructor.
   *
   * @param rules a rules table
   */
  private NetworkRules(Table<Direction, AccessType, Iterable<NetworkRule>> rules) {
    Preconditions.checkNotNull(rules, "rules is null");
    this.rules = rules;
  }

  /**
   * Constructs a NetworkRules object using the given HOCON config.
   *
   * @param config a HOCON config
   * @return a NetworkRules object
   */
  public static NetworkRules fromConfig(Config config) {
    if (config == null) {
      return EMPTY_RULES;
    }

    try {
      return new NetworkRules(parse(config.getConfig(ROOT_KEY)));
    } catch (ConfigException | IllegalArgumentException e) {
      String msg = "Failed to parse network rules";
      LOG.error(msg, e);
      throw new IllegalArgumentException(msg, e);
    }
  }

  /**
   * Parses a HOCON config to a table of network rules.
   */
  private static Table<Direction, AccessType, Iterable<NetworkRule>> parse(Config config) {
    Table<Direction, AccessType, Iterable<NetworkRule>> rules =
        ArrayTable.create(EnumSet.allOf(Direction.class), EnumSet.allOf(AccessType.class));
    for (Direction direction : Direction.values()) {
      for (AccessType access : AccessType.values()) {
        populateRules(config, direction, access, rules);
      }
    }
    return rules;
  }

  /**
   * Populates the rule table.
   *
   * @param config    the HOCON config
   * @param direction the network direction
   * @param access    the network rule access type
   * @param rules     the rules table
   */
  private static void populateRules(Config config, Direction direction, AccessType access,
      Table<Direction, AccessType, Iterable<NetworkRule>> rules) {
    List<? extends Config> configList =
        config.getConfig(direction.toString().toLowerCase()).
            getConfigList(access.toString().toLowerCase());
    Set<NetworkRule> cell = Sets.newHashSet();
    for (Config val : configList) {
      Iterables.addAll(cell, NetworkRule.fromConfig(val, access));
    }
    rules.put(direction, access, cell);
  }

  /**
   * Returns the number of network rules.
   *
   * @return the number of network rules
   */
  public int size() {
    return this.rules.size();
  }

  /**
   * Checks if the network rule set is empty.
   *
   * @return true if it contains no network rules
   */
  public boolean isEmpty() {
    return this.rules.isEmpty();
  }

  /**
   * Retrieves rules based on the direction of network traffic.
   * <p>
   * <p>The returned rules are just copies of actual network rules. Changes to the
   * returned collection will not update the underlying NetworkRules, and vice versa.
   *
   * @param direction the network direction
   * @return the associated list of network rules
   */
  public Iterable<NetworkRule> getRules(Direction direction) {
    if (this.rules.isEmpty()) {
      return Collections.emptySet();
    }
    Set<NetworkRule> ruleSet = Sets.newHashSet();
    for (AccessType access : AccessType.values()) {
      Iterables.addAll(ruleSet, getRules(direction, access));
    }
    return ruleSet;
  }

  /**
   * Retrieves rules based on the network direction and access type.
   * <p>
   * <p>The returned rules are just copies of actual network rules. Changes to the
   * returned collection will not update the underlying NetworkRules, and vice versa.
   *
   * @param direction the network direction
   * @param access    the network rule access type
   * @return the associated list of network rules
   */
  public Iterable<NetworkRule> getRules(Direction direction, AccessType access) {
    return this.rules.isEmpty()
        ? Collections.emptySet()
        : Sets.newHashSet(this.rules.get(direction, access));
  }
}
import com.cloudera.director.aws.AWSFilters;
import com.cloudera.director.aws.AWSTimeouts;
import com.cloudera.director.aws.CustomTagMappings;
import com.cloudera.director.aws.Tags;
import com.cloudera.director.aws.clientprovider.AWSKMSClientProvider;
import com.cloudera.director.aws.clientprovider.AWSSTSClientProvider;
import com.cloudera.director.aws.clientprovider.AmazonAutoScalingClientProvider;
import com.cloudera.director.aws.clientprovider.AmazonEC2ClientProvider;
import com.cloudera.director.aws.clientprovider.AmazonIdentityManagementClientProvider;
import com.cloudera.director.aws.ec2.EC2InstanceTemplate;
import com.cloudera.director.aws.ec2.provider.EC2Provider;
import com.cloudera.director.aws.ec2.provider.EC2ProviderParameterizedLiveTest;
import com.cloudera.director.aws.ec2.EphemeralDeviceMappings;
import com.cloudera.director.aws.ec2.VirtualizationMappings;
import com.cloudera.director.aws.ec2.ebs.EBSDeviceMappings;
import com.cloudera.director.aws.ec2.ebs.EBSMetadata;
import com.cloudera.director.aws.network.NetworkRules;
import com.cloudera.director.aws.shaded.com.amazonaws.ClientConfiguration;
import com.cloudera.director.aws.shaded.com.amazonaws.ClientConfigurationFactory;
import com.cloudera.director.aws.shaded.com.amazonaws.auth.AWSCredentialsProvider;
import com.cloudera.director.aws.shaded.com.typesafe.config.Config;
import com.cloudera.director.aws.shaded.com.typesafe.config.ConfigValueFactory;
import com.cloudera.director.spi.v2.model.ConfigurationPropertyToken;
import com.cloudera.director.spi.v2.model.Configured;
import com.cloudera.director.spi.v2.model.InstanceState;
import com.cloudera.director.spi.v2.model.InstanceStatus;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import java.util.Arrays;
import java.util.Collection;
import java.util.EnumSet;
import java.util.Map;
import java.util.Set;
import java.util.logging.Logger;
import static com.cloudera.director.spi.v2.provider.Launcher.DEFAULT_PLUGIN_LOCALIZATION_CONTEXT;
// (c) Copyright 2017 Cloudera, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package com.cloudera.director.aws.test;




/**
 * Contains common functions used to test EC2Provider.
 */
public class EC2ProviderCommon {

  private static final Logger LOG = Logger.getLogger(EC2ProviderParameterizedLiveTest.class.getName());

  private static final String CUSTOM_TEST_NAME_TAG = "TestName";
  public static final String EXTRA_TAG = "extra";
  private static final String CUSTOM_TEST_EXTRA_TAG = "TestExtra";

  private static final Config CUSTOM_TAG_MAPPING_CONFIG =
      ConfigValueFactory.fromMap(ImmutableMap.<String, String>builder()
          .put(Tags.ResourceTags.RESOURCE_NAME.getTagKey(), CUSTOM_TEST_NAME_TAG)
          .put(EXTRA_TAG, CUSTOM_TEST_EXTRA_TAG)
          .build()).toConfig();

  /**
   * Returns whether the live test property is true
   *
   * @return whether the live test property is true
   */
  public static boolean isLive() {
    String liveString = System.getProperty("test.aws.live");
    return Boolean.parseBoolean(liveString);
  }

  /**
   * Waits until the instance state is running or has a terminal state. Returns
   * the resulting instance states.
   *
   * @param ec2Provider              the EC2 provider
   * @param instanceTemplate         the instance template
   * @param virtualInstanceIds       the virtual instance ids
   * @return the map from virtual instance ID to instance state
   * @throws InterruptedException if the process is interrupted
   */
  public static Map<String, InstanceState> waitUntilRunningOrTerminal(EC2Provider ec2Provider,
      EC2InstanceTemplate instanceTemplate, Collection<String> virtualInstanceIds) throws InterruptedException {
    return waitForInstanceStates(
        ec2Provider,
        instanceTemplate,
        virtualInstanceIds,
        EnumSet.of(InstanceStatus.RUNNING),
        EnumSet.of(InstanceStatus.DELETED, InstanceStatus.FAILED, InstanceStatus.STOPPED)
    );
  }

  /**
   * Determines instance state for the specified instances, waiting until
   * all instances have a desired or terminal status.
   *
   * @param ec2Provider              the EC2 provider
   * @param instanceTemplate         the instance template
   * @param virtualInstanceIds       the virtual instance ids
   * @param desiredInstanceStatuses  the desired instance statuses
   * @param terminalInstanceStatuses the terminal instance statuses
   * @return the map from virtual instance ID to instance state
   * @throws InterruptedException if the process is interrupted
   */
  public static Map<String, InstanceState> waitForInstanceStates(EC2Provider ec2Provider,
                                                                 EC2InstanceTemplate instanceTemplate,
                                                                 Collection<String> virtualInstanceIds,
     Collection<InstanceStatus> desiredInstanceStatuses, Collection<InstanceStatus> terminalInstanceStatuses)
      throws InterruptedException {
    Map<String, InstanceState> instanceStates = Maps.newHashMap();
    Set<String> pendingInstanceIds = Sets.newHashSet(virtualInstanceIds);
    while (!pendingInstanceIds.isEmpty()) {
      Map<String, InstanceState> currentInstanceStates =
          ec2Provider.getInstanceState(instanceTemplate, pendingInstanceIds);

      for (Map.Entry<String, InstanceState> entry : currentInstanceStates.entrySet()) {
        String virtualInstanceId = entry.getKey();
        InstanceState instanceState = entry.getValue();
        InstanceStatus instanceStatus = instanceState.getInstanceStatus();

        if (desiredInstanceStatuses.contains(instanceStatus)) {
          LOG.info("Instance: " + virtualInstanceId + " has desired status: " + instanceStatus);
          instanceStates.put(virtualInstanceId, instanceState);
          pendingInstanceIds.remove(virtualInstanceId);

        } else if (terminalInstanceStatuses.contains(instanceStatus)) {
          LOG.info("Instance: " + virtualInstanceId + " has terminal status: " + instanceStatus);
          instanceStates.put(virtualInstanceId, instanceState);
          pendingInstanceIds.remove(virtualInstanceId);
        }
      }
      if (!pendingInstanceIds.isEmpty()) {
        LOG.info(pendingInstanceIds.size()
            + " instance(s) still pending. Sleeping for 15 seconds...");
        Thread.sleep(15000);
      }
    }
    return instanceStates;
  }

  /**
   * Puts a configuration property and value into a configuration map
   *
   * @param configMap the configuration map
   * @param propertyToken the configuration property token
   * @param value the configuration value
   */
  public static void putConfig(Map<String, String> configMap, ConfigurationPropertyToken propertyToken,
                                String value) {
    if (value != null) {
      configMap.put(propertyToken.unwrap().getConfigKey(), value);
    }
  }

  /**
   * Returns an instance of the @{@link EC2Provider} with the specified configs.
   *
   * @param configuration the configurations
   * @param credentialsProvider the AWS credentials provider
   * @return an instance of the @{@link EC2Provider}
   */
  public static EC2Provider getEc2Provider(Configured configuration,
                                           AWSCredentialsProvider credentialsProvider) {
    return getEc2Provider(configuration, credentialsProvider,
        new AmazonEC2ClientProvider(credentialsProvider, new ClientConfigurationFactory().getConfig()));
  }

  /**
   * Returns an instance of the @{@link EC2Provider} with the specified configs.
   *
   * @param configuration the configurations
   * @param credentialsProvider the AWS credentials provider
   * @param amazonEC2ClientProvider the AWS EC2 client provider
   * @return an instance of the @{@link EC2Provider}
   */
  public static EC2Provider getEc2Provider(Configured configuration,
      AWSCredentialsProvider credentialsProvider, AmazonEC2ClientProvider amazonEC2ClientProvider) {
    // Configure ephemeral device mappings
    EphemeralDeviceMappings ephemeralDeviceMappings =
        EphemeralDeviceMappings.getTestInstance(ImmutableMap.of("m3.medium", 1, "i3.4xlarge", 2),
            DEFAULT_PLUGIN_LOCALIZATION_CONTEXT);

    // Configure ebs device mappings
    EBSDeviceMappings ebsDeviceMappings =
        EBSDeviceMappings.getDefaultInstance(ImmutableMap.<String, String>of(),
            DEFAULT_PLUGIN_LOCALIZATION_CONTEXT);

    // Configure ebs metadata
    EBSMetadata ebsMetadata =