Improve Metrics pull export mode (#2389)

docs/metrics/extending-the-sdk/MyExporter.cs

@@ -43,7 +43,6 @@ public override ExportResult Export(in Batch<Metric> batch)
             }
 
             sb.Append($"{record}");
-            sb.Append(')');
         }
 
         Console.WriteLine($"{this.name}.Export([{sb.ToString()}])");

src/OpenTelemetry.Exporter.Prometheus/MeterProviderBuilderExtensions.cs

@@ -39,13 +39,11 @@ public static MeterProviderBuilder AddPrometheusExporter(this MeterProviderBuild
             configure?.Invoke(options);
 
             var exporter = new PrometheusExporter(options);
-
-            var metricReader = new BaseExportingMetricReader(exporter);
-            exporter.CollectMetric = metricReader.Collect;
+            var reader = new BaseExportingMetricReader(exporter);
 
             var metricsHttpServer = new PrometheusExporterMetricsHttpServer(exporter);
             metricsHttpServer.Start();
-            return builder.AddMetricReader(metricReader);
+            return builder.AddMetricReader(reader);
         }
     }
 }

src/OpenTelemetry.Exporter.Prometheus/PrometheusExporter.cs

@@ -25,10 +25,11 @@ namespace OpenTelemetry.Exporter
     /// </summary>
     [AggregationTemporality(AggregationTemporality.Cumulative)]
     [ExportModes(ExportModes.Pull)]
-    public class PrometheusExporter : BaseExporter<Metric>
+    public class PrometheusExporter : BaseExporter<Metric>, IPullMetricExporter
     {
         internal readonly PrometheusExporterOptions Options;
         internal Batch<Metric> Metrics;
+        private Func<int, bool> funcCollect;
 
         /// <summary>
         /// Initializes a new instance of the <see cref="PrometheusExporter"/> class.
@@ -39,7 +40,11 @@ public PrometheusExporter(PrometheusExporterOptions options)
             this.Options = options;
         }
 
-        internal Func<int, bool> CollectMetric { get; set; }
+        public Func<int, bool> Collect
+        {
+            get => this.funcCollect;
+            set { this.funcCollect = value; }
+        }
 
         public override ExportResult Export(in Batch<Metric> metrics)
         {

src/OpenTelemetry.Exporter.Prometheus/PrometheusExporterMetricsHttpServer.cs

@@ -124,7 +124,7 @@ private void WorkerThread()
 
                     using var output = ctx.Response.OutputStream;
                     using var writer = new StreamWriter(output);
-                    this.exporter.CollectMetric(Timeout.Infinite);
+                    this.exporter.Collect(Timeout.Infinite);
                     this.exporter.WriteMetricsCollection(writer);
                 }
             }

src/OpenTelemetry/Metrics/BaseExportingMetricReader.cs

@@ -43,6 +43,24 @@ public BaseExportingMetricReader(BaseExporter<Metric> exporter)
                 var attr = (ExportModesAttribute)attributes[attributes.Length - 1];
                 this.supportedExportModes = attr.Supported;
             }
+
+            if (exporter is IPullMetricExporter pullExporter)
+            {
+                if (this.supportedExportModes.HasFlag(ExportModes.Push))
+                {
+                    pullExporter.Collect = this.Collect;
+                }
+                else
+                {
+                    pullExporter.Collect = (timeoutMilliseconds) =>
+                    {
+                        using (PullMetricScope.Begin())
+                        {
+                            return this.Collect(timeoutMilliseconds);
+                        }
+                    };
+                }
+            }
         }
 
         protected ExportModes SupportedExportModes => this.supportedExportModes;
@@ -59,6 +77,23 @@ protected override bool ProcessMetrics(Batch<Metric> metrics, int timeoutMillise
             return this.exporter.Export(metrics) == ExportResult.Success;
         }
 
+        /// <inheritdoc />
+        protected override bool OnCollect(int timeoutMilliseconds)
+        {
+            if (this.supportedExportModes.HasFlag(ExportModes.Push))
+            {
+                return base.OnCollect(timeoutMilliseconds);
+            }
+
+            if (this.supportedExportModes.HasFlag(ExportModes.Pull) && PullMetricScope.IsPullAllowed)
+            {
+                return base.OnCollect(timeoutMilliseconds);
+            }
+
+            // TODO: add some error log
+            return false;
+        }
+
         /// <inheritdoc />
         protected override bool OnShutdown(int timeoutMilliseconds)
         {
@@ -77,6 +112,11 @@ protected override void Dispose(bool disposing)
             {
                 try
                 {
+                    if (this.exporter is IPullMetricExporter pullExporter)
+                    {
+                        pullExporter.Collect = null;
+                    }
+
                     this.exporter.Dispose();
                 }
                 catch (Exception)

src/OpenTelemetry/Metrics/IPullMetricExporter.cs

@@ -0,0 +1,28 @@
+// <copyright file="IPullMetricExporter.cs" company="OpenTelemetry Authors">
+// Copyright The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+// </copyright>
+
+using System;
+
+namespace OpenTelemetry.Metrics
+{
+    /// <summary>
+    /// Describes a type of <see cref="BaseExporter{Metric}"/> which supports <see cref="ExportModes.Pull"/>.
+    /// </summary>
+    public interface IPullMetricExporter
+    {
+        Func<int, bool> Collect { get; set; }
+    }
+}

src/OpenTelemetry/Metrics/PullMetricScope.cs

@@ -0,0 +1,52 @@
+// <copyright file="PullMetricScope.cs" company="OpenTelemetry Authors">
+// Copyright The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+// </copyright>
+
+using System;
+using OpenTelemetry.Context;
+
+namespace OpenTelemetry.Metrics
+{
+    internal sealed class PullMetricScope : IDisposable
+    {
+        private static readonly RuntimeContextSlot<bool> Slot = RuntimeContext.RegisterSlot<bool>("otel.pull_metric");
+
+        private readonly bool previousValue;
+        private bool disposed;
+
+        internal PullMetricScope(bool value = true)
+        {
+            this.previousValue = Slot.Get();
+            Slot.Set(value);
+        }
+
+        internal static bool IsPullAllowed => Slot.Get();
+
+        public static IDisposable Begin(bool value = true)
+        {
+            return new PullMetricScope(value);
+        }
+
+        /// <inheritdoc/>
+        public void Dispose()
+        {
+            if (!this.disposed)
+            {
+                Slot.Set(this.previousValue);
+                this.disposed = true;
+            }
+        }
+    }
+}

test/OpenTelemetry.Tests/Metrics/MetricExporterTests.cs

@@ -0,0 +1,105 @@
+// <copyright file="MetricExporterTests.cs" company="OpenTelemetry Authors">
+// Copyright The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+// </copyright>
+
+using System;
+using System.Collections.Generic;
+using System.Diagnostics.Metrics;
+using Xunit;
+
+namespace OpenTelemetry.Metrics.Tests
+{
+    public class MetricExporterTests
+    {
+        [Theory]
+        [InlineData(ExportModes.Push)]
+        [InlineData(ExportModes.Pull)]
+        [InlineData(ExportModes.Pull | ExportModes.Push)]
+        public void FlushMetricExporterTest(ExportModes mode)
+        {
+            BaseExporter<Metric> exporter = null;
+
+            switch (mode)
+            {
+                case ExportModes.Push:
+                    exporter = new PushOnlyMetricExporter();
+                    break;
+                case ExportModes.Pull:
+                    exporter = new PullOnlyMetricExporter();
+                    break;
+                case ExportModes.Pull | ExportModes.Push:
+                    exporter = new PushPullMetricExporter();
+                    break;
+            }
+
+            var reader = new BaseExportingMetricReader(exporter);
+            using var meterProvider = Sdk.CreateMeterProviderBuilder()
+                .AddMetricReader(reader)
+                .Build();
+
+            switch (mode)
+            {
+                case ExportModes.Push:
+                    Assert.True(reader.Collect());
+                    Assert.True(meterProvider.ForceFlush());
+                    break;
+                case ExportModes.Pull:
+                    Assert.False(reader.Collect());
+                    Assert.False(meterProvider.ForceFlush());
+                    Assert.True((exporter as IPullMetricExporter).Collect(-1));
+                    break;
+                case ExportModes.Pull | ExportModes.Push:
+                    Assert.True(reader.Collect());
+                    Assert.True(meterProvider.ForceFlush());
+                    break;
+            }
+        }
+
+        [ExportModes(ExportModes.Push)]
+        private class PushOnlyMetricExporter : BaseExporter<Metric>
+        {
+            public override ExportResult Export(in Batch<Metric> batch)
+            {
+                return ExportResult.Success;
+            }
+        }
+
+        [ExportModes(ExportModes.Pull)]
+        private class PullOnlyMetricExporter : BaseExporter<Metric>, IPullMetricExporter
+        {
+            private Func<int, bool> funcCollect;
+
+            public Func<int, bool> Collect
+            {
+                get => this.funcCollect;
+                set { this.funcCollect = value; }
+            }
+
+            public override ExportResult Export(in Batch<Metric> batch)
+            {
+                return ExportResult.Success;
+            }
+        }
+
+        [ExportModes(ExportModes.Pull | ExportModes.Push)]
+        private class PushPullMetricExporter : BaseExporter<Metric>
+        {
+            public override ExportResult Export(in Batch<Metric> batch)
+            {
+                return ExportResult.Success;
+            }
+        }
+    }
+}