#20365

Committed 27 Jul 2026 06:04AM UTC coverage: 89.199% (+0.07%) from 89.125%

Build # #20365

Build Type

push

github

Committed by

web-flow

Commit Message

core: Implement LB Delay Observability (Proposal A121) (#12807)

This PR implements **Attempt-Level RPC Delay Observability** across the core channel transport, built-in load balancers, xDS policies, and the OpenTelemetry telemetry plugin, aligned with [gRPC Proposal A121](https://github.com/grpc/proposal/pull/556).

Coverage Stats

38276 of 42911 relevant lines covered (89.2%)

0.89 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

94.04

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

/*
 * Copyright 2024 The gRPC Authors
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package io.grpc.opentelemetry;

import static com.google.common.base.Preconditions.checkNotNull;
import static io.grpc.ClientStreamTracer.NAME_RESOLUTION_DELAYED;
import static io.grpc.internal.GrpcUtil.IMPLEMENTATION_VERSION;
import static io.grpc.opentelemetry.internal.OpenTelemetryConstants.BAGGAGE_KEY;

import com.google.common.annotations.VisibleForTesting;
import io.grpc.Attributes;
import io.grpc.CallOptions;
import io.grpc.Channel;
import io.grpc.ClientCall;
import io.grpc.ClientInterceptor;
import io.grpc.ClientStreamTracer;
import io.grpc.ForwardingClientCall.SimpleForwardingClientCall;
import io.grpc.ForwardingClientCallListener.SimpleForwardingClientCallListener;
import io.grpc.ForwardingServerCallListener;
import io.grpc.Metadata;
import io.grpc.MethodDescriptor;
import io.grpc.ServerCall;
import io.grpc.ServerCallHandler;
import io.grpc.ServerInterceptor;
import io.grpc.ServerStreamTracer;
import io.grpc.internal.GrpcUtil;
import io.grpc.opentelemetry.internal.OpenTelemetryConstants;
import io.opentelemetry.api.OpenTelemetry;
import io.opentelemetry.api.baggage.Baggage;
import io.opentelemetry.api.common.AttributeKey;
import io.opentelemetry.api.common.AttributesBuilder;
import io.opentelemetry.api.trace.Span;
import io.opentelemetry.api.trace.StatusCode;
import io.opentelemetry.api.trace.Tracer;
import io.opentelemetry.context.Context;
import io.opentelemetry.context.Scope;
import io.opentelemetry.context.propagation.ContextPropagators;
import java.util.Objects;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import java.util.logging.Level;
import java.util.logging.Logger;
import javax.annotation.Nullable;

/**
 * Provides factories for {@link io.grpc.StreamTracer} that records tracing to OpenTelemetry.
 */
final class OpenTelemetryTracingModule {
  private static final Logger logger = Logger.getLogger(OpenTelemetryTracingModule.class.getName());

  @VisibleForTesting
  final io.grpc.Context.Key<Span> otelSpan = io.grpc.Context.key("opentelemetry-span-key");

  @Nullable
  private static final AtomicIntegerFieldUpdater<CallAttemptsTracerFactory> callEndedUpdater;
  @Nullable
  private static final AtomicIntegerFieldUpdater<ServerTracer> streamClosedUpdater;

  /*
   * When using Atomic*FieldUpdater, some Samsung Android 5.0.x devices encounter a bug in their JDK
   * reflection API that triggers a NoSuchFieldException. When this occurs, we fallback to
   * (potentially racy) direct updates of the volatile variables.
   */
  static {
    AtomicIntegerFieldUpdater<CallAttemptsTracerFactory> tmpCallEndedUpdater;
    AtomicIntegerFieldUpdater<ServerTracer> tmpStreamClosedUpdater;
    try {
      tmpCallEndedUpdater =
          AtomicIntegerFieldUpdater.newUpdater(CallAttemptsTracerFactory.class, "callEnded");
      tmpStreamClosedUpdater =
          AtomicIntegerFieldUpdater.newUpdater(ServerTracer.class, "streamClosed");
    } catch (Throwable t) {
      logger.log(Level.SEVERE, "Creating atomic field updaters failed", t);
      tmpCallEndedUpdater = null;
      tmpStreamClosedUpdater = null;
    }
    callEndedUpdater = tmpCallEndedUpdater;
    streamClosedUpdater = tmpStreamClosedUpdater;
  }

  private final Tracer otelTracer;
  private final ContextPropagators contextPropagators;
  private final MetadataGetter metadataGetter = MetadataGetter.getInstance();
  private final MetadataSetter metadataSetter = MetadataSetter.getInstance();
  private final TracingClientInterceptor clientInterceptor = new TracingClientInterceptor();
  private final ServerInterceptor serverSpanPropagationInterceptor =
      new TracingServerSpanPropagationInterceptor();
  private final ServerTracerFactory serverTracerFactory = new ServerTracerFactory();

  OpenTelemetryTracingModule(OpenTelemetry openTelemetry) {
    this.otelTracer = checkNotNull(openTelemetry.getTracerProvider(), "tracerProvider")
        .tracerBuilder(OpenTelemetryConstants.INSTRUMENTATION_SCOPE)
        .setInstrumentationVersion(IMPLEMENTATION_VERSION)
        .build();
    this.contextPropagators = checkNotNull(openTelemetry.getPropagators(), "contextPropagators");
  }

  @VisibleForTesting
  Tracer getTracer() {
    return otelTracer;
  }

  /**
   * Creates a {@link CallAttemptsTracerFactory} for a new call.
   */
  @VisibleForTesting
  CallAttemptsTracerFactory newClientCallTracer(Span clientSpan, MethodDescriptor<?, ?> method) {
    return new CallAttemptsTracerFactory(clientSpan, method);
  }

  /**
   * Returns the server tracer factory.
   */
  ServerStreamTracer.Factory getServerTracerFactory() {
    return serverTracerFactory;
  }

  /**
   * Returns the client interceptor that facilitates otel tracing reporting.
   */
  ClientInterceptor getClientInterceptor() {
    return clientInterceptor;
  }

  ServerInterceptor getServerSpanPropagationInterceptor() {
    return serverSpanPropagationInterceptor;
  }

  @VisibleForTesting
  final class CallAttemptsTracerFactory extends ClientStreamTracer.Factory {
    volatile int callEnded;
    private final Span clientSpan;
    private final String fullMethodName;

    CallAttemptsTracerFactory(Span clientSpan, MethodDescriptor<?, ?> method) {
      checkNotNull(method, "method");
      this.fullMethodName = checkNotNull(method.getFullMethodName(), "fullMethodName");
      this.clientSpan = checkNotNull(clientSpan, "clientSpan");
    }

    @Override
    public ClientStreamTracer newClientStreamTracer(
        ClientStreamTracer.StreamInfo info, Metadata headers) {
      Span attemptSpan = otelTracer.spanBuilder(
              "Attempt." + fullMethodName.replace('/', '.'))
          .setParent(Context.current().with(clientSpan))
          .startSpan();
      attemptSpan.setAttribute(
          "previous-rpc-attempts", info.getPreviousAttempts());
      attemptSpan.setAttribute(
          "transparent-retry",info.isTransparentRetry());
      if (info.getCallOptions().getOption(NAME_RESOLUTION_DELAYED) != null) {
        clientSpan.addEvent("Delayed name resolution complete");
      }
      return new ClientTracer(attemptSpan, clientSpan);
    }

    /**
     * Record a finished call and mark the current time as the end time.
     *
     * <p>Can be called from any thread without synchronization.  Calling it the second time or more
     * is a no-op.
     */
    void callEnded(io.grpc.Status status) {
      if (callEndedUpdater != null) {
        if (callEndedUpdater.getAndSet(this, 1) != 0) {
          return;
        }
      } else {
        if (callEnded != 0) {
          return;
        }
        callEnded = 1;
      }
      endSpanWithStatus(clientSpan, status);
    }
  }

  private final class ClientTracer extends ClientStreamTracer {
    private final Span span;
    private final Span parentSpan;
    volatile int seqNo;
    boolean isPendingStream;
    @Nullable private volatile Span activeDelaySpan;
    @Nullable private volatile String activeDelayType;

    ClientTracer(Span span, Span parentSpan) {
      this.span = checkNotNull(span, "span");
      this.parentSpan = checkNotNull(parentSpan, "parent span");
    }

    @Override
    public void streamCreated(Attributes transportAtts, Metadata headers) {
      recordAttemptDelayEnd();
      contextPropagators.getTextMapPropagator().inject(Context.current().with(span), headers,
          metadataSetter);
      if (isPendingStream) {
        span.addEvent("Delayed LB pick complete");
      }
    }

    @Override
    public void createPendingStream() {
      isPendingStream = true;
    }

    @Override
    public void recordAttemptDelayStart(String delayType, String delayReason) {
      if (!GrpcOpenTelemetry.isDelayObservabilityEnabled()) {
        return;
      }
      if (activeDelaySpan != null && Objects.equals(activeDelayType, delayType)) {
        // Do not recreate the span if the delay type is unchanged (e.g., priority failover).
        recordAttemptDelayReasonChanged(delayReason);
        return;
      }
      // Close any previous delay segment before starting a new canonical segment.
      recordAttemptDelayEnd();
      activeDelayType = delayType;
      // All attempt queuing segments use the strict child span name "Attempt Delay".
      Span delaySpan = otelTracer.spanBuilder("Attempt Delay")
          .setParent(Context.current().with(span))
          .setAttribute("grpc.delay_type", delayType)
          .startSpan();
      activeDelaySpan = delaySpan;
      delaySpan.addEvent(
          "Delay state transition",
          io.opentelemetry.api.common.Attributes.of(
              AttributeKey.stringKey("grpc.delay_type"), delayType,
              AttributeKey.stringKey("grpc.delay_reason"), delayReason));
    }

    @Override
    public void recordAttemptDelayReasonChanged(String delayReason) {
      if (!GrpcOpenTelemetry.isDelayObservabilityEnabled() || activeDelaySpan == null) {
        return;
      }
      String type = activeDelayType;
      activeDelaySpan.addEvent(
          "Delay state transition",
          io.opentelemetry.api.common.Attributes.of(
              AttributeKey.stringKey("grpc.delay_type"), type != null ? type : "",
              AttributeKey.stringKey("grpc.delay_reason"), delayReason));
    }

    @Override
    public void recordAttemptDelayEnd() {
      Span delaySpan = activeDelaySpan;
      if (delaySpan != null) {
        // End active child span upon pick completion or transport cancellation.
        delaySpan.end();
        activeDelaySpan = null;
        activeDelayType = null;
      }
    }

    @Override
    public void outboundMessageSent(
        int seqNo, long optionalWireSize, long optionalUncompressedSize) {
      recordOutboundMessageSentEvent(span, seqNo, optionalWireSize, optionalUncompressedSize);
    }

    @Override
    public void inboundMessageRead(
        int seqNo, long optionalWireSize, long optionalUncompressedSize) {
      if (optionalWireSize != optionalUncompressedSize) {
        recordInboundCompressedMessage(span, seqNo, optionalWireSize);
      }
    }

    @Override
    public void inboundMessage(int seqNo) {
      this.seqNo = seqNo;
    }

    @Override
    public void inboundUncompressedSize(long bytes) {
      recordInboundMessageSize(parentSpan, seqNo, bytes);
    }

    @Override
    public void streamClosed(io.grpc.Status status) {
      recordAttemptDelayEnd();
      endSpanWithStatus(span, status);
    }
  }

  private final class ServerTracer extends ServerStreamTracer {
    private final Span span;
    volatile int streamClosed;
    private int seqNo;
    private Baggage baggage;

    ServerTracer(String fullMethodName, @Nullable Span remoteSpan, Baggage baggage) {
      checkNotNull(fullMethodName, "fullMethodName");
      this.span =
          otelTracer.spanBuilder(generateTraceSpanName(true, fullMethodName))
              .setParent(remoteSpan == null ? null : Context.current().with(remoteSpan))
              .startSpan();
      this.baggage = baggage;
    }

    /**
     * Record a finished stream and mark the current time as the end time.
     *
     * <p>Can be called from any thread without synchronization.  Calling it the second time or more
     * is a no-op.
     */
    @Override
    public void streamClosed(io.grpc.Status status) {
      if (streamClosedUpdater != null) {
        if (streamClosedUpdater.getAndSet(this, 1) != 0) {
          return;
        }
      } else {
        if (streamClosed != 0) {
          return;
        }
        streamClosed = 1;
      }
      endSpanWithStatus(span, status);
    }

    @Override
    public io.grpc.Context filterContext(io.grpc.Context context) {
      return context
          .withValue(otelSpan, span)
          .withValue(BAGGAGE_KEY, baggage);
    }

    @Override
    public void outboundMessageSent(
        int seqNo, long optionalWireSize, long optionalUncompressedSize) {
      recordOutboundMessageSentEvent(span, seqNo, optionalWireSize, optionalUncompressedSize);
    }

    @Override
    public void inboundMessageRead(
        int seqNo, long optionalWireSize, long optionalUncompressedSize) {
      if (optionalWireSize != optionalUncompressedSize) {
        recordInboundCompressedMessage(span, seqNo, optionalWireSize);
      }
    }

    @Override
    public void inboundMessage(int seqNo) {
      this.seqNo = seqNo;
    }

    @Override
    public void inboundUncompressedSize(long bytes) {
      recordInboundMessageSize(span, seqNo, bytes);
    }
  }

  @VisibleForTesting
  final class ServerTracerFactory extends ServerStreamTracer.Factory {
    @SuppressWarnings("ReferenceEquality")
    @Override
    public ServerStreamTracer newServerStreamTracer(String fullMethodName, Metadata headers) {
      Context context = contextPropagators.getTextMapPropagator().extract(
          Context.current(), headers, metadataGetter
      );
      Span remoteSpan = Span.fromContext(context);
      if (remoteSpan == Span.getInvalid()) {
        remoteSpan = null;
      }
      Baggage baggage = Baggage.fromContext(context);
      return new ServerTracer(fullMethodName, remoteSpan, baggage);
    }
  }

  @VisibleForTesting
  final class TracingServerSpanPropagationInterceptor implements ServerInterceptor {
    @Override
    public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(ServerCall<ReqT, RespT> call,
        Metadata headers, ServerCallHandler<ReqT, RespT> next) {
      Span span = otelSpan.get(io.grpc.Context.current());
      if (span == null) {
        logger.log(Level.FINE, "Server span not found. ServerTracerFactory for server "
            + "tracing must be set.");
        return next.startCall(call, headers);
      }
      Context serverCallContext = Context.current();
      serverCallContext = serverCallContext.with(span);
      Baggage baggage = BAGGAGE_KEY.get();
      if (baggage != null) {
        serverCallContext = serverCallContext.with(baggage);
      } else {
        logger.log(Level.WARNING, "Server baggage not found which is unexpected, "
            + "as it is being added unconditionally in filterContext().");
      }
      try (Scope scope = serverCallContext.makeCurrent()) {
        return new ContextServerCallListener<>(next.startCall(call, headers), serverCallContext);
      }
    }
  }

  private static class ContextServerCallListener<ReqT> extends
      ForwardingServerCallListener.SimpleForwardingServerCallListener<ReqT> {
    private final Context context;

    protected ContextServerCallListener(ServerCall.Listener<ReqT> delegate, Context context) {
      super(delegate);
      this.context = checkNotNull(context, "context");
    }

    @Override
    public void onMessage(ReqT message) {
      try (Scope scope = context.makeCurrent()) {
        delegate().onMessage(message);
      }
    }

    @Override
    public void onHalfClose() {
      try (Scope scope = context.makeCurrent()) {
        delegate().onHalfClose();
      }
    }

    @Override
    public void onCancel() {
      try (Scope scope = context.makeCurrent()) {
        delegate().onCancel();
      }
    }

    @Override
    public void onComplete() {
      try (Scope scope = context.makeCurrent()) {
        delegate().onComplete();
      }
    }

    @Override
    public void onReady() {
      try (Scope scope = context.makeCurrent()) {
        delegate().onReady();
      }
    }
  }

  @VisibleForTesting
  final class TracingClientInterceptor implements ClientInterceptor {

    @Override
    public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
        MethodDescriptor<ReqT, RespT> method, CallOptions callOptions, Channel next) {
      Span clientSpan = otelTracer.spanBuilder(
          generateTraceSpanName(false, method.getFullMethodName()))
          .startSpan();

      final CallAttemptsTracerFactory tracerFactory = newClientCallTracer(clientSpan, method);
      ClientCall<ReqT, RespT> call =
          next.newCall(
              method,
              callOptions.withStreamTracerFactory(tracerFactory));
      return new SimpleForwardingClientCall<ReqT, RespT>(call) {
        @Override
        public void start(Listener<RespT> responseListener, Metadata headers) {
          delegate().start(
              new SimpleForwardingClientCallListener<RespT>(responseListener) {
                @Override
                public void onClose(io.grpc.Status status, Metadata trailers) {
                  tracerFactory.callEnded(status);
                  super.onClose(status, trailers);
                }
              },
              headers);
        }
      };
    }
  }

  // Attribute named "message-size" always means the message size the application sees.
  // If there was compression, additional event reports "message-size-compressed".
  //
  // An example trace with message compression:
  //
  // Sending:
  // |-- Event 'Outbound message sent', attributes('sequence-numer' = 0, 'message-size' = 7854,
  //                                               'message-size-compressed' = 5493) ----|
  //
  // Receiving:
  // |-- Event 'Inbound compressed message', attributes('sequence-numer' = 0,
  //                                                    'message-size-compressed' = 5493 ) ----|
  // |-- Event 'Inbound message received', attributes('sequence-numer' = 0,
  //                                                  'message-size' = 7854) ----|
  //
  // An example trace with no message compression:
  //
  // Sending:
  // |-- Event 'Outbound message sent', attributes('sequence-numer' = 0, 'message-size' = 7854) ---|
  //
  // Receiving:
  // |-- Event 'Inbound message received', attributes('sequence-numer' = 0,
  //                                                  'message-size' = 7854) ----|
  private void recordOutboundMessageSentEvent(Span span,
      int seqNo, long optionalWireSize, long optionalUncompressedSize) {
    AttributesBuilder attributesBuilder = io.opentelemetry.api.common.Attributes.builder();
    attributesBuilder.put("sequence-number", seqNo);
    if (optionalUncompressedSize != -1) {
      attributesBuilder.put("message-size", optionalUncompressedSize);
    }
    if (optionalWireSize != -1 && optionalWireSize != optionalUncompressedSize) {
      attributesBuilder.put("message-size-compressed", optionalWireSize);
    }
    span.addEvent("Outbound message", attributesBuilder.build());
  }

  private void recordInboundCompressedMessage(Span span, int seqNo, long optionalWireSize) {
    AttributesBuilder attributesBuilder = io.opentelemetry.api.common.Attributes.builder();
    attributesBuilder.put("sequence-number", seqNo);
    attributesBuilder.put("message-size-compressed", optionalWireSize);
    span.addEvent("Inbound compressed message", attributesBuilder.build());
  }

  private void recordInboundMessageSize(Span span, int seqNo, long bytes) {
    AttributesBuilder attributesBuilder = io.opentelemetry.api.common.Attributes.builder();
    attributesBuilder.put("sequence-number", seqNo);
    attributesBuilder.put("message-size", bytes);
    span.addEvent("Inbound message", attributesBuilder.build());
  }

  private void endSpanWithStatus(Span span, io.grpc.Status status) {
    if (status.isOk()) {
      span.setStatus(StatusCode.OK);
    } else {
      span.setStatus(StatusCode.ERROR, GrpcUtil.statusToPrettyString(status));
    }
    span.end();
  }

  /**
   * Convert a full method name to a tracing span name.
   *
   * @param isServer {@code false} if the span is on the client-side, {@code true} if on the
   *                 server-side
   * @param fullMethodName the method name as returned by
   *        {@link MethodDescriptor#getFullMethodName}.
   */
  @VisibleForTesting
  static String generateTraceSpanName(boolean isServer, String fullMethodName) {
    String prefix = isServer ? "Recv" : "Sent";
    return prefix + "." + fullMethodName.replace('/', '.');
  }
}

1	/*
2	* Copyright 2024 The gRPC Authors
3	*
4	* Licensed under the Apache License, Version 2.0 (the "License");
5	* you may not use this file except in compliance with the License.
6	* You may obtain a copy of the License at
7	*
8	* http://www.apache.org/licenses/LICENSE-2.0
9	*
10	* Unless required by applicable law or agreed to in writing, software
11	* distributed under the License is distributed on an "AS IS" BASIS,
12	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13	* See the License for the specific language governing permissions and
14	* limitations under the License.
15	*/
16
17	package io.grpc.opentelemetry;
18
19	import static com.google.common.base.Preconditions.checkNotNull;
20	import static io.grpc.ClientStreamTracer.NAME_RESOLUTION_DELAYED;
21	import static io.grpc.internal.GrpcUtil.IMPLEMENTATION_VERSION;
22	import static io.grpc.opentelemetry.internal.OpenTelemetryConstants.BAGGAGE_KEY;
23
24	import com.google.common.annotations.VisibleForTesting;
25	import io.grpc.Attributes;
26	import io.grpc.CallOptions;
27	import io.grpc.Channel;
28	import io.grpc.ClientCall;
29	import io.grpc.ClientInterceptor;
30	import io.grpc.ClientStreamTracer;
31	import io.grpc.ForwardingClientCall.SimpleForwardingClientCall;
32	import io.grpc.ForwardingClientCallListener.SimpleForwardingClientCallListener;
33	import io.grpc.ForwardingServerCallListener;
34	import io.grpc.Metadata;
35	import io.grpc.MethodDescriptor;
36	import io.grpc.ServerCall;
37	import io.grpc.ServerCallHandler;
38	import io.grpc.ServerInterceptor;
39	import io.grpc.ServerStreamTracer;
40	import io.grpc.internal.GrpcUtil;
41	import io.grpc.opentelemetry.internal.OpenTelemetryConstants;
42	import io.opentelemetry.api.OpenTelemetry;
43	import io.opentelemetry.api.baggage.Baggage;
44	import io.opentelemetry.api.common.AttributeKey;
45	import io.opentelemetry.api.common.AttributesBuilder;
46	import io.opentelemetry.api.trace.Span;
47	import io.opentelemetry.api.trace.StatusCode;
48	import io.opentelemetry.api.trace.Tracer;
49	import io.opentelemetry.context.Context;
50	import io.opentelemetry.context.Scope;
51	import io.opentelemetry.context.propagation.ContextPropagators;
52	import java.util.Objects;
53	import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
54	import java.util.logging.Level;
55	import java.util.logging.Logger;
56	import javax.annotation.Nullable;
57
58	/**
59	* Provides factories for {@link io.grpc.StreamTracer} that records tracing to OpenTelemetry.
60	*/
61	final class OpenTelemetryTracingModule {
62	private static final Logger logger = Logger.getLogger(OpenTelemetryTracingModule.class.getName());	1✔
63
64	@VisibleForTesting	1✔
65	final io.grpc.Context.Key<Span> otelSpan = io.grpc.Context.key("opentelemetry-span-key");	1✔
66
67	@Nullable
68	private static final AtomicIntegerFieldUpdater<CallAttemptsTracerFactory> callEndedUpdater;
69	@Nullable
70	private static final AtomicIntegerFieldUpdater<ServerTracer> streamClosedUpdater;
71
72	/*
73	* When using Atomic*FieldUpdater, some Samsung Android 5.0.x devices encounter a bug in their JDK
74	* reflection API that triggers a NoSuchFieldException. When this occurs, we fallback to
75	* (potentially racy) direct updates of the volatile variables.
76	*/
77	static {
78	AtomicIntegerFieldUpdater<CallAttemptsTracerFactory> tmpCallEndedUpdater;
79	AtomicIntegerFieldUpdater<ServerTracer> tmpStreamClosedUpdater;
80	try {
81	tmpCallEndedUpdater =	1✔
82	AtomicIntegerFieldUpdater.newUpdater(CallAttemptsTracerFactory.class, "callEnded");	1✔
83	tmpStreamClosedUpdater =	1✔
84	AtomicIntegerFieldUpdater.newUpdater(ServerTracer.class, "streamClosed");	1✔
85	} catch (Throwable t) {	×
86	logger.log(Level.SEVERE, "Creating atomic field updaters failed", t);	×
87	tmpCallEndedUpdater = null;	×
88	tmpStreamClosedUpdater = null;	×
89	}	1✔
90	callEndedUpdater = tmpCallEndedUpdater;	1✔
91	streamClosedUpdater = tmpStreamClosedUpdater;	1✔
92	}	1✔
93
94	private final Tracer otelTracer;
95	private final ContextPropagators contextPropagators;
96	private final MetadataGetter metadataGetter = MetadataGetter.getInstance();	1✔
97	private final MetadataSetter metadataSetter = MetadataSetter.getInstance();	1✔
98	private final TracingClientInterceptor clientInterceptor = new TracingClientInterceptor();	1✔
99	private final ServerInterceptor serverSpanPropagationInterceptor =	1✔
100	new TracingServerSpanPropagationInterceptor();
101	private final ServerTracerFactory serverTracerFactory = new ServerTracerFactory();	1✔
102
103	OpenTelemetryTracingModule(OpenTelemetry openTelemetry) {	1✔
104	this.otelTracer = checkNotNull(openTelemetry.getTracerProvider(), "tracerProvider")	1✔
105	.tracerBuilder(OpenTelemetryConstants.INSTRUMENTATION_SCOPE)	1✔
106	.setInstrumentationVersion(IMPLEMENTATION_VERSION)	1✔
107	.build();	1✔
108	this.contextPropagators = checkNotNull(openTelemetry.getPropagators(), "contextPropagators");	1✔
109	}	1✔
110
111	@VisibleForTesting
112	Tracer getTracer() {
113	return otelTracer;	1✔
114	}
115
116	/**
117	* Creates a {@link CallAttemptsTracerFactory} for a new call.
118	*/
119	@VisibleForTesting
120	CallAttemptsTracerFactory newClientCallTracer(Span clientSpan, MethodDescriptor<?, ?> method) {
121	return new CallAttemptsTracerFactory(clientSpan, method);	1✔
122	}
123
124	/**
125	* Returns the server tracer factory.
126	*/
127	ServerStreamTracer.Factory getServerTracerFactory() {
128	return serverTracerFactory;	1✔
129	}
130
131	/**
132	* Returns the client interceptor that facilitates otel tracing reporting.
133	*/
134	ClientInterceptor getClientInterceptor() {
135	return clientInterceptor;	1✔
136	}
137
138	ServerInterceptor getServerSpanPropagationInterceptor() {
139	return serverSpanPropagationInterceptor;	1✔
140	}
141
142	@VisibleForTesting
143	final class CallAttemptsTracerFactory extends ClientStreamTracer.Factory {
144	volatile int callEnded;
145	private final Span clientSpan;
146	private final String fullMethodName;
147
148	CallAttemptsTracerFactory(Span clientSpan, MethodDescriptor<?, ?> method) {	1✔
149	checkNotNull(method, "method");	1✔
150	this.fullMethodName = checkNotNull(method.getFullMethodName(), "fullMethodName");	1✔
151	this.clientSpan = checkNotNull(clientSpan, "clientSpan");	1✔
152	}	1✔
153
154	@Override
155	public ClientStreamTracer newClientStreamTracer(
156	ClientStreamTracer.StreamInfo info, Metadata headers) {
157	Span attemptSpan = otelTracer.spanBuilder(	1✔
158	"Attempt." + fullMethodName.replace('/', '.'))	1✔
159	.setParent(Context.current().with(clientSpan))	1✔
160	.startSpan();	1✔
161	attemptSpan.setAttribute(	1✔
162	"previous-rpc-attempts", info.getPreviousAttempts());	1✔
163	attemptSpan.setAttribute(	1✔
164	"transparent-retry",info.isTransparentRetry());	1✔
165	if (info.getCallOptions().getOption(NAME_RESOLUTION_DELAYED) != null) {	1✔
166	clientSpan.addEvent("Delayed name resolution complete");	1✔
167	}
168	return new ClientTracer(attemptSpan, clientSpan);	1✔
169	}
170
171	/**
172	* Record a finished call and mark the current time as the end time.
173	*
174	* <p>Can be called from any thread without synchronization. Calling it the second time or more
175	* is a no-op.
176	*/
177	void callEnded(io.grpc.Status status) {
178	if (callEndedUpdater != null) {	1✔
179	if (callEndedUpdater.getAndSet(this, 1) != 0) {	1✔
180	return;	×
181	}
182	} else {
183	if (callEnded != 0) {	×
184	return;	×
185	}
186	callEnded = 1;	×
187	}
188	endSpanWithStatus(clientSpan, status);	1✔
189	}	1✔
190	}
191
192	private final class ClientTracer extends ClientStreamTracer {
193	private final Span span;
194	private final Span parentSpan;
195	volatile int seqNo;
196	boolean isPendingStream;
197	@Nullable private volatile Span activeDelaySpan;
198	@Nullable private volatile String activeDelayType;
199
200	ClientTracer(Span span, Span parentSpan) {	1✔
201	this.span = checkNotNull(span, "span");	1✔
202	this.parentSpan = checkNotNull(parentSpan, "parent span");	1✔
203	}	1✔
204
205	@Override
206	public void streamCreated(Attributes transportAtts, Metadata headers) {
207	recordAttemptDelayEnd();	1✔
208	contextPropagators.getTextMapPropagator().inject(Context.current().with(span), headers,	1✔
209	metadataSetter);	1✔
210	if (isPendingStream) {	1✔
211	span.addEvent("Delayed LB pick complete");	1✔
212	}
213	}	1✔
214
215	@Override
216	public void createPendingStream() {
217	isPendingStream = true;	1✔
218	}	1✔
219
220	@Override
221	public void recordAttemptDelayStart(String delayType, String delayReason) {
222	if (!GrpcOpenTelemetry.isDelayObservabilityEnabled()) {	1✔
223	return;	1✔
224	}
225	if (activeDelaySpan != null && Objects.equals(activeDelayType, delayType)) {	1✔
226	// Do not recreate the span if the delay type is unchanged (e.g., priority failover).
227	recordAttemptDelayReasonChanged(delayReason);	1✔
228	return;	1✔
229	}
230	// Close any previous delay segment before starting a new canonical segment.
231	recordAttemptDelayEnd();	1✔
232	activeDelayType = delayType;	1✔
233	// All attempt queuing segments use the strict child span name "Attempt Delay".
234	Span delaySpan = otelTracer.spanBuilder("Attempt Delay")	1✔
235	.setParent(Context.current().with(span))	1✔
236	.setAttribute("grpc.delay_type", delayType)	1✔
237	.startSpan();	1✔
238	activeDelaySpan = delaySpan;	1✔
239	delaySpan.addEvent(	1✔
240	"Delay state transition",
241	io.opentelemetry.api.common.Attributes.of(	1✔
242	AttributeKey.stringKey("grpc.delay_type"), delayType,	1✔
243	AttributeKey.stringKey("grpc.delay_reason"), delayReason));	1✔
244	}	1✔
245
246	@Override
247	public void recordAttemptDelayReasonChanged(String delayReason) {
248	if (!GrpcOpenTelemetry.isDelayObservabilityEnabled() \|\| activeDelaySpan == null) {	1✔
249	return;	×
250	}
251	String type = activeDelayType;	1✔
252	activeDelaySpan.addEvent(	1✔
253	"Delay state transition",
254	io.opentelemetry.api.common.Attributes.of(	1✔
255	AttributeKey.stringKey("grpc.delay_type"), type != null ? type : "",	1✔
256	AttributeKey.stringKey("grpc.delay_reason"), delayReason));	1✔
257	}	1✔
258
259	@Override
260	public void recordAttemptDelayEnd() {
261	Span delaySpan = activeDelaySpan;	1✔
262	if (delaySpan != null) {	1✔
263	// End active child span upon pick completion or transport cancellation.
264	delaySpan.end();	1✔
265	activeDelaySpan = null;	1✔
266	activeDelayType = null;	1✔
267	}
268	}	1✔
269
270	@Override
271	public void outboundMessageSent(
272	int seqNo, long optionalWireSize, long optionalUncompressedSize) {
273	recordOutboundMessageSentEvent(span, seqNo, optionalWireSize, optionalUncompressedSize);	1✔
274	}	1✔
275
276	@Override
277	public void inboundMessageRead(
278	int seqNo, long optionalWireSize, long optionalUncompressedSize) {
279	if (optionalWireSize != optionalUncompressedSize) {	1✔
280	recordInboundCompressedMessage(span, seqNo, optionalWireSize);	1✔
281	}
282	}	1✔
283
284	@Override
285	public void inboundMessage(int seqNo) {
286	this.seqNo = seqNo;	1✔
287	}	1✔
288
289	@Override
290	public void inboundUncompressedSize(long bytes) {
291	recordInboundMessageSize(parentSpan, seqNo, bytes);	1✔
292	}	1✔
293
294	@Override
295	public void streamClosed(io.grpc.Status status) {
296	recordAttemptDelayEnd();	1✔
297	endSpanWithStatus(span, status);	1✔
298	}	1✔
299	}
300
301	private final class ServerTracer extends ServerStreamTracer {
302	private final Span span;
303	volatile int streamClosed;
304	private int seqNo;
305	private Baggage baggage;
306
307	ServerTracer(String fullMethodName, @Nullable Span remoteSpan, Baggage baggage) {	1✔
308	checkNotNull(fullMethodName, "fullMethodName");	1✔
309	this.span =	1✔
310	otelTracer.spanBuilder(generateTraceSpanName(true, fullMethodName))	1✔
311	.setParent(remoteSpan == null ? null : Context.current().with(remoteSpan))	1✔
312	.startSpan();	1✔
313	this.baggage = baggage;	1✔
314	}	1✔
315
316	/**
317	* Record a finished stream and mark the current time as the end time.
318	*
319	* <p>Can be called from any thread without synchronization. Calling it the second time or more
320	* is a no-op.
321	*/
322	@Override
323	public void streamClosed(io.grpc.Status status) {
324	if (streamClosedUpdater != null) {	1✔
325	if (streamClosedUpdater.getAndSet(this, 1) != 0) {	1✔
326	return;	×
327	}
328	} else {
329	if (streamClosed != 0) {	×
330	return;	×
331	}
332	streamClosed = 1;	×
333	}
334	endSpanWithStatus(span, status);	1✔
335	}	1✔
336
337	@Override
338	public io.grpc.Context filterContext(io.grpc.Context context) {
339	return context	1✔
340	.withValue(otelSpan, span)	1✔
341	.withValue(BAGGAGE_KEY, baggage);	1✔
342	}
343
344	@Override
345	public void outboundMessageSent(
346	int seqNo, long optionalWireSize, long optionalUncompressedSize) {
347	recordOutboundMessageSentEvent(span, seqNo, optionalWireSize, optionalUncompressedSize);	1✔
348	}	1✔
349
350	@Override
351	public void inboundMessageRead(
352	int seqNo, long optionalWireSize, long optionalUncompressedSize) {
353	if (optionalWireSize != optionalUncompressedSize) {	1✔
354	recordInboundCompressedMessage(span, seqNo, optionalWireSize);	1✔
355	}
356	}	1✔
357
358	@Override
359	public void inboundMessage(int seqNo) {
360	this.seqNo = seqNo;	1✔
361	}	1✔
362
363	@Override
364	public void inboundUncompressedSize(long bytes) {
365	recordInboundMessageSize(span, seqNo, bytes);	1✔
366	}	1✔
367	}
368
369	@VisibleForTesting
370	final class ServerTracerFactory extends ServerStreamTracer.Factory {	1✔
371	@SuppressWarnings("ReferenceEquality")
372	@Override
373	public ServerStreamTracer newServerStreamTracer(String fullMethodName, Metadata headers) {
374	Context context = contextPropagators.getTextMapPropagator().extract(	1✔
375	Context.current(), headers, metadataGetter	1✔
376	);
377	Span remoteSpan = Span.fromContext(context);	1✔
378	if (remoteSpan == Span.getInvalid()) {	1✔
379	remoteSpan = null;	1✔
380	}
381	Baggage baggage = Baggage.fromContext(context);	1✔
382	return new ServerTracer(fullMethodName, remoteSpan, baggage);	1✔
383	}
384	}
385
386	@VisibleForTesting
387	final class TracingServerSpanPropagationInterceptor implements ServerInterceptor {	1✔
388	@Override
389	public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(ServerCall<ReqT, RespT> call,
390	Metadata headers, ServerCallHandler<ReqT, RespT> next) {
391	Span span = otelSpan.get(io.grpc.Context.current());	1✔
392	if (span == null) {	1✔
393	logger.log(Level.FINE, "Server span not found. ServerTracerFactory for server "	1✔
394	+ "tracing must be set.");
395	return next.startCall(call, headers);	1✔
396	}
397	Context serverCallContext = Context.current();	1✔
398	serverCallContext = serverCallContext.with(span);	1✔
399	Baggage baggage = BAGGAGE_KEY.get();	1✔
400	if (baggage != null) {	1✔
401	serverCallContext = serverCallContext.with(baggage);	1✔
402	} else {
403	logger.log(Level.WARNING, "Server baggage not found which is unexpected, "	1✔
404	+ "as it is being added unconditionally in filterContext().");
405	}
406	try (Scope scope = serverCallContext.makeCurrent()) {	1✔
407	return new ContextServerCallListener<>(next.startCall(call, headers), serverCallContext);	1✔
408	}
409	}
410	}
411
412	private static class ContextServerCallListener<ReqT> extends
413	ForwardingServerCallListener.SimpleForwardingServerCallListener<ReqT> {
414	private final Context context;
415
416	protected ContextServerCallListener(ServerCall.Listener<ReqT> delegate, Context context) {
417	super(delegate);	1✔
418	this.context = checkNotNull(context, "context");	1✔
419	}	1✔
420
421	@Override
422	public void onMessage(ReqT message) {
423	try (Scope scope = context.makeCurrent()) {	1✔
424	delegate().onMessage(message);	1✔
425	}
426	}	1✔
427
428	@Override
429	public void onHalfClose() {
430	try (Scope scope = context.makeCurrent()) {	1✔
431	delegate().onHalfClose();	1✔
432	}
433	}	1✔
434
435	@Override
436	public void onCancel() {
437	try (Scope scope = context.makeCurrent()) {	1✔
438	delegate().onCancel();	1✔
439	}
440	}	1✔
441
442	@Override
443	public void onComplete() {
444	try (Scope scope = context.makeCurrent()) {	1✔
445	delegate().onComplete();	1✔
446	}
447	}	1✔
448
449	@Override
450	public void onReady() {
451	try (Scope scope = context.makeCurrent()) {	1✔
452	delegate().onReady();	1✔
453	}
454	}	1✔
455	}
456
457	@VisibleForTesting
458	final class TracingClientInterceptor implements ClientInterceptor {	1✔
459
460	@Override
461	public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
462	MethodDescriptor<ReqT, RespT> method, CallOptions callOptions, Channel next) {
463	Span clientSpan = otelTracer.spanBuilder(	1✔
464	generateTraceSpanName(false, method.getFullMethodName()))	1✔
465	.startSpan();	1✔
466
467	final CallAttemptsTracerFactory tracerFactory = newClientCallTracer(clientSpan, method);	1✔
468	ClientCall<ReqT, RespT> call =	1✔
469	next.newCall(	1✔
470	method,
471	callOptions.withStreamTracerFactory(tracerFactory));	1✔
472	return new SimpleForwardingClientCall<ReqT, RespT>(call) {	1✔
473	@Override
474	public void start(Listener<RespT> responseListener, Metadata headers) {
475	delegate().start(	1✔
476	new SimpleForwardingClientCallListener<RespT>(responseListener) {	1✔
477	@Override
478	public void onClose(io.grpc.Status status, Metadata trailers) {
479	tracerFactory.callEnded(status);	1✔
480	super.onClose(status, trailers);	1✔
481	}	1✔
482	},
483	headers);
484	}	1✔
485	};
486	}
487	}
488
489	// Attribute named "message-size" always means the message size the application sees.
490	// If there was compression, additional event reports "message-size-compressed".
491	//
492	// An example trace with message compression:
493	//
494	// Sending:
495	// \|-- Event 'Outbound message sent', attributes('sequence-numer' = 0, 'message-size' = 7854,
496	// 'message-size-compressed' = 5493) ----\|
497	//
498	// Receiving:
499	// \|-- Event 'Inbound compressed message', attributes('sequence-numer' = 0,
500	// 'message-size-compressed' = 5493 ) ----\|
501	// \|-- Event 'Inbound message received', attributes('sequence-numer' = 0,
502	// 'message-size' = 7854) ----\|
503	//
504	// An example trace with no message compression:
505	//
506	// Sending:
507	// \|-- Event 'Outbound message sent', attributes('sequence-numer' = 0, 'message-size' = 7854) ---\|
508	//
509	// Receiving:
510	// \|-- Event 'Inbound message received', attributes('sequence-numer' = 0,
511	// 'message-size' = 7854) ----\|
512	private void recordOutboundMessageSentEvent(Span span,
513	int seqNo, long optionalWireSize, long optionalUncompressedSize) {
514	AttributesBuilder attributesBuilder = io.opentelemetry.api.common.Attributes.builder();	1✔
515	attributesBuilder.put("sequence-number", seqNo);	1✔
516	if (optionalUncompressedSize != -1) {	1✔
517	attributesBuilder.put("message-size", optionalUncompressedSize);	1✔
518	}
519	if (optionalWireSize != -1 && optionalWireSize != optionalUncompressedSize) {	1✔
520	attributesBuilder.put("message-size-compressed", optionalWireSize);	1✔
521	}
522	span.addEvent("Outbound message", attributesBuilder.build());	1✔
523	}	1✔
524
525	private void recordInboundCompressedMessage(Span span, int seqNo, long optionalWireSize) {
526	AttributesBuilder attributesBuilder = io.opentelemetry.api.common.Attributes.builder();	1✔
527	attributesBuilder.put("sequence-number", seqNo);	1✔
528	attributesBuilder.put("message-size-compressed", optionalWireSize);	1✔
529	span.addEvent("Inbound compressed message", attributesBuilder.build());	1✔
530	}	1✔
531
532	private void recordInboundMessageSize(Span span, int seqNo, long bytes) {
533	AttributesBuilder attributesBuilder = io.opentelemetry.api.common.Attributes.builder();	1✔
534	attributesBuilder.put("sequence-number", seqNo);	1✔
535	attributesBuilder.put("message-size", bytes);	1✔
536	span.addEvent("Inbound message", attributesBuilder.build());	1✔
537	}	1✔
538
539	private void endSpanWithStatus(Span span, io.grpc.Status status) {
540	if (status.isOk()) {	1✔
541	span.setStatus(StatusCode.OK);	1✔
542	} else {
543	span.setStatus(StatusCode.ERROR, GrpcUtil.statusToPrettyString(status));	1✔
544	}
545	span.end();	1✔
546	}	1✔
547
548	/**
549	* Convert a full method name to a tracing span name.
550	*
551	* @param isServer {@code false} if the span is on the client-side, {@code true} if on the
552	* server-side
553	* @param fullMethodName the method name as returned by
554	* {@link MethodDescriptor#getFullMethodName}.
555	*/
556	@VisibleForTesting
557	static String generateTraceSpanName(boolean isServer, String fullMethodName) {
558	String prefix = isServer ? "Recv" : "Sent";	1✔
559	return prefix + "." + fullMethodName.replace('/', '.');	1✔
560	}
561	}

grpc / grpc-java / #20365

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous