How to trace through calls between http services, gRPC, and queues (aws sns / sqs).

[victorpacheco3107]

Hi everybody!

Currently in my company we were able to implement opentelemetry for distributed traceability between different calls of microservices based on http. We use a collector and jeager. Java as programming language and spring boot as development framework. That implementation was done a couple of months ago, so we used version 0.9.1.
We implement this using aspects to intercept http requests, and in clients we capture and inject the headers, as follows:

Server side

The aspect:

Aspect
@Component
@Slf4j
public class ClsTracingAspect {
  private static final String RESPONSE_ATTRIBUTE = "response";
  private static final String ERROR_EVENT = "Error";

  public static final String TRACER_INSTRUMENTATION_NAME = "io.otel.habilitadora";
  public static final String X_APPROVALID = "X-ApprovalId";
  private static final Integer DEADLINE_TIMEOUT_GRPC = 6000;
  private static final Integer MAX_NUMBER_OF_ATTRIBUTES = 45;
  private final OtelConfigProperties otelConfig;

  @Autowired
  public ClsTracingAspect(OtelConfigProperties otelConfig) {
    this.otelConfig = otelConfig;
  }

  @Pointcut(
      "within(co.com.avaldigitallabs.digital.ms.controller..*)  && "
          + "@within(org.springframework.web.bind.annotation.RestController) ")
  public void restControllers() {
    // This method define which class are taken by this aspect
  }

  @Around("restControllers()")
  public Object restTracerRequest(ProceedingJoinPoint joinPoint) throws Throwable {
    log.info(
        "****LoggingAspect.logAroundAllMethods() : "
            + joinPoint.getSignature().getName()
            + ": Before Method Execution");
    Object proceed = null;
    Gson gson = new Gson();
    HttpServletRequest request = getRequest();
    ContextPropagators contextPropagators =
        DefaultContextPropagators.builder()
            .addTextMapPropagator(HttpTraceContext.getInstance())
            .build();
    OpenTelemetry.setPropagators(contextPropagators);
    Context extractContextFromRequest = ClsUtilPropagator.extractContextFromHttpRequest(request); // <---------- The propagator -->
    String bodyIntoJson = extractBodyIntoJson(joinPoint);
    TracerSdkManagement tracerProvider = OpenTelemetrySdk.getTracerManagement();
    TraceConfig tracerConfig =
        tracerProvider
            .getActiveTraceConfig()
            .toBuilder()
            .setMaxNumberOfAttributes(MAX_NUMBER_OF_ATTRIBUTES)
            .build();
    tracerProvider.updateActiveTraceConfig(tracerConfig);

    Tracer tracer = OpenTelemetry.getTracer(TRACER_INSTRUMENTATION_NAME);
    OtlpGrpcSpanExporter spanExporter =
        OtlpGrpcSpanExporter.newBuilder()
            .setEndpoint(otelConfig.getEndpoint())
            .setDeadlineMs(DEADLINE_TIMEOUT_GRPC)
            .build();
    BatchSpanProcessor spanProcessor = BatchSpanProcessor.newBuilder(spanExporter).build();
    tracerProvider.addSpanProcessor(spanProcessor);
    Span span =
        tracer
            .spanBuilder(createNameForSpan(joinPoint))
            .setParent(extractContextFromRequest)
            .startSpan();
    try (Scope scope = tracer.withSpan(span)) {
      span.addEvent("Request");
      if (bodyIntoJson != null) {
        span.setAttribute("request", bodyIntoJson);
      }
      span.setAttribute("requestURL", request.getRequestURL().toString());
      traceHeaders(request, span);
      proceed = joinPoint.proceed();
      span.addEvent("Response");
      Object responseBody = ((ResponseEntity) proceed).getBody();
      if (responseBody != null) {
        String jsonResponse = gson.toJson(((ResponseEntity) proceed).getBody());
        span.setAttribute(RESPONSE_ATTRIBUTE, jsonResponse);
      }
      String approvalId = ((ResponseEntity) proceed).getHeaders().getFirst(X_APPROVALID);
      if (approvalId != null) {
        span.setAttribute("x-approvalid", approvalId);
      }
      span.setAttribute("responseCode", ((ResponseEntity) proceed).getStatusCode().value());
    } catch (ClsApiUnexpectedException unexpectedException) {
      span.addEvent(ERROR_EVENT);
      String jsonResponse = gson.toJson(unexpectedException.getClsUnexpectedErrorModel());
      span.setAttribute(RESPONSE_ATTRIBUTE, jsonResponse);
      span.recordException(unexpectedException);
      throw unexpectedException;
    } catch (ClsApiBusinessException businessException) {
      span.addEvent(ERROR_EVENT);
      String jsonResponse = gson.toJson(businessException.getClsBusinessExceptionModel());
      span.setAttribute(RESPONSE_ATTRIBUTE, jsonResponse);
      span.recordException(businessException);
      throw businessException;
    } catch (ClsApiNotFoundException notFoundException) {
      span.addEvent(ERROR_EVENT);
      String jsonResponse = gson.toJson(notFoundException.getClsUnexpectedErrorModel());
      span.setAttribute(RESPONSE_ATTRIBUTE, jsonResponse);
      span.recordException(notFoundException);
      throw notFoundException;
    } catch (Throwable throwable) {
      log.info("catch en proceed");
      span.addEvent(ERROR_EVENT);
      span.setAttribute("Response ERROR", throwable.getMessage());
      span.recordException(throwable);
      throw throwable;
    } finally {
      span.end();
      spanProcessor.shutdown();
    }
    return proceed;
  }

  private String createNameForSpan(JoinPoint point) {
    return point.getTarget().getClass().getSimpleName() + " " + point.getSignature().getName();
  }

  public HttpServletRequest getRequest() {
    return ((ServletRequestAttributes) RequestContextHolder.currentRequestAttributes())
        .getRequest();
  }

  private void traceHeaders(HttpServletRequest request, Span span) {
    log.info(":::Request Headers:::");
    Enumeration<String> headerNames = request.getHeaderNames();
    while (headerNames.hasMoreElements()) {
      String key = headerNames.nextElement();
      String value = request.getHeader(key);
      log.info(key + " : " + value);
      span.setAttribute(key, value);
    }
  }

  private String extractBodyIntoJson(JoinPoint point) throws JsonProcessingException {
    String jsonBody = null;
    MethodSignature methodSignature = (MethodSignature) point.getSignature();
    Annotation[][] annotationMatrix = methodSignature.getMethod().getParameterAnnotations();
    int index = -1;
    for (Annotation[] annotations : annotationMatrix) {
      index++;
      for (Annotation annotation : annotations) {
        if (!(annotation instanceof RequestBody)) {
          continue;
        }
        Object requestBody = point.getArgs()[index];
        ObjectWriter ow = new ObjectMapper().writer().withDefaultPrettyPrinter();
        jsonBody = ow.writeValueAsString(requestBody);
        break;
      }
    }
    return jsonBody;
  }
}

ClsUtilPropagator:

public class ClsUtilPropagator {

  private ClsUtilPropagator() {
  }

  public static Context extractContextFromHttpRequest(HttpServletRequest request) {
    return OpenTelemetry.getPropagators()
        .getTextMapPropagator()
        .extract(Context.current(), request, new TextMapPropagatorImpl());
  }

}

TextMapPropagatorImpl:

public class TextMapPropagatorImpl implements TextMapPropagator.Getter<HttpServletRequest> {

  private static final int MIN_LENGTH_FOR_HEADER = 1;

  @Nullable
  @Override
  public String get(HttpServletRequest carrier, String key) {
    Enumeration<String> headers = carrier.getHeaders(key);
    if (headers == null || !headers.hasMoreElements()) {
      return null;
    }
    List<String> values = new ArrayList<>();
    while (headers.hasMoreElements()) {
      String nextElement = headers.nextElement();
      if (!nextElement.trim().chars().allMatch(Character::isWhitespace)) {
        values.add(nextElement);
      }
    }
    if (values.isEmpty()) {
      return null;
    }
    if (values.size() == MIN_LENGTH_FOR_HEADER) {
      return values.get(0);
    }
    StringBuilder builder = new StringBuilder(values.get(0));
    for (int i = 1; i < values.size(); i++) {
      builder.append(',').append(values.get(i));
    }
    return builder.toString();
  }
}

Client side:

In all clients, before making a request to a microservice, we call this method:

private Map<String, Object> captureBraveHeaders(Map<String, Object> headers) {
    OpenTelemetry.getPropagators()
        .getTextMapPropagator()
        .inject(
            Context.current(),
            headers,
            new TextMapPropagator.Setter<Map<String, Object>>() {
              @Override
              public void set(Map<String, Object> carrier, String key, String value) {
                carrier.put(key, value);
              }
            });
    return headers;
  }

All this works perfectly, the spans are perfectly correlated between the different calls to the microservices.

But now we face a new challenge. From any microservice you can call other microservices exposed in gRPC. And from any microservice you can write to an SNS topic (aws), and from another microservice in charge of reading from the SQS queue (aws). As if that were not enough, we have new services developed in Golang.

Is there a way to be able to correlated the spans between java/go microservices through calls http, gRPC and message queues?

If so, what would be the way to achieve this? Could you do something like a generic propagator to work with http, gRPC and queues at the same time?

Thanks!

[jkwatson]

In the java-instrumentation repository, there's already support for propagation of grpc calls, and I know there's ongoing work for propagation through AWS services as well.

Here's the grpc support: https://github.com/open-telemetry/opentelemetry-java-instrumentation/tree/main/instrumentation/grpc-1.5/library/src/main/java/io/opentelemetry/instrumentation/grpc/v1_5

And here's the AWS support (I think): https://github.com/open-telemetry/opentelemetry-java-instrumentation/tree/main/instrumentation/aws-sdk/aws-sdk-2.2/library/src/main/java/io/opentelemetry/instrumentation/awssdk/v2_2

Does that help?

[victorpacheco3107]

Thank you very much! I'm going to check the links, I hope they work for me.