Skip to content

feat(storage): add gcp.storage.client.request.body.size metric for write operations #14746

New issue

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

Sign up for GitHub

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

Already on GitHub? Sign in to your account

Draft
nidhiii-27 wants to merge 2 commits into
base: feat/storage-ttfb-metric
Choose a base branch
from
Draft

feat(storage): add gcp.storage.client.request.body.size metric for write operations #14746

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
0f50161
feat: add gcp.storage.client.request.body.size metric for write opera…
nidhiii-27 Oct 22, 2026
209a756
fix(storage): address review feedback and complete request.body.size …
nidhiii-27 Nov 3, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions .../google-cloud-storage/src/main/java/com/google/cloud/storage/BaseStorageWriteChannel.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down
32 changes: 28 additions & 4 deletions ...-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<ObjectTargetOpt> 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);
}

Expand Down Expand Up @@ -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<WriteObjectResponse> responseApiFuture = session.getResult();
return this.getBlob(responseApiFuture);
Expand Down Expand Up @@ -317,6 +324,9 @@ public Blob internalCreateFrom(Path path, BlobInfo info, Opts<ObjectTargetOpt> 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);
Expand Down Expand Up @@ -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);
}
Expand Down Expand Up @@ -930,14 +943,25 @@ public GrpcBlobWriteChannel writer(BlobInfo blobInfo, BlobWriteOption... options
@Override
public BlobInfo internalDirectUpload(
BlobInfo blobInfo, Opts<ObjectTargetOpt> opts, ByteBuffer buf) {
return internalDirectUpload(blobInfo, opts, buf, null);
}

private BlobInfo internalDirectUpload(
BlobInfo blobInfo,
Opts<ObjectTargetOpt> opts,
ByteBuffer buf,
@Nullable StorageMetricsContext metricsContext) {
requireNonNull(blobInfo, "blobInfo must be non null");
requireNonNull(buf, "content must be non null");
Opts<ObjectTargetOpt> optsWithDefaults = opts.prepend(defaultOpts);
GrpcCallContext grpcCallContext =
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(
Expand Down
145 changes: 100 additions & 45 deletions java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -157,21 +158,32 @@ public Bucket create(BucketInfo bucketInfo, BucketTargetOption... options) {
@Override
public Blob create(BlobInfo blobInfo, BlobTargetOption... options) {
Opts<ObjectTargetOpt> 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<ObjectTargetOpt> 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<ObjectTargetOpt> 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);
}

Expand All @@ -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
Expand All @@ -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<ObjectTargetOpt> opts = Opts.unwrap(options).resolveFrom(blobInfo);
final Map<StorageRpc.Option, ?> optionsMap = opts.getRpcOptions();
BlobInfo.Builder builder = blobInfo.toBuilder().setMd5(null).setCrc32c(null);
Expand Down Expand Up @@ -1879,48 +1905,74 @@ public BlobInfo internalCreateFrom(Path path, BlobInfo info, Opts<ObjectTargetOp
}
long size = Files.size(path);
if (size == 0L) {
return internalDirectUpload(info, opts, ByteBuffer.allocate(0));
return internalDirectUpload(
info, opts, ByteBuffer.allocate(0), getOptions().createMetricsContext("WriteObject"));
}
final Map<StorageRpc.Option, ?> 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<StorageRpc.Option, ?> 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<String> 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<String> 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<StorageObject> 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<StorageObject> 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<StorageObject> 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<StorageObject> 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<ObjectTargetOpt> opts, ByteBuffer buf) {
return internalDirectUpload(info, opts, buf, null);
}

private BlobInfo internalDirectUpload(
BlobInfo info,
Opts<ObjectTargetOpt> 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());
Expand All @@ -1934,14 +1986,17 @@ public BlobInfo internalDirectUpload(BlobInfo info, Opts<ObjectTargetOpt> 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);
}

/**
Expand Down
41 changes: 41 additions & 0 deletions ...ge/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageMetricsContext.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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<AttributeKey<?>> 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;
Expand Down Expand Up @@ -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<Object> objKey = (AttributeKey<Object>) 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;
Expand Down Expand Up @@ -394,6 +431,10 @@ long addRequestBytes(long bytes) {
return requestBytes.addAndGet(bytes);
}

void setRequestBytes(long bytes) {
requestBytes.set(bytes);
}

long getRequestBytes() {
return requestBytes.get();
}
Expand Down
Loading
Loading