feat: add activity on connection

Author: aygalinc

projects/RabbitMQ.Client/Impl/Connection.cs

@@ -226,11 +226,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
@@ -248,8 +247,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

@@ -38,10 +38,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;
@@ -56,6 +61,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? Send(string routingKey, string exchange, int bodySize,
             ActivityContext linkedContext = default)
         {
@@ -66,18 +84,21 @@ public static class RabbitMQActivitySource
 
             Activity? activity = linkedContext == default
                 ? s_publisherSource.StartRabbitMQActivity(
-                    UseRoutingKeyAsOperationName ? $"{routingKey} {MessagingOperationTypeSend}" : MessagingOperationTypeSend,
+                    UseRoutingKeyAsOperationName
+                        ? $"{routingKey} {MessagingOperationTypeSend}"
+                        : MessagingOperationTypeSend,
                     ActivityKind.Producer)
                 : s_publisherSource.StartLinkedRabbitMQActivity(
-                    UseRoutingKeyAsOperationName ? $"{routingKey} {MessagingOperationTypeSend}" : MessagingOperationTypeSend,
+                    UseRoutingKeyAsOperationName
+                        ? $"{routingKey} {MessagingOperationTypeSend}"
+                        : MessagingOperationTypeSend,
                     ActivityKind.Producer, linkedContext);
             if (activity != null && activity.IsAllDataRequested)
             {
                 PopulateMessagingTags(MessagingOperationTypeSend, routingKey, exchange, 0, bodySize, activity);
             }
 
             return activity;
-
         }
 
         internal static Activity? ReceiveEmpty(string queue)
@@ -88,7 +109,9 @@ public static class RabbitMQActivitySource
             }
 
             Activity? activity = s_subscriberSource.StartRabbitMQActivity(
-                UseRoutingKeyAsOperationName ? $"{queue} {MessagingOperationTypeReceive}" : MessagingOperationTypeReceive,
+                UseRoutingKeyAsOperationName
+                    ? $"{queue} {MessagingOperationTypeReceive}"
+                    : MessagingOperationTypeReceive,
                 ActivityKind.Consumer);
             if (activity != null && activity.IsAllDataRequested)
             {
@@ -110,11 +133,14 @@ public static class RabbitMQActivitySource
 
             // Extract the PropagationContext of the upstream parent from the message headers.
             Activity? activity = s_subscriberSource.StartLinkedRabbitMQActivity(
-                UseRoutingKeyAsOperationName ? $"{routingKey} {MessagingOperationTypeReceive}" : MessagingOperationTypeReceive, ActivityKind.Consumer,
+                UseRoutingKeyAsOperationName
+                    ? $"{routingKey} {MessagingOperationTypeReceive}"
+                    : MessagingOperationTypeReceive, ActivityKind.Consumer,
                 ContextExtractor(readOnlyBasicProperties));
             if (activity != null && activity.IsAllDataRequested)
             {
-                PopulateMessagingTags(MessagingOperationTypeReceive, routingKey, exchange, deliveryTag, readOnlyBasicProperties,
+                PopulateMessagingTags(MessagingOperationTypeReceive, routingKey, exchange, deliveryTag,
+                    readOnlyBasicProperties,
                     bodySize, activity);
             }
 
@@ -131,7 +157,9 @@ public static class RabbitMQActivitySource
 
             // Extract the PropagationContext of the upstream parent from the message headers.
             Activity? activity = s_subscriberSource.StartLinkedRabbitMQActivity(
-                UseRoutingKeyAsOperationName ? $"{routingKey} {MessagingOperationTypeProcess}" : MessagingOperationTypeProcess,
+                UseRoutingKeyAsOperationName
+                    ? $"{routingKey} {MessagingOperationTypeProcess}"
+                    : MessagingOperationTypeProcess,
                 ActivityKind.Consumer, ContextExtractor(basicProperties));
             if (activity != null && activity.IsAllDataRequested)
             {
@@ -142,10 +170,19 @@ 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)
         {
-            return source.CreateActivity(name, kind, parentContext, idFormat: ActivityIdFormat.W3C, tags: CreationTags)?.Start();
+            return source.CreateActivity(name, kind, parentContext, idFormat: ActivityIdFormat.W3C, tags: CreationTags)
+                ?.Start();
         }
 
         private static Activity? StartLinkedRabbitMQActivity(this ActivitySource source, string name, ActivityKind kind,
@@ -273,7 +310,8 @@ private static ActivityContext DefaultContextExtractor(IReadOnlyBasicProperties
                 return default;
             }
 
-            DistributedContextPropagator.Current.ExtractTraceIdAndState(props.Headers, DefaultContextGetter, out string? traceParent, out string? traceState);
+            DistributedContextPropagator.Current.ExtractTraceIdAndState(props.Headers, DefaultContextGetter,
+                out string? traceParent, out string? traceState);
             return ActivityContext.TryParse(traceParent, traceState, out ActivityContext context) ? context : default;
         }
 
@@ -288,7 +326,8 @@ private static void DefaultContextSetter(object? carrier, string name, string va
             carrierDictionary[name] = value;
         }
 
-        private static void DefaultContextGetter(object? carrier, string name, out string? value, out IEnumerable<string>? values)
+        private static void DefaultContextGetter(object? carrier, string name, out string? value,
+            out IEnumerable<string>? values)
         {
             if (carrier is IDictionary<string, object> carrierDict &&
                 carrierDict.TryGetValue(name, out object? propsVal) && propsVal is byte[] bytes)