Compare commits
3 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 57aa866144 | |||
| ac96c6451e | |||
| 39664957fe |
@@ -47,7 +47,7 @@ public class AthenaJsonHandler {
|
|||||||
resultConfiguration = mapper.treeToValue(request.get("ResultConfiguration"), ResultConfiguration.class);
|
resultConfiguration = mapper.treeToValue(request.get("ResultConfiguration"), ResultConfiguration.class);
|
||||||
}
|
}
|
||||||
|
|
||||||
String id = athenaService.startQueryExecution(query, workGroup, context, resultConfiguration);
|
String id = athenaService.startQueryExecution(query, workGroup, context, resultConfiguration, region);
|
||||||
yield Response.ok(Map.of("QueryExecutionId", id)).build();
|
yield Response.ok(Map.of("QueryExecutionId", id)).build();
|
||||||
}
|
}
|
||||||
case "GetQueryExecution" -> {
|
case "GetQueryExecution" -> {
|
||||||
@@ -62,7 +62,7 @@ public class AthenaJsonHandler {
|
|||||||
}
|
}
|
||||||
case "ListQueryExecutions" -> {
|
case "ListQueryExecutions" -> {
|
||||||
yield Response.ok(Map.of("QueryExecutionIds",
|
yield Response.ok(Map.of("QueryExecutionIds",
|
||||||
athenaService.listQueryExecutions().stream()
|
athenaService.listQueryExecutions(request.has("WorkGroup") ? request.get("WorkGroup").asText() : null).stream()
|
||||||
.map(QueryExecution::getQueryExecutionId).toList())).build();
|
.map(QueryExecution::getQueryExecutionId).toList())).build();
|
||||||
}
|
}
|
||||||
case "StopQueryExecution" -> {
|
case "StopQueryExecution" -> {
|
||||||
|
|||||||
@@ -59,6 +59,9 @@ public class AthenaService {
|
|||||||
public static final String DEFAULT_CATALOG = "AwsDataCatalog";
|
public static final String DEFAULT_CATALOG = "AwsDataCatalog";
|
||||||
private static final String DEFAULT_OUTPUT_BUCKET = "floci-athena-results";
|
private static final String DEFAULT_OUTPUT_BUCKET = "floci-athena-results";
|
||||||
private static final String DEFAULT_WORKGROUP = "primary";
|
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 String DEFAULT_ENGINE_VERSION = "Athena engine version 3";
|
||||||
private static final long MIN_BYTES_SCANNED_CUTOFF_PER_QUERY = 10_000_000L;
|
private static final long MIN_BYTES_SCANNED_CUTOFF_PER_QUERY = 10_000_000L;
|
||||||
private static final String WORKGROUP_RESOURCE = "workgroup/";
|
private static final String WORKGROUP_RESOURCE = "workgroup/";
|
||||||
@@ -122,6 +125,20 @@ public class AthenaService {
|
|||||||
String workGroup,
|
String workGroup,
|
||||||
QueryExecutionContext context,
|
QueryExecutionContext context,
|
||||||
ResultConfiguration resultConfiguration) {
|
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 id = UUID.randomUUID().toString();
|
||||||
String database = context != null && context.getDatabase() != null ? context.getDatabase() : "default";
|
String database = context != null && context.getDatabase() != null ? context.getDatabase() : "default";
|
||||||
QueryExecutionContext resolvedContext = context != null ? context : new QueryExecutionContext();
|
QueryExecutionContext resolvedContext = context != null ? context : new QueryExecutionContext();
|
||||||
@@ -135,7 +152,7 @@ public class AthenaService {
|
|||||||
// directory prefix — the same key is written and returned to the client.
|
// 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.
|
// Statements that do not return rows must not be wrapped in a result COPY.
|
||||||
String outputLocation = producesResultRows(query)
|
String outputLocation = producesResultRows(query)
|
||||||
? resolveOutputLocation(resultConfiguration, id)
|
? resolveOutputLocation(resultConfiguration, workGroupConfiguration, id)
|
||||||
: null;
|
: null;
|
||||||
ResultConfiguration resolvedResult = outputLocation != null
|
ResultConfiguration resolvedResult = outputLocation != null
|
||||||
? new ResultConfiguration(outputLocation)
|
? new ResultConfiguration(outputLocation)
|
||||||
@@ -177,7 +194,7 @@ public class AthenaService {
|
|||||||
if (outputLocation != null) {
|
if (outputLocation != null) {
|
||||||
ensureOutputBucket(outputLocation);
|
ensureOutputBucket(outputLocation);
|
||||||
}
|
}
|
||||||
duckClient.execute(query, setupDdl, outputLocation);
|
duckClient.execute(withoutTrailingSemicolons(query), setupDdl, outputLocation);
|
||||||
return null;
|
return null;
|
||||||
}).onSuccess(v -> {
|
}).onSuccess(v -> {
|
||||||
markSucceeded(id, execution);
|
markSucceeded(id, execution);
|
||||||
@@ -214,6 +231,16 @@ public class AthenaService {
|
|||||||
return queryStore.scan(k -> true);
|
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) {
|
public void stopQueryExecution(String id) {
|
||||||
QueryExecution execution = getQueryExecution(id);
|
QueryExecution execution = getQueryExecution(id);
|
||||||
execution.getStatus().setState(QueryExecutionState.CANCELLED);
|
execution.getStatus().setState(QueryExecutionState.CANCELLED);
|
||||||
@@ -531,6 +558,15 @@ public class AthenaService {
|
|||||||
return properties;
|
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) {
|
private static boolean producesResultRows(String query) {
|
||||||
return RESULT_STATEMENT_PATTERN.matcher(statementText(query)).find();
|
return RESULT_STATEMENT_PATTERN.matcher(statementText(query)).find();
|
||||||
}
|
}
|
||||||
@@ -576,17 +612,53 @@ public class AthenaService {
|
|||||||
|
|
||||||
private void markFailed(String id, QueryExecution execution, Throwable failure) {
|
private void markFailed(String id, QueryExecution execution, Throwable failure) {
|
||||||
execution.getStatus().setState(QueryExecutionState.FAILED);
|
execution.getStatus().setState(QueryExecutionState.FAILED);
|
||||||
execution.getStatus().setStateChangeReason(failure.getMessage());
|
execution.getStatus().setStateChangeReason(failureReason(failure.getMessage()));
|
||||||
queryStore.put(id, execution);
|
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,
|
private record CreateDatabaseDdl(String name, boolean ifNotExists, String comment,
|
||||||
String location, Map<String, String> properties) {
|
String location, Map<String, String> properties) {
|
||||||
}
|
}
|
||||||
|
|
||||||
private String resolveOutputLocation(ResultConfiguration rc, String queryId) {
|
private String resolveOutputLocation(ResultConfiguration rc, WorkGroupConfiguration workGroupConfiguration, String queryId) {
|
||||||
String base = (rc != null && rc.getOutputLocation() != null && !rc.getOutputLocation().isBlank())
|
String requested = rc != null && rc.getOutputLocation() != null && !rc.getOutputLocation().isBlank()
|
||||||
? rc.getOutputLocation()
|
? 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/";
|
: "s3://" + DEFAULT_OUTPUT_BUCKET + "/results/";
|
||||||
return base.endsWith("/") ? base + queryId + ".csv" : base + "/" + queryId + ".csv";
|
return base.endsWith("/") ? base + queryId + ".csv" : base + "/" + queryId + ".csv";
|
||||||
}
|
}
|
||||||
@@ -692,6 +764,12 @@ public class AthenaService {
|
|||||||
return engineVersion;
|
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) {
|
private WorkGroup primaryWorkGroup(String region) {
|
||||||
return workGroupStore.get(workGroupKey(region, DEFAULT_WORKGROUP)).orElseGet(() -> {
|
return workGroupStore.get(workGroupKey(region, DEFAULT_WORKGROUP)).orElseGet(() -> {
|
||||||
WorkGroup workGroup = new WorkGroup();
|
WorkGroup workGroup = new WorkGroup();
|
||||||
|
|||||||
@@ -29,10 +29,6 @@ public class EfsException extends AwsException {
|
|||||||
return new EfsException(Status.NOT_FOUND, "MountTargetNotFound", "Mount target " + mountTargetId + " does not exist.");
|
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) {
|
public static EfsException mountTargetConflict(String message) {
|
||||||
return new EfsException(Status.CONFLICT, "MountTargetConflict", message);
|
return new EfsException(Status.CONFLICT, "MountTargetConflict", message);
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -2,10 +2,7 @@ package io.github.hectorvent.floci.services.efs;
|
|||||||
|
|
||||||
import io.github.hectorvent.floci.core.storage.StorageBackend;
|
import io.github.hectorvent.floci.core.storage.StorageBackend;
|
||||||
import io.github.hectorvent.floci.core.storage.StorageFactory;
|
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.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 io.github.hectorvent.floci.services.efs.model.*;
|
||||||
import jakarta.enterprise.context.ApplicationScoped;
|
import jakarta.enterprise.context.ApplicationScoped;
|
||||||
import jakarta.inject.Inject;
|
import jakarta.inject.Inject;
|
||||||
@@ -36,16 +33,13 @@ public class EfsService implements Resettable {
|
|||||||
private final ConcurrentHashMap<String, Object> syncLocks = new ConcurrentHashMap<>();
|
private final ConcurrentHashMap<String, Object> syncLocks = new ConcurrentHashMap<>();
|
||||||
private final RegionResolver regionResolver;
|
private final RegionResolver regionResolver;
|
||||||
|
|
||||||
private final Ec2Service ec2Service;
|
|
||||||
|
|
||||||
private Object lockFor(String key) {
|
private Object lockFor(String key) {
|
||||||
return syncLocks.computeIfAbsent(key, k -> new Object());
|
return syncLocks.computeIfAbsent(key, k -> new Object());
|
||||||
}
|
}
|
||||||
|
|
||||||
@Inject
|
@Inject
|
||||||
public EfsService(StorageFactory storageFactory, RegionResolver regionResolver, Ec2Service ec2Service) {
|
public EfsService(StorageFactory storageFactory, RegionResolver regionResolver) {
|
||||||
this.regionResolver = regionResolver;
|
this.regionResolver = regionResolver;
|
||||||
this.ec2Service = ec2Service;
|
|
||||||
this.fileSystemStore = storageFactory.create("efs", "efs-filesystems.json",
|
this.fileSystemStore = storageFactory.create("efs", "efs-filesystems.json",
|
||||||
new com.fasterxml.jackson.core.type.TypeReference<java.util.Map<String, FileSystem>>() {});
|
new com.fasterxml.jackson.core.type.TypeReference<java.util.Map<String, FileSystem>>() {});
|
||||||
this.mountTargetStore = storageFactory.create("efs", "efs-mounttargets.json",
|
this.mountTargetStore = storageFactory.create("efs", "efs-mounttargets.json",
|
||||||
@@ -367,24 +361,14 @@ public class EfsService implements Resettable {
|
|||||||
public MountTarget createMountTarget(CreateMountTargetRequest request, String region) {
|
public MountTarget createMountTarget(CreateMountTargetRequest request, String region) {
|
||||||
synchronized (lockFor(regionKey(region, request.getFileSystemId()))) {
|
synchronized (lockFor(regionKey(region, request.getFileSystemId()))) {
|
||||||
FileSystem fs = getFileSystem(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.getFileSystemId().equals(request.getFileSystemId())) {
|
||||||
if (existing.getSubnetId().equals(request.getSubnetId())) {
|
if (existing.getSubnetId().equals(request.getSubnetId())) {
|
||||||
throw EfsException.mountTargetConflict("Mount target already exists in this subnet");
|
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);
|
String mtId = "fsmt-" + UUID.randomUUID().toString().replace("-", "").substring(0, 17);
|
||||||
|
|
||||||
@@ -392,11 +376,11 @@ public class EfsService implements Resettable {
|
|||||||
mt.setMountTargetId(mtId);
|
mt.setMountTargetId(mtId);
|
||||||
mt.setFileSystemId(request.getFileSystemId());
|
mt.setFileSystemId(request.getFileSystemId());
|
||||||
mt.setSubnetId(request.getSubnetId());
|
mt.setSubnetId(request.getSubnetId());
|
||||||
mt.setIpAddress(request.getIpAddress() != null ? request.getIpAddress() : hostAddress(subnet.getCidrBlock(), inSubnet));
|
mt.setIpAddress(request.getIpAddress() != null ? request.getIpAddress() : "10.0.0.10");
|
||||||
mt.setLifeCycleState(LifeCycleState.available);
|
mt.setLifeCycleState(LifeCycleState.available);
|
||||||
mt.setVpcId(subnet.getVpcId());
|
mt.setVpcId("vpc-12345678");
|
||||||
mt.setAvailabilityZoneName(subnet.getAvailabilityZone());
|
mt.setAvailabilityZoneName(region + "a");
|
||||||
mt.setAvailabilityZoneId(generateAzId(subnet.getAvailabilityZone()));
|
mt.setAvailabilityZoneId(generateAzId(region + "a"));
|
||||||
mt.setNetworkInterfaceId("eni-" + UUID.randomUUID().toString().replace("-", "").substring(0, 17));
|
mt.setNetworkInterfaceId("eni-" + UUID.randomUUID().toString().replace("-", "").substring(0, 17));
|
||||||
if (request.getSecurityGroups() != null) {
|
if (request.getSecurityGroups() != null) {
|
||||||
mt.setSecurityGroups(new ArrayList<>(request.getSecurityGroups()));
|
mt.setSecurityGroups(new ArrayList<>(request.getSecurityGroups()));
|
||||||
@@ -629,32 +613,6 @@ public class EfsService implements Resettable {
|
|||||||
return region + "::" + id;
|
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) {
|
private String generateAzId(String azName) {
|
||||||
if (azName == null || azName.length() < 2) return azName + "-id";
|
if (azName == null || azName.length() < 2) return azName + "-id";
|
||||||
String region = azName.substring(0, azName.length() - 1);
|
String region = azName.substring(0, azName.length() - 1);
|
||||||
|
|||||||
@@ -0,0 +1,22 @@
|
|||||||
|
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
@@ -0,0 +1,109 @@
|
|||||||
|
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("""
|
.body("""
|
||||||
{
|
{
|
||||||
"FileSystemId": "%s",
|
"FileSystemId": "%s",
|
||||||
"SubnetId": "subnet-default-us-east-1-a",
|
"SubnetId": "subnet-12345",
|
||||||
"SecurityGroups": ["sg-11111"]
|
"SecurityGroups": ["sg-11111"]
|
||||||
}
|
}
|
||||||
""".formatted(fileSystemId))
|
""".formatted(fileSystemId))
|
||||||
@@ -100,10 +100,7 @@ class EfsIntegrationTest {
|
|||||||
.statusCode(200)
|
.statusCode(200)
|
||||||
.body("MountTargetId", startsWith("fsmt-"))
|
.body("MountTargetId", startsWith("fsmt-"))
|
||||||
.body("FileSystemId", equalTo(fileSystemId))
|
.body("FileSystemId", equalTo(fileSystemId))
|
||||||
.body("SubnetId", equalTo("subnet-default-us-east-1-a"))
|
.body("SubnetId", equalTo("subnet-12345"))
|
||||||
.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");
|
.extract().jsonPath().getString("MountTargetId");
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -1,91 +0,0 @@
|
|||||||
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