Merge pull request #1017 from rabbitmq/marcingrzejszczak-observation

Support Micrometer observation

Author: acogoluegnes

generate-observation-documentation.sh

@@ -0,0 +1,6 @@
+#!/usr/bin/env bash
+
+./mvnw -q -P '!setup-test-cluster' test-compile exec:java \
+  -Dexec.mainClass=io.micrometer.docs.DocsGeneratorCommand \
+  -Dexec.classpathScope="test" \
+  -Dexec.args='src/main/java/com/rabbitmq/client/observation/micrometer .* target/micrometer-observation-docs'

pom.xml

@@ -54,15 +54,19 @@
     <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
     <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
 
+    <spotless.check.skip>true</spotless.check.skip>
     <slf4j.version>1.7.36</slf4j.version>
     <metrics.version>4.2.19</metrics.version>
     <micrometer.version>1.11.2</micrometer.version>
+    <micrometer-tracing.version>1.1.4</micrometer-tracing.version>
     <opentelemetry.version>1.28.0</opentelemetry.version>
     <jackson.version>2.15.2</jackson.version>
     <logback.version>1.2.12</logback.version>
     <junit.jupiter.version>5.9.3</junit.jupiter.version>
     <mockito.version>5.4.0</mockito.version>
     <assertj.version>3.24.2</assertj.version>
+    <micrometer-tracing-test.version>1.1.2</micrometer-tracing-test.version>
+    <micrometer-docs-generator.version>1.0.2</micrometer-docs-generator.version>
     <jetty.version>9.4.51.v20230217</jetty.version>
     <bouncycastle.version>1.70</bouncycastle.version>
     <netcrusher.version>0.10</netcrusher.version>
@@ -87,7 +91,8 @@
     <nexus-staging-maven-plugin.version>1.6.13</nexus-staging-maven-plugin.version>
     <checksum.maven.plugin.version>1.11</checksum.maven.plugin.version>
     <jshell-maven-plugin.version>1.3</jshell-maven-plugin.version>
-
+    <spotless.version>2.35.0</spotless.version>
+    <google-java-format.version>1.17.0</google-java-format.version>
     <!--
     These groovy scripts are used later in this POM file to generate
     source files and resources for the library itself and for the
@@ -791,10 +796,21 @@
       <version>${gson.version}</version>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>io.micrometer</groupId>
+      <artifactId>micrometer-tracing-integration-test</artifactId>
+      <version>${micrometer-tracing-test.version}</version>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>io.micrometer</groupId>
+      <artifactId>micrometer-docs-generator</artifactId>
+      <version>${micrometer-docs-generator.version}</version>
+      <scope>test</scope>
+    </dependency>
 
   </dependencies>
 
-
   <dependencyManagement>
     <dependencies>
       <dependency>
@@ -1091,6 +1107,42 @@
         </configuration>
       </plugin>
 
+      <plugin>
+        <groupId>com.diffplug.spotless</groupId>
+        <artifactId>spotless-maven-plugin</artifactId>
+        <version>${spotless.version}</version>
+        <configuration>
+          <java>
+            <includes>
+              <include>src/main/java/com/rabbitmq/client/observation/**/*.java</include>
+              <include>src/test/java/com/rabbitmq/client/test/functional/MicrometerObservationCollectorMetrics.java</include>
+            </includes>
+            <googleJavaFormat>
+              <version>${google-java-format.version}</version>
+              <style>GOOGLE</style>
+            </googleJavaFormat>
+          </java>
+          <!--          <ratchetFrom>origin/main</ratchetFrom>-->
+          <licenseHeader> <!-- specify either content or file, but not both -->
+            <content>// Copyright (c) $YEAR VMware, Inc. or its affiliates.  All rights reserved.
+              //
+              // This software, the RabbitMQ Java client library, is triple-licensed under the
+              // Mozilla Public License 2.0 ("MPL"), the GNU General Public License version 2
+              // ("GPL") and the Apache License version 2 ("ASL"). For the MPL, please see
+              // LICENSE-MPL-RabbitMQ. For the GPL, please see LICENSE-GPL2.  For the ASL,
+              // please see LICENSE-APACHE2.
+              //
+              // This software is distributed on an "AS IS" basis, WITHOUT WARRANTY OF ANY KIND,
+              // either express or implied. See the LICENSE file for specific language governing
+              // rights and limitations of this software.
+              //
+              // If you have any questions regarding licensing, please contact us at
+              // [email protected].
+            </content>
+          </licenseHeader>
+        </configuration>
+      </plugin>
+
     </plugins>
     <extensions>
       <extension>

src/main/java/com/rabbitmq/client/ConnectionFactory.java

@@ -22,6 +22,7 @@
 import com.rabbitmq.client.impl.recovery.RecoveredQueueNameSupplier;
 import com.rabbitmq.client.impl.recovery.RetryHandler;
 import com.rabbitmq.client.impl.recovery.TopologyRecoveryFilter;
+import com.rabbitmq.client.observation.ObservationCollector;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -52,7 +53,6 @@
  */
 public class ConnectionFactory implements Cloneable {
 
-    private static final Logger LOGGER = LoggerFactory.getLogger(ConnectionFactory.class);
     private static final int MAX_UNSIGNED_SHORT = 65535;
 
     /** Default user name */
@@ -138,6 +138,7 @@ public class ConnectionFactory implements Cloneable {
     private RecoveryDelayHandler recoveryDelayHandler;
 
     private MetricsCollector metricsCollector;
+    private ObservationCollector observationCollector = ObservationCollector.NO_OP;
 
     private boolean nio = false;
     private FrameHandlerFactory frameHandlerFactory;
@@ -978,6 +979,18 @@ public MetricsCollector getMetricsCollector() {
         return metricsCollector;
     }
 
+  /**
+   * Set observation collector.
+   *
+   * @param observationCollector the collector instance
+   * @since 5.19.0
+   * @see ObservationCollector
+   * @see com.rabbitmq.client.observation.micrometer.MicrometerObservationCollectorBuilder
+   */
+  public void setObservationCollector(ObservationCollector observationCollector) {
+        this.observationCollector = observationCollector;
+    }
+
     /**
      * Set a {@link CredentialsRefreshService} instance to handle credentials refresh if appropriate.
      * <p>
@@ -1249,7 +1262,8 @@ public Connection newConnection(ExecutorService executor, AddressResolver addres
             // see com.rabbitmq.client.impl.recovery.RecoveryAwareAMQConnectionFactory#newConnection
             // No Sonar: no need to close this resource because we're the one that creates it
             // and hands it over to the user
-            AutorecoveringConnection conn = new AutorecoveringConnection(params, fhFactory, addressResolver, metricsCollector); //NOSONAR
+            AutorecoveringConnection conn = new AutorecoveringConnection(
+                params, fhFactory, addressResolver, metricsCollector, observationCollector); //NOSONAR
 
             conn.init();
             return conn;
@@ -1316,7 +1330,7 @@ public ConnectionParams params(ExecutorService consumerWorkServiceExecutor) {
     }
 
     protected AMQConnection createConnection(ConnectionParams params, FrameHandler frameHandler, MetricsCollector metricsCollector) {
-        return new AMQConnection(params, frameHandler, metricsCollector);
+        return new AMQConnection(params, frameHandler, metricsCollector, observationCollector);
     }
 
     /**

src/main/java/com/rabbitmq/client/MetricsCollector.java

@@ -66,4 +66,4 @@ default void basicPublishUnrouted(Channel channel) {
 
     void basicCancel(Channel channel, String consumerTag);
 
-}
+}

src/main/java/com/rabbitmq/client/impl/AMQChannel.java

@@ -23,6 +23,7 @@
 import com.rabbitmq.client.AMQP.Queue;
 import com.rabbitmq.client.AMQP.Tx;
 import com.rabbitmq.client.Method;
+import com.rabbitmq.client.observation.ObservationCollector;
 import com.rabbitmq.utility.BlockingValueOrException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -78,6 +79,8 @@ public abstract class AMQChannel extends ShutdownNotifierComponent {
     private final TrafficListener _trafficListener;
     private final int maxInboundMessageBodySize;
 
+    private final ObservationCollector.ConnectionInfo connectionInfo;
+
     /**
      * Construct a channel on the given connection, with the given channel number.
      * @param connection the underlying connection for this channel
@@ -94,6 +97,7 @@ public AMQChannel(AMQConnection connection, int channelNumber) {
         this._trafficListener = connection.getTrafficListener();
         this.maxInboundMessageBodySize = connection.getMaxInboundMessageBodySize();
         this._command = new AMQCommand(this.maxInboundMessageBodySize);
+        this.connectionInfo = connection.connectionInfo();
     }
 
     /**
@@ -584,4 +588,8 @@ public AMQCommand transformReply(AMQCommand command) {
             return command;
         }
     }
+
+    protected ObservationCollector.ConnectionInfo connectionInfo() {
+        return this.connectionInfo;
+    }
 }

src/main/java/com/rabbitmq/client/impl/AMQConnection.java

@@ -19,6 +19,7 @@
 import com.rabbitmq.client.*;
 import com.rabbitmq.client.impl.AMQChannel.BlockingRpcContinuation;
 import com.rabbitmq.client.impl.recovery.RecoveryCanBeginListener;
+import com.rabbitmq.client.observation.ObservationCollector;
 import com.rabbitmq.utility.BlockingCell;
 import com.rabbitmq.utility.Utility;
 import org.slf4j.Logger;
@@ -68,6 +69,7 @@ public class AMQConnection extends ShutdownNotifierComponent implements Connecti
     private final int workPoolTimeout;
 
     private final AtomicBoolean finalShutdownStarted = new AtomicBoolean(false);
+    private volatile ObservationCollector.ConnectionInfo connectionInfo;
 
     /**
      * Retrieve a copy of the default table of client properties that
@@ -140,6 +142,7 @@ public static Map<String, Object> defaultClientProperties() {
     private final CredentialsProvider credentialsProvider;
     private final Collection<BlockedListener> blockedListeners = new CopyOnWriteArrayList<>();
     protected final MetricsCollector metricsCollector;
+    protected final ObservationCollector observationCollector;
     private final int channelRpcTimeout;
     private final boolean channelShouldCheckRpcResponseType;
     private final TrafficListener trafficListener;
@@ -210,13 +213,14 @@ public Map<String, Object> getServerProperties() {
     }
 
     public AMQConnection(ConnectionParams params, FrameHandler frameHandler) {
-        this(params, frameHandler, new NoOpMetricsCollector());
+        this(params, frameHandler, new NoOpMetricsCollector(), ObservationCollector.NO_OP);
     }
 
     /** Construct a new connection
      * @param params parameters for it
      */
-    public AMQConnection(ConnectionParams params, FrameHandler frameHandler, MetricsCollector metricsCollector)
+    public AMQConnection(ConnectionParams params, FrameHandler frameHandler,
+                         MetricsCollector metricsCollector, ObservationCollector observationCollector)
     {
         checkPreconditions();
         this.credentialsProvider = params.getCredentialsProvider();
@@ -255,6 +259,7 @@ public AMQConnection(ConnectionParams params, FrameHandler frameHandler, Metrics
         this._inConnectionNegotiation = true; // we start out waiting for the first protocol response
 
         this.metricsCollector = metricsCollector;
+        this.observationCollector = observationCollector;
 
         this.errorOnWriteListener = params.getErrorOnWriteListener() != null ? params.getErrorOnWriteListener() :
             (connection, exception) -> { throw exception; }; // we just propagate the exception for non-recoverable connections
@@ -425,6 +430,11 @@ public void start()
 
             setHeartbeat(negotiatedHeartbeat);
 
+            this.connectionInfo = new DefaultConnectionInfo(
+                this._frameHandler.getAddress().getHostAddress(),
+                this._frameHandler.getPort()
+            );
+
             _channel0.transmit(new AMQP.Connection.TuneOk.Builder()
                                 .channelMax(negotiatedChannelMax)
                                 .frameMax(frameMax)
@@ -475,7 +485,9 @@ public void start()
     }
 
     protected ChannelManager instantiateChannelManager(int channelMax, ThreadFactory threadFactory) {
-        ChannelManager result = new ChannelManager(this._workService, channelMax, threadFactory, this.metricsCollector);
+        ChannelManager result = new ChannelManager(
+            this._workService, channelMax, threadFactory,
+            this.metricsCollector, this.observationCollector);
         configureChannelManager(result);
         return result;
     }
@@ -1198,4 +1210,30 @@ public TrafficListener getTrafficListener() {
     int getMaxInboundMessageBodySize() {
         return maxInboundMessageBodySize;
     }
+
+    private static class DefaultConnectionInfo implements ObservationCollector.ConnectionInfo {
+
+        private final String peerAddress;
+        private final int peerPort;
+
+        private DefaultConnectionInfo(String peerAddress, int peerPort) {
+            this.peerAddress = peerAddress;
+            this.peerPort = peerPort;
+        }
+
+        @Override
+        public String   getPeerAddress() {
+            return peerAddress;
+        }
+
+        @Override
+        public int getPeerPort() {
+            return this.peerPort;
+        }
+
+    }
+
+    ObservationCollector.ConnectionInfo connectionInfo() {
+        return this.connectionInfo;
+    }
 }

src/main/java/com/rabbitmq/client/impl/ChannelManager.java

@@ -1,4 +1,4 @@
-// Copyright (c) 2007-2020 VMware, Inc. or its affiliates.  All rights reserved.
+// Copyright (c) 2007-2023 VMware, Inc. or its affiliates.  All rights reserved.
 //
 // This software, the RabbitMQ Java client library, is triple-licensed under the
 // Mozilla Public License 2.0 ("MPL"), the GNU General Public License version 2
@@ -19,6 +19,7 @@
 import com.rabbitmq.client.MetricsCollector;
 import com.rabbitmq.client.NoOpMetricsCollector;
 import com.rabbitmq.client.ShutdownSignalException;
+import com.rabbitmq.client.observation.ObservationCollector;
 import com.rabbitmq.utility.IntAllocator;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -55,6 +56,7 @@ public class ChannelManager {
     private int channelShutdownTimeout = (int) ((ConnectionFactory.DEFAULT_HEARTBEAT * AMQConnection.CHANNEL_SHUTDOWN_TIMEOUT_MULTIPLIER) * 1000);
 
     protected final MetricsCollector metricsCollector;
+    protected final ObservationCollector observationCollector;
 
     public int getChannelMax(){
       return _channelMax;
@@ -65,11 +67,13 @@ public ChannelManager(ConsumerWorkService workService, int channelMax) {
     }
 
     public ChannelManager(ConsumerWorkService workService, int channelMax, ThreadFactory threadFactory) {
-        this(workService, channelMax, threadFactory, new NoOpMetricsCollector());
+        this(workService, channelMax, threadFactory,
+             new NoOpMetricsCollector(), ObservationCollector.NO_OP);
     }
 
 
-    public ChannelManager(ConsumerWorkService workService, int channelMax, ThreadFactory threadFactory, MetricsCollector metricsCollector) {
+    public ChannelManager(ConsumerWorkService workService, int channelMax, ThreadFactory threadFactory,
+                          MetricsCollector metricsCollector, ObservationCollector observationCollector) {
         if (channelMax < 0)
             throw new IllegalArgumentException("create ChannelManager: 'channelMax' must be greater or equal to 0.");
         if (channelMax == 0) {
@@ -83,6 +87,7 @@ public ChannelManager(ConsumerWorkService workService, int channelMax, ThreadFac
         this.workService = workService;
         this.threadFactory = threadFactory;
         this.metricsCollector = metricsCollector;
+        this.observationCollector = observationCollector;
     }
 
     /**
@@ -214,7 +219,8 @@ private ChannelN addNewChannel(AMQConnection connection, int channelNumber) {
     }
 
     protected ChannelN instantiateChannel(AMQConnection connection, int channelNumber, ConsumerWorkService workService) {
-        return new ChannelN(connection, channelNumber, workService, this.metricsCollector);
+        return new ChannelN(connection, channelNumber, workService,
+                            this.metricsCollector, this.observationCollector);
     }
 
     /**