import com.amazonaws.services.ec2.model.SpotInstanceRequest; //导入依赖的package包/类
/**
 * 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;
}
 

import com.amazonaws.services.ec2.model.SpotInstanceRequest; //导入依赖的package包/类
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;
	}
}
 

import com.amazonaws.services.ec2.model.SpotInstanceRequest; //导入依赖的package包/类
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;
}
 

import com.amazonaws.services.ec2.model.SpotInstanceRequest; //导入依赖的package包/类
@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;
}
 

import com.amazonaws.services.ec2.model.SpotInstanceRequest; //导入依赖的package包/类
/**
 * 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);
    }
}
 

import com.amazonaws.services.ec2.model.SpotInstanceRequest; //导入依赖的package包/类
@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;
}
 

import com.amazonaws.services.ec2.model.SpotInstanceRequest; //导入依赖的package包/类
@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;
}
 

import com.amazonaws.services.ec2.model.SpotInstanceRequest; //导入依赖的package包/类
/**
 * 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);
}
 

import com.amazonaws.services.ec2.model.SpotInstanceRequest; //导入依赖的package包/类
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.");
    }
}
 

import com.amazonaws.services.ec2.model.SpotInstanceRequest; //导入依赖的package包/类
/**
 * 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;
    };
}
 

import com.amazonaws.services.ec2.model.SpotInstanceRequest; //导入依赖的package包/类
/**
 * 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);
    }
}
 

import com.amazonaws.services.ec2.model.SpotInstanceRequest; //导入依赖的package包/类
/**
 * 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);
    }
}
 

import com.amazonaws.services.ec2.model.SpotInstanceRequest; //导入依赖的package包/类
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());
                }
            }

        }
    }
}
 

import com.amazonaws.services.ec2.model.SpotInstanceRequest; //导入依赖的package包/类
@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);
}
 

import com.amazonaws.services.ec2.model.SpotInstanceRequest; //导入依赖的package包/类
@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);
    }
}
 

import com.amazonaws.services.ec2.model.SpotInstanceRequest; //导入依赖的package包/类
@Override
public String getInstanceId() {
	if (instanceId == null || instanceId.equals("")) {
		SpotInstanceRequest sr = getSpotRequest(spotInstanceRequestId);
		if (sr != null)
			instanceId = sr.getInstanceId();
	}
	return instanceId;
}
 

import com.amazonaws.services.ec2.model.SpotInstanceRequest; //导入依赖的package包/类
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);
	}
}
 

import com.amazonaws.services.ec2.model.SpotInstanceRequest; //导入依赖的package包/类
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;
}
 

import com.amazonaws.services.ec2.model.SpotInstanceRequest; //导入依赖的package包/类
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;
}
 

import com.amazonaws.services.ec2.model.SpotInstanceRequest; //导入依赖的package包/类
@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);
    }
}