diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java index 4f33b7c6c3f1..7d63ba367131 100644 --- a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java +++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java @@ -27,6 +27,7 @@ import static com.google.cloud.storage.Utils.ifNonNull; import static com.google.cloud.storage.Utils.projectNameCodec; import static com.google.common.base.MoreObjects.firstNonNull; +import static com.google.common.base.Preconditions.checkArgument; import static java.util.Objects.requireNonNull; import com.google.api.core.ApiFuture; @@ -55,6 +56,7 @@ import com.google.cloud.storage.HmacKey.HmacKeyState; import com.google.cloud.storage.PostPolicyV4.PostConditionsV4; import com.google.cloud.storage.PostPolicyV4.PostFieldsV4; +import com.google.cloud.storage.ReadProjectionConfig.ProjectionType; import com.google.cloud.storage.Retrying.Retrier; import com.google.cloud.storage.Retrying.RetrierWithAlg; import com.google.cloud.storage.Storage.ComposeRequest.SourceBlob; @@ -1468,6 +1470,26 @@ public Blob moveBlob(MoveBlobRequest request) { @Override public ApiFuture blobReadSession(BlobId id, BlobSourceOption... options) { + return blobReadSession(id, ImmutableList.of(), options); + } + + @Override + public ApiFuture blobReadSession( + BlobId id, + @Nullable ReadProjectionConfig fastOpenConfig, + BlobSourceOption... options) { + if (fastOpenConfig == null) { + return blobReadSession(id, ImmutableList.of(), options); + } + checkArgument( + fastOpenConfig.getType() == ProjectionType.STREAM_READ, + "fastOpenConfig must be asFutureBytes(), asFutureByteString() or asChannel(), but was %s", + fastOpenConfig); + return blobReadSession(id, ImmutableList.of(fastOpenConfig), options); + } + + private ApiFuture blobReadSession( + BlobId id, List> fastOpenConfigs, BlobSourceOption... options) { Opts opts = Opts.unwrap(options); Object object = codecs.blobId().encode(id); @@ -1484,7 +1506,8 @@ public ApiFuture blobReadSession(BlobId id, BlobSourceOption... BidiReadObjectRequest req = b.build(); GrpcCallContext context = opts.grpcMetadataMapper().apply(GrpcCallContext.createDefault()); - ApiFuture session = storageDataClient.readSession(req, context); + ApiFuture session = + storageDataClient.readSession(req, context, fastOpenConfigs); return BlobReadSessionAdapter.wrap(session); } diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/OtelStorageDecorator.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/OtelStorageDecorator.java index 9095a8a22cd7..3546027c2fdd 100644 --- a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/OtelStorageDecorator.java +++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/OtelStorageDecorator.java @@ -1476,6 +1476,14 @@ public Blob moveBlob(MoveBlobRequest request) { @Override public ApiFuture blobReadSession(BlobId id, BlobSourceOption... options) { + return blobReadSession(id, null, options); + } + + @Override + public ApiFuture blobReadSession( + BlobId id, + @Nullable ReadProjectionConfig fastOpenConfig, + BlobSourceOption... options) { Span blobReadSessionSpan = tracer .spanBuilder(BLOB_READ_SESSION) @@ -1484,7 +1492,8 @@ public ApiFuture blobReadSession(BlobId id, BlobSourceOption... try (Scope ignore1 = blobReadSessionSpan.makeCurrent()) { Context blobReadSessionContext = Context.current(); Span ready = tracer.spanBuilder(BLOB_READ_SESSION + "/ready").startSpan(); - ApiFuture blobReadSessionApiFuture = delegate.blobReadSession(id, options); + ApiFuture blobReadSessionApiFuture = + delegate.blobReadSession(id, fastOpenConfig, options); ApiFuture futureDecorated = ApiFutures.transform( blobReadSessionApiFuture, diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/Storage.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/Storage.java index 9d71d8f73711..3088bcb3afc0 100644 --- a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/Storage.java +++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/Storage.java @@ -5955,6 +5955,44 @@ default ApiFuture blobReadSession(BlobId id, BlobSourceOption.. return throwGrpcOnly(fmtMethodName("blobReadSession", BlobId.class, BlobSourceOption.class)); } + /** + * Asynchronously set up a new {@link BlobReadSession} for the specified {@link BlobId} and {@code + * options}, sending the read described by {@code fastOpenConfig} in the same request that opens + * the session ("fast open"). This saves the round trip a separate first read would take. + * + *

The first {@link BlobReadSession#readAs(ReadProjectionConfig)} with a config equal to {@code + * fastOpenConfig} is given that read; any other {@code readAs} sends a new read. If the fast open + * read fails, the error is reported by the returned future or by that first {@code readAs}. + * + *

Example of reading the first 1 KiB of an object with fast open:

+ * + * 
{@code
+   * ReadAsFutureBytes first1KiB =
+   *     ReadProjectionConfigs.asFutureBytes().withRangeSpec(RangeSpec.of(0, 1024));
+   * try (BlobReadSession session =
+   *     storage.blobReadSession(blobId, first1KiB).get(10, TimeUnit.SECONDS)) {
+   *   byte[] bytes = session.readAs(first1KiB).get(10, TimeUnit.SECONDS);
+   * }
+   * }
+ * + * @param id the blob to read from + * @param fastOpenConfig the read to send with the open, or {@code null} to only open the session. + * Must be {@link ReadProjectionConfigs#asFutureBytes()}, {@link + * ReadProjectionConfigs#asFutureByteString()} or {@link ReadProjectionConfigs#asChannel()}, + * optionally with a range. + * @since 2.77.0 This new api is in preview and is subject to breaking changes. + */ + @BetaApi + @TransportCompatibility({Transport.GRPC}) + default ApiFuture blobReadSession( + BlobId id, + @Nullable ReadProjectionConfig fastOpenConfig, + BlobSourceOption... options) { + return throwGrpcOnly( + fmtMethodName( + "blobReadSession", BlobId.class, ReadProjectionConfig.class, BlobSourceOption.class)); + } + /** * Create a new {@link BlobAppendableUpload} for the specified {@code blobInfo} and {@code * options}. diff --git a/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/ITFastOpenFakeTest.java b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/ITFastOpenFakeTest.java index 73ac5780bf67..694ce957e903 100644 --- a/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/ITFastOpenFakeTest.java +++ b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/ITFastOpenFakeTest.java @@ -32,7 +32,9 @@ import com.google.api.gax.rpc.OutOfRangeException; import com.google.cloud.storage.ITObjectReadSessionFakeTest.FakeStorage; import com.google.cloud.storage.ReadProjectionConfigs.BaseConfig; +import com.google.cloud.storage.Storage.BlobSourceOption; import com.google.cloud.storage.it.ChecksummedTestContent; +import com.google.cloud.storage.otel.TestExporter; import com.google.common.base.Throwables; import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableMap; @@ -48,6 +50,11 @@ import io.grpc.Metadata; import io.grpc.Status; import io.grpc.stub.StreamObserver; +import io.opentelemetry.api.trace.StatusCode; +import io.opentelemetry.sdk.OpenTelemetrySdk; +import io.opentelemetry.sdk.trace.SdkTracerProvider; +import io.opentelemetry.sdk.trace.data.SpanData; +import io.opentelemetry.sdk.trace.export.SimpleSpanProcessor; import java.io.IOException; import java.io.UncheckedIOException; import java.util.ArrayList; @@ -72,6 +79,7 @@ public final class ITFastOpenFakeTest { .setSize(_2MiB) .build(); private static final GrpcCallContext CTX = GrpcCallContext.createDefault(); + private static final BlobId BLOB_ID = BlobId.of("b", "o"); private static final BidiReadObjectRequest OPEN = BidiReadObjectRequest.newBuilder() @@ -364,36 +372,7 @@ public void open_whenTheServerFailsTheStreamWithOutOfRange_failsWithoutRetrying( @Test public void open_whenTheServerRejectsOnlyTheFastOpenRange_reopensWithoutItAndReadAsFails() throws Exception { - // OUT_OF_RANGE that names the failing read, as the server reports a bad range on a bidi read - BidiReadObjectError rangeError = - BidiReadObjectError.newBuilder() - .addReadRangeErrors( - ReadRangeError.newBuilder() - .setReadId(1) - .setStatus( - com.google.rpc.Status.newBuilder() - .setCode(com.google.rpc.Code.OUT_OF_RANGE_VALUE))) - .build(); - Metadata trailers = new Metadata(); - trailers.put( - GRPC_STATUS_DETAILS_KEY, - com.google.rpc.Status.newBuilder() - .setCode(com.google.rpc.Code.OUT_OF_RANGE_VALUE) - .addDetails(Any.pack(rangeError)) - .build()); - FakeStorage server = - FakeStorage.of( - ImmutableMap.of( - OPEN_WITH_READ_OF_10_TO_20, - client -> { - requestsReceived.add(OPEN_WITH_READ_OF_10_TO_20); - client.onError(Status.OUT_OF_RANGE.asRuntimeException(trailers)); - }, - OPEN, - client -> { - requestsReceived.add(OPEN); - client.onNext(metadataOnly()); - })); + FakeStorage server = rejectsOnlyTheFastOpenRange(); try (FakeServer fakeServer = FakeServer.of(server); Storage storage = fakeServer.getGrpcStorageOptions().getService()) { @@ -493,6 +472,168 @@ public void open_whenTheConnectionDropsHalfway_fetchesOnlyTheRestOfTheRange() th .inOrder(); } + @Test + public void blobReadSession_withAFastOpenConfig_doesNotSendAnotherRequest() throws Exception { + FakeStorage server = + fakeServer(ImmutableMap.of(OPEN_WITH_READ_OF_10_TO_20, metadataAndBytes(1, 10, 10))); + + try (FakeServer fakeServer = FakeServer.of(server); + Storage storage = fakeServer.getGrpcStorageOptions().getService()) { + try (BlobReadSession session = + storage.blobReadSession(BLOB_ID, READ_10_TO_20).get(5, SECONDS)) { + byte[] actual = session.readAs(READ_10_TO_20).get(2, SECONDS); + assertThat(xxd(actual)).isEqualTo(xxd(slice(10, 10))); + } + } + + assertThat(requestsReceived).containsExactly(OPEN_WITH_READ_OF_10_TO_20); + } + + @Test + public void blobReadSession_withTracing_doesNotSendAnotherRequest_andEndsTheSpans() + throws Exception { + FakeStorage server = + fakeServer(ImmutableMap.of(OPEN_WITH_READ_OF_10_TO_20, metadataAndBytes(1, 10, 10))); + TestExporter exporter = new TestExporter(); + OpenTelemetrySdk otel = + OpenTelemetrySdk.builder() + .setTracerProvider( + SdkTracerProvider.builder() + .addSpanProcessor(SimpleSpanProcessor.create(exporter)) + .build()) + .build(); + + try (FakeServer fakeServer = FakeServer.of(server); + Storage storage = + fakeServer.getGrpcStorageOptions().toBuilder() + .setOpenTelemetry(otel) + .build() + .getService()) { + try (BlobReadSession session = + storage.blobReadSession(BLOB_ID, READ_10_TO_20).get(5, SECONDS)) { + byte[] actual = session.readAs(READ_10_TO_20).get(2, SECONDS); + assertThat(xxd(actual)).isEqualTo(xxd(slice(10, 10))); + } + } + + assertThat(requestsReceived).containsExactly(OPEN_WITH_READ_OF_10_TO_20); + List endedSpans = + exporter.getExportedSpans().stream().map(SpanData::getName).collect(toList()); + assertThat(endedSpans) + .containsAtLeast( + "com.google.cloud.storage.Storage/blobReadSession", + "com.google.cloud.storage.Storage/blobReadSession/readAs"); + } + + @Test + public void blobReadSession_withTracing_whenTheFastOpenReadFails_recordsTheErrorOnReadAs() + throws Exception { + FakeStorage server = rejectsOnlyTheFastOpenRange(); + TestExporter exporter = new TestExporter(); + OpenTelemetrySdk otel = + OpenTelemetrySdk.builder() + .setTracerProvider( + SdkTracerProvider.builder() + .addSpanProcessor(SimpleSpanProcessor.create(exporter)) + .build()) + .build(); + + try (FakeServer fakeServer = FakeServer.of(server); + Storage storage = + fakeServer.getGrpcStorageOptions().toBuilder() + .setOpenTelemetry(otel) + .build() + .getService()) { + try (BlobReadSession session = + storage.blobReadSession(BLOB_ID, READ_10_TO_20).get(10, SECONDS)) { + ApiFuture read = session.readAs(READ_10_TO_20); + assertThrows(ExecutionException.class, () -> read.get(5, SECONDS)); + } + } + + List readAsStatus = + exporter.getExportedSpans().stream() + .filter( + s -> s.getName().equals("com.google.cloud.storage.Storage/blobReadSession/readAs")) + .map(s -> s.getStatus().getStatusCode()) + .collect(toList()); + assertThat(readAsStatus).containsExactly(StatusCode.ERROR); + } + + @Test + public void blobReadSession_withANullFastOpenConfig_isAPlainOpen() throws Exception { + FakeStorage server = fakeServer(ImmutableMap.of(OPEN, metadataOnly())); + + try (FakeServer fakeServer = FakeServer.of(server); + Storage storage = fakeServer.getGrpcStorageOptions().getService()) { + BlobReadSession session = + storage.blobReadSession(BLOB_ID, (ReadAsFutureBytes) null).get(5, SECONDS); + session.close(); + } + + assertThat(requestsReceived).containsExactly(OPEN); + } + + @Test + public void blobReadSession_withASeekableChannelConfig_isRejected() throws Exception { + FakeStorage server = fakeServer(ImmutableMap.of()); + + try (FakeServer fakeServer = FakeServer.of(server); + Storage storage = fakeServer.getGrpcStorageOptions().getService()) { + IllegalArgumentException e = + assertThrows( + IllegalArgumentException.class, + () -> storage.blobReadSession(BLOB_ID, ReadProjectionConfigs.asSeekableChannel())); + assertThat(e).hasMessageThat().startsWith("fastOpenConfig must be"); + } + + assertThat(requestsReceived).isEmpty(); + } + + @Test + public void blobReadSession_whenTheFastOpenReadIsOutOfRange_failsTheOpen() throws Exception { + FakeStorage server = + FakeStorage.of( + ImmutableMap.of( + OPEN_WITH_READ_OF_10_TO_20, + client -> { + requestsReceived.add(OPEN_WITH_READ_OF_10_TO_20); + client.onError(Status.OUT_OF_RANGE.asRuntimeException()); + })); + + try (FakeServer fakeServer = FakeServer.of(server); + Storage storage = fakeServer.getGrpcStorageOptions().getService()) { + ApiFuture open = storage.blobReadSession(BLOB_ID, READ_10_TO_20); + + ExecutionException e = assertThrows(ExecutionException.class, () -> open.get(5, SECONDS)); + assertThat(e).hasCauseThat().isInstanceOf(StorageException.class); + assertThat(((StorageException) e.getCause()).getCode()).isEqualTo(400); + } + + assertThat(requestsReceived).containsExactly(OPEN_WITH_READ_OF_10_TO_20); + } + + @Test + public void blobReadSession_withAFastOpenConfig_stillAppliesBlobSourceOptions() throws Exception { + BidiReadObjectRequest openWithGenerationMatchAndRead = + OPEN_WITH_READ_OF_10_TO_20.toBuilder() + .setReadObjectSpec(OPEN.getReadObjectSpec().toBuilder().setIfGenerationMatch(1)) + .build(); + FakeStorage server = + fakeServer(ImmutableMap.of(openWithGenerationMatchAndRead, metadataAndBytes(1, 10, 10))); + + try (FakeServer fakeServer = FakeServer.of(server); + Storage storage = fakeServer.getGrpcStorageOptions().getService()) { + BlobReadSession session = + storage + .blobReadSession(BLOB_ID, READ_10_TO_20, BlobSourceOption.generationMatch(1)) + .get(5, SECONDS); + session.close(); + } + + assertThat(requestsReceived).containsExactly(openWithGenerationMatchAndRead); + } + /** * A fake server that answers each request in {@code responses} and adds it to {@link * #requestsReceived}. Any other request is answered with an error. @@ -515,6 +656,41 @@ private static BidiReadObjectResponse metadataOnly() { return BidiReadObjectResponse.newBuilder().setMetadata(METADATA).build(); } + /** + * A fake server that fails the open with an OUT_OF_RANGE that names only the fast open read (read + * id 1), as the server reports a bad range on a bidi read, and accepts the plain re-open. + */ + private FakeStorage rejectsOnlyTheFastOpenRange() { + BidiReadObjectError rangeError = + BidiReadObjectError.newBuilder() + .addReadRangeErrors( + ReadRangeError.newBuilder() + .setReadId(1) + .setStatus( + com.google.rpc.Status.newBuilder() + .setCode(com.google.rpc.Code.OUT_OF_RANGE_VALUE))) + .build(); + Metadata trailers = new Metadata(); + trailers.put( + GRPC_STATUS_DETAILS_KEY, + com.google.rpc.Status.newBuilder() + .setCode(com.google.rpc.Code.OUT_OF_RANGE_VALUE) + .addDetails(Any.pack(rangeError)) + .build()); + return FakeStorage.of( + ImmutableMap.of( + OPEN_WITH_READ_OF_10_TO_20, + client -> { + requestsReceived.add(OPEN_WITH_READ_OF_10_TO_20); + client.onError(Status.OUT_OF_RANGE.asRuntimeException(trailers)); + }, + OPEN, + client -> { + requestsReceived.add(OPEN); + client.onNext(metadataOnly()); + })); + } + /** The response to an open request that has a read of {@code [offset, offset + length)}. */ private static BidiReadObjectResponse metadataAndBytes(int readId, int offset, int length) { return BidiReadObjectResponse.newBuilder() diff --git a/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/it/runner/registry/AbstractStorageProxy.java b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/it/runner/registry/AbstractStorageProxy.java index eeae23382551..bb37e88017a4 100644 --- a/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/it/runner/registry/AbstractStorageProxy.java +++ b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/it/runner/registry/AbstractStorageProxy.java @@ -41,6 +41,7 @@ import com.google.cloud.storage.PostPolicyV4; import com.google.cloud.storage.PostPolicyV4.PostConditionsV4; import com.google.cloud.storage.PostPolicyV4.PostFieldsV4; +import com.google.cloud.storage.ReadProjectionConfig; import com.google.cloud.storage.ServiceAccount; import com.google.cloud.storage.Storage; import com.google.cloud.storage.StorageBatch; @@ -498,6 +499,12 @@ public ApiFuture blobReadSession(BlobId id, BlobSourceOption... return delegate.blobReadSession(id, options); } + @Override + public ApiFuture blobReadSession( + BlobId id, ReadProjectionConfig fastOpenConfig, BlobSourceOption... options) { + return delegate.blobReadSession(id, fastOpenConfig, options); + } + @Override public BlobAppendableUpload blobAppendableUpload( BlobInfo blobInfo, BlobAppendableUploadConfig uploadConfig, BlobWriteOption... options) {