diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BaseStorageWriteChannel.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BaseStorageWriteChannel.java index 4751f671101b..0c4394da3914 100644 --- a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BaseStorageWriteChannel.java +++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BaseStorageWriteChannel.java @@ -138,6 +138,9 @@ public final int write(ByteBuffer src) throws IOException { return 0; } int write = tmp.write(src); + if (write > 0 && metricsContext != null) { + metricsContext.addRequestBytes(write); + } return write; } catch (StorageException e) { throw new IOException(e); 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 39152543420e..013c7d1b3ef3 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 @@ -253,7 +253,11 @@ public Blob create(BlobInfo blobInfo, byte[] content, BlobTargetOption... option public Blob create( BlobInfo blobInfo, byte[] content, int offset, int length, BlobTargetOption... options) { Opts opts = Opts.unwrap(options).resolveFrom(blobInfo); - return internalDirectUpload(blobInfo, opts, ByteBuffer.wrap(content, offset, length)) + return internalDirectUpload( + blobInfo, + opts, + ByteBuffer.wrap(content, offset, length), + getOptions().createMetricsContext("WriteObject")) .asBlob(this); } @@ -284,7 +288,10 @@ public Blob create(BlobInfo blobInfo, InputStream content, BlobWriteOption... op .build(); try (UnbufferedWritableByteChannel c = session.open()) { - ByteStreams.copy(Channels.newChannel(inputStreamParam), c); + long copied = ByteStreams.copy(Channels.newChannel(inputStreamParam), c); + if (metricsContext != null) { + metricsContext.setRequestBytes(copied); + } } ApiFuture responseApiFuture = session.getResult(); return this.getBlob(responseApiFuture); @@ -317,6 +324,9 @@ public Blob internalCreateFrom(Path path, BlobInfo info, Opts o StorageMetricsContext multiContext = getOptions().createMultiStepMetricsContext("WriteObject"); Throwable error = null; try { + if (multiContext != null) { + multiContext.setRequestBytes(Files.size(path)); + } GrpcCallContext grpcCallContext = opts.grpcMetadataMapper().apply(GrpcCallContext.createDefault()); WriteObjectRequest req = getWriteObjectRequest(info, opts); @@ -400,7 +410,10 @@ public Blob createFrom( ReadableByteChannel src = Channels.newChannel(firstNonNull(in, new ByteArrayInputStream(ZERO_BYTES))); try (BufferedWritableByteChannel dst = session.open()) { - ByteStreams.copy(src, dst); + long copied = ByteStreams.copy(src, dst); + if (multiContext != null) { + multiContext.setRequestBytes(copied); + } } catch (Exception e) { throw StorageException.coalesce(e); } @@ -930,6 +943,14 @@ public GrpcBlobWriteChannel writer(BlobInfo blobInfo, BlobWriteOption... options @Override public BlobInfo internalDirectUpload( BlobInfo blobInfo, Opts opts, ByteBuffer buf) { + return internalDirectUpload(blobInfo, opts, buf, null); + } + + private BlobInfo internalDirectUpload( + BlobInfo blobInfo, + Opts opts, + ByteBuffer buf, + @Nullable StorageMetricsContext metricsContext) { requireNonNull(blobInfo, "blobInfo must be non null"); requireNonNull(buf, "content must be non null"); Opts optsWithDefaults = opts.prepend(defaultOpts); @@ -937,7 +958,10 @@ public BlobInfo internalDirectUpload( optsWithDefaults.grpcMetadataMapper().apply(GrpcCallContext.createDefault()); WriteObjectRequest req = getWriteObjectRequest(blobInfo, optsWithDefaults); Hasher hasher = opts.getHasher(); - StorageMetricsContext metricsContext = getOptions().createMetricsContext("WriteObject"); + int totalBytes = buf.remaining(); + if (metricsContext != null) { + metricsContext.setRequestBytes(totalBytes); + } GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext(metricsContext)); RewindableContent content = RewindableContent.of(buf); return retrier.run( diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageImpl.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageImpl.java index cba4437e6e04..ce7310f6751b 100644 --- a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageImpl.java +++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageImpl.java @@ -64,6 +64,7 @@ import com.google.common.collect.Lists; import com.google.common.collect.Maps; import com.google.common.io.BaseEncoding; +import com.google.common.io.CountingInputStream; import com.google.common.io.CountingOutputStream; import java.io.ByteArrayInputStream; import java.io.IOException; @@ -157,21 +158,32 @@ public Bucket create(BucketInfo bucketInfo, BucketTargetOption... options) { @Override public Blob create(BlobInfo blobInfo, BlobTargetOption... options) { Opts opts = Opts.unwrap(options).resolveFrom(blobInfo); - return internalDirectUpload(blobInfo, opts, Buffers.allocate(0)).asBlob(this); + return internalDirectUpload( + blobInfo, opts, Buffers.allocate(0), getOptions().createMetricsContext("WriteObject")) + .asBlob(this); } @Override public Blob create(BlobInfo blobInfo, byte[] content, BlobTargetOption... options) { content = firstNonNull(content, EMPTY_BYTE_ARRAY); Opts opts = Opts.unwrap(options).resolveFrom(blobInfo); - return internalDirectUpload(blobInfo, opts, ByteBuffer.wrap(content)).asBlob(this); + return internalDirectUpload( + blobInfo, + opts, + ByteBuffer.wrap(content), + getOptions().createMetricsContext("WriteObject")) + .asBlob(this); } @Override public Blob create( BlobInfo blobInfo, byte[] content, int offset, int length, BlobTargetOption... options) { Opts opts = Opts.unwrap(options).resolveFrom(blobInfo); - return internalDirectUpload(blobInfo, opts, ByteBuffer.wrap(content, offset, length)) + return internalDirectUpload( + blobInfo, + opts, + ByteBuffer.wrap(content, offset, length), + getOptions().createMetricsContext("WriteObject")) .asBlob(this); } @@ -185,12 +197,23 @@ public Blob create(BlobInfo blobInfo, InputStream content, BlobWriteOption... op StorageObject blobPb = codecs.blobInfo().encode(updated); InputStream inputStreamParam = firstNonNull(content, new ByteArrayInputStream(EMPTY_BYTE_ARRAY)); + StorageMetricsContext metricsContext = getOptions().createMetricsContext("WriteObject"); // retries are not safe when the input is an InputStream, so we can't retry. - return run( - "WriteObject", + return retrier.run( Retrying.neverRetry(), - () -> storageRpc.create(blobPb, inputStreamParam, optionsMap), - (x) -> Conversions.json().blobInfo().decode(x).asBlob(this)); + () -> { + if (metricsContext == null) { + return storageRpc.create(blobPb, inputStreamParam, optionsMap); + } + CountingInputStream countingStream = new CountingInputStream(inputStreamParam); + try { + return storageRpc.create(blobPb, countingStream, optionsMap); + } finally { + metricsContext.setRequestBytes(countingStream.getCount()); + } + }, + (x) -> Conversions.json().blobInfo().decode(x).asBlob(this), + metricsContext); } @Override @@ -212,6 +235,9 @@ public Blob createFrom(BlobInfo blobInfo, Path path, int bufferSize, BlobWriteOp StorageMetricsContext multiContext = getOptions().createMultiStepMetricsContext("WriteObject"); Throwable error = null; try { + if (multiContext != null) { + multiContext.setRequestBytes(size); + } Opts opts = Opts.unwrap(options).resolveFrom(blobInfo); final Map optionsMap = opts.getRpcOptions(); BlobInfo.Builder builder = blobInfo.toBuilder().setMd5(null).setCrc32c(null); @@ -1879,48 +1905,74 @@ public BlobInfo internalCreateFrom(Path path, BlobInfo info, Opts optionsMap = opts.getRpcOptions(); - BlobInfo.Builder builder = info.toBuilder().setMd5(null).setCrc32c(null); - BlobInfo updated = opts.blobInfoMapper().apply(builder).build(); - StorageObject encode = codecs.blobInfo().encode(updated); + StorageMetricsContext multiContext = getOptions().createMultiStepMetricsContext("WriteObject"); + Throwable error = null; + try { + if (multiContext != null) { + multiContext.setRequestBytes(size); + } + final Map optionsMap = opts.getRpcOptions(); + BlobInfo.Builder builder = info.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, + retrier.withAlg( + retryAlgorithmManager.getForResumableUploadSessionCreate(optionsMap), + multiContext)); + JsonResumableWrite jsonResumableWrite = + JsonResumableWrite.of( + encode, + optionsMap, + uploadIdSupplier.get(), + 0, + opts.getHasher(), + opts.getHasher().initialValue()); - Supplier uploadIdSupplier = - ResumableMedia.startUploadForBlobInfo( - getOptions(), - updated, - optionsMap, - retrier.withAlg(retryAlgorithmManager.getForResumableUploadSessionCreate(optionsMap))); - JsonResumableWrite jsonResumableWrite = - JsonResumableWrite.of( - encode, - optionsMap, - uploadIdSupplier.get(), - 0, - opts.getHasher(), - opts.getHasher().initialValue()); - - JsonResumableSession session = - ResumableSession.json( - HttpClientContext.from(storageRpc), - retrier.withAlg(retryAlgorithmManager.idempotent()), - jsonResumableWrite); - HttpContentRange contentRange = HttpContentRange.of(ByteRangeSpec.explicit(0L, size), size); - ResumableOperationResult put = - session.put(RewindableContent.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 query = session.query(); - object = query.getObject(); + JsonResumableSession session = + ResumableSession.json( + HttpClientContext.from(storageRpc), + retrier.withAlg(retryAlgorithmManager.idempotent(), multiContext), + jsonResumableWrite); + HttpContentRange contentRange = HttpContentRange.of(ByteRangeSpec.explicit(0L, size), size); + ResumableOperationResult put = + session.put(RewindableContent.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 query = session.query(); + object = query.getObject(); + } + return codecs.blobInfo().decode(object); + } catch (Throwable t) { + error = t; + throw t; + } finally { + if (multiContext != null) { + multiContext.recordOperation(getOptions().getStorageClientMetrics(), error); + } } - return codecs.blobInfo().decode(object); } @Override public BlobInfo internalDirectUpload(BlobInfo info, Opts opts, ByteBuffer buf) { + return internalDirectUpload(info, opts, buf, null); + } + + private BlobInfo internalDirectUpload( + BlobInfo info, + Opts opts, + ByteBuffer buf, + @Nullable StorageMetricsContext metricsContext) { + int totalBytes = buf.remaining(); BlobInfo.Builder builder = opts.blobInfoMapper().apply(info.toBuilder().clearMd5().clearCrc32c()); @Nullable Crc32cLengthKnown hash = opts.getHasher().hash(buf.duplicate()); @@ -1934,14 +1986,17 @@ public BlobInfo internalDirectUpload(BlobInfo info, Opts opts, ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForObjectsCreate(encoded, optionsMap); RewindableContent content = RewindableContent.of(buf); - return run( - "WriteObject", + if (metricsContext != null) { + metricsContext.setRequestBytes(totalBytes); + } + return retrier.run( algorithm, () -> { content.rewindTo(0); return storageRpc.create(encoded, new RewindableContentInputStream(content), optionsMap); }, - Conversions.json().blobInfo()); + Conversions.json().blobInfo(), + metricsContext); } /** diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageMetricsContext.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageMetricsContext.java index 9f0b046507f2..b02c3f43035f 100644 --- a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageMetricsContext.java +++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageMetricsContext.java @@ -20,6 +20,7 @@ import com.google.api.gax.grpc.GrpcCallContext; import com.google.api.gax.rpc.ApiCallContext; +import com.google.common.collect.ImmutableSet; import io.grpc.CallOptions; import io.opentelemetry.api.common.AttributeKey; import io.opentelemetry.api.common.Attributes; @@ -44,6 +45,18 @@ final class StorageMetricsContext { ApiCallContext.Key.create("storage-metrics-context"); static final String HTTP_REQUEST_PROPERTY_KEY = "com.google.cloud.storage.StorageMetricsContext"; + private static final ImmutableSet> BODY_SIZE_EXCLUDED_KEYS = + ImmutableSet.of( + StorageClientMetrics.KEY_NETWORK_PATH, + StorageClientMetrics.KEY_RPC_RESPONSE_STATUS_CODE, + StorageClientMetrics.KEY_HTTP_RESPONSE_STATUS_CODE, + StorageClientMetrics.KEY_HTTP_REQUEST_METHOD, + StorageClientMetrics.KEY_URL_TEMPLATE, + StorageClientMetrics.KEY_ERROR_TYPE, + StorageClientMetrics.KEY_GCP_STORAGE_OPERATION, + StorageClientMetrics.KEY_GCP_CLIENT_METHOD, + StorageClientMetrics.KEY_RPC_METHOD); + private final String operationName; private final long startTimeNanos; private final Context otelContext; @@ -182,9 +195,33 @@ void recordOperation( .build(); metrics.getGcpClientRequestDuration().record(getElapsedSeconds(), durationAttributes); } + + long reqBytes = requestBytes.get(); + if ("WriteObject".equals(operationName) + && reqBytes > 0 + && metrics.getRequestBodySize() != null) { + metrics + .getRequestBodySize() + .record(reqBytes, buildBodySizeAttributes(metrics), otelContext); + } } } + private Attributes buildBodySizeAttributes(StorageClientMetrics metrics) { + AttributesBuilder bodyAttrBuilder = Attributes.builder(); + bodyAttrBuilder.putAll(metrics.getClientAttributes()); + baseAttributes.forEach( + (key, val) -> { + if (!BODY_SIZE_EXCLUDED_KEYS.contains(key)) { + @SuppressWarnings("unchecked") + AttributeKey objKey = (AttributeKey) key; + bodyAttrBuilder.put(objKey, val); + } + }); + bodyAttrBuilder.put(StorageClientMetrics.KEY_RPC_METHOD, operationName); + return bodyAttrBuilder.build(); + } + void recordAttemptSuccess(@Nullable StorageClientMetrics metrics) { if (metrics == null || metrics.getAttempts() == null) { return; @@ -394,6 +431,10 @@ long addRequestBytes(long bytes) { return requestBytes.addAndGet(bytes); } + void setRequestBytes(long bytes) { + requestBytes.set(bytes); + } + long getRequestBytes() { return requestBytes.get(); } diff --git a/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageMetricsContextTest.java b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageMetricsContextTest.java index e8beec487a8e..ff79f327bd11 100644 --- a/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageMetricsContextTest.java +++ b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageMetricsContextTest.java @@ -18,32 +18,77 @@ import static com.google.common.truth.Truth.assertThat; import static org.junit.Assert.assertThrows; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; import com.google.api.client.http.HttpHeaders; import com.google.api.client.http.HttpResponseException; +import com.google.api.client.http.LowLevelHttpRequest; +import com.google.api.client.http.LowLevelHttpResponse; +import com.google.api.client.json.gson.GsonFactory; +import com.google.api.client.testing.http.MockHttpTransport; +import com.google.api.client.testing.http.MockLowLevelHttpRequest; +import com.google.api.client.testing.http.MockLowLevelHttpResponse; +import com.google.api.core.ApiFuture; +import com.google.api.core.ApiFutures; import com.google.api.gax.grpc.GrpcCallContext; import com.google.api.gax.grpc.GrpcStatusCode; import com.google.api.gax.rpc.ApiCallContext; import com.google.api.gax.rpc.ApiException; +import com.google.api.services.storage.model.StorageObject; +import com.google.cloud.NoCredentials; +import com.google.cloud.RestorableState; +import com.google.cloud.ServiceOptions; +import com.google.cloud.WriteChannel; +import com.google.cloud.storage.BufferedWritableByteChannelSession.BufferedWritableByteChannel; +import com.google.cloud.storage.Conversions.Decoder; +import com.google.cloud.storage.UnifiedOpts.Opts; +import com.google.cloud.storage.spi.StorageRpcFactory; +import com.google.cloud.storage.spi.v1.StorageRpc; +import com.google.common.io.ByteStreams; +import com.google.storage.v2.StartResumableWriteRequest; +import com.google.storage.v2.StartResumableWriteResponse; +import com.google.storage.v2.StorageGrpc; +import com.google.storage.v2.WriteObjectRequest; +import com.google.storage.v2.WriteObjectResponse; import io.grpc.CallOptions; +import io.grpc.Server; import io.grpc.Status; import io.grpc.StatusException; import io.grpc.StatusRuntimeException; +import io.grpc.netty.shaded.io.grpc.netty.NettyServerBuilder; +import io.grpc.stub.StreamObserver; import io.opentelemetry.api.common.AttributeKey; import io.opentelemetry.api.common.Attributes; import io.opentelemetry.context.Context; import io.opentelemetry.sdk.metrics.SdkMeterProvider; +import io.opentelemetry.sdk.metrics.data.HistogramPointData; import io.opentelemetry.sdk.metrics.data.LongPointData; import io.opentelemetry.sdk.metrics.data.MetricData; import io.opentelemetry.sdk.testing.exporter.InMemoryMetricReader; +import java.io.ByteArrayInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.net.InetSocketAddress; +import java.nio.ByteBuffer; +import java.nio.file.Files; +import java.nio.file.NoSuchFileException; +import java.nio.file.Path; import java.util.Collection; import java.util.HashMap; +import java.util.Locale; import java.util.Map; +import java.util.concurrent.TimeUnit; import org.checkerframework.checker.nullness.qual.Nullable; +import org.junit.Rule; import org.junit.Test; +import org.junit.rules.TemporaryFolder; public final class StorageMetricsContextTest { + @Rule public final TemporaryFolder tempFolder = new TemporaryFolder(); + @Test public void create_initializesDefaultState() { StorageMetricsContext context = StorageMetricsContext.create("storage.objects.get"); @@ -568,4 +613,428 @@ private static LongPointData getSinglePoint(Collection metrics, Stri assertThat(metric.getLongSumData().getPoints()).hasSize(1); return metric.getLongSumData().getPoints().iterator().next(); } + + private static HistogramPointData getSingleHistogramPoint( + Collection metrics, String metricName) { + MetricData metric = findMetric(metrics, metricName); + assertThat(metric).isNotNull(); + assertThat(metric.getHistogramData().getPoints()).hasSize(1); + return metric.getHistogramData().getPoints().iterator().next(); + } + + @Test + public void recordOperation_recordsRequestBodySize_whenWriteObjectAndBytesGreaterThanZero() { + InMemoryMetricReader reader = InMemoryMetricReader.create(); + SdkMeterProvider provider = SdkMeterProvider.builder().registerMetricReader(reader).build(); + StorageClientMetrics metrics = StorageClientMetrics.create(provider, true, false); + + StorageMetricsContext context = StorageMetricsContext.create("WriteObject"); + context.setRequestBytes(2048L); + context.recordOperation(metrics, null); + + Collection collected = reader.collectAllMetrics(); + HistogramPointData point = + getSingleHistogramPoint( + collected, StorageClientMetrics.METRIC_GCP_STORAGE_CLIENT_REQUEST_BODY_SIZE); + assertThat(point.getCount()).isEqualTo(1L); + assertThat(point.getSum()).isEqualTo(2048.0); + assertThat(point.getAttributes().get(StorageClientMetrics.KEY_RPC_METHOD)) + .isEqualTo("WriteObject"); + } + + @Test + public void recordOperation_requestBodySize_excludesAttemptAndHttpOnlyAttributes() { + InMemoryMetricReader reader = InMemoryMetricReader.create(); + SdkMeterProvider provider = SdkMeterProvider.builder().registerMetricReader(reader).build(); + StorageClientMetrics metrics = StorageClientMetrics.create(provider, true, false); + + Attributes base = + Attributes.builder() + .put(StorageClientMetrics.KEY_RPC_SYSTEM_NAME, "http") + .put(StorageClientMetrics.KEY_SERVER_ADDRESS, "storage.googleapis.com") + .put(StorageClientMetrics.KEY_SERVER_PORT, 443L) + .put(StorageClientMetrics.KEY_HTTP_REQUEST_METHOD, "POST") + .put(StorageClientMetrics.KEY_URL_TEMPLATE, "/upload/storage/v1/b/{bucket}/o") + .put(StorageClientMetrics.KEY_HTTP_RESPONSE_STATUS_CODE, 200L) + .put(StorageClientMetrics.KEY_RPC_RESPONSE_STATUS_CODE, "OK") + .put(StorageClientMetrics.KEY_NETWORK_PATH, StorageClientMetrics.NETWORK_PATH_CLOUDPATH) + .put(StorageClientMetrics.KEY_ERROR_TYPE, "OK") + .put(StorageClientMetrics.KEY_GCP_STORAGE_OPERATION, "WriteObject") + .put(StorageClientMetrics.KEY_GCP_CLIENT_METHOD, "Storage.create") + .put(StorageClientMetrics.KEY_RPC_METHOD, "google.storage.v2.Storage/WriteObject") + .build(); + StorageMetricsContext context = StorageMetricsContext.create("WriteObject", base); + context.setRequestBytes(512L); + context.recordOperation(metrics, null); + + Collection collected = reader.collectAllMetrics(); + HistogramPointData point = + getSingleHistogramPoint( + collected, StorageClientMetrics.METRIC_GCP_STORAGE_CLIENT_REQUEST_BODY_SIZE); + Attributes recordedAttrs = point.getAttributes(); + assertThat(recordedAttrs.get(StorageClientMetrics.KEY_RPC_METHOD)).isEqualTo("WriteObject"); + assertThat(recordedAttrs.get(StorageClientMetrics.KEY_RPC_SYSTEM_NAME)).isEqualTo("http"); + assertThat(recordedAttrs.get(StorageClientMetrics.KEY_SERVER_ADDRESS)) + .isEqualTo("storage.googleapis.com"); + assertThat(recordedAttrs.get(StorageClientMetrics.KEY_SERVER_PORT)).isEqualTo(443L); + assertThat(recordedAttrs.get(StorageClientMetrics.KEY_HTTP_REQUEST_METHOD)).isNull(); + assertThat(recordedAttrs.get(StorageClientMetrics.KEY_URL_TEMPLATE)).isNull(); + assertThat(recordedAttrs.get(StorageClientMetrics.KEY_HTTP_RESPONSE_STATUS_CODE)).isNull(); + assertThat(recordedAttrs.get(StorageClientMetrics.KEY_RPC_RESPONSE_STATUS_CODE)).isNull(); + assertThat(recordedAttrs.get(StorageClientMetrics.KEY_NETWORK_PATH)).isNull(); + assertThat(recordedAttrs.get(StorageClientMetrics.KEY_ERROR_TYPE)).isNull(); + assertThat(recordedAttrs.get(StorageClientMetrics.KEY_GCP_STORAGE_OPERATION)).isNull(); + assertThat(recordedAttrs.get(StorageClientMetrics.KEY_GCP_CLIENT_METHOD)).isNull(); + } + + @Test + public void recordOperation_doesNotRecordRequestBodySize_whenBytesIsZero() { + InMemoryMetricReader reader = InMemoryMetricReader.create(); + SdkMeterProvider provider = SdkMeterProvider.builder().registerMetricReader(reader).build(); + StorageClientMetrics metrics = StorageClientMetrics.create(provider, true, false); + + StorageMetricsContext context = StorageMetricsContext.create("WriteObject"); + context.setRequestBytes(0L); + context.recordOperation(metrics, null); + + Collection collected = reader.collectAllMetrics(); + assertThat( + findMetric(collected, StorageClientMetrics.METRIC_GCP_STORAGE_CLIENT_REQUEST_BODY_SIZE)) + .isNull(); + } + + @Test + public void recordOperation_doesNotRecordRequestBodySize_whenNotWriteObject() { + InMemoryMetricReader reader = InMemoryMetricReader.create(); + SdkMeterProvider provider = SdkMeterProvider.builder().registerMetricReader(reader).build(); + StorageClientMetrics metrics = StorageClientMetrics.create(provider, true, false); + + StorageMetricsContext context = StorageMetricsContext.create("ReadObject"); + context.setRequestBytes(2048L); + context.recordOperation(metrics, null); + + Collection collected = reader.collectAllMetrics(); + assertThat( + findMetric(collected, StorageClientMetrics.METRIC_GCP_STORAGE_CLIENT_REQUEST_BODY_SIZE)) + .isNull(); + } + + @Test + public void baseStorageWriteChannel_recordsRequestBodySizeOnClose_afterMultipleWrites() + throws IOException { + InMemoryMetricReader reader = InMemoryMetricReader.create(); + SdkMeterProvider provider = SdkMeterProvider.builder().registerMetricReader(reader).build(); + StorageClientMetrics metrics = StorageClientMetrics.create(provider, true, false); + StorageMetricsContext context = StorageMetricsContext.createForMultiStep("WriteObject"); + + BaseStorageWriteChannel channel = createTestWriteChannel(context, metrics, false); + channel.write(ByteBuffer.wrap(new byte[128])); + channel.write(ByteBuffer.wrap(new byte[384])); + channel.close(); + + Collection collected = reader.collectAllMetrics(); + HistogramPointData point = + getSingleHistogramPoint( + collected, StorageClientMetrics.METRIC_GCP_STORAGE_CLIENT_REQUEST_BODY_SIZE); + assertThat(point.getCount()).isEqualTo(1L); + assertThat(point.getSum()).isEqualTo(512.0); + assertThat(point.getAttributes().get(StorageClientMetrics.KEY_RPC_METHOD)) + .isEqualTo("WriteObject"); + } + + @Test + public void baseStorageWriteChannel_doesNotRecordRequestBodySize_whenZeroBytesWritten() + throws IOException { + InMemoryMetricReader reader = InMemoryMetricReader.create(); + SdkMeterProvider provider = SdkMeterProvider.builder().registerMetricReader(reader).build(); + StorageClientMetrics metrics = StorageClientMetrics.create(provider, true, false); + StorageMetricsContext context = StorageMetricsContext.createForMultiStep("WriteObject"); + + BaseStorageWriteChannel channel = createTestWriteChannel(context, metrics, false); + channel.close(); + + Collection collected = reader.collectAllMetrics(); + assertThat( + findMetric(collected, StorageClientMetrics.METRIC_GCP_STORAGE_CLIENT_REQUEST_BODY_SIZE)) + .isNull(); + LongPointData opPoint = + getSinglePoint(collected, StorageClientMetrics.METRIC_GCP_STORAGE_CLIENT_OPERATIONS); + assertThat(opPoint.getValue()).isEqualTo(1L); + } + + @Test + public void baseStorageWriteChannel_recordsRequestBodySizeEvenWhenCloseFails() + throws IOException { + InMemoryMetricReader reader = InMemoryMetricReader.create(); + SdkMeterProvider provider = SdkMeterProvider.builder().registerMetricReader(reader).build(); + StorageClientMetrics metrics = StorageClientMetrics.create(provider, true, false); + StorageMetricsContext context = StorageMetricsContext.createForMultiStep("WriteObject"); + + BaseStorageWriteChannel channel = createTestWriteChannel(context, metrics, true); + channel.write(ByteBuffer.wrap(new byte[256])); + assertThrows(StorageException.class, channel::close); + + Collection collected = reader.collectAllMetrics(); + HistogramPointData point = + getSingleHistogramPoint( + collected, StorageClientMetrics.METRIC_GCP_STORAGE_CLIENT_REQUEST_BODY_SIZE); + assertThat(point.getSum()).isEqualTo(256.0); + } + + @Test + @SuppressWarnings("deprecation") + public void storageImpl_writeMethods_recordRequestBodySize_andPcuHookSuppresses() + throws Exception { + InMemoryMetricReader reader = InMemoryMetricReader.create(); + SdkMeterProvider provider = SdkMeterProvider.builder().registerMetricReader(reader).build(); + + StorageRpcFactory rpcFactoryMock = mock(StorageRpcFactory.class); + StorageRpc storageRpcMock = mock(StorageRpc.class); + when(rpcFactoryMock.create(any(StorageOptions.class))).thenReturn(storageRpcMock); + when(storageRpcMock.create(any(StorageObject.class), any(InputStream.class), any())) + .thenAnswer( + inv -> { + InputStream in = inv.getArgument(1); + ByteStreams.exhaust(in); + return new StorageObject().setBucket("b").setName("n"); + }); + when(storageRpcMock.open(any(StorageObject.class), any())) + .thenReturn("http://localhost/upload-session-1"); + + MockHttpTransport mockTransport = + new MockHttpTransport() { + @Override + public LowLevelHttpRequest buildRequest(String method, String url) { + return new MockLowLevelHttpRequest(url) { + @Override + public LowLevelHttpResponse execute() { + long uploadedLen = getContentLength(); + String json = + String.format( + Locale.US, + "{\"bucket\":\"b\",\"name\":\"n\",\"size\":\"%d\"}", + uploadedLen); + return new MockLowLevelHttpResponse() + .setStatusCode(200) + .setContentType("application/json; charset=UTF-8") + .addHeader("content-type", "application/json; charset=UTF-8") + .setContent(json); + } + }; + } + }; + com.google.api.services.storage.Storage apiary = + new com.google.api.services.storage.Storage( + mockTransport, GsonFactory.getDefaultInstance(), null); + when(storageRpcMock.getStorage()).thenReturn(apiary); + + HttpStorageOptions options = + HttpStorageOptions.newBuilder() + .setProjectId("test-proj") + .setCredentials(NoCredentials.getInstance()) + .setServiceRpcFactory(rpcFactoryMock) + .setRetrySettings(ServiceOptions.getNoRetrySettings()) + .setEnableOtelMetrics(true) + .setMeterProvider(provider) + .build(); + StorageImpl storage = (StorageImpl) options.getService(); + BlobInfo blobInfo = BlobInfo.newBuilder("b", "n").build(); + + // 1. create(BlobInfo, byte[]) -> 100 bytes + storage.create(blobInfo, new byte[100]); + // 2. create(BlobInfo, InputStream) -> 200 bytes + storage.create(blobInfo, new ByteArrayInputStream(new byte[200])); + // 3. createFrom(BlobInfo, Path) -> 300 bytes + Path file1 = tempFolder.newFile("f1.bin").toPath(); + Files.write(file1, new byte[300]); + storage.createFrom(blobInfo, file1); + // 4. internalCreateFrom(Path, BlobInfo, Opts) -> 400 bytes + Path file2 = tempFolder.newFile("f2.bin").toPath(); + Files.write(file2, new byte[400]); + storage.internalCreateFrom(file2, blobInfo, Opts.empty()); + // 5. 3-arg internalDirectUpload (PCU part upload hook) -> must NOT record request.body.size + storage.internalDirectUpload(blobInfo, Opts.empty(), ByteBuffer.wrap(new byte[999])); + + Collection collected = reader.collectAllMetrics(); + HistogramPointData point = + getSingleHistogramPoint( + collected, StorageClientMetrics.METRIC_GCP_STORAGE_CLIENT_REQUEST_BODY_SIZE); + assertThat(point.getCount()).isEqualTo(4L); + assertThat(point.getSum()).isEqualTo(1000.0); + } + + @Test + public void grpcStorageImpl_writeMethods_recordRequestBodySize_andPcuHookSuppresses() + throws Exception { + InMemoryMetricReader reader = InMemoryMetricReader.create(); + SdkMeterProvider provider = SdkMeterProvider.builder().registerMetricReader(reader).build(); + + StorageGrpc.StorageImplBase service = + new StorageGrpc.StorageImplBase() { + @Override + public void startResumableWrite( + StartResumableWriteRequest request, + StreamObserver responseObserver) { + responseObserver.onNext( + StartResumableWriteResponse.newBuilder().setUploadId("upload-1").build()); + responseObserver.onCompleted(); + } + + @Override + public StreamObserver writeObject( + StreamObserver responseObserver) { + return new StreamObserver() { + long totalSize = 0; + + @Override + public void onNext(WriteObjectRequest value) { + totalSize += value.getChecksummedData().getContent().size(); + } + + @Override + public void onError(Throwable t) {} + + @Override + public void onCompleted() { + responseObserver.onNext( + WriteObjectResponse.newBuilder() + .setResource( + com.google.storage.v2.Object.newBuilder() + .setBucket("projects/_/buckets/b") + .setName("n") + .setSize(totalSize) + .build()) + .build()); + responseObserver.onCompleted(); + } + }; + } + }; + + Server server = + NettyServerBuilder.forAddress(new InetSocketAddress("127.0.0.1", 0)) + .addService(service) + .build() + .start(); + try { + String endpoint = String.format(Locale.US, "http://127.0.0.1:%d", server.getPort()); + GrpcStorageOptions options = + StorageOptions.grpc() + .setHost(endpoint) + .setProjectId("test-proj") + .setCredentials(NoCredentials.getInstance()) + .setEnableGrpcClientMetrics(false) + .setAttemptDirectPath(false) + .setRetrySettings(ServiceOptions.getNoRetrySettings()) + .setEnableOtelMetrics(true) + .setMeterProvider(provider) + .build(); + try (GrpcStorageImpl storage = (GrpcStorageImpl) options.getService()) { + BlobInfo blobInfo = BlobInfo.newBuilder("b", "n").build(); + + // 1. create(BlobInfo, byte[]) -> 120 bytes + storage.create(blobInfo, new byte[120]); + // 2. create(BlobInfo, InputStream) -> 230 bytes + storage.create(blobInfo, new ByteArrayInputStream(new byte[230])); + // 3. createFrom(BlobInfo, InputStream) -> 340 bytes + storage.createFrom(blobInfo, new ByteArrayInputStream(new byte[340])); + // 4. createFrom(BlobInfo, Path) / internalCreateFrom -> 410 bytes + Path file = tempFolder.newFile("grpc.bin").toPath(); + Files.write(file, new byte[410]); + storage.createFrom(blobInfo, file); + // 5. writer(BlobInfo) -> 500 bytes + try (WriteChannel writer = storage.writer(blobInfo)) { + writer.write(ByteBuffer.wrap(new byte[500])); + } + // 6. 3-arg internalDirectUpload (PCU hook) -> must NOT record request.body.size + storage.internalDirectUpload(blobInfo, Opts.empty(), ByteBuffer.wrap(new byte[777])); + + // 7. internalCreateFrom with nonexistent Path -> throws NoSuchFileException inside try + // block and still records the failed WriteObject operation metric + Path missingPath = tempFolder.getRoot().toPath().resolve("does-not-exist.bin"); + assertThrows( + NoSuchFileException.class, + () -> storage.internalCreateFrom(missingPath, blobInfo, Opts.empty())); + + Collection collected = reader.collectAllMetrics(); + HistogramPointData bodyPoint = + getSingleHistogramPoint( + collected, StorageClientMetrics.METRIC_GCP_STORAGE_CLIENT_REQUEST_BODY_SIZE); + assertThat(bodyPoint.getCount()).isEqualTo(5L); + assertThat(bodyPoint.getSum()).isEqualTo(1600.0); + + MetricData opsMetric = + findMetric(collected, StorageClientMetrics.METRIC_GCP_STORAGE_CLIENT_OPERATIONS); + assertThat(opsMetric).isNotNull(); + long totalOps = + opsMetric.getLongSumData().getPoints().stream() + .mapToLong(LongPointData::getValue) + .sum(); + // 5 successful WriteObject ops + 1 failed internalCreateFrom(missingPath) op = 6 ops + assertThat(totalOps).isEqualTo(6L); + } + } finally { + server.shutdownNow().awaitTermination(5, TimeUnit.SECONDS); + } + } + + private static BaseStorageWriteChannel createTestWriteChannel( + StorageMetricsContext context, StorageClientMetrics metrics, boolean failOnClose) { + BlobInfo resultBlob = BlobInfo.newBuilder("b", "n").build(); + BufferedWritableByteChannel bufferedChannel = + new BufferedWritableByteChannel() { + private boolean open = true; + + @Override + public void flush() {} + + @Override + public int write(ByteBuffer src) { + int remaining = src.remaining(); + src.position(src.limit()); + return remaining; + } + + @Override + public boolean isOpen() { + return open; + } + + @Override + public void close() throws IOException { + open = false; + if (failOnClose) { + throw new IOException("simulated close failure"); + } + } + }; + BufferedWritableByteChannelSession session = + new BufferedWritableByteChannelSession() { + @Override + public ApiFuture openAsync() { + return ApiFutures.immediateFuture(bufferedChannel); + } + + @Override + public BufferedWritableByteChannel open() { + return bufferedChannel; + } + + @Override + public ApiFuture getResult() { + return ApiFutures.immediateFuture(resultBlob); + } + }; + return new BaseStorageWriteChannel(Decoder.identity(), context, metrics) { + @Override + public RestorableState capture() { + return null; + } + + @Override + protected LazyWriteChannel newLazyWriteChannel() { + return new LazyWriteChannel<>(() -> session); + } + }; + } }