How to correlate log events in distributed Vertx system

user_4386547 0
Viewed 3225

while doing logs in the multiple module of vertx, it is a basic requirement that we should be able to correlate all the logs for a single request.

as vertx being asynchronous what will be the best place to keep logid, conversationid, eventid.

any solution or patterns we can implement?

vert.x distributed logging
5 Answers

There's a surprising lack of good answers published about this, which is odd, given how easy it is.

Assuming you set the correlationId in your MDC context on receipt of a request or message, the simplest way I've found to propagate it is to use interceptors to pass the value between contexts:

vertx.eventBus()
        .addInboundInterceptor(deliveryContext -> {
            MultiMap headers = deliveryContext.message().headers();
            if (headers.contains("correlationId")) {
                MDC.put("correlationId", headers.get("correlationId"));
                deliveryContext.next();
            }
        })
        .addOutboundInterceptor(deliveryContext -> {
            deliveryContext.message().headers().add("correlationId", MDC.get("correlationId"));
            deliveryContext.next();
        });
user_5984070

The interceptor example presented by Clive Evans works great. I added a more details example showing how this might work:

import io.vertx.core.AbstractVerticle;
import io.vertx.core.DeploymentOptions;
import io.vertx.core.MultiMap;
import io.vertx.core.Promise;
import io.vertx.core.Vertx;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;

import java.time.Duration;
import java.util.UUID;

public class PublisherSubscriberInterceptor {

  private static final Logger LOG = LoggerFactory.getLogger(PublisherSubscriberInterceptor.class);
  public static final String ADRESS = "sender.address";


  public static void main(String[] args) {
    Vertx vertx = Vertx.vertx();
    createInterceptors(vertx);

    vertx.deployVerticle(new Publisher());
    vertx.deployVerticle(new Subscriber1());

    //For our example lets deploy subscriber2 2 times.
    vertx.deployVerticle(Subscriber2.class.getName(), new DeploymentOptions().setInstances(2));
  }

  private static void createInterceptors(Vertx vertx) {
    vertx.eventBus()
      .addInboundInterceptor(deliveryContext -> {
        MultiMap headers = deliveryContext.message().headers();
        if (headers.contains("myId")) {
          MDC.put("myId", headers.get("myId"));
          deliveryContext.next();
        }
      })
      .addOutboundInterceptor(deliveryContext -> {
        deliveryContext.message().headers().add("myId", MDC.get("myId"));
        deliveryContext.next();
      });
  }

  public static class Publisher extends AbstractVerticle {

    @Override
    public void start(Promise<Void> startPromise) throws Exception {
      startPromise.complete();
      vertx.setPeriodic(Duration.ofSeconds(5).toMillis(), id -> {
        MDC.put("myId", UUID.randomUUID().toString());
      vertx.eventBus().publish(Publish.class.getName(), "A message for all");
      });
    }
  }

  public static class Subscriber1 extends AbstractVerticle {
    private static final Logger LOG = LoggerFactory.getLogger(Subscriber1.class);

    @Override
    public void start(Promise<Void> startPromise) throws Exception {
      startPromise.complete();
      vertx.eventBus().consumer(Publish.class.getName(), message-> {
        LOG.debug("Subscriber1 Received: {}", message.body());
      });
    }
  }

  public static class Subscriber2 extends AbstractVerticle {
    private static final Logger LOG = LoggerFactory.getLogger(Subscriber2.class);

    @Override
    public void start(Promise<Void> startPromise) throws Exception {
      startPromise.complete();
      vertx.eventBus().consumer(Publish.class.getName(), message-> {
        LOG.debug("Subscriber2 Received: {}", message.body());
      });
    }
  }
}

you can see the log example for publishing 2 messages:

    13:37:14.315 [vert.x-eventloop-thread-3][myId=a2f0584c-9d4e-48a8-a724-a24ea12f7d80] DEBUG o.s.v.l.PublishSubscribeInterceptor$Subscriber2 - Subscriber2 Received: A message for all
    13:37:14.315 [vert.x-eventloop-thread-1][myId=a2f0584c-9d4e-48a8-a724-a24ea12f7d80] DEBUG o.s.v.l.PublishSubscribeInterceptor$Subscriber1 - Subscriber1 Received: A message for all
    13:37:14.315 [vert.x-eventloop-thread-4][myId=a2f0584c-9d4e-48a8-a724-a24ea12f7d80] DEBUG o.s.v.l.PublishSubscribeInterceptor$Subscriber2 - Subscriber2 Received: A message for all
    13:37:19.295 [vert.x-eventloop-thread-1][myId=63b5839e-3b0b-43a5-b379-92bd1466b870] DEBUG o.s.v.l.PublishSubscribeInterceptor$Subscriber1 - Subscriber1 Received: A message for all
    13:37:19.295 [vert.x-eventloop-thread-3][myId=63b5839e-3b0b-43a5-b379-92bd1466b870] DEBUG o.s.v.l.PublishSubscribeInterceptor$Subscriber2 - Subscriber2 Received: A message for all
    13:37:19.295 [vert.x-eventloop-thread-4][myId=63b5839e-3b0b-43a5-b379-92bd1466b870] DEBUG o.s.v.l.PublishSubscribeInterceptor$Subscriber2 - Subscriber2 Received: A message for all
user_77390830

Use vertx-sync and a ThreadLocal for the correlation ID. (i.e., a "FiberLocal"). Works great for me.

user_17579940
Related