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..f92182347aa --- /dev/null +++ b/src/Libraries/Microsoft.Extensions.Telemetry/Sampling/Cckr.cs @@ -0,0 +1,426 @@ +// 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 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; + + // 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 = Tau; + + // (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; + } + + /// + 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(); + + // 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); + } + } + + // The (T+1)-th smallest rank is gone; the new root is the largest of the remaining T + // smallest, which is the new threshold. + 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..a7147343db4 --- /dev/null +++ b/src/Libraries/Microsoft.Extensions.Telemetry/Sampling/CckrLogBuffer.cs @@ -0,0 +1,238 @@ +// 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.Threading; +using Microsoft.Extensions.Diagnostics.Buffering; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Shared.Diagnostics; + +namespace Microsoft.Extensions.Diagnostics.Sampling; + +/// +/// A implementation backed by the CCKR adaptive reservoir. It plugs into the +/// existing logging pipeline through the standard buffer seam: holds an +/// admitted record in a per-category reservoir instead of writing it, and emits +/// the period's kept records — each carrying its Horvitz-Thompson sampling.count weight +/// — through the same callback the global buffer uses. +/// +/// +/// The paired makes the admission decision at the +/// seam and stashes the result for this thread; +/// reuses it so the reservoir is consulted once per record. When used without that sampler, +/// makes the admission decision itself. +/// +internal sealed class CckrLogBuffer : LogBuffer, IDisposable +{ + private readonly ConcurrentDictionary _categories = new(StringComparer.Ordinal); + private readonly ReservoirSamplingConfig _config; + private readonly TimeProvider _timeProvider; + private readonly ThreadLocal _pending = new(); + private readonly object _flushClock = new(); + + private DateTimeOffset _nextFlush; + + public CckrLogBuffer(ReservoirSamplingConfig config, TimeProvider timeProvider) + { + _config = config; + _timeProvider = timeProvider; + _nextFlush = timeProvider.GetUtcNow() + config.FlushInterval; + } + + /// + /// Makes and records this thread's admission decision for a callsite. Called from the paired + /// at the sampling seam, before . + /// + /// if the record should be processed and held; otherwise . + public bool Admit(string category, EventId eventId) + { + CategoryReservoir reservoir = GetCategory(category); + Admission admission = reservoir.Admit(eventId); + _pending.Value = new PendingAdmission(category, eventId.Id, admission); + return admission.Kind != AdmissionKind.Skip; + } + + /// + public override bool TryEnqueue(IBufferedLogger bufferedLogger, in LogEntry logEntry) + { + string category = logEntry.Category; + CategoryReservoir reservoir = GetCategory(category); + + // Reuse the admission computed by the paired sampler on this thread; otherwise decide now. + Admission admission; + PendingAdmission pending = _pending.Value; + if (pending.HasValue && pending.EventId == logEntry.EventId.Id && string.Equals(pending.Category, category, StringComparison.Ordinal)) + { + admission = pending.Admission; + _pending.Value = default; + } + else + { + admission = reservoir.Admit(logEntry.EventId); + } + + if (admission.Kind == AdmissionKind.Skip) + { + // Consumed by the reservoir (counted) but not kept: drop without writing. + MaybeFlush(); + 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)); + + reservoir.Insert(bufferedLogger, logEntry.EventId, admission, record); + + MaybeFlush(); + return true; + } + + /// + public override void Flush() + { + foreach (CategoryReservoir reservoir in _categories.Values) + { + reservoir.Flush(); + } + + lock (_flushClock) + { + _nextFlush = _timeProvider.GetUtcNow() + _config.FlushInterval; + } + } + + public void Dispose() => _pending.Dispose(); + + private CategoryReservoir GetCategory(string category) + => _categories.GetOrAdd(category, static (_, cfg) => new CategoryReservoir(cfg), _config); + + private void MaybeFlush() + { + DateTimeOffset now = _timeProvider.GetUtcNow(); + lock (_flushClock) + { + if (now < _nextFlush) + { + return; + } + + _nextFlush = now + _config.FlushInterval; + } + + foreach (CategoryReservoir reservoir in _categories.Values) + { + reservoir.Flush(); + } + } + + /// + /// This thread's admission decision, carried from the sampler seam to . + /// + private readonly struct PendingAdmission + { + public PendingAdmission(string category, int eventId, Admission admission) + { + Category = category; + EventId = eventId; + Admission = admission; + } + + public bool HasValue => Category is not null; + + public string? Category { get; } + + public int EventId { get; } + + public Admission Admission { get; } + } + + /// + /// One category's reservoir plus the buffered-logger callback used to emit its flushed records. + /// + private sealed class CategoryReservoir + { + private readonly Cckr _reservoir; + private readonly object _lock = new(); + private IBufferedLogger? _bufferedLogger; + + public CategoryReservoir(ReservoirSamplingConfig config) + { + _reservoir = new Cckr( + config.Capacity, + config.PreserveCapacity, + config.MinPeriodCount, + config.UnseenWeightMode, + seed: null); + } + + public Admission Admit(EventId eventId) + { + lock (_lock) + { + return _reservoir.Admit(eventId.Id); + } + } + + public void Insert(IBufferedLogger bufferedLogger, EventId eventId, Admission admission, SerializedLogRecord record) + { + lock (_lock) + { + _bufferedLogger = bufferedLogger; + _reservoir.Insert(eventId.Id, admission, record); + } + } + + public void Flush() + { + List> drained; + IBufferedLogger? bufferedLogger; + lock (_lock) + { + bufferedLogger = _bufferedLogger; + drained = _reservoir.Flush(); + } + + if (bufferedLogger 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); + attributes.AddRange(serialized.Attributes); + attributes.Add(new KeyValuePair("sampling.count", sampled.SamplingCount)); + + records.Add(new DeserializedLogRecord( + serialized.Timestamp, + serialized.LogLevel, + serialized.EventId, + serialized.Exception, + serialized.FormattedMessage, + attributes)); + } + + bufferedLogger.LogRecords(records); + } + } +} +#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..f94b7481695 --- /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.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..ec687cccd2b --- /dev/null +++ b/src/Libraries/Microsoft.Extensions.Telemetry/Sampling/CckrSamplingLoggingBuilderExtensions.cs @@ -0,0 +1,43 @@ +// 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 Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.DependencyInjection.Extensions; +using Microsoft.Extensions.Diagnostics.Buffering; +using Microsoft.Extensions.Diagnostics.Sampling; +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. Registers a single + /// reservoir as both the pipeline's and its . + /// + /// The logging builder. + /// An optional delegate to configure the reservoir. + /// The value of . + public static ILoggingBuilder AddCckrLogSampling(this ILoggingBuilder builder, Action? configure = null) + { + _ = Throw.IfNull(builder); + + var config = new ReservoirSamplingConfig(); + configure?.Invoke(config); + + // Register one reservoir instance and expose it through both pipeline seams. The DI container + // owns its lifetime (and disposal); the LoggingSampler resolves the same instance. + builder.Services.TryAddSingleton(_ => new CckrLogBuffer(config, 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..cbfddf45b2e --- /dev/null +++ b/src/Libraries/Microsoft.Extensions.Telemetry/Sampling/ReservoirSamplingConfig.cs @@ -0,0 +1,39 @@ +// 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; + +/// +/// Configuration for the adaptive (CCKR) log reservoir sampler wired into the logging pipeline. +/// +public sealed class ReservoirSamplingConfig +{ + /// + /// Gets or sets the per-period reservoir capacity (T). + /// + public int Capacity { get; set; } = 128; + + /// + /// Gets or sets the per-period novelty-preserve capacity (R). 0 disables the preserve. + /// + 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. + /// + 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; +} 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, +}