Add Metric Reader and use IEnumerable for Metric collect instead of Batch

examples/Console/TestMetrics.cs

@@ -76,7 +76,7 @@ internal static object Run(MetricsOptions options)
                     .AddConsoleExporter(o =>
                     {
                         o.MetricExportIntervalMilliseconds = options.DefaultCollectionPeriodMilliseconds;
-                        o.IsDelta = options.IsDelta;
+                        o.AggregationTemporality = options.IsDelta ? AggregationTemporality.Delta : AggregationTemporality.Cumulative;
                     });
             }
 

src/OpenTelemetry.Exporter.Console/ConsoleExporterMetricHelperExtensions.cs

@@ -37,7 +37,9 @@ public static MeterProviderBuilder AddConsoleExporter(this MeterProviderBuilder
 
             var options = new ConsoleExporterOptions();
             configure?.Invoke(options);
-            return builder.AddMetricProcessor(new PushMetricProcessor(new ConsoleMetricExporter(options), options.MetricExportIntervalMilliseconds, options.IsDelta));
+
+            var exporter = new ConsoleMetricExporter(options);
+            return builder.AddMetricReader(new PeriodicExportingMetricReader(exporter, options.MetricExportIntervalMilliseconds));
         }
     }
 }

src/OpenTelemetry.Exporter.Console/ConsoleExporterOptions.cs

@@ -14,6 +14,8 @@
 // limitations under the License.
 // </copyright>
 
+using OpenTelemetry.Metrics;
+
 namespace OpenTelemetry.Exporter
 {
     public class ConsoleExporterOptions
@@ -29,9 +31,9 @@ public class ConsoleExporterOptions
         public int MetricExportIntervalMilliseconds { get; set; } = 1000;
 
         /// <summary>
-        /// Gets or sets a value indicating whether to export Delta
-        /// values or not (Cumulative).
+        /// Gets or sets the AggregationTemporality used for Sum, Histogram
+        /// metrics.
         /// </summary>
-        public bool IsDelta { get; set; } = false;
+        public AggregationTemporality AggregationTemporality { get; set; } = AggregationTemporality.Delta;
     }
 }

src/OpenTelemetry.Exporter.Console/ConsoleMetricExporter.cs

@@ -15,6 +15,7 @@
 // </copyright>
 
 using System;
+using System.Collections.Generic;
 using System.Globalization;
 using System.Linq;
 using System.Text;
@@ -23,16 +24,22 @@
 
 namespace OpenTelemetry.Exporter
 {
-    public class ConsoleMetricExporter : ConsoleExporter<Metric>
+    public class ConsoleMetricExporter : MetricExporter
     {
         private Resource resource;
+        private ConsoleExporterOptions options;
 
         public ConsoleMetricExporter(ConsoleExporterOptions options)
-            : base(options)
         {
+            this.options = options;
         }
 
-        public override ExportResult Export(in Batch<Metric> batch)
+        public override AggregationTemporality GetAggregationTemporality()
+        {
+            return this.options.AggregationTemporality;
+        }
+
+        public override ExportResult Export(IEnumerable<Metric> metrics)
         {
             if (this.resource == null)
             {
@@ -49,7 +56,7 @@ public override ExportResult Export(in Batch<Metric> batch)
                 }
             }
 
-            foreach (var metric in batch)
+            foreach (var metric in metrics)
             {
                 var msg = new StringBuilder($"\nExport ");
                 msg.Append(metric.Name);
@@ -160,6 +167,7 @@ public override ExportResult Export(in Batch<Metric> batch)
                     }
 
                     msg = new StringBuilder();
+                    msg.Append("(");
                     msg.Append(metricPoint.StartTime.ToString("yyyy-MM-ddTHH:mm:ss.fffffffZ", CultureInfo.InvariantCulture));
                     msg.Append(", ");
                     msg.Append(metricPoint.EndTime.ToString("yyyy-MM-ddTHH:mm:ss.fffffffZ", CultureInfo.InvariantCulture));

src/OpenTelemetry.Exporter.Prometheus/MeterProviderBuilderExtensions.cs

@@ -38,12 +38,13 @@ public static MeterProviderBuilder AddPrometheusExporter(this MeterProviderBuild
             var options = new PrometheusExporterOptions();
             configure?.Invoke(options);
             var exporter = new PrometheusExporter(options);
-            var pullMetricProcessor = new PullMetricProcessor(exporter, false);
-            exporter.MakePullRequest = pullMetricProcessor.PullRequest;
+
+            var metricReader = new BaseExportingMetricReader(exporter);
+            exporter.CollectMetric = metricReader.Collect;
 
             var metricsHttpServer = new PrometheusExporterMetricsHttpServer(exporter);
             metricsHttpServer.Start();
-            return builder.AddMetricProcessor(pullMetricProcessor);
+            return builder.AddMetricReader(metricReader);
         }
     }
 }

src/OpenTelemetry.Exporter.Prometheus/PrometheusExporter.cs

@@ -15,17 +15,18 @@
 // </copyright>
 
 using System;
+using System.Collections.Generic;
 using OpenTelemetry.Metrics;
 
 namespace OpenTelemetry.Exporter
 {
     /// <summary>
     /// Exporter of OpenTelemetry metrics to Prometheus.
     /// </summary>
-    public class PrometheusExporter : BaseExporter<Metric>
+    public class PrometheusExporter : MetricExporter
     {
         internal readonly PrometheusExporterOptions Options;
-        internal Batch<Metric> Batch;
+        internal IEnumerable<Metric> Metrics;
 
         /// <summary>
         /// Initializes a new instance of the <see cref="PrometheusExporter"/> class.
@@ -36,11 +37,11 @@ public PrometheusExporter(PrometheusExporterOptions options)
             this.Options = options;
         }
 
-        internal Action MakePullRequest { get; set; }
+        internal Action CollectMetric { get; set; }
 
-        public override ExportResult Export(in Batch<Metric> batch)
+        public override ExportResult Export(IEnumerable<Metric> metrics)
         {
-            this.Batch = batch;
+            this.Metrics = metrics;
             return ExportResult.Success;
         }
     }

src/OpenTelemetry.Exporter.Prometheus/PrometheusExporterExtensions.cs

@@ -44,7 +44,7 @@ public static class PrometheusExporterExtensions
         /// <param name="writer">StreamWriter to write to.</param>
         public static void WriteMetricsCollection(this PrometheusExporter exporter, StreamWriter writer)
         {
-            foreach (var metric in exporter.Batch)
+            foreach (var metric in exporter.Metrics)
             {
                 var builder = new PrometheusMetricBuilder()
                     .WithName(metric.Name)

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.MakePullRequest();
+                    this.exporter.CollectMetric();
                     this.exporter.WriteMetricsCollection(writer);
                 }
             }

src/OpenTelemetry/Batch.cs

@@ -19,7 +19,6 @@
 using System.Collections.Generic;
 using System.Diagnostics;
 using OpenTelemetry.Internal;
-using OpenTelemetry.Metrics;
 
 namespace OpenTelemetry
 {
@@ -32,14 +31,12 @@ namespace OpenTelemetry
     {
         private readonly T item;
         private readonly CircularBuffer<T> circularBuffer;
-        private readonly T[] metrics;
         private readonly long targetCount;
 
         internal Batch(T item)
         {
             this.item = item ?? throw new ArgumentNullException(nameof(item));
             this.circularBuffer = null;
-            this.metrics = null;
             this.targetCount = 1;
         }
 
@@ -48,21 +45,10 @@ internal Batch(CircularBuffer<T> circularBuffer, int maxSize)
             Debug.Assert(maxSize > 0, $"{nameof(maxSize)} should be a positive number.");
 
             this.item = null;
-            this.metrics = null;
             this.circularBuffer = circularBuffer ?? throw new ArgumentNullException(nameof(circularBuffer));
             this.targetCount = circularBuffer.RemovedCount + Math.Min(maxSize, circularBuffer.Count);
         }
 
-        internal Batch(T[] metrics, int maxSize)
-        {
-            Debug.Assert(maxSize > 0, $"{nameof(maxSize)} should be a positive number.");
-
-            this.item = null;
-            this.circularBuffer = null;
-            this.metrics = metrics ?? throw new ArgumentNullException(nameof(metrics));
-            this.targetCount = maxSize;
-        }
-
         /// <inheritdoc/>
         public void Dispose()
         {
@@ -84,8 +70,6 @@ public Enumerator GetEnumerator()
         {
             return this.circularBuffer != null
                 ? new Enumerator(this.circularBuffer, this.targetCount)
-                : this.metrics != null
-                ? new Enumerator(this.metrics, this.targetCount)
                 : new Enumerator(this.item);
         }
 
@@ -95,35 +79,20 @@ public Enumerator GetEnumerator()
         public struct Enumerator : IEnumerator<T>
         {
             private readonly CircularBuffer<T> circularBuffer;
-            private readonly T[] metrics;
             private long targetCount;
-            private int metricIndex;
 
             internal Enumerator(T item)
             {
                 this.Current = item;
                 this.circularBuffer = null;
-                this.metrics = null;
                 this.targetCount = -1;
-                this.metricIndex = 0;
             }
 
             internal Enumerator(CircularBuffer<T> circularBuffer, long targetCount)
             {
                 this.Current = null;
-                this.metrics = null;
                 this.circularBuffer = circularBuffer;
                 this.targetCount = targetCount;
-                this.metricIndex = 0;
-            }
-
-            internal Enumerator(T[] metrics, long targetCount)
-            {
-                this.Current = null;
-                this.circularBuffer = null;
-                this.metrics = metrics;
-                this.targetCount = targetCount;
-                this.metricIndex = 0;
             }
 
             /// <inheritdoc/>
@@ -141,9 +110,8 @@ public void Dispose()
             public bool MoveNext()
             {
                 var circularBuffer = this.circularBuffer;
-                var metrics = this.metrics;
 
-                if (circularBuffer == null && metrics == null)
+                if (circularBuffer == null)
                 {
                     if (this.targetCount >= 0)
                     {
@@ -155,23 +123,10 @@ public bool MoveNext()
                     return true;
                 }
 
-                if (circularBuffer != null)
-                {
-                    if (circularBuffer.RemovedCount < this.targetCount)
-                    {
-                        this.Current = circularBuffer.Read();
-                        return true;
-                    }
-                }
-
-                if (metrics != null)
+                if (circularBuffer.RemovedCount < this.targetCount)
                 {
-                    if (this.metricIndex < this.targetCount)
-                    {
-                        this.Current = metrics[this.metricIndex];
-                        this.metricIndex++;
-                        return true;
-                    }
+                    this.Current = circularBuffer.Read();
+                    return true;
                 }
 
                 this.Current = null;

src/OpenTelemetry/Metrics/BaseExportingMetricReader.cs

@@ -0,0 +1,41 @@
+// <copyright file="BaseExportingMetricReader.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;
+
+namespace OpenTelemetry.Metrics
+{
+    public class BaseExportingMetricReader : MetricReader
+    {
+        private MetricExporter exporter;
+
+        public BaseExportingMetricReader(MetricExporter exporter)
+        {
+            this.exporter = exporter;
+        }
+
+        public override void OnCollect(IEnumerable<Metric> metrics)
+        {
+            this.exporter.Export(metrics);
+        }
+
+        public override AggregationTemporality GetAggregationTemporality()
+        {
+            return this.exporter.GetAggregationTemporality();
+        }
+    }
+}

src/OpenTelemetry/Metrics/MeterProviderBuilderExtensions.cs

@@ -39,6 +39,22 @@ public static MeterProviderBuilder AddMetricProcessor(this MeterProviderBuilder
             return meterProviderBuilder;
         }
 
+        /// <summary>
+        /// Add metric reader.
+        /// </summary>
+        /// <param name="meterProviderBuilder"><see cref="MeterProviderBuilder"/>.</param>
+        /// <param name="metricReader">Metricreader.</param>
+        /// <returns><see cref="MeterProvider"/>.</returns>
+        public static MeterProviderBuilder AddMetricReader(this MeterProviderBuilder meterProviderBuilder, MetricReader metricReader)
+        {
+            if (meterProviderBuilder is MeterProviderBuilderSdk meterProviderBuilderSdk)
+            {
+                return meterProviderBuilderSdk.AddMetricReader(metricReader);
+            }
+
+            return meterProviderBuilder;
+        }
+
         /// <summary>
         /// Sets the <see cref="ResourceBuilder"/> from which the Resource associated with
         /// this provider is built from. Overwrites currently set ResourceBuilder.

src/OpenTelemetry/Metrics/MeterProviderBuilderSdk.cs

@@ -32,6 +32,8 @@ internal MeterProviderBuilderSdk()
 
         internal List<MetricProcessor> MetricProcessors { get; } = new List<MetricProcessor>();
 
+        internal List<MetricReader> MetricReaders { get; } = new List<MetricReader>();
+
         public override MeterProviderBuilder AddInstrumentation<TInstrumentation>(Func<TInstrumentation> instrumentationFactory)
         {
             if (instrumentationFactory == null)
@@ -79,6 +81,12 @@ internal MeterProviderBuilderSdk AddMetricProcessor(MetricProcessor processor)
             return this;
         }
 
+        internal MeterProviderBuilderSdk AddMetricReader(MetricReader metricReader)
+        {
+            this.MetricReaders.Add(metricReader);
+            return this;
+        }
+
         internal MeterProviderBuilderSdk SetResourceBuilder(ResourceBuilder resourceBuilder)
         {
             this.resourceBuilder = resourceBuilder ?? throw new ArgumentNullException(nameof(resourceBuilder));
@@ -91,7 +99,8 @@ internal MeterProvider Build()
                 this.resourceBuilder.Build(),
                 this.meterSources,
                 this.instrumentationFactories,
-                this.MetricProcessors.ToArray());
+                this.MetricProcessors.ToArray(),
+                this.MetricReaders.ToArray());
         }
 
         // TODO: This is copied from TracerProviderBuilderSdk. Move to common location.