Add tracing support using Micrometer

buildSrc/src/main/kotlin/project/Companion.kt

@@ -23,4 +23,4 @@ import org.gradle.kotlin.dsl.getByType
 internal val Project.libs: LibrariesForLibs
     get() = extensions.getByType()
 
-internal const val DEFAULT_JAVA_VERSION = 17
+const val DEFAULT_JAVA_VERSION = 17

config/checkstyle/suppressions.xml

@@ -60,6 +60,7 @@
 
     <!-- Allow printStackTrace in this file -->
     <suppress checks="Regexp" files="CallbackResultHolder"/>
+    <suppress checks="Regexp" files="MicrometerTracer"/>
 
     <!--Do not check documentation tests classes -->
     <suppress checks="Javadoc*" files=".*documentation.*"/>

driver-core/build.gradle.kts

@@ -55,6 +55,9 @@ dependencies {
     optionalImplementation(libs.snappy.java)
     optionalImplementation(libs.zstd.jni)
 
+    optionalImplementation(platform(libs.micrometer.observation.bom))
+    optionalImplementation(libs.micrometer.observation)
+
     testImplementation(project(path = ":bson", configuration = "testArtifacts"))
     testImplementation(libs.reflections)
     testImplementation(libs.netty.tcnative.boringssl.static)

driver-core/src/main/com/mongodb/MongoClientSettings.java

@@ -31,6 +31,7 @@
 import com.mongodb.connection.TransportSettings;
 import com.mongodb.event.CommandListener;
 import com.mongodb.lang.Nullable;
+import com.mongodb.observability.ObservabilitySettings;
 import com.mongodb.spi.dns.DnsClient;
 import com.mongodb.spi.dns.InetAddressResolver;
 import org.bson.UuidRepresentation;
@@ -116,6 +117,7 @@ public final class MongoClientSettings {
     private final ContextProvider contextProvider;
     private final DnsClient dnsClient;
     private final InetAddressResolver inetAddressResolver;
+    private final ObservabilitySettings observabilitySettings;
     @Nullable
     private final Long timeoutMS;
 
@@ -215,6 +217,7 @@ public static final class Builder {
         private ReadConcern readConcern = ReadConcern.DEFAULT;
         private CodecRegistry codecRegistry = MongoClientSettings.getDefaultCodecRegistry();
         private TransportSettings transportSettings;
+        private ObservabilitySettings observabilitySettings;
         private List<CommandListener> commandListeners = new ArrayList<>();
 
         private final LoggerSettings.Builder loggerSettingsBuilder = LoggerSettings.builder();
@@ -260,6 +263,7 @@ private Builder(final MongoClientSettings settings) {
             timeoutMS = settings.getTimeout(MILLISECONDS);
             inetAddressResolver = settings.getInetAddressResolver();
             transportSettings = settings.getTransportSettings();
+            observabilitySettings = settings.getObservabilitySettings();
             autoEncryptionSettings = settings.getAutoEncryptionSettings();
             contextProvider = settings.getContextProvider();
             loggerSettingsBuilder.applySettings(settings.getLoggerSettings());
@@ -506,6 +510,19 @@ public Builder transportSettings(final TransportSettings transportSettings) {
             return this;
         }
 
+        /** Sets the {@link ObservabilitySettings} to apply.
+         *
+         * @param observabilitySettings the observability settings
+         * @return this
+         * @see #getObservabilitySettings()
+         * @since 5.7
+         */
+        @Alpha(Reason.CLIENT)
+        public Builder observabilitySettings(final ObservabilitySettings observabilitySettings) {
+            this.observabilitySettings = notNull("observabilitySettings", observabilitySettings);
+            return this;
+        }
+
         /**
          * Adds the given command listener.
          *
@@ -1040,6 +1057,18 @@ public ContextProvider getContextProvider() {
         return contextProvider;
     }
 
+
+    /**
+     * Get the observability settings.
+     * @return the observability settings
+     * @since 5.7
+     */
+    @Alpha(Reason.CLIENT)
+    @Nullable
+    public ObservabilitySettings getObservabilitySettings() {
+        return observabilitySettings;
+    }
+
     @Override
     public boolean equals(final Object o) {
         if (this == o) {
@@ -1137,6 +1166,7 @@ private MongoClientSettings(final Builder builder) {
         socketSettings = builder.socketSettingsBuilder.build();
         connectionPoolSettings = builder.connectionPoolSettingsBuilder.build();
         sslSettings = builder.sslSettingsBuilder.build();
+        observabilitySettings = builder.observabilitySettings;
         compressorList = builder.compressorList;
         uuidRepresentation = builder.uuidRepresentation;
         serverApi = builder.serverApi;

driver-core/src/main/com/mongodb/MongoNamespace.java

@@ -17,6 +17,7 @@
 package com.mongodb;
 
 import com.mongodb.annotations.Immutable;
+import com.mongodb.annotations.Internal;
 import org.bson.codecs.pojo.annotations.BsonCreator;
 import org.bson.codecs.pojo.annotations.BsonIgnore;
 import org.bson.codecs.pojo.annotations.BsonProperty;
@@ -40,9 +41,12 @@ public final class MongoNamespace {
      * @deprecated there is no replacement for this constant, as it is only needed for the OP_QUERY wire protocol message, which has
      * been replaced by OP_MSG
      */
-    @Deprecated
+    @Internal
     public static final String COMMAND_COLLECTION_NAME = "$cmd";
 
+    @Internal
+    public static final MongoNamespace ADMIN_DB_COMMAND_NAMESPACE = new MongoNamespace("admin", COMMAND_COLLECTION_NAME);
+
     private static final Set<Character> PROHIBITED_CHARACTERS_IN_DATABASE_NAME =
             new HashSet<>(asList('\0', '/', '\\', ' ', '"', '.'));
 

driver-core/src/main/com/mongodb/internal/connection/InternalStreamConnection.java

@@ -51,6 +51,7 @@
 import com.mongodb.internal.logging.StructuredLogger;
 import com.mongodb.internal.session.SessionContext;
 import com.mongodb.internal.time.Timeout;
+import com.mongodb.internal.tracing.Span;
 import com.mongodb.lang.Nullable;
 import org.bson.BsonBinaryReader;
 import org.bson.BsonDocument;
@@ -94,6 +95,8 @@
 import static com.mongodb.internal.connection.ProtocolHelper.isCommandOk;
 import static com.mongodb.internal.logging.LogMessage.Level.DEBUG;
 import static com.mongodb.internal.thread.InterruptionUtil.translateInterruptedException;
+import static com.mongodb.internal.tracing.MongodbObservation.HighCardinalityKeyNames.QUERY_TEXT;
+import static com.mongodb.internal.tracing.MongodbObservation.LowCardinalityKeyNames.RESPONSE_STATUS_CODE;
 import static java.util.Arrays.asList;
 
 /**
@@ -432,22 +435,59 @@ public boolean reauthenticationIsTriggered(@Nullable final Throwable t) {
     private <T> T sendAndReceiveInternal(final CommandMessage message, final Decoder<T> decoder,
             final OperationContext operationContext) {
         CommandEventSender commandEventSender;
+        Span tracingSpan;
         try (ByteBufferBsonOutput bsonOutput = new ByteBufferBsonOutput(this)) {
             message.encode(bsonOutput, operationContext);
-            commandEventSender = createCommandEventSender(message, bsonOutput, operationContext);
-            commandEventSender.sendStartedEvent();
+            tracingSpan = operationContext
+                    .getTracingManager()
+                    .createTracingSpan(message,
+                            operationContext,
+                            () -> message.getCommandDocument(bsonOutput),
+                            cmdName -> SECURITY_SENSITIVE_COMMANDS.contains(cmdName)
+                                    || SECURITY_SENSITIVE_HELLO_COMMANDS.contains(cmdName),
+                            () -> getDescription().getServerAddress(),
+                            () -> getDescription().getConnectionId()
+                    );
+
+            boolean isLoggingCommandNeeded = isLoggingCommandNeeded();
+            boolean isTracingCommandPayloadNeeded = tracingSpan != null && operationContext.getTracingManager().isCommandPayloadEnabled();
+
+            // Only hydrate the command document if necessary
+            BsonDocument commandDocument = null;
+            if (isLoggingCommandNeeded || isTracingCommandPayloadNeeded) {
+                commandDocument = message.getCommandDocument(bsonOutput);
+            }
+            if (isLoggingCommandNeeded) {
+                commandEventSender = new LoggingCommandEventSender(
+                        SECURITY_SENSITIVE_COMMANDS, SECURITY_SENSITIVE_HELLO_COMMANDS, description, commandListener,
+                        operationContext, message, commandDocument,
+                        COMMAND_PROTOCOL_LOGGER, loggerSettings);
+                commandEventSender.sendStartedEvent();
+            } else {
+                commandEventSender = new NoOpCommandEventSender();
+            }
+            if (isTracingCommandPayloadNeeded) {
+                tracingSpan.tagHighCardinality(QUERY_TEXT.asString(), commandDocument);
+            }
+
             try {
                 sendCommandMessage(message, bsonOutput, operationContext);
             } catch (Exception e) {
+                if (tracingSpan != null) {
+                    tracingSpan.error(e);
+                }
                 commandEventSender.sendFailedEvent(e);
                 throw e;
             }
         }
 
         if (message.isResponseExpected()) {
-            return receiveCommandMessageResponse(decoder, commandEventSender, operationContext);
+            return receiveCommandMessageResponse(decoder, commandEventSender, operationContext, tracingSpan);
         } else {
             commandEventSender.sendSucceededEventForOneWayCommand();
+            if (tracingSpan != null) {
+                tracingSpan.end();
+            }
             return null;
         }
     }
@@ -466,7 +506,7 @@ public <T> void send(final CommandMessage message, final Decoder<T> decoder, fin
     @Override
     public <T> T receive(final Decoder<T> decoder, final OperationContext operationContext) {
         isTrue("Response is expected", hasMoreToCome);
-        return receiveCommandMessageResponse(decoder, new NoOpCommandEventSender(), operationContext);
+        return receiveCommandMessageResponse(decoder, new NoOpCommandEventSender(), operationContext, null);
     }
 
     @Override
@@ -512,7 +552,7 @@ private void trySendMessage(final CommandMessage message, final ByteBufferBsonOu
     }
 
     private <T> T receiveCommandMessageResponse(final Decoder<T> decoder, final CommandEventSender commandEventSender,
-            final OperationContext operationContext) {
+            final OperationContext operationContext, @Nullable final Span tracingSpan) {
         boolean commandSuccessful = false;
         try (ResponseBuffers responseBuffers = receiveResponseBuffers(operationContext)) {
             updateSessionContext(operationContext.getSessionContext(), responseBuffers);
@@ -537,7 +577,17 @@ private <T> T receiveCommandMessageResponse(final Decoder<T> decoder, final Comm
             if (!commandSuccessful) {
                 commandEventSender.sendFailedEvent(e);
             }
+            if (tracingSpan != null) {
+                if (e instanceof MongoCommandException) {
+                    tracingSpan.tagLowCardinality(RESPONSE_STATUS_CODE.withValue(String.valueOf(((MongoCommandException) e).getErrorCode())));
+                }
+                tracingSpan.error(e);
+            }
             throw e;
+        } finally {
+            if (tracingSpan != null) {
+                tracingSpan.end();
+            }
         }
     }
 
@@ -553,7 +603,18 @@ private <T> void sendAndReceiveAsyncInternal(final CommandMessage message, final
 
         try {
             message.encode(bsonOutput, operationContext);
-            CommandEventSender commandEventSender = createCommandEventSender(message, bsonOutput, operationContext);
+
+            CommandEventSender commandEventSender;
+            if (isLoggingCommandNeeded()) {
+                BsonDocument commandDocument = message.getCommandDocument(bsonOutput);
+                commandEventSender = new LoggingCommandEventSender(
+                        SECURITY_SENSITIVE_COMMANDS, SECURITY_SENSITIVE_HELLO_COMMANDS, description, commandListener,
+                        operationContext, message, commandDocument,
+                        COMMAND_PROTOCOL_LOGGER, loggerSettings);
+            } else {
+                commandEventSender = new NoOpCommandEventSender();
+            }
+
             commandEventSender.sendStartedEvent();
             Compressor localSendCompressor = sendCompressor;
             if (localSendCompressor == null || SECURITY_SENSITIVE_COMMANDS.contains(message.getCommandDocument(bsonOutput).getFirstKey())) {
@@ -952,19 +1013,13 @@ public void onResult(@Nullable final ByteBuf result, @Nullable final Throwable t
 
     private static final StructuredLogger COMMAND_PROTOCOL_LOGGER = new StructuredLogger("protocol.command");
 
-    private CommandEventSender createCommandEventSender(final CommandMessage message, final ByteBufferBsonOutput bsonOutput,
-                                                        final OperationContext operationContext) {
+    private boolean isLoggingCommandNeeded() {
         boolean listensOrLogs = commandListener != null || COMMAND_PROTOCOL_LOGGER.isRequired(DEBUG, getClusterId());
-        if (!recordEverything && (isMonitoringConnection || !opened() || !authenticated.get() || !listensOrLogs)) {
-            return new NoOpCommandEventSender();
-        }
-        return new LoggingCommandEventSender(
-                SECURITY_SENSITIVE_COMMANDS, SECURITY_SENSITIVE_HELLO_COMMANDS, description, commandListener,
-                operationContext, message, bsonOutput,
-                COMMAND_PROTOCOL_LOGGER, loggerSettings);
+        return recordEverything || (!isMonitoringConnection && opened() && authenticated.get() && listensOrLogs);
     }
 
     private ClusterId getClusterId() {
         return description.getConnectionId().getServerId().getClusterId();
     }
+
 }

driver-core/src/main/com/mongodb/internal/connection/LoggingCommandEventSender.java

@@ -78,7 +78,7 @@ class LoggingCommandEventSender implements CommandEventSender {
             @Nullable final CommandListener commandListener,
             final OperationContext operationContext,
             final CommandMessage message,
-            final ByteBufferBsonOutput bsonOutput,
+            final BsonDocument commandDocument,
             final StructuredLogger logger,
             final LoggerSettings loggerSettings) {
         this.description = description;
@@ -88,7 +88,7 @@ class LoggingCommandEventSender implements CommandEventSender {
         this.loggerSettings = loggerSettings;
         this.startTimeNanos = System.nanoTime();
         this.message = message;
-        this.commandDocument = message.getCommandDocument(bsonOutput);
+        this.commandDocument = commandDocument;
         this.commandName = commandDocument.getFirstKey();
         this.redactionRequired = securitySensitiveCommands.contains(commandName)
                 || (securitySensitiveHelloCommands.contains(commandName) && commandDocument.containsKey("speculativeAuthenticate"));