diff --git a/bench/Libraries/Microsoft.Extensions.Telemetry.PerformanceTests/BufferingImpactBench.cs b/bench/Libraries/Microsoft.Extensions.Telemetry.PerformanceTests/BufferingImpactBench.cs new file mode 100644 index 00000000000..c4a3f92c979 --- /dev/null +++ b/bench/Libraries/Microsoft.Extensions.Telemetry.PerformanceTests/BufferingImpactBench.cs @@ -0,0 +1,88 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. + +using System; +using BenchmarkDotNet.Attributes; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Diagnostics.Buffering; +using Microsoft.Extensions.Logging; + +namespace Microsoft.Extensions.Telemetry.Bench; + +[MemoryDiagnoser] +[InvocationCount(1)] +public class BufferingImpactBench +{ + private ServiceProvider _baselineServices = null!; + private ServiceProvider _bufferedServices = null!; + private ILogger[] _baselineLoggers = null!; + private ILogger[] _bufferedLoggers = null!; + private GlobalLogBuffer _buffer = null!; + + [Params(10_000, 20_000)] + public int RecordsPerMinute { get; set; } + + [GlobalSetup] + public void GlobalSetup() + { + _baselineServices = CreateServices(bufferingEnabled: false); + _bufferedServices = CreateServices(bufferingEnabled: true); + _baselineLoggers = LoggingBenchmarkWorkload.CreateLoggers(_baselineServices.GetRequiredService()); + _bufferedLoggers = LoggingBenchmarkWorkload.CreateLoggers(_bufferedServices.GetRequiredService()); + _buffer = _bufferedServices.GetRequiredService(); + } + + [GlobalCleanup] + public void GlobalCleanup() + { + _bufferedServices.Dispose(); + _baselineServices.Dispose(); + } + + [IterationCleanup] + public void FlushBuffer() + { + _buffer.Flush(); + } + + [Benchmark(Baseline = true)] + public void NoBuffering() + { + LoggingBenchmarkWorkload.LogBatch(_baselineLoggers, RecordsPerMinute); + } + + [Benchmark] + public void BufferOnly() + { + LoggingBenchmarkWorkload.LogBatch(_bufferedLoggers, RecordsPerMinute); + } + + [Benchmark] + public void BufferAndFlush() + { + LoggingBenchmarkWorkload.LogBatch(_bufferedLoggers, RecordsPerMinute); + _buffer.Flush(); + } + + private static ServiceProvider CreateServices(bool bufferingEnabled) + { + var services = new ServiceCollection(); + + services.AddLogging(builder => + { + builder.AddProvider(new BenchLoggerProvider()); + + if (bufferingEnabled) + { + builder.AddGlobalBuffer(options => + { + options.AutoFlushDuration = TimeSpan.Zero; + options.MaxBufferSizeInBytes = 512 * 1024 * 1024; + options.Rules.Add(new LogBufferingFilterRule(logLevel: LogLevel.Information)); + }); + } + }); + + return services.BuildServiceProvider(); + } +} diff --git a/bench/Libraries/Microsoft.Extensions.Telemetry.PerformanceTests/CategoryCardinalityImpactBench.cs b/bench/Libraries/Microsoft.Extensions.Telemetry.PerformanceTests/CategoryCardinalityImpactBench.cs new file mode 100644 index 00000000000..a10994c0b9b --- /dev/null +++ b/bench/Libraries/Microsoft.Extensions.Telemetry.PerformanceTests/CategoryCardinalityImpactBench.cs @@ -0,0 +1,106 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. + +using System; +using BenchmarkDotNet.Attributes; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Diagnostics.Buffering; +using Microsoft.Extensions.Logging; + +namespace Microsoft.Extensions.Telemetry.Bench; + +[MemoryDiagnoser] +[InvocationCount(1)] +public class CategoryCardinalityImpactBench +{ + private const int RecordsPerMinute = 20_000; + + private ServiceProvider _baselineServices = null!; + private ServiceProvider _strategyServices = null!; + private ILogger[] _baselineLoggers = null!; + private ILogger[] _strategyLoggers = null!; + private LogBuffer? _strategyBuffer; + + public enum SamplingStrategy + { + RandomOnePercent, + CckrOnePercent + } + + [Params(50, 100, 200)] + public int CategoryCount { get; set; } + + [Params(SamplingStrategy.RandomOnePercent, SamplingStrategy.CckrOnePercent)] + public SamplingStrategy Strategy { get; set; } + + [GlobalSetup] + public void GlobalSetup() + { + _baselineServices = CreateServices(); + _strategyServices = CreateServices(Strategy, CategoryCount); + _baselineLoggers = CreateLoggers(_baselineServices); + _strategyLoggers = CreateLoggers(_strategyServices); + _strategyBuffer = _strategyServices.GetService(); + } + + [GlobalCleanup] + public void GlobalCleanup() + { + _strategyServices.Dispose(); + _baselineServices.Dispose(); + } + + [IterationCleanup] + public void FlushBuffer() + { + _strategyBuffer?.Flush(); + } + + [Benchmark(Baseline = true)] + public void NoSampling() + { + LoggingBenchmarkWorkload.LogBatch(_baselineLoggers, RecordsPerMinute); + } + + [Benchmark] + public void WithStrategy() + { + LoggingBenchmarkWorkload.LogBatch(_strategyLoggers, RecordsPerMinute); + _strategyBuffer?.Flush(); + } + + private static ServiceProvider CreateServices( + SamplingStrategy? strategy = null, + int categoryCount = 1) + { + var services = new ServiceCollection(); + + services.AddLogging(builder => + { + builder.AddProvider(new SerializedExporterLoggerProvider()); + + switch (strategy) + { + case SamplingStrategy.RandomOnePercent: + builder.AddRandomProbabilisticSampler(0.01); + break; + + case SamplingStrategy.CckrOnePercent: + builder.AddCckrLogSampling(options => + { + options.Capacity = RecordsPerMinute / 100 / categoryCount; + options.PreserveCapacity = 0; + options.FlushInterval = TimeSpan.FromDays(1); + }); + break; + } + }); + + return services.BuildServiceProvider(); + } + + private ILogger[] CreateLoggers(ServiceProvider services) + => LoggingBenchmarkWorkload.CreateLoggers( + services.GetRequiredService(), + CategoryCount); +} diff --git a/bench/Libraries/Microsoft.Extensions.Telemetry.PerformanceTests/CckrImpactBench.cs b/bench/Libraries/Microsoft.Extensions.Telemetry.PerformanceTests/CckrImpactBench.cs new file mode 100644 index 00000000000..ab700c03767 --- /dev/null +++ b/bench/Libraries/Microsoft.Extensions.Telemetry.PerformanceTests/CckrImpactBench.cs @@ -0,0 +1,148 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. + +using System; +using BenchmarkDotNet.Attributes; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Diagnostics.Buffering; +using Microsoft.Extensions.Diagnostics.Sampling; +using Microsoft.Extensions.Logging; + +namespace Microsoft.Extensions.Telemetry.Bench; + +[MemoryDiagnoser] +[InvocationCount(1)] +public class CckrImpactBench +{ + private const int AdaptiveCapacity = 128; + + private ServiceProvider _baselineServices = null!; + private ServiceProvider _disabledServices = null!; + private ServiceProvider _retainAllPolicyServices = null!; + private ServiceProvider _retainAllServices = null!; + private ServiceProvider _adaptiveServices = null!; + private ILogger[] _baselineLoggers = null!; + private ILogger[] _disabledLoggers = null!; + private ILogger[] _retainAllPolicyLoggers = null!; + private ILogger[] _retainAllLoggers = null!; + private ILogger[] _adaptiveLoggers = null!; + private LogBuffer _retainAllBuffer = null!; + private LogBuffer _adaptiveBuffer = null!; + + [Params(10_000, 20_000)] + public int RecordsPerMinute { get; set; } + + [GlobalSetup] + public void GlobalSetup() + { + _baselineServices = CreateServices(); + _disabledServices = CreateServices(options => options.Enabled = false); + _retainAllPolicyServices = CreateServices(options => options.RetainAllCategories.Add("Benchmark.*")); + _retainAllServices = CreateServices(RecordsPerMinute / LoggingBenchmarkWorkload.CategoryCount); + _adaptiveServices = CreateServices(AdaptiveCapacity); + + _baselineLoggers = CreateLoggers(_baselineServices); + _disabledLoggers = CreateLoggers(_disabledServices); + _retainAllPolicyLoggers = CreateLoggers(_retainAllPolicyServices); + _retainAllLoggers = CreateLoggers(_retainAllServices); + _adaptiveLoggers = CreateLoggers(_adaptiveServices); + _retainAllBuffer = _retainAllServices.GetRequiredService(); + _adaptiveBuffer = _adaptiveServices.GetRequiredService(); + } + + [GlobalCleanup] + public void GlobalCleanup() + { + _adaptiveServices.Dispose(); + _retainAllServices.Dispose(); + _retainAllPolicyServices.Dispose(); + _disabledServices.Dispose(); + _baselineServices.Dispose(); + } + + [IterationCleanup] + public void FlushBuffers() + { + _retainAllBuffer.Flush(); + _adaptiveBuffer.Flush(); + } + + [Benchmark(Baseline = true)] + public void NoSampling() + { + LoggingBenchmarkWorkload.LogBatch(_baselineLoggers, RecordsPerMinute); + } + + [Benchmark] + public void CckrDisabled() + { + LoggingBenchmarkWorkload.LogBatch(_disabledLoggers, RecordsPerMinute); + } + + [Benchmark] + public void CckrRetainAllPolicy() + { + LoggingBenchmarkWorkload.LogBatch(_retainAllPolicyLoggers, RecordsPerMinute); + } + + [Benchmark] + public void CckrRetainAll() + { + LoggingBenchmarkWorkload.LogBatch(_retainAllLoggers, RecordsPerMinute); + } + + [Benchmark] + public void CckrRetainAllAndFlush() + { + LoggingBenchmarkWorkload.LogBatch(_retainAllLoggers, RecordsPerMinute); + _retainAllBuffer.Flush(); + } + + [Benchmark] + public void CckrAdaptive() + { + LoggingBenchmarkWorkload.LogBatch(_adaptiveLoggers, RecordsPerMinute); + } + + [Benchmark] + public void CckrAdaptiveAndFlush() + { + LoggingBenchmarkWorkload.LogBatch(_adaptiveLoggers, RecordsPerMinute); + _adaptiveBuffer.Flush(); + } + + private static ServiceProvider CreateServices(int? capacity = null) + { + if (!capacity.HasValue) + { + return CreateServices((Action?)null); + } + + return CreateServices(options => + { + options.Capacity = capacity.Value; + options.PreserveCapacity = 0; + options.FlushInterval = TimeSpan.FromDays(1); + }); + } + + private static ServiceProvider CreateServices(Action? configure) + { + var services = new ServiceCollection(); + + services.AddLogging(builder => + { + builder.AddProvider(new BenchLoggerProvider()); + + if (configure is not null) + { + builder.AddCckrLogSampling(configure); + } + }); + + return services.BuildServiceProvider(); + } + + private static ILogger[] CreateLoggers(ServiceProvider services) + => LoggingBenchmarkWorkload.CreateLoggers(services.GetRequiredService()); +} diff --git a/bench/Libraries/Microsoft.Extensions.Telemetry.PerformanceTests/LoggingBenchmarkWorkload.cs b/bench/Libraries/Microsoft.Extensions.Telemetry.PerformanceTests/LoggingBenchmarkWorkload.cs new file mode 100644 index 00000000000..0cfb3c37550 --- /dev/null +++ b/bench/Libraries/Microsoft.Extensions.Telemetry.PerformanceTests/LoggingBenchmarkWorkload.cs @@ -0,0 +1,125 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. + +using System; +using Microsoft.Extensions.Logging; + +namespace Microsoft.Extensions.Telemetry.Bench; + +internal static class LoggingBenchmarkWorkload +{ + public const int CategoryCount = 4; + public const int SamplingInvocationsPerIteration = 8; + public const string CriticalCategoryPrefix = "Benchmark.Critical."; + public const string HighVolumeCategoryPrefix = "Benchmark.HighVolume."; + + private const int EventCount = 16; + + private static readonly string[] _categories = + [ + $"{HighVolumeCategoryPrefix}Orders", + $"{HighVolumeCategoryPrefix}Payments", + $"{CriticalCategoryPrefix}Audit", + "Benchmark.Diagnostics" + ]; + + private static readonly Action[] _messages = CreateMessages(); + + public static ILogger[] CreateLoggers(ILoggerFactory factory) + { + var loggers = new ILogger[CategoryCount]; + for (int i = 0; i < loggers.Length; i++) + { + loggers[i] = factory.CreateLogger(_categories[i]); + } + + return loggers; + } + + public static ILogger[] CreateLoggers(ILoggerFactory factory, int categoryCount) + { + var loggers = new ILogger[categoryCount]; + for (int i = 0; i < loggers.Length; i++) + { + loggers[i] = factory.CreateLogger($"Benchmark.Category.{i:D3}"); + } + + return loggers; + } + + public static void LogBatch(ILogger[] loggers, int recordCount) + { + for (int i = 0; i < recordCount; i++) + { + LogRecord(loggers, i); + } + } + + public static void LogBatchWithObserver(ILogger[] loggers, int recordCount, Action observer) + { + for (int i = 0; i < recordCount; i++) + { + LogRecord(loggers, i); + if ((i & 255) == 255) + { + observer(); + } + } + + observer(); + } + + public static void LogInterleaved( + ILogger[] loggers, + int firstRecord, + int recordStride, + int recordCount) + { + for (int i = firstRecord; i < recordCount; i += recordStride) + { + LogRecord(loggers, i); + } + } + + private static Action[] CreateMessages() + { + var messages = new Action[EventCount]; + for (int i = 0; i < messages.Length; i++) + { + messages[i] = LoggerMessage.Define( + LogLevel.Information, + new EventId(i + 1, $"BenchmarkEvent{i + 1}"), + "Benchmark message {Value}"); + } + + return messages; + } + + private static int SelectEventIndex(int recordIndex) + { + int percentile = recordIndex % 100; + if (percentile < 70) + { + return 0; + } + + if (percentile < 85) + { + return 1; + } + + if (percentile < 93) + { + return 2; + } + + return 3 + (recordIndex % (EventCount - 3)); + } + + private static void LogRecord(ILogger[] loggers, int recordIndex) + { + int eventIndex = SelectEventIndex(recordIndex); + int categoryIndex = (recordIndex / 100) % loggers.Length; + _messages[eventIndex](loggers[categoryIndex], recordIndex, null); + } +} diff --git a/bench/Libraries/Microsoft.Extensions.Telemetry.PerformanceTests/MultithreadedSamplingImpactBench.cs b/bench/Libraries/Microsoft.Extensions.Telemetry.PerformanceTests/MultithreadedSamplingImpactBench.cs new file mode 100644 index 00000000000..5dea1e3fe10 --- /dev/null +++ b/bench/Libraries/Microsoft.Extensions.Telemetry.PerformanceTests/MultithreadedSamplingImpactBench.cs @@ -0,0 +1,121 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. + +using System; +using System.Threading.Tasks; +using BenchmarkDotNet.Attributes; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Diagnostics.Buffering; +using Microsoft.Extensions.Logging; + +namespace Microsoft.Extensions.Telemetry.Bench; + +[MemoryDiagnoser] +[InvocationCount(1)] +public class MultithreadedSamplingImpactBench +{ + private const int CategoryCount = 100; + private const int RecordsPerMinute = 20_000; + + private ServiceProvider _baselineServices = null!; + private ServiceProvider _strategyServices = null!; + private ILogger[] _baselineLoggers = null!; + private ILogger[] _strategyLoggers = null!; + private LogBuffer? _strategyBuffer; + private ParallelOptions _parallelOptions = null!; + + public enum SamplingStrategy + { + RandomOnePercent, + CckrOnePercent + } + + [Params(1, 4, 8)] + public int WorkerCount { get; set; } + + [Params(SamplingStrategy.RandomOnePercent, SamplingStrategy.CckrOnePercent)] + public SamplingStrategy Strategy { get; set; } + + [GlobalSetup] + public void GlobalSetup() + { + _baselineServices = CreateServices(); + _strategyServices = CreateServices(Strategy); + _baselineLoggers = CreateLoggers(_baselineServices); + _strategyLoggers = CreateLoggers(_strategyServices); + _strategyBuffer = _strategyServices.GetService(); + _parallelOptions = new ParallelOptions { MaxDegreeOfParallelism = WorkerCount }; + } + + [GlobalCleanup] + public void GlobalCleanup() + { + _strategyServices.Dispose(); + _baselineServices.Dispose(); + } + + [IterationCleanup] + public void FlushBuffer() + { + _strategyBuffer?.Flush(); + } + + [Benchmark(Baseline = true)] + public void NoSampling() + { + LogConcurrently(_baselineLoggers); + } + + [Benchmark] + public void WithStrategy() + { + LogConcurrently(_strategyLoggers); + _strategyBuffer?.Flush(); + } + + private static ServiceProvider CreateServices(SamplingStrategy? strategy = null) + { + var services = new ServiceCollection(); + + services.AddLogging(builder => + { + builder.AddProvider(new BenchLoggerProvider()); + + switch (strategy) + { + case SamplingStrategy.RandomOnePercent: + builder.AddRandomProbabilisticSampler(0.01); + break; + + case SamplingStrategy.CckrOnePercent: + builder.AddCckrLogSampling(options => + { + options.Capacity = RecordsPerMinute / 100 / CategoryCount; + options.PreserveCapacity = 0; + options.FlushInterval = TimeSpan.FromDays(1); + }); + break; + } + }); + + return services.BuildServiceProvider(); + } + + private static ILogger[] CreateLoggers(ServiceProvider services) + => LoggingBenchmarkWorkload.CreateLoggers( + services.GetRequiredService(), + CategoryCount); + + private void LogConcurrently(ILogger[] loggers) + { + _ = Parallel.For( + 0, + WorkerCount, + _parallelOptions, + workerIndex => LoggingBenchmarkWorkload.LogInterleaved( + loggers, + workerIndex, + WorkerCount, + RecordsPerMinute)); + } +} diff --git a/bench/Libraries/Microsoft.Extensions.Telemetry.PerformanceTests/Program.cs b/bench/Libraries/Microsoft.Extensions.Telemetry.PerformanceTests/Program.cs index ad29c230ca5..b01ee0b2dc2 100644 --- a/bench/Libraries/Microsoft.Extensions.Telemetry.PerformanceTests/Program.cs +++ b/bench/Libraries/Microsoft.Extensions.Telemetry.PerformanceTests/Program.cs @@ -1,6 +1,7 @@ // Licensed to the .NET Foundation under one or more agreements. // The .NET Foundation licenses this file to you under the MIT license. +using System; using BenchmarkDotNet.Configs; using BenchmarkDotNet.Jobs; using BenchmarkDotNet.Running; @@ -11,9 +12,51 @@ internal static class Program { public static void Main(string[] args) { + if (args.Length == 1 && string.Equals(args[0], "--retained-memory", StringComparison.Ordinal)) + { + RetainedMemoryMeasurement.Run(); + return; + } + + if (args.Length == 3 && string.Equals(args[0], "--retained-memory-worker", StringComparison.Ordinal)) + { + RetainedMemoryMeasurement.RunWorker(args[1], args[2]); + return; + } + + if (args.Length == 1 && string.Equals(args[0], "--serialized-exporter-volume", StringComparison.Ordinal)) + { + SerializedExporterVolumeMeasurement.Run(); + return; + } + + if (args.Length == 1 && string.Equals(args[0], "--sustained-gc", StringComparison.Ordinal)) + { + SustainedGcPressureMeasurement.Run(); + return; + } + + if (args.Length == 2 && string.Equals(args[0], "--sustained-gc", StringComparison.Ordinal)) + { + SustainedGcPressureMeasurement.Run(args[1]); + return; + } + + if (args.Length == 3 && string.Equals(args[0], "--sustained-gc", StringComparison.Ordinal)) + { + SustainedGcPressureMeasurement.Run(args[1], args[2]); + return; + } + + if (args.Length == 5 && string.Equals(args[0], "--sustained-gc-worker", StringComparison.Ordinal)) + { + SustainedGcPressureMeasurement.RunWorker(args[1], args[2], args[3], args[4]); + return; + } + var dontRequireSlnToRunBenchmarks = ManualConfig .Create(DefaultConfig.Instance) - .AddJob(Job.MediumRun); + .AddJob(Job.MediumRun.WithEnvironmentVariable("DOTNET_TieredCompilation", "0")); BenchmarkSwitcher.FromAssembly(typeof(Program).Assembly).Run(args, dontRequireSlnToRunBenchmarks); } diff --git a/bench/Libraries/Microsoft.Extensions.Telemetry.PerformanceTests/RetainedMemoryMeasurement.cs b/bench/Libraries/Microsoft.Extensions.Telemetry.PerformanceTests/RetainedMemoryMeasurement.cs new file mode 100644 index 00000000000..d9130fa4ded --- /dev/null +++ b/bench/Libraries/Microsoft.Extensions.Telemetry.PerformanceTests/RetainedMemoryMeasurement.cs @@ -0,0 +1,218 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. + +using System; +using System.Diagnostics; +using System.Globalization; +using System.IO; +using System.Reflection; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Diagnostics.Buffering; +using Microsoft.Extensions.Diagnostics.Sampling; +using Microsoft.Extensions.Logging; + +namespace Microsoft.Extensions.Telemetry.Bench; + +internal static class RetainedMemoryMeasurement +{ + private enum Pipeline + { + NoSampling, + RandomOnePercent, + RandomByCategory, + GlobalBuffer, + CckrAdaptive, + CckrRetainAll + } + + public static void Run() + { + Console.WriteLine("| Pipeline | Records | Retained managed | Peak managed | Peak working set | Peak private bytes |"); + Console.WriteLine("|---|---:|---:|---:|---:|---:|"); + + foreach (int recordCount in new[] { 10_000, 20_000 }) + { + foreach (Pipeline pipeline in Enum.GetValues()) + { + RunIsolatedWorker(pipeline, recordCount); + } + } + } + + public static void RunWorker(string pipelineValue, string recordCountValue) + { + if (!Enum.TryParse(pipelineValue, out Pipeline pipeline) || + !int.TryParse(recordCountValue, NumberStyles.None, CultureInfo.InvariantCulture, out int recordCount)) + { +#pragma warning disable LA0001 // Worker argument validation is outside the measured path. + throw new ArgumentException("Invalid retained-memory worker arguments."); +#pragma warning restore LA0001 + } + + Result result = Measure(pipeline, recordCount); + Console.WriteLine( + $"| {pipeline} | {recordCount:N0} | {FormatBytes(result.RetainedManagedBytes)} | " + + $"{FormatBytes(result.PeakManagedBytes)} | {FormatBytes(result.PeakWorkingSetBytes)} | " + + $"{FormatBytes(result.PeakPrivateBytes)} |"); + } + + private static Result Measure(Pipeline pipeline, int recordCount) + { + ForceFullCollection(); + + using ServiceProvider services = CreateServices(pipeline, recordCount); + ILogger[] loggers = LoggingBenchmarkWorkload.CreateLoggers( + services.GetRequiredService()); + + ForceFullCollection(); + + using Process process = Process.GetCurrentProcess(); + process.Refresh(); + + long managedBefore = GC.GetTotalMemory(forceFullCollection: false); + long workingSetBefore = process.WorkingSet64; + long privateBytesBefore = process.PrivateMemorySize64; + long peakManaged = managedBefore; + long peakWorkingSet = workingSetBefore; + long peakPrivateBytes = privateBytesBefore; + + LoggingBenchmarkWorkload.LogBatchWithObserver(loggers, recordCount, ObserveMemory); + + ForceFullCollection(); + long retainedManaged = GC.GetTotalMemory(forceFullCollection: false) - managedBefore; + + return new Result( + retainedManaged, + peakManaged - managedBefore, + peakWorkingSet - workingSetBefore, + peakPrivateBytes - privateBytesBefore); + + void ObserveMemory() + { + peakManaged = Math.Max(peakManaged, GC.GetTotalMemory(forceFullCollection: false)); + process.Refresh(); + peakWorkingSet = Math.Max(peakWorkingSet, process.WorkingSet64); + peakPrivateBytes = Math.Max(peakPrivateBytes, process.PrivateMemorySize64); + } + } + + private static ServiceProvider CreateServices(Pipeline pipeline, int recordCount) + { + var services = new ServiceCollection(); + + services.AddLogging(builder => + { + builder.AddProvider(new BenchLoggerProvider()); + + switch (pipeline) + { + case Pipeline.RandomOnePercent: + builder.AddRandomProbabilisticSampler(0.01); + break; + + case Pipeline.RandomByCategory: + builder.AddRandomProbabilisticSampler(options => + { + options.Rules.Add(new RandomProbabilisticSamplerFilterRule( + 0.01, + categoryName: $"{LoggingBenchmarkWorkload.HighVolumeCategoryPrefix}*")); + options.Rules.Add(new RandomProbabilisticSamplerFilterRule( + 1.0, + categoryName: $"{LoggingBenchmarkWorkload.CriticalCategoryPrefix}*")); + options.Rules.Add(new RandomProbabilisticSamplerFilterRule(0.1)); + }); + break; + + case Pipeline.GlobalBuffer: + builder.AddGlobalBuffer(options => + { + options.AutoFlushDuration = TimeSpan.Zero; + options.MaxBufferSizeInBytes = 512 * 1024 * 1024; + options.Rules.Add(new LogBufferingFilterRule(logLevel: LogLevel.Information)); + }); + break; + + case Pipeline.CckrAdaptive: + AddCckr(builder, capacity: 128); + break; + + case Pipeline.CckrRetainAll: + AddCckr(builder, capacity: recordCount); + break; + } + }); + + return services.BuildServiceProvider(); + } + + private static void AddCckr(ILoggingBuilder builder, int capacity) + { + builder.AddCckrLogSampling(options => + { + options.Capacity = capacity; + options.PreserveCapacity = 0; + options.FlushInterval = TimeSpan.FromDays(1); + }); + } + + private static void ForceFullCollection() + { +#pragma warning disable S1215 // Full collections establish the live-memory baseline for this diagnostic. + GC.Collect(); + GC.WaitForPendingFinalizers(); + GC.Collect(); +#pragma warning restore S1215 + } + + private static void RunIsolatedWorker(Pipeline pipeline, int recordCount) + { + string executable = Environment.ProcessPath + ?? throw new InvalidOperationException("Unable to locate the current executable."); + + var startInfo = new ProcessStartInfo(executable) + { + RedirectStandardOutput = true, + RedirectStandardError = true, + UseShellExecute = false + }; + + if (string.Equals(Path.GetFileNameWithoutExtension(executable), "dotnet", StringComparison.OrdinalIgnoreCase)) + { + startInfo.ArgumentList.Add( + Assembly.GetEntryAssembly()?.Location + ?? throw new InvalidOperationException("Unable to locate the entry assembly.")); + } + + startInfo.ArgumentList.Add("--retained-memory-worker"); + startInfo.ArgumentList.Add(pipeline.ToString()); +#pragma warning disable LA0002 // Worker process setup is outside the measured path. + startInfo.ArgumentList.Add(recordCount.ToString(CultureInfo.InvariantCulture)); +#pragma warning restore LA0002 + + using Process worker = Process.Start(startInfo) + ?? throw new InvalidOperationException("Unable to start the retained-memory worker."); + string output = worker.StandardOutput.ReadToEnd(); + string error = worker.StandardError.ReadToEnd(); + worker.WaitForExit(); + + if (worker.ExitCode != 0) + { + throw new InvalidOperationException( + $"Retained-memory worker failed with exit code {worker.ExitCode}: {error}"); + } + + Console.Write(output); + } + + private static string FormatBytes(long bytes) + { + const double BytesPerMiB = 1024 * 1024; + return $"{bytes / BytesPerMiB:N2} MiB"; + } + + private readonly record struct Result( + long RetainedManagedBytes, + long PeakManagedBytes, + long PeakWorkingSetBytes, + long PeakPrivateBytes); +} diff --git a/bench/Libraries/Microsoft.Extensions.Telemetry.PerformanceTests/SamplingImpactBench.cs b/bench/Libraries/Microsoft.Extensions.Telemetry.PerformanceTests/SamplingImpactBench.cs new file mode 100644 index 00000000000..c3191ebd802 --- /dev/null +++ b/bench/Libraries/Microsoft.Extensions.Telemetry.PerformanceTests/SamplingImpactBench.cs @@ -0,0 +1,124 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. + +using System.Diagnostics; +using BenchmarkDotNet.Attributes; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Diagnostics.Sampling; +using Microsoft.Extensions.Logging; + +namespace Microsoft.Extensions.Telemetry.Bench; + +[MemoryDiagnoser] +[InvocationCount(LoggingBenchmarkWorkload.SamplingInvocationsPerIteration)] +public class SamplingImpactBench +{ + private ServiceProvider _baselineServices = null!; + private ServiceProvider _sampledServices = null!; + private ILogger[] _baselineLoggers = null!; + private ILogger[] _sampledLoggers = null!; + private Activity? _activity; + + public enum SamplingScenario + { + RandomSampleAll, + RandomSampleOnePercent, + RandomDropAll, + RandomByCategory, + TraceSample, + TraceDrop + } + + [Params(10_000, 20_000)] + public int RecordsPerMinute { get; set; } + + [Params( + SamplingScenario.RandomSampleAll, + SamplingScenario.RandomSampleOnePercent, + SamplingScenario.RandomDropAll, + SamplingScenario.RandomByCategory, + SamplingScenario.TraceSample, + SamplingScenario.TraceDrop)] + public SamplingScenario Scenario { get; set; } + + [GlobalSetup] + public void GlobalSetup() + { + _baselineServices = CreateServices(); + _sampledServices = CreateServices(Scenario); + _baselineLoggers = LoggingBenchmarkWorkload.CreateLoggers(_baselineServices.GetRequiredService()); + _sampledLoggers = LoggingBenchmarkWorkload.CreateLoggers(_sampledServices.GetRequiredService()); + + if (Scenario is SamplingScenario.TraceSample or SamplingScenario.TraceDrop) + { + _activity = new Activity("SamplingBenchmark") + { + ActivityTraceFlags = Scenario == SamplingScenario.TraceSample + ? ActivityTraceFlags.Recorded + : ActivityTraceFlags.None + }; + _activity.Start(); + } + } + + [GlobalCleanup] + public void GlobalCleanup() + { + _activity?.Stop(); + _sampledServices.Dispose(); + _baselineServices.Dispose(); + } + + [Benchmark(Baseline = true)] + public void NoSampling() + { + LoggingBenchmarkWorkload.LogBatch(_baselineLoggers, RecordsPerMinute); + } + + [Benchmark] + public void WithSampling() + { + LoggingBenchmarkWorkload.LogBatch(_sampledLoggers, RecordsPerMinute); + } + + private static ServiceProvider CreateServices(SamplingScenario? scenario = null) + { + var services = new ServiceCollection(); + + services.AddLogging(builder => + { + builder.AddProvider(new BenchLoggerProvider()); + + switch (scenario) + { + case SamplingScenario.RandomSampleAll: + builder.AddRandomProbabilisticSampler(1.0); + break; + case SamplingScenario.RandomSampleOnePercent: + builder.AddRandomProbabilisticSampler(0.01); + break; + case SamplingScenario.RandomDropAll: + builder.AddRandomProbabilisticSampler(0.0); + break; + case SamplingScenario.RandomByCategory: + builder.AddRandomProbabilisticSampler(options => + { + options.Rules.Add(new RandomProbabilisticSamplerFilterRule( + 0.01, + categoryName: $"{LoggingBenchmarkWorkload.HighVolumeCategoryPrefix}*")); + options.Rules.Add(new RandomProbabilisticSamplerFilterRule( + 1.0, + categoryName: $"{LoggingBenchmarkWorkload.CriticalCategoryPrefix}*")); + options.Rules.Add(new RandomProbabilisticSamplerFilterRule(0.1)); + }); + break; + case SamplingScenario.TraceSample: + case SamplingScenario.TraceDrop: + builder.AddTraceBasedSampler(); + break; + } + }); + + return services.BuildServiceProvider(); + } +} diff --git a/bench/Libraries/Microsoft.Extensions.Telemetry.PerformanceTests/SerializedExporterImpactBench.cs b/bench/Libraries/Microsoft.Extensions.Telemetry.PerformanceTests/SerializedExporterImpactBench.cs new file mode 100644 index 00000000000..d9b21fcf3f5 --- /dev/null +++ b/bench/Libraries/Microsoft.Extensions.Telemetry.PerformanceTests/SerializedExporterImpactBench.cs @@ -0,0 +1,146 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. + +using System; +using System.Diagnostics; +using BenchmarkDotNet.Attributes; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Diagnostics.Buffering; +using Microsoft.Extensions.Diagnostics.Sampling; +using Microsoft.Extensions.Logging; + +namespace Microsoft.Extensions.Telemetry.Bench; + +[MemoryDiagnoser] +[InvocationCount(1)] +public class SerializedExporterImpactBench +{ + private ServiceProvider _baselineServices = null!; + private ServiceProvider _strategyServices = null!; + private ILogger[] _baselineLoggers = null!; + private ILogger[] _strategyLoggers = null!; + private LogBuffer? _strategyBuffer; + private Activity? _activity; + + public enum ExportStrategy + { + RandomOnePercent, + RandomByCategory, + TraceRetain, + TraceDrop, + CckrOnePercent + } + + [Params(10_000, 20_000)] + public int RecordsPerMinute { get; set; } + + [Params( + ExportStrategy.RandomOnePercent, + ExportStrategy.RandomByCategory, + ExportStrategy.TraceRetain, + ExportStrategy.TraceDrop, + ExportStrategy.CckrOnePercent)] + public ExportStrategy Strategy { get; set; } + + [GlobalSetup] + public void GlobalSetup() + { + _baselineServices = CreateServices(); + _strategyServices = CreateServices(Strategy, RecordsPerMinute); + _baselineLoggers = CreateLoggers(_baselineServices); + _strategyLoggers = CreateLoggers(_strategyServices); + _strategyBuffer = _strategyServices.GetService(); + + if (Strategy is ExportStrategy.TraceRetain or ExportStrategy.TraceDrop) + { + _activity = new Activity("SerializedExporterBenchmark") + { + ActivityTraceFlags = Strategy == ExportStrategy.TraceRetain + ? ActivityTraceFlags.Recorded + : ActivityTraceFlags.None + }; + _activity.Start(); + } + } + + [GlobalCleanup] + public void GlobalCleanup() + { + _activity?.Stop(); + _strategyServices.Dispose(); + _baselineServices.Dispose(); + } + + [IterationCleanup] + public void FlushBuffer() + { + _strategyBuffer?.Flush(); + } + + [Benchmark(Baseline = true)] + public void NoSampling() + { + LoggingBenchmarkWorkload.LogBatch(_baselineLoggers, RecordsPerMinute); + } + + [Benchmark] + public void WithStrategy() + { + LoggingBenchmarkWorkload.LogBatch(_strategyLoggers, RecordsPerMinute); + _strategyBuffer?.Flush(); + } + + internal static ServiceProvider CreateServices( + ExportStrategy? strategy = null, + int recordCount = 0, + SerializedExporterMetrics? metrics = null) + { + var services = new ServiceCollection(); + + services.AddLogging(builder => + { + builder.AddProvider(new SerializedExporterLoggerProvider(metrics)); + + switch (strategy) + { + case ExportStrategy.RandomOnePercent: + builder.AddRandomProbabilisticSampler(0.01); + break; + + case ExportStrategy.RandomByCategory: + builder.AddRandomProbabilisticSampler(options => + { + options.Rules.Add(new RandomProbabilisticSamplerFilterRule( + 0.01, + categoryName: $"{LoggingBenchmarkWorkload.HighVolumeCategoryPrefix}*")); + options.Rules.Add(new RandomProbabilisticSamplerFilterRule( + 1.0, + categoryName: $"{LoggingBenchmarkWorkload.CriticalCategoryPrefix}*")); + options.Rules.Add(new RandomProbabilisticSamplerFilterRule(0.1)); + }); + break; + + case ExportStrategy.TraceRetain: + case ExportStrategy.TraceDrop: + builder.AddTraceBasedSampler(); + break; + + case ExportStrategy.CckrOnePercent: + builder.AddCckrLogSampling(options => + { + options.Capacity = Math.Max( + 1, + recordCount / 100 / LoggingBenchmarkWorkload.CategoryCount); + options.PreserveCapacity = 0; + options.FlushInterval = TimeSpan.FromDays(1); + }); + break; + } + }); + + return services.BuildServiceProvider(); + } + + private static ILogger[] CreateLoggers(ServiceProvider services) + => LoggingBenchmarkWorkload.CreateLoggers(services.GetRequiredService()); +} diff --git a/bench/Libraries/Microsoft.Extensions.Telemetry.PerformanceTests/SerializedExporterLoggerProvider.cs b/bench/Libraries/Microsoft.Extensions.Telemetry.PerformanceTests/SerializedExporterLoggerProvider.cs new file mode 100644 index 00000000000..f168655d0f9 --- /dev/null +++ b/bench/Libraries/Microsoft.Extensions.Telemetry.PerformanceTests/SerializedExporterLoggerProvider.cs @@ -0,0 +1,152 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. + +using System; +using System.Buffers; +using System.Collections.Generic; +using System.Text.Json; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; + +namespace Microsoft.Extensions.Telemetry.Bench; + +internal sealed class SerializedExporterLoggerProvider : ILoggerProvider +{ + private readonly SerializedExporterMetrics? _metrics; + + public SerializedExporterLoggerProvider(SerializedExporterMetrics? metrics = null) + { + _metrics = metrics; + } + + public ILogger CreateLogger(string categoryName) => new SerializedExporterLogger(categoryName, _metrics); + + public void Dispose() + { + } + + private sealed class SerializedExporterLogger : ILogger, IBufferedLogger + { + private sealed class Scope : IDisposable + { + public static Scope Instance { get; } = new(); + + public void Dispose() + { + } + } + + private static void WriteValue(Utf8JsonWriter writer, string name, object? value) + { + switch (value) + { + case null: + writer.WriteNull(name); + break; + case bool boolean: + writer.WriteBoolean(name, boolean); + break; + case int integer: + writer.WriteNumber(name, integer); + break; + case long longInteger: + writer.WriteNumber(name, longInteger); + break; + case double doubleValue: + writer.WriteNumber(name, doubleValue); + break; + case string text: + writer.WriteString(name, text); + break; + default: + writer.WriteString(name, value.ToString()); + break; + } + } + + private readonly ArrayBufferWriter _output = new(); + private readonly string _category; + private readonly SerializedExporterMetrics? _metrics; + + public SerializedExporterLogger(string category, SerializedExporterMetrics? metrics) + { + _category = category; + _metrics = metrics; + } + + public IDisposable? BeginScope(TState state) + where TState : notnull + => Scope.Instance; + + public bool IsEnabled(LogLevel logLevel) => true; + + public void Log( + LogLevel logLevel, + EventId eventId, + TState state, + Exception? exception, + Func formatter) + { + Serialize( + logLevel, + eventId, + state as IReadOnlyList>, + formatter(state, exception), + exception?.ToString()); + } + + public void LogRecords(IEnumerable records) + { + _metrics?.RecordBatch(); + + foreach (BufferedLogRecord record in records) + { + Serialize( + record.LogLevel, + record.EventId, + record.Attributes, + record.FormattedMessage ?? string.Empty, + record.Exception); + } + } + + private void Serialize( + LogLevel logLevel, + EventId eventId, + IReadOnlyList>? attributes, + string formattedMessage, + string? exception) + { + _output.Clear(); + + using var writer = new Utf8JsonWriter(_output); + writer.WriteStartObject(); + writer.WriteString("category", _category); + writer.WriteString("level", logLevel.ToString()); + writer.WriteNumber("eventId", eventId.Id); + writer.WriteString("eventName", eventId.Name); + writer.WriteString("message", formattedMessage); + + if (exception is not null) + { + writer.WriteString("exception", exception); + } + + writer.WriteStartObject("attributes"); + if (attributes is not null) + { + for (int i = 0; i < attributes.Count; i++) + { + KeyValuePair attribute = attributes[i]; + WriteValue(writer, attribute.Key, attribute.Value); + } + } + + writer.WriteEndObject(); + writer.WriteEndObject(); + writer.Flush(); + + _metrics?.Record(_output.WrittenCount); + } + } +} diff --git a/bench/Libraries/Microsoft.Extensions.Telemetry.PerformanceTests/SerializedExporterMetrics.cs b/bench/Libraries/Microsoft.Extensions.Telemetry.PerformanceTests/SerializedExporterMetrics.cs new file mode 100644 index 00000000000..f79a099e8d2 --- /dev/null +++ b/bench/Libraries/Microsoft.Extensions.Telemetry.PerformanceTests/SerializedExporterMetrics.cs @@ -0,0 +1,37 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. + +using System.Threading; + +namespace Microsoft.Extensions.Telemetry.Bench; + +internal sealed class SerializedExporterMetrics +{ + private long _batchesEmitted; + private long _bytesEmitted; + private long _recordsEmitted; + + public long BatchesEmitted => Interlocked.Read(ref _batchesEmitted); + + public long BytesEmitted => Interlocked.Read(ref _bytesEmitted); + + public long RecordsEmitted => Interlocked.Read(ref _recordsEmitted); + + public void Record(int bytes) + { + Interlocked.Increment(ref _recordsEmitted); + Interlocked.Add(ref _bytesEmitted, bytes); + } + + public void RecordBatch() + { + Interlocked.Increment(ref _batchesEmitted); + } + + public void Reset() + { + Interlocked.Exchange(ref _batchesEmitted, 0); + Interlocked.Exchange(ref _bytesEmitted, 0); + Interlocked.Exchange(ref _recordsEmitted, 0); + } +} diff --git a/bench/Libraries/Microsoft.Extensions.Telemetry.PerformanceTests/SerializedExporterVolumeMeasurement.cs b/bench/Libraries/Microsoft.Extensions.Telemetry.PerformanceTests/SerializedExporterVolumeMeasurement.cs new file mode 100644 index 00000000000..d4d1e8f2f58 --- /dev/null +++ b/bench/Libraries/Microsoft.Extensions.Telemetry.PerformanceTests/SerializedExporterVolumeMeasurement.cs @@ -0,0 +1,71 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. + +using System; +using System.Diagnostics; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Diagnostics.Buffering; +using Microsoft.Extensions.Logging; + +namespace Microsoft.Extensions.Telemetry.Bench; + +internal static class SerializedExporterVolumeMeasurement +{ + public static void Run() + { + Console.WriteLine("| Strategy | Input | Emitted | Retention | UTF-8 bytes | Bytes/record |"); + Console.WriteLine("|---|---:|---:|---:|---:|---:|"); + + foreach (int recordCount in new[] { 10_000, 20_000 }) + { + Measure(strategy: null, recordCount); + foreach (SerializedExporterImpactBench.ExportStrategy strategy + in Enum.GetValues()) + { + Measure(strategy, recordCount); + } + } + } + + private static void Measure(SerializedExporterImpactBench.ExportStrategy? strategy, int recordCount) + { + var metrics = new SerializedExporterMetrics(); + using ServiceProvider services = + SerializedExporterImpactBench.CreateServices(strategy, recordCount, metrics); + ILogger[] loggers = LoggingBenchmarkWorkload.CreateLoggers( + services.GetRequiredService()); + using Activity? activity = StartActivity(strategy); + + LoggingBenchmarkWorkload.LogBatch(loggers, recordCount); + services.GetService()?.Flush(); + + double retention = (double)metrics.RecordsEmitted / recordCount; + double bytesPerRecord = metrics.RecordsEmitted == 0 + ? 0 + : (double)metrics.BytesEmitted / metrics.RecordsEmitted; + + Console.WriteLine( + $"| {strategy?.ToString() ?? "NoSampling"} | {recordCount:N0} | " + + $"{metrics.RecordsEmitted:N0} | {retention:P2} | {metrics.BytesEmitted:N0} | " + + $"{bytesPerRecord:N1} |"); + } + + private static Activity? StartActivity(SerializedExporterImpactBench.ExportStrategy? strategy) + { + if (strategy is not ( + SerializedExporterImpactBench.ExportStrategy.TraceRetain or + SerializedExporterImpactBench.ExportStrategy.TraceDrop)) + { + return null; + } + + var activity = new Activity("SerializedExporterVolume") + { + ActivityTraceFlags = strategy == SerializedExporterImpactBench.ExportStrategy.TraceRetain + ? ActivityTraceFlags.Recorded + : ActivityTraceFlags.None + }; + activity.Start(); + return activity; + } +} diff --git a/bench/Libraries/Microsoft.Extensions.Telemetry.PerformanceTests/SustainedGcPressureMeasurement.cs b/bench/Libraries/Microsoft.Extensions.Telemetry.PerformanceTests/SustainedGcPressureMeasurement.cs new file mode 100644 index 00000000000..77e5877c25b --- /dev/null +++ b/bench/Libraries/Microsoft.Extensions.Telemetry.PerformanceTests/SustainedGcPressureMeasurement.cs @@ -0,0 +1,355 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. + +using System; +using System.Diagnostics; +using System.Globalization; +using System.IO; +using System.Reflection; +using System.Threading; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Diagnostics.Buffering; +using Microsoft.Extensions.Logging; + +namespace Microsoft.Extensions.Telemetry.Bench; + +internal static class SustainedGcPressureMeasurement +{ + private const int CategoryCount = 4; + private const int DefaultLogsPerMinute = 10_000; + private const int WarmupRecordCount = 1_000; + private static readonly TimeSpan _defaultDuration = TimeSpan.FromMinutes(2); + private static readonly TimeSpan _defaultFlushInterval = TimeSpan.FromSeconds(30); + + private enum Strategy + { + RandomOnePercent, + CckrOnePercent + } + + public static void Run() + => Run(DefaultLogsPerMinute, _defaultFlushInterval); + + public static void Run(string logsPerMinuteValue) + => Run(logsPerMinuteValue, _defaultFlushInterval.TotalSeconds.ToString(CultureInfo.InvariantCulture)); + + public static void Run(string logsPerMinuteValue, string flushIntervalSecondsValue) + { + if (!int.TryParse( + logsPerMinuteValue, + NumberStyles.None, + CultureInfo.InvariantCulture, + out int logsPerMinute) || + logsPerMinute <= 0) + { +#pragma warning disable LA0001 // Command-line argument validation is outside the measured path. + throw new ArgumentException("Logs per minute must be a positive integer.", nameof(logsPerMinuteValue)); +#pragma warning restore LA0001 + } + + if (!double.TryParse( + flushIntervalSecondsValue, + NumberStyles.AllowDecimalPoint, + CultureInfo.InvariantCulture, + out double flushIntervalSeconds) || + flushIntervalSeconds <= 0) + { +#pragma warning disable LA0001 // Command-line argument validation is outside the measured path. + throw new ArgumentException( + "Flush interval seconds must be positive.", + nameof(flushIntervalSecondsValue)); +#pragma warning restore LA0001 + } + + Run(logsPerMinute, TimeSpan.FromSeconds(flushIntervalSeconds)); + } + + private static void Run(int logsPerMinute, TimeSpan flushInterval) + { + Console.WriteLine( + $"Sustained logging: {logsPerMinute:N0} logs/min, {_defaultDuration.TotalMinutes:N0} min, " + + $"{flushInterval.TotalSeconds:N1} s CCKR flush interval"); + Console.WriteLine(); + Console.WriteLine( + "| Strategy | Input | Emitted | Retention | Export batches | Allocated | Gen0 | Gen1 | Gen2 | " + + "GC pause | CPU time | CPU/wall | Peak managed | Retained managed |"); + Console.WriteLine( + "|---|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|"); + + foreach (Strategy strategy in Enum.GetValues()) + { + RunIsolatedWorker(strategy, _defaultDuration, logsPerMinute, flushInterval); + } + } + + public static void RunWorker( + string strategyValue, + string durationSecondsValue, + string logsPerMinuteValue, + string flushIntervalSecondsValue) + { + bool validStrategy = Enum.TryParse(strategyValue, out Strategy strategy); + bool validDuration = double.TryParse( + durationSecondsValue, + NumberStyles.AllowDecimalPoint, + CultureInfo.InvariantCulture, + out double durationSeconds); + bool validRate = int.TryParse( + logsPerMinuteValue, + NumberStyles.None, + CultureInfo.InvariantCulture, + out int logsPerMinute); + bool validFlushInterval = double.TryParse( + flushIntervalSecondsValue, + NumberStyles.AllowDecimalPoint, + CultureInfo.InvariantCulture, + out double flushIntervalSeconds); + + if (!validStrategy || !validDuration || !validRate || !validFlushInterval) + { + ThrowInvalidWorkerArguments(); + } + + if (durationSeconds <= 0 || logsPerMinute <= 0 || flushIntervalSeconds <= 0) + { + ThrowInvalidWorkerArguments(); + } + + Result result = Measure( + strategy, + TimeSpan.FromSeconds(durationSeconds), + logsPerMinute, + TimeSpan.FromSeconds(flushIntervalSeconds)); + Console.WriteLine( + $"| {strategy} | {result.InputRecords:N0} | {result.EmittedRecords:N0} | " + + $"{result.Retention:P2} | {result.ExportBatches:N0} | {FormatBytes(result.AllocatedBytes)} | " + + $"{result.Gen0Collections:N0} | {result.Gen1Collections:N0} | {result.Gen2Collections:N0} | " + + $"{result.GcPause.TotalMilliseconds:N1} ms | {result.CpuTime.TotalMilliseconds:N0} ms | " + + $"{result.CpuPercent:N1}% | {FormatBytes(result.PeakManagedBytes)} | " + + $"{FormatBytes(result.RetainedManagedBytes)} |"); + } + + private static Result Measure( + Strategy strategy, + TimeSpan duration, + int logsPerMinute, + TimeSpan flushInterval) + { + var metrics = new SerializedExporterMetrics(); + using ServiceProvider services = CreateServices(strategy, metrics, logsPerMinute, flushInterval); + ILogger[] loggers = LoggingBenchmarkWorkload.CreateLoggers( + services.GetRequiredService(), + CategoryCount); + LogBuffer? buffer = services.GetService(); + + LoggingBenchmarkWorkload.LogBatch(loggers, WarmupRecordCount); + buffer?.Flush(); + metrics.Reset(); + ForceFullCollection(); + + using Process process = Process.GetCurrentProcess(); + + long managedBefore = GC.GetTotalMemory(forceFullCollection: false); + long allocatedBefore = GC.GetTotalAllocatedBytes(precise: true); + long peakManaged = managedBefore; + int gen0Before = GC.CollectionCount(0); + int gen1Before = GC.CollectionCount(1); + int gen2Before = GC.CollectionCount(2); + TimeSpan pauseBefore = GC.GetTotalPauseDuration(); + TimeSpan cpuBefore = process.TotalProcessorTime; + + int targetRecordCount = checked((int)Math.Round( + logsPerMinute * duration.TotalMinutes, + MidpointRounding.AwayFromZero)); + int inputRecords = 0; + var stopwatch = Stopwatch.StartNew(); + TimeSpan nextMemoryObservation = TimeSpan.Zero; + + while (stopwatch.Elapsed < duration) + { + int expectedRecords = Math.Min( + targetRecordCount, + (int)(logsPerMinute * stopwatch.Elapsed.TotalMinutes)); + + if (expectedRecords > inputRecords) + { + LoggingBenchmarkWorkload.LogInterleaved( + loggers, + inputRecords, + recordStride: 1, + expectedRecords); + inputRecords = expectedRecords; + } + + if (stopwatch.Elapsed >= nextMemoryObservation) + { + ObserveMemory(); + nextMemoryObservation += TimeSpan.FromSeconds(1); + } + + Thread.Sleep(TimeSpan.FromMilliseconds(10)); + } + + if (inputRecords < targetRecordCount) + { + LoggingBenchmarkWorkload.LogInterleaved( + loggers, + inputRecords, + recordStride: 1, + targetRecordCount); + inputRecords = targetRecordCount; + } + + buffer?.Flush(); + ObserveMemory(); + stopwatch.Stop(); + + long allocatedBytes = GC.GetTotalAllocatedBytes(precise: true) - allocatedBefore; + int gen0Collections = GC.CollectionCount(0) - gen0Before; + int gen1Collections = GC.CollectionCount(1) - gen1Before; + int gen2Collections = GC.CollectionCount(2) - gen2Before; + TimeSpan gcPause = GC.GetTotalPauseDuration() - pauseBefore; + TimeSpan cpuTime = process.TotalProcessorTime - cpuBefore; + double cpuPercent = cpuTime.TotalMilliseconds / stopwatch.Elapsed.TotalMilliseconds * 100; + + ForceFullCollection(); + long retainedManaged = GC.GetTotalMemory(forceFullCollection: false) - managedBefore; + + return new Result( + inputRecords, + metrics.RecordsEmitted, + metrics.BatchesEmitted, + allocatedBytes, + gen0Collections, + gen1Collections, + gen2Collections, + gcPause, + cpuTime, + cpuPercent, + peakManaged - managedBefore, + retainedManaged); + + void ObserveMemory() + => peakManaged = Math.Max(peakManaged, GC.GetTotalMemory(forceFullCollection: false)); + } + + private static ServiceProvider CreateServices( + Strategy strategy, + SerializedExporterMetrics metrics, + int logsPerMinute, + TimeSpan flushInterval) + { + var services = new ServiceCollection(); + + services.AddLogging(builder => + { + builder.AddProvider(new SerializedExporterLoggerProvider(metrics)); + + switch (strategy) + { + case Strategy.RandomOnePercent: + builder.AddRandomProbabilisticSampler(0.01); + break; + + case Strategy.CckrOnePercent: + builder.AddCckrLogSampling(options => + { + options.Capacity = Math.Max( + 1, + (int)(logsPerMinute * flushInterval.TotalMinutes / 100 / CategoryCount)); + options.PreserveCapacity = 0; + options.FlushInterval = flushInterval; + }); + break; + } + }); + + return services.BuildServiceProvider(); + } + + private static void ForceFullCollection() + { +#pragma warning disable S1215 // Full collections establish comparable worker baselines. + GC.Collect(); + GC.WaitForPendingFinalizers(); + GC.Collect(); +#pragma warning restore S1215 + } + + private static void ThrowInvalidWorkerArguments() + { +#pragma warning disable LA0001 // Worker argument validation is outside the measured path. + throw new ArgumentException("Invalid sustained-GC worker arguments."); +#pragma warning restore LA0001 + } + + private static void RunIsolatedWorker( + Strategy strategy, + TimeSpan duration, + int logsPerMinute, + TimeSpan flushInterval) + { + string executable = Environment.ProcessPath + ?? throw new InvalidOperationException("Unable to locate the current executable."); + + var startInfo = new ProcessStartInfo(executable) + { + RedirectStandardOutput = true, + RedirectStandardError = true, + UseShellExecute = false + }; + startInfo.Environment["DOTNET_TieredCompilation"] = "0"; + + if (string.Equals(Path.GetFileNameWithoutExtension(executable), "dotnet", StringComparison.OrdinalIgnoreCase)) + { + startInfo.ArgumentList.Add( + Assembly.GetEntryAssembly()?.Location + ?? throw new InvalidOperationException("Unable to locate the entry assembly.")); + } + + startInfo.ArgumentList.Add("--sustained-gc-worker"); + startInfo.ArgumentList.Add(strategy.ToString()); + startInfo.ArgumentList.Add(duration.TotalSeconds.ToString(CultureInfo.InvariantCulture)); +#pragma warning disable LA0002 // Worker process setup is outside the measured path. + startInfo.ArgumentList.Add(logsPerMinute.ToString(CultureInfo.InvariantCulture)); +#pragma warning restore LA0002 + startInfo.ArgumentList.Add(flushInterval.TotalSeconds.ToString(CultureInfo.InvariantCulture)); + + using Process worker = Process.Start(startInfo) + ?? throw new InvalidOperationException("Unable to start the sustained-GC worker."); + string output = worker.StandardOutput.ReadToEnd(); + string error = worker.StandardError.ReadToEnd(); + worker.WaitForExit(); + + if (worker.ExitCode != 0) + { + throw new InvalidOperationException( + $"Sustained-GC worker failed with exit code {worker.ExitCode}: {error}"); + } + + Console.Write(output); + } + + private static string FormatBytes(long bytes) + { + const double BytesPerMiB = 1024 * 1024; + return $"{bytes / BytesPerMiB:N2} MiB"; + } + + private readonly record struct Result( + int InputRecords, + long EmittedRecords, + long ExportBatches, + long AllocatedBytes, + int Gen0Collections, + int Gen1Collections, + int Gen2Collections, + TimeSpan GcPause, + TimeSpan CpuTime, + double CpuPercent, + long PeakManagedBytes, + long RetainedManagedBytes) + { + public double Retention => (double)EmittedRecords / InputRecords; + } +} diff --git a/bench/Libraries/Microsoft.Extensions.Telemetry.PerformanceTests/design.md b/bench/Libraries/Microsoft.Extensions.Telemetry.PerformanceTests/design.md new file mode 100644 index 00000000000..597ad20201a --- /dev/null +++ b/bench/Libraries/Microsoft.Extensions.Telemetry.PerformanceTests/design.md @@ -0,0 +1,181 @@ +# Sampling and buffering performance tests + +## Goal + +Measure the incremental CPU and managed-memory cost of log sampling and buffering relative to the same logging pipeline without either feature. The sampling comparison must also show when dropping logs offsets the sampler's own cost. + +BenchmarkDotNet mean time per operation is used as the CPU-cost proxy. `MemoryDiagnoser` reports managed allocations and garbage collections per operation. These measurements intentionally do not represent process working set or machine-wide CPU utilization. + +Each benchmark invocation processes either 10,000 or 20,000 logs, representing one minute of traffic at the selected rate. BenchmarkDotNet reports time and allocation for the complete one-minute batch. Divide those values by `RecordsPerMinute` when a per-log value is needed. + +The shared deterministic workload uses four categories and sixteen event IDs. Event frequency is intentionally skewed: event 1 accounts for 70% of records, event 2 for 15%, event 3 for 8%, and the remaining 7% is spread across events 4 through 16. This exercises category/event rule caches and gives CCKR both frequent and rare callsites. + +.NET sampling benchmarks invoke each method eight times per iteration. Combined with the configured ten warmup iterations, each sampling benchmark process executes 80 warmup batches before measurement. Buffering and CCKR use one invocation per iteration so `IterationCleanup` empties retained state after every batch and each invocation continues to represent one minute of traffic. + +Benchmark worker processes disable tiered compilation with `DOTNET_TieredCompilation=0`. This makes every launch use optimized JIT-generated code from the start, avoiding tier-promotion differences between launches while retaining BenchmarkDotNet's ten warmup iterations. + +## Retained-memory measurement + +`--retained-memory` runs each pipeline outside BenchmarkDotNet, fills it with both record counts, and reports: + +- `Retained managed`: the change in live managed memory after a forced full collection. +- `Peak managed`: the largest observed managed heap increase while filling the pipeline, including objects that may later be collected. +- `Peak working set`: the largest observed increase in physical process memory. +- `Peak private bytes`: the largest observed increase in committed private process memory. + +Memory is sampled every 256 records. Managed retained memory is the most useful comparison for buffers and sampler caches. Working-set and private-byte deltas are process-level measurements and can be affected by runtime heap reservation, operating-system paging, and earlier scenarios in the same process. + +The benchmarks process the minute's traffic as a batch rather than sleeping between logs. This keeps wall-clock waiting out of the measurements and isolates logging pipeline cost. + +## Sampling benchmark + +`SamplingImpactBench` builds two otherwise identical logging pipelines: + +- `NoSampling` is the baseline and sends every log to `BenchLogger`. +- `WithSampling` adds one sampler before the same provider. + +Both methods use a cached `LoggerMessage` delegate with one primitive structured argument. This avoids call-site formatting and boxing allocations while still exercising the structured log state that passes through the sampling pipeline, making sampler-introduced allocations visible. + +The benchmark runs the following scenarios: + +| Scenario | Configuration | Purpose | +| --- | --- | --- | +| `RandomSampleAll` | Probability `1.0` | Measures random sampler overhead when provider work is unchanged. | +| `RandomSampleOnePercent` | Probability `0.01` | Represents a high-volume production configuration where most logs are discarded. | +| `RandomDropAll` | Probability `0.0` | Establishes the lower bound when all provider work is avoided. | +| `RandomByCategory` | High-volume categories `0.01`, critical categories `1.0`, fallback `0.1` | Measures category-rule selection with different retention policies. | +| `TraceSample` | Current activity has the `Recorded` flag | Measures trace-based sampler overhead when the log is retained. | +| `TraceDrop` | Current activity is not recorded | Measures trace-based sampling when provider work is avoided. | + +Each scenario is a separate BenchmarkDotNet parameter, so its `WithSampling` result is compared directly with a `NoSampling` baseline created under the same process and job settings. Setup, dependency injection, logger creation, and activity creation are outside the measured operations. + +## Buffering benchmark + +`BufferingImpactBench` builds two otherwise identical logging pipelines: + +- `NoBuffering` is the baseline and sends the selected record count directly to `BenchLogger`. +- `BufferOnly` measures serialization and insertion of the selected record count into the global buffer. Its flush runs during iteration cleanup and is excluded from the measurement. +- `BufferAndFlush` measures the end-to-end cost of buffering and then emitting the selected record count to `BenchLogger`. + +The buffer is sized to retain all batches from a measured iteration and automatic post-flush bypass is disabled. `BufferOnly` is flushed after each benchmark iteration, outside the measurement. This avoids capacity eviction and ensures every measured iteration starts with an empty buffer. + +## CCKR benchmark + +`CckrImpactBench` is available on the CCKR integration branch. CCKR combines the sampling decision and reservoir buffering in one pipeline, so its decision and buffer costs cannot be isolated through the public logging integration. + +- `NoSampling` is the baseline and sends every log directly to `BenchLogger`. +- `CckrRetainAll` gives the reservoir enough capacity for the selected record count and measures admission plus buffering with no drops. +- `CckrRetainAllAndFlush` adds emission of every retained log, making downstream provider work equivalent to the baseline. +- `CckrDisabled` measures the options-monitor and policy-check overhead of a registered CCKR pipeline with its kill switch disabled. +- `CckrRetainAllPolicy` measures the category-pattern bypass used for protected telemetry. +- `CckrAdaptive` uses a representative fixed capacity of 128 per category (up to 512 records across the four-category workload) for the selected one-minute period and measures the adaptive high-volume path without flush cost. +- `CckrAdaptiveAndFlush` includes emission of the adaptive reservoir at the period boundary. + +The novelty preserve is disabled so retained records are controlled only by the configured reservoir capacity. Automatic time-based flushing is moved beyond the benchmark duration, and iteration cleanup flushes both reservoirs outside the measurement. CCKR uses random ranks, so adaptive results should be interpreted from the full BenchmarkDotNet run rather than a single invocation. + +## Serialized-exporter benchmark + +`SerializedExporterImpactBench` is independent from the sampler and buffering microbenchmarks. It uses a dedicated provider that formats retained messages and serializes their category, level, EventId, event name, formatted message, exception, and structured attributes to UTF-8 JSON in a reusable buffer. It performs no terminal, disk, or network I/O. + +Each strategy is compared with a no-sampling pipeline using the same serialized exporter: + +- `RandomOnePercent` immediately exports approximately 1% of records. +- `RandomByCategory` applies the same 1% high-volume, 100% critical, and 10% fallback rules as the sampling microbenchmark. +- `TraceRetain` exports every record from a recorded activity. +- `TraceDrop` drops every record from an unrecorded activity. +- `CckrOnePercent` uses a per-category reservoir sized to retain approximately 1% across all four categories and includes the required flush and weighted-record serialization in the measured operation. + +This suite measures whether avoided formatting and serialization offset sampling overhead. The exporter retains the last serialized payload in its reusable output buffer, preserving all serialization work while keeping external I/O noise out of the results. + +`--serialized-exporter-volume` runs the same pipelines with exporter counters enabled and reports the actual number of records and UTF-8 bytes emitted. Counters are disabled in the timed benchmark so metric collection does not affect performance results. + +## Category-cardinality benchmark + +`CategoryCardinalityImpactBench` is an independent serialized-exporter suite for 20,000 records and 50, 100, or 200 categories. Both random sampling and CCKR use a 1% output budget. At 20,000 records each category receives enough traffic for an integral CCKR capacity of 4, 2, or 1 respectively. This measures rule-cache, logger, per-category reservoir, flush, and serialization scaling as category cardinality grows. + +## Multithreaded contention benchmark + +`MultithreadedSamplingImpactBench` processes 20,000 records across 100 shared categories using 1, 4, or 8 workers. Worker record indices are interleaved so workers concurrently target the same category instead of operating on disjoint category ranges. It uses the lightweight provider to emphasize sampler synchronization and compares random 1% with CCKR 1%, including the CCKR flush. + +## Sustained GC-pressure measurement + +`--sustained-gc` compares random 1% sampling with CCKR under a production-shaped, two-minute run at 10,000 logs per minute. Each strategy runs in an isolated worker process after pipeline warmup and a forced-GC baseline. CCKR uses its configurable 30-second automatic flush interval, producing four sampling periods, and both strategies use the serialized exporter with output counters enabled. + +An optional positive integer argument overrides the logging rate while retaining the two-minute duration and 30-second flush interval. A second optional argument overrides the CCKR flush interval in seconds. This supports higher-volume stress runs that generate enough allocation traffic to observe collections and compares the effect of record lifetime: + +```powershell +dotnet run -c Release --project .\bench\Libraries\Microsoft.Extensions.Telemetry.PerformanceTests\Microsoft.Extensions.Telemetry.PerformanceTests.csproj -- --sustained-gc 1000000 1 +``` + +The diagnostic reports input and emitted volume, exporter batch callbacks, total allocated bytes, Gen0/Gen1/Gen2 collection counts, cumulative GC pause time, process CPU time, CPU time as a percentage of wall time, peak managed-memory growth, and retained managed memory after a final forced collection. The final forced collection is performed after the collection counts, pause time, and CPU time are captured, so it does not contribute to those pressure measurements. Process working-set polling is intentionally left to the separate retained-memory diagnostic because repeatedly refreshing operating-system process counters would distort CPU measurements at this logging rate. + +At four categories, an exact 1% CCKR budget would require 12.5 records per category per 30-second period. The integer capacity is therefore 12 records per category, or 48 of each 5,000-record period (0.96%). Actual emitted volume is reported beside random sampling's probabilistic output. + +## Running + +From the repository root: + +```powershell +dotnet run -c Release --project .\bench\Libraries\Microsoft.Extensions.Telemetry.PerformanceTests\Microsoft.Extensions.Telemetry.PerformanceTests.csproj -- --filter *SamplingImpactBench* *BufferingImpactBench* *CckrImpactBench* +``` + +Run the independent serialized-exporter comparison: + +```powershell +dotnet run -c Release --project .\bench\Libraries\Microsoft.Extensions.Telemetry.PerformanceTests\Microsoft.Extensions.Telemetry.PerformanceTests.csproj -- --filter *SerializedExporterImpactBench* +``` + +Validate actual serialized output volume: + +```powershell +dotnet run -c Release --project .\bench\Libraries\Microsoft.Extensions.Telemetry.PerformanceTests\Microsoft.Extensions.Telemetry.PerformanceTests.csproj -- --serialized-exporter-volume +``` + +Run category-cardinality and multithreaded contention benchmarks: + +```powershell +dotnet run -c Release --project .\bench\Libraries\Microsoft.Extensions.Telemetry.PerformanceTests\Microsoft.Extensions.Telemetry.PerformanceTests.csproj -- --filter *CategoryCardinalityImpactBench* *MultithreadedSamplingImpactBench* +``` + +Measure retained and peak memory separately: + +```powershell +dotnet run -c Release --project .\bench\Libraries\Microsoft.Extensions.Telemetry.PerformanceTests\Microsoft.Extensions.Telemetry.PerformanceTests.csproj -- --retained-memory +``` + +Run the isolated two-minute-per-strategy sustained GC-pressure comparison: + +```powershell +dotnet run -c Release --project .\bench\Libraries\Microsoft.Extensions.Telemetry.PerformanceTests\Microsoft.Extensions.Telemetry.PerformanceTests.csproj -- --sustained-gc +``` + +Run on an otherwise idle machine with a fixed power plan. Compare `Mean`, `Ratio`, `Allocated`, and GC columns. Retain the generated BenchmarkDotNet artifacts with the machine, OS, runtime, and processor metadata when comparing changes over time. + +## Interpretation + +- `RandomSampleAll` and `TraceSample` isolate the cost of making a sampling decision because both paths still invoke the provider. +- `RandomDropAll` and `TraceDrop` show the best-case savings when sampling bypasses provider processing. +- `RandomSampleOnePercent` captures the combined decision cost and expected provider savings, but individual invocations are nondeterministic. BenchmarkDotNet's repeated operations provide the aggregate result. +- `RandomByCategory` retains approximately 28% of this evenly distributed four-category workload: 1% for two high-volume categories, 100% for the critical category, and 10% for the fallback category. +- Results depend on provider cost. `BenchLogger` is deliberately lightweight, so dropped-path savings are conservative relative to providers that format, serialize, buffer, or export logs. +- `BufferOnly` isolates the cost and managed allocations required to retain logs in memory. +- `BufferAndFlush` includes deserialization and downstream provider work, so it represents the complete buffering lifecycle. +- `CckrRetainAll` shows the combined admission and buffering overhead when sampling provides no volume reduction. +- `CckrAdaptive` shows when dropped-log savings offset reservoir decision cost, while the `AndFlush` variants include the cost of emitting retained records. + +## Follow-up plan + +1. Record a baseline on each supported performance-test platform. +2. Track accepted-path CPU and allocation regressions separately from dropped-path throughput gains. +3. Add a representative production exporter benchmark only if end-to-end exporter savings are needed; keep it separate so I/O and serialization do not hide sampler regressions. +4. Add multithreaded contention coverage if rule-cache or random-number generation changes, because this initial benchmark isolates steady-state single-thread overhead. + +## Acceptance criteria + +- Every sampling implementation has both a retained and discarded-log scenario. +- Each sampling result has an equivalent no-sampling baseline. +- Buffer insertion and buffer insertion plus flush are both compared with direct logging. +- CCKR covers retain-all and adaptive-drop paths, both before and through flush. +- Every measured invocation represents either 10,000 or 20,000 logs and reports the complete one-minute batch cost. +- Reports include time per operation, ratio, managed allocation per operation, and GC counts. +- Benchmark setup and the post-iteration `BufferOnly` flush do not contribute to measured CPU or allocation results. diff --git a/src/Libraries/Microsoft.Extensions.Telemetry/Sampling/Admission.cs b/src/Libraries/Microsoft.Extensions.Telemetry/Sampling/Admission.cs new file mode 100644 index 00000000000..606479f2c72 --- /dev/null +++ b/src/Libraries/Microsoft.Extensions.Telemetry/Sampling/Admission.cs @@ -0,0 +1,67 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. + +using System; + +namespace Microsoft.Extensions.Diagnostics.Sampling; + +/// +/// The result of an admission attempt. For an outcome it also +/// carries the EXP rank that must be handed back verbatim to +/// so the sampler can order its heap. +/// +internal readonly struct Admission : IEquatable +{ + private Admission(AdmissionKind kind, double key) + { + Kind = kind; + Key = key; + } + + /// + /// Gets a shared admission. + /// + public static Admission Skip { get; } = new(AdmissionKind.Skip, double.NaN); + + /// + /// Gets a shared admission. + /// + public static Admission Preserve { get; } = new(AdmissionKind.Preserve, double.NaN); + + /// + /// Gets the admission category. + /// + public AdmissionKind Kind { get; } + + /// + /// Gets the EXP rank -ln(u) / w_c used for bottom-K heap ordering. Only meaningful when + /// is ; otherwise . + /// + public double Key { get; } + + public static bool operator ==(Admission left, Admission right) + { + return left.Equals(right); + } + + public static bool operator !=(Admission left, Admission right) + { + return !left.Equals(right); + } + + /// + /// Creates an admission carrying its EXP rank. + /// + /// The EXP rank -ln(u) / w_c for heap ordering. + /// An admit admission. + public static Admission Admit(double key) => new(AdmissionKind.Admit, key); + + /// + public bool Equals(Admission other) => Kind == other.Kind && Key.Equals(other.Key); + + /// + public override bool Equals(object? obj) => obj is Admission other && Equals(other); + + /// + public override int GetHashCode() => (Kind, Key).GetHashCode(); +} diff --git a/src/Libraries/Microsoft.Extensions.Telemetry/Sampling/AdmissionKind.cs b/src/Libraries/Microsoft.Extensions.Telemetry/Sampling/AdmissionKind.cs new file mode 100644 index 00000000000..5f1653633ec --- /dev/null +++ b/src/Libraries/Microsoft.Extensions.Telemetry/Sampling/AdmissionKind.cs @@ -0,0 +1,29 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. + +namespace Microsoft.Extensions.Diagnostics.Sampling; + +/// +/// The category of an decision. +/// +internal enum AdmissionKind +{ + /// + /// The event must be dropped. The caller must not format the log record; this is the fast path + /// that yields the CPU and allocation savings. + /// + Skip, + + /// + /// The event was admitted into the statistical (bottom-K) sample. The caller must format the + /// payload and call with the admission. + /// + Admit, + + /// + /// The event was rejected by the statistical sample but accepted by the bounded novelty preserve + /// as a weight-0 observational record. The caller must format the payload and call + /// with the admission. + /// + Preserve, +} diff --git a/src/Libraries/Microsoft.Extensions.Telemetry/Sampling/Cckr.cs b/src/Libraries/Microsoft.Extensions.Telemetry/Sampling/Cckr.cs new file mode 100644 index 00000000000..1f1b3b6209d --- /dev/null +++ b/src/Libraries/Microsoft.Extensions.Telemetry/Sampling/Cckr.cs @@ -0,0 +1,430 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. + +using System; +using System.Collections.Generic; +using Microsoft.Shared.Diagnostics; + +namespace Microsoft.Extensions.Diagnostics.Sampling; + +#pragma warning disable CA5394 // Do not use insecure randomness - acceptable for the purposes of sampling + +/// +/// CCKR — the Chao-Cohen-Kaplan-Reservoir adaptive log sampler. +/// +/// +/// +/// CCKR is a bottom-(K+1) weighted reservoir sketch with exponential ranks (a WS-sketch, Cohen & +/// Kaplan 2007, built on the priority sampling of Duffield, Lund & Thorup 2007). Across periods it +/// feeds the previous period's per-callsite arrival counts back as inverse-frequency weights, so +/// chatty callsites are sampled hard while rare ones are kept. A Chao1 / Good-Turing estimate (Chao +/// 1984) weights as-yet-unseen callsites, and a bounded novelty preserve keeps one example of each +/// first-rejected callsite as a weight-0 observational record for tail coverage. +/// +/// +/// The type is single-threaded by design: use one instance per thread. The fast +/// path avoids formatting the payload for dropped events, which is the source of the CPU and +/// allocation savings. +/// +/// +/// The callsite identifier type (in production, the durable ID). +/// The formatted log payload type. +internal sealed class Cckr : ILogSampler + where TCallsite : notnull +{ + private const long DefaultMinPeriodCount = 32; + + private readonly int _reservoirCapacity; + private readonly int _preserveCapacity; + private readonly long _minPeriodCount; + private readonly UnseenWeightMode _unseenWeightMode; + private readonly Random _rng; + private readonly List _heap; + private readonly Dictionary _states; + + private Dictionary _freqPrev; + private Dictionary _freqCurr; + private double _samplingTau; + private long _seqCounter; + + /// + /// Initializes a new instance of the class using the + /// default preserve capacity (equal to ), the default Chao1 + /// stability threshold, , and an OS-derived seed. + /// + /// The sample size T per period. Must be at least 1. + public Cckr(int reservoirCapacity) + : this(reservoirCapacity, reservoirCapacity, DefaultMinPeriodCount, UnseenWeightMode.Chao1, null) + { + } + + /// + /// Initializes a new instance of the class with explicit + /// configuration. + /// + /// The sample size T per period. Must be at least 1. + /// The novelty-preserve capacity R (0 disables the preserve). Must not be negative. + /// The minimum prior-period arrival count below which the frozen table is discarded and the next period is treated as warmup. Must not be negative. + /// The strategy used to weight unseen callsites. + /// An optional RNG seed for deterministic behavior; uses an OS-derived seed. + public Cckr(int reservoirCapacity, int preserveCapacity, long minPeriodCount, UnseenWeightMode unseenWeightMode, int? seed) + { + _reservoirCapacity = Throw.IfLessThan(reservoirCapacity, 1); + _preserveCapacity = Throw.IfLessThan(preserveCapacity, 0); + _minPeriodCount = Throw.IfLessThan(minPeriodCount, 0); + _unseenWeightMode = unseenWeightMode; + _rng = seed.HasValue ? new Random(seed.Value) : new Random(); + _heap = new List(reservoirCapacity + 1); + _states = new Dictionary(reservoirCapacity + preserveCapacity); + _freqPrev = new Dictionary(reservoirCapacity); + _freqCurr = new Dictionary(reservoirCapacity); + _seqCounter = 0; + ReserveLength = 0; + Tau = double.PositiveInfinity; + _samplingTau = double.PositiveInfinity; + + // Until the first flush every callsite is "unseen" with weight 1.0, i.e. we behave as a + // uniform reservoir. + UnseenWeight = 1.0; + } + + /// + /// Gets the current threshold tau. Decreases monotonically within a period and resets to + /// at flush. Exposed for tests and metrics. + /// + public double Tau { get; private set; } + + /// + /// Gets the current unseen-callsite weight, frozen at the last flush. Exposed for tests and metrics. + /// + public double UnseenWeight { get; private set; } + + /// + /// Gets the number of callsites in the prior-period frozen table. + /// + public int FrozenCallsites => _freqPrev.Count; + + /// + /// Gets the current novelty-preserve occupancy. + /// + public int ReserveLength { get; private set; } + + /// + public Admission Admit(TCallsite callsite) + { + // Always count arrivals for the next period's frozen table, even when skipped. + _freqCurr[callsite] = _freqCurr.TryGetValue(callsite, out var current) ? current + 1 : 1; + + double wC = WeightFor(_freqPrev, callsite, UnseenWeight); + double u = _rng.NextDouble(); + if (u <= 0.0) + { + u = 1e-300; + } + + // EXP rank: -ln(U(0,1]) / w ~ Exp(w). See Cohen & Kaplan for the rank-function family. + double k = -Math.Log(u) / wC; + + if (k < Tau) + { + return Admission.Admit(k); + } + + // K-rejected. Consider the novelty preserve. Skip when the callsite is already represented + // anywhere in the state map (heap presence would violate disjointness; preserve presence is + // the first-rejection-wins rule). + if (ReserveLength < _preserveCapacity && !_states.ContainsKey(callsite)) + { + return Admission.Preserve; + } + + return Admission.Skip; + } + + /// + public void Insert(TCallsite callsite, Admission admission, TPayload payload) + { + switch (admission.Kind) + { + case AdmissionKind.Admit: + InsertAdmit(callsite, admission.Key, payload); + break; + + case AdmissionKind.Preserve: + InsertPreserve(callsite, payload); + break; + + default: + // Skip admissions must never reach Insert. + Throw.ArgumentException(nameof(admission), "Insert must not be called for a Skip admission."); + break; + } + } + + /// + public void FlushInto(ICollection> output) + { + _ = Throw.IfNull(output); + + double finalTau = _samplingTau; + + // (1) Drain the bottom-T heap with Horvitz-Thompson weights. + foreach (var entry in _heap) + { + double wC = WeightFor(_freqPrev, entry.Callsite, UnseenWeight); + double samplingCount; + if (double.IsInfinity(finalTau)) + { + samplingCount = 1.0; + } + else + { + // EXP-rank inclusion probability: pi = 1 - exp(-tau * w_c). + double pi = -Expm1(-finalTau * wC); + samplingCount = pi > 0.0 ? Math.Max(1.0, 1.0 / pi) : 1.0; + } + + output.Add(new SampledRecord(entry.Callsite, entry.Payload, samplingCount)); + } + + _heap.Clear(); + + // (2) Drain the preserve slots as weight-0 observational novelty records. Heap entries were + // already emitted above, and the heap/preserve disjointness invariant guarantees these + // callsites are not double-counted. + foreach (var pair in _states) + { + if (pair.Value.Preserve is { } preserve) + { + output.Add(new SampledRecord(pair.Key, preserve.Payload, 0.0)); + } + } + + _states.Clear(); + ReserveLength = 0; + + // (3) Period-boundary bookkeeping. + long observed = 0; + foreach (var value in _freqCurr.Values) + { + observed += value; + } + + if (observed < _minPeriodCount) + { + _freqPrev.Clear(); + UnseenWeight = 1.0; + } + else + { + UnseenWeight = ComputeUnseenWeight(); + (_freqPrev, _freqCurr) = (_freqCurr, _freqPrev); + } + + _freqCurr.Clear(); + Tau = double.PositiveInfinity; + _samplingTau = double.PositiveInfinity; + } + + /// + public List> Flush() + { + var output = new List>(_heap.Count + ReserveLength); + FlushInto(output); + return output; + } + + private static double WeightFor(Dictionary freqPrev, TCallsite callsite, double unseenWeight) + => freqPrev.TryGetValue(callsite, out var frequency) ? 1.0 / frequency : unseenWeight; + + /// + /// A netstandard2.0-safe exp(x) - 1 that stays accurate near zero, where the inclusion + /// probability would otherwise suffer catastrophic cancellation. + /// + /// The exponent. + /// exp(x) - 1. + private static double Expm1(double x) + { + if (Math.Abs(x) < 1e-5) + { + // Two-term Taylor series; the truncation error is O(x^3) which is negligible here. + return x + (0.5 * x * x); + } + + return Math.Exp(x) - 1.0; + } + + private void InsertAdmit(TCallsite callsite, double key, TPayload payload) + { + if (!_states.TryGetValue(callsite, out var state)) + { + state = new CallsiteState(); + _states[callsite] = state; + } + + // A heap admission supplants any pre-existing preserve slot for this callsite. + if (state.Preserve.HasValue) + { + state.Preserve = null; + ReserveLength--; + } + + state.HeapCount++; + HeapPush(new HeapEntry(key, callsite, payload)); + + if (_heap.Count > _reservoirCapacity) + { + HeapEntry evicted = HeapPopMax(); + _samplingTau = evicted.Key; + + // The evicted entry may be the one just pushed (when its key is the new maximum), in which + // case the increment and decrement cancel. + if (_states.TryGetValue(evicted.Callsite, out var evictedState)) + { + evictedState.HeapCount--; + if (evictedState.IsEmpty) + { + _ = _states.Remove(evicted.Callsite); + } + } + + // Admission uses the largest retained rank. Estimation separately uses the evicted + // (T+1)-th rank, which is the inclusion cutoff for the retained records. + Tau = _heap[0].Key; + } + } + + private void InsertPreserve(TCallsite callsite, TPayload payload) + { + long seq = _seqCounter; + _seqCounter++; + + // Only create the preserve slot when the callsite has no current heap or preserve entry: heap + // presence wins by disjointness, preserve presence by first-rejection-wins. + if (!_states.ContainsKey(callsite)) + { + _states[callsite] = new CallsiteState { Preserve = (payload, seq) }; + ReserveLength++; + } + } + + private double ComputeUnseenWeight() + { + if (_unseenWeightMode == UnseenWeightMode.RarestSeen) + { + // The rarest seen callsite has the smallest frequency, hence the largest weight. + long minFrequency = 0; + foreach (var value in _freqCurr.Values) + { + if (value > 0 && (minFrequency == 0 || value < minFrequency)) + { + minFrequency = value; + } + } + + if (minFrequency == 0) + { + return 1.0; + } + + double weight = 1.0 / minFrequency; + return weight < 1.0 ? weight : 1.0; + } + + return ChaoEstimator.Chao1UnseenWeight(_freqCurr.Values); + } + + private void HeapPush(HeapEntry entry) + { + _heap.Add(entry); + int i = _heap.Count - 1; + while (i > 0) + { + int parent = (i - 1) / 2; + if (_heap[parent].Key >= _heap[i].Key) + { + break; + } + + (_heap[parent], _heap[i]) = (_heap[i], _heap[parent]); + i = parent; + } + } + + private HeapEntry HeapPopMax() + { + HeapEntry root = _heap[0]; + int last = _heap.Count - 1; + _heap[0] = _heap[last]; + _heap.RemoveAt(last); + if (_heap.Count > 0) + { + SiftDown(0); + } + + return root; + } + + private void SiftDown(int start) + { + int i = start; + int count = _heap.Count; + while (true) + { + int left = (2 * i) + 1; + int right = (2 * i) + 2; + int largest = i; + + if (left < count && _heap[left].Key > _heap[largest].Key) + { + largest = left; + } + + if (right < count && _heap[right].Key > _heap[largest].Key) + { + largest = right; + } + + if (largest == i) + { + break; + } + + (_heap[i], _heap[largest]) = (_heap[largest], _heap[i]); + i = largest; + } + } + + /// + /// One heap entry. is the EXP rank; the containing list is maintained as a + /// max-heap so its root is the current threshold tau. + /// + private readonly struct HeapEntry + { + public HeapEntry(double key, TCallsite callsite, TPayload payload) + { + Key = key; + Callsite = callsite; + Payload = payload; + } + + public double Key { get; } + + public TCallsite Callsite { get; } + + public TPayload Payload { get; } + } + + /// + /// Per-callsite live state: reservoir multiplicity plus an optional novelty-preserve slot. The two + /// are mutually exclusive. + /// + private sealed class CallsiteState + { + public uint HeapCount { get; set; } + + public (TPayload Payload, long Seq)? Preserve { get; set; } + + public bool IsEmpty => HeapCount == 0 && !Preserve.HasValue; + } +} diff --git a/src/Libraries/Microsoft.Extensions.Telemetry/Sampling/CckrLogBuffer.cs b/src/Libraries/Microsoft.Extensions.Telemetry/Sampling/CckrLogBuffer.cs new file mode 100644 index 00000000000..c20ded32f9f --- /dev/null +++ b/src/Libraries/Microsoft.Extensions.Telemetry/Sampling/CckrLogBuffer.cs @@ -0,0 +1,491 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. +#if NET9_0_OR_GREATER + +using System; +using System.Collections.Concurrent; +using System.Collections.Generic; +using System.Diagnostics.CodeAnalysis; +using System.Threading; +using Microsoft.Extensions.Diagnostics.Buffering; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Options; +using Microsoft.Shared.Diagnostics; + +namespace Microsoft.Extensions.Diagnostics.Sampling; + +/// +/// Buffers records admitted by the CCKR sampler and emits weighted records at period boundaries. +/// +/// +/// Configuration is read through . Retain-all policies bypass +/// this buffer so protected records continue through the ordinary logging providers unchanged. +/// +[SuppressMessage("Usage", "CA2213:Disposable fields should be disposed", Justification = "The thread-local values do not own disposable resources and remain available for shutdown-time logging.")] +internal sealed class CckrLogBuffer : LogBuffer, IDisposable +{ + private readonly ConcurrentDictionary _categories = new(StringComparer.Ordinal); + private readonly IDisposable? _optionsChangeToken; + private readonly TimeProvider _timeProvider; + private readonly ThreadLocal _pending = new(); + private readonly object _flushClock = new(); + + private volatile ReservoirSamplingConfig _currentOptions; + private DateTimeOffset _lastFlush; + private int _disposed; + + public CckrLogBuffer(IOptionsMonitor options, TimeProvider timeProvider) + { + _ = Throw.IfNull(options); + _timeProvider = Throw.IfNull(timeProvider); + _currentOptions = Throw.IfMemberNull(options, options.CurrentValue); + _optionsChangeToken = options.OnChange(OnOptionsChanged); + _lastFlush = timeProvider.GetUtcNow(); + } + + public CckrLogBuffer(ReservoirSamplingConfig config, TimeProvider timeProvider) + : this(new FixedOptionsMonitor(Throw.IfNull(config)), timeProvider) + { + } + + /// + /// Determines whether a record should proceed through the logging pipeline and stores the + /// decision for the paired buffer operation. + /// + /// The logger category. + /// The log level. + /// The event identifier. + /// when the record must continue through the pipeline. + public bool Admit(string category, LogLevel logLevel, EventId eventId) + { + ReservoirSamplingConfig options = _currentOptions; + MaybeFlush(options.FlushInterval); + + PendingAdmission pending = CreateAdmission(category, logLevel, eventId, options); + _pending.Value = pending; + + return pending.Bypass || pending.Admission.Admission.Kind != AdmissionKind.Skip; + } + + /// + /// Determines whether an information-level record should proceed through the logging pipeline. + /// + /// The logger category. + /// The event identifier. + /// when the record must continue through the pipeline. + public bool Admit(string category, EventId eventId) + => Admit(category, LogLevel.Information, eventId); + + /// + public override bool TryEnqueue(IBufferedLogger bufferedLogger, in LogEntry logEntry) + { + PendingAdmission pending = _pending.Value; + _pending.Value = default; + + if (!pending.Matches(logEntry.Category, logEntry.LogLevel, logEntry.EventId)) + { + ReservoirSamplingConfig options = _currentOptions; + MaybeFlush(options.FlushInterval); + pending = CreateAdmission(logEntry.Category, logEntry.LogLevel, logEntry.EventId, options); + } + + if (pending.Bypass) + { + return false; + } + + if (pending.Admission.Admission.Kind == AdmissionKind.Skip) + { + return true; + } + + IReadOnlyList>? attributes = + logEntry.State as IReadOnlyList>; + if (attributes is null) + { + Throw.InvalidOperationException( + $"Unsupported type of log state detected: {typeof(TState)}, expected IReadOnlyList>"); + } + + SerializedLogRecord record = SerializedLogRecordFactory.Create( + logEntry.LogLevel, + logEntry.EventId, + _timeProvider.GetUtcNow(), + attributes, + logEntry.Exception, + logEntry.Formatter(logEntry.State, logEntry.Exception)); + + if (!pending.Reservoir!.Insert(bufferedLogger, pending.Admission, record)) + { + SerializedLogRecordFactory.Return(record); + } + + return true; + } + + /// + public override void Flush() + { + foreach (CategoryReservoir reservoir in _categories.Values) + { + reservoir.Flush(); + } + + lock (_flushClock) + { + _lastFlush = _timeProvider.GetUtcNow(); + } + } + + /// + /// Flushes retained records and releases per-thread admission state. + /// + public void Dispose() + { + if (Interlocked.Exchange(ref _disposed, 1) != 0) + { + return; + } + + Flush(); + _optionsChangeToken?.Dispose(); + } + + private static bool MatchesCategory(string category, string pattern) + { + int wildcard = pattern.IndexOf("*", StringComparison.Ordinal); + if (wildcard < 0) + { + return string.Equals(category, pattern, StringComparison.OrdinalIgnoreCase); + } + + return category.Length >= pattern.Length - 1 + && category.AsSpan().StartsWith(pattern.AsSpan(0, wildcard), StringComparison.OrdinalIgnoreCase) + && category.AsSpan().EndsWith(pattern.AsSpan(wildcard + 1), StringComparison.OrdinalIgnoreCase); + } + + private static bool MatchesAnyCategory(string category, IList patterns) + { + foreach (string pattern in patterns) + { + if (MatchesCategory(category, pattern)) + { + return true; + } + } + + return false; + } + + private static bool ShouldBypass( + string category, + LogLevel logLevel, + EventId eventId, + ReservoirSamplingConfig options) + { + if (!options.Enabled + || options.RetainAllLogLevels.Contains(logLevel) + || options.RetainAllEventIds.Contains(eventId.Id) + || MatchesAnyCategory(category, options.RetainAllCategories)) + { + return true; + } + + return options.SampledCategories.Count > 0 + && !MatchesAnyCategory(category, options.SampledCategories); + } + + private PendingAdmission CreateAdmission( + string category, + LogLevel logLevel, + EventId eventId, + ReservoirSamplingConfig options) + { + if (ShouldBypass(category, logLevel, eventId, options)) + { + return PendingAdmission.CreateBypass(category, logLevel, eventId.Id); + } + + CategoryReservoir reservoir = GetCategory(category); + CckrAdmission admission = reservoir.Admit(eventId, options); + return PendingAdmission.CreateAdaptive(category, logLevel, eventId.Id, reservoir, admission); + } + + private CategoryReservoir GetCategory(string category) + => _categories.GetOrAdd(category, static _ => new CategoryReservoir()); + + private void OnOptionsChanged(ReservoirSamplingConfig? options, string? name) + { + if (options is not null && string.IsNullOrEmpty(name)) + { + _currentOptions = options; + } + } + + private void MaybeFlush(TimeSpan flushInterval) + { + DateTimeOffset now = _timeProvider.GetUtcNow(); + lock (_flushClock) + { + if (now < _lastFlush + flushInterval) + { + return; + } + + _lastFlush = now; + } + + foreach (CategoryReservoir reservoir in _categories.Values) + { + reservoir.Flush(); + } + } + + private readonly struct CckrAdmission + { + public CckrAdmission(Admission admission, long generation) + { + Admission = admission; + Generation = generation; + } + + public Admission Admission { get; } + + public long Generation { get; } + } + + private readonly struct PendingAdmission + { + private PendingAdmission( + string category, + LogLevel logLevel, + int eventId, + bool bypass, + CategoryReservoir? reservoir, + CckrAdmission admission) + { + Category = category; + LogLevel = logLevel; + EventId = eventId; + Bypass = bypass; + Reservoir = reservoir; + Admission = admission; + } + + public CckrAdmission Admission { get; } + + public bool Bypass { get; } + + public string? Category { get; } + + public int EventId { get; } + + public LogLevel LogLevel { get; } + + public CategoryReservoir? Reservoir { get; } + + public static PendingAdmission CreateAdaptive( + string category, + LogLevel logLevel, + int eventId, + CategoryReservoir reservoir, + CckrAdmission admission) + => new(category, logLevel, eventId, false, reservoir, admission); + + public static PendingAdmission CreateBypass(string category, LogLevel logLevel, int eventId) + => new(category, logLevel, eventId, true, null, default); + + public bool Matches(string category, LogLevel logLevel, EventId eventId) + => Category is not null + && EventId == eventId.Id + && LogLevel == logLevel + && string.Equals(Category, category, StringComparison.Ordinal); + } + + private sealed class CategoryReservoir + { + private readonly object _lock = new(); + private Cckr? _reservoir; + private IBufferedLogger? _bufferedLogger; + private AlgorithmConfiguration _configuration; + private long _generation; + + public CckrAdmission Admit(EventId eventId, ReservoirSamplingConfig options) + { + List>? drained = null; + IBufferedLogger? bufferedLogger = null; + Admission admission; + long generation; + + lock (_lock) + { + AlgorithmConfiguration configuration = new(options); + if (_reservoir is null || !_configuration.Equals(configuration)) + { + if (_reservoir is not null) + { + drained = _reservoir.Flush(); + bufferedLogger = _bufferedLogger; + } + + _configuration = configuration; + _reservoir = configuration.CreateReservoir(); + _generation++; + } + + admission = _reservoir.Admit(eventId.Id); + generation = _generation; + } + + Emit(bufferedLogger, drained); + return new CckrAdmission(admission, generation); + } + + public bool Insert( + IBufferedLogger bufferedLogger, + CckrAdmission pending, + SerializedLogRecord record) + { + lock (_lock) + { + if (pending.Generation != _generation || pending.Admission.Kind == AdmissionKind.Skip) + { + return false; + } + + _bufferedLogger = bufferedLogger; + _reservoir!.Insert(record.EventId.Id, pending.Admission, record); + return true; + } + } + + public void Flush() + { + List>? drained; + IBufferedLogger? bufferedLogger; + + lock (_lock) + { + if (_reservoir is null) + { + return; + } + + bufferedLogger = _bufferedLogger; + drained = _reservoir.Flush(); + _generation++; + } + + Emit(bufferedLogger, drained); + } + + private static void Emit( + IBufferedLogger? bufferedLogger, + List>? drained) + { + if (bufferedLogger is null || drained is null || drained.Count == 0) + { + return; + } + + var records = new List(drained.Count); + foreach (SampledRecord sampled in drained) + { + SerializedLogRecord serialized = sampled.Payload; + var attributes = new List>(serialized.Attributes.Count + 1); + + int originalFormatIndex = serialized.Attributes.Count; + for (int i = 0; i < serialized.Attributes.Count; i++) + { + if (string.Equals(serialized.Attributes[i].Key, "{OriginalFormat}", StringComparison.Ordinal)) + { + originalFormatIndex = i; + break; + } + } + + for (int i = 0; i < originalFormatIndex; i++) + { + attributes.Add(serialized.Attributes[i]); + } + + attributes.Add(new KeyValuePair("sampling.count", sampled.SamplingCount)); + + for (int i = originalFormatIndex; i < serialized.Attributes.Count; i++) + { + attributes.Add(serialized.Attributes[i]); + } + + records.Add(new DeserializedLogRecord( + serialized.Timestamp, + serialized.LogLevel, + serialized.EventId, + serialized.Exception, + serialized.FormattedMessage, + attributes)); + } + + try + { + bufferedLogger.LogRecords(records); + } + finally + { + foreach (SampledRecord sampled in drained) + { + SerializedLogRecordFactory.Return(sampled.Payload); + } + } + } + } + + private readonly struct AlgorithmConfiguration : IEquatable + { + public AlgorithmConfiguration(ReservoirSamplingConfig options) + { + Capacity = options.Capacity; + PreserveCapacity = options.PreserveCapacity; + MinPeriodCount = options.MinPeriodCount; + UnseenWeightMode = options.UnseenWeightMode; + } + + public int Capacity { get; } + + public long MinPeriodCount { get; } + + public int PreserveCapacity { get; } + + public UnseenWeightMode UnseenWeightMode { get; } + + public Cckr CreateReservoir() + => new(Capacity, PreserveCapacity, MinPeriodCount, UnseenWeightMode, seed: null); + + public bool Equals(AlgorithmConfiguration other) + => Capacity == other.Capacity + && PreserveCapacity == other.PreserveCapacity + && MinPeriodCount == other.MinPeriodCount + && UnseenWeightMode == other.UnseenWeightMode; + + public override bool Equals(object? obj) + => obj is AlgorithmConfiguration other && Equals(other); + + public override int GetHashCode() + => HashCode.Combine(Capacity, PreserveCapacity, MinPeriodCount, UnseenWeightMode); + } + + private sealed class FixedOptionsMonitor : IOptionsMonitor + { + public FixedOptionsMonitor(ReservoirSamplingConfig value) + { + CurrentValue = value; + } + + public ReservoirSamplingConfig CurrentValue { get; } + + public ReservoirSamplingConfig Get(string? name) => CurrentValue; + + public IDisposable? OnChange(Action listener) => null; + } +} +#endif diff --git a/src/Libraries/Microsoft.Extensions.Telemetry/Sampling/CckrLoggingSampler.cs b/src/Libraries/Microsoft.Extensions.Telemetry/Sampling/CckrLoggingSampler.cs new file mode 100644 index 00000000000..2c9bc6d162f --- /dev/null +++ b/src/Libraries/Microsoft.Extensions.Telemetry/Sampling/CckrLoggingSampler.cs @@ -0,0 +1,30 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. +#if NET9_0_OR_GREATER + +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Shared.Diagnostics; + +namespace Microsoft.Extensions.Diagnostics.Sampling; + +/// +/// A that makes the CCKR admission decision at the sampling seam — +/// dropping records the reservoir rejects before they are buffered — and shares its reservoir +/// with the paired , which holds the admitted records and emits them, +/// weighted, at each flush. +/// +internal sealed class CckrLoggingSampler : LoggingSampler +{ + private readonly CckrLogBuffer _buffer; + + public CckrLoggingSampler(CckrLogBuffer buffer) + { + _buffer = Throw.IfNull(buffer); + } + + /// + public override bool ShouldSample(in LogEntry logEntry) + => _buffer.Admit(logEntry.Category, logEntry.LogLevel, logEntry.EventId); +} +#endif diff --git a/src/Libraries/Microsoft.Extensions.Telemetry/Sampling/CckrSamplingLoggingBuilderExtensions.cs b/src/Libraries/Microsoft.Extensions.Telemetry/Sampling/CckrSamplingLoggingBuilderExtensions.cs new file mode 100644 index 00000000000..a87fd423e7a --- /dev/null +++ b/src/Libraries/Microsoft.Extensions.Telemetry/Sampling/CckrSamplingLoggingBuilderExtensions.cs @@ -0,0 +1,102 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. +#if NET9_0_OR_GREATER + +using System; +using System.Linq; +using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.DependencyInjection.Extensions; +using Microsoft.Extensions.Diagnostics.Buffering; +using Microsoft.Extensions.Diagnostics.Sampling; +using Microsoft.Extensions.Options; +using Microsoft.Shared.Diagnostics; + +namespace Microsoft.Extensions.Logging; + +/// +/// Registers the CCKR adaptive log sampler, which reuses the existing logging pipeline seams: the +/// for the admit/drop decision and the for +/// holding admitted records and emitting them — weighted — at each period flush. +/// +public static class CckrSamplingLoggingBuilderExtensions +{ + /// + /// Adds the CCKR adaptive reservoir sampler to the logging infrastructure with default options. + /// + /// The logging builder. + /// The value of . + public static ILoggingBuilder AddCckrLogSampling(this ILoggingBuilder builder) + { + _ = Throw.IfNull(builder); + + return builder.AddCckrLogSamplingCore(); + } + + /// + /// Adds the CCKR adaptive reservoir sampler to the logging infrastructure. + /// + /// The logging builder. + /// The delegate used to configure CCKR. + /// The value of . + public static ILoggingBuilder AddCckrLogSampling( + this ILoggingBuilder builder, + Action configure) + { + _ = Throw.IfNull(builder); + _ = Throw.IfNull(configure); + + _ = builder.Services.Configure(configure); + + return builder.AddCckrLogSamplingCore(); + } + + /// + /// Adds the CCKR adaptive reservoir sampler to the logging infrastructure. + /// + /// The logging builder. + /// The configuration section used to configure CCKR. + /// The value of . + public static ILoggingBuilder AddCckrLogSampling( + this ILoggingBuilder builder, + IConfigurationSection section) + { + _ = Throw.IfNull(builder); + _ = Throw.IfNull(section); + + _ = builder.Services.Configure(section); + + return builder.AddCckrLogSamplingCore(); + } + + private static ILoggingBuilder AddCckrLogSamplingCore(this ILoggingBuilder builder) + { + bool cckrRegistered = builder.Services.Any(static descriptor => + descriptor.ServiceType == typeof(CckrLogBuffer)); + bool logBufferRegistered = builder.Services.Any(static descriptor => + descriptor.ServiceType == typeof(LogBuffer)); + + if (cckrRegistered) + { + return builder; + } + + if (logBufferRegistered) + { + Throw.InvalidOperationException( + "CCKR log sampling cannot be combined with another log buffer in the same logging pipeline."); + } + + _ = builder.Services + .AddOptionsWithValidateOnStart() + .Services.AddOptionsWithValidateOnStart(); + + builder.Services.TryAddSingleton(static services => new CckrLogBuffer( + services.GetRequiredService>(), + TimeProvider.System)); + builder.Services.TryAddSingleton(static sp => sp.GetRequiredService()); + + return builder.AddSampler(); + } +} +#endif diff --git a/src/Libraries/Microsoft.Extensions.Telemetry/Sampling/ChaoEstimator.cs b/src/Libraries/Microsoft.Extensions.Telemetry/Sampling/ChaoEstimator.cs new file mode 100644 index 00000000000..d29a24e5825 --- /dev/null +++ b/src/Libraries/Microsoft.Extensions.Telemetry/Sampling/ChaoEstimator.cs @@ -0,0 +1,87 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. + +using System.Collections.Generic; + +namespace Microsoft.Extensions.Diagnostics.Sampling; + +/// +/// Chao1 / Good-Turing estimators used to weight as-yet-unseen callsites so the tail of the +/// distribution is not systematically under-sampled. See Chao, Scand. J. Statist. (1984). +/// +internal static class ChaoEstimator +{ + /// + /// Computes the Good-Turing-derived unseen_weight from a sample of per-callsite frequencies + /// using Chao1's species-richness lower bound. The returned value is 1 / f_unseen where + /// f_unseen is the expected per-unseen-callsite frequency. Falls back to 1.0 (treat + /// unseen callsites as singletons) in degenerate cases. + /// + /// The per-callsite arrival counts observed in the period. + /// The weight to assign to callsites not seen in the frozen table. + public static double Chao1UnseenWeight(IEnumerable frequencies) + { + long n = 0; + long seen = 0; + long f1 = 0; + long f2 = 0; + foreach (var f in frequencies) + { + if (f == 0) + { + continue; + } + + n += f; + seen++; + if (f == 1) + { + f1++; + } + else if (f == 2) + { + f2++; + } + } + + if (n == 0 || f1 == 0) + { + return 1.0; + } + + double nf = n; + double f1f = f1; + double f2f = f2; + double seenf = seen; + + // Chao1 richness (lower bound on total number of distinct callsites). + double chao1 = f2 > 0 + ? seenf + (((nf - 1.0) / nf) * (f1f * f1f) / (2.0 * f2f)) + : seenf + (((nf - 1.0) / nf) * (f1f * (f1f - 1.0)) / 2.0); + + double unseenSpecies = chao1 - seenf; + if (unseenSpecies <= 0.0) + { + return 1.0; + } + + // Good-Turing missing-mass estimate: f1 / n. + double missingMass = f1f / nf; + if (missingMass <= 0.0) + { + return 1.0; + } + + double perUnseenProbability = missingMass / unseenSpecies; + double unseenFrequency = nf * perUnseenProbability; + if (double.IsNaN(unseenFrequency) || double.IsInfinity(unseenFrequency) || unseenFrequency <= 0.0) + { + return 1.0; + } + + // Cap at 1.0: an unseen callsite should never be weighted as if it had been seen more than + // once on average. + double weight = 1.0 / unseenFrequency; + return weight < 1.0 ? weight : 1.0; + } +} diff --git a/src/Libraries/Microsoft.Extensions.Telemetry/Sampling/ILogSampler.cs b/src/Libraries/Microsoft.Extensions.Telemetry/Sampling/ILogSampler.cs new file mode 100644 index 00000000000..14cce78f408 --- /dev/null +++ b/src/Libraries/Microsoft.Extensions.Telemetry/Sampling/ILogSampler.cs @@ -0,0 +1,55 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. + +using System.Collections.Generic; + +namespace Microsoft.Extensions.Diagnostics.Sampling; + +/// +/// A per-thread adaptive log-event sampler. Callers first ask whether +/// an event should be kept; only for a non- result do they format the +/// payload and call . Periodically the caller drains the current period via +/// / . +/// +/// +/// The callsite identifier type. In production this is the durable log identifier: a stable key per +/// logging statement is what makes adaptive, per-callsite sampling possible. +/// +/// The formatted log payload type. +internal interface ILogSampler + where TCallsite : notnull +{ + /// + /// Hot path: decide whether an event for should be kept. When the + /// result is the caller must drop the event without formatting + /// it. Otherwise the caller must format the payload and pass the returned admission verbatim to + /// . + /// + /// The callsite identifier (durable ID). + /// The admission decision. + Admission Admit(TCallsite callsite); + + /// + /// Store a formatted payload previously approved by . + /// + /// The callsite identifier. + /// The admission returned by . + /// The formatted payload. + void Insert(TCallsite callsite, Admission admission, TPayload payload); + + /// + /// Drain the current period's sample into a caller-supplied buffer, avoiding the per-flush + /// allocation of . The sum of + /// is an unbiased Horvitz-Thompson + /// estimator of the period's total arrival count. + /// + /// The buffer to append records to. + void FlushInto(ICollection> output); + + /// + /// Drain the current period's sample. Allocates a fresh list each call; use + /// to recycle a buffer. + /// + /// The sampled records for the period. + List> Flush(); +} diff --git a/src/Libraries/Microsoft.Extensions.Telemetry/Sampling/ReservoirSamplingConfig.cs b/src/Libraries/Microsoft.Extensions.Telemetry/Sampling/ReservoirSamplingConfig.cs new file mode 100644 index 00000000000..7dfcf627d68 --- /dev/null +++ b/src/Libraries/Microsoft.Extensions.Telemetry/Sampling/ReservoirSamplingConfig.cs @@ -0,0 +1,81 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. + +using System; +using System.Collections.Generic; +using System.ComponentModel.DataAnnotations; +using Microsoft.Extensions.Logging; + +namespace Microsoft.Extensions.Diagnostics.Sampling; + +/// +/// Provides configuration for the adaptive CCKR log sampler. +/// +public sealed class ReservoirSamplingConfig +{ + /// + /// Gets or sets a value indicating whether CCKR sampling is enabled. + /// + public bool Enabled { get; set; } = true; + + /// + /// Gets or sets the per-period reservoir capacity (T). + /// + [Range(1, int.MaxValue)] + public int Capacity { get; set; } = 128; + + /// + /// Gets or sets the per-period novelty-preserve capacity (R). 0 disables the preserve. + /// + [Range(0, int.MaxValue)] + public int PreserveCapacity { get; set; } = 128; + + /// + /// Gets or sets the minimum prior-period arrival count below which the frozen frequency table is + /// discarded and the next period is treated as warmup. + /// + [Range(0, long.MaxValue)] + public long MinPeriodCount { get; set; } = 32; + + /// + /// Gets or sets the period length. When this much time has elapsed the reservoir is flushed and a + /// new period begins. + /// + public TimeSpan FlushInterval { get; set; } = TimeSpan.FromSeconds(1); + + /// + /// Gets or sets the strategy used to weight callsites unseen in the frozen table. + /// + public UnseenWeightMode UnseenWeightMode { get; set; } = UnseenWeightMode.Chao1; + + /// + /// Gets or sets the log levels that bypass CCKR and are emitted normally. + /// + [Required] + public IList RetainAllLogLevels { get; set; } = [LogLevel.Error, LogLevel.Critical]; + + /// + /// Gets or sets category patterns that bypass CCKR and are emitted normally. + /// + /// + /// Matching is case-insensitive. A pattern can contain one * wildcard. + /// + [Required] + public IList RetainAllCategories { get; set; } = []; + + /// + /// Gets or sets category patterns eligible for CCKR sampling. + /// + /// + /// An empty collection applies CCKR to every category not covered by a retain-all policy. + /// Matching is case-insensitive. A pattern can contain one * wildcard. + /// + [Required] + public IList SampledCategories { get; set; } = []; + + /// + /// Gets or sets event identifiers that bypass CCKR and are emitted normally. + /// + [Required] + public IList RetainAllEventIds { get; set; } = []; +} diff --git a/src/Libraries/Microsoft.Extensions.Telemetry/Sampling/ReservoirSamplingConfigCustomValidator.cs b/src/Libraries/Microsoft.Extensions.Telemetry/Sampling/ReservoirSamplingConfigCustomValidator.cs new file mode 100644 index 00000000000..dae79a1ced1 --- /dev/null +++ b/src/Libraries/Microsoft.Extensions.Telemetry/Sampling/ReservoirSamplingConfigCustomValidator.cs @@ -0,0 +1,80 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. +#if NET9_0_OR_GREATER + +using System; +using System.Collections.Generic; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Options; + +namespace Microsoft.Extensions.Diagnostics.Sampling; + +/// +/// Validates CCKR configuration constraints that cannot be expressed with data annotations. +/// +internal sealed class ReservoirSamplingConfigCustomValidator : IValidateOptions +{ + public ValidateOptionsResult Validate(string? name, ReservoirSamplingConfig options) + { + ValidateOptionsResultBuilder result = new(); + + if (options.FlushInterval <= TimeSpan.Zero) + { + result.AddError("FlushInterval must be greater than zero.", nameof(options.FlushInterval)); + } + + if (!Enum.IsDefined(options.UnseenWeightMode)) + { + result.AddError("UnseenWeightMode must be a defined value.", nameof(options.UnseenWeightMode)); + } + + ValidateLogLevels(options.RetainAllLogLevels, result); + ValidateCategoryPatterns(options.RetainAllCategories, nameof(options.RetainAllCategories), result); + ValidateCategoryPatterns(options.SampledCategories, nameof(options.SampledCategories), result); + + return result.Build(); + } + + private static void ValidateLogLevels(IList? levels, ValidateOptionsResultBuilder result) + { + if (levels is null) + { + return; + } + + foreach (LogLevel level in levels) + { + if (!Enum.IsDefined(level)) + { + result.AddError("RetainAllLogLevels must contain only defined values.", nameof(ReservoirSamplingConfig.RetainAllLogLevels)); + } + } + } + + private static void ValidateCategoryPatterns( + IList? patterns, + string memberName, + ValidateOptionsResultBuilder result) + { + if (patterns is null) + { + return; + } + + foreach (string pattern in patterns) + { + if (string.IsNullOrWhiteSpace(pattern)) + { + result.AddError("Category patterns cannot be empty.", memberName); + continue; + } + + int wildcard = pattern.IndexOf("*", StringComparison.Ordinal); + if (wildcard >= 0 && pattern.IndexOf("*", wildcard + 1, StringComparison.Ordinal) >= 0) + { + result.AddError("Only one wildcard character is allowed in a category pattern.", memberName); + } + } + } +} +#endif diff --git a/src/Libraries/Microsoft.Extensions.Telemetry/Sampling/ReservoirSamplingConfigValidator.cs b/src/Libraries/Microsoft.Extensions.Telemetry/Sampling/ReservoirSamplingConfigValidator.cs new file mode 100644 index 00000000000..9bce307a46d --- /dev/null +++ b/src/Libraries/Microsoft.Extensions.Telemetry/Sampling/ReservoirSamplingConfigValidator.cs @@ -0,0 +1,16 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. +#if NET9_0_OR_GREATER + +using Microsoft.Extensions.Options; + +namespace Microsoft.Extensions.Diagnostics.Sampling; + +/// +/// Validates data annotations on . +/// +[OptionsValidator] +internal sealed partial class ReservoirSamplingConfigValidator : IValidateOptions +{ +} +#endif diff --git a/src/Libraries/Microsoft.Extensions.Telemetry/Sampling/SampledRecord.cs b/src/Libraries/Microsoft.Extensions.Telemetry/Sampling/SampledRecord.cs new file mode 100644 index 00000000000..ead4e3e61a4 --- /dev/null +++ b/src/Libraries/Microsoft.Extensions.Telemetry/Sampling/SampledRecord.cs @@ -0,0 +1,74 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. + +using System; +using System.Collections.Generic; + +namespace Microsoft.Extensions.Diagnostics.Sampling; + +/// +/// A single record produced when a sampling period is flushed. The is the +/// Horvitz-Thompson weight: summed across all records of a callsite it is an unbiased estimate of that +/// callsite's true arrival count for the period. +/// +/// The callsite identifier type (in production, the durable ID). +/// The formatted log payload type. +internal readonly struct SampledRecord : IEquatable> +{ + /// + /// Initializes a new instance of the struct. + /// + /// The callsite identifier. + /// The formatted payload. + /// + /// The Horvitz-Thompson weight. A value of 0 marks an observational novelty-preserve record + /// that does not contribute to count estimates. + /// + public SampledRecord(TCallsite callsite, TPayload payload, double samplingCount) + { + Callsite = callsite; + Payload = payload; + SamplingCount = samplingCount; + } + + /// + /// Gets the callsite identifier. + /// + public TCallsite Callsite { get; } + + /// + /// Gets the formatted payload. + /// + public TPayload Payload { get; } + + /// + /// Gets the Horvitz-Thompson sampling weight. A value greater than or equal to 1 means the + /// record stands in for that many events; a value of 0 is an observational novelty record + /// that does not contribute to estimates. + /// + public double SamplingCount { get; } + + public static bool operator ==(SampledRecord left, SampledRecord right) + { + return left.Equals(right); + } + + public static bool operator !=(SampledRecord left, SampledRecord right) + { + return !left.Equals(right); + } + + /// + public bool Equals(SampledRecord other) + { + return EqualityComparer.Default.Equals(Callsite, other.Callsite) + && EqualityComparer.Default.Equals(Payload, other.Payload) + && SamplingCount.Equals(other.SamplingCount); + } + + /// + public override bool Equals(object? obj) => obj is SampledRecord other && Equals(other); + + /// + public override int GetHashCode() => (Callsite, Payload, SamplingCount).GetHashCode(); +} diff --git a/src/Libraries/Microsoft.Extensions.Telemetry/Sampling/UnseenWeightMode.cs b/src/Libraries/Microsoft.Extensions.Telemetry/Sampling/UnseenWeightMode.cs new file mode 100644 index 00000000000..89de2ac32df --- /dev/null +++ b/src/Libraries/Microsoft.Extensions.Telemetry/Sampling/UnseenWeightMode.cs @@ -0,0 +1,22 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. + +namespace Microsoft.Extensions.Diagnostics.Sampling; + +/// +/// Strategy for weighting callsites that were not present in the previous period's frozen +/// frequency table. +/// +public enum UnseenWeightMode +{ + /// + /// Chao1 / Good-Turing missing-mass estimate. + /// + Chao1, + + /// + /// Rarest-seen rule: an unseen callsite is weighted the same as the rarest callsite already + /// observed (the inverse of the smallest observed frequency). + /// + RarestSeen, +} diff --git a/test/Libraries/Microsoft.Extensions.Telemetry.Tests/Sampling/CckrLogBufferTests.cs b/test/Libraries/Microsoft.Extensions.Telemetry.Tests/Sampling/CckrLogBufferTests.cs new file mode 100644 index 00000000000..00bde48fae4 --- /dev/null +++ b/test/Libraries/Microsoft.Extensions.Telemetry.Tests/Sampling/CckrLogBufferTests.cs @@ -0,0 +1,191 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. +#if NET9_0_OR_GREATER + +using System; +using System.Collections.Generic; +using System.Linq; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Diagnostics.Buffering; +using Microsoft.Extensions.Diagnostics.Enrichment; +using Microsoft.Extensions.Diagnostics.Sampling; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Logging.Test; +using Xunit; + +namespace Microsoft.Extensions.Telemetry.Sampling; + +public class CckrLogBufferTests +{ + private static readonly Func>, Exception?, string> _formatter = + static (_, _) => "message"; + + [Fact] + public void Admit_WhenIntervalElapsed_FlushesBeforeNewAdmission() + { + var timeProvider = new TestTimeProvider(); + var destination = new RecordingBufferedLogger(); + using var buffer = CreateBuffer(timeProvider); + + Enqueue(buffer, destination, [new("{OriginalFormat}", "message")]); + Assert.Empty(destination.Records); + + timeProvider.Advance(TimeSpan.FromSeconds(2)); + _ = buffer.Admit("category", new EventId(1)); + + Assert.Single(destination.Records); + } + + [Fact] + public void Dispose_FlushesRetainedRecords() + { + var destination = new RecordingBufferedLogger(); + var buffer = CreateBuffer(new TestTimeProvider()); + + Enqueue(buffer, destination, [new("{OriginalFormat}", "message")]); + buffer.Dispose(); + + Assert.Single(destination.Records); + } + + [Fact] + public void Flush_AddsSamplingCountBeforeOriginalFormat() + { + var destination = new RecordingBufferedLogger(); + using var buffer = CreateBuffer(new TestTimeProvider()); + + Enqueue( + buffer, + destination, + [ + new("property", 42), + new("{OriginalFormat}", "message {property}"), + ]); + buffer.Flush(); + + BufferedLogRecord record = Assert.Single(destination.Records); + Assert.Equal("sampling.count", record.Attributes[^2].Key); + Assert.Equal("{OriginalFormat}", record.Attributes[^1].Key); + Assert.True(double.IsFinite(Assert.IsType(record.Attributes[^2].Value))); + } + + [Fact] + public void LoggingPipeline_PreservesEnrichmentAndSupportsOrdinaryProvider() + { + var provider = new CapturingProvider(); + using ILoggerFactory factory = Utils.CreateLoggerFactory(builder => + { + builder.AddProvider(provider); + builder.Services.AddSingleton(new TestEnricher()); + builder.AddCckrLogSampling(options => + { + options.Capacity = 1; + options.PreserveCapacity = 0; + }); + }); + + ILogger logger = factory.CreateLogger("category"); + logger.LogInformation("message {property}", 42); + + var disposingFactory = Assert.IsType(factory); + disposingFactory.ServiceProvider.GetRequiredService().Flush(); + + IReadOnlyList> state = Assert.Single(provider.States); + Assert.Contains(state, pair => pair.Key == "enriched" && Equals(pair.Value, "value")); + Assert.Contains(state, pair => pair.Key == "sampling.count" && pair.Value is double weight && weight >= 1.0); + Assert.Equal("{OriginalFormat}", state[^1].Key); + } + + private static CckrLogBuffer CreateBuffer(TimeProvider timeProvider) + => new( + new ReservoirSamplingConfig + { + Capacity = 1, + PreserveCapacity = 0, + FlushInterval = TimeSpan.FromSeconds(1), + }, + timeProvider); + + private static void Enqueue( + CckrLogBuffer buffer, + IBufferedLogger destination, + IReadOnlyList> state) + { + var eventId = new EventId(1); + Assert.True(buffer.Admit("category", eventId)); + + var entry = new LogEntry>>( + LogLevel.Information, + "category", + eventId, + state, + null, + _formatter); + + Assert.True(buffer.TryEnqueue(destination, entry)); + } + + private sealed class RecordingBufferedLogger : IBufferedLogger + { + public List Records { get; } = []; + + public void LogRecords(IEnumerable records) + => Records.AddRange(records); + } + + private sealed class TestTimeProvider : TimeProvider + { + private DateTimeOffset _now = DateTimeOffset.UnixEpoch; + + public override DateTimeOffset GetUtcNow() => _now; + + public void Advance(TimeSpan value) => _now += value; + } + + private sealed class TestEnricher : ILogEnricher + { + public void Enrich(IEnrichmentTagCollector collector) + => collector.Add("enriched", "value"); + } + + private sealed class CapturingProvider : ILoggerProvider + { + public List>> States { get; } = []; + + public ILogger CreateLogger(string categoryName) => new CapturingLogger(this); + + public void Dispose() + { + } + + private sealed class CapturingLogger : ILogger + { + private readonly CapturingProvider _provider; + + public CapturingLogger(CapturingProvider provider) + { + _provider = provider; + } + + public IDisposable? BeginScope(TState state) + where TState : notnull => null; + + public bool IsEnabled(LogLevel logLevel) => true; + + public void Log( + LogLevel logLevel, + EventId eventId, + TState state, + Exception? exception, + Func formatter) + { + if (state is IReadOnlyList> attributes) + { + _provider.States.Add(attributes); + } + } + } + } +} +#endif diff --git a/test/Libraries/Microsoft.Extensions.Telemetry.Tests/Sampling/CckrTests.cs b/test/Libraries/Microsoft.Extensions.Telemetry.Tests/Sampling/CckrTests.cs new file mode 100644 index 00000000000..1c74e0b2890 --- /dev/null +++ b/test/Libraries/Microsoft.Extensions.Telemetry.Tests/Sampling/CckrTests.cs @@ -0,0 +1,132 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. +#if NET9_0_OR_GREATER + +using System; +using System.Collections.Generic; +using System.Linq; +using Microsoft.Extensions.Diagnostics.Sampling; +using Xunit; + +namespace Microsoft.Extensions.Telemetry.Sampling; + +public class CckrTests +{ + [Fact] + public void Flush_UsesFirstExcludedRankForSamplingCount() + { + const int Seed = 42; + var sampler = new Cckr(1, 0, 0, UnseenWeightMode.RarestSeen, Seed); + + Add(sampler, 0, 0); + Add(sampler, 0, 1); + + SampledRecord record = Assert.Single(sampler.Flush()); + + var random = new Random(Seed); + double firstRank = -Math.Log(random.NextDouble()); + double secondRank = -Math.Log(random.NextDouble()); + double firstExcludedRank = Math.Max(firstRank, secondRank); + double expectedWeight = 1.0 / (1.0 - Math.Exp(-firstExcludedRank)); + + Assert.Equal(expectedWeight, record.SamplingCount, 12); + Assert.Equal(firstRank < secondRank ? 0 : 1, record.Payload); + } + + [Fact] + public void Flush_WhenInputFitsCapacity_UsesUnitWeights() + { + var sampler = new Cckr(4, 0, 0, UnseenWeightMode.RarestSeen, 42); + + Add(sampler, 0, 0); + Add(sampler, 1, 1); + Add(sampler, 2, 2); + + List> records = sampler.Flush(); + + Assert.Equal(3, records.Count); + Assert.All(records, record => Assert.Equal(1.0, record.SamplingCount)); + } + + [Theory] + [InlineData(1, 0.08)] + [InlineData(8, 0.03)] + [InlineData(32, 0.02)] + [InlineData(128, 0.01)] + public void SamplingCount_MeanConvergesToUniformInput(int capacity, double tolerance) + { + const int Arrivals = 256; + const int Trials = 4_000; + double estimatedTotal = 0.0; + + for (int seed = 0; seed < Trials; seed++) + { + var sampler = new Cckr(capacity, 0, 0, UnseenWeightMode.RarestSeen, seed); + for (int i = 0; i < Arrivals; i++) + { + Add(sampler, 0, i); + } + + List> records = sampler.Flush(); + Assert.All(records, record => + { + Assert.True(double.IsFinite(record.SamplingCount)); + Assert.True(record.SamplingCount >= 1.0); + }); + + estimatedTotal += records.Sum(record => record.SamplingCount); + } + + double mean = estimatedTotal / Trials; + Assert.InRange(mean, Arrivals * (1.0 - tolerance), Arrivals * (1.0 + tolerance)); + } + + [Fact] + public void SamplingCount_MeanConvergesForSkewedKnownCallsites() + { + const int Trials = 2_000; + int[] arrivals = [900, 90, 10]; + double[] estimatedByCallsite = new double[arrivals.Length]; + + for (int seed = 0; seed < Trials; seed++) + { + var sampler = new Cckr(32, 0, 0, UnseenWeightMode.RarestSeen, seed); + + AddPeriod(sampler, arrivals); + _ = sampler.Flush(); + + AddPeriod(sampler, arrivals); + foreach (SampledRecord record in sampler.Flush()) + { + estimatedByCallsite[record.Callsite] += record.SamplingCount; + } + } + + for (int callsite = 0; callsite < arrivals.Length; callsite++) + { + double mean = estimatedByCallsite[callsite] / Trials; + Assert.InRange(mean, arrivals[callsite] * 0.9, arrivals[callsite] * 1.1); + } + } + + private static void AddPeriod(Cckr sampler, int[] arrivals) + { + for (int callsite = 0; callsite < arrivals.Length; callsite++) + { + for (int i = 0; i < arrivals[callsite]; i++) + { + Add(sampler, callsite, i); + } + } + } + + private static void Add(Cckr sampler, int callsite, int payload) + { + Admission admission = sampler.Admit(callsite); + if (admission.Kind != AdmissionKind.Skip) + { + sampler.Insert(callsite, admission, payload); + } + } +} +#endif