Skip to content

Commit 3f90b7b

Browse files
DaveCTurnerDaveCTurner
committed
Reinstate S3BlobContainerRetriesTests
1 parent 47860ac commit 3f90b7b

File tree

3 files changed

+115
-61
lines changed

3 files changed

+115
-61
lines changed

‎modules/repository-s3/src/main/java/org/elasticsearch/repositories/s3/S3BlobStore.java‎

Lines changed: 10 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -51,10 +51,12 @@
5151
import java.util.Locale;
5252
import java.util.Map;
5353
import java.util.Objects;
54+
import java.util.Set;
5455
import java.util.concurrent.ConcurrentHashMap;
5556
import java.util.concurrent.Executor;
5657
import java.util.concurrent.TimeUnit;
5758
import java.util.concurrent.atomic.LongAdder;
59+
import java.util.function.Predicate;
5860
import java.util.stream.Collectors;
5961

6062
class S3BlobStore implements BlobStore {
@@ -564,15 +566,18 @@ static Operation parse(String s) {
564566
);
565567
}
566568

569+
private static final Predicate<String> IS_PUT_MULTIPART_OPERATION = Set.of("CreateMultipartUpload", "UploadPart")::contains;
570+
567571
boolean assertConsistentOperationName(MetricCollection metricCollection) {
568572
final var operationNameMetrics = metricCollection.metricValues(CoreMetric.OPERATION_NAME);
569573
assert operationNameMetrics.size() == 1 : operationNameMetrics;
570-
final var expectedOperationName = switch (this) {
571-
case LIST_OBJECTS -> "ListObjectsV2";
572-
case PUT_MULTIPART_OBJECT -> "CreateMultipartUpload";
573-
default -> key;
574+
final Predicate<String> expectedOperationPredicate = switch (this) {
575+
case LIST_OBJECTS -> "ListObjectsV2"::equals;
576+
case PUT_MULTIPART_OBJECT -> IS_PUT_MULTIPART_OPERATION;
577+
case ABORT_MULTIPART_OBJECT -> "AbortMultipartUpload"::equals;
578+
default -> key::equals;
574579
};
575-
assert expectedOperationName.equals(operationNameMetrics.get(0)) : expectedOperationName + " vs " + operationNameMetrics;
580+
assert expectedOperationPredicate.test(operationNameMetrics.get(0)) : this + " vs " + operationNameMetrics;
576581
return true;
577582
}
578583
}

‎modules/repository-s3/src/test/java/org/elasticsearch/repositories/s3/S3BlobContainerRetriesTests.javaTODO‎ renamed to ‎modules/repository-s3/src/test/java/org/elasticsearch/repositories/s3/S3BlobContainerRetriesTests.java‎

Lines changed: 105 additions & 55 deletions
Original file line numberDiff line numberDiff line change
@@ -9,13 +9,10 @@
99
package org.elasticsearch.repositories.s3;
1010

1111
import fixture.s3.S3HttpHandler;
12-
import software.amazon.awssdk.http.apache.ApacheHttpClient;
12+
import software.amazon.awssdk.core.exception.SdkClientException;
13+
import software.amazon.awssdk.core.exception.SdkException;
14+
import software.amazon.awssdk.services.s3.model.S3Exception;
1315

14-
import com.amazonaws.AbortedException;
15-
import com.amazonaws.SdkClientException;
16-
import com.amazonaws.services.s3.internal.MD5DigestCalculatingInputStream;
17-
import com.amazonaws.services.s3.model.AmazonS3Exception;
18-
import com.amazonaws.util.Base16;
1916
import com.sun.net.httpserver.HttpExchange;
2017
import com.sun.net.httpserver.HttpHandler;
2118

@@ -32,6 +29,7 @@
3229
import org.elasticsearch.common.blobstore.OperationPurpose;
3330
import org.elasticsearch.common.blobstore.OptionalBytesReference;
3431
import org.elasticsearch.common.bytes.BytesReference;
32+
import org.elasticsearch.common.hash.MessageDigests;
3533
import org.elasticsearch.common.io.Streams;
3634
import org.elasticsearch.common.lucene.store.ByteArrayIndexInput;
3735
import org.elasticsearch.common.lucene.store.InputStreamIndexInput;
@@ -48,6 +46,8 @@
4846
import org.elasticsearch.core.SuppressForbidden;
4947
import org.elasticsearch.core.TimeValue;
5048
import org.elasticsearch.env.Environment;
49+
import org.elasticsearch.logging.LogManager;
50+
import org.elasticsearch.logging.Logger;
5151
import org.elasticsearch.repositories.RepositoriesMetrics;
5252
import org.elasticsearch.repositories.blobstore.AbstractBlobContainerRetriesTestCase;
5353
import org.elasticsearch.repositories.blobstore.BlobStoreTestUtil;
@@ -70,9 +70,9 @@
7070
import java.io.InputStreamReader;
7171
import java.net.InetSocketAddress;
7272
import java.net.SocketTimeoutException;
73-
import java.net.UnknownHostException;
7473
import java.nio.charset.StandardCharsets;
7574
import java.nio.file.NoSuchFileException;
75+
import java.security.MessageDigest;
7676
import java.util.ArrayList;
7777
import java.util.Arrays;
7878
import java.util.List;
@@ -116,33 +116,12 @@ public class S3BlobContainerRetriesTests extends AbstractBlobContainerRetriesTes
116116

117117
private static final int MAX_NUMBER_SNAPSHOT_DELETE_RETRIES = 10;
118118
private S3Service service;
119-
private AtomicBoolean shouldErrorOnDns;
119+
// private AtomicBoolean shouldErrorOnDns; // TODO NOMERGE do we need to cover this too?
120120
private RecordingMeterRegistry recordingMeterRegistry;
121121

122122
@Before
123123
public void setUp() throws Exception {
124-
shouldErrorOnDns = new AtomicBoolean(false);
125-
service = new S3Service(Mockito.mock(Environment.class), Settings.EMPTY, Mockito.mock(ResourceWatcherService.class)) {
126-
@Override
127-
ApacheHttpClient.Builder buildHttpClient(S3ClientSettings clientSettings) {
128-
// override http server builder dnsResolver
129-
final ApacheHttpClient.Builder builder = super.buildHttpClient(clientSettings);
130-
131-
// NOMERGE: TODO: There doesn't appear to be access to the default DNS Resolver in the HttpServer builder or elsewhere.
132-
// Need to find some alternate way to force request errors to test retries...
133-
// final DnsResolver defaultDnsResolver = builder.getClientConfiguration().getDnsResolver();
134-
// DnsResolver defaultDnsResolver = SystemDefaultRoutePlanner.getSystemDefaultDnsResolver();
135-
136-
builder.dnsResolver(host -> {
137-
if (shouldErrorOnDns.get() && randomBoolean() && randomBoolean()) {
138-
throw new UnknownHostException(host);
139-
}
140-
return defaultDnsResolver.resolve(host);
141-
});
142-
143-
return builder;
144-
}
145-
};
124+
service = new S3Service(Mockito.mock(Environment.class), Settings.EMPTY, Mockito.mock(ResourceWatcherService.class));
146125
recordingMeterRegistry = new RecordingMeterRegistry();
147126
super.setUp();
148127
}
@@ -165,7 +144,8 @@ protected String bytesContentType() {
165144

166145
@Override
167146
protected Class<? extends Exception> unresponsiveExceptionType() {
168-
return SdkClientException.class;
147+
// TODO NOMERGE can we be more precise?
148+
return SdkException.class;
169149
}
170150

171151
@Override
@@ -351,6 +331,7 @@ public void testWriteBlobWithReadTimeouts() {
351331
* This test shows that the AWS SDKv1 defers the closing of the InputStream used to upload a blob after the HTTP request has been sent
352332
* to S3, swallowing any exception thrown at closing time.
353333
*/
334+
@AwaitsFix(bugUrl = "TODO NOMERGE")
354335
public void testWriteBlobWithExceptionThrownAtClosingTime() throws Exception {
355336
var maxRetries = randomInt(3);
356337
var blobLength = randomIntBetween(1, 4096 * 3);
@@ -398,6 +379,7 @@ public void close() throws IOException {
398379
assertArrayEquals(bytes, BytesReference.toBytes(uploadedBytes.get()));
399380
}
400381

382+
@AwaitsFix(bugUrl = "TODO NOMERGE")
401383
public void testWriteLargeBlob() throws Exception {
402384
final boolean useTimeout = rarely();
403385
final TimeValue readTimeout = useTimeout ? TimeValue.timeValueMillis(randomIntBetween(100, 500)) : null;
@@ -445,7 +427,7 @@ public void testWriteLargeBlob() throws Exception {
445427
assertThat(contentLength, anyOf(equalTo(lastPartSize), equalTo(bufferSize.getBytes())));
446428

447429
if (countDownUploads.decrementAndGet() % 2 == 0) {
448-
exchange.getResponseHeaders().add("ETag", Base16.encodeAsString(md5.getMd5Digest()));
430+
exchange.getResponseHeaders().add("ETag", md5.getBase16Md5Digest());
449431
exchange.sendResponseHeaders(HttpStatus.SC_OK, -1);
450432
exchange.close();
451433
return;
@@ -499,6 +481,7 @@ public void testWriteLargeBlob() throws Exception {
499481
assertThat(countDownComplete.isCountedDown(), is(true));
500482
}
501483

484+
@AwaitsFix(bugUrl = "TODO NOMERGE")
502485
public void testWriteLargeBlobStreaming() throws Exception {
503486
final boolean useTimeout = rarely();
504487
final TimeValue readTimeout = useTimeout ? TimeValue.timeValueMillis(randomIntBetween(100, 500)) : null;
@@ -546,7 +529,7 @@ public void testWriteLargeBlobStreaming() throws Exception {
546529

547530
if (counterUploads.incrementAndGet() % 2 == 0) {
548531
bytesReceived.addAndGet(bytes.length());
549-
exchange.getResponseHeaders().add("ETag", Base16.encodeAsString(md5.getMd5Digest()));
532+
exchange.getResponseHeaders().add("ETag", md5.getBase16Md5Digest());
550533
exchange.sendResponseHeaders(HttpStatus.SC_OK, -1);
551534
exchange.close();
552535
return;
@@ -733,7 +716,7 @@ public void testReadWithIndicesPurposeRetriesForever() throws IOException {
733716

734717
final byte[] bytes = randomBlobContent(512);
735718

736-
shouldErrorOnDns.set(true);
719+
// shouldErrorOnDns.set(true); // TODO NOMERGE do we need to cover this too?
737720
final AtomicInteger failures = new AtomicInteger();
738721
@SuppressForbidden(reason = "use a http server")
739722
class FlakyReadHandler implements HttpHandler {
@@ -899,7 +882,7 @@ public void testSnapshotDeletesAbortRetriesWhenThreadIsInterrupted() {
899882
blobsToDelete.iterator()
900883
)
901884
);
902-
assertThat(exception.getCause(), instanceOf(AbortedException.class));
885+
assertThat(exception.getCause(), instanceOf(SdkException.class /* TODO NOMERGE can we be more precise? */));
903886
assertThat(handler.numberOfDeleteAttempts.get(), equalTo(interruptBeforeAttempt + 1));
904887
assertThat(handler.numberOfSuccessfulDeletes.get(), equalTo(0));
905888
} finally {
@@ -931,7 +914,7 @@ public void testNonSnapshotDeletesAreNotRetried() {
931914
);
932915
assertEquals(
933916
ThrottlingDeleteHandler.THROTTLING_ERROR_CODE,
934-
asInstanceOf(AmazonS3Exception.class, exception.getCause()).getErrorCode()
917+
asInstanceOf(S3Exception.class /* TODO NOMERGE can we be more precise? */, exception.getCause()).awsErrorDetails().errorCode()
935918
);
936919
assertThat(handler.numberOfDeleteAttempts.get(), equalTo(expectedNumberOfBatches(numBlobsToDelete)));
937920
assertThat(handler.numberOfSuccessfulDeletes.get(), equalTo(0));
@@ -961,7 +944,7 @@ private int expectedNumberOfBatches(int blobsToDelete) {
961944
}
962945

963946
@SuppressForbidden(reason = "use a http server")
964-
private class ThrottlingDeleteHandler extends S3HttpHandler {
947+
private static class ThrottlingDeleteHandler extends S3HttpHandler {
965948

966949
private static final String THROTTLING_ERROR_CODE = "SlowDown";
967950

@@ -986,7 +969,7 @@ private class ThrottlingDeleteHandler extends S3HttpHandler {
986969

987970
@Override
988971
public void handle(HttpExchange exchange) throws IOException {
989-
if (exchange.getRequestMethod().equals("POST") && exchange.getRequestURI().toString().startsWith("/bucket/?delete")) {
972+
if (isMultiDeleteRequest(exchange)) {
990973
onAttemptCallback.accept(numberOfDeleteAttempts.get());
991974
numberOfDeleteAttempts.incrementAndGet();
992975
if (throttleTimesBeforeSuccess.getAndDecrement() > 0) {
@@ -1048,9 +1031,12 @@ interface FailingHandlerFactory {
10481031
OptionalBytesReference.class,
10491032
l -> blobContainer.getRegister(randomRetryingPurpose(), "test_register_internal_retries", l)
10501033
);
1051-
assertThat(exceptionWithInternalRetries, instanceOf(AmazonS3Exception.class));
1034+
assertEquals(Integer.valueOf(maxRetries + 1), asInstanceOf(S3Exception.class, exceptionWithInternalRetries).numAttempts());
10521035
assertEquals((maxRetries + 1) * (maxRetries + 1), requestCounter.get());
1053-
assertEquals(maxRetries, exceptionWithInternalRetries.getSuppressed().length);
1036+
assertEquals(
1037+
maxRetries * 2 /* each failure yields a suppressed S3Exception and a suppressed SdkClientException */,
1038+
exceptionWithInternalRetries.getSuppressed().length
1039+
);
10541040
}
10551041

10561042
{
@@ -1059,7 +1045,7 @@ interface FailingHandlerFactory {
10591045
OptionalBytesReference.class,
10601046
l -> blobContainer.getRegister(randomRetryingPurpose(), "test_register_no_internal_retries", l)
10611047
);
1062-
assertThat(exceptionWithoutInternalRetries, instanceOf(AmazonS3Exception.class));
1048+
assertEquals(Integer.valueOf(1), asInstanceOf(S3Exception.class, exceptionWithoutInternalRetries).numAttempts());
10631049
assertEquals(maxRetries + 1, requestCounter.get());
10641050
assertEquals(maxRetries, exceptionWithoutInternalRetries.getSuppressed().length);
10651051
}
@@ -1070,7 +1056,7 @@ interface FailingHandlerFactory {
10701056
OptionalBytesReference.class,
10711057
l -> blobContainer.getRegister(OperationPurpose.REPOSITORY_ANALYSIS, "test_register_no_internal_retries", l)
10721058
);
1073-
assertThat(repoAnalysisException, instanceOf(AmazonS3Exception.class));
1059+
assertEquals(Integer.valueOf(1), asInstanceOf(S3Exception.class, repoAnalysisException).numAttempts());
10741060
assertEquals(1, requestCounter.get());
10751061
assertEquals(0, repoAnalysisException.getSuppressed().length);
10761062
}
@@ -1085,12 +1071,13 @@ interface FailingHandlerFactory {
10851071
}
10861072
}
10871073

1074+
@AwaitsFix(bugUrl = "TODO NOMERGE")
10881075
public void testSuppressedDeletionErrorsAreCapped() {
10891076
final TimeValue readTimeout = TimeValue.timeValueMillis(randomIntBetween(100, 500));
10901077
int maxBulkDeleteSize = randomIntBetween(1, 10);
10911078
final BlobContainer blobContainer = createBlobContainer(1, readTimeout, true, null, maxBulkDeleteSize);
10921079
httpServer.createContext("/", exchange -> {
1093-
if (exchange.getRequestMethod().equals("POST") && exchange.getRequestURI().toString().startsWith("/bucket/?delete")) {
1080+
if (isMultiDeleteRequest(exchange)) {
10941081
exchange.sendResponseHeaders(
10951082
randomFrom(
10961083
HttpStatus.SC_INTERNAL_SERVER_ERROR,
@@ -1114,17 +1101,25 @@ public void testSuppressedDeletionErrorsAreCapped() {
11141101
"deletion should not succeed",
11151102
() -> blobContainer.deleteBlobsIgnoringIfNotExists(randomPurpose(), blobs.iterator())
11161103
);
1104+
logger.info("--> deletion exception", exception);
11171105
assertThat(exception.getCause().getSuppressed().length, lessThan(S3BlobStore.MAX_DELETE_EXCEPTIONS));
11181106
}
11191107

1108+
private static boolean isMultiDeleteRequest(HttpExchange exchange) {
1109+
// TODO NOMERGE use S3HttpHandler#isMultiObjectDeleteRequest
1110+
return exchange.getRequestMethod().equals("POST")
1111+
&& (exchange.getRequestURI().toString().startsWith("/bucket/?delete")
1112+
|| exchange.getRequestURI().toString().startsWith("/bucket?delete"));
1113+
}
1114+
11201115
public void testTrimmedLogAndCappedSuppressedErrorOnMultiObjectDeletionException() {
11211116
final TimeValue readTimeout = TimeValue.timeValueMillis(randomIntBetween(100, 500));
11221117
int maxBulkDeleteSize = randomIntBetween(10, 30);
11231118
final BlobContainer blobContainer = createBlobContainer(1, readTimeout, true, null, maxBulkDeleteSize);
11241119

11251120
final Pattern pattern = Pattern.compile("<Key>(.+?)</Key>");
11261121
httpServer.createContext("/", exchange -> {
1127-
if (exchange.getRequestMethod().equals("POST") && exchange.getRequestURI().toString().startsWith("/bucket/?delete")) {
1122+
if (isMultiDeleteRequest(exchange)) {
11281123
final String requestBody = Streams.copyToString(new InputStreamReader(exchange.getRequestBody(), StandardCharsets.UTF_8));
11291124
final var matcher = pattern.matcher(requestBody);
11301125
final StringBuilder deletes = new StringBuilder();
@@ -1171,6 +1166,21 @@ public void testTrimmedLogAndCappedSuppressedErrorOnMultiObjectDeletionException
11711166
}
11721167
}
11731168

1169+
public void testMd5DigestCalculatingInputStream() throws IOException {
1170+
// from Wikipedia
1171+
doMD5DigestCalculatingInputStreamTest("", "d41d8cd98f00b204e9800998ecf8427e");
1172+
doMD5DigestCalculatingInputStreamTest("The quick brown fox jumps over the lazy dog", "9e107d9d372bb6826bd81d3542a419d6");
1173+
doMD5DigestCalculatingInputStreamTest("The quick brown fox jumps over the lazy dog.", "e4d909c290d0fb1ca068ffaddf22cbd0");
1174+
}
1175+
1176+
private static void doMD5DigestCalculatingInputStreamTest(String input, String expectedDigestString) throws IOException {
1177+
final var bytes = input.getBytes(StandardCharsets.UTF_8);
1178+
try (var s = new ByteArrayInputStream(bytes); var m = new MD5DigestCalculatingInputStream(s)) {
1179+
assertArrayEquals(bytes, m.readAllBytes());
1180+
assertEquals(expectedDigestString, m.getBase16Md5Digest());
1181+
}
1182+
}
1183+
11741184
@Override
11751185
protected Matcher<Integer> getMaxRetriesMatcher(int maxRetries) {
11761186
// some attempts make meaningful progress and do not count towards the max retry limit
@@ -1304,21 +1314,61 @@ public String toString() {
13041314
public void close() throws IOException {
13051315
super.close();
13061316
if (in instanceof final S3RetryingInputStream s3Stream) {
1307-
assertTrue(
1308-
"Stream "
1309-
+ toString()
1310-
+ " should have reached EOF or should have been aborted but got [eof="
1311-
+ s3Stream.isEof()
1312-
+ ", aborted="
1313-
+ s3Stream.isAborted()
1314-
+ ']',
1315-
s3Stream.isEof() || s3Stream.isAborted()
1316-
);
1317+
// assertTrue(
1318+
// "Stream "
1319+
// + toString()
1320+
// + " should have reached EOF or should have been aborted but got [eof="
1321+
// + s3Stream.isEof()
1322+
// + ", aborted="
1323+
// + s3Stream.isAborted()
1324+
// + ']',
1325+
// s3Stream.isEof() || s3Stream.isAborted()
1326+
// );
13171327
} else {
13181328
assertThat(in, instanceOf(ByteArrayInputStream.class));
13191329
assertThat(((ByteArrayInputStream) in).available(), equalTo(0));
13201330
}
13211331
}
13221332
}
1333+
1334+
private static final Logger logger = LogManager.getLogger(S3BlobContainerRetriesTests.class);
1335+
1336+
private static class MD5DigestCalculatingInputStream extends InputStream {
1337+
1338+
private final MessageDigest messageDigest = MessageDigests.md5();
1339+
private final InputStream delegate;
1340+
1341+
private MD5DigestCalculatingInputStream(InputStream delegate) {
1342+
this.delegate = delegate;
1343+
}
1344+
1345+
@Override
1346+
public int read() throws IOException {
1347+
final var b = delegate.read();
1348+
if (b >= 0) {
1349+
messageDigest.update((byte) b);
1350+
}
1351+
return b;
1352+
}
1353+
1354+
@Override
1355+
public int read(byte[] b, int off, int len) throws IOException {
1356+
final var readLen = delegate.read(b, off, len);
1357+
if (readLen > 0) {
1358+
messageDigest.update(b, off, readLen);
1359+
}
1360+
return readLen;
1361+
}
1362+
1363+
public String getBase16Md5Digest() {
1364+
final var digestBytes = messageDigest.digest();
1365+
final var stringChars = new char[digestBytes.length * 2];
1366+
for (int i = 0; i < digestBytes.length; i++) {
1367+
final var digestByte = digestBytes[i];
1368+
stringChars[2 * i] = Character.forDigit((digestByte >> 4) & 0xF, 16);
1369+
stringChars[2 * i + 1] = Character.forDigit(digestByte & 0xF, 16);
1370+
}
1371+
return new String(stringChars);
1372+
}
1373+
}
13231374
}
1324-
// TODO NOMERGE bring these tests back

‎test/fixtures/s3-fixture/src/main/java/fixture/s3/S3HttpHandler.java‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -27,7 +27,6 @@
2727
import org.elasticsearch.logging.Logger;
2828
import org.elasticsearch.rest.RestStatus;
2929
import org.elasticsearch.rest.RestUtils;
30-
import org.elasticsearch.test.ESTestCase;
3130
import org.elasticsearch.test.fixture.HttpHeaderParser;
3231

3332
import java.io.IOException;

0 commit comments

Comments
 (0)