Add OTel metrics support

examples/Example.OpenTelemetry/ClientApp.cs

@@ -1,6 +1,9 @@
 using System.Diagnostics;
+using Microsoft.Extensions.Logging;
 using NATS.Client.Core;
 using OpenTelemetry;
+using OpenTelemetry.Logs;
+using OpenTelemetry.Metrics;
 using OpenTelemetry.Resources;
 using OpenTelemetry.Trace;
 
@@ -13,19 +16,41 @@ public static async Task Run()
         var serviceName = "ClientApp";
         var serviceVersion = "1.0.0";
 
+        var resourceBuilder = ResourceBuilder.CreateDefault().AddService(serviceName: serviceName, serviceVersion: serviceVersion);
+
         using var tracerProvider = Sdk.CreateTracerProviderBuilder()
             .AddOtlpExporter()
-            .SetResourceBuilder(ResourceBuilder.CreateDefault().AddService(serviceName: serviceName, serviceVersion: serviceVersion))
-            .AddSource("NATS.Net")
+            .SetResourceBuilder(resourceBuilder)
+            .AddSource(NatsTelemetry.SourceName)
             .AddSource("MyClientSource")
             .Build();
 
+        using var meterProvider = Sdk.CreateMeterProviderBuilder()
+            .AddOtlpExporter()
+            .SetResourceBuilder(resourceBuilder)
+            .AddMeter(NatsTelemetry.SourceName)
+            .Build();
+
+        using var loggerFactory = LoggerFactory.Create(builder =>
+        {
+            builder.AddOpenTelemetry(options =>
+            {
+                options.SetResourceBuilder(resourceBuilder);
+                options.IncludeFormattedMessage = true;
+                options.IncludeScopes = true;
+                options.ParseStateValues = true;
+                options.AddOtlpExporter();
+            });
+        });
+        var logger = loggerFactory.CreateLogger(serviceName);
+
         ActivitySource activitySource = new("MyClientSource");
 
-        Console.WriteLine("Client App is starting...");
+        logger.LogInformation("Client App is starting...");
 
         await using var nats = new NatsConnection(new NatsOpts
         {
+            LoggerFactory = loggerFactory,
             RequestReplyMode = NatsRequestReplyMode.Direct,
         });
 
@@ -34,7 +59,7 @@ public static async Task Run()
             await nats.PublishAsync("greet.presence.client.app", "ClientApp is here!");
 
             var response = await nats.RequestAsync<string, string>("greet.hi", "Hi, telemetry!");
-            Console.WriteLine($"Response: {response}");
+            logger.LogInformation("Response: {Response}", response);
         }
     }
 }

examples/Example.OpenTelemetry/Program.cs

@@ -2,26 +2,28 @@
 
                          OpenTelemetry Example
 
-(1) Run Jaeger locally and then run the client and server apps.
+Both apps export traces and metrics via OTLP. Point them at any backend that
+accepts OTLP (Aspire dashboard, Jaeger for traces, Grafana stack, etc.).
 
-https://www.jaegertracing.io/download/
-
-https://medium.com/jaegertracing/introducing-native-support-for-opentelemetry-in-jaeger-eb661be8183c
+(1) Start Aspire dashboard:
 
 ```powershell
-> $env:COLLECTOR_OTLP_ENABLED=true
-> jaeger-all-in-one.exe
+> docker run --rm -it `
+  -p 18888:18888 -p 4317:18889 `
+  -e DASHBOARD__OTLP__AUTHMODE=Unsecured `
+  mcr.microsoft.com/dotnet/aspire-dashboard:latest
 ```
 
 or
 
 ```bash
-$ COLLECTOR_OTLP_ENABLED=true jaeger-all-in-one
+$ docker run --rm -it \
+  -p 18888:18888 -p 4317:18889 \
+  -e DASHBOARD__OTLP__AUTHMODE=Unsecured \
+  mcr.microsoft.com/dotnet/aspire-dashboard:latest
 ```
 
-(2) Jaeger UI default URL http://localhost:16686/search
-
-(3) In different terminals run:
+(2) In different terminals run:
 
 ```
 nats-server

examples/Example.OpenTelemetry/ServiceApp.cs

@@ -1,6 +1,9 @@
 using System.Diagnostics;
+using Microsoft.Extensions.Logging;
 using NATS.Client.Core;
 using OpenTelemetry;
+using OpenTelemetry.Logs;
+using OpenTelemetry.Metrics;
 using OpenTelemetry.Resources;
 using OpenTelemetry.Trace;
 
@@ -13,19 +16,41 @@ public static async Task Run()
         var serviceName = "ServiceApp";
         var serviceVersion = "1.0.0";
 
+        var resourceBuilder = ResourceBuilder.CreateDefault().AddService(serviceName: serviceName, serviceVersion: serviceVersion);
+
         using var tracerProvider = Sdk.CreateTracerProviderBuilder()
             .AddOtlpExporter()
-            .SetResourceBuilder(ResourceBuilder.CreateDefault().AddService(serviceName: serviceName, serviceVersion: serviceVersion))
-            .AddSource("NATS.Net")
+            .SetResourceBuilder(resourceBuilder)
+            .AddSource(NatsTelemetry.SourceName)
             .AddSource("MyServiceSource")
             .Build();
 
+        using var meterProvider = Sdk.CreateMeterProviderBuilder()
+            .AddOtlpExporter()
+            .SetResourceBuilder(resourceBuilder)
+            .AddMeter(NatsTelemetry.SourceName)
+            .Build();
+
+        using var loggerFactory = LoggerFactory.Create(builder =>
+        {
+            builder.AddOpenTelemetry(options =>
+            {
+                options.SetResourceBuilder(resourceBuilder);
+                options.IncludeFormattedMessage = true;
+                options.IncludeScopes = true;
+                options.ParseStateValues = true;
+                options.AddOtlpExporter();
+            });
+        });
+        var logger = loggerFactory.CreateLogger(serviceName);
+
         ActivitySource activitySource = new("MyServiceSource");
 
-        Console.WriteLine("Service App is starting...");
+        logger.LogInformation("Service App is starting...");
 
         await using var nats = new NatsConnection(new NatsOpts
         {
+            LoggerFactory = loggerFactory,
             RequestReplyMode = NatsRequestReplyMode.Direct,
         });
 
@@ -35,7 +60,7 @@ public static async Task Run()
 
             if (msg.Subject.StartsWith("greet.presence"))
             {
-                Console.WriteLine($"{msg.Data} is here!");
+                logger.LogInformation("{Data} is here!", msg.Data);
 
                 activity?.AddEvent(new ActivityEvent("Presence", tags: new()
                 {

src/NATS.Client.Core/Commands/CommandWriter.cs

@@ -349,6 +349,21 @@ public ValueTask PublishAsync<T>(string subject, T? value, NatsHeaders? headers,
             {
                 throw new NatsPayloadTooLargeException($"Payload size {size} exceeds server's maximum payload size {info.MaxPayload}");
             }
+
+            // Per OTel messaging semconv, messaging.client.published.messages counts publish
+            // attempts, not successful wire writes -- so we increment here, before the
+            // _semLock / _disposed / state-machine paths that might fail to enqueue.
+            // sent.bytes follows the same "attempted" convention for consistency. Operators
+            // wanting a "successfully published" view subtract operation.duration samples
+            // that carry an error.type tag.
+            if (Telemetry.PublishedMessages.Enabled || Telemetry.SentBytes.Enabled)
+            {
+                var tags = Telemetry.BuildMetricTags(_connection, Telemetry.Constants.OpPub);
+                if (Telemetry.PublishedMessages.Enabled)
+                    Telemetry.PublishedMessages.Add(1, tags);
+                if (Telemetry.SentBytes.Enabled)
+                    Telemetry.SentBytes.Add(size, tags);
+            }
         }
         catch
         {

src/NATS.Client.Core/Internal/ReplyTask.cs

@@ -19,6 +19,8 @@ internal sealed class ReplyTask<T> : ReplyTaskBase, IDisposable
     private readonly TimeSpan _requestTimeout;
     private readonly TaskCompletionSource _tcs;
     private NatsMsg<T> _msg;
+    private long _replyBytes;
+    private bool _isNoResponders;
 
     public ReplyTask(ReplyTaskFactory factory, long id, string subject, NatsConnection connection, INatsDeserialize<T> deserializer, TimeSpan requestTimeout)
     {
@@ -51,17 +53,39 @@ await _tcs.Task
             NatsNoReplyException.Throw();
         }
 
+        NatsMsg<T> msg;
+        long bytes;
+        bool isNoResponders;
         lock (_gate)
         {
-            return _msg;
+            msg = _msg;
+            bytes = _replyBytes;
+            isNoResponders = _isNoResponders;
         }
+
+        // Count only messages actually delivered to the caller. Late replies that arrive
+        // after a timeout still hit SetResult, but the user never sees them, so the
+        // counters belong here on the success path, not in SetResult. 503 NoResponders
+        // sentinels are also excluded for parity with the SharedInbox path.
+        if (!isNoResponders && (Telemetry.ConsumedMessages.Enabled || Telemetry.ReceivedBytes.Enabled))
+        {
+            var tags = Telemetry.BuildMetricTags(_connection, Telemetry.Constants.OpRec);
+            if (Telemetry.ConsumedMessages.Enabled)
+                Telemetry.ConsumedMessages.Add(1, tags);
+            if (Telemetry.ReceivedBytes.Enabled)
+                Telemetry.ReceivedBytes.Add(bytes, tags);
+        }
+
+        return msg;
     }
 
     public override void SetResult(string? replyTo, ReadOnlySequence<byte> payload, ReadOnlySequence<byte>? headersBuffer)
     {
         lock (_gate)
         {
             _msg = NatsMsg<T>.Build(Subject, replyTo, headersBuffer, payload, _connection, _connection.HeaderParser, _deserializer);
+            _isNoResponders = payload.Length == 0 && NatsSubBase.IsHeader503(headersBuffer);
+            _replyBytes = payload.Length + (headersBuffer?.Length ?? 0);
         }
 
         _tcs.TrySetResult();

src/NATS.Client.Core/Internal/Telemetry.cs

@@ -1,17 +1,114 @@
 using System.Diagnostics;
+using System.Diagnostics.Metrics;
 
 namespace NATS.Client.Core.Internal;
 
 // https://opentelemetry.io/docs/specs/semconv/attributes-registry/messaging/
 // https://opentelemetry.io/docs/specs/semconv/messaging/messaging-spans/#messaging-attributes
+// https://opentelemetry.io/docs/specs/semconv/messaging/messaging-metrics/
 internal static class Telemetry
 {
-    public const string NatsActivitySource = "NATS.Net";
+    public const string NatsActivitySource = NatsTelemetry.SourceName;
     public static readonly ActivitySource NatsActivities = new(name: NatsActivitySource);
+
+    public static readonly Meter NatsMeter = new(name: NatsActivitySource);
+
+    public static readonly Counter<long> PublishedMessages =
+        NatsMeter.CreateCounter<long>("messaging.client.published.messages", unit: "{message}");
+
+    public static readonly Counter<long> ConsumedMessages =
+        NatsMeter.CreateCounter<long>("messaging.client.consumed.messages", unit: "{message}");
+
+    // OTel messaging semconv recommends these advisory buckets for messaging.client.operation.duration.
+    // https://opentelemetry.io/docs/specs/semconv/messaging/messaging-metrics/#metric-messagingclientoperationduration
+    public static readonly Histogram<double> OperationDuration =
+        NatsMeter.CreateHistogram<double>(
+            "messaging.client.operation.duration",
+            unit: "s",
+            advice: new InstrumentAdvice<double>
+            {
+                HistogramBucketBoundaries = new[] { 0.005, 0.01, 0.025, 0.05, 0.075, 0.1, 0.25, 0.5, 0.75, 1, 2.5, 5, 7.5, 10 },
+            });
+
+    public static readonly UpDownCounter<long> ActiveSubscriptions =
+        NatsMeter.CreateUpDownCounter<long>("nats.client.active_subscriptions", unit: "{subscription}");
+
+    public static readonly Counter<long> Reconnects =
+        NatsMeter.CreateCounter<long>("nats.client.reconnects", unit: "{reconnect}");
+
+    public static readonly Counter<long> SentBytes =
+        NatsMeter.CreateCounter<long>("nats.client.sent.bytes", unit: "By");
+
+    public static readonly Counter<long> ReceivedBytes =
+        NatsMeter.CreateCounter<long>("nats.client.received.bytes", unit: "By");
+
     private static readonly object BoxedTrue = true;
 
+    /// <summary>
+    /// Don't use this for metrics.
+    /// </summary>
     public static bool HasListeners() => NatsActivities.HasListeners();
 
+    public static void RecordOperationDuration(long startTimestamp, INatsConnection? connection, string operation, Exception? error)
+    {
+        if (!OperationDuration.Enabled)
+            return;
+
+        try
+        {
+            var elapsed = (Stopwatch.GetTimestamp() - startTimestamp) / (double)Stopwatch.Frequency;
+            var tags = BuildMetricTags(connection, operation);
+            if (error is not null)
+                tags.Add(Constants.ErrorTypeKey, error.GetType().FullName ?? "unknown");
+
+            OperationDuration.Record(elapsed, tags);
+        }
+        catch
+        {
+            // Instrumentation must never break the calling operation. A buggy MeterListener
+            // or tag construction failure here would otherwise replace the in-flight messaging
+            // exception (catch/finally semantics), hiding the real failure from the caller.
+        }
+    }
+
+    public static async ValueTask MeasureOperationAsync(ValueTask task, long startTimestamp, INatsConnection? connection, string operation)
+    {
+        try
+        {
+            await task.ConfigureAwait(false);
+            RecordOperationDuration(startTimestamp, connection, operation, null);
+        }
+        catch (Exception ex)
+        {
+            RecordOperationDuration(startTimestamp, connection, operation, ex);
+            throw;
+        }
+    }
+
+    public static TagList BuildMetricTags(INatsConnection? connection, string operation)
+    {
+        var tags = default(TagList);
+
+        // MetricTagsPrefix is read off the concrete NatsConnection by design: exposing it on
+        // INatsConnection would be a breaking change for external implementers, and default
+        // interface members aren't available on netstandard2.0. Custom INatsConnection wrappers
+        // therefore fall through to the minimal tag set below. Normal usage (including via
+        // NatsJSContext) hits this branch because the runtime type is NatsConnection regardless
+        // of the static type the caller holds.
+        if (connection is NatsConnection { MetricTagsPrefix: { } prefix })
+        {
+            for (var i = 0; i < prefix.Length; i++)
+                tags.Add(prefix[i]);
+        }
+        else
+        {
+            tags.Add(Constants.SystemKey, Constants.SystemVal);
+        }
+
+        tags.Add(Constants.OpKey, operation);
+        return tags;
+    }
+
     public static Activity? StartSendActivity(
         string name,
         INatsConnection? connection,
@@ -300,6 +397,10 @@ public class Constants
         public const string OpKey = "messaging.operation";
         public const string OpPub = "publish";
         public const string OpRec = "receive";
+        public const string OpSub = "subscribe";
+        public const string OpReq = "request";
+        public const string OpReconnect = "reconnect";
+        public const string ErrorTypeKey = "error.type";
         public const string MsgBodySize = "messaging.message.body.size";
         public const string MsgTotalSize = "messaging.message.envelope.size";
 

src/NATS.Client.Core/NATS.Client.Core.csproj

@@ -20,13 +20,18 @@
     <!--  Dependencies for prior to net5.0-->
     <ItemGroup Condition="!$([MSBuild]::IsTargetFrameworkCompatible('$(TargetFramework)', 'net5.0'))">
         <PackageReference Include="System.Threading.Channels" Version="8.0.0"/>
-        <PackageReference Include="System.Diagnostics.DiagnosticSource" Version="8.0.1"/>
         <PackageReference Include="Nullable" Version="1.3.1">
             <PrivateAssets>all</PrivateAssets>
             <IncludeAssets>runtime; build; native; contentfiles; analyzers</IncludeAssets>
         </PackageReference>
     </ItemGroup>
 
+    <!--  Dependencies for prior to net10.0 (InstrumentAdvice<T> requires DiagnosticSource 9.0+) -->
+    <!--  9.0.1 is the minimum floor; 9.0.0 trims ActivitySource ctor under NativeAOT (dotnet/runtime#109872). -->
+    <ItemGroup Condition="!$([MSBuild]::IsTargetFrameworkCompatible('$(TargetFramework)', 'net10.0'))">
+        <PackageReference Include="System.Diagnostics.DiagnosticSource" Version="9.0.1"/>
+    </ItemGroup>
+
     <!--  Dependencies for netstandard2.0 only -->
     <ItemGroup Condition="'$(TargetFramework)' == 'netstandard2.0'">
         <PackageReference Include="System.Memory" Version="4.5.5"/>