Add System.Diagnostics.Metrics for per-connection Zebus monitoring via net10.0 multi-targeting

src/Abc.Zebus/Abc.Zebus.csproj

@@ -1,6 +1,6 @@
 <Project Sdk="Microsoft.NET.Sdk">
   <PropertyGroup>
-    <TargetFramework>netstandard2.0</TargetFramework>
+    <TargetFrameworks>netstandard2.0;net10.0</TargetFrameworks>
     <PackageId>Zebus</PackageId>
     <Version>$(ZebusVersion)</Version>
   </PropertyGroup>

src/Abc.Zebus/Directory/PeerDirectoryClient.cs

@@ -4,6 +4,7 @@
 using System.Diagnostics;
 using System.Linq;
 using System.Threading.Tasks;
+using Abc.Zebus.Monitoring;
 using Abc.Zebus.Util;
 using Abc.Zebus.Util.Extensions;
 using Microsoft.Extensions.Logging;
@@ -259,13 +260,24 @@ private void AddOrUpdatePeerEntry(PeerDescriptor peerDescriptor, bool shouldRais
         peerEntry.SetSubscriptions(subscriptions, peerDescriptor.TimestampUtc);
 
         if (shouldRaisePeerUpdated)
+        {
             PeerUpdated?.Invoke(peerDescriptor.Peer.Id, PeerUpdateAction.Started);
+#if NET10_0_OR_GREATER
+            DirectoryMetrics.PeerUpdates.Add(1);
+#endif
+        }
 
         var observedSubscriptions = GetObservedSubscriptions(subscriptions);
         if (observedSubscriptions.Count > 0)
             PeerSubscriptionsUpdated?.Invoke(peerDescriptor.PeerId, observedSubscriptions);
 
-        PeerEntry CreatePeerEntry() => new(peerDescriptor, _globalSubscriptionsIndex);
+        PeerEntry CreatePeerEntry()
+        {
+#if NET10_0_OR_GREATER
+            DirectoryMetrics.KnownPeerCount.Add(1);
+#endif
+            return new(peerDescriptor, _globalSubscriptionsIndex);
+        }
 
         PeerEntry UpdatePeerEntry(PeerEntry entry)
         {
@@ -325,6 +337,9 @@ public void Handle(PeerStopped message)
         peer.Value.TimestampUtc = message.TimestampUtc ?? DateTime.UtcNow;
 
         PeerUpdated?.Invoke(message.PeerId, PeerUpdateAction.Stopped);
+#if NET10_0_OR_GREATER
+        DirectoryMetrics.PeerUpdates.Add(1);
+#endif
     }
 
     public void Handle(PeerDecommissioned message)
@@ -336,8 +351,14 @@ public void Handle(PeerDecommissioned message)
             return;
 
         removedPeer.RemoveSubscriptions();
+#if NET10_0_OR_GREATER
+        DirectoryMetrics.KnownPeerCount.Add(-1);
+#endif
 
         PeerUpdated?.Invoke(message.PeerId, PeerUpdateAction.Decommissioned);
+#if NET10_0_OR_GREATER
+        DirectoryMetrics.PeerUpdates.Add(1);
+#endif
     }
 
     public void Handle(PeerSubscriptionsUpdated message)
@@ -358,6 +379,9 @@ public void Handle(PeerSubscriptionsUpdated message)
         peer.Value.TimestampUtc = message.PeerDescriptor.TimestampUtc ?? DateTime.UtcNow;
 
         PeerUpdated?.Invoke(message.PeerDescriptor.PeerId, PeerUpdateAction.Updated);
+#if NET10_0_OR_GREATER
+        DirectoryMetrics.PeerUpdates.Add(1);
+#endif
 
         var observedSubscriptions = GetObservedSubscriptions(subscriptions);
         if (observedSubscriptions.Count > 0)
@@ -397,6 +421,9 @@ public void Handle(PeerSubscriptionsForTypesUpdated message)
         peer.Value.SetSubscriptionsForType(subscriptionsForTypes, message.TimestampUtc);
 
         PeerUpdated?.Invoke(message.PeerId, PeerUpdateAction.Updated);
+#if NET10_0_OR_GREATER
+        DirectoryMetrics.PeerUpdates.Add(1);
+#endif
 
         var observedSubscriptions = GetObservedSubscriptions(subscriptionsForTypes);
         if (observedSubscriptions.Count > 0)
@@ -446,6 +473,9 @@ private void HandlePeerRespondingChange(PeerId peerId, bool isResponding)
         peer.Peer.IsResponding = isResponding;
 
         PeerUpdated?.Invoke(peerId, PeerUpdateAction.Updated);
+#if NET10_0_OR_GREATER
+        DirectoryMetrics.PeerUpdates.Add(1);
+#endif
     }
 
     private PeerEntryResult GetPeerCheckTimestamp(PeerId peerId, DateTime? timestampUtc)

src/Abc.Zebus/DomainException.cs

@@ -48,10 +48,12 @@ public DomainException(Expression<Func<int>> errorCodeExpression, params object[
     {
     }
 
+#if !NETCOREAPP
     protected DomainException(SerializationInfo info, StreamingContext context)
         : base(info, context)
     {
     }
+#endif
 
     private static string ReadDescriptionFromAttribute(Expression<Func<int>> errorCodeExpression)
     {

src/Abc.Zebus/MessageId.cs

@@ -62,7 +62,9 @@ public static IDisposable PauseIdGenerationAtDate(DateTime utcDatetime)
     private class TimeGuidGenerator
     {
         private static readonly long _gregorianCalendarTimeTicks = new DateTime(1582, 10, 15, 0, 0, 0, DateTimeKind.Utc).Ticks;
+#if !NETCOREAPP
         private static readonly RNGCryptoServiceProvider _cryptoServiceProvider = new();
+#endif
 
         private readonly uint _nodeIdPart1;
         private readonly ushort _nodeIdPart2;
@@ -139,15 +141,23 @@ public unsafe Guid NewGuid(long absoluteTimestamp)
         private static byte[] GetRandomNodeId()
         {
             var nodeId = new byte[6];
+#if !NETCOREAPP
             _cryptoServiceProvider.GetBytes(nodeId);
+#else
+            RandomNumberGenerator.Fill(nodeId);
+#endif
 
             return nodeId;
         }
 
         private static ushort GetRandomClockId()
         {
             var clockId = new byte[2];
+#if !NETCOREAPP
             _cryptoServiceProvider.GetBytes(clockId);
+#else
+            RandomNumberGenerator.Fill(clockId);
+#endif
             return BitConverter.ToUInt16(clockId, 0);
         }
 

src/Abc.Zebus/MessageProcessingException.cs

@@ -32,8 +32,10 @@ public MessageProcessingException(string message, Exception? inner)
     {
     }
 
+#if !NETCOREAPP
     protected MessageProcessingException(SerializationInfo info, StreamingContext context)
         : base(info, context)
     {
     }
+#endif
 }

src/Abc.Zebus/Monitoring/DirectoryMetrics.cs

@@ -0,0 +1,23 @@
+#if NET10_0_OR_GREATER
+using System.Diagnostics.Metrics;
+#endif
+
+namespace Abc.Zebus.Monitoring;
+
+/// <summary>
+/// Provides <see cref="System.Diagnostics.Metrics"/> instruments for monitoring Zebus peer directory.
+/// </summary>
+internal static class DirectoryMetrics
+{
+#if NET10_0_OR_GREATER
+    internal static readonly UpDownCounter<int> KnownPeerCount = ZebusMetrics.Meter.CreateUpDownCounter<int>(
+        "zebus.directory.known_peers",
+        unit: "{peer}",
+        description: "Current number of known peers in the directory");
+
+    internal static readonly Counter<long> PeerUpdates = ZebusMetrics.Meter.CreateCounter<long>(
+        "zebus.directory.peer_updates",
+        unit: "{update}",
+        description: "Number of peer update events received");
+#endif
+}

src/Abc.Zebus/Monitoring/TransportMetrics.cs

@@ -0,0 +1,62 @@
+#if NET10_0_OR_GREATER
+using System.Diagnostics.Metrics;
+#endif
+
+namespace Abc.Zebus.Monitoring;
+
+/// <summary>
+/// Provides <see cref="System.Diagnostics.Metrics"/> instruments for monitoring Zebus transport layer.
+/// All per-connection metrics include a <c>zebus.peer.id</c> tag for filtering by individual peer.
+/// </summary>
+internal static class TransportMetrics
+{
+#if NET10_0_OR_GREATER
+    // Per-connection message counters
+    internal static readonly Counter<long> MessagesSent = ZebusMetrics.Meter.CreateCounter<long>(
+        "zebus.transport.messages.sent",
+        unit: "{message}",
+        description: "Number of transport messages sent to peers");
+
+    internal static readonly Counter<long> MessagesReceived = ZebusMetrics.Meter.CreateCounter<long>(
+        "zebus.transport.messages.received",
+        unit: "{message}",
+        description: "Number of transport messages received");
+
+    internal static readonly Counter<long> BytesSent = ZebusMetrics.Meter.CreateCounter<long>(
+        "zebus.transport.bytes.sent",
+        unit: "By",
+        description: "Number of bytes sent to peers");
+
+    internal static readonly Counter<long> BytesReceived = ZebusMetrics.Meter.CreateCounter<long>(
+        "zebus.transport.bytes.received",
+        unit: "By",
+        description: "Number of bytes received from peers");
+
+    internal static readonly Counter<long> MessageSendFailures = ZebusMetrics.Meter.CreateCounter<long>(
+        "zebus.transport.messages.send_failures",
+        unit: "{message}",
+        description: "Number of transport message send failures");
+
+    // Per-connection state
+    internal static readonly Counter<long> PeerConnections = ZebusMetrics.Meter.CreateCounter<long>(
+        "zebus.transport.peer_connections",
+        unit: "{connection}",
+        description: "Number of outbound peer socket connections established");
+
+    internal static readonly Counter<long> PeerDisconnections = ZebusMetrics.Meter.CreateCounter<long>(
+        "zebus.transport.peer_disconnections",
+        unit: "{disconnection}",
+        description: "Number of outbound peer socket disconnections");
+
+    internal static readonly Counter<long> PeerConnectionFailures = ZebusMetrics.Meter.CreateCounter<long>(
+        "zebus.transport.peer_connection_failures",
+        unit: "{failure}",
+        description: "Number of outbound peer socket connection failures");
+
+    // Aggregate outbound socket count
+    internal static readonly UpDownCounter<int> OutboundSocketCount = ZebusMetrics.Meter.CreateUpDownCounter<int>(
+        "zebus.transport.outbound_sockets",
+        unit: "{socket}",
+        description: "Current number of active outbound sockets");
+#endif
+}

src/Abc.Zebus/Monitoring/ZebusMetrics.cs

@@ -0,0 +1,22 @@
+#if NET10_0_OR_GREATER
+using System.Collections.Generic;
+using System.Diagnostics.Metrics;
+#endif
+
+namespace Abc.Zebus.Monitoring;
+
+/// <summary>
+/// Shared <see cref="System.Diagnostics.Metrics.Meter"/> and tag helpers for Zebus metrics.
+/// Service-specific instruments are defined in <see cref="TransportMetrics"/> and <see cref="DirectoryMetrics"/>.
+/// </summary>
+internal static class ZebusMetrics
+{
+#if NET10_0_OR_GREATER
+    internal static readonly Meter Meter = new("Abc.Zebus", typeof(ZebusMetrics).Assembly.GetName().Version?.ToString());
+
+    private const string PeerIdTag = "zebus.peer.id";
+
+    internal static KeyValuePair<string, object?> PeerTag(PeerId peerId)
+        => new(PeerIdTag, peerId.ToString());
+#endif
+}

src/Abc.Zebus/Routing/BindingKey.cs

@@ -114,7 +114,8 @@ internal static BindingKey Create(Type messageType, IDictionary<string, string>
         var parts = new string[routingMembers.Length];
         for (var tokenIndex = 0; tokenIndex < routingMembers.Length; ++tokenIndex)
         {
-            parts[tokenIndex] = fieldValues.GetValueOrDefault(routingMembers[tokenIndex].Member.Name, BindingKeyPart.StarToken);
+            var memberName = routingMembers[tokenIndex].Member.Name;
+            parts[tokenIndex] = fieldValues.TryGetValue(memberName, out var value) ? value : BindingKeyPart.StarToken;
         }
 
         return new BindingKey(parts);

src/Abc.Zebus/Serialization/ProtoBufConvert.cs

@@ -3,7 +3,11 @@
 using System.Diagnostics.CodeAnalysis;
 using System.IO;
 using System.Reflection;
+#if !NETCOREAPP
 using System.Runtime.Serialization;
+#else
+using System.Runtime.CompilerServices;
+#endif
 using ProtoBuf.Meta;
 
 namespace Abc.Zebus.Serialization;
@@ -49,7 +53,11 @@ public static object Deserialize(Type messageType, Stream stream)
     private static object? CreateMessageIfRequired(Type messageType)
     {
         if (!HasParameterLessConstructor(messageType) && messageType != typeof(string))
+#if !NETCOREAPP
             return FormatterServices.GetUninitializedObject(messageType);
+#else
+            return RuntimeHelpers.GetUninitializedObject(messageType);
+#endif
 
         return null;
     }

src/Abc.Zebus/Serialization/Protobuf/ProtoBufferReader.cs

@@ -104,7 +104,10 @@ public bool TrySkipString()
     public bool TryReadGuid(out Guid value)
     {
         if (!TryReadLength(out var length) || !CanRead(length) || length != ProtoBufferWriter.GuidSize)
+        {
+            value = default;
             return false;
+        }
 
         // Skip tag
         _buffer.AsSpan(_position + 1, 8).CopyTo(_guidBuffer.AsSpan(0));

src/Abc.Zebus/Transport/ZmqOutboundSocket.cs

@@ -1,5 +1,6 @@
 using System;
 using System.Diagnostics;
+using Abc.Zebus.Monitoring;
 using Abc.Zebus.Transport.Zmq;
 using Microsoft.Extensions.Logging;
 
@@ -42,6 +43,9 @@ public void ConnectFor(TransportMessage message)
             _socket.Connect(EndPoint);
 
             IsConnected = true;
+#if NET10_0_OR_GREATER
+            TransportMetrics.PeerConnections.Add(1, ZebusMetrics.PeerTag(PeerId));
+#endif
 
             _logger.LogInformation($"Socket connected, Peer: {PeerId}, EndPoint: {EndPoint}");
         }
@@ -53,6 +57,9 @@ public void ConnectFor(TransportMessage message)
 
             _logger.LogError(ex, $"Unable to connect socket, Peer: {PeerId}, EndPoint: {EndPoint}");
             _errorHandler.OnConnectException(PeerId, EndPoint, ex);
+#if NET10_0_OR_GREATER
+            TransportMetrics.PeerConnectionFailures.Add(1, ZebusMetrics.PeerTag(PeerId));
+#endif
 
             SwitchToClosedState(_options.ClosedStateDurationAfterConnectFailure);
         }
@@ -95,6 +102,9 @@ public void Disconnect()
         {
             _socket!.SetOption(ZmqSocketOption.LINGER, 0);
             _socket!.Dispose();
+#if NET10_0_OR_GREATER
+            TransportMetrics.PeerDisconnections.Add(1, ZebusMetrics.PeerTag(PeerId));
+#endif
 
             _logger.LogInformation($"Socket disconnected, Peer: {PeerId}");
         }
@@ -123,6 +133,9 @@ public void Send(byte[] buffer, int length, TransportMessage message)
 
         _logger.LogError($"Unable to send message, destination peer: {PeerId}, MessageTypeId: {message.MessageTypeId}, MessageId: {message.Id}, Error: {errorMessage}");
         _errorHandler.OnSendFailed(PeerId, EndPoint, message.MessageTypeId, message.Id);
+#if NET10_0_OR_GREATER
+        TransportMetrics.MessageSendFailures.Add(1, ZebusMetrics.PeerTag(PeerId));
+#endif
 
         if (_failedSendCount >= _options.SendRetriesBeforeSwitchingToClosedState)
             SwitchToClosedState(_options.ClosedStateDurationAfterSendFailure);