impl(o11y): introduce http.response.body_size

gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/HttpJsonCallContext.java

@@ -607,10 +607,11 @@ public ApiTracer getTracer() {
   @Override
   public HttpJsonCallContext withTracer(@Nonnull ApiTracer newTracer) {
     Preconditions.checkNotNull(newTracer);
+    HttpJsonCallOptions newCallOptions = callOptions.toBuilder().setTracer(newTracer).build();
 
     return new HttpJsonCallContext(
         this.channel,
-        this.callOptions,
+        newCallOptions,
         this.timeout,
         this.streamWaitTimeout,
         this.streamIdleTimeout,

gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/HttpJsonCallOptions.java

@@ -35,6 +35,7 @@
 import static com.google.api.gax.util.TimeConversionUtils.toThreetenInstant;
 
 import com.google.api.core.ObsoleteApi;
+import com.google.api.gax.tracing.ApiTracer;
 import com.google.auth.Credentials;
 import com.google.auto.value.AutoValue;
 import com.google.protobuf.TypeRegistry;
@@ -71,6 +72,9 @@ public final org.threeten.bp.Instant getDeadline() {
   @Nullable
   public abstract TypeRegistry getTypeRegistry();
 
+  @Nullable
+  public abstract ApiTracer getTracer();
+
   public abstract Builder toBuilder();
 
   public static Builder newBuilder() {
@@ -106,6 +110,11 @@ public HttpJsonCallOptions merge(HttpJsonCallOptions inputOptions) {
       builder.setTypeRegistry(newTypeRegistry);
     }
 
+    ApiTracer newTracer = inputOptions.getTracer();
+    if (newTracer != null) {
+      builder.setTracer(newTracer);
+    }
+
     return builder.build();
   }
 
@@ -131,6 +140,8 @@ public final Builder setDeadline(org.threeten.bp.Instant value) {
 
     public abstract Builder setTypeRegistry(TypeRegistry value);
 
+    public abstract Builder setTracer(ApiTracer value);
+
     public abstract HttpJsonCallOptions build();
   }
 }

gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/HttpJsonClientCallImpl.java

@@ -34,7 +34,9 @@
 import com.google.api.gax.httpjson.HttpRequestRunnable.ResultListener;
 import com.google.api.gax.httpjson.HttpRequestRunnable.RunnableResult;
 import com.google.api.gax.rpc.StatusCode;
+import com.google.api.gax.tracing.ApiTracer;
 import com.google.common.base.Preconditions;
+import com.google.common.io.CountingInputStream;
 import com.google.errorprone.annotations.concurrent.GuardedBy;
 import java.io.IOException;
 import java.io.InputStreamReader;
@@ -118,6 +120,9 @@ final class HttpJsonClientCallImpl<RequestT, ResponseT>
   @GuardedBy("lock")
   private ProtoMessageJsonStreamIterator responseStreamIterator;
 
+  @GuardedBy("lock")
+  private CountingInputStream responseCountingStream;
+
   @GuardedBy("lock")
   private volatile boolean closed;
 
@@ -400,14 +405,19 @@ private boolean consumeMessageFromStream() throws IOException {
       return true;
     }
 
+    if (responseCountingStream == null) {
+      responseCountingStream = new CountingInputStream(runnableResult.getResponseContent());
+    }
+
     boolean allMessagesConsumed;
     Reader responseReader;
+    long responseBodySizeStart = responseCountingStream.getCount();
     if (methodDescriptor.getType() == MethodType.SERVER_STREAMING) {
       // Lazily initialize responseStreamIterator in case if it is a server streaming response
       if (responseStreamIterator == null) {
         responseStreamIterator =
             new ProtoMessageJsonStreamIterator(
-                new InputStreamReader(runnableResult.getResponseContent(), StandardCharsets.UTF_8));
+                new InputStreamReader(responseCountingStream, StandardCharsets.UTF_8));
       }
       if (responseStreamIterator.hasNext()) {
         responseReader = responseStreamIterator.next();
@@ -419,15 +429,20 @@ private boolean consumeMessageFromStream() throws IOException {
       // from the client to check if there is anything else left in the stream).
       allMessagesConsumed = !responseStreamIterator.hasNext();
     } else {
-      responseReader =
-          new InputStreamReader(runnableResult.getResponseContent(), StandardCharsets.UTF_8);
+      responseReader = new InputStreamReader(responseCountingStream, StandardCharsets.UTF_8);
       // Unary calls have only one message in their response, so we should be ready to close
       // immediately after delivering a single response message.
       allMessagesConsumed = true;
     }
 
     ResponseT message =
         methodDescriptor.getResponseParser().parse(responseReader, callOptions.getTypeRegistry());
+    long responseBodySizeEnd = responseCountingStream.getCount();
+
+    ApiTracer tracer = callOptions.getTracer();
+    if (tracer != null) {
+      tracer.responseReceived(responseBodySizeEnd - responseBodySizeStart);
+    }
     pendingNotifications.offer(new OnMessageNotificationTask<>(listener, message));
 
     return allMessagesConsumed;

gax-java/gax-httpjson/src/test/java/com/google/api/gax/httpjson/BodySizeRecordingTest.java

@@ -0,0 +1,225 @@
+/*
+ * Copyright 2026 Google LLC
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ *     * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ *     * Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following disclaimer
+ * in the documentation and/or other materials provided with the
+ * distribution.
+ *     * Neither the name of Google LLC nor the names of its
+ * contributors may be used to endorse or promote products derived from
+ * this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+package com.google.api.gax.httpjson;
+
+import static com.google.common.truth.Truth.assertThat;
+
+import com.google.api.gax.httpjson.testing.MockHttpService;
+import com.google.api.gax.httpjson.testing.TestApiTracer;
+import com.google.api.gax.rpc.EndpointContext;
+import com.google.api.gax.rpc.ResponseObserver;
+import com.google.api.gax.rpc.StreamController;
+import com.google.auth.Credentials;
+import com.google.protobuf.Field;
+import java.io.IOException;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+class BodySizeRecordingTest {
+  private static final ApiMethodDescriptor<Field, Field> FAKE_METHOD_DESCRIPTOR =
+      ApiMethodDescriptor.<Field, Field>newBuilder()
+          .setFullMethodName("google.cloud.v1.Fake/FakeMethod")
+          .setHttpMethod("POST")
+          .setRequestFormatter(
+              ProtoMessageRequestFormatter.<Field>newBuilder()
+                  .setPath(
+                      "/fake/v1/name/{name}",
+                      request -> {
+                        Map<String, String> fields = new HashMap<>();
+                        ProtoRestSerializer<Field> serializer = ProtoRestSerializer.create();
+                        serializer.putPathParam(fields, "name", request.getName());
+                        return fields;
+                      })
+                  .setQueryParamsExtractor(request -> new HashMap<>())
+                  .setRequestBodyExtractor(
+                      request ->
+                          ProtoRestSerializer.create()
+                              .toBody("*", request.toBuilder().clearName().build(), false))
+                  .build())
+          .setResponseParser(
+              ProtoMessageResponseParser.<Field>newBuilder()
+                  .setDefaultInstance(Field.getDefaultInstance())
+                  .build())
+          .build();
+
+  private static final MockHttpService MOCK_SERVICE =
+      new MockHttpService(Collections.singletonList(FAKE_METHOD_DESCRIPTOR), "google.com:443");
+
+  private static ExecutorService executorService;
+  private ManagedHttpJsonChannel channel;
+  private TestApiTracer tracer;
+
+  @BeforeAll
+  public static void initialize() {
+    executorService = Executors.newFixedThreadPool(2);
+  }
+
+  @AfterAll
+  public static void destroy() {
+    executorService.shutdownNow();
+  }
+
+  @BeforeEach
+  void setUp() throws IOException {
+    channel =
+        ManagedHttpJsonChannel.newBuilder()
+            .setEndpoint("google.com:443")
+            .setExecutor(executorService)
+            .setHttpTransport(MOCK_SERVICE)
+            .build();
+    tracer = new TestApiTracer();
+  }
+
+  @AfterEach
+  void tearDown() {
+    MOCK_SERVICE.reset();
+  }
+
+  @Test
+  void testBodySizeRecording() throws Exception {
+    HttpJsonDirectCallable<Field, Field> callable =
+        new HttpJsonDirectCallable<>(FAKE_METHOD_DESCRIPTOR);
+
+    EndpointContext endpointContext = Mockito.mock(EndpointContext.class);
+    Mockito.doNothing()
+        .when(endpointContext)
+        .validateUniverseDomain(
+            Mockito.any(Credentials.class), Mockito.any(HttpJsonStatusCode.class));
+
+    HttpJsonCallContext callContext =
+        HttpJsonCallContext.createDefault()
+            .withChannel(channel)
+            .withEndpointContext(endpointContext)
+            .withTracer(tracer);
+
+    Field request = Field.newBuilder().setName("bob").setNumber(42).build();
+    Field response = Field.newBuilder().setName("alice").setNumber(43).build();
+
+    MOCK_SERVICE.addResponse(response);
+
+    callable.futureCall(request, callContext).get();
+
+    // Verify response size
+    // MockHttpService uses ProtoRestSerializer which pretty-prints.
+    String expectedResponseBody = ProtoRestSerializer.create().toBody("*", response, false);
+    long expectedResponseSize = expectedResponseBody.getBytes("UTF-8").length;
+    assertThat(tracer.getResponseReceivedSize()).isEqualTo(expectedResponseSize);
+  }
+
+  @Test
+  void testBodySizeRecordingServerStreaming() throws Exception {
+    ApiMethodDescriptor<Field, Field> methodServerStreaming =
+        FAKE_METHOD_DESCRIPTOR.toBuilder()
+            .setType(ApiMethodDescriptor.MethodType.SERVER_STREAMING)
+            .build();
+
+    MockHttpService streamingMockService =
+        new MockHttpService(Collections.singletonList(methodServerStreaming), "google.com:443");
+    ManagedHttpJsonChannel streamingChannel =
+        ManagedHttpJsonChannel.newBuilder()
+            .setEndpoint("google.com:443")
+            .setExecutor(executorService)
+            .setHttpTransport(streamingMockService)
+            .build();
+
+    HttpJsonDirectServerStreamingCallable<Field, Field> callable =
+        new HttpJsonDirectServerStreamingCallable<>(methodServerStreaming);
+
+    EndpointContext endpointContext = Mockito.mock(EndpointContext.class);
+    Mockito.doNothing()
+        .when(endpointContext)
+        .validateUniverseDomain(
+            Mockito.any(Credentials.class), Mockito.any(HttpJsonStatusCode.class));
+
+    HttpJsonCallContext callContext =
+        HttpJsonCallContext.createDefault()
+            .withChannel(streamingChannel)
+            .withEndpointContext(endpointContext)
+            .withTracer(tracer);
+
+    Field request = Field.newBuilder().setName("bob").setNumber(42).build();
+    Field response1 = Field.newBuilder().setName("alice1").setNumber(43).build();
+    Field response2 = Field.newBuilder().setName("alice2").setNumber(44).build();
+
+    streamingMockService.addResponse(new Field[] {response1, response2});
+
+    final List<Field> receivedResponses = new java.util.ArrayList<>();
+    final CountDownLatch latch = new CountDownLatch(1);
+
+    callable.call(
+        request,
+        new ResponseObserver<Field>() {
+          @Override
+          public void onStart(StreamController controller) {}
+
+          @Override
+          public void onResponse(Field response) {
+            receivedResponses.add(response);
+          }
+
+          @Override
+          public void onError(Throwable t) {
+            latch.countDown();
+          }
+
+          @Override
+          public void onComplete() {
+            latch.countDown();
+          }
+        },
+        callContext);
+
+    latch.await(10, TimeUnit.SECONDS);
+
+    assertThat(receivedResponses).hasSize(2);
+
+    // Verify response size
+    // MockHttpService server-streaming response construction adds [ ] and ,
+    String resp1Json = methodServerStreaming.getResponseParser().serialize(response1);
+    String resp2Json = methodServerStreaming.getResponseParser().serialize(response2);
+    long expectedTotalResponseSize =
+        ("[" + resp1Json + "," + resp2Json + "]").getBytes("UTF-8").length;
+
+    assertThat(tracer.getResponseReceivedSize()).isEqualTo(expectedTotalResponseSize);
+    streamingChannel.shutdownNow();
+  }
+}

gax-java/gax-httpjson/src/test/java/com/google/api/gax/httpjson/testing/TestApiTracer.java

@@ -32,6 +32,7 @@
 import com.google.api.gax.tracing.ApiTracer;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
 import org.threeten.bp.Duration;
 
 /**
@@ -43,6 +44,7 @@ public class TestApiTracer implements ApiTracer {
   private final AtomicInteger attemptsStarted = new AtomicInteger();
   private final AtomicInteger attemptsFailed = new AtomicInteger();
   private final AtomicBoolean retriesExhausted = new AtomicBoolean(false);
+  private final AtomicLong responseReceivedSize = new AtomicLong();
 
   public TestApiTracer() {}
 
@@ -58,6 +60,10 @@ public AtomicBoolean getRetriesExhausted() {
     return retriesExhausted;
   }
 
+  public long getResponseReceivedSize() {
+    return responseReceivedSize.get();
+  }
+
   @Override
   public void attemptStarted(int attemptNumber) {
     attemptsStarted.incrementAndGet();
@@ -78,5 +84,10 @@ public void attemptFailedRetriesExhausted(Throwable error) {
     attemptsFailed.incrementAndGet();
     retriesExhausted.set(true);
   }
+
+  @Override
+  public void responseReceived(long responseSize) {
+    responseReceivedSize.addAndGet(responseSize);
+  }
 }
 ;

gax-java/gax/src/main/java/com/google/api/gax/tracing/ApiTracer.java

@@ -179,9 +179,13 @@
   default void responseReceived() {}
   ;
 
+  /** Adds an annotation that a streaming response has been received with size. */
+  default void responseReceived(long responseSize) {}
+  ;
+
   /** Adds an annotation that a streaming request has been sent. */
   default void requestSent() {}
   ;
 
   /**
    * Adds an annotation that a batch of writes has been flushed.