Add Initial Metrics

Directory.Packages.props

@@ -12,6 +12,7 @@
     <PackageVersion Include="BenchmarkDotNet" Version="0.13.1" />
     <PackageVersion Include="GitHubActionsTestLogger" Version="2.0.0-alpha" />
     <PackageVersion Include="Microsoft.CodeAnalysis.PublicApiAnalyzers" Version="3.3.4-beta1.22362.3" />
+    <PackageVersion Include="Microsoft.Extensions.Telemetry.Testing" Version="8.0.0-preview.6.23360.2" />
     <PackageVersion Include="Microsoft.NETFramework.ReferenceAssemblies" Version="1.0.2" />
     <PackageVersion Include="Microsoft.NET.Test.Sdk" Version="17.1.0" />
     <PackageVersion Include="Microsoft.SourceLink.GitHub" Version="1.1.1" />

src/StackExchange.Redis/ConfigurationOptions.cs

@@ -12,6 +12,10 @@
 using System.Threading.Tasks;
 using StackExchange.Redis.Configuration;
 
+#if NET6_0_OR_GREATER
+using System.Diagnostics.Metrics;
+#endif
+
 namespace StackExchange.Redis
 {
     /// <summary>
@@ -619,6 +623,14 @@ public int ConfigCheckSeconds
             set => configCheckSeconds = value;
         }
 
+#if NET6_0_OR_GREATER
+        /// <summary>
+        /// A factory method able to inject the <see cref="Meter"/> object to use
+        /// when emitting metrics. Used by tests.
+        /// </summary>
+        internal Func<Meter>? MeterFactory { get; set; }
+#endif
+
         /// <summary>
         /// Parse the configuration from a comma-delimited configuration string.
         /// </summary>

src/StackExchange.Redis/ConnectionMultiplexer.Metrics.cs

@@ -0,0 +1,19 @@
+#if NET6_0_OR_GREATER
+
+namespace StackExchange.Redis;
+
+public partial class ConnectionMultiplexer
+{
+    internal RedisMetrics Metrics { get; }
+
+    private static RedisMetrics GetMetrics(ConfigurationOptions configuration)
+    {
+        if (configuration.MeterFactory is not null)
+        {
+            return new RedisMetrics(configuration.MeterFactory());
+        }
+
+        return RedisMetrics.Default;
+    }
+}
+#endif

src/StackExchange.Redis/ConnectionMultiplexer.cs

@@ -128,6 +128,9 @@ static ConnectionMultiplexer()
         private ConnectionMultiplexer(ConfigurationOptions configuration, ServerType? serverType = null, EndPointCollection? endpoints = null)
         {
             RawConfig = configuration ?? throw new ArgumentNullException(nameof(configuration));
+#if NET6_0_OR_GREATER
+            Metrics = GetMetrics(configuration);
+#endif
             EndPoints = endpoints ?? RawConfig.EndPoints.Clone();
             EndPoints.SetDefaultPorts(serverType, ssl: RawConfig.Ssl);
 
@@ -1939,6 +1942,9 @@ internal void UpdateClusterRange(ClusterConfiguration configuration)
         private bool PrepareToPushMessageToBridge<T>(Message message, ResultProcessor<T>? processor, IResultBox<T>? resultBox, [NotNullWhen(true)] ref ServerEndPoint? server)
         {
             message.SetSource(processor, resultBox);
+#if NET6_0_OR_GREATER
+            message.SetMetrics(Metrics);
+#endif
 
             if (server == null)
             {

src/StackExchange.Redis/Message.cs

@@ -84,6 +84,10 @@ internal abstract class Message : ICompletable
         internal DateTime CreatedDateTime;
         internal long CreatedTimestamp;
 
+#if NET6_0_OR_GREATER
+        private RedisMetrics? metrics;
+#endif
+
         protected Message(int db, CommandFlags flags, RedisCommand command)
         {
             bool dbNeeded = RequiresDatabase(command);
@@ -134,6 +138,13 @@ internal void SetPrimaryOnly()
             }
         }
 
+#if NET6_0_OR_GREATER
+        internal void SetMetrics(RedisMetrics redisMetrics)
+        {
+            metrics = redisMetrics;
+        }
+#endif
+
         internal void SetProfileStorage(ProfiledCommand storage)
         {
             performance = storage;
@@ -374,6 +385,9 @@ public void Complete()
 
             // set the completion/performance data
             performance?.SetCompleted();
+#if NET6_0_OR_GREATER
+            metrics?.OnMessageComplete(this, currBox);
+#endif
 
             currBox?.ActivateContinuations();
         }
@@ -441,7 +455,7 @@ internal static Message Create(int db, CommandFlags flags, RedisCommand command,
                 3 => new CommandKeyKeyValueValueValueMessage(db, flags, command, key0, key1, values[0], values[1], values[2]),
                 4 => new CommandKeyKeyValueValueValueValueMessage(db, flags, command, key0, key1, values[0], values[1], values[2], values[3]),
                 5 => new CommandKeyKeyValueValueValueValueValueMessage(db, flags, command, key0, key1, values[0], values[1], values[2], values[3], values[4]),
-                6 => new CommandKeyKeyValueValueValueValueValueValueMessage(db, flags, command, key0, key1, values[0], values[1], values[2], values[3],values[4],values[5]),
+                6 => new CommandKeyKeyValueValueValueValueValueValueMessage(db, flags, command, key0, key1, values[0], values[1], values[2], values[3], values[4], values[5]),
                 7 => new CommandKeyKeyValueValueValueValueValueValueValueMessage(db, flags, command, key0, key1, values[0], values[1], values[2], values[3], values[4], values[5], values[6]),
                 _ => new CommandKeyKeyValuesMessage(db, flags, command, key0, key1, values),
             };

src/StackExchange.Redis/PhysicalBridge.cs

@@ -350,6 +350,9 @@ internal string GetStormLog()
         internal void IncrementOpCount()
         {
             Interlocked.Increment(ref operationCount);
+#if NET6_0_OR_GREATER
+            Multiplexer.Metrics.IncrementOperationCount(Name);
+#endif
         }
 
         internal void KeepAlive()
@@ -1404,12 +1407,22 @@ private void LogNonPreferred(CommandFlags flags, bool isReplica)
                 if (isReplica)
                 {
                     if (Message.GetPrimaryReplicaFlags(flags) == CommandFlags.PreferMaster)
+                    {
                         Interlocked.Increment(ref nonPreferredEndpointCount);
+#if NET6_0_OR_GREATER
+                        Multiplexer.Metrics.IncrementNonPreferredEndpointCount(Name);
+#endif
+                    }
                 }
                 else
                 {
                     if (Message.GetPrimaryReplicaFlags(flags) == CommandFlags.PreferReplica)
+                    {
                         Interlocked.Increment(ref nonPreferredEndpointCount);
+#if NET6_0_OR_GREATER
+                        Multiplexer.Metrics.IncrementNonPreferredEndpointCount(Name);
+#endif
+                    }
                 }
             }
         }

src/StackExchange.Redis/RedisMetrics.cs

@@ -0,0 +1,79 @@
+#if NET6_0_OR_GREATER
+
+using System;
+using System.Collections.Generic;
+using System.Diagnostics;
+using System.Diagnostics.Metrics;
+
+namespace StackExchange.Redis;
+
+internal sealed class RedisMetrics
+{
+    private static readonly double s_tickFrequency = (double)TimeSpan.TicksPerSecond / Stopwatch.Frequency;
+    // cache these boxed boolean values so we don't allocate on each usage.
+    private static readonly object s_trueBox = true;
+    private static readonly object s_falseBox = false;
+
+    private readonly Meter _meter;
+    private readonly Counter<long> _operationCount;
+    private readonly Histogram<double> _messageDuration;
+    private readonly Counter<long> _nonPreferredEndpointCount;
+
+    public static readonly RedisMetrics Default = new RedisMetrics();
+
+    public RedisMetrics(Meter? meter = null)
+    {
+        _meter = meter ?? new Meter("StackExchange.Redis");
+
+        _operationCount = _meter.CreateCounter<long>(
+            "db.redis.operation.count",
+            description: "The number of operations performed.");
+
+        _messageDuration = _meter.CreateHistogram<double>(
+            "db.redis.duration",
+            unit: "s",
+            description: "Measures the duration of outbound message requests.");
+
+        _nonPreferredEndpointCount = _meter.CreateCounter<long>(
+            "db.redis.non_preferred_endpoint.count",
+            description: "Indicates the total number of messages dispatched to a non-preferred endpoint, for example sent to a primary when the caller stated a preference of replica.");
+    }
+
+    public void IncrementOperationCount(string endpoint)
+    {
+        _operationCount.Add(1,
+            new KeyValuePair<string, object?>("endpoint", endpoint));
+    }
+
+    public void OnMessageComplete(Message message, IResultBox? result)
+    {
+        // The caller ensures we can don't record on the same resultBox from two threads.
+        // 'result' can be null if this method is called for the same message more than once.
+        if (result is not null && _messageDuration.Enabled)
+        {
+            // Stopwatch.GetElapsedTime is only available in net7.0+
+            // https://github.com/dotnet/runtime/blob/ae068fec6ede58d2a5b343c5ac41c9ca8715fa47/src/libraries/System.Private.CoreLib/src/System/Diagnostics/Stopwatch.cs#L129-L137
+            var now = Stopwatch.GetTimestamp();
+            var duration = new TimeSpan((long)((now - message.CreatedTimestamp) * s_tickFrequency));
+
+            var tags = new TagList
+            {
+                { "db.redis.async", result.IsAsync ? s_trueBox : s_falseBox },
+                { "db.redis.faulted", result.IsFaulted ? s_trueBox : s_falseBox }
+                // TODO: can we pass endpoint here?
+                // should we log the Db?
+                // { "db.redis.database_index", message.Db },
+            };
+
+            _messageDuration.Record(duration.TotalSeconds, tags);
+        }
+    }
+
+    public void IncrementNonPreferredEndpointCount(string endpoint)
+    {
+        _nonPreferredEndpointCount.Add(1,
+            new KeyValuePair<string, object?>("endpoint", endpoint));
+    }
+}
+
+#endif

tests/StackExchange.Redis.Tests/MetricsTests.cs

@@ -0,0 +1,44 @@
+#if NET6_0_OR_GREATER
+#pragma warning disable TBD // MetricCollector is for evaluation purposes only and is subject to change or removal in future updates.
+
+using Microsoft.Extensions.Telemetry.Testing.Metering;
+using System.Diagnostics.Metrics;
+using System.Threading.Tasks;
+using Xunit;
+using Xunit.Abstractions;
+
+namespace StackExchange.Redis.Tests;
+
+public class MetricsTests : TestBase
+{
+    public MetricsTests(ITestOutputHelper output) : base(output) { }
+
+    [Fact]
+    public async Task SimpleCommandDuration()
+    {
+        var options = ConfigurationOptions.Parse(GetConfiguration());
+
+        using var meter = new Meter("StackExchange.Redis.Tests");
+        using var collector = new MetricCollector<double>(meter, "db.redis.duration");
+
+        options.MeterFactory = () => meter;
+
+        using var conn = await ConnectionMultiplexer.ConnectAsync(options, Writer);
+        var db = conn.GetDatabase();
+
+        RedisKey key = Me();
+        string? g1 = await db.StringGetAsync(key);
+        Assert.Null(g1);
+
+        await collector.WaitForMeasurementsAsync(1);
+
+        Assert.Collection(collector.GetMeasurementSnapshot(),
+            measurement =>
+            {
+                // Built-in
+                Assert.Equal(true, measurement.Tags["db.redis.async"]);
+                Assert.Equal(false, measurement.Tags["db.redis.faulted"]);
+            });
+    }
+}
+#endif

tests/StackExchange.Redis.Tests/StackExchange.Redis.Tests.csproj

@@ -33,4 +33,8 @@
     <Reference Include="Microsoft.CSharp" />
     <PackageReference Include="System.Runtime.CompilerServices.Unsafe" Version="5.0.0" />
   </ItemGroup>
+
+  <ItemGroup Condition="'$(TargetFrameworkIdentifier)' == '.NETCoreApp'">
+    <PackageReference Include="Microsoft.Extensions.Telemetry.Testing" />
+  </ItemGroup>
 </Project>