[OpenTelementry] Fix observable callbacks

src/OpenTelemetry/CHANGELOG.md

@@ -10,6 +10,10 @@ Notes](../../RELEASENOTES.md).
   variables.
   ([#7187](https://github.com/open-telemetry/opentelemetry-dotnet/pull/7187))
 
+* Fix observable instrument callbacks running once per reader instead of
+  once per collection cycle.
+  ([#7188](https://github.com/open-telemetry/opentelemetry-dotnet/pull/7188))
+
 ## 1.15.3
 
 Released 2026-Apr-21

src/OpenTelemetry/Metrics/MeterProviderSdk.cs

@@ -94,6 +94,8 @@ internal MeterProviderSdk(
                 this.Reader = new CompositeMetricReader([this.Reader, reader]);
             }
 
+            this.Reader.SetParentProvider(this);
+
             if (reader is PeriodicExportingMetricReader periodicExportingMetricReader)
             {
                 exportersAdded.Append(periodicExportingMetricReader.Exporter);

src/OpenTelemetry/Metrics/Reader/BaseExportingMetricReader.cs

@@ -103,6 +103,42 @@ internal override bool ProcessMetrics(in Batch<Metric> metrics, int timeoutMilli
         }
     }
 
+    /// <inheritdoc />
+    internal override bool OnCollectFromComposite(int timeoutMilliseconds)
+    {
+        if (this.SupportedExportModes.HasFlag(ExportModes.Push))
+        {
+            return base.OnCollectFromComposite(timeoutMilliseconds);
+        }
+        else if (this.SupportedExportModes.HasFlag(ExportModes.Pull) && PullMetricScope.IsPullAllowed)
+        {
+            return base.OnCollectFromComposite(timeoutMilliseconds);
+        }
+
+        return false;
+    }
+
+    /// <inheritdoc />
+    internal override bool OnShutdownFromComposite(int timeoutMilliseconds)
+    {
+        var result = true;
+
+        if (timeoutMilliseconds == Timeout.Infinite)
+        {
+            result = this.CollectFromComposite(Timeout.Infinite) && result;
+            result = this.exporter.Shutdown(Timeout.Infinite) && result;
+        }
+        else
+        {
+            var sw = Stopwatch.StartNew();
+            result = this.CollectFromComposite(timeoutMilliseconds) && result;
+            var timeout = timeoutMilliseconds - sw.ElapsedMilliseconds;
+            result = this.exporter.Shutdown((int)Math.Max(timeout, 0)) && result;
+        }
+
+        return result;
+    }
+
     /// <inheritdoc />
     protected override bool OnCollect(int timeoutMilliseconds)
     {

src/OpenTelemetry/Metrics/Reader/CompositeMetricReader.cs

@@ -54,34 +54,35 @@ public CompositeMetricReader AddReader(MetricReader reader)
 
     public Enumerator GetEnumerator() => new(this.Head);
 
+    // CompositeMetricReader delegates the work to its underlying readers,
+    // so CompositeMetricReader.ProcessMetrics should never be called.
+
     /// <inheritdoc/>
     internal override bool ProcessMetrics(in Batch<Metric> metrics, int timeoutMilliseconds)
-    {
-        // CompositeMetricReader delegates the work to its underlying readers,
-        // so CompositeMetricReader.ProcessMetrics should never be called.
-        throw new NotSupportedException();
-    }
+        => throw new NotSupportedException();
 
     /// <inheritdoc/>
-    protected override bool OnCollect(int timeoutMilliseconds = Timeout.Infinite)
+    protected override bool OnCollect(int timeoutMilliseconds)
     {
         var result = true;
         var sw = timeoutMilliseconds == Timeout.Infinite
             ? null
             : Stopwatch.StartNew();
 
+        this.CollectObservableInstruments();
+
         for (var cur = this.Head; cur != null; cur = cur.Next)
         {
             if (sw == null)
             {
-                result = cur.Value.Collect(Timeout.Infinite) && result;
+                result = cur.Value.CollectFromComposite(Timeout.Infinite) && result;
             }
             else
             {
                 var timeout = timeoutMilliseconds - sw.ElapsedMilliseconds;
 
                 // notify all the readers, even if we run overtime
-                result = cur.Value.Collect((int)Math.Max(timeout, 0)) && result;
+                result = cur.Value.CollectFromComposite((int)Math.Max(timeout, 0)) && result;
             }
         }
 
@@ -96,18 +97,20 @@ protected override bool OnShutdown(int timeoutMilliseconds)
             ? null
             : Stopwatch.StartNew();
 
+        this.CollectObservableInstruments();
+
         for (var cur = this.Head; cur != null; cur = cur.Next)
         {
             if (sw == null)
             {
-                result = cur.Value.Shutdown(Timeout.Infinite) && result;
+                result = cur.Value.ShutdownFromComposite(Timeout.Infinite) && result;
             }
             else
             {
                 var timeout = timeoutMilliseconds - sw.ElapsedMilliseconds;
 
                 // notify all the readers, even if we run overtime
-                result = cur.Value.Shutdown((int)Math.Max(timeout, 0)) && result;
+                result = cur.Value.ShutdownFromComposite((int)Math.Max(timeout, 0)) && result;
             }
         }