Implement otel retry metrics

api/src/main/java/io/grpc/ClientStreamTracer.java

@@ -132,12 +132,15 @@
     private final CallOptions callOptions;
     private final int previousAttempts;
     private final boolean isTransparentRetry;
+    private final boolean isHedging;
 
     StreamInfo(
-        CallOptions callOptions, int previousAttempts, boolean isTransparentRetry) {
+        CallOptions callOptions, int previousAttempts, boolean isTransparentRetry,
+        boolean isHedging) {
       this.callOptions = checkNotNull(callOptions, "callOptions");
       this.previousAttempts = previousAttempts;
       this.isTransparentRetry = isTransparentRetry;
+      this.isHedging = isHedging;
     }
 
     /**
@@ -165,6 +168,15 @@
       return isTransparentRetry;
     }
 
+    /**
+     * Whether the stream is hedging.
+     *
+     * @since 1.74.0
+     */
+    public boolean isHedging() {
+      return isHedging;
+    }
+
     /**
      * Converts this StreamInfo into a new Builder.
      *
@@ -174,7 +186,9 @@
       return new Builder()
           .setCallOptions(callOptions)
           .setPreviousAttempts(previousAttempts)
-          .setIsTransparentRetry(isTransparentRetry);
+          .setIsTransparentRetry(isTransparentRetry)
+          .setIsHedging(isHedging);
+
     }
 
     /**
@@ -192,6 +206,7 @@
           .add("callOptions", callOptions)
           .add("previousAttempts", previousAttempts)
           .add("isTransparentRetry", isTransparentRetry)
+          .add("isHedging", isHedging)
           .toString();
     }
 
@@ -204,6 +219,7 @@
       private CallOptions callOptions = CallOptions.DEFAULT;
       private int previousAttempts;
       private boolean isTransparentRetry;
+      private boolean isHedging;
 
       Builder() {
       }
@@ -236,11 +252,21 @@
         return this;
       }
 
+      /**
+       * Sets whether the stream is hedging.
+       *
+       * @since 1.74.0
+       */
+      public Builder setIsHedging(boolean isHedging) {
+        this.isHedging = isHedging;
+        return this;
+      }
+
       /**
        * Builds a new StreamInfo.
        */
       public StreamInfo build() {
-        return new StreamInfo(callOptions, previousAttempts, isTransparentRetry);
+        return new StreamInfo(callOptions, previousAttempts, isTransparentRetry, isHedging);
       }
     }
   }

core/src/main/java/io/grpc/internal/ClientCallImpl.java

@@ -250,7 +250,8 @@ public void runInContext() {
       stream = clientStreamProvider.newStream(method, callOptions, headers, context);
     } else {
       ClientStreamTracer[] tracers =
-          GrpcUtil.getClientStreamTracers(callOptions, headers, 0, false);
+          GrpcUtil.getClientStreamTracers(callOptions, headers, 0,
+              false, false);
       String deadlineName = contextIsDeadlineSource ? "Context" : "CallOptions";
       Long nameResolutionDelay = callOptions.getOption(NAME_RESOLUTION_DELAYED);
       String description = String.format(

core/src/main/java/io/grpc/internal/GrpcUtil.java

@@ -757,13 +757,15 @@ public ListenableFuture<SocketStats> getStats() {
 
   /** Gets stream tracers based on CallOptions. */
   public static ClientStreamTracer[] getClientStreamTracers(
-      CallOptions callOptions, Metadata headers, int previousAttempts, boolean isTransparentRetry) {
+      CallOptions callOptions, Metadata headers, int previousAttempts, boolean isTransparentRetry,
+      boolean isHedging) {
     List<ClientStreamTracer.Factory> factories = callOptions.getStreamTracerFactories();
     ClientStreamTracer[] tracers = new ClientStreamTracer[factories.size() + 1];
     StreamInfo streamInfo = StreamInfo.newBuilder()
         .setCallOptions(callOptions)
         .setPreviousAttempts(previousAttempts)
         .setIsTransparentRetry(isTransparentRetry)
+        .setIsHedging(isHedging)
         .build();
     for (int i = 0; i < factories.size(); i++) {
       tracers[i] = factories.get(i).newClientStreamTracer(streamInfo, headers);

core/src/main/java/io/grpc/internal/ManagedChannelImpl.java

@@ -479,7 +479,8 @@ public ClientStream newStream(
       // the delayed transport or a real transport will go in-use and cancel the idle timer.
       if (!retryEnabled) {
         ClientStreamTracer[] tracers = GrpcUtil.getClientStreamTracers(
-            callOptions, headers, 0, /* isTransparentRetry= */ false);
+            callOptions, headers, 0, /* isTransparentRetry= */ false,
+            /* isHedging= */false);
         Context origContext = context.attach();
         try {
           return delayedTransport.newStream(method, headers, callOptions, tracers);
@@ -519,10 +520,10 @@ void postCommit() {
           @Override
           ClientStream newSubstream(
               Metadata newHeaders, ClientStreamTracer.Factory factory, int previousAttempts,
-              boolean isTransparentRetry) {
+              boolean isTransparentRetry, boolean isHedging) {
             CallOptions newOptions = callOptions.withStreamTracerFactory(factory);
             ClientStreamTracer[] tracers = GrpcUtil.getClientStreamTracers(
-                newOptions, newHeaders, previousAttempts, isTransparentRetry);
+                newOptions, newHeaders, previousAttempts, isTransparentRetry, isHedging);
             Context origContext = context.attach();
             try {
               return delayedTransport.newStream(method, newHeaders, newOptions, tracers);

core/src/main/java/io/grpc/internal/OobChannel.java

@@ -88,7 +88,8 @@ final class OobChannel extends ManagedChannel implements InternalInstrumented<Ch
     public ClientStream newStream(MethodDescriptor<?, ?> method,
         CallOptions callOptions, Metadata headers, Context context) {
       ClientStreamTracer[] tracers = GrpcUtil.getClientStreamTracers(
-          callOptions, headers, 0, /* isTransparentRetry= */ false);
+          callOptions, headers, 0, /* isTransparentRetry= */ false,
+          /* isHedging= */ false);
       Context origContext = context.attach();
       // delayed transport's newStream() always acquires a lock, but concurrent performance doesn't
       // matter here because OOB communication should be sparse, and it's not on application RPC's

core/src/main/java/io/grpc/internal/RetriableStream.java

@@ -266,7 +266,8 @@ public ClientStreamTracer newClientStreamTracer(
 
     Metadata newHeaders = updateHeaders(headers, previousAttemptCount);
     // NOTICE: This set _must_ be done before stream.start() and it actually is.
-    sub.stream = newSubstream(newHeaders, tracerFactory, previousAttemptCount, isTransparentRetry);
+    sub.stream = newSubstream(newHeaders, tracerFactory, previousAttemptCount, isTransparentRetry,
+        isHedging);
     return sub;
   }
 
@@ -276,7 +277,7 @@ public ClientStreamTracer newClientStreamTracer(
    */
   abstract ClientStream newSubstream(
       Metadata headers, ClientStreamTracer.Factory tracerFactory, int previousAttempts,
-      boolean isTransparentRetry);
+      boolean isTransparentRetry, boolean isHedging);
 
   /** Adds grpc-previous-rpc-attempts in the headers of a retry/hedging RPC. */
   @VisibleForTesting

core/src/main/java/io/grpc/internal/SubchannelChannel.java

@@ -59,7 +59,8 @@ public ClientStream newStream(MethodDescriptor<?, ?> method,
           transport = notReadyTransport;
         }
         ClientStreamTracer[] tracers = GrpcUtil.getClientStreamTracers(
-            callOptions, headers, 0, /* isTransparentRetry= */ false);
+            callOptions, headers, 0, /* isTransparentRetry= */ false,
+            /* isHedging= */ false);
         Context origContext = context.attach();
         try {
           return transport.newStream(method, headers, callOptions, tracers);

core/src/test/java/io/grpc/internal/RetriableStreamTest.java

@@ -186,7 +186,8 @@ ClientStream newSubstream(
         Metadata metadata,
         ClientStreamTracer.Factory tracerFactory,
         int previousAttempts,
-        boolean isTransparentRetry) {
+        boolean isTransparentRetry,
+        boolean isHedging) {
       bufferSizeTracer =
           tracerFactory.newClientStreamTracer(STREAM_INFO, metadata);
       int actualPreviousRpcAttemptsInHeader = metadata.get(GRPC_PREVIOUS_RPC_ATTEMPTS) == null

opentelemetry/src/main/java/io/grpc/opentelemetry/GrpcOpenTelemetry.java

@@ -18,8 +18,11 @@
 
 import static com.google.common.base.Preconditions.checkNotNull;
 import static io.grpc.internal.GrpcUtil.IMPLEMENTATION_VERSION;
+import static io.grpc.opentelemetry.internal.OpenTelemetryConstants.HEDGE_BUCKETS;
 import static io.grpc.opentelemetry.internal.OpenTelemetryConstants.LATENCY_BUCKETS;
+import static io.grpc.opentelemetry.internal.OpenTelemetryConstants.RETRY_BUCKETS;
 import static io.grpc.opentelemetry.internal.OpenTelemetryConstants.SIZE_BUCKETS;
+import static io.grpc.opentelemetry.internal.OpenTelemetryConstants.TRANSPARENT_RETRY_BUCKETS;
 
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Stopwatch;
@@ -64,8 +67,8 @@ public Stopwatch get() {
   };
 
   @VisibleForTesting
-  static boolean ENABLE_OTEL_TRACING = GrpcUtil.getFlag("GRPC_EXPERIMENTAL_ENABLE_OTEL_TRACING",
-      false);
+  static boolean ENABLE_OTEL_TRACING =
+      GrpcUtil.getFlag("GRPC_EXPERIMENTAL_ENABLE_OTEL_TRACING", false);
 
   private final OpenTelemetry openTelemetrySdk;
   private final MeterProvider meterProvider;
@@ -241,6 +244,54 @@ static OpenTelemetryMetricsResource createMetricInstruments(Meter meter,
               .build());
     }
 
+    if (isMetricEnabled("grpc.client.call.retries", enableMetrics, disableDefault)) {
+      builder.clientCallRetriesCounter(
+          meter.histogramBuilder(
+                  "grpc.client.call.retries")
+              .setUnit("{retry}")
+              .setDescription("Number of retries during the client call. "
+                  + "If there were no retries, 0 is not reported.")
+              .ofLongs()
+              .setExplicitBucketBoundariesAdvice(RETRY_BUCKETS)
+              .build());
+    }
+
+    if (isMetricEnabled("grpc.client.call.transparent_retries", enableMetrics,
+        disableDefault)) {
+      builder.clientCallTransparentRetriesCounter(
+          meter.histogramBuilder(
+                  "grpc.client.call.transparent_retries")
+              .setUnit("{transparent_retry}")
+              .setDescription("Number of transparent retries during the client call. "
+                  + "If there were no transparent retries, 0 is not reported.")
+              .ofLongs()
+              .setExplicitBucketBoundariesAdvice(TRANSPARENT_RETRY_BUCKETS)
+              .build());
+    }
+
+    if (isMetricEnabled("grpc.client.call.hedges", enableMetrics, disableDefault)) {
+      builder.clientCallHedgesCounter(
+          meter.histogramBuilder(
+                  "grpc.client.call.hedges")
+              .setUnit("{hedge}")
+              .setDescription("Number of hedges during the client call. "
+                  + "If there were no hedges, 0 is not reported.")
+              .ofLongs()
+              .setExplicitBucketBoundariesAdvice(HEDGE_BUCKETS)
+              .build());
+    }
+
+    if (isMetricEnabled("grpc.client.call.retry_delay", enableMetrics, disableDefault)) {
+      builder.clientCallRetryDelayCounter(
+          meter.histogramBuilder(
+                  "grpc.client.call.retry_delay")
+              .setUnit("s")
+              .setDescription("Total time of delay while there is no active attempt during the "
+                  + "client call")
+              .setExplicitBucketBoundariesAdvice(LATENCY_BUCKETS)
+              .build());
+    }
+
     if (isMetricEnabled("grpc.server.call.started", enableMetrics, disableDefault)) {
       builder.serverCallCountCounter(
           meter.counterBuilder("grpc.server.call.started")
@@ -259,8 +310,8 @@ static OpenTelemetryMetricsResource createMetricInstruments(Meter meter,
               .build());
     }
 
-    if (isMetricEnabled("grpc.server.call.sent_total_compressed_message_size", enableMetrics,
-        disableDefault)) {
+    if (isMetricEnabled("grpc.server.call.sent_total_compressed_message_size",
+        enableMetrics, disableDefault)) {
       builder.serverTotalSentCompressedMessageSizeCounter(
           meter.histogramBuilder(
                   "grpc.server.call.sent_total_compressed_message_size")
@@ -271,8 +322,8 @@ static OpenTelemetryMetricsResource createMetricInstruments(Meter meter,
               .build());
     }
 
-    if (isMetricEnabled("grpc.server.call.rcvd_total_compressed_message_size", enableMetrics,
-        disableDefault)) {
+    if (isMetricEnabled("grpc.server.call.rcvd_total_compressed_message_size",
+        enableMetrics, disableDefault)) {
       builder.serverTotalReceivedCompressedMessageSizeCounter(
           meter.histogramBuilder(
                   "grpc.server.call.rcvd_total_compressed_message_size")

opentelemetry/src/main/java/io/grpc/opentelemetry/OpenTelemetryMetricsModule.java

@@ -71,6 +71,7 @@
  */
 final class OpenTelemetryMetricsModule {
   private static final Logger logger = Logger.getLogger(OpenTelemetryMetricsModule.class.getName());
+  private static final double NANOS_PER_SEC = 1_000_000_000.0;
   public static final ImmutableSet<String> DEFAULT_PER_CALL_METRICS_SET =
       ImmutableSet.of(
           "grpc.client.attempt.started",
@@ -292,9 +293,12 @@ static final class CallAttemptsTracerFactory extends ClientStreamTracer.Factory
     private final String fullMethodName;
     private final List<OpenTelemetryPlugin.ClientCallPlugin> callPlugins;
     private Status status;
+    private long retryDelayNanos;
     private long callLatencyNanos;
     private final Object lock = new Object();
     private final AtomicLong attemptsPerCall = new AtomicLong();
+    private final AtomicLong hedgedAttemptsPerCall = new AtomicLong();
+    private final AtomicLong transparentRetriesPerCall = new AtomicLong();
     @GuardedBy("lock")
     private int activeStreams;
     @GuardedBy("lock")
@@ -331,6 +335,7 @@ public ClientStreamTracer newClientStreamTracer(StreamInfo info, Metadata metada
         }
         if (++activeStreams == 1 && attemptStopwatch.isRunning()) {
           attemptStopwatch.stop();
+          retryDelayNanos = attemptStopwatch.elapsed(TimeUnit.NANOSECONDS);
         }
       }
       // Skip recording for the first time, since it is already recorded in
@@ -344,7 +349,11 @@ public ClientStreamTracer newClientStreamTracer(StreamInfo info, Metadata metada
           module.resource.clientAttemptCountCounter().add(1, attribute);
         }
       }
-      if (!info.isTransparentRetry()) {
+      if (info.isTransparentRetry()) {
+        transparentRetriesPerCall.incrementAndGet();
+      } else if (info.isHedging()) {
+        hedgedAttemptsPerCall.incrementAndGet();
+      } else {
         attemptsPerCall.incrementAndGet();
       }
       return newClientTracer(info);
@@ -407,14 +416,53 @@ void recordFinishedCall() {
         tracer.recordFinishedAttempt();
       }
       callLatencyNanos = callStopWatch.elapsed(TimeUnit.NANOSECONDS);
-      io.opentelemetry.api.common.Attributes attribute =
-          io.opentelemetry.api.common.Attributes.of(METHOD_KEY, fullMethodName,
-              TARGET_KEY, target,
-              STATUS_KEY, status.getCode().toString());
 
+      // Base attributes
+      io.opentelemetry.api.common.Attributes baseAttributes =
+          io.opentelemetry.api.common.Attributes.of(
+              METHOD_KEY, fullMethodName,
+              TARGET_KEY, target
+          );
+
+      // Duration
       if (module.resource.clientCallDurationCounter() != null) {
-        module.resource.clientCallDurationCounter()
-            .record(callLatencyNanos * SECONDS_PER_NANO, attribute);
+        module.resource.clientCallDurationCounter().record(
+            callLatencyNanos * SECONDS_PER_NANO,
+            baseAttributes.toBuilder()
+                .put(STATUS_KEY, status.getCode().toString())
+                .build()
+        );
+      }
+
+      // Retry counts
+      if (module.resource.clientCallRetriesCounter() != null) {
+        long retriesPerCall = attemptsPerCall.get() - 1 >= 0 ? attemptsPerCall.get() - 1 : 0;
+        if (retriesPerCall > 0) {
+          module.resource.clientCallRetriesCounter().record(retriesPerCall, baseAttributes);
+        }
+      }
+
+      // Hedge counts
+      if (module.resource.clientCallHedgesCounter() != null) {
+        if (hedgedAttemptsPerCall.get() > 0) {
+          module.resource.clientCallHedgesCounter()
+              .record(hedgedAttemptsPerCall.get(), baseAttributes);
+        }
+      }
+
+      // Transparent Retry counts
+      if (module.resource.clientCallTransparentRetriesCounter() != null
+          && transparentRetriesPerCall.get() > 0) {
+        module.resource.clientCallTransparentRetriesCounter().record(
+            transparentRetriesPerCall.get(), baseAttributes);
+      }
+
+      // Retry delay
+      if (module.resource.clientCallRetryDelayCounter() != null) {
+        module.resource.clientCallRetryDelayCounter().record(
+            retryDelayNanos / NANOS_PER_SEC,
+            baseAttributes
+        );
       }
     }
   }