feat: add otel instrumentation

ascott18 · ascott18 · commit 766fd9a914f4 · 2025-10-02T13:54:37.000-07:00
diff --git a/README.md b/README.md
@@ -75,6 +75,64 @@ Alternatively, you may configure `IntelliTect.AspNetCore.SignalR.SqlServer.SqlSe
 services.Configure<SqlServerOptions>(Configuration.GetSection("SignalR:SqlServer"));
 ```
 
+## OpenTelemetry Support
+
+This library includes OpenTelemetry instrumentation that wraps background database queries in activities, making them more easily identified and grouped in your collected telemetry.
+
+### Setup
+
+To enable OpenTelemetry collection of these trace spans, add the source to your tracing configuration:
+
+``` cs
+builder.Services.AddOpenTelemetry()
+    .WithTracing(tracing => tracing
+        .AddSource("IntelliTect.AspNetCore.SignalR.SqlServer")
+        // ... other instrumentation
+    );
+```
+
+### Activity Names
+
+The library creates activities for the following operations:
+- `SignalR.SqlServer.Install` - Database schema installation/setup
+- `SignalR.SqlServer.Start` - Receiver startup operations
+- `SignalR.SqlServer.Listen` - Service Broker listening operations (database reads)
+- `SignalR.SqlServer.Poll` - Polling operations (database reads, when Service Broker is not used)
+- `SignalR.SqlServer.Publish` - Message publishing operations (database writes)
+
+### Filtering Noise
+
+Since the SQL Server backplane performs frequent polling operations, you may want to filter out successful, fast queries to reduce trace noise. 
+
+The following example assumes using package `OpenTelemetry.Instrumentation.SqlClient` for SqlClient instrumentation. There are currently [4 different packages for SqlClient instrumentation](https://github.com/dotnet/aspire/issues/2427#issuecomment-3259572206), so your method of collecting or filtering the command details may vary if you're using Aspire's instrumentation or Azure Monitor's instrumentation. Be sure to update the CommandText filter if you customize the schema name:
+
+``` cs
+builder.Services.AddOpenTelemetry()
+    .WithTracing(tracing => tracing
+        .AddSqlClientInstrumentation(options =>
+        {
+            options.Enrich = (activity, _, cmd) => activity.SetCustomProperty("sqlCommand", cmd);
+        })
+        .AddSource("IntelliTect.AspNetCore.SignalR.SqlServer")
+        .AddProcessor<SignalRTelemetryNoiseFilter>()
+    );
+
+internal sealed class SignalRTelemetryNoiseFilter : BaseProcessor<Activity>
+{
+    public override void OnEnd(Activity activity)
+    {
+        if (activity.Status != ActivityStatusCode.Error &&
+            activity.Duration.TotalMilliseconds < 100 &&
+            activity.GetCustomProperty("sqlCommand") is DbCommand command &&
+            command.CommandText.StartsWith("SELECT [PayloadId], [Payload], [InsertedOn] FROM [SignalR]") == true)
+        {
+            // Sample out successful and fast SignalR queries
+            activity.ActivityTraceFlags &= ~ActivityTraceFlags.Recorded;
+        }
+    }
+}
+```
+
 ## Caveats
 
 As mentioned above, if SQL Server Service Broker is not available, messages will not always be transmitted immediately since a fallback of periodic querying must be used.
diff --git a/demo/DemoServer/Program.cs b/demo/DemoServer/Program.cs
@@ -1,6 +1,7 @@
+using System.Data.Common;
+using System.Diagnostics;
 using DemoServer;
 using IntelliTect.AspNetCore.SignalR.SqlServer;
-using IntelliTect.AspNetCore.SignalR.SqlServer.OpenTelemetry;
 using Microsoft.AspNetCore.SignalR;
 using OpenTelemetry;
 using OpenTelemetry.Metrics;
@@ -36,11 +37,11 @@
     )
     .WithTracing(tracing => tracing
         .AddSource(builder.Environment.ApplicationName)
+        .AddSource("IntelliTect.AspNetCore.SignalR.SqlServer")
+        .AddProcessor<SignalRMessagesNoiseFilterProcessor>()
         .AddAspNetCoreInstrumentation()
-       // .AddSignalRSqlServerInstrumentation()
         .AddSqlClientInstrumentation(tracing =>
         {
-            tracing.SetDbStatementForText = true;
             tracing.Enrich = (activity, _, cmd) => activity.SetCustomProperty("sqlCommand", cmd);
         })
 );
@@ -99,3 +100,19 @@
 app.Run();
 
 public record BroadcastRequest(string Message);
+
+
+internal sealed class SignalRMessagesNoiseFilterProcessor : BaseProcessor<Activity>
+{
+    public override void OnEnd(Activity activity)
+    {
+        if (activity.Status != ActivityStatusCode.Error &&
+            activity.Duration.TotalMilliseconds < 100 &&
+            activity.GetCustomProperty("sqlCommand") is DbCommand command &&
+            command.CommandText.StartsWith("SELECT [PayloadId], [Payload], [InsertedOn] FROM [SignalR") == true)
+        {
+            // Sample out successful and fast SignalR queries
+            activity.ActivityTraceFlags &= ~ActivityTraceFlags.Recorded;
+        }
+    }
+}
diff --git a/src/IntelliTect.AspNetCore.SignalR.SqlServer/Internal/SqlServer/SqlInstaller.cs b/src/IntelliTect.AspNetCore.SignalR.SqlServer/Internal/SqlServer/SqlInstaller.cs
@@ -22,6 +22,9 @@ public async Task Install()
                 return;
             }
 
+            using var activity = SqlServerOptions.ActivitySource.StartActivity("SignalR.SqlServer.Install");
+            activity?.SetTag("signalr.hub", tracePrefix);
+
             await options.InstallLock.WaitAsync();
             logger.LogInformation("{HubName}: Start installing SignalR SQL objects", tracePrefix);
             try
@@ -55,12 +58,13 @@ public async Task Install()
                     command.CommandText = script;
                     await command.ExecuteNonQueryAsync();
 
-                    logger.LogInformation("{HubName}: SignalR SQL objects installed", messagesTableNamePrefix);
+                    logger.LogInformation("{HubName}: SignalR SQL objects installed", tracePrefix);
                 }
             }
             catch (Exception ex)
             {
-                logger.LogError(ex, "{HubName}: Unable to install SignalR SQL objects", messagesTableNamePrefix);
+                activity?.SetStatus(ActivityStatusCode.Error, ex.Message);
+                logger.LogError(ex, "{HubName}: Unable to install SignalR SQL objects", tracePrefix);
                 throw;
             }
             finally
diff --git a/src/IntelliTect.AspNetCore.SignalR.SqlServer/Internal/SqlServer/SqlReceiver.cs b/src/IntelliTect.AspNetCore.SignalR.SqlServer/Internal/SqlServer/SqlReceiver.cs
@@ -42,6 +42,8 @@ internal class SqlReceiver : IDisposable
         private bool _disposed;
         private readonly string _maxIdSql = "SELECT [PayloadId] FROM [{0}].[{1}_Id]";
         private readonly string _selectSql = "SELECT [PayloadId], [Payload], [InsertedOn] FROM [{0}].[{1}] WHERE [PayloadId] > @PayloadId";
+        private readonly TimeSpan _activityMaxDuration = TimeSpan.FromMinutes(10);
+            
 
         public SqlReceiver(SqlServerOptions options, ILogger logger, string tableName, string tracePrefix)
         {
@@ -73,14 +75,26 @@ private async Task StartLoop()
         {
             if (_cts.IsCancellationRequested) return;
 
-            if (!_lastPayloadId.HasValue)
+            bool useBroker = _options.Mode.HasFlag(SqlServerMessageMode.ServiceBroker);
+
+            using (var activity = SqlServerOptions.ActivitySource.StartActivity("SignalR.SqlServer.Start"))
             {
-                _lastPayloadId = await GetLastPayloadId();
+                activity?.SetTag("signalr.hub", _tracePrefix);
+                activity?.SetTag("signalr.sql.mode", _options.Mode.ToString());
+
+                if (!_lastPayloadId.HasValue)
+                {
+                    _lastPayloadId = await GetLastPayloadId();
+                }
+                if (useBroker)
+                {
+                    useBroker = StartSqlDependencyListener();
+                }
             }
 
             if (_cts.IsCancellationRequested) return;
 
-            if (_options.Mode.HasFlag(SqlServerMessageMode.ServiceBroker) && StartSqlDependencyListener())
+            if (useBroker)
             {
                 await NotificationLoop(_cts.Token);
             }
@@ -103,6 +117,9 @@ private async Task NotificationLoop(CancellationToken cancellationToken)
             {
                 if (cancellationToken.IsCancellationRequested) return;
 
+                using var activity = SqlServerOptions.ActivitySource.StartActivity("SignalR.SqlServer.Listen");
+                activity?.SetTag("signalr.hub", _tracePrefix);
+
                 try
                 {
                     _logger.LogDebug("{HubStream}: Setting up SQL notification", _tracePrefix);
@@ -170,6 +187,9 @@ private async Task NotificationLoop(CancellationToken cancellationToken)
                             else
                             {
                                 // Unknown subscription error, let's stop using query notifications
+                                activity?.SetStatus(ActivityStatusCode.Error);
+                                // Dispose so this activity doesn't become the parent of the next loop.
+                                activity?.Dispose();
                                 _notificationsDisabled = true;
                                 await StartLoop();
                                 return;
@@ -180,6 +200,7 @@ private async Task NotificationLoop(CancellationToken cancellationToken)
                 catch (TaskCanceledException) { return; }
                 catch (Exception ex)
                 {
+                    activity?.SetStatus(ActivityStatusCode.Error);
                     _logger.LogError(ex, "{HubStream}: Error in SQL notification loop", _tracePrefix);
 
                     await Task.Delay(1000, cancellationToken);
@@ -196,56 +217,75 @@ private async Task NotificationLoop(CancellationToken cancellationToken)
         /// </summary>
         private async Task PollingLoop(CancellationToken cancellationToken)
         {
-            var delays = _updateLoopRetryDelays;
-            for (var retrySetIndex = 0; retrySetIndex < delays.Length; retrySetIndex++)
+            var activity = SqlServerOptions.ActivitySource.StartActivity("SignalR.SqlServer.Poll");
+            activity?.SetTag("signalr.hub", _tracePrefix);
+
+            try
             {
-                Tuple<int, int> retry = delays[retrySetIndex];
-                var retryDelay = retry.Item1;
-                var numRetries = retry.Item2;
 
-                for (var retryIndex = 0; retryIndex < numRetries; retryIndex++)
+                var delays = _updateLoopRetryDelays;
+                for (var retrySetIndex = 0; retrySetIndex < delays.Length; retrySetIndex++)
                 {
-                    if (cancellationToken.IsCancellationRequested) return;
+                    Tuple<int, int> retry = delays[retrySetIndex];
+                    var retryDelay = retry.Item1;
+                    var numRetries = retry.Item2;
 
-                    var recordCount = 0;
-                    try
+                    for (var retryIndex = 0; retryIndex < numRetries; retryIndex++)
                     {
-                        if (retryDelay > 0)
-                        {
-                            _logger.LogTrace("{HubStream}: Waiting {1}ms before checking for messages again", _tracePrefix, retryDelay);
+                        if (cancellationToken.IsCancellationRequested) return;
 
-                            await Task.Delay(retryDelay, cancellationToken);
+                        // Restart activity every 10 minutes to prevent long-running activities
+                        if (activity != null && DateTime.UtcNow - activity.StartTimeUtc > _activityMaxDuration)
+                        {
+                            activity?.Dispose();
+                            activity = SqlServerOptions.ActivitySource.StartActivity("SignalR.SqlServer.Poll");
+                            activity?.SetTag("signalr.hub", _tracePrefix);
                         }
 
-                        recordCount = await ReadRows(null);
-                    }
-                    catch (TaskCanceledException) { return; }
-                    catch (Exception ex)
-                    {
-                        _logger.LogError(ex, "{HubStream}: Error in SQL polling loop", _tracePrefix);
-                    }
+                        var recordCount = 0;
+                        try
+                        {
+                            if (retryDelay > 0)
+                            {
+                                _logger.LogTrace("{HubStream}: Waiting {Delay}ms before checking for messages again", _tracePrefix, retryDelay);
 
-                    if (recordCount > 0)
-                    {
-                        _logger.LogDebug("{HubStream}: {RecordCount} records received", _tracePrefix, recordCount);
+                                await Task.Delay(retryDelay, cancellationToken);
+                            }
 
-                        // We got records so start the retry loop again
-                        // at the lowest delay.
-                        retrySetIndex = -1;
-                        break;
-                    }
+                            recordCount = await ReadRows(null);
+                        }
+                        catch (TaskCanceledException) { return; }
+                        catch (Exception ex)
+                        {
+                            _logger.LogError(ex, "{HubStream}: Error in SQL polling loop", _tracePrefix);
+                        }
 
-                    _logger.LogTrace("{HubStream}: No records received", _tracePrefix);
+                        if (recordCount > 0)
+                        {
+                            _logger.LogDebug("{HubStream}: {RecordCount} records received", _tracePrefix, recordCount);
 
-                    var isLastRetry = retrySetIndex == delays.Length - 1 && retryIndex == numRetries - 1;
+                            // We got records so start the retry loop again
+                            // at the lowest delay.
+                            retrySetIndex = -1;
+                            break;
+                        }
 
-                    if (isLastRetry)
-                    {
-                        // Last retry loop so just stay looping on the last retry delay
-                        retryIndex--;
+                        _logger.LogTrace("{HubStream}: No records received", _tracePrefix);
+
+                        var isLastRetry = retrySetIndex == delays.Length - 1 && retryIndex == numRetries - 1;
+
+                        if (isLastRetry)
+                        {
+                            // Last retry loop so just stay looping on the last retry delay
+                            retryIndex--;
+                        }
                     }
                 }
             }
+            finally
+            {
+                activity?.Dispose();
+            }
 
             _logger.LogDebug("{HubStream}: SQL polling loop fell out", _tracePrefix);
             await StartLoop();
diff --git a/src/IntelliTect.AspNetCore.SignalR.SqlServer/SqlServerHubLifetimeManager.cs b/src/IntelliTect.AspNetCore.SignalR.SqlServer/SqlServerHubLifetimeManager.cs
@@ -297,6 +297,14 @@ public override Task SendUserAsync(string userId, string methodName, object?[]?
 
         private async Task PublishAsync(MessageType type, byte[] payload)
         {
+            using var activity = SqlServerOptions.ActivitySource.StartActivity("SignalR.SqlServer.Publish");
+            if (activity != null)
+            {
+                activity.SetTag("signalr.hub", typeof(THub).FullName!);
+                activity.SetTag("signalr.sql.message_type", type.ToString());
+                activity.SetTag("signalr.sql.message_size_bytes", payload.Length);
+            }
+
             await EnsureSqlServerConnection();
             _logger.Published(type.ToString());
 
diff --git a/src/IntelliTect.AspNetCore.SignalR.SqlServer/SqlServerOptions.cs b/src/IntelliTect.AspNetCore.SignalR.SqlServer/SqlServerOptions.cs
@@ -3,6 +3,7 @@
 
 using System;
 using System.ComponentModel.DataAnnotations;
+using System.Diagnostics;
 using System.IO;
 using System.Net;
 using System.Threading.Tasks;
@@ -14,6 +15,8 @@ namespace IntelliTect.AspNetCore.SignalR.SqlServer
     /// </summary>
     public class SqlServerOptions
     {
+        internal static readonly ActivitySource ActivitySource = new("IntelliTect.AspNetCore.SignalR.SqlServer");
+        
         /// <summary>
         /// Shared lock to prevent multiple concurrent installs against the same DB.
         /// This prevents auto-enable of service broker from deadlocking 

Original file line number	Diff line number	Diff line change
`@@ -22,6 +22,9 @@ public async Task Install()`
`22`	`22`	`return;`
`23`	`23`	`}`
`24`	`24`
	`25`	`+ using var activity = SqlServerOptions.ActivitySource.StartActivity("SignalR.SqlServer.Install");`
	`26`	`+ activity?.SetTag("signalr.hub", tracePrefix);`
	`27`	`+`
`25`	`28`	`await options.InstallLock.WaitAsync();`
`26`	`29`	`logger.LogInformation("{HubName}: Start installing SignalR SQL objects", tracePrefix);`
`27`	`30`	`try`
`@@ -55,12 +58,13 @@ public async Task Install()`
`55`	`58`	`command.CommandText = script;`
`56`	`59`	`await command.ExecuteNonQueryAsync();`
`57`	`60`
`58`		`- logger.LogInformation("{HubName}: SignalR SQL objects installed", messagesTableNamePrefix);`
	`61`	`+ logger.LogInformation("{HubName}: SignalR SQL objects installed", tracePrefix);`
`59`	`62`	`}`
`60`	`63`	`}`
`61`	`64`	`catch (Exception ex)`
`62`	`65`	`{`
`63`		`- logger.LogError(ex, "{HubName}: Unable to install SignalR SQL objects", messagesTableNamePrefix);`
	`66`	`+ activity?.SetStatus(ActivityStatusCode.Error, ex.Message);`
	`67`	`+ logger.LogError(ex, "{HubName}: Unable to install SignalR SQL objects", tracePrefix);`
`64`	`68`	`throw;`
`65`	`69`	`}`
`66`	`70`	`finally`