Adding distinct activity for distributed tracing to YARP

src/ReverseProxy/Forwarder/ForwarderMiddleware.cs

@@ -4,6 +4,7 @@
 using System;
 using System.Threading.Tasks;
 using Microsoft.AspNetCore.Http;
+using Microsoft.AspNetCore.Http.Connections;
 using Microsoft.Extensions.Logging;
 using Yarp.ReverseProxy.Model;
 using Yarp.ReverseProxy.Utilities;
@@ -64,11 +65,19 @@ public async Task Invoke(HttpContext context)
             throw new InvalidOperationException($"Chosen destination has no model set: '{destination.DestinationId}'");
         }
 
+        var activityForTracing = reverseProxyFeature.ActivityForTracing;
+        if (activityForTracing != null)
+        {
+            activityForTracing.AddTag("Route", route.Config.RouteId);
+            activityForTracing.AddTag("Cluster", cluster.ClusterId);
+            activityForTracing.AddTag("Destination", destination.DestinationId);
+        }
+
         try
         {
             cluster.ConcurrencyCounter.Increment();
             destination.ConcurrencyCounter.Increment();
-
+            activityForTracing?.Start();
             ForwarderTelemetry.Log.ForwarderInvoke(cluster.ClusterId, route.Config.RouteId, destination.DestinationId);
 
             var clusterConfig = reverseProxyFeature.Cluster;
@@ -79,6 +88,7 @@ await _forwarder.SendAsync(context, destinationModel.Config.Address, clusterConf
         {
             destination.ConcurrencyCounter.Decrement();
             cluster.ConcurrencyCounter.Decrement();
+            activityForTracing?.Stop();
         }
     }
 

src/ReverseProxy/Health/ActiveHealthCheckMonitor.cs

@@ -10,6 +10,7 @@
 using Microsoft.Extensions.Options;
 using Yarp.ReverseProxy.Model;
 using Yarp.ReverseProxy.Utilities;
+using System.Diagnostics;
 
 namespace Yarp.ReverseProxy.Health;
 
@@ -108,6 +109,8 @@ private async Task ProbeCluster(ClusterState cluster)
             return;
         }
 
+        var activity = Observability.YarpActivitySource.CreateActivity("Yarp.ActiveHealthCheck", ActivityKind.Internal);
+
         Log.StartingActiveHealthProbingOnCluster(_logger, cluster.ClusterId);
 
         var allDestinations = cluster.DestinationsState.AllDestinations;
@@ -118,7 +121,7 @@ private async Task ProbeCluster(ClusterState cluster)
 
         for (var i = 0; i < probeTasks.Length; i++)
         {
-            probeTasks[i] = ProbeDestinationAsync(cluster, allDestinations[i], timeout);
+            probeTasks[i] = ProbeDestinationAsync(cluster, allDestinations[i], timeout, activity);
         }
 
         for (var i = 0; i < probeResults.Length; i++)
@@ -150,10 +153,11 @@ private async Task ProbeCluster(ClusterState cluster)
             }
 
             Log.StoppedActiveHealthProbingOnCluster(_logger, cluster.ClusterId);
+            activity?.Stop();
         }
     }
 
-    private async Task<DestinationProbingResult> ProbeDestinationAsync(ClusterState cluster, DestinationState destination, TimeSpan timeout)
+    private async Task<DestinationProbingResult> ProbeDestinationAsync(ClusterState cluster, DestinationState destination, TimeSpan timeout, Activity? healthCheckActivity)
     {
         HttpRequestMessage request;
         try
@@ -167,19 +171,28 @@ private async Task<DestinationProbingResult> ProbeDestinationAsync(ClusterState
             return new DestinationProbingResult(destination, null, ex);
         }
 
+        var probeActivity = Observability.YarpActivitySource.StartActivity("Probe Destination", ActivityKind.Internal);
+        probeActivity?.AddTag("Cluster", cluster.ClusterId);
+        probeActivity?.AddTag("Destination", destination.DestinationId);
         var cts = new CancellationTokenSource(timeout);
         try
         {
             Log.SendingHealthProbeToEndpointOfDestination(_logger, request.RequestUri, destination.DestinationId, cluster.ClusterId);
             var response = await cluster.Model.HttpClient.SendAsync(request, cts.Token);
             Log.DestinationProbingCompleted(_logger, destination.DestinationId, cluster.ClusterId, (int)response.StatusCode);
 
+            probeActivity?.SetStatus(ActivityStatusCode.Ok);
+            probeActivity?.Stop();
+
             return new DestinationProbingResult(destination, response, null);
         }
         catch (Exception ex)
         {
             Log.DestinationProbingFailed(_logger, destination.DestinationId, cluster.ClusterId, ex);
 
+            probeActivity?.SetStatus(ActivityStatusCode.Error);
+            probeActivity?.Stop();
+
             return new DestinationProbingResult(destination, null, ex);
         }
         finally

src/ReverseProxy/Model/IReverseProxyFeature.cs

@@ -2,6 +2,7 @@
 // Licensed under the MIT License.
 
 using System.Collections.Generic;
+using System.Diagnostics;
 
 namespace Yarp.ReverseProxy.Model;
 
@@ -35,4 +36,6 @@ public interface IReverseProxyFeature
     /// The actual destination that the request was proxied to.
     /// </summary>
     DestinationState? ProxiedDestination { get; set; }
+
+    Activity? ActivityForTracing { get; } 
 }

src/ReverseProxy/Model/ReverseProxyFeature.cs

@@ -3,6 +3,8 @@
 
 using System;
 using System.Collections.Generic;
+using System.Diagnostics;
+using Yarp.ReverseProxy.Utilities;
 
 namespace Yarp.ReverseProxy.Model;
 
@@ -31,4 +33,12 @@ public IReadOnlyList<DestinationState> AvailableDestinations
 
     /// <inheritdoc/>
     public DestinationState? ProxiedDestination { get; set; }
+
+    public Activity? ActivityForTracing { get; init; }
+
+    public ReverseProxyFeature()
+    {
+        if (Observability.IsListening)
+        ActivityForTracing = (Observability.IsListening) ? Observability.YarpActivitySource.CreateActivity("Forward", ActivityKind.Server) : null;
+    }
 }

src/ReverseProxy/Utilities/Observability.cs

@@ -0,0 +1,18 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+using System.Diagnostics;
+
+namespace Yarp.ReverseProxy.Utilities
+{
+    internal class Observability
+    {
+        private static ActivitySource _activitySource = new ActivitySource("Yarp.ReverseProxy");
+
+        public static ActivitySource YarpActivitySource => _activitySource;
+
+        public static bool IsListening => _activitySource.HasListeners();
+    }
+}