#316 Start work on metrics

Signed-off-by: James Thompson <thompson.tomo@outlook.com>

Author: thompson-tomo

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

@@ -37,7 +37,6 @@ internal sealed class CommandWriter : IAsyncDisposable
     private readonly int _arrayPoolInitialSize;
     private readonly object _lock = new();
     private readonly CancellationTokenSource _cts;
-    private readonly ConnectionStatsCounter _counter;
     private readonly Memory<byte> _consolidateMem = new byte[SendMemSize].AsMemory();
     private readonly TimeSpan _defaultCommandTimeout;
     private readonly Action<PingCommand> _enqueuePing;
@@ -55,7 +54,7 @@ internal sealed class CommandWriter : IAsyncDisposable
     private CancellationTokenSource? _ctsReader;
     private volatile bool _disposed;
 
-    public CommandWriter(string name, NatsConnection connection, ObjectPool pool, NatsOpts opts, ConnectionStatsCounter counter, Action<PingCommand> enqueuePing, TimeSpan? overrideCommandTimeout = default)
+    public CommandWriter(string name, NatsConnection connection, ObjectPool pool, NatsOpts opts, Action<PingCommand> enqueuePing, TimeSpan? overrideCommandTimeout = default)
     {
         _logger = opts.LoggerFactory.CreateLogger<CommandWriter>();
         _trace = _logger.IsEnabled(LogLevel.Trace);
@@ -67,7 +66,6 @@ public CommandWriter(string name, NatsConnection connection, ObjectPool pool, Na
         // avoid defining another option.
         _arrayPoolInitialSize = opts.WriterBufferSize / 256;
 
-        _counter = counter;
         _defaultCommandTimeout = overrideCommandTimeout ?? opts.CommandTimeout;
         _enqueuePing = enqueuePing;
         _protocolWriter = new ProtocolWriter(opts.SubjectEncoding);
@@ -321,6 +319,8 @@ public ValueTask PublishAsync<T>(string subject, T? value, NatsHeaders? headers,
                 serializer.Serialize(payloadBuffer, value);
 
             var size = payloadBuffer.WrittenMemory.Length + (headersBuffer?.WrittenMemory.Length ?? 0);
+            Activity.Current?.AddTag("messaging.message.body.size", payloadBuffer.WrittenMemory.Length);
+            Activity.Current?.AddTag("messaging.message.envelope.size", size);
             if (_connection.ServerInfo is { } info && size > info.MaxPayload)
             {
                 throw new NatsPayloadTooLargeException($"Payload size {size} exceeds server's maximum payload size {info.MaxPayload}");
@@ -693,11 +693,19 @@ private void EnqueueCommand()
             return;
         }
 
-        Interlocked.Add(ref _counter.PendingMessages, 1);
+        Telemetry.AddPendingMessages(1, Activity.Current.TagObjects.ToArray());
 
         _channelSize.Writer.TryWrite(size);
         var flush = _pipeWriter.FlushAsync();
-        _flushTask = flush.IsCompletedSuccessfully ? null : flush.AsTask();
+        if (flush.IsCompletedSuccessfully)
+        {
+            _flushTask = null;
+            Telemetry.AddPendingMessages(-1, Activity.Current.TagObjects.ToArray());
+        }
+        else
+        {
+            _flushTask = flush.AsTask();
+        }
     }
 
     private async ValueTask ConnectStateMachineAsync(bool lockHeld, ClientOpts connectOpts, CancellationToken cancellationToken)

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

@@ -6,9 +6,9 @@ internal sealed class PriorityCommandWriter : IAsyncDisposable
 {
     private int _disposed;
 
-    public PriorityCommandWriter(NatsConnection connection, ObjectPool pool, SocketConnectionWrapper socketConnection, NatsOpts opts, ConnectionStatsCounter counter, Action<PingCommand> enqueuePing)
+    public PriorityCommandWriter(NatsConnection connection, ObjectPool pool, SocketConnectionWrapper socketConnection, NatsOpts opts, Action<PingCommand> enqueuePing)
     {
-        CommandWriter = new CommandWriter("init", connection, pool, opts, counter, enqueuePing);
+        CommandWriter = new CommandWriter("init", connection, pool, opts, enqueuePing);
         CommandWriter.Reset(socketConnection);
     }
 

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

@@ -33,7 +33,7 @@ public NatsReadProtocolProcessor(SocketConnectionWrapper socketConnection, NatsC
         _waitForPongOrErrorSignal = waitForPongOrErrorSignal;
         _infoParsed = infoParsed;
         _pingCommands = new ConcurrentQueue<PingCommand>();
-        _socketReader = new SocketReader(socketConnection, connection.Opts.ReaderBufferSize, connection.Counter, connection.Opts.LoggerFactory);
+        _socketReader = new SocketReader(socketConnection, connection.Opts.ReaderBufferSize, connection.Opts.LoggerFactory);
         _readLoop = Task.Run(ReadLoopAsync);
     }
 
@@ -156,7 +156,7 @@ private async Task ReadLoopAsync()
                         code = GetCode(buffer);
                     }
 
-                    Interlocked.Increment(ref _connection.Counter.ReceivedMessages);
+                    Telemetry.AddReceivedMessages(1, Activity.Current.TagObjects.ToArray());
 
                     // Optimize for Msg parsing, Inline async code
                     if (code == ServerOpCodes.Msg)
@@ -445,7 +445,7 @@ private async ValueTask<ReadOnlySequence<byte>> DispatchCommandAsync(int code, R
         {
             // reaches invalid line, log warn and try to get newline and go to nextloop.
             _logger.LogWarning(NatsLogEvents.Protocol, "Reached invalid line");
-            Interlocked.Decrement(ref _connection.Counter.ReceivedMessages);
+            Telemetry.AddReceivedMessages(-1, Activity.Current.TagObjects.ToArray());
 
             var position = buffer.PositionOf((byte)'\n');
             if (position == null)

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

@@ -9,7 +9,6 @@ namespace NATS.Client.Core.Internal;
 internal sealed class SocketReader
 {
     private readonly int _minimumBufferSize;
-    private readonly ConnectionStatsCounter _counter;
     private readonly SeqeunceBuilder _seqeunceBuilder = new SeqeunceBuilder();
     private readonly Stopwatch _stopwatch = new Stopwatch();
     private readonly ILogger<SocketReader> _logger;
@@ -18,11 +17,10 @@ internal sealed class SocketReader
 
     private Memory<byte> _availableMemory;
 
-    public SocketReader(SocketConnectionWrapper socketConnection, int minimumBufferSize, ConnectionStatsCounter counter, ILoggerFactory loggerFactory)
+    public SocketReader(SocketConnectionWrapper socketConnection, int minimumBufferSize, ILoggerFactory loggerFactory)
     {
         _socketConnection = socketConnection;
         _minimumBufferSize = minimumBufferSize;
-        _counter = counter;
         _logger = loggerFactory.CreateLogger<SocketReader>();
         _isTraceLogging = _logger.IsEnabled(LogLevel.Trace);
     }
@@ -66,7 +64,7 @@ public async ValueTask<ReadOnlySequence<byte>> ReadAtLeastAsync(int minimumSize)
             }
 
             totalRead += read;
-            Interlocked.Add(ref _counter.ReceivedBytes, read);
+            Telemetry.RecordReceivedBytes(read, Activity.Current.TagObjects.ToArray());
             _seqeunceBuilder.Append(_availableMemory.Slice(0, read));
             _availableMemory = _availableMemory.Slice(read);
         }
@@ -112,7 +110,7 @@ public async ValueTask<ReadOnlySequence<byte>> ReadUntilReceiveNewLineAsync()
                 throw ex;
             }
 
-            Interlocked.Add(ref _counter.ReceivedBytes, read);
+            Telemetry.RecordReceivedBytes(read, Activity.Current.TagObjects.ToArray());
             var appendMemory = _availableMemory.Slice(0, read);
             _seqeunceBuilder.Append(appendMemory);
             _availableMemory = _availableMemory.Slice(read);

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

@@ -1,6 +1,7 @@
 using System.Buffers;
 using System.Collections.Concurrent;
 using System.Runtime.CompilerServices;
+using System.Threading;
 using Microsoft.Extensions.Logging;
 using NATS.Client.Core.Commands;
 
@@ -51,41 +52,27 @@ public SubscriptionManager(NatsConnection connection, string inboxPrefix)
 
     public ValueTask SubscribeAsync(NatsSubBase sub, CancellationToken cancellationToken)
     {
-        if (Telemetry.HasListeners())
+        ValueTask task;
+        using var activity = Telemetry.StartSendActivity($"{_connection.SpanDestinationName(sub.Subject)} {Telemetry.Constants.SubscribeActivityName}", _connection, sub.Subject, null, null);
+        if (activity != null)
         {
-            using var activity = Telemetry.StartSendActivity($"{_connection.SpanDestinationName(sub.Subject)} {Telemetry.Constants.SubscribeActivityName}", _connection, sub.Subject, null, null);
             try
             {
-                if (IsInboxSubject(sub.Subject))
-                {
-                    if (sub.QueueGroup != null)
-                    {
-                        throw new NatsException("Inbox subscriptions don't support queue groups");
-                    }
-
-                    return SubscribeInboxAsync(sub, cancellationToken);
-                }
-
-                return SubscribeInternalAsync(sub.Subject, sub.QueueGroup, sub.Opts, sub, cancellationToken);
+                task = SubscribeImpAsync(sub, cancellationToken);
+                Telemetry.RecordOperationDuration(activity.Duration.TotalSeconds, activity.TagObjects.ToArray());
             }
             catch (Exception ex)
             {
                 Telemetry.SetException(activity, ex);
                 throw;
             }
         }
-
-        if (IsInboxSubject(sub.Subject))
+        else
         {
-            if (sub.QueueGroup != null)
-            {
-                throw new NatsException("Inbox subscriptions don't support queue groups");
-            }
-
-            return SubscribeInboxAsync(sub, cancellationToken);
+            task = SubscribeImpAsync(sub, cancellationToken);
         }
 
-        return SubscribeInternalAsync(sub.Subject, sub.QueueGroup, sub.Opts, sub, cancellationToken);
+        return task;
     }
 
     public ValueTask PublishToClientHandlersAsync(string subject, string? replyTo, int sid, in ReadOnlySequence<byte>? headersBuffer, in ReadOnlySequence<byte> payloadBuffer)
@@ -262,6 +249,21 @@ await SubscribeInternalAsync(
         }
     }
 
+    private ValueTask SubscribeImpAsync(NatsSubBase sub, CancellationToken cancellationToken)
+    {
+        if (IsInboxSubject(sub.Subject))
+        {
+            if (sub.QueueGroup != null)
+            {
+                throw new NatsException("Inbox subscriptions don't support queue groups");
+            }
+
+            return SubscribeInboxAsync(sub, cancellationToken);
+        }
+
+        return SubscribeInternalAsync(sub.Subject, sub.QueueGroup, sub.Opts, sub, cancellationToken);
+    }
+
     private async ValueTask SubscribeInboxAsync(NatsSubBase sub, CancellationToken cancellationToken)
     {
         await InitializeInboxSubscriptionAsync(cancellationToken).ConfigureAwait(false);

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

@@ -1,17 +1,58 @@
 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
 internal static class Telemetry
 {
-    internal static readonly ActivitySource NatsActivities = new(name: NatsActivitySource);
-
     private const string NatsActivitySource = "NATS.Net";
-    private static readonly object BoxedTrue = true;
+    private const string MeterName = "NATS.Net";
+    private static readonly ActivitySource NatsActivities = new(name: NatsActivitySource);
+    private static readonly Meter NatsMeter = new Meter(MeterName);
+
+    private static readonly Counter<long> _subscriptionCounter = NatsMeter.CreateCounter<long>(
+            Constants.SubscriptionInstrumentName,
+            unit: "{subscriptions}",
+            description: "Number of subscriptions");
+
+    private static readonly Counter<long> _pendingMessagesCounter = NatsMeter.CreateCounter<long>(
+            Constants.PendingMessagesInstrumentName,
+            unit: "{messages}",
+            description: "Number of pending messages");
+
+    private static readonly Histogram<long> _sentBytesHistogram = NatsMeter.CreateHistogram<long>(
+            Constants.SentBytesInstrumentName,
+            unit: "{bytes}",
+            description: "Number of bytes sent");
+
+    private static readonly Histogram<long> _receivedBytesHistogram = NatsMeter.CreateHistogram<long>(
+            Constants.ReceivedBytesInstrumentName,
+            unit: "{bytes}",
+            description: "Number of bytes received");
+
+    private static readonly Counter<long> _sentMessagesCounter = NatsMeter.CreateCounter<long>(
+            Constants.SentMessagesInstrumentName,
+            unit: "{messages}",
+            description: "Number of messages sent");
+
+    private static readonly Counter<long> _receivedMessagesCounter = NatsMeter.CreateCounter<long>(
+            Constants.ReceivedMessagesInstrumentName,
+            unit: "{messages}",
+            description: "Number of messages received");
+
+    private static readonly Histogram<double> _durationOperationHistogram = NatsMeter.CreateHistogram<double>(
+        Constants.DurationOperationInstrumentName,
+        unit: "{s}",
+        description: "Duration of messaging operation initiated by a producer or consumer client.");
+
+    private static readonly Histogram<double> _durationProcessHistogram = NatsMeter.CreateHistogram<double>(
+        Constants.DurationProcessInstrumentName,
+        unit: "{s}",
+        description: "Duration of processing operation within client.");
 
-    internal static bool HasListeners() => NatsActivities.HasListeners();
+    private static readonly object BoxedTrue = true;
 
     internal static Activity? StartSendActivity(
         string name,
@@ -20,9 +61,6 @@ internal static class Telemetry
         string? replyTo,
         ActivityContext? parentContext = null)
     {
-        if (!NatsActivities.HasListeners())
-            return null;
-
         KeyValuePair<string, object?>[] tags;
         if (connection is NatsConnection { ServerInfo: not null } conn)
         {
@@ -106,9 +144,6 @@ internal static void AddTraceContextHeaders(Activity? activity, ref NatsHeaders?
         long size,
         NatsHeaders? headers)
     {
-        if (!NatsActivities.HasListeners())
-            return null;
-
         KeyValuePair<string, object?>[] tags;
         if (connection is NatsConnection { ServerInfo: not null } conn)
         {
@@ -216,6 +251,51 @@ static string GetStackTrace(Exception? exception)
         }
     }
 
+    internal static void IncrementSubscriptionCount(KeyValuePair<string, object?>[] tags)
+    {
+        _subscriptionCounter.Add(1, tags);
+    }
+
+    internal static void DecrementSubscriptionCount(KeyValuePair<string, object?>[] tags)
+    {
+        _subscriptionCounter.Add(-1, tags);
+    }
+
+    internal static void AddPendingMessages(long messages, KeyValuePair<string, object?>[] tags)
+    {
+        _pendingMessagesCounter.Add(messages, tags);
+    }
+
+    internal static void RecordSentBytes(long bytes, KeyValuePair<string, object?>[] tags)
+    {
+        _sentBytesHistogram.Record(bytes, tags);
+    }
+
+    internal static void RecordReceivedBytes(long bytes, KeyValuePair<string, object?>[] tags)
+    {
+        _receivedBytesHistogram.Record(bytes, tags);
+    }
+
+    internal static void AddSentMessages(long messages, KeyValuePair<string, object?>[] tags)
+    {
+        _sentMessagesCounter.Add(messages, tags);
+    }
+
+    internal static void AddReceivedMessages(long messages, KeyValuePair<string, object?>[] tags)
+    {
+        _receivedMessagesCounter.Add(messages, tags);
+    }
+
+    internal static void RecordOperationDuration(double duration, KeyValuePair<string, object?>[] tags)
+    {
+        _durationOperationHistogram.Record(duration, tags);
+    }
+
+    internal static void RecordProcessDuration(long bytes, KeyValuePair<string, object?>[] tags)
+    {
+        _durationProcessHistogram.Record(bytes, tags);
+    }
+
     private static bool TryParseTraceContext(NatsHeaders headers, out ActivityContext context)
     {
         DistributedContextPropagator.Current.ExtractTraceIdAndState(
@@ -278,7 +358,7 @@ internal class Constants
         public const string DestIsTemporary = "messaging.destination.temporary";
         public const string DestPubName = "messaging.destination_publish.name";
 
-        public const string QueueGroup = "messaging.nats.consumer.group";
+        public const string QueueGroup = "messaging.consumer.group.name";
         public const string ReplyTo = "messaging.nats.message.reply_to";
         public const string Subject = "messaging.nats.message.subject";
 
@@ -289,5 +369,14 @@ internal class Constants
         public const string NetworkPeerAddress = "network.peer.address";
         public const string NetworkPeerPort = "network.peer.port";
         public const string NetworkLocalAddress = "network.local.address";
+
+        public const string PendingMessagesInstrumentName = $"messaging.client.pending.messages";
+        public const string SentBytesInstrumentName = $"messaging.client.sent.bytes";
+        public const string ReceivedBytesInstrumentName = $"messaging.client.consumed.bytes";
+        public const string SentMessagesInstrumentName = $"messaging.client.sent.messages";
+        public const string ReceivedMessagesInstrumentName = $"messaging.client.consumed.messages";
+        public const string SubscriptionInstrumentName = $"messaging.client.nats.subscription.count";
+        public const string DurationOperationInstrumentName = $"messaging.client.operation.duration";
+        public const string DurationProcessInstrumentName = $"messaging.process.duration";
     }
 }