Skip to content

Refactor gcp-fixture multipart parser #125828

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 21 commits into from
Apr 15, 2025
Merged

Refactor gcp-fixture multipart parser #125828

Show file tree
Hide file tree
Changes from 6 commits
Commits
Show all changes
21 commits
Select commit Hold shift + click to select a range
2fb738d
refactor gcp-fixture multipart parser
mhl-b Mar 28, 2025
28d4481
remove ring-buffer
mhl-b Mar 28, 2025
4ef1236
naming
mhl-b Mar 28, 2025
5105d4c
Merge remote-tracking branch 'upstream/main' into gcp-fixture-multipa…
mhl-b Mar 28, 2025
0dfddd6
add tests
mhl-b Mar 30, 2025
1008194
Merge branch 'main' into gcp-fixture-multipart-refactor
mhl-b Mar 30, 2025
771e069
Merge remote-tracking branch 'upstream/main' into gcp-fixture-multipa…
mhl-b Apr 3, 2025
4eb8b23
merge main
mhl-b Apr 10, 2025
3381864
generic multipart parser
mhl-b Apr 10, 2025
f8eb315
Merge branch 'gcp-fixture-multipart-refactor' of github.com:mhl-b/ela…
mhl-b Apr 10, 2025
f3210e4
[CI] Auto commit changes from spotless
Apr 10, 2025
9714b32
exceptions
mhl-b Apr 10, 2025
019a04f
Merge branch 'gcp-fixture-multipart-refactor' of github.com:mhl-b/ela…
mhl-b Apr 10, 2025
b4869f5
test zero-sized blob
mhl-b Apr 11, 2025
47912f5
sanitize content
mhl-b Apr 11, 2025
8de1110
Merge branch 'main' into gcp-fixture-multipart-refactor
mhl-b Apr 11, 2025
1505b48
Merge remote-tracking branch 'upstream/main' into gcp-fixture-multipa…
mhl-b Apr 14, 2025
b7ef093
Merge branch 'gcp-fixture-multipart-refactor' of github.com:mhl-b/ela…
mhl-b Apr 14, 2025
905620d
random multipart boundary
mhl-b Apr 14, 2025
efda0e7
cleanup
mhl-b Apr 14, 2025
797d27d
Merge branch 'main' into gcp-fixture-multipart-refactor
mhl-b Apr 15, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 10 additions & 14 deletions .../java/org/elasticsearch/repositories/gcs/GoogleCloudStorageBlobContainerRetriesTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@

import fixture.gcs.FakeOAuth2HttpHandler;
import fixture.gcs.GoogleCloudStorageHttpHandler;
import fixture.gcs.MultipartUpload;

import com.google.api.client.http.HttpExecuteInterceptor;
import com.google.api.client.http.HttpRequestInitializer;
Expand Down Expand Up @@ -42,7 +43,6 @@
import org.elasticsearch.core.Nullable;
import org.elasticsearch.core.SuppressForbidden;
import org.elasticsearch.core.TimeValue;
import org.elasticsearch.core.Tuple;
import org.elasticsearch.http.ResponseInjectingHttpHandler;
import org.elasticsearch.repositories.blobstore.AbstractBlobContainerRetriesTestCase;
import org.elasticsearch.repositories.blobstore.ESMockAPIBasedRepositoryIntegTestCase;
Expand All @@ -60,16 +60,13 @@
import java.util.Iterator;
import java.util.Locale;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Queue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;

import static fixture.gcs.GoogleCloudStorageHttpHandler.parseMultipartRequestBody;
import static fixture.gcs.TestUtils.createServiceAccount;
import static java.nio.charset.StandardCharsets.UTF_8;
import static org.elasticsearch.repositories.blobstore.BlobStoreTestUtil.randomPurpose;
Expand All @@ -79,7 +76,6 @@
import static org.elasticsearch.repositories.gcs.GoogleCloudStorageClientSettings.ENDPOINT_SETTING;
import static org.elasticsearch.repositories.gcs.GoogleCloudStorageClientSettings.READ_TIMEOUT_SETTING;
import static org.elasticsearch.repositories.gcs.GoogleCloudStorageClientSettings.TOKEN_URI_SETTING;
import static org.elasticsearch.test.hamcrest.OptionalMatchers.isPresent;
import static org.hamcrest.Matchers.anyOf;
import static org.hamcrest.Matchers.containsString;
import static org.hamcrest.Matchers.equalTo;
Expand Down Expand Up @@ -261,17 +257,17 @@ public void testWriteBlobWithRetries() throws Exception {
final CountDown countDown = new CountDown(maxRetries);

final BlobContainer blobContainer = createBlobContainer(maxRetries, null, null, null, null);
final byte[] bytes = randomBlobContent();
final byte[] blobContent = randomBlobContent();
httpServer.createContext("/upload/storage/v1/b/bucket/o", safeHandler(exchange -> {
assertThat(exchange.getRequestURI().getQuery(), containsString("uploadType=multipart"));
if (countDown.countDown()) {
Optional<Tuple<String, BytesReference>> content = parseMultipartRequestBody(exchange.getRequestBody());
assertThat(content, isPresent());
assertThat(content.get().v1(), equalTo(blobContainer.path().buildAsString() + "write_blob_max_retries"));
if (Objects.deepEquals(bytes, BytesReference.toBytes(content.get().v2()))) {
MultipartUpload multipartUpload = MultipartUpload.parseBody(exchange, exchange.getRequestBody());
assertTrue(multipartUpload.content().length() > 0);
Copy link
Contributor

@DaveCTurner DaveCTurner Mar 31, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It's legitimate to upload an empty blob comprising a single empty part isn't it? Certainly is in S3. We should be covering this case in these tests.

Copy link
Contributor Author

@mhl-b mhl-b Mar 31, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This test explicitly set content length to be at least 1 byte long. Empty content is not expected.

    protected static byte[] randomBlobContent() {
        return randomBlobContent(1);
    }

Copy link
Contributor Author

@mhl-b mhl-b Mar 31, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It's legitimate to send 0 byte blob. It will produce two body parts, json metadata and empty part with headers.

Copy link
Contributor

@nicktindall nicktindall Apr 11, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think we could do away with that assertion (it might trip in the event someone changes randomBlobContent() down the track). It's not important for the test. I think the previous assertion (assertThat(content, isPresent());) was only there because the old parseMultipartRequestBody returned Optional.empty() when the parse failed.

I don't have strong feelings about it but given we check that the content equals the uploaded bytes further down it seems redundant?

Copy link
Contributor Author

@mhl-b mhl-b Apr 11, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Changed test with min-size 0 and removed content length assertion b4869f5. Manually tested with 0 sized blob, it works.

assertEquals(multipartUpload.name(), blobContainer.path().buildAsString() + "write_blob_max_retries");
if (multipartUpload.content().equals(new BytesArray(blobContent))) {
byte[] response = Strings.format("""
{"bucket":"bucket","name":"%s"}
""", content.get().v1()).getBytes(UTF_8);
""", multipartUpload.name()).getBytes(UTF_8);
exchange.getResponseHeaders().add("Content-Type", "application/json");
exchange.sendResponseHeaders(RestStatus.OK.getStatus(), response.length);
exchange.getResponseBody().write(response);
Expand All @@ -284,7 +280,7 @@ public void testWriteBlobWithRetries() throws Exception {
if (randomBoolean()) {
org.elasticsearch.core.Streams.readFully(
exchange.getRequestBody(),
new byte[randomIntBetween(1, Math.max(1, bytes.length - 1))]
new byte[randomIntBetween(1, Math.max(1, blobContent.length - 1))]
);
} else {
Streams.readFully(exchange.getRequestBody());
Expand All @@ -293,8 +289,8 @@ public void testWriteBlobWithRetries() throws Exception {
}
}));

try (InputStream stream = new InputStreamIndexInput(new ByteArrayIndexInput("desc", bytes), bytes.length)) {
blobContainer.writeBlob(randomPurpose(), "write_blob_max_retries", stream, bytes.length, false);
try (InputStream stream = new InputStreamIndexInput(new ByteArrayIndexInput("desc", blobContent), blobContent.length)) {
blobContainer.writeBlob(randomPurpose(), "write_blob_max_retries", stream, blobContent.length, false);
}
assertThat(countDown.isCountedDown(), is(true));
}
Expand Down
113 changes: 8 additions & 105 deletions test/fixtures/gcs-fixture/src/main/java/fixture/gcs/GoogleCloudStorageHttpHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,6 @@
import org.elasticsearch.common.io.Streams;
import org.elasticsearch.common.regex.Regex;
import org.elasticsearch.core.SuppressForbidden;
import org.elasticsearch.core.Tuple;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.rest.RestUtils;
import org.elasticsearch.test.fixture.HttpHeaderParser;
Expand All @@ -27,26 +26,17 @@
import org.elasticsearch.xcontent.XContentFactory;
import org.elasticsearch.xcontent.XContentType;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.net.URLDecoder;
import java.util.HashMap;
import java.util.Locale;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
import java.util.zip.GZIPInputStream;

import static fixture.gcs.MockGcsBlobStore.failAndThrow;
import static java.nio.charset.StandardCharsets.UTF_8;
import static java.util.stream.Collectors.joining;
import static org.elasticsearch.core.Strings.format;

/**
* Minimal HTTP handler that acts as a Google Cloud Storage compliant server
Expand Down Expand Up @@ -186,26 +176,14 @@ public void handle(final HttpExchange exchange) throws IOException {
exchange.getResponseBody().write(response);

} else if (Regex.simpleMatch("POST /upload/storage/v1/b/" + bucket + "/*uploadType=multipart*", request)) {
// Multipart upload
Optional<Tuple<String, BytesReference>> content = parseMultipartRequestBody(requestBody.streamInput());
if (content.isPresent()) {
final Long ifGenerationMatch = parseOptionalLongParameter(exchange, IF_GENERATION_MATCH);
final MockGcsBlobStore.BlobVersion newBlobVersion = mockGcsBlobStore.updateBlob(
content.get().v1(),
ifGenerationMatch,
content.get().v2()
);
writeBlobVersionAsJson(exchange, newBlobVersion);
} else {
throw new AssertionError(
Copy link
Contributor

@DaveCTurner DaveCTurner Mar 28, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I would like to keep the behaviour where an invalid body causes a test failure rather than just an IllegalStateException

Copy link
Contributor Author

@mhl-b mhl-b Mar 30, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Throw AssertionError errors from the MultipartUpload.parseBody now.

Copy link
Contributor

@DaveCTurner DaveCTurner Mar 31, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I was sort of hoping to see tests that ensured that this kind of malformed body would throw an IAE from the parser. Throwing AssertionError from the parser itself basically means you can only test the happy path (or else catch AssertionError in tests which is best avoided if possible).

"Could not read multi-part request to ["
+ request
+ "] with headers ["
+ new HashMap<>(exchange.getRequestHeaders())
+ "]"
);
}

final var multipartUpload = MultipartUpload.parseBody(exchange, requestBody.streamInput());
final Long ifGenerationMatch = parseOptionalLongParameter(exchange, IF_GENERATION_MATCH);
final MockGcsBlobStore.BlobVersion newBlobVersion = mockGcsBlobStore.updateBlob(
multipartUpload.name(),
ifGenerationMatch,
multipartUpload.content()
);
writeBlobVersionAsJson(exchange, newBlobVersion);
} else if (Regex.simpleMatch("POST /upload/storage/v1/b/" + bucket + "/*uploadType=resumable*", request)) {
// Resumable upload initialization https://cloud.google.com/storage/docs/json_api/v1/how-tos/resumable-upload
final Map<String, String> params = new HashMap<>();
Expand Down Expand Up @@ -331,81 +309,6 @@ private static String httpServerUrl(final HttpExchange exchange) {
return "http://" + exchange.getRequestHeaders().get("HOST").get(0);
}

private static final Pattern NAME_PATTERN = Pattern.compile("\"name\":\"([^\"]*)\"");

public static Optional<Tuple<String, BytesReference>> parseMultipartRequestBody(final InputStream requestBody) throws IOException {
Tuple<String, BytesReference> content = null;
final BytesReference fullRequestBody;
try (InputStream in = new GZIPInputStream(requestBody)) {
fullRequestBody = Streams.readFully(in);
}
String name = null;
boolean skippedEmptyLine = false;
int startPos = 0;
int endPos = 0;
while (startPos < fullRequestBody.length()) {
do {
endPos = fullRequestBody.indexOf((byte) '\r', endPos + 1);
} while (endPos >= 0 && fullRequestBody.get(endPos + 1) != '\n');
boolean markAndContinue = false;
final String bucketPrefix = "{\"bucket\":";
if (startPos > 0) {
startPos += 2;
}
if (name == null || skippedEmptyLine == false) {
if ((skippedEmptyLine == false && endPos == startPos)
|| (fullRequestBody.get(startPos) == '-' && fullRequestBody.get(startPos + 1) == '-')) {
markAndContinue = true;
} else {
final String start = fullRequestBody.slice(startPos, Math.min(endPos - startPos, bucketPrefix.length())).utf8ToString();
if (start.toLowerCase(Locale.ROOT).startsWith("content")) {
markAndContinue = true;
} else if (start.startsWith(bucketPrefix)) {
markAndContinue = true;
final String line = fullRequestBody.slice(
startPos + bucketPrefix.length(),
endPos - startPos - bucketPrefix.length()
).utf8ToString();
Matcher matcher = NAME_PATTERN.matcher(line);
if (matcher.find()) {
name = matcher.group(1);
}
}
}
skippedEmptyLine = markAndContinue && endPos == startPos;
startPos = endPos;
} else {
while (isEndOfPart(fullRequestBody, endPos) == false) {
endPos = fullRequestBody.indexOf((byte) '\r', endPos + 1);
}
content = Tuple.tuple(name, fullRequestBody.slice(startPos, endPos - startPos));
break;
}
}
if (content == null) {
final InputStream stream = fullRequestBody.streamInput();
logger.warn(
() -> format(
"Failed to find multi-part upload in [%s]",
new BufferedReader(new InputStreamReader(stream)).lines().collect(joining("\n"))
)
);
}
return Optional.ofNullable(content);
}

private static final byte[] END_OF_PARTS_MARKER = "\r\n--__END_OF_PART__".getBytes(UTF_8);

private static boolean isEndOfPart(BytesReference fullRequestBody, int endPos) {
for (int i = 0; i < END_OF_PARTS_MARKER.length; i++) {
final byte b = END_OF_PARTS_MARKER[i];
if (fullRequestBody.get(endPos + i) != b) {
return false;
}
}
return true;
}

private static String requireHeader(HttpExchange exchange, String headerName) {
final String headerValue = exchange.getRequestHeaders().getFirst(headerName);
if (headerValue != null) {
Expand Down
Loading