Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 5dd9a1e346 |
@@ -47,7 +47,7 @@ public class AthenaJsonHandler {
|
||||
resultConfiguration = mapper.treeToValue(request.get("ResultConfiguration"), ResultConfiguration.class);
|
||||
}
|
||||
|
||||
String id = athenaService.startQueryExecution(query, workGroup, context, resultConfiguration, region);
|
||||
String id = athenaService.startQueryExecution(query, workGroup, context, resultConfiguration);
|
||||
yield Response.ok(Map.of("QueryExecutionId", id)).build();
|
||||
}
|
||||
case "GetQueryExecution" -> {
|
||||
@@ -62,7 +62,7 @@ public class AthenaJsonHandler {
|
||||
}
|
||||
case "ListQueryExecutions" -> {
|
||||
yield Response.ok(Map.of("QueryExecutionIds",
|
||||
athenaService.listQueryExecutions(request.has("WorkGroup") ? request.get("WorkGroup").asText() : null).stream()
|
||||
athenaService.listQueryExecutions().stream()
|
||||
.map(QueryExecution::getQueryExecutionId).toList())).build();
|
||||
}
|
||||
case "StopQueryExecution" -> {
|
||||
|
||||
@@ -59,9 +59,6 @@ public class AthenaService {
|
||||
public static final String DEFAULT_CATALOG = "AwsDataCatalog";
|
||||
private static final String DEFAULT_OUTPUT_BUCKET = "floci-athena-results";
|
||||
private static final String DEFAULT_WORKGROUP = "primary";
|
||||
private static final Pattern ENGINE_FAILURE = Pattern.compile("^floci-duck \\w+ returned HTTP \\d+: (\\{.*})$", Pattern.DOTALL);
|
||||
private static final com.fasterxml.jackson.databind.ObjectMapper FAILURE_READER = new com.fasterxml.jackson.databind.ObjectMapper();
|
||||
private static final Pattern TRAILING_SEMICOLONS = Pattern.compile("[\\s;]+$");
|
||||
private static final String DEFAULT_ENGINE_VERSION = "Athena engine version 3";
|
||||
private static final long MIN_BYTES_SCANNED_CUTOFF_PER_QUERY = 10_000_000L;
|
||||
private static final String WORKGROUP_RESOURCE = "workgroup/";
|
||||
@@ -125,20 +122,6 @@ public class AthenaService {
|
||||
String workGroup,
|
||||
QueryExecutionContext context,
|
||||
ResultConfiguration resultConfiguration) {
|
||||
return startQueryExecution(query, workGroup, context, resultConfiguration, null);
|
||||
}
|
||||
|
||||
/**
|
||||
* Runs a query in a workgroup. As in AWS, the workgroup's result location applies when the
|
||||
* request names none, and always when the workgroup enforces its configuration; an unknown
|
||||
* workgroup is refused. A null region skips the workgroup lookup.
|
||||
*/
|
||||
public String startQueryExecution(String query,
|
||||
String workGroup,
|
||||
QueryExecutionContext context,
|
||||
ResultConfiguration resultConfiguration,
|
||||
String region) {
|
||||
WorkGroupConfiguration workGroupConfiguration = region == null ? null : workGroupFor(region, workGroup).getConfiguration();
|
||||
String id = UUID.randomUUID().toString();
|
||||
String database = context != null && context.getDatabase() != null ? context.getDatabase() : "default";
|
||||
QueryExecutionContext resolvedContext = context != null ? context : new QueryExecutionContext();
|
||||
@@ -152,7 +135,7 @@ public class AthenaService {
|
||||
// directory prefix — the same key is written and returned to the client.
|
||||
// Statements that do not return rows must not be wrapped in a result COPY.
|
||||
String outputLocation = producesResultRows(query)
|
||||
? resolveOutputLocation(resultConfiguration, workGroupConfiguration, id)
|
||||
? resolveOutputLocation(resultConfiguration, id)
|
||||
: null;
|
||||
ResultConfiguration resolvedResult = outputLocation != null
|
||||
? new ResultConfiguration(outputLocation)
|
||||
@@ -194,7 +177,7 @@ public class AthenaService {
|
||||
if (outputLocation != null) {
|
||||
ensureOutputBucket(outputLocation);
|
||||
}
|
||||
duckClient.execute(withoutTrailingSemicolons(query), setupDdl, outputLocation);
|
||||
duckClient.execute(query, setupDdl, outputLocation);
|
||||
return null;
|
||||
}).onSuccess(v -> {
|
||||
markSucceeded(id, execution);
|
||||
@@ -231,16 +214,6 @@ public class AthenaService {
|
||||
return queryStore.scan(k -> true);
|
||||
}
|
||||
|
||||
/** A workgroup's queries, most recent first, as ListQueryExecutions returns them. */
|
||||
public List<QueryExecution> listQueryExecutions(String workGroup) {
|
||||
String name = workGroup == null || workGroup.isBlank() ? DEFAULT_WORKGROUP : workGroup;
|
||||
return queryStore.scan(k -> true).stream()
|
||||
.filter(execution -> name.equals(execution.getWorkGroup() == null ? DEFAULT_WORKGROUP : execution.getWorkGroup()))
|
||||
.sorted(Comparator.comparing((QueryExecution execution) -> execution.getStatus().getSubmissionDateTime(),
|
||||
Comparator.nullsLast(Comparator.naturalOrder())).reversed())
|
||||
.toList();
|
||||
}
|
||||
|
||||
public void stopQueryExecution(String id) {
|
||||
QueryExecution execution = getQueryExecution(id);
|
||||
execution.getStatus().setState(QueryExecutionState.CANCELLED);
|
||||
@@ -558,15 +531,6 @@ public class AthenaService {
|
||||
return properties;
|
||||
}
|
||||
|
||||
/**
|
||||
* Athena accepts one statement with or without a closing semicolon. The engine wraps the
|
||||
* statement in {@code COPY (...) TO ...}, where a semicolon is a syntax error, so it is
|
||||
* dropped before the query is handed over.
|
||||
*/
|
||||
static String withoutTrailingSemicolons(String query) {
|
||||
return query == null ? null : TRAILING_SEMICOLONS.matcher(query).replaceFirst("");
|
||||
}
|
||||
|
||||
private static boolean producesResultRows(String query) {
|
||||
return RESULT_STATEMENT_PATTERN.matcher(statementText(query)).find();
|
||||
}
|
||||
@@ -612,53 +576,17 @@ public class AthenaService {
|
||||
|
||||
private void markFailed(String id, QueryExecution execution, Throwable failure) {
|
||||
execution.getStatus().setState(QueryExecutionState.FAILED);
|
||||
execution.getStatus().setStateChangeReason(failureReason(failure.getMessage()));
|
||||
execution.getStatus().setStateChangeReason(failure.getMessage());
|
||||
queryStore.put(id, execution);
|
||||
}
|
||||
|
||||
/**
|
||||
* The engine's own explanation of a failed query, as Athena reports it: without the HTTP
|
||||
* envelope of the engine call, and without the echoed statement, which shows the
|
||||
* internal {@code COPY ... TO} wrapper rather than the query that was sent.
|
||||
*/
|
||||
static String failureReason(String message) {
|
||||
if (message == null) {
|
||||
return null;
|
||||
}
|
||||
Matcher envelope = ENGINE_FAILURE.matcher(message);
|
||||
if (!envelope.matches()) {
|
||||
return message;
|
||||
}
|
||||
try {
|
||||
String engineMessage = FAILURE_READER.readTree(envelope.group(1)).path("message").asText("");
|
||||
if (engineMessage.isBlank()) {
|
||||
return message;
|
||||
}
|
||||
int echo = engineMessage.indexOf("\n\nLINE ");
|
||||
return (echo >= 0 ? engineMessage.substring(0, echo) : engineMessage).strip();
|
||||
} catch (Exception e) {
|
||||
return message;
|
||||
}
|
||||
}
|
||||
|
||||
private record CreateDatabaseDdl(String name, boolean ifNotExists, String comment,
|
||||
String location, Map<String, String> properties) {
|
||||
}
|
||||
|
||||
private String resolveOutputLocation(ResultConfiguration rc, WorkGroupConfiguration workGroupConfiguration, String queryId) {
|
||||
String requested = rc != null && rc.getOutputLocation() != null && !rc.getOutputLocation().isBlank()
|
||||
private String resolveOutputLocation(ResultConfiguration rc, String queryId) {
|
||||
String base = (rc != null && rc.getOutputLocation() != null && !rc.getOutputLocation().isBlank())
|
||||
? rc.getOutputLocation()
|
||||
: null;
|
||||
String workGroupLocation = workGroupConfiguration != null
|
||||
&& workGroupConfiguration.getResultConfiguration() != null
|
||||
&& workGroupConfiguration.getResultConfiguration().getOutputLocation() != null
|
||||
&& !workGroupConfiguration.getResultConfiguration().getOutputLocation().isBlank()
|
||||
? workGroupConfiguration.getResultConfiguration().getOutputLocation()
|
||||
: null;
|
||||
boolean enforced = workGroupConfiguration != null && Boolean.TRUE.equals(workGroupConfiguration.getEnforceWorkGroupConfiguration());
|
||||
String base = enforced && workGroupLocation != null ? workGroupLocation
|
||||
: requested != null ? requested
|
||||
: workGroupLocation != null ? workGroupLocation
|
||||
: "s3://" + DEFAULT_OUTPUT_BUCKET + "/results/";
|
||||
return base.endsWith("/") ? base + queryId + ".csv" : base + "/" + queryId + ".csv";
|
||||
}
|
||||
@@ -764,12 +692,6 @@ public class AthenaService {
|
||||
return engineVersion;
|
||||
}
|
||||
|
||||
private WorkGroup workGroupFor(String region, String name) {
|
||||
return name == null || name.isBlank() || DEFAULT_WORKGROUP.equals(name)
|
||||
? primaryWorkGroup(region)
|
||||
: requireWorkGroup(region, name);
|
||||
}
|
||||
|
||||
private WorkGroup primaryWorkGroup(String region) {
|
||||
return workGroupStore.get(workGroupKey(region, DEFAULT_WORKGROUP)).orElseGet(() -> {
|
||||
WorkGroup workGroup = new WorkGroup();
|
||||
|
||||
@@ -29,6 +29,10 @@ public class EfsException extends AwsException {
|
||||
return new EfsException(Status.NOT_FOUND, "MountTargetNotFound", "Mount target " + mountTargetId + " does not exist.");
|
||||
}
|
||||
|
||||
public static EfsException subnetNotFound(String subnetId) {
|
||||
return new EfsException(Status.BAD_REQUEST, "SubnetNotFound", "The subnet ID '" + subnetId + "' does not exist");
|
||||
}
|
||||
|
||||
public static EfsException mountTargetConflict(String message) {
|
||||
return new EfsException(Status.CONFLICT, "MountTargetConflict", message);
|
||||
}
|
||||
|
||||
@@ -2,7 +2,10 @@ package io.github.hectorvent.floci.services.efs;
|
||||
|
||||
import io.github.hectorvent.floci.core.storage.StorageBackend;
|
||||
import io.github.hectorvent.floci.core.storage.StorageFactory;
|
||||
import io.github.hectorvent.floci.core.common.AwsException;
|
||||
import io.github.hectorvent.floci.core.common.Resettable;
|
||||
import io.github.hectorvent.floci.services.ec2.Ec2Service;
|
||||
import io.github.hectorvent.floci.services.ec2.model.Subnet;
|
||||
import io.github.hectorvent.floci.services.efs.model.*;
|
||||
import jakarta.enterprise.context.ApplicationScoped;
|
||||
import jakarta.inject.Inject;
|
||||
@@ -33,13 +36,16 @@ public class EfsService implements Resettable {
|
||||
private final ConcurrentHashMap<String, Object> syncLocks = new ConcurrentHashMap<>();
|
||||
private final RegionResolver regionResolver;
|
||||
|
||||
private final Ec2Service ec2Service;
|
||||
|
||||
private Object lockFor(String key) {
|
||||
return syncLocks.computeIfAbsent(key, k -> new Object());
|
||||
}
|
||||
|
||||
@Inject
|
||||
public EfsService(StorageFactory storageFactory, RegionResolver regionResolver) {
|
||||
public EfsService(StorageFactory storageFactory, RegionResolver regionResolver, Ec2Service ec2Service) {
|
||||
this.regionResolver = regionResolver;
|
||||
this.ec2Service = ec2Service;
|
||||
this.fileSystemStore = storageFactory.create("efs", "efs-filesystems.json",
|
||||
new com.fasterxml.jackson.core.type.TypeReference<java.util.Map<String, FileSystem>>() {});
|
||||
this.mountTargetStore = storageFactory.create("efs", "efs-mounttargets.json",
|
||||
@@ -361,14 +367,24 @@ public class EfsService implements Resettable {
|
||||
public MountTarget createMountTarget(CreateMountTargetRequest request, String region) {
|
||||
synchronized (lockFor(regionKey(region, request.getFileSystemId()))) {
|
||||
FileSystem fs = getFileSystem(region, request.getFileSystemId());
|
||||
Subnet subnet = requireSubnet(region, request.getSubnetId());
|
||||
|
||||
for (MountTarget existing : mountTargetStore.scan(k -> k.startsWith(region + "::"))) {
|
||||
List<MountTarget> regionMountTargets = mountTargetStore.scan(k -> k.startsWith(region + "::"));
|
||||
for (MountTarget existing : regionMountTargets) {
|
||||
if (existing.getFileSystemId().equals(request.getFileSystemId())) {
|
||||
if (existing.getSubnetId().equals(request.getSubnetId())) {
|
||||
throw EfsException.mountTargetConflict("Mount target already exists in this subnet");
|
||||
}
|
||||
// AWS allows one mount target per Availability Zone, all in one VPC.
|
||||
if (subnet.getAvailabilityZone().equals(existing.getAvailabilityZoneName())) {
|
||||
throw EfsException.mountTargetConflict("Mount target already exists in this AZ");
|
||||
}
|
||||
if (!subnet.getVpcId().equals(existing.getVpcId())) {
|
||||
throw EfsException.mountTargetConflict("Mount targets of a file system must be in the same VPC");
|
||||
}
|
||||
}
|
||||
}
|
||||
long inSubnet = regionMountTargets.stream().filter(existing -> request.getSubnetId().equals(existing.getSubnetId())).count();
|
||||
|
||||
String mtId = "fsmt-" + UUID.randomUUID().toString().replace("-", "").substring(0, 17);
|
||||
|
||||
@@ -376,11 +392,11 @@ public class EfsService implements Resettable {
|
||||
mt.setMountTargetId(mtId);
|
||||
mt.setFileSystemId(request.getFileSystemId());
|
||||
mt.setSubnetId(request.getSubnetId());
|
||||
mt.setIpAddress(request.getIpAddress() != null ? request.getIpAddress() : "10.0.0.10");
|
||||
mt.setIpAddress(request.getIpAddress() != null ? request.getIpAddress() : hostAddress(subnet.getCidrBlock(), inSubnet));
|
||||
mt.setLifeCycleState(LifeCycleState.available);
|
||||
mt.setVpcId("vpc-12345678");
|
||||
mt.setAvailabilityZoneName(region + "a");
|
||||
mt.setAvailabilityZoneId(generateAzId(region + "a"));
|
||||
mt.setVpcId(subnet.getVpcId());
|
||||
mt.setAvailabilityZoneName(subnet.getAvailabilityZone());
|
||||
mt.setAvailabilityZoneId(generateAzId(subnet.getAvailabilityZone()));
|
||||
mt.setNetworkInterfaceId("eni-" + UUID.randomUUID().toString().replace("-", "").substring(0, 17));
|
||||
if (request.getSecurityGroups() != null) {
|
||||
mt.setSecurityGroups(new ArrayList<>(request.getSecurityGroups()));
|
||||
@@ -613,6 +629,32 @@ public class EfsService implements Resettable {
|
||||
return region + "::" + id;
|
||||
}
|
||||
|
||||
private Subnet requireSubnet(String region, String subnetId) {
|
||||
if (subnetId == null || subnetId.isBlank()) {
|
||||
throw EfsException.badRequest("SubnetId is required.");
|
||||
}
|
||||
try {
|
||||
return ec2Service.requireSubnet(region, subnetId);
|
||||
} catch (AwsException e) {
|
||||
throw EfsException.subnetNotFound(subnetId);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* An address inside the subnet for its n-th mount target, skipping the first four
|
||||
* addresses AWS reserves in every subnet.
|
||||
*/
|
||||
static String hostAddress(String cidrBlock, long index) {
|
||||
String[] parts = cidrBlock.split("/");
|
||||
String[] octets = parts[0].split("\\.");
|
||||
long base = 0;
|
||||
for (String octet : octets) {
|
||||
base = (base << 8) | Long.parseLong(octet);
|
||||
}
|
||||
long address = base + 4 + index;
|
||||
return ((address >> 24) & 0xff) + "." + ((address >> 16) & 0xff) + "." + ((address >> 8) & 0xff) + "." + (address & 0xff);
|
||||
}
|
||||
|
||||
private String generateAzId(String azName) {
|
||||
if (azName == null || azName.length() < 2) return azName + "-id";
|
||||
String region = azName.substring(0, azName.length() - 1);
|
||||
|
||||
@@ -1,22 +0,0 @@
|
||||
package io.github.hectorvent.floci.services.athena;
|
||||
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
|
||||
class AthenaQueryTextTest {
|
||||
|
||||
@Test
|
||||
void dropsTheClosingSemicolonsAthenaAccepts() {
|
||||
assertEquals("SELECT * FROM orders LIMIT 10", AthenaService.withoutTrailingSemicolons("SELECT * FROM orders LIMIT 10;"));
|
||||
assertEquals("SELECT 1", AthenaService.withoutTrailingSemicolons("SELECT 1 ; ;\n"));
|
||||
assertEquals("SELECT ';' AS s", AthenaService.withoutTrailingSemicolons("SELECT ';' AS s"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void reportsTheEngineErrorWithoutItsEnvelopeOrTheWrappedStatement() {
|
||||
String engine = "floci-duck execute returned HTTP 500: {\"status\":\"error\",\"message\":\"Catalog Error: Table with name nope does not exist!\\nDid you mean \\\"orders\\\"?\\n\\nLINE 1: COPY (SELECT * FROM nope) TO 's3://results/q.csv'\"}";
|
||||
assertEquals("Catalog Error: Table with name nope does not exist!\nDid you mean \"orders\"?", AthenaService.failureReason(engine));
|
||||
assertEquals("Failed to call floci-duck execute: timeout", AthenaService.failureReason("Failed to call floci-duck execute: timeout"));
|
||||
}
|
||||
}
|
||||
-109
@@ -1,109 +0,0 @@
|
||||
package io.github.hectorvent.floci.services.athena;
|
||||
|
||||
import io.github.hectorvent.floci.testing.RestAssuredJsonUtils;
|
||||
import io.quarkus.test.junit.QuarkusTest;
|
||||
import org.junit.jupiter.api.BeforeAll;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.UUID;
|
||||
|
||||
import static io.restassured.RestAssured.given;
|
||||
import static org.hamcrest.Matchers.equalTo;
|
||||
import static org.hamcrest.Matchers.hasItem;
|
||||
import static org.hamcrest.Matchers.not;
|
||||
import static org.hamcrest.Matchers.startsWith;
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
|
||||
/** Queries belong to their workgroup: they use its result location and are listed under it. */
|
||||
@QuarkusTest
|
||||
class AthenaWorkGroupQueriesIntegrationTest {
|
||||
|
||||
private static final String CONTENT_TYPE = "application/x-amz-json-1.1";
|
||||
|
||||
@BeforeAll
|
||||
static void configureRestAssured() {
|
||||
RestAssuredJsonUtils.configureAwsContentTypes();
|
||||
}
|
||||
|
||||
private static io.restassured.response.ValidatableResponse athena(String action, String body) {
|
||||
return given()
|
||||
.header("X-Amz-Target", "AmazonAthena." + action)
|
||||
.contentType(CONTENT_TYPE)
|
||||
.body(body)
|
||||
.when().post("/")
|
||||
.then();
|
||||
}
|
||||
|
||||
private static String workGroup(String location, boolean enforce) {
|
||||
String name = "wg-" + UUID.randomUUID().toString().substring(0, 8);
|
||||
athena("CreateWorkGroup", """
|
||||
{"Name": "%s", "Configuration": {"ResultConfiguration": {"OutputLocation": "%s"}, "EnforceWorkGroupConfiguration": %s}}
|
||||
""".formatted(name, location, enforce)).statusCode(200);
|
||||
return name;
|
||||
}
|
||||
|
||||
private static String start(String workGroup, String outputLocation) {
|
||||
String result = outputLocation == null ? "" : ", \"ResultConfiguration\": {\"OutputLocation\": \"" + outputLocation + "\"}";
|
||||
return athena("StartQueryExecution", "{\"QueryString\": \"SELECT 1\", \"WorkGroup\": \"" + workGroup + "\"" + result + "}")
|
||||
.statusCode(200)
|
||||
.extract().jsonPath().getString("QueryExecutionId");
|
||||
}
|
||||
|
||||
private static String outputLocation(String queryId) {
|
||||
return athena("GetQueryExecution", "{\"QueryExecutionId\": \"" + queryId + "\"}")
|
||||
.statusCode(200)
|
||||
.body("QueryExecution.WorkGroup", not(equalTo(null)))
|
||||
.extract().jsonPath().getString("QueryExecution.ResultConfiguration.OutputLocation");
|
||||
}
|
||||
|
||||
@Test
|
||||
void usesTheWorkGroupResultLocationWhenTheQueryNamesNone() {
|
||||
String name = workGroup("s3://team-results/athena/", false);
|
||||
String queryId = start(name, null);
|
||||
|
||||
assertEquals("s3://team-results/athena/" + queryId + ".csv", outputLocation(queryId));
|
||||
}
|
||||
|
||||
@Test
|
||||
void aRequestedLocationWinsUnlessTheWorkGroupEnforcesItsOwn() {
|
||||
String open = workGroup("s3://team-results/open/", false);
|
||||
String openQuery = start(open, "s3://mine/results/");
|
||||
assertEquals("s3://mine/results/" + openQuery + ".csv", outputLocation(openQuery));
|
||||
|
||||
String enforced = workGroup("s3://team-results/enforced/", true);
|
||||
String enforcedQuery = start(enforced, "s3://mine/results/");
|
||||
assertEquals("s3://team-results/enforced/" + enforcedQuery + ".csv", outputLocation(enforcedQuery));
|
||||
}
|
||||
|
||||
@Test
|
||||
void refusesAWorkGroupThatDoesNotExist() {
|
||||
athena("StartQueryExecution", "{\"QueryString\": \"SELECT 1\", \"WorkGroup\": \"no-such-group\"}")
|
||||
.statusCode(400)
|
||||
.body("__type", equalTo("InvalidRequestException"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void listsOnlyTheWorkGroupQueriesNewestFirst() {
|
||||
String name = workGroup("s3://team-results/listed/", false);
|
||||
String first = start(name, null);
|
||||
String second = start(name, null);
|
||||
String elsewhere = start("primary", null);
|
||||
|
||||
List<String> listed = athena("ListQueryExecutions", "{\"WorkGroup\": \"" + name + "\"}")
|
||||
.statusCode(200)
|
||||
.extract().jsonPath().getList("QueryExecutionIds", String.class);
|
||||
assertEquals(List.of(second, first), listed);
|
||||
|
||||
athena("ListQueryExecutions", "{}")
|
||||
.statusCode(200)
|
||||
.body("QueryExecutionIds", hasItem(elsewhere))
|
||||
.body("QueryExecutionIds", not(hasItem(first)));
|
||||
}
|
||||
|
||||
@Test
|
||||
void thePrimaryWorkGroupKeepsItsDefaultLocation() {
|
||||
String queryId = start("primary", null);
|
||||
org.hamcrest.MatcherAssert.assertThat(outputLocation(queryId), startsWith("s3://floci-athena-results/results/"));
|
||||
}
|
||||
}
|
||||
@@ -90,7 +90,7 @@ class EfsIntegrationTest {
|
||||
.body("""
|
||||
{
|
||||
"FileSystemId": "%s",
|
||||
"SubnetId": "subnet-12345",
|
||||
"SubnetId": "subnet-default-us-east-1-a",
|
||||
"SecurityGroups": ["sg-11111"]
|
||||
}
|
||||
""".formatted(fileSystemId))
|
||||
@@ -100,7 +100,10 @@ class EfsIntegrationTest {
|
||||
.statusCode(200)
|
||||
.body("MountTargetId", startsWith("fsmt-"))
|
||||
.body("FileSystemId", equalTo(fileSystemId))
|
||||
.body("SubnetId", equalTo("subnet-12345"))
|
||||
.body("SubnetId", equalTo("subnet-default-us-east-1-a"))
|
||||
.body("VpcId", equalTo("vpc-default-us-east-1"))
|
||||
.body("AvailabilityZoneName", equalTo("us-east-1a"))
|
||||
.body("IpAddress", equalTo("172.31.0.4"))
|
||||
.extract().jsonPath().getString("MountTargetId");
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,91 @@
|
||||
package io.github.hectorvent.floci.services.efs;
|
||||
|
||||
import io.github.hectorvent.floci.testing.RestAssuredJsonUtils;
|
||||
import io.quarkus.test.junit.QuarkusTest;
|
||||
import org.junit.jupiter.api.BeforeAll;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import static io.restassured.RestAssured.given;
|
||||
import static org.hamcrest.Matchers.equalTo;
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
|
||||
/** Mount targets take their VPC, zone and address from the EC2 subnet they are created in. */
|
||||
@QuarkusTest
|
||||
class EfsMountTargetSubnetTest {
|
||||
|
||||
@BeforeAll
|
||||
static void configureRestAssured() {
|
||||
RestAssuredJsonUtils.configureAwsContentTypes();
|
||||
}
|
||||
|
||||
private String createFileSystem(String token) {
|
||||
return given()
|
||||
.contentType("application/json")
|
||||
.body("{\"CreationToken\": \"" + token + "\"}")
|
||||
.when()
|
||||
.post("/2015-02-01/file-systems")
|
||||
.then()
|
||||
.statusCode(201)
|
||||
.extract().jsonPath().getString("FileSystemId");
|
||||
}
|
||||
|
||||
private io.restassured.response.ValidatableResponse createMountTarget(String fileSystemId, String subnetId) {
|
||||
return given()
|
||||
.contentType("application/json")
|
||||
.body("{\"FileSystemId\": \"" + fileSystemId + "\", \"SubnetId\": \"" + subnetId + "\"}")
|
||||
.when()
|
||||
.post("/2015-02-01/mount-targets")
|
||||
.then();
|
||||
}
|
||||
|
||||
@Test
|
||||
void usesTheZoneAndVpcOfEachSubnet() {
|
||||
String fileSystemId = createFileSystem("subnet-zones");
|
||||
|
||||
createMountTarget(fileSystemId, "subnet-default-us-east-1-b")
|
||||
.statusCode(200)
|
||||
.body("VpcId", equalTo("vpc-default-us-east-1"))
|
||||
.body("AvailabilityZoneName", equalTo("us-east-1b"))
|
||||
.body("IpAddress", equalTo("172.31.16.4"));
|
||||
createMountTarget(fileSystemId, "subnet-default-us-east-1-c")
|
||||
.statusCode(200)
|
||||
.body("AvailabilityZoneName", equalTo("us-east-1c"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void refusesASubnetThatDoesNotExist() {
|
||||
String fileSystemId = createFileSystem("subnet-missing");
|
||||
|
||||
createMountTarget(fileSystemId, "subnet-nope")
|
||||
.statusCode(400)
|
||||
.body("__type", equalTo("SubnetNotFound"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void allowsOneMountTargetPerZone() {
|
||||
String fileSystemId = createFileSystem("subnet-one-per-zone");
|
||||
String vpcId = "vpc-default-us-east-1";
|
||||
String secondSubnet = given()
|
||||
.formParam("Action", "CreateSubnet")
|
||||
.formParam("VpcId", vpcId)
|
||||
.formParam("CidrBlock", "172.31.200.0/24")
|
||||
.formParam("AvailabilityZone", "us-east-1a")
|
||||
.header("Authorization", "AWS4-HMAC-SHA256 Credential=test/20260101/us-east-1/ec2/aws4_request")
|
||||
.when()
|
||||
.post("/")
|
||||
.then()
|
||||
.statusCode(200)
|
||||
.extract().path("CreateSubnetResponse.subnet.subnetId");
|
||||
|
||||
createMountTarget(fileSystemId, "subnet-default-us-east-1-a").statusCode(200);
|
||||
createMountTarget(fileSystemId, secondSubnet)
|
||||
.statusCode(409)
|
||||
.body("__type", equalTo("MountTargetConflict"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void picksAddressesAfterTheFourAwsReserves() {
|
||||
assertEquals("10.0.1.4", EfsService.hostAddress("10.0.1.0/24", 0));
|
||||
assertEquals("10.0.2.1", EfsService.hostAddress("10.0.1.0/24", 253));
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user