feat: add activity on connection

Author: aygalinc

projects/RabbitMQ.Client/Impl/Connection.cs

@@ -228,11 +228,10 @@ internal void TakeOver(Connection other)
         internal async ValueTask<IConnection> OpenAsync(CancellationToken cancellationToken)
         {
             cancellationToken.ThrowIfCancellationRequested();
-
+            Activity? _connectionActivity = RabbitMQActivitySource.OpenConnection(_frameHandler);
             try
             {
                 RabbitMqClientEventSource.Log.ConnectionOpened();
-
                 cancellationToken.ThrowIfCancellationRequested();
 
                 // Note: this must happen *after* the frame handler is started
@@ -250,8 +249,9 @@ await _channel0.ConnectionOpenAsync(_config.VirtualHost, cancellationToken)
 
                 return this;
             }
-            catch
+            catch(Exception ex)
             {
+                _connectionActivity?.ReportException(ex);
                 try
                 {
                     var ea = new ShutdownEventArgs(ShutdownInitiator.Library, Constants.InternalError, "FailedOpen");

projects/RabbitMQ.Client/Impl/RabbitMQActivitySource.cs

@@ -43,10 +43,15 @@ public static class RabbitMQActivitySource
         private static readonly ActivitySource s_subscriberSource =
             new ActivitySource(SubscriberSourceName, AssemblyVersion);
 
+        private static readonly ActivitySource s_connectionSource =
+            new ActivitySource(ConnectionSourceName, AssemblyVersion);
+
         public const string PublisherSourceName = "RabbitMQ.Client.Publisher";
         public const string SubscriberSourceName = "RabbitMQ.Client.Subscriber";
+        public const string ConnectionSourceName = "RabbitMQ.Client.Connection";
 
-        public static Action<Activity, IDictionary<string, object?>> ContextInjector { get; set; } = DefaultContextInjector;
+        public static Action<Activity, IDictionary<string, object?>> ContextInjector { get; set; } =
+            DefaultContextInjector;
 
         public static Func<IReadOnlyBasicProperties, ActivityContext> ContextExtractor { get; set; } =
             DefaultContextExtractor;
@@ -61,6 +66,19 @@ public static class RabbitMQActivitySource
             new KeyValuePair<string, object?>(ProtocolVersion, "0.9.1")
         };
 
+        internal static Activity? OpenConnection(IFrameHandler frameHandler)
+        {
+            if (!s_connectionSource.HasListeners())
+            {
+                return null;
+            }
+            Activity? connectionActivity =
+                s_connectionSource.StartRabbitMQActivity("rabbitmq connect", ActivityKind.Client);
+            connectionActivity?
+                .SetNetworkTags(frameHandler);
+            return connectionActivity;
+        }
+
         internal static Activity? BasicPublish(string routingKey, string exchange, int bodySize,
             ActivityContext linkedContext = default)
         {
@@ -82,7 +100,6 @@ public static class RabbitMQActivitySource
             }
 
             return activity;
-
         }
 
         internal static Activity? BasicGetEmpty(string queue)
@@ -148,6 +165,14 @@ public static class RabbitMQActivitySource
             return activity;
         }
 
+        internal static void ReportException(this Activity? activity, Exception exception)
+        {
+            activity?.AddTag("exception.message", exception.Message);
+            activity?.AddTag("exception.stacktrace", exception.ToString());
+            activity?.AddTag("exception.type", exception.GetType().FullName);
+            activity?.SetStatus(ActivityStatusCode.Error);
+        }
+        
         private static Activity? StartRabbitMQActivity(this ActivitySource source, string name, ActivityKind kind,
             ActivityContext parentContext = default)
         {