From 4c2f44e28a1ff19ffb2a02e3cefc062a1dd98fdc Mon Sep 17 00:00:00 2001 From: BenWhitehead Date: Thu, 13 Jul 2023 16:06:32 -0400 Subject: [PATCH] feat: update Storage.createFrom(BlobInfo, Path) to have 150% higher throughput (#2059) When uploading a file where we are able to rewind to an arbitrary offset, we can be more optimistic in the way we send requests to GCS. Add new code middleware to allow PUTing an entire file to GCS in a single request, and using query resumable session to recover from the specific offset in the case of retryable error. ### Benchmark Results #### Methodology Generate a random file on disk of size `128KiB..2GiB` from `/dev/urandom`, then upload the generated file using `Storage.createFrom(BlobInfo, Path)`. Perform each 4096 times. Run on a c2-standard-60 instance is us-central1 against a regional bucket located in us-central1. #### Results The following summary of throughput in MiB/s as observed between the existing implementation, and the new implementation proposed in this PR. ``` count mean std min 50% 75% 90% 99% max runId ApiName createFrom - existing JSON 4096.0 66.754 10.988 3.249 67.317 73.476 78.961 91.197 107.247 createFrom - new JSON 4096.0 158.769 67.105 4.600 170.680 218.618 240.992 266.297 305.205 ``` #### Comparison When comparing the new implementation to the existing implementation we get the following improvement to throughput (higher is better): ``` stat pct mean 137.841 50% 153.547 90% 205.204 99% 192.003 ``` --- .../google/cloud/storage/ByteRangeSpec.java | 17 + .../cloud/storage/ByteSizeConstants.java | 6 + .../cloud/storage/HttpClientContext.java | 85 ++ .../cloud/storage/HttpContentRange.java | 249 ++++++ .../cloud/storage/HttpStorageOptions.java | 27 + .../cloud/storage/JsonResumableSession.java | 83 ++ .../JsonResumableSessionFailureScenario.java | 233 +++++ .../storage/JsonResumableSessionPutTask.java | 218 +++++ .../JsonResumableSessionQueryTask.java | 136 +++ .../cloud/storage/JsonResumableWrite.java | 88 ++ .../storage/ResumableOperationResult.java | 90 ++ .../cloud/storage/ResumableSession.java | 39 + .../cloud/storage/RewindableHttpContent.java | 106 +++ .../com/google/cloud/storage/StorageImpl.java | 37 +- .../google/cloud/storage/FakeHttpServer.java | 144 ++++ .../ITJsonResumableSessionPutTaskTest.java | 814 ++++++++++++++++++ .../ITJsonResumableSessionQueryTaskTest.java | 235 +++++ .../storage/ITJsonResumableSessionTest.java | 134 +++ ...onResumableSessionFailureScenarioTest.java | 155 ++++ .../RewindableHttpContentPropertyTest.java | 201 +++++ .../cloud/storage/StorageImplMockitoTest.java | 11 - .../storage/conformance/retry/Functions.java | 4 + .../retry/ITRetryConformanceTest.java | 13 +- .../conformance/retry/RpcMethodMappings.java | 26 +- .../it/ITObjectChecksumSupportTest.java | 53 +- 25 files changed, 3184 insertions(+), 20 deletions(-) create mode 100644 google-cloud-storage/src/main/java/com/google/cloud/storage/HttpClientContext.java create mode 100644 google-cloud-storage/src/main/java/com/google/cloud/storage/HttpContentRange.java create mode 100644 google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableSession.java create mode 100644 google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableSessionFailureScenario.java create mode 100644 google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableSessionPutTask.java create mode 100644 google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableSessionQueryTask.java create mode 100644 google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableWrite.java create mode 100644 google-cloud-storage/src/main/java/com/google/cloud/storage/ResumableOperationResult.java create mode 100644 google-cloud-storage/src/main/java/com/google/cloud/storage/ResumableSession.java create mode 100644 google-cloud-storage/src/main/java/com/google/cloud/storage/RewindableHttpContent.java create mode 100644 google-cloud-storage/src/test/java/com/google/cloud/storage/FakeHttpServer.java create mode 100644 google-cloud-storage/src/test/java/com/google/cloud/storage/ITJsonResumableSessionPutTaskTest.java create mode 100644 google-cloud-storage/src/test/java/com/google/cloud/storage/ITJsonResumableSessionQueryTaskTest.java create mode 100644 google-cloud-storage/src/test/java/com/google/cloud/storage/ITJsonResumableSessionTest.java create mode 100644 google-cloud-storage/src/test/java/com/google/cloud/storage/JsonResumableSessionFailureScenarioTest.java create mode 100644 google-cloud-storage/src/test/java/com/google/cloud/storage/RewindableHttpContentPropertyTest.java diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/ByteRangeSpec.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/ByteRangeSpec.java index ca4cd64d71..524bfe3c2e 100644 --- a/google-cloud-storage/src/main/java/com/google/cloud/storage/ByteRangeSpec.java +++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/ByteRangeSpec.java @@ -16,6 +16,9 @@ package com.google.cloud.storage; +import static com.google.api.client.util.Preconditions.checkNotNull; +import static com.google.common.base.Preconditions.checkArgument; + import com.google.api.core.InternalApi; import com.google.common.base.MoreObjects; import com.google.common.base.MoreObjects.ToStringHelper; @@ -124,6 +127,20 @@ static ByteRangeSpec explicitClosed( return create(beginOffset, endOffsetInclusive, LeftClosedRightClosedByteRangeSpec::new); } + static ByteRangeSpec parse(String string) { + checkNotNull(string, "Range header is null"); + checkArgument(string.startsWith("bytes="), "malformed Range header value: %s", string); + + int i = string.indexOf('-'); + String minS = string.substring(6, i); + String maxS = string.substring(i + 1); + + long min = Long.parseLong(minS); + long max = Long.parseLong(maxS); + + return explicitClosed(min, max); + } + private static ByteRangeSpec create( @Nullable Long beginOffset, @Nullable Long length, diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/ByteSizeConstants.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/ByteSizeConstants.java index cac56927fa..98b31a3ded 100644 --- a/google-cloud-storage/src/main/java/com/google/cloud/storage/ByteSizeConstants.java +++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/ByteSizeConstants.java @@ -23,10 +23,16 @@ final class ByteSizeConstants { static final int _256KiB = 256 * _1KiB; static final int _384KiB = 384 * _1KiB; static final int _512KiB = 512 * _1KiB; + static final int _768KiB = 768 * _1KiB; static final int _1MiB = 1024 * _1KiB; static final int _2MiB = 2 * _1MiB; static final int _16MiB = 16 * _1MiB; static final int _32MiB = 32 * _1MiB; + static final long _128KiBL = 131072L; + static final long _256KiBL = 262144L; + static final long _512KiBL = 524288L; + static final long _768KiBL = 786432L; + private ByteSizeConstants() {} } diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/HttpClientContext.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/HttpClientContext.java new file mode 100644 index 0000000000..35c93c02f5 --- /dev/null +++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/HttpClientContext.java @@ -0,0 +1,85 @@ +/* + * Copyright 2023 Google LLC + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.google.cloud.storage; + +import com.google.api.client.http.HttpHeaders; +import com.google.api.client.http.HttpRequestFactory; +import com.google.api.client.json.JsonObjectParser; +import com.google.api.client.util.ObjectParser; +import com.google.cloud.storage.spi.v1.StorageRpc; +import io.opencensus.trace.Span; +import io.opencensus.trace.Tracer; +import io.opencensus.trace.Tracing; +import java.util.List; +import org.checkerframework.checker.nullness.qual.NonNull; +import org.checkerframework.checker.nullness.qual.Nullable; + +final class HttpClientContext { + + private final HttpRequestFactory requestFactory; + private final ObjectParser objectParser; + private final Tracer tracer; + + private HttpClientContext( + HttpRequestFactory requestFactory, ObjectParser objectParser, Tracer tracer) { + this.requestFactory = requestFactory; + this.objectParser = objectParser; + this.tracer = tracer; + } + + @SuppressWarnings({"unchecked", "SameParameterValue"}) + static @Nullable String firstHeaderValue( + @NonNull HttpHeaders headers, @NonNull String headerName) { + Object v = headers.get(headerName); + // HttpHeaders doesn't type its get method, so we have to jump through hoops here + if (v instanceof List) { + List list = (List) v; + return list.get(0); + } else { + return null; + } + } + + public HttpRequestFactory getRequestFactory() { + return requestFactory; + } + + public ObjectParser getObjectParser() { + return objectParser; + } + + public Tracer getTracer() { + return tracer; + } + + public Span startSpan(String name) { + // record events is hardcoded to true in HttpStorageRpc, preserve it here + return tracer.spanBuilder(name).setRecordEvents(true).startSpan(); + } + + static HttpClientContext from(StorageRpc storageRpc) { + return new HttpClientContext( + storageRpc.getStorage().getRequestFactory(), + storageRpc.getStorage().getObjectParser(), + Tracing.getTracer()); + } + + public static HttpClientContext of( + HttpRequestFactory requestFactory, JsonObjectParser jsonObjectParser) { + return new HttpClientContext(requestFactory, jsonObjectParser, Tracing.getTracer()); + } +} diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/HttpContentRange.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/HttpContentRange.java new file mode 100644 index 0000000000..8961f52393 --- /dev/null +++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/HttpContentRange.java @@ -0,0 +1,249 @@ +/* + * Copyright 2023 Google LLC + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.google.cloud.storage; + +import static com.google.common.base.Preconditions.checkArgument; + +import com.google.common.base.MoreObjects; +import java.util.Objects; +import java.util.function.UnaryOperator; + +abstract class HttpContentRange { + + private final boolean finalizing; + + private HttpContentRange(boolean finalizing) { + this.finalizing = finalizing; + } + + public abstract String getHeaderValue(); + + public boolean isFinalizing() { + return finalizing; + } + + static Total of(ByteRangeSpec spec, long size) { + checkArgument(size >= 0, "size must be >= 0"); + checkArgument(size >= spec.endOffsetInclusive(), "size must be >= end"); + return new Total(spec, size); + } + + static Incomplete of(ByteRangeSpec spec) { + return new Incomplete(spec); + } + + static Size of(long size) { + checkArgument(size >= 0, "size must be >= 0"); + return new Size(size); + } + + static Query query() { + return Query.INSTANCE; + } + + static HttpContentRange parse(String string) { + if ("bytes */*".equals(string)) { + return HttpContentRange.query(); + } else if (string.startsWith("bytes */")) { + return HttpContentRange.of(Long.parseLong(string.substring(8))); + } else { + int idxDash = string.indexOf('-'); + int idxSlash = string.indexOf('/'); + + String beginS = string.substring(6, idxDash); + String endS = string.substring(idxDash + 1, idxSlash); + long begin = Long.parseLong(beginS); + long end = Long.parseLong(endS); + if (string.endsWith("/*")) { + return HttpContentRange.of(ByteRangeSpec.explicitClosed(begin, end)); + } else { + String sizeS = string.substring(idxSlash + 1); + long size = Long.parseLong(sizeS); + return HttpContentRange.of(ByteRangeSpec.explicitClosed(begin, end), size); + } + } + } + + static final class Incomplete extends HttpContentRange implements HasRange { + + private final ByteRangeSpec spec; + + private Incomplete(ByteRangeSpec spec) { + super(false); + this.spec = spec; + } + + @Override + public String getHeaderValue() { + return String.format("bytes %d-%d/*", spec.beginOffset(), spec.endOffsetInclusive()); + } + + @Override + public ByteRangeSpec range() { + return spec; + } + + @Override + public Incomplete map(UnaryOperator f) { + return new Incomplete(f.apply(spec)); + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (!(o instanceof Incomplete)) { + return false; + } + Incomplete that = (Incomplete) o; + return Objects.equals(spec, that.spec); + } + + @Override + public int hashCode() { + return Objects.hash(spec); + } + + @Override + public String toString() { + return MoreObjects.toStringHelper(this).add("spec", spec).toString(); + } + } + + static final class Total extends HttpContentRange implements HasRange, HasSize { + + private final ByteRangeSpec spec; + private final long size; + + private Total(ByteRangeSpec spec, long size) { + super(true); + this.spec = spec; + this.size = size; + } + + @Override + public String getHeaderValue() { + return String.format("bytes %d-%d/%d", spec.beginOffset(), spec.endOffsetInclusive(), size); + } + + @Override + public long getSize() { + return size; + } + + @Override + public ByteRangeSpec range() { + return spec; + } + + @Override + public Total map(UnaryOperator f) { + return new Total(f.apply(spec), size); + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (!(o instanceof Total)) { + return false; + } + Total total = (Total) o; + return size == total.size && Objects.equals(spec, total.spec); + } + + @Override + public int hashCode() { + return Objects.hash(spec, size); + } + + @Override + public String toString() { + return MoreObjects.toStringHelper(this).add("spec", spec).add("size", size).toString(); + } + } + + static final class Size extends HttpContentRange implements HasSize { + + private final long size; + + private Size(long size) { + super(true); + this.size = size; + } + + @Override + public String getHeaderValue() { + return String.format("bytes */%d", size); + } + + @Override + public long getSize() { + return size; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (!(o instanceof Size)) { + return false; + } + Size size1 = (Size) o; + return size == size1.size; + } + + @Override + public int hashCode() { + return Objects.hash(size); + } + + @Override + public String toString() { + return MoreObjects.toStringHelper(this).add("size", size).toString(); + } + } + + static final class Query extends HttpContentRange { + + private static final Query INSTANCE = new Query(); + + private Query() { + super(false); + } + + @Override + public String getHeaderValue() { + return "bytes */*"; + } + } + + interface HasRange { + + ByteRangeSpec range(); + + T map(UnaryOperator f); + } + + interface HasSize { + + long getSize(); + } +} diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/HttpStorageOptions.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/HttpStorageOptions.java index 3483dd9074..b525cf75b8 100644 --- a/google-cloud-storage/src/main/java/com/google/cloud/storage/HttpStorageOptions.java +++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/HttpStorageOptions.java @@ -29,6 +29,7 @@ import com.google.cloud.TransportOptions; import com.google.cloud.http.HttpTransportOptions; import com.google.cloud.spi.ServiceRpcFactory; +import com.google.cloud.storage.Retrying.RetryingDependencies; import com.google.cloud.storage.TransportCompatibility.Transport; import com.google.cloud.storage.spi.StorageRpcFactory; import com.google.cloud.storage.spi.v1.HttpStorageRpc; @@ -51,6 +52,7 @@ public class HttpStorageOptions extends StorageOptions { private static final String DEFAULT_HOST = "https://storage.googleapis.com"; private final HttpRetryAlgorithmManager retryAlgorithmManager; + private final transient RetryDependenciesAdapter retryDepsAdapter; private HttpStorageOptions(Builder builder, StorageDefaults serviceDefaults) { super(builder, serviceDefaults); @@ -58,6 +60,7 @@ private HttpStorageOptions(Builder builder, StorageDefaults serviceDefaults) { new HttpRetryAlgorithmManager( MoreObjects.firstNonNull( builder.storageRetryStrategy, defaults().getStorageRetryStrategy())); + retryDepsAdapter = new RetryDependenciesAdapter(); } @Override @@ -102,6 +105,11 @@ public static HttpStorageDefaults defaults() { return HttpStorageDefaults.INSTANCE; } + @InternalApi + RetryingDependencies asRetryDependencies() { + return retryDepsAdapter; + } + public static class Builder extends StorageOptions.Builder { private StorageRetryStrategy storageRetryStrategy; @@ -321,4 +329,23 @@ public ServiceRpc create(StorageOptions options) { } } } + + /** + * We don't yet want to make HttpStorageOptions itself implement {@link RetryingDependencies} but + * we do need use it in a couple places, for those we create this adapter. + */ + private final class RetryDependenciesAdapter implements RetryingDependencies { + + private RetryDependenciesAdapter() {} + + @Override + public RetrySettings getRetrySettings() { + return HttpStorageOptions.this.getRetrySettings(); + } + + @Override + public ApiClock getClock() { + return HttpStorageOptions.this.getClock(); + } + } } diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableSession.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableSession.java new file mode 100644 index 0000000000..f59355502b --- /dev/null +++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableSession.java @@ -0,0 +1,83 @@ +/* + * Copyright 2023 Google LLC + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.google.cloud.storage; + +import com.google.api.gax.retrying.ResultRetryAlgorithm; +import com.google.api.services.storage.model.StorageObject; +import com.google.cloud.storage.Conversions.Decoder; +import com.google.cloud.storage.Retrying.RetryingDependencies; +import com.google.cloud.storage.spi.v1.HttpStorageRpc; +import io.opencensus.trace.EndSpanOptions; +import java.util.concurrent.atomic.AtomicBoolean; +import org.checkerframework.checker.nullness.qual.Nullable; + +final class JsonResumableSession extends ResumableSession { + + static final String SPAN_NAME_WRITE = + String.format("Sent.%s.write", HttpStorageRpc.class.getName()); + static final EndSpanOptions END_SPAN_OPTIONS = + EndSpanOptions.builder().setSampleToLocalSpanStore(true).build(); + + private final HttpClientContext context; + private final RetryingDependencies deps; + private final ResultRetryAlgorithm alg; + private final JsonResumableWrite resumableWrite; + + JsonResumableSession( + HttpClientContext context, + RetryingDependencies deps, + ResultRetryAlgorithm alg, + JsonResumableWrite resumableWrite) { + this.context = context; + this.deps = deps; + this.alg = alg; + this.resumableWrite = resumableWrite; + } + + /** + * Not automatically retried. Usually called from within another retrying context. We don't yet + * have the concept of nested retry handling. + */ + @Override + ResumableOperationResult<@Nullable StorageObject> query() { + return new JsonResumableSessionQueryTask(context, resumableWrite.getUploadId()).call(); + } + + @Override + ResumableOperationResult<@Nullable StorageObject> put( + RewindableHttpContent content, HttpContentRange contentRange) { + JsonResumableSessionPutTask task = + new JsonResumableSessionPutTask( + context, resumableWrite.getUploadId(), content, contentRange); + AtomicBoolean dirty = new AtomicBoolean(false); + return Retrying.run( + deps, + alg, + () -> { + if (dirty.getAndSet(true)) { + ResumableOperationResult<@Nullable StorageObject> query = query(); + if (query.getObject() != null) { + return query; + } else { + task.rewindTo(query.getPersistedSize()); + } + } + return task.call(); + }, + Decoder.identity()); + } +} diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableSessionFailureScenario.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableSessionFailureScenario.java new file mode 100644 index 0000000000..2b6e8d569c --- /dev/null +++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableSessionFailureScenario.java @@ -0,0 +1,233 @@ +/* + * Copyright 2023 Google LLC + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.google.cloud.storage; + +import com.google.api.client.http.HttpHeaders; +import com.google.api.client.http.HttpResponse; +import com.google.api.client.http.HttpResponseException; +import com.google.cloud.BaseServiceException; +import com.google.cloud.storage.StorageException.IOExceptionCallable; +import com.google.common.io.CharStreams; +import java.io.IOException; +import java.io.InputStreamReader; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.function.Predicate; +import javax.annotation.ParametersAreNonnullByDefault; +import org.checkerframework.checker.nullness.qual.Nullable; + +@ParametersAreNonnullByDefault +enum JsonResumableSessionFailureScenario { + // TODO: send more bytes than are in the Content-Range header + SCENARIO_0(BaseServiceException.UNKNOWN_CODE, null, "Unknown Error"), + SCENARIO_0_1(BaseServiceException.UNKNOWN_CODE, null, "Response not application/json."), + SCENARIO_1( + BaseServiceException.UNKNOWN_CODE, + "invalid", + "Attempt to append to already finalized resumable session."), + SCENARIO_2( + BaseServiceException.UNKNOWN_CODE, + "invalid", + "Attempt to finalize resumable session with fewer bytes than the backend has received."), + SCENARIO_3( + BaseServiceException.UNKNOWN_CODE, + "dataLoss", + "Attempt to finalize resumable session with more bytes than the backend has received."), + SCENARIO_4(200, "ok", "Attempt to finalize an already finalized session with same object size"), + SCENARIO_4_1( + BaseServiceException.UNKNOWN_CODE, + "dataLoss", + "Finalized resumable session, but object size less than expected."), + SCENARIO_4_2( + BaseServiceException.UNKNOWN_CODE, + "dataLoss", + "Finalized resumable session, but object size greater than expected."), + SCENARIO_5( + BaseServiceException.UNKNOWN_CODE, + "dataLoss", + "Client side data loss detected. Attempt to append to a resumable session with an offset higher than the backend has"), + SCENARIO_7( + BaseServiceException.UNKNOWN_CODE, + "dataLoss", + "Client side data loss detected. Bytes acked is more than client sent."), + SCENARIO_9(503, "backendNotConnected", "Ack less than bytes sent"), + QUERY_SCENARIO_1(503, "", "Missing Range header in response"); + + private static final String PREFIX_I = "\t|< "; + private static final String PREFIX_O = "\t|> "; + private static final String PREFIX_X = "\t| "; + + private static final Predicate includedHeaders = + matches("Content-Length") + .or(matches("Content-Encoding")) + .or(matches("Content-Range")) + .or(matches("Content-Type")) + .or(matches("Range")) + .or(startsWith("X-Goog-Stored-")) + .or(matches("X-GUploader-UploadID")); + + private static final Predicate> includeHeader = + e -> includedHeaders.test(e.getKey()); + + private final int code; + @Nullable private final String reason; + private final String message; + + JsonResumableSessionFailureScenario(int code, @Nullable String reason, String message) { + this.code = code; + this.reason = reason; + this.message = message; + } + + StorageException toStorageException(String uploadId, HttpResponse resp) { + return toStorageException( + uploadId, resp, null, () -> CharStreams.toString(new InputStreamReader(resp.getContent()))); + } + + StorageException toStorageException( + String uploadId, @Nullable HttpResponse resp, @Nullable Throwable cause) { + if (resp != null) { + // an exception caused this, do not try to read the content from the response. + return toStorageException(uploadId, resp, cause, () -> null); + } else { + return new StorageException(code, message, reason, cause); + } + } + + StorageException toStorageException( + String uploadId, + HttpResponse resp, + @Nullable Throwable cause, + IOExceptionCallable<@Nullable String> contentCallable) { + return toStorageException(code, message, reason, uploadId, resp, cause, contentCallable); + } + + static StorageException toStorageException( + HttpResponse response, HttpResponseException cause, String uploadId) { + String statusMessage = cause.getStatusMessage(); + StorageException se = + JsonResumableSessionFailureScenario.toStorageException( + cause.getStatusCode(), + String.format( + "%d %s", cause.getStatusCode(), statusMessage == null ? "" : statusMessage), + "", + uploadId, + response, + cause, + () -> null); + return se; + } + + static StorageException toStorageException( + int overrideCode, + String message, + @Nullable String reason, + String uploadId, + HttpResponse resp, + @Nullable Throwable cause, + IOExceptionCallable<@Nullable String> contentCallable) { + Throwable suppress = null; + StringBuilder sb = new StringBuilder(); + sb.append(message); + // add request context + sb.append("\n").append(PREFIX_O).append("PUT ").append(uploadId); + recordHeaderTo(resp.getRequest().getHeaders(), PREFIX_O, sb); + + sb.append("\n").append(PREFIX_X); + // add response context + { + int code = resp.getStatusCode(); + sb.append("\n").append(PREFIX_I).append("HTTP/1.1 ").append(code); + if (resp.getStatusMessage() != null) { + sb.append(" ").append(resp.getStatusMessage()); + } + + recordHeaderTo(resp.getHeaders(), PREFIX_I, sb); + // try to include any body that we can handle + if (isOk(code) || code == 503 || code == 400) { + try { + String content = contentCallable.call(); + if (content != null) { + sb.append("\n").append(PREFIX_I); + if (content.contains("\n") || content.contains("\r\n")) { + sb.append("\n").append(PREFIX_I).append(content.replaceAll("\r?\n", "\n" + PREFIX_I)); + } else { + sb.append("\n").append(PREFIX_I).append(content); + } + } + } catch (IOException e) { + // com.google.api.client.http.HttpResponseException.Builder.Builder + // prints an exception which might occur while attempting to resolve the content + // this can lose the context about the request it was for, instead we register it + // as a suppressed exception + suppress = new StorageException(0, "Error reading response content for diagnostics.", e); + } + } + + sb.append("\n").append(PREFIX_X); + } + StorageException storageException = + new StorageException(overrideCode, sb.toString(), reason, cause); + if (suppress != null) { + storageException.addSuppressed(suppress); + } + return storageException; + } + + static boolean isOk(int code) { + return code == 200 || code == 201; + } + + static boolean isContinue(int code) { + return code == 308; + } + + // The header names from HttpHeaders are lower cased, define some utility methods to create + // predicates where we can specify values ignoring case + private static Predicate matches(String expected) { + String lower = expected.toLowerCase(Locale.US); + return lower::equals; + } + + private static Predicate startsWith(String prefix) { + String lower = prefix.toLowerCase(Locale.US); + return s -> s.startsWith(lower); + } + + private static void recordHeaderTo(HttpHeaders h, String prefix, StringBuilder sb) { + h.entrySet().stream() + .filter(includeHeader) + .forEach( + e -> { + String key = e.getKey(); + String value = headerValueToString(e.getValue()); + sb.append("\n").append(prefix).append(key).append(": ").append(value); + }); + } + + private static String headerValueToString(Object o) { + if (o instanceof List) { + List l = (List) o; + if (l.size() == 1) { + return l.get(0).toString(); + } + } + + return o.toString(); + } +} diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableSessionPutTask.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableSessionPutTask.java new file mode 100644 index 0000000000..73b7d14a46 --- /dev/null +++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableSessionPutTask.java @@ -0,0 +1,218 @@ +/* + * Copyright 2023 Google LLC + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.google.cloud.storage; + +import com.google.api.client.http.GenericUrl; +import com.google.api.client.http.HttpRequest; +import com.google.api.client.http.HttpResponse; +import com.google.api.client.http.HttpResponseException; +import com.google.api.services.storage.model.StorageObject; +import com.google.cloud.storage.StorageException.IOExceptionCallable; +import com.google.common.annotations.VisibleForTesting; +import io.opencensus.common.Scope; +import io.opencensus.trace.Span; +import io.opencensus.trace.Status; +import java.io.IOException; +import java.math.BigInteger; +import java.util.Locale; +import java.util.concurrent.Callable; +import org.checkerframework.checker.nullness.qual.Nullable; + +final class JsonResumableSessionPutTask + implements Callable> { + + private final HttpClientContext context; + private final String uploadId; + private final RewindableHttpContent content; + private final HttpContentRange originalContentRange; + + private HttpContentRange contentRange; + + @VisibleForTesting + JsonResumableSessionPutTask( + HttpClientContext httpClientContext, + String uploadId, + RewindableHttpContent content, + HttpContentRange originalContentRange) { + this.context = httpClientContext; + this.uploadId = uploadId; + this.content = content; + this.originalContentRange = originalContentRange; + this.contentRange = originalContentRange; + } + + public void rewindTo(long offset) { + content.rewindTo(offset); + if (contentRange instanceof HttpContentRange.HasRange) { + HttpContentRange.HasRange range = (HttpContentRange.HasRange) contentRange; + contentRange = range.map(s -> s.withNewBeginOffset(offset)); + } + } + + public ResumableOperationResult<@Nullable StorageObject> call() throws IOException { + Span span = context.startSpan(JsonResumableSession.SPAN_NAME_WRITE); + Scope scope = context.getTracer().withSpan(span); + + boolean success = false; + boolean finalizing = originalContentRange.isFinalizing(); + + HttpRequest req = + context + .getRequestFactory() + .buildPutRequest(new GenericUrl(uploadId), content) + .setParser(context.getObjectParser()); + req.setThrowExceptionOnExecuteError(false); + req.getHeaders().setContentRange(contentRange.getHeaderValue()); + + HttpResponse response = null; + try { + response = req.execute(); + + int code = response.getStatusCode(); + + if (!finalizing && JsonResumableSessionFailureScenario.isContinue(code)) { + long effectiveEnd = ((HttpContentRange.HasRange) contentRange).range().endOffset(); + @Nullable String range = response.getHeaders().getRange(); + ByteRangeSpec ackRange = ByteRangeSpec.parse(range); + if (ackRange.endOffset() == effectiveEnd) { + success = true; + return ResumableOperationResult.incremental(ackRange.endOffset()); + } else if (ackRange.endOffset() < effectiveEnd) { + StorageException se = + JsonResumableSessionFailureScenario.SCENARIO_9.toStorageException(uploadId, response); + span.setStatus(Status.UNKNOWN.withDescription(se.getMessage())); + throw se; + } else { + StorageException se = + JsonResumableSessionFailureScenario.SCENARIO_7.toStorageException(uploadId, response); + span.setStatus(Status.UNKNOWN.withDescription(se.getMessage())); + throw se; + } + } else if (finalizing && JsonResumableSessionFailureScenario.isOk(code)) { + @Nullable StorageObject storageObject; + @Nullable BigInteger actualSize; + + Long contentLength = response.getHeaders().getContentLength(); + String contentType = response.getHeaders().getContentType(); + String storedContentLength = + HttpClientContext.firstHeaderValue( + response.getHeaders(), "x-goog-stored-content-length"); + boolean isJson = contentType != null && contentType.startsWith("application/json"); + if (isJson) { + storageObject = response.parseAs(StorageObject.class); + actualSize = storageObject != null ? storageObject.getSize() : null; + } else if ((contentLength == null || contentLength == 0) && storedContentLength != null) { + // when a signed url is used, the finalize response is empty + response.ignore(); + actualSize = new BigInteger(storedContentLength, 10); + success = true; + storageObject = null; + } else { + response.ignore(); + StorageException se = + JsonResumableSessionFailureScenario.SCENARIO_0_1.toStorageException( + uploadId, response, null, () -> null); + span.setStatus(Status.UNKNOWN.withDescription(se.getMessage())); + throw se; + } + BigInteger expectedSize = + BigInteger.valueOf(((HttpContentRange.HasSize) contentRange).getSize()); + int compare = expectedSize.compareTo(actualSize); + if (compare == 0) { + success = true; + //noinspection DataFlowIssue compareTo result will filter out actualSize == null + return ResumableOperationResult.complete(storageObject, actualSize.longValue()); + } else if (compare < 0) { + StorageException se = + JsonResumableSessionFailureScenario.SCENARIO_4_1.toStorageException( + uploadId, response, null, toString(storageObject)); + span.setStatus(Status.UNKNOWN.withDescription(se.getMessage())); + throw se; + } else { + StorageException se = + JsonResumableSessionFailureScenario.SCENARIO_4_2.toStorageException( + uploadId, response, null, toString(storageObject)); + span.setStatus(Status.UNKNOWN.withDescription(se.getMessage())); + throw se; + } + } else if (!finalizing && JsonResumableSessionFailureScenario.isOk(code)) { + StorageException se = + JsonResumableSessionFailureScenario.SCENARIO_1.toStorageException(uploadId, response); + span.setStatus(Status.UNKNOWN.withDescription(se.getMessage())); + throw se; + } else if (finalizing && JsonResumableSessionFailureScenario.isContinue(code)) { + // in order to finalize the content range must have a size, cast down to read it + HttpContentRange.HasSize size = (HttpContentRange.HasSize) contentRange; + + ByteRangeSpec range = ByteRangeSpec.parse(response.getHeaders().getRange()); + if (range.endOffsetInclusive() < size.getSize()) { + StorageException se = + JsonResumableSessionFailureScenario.SCENARIO_3.toStorageException(uploadId, response); + span.setStatus(Status.UNKNOWN.withDescription(se.getMessage())); + throw se; + } else { + StorageException se = + JsonResumableSessionFailureScenario.SCENARIO_2.toStorageException(uploadId, response); + span.setStatus(Status.UNKNOWN.withDescription(se.getMessage())); + throw se; + } + } else { + HttpResponseException cause = new HttpResponseException(response); + String contentType = response.getHeaders().getContentType(); + // If the content-range header value has run ahead of the backend, it will respond with + // a 503 with plain text content + // Attempt to detect this very loosely as to minimize impact of modified error message + // This is accurate circa 2023-06 + if ((!JsonResumableSessionFailureScenario.isOk(code) + && !JsonResumableSessionFailureScenario.isContinue(code)) + && contentType != null + && contentType.startsWith("text/plain")) { + String errorMessage = cause.getContent().toLowerCase(Locale.US); + if (errorMessage.contains("content-range")) { + StorageException se = + JsonResumableSessionFailureScenario.SCENARIO_5.toStorageException( + uploadId, response, cause, cause::getContent); + span.setStatus(Status.UNKNOWN.withDescription(se.getMessage())); + throw se; + } + } + StorageException se = + JsonResumableSessionFailureScenario.toStorageException(response, cause, uploadId); + span.setStatus(Status.UNKNOWN.withDescription(se.getMessage())); + throw se; + } + } catch (StorageException e) { + span.setStatus(Status.UNKNOWN.withDescription(e.getMessage())); + throw e; + } catch (Exception e) { + StorageException se = + JsonResumableSessionFailureScenario.SCENARIO_0.toStorageException(uploadId, response, e); + span.setStatus(Status.UNKNOWN.withDescription(se.getMessage())); + throw se; + } finally { + if (success && !finalizing && response != null) { + response.ignore(); + } + scope.close(); + span.end(JsonResumableSession.END_SPAN_OPTIONS); + } + } + + static IOExceptionCallable<@Nullable String> toString(@Nullable Object o) { + return () -> o != null ? o.toString() : null; + } +} diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableSessionQueryTask.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableSessionQueryTask.java new file mode 100644 index 0000000000..b37d2396d3 --- /dev/null +++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableSessionQueryTask.java @@ -0,0 +1,136 @@ +/* + * Copyright 2023 Google LLC + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.google.cloud.storage; + +import static com.google.cloud.storage.HttpClientContext.firstHeaderValue; + +import com.google.api.client.http.EmptyContent; +import com.google.api.client.http.GenericUrl; +import com.google.api.client.http.HttpRequest; +import com.google.api.client.http.HttpResponse; +import com.google.api.client.http.HttpResponseException; +import com.google.api.services.storage.model.StorageObject; +import java.io.IOException; +import java.math.BigInteger; +import java.util.Locale; +import java.util.concurrent.Callable; +import org.checkerframework.checker.nullness.qual.Nullable; + +final class JsonResumableSessionQueryTask + implements Callable> { + + private final HttpClientContext context; + private final String uploadId; + + JsonResumableSessionQueryTask(HttpClientContext context, String uploadId) { + this.context = context; + this.uploadId = uploadId; + } + + public ResumableOperationResult<@Nullable StorageObject> call() { + HttpResponse response = null; + try { + HttpRequest req = + context + .getRequestFactory() + .buildPutRequest(new GenericUrl(uploadId), new EmptyContent()) + .setParser(context.getObjectParser()); + req.setThrowExceptionOnExecuteError(false); + req.getHeaders().setContentRange(HttpContentRange.query().getHeaderValue()); + + response = req.execute(); + + int code = response.getStatusCode(); + if (JsonResumableSessionFailureScenario.isOk(code)) { + @Nullable StorageObject storageObject; + @Nullable BigInteger actualSize; + + Long contentLength = response.getHeaders().getContentLength(); + String contentType = response.getHeaders().getContentType(); + String storedContentLength = + firstHeaderValue(response.getHeaders(), "x-goog-stored-content-length"); + boolean isJson = contentType != null && contentType.startsWith("application/json"); + if (isJson) { + storageObject = response.parseAs(StorageObject.class); + actualSize = storageObject != null ? storageObject.getSize() : null; + } else if ((contentLength == null || contentLength == 0) && storedContentLength != null) { + // when a signed url is used, the finalize response is empty + response.ignore(); + actualSize = new BigInteger(storedContentLength, 10); + storageObject = null; + } else { + response.ignore(); + throw JsonResumableSessionFailureScenario.SCENARIO_0_1.toStorageException( + uploadId, response, null, () -> null); + } + if (actualSize != null) { + if (storageObject != null) { + return ResumableOperationResult.complete(storageObject, actualSize.longValue()); + } else { + return ResumableOperationResult.incremental(actualSize.longValue()); + } + } else { + throw JsonResumableSessionFailureScenario.SCENARIO_0.toStorageException( + uploadId, + response, + null, + () -> storageObject != null ? storageObject.toString() : null); + } + } else if (JsonResumableSessionFailureScenario.isContinue(code)) { + String range1 = response.getHeaders().getRange(); + if (range1 != null) { + ByteRangeSpec range = ByteRangeSpec.parse(range1); + long endOffset = range.endOffset(); + return ResumableOperationResult.incremental(endOffset); + } else { + throw JsonResumableSessionFailureScenario.QUERY_SCENARIO_1.toStorageException( + uploadId, response); + } + } else { + HttpResponseException cause = new HttpResponseException(response); + String contentType = response.getHeaders().getContentType(); + // If the content-range header value has run ahead of the backend, it will respond with + // a 503 with plain text content + // Attempt to detect this very loosely as to minimize impact of modified error message + // This is accurate circa 2023-06 + if ((!JsonResumableSessionFailureScenario.isOk(code) + && !JsonResumableSessionFailureScenario.isContinue(code)) + && contentType != null + && contentType.startsWith("text/plain")) { + String errorMessage = cause.getContent().toLowerCase(Locale.US); + if (errorMessage.contains("content-range")) { + throw JsonResumableSessionFailureScenario.SCENARIO_5.toStorageException( + uploadId, response, cause, cause::getContent); + } + } + throw JsonResumableSessionFailureScenario.toStorageException(response, cause, uploadId); + } + } catch (StorageException se) { + throw se; + } catch (Exception e) { + throw JsonResumableSessionFailureScenario.SCENARIO_0.toStorageException( + uploadId, response, e); + } finally { + if (response != null) { + try { + response.ignore(); + } catch (IOException ignore) { + } + } + } + } +} diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableWrite.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableWrite.java new file mode 100644 index 0000000000..c5e17aad52 --- /dev/null +++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableWrite.java @@ -0,0 +1,88 @@ +/* + * Copyright 2023 Google LLC + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.google.cloud.storage; + +import com.google.api.services.storage.model.StorageObject; +import com.google.cloud.storage.spi.v1.StorageRpc; +import com.google.common.base.MoreObjects; +import java.util.Map; +import java.util.Objects; +import org.checkerframework.checker.nullness.qual.MonotonicNonNull; +import org.checkerframework.checker.nullness.qual.NonNull; + +final class JsonResumableWrite { + @MonotonicNonNull private final StorageObject object; + @MonotonicNonNull private final Map options; + + @MonotonicNonNull private final String signedUrl; + + @NonNull private final String uploadId; + + private JsonResumableWrite( + StorageObject object, + Map options, + String signedUrl, + @NonNull String uploadId) { + this.object = object; + this.options = options; + this.signedUrl = signedUrl; + this.uploadId = uploadId; + } + + public @NonNull String getUploadId() { + return uploadId; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (!(o instanceof JsonResumableWrite)) { + return false; + } + JsonResumableWrite that = (JsonResumableWrite) o; + return Objects.equals(object, that.object) + && Objects.equals(options, that.options) + && Objects.equals(signedUrl, that.signedUrl) + && uploadId.equals(that.uploadId); + } + + @Override + public int hashCode() { + return Objects.hash(object, options, signedUrl, uploadId); + } + + @Override + public String toString() { + return MoreObjects.toStringHelper(this) + .add("object", object) + .add("options", options) + .add("signedUrl", signedUrl) + .add("uploadId", uploadId) + .toString(); + } + + static JsonResumableWrite of( + StorageObject req, Map options, String uploadId) { + return new JsonResumableWrite(req, options, null, uploadId); + } + + static JsonResumableWrite of(String signedUrl, String uploadId) { + return new JsonResumableWrite(null, null, signedUrl, uploadId); + } +} diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/ResumableOperationResult.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/ResumableOperationResult.java new file mode 100644 index 0000000000..88b5b7565e --- /dev/null +++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/ResumableOperationResult.java @@ -0,0 +1,90 @@ +/* + * Copyright 2023 Google LLC + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.google.cloud.storage; + +import com.google.common.base.MoreObjects; +import org.checkerframework.checker.nullness.qual.Nullable; + +abstract class ResumableOperationResult<@Nullable T> { + + private ResumableOperationResult() {} + + abstract @Nullable T getObject(); + + abstract long getPersistedSize(); + + static ResumableOperationResult complete(T t, long persistedSize) { + return new CompletedResult<>(t, persistedSize); + } + + static <@Nullable T> ResumableOperationResult incremental(long persistedSize) { + return new IncrementalResult<>(persistedSize); + } + + private static final class CompletedResult extends ResumableOperationResult { + + private final long persistedSize; + private final T entity; + + private CompletedResult(T entity, long persistedSize) { + this.entity = entity; + this.persistedSize = persistedSize; + } + + @Override + public @Nullable T getObject() { + return entity; + } + + @Override + public long getPersistedSize() { + return persistedSize; + } + + @Override + public String toString() { + return MoreObjects.toStringHelper(this) + .add("persistedSize", persistedSize) + .add("entity", entity) + .toString(); + } + } + + private static final class IncrementalResult<@Nullable T> extends ResumableOperationResult { + + private final long persistedSize; + + private IncrementalResult(long persistedSize) { + this.persistedSize = persistedSize; + } + + @Override + public @Nullable T getObject() { + return null; + } + + @Override + public long getPersistedSize() { + return persistedSize; + } + + @Override + public String toString() { + return MoreObjects.toStringHelper(this).add("persistedSize", persistedSize).toString(); + } + } +} diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/ResumableSession.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/ResumableSession.java new file mode 100644 index 0000000000..7b608c68a4 --- /dev/null +++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/ResumableSession.java @@ -0,0 +1,39 @@ +/* + * Copyright 2023 Google LLC + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.google.cloud.storage; + +import com.google.api.gax.retrying.ResultRetryAlgorithm; +import com.google.cloud.storage.Retrying.RetryingDependencies; +import org.checkerframework.checker.nullness.qual.Nullable; + +abstract class ResumableSession { + + ResumableSession() {} + + abstract ResumableOperationResult<@Nullable T> put( + RewindableHttpContent content, HttpContentRange contentRange); + + abstract ResumableOperationResult<@Nullable T> query(); + + static JsonResumableSession json( + HttpClientContext context, + RetryingDependencies deps, + ResultRetryAlgorithm alg, + JsonResumableWrite resumableWrite) { + return new JsonResumableSession(context, deps, alg, resumableWrite); + } +} diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/RewindableHttpContent.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/RewindableHttpContent.java new file mode 100644 index 0000000000..fd171e0544 --- /dev/null +++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/RewindableHttpContent.java @@ -0,0 +1,106 @@ +/* + * Copyright 2023 Google LLC + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.google.cloud.storage; + +import com.google.api.client.http.AbstractHttpContent; +import com.google.api.client.http.HttpMediaType; +import com.google.common.base.Preconditions; +import com.google.common.io.ByteStreams; +import java.io.IOException; +import java.io.OutputStream; +import java.nio.channels.Channels; +import java.nio.channels.SeekableByteChannel; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.StandardOpenOption; + +abstract class RewindableHttpContent extends AbstractHttpContent { + + private RewindableHttpContent() { + super((HttpMediaType) null); + } + + @Override + public abstract long getLength(); + + abstract void rewindTo(long offset); + + @Override + public final boolean retrySupported() { + return false; + } + + static RewindableHttpContent empty() { + return EmptyRewindableContent.INSTANCE; + } + + static RewindableHttpContent of(Path path) throws IOException { + return new PathRewindableHttpContent(path); + } + + private static final class EmptyRewindableContent extends RewindableHttpContent { + private static final EmptyRewindableContent INSTANCE = new EmptyRewindableContent(); + + @Override + public long getLength() { + return 0L; + } + + @Override + public void writeTo(OutputStream out) throws IOException { + out.flush(); + } + + @Override + protected void rewindTo(long offset) {} + } + + private static final class PathRewindableHttpContent extends RewindableHttpContent { + + private final Path path; + private final long size; + + private long readOffset; + + private PathRewindableHttpContent(Path path) throws IOException { + this.path = path; + this.size = Files.size(path); + this.readOffset = 0; + } + + @Override + public long getLength() { + return size - readOffset; + } + + @Override + protected void rewindTo(long offset) { + Preconditions.checkArgument( + offset < size, "provided offset must be less than size (%d < %d)", offset, size); + this.readOffset = offset; + } + + @Override + public void writeTo(OutputStream out) throws IOException { + try (SeekableByteChannel in = Files.newByteChannel(path, StandardOpenOption.READ)) { + in.position(readOffset); + ByteStreams.copy(in, Channels.newChannel(out)); + out.flush(); + } + } + } +} diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageImpl.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageImpl.java index 8c0768b9de..a97a325c5c 100644 --- a/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageImpl.java +++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageImpl.java @@ -86,6 +86,8 @@ import java.util.concurrent.Callable; import java.util.concurrent.TimeUnit; import java.util.function.Function; +import java.util.function.Supplier; +import org.checkerframework.checker.nullness.qual.Nullable; final class StorageImpl extends BaseService implements Storage { @@ -225,9 +227,40 @@ public Blob createFrom(BlobInfo blobInfo, Path path, int bufferSize, BlobWriteOp if (Files.isDirectory(path)) { throw new StorageException(0, path + " is a directory"); } - try (InputStream input = Files.newInputStream(path)) { - return createFrom(blobInfo, input, bufferSize, options); + Opts opts = Opts.unwrap(options).resolveFrom(blobInfo); + final Map optionsMap = opts.getRpcOptions(); + BlobInfo.Builder builder = blobInfo.toBuilder().setMd5(null).setCrc32c(null); + BlobInfo updated = opts.blobInfoMapper().apply(builder).build(); + StorageObject encode = codecs.blobInfo().encode(updated); + + Supplier uploadIdSupplier = + ResumableMedia.startUploadForBlobInfo( + getOptions(), + updated, + optionsMap, + retryAlgorithmManager.getForResumableUploadSessionCreate(optionsMap)); + JsonResumableWrite jsonResumableWrite = + JsonResumableWrite.of(encode, optionsMap, uploadIdSupplier.get()); + + JsonResumableSession session = + ResumableSession.json( + HttpClientContext.from(storageRpc), + getOptions().asRetryDependencies(), + retryAlgorithmManager.idempotent(), + jsonResumableWrite); + long size = Files.size(path); + HttpContentRange contentRange = + HttpContentRange.of(ByteRangeSpec.relativeLength(0L, size), size); + ResumableOperationResult put = + session.put(RewindableHttpContent.of(path), contentRange); + // all exception translation is taken care of down in the JsonResumableSession + StorageObject object = put.getObject(); + if (object == null) { + // if by some odd chance the put didn't get the StorageObject, query for it + ResumableOperationResult<@Nullable StorageObject> query = session.query(); + object = query.getObject(); } + return codecs.blobInfo().decode(object).asBlob(this); } @Override diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/FakeHttpServer.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/FakeHttpServer.java new file mode 100644 index 0000000000..be739ab188 --- /dev/null +++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/FakeHttpServer.java @@ -0,0 +1,144 @@ +/* + * Copyright 2023 Google LLC + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.google.cloud.storage; + +import static io.grpc.netty.shaded.io.netty.handler.codec.http.HttpHeaderNames.CONNECTION; +import static io.grpc.netty.shaded.io.netty.handler.codec.http.HttpHeaderNames.CONTENT_LENGTH; +import static io.grpc.netty.shaded.io.netty.handler.codec.http.HttpHeaderValues.CLOSE; + +import io.grpc.netty.shaded.io.netty.bootstrap.ServerBootstrap; +import io.grpc.netty.shaded.io.netty.buffer.ByteBuf; +import io.grpc.netty.shaded.io.netty.channel.Channel; +import io.grpc.netty.shaded.io.netty.channel.ChannelFuture; +import io.grpc.netty.shaded.io.netty.channel.ChannelFutureListener; +import io.grpc.netty.shaded.io.netty.channel.ChannelHandlerContext; +import io.grpc.netty.shaded.io.netty.channel.ChannelInitializer; +import io.grpc.netty.shaded.io.netty.channel.ChannelOption; +import io.grpc.netty.shaded.io.netty.channel.ChannelPipeline; +import io.grpc.netty.shaded.io.netty.channel.EventLoopGroup; +import io.grpc.netty.shaded.io.netty.channel.SimpleChannelInboundHandler; +import io.grpc.netty.shaded.io.netty.channel.nio.NioEventLoopGroup; +import io.grpc.netty.shaded.io.netty.channel.socket.SocketChannel; +import io.grpc.netty.shaded.io.netty.channel.socket.nio.NioServerSocketChannel; +import io.grpc.netty.shaded.io.netty.handler.codec.http.FullHttpResponse; +import io.grpc.netty.shaded.io.netty.handler.codec.http.HttpHeaders; +import io.grpc.netty.shaded.io.netty.handler.codec.http.HttpRequest; +import io.grpc.netty.shaded.io.netty.handler.codec.http.HttpServerCodec; +import io.grpc.netty.shaded.io.netty.handler.codec.http.HttpServerExpectContinueHandler; +import io.grpc.netty.shaded.io.netty.handler.logging.LogLevel; +import io.grpc.netty.shaded.io.netty.handler.logging.LoggingHandler; +import java.net.InetSocketAddress; +import java.net.URI; + +final class FakeHttpServer implements AutoCloseable { + + private final URI endpoint; + private final Channel channel; + private final Runnable shutdown; + + private FakeHttpServer(URI endpoint, Channel channel, Runnable shutdown) { + this.endpoint = endpoint; + this.channel = channel; + this.shutdown = shutdown; + } + + public URI getEndpoint() { + return endpoint; + } + + @Override + public void close() throws Exception { + shutdown.run(); + channel.closeFuture().syncUninterruptibly(); + } + + static FakeHttpServer of(HttpRequestHandler server) { + // based on + // https://github.com/netty/netty/blob/59aa6e635b9996cf21cd946e64353270679adc73/example/src/main/java/io/netty/example/http/helloworld/HttpHelloWorldServer.java + InetSocketAddress address = new InetSocketAddress("localhost", 0); + // Configure the server. + EventLoopGroup bossGroup = new NioEventLoopGroup(1); + EventLoopGroup workerGroup = new NioEventLoopGroup(); + ServerBootstrap b = new ServerBootstrap(); + b.option(ChannelOption.SO_BACKLOG, 1024); + b.group(bossGroup, workerGroup) + .channel(NioServerSocketChannel.class) + .handler(new LoggingHandler(LogLevel.DEBUG)) + .childHandler( + new ChannelInitializer() { + @Override + protected void initChannel(SocketChannel ch) { + ChannelPipeline p = ch.pipeline(); + p.addLast(new HttpServerCodec()); + p.addLast(new HttpServerExpectContinueHandler()); + p.addLast(new Handler(server)); + } + }); + + Channel channel = b.bind(address).syncUninterruptibly().channel(); + + InetSocketAddress socketAddress = (InetSocketAddress) channel.localAddress(); + return new FakeHttpServer( + URI.create("http://localhost:" + socketAddress.getPort()), + channel, + () -> { + bossGroup.shutdownGracefully(); + workerGroup.shutdownGracefully(); + }); + } + + interface HttpRequestHandler { + FullHttpResponse apply(HttpRequest req) throws Exception; + } + + /** + * Based on + * https://github.com/netty/netty/blob/59aa6e635b9996cf21cd946e64353270679adc73/example/src/main/java/io/netty/example/http/helloworld/HttpHelloWorldServerHandler.java + */ + private static final class Handler extends SimpleChannelInboundHandler { + + private final HttpRequestHandler server; + + private Handler(HttpRequestHandler server) { + this.server = server; + } + + @Override + public void channelReadComplete(ChannelHandlerContext ctx) { + ctx.flush(); + } + + @Override + protected void channelRead0(ChannelHandlerContext ctx, HttpRequest req) throws Exception { + FullHttpResponse resp = server.apply(req); + HttpHeaders headers = resp.headers(); + if (!headers.contains(CONTENT_LENGTH)) { + ByteBuf content = resp.content(); + headers.setInt(CONTENT_LENGTH, content.readableBytes()); + } + headers.set(CONNECTION, CLOSE); + ChannelFuture future = ctx.writeAndFlush(resp); + future.addListener(ChannelFutureListener.CLOSE); + } + + @Override + public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { + cause.printStackTrace(); + ctx.close(); + } + } +} diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/ITJsonResumableSessionPutTaskTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/ITJsonResumableSessionPutTaskTest.java new file mode 100644 index 0000000000..18704dc288 --- /dev/null +++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/ITJsonResumableSessionPutTaskTest.java @@ -0,0 +1,814 @@ +/* + * Copyright 2023 Google LLC + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.google.cloud.storage; + +import static com.google.cloud.storage.ByteSizeConstants._128KiBL; +import static com.google.cloud.storage.ByteSizeConstants._256KiBL; +import static com.google.cloud.storage.ByteSizeConstants._512KiBL; +import static com.google.cloud.storage.ByteSizeConstants._768KiBL; +import static com.google.common.truth.Truth.assertThat; +import static io.grpc.netty.shaded.io.netty.handler.codec.http.HttpHeaderNames.CONTENT_RANGE; +import static io.grpc.netty.shaded.io.netty.handler.codec.http.HttpHeaderNames.CONTENT_TYPE; +import static io.grpc.netty.shaded.io.netty.handler.codec.http.HttpResponseStatus.OK; +import static org.junit.Assert.assertThrows; + +import com.google.api.client.http.javanet.NetHttpTransport; +import com.google.api.client.json.JsonObjectParser; +import com.google.api.client.json.gson.GsonFactory; +import com.google.api.gax.retrying.ResultRetryAlgorithm; +import com.google.api.services.storage.model.StorageObject; +import com.google.cloud.storage.FakeHttpServer.HttpRequestHandler; +import com.google.cloud.storage.it.runner.StorageITRunner; +import com.google.cloud.storage.it.runner.annotations.Backend; +import com.google.cloud.storage.it.runner.annotations.ParallelFriendly; +import com.google.cloud.storage.it.runner.annotations.SingleBackend; +import com.google.common.collect.ImmutableMap; +import io.grpc.netty.shaded.io.netty.buffer.ByteBuf; +import io.grpc.netty.shaded.io.netty.buffer.Unpooled; +import io.grpc.netty.shaded.io.netty.handler.codec.http.DefaultFullHttpResponse; +import io.grpc.netty.shaded.io.netty.handler.codec.http.FullHttpResponse; +import io.grpc.netty.shaded.io.netty.handler.codec.http.HttpHeaderNames; +import io.grpc.netty.shaded.io.netty.handler.codec.http.HttpResponseStatus; +import java.math.BigInteger; +import java.net.URI; +import java.nio.charset.StandardCharsets; +import java.util.UUID; +import java.util.concurrent.atomic.AtomicLong; +import org.checkerframework.checker.nullness.qual.Nullable; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; +import org.junit.runner.RunWith; + +@RunWith(StorageITRunner.class) +@SingleBackend(Backend.PROD) +@ParallelFriendly +public final class ITJsonResumableSessionPutTaskTest { + private static final GsonFactory gson = GsonFactory.getDefaultInstance(); + private static final NetHttpTransport transport = new NetHttpTransport.Builder().build(); + private static final HttpResponseStatus RESUME_INCOMPLETE = + HttpResponseStatus.valueOf(308, "Resume Incomplete"); + private static final HttpResponseStatus APPEND_GREATER_THAN_CURRENT_SIZE = + HttpResponseStatus.valueOf(503, ""); + private HttpClientContext httpClientContext; + + @Rule public final TemporaryFolder temp = new TemporaryFolder(); + + @Before + public void setUp() throws Exception { + httpClientContext = + HttpClientContext.of(transport.createRequestFactory(), new JsonObjectParser(gson)); + } + + @Test + public void emptyObjectHappyPath() throws Exception { + + HttpRequestHandler handler = + req -> { + StorageObject so = new StorageObject(); + so.setName("object-name").setSize(BigInteger.ZERO); + ByteBuf buf = Unpooled.wrappedBuffer(gson.toByteArray(so)); + + DefaultFullHttpResponse resp = + new DefaultFullHttpResponse(req.protocolVersion(), OK, buf); + resp.headers().set(CONTENT_TYPE, "application/json; charset=utf-8"); + return resp; + }; + + try (FakeHttpServer fakeHttpServer = FakeHttpServer.of(handler)) { + URI endpoint = fakeHttpServer.getEndpoint(); + String uploadUrl = String.format("%s/upload/%s", endpoint.toString(), UUID.randomUUID()); + + JsonResumableSessionPutTask task = + new JsonResumableSessionPutTask( + httpClientContext, + uploadUrl, + RewindableHttpContent.empty(), + HttpContentRange.of(ByteRangeSpec.explicitClosed(0L, 0L), 0)); + + ResumableOperationResult<@Nullable StorageObject> operationResult = task.call(); + StorageObject object = operationResult.getObject(); + assertThat(object).isNotNull(); + assertThat(operationResult.getPersistedSize()).isEqualTo(0L); + } + } + + /** + * + * + *

S.9

+ * + * Partial successful append to session + * + *

The client has sent N bytes, the server confirmed N bytes as committed. The client sends K + * bytes starting at offset N. The server responds with only N + L with 0 <= L < K bytes as + * committed. + */ + @Test + public void scenario9() throws Exception { + + HttpRequestHandler handler = + req -> { + String contentRangeString = req.headers().get(CONTENT_RANGE); + HttpContentRange parse = HttpContentRange.parse(contentRangeString); + long endInclusive = ((HttpContentRange.HasRange) parse).range().endOffsetInclusive(); + FullHttpResponse resp = + new DefaultFullHttpResponse(req.protocolVersion(), RESUME_INCOMPLETE); + ByteRangeSpec range = ByteRangeSpec.explicitClosed(0L, endInclusive - 1); + resp.headers().set(HttpHeaderNames.RANGE, range.getHttpRangeHeader()); + return resp; + }; + + try (FakeHttpServer fakeHttpServer = FakeHttpServer.of(handler)) { + URI endpoint = fakeHttpServer.getEndpoint(); + String uploadUrl = String.format("%s/upload/%s", endpoint.toString(), UUID.randomUUID()); + + AtomicLong confirmedBytes = new AtomicLong(-1L); + + JsonResumableSessionPutTask task = + new JsonResumableSessionPutTask( + httpClientContext, + uploadUrl, + RewindableHttpContent.empty(), + HttpContentRange.of(ByteRangeSpec.explicitClosed(0L, 10L))); + + StorageException se = assertThrows(StorageException.class, task::call); + assertThat(se.getCode()).isEqualTo(503); + assertThat(confirmedBytes.get()).isEqualTo(-1L); + } + } + + /** + * + * + *

S.7

+ * + * GCS Acknowledges more bytes than were sent in the PUT + * + *

The client believes the server offset is N, it sends K bytes and the server responds that N + * + 2K bytes are now committed. + * + *

The client has detected data loss and should raise an error and prevent sending of more + * bytes. + */ + @Test + public void scenario7() throws Exception { + + HttpRequestHandler handler = + req -> { + String contentRangeString = req.headers().get(CONTENT_RANGE); + HttpContentRange parse = HttpContentRange.parse(contentRangeString); + long endInclusive = ((HttpContentRange.HasRange) parse).range().endOffsetInclusive(); + FullHttpResponse resp = + new DefaultFullHttpResponse(req.protocolVersion(), RESUME_INCOMPLETE); + ByteRangeSpec range = ByteRangeSpec.explicitClosed(0L, endInclusive + 1); + resp.headers().set(HttpHeaderNames.RANGE, range.getHttpRangeHeader()); + return resp; + }; + + try (FakeHttpServer fakeHttpServer = FakeHttpServer.of(handler)) { + URI endpoint = fakeHttpServer.getEndpoint(); + String uploadUrl = String.format("%s/upload/%s", endpoint.toString(), UUID.randomUUID()); + + AtomicLong confirmedBytes = new AtomicLong(-1L); + + JsonResumableSessionPutTask task = + new JsonResumableSessionPutTask( + httpClientContext, + uploadUrl, + RewindableHttpContent.empty(), + HttpContentRange.of(ByteRangeSpec.explicitClosed(0L, 10L))); + + StorageException se = assertThrows(StorageException.class, task::call); + assertThat(se.getCode()).isEqualTo(0); + assertThat(se.getReason()).isEqualTo("dataLoss"); + assertThat(confirmedBytes.get()).isEqualTo(-1L); + } + } + + /** + * + * + *

S.1

+ * + * Attempting to append to a session which has already been finalized should raise an error + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + *
server state
+   * resource = { name= obj, persisted_size = 524288 }
+   *
client state
+   * write_offset = 0, data = [0:262144]
+   *
request
+   * PUT $UPLOAD_ID
+   * Content-Range: bytes 0-262143/*
+   *
response
+   * 200 OK
+   * Content-Type: application/json; charset=utf-8
+   *
+   * {"name": "obj", "size": 524288}
+   *
+ */ + @Test + public void scenario1() throws Exception { + HttpRequestHandler handler = + req -> { + StorageObject so = new StorageObject(); + URI uri = URI.create(req.uri()); + so.setName("object") + .setBucket("bucket") + .setGeneration(1L) + .setMetageneration(1L) + .setSize(BigInteger.valueOf(_512KiBL)) + .setMetadata(ImmutableMap.of("upload_id", uri.toString())); + + ByteBuf buf = Unpooled.wrappedBuffer(gson.toByteArray(so)); + DefaultFullHttpResponse resp = + new DefaultFullHttpResponse(req.protocolVersion(), OK, buf); + resp.headers().set(CONTENT_TYPE, "application/json; charset=utf-8"); + return resp; + }; + + try (FakeHttpServer fakeHttpServer = FakeHttpServer.of(handler); + TmpFile tmpFile = + DataGenerator.base64Characters().tempFile(temp.newFolder().toPath(), _256KiBL)) { + URI endpoint = fakeHttpServer.getEndpoint(); + String uploadUrl = String.format("%s/upload/%s", endpoint.toString(), UUID.randomUUID()); + + AtomicLong confirmedBytes = new AtomicLong(-1L); + + JsonResumableSessionPutTask task = + new JsonResumableSessionPutTask( + httpClientContext, + uploadUrl, + RewindableHttpContent.of(tmpFile.getPath()), + HttpContentRange.of(ByteRangeSpec.explicit(0L, _256KiBL))); + + StorageException se = assertThrows(StorageException.class, task::call); + assertThat(se.getCode()).isEqualTo(0); + assertThat(se.getReason()).isEqualTo("invalid"); + assertThat(confirmedBytes.get()).isEqualTo(-1L); + } + } + + /** + * + * + *

S.2

+ * + * Attempting to finalize a session with fewer bytes than GCS acknowledges. + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + *
server state
+   * persisted_size = 524288
+   *
client state
+   * write_offset = 262144, finish = true
+   *
request
+   * PUT $UPLOAD_ID
+   * Content-Range: bytes */262144
+   *
response
+   * 308 Resume Incomplete
+   * Range: bytes=0-524287
+   *
+ */ + @Test + public void scenario2() throws Exception { + + HttpRequestHandler handler = + req -> { + FullHttpResponse resp = + new DefaultFullHttpResponse(req.protocolVersion(), RESUME_INCOMPLETE); + ByteRangeSpec range = ByteRangeSpec.explicit(0L, _512KiBL); + resp.headers().set(HttpHeaderNames.RANGE, range.getHttpRangeHeader()); + return resp; + }; + + try (FakeHttpServer fakeHttpServer = FakeHttpServer.of(handler)) { + URI endpoint = fakeHttpServer.getEndpoint(); + String uploadUrl = String.format("%s/upload/%s", endpoint.toString(), UUID.randomUUID()); + + AtomicLong confirmedBytes = new AtomicLong(-1L); + + JsonResumableSessionPutTask task = + new JsonResumableSessionPutTask( + httpClientContext, + uploadUrl, + RewindableHttpContent.empty(), + HttpContentRange.of(_256KiBL)); + + StorageException se = assertThrows(StorageException.class, task::call); + assertThat(se.getCode()).isEqualTo(0); + assertThat(se.getReason()).isEqualTo("invalid"); + assertThat(confirmedBytes.get()).isEqualTo(-1L); + } + } + + /** + * + * + *

S.3

+ * + * Attempting to finalize a session with more bytes than GCS acknowledges. + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + *
server state
+   * persisted_size = 262144
+   *
client state
+   * write_offset = 524288, finish = true
+   *
request
+   * PUT $UPLOAD_ID
+   * Content-Range: bytes */524288
+   *
response
+   * 308 Resume Incomplete
+   * Range: bytes=0-262143
+   *
+ */ + @Test + public void scenario3() throws Exception { + + HttpRequestHandler handler = + req -> { + FullHttpResponse resp = + new DefaultFullHttpResponse(req.protocolVersion(), RESUME_INCOMPLETE); + ByteRangeSpec range = ByteRangeSpec.explicit(0L, _256KiBL); + resp.headers().set(HttpHeaderNames.RANGE, range.getHttpRangeHeader()); + return resp; + }; + + try (FakeHttpServer fakeHttpServer = FakeHttpServer.of(handler)) { + URI endpoint = fakeHttpServer.getEndpoint(); + String uploadUrl = String.format("%s/upload/%s", endpoint.toString(), UUID.randomUUID()); + + AtomicLong confirmedBytes = new AtomicLong(-1L); + + JsonResumableSessionPutTask task = + new JsonResumableSessionPutTask( + httpClientContext, + uploadUrl, + RewindableHttpContent.empty(), + HttpContentRange.of(_512KiBL)); + + StorageException se = assertThrows(StorageException.class, task::call); + assertThat(se.getCode()).isEqualTo(0); + assertThat(se.getReason()).isEqualTo("dataLoss"); + assertThat(confirmedBytes.get()).isEqualTo(-1L); + } + } + + /** + * + * + *

S.4

+ * + * Attempting to finalize an already finalized session + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + *
server state
+   * resource = {name = obj1, size = 262114}
+   *
client state
+   * write_offset = 262114, finish = true
+   *
request
+   * PUT $UPLOAD_ID
+   * Content-Range: bytes */262114
+   *
response
+   * 200 Ok
+   * Content-Type: application/json; charset=utf-8
+   *
+   * {"name": "obj", "size": 262114}
+   *
+ */ + @Test + public void scenario4() throws Exception { + + HttpRequestHandler handler = + req -> { + StorageObject so = new StorageObject(); + URI uri = URI.create(req.uri()); + so.setName("object") + .setBucket("bucket") + .setGeneration(1L) + .setMetageneration(1L) + .setSize(BigInteger.valueOf(_256KiBL)) + .setMetadata(ImmutableMap.of("upload_id", uri.toString())); + + ByteBuf buf = Unpooled.wrappedBuffer(gson.toByteArray(so)); + DefaultFullHttpResponse resp = + new DefaultFullHttpResponse(req.protocolVersion(), OK, buf); + resp.headers().set(CONTENT_TYPE, "application/json; charset=utf-8"); + return resp; + }; + + try (FakeHttpServer fakeHttpServer = FakeHttpServer.of(handler)) { + URI endpoint = fakeHttpServer.getEndpoint(); + String uploadUrl = String.format("%s/upload/%s", endpoint.toString(), UUID.randomUUID()); + + JsonResumableSessionPutTask task = + new JsonResumableSessionPutTask( + httpClientContext, + uploadUrl, + RewindableHttpContent.empty(), + HttpContentRange.of(_256KiBL)); + + ResumableOperationResult<@Nullable StorageObject> operationResult = task.call(); + StorageObject call = operationResult.getObject(); + assertThat(call).isNotNull(); + assertThat(call.getMetadata()) + .containsEntry("upload_id", uploadUrl.substring(endpoint.toString().length())); + assertThat(operationResult.getPersistedSize()).isEqualTo(_256KiBL); + } + } + + /** + * + * + *

S.4.1

+ * + * Attempting to finalize an already finalized session (ack < expected) + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + *
server state
+   * resource = {name = obj1, size = 262114}
+   *
client state
+   * write_offset = 524288, finish = true
+   *
request
+   * PUT $UPLOAD_ID
+   * Content-Range: bytes */524288
+   *
response
+   * 200 Ok
+   * Content-Type: application/json; charset=utf-8
+   *
+   * {"name": "obj", "size": 262114}
+   *
+ */ + @Test + public void scenario4_1() throws Exception { + + HttpRequestHandler handler = + req -> { + StorageObject so = new StorageObject(); + URI uri = URI.create(req.uri()); + so.setName("object") + .setBucket("bucket") + .setGeneration(1L) + .setMetageneration(1L) + .setSize(BigInteger.valueOf(_256KiBL)) + .setMetadata(ImmutableMap.of("upload_id", uri.toString())); + + ByteBuf buf = Unpooled.wrappedBuffer(gson.toByteArray(so)); + DefaultFullHttpResponse resp = + new DefaultFullHttpResponse(req.protocolVersion(), OK, buf); + resp.headers().set(CONTENT_TYPE, "application/json; charset=utf-8"); + return resp; + }; + + try (FakeHttpServer fakeHttpServer = FakeHttpServer.of(handler)) { + URI endpoint = fakeHttpServer.getEndpoint(); + String uploadUrl = String.format("%s/upload/%s", endpoint.toString(), UUID.randomUUID()); + + AtomicLong confirmedBytes = new AtomicLong(-1L); + + JsonResumableSessionPutTask task = + new JsonResumableSessionPutTask( + httpClientContext, + uploadUrl, + RewindableHttpContent.empty(), + HttpContentRange.of(_512KiBL)); + + StorageException se = assertThrows(StorageException.class, task::call); + assertThat(se.getCode()).isEqualTo(0); + assertThat(se.getReason()).isEqualTo("dataLoss"); + assertThat(confirmedBytes.get()).isEqualTo(-1); + } + } + + /** + * + * + *

S.4.2

+ * + * Attempting to finalize an already finalized session (ack > expected) + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + *
server state
+   * resource = {name = obj1, size = 262114}
+   *
client state
+   * write_offset = 524288, finish = true
+   *
request
+   * PUT $UPLOAD_ID
+   * Content-Range: bytes */131072
+   *
response
+   * 200 Ok
+   * Content-Type: application/json; charset=utf-8
+   *
+   * {"name": "obj", "size": 262114}
+   *
+ */ + @Test + public void scenario4_2() throws Exception { + + HttpRequestHandler handler = + req -> { + StorageObject so = new StorageObject(); + URI uri = URI.create(req.uri()); + so.setName("object") + .setBucket("bucket") + .setGeneration(1L) + .setMetageneration(1L) + .setSize(BigInteger.valueOf(_256KiBL)) + .setMetadata(ImmutableMap.of("upload_id", uri.toString())); + + ByteBuf buf = Unpooled.wrappedBuffer(gson.toByteArray(so)); + DefaultFullHttpResponse resp = + new DefaultFullHttpResponse(req.protocolVersion(), OK, buf); + resp.headers().set(CONTENT_TYPE, "application/json; charset=utf-8"); + return resp; + }; + + try (FakeHttpServer fakeHttpServer = FakeHttpServer.of(handler)) { + URI endpoint = fakeHttpServer.getEndpoint(); + String uploadUrl = String.format("%s/upload/%s", endpoint.toString(), UUID.randomUUID()); + + AtomicLong confirmedBytes = new AtomicLong(-1L); + + JsonResumableSessionPutTask task = + new JsonResumableSessionPutTask( + httpClientContext, + uploadUrl, + RewindableHttpContent.empty(), + HttpContentRange.of(_128KiBL)); + + StorageException se = assertThrows(StorageException.class, task::call); + assertThat(se.getCode()).isEqualTo(0); + assertThat(se.getReason()).isEqualTo("dataLoss"); + assertThat(confirmedBytes.get()).isEqualTo(-1); + } + } + + /** + * + * + *

S.5

+ * + * Attempt to append to a resumable session with an offset higher than GCS expects + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + *
server state
+   * persisted_size = 262144
+   *
client state
+   * write_offset = 524288, data = [524288:786432]
+   *
request
+   * PUT $UPLOAD_ID
+   * Content-Range: bytes 524288-786431/*
+   *
response
+   * 503
+   * Content-Type: text/plain; charset=utf-8
+   *
+   * Invalid request. According to the Content-Range header, the upload offset is 524288 byte(s), which exceeds already uploaded size of 262144 byte(s).
+   *
+ */ + @Test + public void scenario5() throws Exception { + + HttpRequestHandler handler = + req -> { + // error message from GCS circa 2023-02 + ByteBuf buf = + Unpooled.wrappedBuffer( + "Invalid request. According to the Content-Range header, the upload offset is 524288 byte(s), which exceeds already uploaded size of 262144 byte(s)." + .getBytes(StandardCharsets.UTF_8)); + FullHttpResponse resp = + new DefaultFullHttpResponse( + req.protocolVersion(), APPEND_GREATER_THAN_CURRENT_SIZE, buf); + resp.headers().set(CONTENT_TYPE, "text/plain; charset=utf-8"); + return resp; + }; + + try (FakeHttpServer fakeHttpServer = FakeHttpServer.of(handler); + TmpFile tmpFile = + DataGenerator.base64Characters().tempFile(temp.newFolder().toPath(), _256KiBL)) { + URI endpoint = fakeHttpServer.getEndpoint(); + String uploadUrl = String.format("%s/upload/%s", endpoint.toString(), UUID.randomUUID()); + + AtomicLong confirmedBytes = new AtomicLong(-1L); + + JsonResumableSessionPutTask task = + new JsonResumableSessionPutTask( + httpClientContext, + uploadUrl, + RewindableHttpContent.of(tmpFile.getPath()), + HttpContentRange.of(ByteRangeSpec.explicit(_512KiBL, _768KiBL))); + + StorageException se = assertThrows(StorageException.class, task::call); + assertThat(se.getCode()).isEqualTo(0); + assertThat(se.getReason()).isEqualTo("dataLoss"); + assertThat(confirmedBytes.get()).isEqualTo(-1); + } + } + + @Test + public void jsonParseFailure() throws Exception { + + HttpRequestHandler handler = + req -> { + StorageObject so = new StorageObject(); + URI uri = URI.create(req.uri()); + so.setName("object") + .setBucket("bucket") + .setGeneration(1L) + .setMetageneration(1L) + .setSize(BigInteger.ZERO) + .setMetadata(ImmutableMap.of("upload_id", uri.toString())); + + byte[] bytes = gson.toByteArray(so); + ByteBuf buf = Unpooled.wrappedBuffer(bytes, 0, bytes.length / 2); + DefaultFullHttpResponse resp = + new DefaultFullHttpResponse(req.protocolVersion(), OK, buf); + resp.headers().set(CONTENT_TYPE, "application/json; charset=utf-8"); + return resp; + }; + + try (FakeHttpServer fakeHttpServer = FakeHttpServer.of(handler)) { + URI endpoint = fakeHttpServer.getEndpoint(); + String uploadUrl = String.format("%s/upload/%s", endpoint.toString(), UUID.randomUUID()); + + AtomicLong confirmedBytes = new AtomicLong(-1L); + + JsonResumableSessionPutTask task = + new JsonResumableSessionPutTask( + httpClientContext, uploadUrl, RewindableHttpContent.empty(), HttpContentRange.of(0)); + + StorageException se = assertThrows(StorageException.class, task::call); + // the parse error happens while trying to read the success object, make sure we raise it as + // a client side retryable exception + assertThat(se.getCode()).isEqualTo(0); + assertThat(se.getReason()).isEqualTo(null); + // Finalization was successful, but we can't confirm the number of bytes due to the parse + // error + assertThat(confirmedBytes.get()).isEqualTo(-1); + + ResultRetryAlgorithm idempotentHandler = + StorageRetryStrategy.getDefaultStorageRetryStrategy().getIdempotentHandler(); + boolean shouldRetry = idempotentHandler.shouldRetry(se, null); + assertThat(shouldRetry).isTrue(); + } + } + + @Test + public void jsonDeserializationOnlyAttemptedWhenContentPresent() throws Exception { + + HttpRequestHandler handler = + req -> { + DefaultFullHttpResponse resp = new DefaultFullHttpResponse(req.protocolVersion(), OK); + resp.headers().set(CONTENT_TYPE, "text/html; charset=UTF-8"); + resp.headers().set("x-goog-stored-content-length", "0"); + return resp; + }; + + try (FakeHttpServer fakeHttpServer = FakeHttpServer.of(handler)) { + URI endpoint = fakeHttpServer.getEndpoint(); + String uploadUrl = String.format("%s/upload/%s", endpoint.toString(), UUID.randomUUID()); + + JsonResumableSessionPutTask task = + new JsonResumableSessionPutTask( + httpClientContext, uploadUrl, RewindableHttpContent.empty(), HttpContentRange.of(0)); + + ResumableOperationResult<@Nullable StorageObject> operationResult = task.call(); + StorageObject call = operationResult.getObject(); + assertThat(call).isNull(); + assertThat(operationResult.getPersistedSize()).isEqualTo(0L); + } + } +} diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/ITJsonResumableSessionQueryTaskTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/ITJsonResumableSessionQueryTaskTest.java new file mode 100644 index 0000000000..07a04ed61e --- /dev/null +++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/ITJsonResumableSessionQueryTaskTest.java @@ -0,0 +1,235 @@ +/* + * Copyright 2023 Google LLC + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.google.cloud.storage; + +import static com.google.cloud.storage.ByteSizeConstants._256KiBL; +import static com.google.common.truth.Truth.assertThat; +import static io.grpc.netty.shaded.io.netty.handler.codec.http.HttpHeaderNames.CONTENT_TYPE; +import static io.grpc.netty.shaded.io.netty.handler.codec.http.HttpResponseStatus.OK; +import static org.junit.Assert.assertThrows; + +import com.google.api.client.http.javanet.NetHttpTransport; +import com.google.api.client.json.JsonObjectParser; +import com.google.api.client.json.gson.GsonFactory; +import com.google.api.services.storage.model.StorageObject; +import com.google.cloud.storage.FakeHttpServer.HttpRequestHandler; +import com.google.cloud.storage.it.runner.StorageITRunner; +import com.google.cloud.storage.it.runner.annotations.Backend; +import com.google.cloud.storage.it.runner.annotations.ParallelFriendly; +import com.google.cloud.storage.it.runner.annotations.SingleBackend; +import io.grpc.netty.shaded.io.netty.buffer.ByteBuf; +import io.grpc.netty.shaded.io.netty.buffer.Unpooled; +import io.grpc.netty.shaded.io.netty.handler.codec.http.DefaultFullHttpResponse; +import io.grpc.netty.shaded.io.netty.handler.codec.http.FullHttpResponse; +import io.grpc.netty.shaded.io.netty.handler.codec.http.HttpHeaderNames; +import io.grpc.netty.shaded.io.netty.handler.codec.http.HttpResponseStatus; +import java.math.BigInteger; +import java.net.URI; +import java.nio.charset.StandardCharsets; +import java.util.UUID; +import org.checkerframework.checker.nullness.qual.Nullable; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; + +@RunWith(StorageITRunner.class) +@SingleBackend(Backend.PROD) +@ParallelFriendly +public final class ITJsonResumableSessionQueryTaskTest { + private static final GsonFactory gson = GsonFactory.getDefaultInstance(); + private static final NetHttpTransport transport = new NetHttpTransport.Builder().build(); + private static final HttpResponseStatus RESUME_INCOMPLETE = + HttpResponseStatus.valueOf(308, "Resume Incomplete"); + private static final HttpResponseStatus APPEND_GREATER_THAN_CURRENT_SIZE = + HttpResponseStatus.valueOf(503, ""); + + private HttpClientContext httpClientContext; + + @Before + public void setUp() throws Exception { + httpClientContext = + HttpClientContext.of(transport.createRequestFactory(), new JsonObjectParser(gson)); + } + + @Test + public void successfulSession() throws Exception { + HttpRequestHandler handler = + req -> { + StorageObject so = new StorageObject(); + so.setName("object-name").setSize(BigInteger.ZERO); + ByteBuf buf = Unpooled.wrappedBuffer(gson.toByteArray(so)); + + DefaultFullHttpResponse resp = + new DefaultFullHttpResponse(req.protocolVersion(), OK, buf); + resp.headers().set(CONTENT_TYPE, "application/json; charset=utf-8"); + return resp; + }; + + try (FakeHttpServer fakeHttpServer = FakeHttpServer.of(handler)) { + URI endpoint = fakeHttpServer.getEndpoint(); + String uploadUrl = String.format("%s/upload/%s", endpoint.toString(), UUID.randomUUID()); + + JsonResumableSessionQueryTask task = + new JsonResumableSessionQueryTask(httpClientContext, uploadUrl); + + ResumableOperationResult<@Nullable StorageObject> result = task.call(); + StorageObject object = result.getObject(); + assertThat(object).isNotNull(); + assertThat(result.getPersistedSize()).isEqualTo(0L); + } + } + + @Test + public void successfulSession_noObject() throws Exception { + HttpRequestHandler handler = + req -> { + DefaultFullHttpResponse response = new DefaultFullHttpResponse(req.protocolVersion(), OK); + response.headers().set("X-Goog-Stored-Content-Length", 0); + return response; + }; + + try (FakeHttpServer fakeHttpServer = FakeHttpServer.of(handler)) { + URI endpoint = fakeHttpServer.getEndpoint(); + String uploadUrl = String.format("%s/upload/%s", endpoint.toString(), UUID.randomUUID()); + + JsonResumableSessionQueryTask task = + new JsonResumableSessionQueryTask(httpClientContext, uploadUrl); + + ResumableOperationResult<@Nullable StorageObject> result = task.call(); + StorageObject object = result.getObject(); + assertThat(object).isNull(); + assertThat(result.getPersistedSize()).isEqualTo(0L); + } + } + + @Test + public void incompleteSession() throws Exception { + HttpRequestHandler handler = + req -> { + DefaultFullHttpResponse response = + new DefaultFullHttpResponse(req.protocolVersion(), RESUME_INCOMPLETE); + response + .headers() + .set( + HttpHeaderNames.RANGE, + ByteRangeSpec.relativeLength(0L, _256KiBL).getHttpRangeHeader()); + return response; + }; + try (FakeHttpServer fakeHttpServer = FakeHttpServer.of(handler)) { + URI endpoint = fakeHttpServer.getEndpoint(); + String uploadUrl = String.format("%s/upload/%s", endpoint.toString(), UUID.randomUUID()); + + JsonResumableSessionQueryTask task = + new JsonResumableSessionQueryTask(httpClientContext, uploadUrl); + + ResumableOperationResult<@Nullable StorageObject> result = task.call(); + assertThat(result.getPersistedSize()).isEqualTo(_256KiBL); + } + } + + /** + * This is a hard failure from the perspective of GCS as a range header is a required header to be + * included in the response to a query upload request. + */ + @Test + public void incompleteSession_missingRangeHeader() throws Exception { + HttpRequestHandler handler = + req -> new DefaultFullHttpResponse(req.protocolVersion(), RESUME_INCOMPLETE); + try (FakeHttpServer fakeHttpServer = FakeHttpServer.of(handler)) { + URI endpoint = fakeHttpServer.getEndpoint(); + String uploadUrl = String.format("%s/upload/%s", endpoint.toString(), UUID.randomUUID()); + + JsonResumableSessionQueryTask task = + new JsonResumableSessionQueryTask(httpClientContext, uploadUrl); + + StorageException se = assertThrows(StorageException.class, task::call); + assertThat(se.getCode()).isEqualTo(503); + assertThat(se).hasMessageThat().contains("Range"); + } + } + + @Test + public void successfulSession_noJson_noStoredContentLength() throws Exception { + HttpRequestHandler handler = req -> new DefaultFullHttpResponse(req.protocolVersion(), OK); + + try (FakeHttpServer fakeHttpServer = FakeHttpServer.of(handler)) { + URI endpoint = fakeHttpServer.getEndpoint(); + String uploadUrl = String.format("%s/upload/%s", endpoint.toString(), UUID.randomUUID()); + + JsonResumableSessionQueryTask task = + new JsonResumableSessionQueryTask(httpClientContext, uploadUrl); + + StorageException se = assertThrows(StorageException.class, task::call); + assertThat(se.getCode()).isEqualTo(0); + } + } + + @Test + public void successfulSession_noSize() throws Exception { + HttpRequestHandler handler = + req -> { + StorageObject so = new StorageObject(); + so.setName("object-name"); + ByteBuf buf = Unpooled.wrappedBuffer(gson.toByteArray(so)); + + DefaultFullHttpResponse resp = + new DefaultFullHttpResponse(req.protocolVersion(), OK, buf); + resp.headers().set(CONTENT_TYPE, "application/json; charset=utf-8"); + return resp; + }; + + try (FakeHttpServer fakeHttpServer = FakeHttpServer.of(handler)) { + URI endpoint = fakeHttpServer.getEndpoint(); + String uploadUrl = String.format("%s/upload/%s", endpoint.toString(), UUID.randomUUID()); + + JsonResumableSessionQueryTask task = + new JsonResumableSessionQueryTask(httpClientContext, uploadUrl); + + StorageException se = assertThrows(StorageException.class, task::call); + assertThat(se.getCode()).isEqualTo(0); + } + } + + @Test + public void query_badOffset() throws Exception { + HttpRequestHandler handler = + req -> { + // error message from GCS circa 2023-02 + ByteBuf buf = + Unpooled.wrappedBuffer( + "Invalid request. According to the Content-Range header, the upload offset is 524288 byte(s), which exceeds already uploaded size of 262144 byte(s)." + .getBytes(StandardCharsets.UTF_8)); + FullHttpResponse resp = + new DefaultFullHttpResponse( + req.protocolVersion(), APPEND_GREATER_THAN_CURRENT_SIZE, buf); + resp.headers().set(CONTENT_TYPE, "text/plain; charset=utf-8"); + return resp; + }; + + try (FakeHttpServer fakeHttpServer = FakeHttpServer.of(handler)) { + URI endpoint = fakeHttpServer.getEndpoint(); + String uploadUrl = String.format("%s/upload/%s", endpoint.toString(), UUID.randomUUID()); + + JsonResumableSessionQueryTask task = + new JsonResumableSessionQueryTask(httpClientContext, uploadUrl); + + StorageException se = assertThrows(StorageException.class, task::call); + assertThat(se.getCode()).isEqualTo(0); + assertThat(se.getReason()).isEqualTo("dataLoss"); + } + } +} diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/ITJsonResumableSessionTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/ITJsonResumableSessionTest.java new file mode 100644 index 0000000000..a71f5cf493 --- /dev/null +++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/ITJsonResumableSessionTest.java @@ -0,0 +1,134 @@ +/* + * Copyright 2023 Google LLC + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.google.cloud.storage; + +import static com.google.cloud.storage.ByteSizeConstants._256KiBL; +import static com.google.cloud.storage.ByteSizeConstants._512KiBL; +import static com.google.common.truth.Truth.assertThat; +import static io.grpc.netty.shaded.io.netty.handler.codec.http.HttpHeaderNames.CONTENT_RANGE; +import static io.grpc.netty.shaded.io.netty.handler.codec.http.HttpHeaderNames.RANGE; + +import com.google.api.client.http.javanet.NetHttpTransport; +import com.google.api.client.json.JsonObjectParser; +import com.google.api.client.json.gson.GsonFactory; +import com.google.api.core.ApiClock; +import com.google.api.core.NanoClock; +import com.google.api.gax.retrying.ResultRetryAlgorithm; +import com.google.api.gax.retrying.RetrySettings; +import com.google.api.services.storage.model.StorageObject; +import com.google.cloud.storage.FakeHttpServer.HttpRequestHandler; +import com.google.cloud.storage.Retrying.RetryingDependencies; +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableMap; +import io.grpc.netty.shaded.io.netty.handler.codec.http.DefaultFullHttpResponse; +import io.grpc.netty.shaded.io.netty.handler.codec.http.HttpRequest; +import io.grpc.netty.shaded.io.netty.handler.codec.http.HttpResponseStatus; +import java.net.URI; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.UUID; +import java.util.stream.Collectors; +import org.checkerframework.checker.nullness.qual.Nullable; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; + +public final class ITJsonResumableSessionTest { + private static final GsonFactory gson = GsonFactory.getDefaultInstance(); + private static final NetHttpTransport transport = new NetHttpTransport.Builder().build(); + private static final HttpResponseStatus RESUME_INCOMPLETE = + HttpResponseStatus.valueOf(308, "Resume Incomplete"); + private static final HttpResponseStatus APPEND_GREATER_THAN_CURRENT_SIZE = + HttpResponseStatus.valueOf(503, ""); + private static final RetryingDependencies RETRYING_DEPENDENCIES = + new RetryingDependencies() { + @Override + public RetrySettings getRetrySettings() { + return RetrySettings.newBuilder().setMaxAttempts(3).build(); + } + + @Override + public ApiClock getClock() { + return NanoClock.getDefaultClock(); + } + }; + private static final ResultRetryAlgorithm RETRY_ALGORITHM = + StorageRetryStrategy.getUniformStorageRetryStrategy().getIdempotentHandler(); + private HttpClientContext httpClientContext; + + @Rule public final TemporaryFolder temp = new TemporaryFolder(); + + @Before + public void setUp() throws Exception { + httpClientContext = + HttpClientContext.of(transport.createRequestFactory(), new JsonObjectParser(gson)); + } + + @Test + public void rewindWillQueryStatusOnlyWhenDirty() throws Exception { + HttpContentRange range1 = HttpContentRange.of(ByteRangeSpec.explicit(0L, _512KiBL)); + HttpContentRange range2 = HttpContentRange.query(); + HttpContentRange range3 = HttpContentRange.of(ByteRangeSpec.explicit(_256KiBL, _512KiBL)); + + final List requests = Collections.synchronizedList(new ArrayList<>()); + HttpRequestHandler handler = + req -> { + requests.add(req); + String contentRange = req.headers().get(CONTENT_RANGE); + System.out.println("contentRange = " + contentRange); + DefaultFullHttpResponse resp = + new DefaultFullHttpResponse(req.protocolVersion(), RESUME_INCOMPLETE); + if (range1.getHeaderValue().equals(contentRange)) { + resp.headers().set(RANGE, ByteRangeSpec.explicit(0L, _256KiBL).getHttpRangeHeader()); + } else if (range2.getHeaderValue().equals(contentRange)) { + resp.headers().set(RANGE, ByteRangeSpec.explicit(0L, _256KiBL).getHttpRangeHeader()); + } else { + resp.headers().set(RANGE, ByteRangeSpec.explicit(0L, _512KiBL).getHttpRangeHeader()); + } + return resp; + }; + + try (FakeHttpServer fakeHttpServer = FakeHttpServer.of(handler); + TmpFile tmpFile = + DataGenerator.base64Characters().tempFile(temp.newFolder().toPath(), _512KiBL)) { + URI endpoint = fakeHttpServer.getEndpoint(); + String uploadUrl = String.format("%s/upload/%s", endpoint.toString(), UUID.randomUUID()); + + JsonResumableWrite resumableWrite = JsonResumableWrite.of(null, ImmutableMap.of(), uploadUrl); + JsonResumableSession session = + new JsonResumableSession( + httpClientContext, RETRYING_DEPENDENCIES, RETRY_ALGORITHM, resumableWrite); + + ResumableOperationResult<@Nullable StorageObject> operationResult = + session.put(RewindableHttpContent.of(tmpFile.getPath()), range1); + StorageObject call = operationResult.getObject(); + assertThat(call).isNull(); + assertThat(operationResult.getPersistedSize()).isEqualTo(_512KiBL); + } + + assertThat(requests).hasSize(3); + List actual = + requests.stream().map(r -> r.headers().get(CONTENT_RANGE)).collect(Collectors.toList()); + + List expected = + ImmutableList.of(range1.getHeaderValue(), range2.getHeaderValue(), range3.getHeaderValue()); + + assertThat(actual).isEqualTo(expected); + } +} diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/JsonResumableSessionFailureScenarioTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/JsonResumableSessionFailureScenarioTest.java new file mode 100644 index 0000000000..5f23a5aa19 --- /dev/null +++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/JsonResumableSessionFailureScenarioTest.java @@ -0,0 +1,155 @@ +/* + * Copyright 2023 Google LLC + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.google.cloud.storage; + +import static com.google.cloud.storage.JsonResumableSessionFailureScenario.isContinue; +import static com.google.cloud.storage.JsonResumableSessionFailureScenario.isOk; +import static com.google.common.truth.Truth.assertThat; + +import com.google.api.client.http.EmptyContent; +import com.google.api.client.http.GenericUrl; +import com.google.api.client.http.HttpRequest; +import com.google.api.client.http.HttpResponse; +import com.google.api.client.json.gson.GsonFactory; +import com.google.api.client.testing.http.MockHttpTransport; +import com.google.api.services.storage.model.StorageObject; +import com.google.common.collect.ImmutableMap; +import java.io.IOException; +import java.math.BigInteger; +import java.nio.charset.StandardCharsets; +import org.junit.Test; + +public final class JsonResumableSessionFailureScenarioTest { + private static final GsonFactory gson = GsonFactory.getDefaultInstance(); + + @Test + public void isOk_200() { + assertThat(isOk(200)).isTrue(); + } + + @Test + public void isOk_201() { + assertThat(isOk(201)).isTrue(); + } + + @Test + public void isContinue_308() { + assertThat(isContinue(308)).isTrue(); + } + + @Test + public void toStorageException_ioExceptionDuringContentResolutionAddedAsSuppressed() + throws IOException { + HttpRequest req = + new MockHttpTransport() + .createRequestFactory() + .buildPutRequest(new GenericUrl("http://localhost:80980"), new EmptyContent()); + req.getHeaders().setContentLength(0L).setContentRange(HttpContentRange.of(0).getHeaderValue()); + + HttpResponse resp = req.execute(); + resp.getHeaders().setContentType("text/plain; charset=utf-8").setContentLength(5L); + + StorageException storageException = + JsonResumableSessionFailureScenario.SCENARIO_1.toStorageException( + "uploadId", + resp, + new Cause(), + () -> { + throw new Kaboom(); + }); + + assertThat(storageException.getCode()).isEqualTo(400); + assertThat(storageException).hasCauseThat().isInstanceOf(Cause.class); + assertThat(storageException.getSuppressed()).isNotEmpty(); + assertThat(storageException.getSuppressed()[0]).isInstanceOf(StorageException.class); + assertThat(storageException.getSuppressed()[0]).hasCauseThat().isInstanceOf(Kaboom.class); + } + + @Test + public void multilineResponseBodyIsProperlyPrefixed() throws Exception { + StorageObject so = new StorageObject(); + so.setName("object-name") + .setSize(BigInteger.ZERO) + .setGeneration(1L) + .setMetageneration(2L) + .setMetadata( + ImmutableMap.of( + "k1", "v1", + "k2", "v2")); + final String json = gson.toPrettyString(so); + + byte[] bytes = json.getBytes(StandardCharsets.UTF_8); + HttpRequest req = + new MockHttpTransport() + .createRequestFactory() + .buildPutRequest(new GenericUrl("http://localhost:80980"), new EmptyContent()); + req.getHeaders().setContentLength(0L); + + HttpResponse resp = req.execute(); + resp.getHeaders() + .setContentType("application/json; charset=utf-8") + .setContentLength((long) bytes.length); + + StorageException storageException = + JsonResumableSessionFailureScenario.SCENARIO_0.toStorageException( + "uploadId", resp, null, () -> json); + + assertThat(storageException.getCode()).isEqualTo(0); + assertThat(storageException).hasMessageThat().contains("\t|< \"generation\": \"1\",\n"); + } + + @Test + public void xGoogStoredHeadersIncludedIfPresent() throws IOException { + HttpRequest req = + new MockHttpTransport() + .createRequestFactory() + .buildPutRequest(new GenericUrl("http://localhost:80980"), new EmptyContent()); + req.getHeaders().setContentLength(0L); + + HttpResponse resp = req.execute(); + resp.getHeaders() + .set("X-Goog-Stored-Content-Length", "5") + .set("x-goog-stored-content-encoding", "identity") + .set("X-GOOG-STORED-SOMETHING", "blah") + .setContentLength(0L); + + StorageException storageException = + JsonResumableSessionFailureScenario.SCENARIO_0.toStorageException( + "uploadId", resp, null, () -> null); + + assertThat(storageException.getCode()).isEqualTo(0); + assertThat(storageException).hasMessageThat().contains("|< x-goog-stored-content-length: 5"); + assertThat(storageException) + .hasMessageThat() + .contains("|< x-goog-stored-content-encoding: identity"); + assertThat(storageException).hasMessageThat().contains("|< x-goog-stored-something: blah"); + } + + private static final class Cause extends RuntimeException { + + private Cause() { + super("Cause"); + } + } + + private static final class Kaboom extends IOException { + + private Kaboom() { + super("Kaboom!!!"); + } + } +} diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/RewindableHttpContentPropertyTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/RewindableHttpContentPropertyTest.java new file mode 100644 index 0000000000..5d23537771 --- /dev/null +++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/RewindableHttpContentPropertyTest.java @@ -0,0 +1,201 @@ +/* + * Copyright 2023 Google LLC + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.google.cloud.storage; + +import static com.google.cloud.storage.TestUtils.xxd; +import static com.google.common.truth.Truth.assertThat; +import static org.junit.Assert.assertThrows; + +import com.google.common.base.MoreObjects; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.OutputStream; +import java.nio.ByteBuffer; +import java.nio.channels.SeekableByteChannel; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.util.Arrays; +import net.jqwik.api.Arbitraries; +import net.jqwik.api.Arbitrary; +import net.jqwik.api.Combinators; +import net.jqwik.api.ForAll; +import net.jqwik.api.Property; +import net.jqwik.api.Provide; +import net.jqwik.api.RandomDistribution; +import org.checkerframework.checker.nullness.qual.NonNull; + +final class RewindableHttpContentPropertyTest { + + @Property + void path(@ForAll("PathScenario") PathScenario pathScenario) throws Exception { + try (PathScenario s = pathScenario) { + RewindableHttpContent content = RewindableHttpContent.of(s.getPath()); + assertThrows( + IOException.class, + () -> { + try (ErroringOutputStream erroringOutputStream = + new ErroringOutputStream(s.getErrorAtOffset())) { + content.writeTo(erroringOutputStream); + } + }); + content.rewindTo(s.getRewindOffset()); + + ByteArrayOutputStream baos = new ByteArrayOutputStream(); + content.writeTo(baos); + + String actual = xxd(baos.toByteArray()); + + assertThat(actual).isEqualTo(s.getExpectedXxd()); + } + } + + @Provide("PathScenario") + static Arbitrary pathScenario() { + return Arbitraries.lazyOf( + () -> + Arbitraries.oneOf( + bytes(1, 10), + bytes(10, 100), + bytes(100, 1_000), + bytes(1_000, 10_000), + bytes(10_000, 100_000), + bytes(100_000, 1_000_000), + bytes(1_000_000, 10_000_000)) + .flatMap( + bytes -> + Combinators.combine( + Arbitraries.integers().between(0, bytes.length - 1), + Arbitraries.integers().between(0, bytes.length - 1), + Arbitraries.just(bytes)) + .as(PathScenario::of))); + } + + @NonNull + private static Arbitrary bytes(int minFileSize, int maxFileSize) { + return Arbitraries.integers() + .between(minFileSize, maxFileSize) + .withDistribution(RandomDistribution.uniform()) + .map(DataGenerator.base64Characters()::genBytes); + } + + private static final class PathScenario implements AutoCloseable { + + private static final Path TMP_DIR = Paths.get(System.getProperty("java.io.tmpdir")); + + private final int rewindOffset; + private final int errorAtOffset; + private final TmpFile tmpFile; + private final byte[] expectedBytes; + private final String expectedXxd; + + private PathScenario( + int rewindOffset, int errorAtOffset, TmpFile tmpFile, byte[] expectedBytes) { + this.rewindOffset = rewindOffset; + this.errorAtOffset = errorAtOffset; + this.tmpFile = tmpFile; + this.expectedBytes = expectedBytes; + this.expectedXxd = xxd(expectedBytes); + } + + public int getRewindOffset() { + return rewindOffset; + } + + public int getErrorAtOffset() { + return errorAtOffset; + } + + public Path getPath() { + return tmpFile.getPath(); + } + + public String getExpectedXxd() { + return expectedXxd; + } + + public long getFullLength() throws IOException { + return Files.size(tmpFile.getPath()); + } + + @Override + public void close() throws IOException { + tmpFile.close(); + } + + @Override + public String toString() { + return MoreObjects.toStringHelper(this) + .add("expectedXxd", "\n" + expectedXxd) + .add("expectedBytes.length", expectedBytes.length) + .add("rewindOffset", rewindOffset) + .add("errorAtOffset", errorAtOffset) + .add("tmpFile", tmpFile) + .toString(); + } + + private static PathScenario of(int rewindOffset, int errorAtOffset, byte[] bytes) { + try { + TmpFile tmpFile1 = TmpFile.of(TMP_DIR, "PathScenario", ".bin"); + try (SeekableByteChannel writer = tmpFile1.writer()) { + writer.write(ByteBuffer.wrap(bytes)); + } + byte[] expectedBytes = + Arrays.copyOfRange(bytes, Math.min(rewindOffset, bytes.length), bytes.length); + return new PathScenario(rewindOffset, errorAtOffset, tmpFile1, expectedBytes); + } catch (IOException e) { + throw new RuntimeException(e); + } + } + } + + static final class ErroringOutputStream extends OutputStream { + private final long errorAt; + private long totalWritten; + + ErroringOutputStream(long errorAt) { + this.errorAt = errorAt; + this.totalWritten = 0; + } + + @Override + public void write(int b) throws IOException { + if (totalWritten++ >= errorAt) { + throw new IOException("Reached errorAt limit"); + } + } + + @Override + public void write(byte[] b) throws IOException { + if (totalWritten + b.length >= errorAt) { + throw new IOException("Reached errorAt limit"); + } else { + totalWritten += b.length; + } + } + + @Override + public void write(byte[] b, int off, int len) throws IOException { + int diff = len - off; + if (totalWritten + diff >= errorAt) { + throw new IOException("Reached errorAt limit"); + } else { + totalWritten += diff; + } + } + } +} diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageImplMockitoTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageImplMockitoTest.java index ce750154e3..8395c6d0a5 100644 --- a/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageImplMockitoTest.java +++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageImplMockitoTest.java @@ -983,17 +983,6 @@ private BlobInfo initializeUpload( return blobInfo; } - @Test - public void testCreateFromFile() throws Exception { - byte[] dataToSend = {1, 2, 3, 4}; - Path tempFile = Files.createTempFile("testCreateFrom", ".tmp"); - Files.write(tempFile, dataToSend); - - BlobInfo blobInfo = initializeUpload(dataToSend); - Blob blob = storage.createFrom(blobInfo, tempFile); - assertEquals(expectedUpdated, blob); - } - @Test public void testCreateFromStream() throws Exception { byte[] dataToSend = {1, 2, 3, 4, 5}; diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/conformance/retry/Functions.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/conformance/retry/Functions.java index 46d5643e0b..47ab7a8251 100644 --- a/google-cloud-storage/src/test/java/com/google/cloud/storage/conformance/retry/Functions.java +++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/conformance/retry/Functions.java @@ -37,6 +37,10 @@ default CtxFunction andThen(CtxFunction f) { return (Ctx ctx, TestRetryConformance trc) -> f.apply(apply(ctx, trc), trc); } + default CtxFunction compose(CtxFunction f) { + return (Ctx ctx, TestRetryConformance trc) -> apply(f.apply(ctx, trc), trc); + } + static CtxFunction identity() { return (ctx, c) -> ctx; } diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/conformance/retry/ITRetryConformanceTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/conformance/retry/ITRetryConformanceTest.java index 8aa55955a2..d42755ef07 100644 --- a/google-cloud-storage/src/test/java/com/google/cloud/storage/conformance/retry/ITRetryConformanceTest.java +++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/conformance/retry/ITRetryConformanceTest.java @@ -60,6 +60,7 @@ import java.util.Random; import java.util.Set; import java.util.function.BiPredicate; +import java.util.function.Predicate; import java.util.logging.Logger; import java.util.stream.Collectors; import org.junit.After; @@ -209,7 +210,7 @@ private static CtxFunction getReplaceStorageInObjectsFromCtx() { * each defined scenario from google-cloud-conformance-tests and our defined {@link * RpcMethodMappings}. */ - private static final class RetryTestCaseResolver { + static final class RetryTestCaseResolver { private static final String HEX_SHUFFLE_SEED_OVERRIDE = System.getProperty("HEX_SHUFFLE_SEED_OVERRIDE"); @@ -220,7 +221,7 @@ private static final class RetryTestCaseResolver { private final String host; private final String projectId; - RetryTestCaseResolver( + private RetryTestCaseResolver( String retryTestsJsonResourcePath, RpcMethodMappings mappings, BiPredicate testAllowFilter, @@ -383,8 +384,12 @@ static BiPredicate specificMappings(int... mapp return (m, c) -> set.contains(c.getMappingId()); } - static BiPredicate instructionsAre(String... instructions) { - return (m, trc) -> + static BiPredicate lift(Predicate p) { + return (m, trc) -> p.test(trc); + } + + static Predicate instructionsAre(String... instructions) { + return trc -> trc.getInstruction().getInstructionsList().equals(ImmutableList.copyOf(instructions)); } diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/conformance/retry/RpcMethodMappings.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/conformance/retry/RpcMethodMappings.java index 67b8159b44..7a87f33048 100644 --- a/google-cloud-storage/src/test/java/com/google/cloud/storage/conformance/retry/RpcMethodMappings.java +++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/conformance/retry/RpcMethodMappings.java @@ -20,10 +20,12 @@ import static com.google.cloud.storage.conformance.retry.CtxFunctions.ResourceSetup.notificationSetup; import static com.google.cloud.storage.conformance.retry.CtxFunctions.ResourceSetup.pubsubTopicSetup; import static com.google.cloud.storage.conformance.retry.CtxFunctions.ResourceSetup.serviceAccount; +import static com.google.cloud.storage.conformance.retry.ITRetryConformanceTest.RetryTestCaseResolver.instructionsAre; import static com.google.common.base.Predicates.and; import static com.google.common.base.Predicates.not; import static com.google.common.truth.Truth.assertThat; import static org.junit.Assert.assertTrue; +import static org.junit.Assume.assumeFalse; import com.google.cloud.BaseServiceException; import com.google.cloud.Binding; @@ -54,6 +56,7 @@ import com.google.cloud.storage.conformance.retry.CtxFunctions.Local; import com.google.cloud.storage.conformance.retry.CtxFunctions.ResourceSetup; import com.google.cloud.storage.conformance.retry.CtxFunctions.Rpc; +import com.google.cloud.storage.conformance.retry.Functions.CtxFunction; import com.google.cloud.storage.conformance.retry.Functions.EConsumer; import com.google.cloud.storage.conformance.retry.RpcMethod.storage.bucket_acl; import com.google.cloud.storage.conformance.retry.RpcMethod.storage.buckets; @@ -124,6 +127,11 @@ final class RpcMethodMappings { static final int _2MiB = 2 * 1024 * 1024; private static final ImmutableMap MODIFY = ImmutableMap.of("a", "b"); + private static final CtxFunction skipUntil2114Fixed = + temporarilySkipMapping( + "Skipped until https://github.com/googleapis/java-storage/issues/2114 is fixed", + instructionsAre("return-503-after-8192K", "return-408") + .or(instructionsAre("return-503-after-256K"))); final Multimap funcMap; RpcMethodMappings() { @@ -1526,7 +1534,10 @@ private static void insert(ArrayList a) { a.add( RpcMethodMapping.newBuilder(50, objects.insert) .withApplicable(TestRetryConformance::isPreconditionsProvided) - .withSetup(defaultSetup.andThen(Local.blobInfoWithGenerationZero)) + .withSetup( + defaultSetup + .andThen(Local.blobInfoWithGenerationZero) + .compose(skipUntil2114Fixed)) .withTest( (ctx, c) -> ctx.map( @@ -1541,7 +1552,10 @@ private static void insert(ArrayList a) { a.add( RpcMethodMapping.newBuilder(51, objects.insert) .withApplicable(TestRetryConformance::isPreconditionsProvided) - .withSetup(defaultSetup.andThen(Local.blobInfoWithGenerationZero)) + .withSetup( + defaultSetup + .andThen(Local.blobInfoWithGenerationZero) + .compose(skipUntil2114Fixed)) .withTest( (ctx, c) -> ctx.map( @@ -2099,4 +2113,12 @@ private static void put(ArrayList a) {} private static Predicate methodGroupIs(String s) { return (c) -> s.equals(c.getMethod().getGroup()); } + + private static CtxFunction temporarilySkipMapping( + String message, java.util.function.Predicate p) { + return (ctx, trc) -> { + assumeFalse(message, p.test(trc)); + return ctx; + }; + } } diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITObjectChecksumSupportTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITObjectChecksumSupportTest.java index 2bcb78e640..d8e026cbc0 100644 --- a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITObjectChecksumSupportTest.java +++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITObjectChecksumSupportTest.java @@ -28,6 +28,7 @@ import com.google.cloud.storage.Storage; import com.google.cloud.storage.Storage.BlobWriteOption; import com.google.cloud.storage.StorageException; +import com.google.cloud.storage.TmpFile; import com.google.cloud.storage.TransportCompatibility.Transport; import com.google.cloud.storage.it.ITObjectChecksumSupportTest.ChecksummedTestContentProvider; import com.google.cloud.storage.it.runner.StorageITRunner; @@ -42,18 +43,24 @@ import com.google.common.io.ByteStreams; import java.io.ByteArrayInputStream; import java.io.IOException; +import java.nio.ByteBuffer; import java.nio.channels.Channels; import java.nio.channels.ReadableByteChannel; +import java.nio.channels.SeekableByteChannel; +import java.nio.file.Path; +import java.nio.file.Paths; import org.junit.Test; import org.junit.runner.RunWith; @RunWith(StorageITRunner.class) @CrossRun( - transports = {Transport.HTTP, Transport.GRPC}, + transports = {Transport.HTTP /*, Transport.GRPC*/}, backends = Backend.PROD) @Parameterized(ChecksummedTestContentProvider.class) public final class ITObjectChecksumSupportTest { + private static final Path tmpDir = Paths.get(System.getProperty("java.io.tmpdir")); + @Inject public Generator generator; @Inject public Storage storage; @@ -116,6 +123,50 @@ public void testCrc32cValidated_createFrom_expectSuccess() throws IOException { assertThat(blob.getCrc32c()).isEqualTo(content.getCrc32cBase64()); } + @Test + public void testCrc32cValidated_createFrom_path_expectFailure() throws IOException { + String blobName = generator.randomObjectName(); + BlobId blobId = BlobId.of(bucket.getName(), blobName); + BlobInfo blobInfo = BlobInfo.newBuilder(blobId).setCrc32c(content.getCrc32cBase64()).build(); + + try (TmpFile tmpFile = TmpFile.of(tmpDir, "prefix", "bin")) { + try (SeekableByteChannel writer = tmpFile.writer()) { + writer.write(ByteBuffer.wrap(content.concat('x'))); + } + StorageException expected = + assertThrows( + StorageException.class, + () -> + storage.createFrom( + blobInfo, + tmpFile.getPath(), + BlobWriteOption.doesNotExist(), + BlobWriteOption.crc32cMatch())); + assertThat(expected.getCode()).isEqualTo(400); + } + } + + @Test + public void testCrc32cValidated_createFrom_path_expectSuccess() throws IOException { + String blobName = generator.randomObjectName(); + BlobId blobId = BlobId.of(bucket.getName(), blobName); + BlobInfo blobInfo = BlobInfo.newBuilder(blobId).setCrc32c(content.getCrc32cBase64()).build(); + + try (TmpFile tmpFile = TmpFile.of(tmpDir, "prefix", "bin")) { + try (SeekableByteChannel writer = tmpFile.writer()) { + writer.write(ByteBuffer.wrap(content.getBytes())); + } + + Blob blob = + storage.createFrom( + blobInfo, + tmpFile.getPath(), + BlobWriteOption.doesNotExist(), + BlobWriteOption.crc32cMatch()); + assertThat(blob.getCrc32c()).isEqualTo(content.getCrc32cBase64()); + } + } + @Test public void testCrc32cValidated_writer_expectFailure() { String blobName = generator.randomObjectName();