Java 类com.amazonaws.services.ec2.model.SpotInstanceRequest 实例源码

/**
 * Checks whether the specified Spot instance request is visible, <em>i.e.</em> can
 * be located by a describe Spot instance request call.
 *
 * @param spotInstanceRequestId the Spot instance request ID
 * @return whether the Spot instance request is visible, <em>i.e.</em> can
 * be located by a describe Spot instance request call
 */
@VisibleForTesting
protected boolean spotInstanceRequestExists(String spotInstanceRequestId) {
  DescribeSpotInstanceRequestsResult result;
  try {
    result = client.describeSpotInstanceRequests(
        new DescribeSpotInstanceRequestsRequest()
            .withSpotInstanceRequestIds(spotInstanceRequestId)
    );
  } catch (AmazonServiceException e) {
    if ("InvalidSpotInstanceRequestID.NotFound".equals(e.getErrorCode())) {
      return false;
    }
    throw AWSExceptions.propagate(e);
  }

  for (SpotInstanceRequest spotInstanceRequest : result.getSpotInstanceRequests()) {
    if (spotInstanceRequestId.equals(spotInstanceRequest.getSpotInstanceRequestId())) {
      return true;
    }
  }
  return false;
}

public SpotState getSpotStatus() {
    if (spotRequestId == null)
        return SpotState.SPOT_REQUEST_NOT_FOUND;

    com.amazonaws.services.ec2.AmazonEC2 client = AmazonEC2.connect();

    DescribeSpotInstanceRequestsRequest spotInstanceReq = new DescribeSpotInstanceRequestsRequest();
    List<String> spotInstanceRequestIds = new ArrayList<String>();
    spotInstanceRequestIds.add(spotRequestId);
    spotInstanceReq.setSpotInstanceRequestIds(spotInstanceRequestIds);
    DescribeSpotInstanceRequestsResult res = client
            .describeSpotInstanceRequests(spotInstanceReq);

    List<SpotInstanceRequest> reqs = res.getSpotInstanceRequests();
    if (reqs.size() > 0) {
        SpotInstanceRequest req = reqs.get(0);
        id = req.getInstanceId();
        return SpotState.valueFromRequest(req);
    } else {
        return SpotState.SPOT_REQUEST_NOT_FOUND;
    }
}

private List<SpotInstanceOfferResult> toSpotInstanceResults(SpotInstanceOffer request, List<SpotInstanceRequest> spotInstanceRequests)
{
    List<SpotInstanceOfferResult> resultList = new ArrayList<SpotInstanceOfferResult>();

    for (SpotInstanceRequest spotInstanceResult : spotInstanceRequests)
    {
        SpotInstanceOfferResult result = new SpotInstanceOfferResult().setCreateTime(new Date()).setOfferRequest(request);

        result.setState(SpotInstanceOfferStateType.valueOfFrom(spotInstanceResult.getState()));
        result.setSpotRequestId(spotInstanceResult.getSpotInstanceRequestId());

        result.setStatus(new SpotInstanceOfferStatus()
                .setCode(spotInstanceResult.getStatus().getCode())
                .setMessage(spotInstanceResult.getStatus().getMessage())
                .setUpdateTime(spotInstanceResult.getStatus().getUpdateTime()));

        resultList.add(result);
    }

    return resultList;
}

@Override
public List<Machine> startMachines(int count) throws StartMachinesException, CloudPoolDriverException {
    checkState(isConfigured(), "attempt to use unconfigured driver");
    List<Machine> startedMachines = new ArrayList<>();
    try {
        Ec2ProvisioningTemplate template = provisioningTemplate();
        // add pool tag to recognize spot requests as pool members
        template = template.withTag(ScalingTags.CLOUD_POOL_TAG, getPoolName());

        List<SpotInstanceRequest> spotRequests = this.client.placeSpotRequests(cloudApiSettings().getBidPrice(),
                template, count);
        List<String> spotIds = spotRequests.stream().map(SpotInstanceRequest::getSpotInstanceRequestId)
                .collect(Collectors.toList());
        LOG.info("placed spot requests: {}", spotIds);
        for (SpotInstanceRequest spotRequest : spotRequests) {
            InstancePairedSpotRequest pairedSpotRequest = new InstancePairedSpotRequest(spotRequest, null);
            startedMachines.add(InstancePairedSpotRequestToMachine.convert(pairedSpotRequest));
        }
    } catch (Exception e) {
        throw new StartMachinesException(count, startedMachines, e);
    }

    return startedMachines;
}

/**
 * Returns all {@link SpotInstanceRequest}s in the pool that are in any of a
 * given set of states.
 *
 * @param inStates
 *            The spot request states of interest.
 * @return The {@link SpotInstanceRequest}s paired with their
 *         {@link Instance}.
 * @throws CloudPoolDriverException
 */
private List<InstancePairedSpotRequest> getPoolSpotRequests(List<String> states) throws CloudPoolDriverException {
    try {
        // only include spot requests with cloud pool tag
        Filter poolFilter = new Filter().withName(ScalingFilters.CLOUD_POOL_TAG_FILTER).withValues(getPoolName());
        // only include spot requests in any of the given states
        Filter stateFilter = new Filter().withName(ScalingFilters.SPOT_REQUEST_STATE_FILTER).withValues(states);

        List<SpotInstanceRequest> spotRequests = this.client
                .getSpotInstanceRequests(asList(poolFilter, stateFilter));
        List<InstancePairedSpotRequest> requestInstancePairs = pairUpWithInstances(spotRequests);
        return requestInstancePairs;
    } catch (Exception e) {
        throw new CloudPoolDriverException(
                format("failed to retrieve machines in cloud pool \"%s\": %s", getPoolName(), e.getMessage()), e);
    }
}

@Override
public List<SpotInstanceRequest> getSpotInstanceRequests(Collection<Filter> filters) throws AmazonClientException {
    List<SpotInstanceRequest> requests = new ArrayList<>(this.spotRequests.values());
    // filter out requests that don't match all filters
    Iterator<SpotInstanceRequest> iterator = requests.iterator();
    while (iterator.hasNext()) {
        SpotInstanceRequest request = iterator.next();
        for (Filter filter : filters) {
            if (!matches(filter, request)) {
                // filter out
                iterator.remove();
                break;
            }
        }
    }
    // return sorted on identifier to ease verifications in tests
    Collections.sort(requests, (o1, o2) -> o1.getSpotInstanceRequestId().compareTo(o2.getSpotInstanceRequestId()));
    return requests;
}

@Override
public List<SpotInstanceRequest> placeSpotRequests(double bidPrice, Ec2ProvisioningTemplate template, int count)
        throws AmazonClientException {
    List<SpotInstanceRequest> requests = new ArrayList<>();
    for (int i = 0; i < count; i++) {
        String id = "sir-" + (System.currentTimeMillis() + count);
        SpotInstanceRequest request = new SpotInstanceRequest().withSpotInstanceRequestId(id)
                .withLaunchSpecification(new LaunchSpecification().withInstanceType(template.getInstanceType()))
                .withState(SpotInstanceState.Open).withSpotPrice(String.valueOf(bidPrice));

        if (!template.getTags().isEmpty()) {
            List<Tag> tags = new ArrayList<>();
            for (Entry<String, String> tag : template.getTags().entrySet()) {
                tags.add(new Tag(tag.getKey(), tag.getValue()));
            }
            request.withTags(tags);
        }
        this.spotRequests.put(id, request);
        requests.add(request);
    }
    return requests;
}

/**
 * Make sure that in the case when all bid prices are correct no
 * cancellation {@link Alert}s are sent out.
 */
@Test
public void cancellationOfWrongPricedRequestsWhenAllRequestsHaveCorrectPrice() {
    this.driver = new SpotPoolDriver(this.fakeClient, this.executor, this.mockEventBus);
    this.driver.configure(config());
    double currentBidPrice = this.driver.cloudApiSettings().getBidPrice();

    // unfulfilled, right bid price => should not be replaced
    SpotInstanceRequest spot1 = spotRequest("sir-1", "open", null, POOL1_TAG);
    spot1.setSpotPrice(String.valueOf(currentBidPrice));

    List<Instance> instances = asList();
    this.fakeClient.setupFakeAccount(asList(spot1), instances);

    List<String> cancelledRequests = this.driver.cancelWrongPricedRequests();
    assertThat(cancelledRequests.isEmpty(), is(true));

    // verify event posted on event bus
    verifyZeroInteractions(this.mockEventBus);
}

public static void main(String[] args) {
    AwsSpotClient client = new AwsSpotClient();
    client.configure(awsAccessKeyId, awsSecretAccessKey, region, new ClientConfiguration());

    LOG.info("Testing spot request limit in region {}", region);
    BaseCloudPoolConfig config = JsonUtils.toObject(JsonUtils.parseJsonFile(new File("myconfig.json")),
            BaseCloudPoolConfig.class);
    Ec2ProvisioningTemplate provisioningTemplate = JsonUtils.toObject(config.getProvisioningTemplate(),
            Ec2ProvisioningTemplate.class);
    LOG.info("config: {}", provisioningTemplate);
    List<SpotInstanceRequest> placedRequests = new ArrayList<>();
    int MAX = 200;
    try {
        placedRequests.addAll(client.placeSpotRequests(0.001, provisioningTemplate, MAX));
    } catch (Exception e) {
        LOG.error("failed: {}", e.getMessage(), e);
    } finally {
        LOG.info("cancelling all placed requests ...");
        for (SpotInstanceRequest request : placedRequests) {
            client.cancelSpotRequests(Arrays.asList(request.getSpotInstanceRequestId()));
        }
        LOG.info("cancelled all requests.");
    }
}

/**
 * Returns a {@link Predicate} that returns <code>true</code> if every spot
 * instance request in a collection of spot instance requests is in a given
 * set of {@link SpotInstanceState}s.
 *
 * @param states
 *            The acceptable states.
 * @return
 */
public static Predicate<List<SpotInstanceRequest>> allInAnyOfStates(final String... states) {
    // validate states
    for (String state : states) {
        Preconditions.checkArgument(VALID_STATES.contains(state), "unrecognized spot instance request state '%s'",
                state);
    }
    List<String> expectedStates = Arrays.asList(states);
    return spotRequests -> {
        for (SpotInstanceRequest request : spotRequests) {
            if (!expectedStates.contains(request.getState())) {
                return false;
            }
        }
        return true;
    };
}

/**
 * Waits for all placed spot requests to become visible in the API.
 *
 * @param placedSpotRequestIds
 * @return
 */
private List<SpotInstanceRequest> awaitSpotRequests(List<String> placedSpotRequestIds) {

    // wait for placed spot requests to be seen in API when listing *all*
    // spot requests (due to eventual consistency, EC2 will return cached
    // data, which means that the placed requests may not be visible
    // immediately)
    String name = String.format("await-spot-requests{%s}", placedSpotRequestIds);
    Callable<List<SpotInstanceRequest>> requester = new GetSpotInstanceRequests(getAwsCredentials(), getRegion(),
            getClientConfig(), null, null);
    Retryable<List<SpotInstanceRequest>> retryer = Retryers.exponentialBackoffRetryer(name, requester,
            INITIAL_BACKOFF_DELAY, TimeUnit.MILLISECONDS, MAX_RETRIES, contains(placedSpotRequestIds));

    try {
        // only return those spot requests that we actually placed
        List<SpotInstanceRequest> allInstances = retryer.call();
        return allInstances.stream()
                .filter(request -> placedSpotRequestIds.contains(request.getSpotInstanceRequestId()))
                .collect(Collectors.toList());
    } catch (Exception e) {
        throw new RuntimeException(String.format("gave up waiting for spot instance " + "requests to appear %s: %s",
                placedSpotRequestIds, e.getMessage()), e);
    }
}

/**
 * Waits for the cancelled spot instance requests to be reported as
 * cancelled by the Amazon API.
 *
 * @param spotRequestIds
 */
private void awaitCancellation(List<String> spotRequestIds) {
    String name = String.format("await-cancelled{%s}", spotRequestIds);
    GetSpotInstanceRequests requester = new GetSpotInstanceRequests(getAwsCredentials(), getRegion(),
            getClientConfig(), spotRequestIds, null);
    Retryable<List<SpotInstanceRequest>> retryer = Retryers.exponentialBackoffRetryer(name, requester,
            INITIAL_BACKOFF_DELAY, TimeUnit.MILLISECONDS, MAX_RETRIES, allInAnyOfStates(Cancelled.toString()));

    try {
        retryer.call();
    } catch (Exception e) {
        throw new RuntimeException(
                String.format("gave up waiting for spot instance requests " + "to be cancelled %s: %s",
                        spotRequestIds, e.getMessage()),
                e);
    }
}

private void checkForSpotRequest(Activity activity, AmazonEC2Client amazonEC2Client) {
    if (WAIT_FOR_SPOT_INSTANCES_STATUS_CODE.equals(activity.getStatusCode())) {
        Pattern pattern = Pattern.compile(SPOT_ID_PATTERN);
        Matcher matcher = pattern.matcher(activity.getStatusMessage());
        if (matcher.find()) {
            String spotId = matcher.group(0);
            DescribeSpotInstanceRequestsResult spotResult = amazonEC2Client.describeSpotInstanceRequests(
                    new DescribeSpotInstanceRequestsRequest().withSpotInstanceRequestIds(spotId));
            Optional<SpotInstanceRequest> request = spotResult.getSpotInstanceRequests().stream().findFirst();
            if (request.isPresent()) {
                if (LOW_SPOT_PRICE_STATUS_CODE.equals(request.get().getStatus().getCode())) {
                    throw new CloudConnectorException(request.get().getStatus().getMessage());
                }
            }

        }
    }
}

@Override
public void execute(AmazonEC2 client, Pool pool, DelegateExecution execution) {
    LOG.info(">> retrieving instance Ids from spot request Ids");

    @SuppressWarnings("unchecked")
    List<String> requestIds =
        (List<String>) execution.getVariable(ProcessVariables.SPOT_INSTANCE_REQUEST_IDS);
    DescribeSpotInstanceRequestsResult result = client.describeSpotInstanceRequests(
        new DescribeSpotInstanceRequestsRequest().withSpotInstanceRequestIds(requestIds));

    List<String> instanceIds = new ArrayList<String>();
    for (SpotInstanceRequest spotRequest : result.getSpotInstanceRequests()) {
        if (spotRequest.getInstanceId() != null) {
            instanceIds.add(spotRequest.getInstanceId());
        }
    }

    execution.setVariable(ProcessVariables.INSTANCE_IDS, instanceIds);
}

@Override
public void execute(AmazonEC2 client, Pool pool, DelegateExecution execution) {

    LOG.info(">> Checking if all spot requests match predicate {}", predicate);

    @SuppressWarnings("unchecked")
    List<String> requestIds = (List<String>) execution.getVariable(ProcessVariables.SPOT_INSTANCE_REQUEST_IDS);
    checkNotNull(requestIds, "process variable '{}' not found", ProcessVariables.SPOT_INSTANCE_REQUEST_IDS);

    DescribeSpotInstanceRequestsRequest describeRequest = new DescribeSpotInstanceRequestsRequest();
    describeRequest.setSpotInstanceRequestIds(requestIds);

    // Retrieve all of the requests we want to monitor.
    DescribeSpotInstanceRequestsResult describeResult = client.describeSpotInstanceRequests(describeRequest);
    List<SpotInstanceRequest> requests = describeResult.getSpotInstanceRequests();

    if (Iterables.all(requests, predicate)) {
        LOG.info(">> All {} requests match predicate {} ", requests, predicate);
        execution.setVariable(resultVariable, true);
    } else {
        LOG.info("<< Not all requests {} match predicate {}", requests, predicate);
        execution.setVariable(resultVariable, false);
    }
}

@Override
public String getInstanceId() {
    if (instanceId == null || instanceId.equals("")) {
        SpotInstanceRequest sr = getSpotRequest(spotInstanceRequestId);
        if (sr != null)
            instanceId = sr.getInstanceId();
    }
    return instanceId;
}

public void shutdownInstances() {
    AmazonEC2 ec2 = new AmazonEC2Client(getAwsCredentials());
    ec2.setEndpoint(getOrCry("ec2endpoint"));

    try {
        // cancel spot request, so no new instances will be launched
        DescribeSpotInstanceRequestsRequest describeRequest = new DescribeSpotInstanceRequestsRequest();
        DescribeSpotInstanceRequestsResult describeResult = ec2
                .describeSpotInstanceRequests(describeRequest);
        List<SpotInstanceRequest> describeResponses = describeResult
                .getSpotInstanceRequests();
        List<String> spotRequestIds = new ArrayList<String>();
        List<String> instanceIds = new ArrayList<String>();

        for (SpotInstanceRequest describeResponse : describeResponses) {
            spotRequestIds.add(describeResponse.getSpotInstanceRequestId());
            if ("active".equals(describeResponse.getState())) {
                instanceIds.add(describeResponse.getInstanceId());
            }
        }
        ec2.cancelSpotInstanceRequests(new CancelSpotInstanceRequestsRequest()
                .withSpotInstanceRequestIds(spotRequestIds));
        log.info("Cancelled spot request");

        if (instanceIds.size() > 0) {
            ec2.terminateInstances(new TerminateInstancesRequest(
                    instanceIds));
            log.info("Shut down " + instanceIds.size() + " instances");
        }

    } catch (Exception e) {
        log.warn("Failed to shutdown instances - ", e);
    }
}

public List<String> launch(String workerAMI, String instanceType, int num,
        double price, List<String> securityGroups, String keyName, String userData, String charset) throws UnsupportedEncodingException {
    RequestSpotInstancesRequest requestRequest = new RequestSpotInstancesRequest();
    requestRequest.setSpotPrice(Double.toString(price));
    requestRequest.setInstanceCount(Integer.valueOf(num));

    LaunchSpecification launchSpecification = new LaunchSpecification();
    launchSpecification.setImageId(workerAMI);
    launchSpecification.setInstanceType(instanceType);
    launchSpecification.setSecurityGroups(securityGroups);
    launchSpecification.setUserData(new String(Base64.encode(userData.getBytes(charset))));
    launchSpecification.setKeyName(keyName); //for test

    requestRequest.setLaunchSpecification(launchSpecification);
    RequestSpotInstancesResult requestResult = ec2
            .requestSpotInstances(requestRequest);
    List<SpotInstanceRequest> requestResponses = requestResult
            .getSpotInstanceRequests();

    List<String> spotInstanceRequestIds = new ArrayList<String>();

    for (SpotInstanceRequest requestResponse : requestResponses) {
        System.out.println("Created Spot Request: "
                + requestResponse.getSpotInstanceRequestId());
        spotInstanceRequestIds.add(requestResponse
                .getSpotInstanceRequestId());
    }
    return spotInstanceRequestIds;
}

public List<String> getActiveSpotInstanceId(
        List<String> spotInstanceRequestIds) {
    DescribeSpotInstanceRequestsRequest describeRequest = new DescribeSpotInstanceRequestsRequest();
    describeRequest.setSpotInstanceRequestIds(spotInstanceRequestIds);

    System.out
            .println("Checking to determine if Spot Bids have reached the active state...");

    List<String> instanceIds = new ArrayList<String>();

    try {
        DescribeSpotInstanceRequestsResult describeResult = ec2
                .describeSpotInstanceRequests(describeRequest);
        List<SpotInstanceRequest> describeResponses = describeResult
                .getSpotInstanceRequests();

        for (SpotInstanceRequest describeResponse : describeResponses) {
            System.out.println(" "
                    + describeResponse.getSpotInstanceRequestId()
                    + " is in the " + describeResponse.getState()
                    + " state.");

            if (describeResponse.getState().equals("open"))
                return null;

            if (describeResponse.getState().equals("active"))
                instanceIds.add(describeResponse.getInstanceId());
        }
    } catch (AmazonServiceException e) {
        System.out.println("Error when calling describeSpotInstances");
        System.out.println("Caught Exception: " + e.getMessage());
        System.out.println("Reponse Status Code: " + e.getStatusCode());
        System.out.println("Error Code: " + e.getErrorCode());
        System.out.println("Request ID: " + e.getRequestId());

        return null;
    }
    return instanceIds;
}

@Override
public void attachMachine(String spotRequestId) throws NotFoundException, CloudPoolDriverException {
    checkState(isConfigured(), "attempt to use unconfigured driver");

    try {
        SpotInstanceRequest spotRequest = verifySpotRequestExistance(spotRequestId);

        setPoolMembershipTag(spotRequest);
    } catch (Exception e) {
        Throwables.throwIfInstanceOf(e, NotFoundException.class);
        throw new CloudPoolDriverException(
                String.format("failed to attach '%s' to cloud pool: %s", spotRequestId, e.getMessage()), e);
    }
}

/**
 * Pairs up each fulfilled {@link SpotInstanceRequest} with its assigned
 * {@link Instance} in a {@link InstancePairedSpotRequest}. Unfulfilled
 * {@link SpotInstanceRequest}s are returned without a paired
 * {@link Instance}.
 *
 * @param spotRequests
 * @return
 */
private List<InstancePairedSpotRequest> pairUpWithInstances(List<SpotInstanceRequest> spotRequests) {
    List<InstancePairedSpotRequest> pairs = new ArrayList<>();

    for (SpotInstanceRequest spotRequest : spotRequests) {
        String assignedInstanceId = spotRequest.getInstanceId();
        Instance spotInstance = null;
        if (assignedInstanceId != null) {
            spotInstance = this.client.getInstanceMetadata(assignedInstanceId);
        }
        pairs.add(new InstancePairedSpotRequest(spotRequest, spotInstance));
    }
    return pairs;
}

/**
 * Verifies that a particular {@link SpotInstanceRequest} exists and is a
 * member of the pool. If it is not tagged with the pool membership tag a
 * {@link NotFoundException} is thrown.
 *
 * @param spotRequestId
 * @throws NotFoundException
 * @throws AmazonClientException
 */
private void verifyPoolMember(String spotRequestId) throws NotFoundException, AmazonClientException {
    Filter idFilter = new Filter(SPOT_REQUEST_ID_FILTER, asList(spotRequestId));
    Filter poolFilter = new Filter(CLOUD_POOL_TAG_FILTER, asList(getPoolName()));
    List<SpotInstanceRequest> matchingRequests = this.client
            .getSpotInstanceRequests(Arrays.asList(idFilter, poolFilter));
    if (matchingRequests.isEmpty()) {
        throw new NotFoundException(
                String.format("spot instance request %s is not a member of the pool", spotRequestId));
    }
}

/**
 * Verifies that a particular {@link SpotInstanceRequest} exists at all and,
 * if so, returns it. If not, a {@link NotFoundException} is thrown.
 *
 * @param spotRequestId
 * @return The {@link SpotInstanceRequest}, if it exists.
 * @throws NotFoundException
 * @throws AmazonClientException
 */
private SpotInstanceRequest verifySpotRequestExistance(String spotRequestId)
        throws NotFoundException, AmazonClientException {
    Filter idFilter = new Filter(SPOT_REQUEST_ID_FILTER, asList(spotRequestId));
    List<SpotInstanceRequest> matchingRequests = this.client.getSpotInstanceRequests(Arrays.asList(idFilter));
    if (matchingRequests.isEmpty()) {
        throw new NotFoundException(String.format("spot instance request %s does not exist", spotRequestId));
    }
    return matchingRequests.get(0);
}

/**
 * Cleans up any dangling {@link Instance}s (instances whose spot request
 * has been cancelled).
 *
 * @return All {@link Instance}s that were terminated.
 */
List<Instance> cleanupDanglingInstances() {
    LOG.info("cleaning up dangling instances (whose spot requests " + "are cancelled) ...");
    // get all dead spot requests (canceled/closed/failed) spot requests
    // belonging to the pool
    Filter poolFilter = new Filter().withName(CLOUD_POOL_TAG_FILTER).withValues(getPoolName());
    // only include spot requests in state
    Filter spotStateFilter = new Filter().withName(SPOT_REQUEST_STATE_FILTER).withValues(Cancelled.toString(),
            Closed.toString());
    List<SpotInstanceRequest> deadRequests = client().getSpotInstanceRequests(asList(poolFilter, spotStateFilter));
    List<String> deadRequestIds = deadRequests.stream().map(SpotInstanceRequest::getSpotInstanceRequestId)
            .collect(Collectors.toList());

    // get all pending/running instances with a spot instance id equal
    // to any of the dead spot requests
    Filter stateFilter = new Filter().withName(INSTANCE_STATE_FILTER).withValues(Pending.toString(),
            Running.toString());
    Filter spotRequestFilter = new Filter().withName(ScalingFilters.SPOT_REQUEST_ID_FILTER)
            .withValues(deadRequestIds);

    List<Instance> danglingInstances = client().getInstances(asList(stateFilter, spotRequestFilter));
    for (Instance danglingInstance : danglingInstances) {
        LOG.info("terminating dangling instance {} belonging " + "to dead spot request {}",
                danglingInstance.getInstanceId(), danglingInstance.getSpotInstanceRequestId());
        client().terminateInstances(asList(danglingInstance.getInstanceId()));
    }
    return danglingInstances;
}

/**
 * Check bid prices for all unfulfilled spot requests and cancel ones that
 * are not up-to-date with the currently configured bid price. These are to
 * eventually be replaced with a new spot request with the right bid price,
 * as soon as the {@link BaseCloudPool} detects that the pool is short on
 * spot requests.
 *
 * @return Returns the list of wrong-priced spot request identifiers that
 *         were cancelled.
 */
List<String> cancelWrongPricedRequests() {
    double currentBidPrice = cloudApiSettings().getBidPrice();
    LOG.info("cancelling unfulfilled spot requests with bidprice " + "other than {} ...", currentBidPrice);
    List<InstancePairedSpotRequest> unfulfilledRequests = getPoolSpotRequests(asList(Open.toString()));
    List<String> wrongPricedSpotIds = new ArrayList<>();
    for (InstancePairedSpotRequest unfulfilledRequest : unfulfilledRequests) {
        SpotInstanceRequest request = unfulfilledRequest.getRequest();
        double spotPrice = Double.valueOf(request.getSpotPrice());
        if (spotPrice != currentBidPrice) {
            wrongPricedSpotIds.add(request.getSpotInstanceRequestId());
        }
    }
    if (wrongPricedSpotIds.isEmpty()) {
        return Collections.emptyList();
    }

    LOG.info("cancelling unfulfilled spot requests with wrong bid " + "price: {}", wrongPricedSpotIds);
    try {
        // Note: there is a possibility that a wrong-priced spot request has
        // been fulfilled after we decided to cancel it. If so, it will
        // become a dangling instance that gets cleaned up eventually.
        this.client.cancelSpotRequests(wrongPricedSpotIds);
    } catch (Exception e) {
        postCancellationFailureAlert(wrongPricedSpotIds, e);
    }

    postCancellationAlert(wrongPricedSpotIds);
    return wrongPricedSpotIds;
}

/**
 * Extracts the region of the spot request from the availability zone
 * specified in the spot request's launch configuration.
 *
 * @param spotInstanceRequest
 * @return
 */
private String extractRegion(InstancePairedSpotRequest spotInstanceRequest) {
    SpotInstanceRequest request = spotInstanceRequest.getRequest();
    if (request.getLaunchSpecification() == null || request.getLaunchSpecification().getPlacement() == null
            || request.getLaunchSpecification().getPlacement().getAvailabilityZone() == null) {
        LOG.warn("failed to extract region from spot request {}", request.getSpotInstanceRequestId());
        return "unknown";
    }
    // availability zone is region + letter, for example 'us-east-1a'
    String availabilityZone = request.getLaunchSpecification().getPlacement().getAvailabilityZone();
    String region = availabilityZone.substring(0, availabilityZone.length() - 1);
    return region;
}

/**
 * Retrieves a certain tag value from an {@link SpotInstanceRequest}.
 *
 * @param spotRequest
 * @param tagKey
 * @return
 */
private Optional<String> getTagValue(SpotInstanceRequest spotRequest, String tagKey) {
    List<Tag> tags = spotRequest.getTags();
    for (Tag tag : tags) {
        if (tag.getKey().equals(tagKey)) {
            return Optional.of(tag.getValue());
        }
    }
    return Optional.empty();
}

/**
 * Adds a number of {@link SpotInstanceRequest}s and {@link Instance}s to
 * the faked cloud account.
 *
 * @param spotRequests
 * @param instances
 */
public void setupFakeAccount(Collection<SpotInstanceRequest> spotRequests, Collection<Instance> instances) {
    for (SpotInstanceRequest request : spotRequests) {
        this.spotRequests.put(request.getSpotInstanceRequestId(), request);
    }
    for (Instance instance : instances) {
        this.instances.put(instance.getInstanceId(), instance);
    }
}

@Override
public SpotInstanceRequest getSpotInstanceRequest(String spotRequestId) throws AmazonClientException {
    if (!this.spotRequests.containsKey(spotRequestId)) {
        throw new AmazonServiceException(
                String.format("The spot instance request ID '%s' does not exist (Service: AmazonEC2; "
                        + "Status Code: 400; Error Code: " + "InvalidSpotInstanceRequestID.NotFound; "
                        + "Request ID: 5fc1854f-ceb4-4f6a-8178-a9e918371d46)", spotRequestId));
    }
    return this.spotRequests.get(spotRequestId);
}

private boolean matches(Filter filter, SpotInstanceRequest spotRequest) {
    String filterName = filter.getName();
    if (!spotRequestFilterMatchers.containsKey(filterName)) {
        throw new IllegalArgumentException("fake SpotClient does not recognize spot filter " + filterName);
    }
    return spotRequestFilterMatchers.get(filterName).matches(spotRequest, filter);
}

@Override
public boolean matches(SpotInstanceRequest spotRequest, Filter filter) {
    // filter name is "tag:<tagname>"
    String expectedTag = filter.getName().replace("tag:", "");
    List<String> acceptableTagValues = filter.getValues();
    for (Tag tag : spotRequest.getTags()) {
        if (tag.getKey().equals(expectedTag)) {
            return acceptableTagValues.contains(tag.getValue());
        }
    }
    return filter.getValues().contains(spotRequest.getState());
}

/**
 * Verify that spot requests are placed and tagged with group tag on
 * {@link CloudPoolDriver#startMachines}.
 */
@Test
public void testStartMachines() {
    this.driver = new SpotPoolDriver(this.fakeClient, this.executor, this.mockEventBus);
    this.driver.configure(config());

    assertThat(this.driver.listMachines().size(), is(0));
    List<Machine> started = this.driver.startMachines(1);
    assertThat(started.size(), is(1));
    SpotInstanceRequest placedSpotRequest = this.fakeClient.getSpotInstanceRequest(started.get(0).getId());
    assertTrue(placedSpotRequest.getTags().contains(new Tag(CLOUD_POOL_TAG, POOL_NAME)));
}

/**
 * Test that whenever the bid price changes, unfulfilled requests with an
 * incorrect bid price are cancelled (to be replaced by requests with the
 * new bid price on the next pool size update).
 */
@Test
public void testCancelWrongPricedRequests() {
    this.driver = new SpotPoolDriver(this.fakeClient, this.executor, this.mockEventBus);
    this.driver.configure(config());
    double currentBidPrice = this.driver.cloudApiSettings().getBidPrice();

    // unfulfilled, wrong bid price => should be replaced
    SpotInstanceRequest spot1 = spotRequest("sir-1", "open", null, POOL1_TAG);
    spot1.setSpotPrice(String.valueOf(currentBidPrice + 0.01));
    // unfulfilled, right bid price => should not be replaced
    SpotInstanceRequest spot2 = spotRequest("sir-2", "open", null, POOL1_TAG);
    spot2.setSpotPrice(String.valueOf(currentBidPrice));
    // fulfilled, right bid price => should not be replaced
    SpotInstanceRequest spot3 = spotRequest("sir-3", "active", "i-3", POOL1_TAG);
    spot3.setSpotPrice(String.valueOf(currentBidPrice));
    // fulfilled, wrong bid price => should not be replaced
    SpotInstanceRequest spot4 = spotRequest("sir-4", "active", "i-4", POOL1_TAG);
    spot4.setSpotPrice(String.valueOf(currentBidPrice + 0.01));

    this.fakeClient.setupFakeAccount(asList(spot1, spot2, spot3, spot4),
            asList(instance("i-3", Running, "sir-3"), instance("i-4", Running, "sir-4")));

    List<String> cancelledRequests = this.driver.cancelWrongPricedRequests();
    assertThat(cancelledRequests.size(), is(1));
    assertThat(cancelledRequests.get(0), is("sir-1"));
    assertRequestIds(this.driver.listMachines(), asList("sir-2", "sir-3", "sir-4"));

    // verify event posted on event bus
    verify(this.mockEventBus).post(argThat(IsCancelAlert.isCancelAlert("sir-1")));
}

/**
 * An error to complete the operation should result in a
 * {@link CloudPoolDriverException}.
 */
@SuppressWarnings("unchecked")
@Test(expected = CloudPoolDriverException.class)
public void testDetachOnError() {
    this.driver = new SpotPoolDriver(this.mockClient, this.executor, this.mockEventBus);
    this.driver.configure(config());

    List<SpotInstanceRequest> poolMembers = asList(spotRequest("sir-1", "open", null, POOL1_TAG));
    when(this.mockClient.getSpotInstanceRequests(Matchers.anyCollection())).thenReturn(poolMembers);
    Tag poolTag = new Tag(CLOUD_POOL_TAG, POOL_NAME);
    doThrow(new AmazonServiceException("something went wrong")).when(this.mockClient).untagResource("sir-1",
            asList(poolTag));

    this.driver.detachMachine("sir-1");
}

/**
 * An error to complete the operation should result in a
 * {@link CloudPoolDriverException}.
 */
@SuppressWarnings("unchecked")
@Test(expected = CloudPoolDriverException.class)
public void testAttachOnError() {
    this.driver = new SpotPoolDriver(this.mockClient, this.executor, this.mockEventBus);
    this.driver.configure(config());

    List<SpotInstanceRequest> poolMembers = asList(spotRequest("sir-1", "open", null, POOL1_TAG));
    when(this.mockClient.getSpotInstanceRequests(Matchers.anyCollection())).thenReturn(poolMembers);
    Tag poolTag = new Tag(CLOUD_POOL_TAG, POOL_NAME);
    doThrow(new AmazonServiceException("something went wrong")).when(this.mockClient).tagResource("sir-1",
            asList(poolTag));

    this.driver.attachMachine("sir-1");
}

/**
 * An error to complete the operation should result in a
 * {@link CloudPoolDriverException}.
 */
@SuppressWarnings("unchecked")
@Test(expected = CloudPoolDriverException.class)
public void testSetServiceStateOnError() {
    this.driver = new SpotPoolDriver(this.mockClient, this.executor, this.mockEventBus);
    this.driver.configure(config());

    List<SpotInstanceRequest> poolMembers = asList(spotRequest("sir-1", "open", null, POOL1_TAG));
    when(this.mockClient.getSpotInstanceRequests(Matchers.anyCollection())).thenReturn(poolMembers);
    Tag stateTag = new Tag(ScalingTags.SERVICE_STATE_TAG, "BOOTING");
    doThrow(new AmazonServiceException("something went wrong")).when(this.mockClient).tagResource("sir-1",
            asList(stateTag));

    this.driver.setServiceState("sir-1", ServiceState.BOOTING);
}

/**
 * An error to complete the operation should result in a
 * {@link CloudPoolDriverException}.
 */
@SuppressWarnings("unchecked")
@Test(expected = CloudPoolDriverException.class)
public void testSetMembershipStatusOnError() {
    this.driver = new SpotPoolDriver(this.mockClient, this.executor, this.mockEventBus);
    this.driver.configure(config());

    List<SpotInstanceRequest> poolMembers = asList(spotRequest("sir-1", "open", null, POOL1_TAG));
    when(this.mockClient.getSpotInstanceRequests(Matchers.anyCollection())).thenReturn(poolMembers);
    Tag statusTag = new Tag(ScalingTags.MEMBERSHIP_STATUS_TAG, MembershipStatus.blessed().toString());
    doThrow(new AmazonServiceException("something went wrong")).when(this.mockClient).tagResource("sir-1",
            asList(statusTag));

    this.driver.setMembershipStatus("sir-1", MembershipStatus.blessed());
}

public static void main(String[] args) {
    String region = "us-east-1";

    AWSCredentials credentials = new BasicAWSCredentials(awsAccessKeyId, awsSecretAccessKey);

    AmazonEC2 api = AmazonEC2ClientBuilder.standard().withRegion(region)
            .withCredentials(new AWSStaticCredentialsProvider(credentials)).build();

    // no particular availability zone
    String availabilityZone = null;
    String instanceType = "t1.micro";
    String imageId = "ami-3cf8b154";
    List<String> securityGroups = Arrays.asList("webserver");
    String keyPair = "instancekey";

    String bootScript = Joiner.on("\n")
            .join(Arrays.asList("#!/bin/bash", "sudo apt-get update -qy", "sudo apt-get install -qy apache2"));
    int instanceCount = 50;
    String bidPrice = "0.001";

    SpotPlacement placement = new SpotPlacement().withAvailabilityZone(availabilityZone);
    LaunchSpecification launchSpec = new LaunchSpecification().withInstanceType(instanceType).withImageId(imageId)
            .withPlacement(placement).withSecurityGroups(securityGroups).withKeyName(keyPair)
            .withUserData(AmazonApiUtils.base64Encode(bootScript));
    RequestSpotInstancesRequest request = new RequestSpotInstancesRequest().withInstanceCount(instanceCount)
            .withType(SpotInstanceType.Persistent).withSpotPrice(bidPrice).withLaunchSpecification(launchSpec);
    RequestSpotInstancesResult result = api.requestSpotInstances(request);

    for (SpotInstanceRequest spotRequest : result.getSpotInstanceRequests()) {
        LOG.info("placed request: {}", spotRequest.getSpotInstanceRequestId());
    }
}

/**
 * Test {@link MachineState} conversion when {@link SpotInstanceRequest}
 * state is not recognized as a valid state.
 */
@Test(expected = IllegalArgumentException.class)
public void getMachineStateOnUnrecognizedSpotRequestState() {
    InstancePairedSpotRequest badStateRequest = new InstancePairedSpotRequest(
            new SpotInstanceRequest().withSpotInstanceRequestId("sir-1").withState("badstate"), null);
    badStateRequest.getMachineState();
}

@Test
public void testConversionWithServiceStateTag() {
    SpotInstanceRequest spotRequest = SpotTestUtil.spotRequest("sir-1", "open", null);
    spotRequest
            .withTags(new Tag().withKey(ScalingTags.SERVICE_STATE_TAG).withValue(ServiceState.IN_SERVICE.name()));
    InstancePairedSpotRequest request = new InstancePairedSpotRequest(spotRequest, null);

    Machine machine = InstancePairedSpotRequestToMachine.convert(request);
    assertThat(machine.getServiceState(), is(ServiceState.IN_SERVICE));
}