Skip to content
Merged
Show file tree
Hide file tree
Changes from 3 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions src/OpenTelemetry/CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -58,6 +58,10 @@ Notes](../../RELEASENOTES.md).
underlying `MeterListener`).
([#7924](https://github.com/open-telemetry/opentelemetry-dotnet/pull/7924))

* Improved metric collection performance after metric streams are removed by
compacting metric storage during collection.
([#7945](https://github.com/open-telemetry/opentelemetry-dotnet/pull/7945))

## 1.19.1

Released 2026-Sep-21
Expand Down
46 changes: 45 additions & 1 deletion src/OpenTelemetry/Metrics/Reader/MetricReaderExt.cs
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,9 @@ internal static void DeactivateMetric(Metric metric)
// during the next collect/export. The freed storage slot is then
// reused for a subsequently published instrument (see
// TryReserveMetricIndex), so repeatedly creating and disposing
// meters/instruments does not exhaust the metric storage.
// meters/instruments does not exhaust the metric storage. Once the
// removal is processed, the array is compacted so that collection
// keeps scanning only live streams (see CompactMetrics).
metric.Active = false;

OpenTelemetrySdkEventSource.Log.MetricInstrumentDeactivated(
Expand Down Expand Up @@ -272,6 +274,14 @@ private Batch<Metric> GetMetricsBatch()
}
}

// Reclaim the free slots left by removed metrics so that subsequent
// scans only visit live streams. Runs only on the collect path, so
// no other scan can observe the array while entries are moved.
if (!this.availableMetricIndices.IsEmpty)
{
this.CompactMetrics();
}

return (metricCountCurrentBatch > 0) ? new Batch<Metric>(this.metricsCurrentBatch, metricCountCurrentBatch) : default;
}
catch (Exception ex)
Expand Down Expand Up @@ -331,6 +341,40 @@ private void RemoveMetric(ref Metric? metric, int index)
this.availableMetricIndices.Enqueue(index);
}

private void CompactMetrics()
{
// Note: Holding instrumentCreationLock keeps AddMetricWith*Views from
Comment thread
matt-hensley marked this conversation as resolved.
Resolved
// writing to the storage array while entries are moved. Preserving the
// relative order keeps exported metric ordering stable.
lock (this.instrumentCreationLock)
{
var highIndex = this.metricIndex;
var writeIndex = 0;
for (var readIndex = 0; readIndex <= highIndex; readIndex++)
{
var metric = this.metrics[readIndex];
if (metric != null)
{
this.metrics[writeIndex++] = metric;
}
}

// Clear the vacated tail so stale references are not retained.
for (var i = writeIndex; i <= highIndex; i++)
{
this.metrics[i] = null;
}

this.metricIndex = writeIndex - 1;

// Every free slot was below the old high-water mark and has been
// compacted away, so the free list no longer refers to valid slots.
while (this.availableMetricIndices.TryDequeue(out _))
{
}
}
}

private bool TryReserveMetricIndex(out int index)
{
// Note: This is always called while holding instrumentCreationLock, so
Expand Down
171 changes: 171 additions & 0 deletions test/Benchmarks/Metrics/MetricCollectChurnBenchmarks.cs
Original file line number Diff line number Diff line change
@@ -0,0 +1,171 @@
// Copyright The OpenTelemetry Authors
// SPDX-License-Identifier: Apache-2.0

using System.Diagnostics.Metrics;
using BenchmarkDotNet.Attributes;
using OpenTelemetry;
using OpenTelemetry.Metrics;
using OpenTelemetry.Tests;

namespace Benchmarks.Metrics;

#pragma warning disable CA1001 // Types that own disposable fields should be disposed by GlobalCleanup.
[MemoryDiagnoser]
public class MetricCollectChurnBenchmarks
#pragma warning restore CA1001
{
private const string ChurnMeterName = "MetricCollectChurnBenchmarks.Churn";
private const string CleanMeterName = "MetricCollectChurnBenchmarks.Clean";
private const int ActiveMetricCount = 4;

private readonly HashSet<string> churnedExportedMetricNames = new(StringComparer.Ordinal);
private readonly HashSet<string> cleanExportedMetricNames = new(StringComparer.Ordinal);
private readonly List<Meter> meters = [];
private Counter<long>[]? churnedCounters;
private Counter<long>[]? cleanCounters;
private BaseExportingMetricReader? churnedReader;
private BaseExportingMetricReader? cleanReader;
private MeterProvider? churnedProvider;
private MeterProvider? cleanProvider;

[Params(16, 256, 1024)]
public int HistoricalMetricCount { get; set; }

[Params(MetricReaderTemporalityPreference.Cumulative, MetricReaderTemporalityPreference.Delta)]
public MetricReaderTemporalityPreference TemporalityPreference { get; set; }

[GlobalSetup]
public void Setup()
{
this.churnedProvider = this.CreateProvider(ChurnMeterName, this.churnedExportedMetricNames, out var churnedReader, this.HistoricalMetricCount + ActiveMetricCount);
this.churnedReader = churnedReader;
this.cleanProvider = this.CreateProvider(CleanMeterName, this.cleanExportedMetricNames, out var cleanReader, ActiveMetricCount);
this.cleanReader = cleanReader;

// Publish many instruments together so the reader's high-water index grows
// before any slots are released. Disposing the meter completes its
// instruments; the next collection removes them and makes their slots
// available for reuse.
var churnMeter = new Meter(ChurnMeterName);
this.meters.Add(churnMeter);
for (var i = 0; i < this.HistoricalMetricCount; i++)
{
var counter = churnMeter.CreateCounter<long>($"churn_{i}");
counter.Add(1);
}

churnMeter.Dispose();
this.churnedReader.Collect();
if (this.churnedExportedMetricNames.Count != this.HistoricalMetricCount)
{
throw new InvalidOperationException("The churned provider must export every historical metric before removing the disposed meter's streams.");
}

this.churnedExportedMetricNames.Clear();

// Publish the active workload only after the churned streams have been
// removed. The same workload is installed in a fresh provider below.
var activeMeter = new Meter(ChurnMeterName);
this.meters.Add(activeMeter);
this.churnedCounters = CreateActiveCounters(activeMeter);
var cleanMeter = new Meter(CleanMeterName);
this.meters.Add(cleanMeter);
this.cleanCounters = CreateActiveCounters(cleanMeter);

foreach (var counter in this.churnedCounters)
{
counter.Add(1);
}

foreach (var counter in this.cleanCounters)
{
counter.Add(1);
}

this.churnedReader.Collect();
this.cleanReader.Collect();

// Verify via the exporter that the churned provider retained exactly
// the active workload after filling, disposing, and recycling slots.
if (!HasActiveMetricNames(this.churnedExportedMetricNames)
|| !HasActiveMetricNames(this.cleanExportedMetricNames))
{
throw new InvalidOperationException("The churned and clean providers must each export the configured active metrics.");
}
}

[GlobalCleanup]
public void Cleanup()
{
foreach (var meter in this.meters)
{
meter.Dispose();
}

this.churnedProvider?.Dispose();
this.cleanProvider?.Dispose();
}

[Benchmark(Baseline = true)]
public bool CollectWithCleanProvider()
{
return this.cleanReader!.Collect();
}

[Benchmark]
public bool CollectWithHistoricalChurn()
{
return this.churnedReader!.Collect();
}

private static Counter<long>[] CreateActiveCounters(Meter meter)
{
var counters = new Counter<long>[ActiveMetricCount];
for (var i = 0; i < counters.Length; i++)
{
counters[i] = meter.CreateCounter<long>($"active_{i}");
}

return counters;
}

private static bool HasActiveMetricNames(HashSet<string> exportedMetricNames)
{
if (exportedMetricNames.Count != ActiveMetricCount)
{
return false;
}

for (var i = 0; i < ActiveMetricCount; i++)
{
if (!exportedMetricNames.Contains($"active_{i}"))
{
return false;
}
}

return true;
}

private MeterProvider CreateProvider(string meterName, HashSet<string> exportedMetricNames, out BaseExportingMetricReader reader, int maxMetricStreams)
{
#pragma warning disable CA2000 // The provider owns and disposes the reader and exporter.
reader = new BaseExportingMetricReader(new TestExporter<Metric>(batch =>
{
foreach (var metric in batch)
{
exportedMetricNames.Add(metric.Name);
}
}))
{
TemporalityPreference = this.TemporalityPreference,
};
#pragma warning restore CA2000

return Sdk.CreateMeterProviderBuilder()
.AddMeter(meterName)
.SetMaxMetricStreams(maxMetricStreams)
.AddReader(reader)
.Build();
}
}
91 changes: 91 additions & 0 deletions test/OpenTelemetry.Tests/Metrics/MeterProviderSdkTests.cs
Original file line number Diff line number Diff line change
Expand Up @@ -214,6 +214,97 @@ public void TransientMetricStreamNameRegistrationsAreReleasedTest()
Assert.Empty(metricStreamNames);
}

[Fact]
public void TransientMetricStorageIsCompactedAfterCollectTest()
{
var meterName = Utils.GetCurrentMethodName();
var exportedItems = new List<Metric>();

const int MaxMetricStreams = 16;

using var meterProvider = Sdk.CreateMeterProviderBuilder()
.SetMaxMetricStreams(MaxMetricStreams)
.AddMeter(meterName)
.AddInMemoryExporter(exportedItems)
.Build() as MeterProviderSdk;

Assert.NotNull(meterProvider);

// Publish many instruments so the storage high-water mark grows, then
// dispose them all and collect.
using (var meter = new Meter(meterName))
{
for (var i = 0; i < MaxMetricStreams; i++)
{
meter.CreateCounter<int>($"Counter{i}").Add(1);
}
}

meterProvider.ForceFlush();

// Storage must be compacted to the live streams (none), so collection
// does not keep scanning the slots freed by the removed instruments.
var metricIndexField = typeof(MetricReader).GetField("metricIndex", BindingFlags.Instance | BindingFlags.NonPublic);
Assert.NotNull(metricIndexField);
Assert.Equal(-1, Assert.IsType<int>(metricIndexField.GetValue(meterProvider.Reader)));

// The reclaimed storage must still accept new instruments.
exportedItems.Clear();
using (var meter = new Meter(meterName))
{
meter.CreateCounter<int>("Counter").Add(1);
meterProvider.ForceFlush();
}

var item = Assert.Single(exportedItems);
Assert.Equal("Counter", item.Name);
}

[Fact]
public void LiveMetricSurvivesStorageCompactionTest()
{
var meterName = Utils.GetCurrentMethodName();
var exportedItems = new List<Metric>();

const int MaxMetricStreams = 16;
const int TransientCount = 8;

using var meterProvider = Sdk.CreateMeterProviderBuilder()
.SetMaxMetricStreams(MaxMetricStreams)
.AddMeter(meterName)
.AddInMemoryExporter(exportedItems)
.Build() as MeterProviderSdk;

Assert.NotNull(meterProvider);

// The live instrument is published after the transient ones, so it
// occupies a high storage index that compaction must move down.
var transientMeter = new Meter(meterName);
for (var i = 0; i < TransientCount; i++)
{
transientMeter.CreateCounter<int>($"Transient{i}").Add(1);
}

using var liveMeter = new Meter(meterName);
var liveCounter = liveMeter.CreateCounter<int>("Live");
liveCounter.Add(1);

transientMeter.Dispose();
meterProvider.ForceFlush();

var metricIndexField = typeof(MetricReader).GetField("metricIndex", BindingFlags.Instance | BindingFlags.NonPublic);
Assert.NotNull(metricIndexField);
Assert.Equal(0, Assert.IsType<int>(metricIndexField.GetValue(meterProvider.Reader)));

// The live instrument must still be exported after it has been moved.
exportedItems.Clear();
liveCounter.Add(1);
meterProvider.ForceFlush();

var item = Assert.Single(exportedItems);
Assert.Equal("Live", item.Name);
}

[Fact]
public void MetricsListenerDoesNotKeepMeterProviderAliveAfterDisposeTest()
{
Expand Down
Loading