Merge branch 'main' into 1997-multiple-connection-support

src/OpenTelemetry.Extensions/.publicApi/PublicAPI.Unshipped.txt

@@ -8,8 +8,11 @@ OpenTelemetry.Logs.LogToActivityEventConversionOptions.ScopeConverter.get -> Sys
 OpenTelemetry.Logs.LogToActivityEventConversionOptions.ScopeConverter.set -> void
 OpenTelemetry.Logs.LogToActivityEventConversionOptions.StateConverter.get -> System.Action<System.Diagnostics.ActivityTagsCollection!, System.Collections.Generic.IReadOnlyList<System.Collections.Generic.KeyValuePair<string!, object?>>!>!
 OpenTelemetry.Logs.LogToActivityEventConversionOptions.StateConverter.set -> void
+OpenTelemetry.RateLimitingSampler
+OpenTelemetry.RateLimitingSampler.RateLimitingSampler(int maxTracesPerSecond) -> void
 OpenTelemetry.Trace.BaggageActivityProcessor
 OpenTelemetry.Trace.TracerProviderBuilderExtensions
+override OpenTelemetry.RateLimitingSampler.ShouldSample(in OpenTelemetry.Trace.SamplingParameters samplingParameters) -> OpenTelemetry.Trace.SamplingResult
 override OpenTelemetry.Trace.BaggageActivityProcessor.OnStart(System.Diagnostics.Activity! data) -> void
 static Microsoft.Extensions.Logging.OpenTelemetryLoggingExtensions.AttachLogsToActivityEvent(this OpenTelemetry.Logs.OpenTelemetryLoggerOptions! loggerOptions, System.Action<OpenTelemetry.Logs.LogToActivityEventConversionOptions!>? configure = null) -> OpenTelemetry.Logs.OpenTelemetryLoggerOptions!
 static OpenTelemetry.Trace.BaggageActivityProcessor.AllowAllBaggageKeys.get -> System.Predicate<string!>!

src/OpenTelemetry.Extensions/CHANGELOG.md

@@ -8,6 +8,11 @@
 * Updated OpenTelemetry core component version(s) to `1.9.0`.
   ([#1888](https://github.com/open-telemetry/opentelemetry-dotnet-contrib/pull/1888))
 
+* Added rate limiting sampler which limits the number of traces to the specified
+rate per second. For details see
+  [RateLimitingSampler](https://github.com/open-telemetry/opentelemetry-dotnet-contrib/tree/main/src/OpenTelemetry.Extensions#ratelimitingsampler).
+  ([#1996](https://github.com/open-telemetry/opentelemetry-dotnet-contrib/pull/1996))
+
 ## 1.0.0-beta.5
 
 Released 2024-May-08

src/OpenTelemetry.Extensions/Internal/RateLimiter.cs

@@ -0,0 +1,48 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+using System.Diagnostics;
+
+namespace OpenTelemetry.Extensions.Internal;
+
+internal sealed class RateLimiter
+{
+    private readonly Stopwatch stopwatch = Stopwatch.StartNew();
+    private readonly double creditsPerTick;
+    private readonly long maxBalance; // max balance in ticks
+    private long currentBalance; // last op ticks less remaining balance, using long directly with Interlocked for thread safety
+
+    public RateLimiter(double creditsPerSecond, double maxBalance)
+    {
+        this.creditsPerTick = creditsPerSecond / Stopwatch.Frequency;
+        this.maxBalance = (long)(maxBalance / this.creditsPerTick);
+        this.currentBalance = this.stopwatch.ElapsedTicks - this.maxBalance;
+    }
+
+    public bool TrySpend(double itemCost)
+    {
+        long cost = (long)(itemCost / this.creditsPerTick);
+        long currentTicks;
+        long currentBalanceTicks;
+        long availableBalanceAfterWithdrawal;
+        do
+        {
+            currentBalanceTicks = Interlocked.Read(ref this.currentBalance);
+            currentTicks = this.stopwatch.ElapsedTicks;
+            long currentAvailableBalance = currentTicks - currentBalanceTicks;
+            if (currentAvailableBalance > this.maxBalance)
+            {
+                currentAvailableBalance = this.maxBalance;
+            }
+
+            availableBalanceAfterWithdrawal = currentAvailableBalance - cost;
+            if (availableBalanceAfterWithdrawal < 0)
+            {
+                return false;
+            }
+        }
+
+        // CompareExchange will fail if currentBalance has changed since the last read, implying another thread has updated the balance
+        while (Interlocked.CompareExchange(ref this.currentBalance, currentTicks - availableBalanceAfterWithdrawal, currentBalanceTicks) != currentBalanceTicks);
+        return true;
+    }
+}

src/OpenTelemetry.Extensions/README.md

@@ -84,3 +84,27 @@ var tracerProvider = Sdk.CreateTracerProviderBuilder()
 Warning: The baggage key predicate is executed for every baggage entry for each
 started activity.
 Do not use slow or intensive operations.
+
+### RateLimitingSampler
+
+The rate limiting sampler is a sampler that will limit the number of samples to
+the specified rate per second. It is typically used in conjunction with the ParentBasedSampler
+to ensure that the rate limiting sampler is only applied to the root spans. When
+using the ParentBasedSampler, when an Activity creation request comes in without
+a sampling decision, it will delegate to the rate limiting sampler which will
+make a decision based on the rate limit, that way all spans in the trace will use
+the same sampling decision, and the rate will effectively become the number of
+traces per second, irrespective of the number of spans within each trace.
+
+Example of RateLimitingSampler usage:
+
+```cs
+builder.Services.AddOpenTelemetry()
+    .WithTracing(tracing =>
+    {
+        tracing.AddAspNetCoreInstrumentation()
+            .AddHttpClientInstrumentation()
+            // Add the rate limiting sampler with a limit of 3 traces per second
+            .SetSampler(new ParentBasedSampler(new RateLimitingSampler(3)))
+    });
+```

src/OpenTelemetry.Extensions/Trace/RateLimitingSampler.cs

@@ -0,0 +1,68 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+// Based on the jaeger remote sampler for Java from https://github.com/open-telemetry/opentelemetry-java/blob/main/sdk-extensions/jaeger-remote-sampler/src/main/java/io/opentelemetry/sdk/extension/trace/jaeger/sampler/RateLimitingSampler.java
+
+using System.Globalization;
+using OpenTelemetry.Extensions.Internal;
+using OpenTelemetry.Trace;
+
+namespace OpenTelemetry;
+
+/// <summary>
+/// Rate limiting sampler that can be used to sample traces at a constant rate.
+/// </summary>
+public class RateLimitingSampler : Sampler
+{
+    private const string SAMPLERTYPE = "ratelimiting";
+    private const string SAMPLERTYPEKEY = "sampler.type";
+    private const string SAMPLERPARAMKEY = "sampler.param";
+
+    private readonly RateLimiter rateLimiter;
+    private readonly SamplingResult onSamplingResult;
+    private readonly SamplingResult offSamplingResult;
+
+    /// <summary>
+    /// Initializes a new instance of the <see cref="RateLimitingSampler"/> class.
+    /// </summary>
+    /// <param name="maxTracesPerSecond">The maximum number of traces that will be emitted each second.</param>
+    public RateLimitingSampler(int maxTracesPerSecond)
+    {
+        double maxBalance = maxTracesPerSecond < 1.0 ? 1.0 : maxTracesPerSecond;
+        this.rateLimiter = new RateLimiter(maxTracesPerSecond, maxBalance);
+        var attributes = new Dictionary<string, object>()
+        {
+            { SAMPLERTYPEKEY, SAMPLERTYPE },
+            { SAMPLERPARAMKEY, (double)maxTracesPerSecond },
+        };
+        this.onSamplingResult = new SamplingResult(SamplingDecision.RecordAndSample, attributes);
+        this.offSamplingResult = new SamplingResult(SamplingDecision.Drop, attributes);
+        this.Description = $"RateLimitingSampler{{{DecimalFormat(maxTracesPerSecond)}}}";
+    }
+
+    /// <summary>
+    /// Checks whether activity needs to be created and tracked.
+    /// </summary>
+    /// <param name="samplingParameters">
+    /// The OpenTelemetry.Trace.SamplingParameters used by the OpenTelemetry.Trace.Sampler
+    /// to decide if the System.Diagnostics.Activity to be created is going to be sampled
+    /// or not.
+    /// </param>
+    /// <returns>
+    /// Sampling decision on whether activity needs to be sampled or not.
+    /// </returns>
+    public override SamplingResult ShouldSample(in SamplingParameters samplingParameters)
+    {
+        return this.rateLimiter.TrySpend(1.0) ? this.onSamplingResult : this.offSamplingResult;
+    }
+
+    private static string DecimalFormat(double value)
+    {
+        NumberFormatInfo numberFormatInfo = new NumberFormatInfo
+        {
+            NumberDecimalSeparator = ".",
+        };
+
+        return value.ToString("0.00", numberFormatInfo);
+    }
+}

test/OpenTelemetry.Extensions.Tests/Trace/RateLimitingSamplerTests.cs

@@ -0,0 +1,102 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+using OpenTelemetry.Trace;
+using Xunit;
+
+namespace OpenTelemetry.Extensions.Tests.Trace;
+
+public class RateLimitingSamplerTests
+{
+    [Fact]
+    public void ShouldSample_ReturnsRecordAndSample_WhenWithinRateLimit()
+    {
+        // Arrange
+        var samplingParameters = new SamplingParameters(
+            parentContext: default,
+            traceId: default,
+            name: "TestOperation",
+            kind: default,
+            tags: null,
+            links: null);
+
+        var sampler = new RateLimitingSampler(5); // 5 trace per second
+        int sampleIn = 0, sampleOut = 0;
+
+        // Fire in 3 traces with a second, should all be sampled in
+
+        for (var i = 0; i < 3; i++)
+        {
+            // Act
+            var result = sampler.ShouldSample(in samplingParameters);
+            switch (result.Decision)
+            {
+                case SamplingDecision.RecordAndSample:
+                    sampleIn++;
+                    break;
+                case SamplingDecision.RecordOnly:
+                    Assert.Fail("Unexpected decision");
+                    break;
+                case SamplingDecision.Drop:
+                    sampleOut++;
+                    break;
+            }
+
+            Thread.Sleep(333);
+        }
+
+        // Assert
+        Assert.Equal(3, sampleIn);
+        Assert.Equal(0, sampleOut);
+    }
+
+    [Fact]
+    public async Task ShouldFilter_WhenAboveRateLimit()
+    {
+        const int SAMPLE_RATE = 5; // 5 trace per second
+        const int CYCLES = 500;
+
+        var samplingParameters = new SamplingParameters(
+            parentContext: default,
+            traceId: default,
+            name: "TestOperation",
+            kind: default,
+            tags: null,
+            links: null);
+        var sampler = new RateLimitingSampler(SAMPLE_RATE);
+        int sampleIn = 0, sampleOut = 0;
+
+        var startTime = DateTime.UtcNow;
+
+        for (var i = 0; i < CYCLES; i++)
+        {
+            var result = sampler.ShouldSample(in samplingParameters);
+            switch (result.Decision)
+            {
+                case SamplingDecision.RecordAndSample:
+                    sampleIn++;
+                    break;
+                case SamplingDecision.RecordOnly:
+                    Assert.Fail("Unexpected decision");
+                    break;
+                case SamplingDecision.Drop:
+                    sampleOut++;
+                    break;
+            }
+
+            // Task.Delay is limited by the OS Scheduler, so we can't guarantee the exact time
+            await Task.Delay(5);
+        }
+
+        var timeTakenSeconds = (DateTime.UtcNow - startTime).TotalSeconds;
+
+        // Approximate the number of samples we should have taken
+        // Account for the fact that the initial balance is the SampleRate, so they will all be sampled in
+        var approxSamples = Math.Floor(timeTakenSeconds * SAMPLE_RATE) + SAMPLE_RATE;
+
+        // Assert - We should have sampled in 5 traces per second over duration
+        // Adding in a fudge factor
+        Assert.True(sampleIn > (approxSamples * 0.9) && sampleIn < (approxSamples * 1.1));
+        Assert.True(sampleOut == (CYCLES - sampleIn));
+    }
+}