Java 类com.amazonaws.services.ec2.AmazonEC2AsyncClient 实例源码

public void createSnapshotFromTagName(TagNameRequest tagNameRequest, Context context) {

        LambdaLogger logger = context.getLogger();
        logger.log("create ebs snapshot from tag name Start. backup target[" + tagNameRequest + "]");

        String regionName = System.getenv("AWS_DEFAULT_REGION");
        AmazonEC2Async client = RegionUtils.getRegion(regionName).createClient(AmazonEC2AsyncClient.class,
                new DefaultAWSCredentialsProviderChain(), cc);

        try {
            List<Volume> volumes = describeBackupVolumes(client, tagNameRequest);

            for (Volume volume : volumes) {
                createSnapshot(volume.getVolumeId(), tagNameRequest.getGenerationCount(), context);
            }
        } finally {
            client.shutdown();
        }
    }

private void startInstance(AmazonEC2AsyncClient client, DefaultAdapterContext c) {
    StartInstancesRequest startRequest  = new StartInstancesRequest();
    startRequest.withInstanceIds(c.child.id);
    client.startInstancesAsync(startRequest,
            new AWSAsyncHandler<StartInstancesRequest, StartInstancesResult>() {

                @Override
                protected void handleError(Exception e) {
                    c.taskManager.patchTaskToFailure(e);
                }

                @Override
                protected void handleSuccess(StartInstancesRequest request, StartInstancesResult result) {
                    AWSUtils.waitForTransitionCompletion(getHost(),
                            result.getStartingInstances(), "running",
                            client, (is, e) -> {
                                if (e == null) {
                                    c.taskManager.finishTask();
                                } else {
                                    c.taskManager.patchTaskToFailure(e);
                                }
                            });
                }
            });
}

@Override
public void handlePatch(Operation op) {
    if (!op.hasBody()) {
        op.fail(new IllegalArgumentException("body is required"));
        return;
    }
    ComputePowerRequest pr = op.getBody(ComputePowerRequest.class);
    op.complete();
    if (pr.isMockRequest) {
        updateComputeState(pr, new DefaultAdapterContext(this, pr));
    } else {
        new DefaultAdapterContext(this, pr)
                .populateBaseContext(BaseAdapterStage.VMDESC)
                .whenComplete((c, e) -> {
                    AmazonEC2AsyncClient client = this.clientManager.getOrCreateEC2Client(
                            c.parentAuth, c.child.description.regionId, this,
                            (t) -> c.taskManager.patchTaskToFailure(t));
                    if (client == null) {
                        return;
                    }
                    applyPowerOperation(client, pr, c);
                });
    }

}

private void applyPowerOperation(AmazonEC2AsyncClient client, ComputePowerRequest pr,
        DefaultAdapterContext c) {
    switch (pr.powerState) {
    case OFF:
        powerOff(client, pr, c);
        break;
    case ON:
        powerOn(client, pr, c);
        break;
    case SUSPEND:
        // TODO: Not supported yet, so simply patch the state with requested power state.
        updateComputeState(pr, c);
        break;
    case UNKNOWN:
    default:
        c.taskManager.patchTaskToFailure(
                new IllegalArgumentException("Unsupported power state transition requested."));
    }

}

private DeferredResult<Void> validateCredentials(
        AuthCredentialsServiceState credentials, String regionId) {
    AmazonEC2AsyncClient client = AWSUtils.getAsyncClient(credentials, regionId,
            this.clientManager.getExecutor());

    AWSDeferredResultAsyncHandler<DescribeAvailabilityZonesRequest, DescribeAvailabilityZonesResult> asyncHandler = new AWSDeferredResultAsyncHandler<>(
            this, "Validate Credentials");

    client.describeAvailabilityZonesAsync(asyncHandler);

    return asyncHandler
            .toDeferredResult()
            .handle((describeAvailabilityZonesResult, e) -> {
                if (e instanceof AmazonServiceException) {
                    AmazonServiceException ase = (AmazonServiceException) e;
                    if (ase.getStatusCode() == STATUS_CODE_UNAUTHORIZED) {

                        throw new LocalizableValidationException(
                                e,
                                PHOTON_MODEL_ADAPTER_UNAUTHORIZED_MESSAGE,
                                PHOTON_MODEL_ADAPTER_UNAUTHORIZED_MESSAGE_CODE);
                    }
                }
                return null;
            });
}

/**
 * Accesses the client cache to get the EC2 client for the given auth credentials and regionId. If a client
 * is not found to exist, creates a new one and adds an entry in the cache for it.
 * @param credentials
 *         The auth credentials to be used for the client creation
 * @param regionId
 *         The region of the AWS client
 * @param service
 *         The stateless service making the request and for which the executor pool needs to be allocated.
 * @return The AWSClient
 */
public AmazonEC2AsyncClient getOrCreateEC2Client(
        AuthCredentialsServiceState credentials,
        String regionId, StatelessService service, Consumer<Throwable> failConsumer) {
    if (this.awsClientType != AwsClientType.EC2) {
        throw new UnsupportedOperationException(
                "This client manager supports only AWS " + this.awsClientType + " clients.");
    }
    AmazonEC2AsyncClient amazonEC2Client = null;
    String cacheKey = createCredentialRegionCacheKey(credentials, regionId);
    try {
        amazonEC2Client = this.ec2ClientCache.computeIfAbsent(cacheKey, key -> AWSUtils
                .getAsyncClient(credentials, regionId, getExecutor()));
    } catch (Throwable e) {
        service.logSevere(e);
        failConsumer.accept(e);
    }
    return amazonEC2Client;
}

public static void tearDownTestVpc(
        AmazonEC2AsyncClient client, VerificationHost host,
        Map<String, Object> awsTestContext, boolean isMock) {
    if (!isMock && !vpcIdExists(client, AWS_DEFAULT_VPC_ID)) {
        final String vpcId = (String) awsTestContext.get(VPC_KEY);
        final String subnetId = (String) awsTestContext.get(SUBNET_KEY);
        final String internetGatewayId = (String) awsTestContext.get(INTERNET_GATEWAY_KEY);
        final String securityGroupId = (String) awsTestContext.get(SECURITY_GROUP_KEY);
        // clean up VPC and all its dependencies if creating one at setUp
        deleteSecurityGroupUsingEC2Client(client, host, securityGroupId);
        SecurityGroup securityGroup = new AWSSecurityGroupClient(client)
                .getSecurityGroup(AWS_DEFAULT_GROUP_NAME, vpcId);
        if (securityGroup != null) {
            deleteSecurityGroupUsingEC2Client(client, host, securityGroup.getGroupId());
        }
        deleteSubnet(client, subnetId);
        detachInternetGateway(client, vpcId, internetGatewayId);
        deleteInternetGateway(client, internetGatewayId);
        deleteVPC(client, vpcId);
    }
}

public static void tearDownTestDisk(
        AmazonEC2AsyncClient client, VerificationHost host,
        Map<String, Object> awsTestContext, boolean isMock) {
    if (awsTestContext.containsKey(DISK_KEY)) {
        String volumeId = awsTestContext.get(DISK_KEY).toString();
        if (!isMock) {
            deleteVolume(client, volumeId);
        }
        awsTestContext.remove(DISK_KEY);
    }
    if (awsTestContext.containsKey(SNAPSHOT_KEY)) {
        String snapshotId = awsTestContext.get(SNAPSHOT_KEY).toString();
        if (!isMock) {
            deleteSnapshot(client, snapshotId);
        }
        awsTestContext.remove(SNAPSHOT_KEY);
    }
}

/**
 * Method to directly provision instances on the AWS endpoint without the knowledge of the local
 * system. This is used to spawn instances and to test that the discovery of items not
 * provisioned by Xenon happens correctly.
 *
 * @throws Throwable
 */
public static List<String> provisionAWSVMWithEC2Client(AmazonEC2AsyncClient client,
        VerificationHost host, int numberOfInstance, String instanceType, String subnetId,
        String securityGroupId) throws Throwable {
    host.log("Provisioning %d instances on the AWS endpoint using the EC2 client.",
            numberOfInstance);

    RunInstancesRequest runInstancesRequest = new RunInstancesRequest()
            .withSubnetId(subnetId)
            .withImageId(EC2_LINUX_AMI).withInstanceType(instanceType)
            .withMinCount(numberOfInstance).withMaxCount(numberOfInstance)
            .withSecurityGroupIds(securityGroupId);

    // handler invoked once the EC2 runInstancesAsync commands completes
    AWSRunInstancesAsyncHandler creationHandler = new AWSRunInstancesAsyncHandler(
            host);
    client.runInstancesAsync(runInstancesRequest, creationHandler);
    host.waitFor("Waiting for instanceIds to be retured from AWS", () -> {
        return checkInstanceIdsReturnedFromAWS(numberOfInstance, creationHandler.instanceIds);

    });

    return creationHandler.instanceIds;
}

/**
 * Method to get Instance details directly from Amazon
 *
 * @throws Throwable
 */
public static List<Instance> getAwsInstancesByIds(AmazonEC2AsyncClient client,
        VerificationHost host, List<String> instanceIds)
        throws Throwable {
    host.log("Getting instances with ids " + instanceIds
            + " from the AWS endpoint using the EC2 client.");

    DescribeInstancesRequest describeInstancesRequest = new DescribeInstancesRequest()
            .withInstanceIds(instanceIds);

    DescribeInstancesResult describeInstancesResult = client
            .describeInstances(describeInstancesRequest);

    return describeInstancesResult.getReservations().stream()
            .flatMap(r -> r.getInstances().stream()).collect(Collectors.toList());
}

/**
 * Method that polls to see if the instances provisioned have turned ON.This method accepts an
 * error count to allow some room for errors in case all the requested resources are not
 * provisioned correctly.
 *
 * @return boolean if the required instances have been turned ON on AWS with some acceptable
 *         error rate.
 */
public static boolean computeInstancesStartedStateWithAcceptedErrorRate(
        AmazonEC2AsyncClient client,
        VerificationHost host, List<String> instanceIds, int errorRate) throws Throwable {
    // If there are no instanceIds set then return false
    if (instanceIds.size() == 0) {
        return false;
    }
    ArrayList<Boolean> provisioningFlags = new ArrayList<Boolean>(instanceIds.size());
    for (int i = 0; i < instanceIds.size(); i++) {
        provisioningFlags.add(i, Boolean.FALSE);
    }
    // Calls the describe instances API to get the latest state of each machine being
    // provisioned.
    checkInstancesStarted(host, client, instanceIds, provisioningFlags);
    int totalCount = instanceIds.size();
    int passCount = (int) Math.ceil((((100 - errorRate) / HUNDERED) * totalCount));
    int poweredOnCount = 0;
    for (boolean startedFlag : provisioningFlags) {
        if (startedFlag) {
            poweredOnCount++;
        }
    }
    return (poweredOnCount >= passCount);
}

/**
 * Gets the instance count of non-terminated instances on the AWS endpoint. This is used to run
 * the asserts and validate the results for the data that is collected during enumeration.This
 * also calculates the compute descriptions that will be used to represent the instances that
 * were discovered on the AWS endpoint. Further factoring in the
 *
 * @throws Throwable
 */
public static BaseLineState getBaseLineInstanceCount(VerificationHost host,
        AmazonEC2AsyncClient client,
        List<String> testComputeDescriptions)
        throws Throwable {
    BaseLineState baseLineState = new BaseLineState();
    AWSEnumerationAsyncHandler enumerationHandler = new AWSEnumerationAsyncHandler(host,
            AWSEnumerationAsyncHandler.MODE.GET_COUNT, null, null, null,
            testComputeDescriptions,
            baseLineState);
    DescribeInstancesRequest request = new DescribeInstancesRequest();
    Filter runningInstanceFilter = getAWSNonTerminatedInstancesFilter();
    request.getFilters().add(runningInstanceFilter);
    client.describeInstancesAsync(request, enumerationHandler);
    host.waitFor("Error waiting to get base line instance count from AWS in test ", () -> {
        return baseLineState.isCountPopulated;
    });
    return baseLineState;
}

public static void waitForInstancesToBeTerminated(AmazonEC2AsyncClient client,
        VerificationHost host, List<String> instanceIdsToDelete) throws Throwable {
    if (instanceIdsToDelete.size() == 0) {
        return;
    }
    ArrayList<Boolean> deletionFlags = new ArrayList<>(instanceIdsToDelete.size());
    for (int i = 0; i < instanceIdsToDelete.size(); i++) {
        deletionFlags.add(i, Boolean.FALSE);
    }
    host.waitFor("Error waiting for EC2 client delete instances in test ", () -> {
        boolean isDeleted = computeInstancesTerminationState(client,
                host, instanceIdsToDelete, deletionFlags);
        if (isDeleted) {
            return true;
        }

        host.log(Level.INFO, "Waiting for EC2 instance deletion");
        Thread.sleep(TimeUnit.SECONDS.toMillis(10));
        return false;
    });

}

/**
 * Checks if a newly deleted instance has its status set to terminated.
 *
 * @return
 */
public static void checkInstancesDeleted(AmazonEC2AsyncClient client,
        VerificationHost host, List<String> instanceIdsToDelete,
        ArrayList<Boolean> deletionFlags) throws Throwable {
    AWSEnumerationAsyncHandler enumerationHandler = new AWSEnumerationAsyncHandler(host,
            AWSEnumerationAsyncHandler.MODE.CHECK_TERMINATION, null, deletionFlags, null, null,
            null);
    DescribeInstancesRequest request = new DescribeInstancesRequest()
            .withInstanceIds(instanceIdsToDelete);
    client.describeInstancesAsync(request, enumerationHandler);
    // Waiting to get a response from AWS before the state computation is done for the list of
    // VMs.
    host.waitFor("Waiting to get response from AWS ", () -> {
        return enumerationHandler.responseReceived;
    });
}

/**
 * Method to get Disk details directly from Amazon
 */
public static List<Volume> getAwsDisksByIds(AmazonEC2AsyncClient client,
        VerificationHost host, List<String> diskIds) throws Throwable {
    try {

        host.log("Getting disks with ids " + diskIds
                + " from the AWS endpoint using the EC2 client.");

        DescribeVolumesRequest describeVolumesRequest = new DescribeVolumesRequest()
                .withVolumeIds(diskIds);

        DescribeVolumesResult describeVolumesResult = client
                .describeVolumes(describeVolumesRequest);

        return describeVolumesResult.getVolumes();
    } catch (Exception e) {
        if (e instanceof AmazonEC2Exception
                && ((AmazonEC2Exception) e).getErrorCode()
                .equalsIgnoreCase(AWS_INVALID_VOLUME_ID_ERROR_CODE)) {
            return null;
        }
    }
    return new ArrayList<>();
}

@Test
public void testResourceNaming() throws Throwable {
    boolean tagFound = false;
    AmazonEC2AsyncClient client = TestUtils.getClient(this.privateKeyId,this.privateKey,this.region,false);

    //create something to name
    AWSNetworkClient svc = new AWSNetworkClient(client);
    String vpcID = svc.createVPC("10.20.0.0/16");
    AWSUtils.tagResourcesWithName(client, TEST_NAME, vpcID);
    List<TagDescription> tags = AWSUtils.getResourceTags(vpcID,client);

    for (TagDescription tagDesc:tags) {
        if (tagDesc.getKey().equalsIgnoreCase(AWS_TAG_NAME)) {
            assertTrue(tagDesc.getValue().equalsIgnoreCase(TEST_NAME));
            tagFound = true;
            break;
        }
    }
    // ensure we found the tag
    assertTrue(tagFound);
    svc.deleteVPC(vpcID);
}

protected void assertBootDiskConfiguration(AmazonEC2AsyncClient client, Instance awsInstance,
        String diskLink) {
    DiskState diskState = getDiskState(diskLink);

    Volume bootVolume = getVolume(client, awsInstance, awsInstance.getRootDeviceName());

    assertEquals("Boot Disk capacity in diskstate is not matching the boot disk size of the "
            + "vm launched in aws", diskState.capacityMBytes, bootVolume.getSize() * 1024);

    assertEquals(
            "Boot disk type in diskstate is not same as the type of the volume attached to the VM",
            diskState.customProperties.get("volumeType"), bootVolume.getVolumeType());

    assertEquals(
            "Boot disk iops in diskstate is the same as the iops of the volume attached to the VM",
            Integer.parseInt(diskState.customProperties.get("iops")),
            bootVolume.getIops().intValue());

    assertEquals("Boot disk attach status is not matching", DiskService.DiskStatus.ATTACHED, diskState.status);
}

protected Volume getVolume(AmazonEC2AsyncClient client, Instance awsInstance, String deviceName) {
    InstanceBlockDeviceMapping bootDiskMapping = awsInstance.getBlockDeviceMappings().stream()
            .filter(blockDeviceMapping -> blockDeviceMapping.getDeviceName().equals(deviceName))
            .findAny()
            .orElse(null);

    //The ami used in this test is an ebs-backed AMI
    assertNotNull("Device type should be ebs type", bootDiskMapping.getEbs());

    String bootVolumeId = bootDiskMapping.getEbs().getVolumeId();
    DescribeVolumesRequest describeVolumesRequest = new DescribeVolumesRequest()
            .withVolumeIds(bootVolumeId);
    DescribeVolumesResult describeVolumesResult = client
            .describeVolumes(describeVolumesRequest);

    return describeVolumesResult.getVolumes().get(0);
}

public EC2(UserProviderCredentials credentials)
{
    this.credentials_ = checkNotNull(credentials);
    checkState(!isNullOrEmpty(credentials.getLoginCredentials().getCredentialName()));
    checkNotNull(credentials.getRegion());
    checkState(!isNullOrEmpty(credentials.getRegion().getName()));
    checkState(!isNullOrEmpty(credentials.getRegion().getEndpoint()));

    this.awsCredentials_ = new BasicAWSCredentials(credentials.getLoginCredentials().getIdentity(), 
                                                   credentials.getLoginCredentials().getCredential());

    ec2_ = new AmazonEC2AsyncClient(this.awsCredentials_);
    ec2_.setEndpoint(credentials.getRegion().getEndpoint());
    this.defaultUserGroupName_ = System.getProperty("org.excalibur.security.default.group.name", "excalibur-security-group");
    backoffLimitedRetryHandler_ = new BackoffLimitedRetryHandler();

}

@Override
public void handleRequest(InputStream is, OutputStream os, Context context) {

    LambdaLogger logger = context.getLogger();
    String regionName = System.getenv("AWS_DEFAULT_REGION");
    AmazonEC2Async client = RegionUtils.getRegion(regionName).createClient(AmazonEC2AsyncClient.class,
            new DefaultAWSCredentialsProviderChain(), cc);
    try {
        ObjectMapper om = new ObjectMapper();
        DeregisterImageRequest event = om.readValue(is, DeregisterImageRequest.class);
        String imageId = event.getDetail().getRequestParameters().getImageId();

        logger.log("Deregister AMI parge snapshot Start. ImageId[" + imageId + "]");

        List<Snapshot> snapshots = describeSnapshot(client, imageId, context);
        if (snapshots.size() == 0) {
            logger.log("Target of snapshot there is nothing.");
        } else {
            snapshots.stream().forEach(s -> pargeSnapshot(client, s.getSnapshotId(), context));
        }
        logger.log("[SUCCESS][DeregisterImage]has been completed successfully." + imageId);
    } catch (Exception e) {
        logger.log("[ERROR][DeregisterImage]An unexpected error has occurred. message[" + e.getMessage() + "]");
    } finally {
        client.shutdown();
    }
}

private AWSTaskStatusChecker(AmazonEC2AsyncClient amazonEC2Client,
        String instanceId, String desiredState,
        Consumer<Object> consumer, TaskManager taskManager, StatelessService service,
        long expirationTimeMicros) {
    this(amazonEC2Client, instanceId, desiredState, Collections.emptyList(), consumer,
            taskManager, service, expirationTimeMicros);
}

private AWSTaskStatusChecker(AmazonEC2AsyncClient amazonEC2Client,
        String instanceId, String desiredState, List<String> failureStates,
        Consumer<Object> consumer, TaskManager taskManager, StatelessService service,
        long expirationTimeMicros) {
    this.instanceId = instanceId;
    this.amazonEC2Client = amazonEC2Client;
    this.consumer = consumer;
    this.desiredState = desiredState;
    this.failureStates = failureStates;
    this.taskManager = taskManager;
    this.service = service;
    this.expirationTimeMicros = expirationTimeMicros;
}

public static AWSTaskStatusChecker create(
        AmazonEC2AsyncClient amazonEC2Client, String instanceId,
        String desiredState, Consumer<Object> consumer,
        TaskManager taskManager, StatelessService service,
        long expirationTimeMicros) {
    return new AWSTaskStatusChecker(amazonEC2Client, instanceId,
            desiredState, consumer, taskManager, service, expirationTimeMicros);
}

public static AWSTaskStatusChecker create(
        AmazonEC2AsyncClient amazonEC2Client, String instanceId,
        String desiredState, List<String> failureStates, Consumer<Object> consumer,
        TaskManager taskManager, StatelessService service,
        long expirationTimeMicros) {
    return new AWSTaskStatusChecker(amazonEC2Client, instanceId, desiredState, failureStates,
            consumer, taskManager, service, expirationTimeMicros);
}

private void reset(AmazonEC2AsyncClient client, ResourceOperationRequest pr,
                    DefaultAdapterContext c) {
    if (!c.child.powerState.equals(ComputeService.PowerState.ON)) {
        logWarning(() -> String.format("Cannot perform a reset on this EC2 instance. " +
                        "The machine should be in powered on state"));
        c.taskManager.patchTaskToFailure(new IllegalStateException("Incorrect power state. Expected the machine " +
                "to be powered on "));
        return;
    }

    // The stop action for reset is a force stop. So we use the withForce method to set the force parameter to TRUE
    // This is similar to unplugging the machine from the power circuit.
    // The OS and the applications are forcefully stopped.
    StopInstancesRequest stopRequest  = new StopInstancesRequest();
    stopRequest.withInstanceIds(c.child.id).withForce(Boolean.TRUE);
    client.stopInstancesAsync(stopRequest,
            new AWSAsyncHandler<StopInstancesRequest, StopInstancesResult>() {
                @Override
                protected void handleError(Exception e) {
                    c.taskManager.patchTaskToFailure(e);
                }

                @Override
                protected void handleSuccess(StopInstancesRequest request, StopInstancesResult result) {

                    AWSUtils.waitForTransitionCompletion(getHost(),
                            result.getStoppingInstances(), "stopped", client, (is, e) -> {
                                if (e != null) {
                                    onError(e);
                                    return;
                                }
                                //Instances will be started only if they're successfully stopped
                                startInstance(client,c);
                            });
                }
            });
}

@Override
public void handlePatch(Operation op) {
    if (!op.hasBody()) {
        op.fail(new IllegalArgumentException("body is required"));
        return;
    }
    ResourceOperationRequest request = op.getBody(ResourceOperationRequest.class);
    op.complete();

    logInfo(() -> String.format("Handle operation %s for compute %s.",
            request.operation, request.resourceLink()));

    if (request.isMockRequest) {
        updateComputeState(new DefaultAdapterContext(this, request));
    } else {
        new DefaultAdapterContext(this, request)
                .populateBaseContext(BaseAdapterStage.VMDESC)
                .whenComplete((c, e) -> {
                    AmazonEC2AsyncClient client = this.clientManager.getOrCreateEC2Client(
                            c.parentAuth, c.child.description.regionId, this,
                            (t) -> c.taskManager.patchTaskToFailure(t));
                    if (client != null) {
                        reset(client,request,c);
                    }
                    // if the client is found to be null, it implies the task is already patched to
                    // failure in the catch block of getOrCreateEC2Client method (failConsumer.accept()).
                    // So it is not required to patch it again.
                });
    }
}

public static AmazonEC2AsyncClient getAsyncClient(
        AuthCredentialsServiceState credentials, String region,
        ExecutorService executorService) {

    ClientConfiguration configuration = new ClientConfiguration();
    configuration.setMaxConnections(100);
    configuration.withRetryPolicy(new RetryPolicy(new CustomRetryCondition(),
            DEFAULT_BACKOFF_STRATEGY,
            DEFAULT_MAX_ERROR_RETRY,
            false));

    AWSStaticCredentialsProvider awsStaticCredentialsProvider = new AWSStaticCredentialsProvider(
            new BasicAWSCredentials(credentials.privateKeyId,
                    EncryptionUtils.decrypt(credentials.privateKey)));

    AmazonEC2AsyncClientBuilder ec2AsyncClientBuilder = AmazonEC2AsyncClientBuilder
            .standard()
            .withClientConfiguration(configuration)
            .withCredentials(awsStaticCredentialsProvider)
            .withExecutorFactory(() -> executorService);

    if (region == null) {
        region = Regions.DEFAULT_REGION.getName();
    }

    if (isAwsClientMock()) {
        configuration.addHeader(AWS_REGION_HEADER, region);
        ec2AsyncClientBuilder.setClientConfiguration(configuration);
        AwsClientBuilder.EndpointConfiguration endpointConfiguration = new AwsClientBuilder.EndpointConfiguration(
                getAWSMockHost() + AWS_MOCK_EC2_ENDPOINT, region);
        ec2AsyncClientBuilder.setEndpointConfiguration(endpointConfiguration);
    } else {
        ec2AsyncClientBuilder.setRegion(region);
    }

    return (AmazonEC2AsyncClient) ec2AsyncClientBuilder.build();
}

public static void validateCredentials(AmazonEC2AsyncClient ec2Client,
        AWSClientManager clientManager, AuthCredentialsServiceState credentials,
        ComputeEnumerateAdapterRequest context, Operation op, StatelessService service,
        Consumer<DescribeAvailabilityZonesResult> onSuccess, Consumer<Throwable> onFail) {

    if (clientManager.isEc2ClientInvalid(credentials, context.regionId)) {
        op.complete();
        return;
    }

    ec2Client.describeAvailabilityZonesAsync(new DescribeAvailabilityZonesRequest(),
            new AsyncHandler<DescribeAvailabilityZonesRequest, DescribeAvailabilityZonesResult>() {

                @Override
                public void onError(Exception e) {
                    if (e instanceof AmazonServiceException) {
                        AmazonServiceException ase = (AmazonServiceException) e;
                        if (ase.getStatusCode() == STATUS_CODE_UNAUTHORIZED) {
                            clientManager.markEc2ClientInvalid(service, credentials,
                                    context.regionId);
                            op.complete();
                            return;
                        }
                        onFail.accept(e);
                    }
                }

                @Override
                public void onSuccess(DescribeAvailabilityZonesRequest request,
                        DescribeAvailabilityZonesResult describeAvailabilityZonesResult) {
                    onSuccess.accept(describeAvailabilityZonesResult);
                }
            });
}

/**
 * Synchronous UnTagging of one or many AWS resources with the provided tags.
 */
public static void unTagResources(AmazonEC2AsyncClient client, Collection<Tag> tags,
        String... resourceIds) {
    if (isAwsClientMock()) {
        return;
    }

    DeleteTagsRequest req = new DeleteTagsRequest()
            .withTags(tags)
            .withResources(resourceIds);

    client.deleteTags(req);
}

/**
 * Synchronous Tagging of one or many AWS resources with the provided tags.
 */
public static void tagResources(AmazonEC2AsyncClient client,
        Collection<Tag> tags, String... resourceIds) {
    if (isAwsClientMock()) {
        return;
    }

    CreateTagsRequest req = new CreateTagsRequest()
            .withResources(resourceIds).withTags(tags);

    client.createTags(req);
}

public static List<TagDescription> getResourceTags(String resourceID,
        AmazonEC2AsyncClient client) {
    Filter resource = new Filter().withName(AWS_FILTER_RESOURCE_ID)
            .withValues(resourceID);
    DescribeTagsRequest req = new DescribeTagsRequest()
            .withFilters(resource);
    DescribeTagsResult result = client.describeTags(req);
    return result.getTags();
}

public static List<String> getOrCreateDefaultSecurityGroup(AmazonEC2AsyncClient amazonEC2Client,
        AWSNicContext nicCtx) {

    AWSSecurityGroupClient client = new AWSSecurityGroupClient(amazonEC2Client);
    // in case no group is configured in the properties, attempt to discover the default one
    if (nicCtx != null && nicCtx.vpc != null) {
        try {
            SecurityGroup group = client.getSecurityGroup(
                    DEFAULT_SECURITY_GROUP_NAME,
                    nicCtx.vpc.getVpcId());
            if (group != null) {
                return Arrays.asList(group.getGroupId());
            }
        } catch (AmazonServiceException t) {
            if (!t.getMessage().contains(
                    DEFAULT_SECURITY_GROUP_NAME)) {
                throw t;
            }
        }
    }

    // if the group doesn't exist an exception is thrown. We won't throw a
    // missing group exception
    // we will continue and create the group
    String groupId = client.createDefaultSecurityGroupWithDefaultRules(nicCtx.vpc);

    return Collections.singletonList(groupId);
}

public static void waitForTransitionCompletion(ServiceHost host,
        List<InstanceStateChange> stateChangeList,
        final String desiredState, AmazonEC2AsyncClient client,
        BiConsumer<InstanceState, Exception> callback) {
    InstanceStateChange stateChange = stateChangeList.get(0);

    try {
        DescribeInstancesRequest request = new DescribeInstancesRequest();
        request.withInstanceIds(stateChange.getInstanceId());
        DescribeInstancesResult result = client.describeInstances(request);
        Instance instance = result.getReservations()
                .stream()
                .flatMap(r -> r.getInstances().stream())
                .filter(i -> i.getInstanceId()
                        .equalsIgnoreCase(stateChange.getInstanceId()))
                .findFirst().orElseThrow(() -> new IllegalArgumentException(
                        String.format("%s instance not found", stateChange.getInstanceId())));

        String state = instance.getState().getName();

        if (state.equals(desiredState)) {
            callback.accept(instance.getState(), null);
        } else {
            host.schedule(() -> waitForTransitionCompletion(host, stateChangeList, desiredState,
                    client, callback), 5, TimeUnit.SECONDS);
        }

    } catch (AmazonServiceException | IllegalArgumentException ase) {
        callback.accept(null, ase);
    }

}

@Override
public void handlePatch(Operation op) {
    if (!op.hasBody()) {
        op.fail(new IllegalArgumentException("body is required"));
        return;
    }
    ResourceOperationRequest request = op.getBody(ResourceOperationRequest.class);
    op.complete();

    logInfo("Handle operation %s for compute %s.",
            request.operation, request.resourceLink());

    if (request.isMockRequest) {
        updateComputeState(request, new DefaultAdapterContext(this, request));
    } else {
        new DefaultAdapterContext(this, request)
                .populateBaseContext(BaseAdapterStage.VMDESC)
                .whenComplete((c, e) -> {
                    AmazonEC2AsyncClient client = this.clientManager.getOrCreateEC2Client(
                            c.parentAuth, c.child.description.regionId, this,
                            (t) -> c.taskManager.patchTaskToFailure(t));
                    if (client != null) {
                        reboot(client,request,c);
                    }
                    // if the client is found to be null, it implies the task is already patched to
                    // failure in the catch block of getOrCreateEC2Client method (failConsumer.accept()).
                    // So it is not required to patch it again.
                });
    }
}

/**
 * start the instance and on success updates the disk and compute state to reflect the detach information.
 */
private void startInstance(AmazonEC2AsyncClient client, DiskContext c, DeferredResult<DiskContext> dr) {
    StartInstancesRequest startRequest  = new StartInstancesRequest();
    startRequest.withInstanceIds(c.baseAdapterContext.child.id);
    client.startInstancesAsync(startRequest,
            new AWSAsyncHandler<StartInstancesRequest, StartInstancesResult>() {

                @Override
                protected void handleError(Exception e) {
                    dr.fail(e);
                }

                @Override
                protected void handleSuccess(StartInstancesRequest request, StartInstancesResult result) {
                    AWSUtils.waitForTransitionCompletion(getHost(),
                            result.getStartingInstances(), "running",
                            client, (is, e) -> {
                                if (e != null) {
                                    dr.fail(e);
                                    return;
                                }

                                logInfo(() -> String.format(
                                        "[AWSComputeDiskDay2Service] Successfully started the "
                                                + "instance %s",
                                        result.getStartingInstances().get(0).getInstanceId()));
                                updateComputeAndDiskState(dr, c);
                            });
                }
            });
}

public static void setUpTestVolume(VerificationHost host, AmazonEC2AsyncClient client,
        Map<String, Object> awsTestContext, boolean isMock) {
    if (!isMock) {
        String volumeId = createVolume(host, client);
        awsTestContext.put(DISK_KEY, volumeId);
        String snapshotId = createSnapshot(host, client, volumeId);
        awsTestContext.put(SNAPSHOT_KEY, snapshotId);
    }
}

public static void setUpTestVpc(AmazonEC2AsyncClient client, Map<String, Object> awsTestContext, boolean isMock, String zoneId) {
    awsTestContext.put(VPC_KEY, AWS_DEFAULT_VPC_ID);
    awsTestContext.put(NIC_SPECS_KEY, SINGLE_NIC_SPEC);
    awsTestContext.put(SUBNET_KEY, AWS_DEFAULT_SUBNET_ID);
    awsTestContext.put(SECURITY_GROUP_KEY, AWS_DEFAULT_GROUP_ID);

    // create new VPC, Subnet, InternetGateway if the default VPC doesn't exist
    if (!isMock && !vpcIdExists(client, AWS_DEFAULT_VPC_ID)) {
        String vpcId = createVPC(client, AWS_DEFAULT_VPC_CIDR);
        awsTestContext.put(VPC_KEY, vpcId);
        String subnetId = createOrGetSubnet(client, AWS_DEFAULT_VPC_CIDR, vpcId, zoneId);
        awsTestContext.put(SUBNET_KEY, subnetId);

        String internetGatewayId = createInternetGateway(client);
        awsTestContext.put(INTERNET_GATEWAY_KEY, internetGatewayId);
        attachInternetGateway(client, vpcId, internetGatewayId);
        awsTestContext.put(SECURITY_GROUP_KEY, new AWSSecurityGroupClient(client)
                .createDefaultSecurityGroup(vpcId));

        NetSpec network = new NetSpec(vpcId, vpcId, AWS_DEFAULT_VPC_CIDR);

        List<NetSpec> subnets = new ArrayList<>();

        subnets.add(new NetSpec(subnetId,
                AWS_DEFAULT_SUBNET_NAME,
                AWS_DEFAULT_SUBNET_CIDR,
                zoneId == null ? TestAWSSetupUtils.zoneId + avalabilityZoneIdentifier : zoneId));

        NicSpec nicSpec = NicSpec.create()
                .withSubnetSpec(subnets.get(0))
                .withDynamicIpAssignment();

        awsTestContext.put(NIC_SPECS_KEY, new AwsNicSpecs(network, Collections.singletonList(nicSpec)));
    }
}

/**
 * Return true if vpcId exists.
 */
public static boolean vpcIdExists(AmazonEC2AsyncClient client, String vpcId) {
    List<Vpc> vpcs = client.describeVpcs()
            .getVpcs()
            .stream()
            .filter(vpc -> vpc.getVpcId().equals(vpcId))
            .collect(Collectors.toList());
    return vpcs != null && !vpcs.isEmpty();
}

/**
 * Return true if volumeId exists.
 */
public static boolean volumeIdExists(AmazonEC2AsyncClient client, String volumeId) {
    List<Volume> volumes = client.describeVolumes()
            .getVolumes()
            .stream()
            .filter(volume -> volume.getVolumeId().equals(volumeId))
            .collect(Collectors.toList());
    return volumes != null && !volumes.isEmpty();
}

/**
 * Return true if snapshotId exists.
 */
public static boolean snapshotIdExists(AmazonEC2AsyncClient client, String snapshotId) {
    List<Snapshot> snapshots = client.describeSnapshots()
            .getSnapshots()
            .stream()
            .filter(snapshot -> snapshot.getSnapshotId().equals(snapshotId))
            .collect(Collectors.toList());
    return snapshots != null && !snapshots.isEmpty();
}