Asynchronous and reactive Mongo Java Driver instrumentation

[csabakos]

Feature: https://github.com/openzipkin/brave/pull/1090 added MongoDB tracing for the synchronous clients (com.mongodb.MongoClient and com.mongodb.client.MongoClient) but left the instrumentation unimplemented for the asynchronous (com.mongodb.async.MongoClient) and reactive (com.mongodb.reactivestreams.client.MongoClient) clients.

Implement tracing for the asynchronous and reactive Mongo Java Drivers.

Rational / Example Scenario Spring Data supports Reactive MongoDB repositories and it would be nice to get tracing for it.

Prior Art

• https://github.com/openzipkin/brave/pull/1090 added tracing for the synchronous clients
• README.md section that will need to be updated after fixing this issue

Implementation notes regarding the asynchronous MongoDB clients

The asynchronous clients use threads for the async completion handlers (meaning that #commandStarted(CommandStartedEvent) and #commandSucceeded(CommandSucceededEvent)/ #commandFailed(CommandFailedEvent) may get called from background threads and also not necessarily from the same thread).

It should be possible to set a custom com.mongodb.connection.StreamFactoryFactory on the com.mongodb.MongoClientSettings.Builder which can propagate the tracing context correctly between those handlers, but it is unknown if this would be sufficient.

[codefromthecrypt]

As mongodb reactive is a different library, with a different version, we shouldn't embed it into brave-instrumentation-mongodb, rather brave-instrumentation-mongodb-reactivestreams

this is similar to apachehttpclient and apachehttpasyncclient which have different versions though share a base core library

Here's a test case example, which aimed to start this work.

  MongoClient mongoClient;
  MongoDatabase database;

  @Before public void init() {
    CommandListener listener = MongoDBTracing.newBuilder(tracing)
      .build()
      .commandListener();
    MongoClientSettings settings = mongoClientSettingsBuilder()
      .addCommandListener(listener)
      .build();
    mongoClient = MongoClients.create(settings);
    database = mongoClient.getDatabase("testDatabase");

    spans.clear();
  }

  @After public void close() {
    Tracing.current().close();
    if (mongoClient != null) mongoClient.close();
  }

  @Test public void makesChildOfCurrentSpan() throws InterruptedException {
    ScopedSpan parent = tracing.tracer().startScopedSpan("test");
    try {
      blockUntilComplete(database.getCollection(COLLECTION_NAME).find().first());
    } finally {
      parent.finish();
    }

    assertThat(spans)
      .hasSize(2);
  }

  static void blockUntilComplete(Publisher<?> publisher) throws InterruptedException {
    CountDownLatch countDownLatch = new CountDownLatch(1);

    publisher.subscribe(new Subscriber<Object>() {
      @Override public void onSubscribe(Subscription subscription) {
        subscription.request(Long.MAX_VALUE);
      }

      @Override public void onNext(Object o) {

      }

      @Override public void onError(Throwable throwable) {

      }

      @Override
      public void onComplete() {
        countDownLatch.countDown();
      }
    });
    countDownLatch.await();
  }
}

[codefromthecrypt]

Note: the reactive driver v4 is out. the current code is v3 and I'm not sure about its future. If someone knows what's the deal, let's try to brainstorm prior to release. Ex is the normal driver deprecated (ex dying at v3) or are they aiming to continue it in the v4 line? cc @kojilin