[prometheus] Fixed a race condition happening in simultaneous OpenMetrics and PlainText requests

src/OpenTelemetry.Exporter.Prometheus.AspNetCore/CHANGELOG.md

@@ -2,6 +2,9 @@
 
 ## Unreleased
 
+* Fixed a race condition happening in simultaneous OpenMetrics and PlainText requests which could cause malformed response
+  ([#5517](https://github.com/open-telemetry/opentelemetry-dotnet/pull/5517))
+
 ## 1.8.0-rc.1
 
 Released 2024-Mar-27

src/OpenTelemetry.Exporter.Prometheus.AspNetCore/PrometheusExporterMiddleware.cs

@@ -53,11 +53,13 @@ public async Task InvokeAsync(HttpContext httpContext)
         try
         {
             var openMetricsRequested = AcceptsOpenMetrics(httpContext.Request);
-            var collectionResponse = await this.exporter.CollectionManager.EnterCollect(openMetricsRequested).ConfigureAwait(false);
+            var collectionResponse = await this.exporter.CollectionManager.EnterCollect().ConfigureAwait(false);
 
             try
             {
-                if (collectionResponse.View.Count > 0)
+                var dataView = openMetricsRequested ? collectionResponse.OpenMetricsView : collectionResponse.PlainTextView;
+
+                if (dataView.Count > 0)
                 {
                     response.StatusCode = 200;
 #if NET8_0_OR_GREATER
@@ -69,7 +71,7 @@ public async Task InvokeAsync(HttpContext httpContext)
                         ? "application/openmetrics-text; version=1.0.0; charset=utf-8"
                         : "text/plain; charset=utf-8; version=0.0.4";
 
-                    await response.Body.WriteAsync(collectionResponse.View.Array, 0, collectionResponse.View.Count).ConfigureAwait(false);
+                    await response.Body.WriteAsync(dataView.Array, 0, dataView.Count).ConfigureAwait(false);
                 }
                 else
                 {
@@ -91,8 +93,6 @@ public async Task InvokeAsync(HttpContext httpContext)
                 response.StatusCode = 500;
             }
         }
-
-        this.exporter.OnExport = null;
     }
 
     private static bool AcceptsOpenMetrics(HttpRequest request)

src/OpenTelemetry.Exporter.Prometheus.HttpListener/Internal/PrometheusCollectionManager.cs

@@ -16,10 +16,12 @@ internal sealed class PrometheusCollectionManager
     private readonly Dictionary<Metric, PrometheusMetric> metricsCache;
     private readonly HashSet<string> scopes;
     private int metricsCacheCount;
-    private byte[] buffer = new byte[85000]; // encourage the object to live in LOH (large object heap)
+    private byte[] plainTextBuffer = new byte[85000]; // encourage the object to live in LOH (large object heap)
+    private byte[] openMetricsBuffer = new byte[85000]; // encourage the object to live in LOH (large object heap)
     private int targetInfoBufferLength = -1; // zero or positive when target_info has been written for the first time
     private int globalLockState;
-    private ArraySegment<byte> previousDataView;
+    private ArraySegment<byte> previousPlainTextDataView;
+    private ArraySegment<byte> previousOpenMetricsDataView;
     private DateTime? previousDataViewGeneratedAtUtc;
     private int readerCount;
     private bool collectionRunning;
@@ -35,9 +37,9 @@ public PrometheusCollectionManager(PrometheusExporter exporter)
     }
 
 #if NET6_0_OR_GREATER
-    public ValueTask<CollectionResponse> EnterCollect(bool openMetricsRequested)
+    public ValueTask<CollectionResponse> EnterCollect()
 #else
-    public Task<CollectionResponse> EnterCollect(bool openMetricsRequested)
+    public Task<CollectionResponse> EnterCollect()
 #endif
     {
         this.EnterGlobalLock();
@@ -51,9 +53,9 @@ public Task<CollectionResponse> EnterCollect(bool openMetricsRequested)
             Interlocked.Increment(ref this.readerCount);
             this.ExitGlobalLock();
 #if NET6_0_OR_GREATER
-            return new ValueTask<CollectionResponse>(new CollectionResponse(this.previousDataView, this.previousDataViewGeneratedAtUtc.Value, fromCache: true));
+            return new ValueTask<CollectionResponse>(new CollectionResponse(this.previousOpenMetricsDataView, this.previousPlainTextDataView, this.previousDataViewGeneratedAtUtc.Value, fromCache: true));
 #else
-            return Task.FromResult(new CollectionResponse(this.previousDataView, this.previousDataViewGeneratedAtUtc.Value, fromCache: true));
+            return Task.FromResult(new CollectionResponse(this.previousOpenMetricsDataView, this.previousPlainTextDataView, this.previousDataViewGeneratedAtUtc.Value, fromCache: true));
 #endif
         }
 
@@ -65,12 +67,13 @@ public Task<CollectionResponse> EnterCollect(bool openMetricsRequested)
                 this.collectionTcs = new TaskCompletionSource<CollectionResponse>(TaskCreationOptions.RunContinuationsAsynchronously);
             }
 
+            var task = this.collectionTcs.Task;
             Interlocked.Increment(ref this.readerCount);
             this.ExitGlobalLock();
 #if NET6_0_OR_GREATER
-            return new ValueTask<CollectionResponse>(this.collectionTcs.Task);
+            return new ValueTask<CollectionResponse>(task);
 #else
-            return this.collectionTcs.Task;
+            return task;
 #endif
         }
 
@@ -83,11 +86,11 @@ public Task<CollectionResponse> EnterCollect(bool openMetricsRequested)
         this.ExitGlobalLock();
 
         CollectionResponse response;
-        var result = this.ExecuteCollect(openMetricsRequested);
+        var result = this.ExecuteCollect();
         if (result)
         {
             this.previousDataViewGeneratedAtUtc = DateTime.UtcNow;
-            response = new CollectionResponse(this.previousDataView, this.previousDataViewGeneratedAtUtc.Value, fromCache: false);
+            response = new CollectionResponse(this.previousOpenMetricsDataView, this.previousPlainTextDataView, this.previousDataViewGeneratedAtUtc.Value, fromCache: false);
         }
         else
         {
@@ -147,7 +150,7 @@ private void WaitForReadersToComplete()
         SpinWait readWait = default;
         while (true)
         {
-            if (Interlocked.CompareExchange(ref this.readerCount, 0, this.readerCount) != 0)
+            if (Interlocked.CompareExchange(ref this.readerCount, 0, 0) != 0)
             {
                 readWait.SpinOnce();
                 continue;
@@ -158,56 +161,51 @@ private void WaitForReadersToComplete()
     }
 
     [MethodImpl(MethodImplOptions.AggressiveInlining)]
-    private bool ExecuteCollect(bool openMetricsRequested)
+    private bool ExecuteCollect()
     {
         this.exporter.OnExport = this.onCollectRef;
-        this.exporter.OpenMetricsRequested = openMetricsRequested;
         var result = this.exporter.Collect(Timeout.Infinite);
         this.exporter.OnExport = null;
         return result;
     }
 
     private ExportResult OnCollect(Batch<Metric> metrics)
     {
-        var cursor = 0;
-
         try
         {
-            if (this.exporter.OpenMetricsRequested)
-            {
-                cursor = this.WriteTargetInfo();
+            var openMetricsCursor = this.WriteTargetInfo();
+            var plainTextCursor = 0;
 
-                this.scopes.Clear();
+            this.scopes.Clear();
 
-                foreach (var metric in metrics)
+            foreach (var metric in metrics)
+            {
+                if (!PrometheusSerializer.CanWriteMetric(metric))
                 {
-                    if (!PrometheusSerializer.CanWriteMetric(metric))
-                    {
-                        continue;
-                    }
+                    continue;
+                }
 
-                    if (this.scopes.Add(metric.MeterName))
+                if (this.scopes.Add(metric.MeterName))
+                {
+                    while (true)
                     {
-                        while (true)
+                        try
                         {
-                            try
-                            {
-                                cursor = PrometheusSerializer.WriteScopeInfo(this.buffer, cursor, metric.MeterName);
+                            openMetricsCursor = PrometheusSerializer.WriteScopeInfo(this.openMetricsBuffer, openMetricsCursor, metric.MeterName);
 
-                                break;
-                            }
-                            catch (IndexOutOfRangeException)
+                            break;
+                        }
+                        catch (IndexOutOfRangeException)
+                        {
+                            if (!this.IncreaseBufferSize(ref this.openMetricsBuffer))
                             {
-                                if (!this.IncreaseBufferSize())
-                                {
-                                    // there are two cases we might run into the following condition:
-                                    // 1. we have many metrics to be exported - in this case we probably want
-                                    //    to put some upper limit and allow the user to configure it.
-                                    // 2. we got an IndexOutOfRangeException which was triggered by some other
-                                    //    code instead of the buffer[cursor++] - in this case we should give up
-                                    //    at certain point rather than allocating like crazy.
-                                    throw;
-                                }
+                                // there are two cases we might run into the following condition:
+                                // 1. we have many metrics to be exported - in this case we probably want
+                                //    to put some upper limit and allow the user to configure it.
+                                // 2. we got an IndexOutOfRangeException which was triggered by some other
+                                //    code instead of the buffer[cursor++] - in this case we should give up
+                                //    at certain point rather than allocating like crazy.
+                                throw;
                             }
                         }
                     }
@@ -221,22 +219,46 @@ private ExportResult OnCollect(Batch<Metric> metrics)
                     continue;
                 }
 
+                var prometheusMetric = this.GetPrometheusMetric(metric);
+
+                while (true)
+                {
+                    try
+                    {
+                        openMetricsCursor = PrometheusSerializer.WriteMetric(
+                            this.openMetricsBuffer,
+                            openMetricsCursor,
+                            metric,
+                            prometheusMetric,
+                            true);
+
+                        break;
+                    }
+                    catch (IndexOutOfRangeException)
+                    {
+                        if (!this.IncreaseBufferSize(ref this.openMetricsBuffer))
+                        {
+                            throw;
+                        }
+                    }
+                }
+
                 while (true)
                 {
                     try
                     {
-                        cursor = PrometheusSerializer.WriteMetric(
-                            this.buffer,
-                            cursor,
+                        plainTextCursor = PrometheusSerializer.WriteMetric(
+                            this.plainTextBuffer,
+                            plainTextCursor,
                             metric,
-                            this.GetPrometheusMetric(metric),
-                            this.exporter.OpenMetricsRequested);
+                            prometheusMetric,
+                            false);
 
                         break;
                     }
                     catch (IndexOutOfRangeException)
                     {
-                        if (!this.IncreaseBufferSize())
+                        if (!this.IncreaseBufferSize(ref this.plainTextBuffer))
                         {
                             throw;
                         }
@@ -248,24 +270,41 @@ private ExportResult OnCollect(Batch<Metric> metrics)
             {
                 try
                 {
-                    cursor = PrometheusSerializer.WriteEof(this.buffer, cursor);
+                    openMetricsCursor = PrometheusSerializer.WriteEof(this.openMetricsBuffer, openMetricsCursor);
                     break;
                 }
                 catch (IndexOutOfRangeException)
                 {
-                    if (!this.IncreaseBufferSize())
+                    if (!this.IncreaseBufferSize(ref this.openMetricsBuffer))
                     {
                         throw;
                     }
                 }
             }
 
-            this.previousDataView = new ArraySegment<byte>(this.buffer, 0, cursor);
+            while (true)
+            {
+                try
+                {
+                    plainTextCursor = PrometheusSerializer.WriteEof(this.plainTextBuffer, plainTextCursor);
+                    break;
+                }
+                catch (IndexOutOfRangeException)
+                {
+                    if (!this.IncreaseBufferSize(ref this.plainTextBuffer))
+                    {
+                        throw;
+                    }
+                }
+            }
+
+            this.previousOpenMetricsDataView = new ArraySegment<byte>(this.openMetricsBuffer, 0, openMetricsCursor);
+            this.previousPlainTextDataView = new ArraySegment<byte>(this.plainTextBuffer, 0, plainTextCursor);
             return ExportResult.Success;
         }
         catch (Exception)
         {
-            this.previousDataView = new ArraySegment<byte>(Array.Empty<byte>(), 0, 0);
+            this.previousOpenMetricsDataView = this.previousPlainTextDataView = new ArraySegment<byte>(Array.Empty<byte>(), 0, 0);
             return ExportResult.Failure;
         }
     }
@@ -278,13 +317,13 @@ private int WriteTargetInfo()
             {
                 try
                 {
-                    this.targetInfoBufferLength = PrometheusSerializer.WriteTargetInfo(this.buffer, 0, this.exporter.Resource);
+                    this.targetInfoBufferLength = PrometheusSerializer.WriteTargetInfo(this.openMetricsBuffer, 0, this.exporter.Resource);
 
                     break;
                 }
                 catch (IndexOutOfRangeException)
                 {
-                    if (!this.IncreaseBufferSize())
+                    if (!this.IncreaseBufferSize(ref this.openMetricsBuffer))
                     {
                         throw;
                     }
@@ -295,18 +334,18 @@ private int WriteTargetInfo()
         return this.targetInfoBufferLength;
     }
 
-    private bool IncreaseBufferSize()
+    private bool IncreaseBufferSize(ref byte[] buffer)
     {
-        var newBufferSize = this.buffer.Length * 2;
+        var newBufferSize = buffer.Length * 2;
 
         if (newBufferSize > 100 * 1024 * 1024)
         {
             return false;
         }
 
         var newBuffer = new byte[newBufferSize];
-        this.buffer.CopyTo(newBuffer, 0);
-        this.buffer = newBuffer;
+        buffer.CopyTo(newBuffer, 0);
+        buffer = newBuffer;
 
         return true;
     }
@@ -331,14 +370,17 @@ private PrometheusMetric GetPrometheusMetric(Metric metric)
 
     public readonly struct CollectionResponse
     {
-        public CollectionResponse(ArraySegment<byte> view, DateTime generatedAtUtc, bool fromCache)
+        public CollectionResponse(ArraySegment<byte> openMetricsView, ArraySegment<byte> plainTextView, DateTime generatedAtUtc, bool fromCache)
         {
-            this.View = view;
+            this.OpenMetricsView = openMetricsView;
+            this.PlainTextView = plainTextView;
             this.GeneratedAtUtc = generatedAtUtc;
             this.FromCache = fromCache;
         }
 
-        public ArraySegment<byte> View { get; }
+        public ArraySegment<byte> OpenMetricsView { get; }
+
+        public ArraySegment<byte> PlainTextView { get; }
 
         public DateTime GeneratedAtUtc { get; }
 

src/OpenTelemetry.Exporter.Prometheus.HttpListener/PrometheusHttpListener.cs

@@ -148,20 +148,21 @@ private async Task ProcessRequestAsync(HttpListenerContext context)
         try
         {
             var openMetricsRequested = AcceptsOpenMetrics(context.Request);
-            var collectionResponse = await this.exporter.CollectionManager.EnterCollect(openMetricsRequested).ConfigureAwait(false);
+            var collectionResponse = await this.exporter.CollectionManager.EnterCollect().ConfigureAwait(false);
 
             try
             {
                 context.Response.Headers.Add("Server", string.Empty);
-                if (collectionResponse.View.Count > 0)
+                var dataView = openMetricsRequested ? collectionResponse.OpenMetricsView : collectionResponse.PlainTextView;
+                if (dataView.Count > 0)
                 {
                     context.Response.StatusCode = 200;
                     context.Response.Headers.Add("Last-Modified", collectionResponse.GeneratedAtUtc.ToString("R"));
                     context.Response.ContentType = openMetricsRequested
                         ? "application/openmetrics-text; version=1.0.0; charset=utf-8"
                         : "text/plain; charset=utf-8; version=0.0.4";
 
-                    await context.Response.OutputStream.WriteAsync(collectionResponse.View.Array, 0, collectionResponse.View.Count).ConfigureAwait(false);
+                    await context.Response.OutputStream.WriteAsync(dataView.Array, 0, dataView.Count).ConfigureAwait(false);
                 }
                 else
                 {