[otlp] Expand array buffer / add tests to existing base buffer (#6013)

Co-authored-by: Mikel Blanchard <mblanchard@macrosssoftware.com>
open-telemetry · Dec 10, 2024 · be82099 · be82099
1 parent 145e7ad
commit be82099
Show file tree

Hide file tree

Showing 16 changed files with 525 additions and 42 deletions.
diff --git a/...Exporter.OpenTelemetryProtocol/Implementation/OpenTelemetryProtocolExporterEventSource.cs b/...Exporter.OpenTelemetryProtocol/Implementation/OpenTelemetryProtocolExporterEventSource.cs
@@ -220,6 +220,12 @@ public void GrpcRetryDelayParsingFailed(string grpcStatusDetailsHeader, string e
         this.WriteEvent(24, grpcStatusDetailsHeader, exception);
     }
 
+    [Event(25, Message = "The array tag buffer exceeded the maximum allowed size. The array tag value was replaced with 'TRUNCATED'", Level = EventLevel.Warning)]
+    public void ArrayBufferExceededMaxSize()
+    {
+        this.WriteEvent(25);
+    }
+
     void IConfigurationExtensionsLogger.LogInvalidConfigurationValue(string key, string value)
     {
         this.InvalidConfigurationValue(key, value);

diff --git a/...try.Exporter.OpenTelemetryProtocol/Implementation/Serializer/ProtobufOtlpLogSerializer.cs b/...try.Exporter.OpenTelemetryProtocol/Implementation/Serializer/ProtobufOtlpLogSerializer.cs
@@ -20,7 +20,7 @@ internal static class ProtobufOtlpLogSerializer
     [ThreadStatic]
     private static SerializationState? threadSerializationState;
 
-    internal static int WriteLogsData(byte[] buffer, int writePosition, SdkLimitOptions sdkLimitOptions, ExperimentalOptions experimentalOptions, Resources.Resource? resource, in Batch<LogRecord> logRecordBatch)
+    internal static int WriteLogsData(ref byte[] buffer, int writePosition, SdkLimitOptions sdkLimitOptions, ExperimentalOptions experimentalOptions, Resources.Resource? resource, in Batch<LogRecord> logRecordBatch)
     {
         writePosition = ProtobufSerializer.WriteTag(buffer, writePosition, ProtobufOtlpLogFieldNumberConstants.LogsData_Resource_Logs, ProtobufWireType.LEN);
         int logsDataLengthPosition = writePosition;
@@ -47,27 +47,27 @@ internal static int WriteLogsData(byte[] buffer, int writePosition, SdkLimitOpti
             logRecords.Add(logRecord);
         }
 
-        writePosition = TryWriteResourceLogs(buffer, writePosition, sdkLimitOptions, experimentalOptions, resource, ScopeLogsList);
+        writePosition = TryWriteResourceLogs(ref buffer, writePosition, sdkLimitOptions, experimentalOptions, resource, ScopeLogsList);
         ProtobufSerializer.WriteReservedLength(buffer, logsDataLengthPosition, writePosition - (logsDataLengthPosition + ReserveSizeForLength));
         ReturnLogRecordListToPool();
 
         return writePosition;
     }
 
-    internal static int TryWriteResourceLogs(byte[] buffer, int writePosition, SdkLimitOptions sdkLimitOptions, ExperimentalOptions experimentalOptions, Resources.Resource? resource, Dictionary<string, List<LogRecord>> scopeLogs)
+    internal static int TryWriteResourceLogs(ref byte[] buffer, int writePosition, SdkLimitOptions sdkLimitOptions, ExperimentalOptions experimentalOptions, Resources.Resource? resource, Dictionary<string, List<LogRecord>> scopeLogs)
     {
         try
         {
             writePosition = WriteResourceLogs(buffer, writePosition, sdkLimitOptions, experimentalOptions, resource, scopeLogs);
         }
-        catch (IndexOutOfRangeException)
+        catch (Exception ex) when (ex is IndexOutOfRangeException || ex is ArgumentException)
         {
             if (!ProtobufSerializer.IncreaseBufferSize(ref buffer, OtlpSignalType.Logs))
             {
                 throw;
             }
 
-            return TryWriteResourceLogs(buffer, writePosition, sdkLimitOptions, experimentalOptions, resource, scopeLogs);
+            return TryWriteResourceLogs(ref buffer, writePosition, sdkLimitOptions, experimentalOptions, resource, scopeLogs);
         }
 
         return writePosition;

diff --git a/....Exporter.OpenTelemetryProtocol/Implementation/Serializer/ProtobufOtlpMetricSerializer.cs b/....Exporter.OpenTelemetryProtocol/Implementation/Serializer/ProtobufOtlpMetricSerializer.cs
@@ -17,7 +17,7 @@ internal static class ProtobufOtlpMetricSerializer
 
     private delegate int WriteExemplarFunc(byte[] buffer, int writePosition, in Exemplar exemplar);
 
-    internal static int WriteMetricsData(byte[] buffer, int writePosition, Resources.Resource? resource, in Batch<Metric> batch)
+    internal static int WriteMetricsData(ref byte[] buffer, int writePosition, Resources.Resource? resource, in Batch<Metric> batch)
     {
         writePosition = ProtobufSerializer.WriteTag(buffer, writePosition, ProtobufOtlpMetricFieldNumberConstants.MetricsData_Resource_Metrics, ProtobufWireType.LEN);
         int mericsDataLengthPosition = writePosition;
@@ -35,27 +35,27 @@ internal static int WriteMetricsData(byte[] buffer, int writePosition, Resources
             metrics.Add(metric);
         }
 
-        writePosition = TryWriteResourceMetrics(buffer, writePosition, resource, ScopeMetricsList);
+        writePosition = TryWriteResourceMetrics(ref buffer, writePosition, resource, ScopeMetricsList);
         ProtobufSerializer.WriteReservedLength(buffer, mericsDataLengthPosition, writePosition - (mericsDataLengthPosition + ReserveSizeForLength));
         ReturnMetricListToPool();
 
         return writePosition;
     }
 
-    internal static int TryWriteResourceMetrics(byte[] buffer, int writePosition, Resources.Resource? resource, Dictionary<string, List<Metric>> scopeMetrics)
+    internal static int TryWriteResourceMetrics(ref byte[] buffer, int writePosition, Resources.Resource? resource, Dictionary<string, List<Metric>> scopeMetrics)
     {
         try
         {
             writePosition = WriteResourceMetrics(buffer, writePosition, resource, scopeMetrics);
         }
-        catch (IndexOutOfRangeException)
+        catch (Exception ex) when (ex is IndexOutOfRangeException || ex is ArgumentException)
         {
             if (!ProtobufSerializer.IncreaseBufferSize(ref buffer, OtlpSignalType.Metrics))
             {
                 throw;
             }
 
-            return TryWriteResourceMetrics(buffer, writePosition, resource, scopeMetrics);
+            return TryWriteResourceMetrics(ref buffer, writePosition, resource, scopeMetrics);
         }
 
         return writePosition;

diff --git a/...lemetry.Exporter.OpenTelemetryProtocol/Implementation/Serializer/ProtobufOtlpTagWriter.cs b/...lemetry.Exporter.OpenTelemetryProtocol/Implementation/Serializer/ProtobufOtlpTagWriter.cs
@@ -1,6 +1,7 @@
 // Copyright The OpenTelemetry Authors
 // SPDX-License-Identifier: Apache-2.0
 
+using System.Diagnostics;
 using OpenTelemetry.Internal;
 
 namespace OpenTelemetry.Exporter.OpenTelemetryProtocol.Implementation.Serializer;
@@ -61,7 +62,6 @@ protected override void WriteStringTag(ref OtlpTagWriterState state, string key,
 
     protected override void WriteArrayTag(ref OtlpTagWriterState state, string key, ref OtlpTagWriterArrayState value)
     {
-        // TODO: Expand OtlpTagWriterArrayState.Buffer on IndexOutOfRangeException.
         // Write KeyValue tag
         state.WritePosition = ProtobufSerializer.WriteStringWithTag(state.Buffer, state.WritePosition, ProtobufOtlpCommonFieldNumberConstants.KeyValue_Key, key);
 
@@ -95,18 +95,19 @@ internal struct OtlpTagWriterArrayState
         public int WritePosition;
     }
 
-    private sealed class OtlpArrayTagWriter : ArrayTagWriter<OtlpTagWriterArrayState>
+    internal sealed class OtlpArrayTagWriter : ArrayTagWriter<OtlpTagWriterArrayState>
     {
         [ThreadStatic]
-        private static byte[]? threadBuffer;
+        internal static byte[]? ThreadBuffer;
+        private const int MaxBufferSize = 2 * 1024 * 1024;
 
         public override OtlpTagWriterArrayState BeginWriteArray()
         {
-            threadBuffer ??= new byte[2048];
+            ThreadBuffer ??= new byte[2048];
 
             return new OtlpTagWriterArrayState
             {
-                Buffer = threadBuffer,
+                Buffer = ThreadBuffer,
                 WritePosition = 0,
             };
         }
@@ -149,5 +150,29 @@ public override void WriteStringValue(ref OtlpTagWriterArrayState state, ReadOnl
         public override void EndWriteArray(ref OtlpTagWriterArrayState state)
         {
         }
+
+        public override bool TryResize()
+        {
+            var buffer = ThreadBuffer;
+
+            Debug.Assert(buffer != null, "buffer was null");
+
+            if (buffer!.Length >= MaxBufferSize)
+            {
+                OpenTelemetryProtocolExporterEventSource.Log.ArrayBufferExceededMaxSize();
+                return false;
+            }
+
+            try
+            {
+                ThreadBuffer = new byte[buffer.Length * 2];
+                return true;
+            }
+            catch (OutOfMemoryException)
+            {
+                OpenTelemetryProtocolExporterEventSource.Log.BufferResizeFailedDueToMemory(nameof(OtlpArrayTagWriter));
+                return false;
+            }
+        }
     }
 }
diff --git a/...y.Exporter.OpenTelemetryProtocol/Implementation/Serializer/ProtobufOtlpTraceSerializer.cs b/...y.Exporter.OpenTelemetryProtocol/Implementation/Serializer/ProtobufOtlpTraceSerializer.cs
@@ -18,7 +18,7 @@ internal static class ProtobufOtlpTraceSerializer
     private static readonly Stack<List<Activity>> ActivityListPool = [];
     private static readonly Dictionary<string, List<Activity>> ScopeTracesList = [];
 
-    internal static int WriteTraceData(byte[] buffer, int writePosition, SdkLimitOptions sdkLimitOptions, Resources.Resource? resource, in Batch<Activity> batch)
+    internal static int WriteTraceData(ref byte[] buffer, int writePosition, SdkLimitOptions sdkLimitOptions, Resources.Resource? resource, in Batch<Activity> batch)
     {
         writePosition = ProtobufSerializer.WriteTag(buffer, writePosition, ProtobufOtlpTraceFieldNumberConstants.TracesData_Resource_Spans, ProtobufWireType.LEN);
         int resourceSpansScopeSpansLengthPosition = writePosition;
@@ -36,20 +36,20 @@ internal static int WriteTraceData(byte[] buffer, int writePosition, SdkLimitOpt
             activities.Add(activity);
         }
 
-        writePosition = TryWriteResourceSpans(buffer, writePosition, sdkLimitOptions, resource);
+        writePosition = TryWriteResourceSpans(ref buffer, writePosition, sdkLimitOptions, resource);
         ReturnActivityListToPool();
         ProtobufSerializer.WriteReservedLength(buffer, resourceSpansScopeSpansLengthPosition, writePosition - (resourceSpansScopeSpansLengthPosition + ReserveSizeForLength));
 
         return writePosition;
     }
 
-    internal static int TryWriteResourceSpans(byte[] buffer, int writePosition, SdkLimitOptions sdkLimitOptions, Resources.Resource? resource)
+    internal static int TryWriteResourceSpans(ref byte[] buffer, int writePosition, SdkLimitOptions sdkLimitOptions, Resources.Resource? resource)
     {
         try
         {
             writePosition = WriteResourceSpans(buffer, writePosition, sdkLimitOptions, resource);
         }
-        catch (IndexOutOfRangeException)
+        catch (Exception ex) when (ex is IndexOutOfRangeException || ex is ArgumentException)
         {
             // Attempt to increase the buffer size
             if (!ProtobufSerializer.IncreaseBufferSize(ref buffer, OtlpSignalType.Traces))
@@ -61,7 +61,7 @@ internal static int TryWriteResourceSpans(byte[] buffer, int writePosition, SdkL
             // The recursion depth is limited to a maximum of 7 calls, as the buffer size starts at ~732 KB
             // and doubles until it reaches the maximum size of 100 MB. This ensures the recursion remains safe
             // and avoids stack overflow.
-            return TryWriteResourceSpans(buffer, writePosition, sdkLimitOptions, resource);
+            return TryWriteResourceSpans(ref buffer, writePosition, sdkLimitOptions, resource);
         }
 
         return writePosition;

diff --git a/...nTelemetry.Exporter.OpenTelemetryProtocol/Implementation/Serializer/ProtobufSerializer.cs b/...nTelemetry.Exporter.OpenTelemetryProtocol/Implementation/Serializer/ProtobufSerializer.cs
@@ -321,6 +321,13 @@ internal static int WriteStringWithTag(byte[] buffer, int writePosition, int fie
         writePosition = WriteLength(buffer, writePosition, numberOfUtf8CharsInString);
 
 #if NETFRAMEWORK || NETSTANDARD2_0
+        if (buffer.Length - writePosition < numberOfUtf8CharsInString)
+        {
+            // Note: Validate there is enough space in the buffer to hold the
+            // string otherwise throw to trigger a resize of the buffer.
+            throw new IndexOutOfRangeException();
+        }
+
         unsafe
         {
             fixed (char* strPtr = &GetNonNullPinnableReference(value))

diff --git a/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/OtlpLogExporter.cs b/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/OtlpLogExporter.cs
@@ -72,7 +72,7 @@ public override ExportResult Export(in Batch<LogRecord> logRecordBatch)
 
         try
         {
-            int writePosition = ProtobufOtlpLogSerializer.WriteLogsData(this.buffer, this.startWritePosition, this.sdkLimitOptions, this.experimentalOptions, this.Resource, logRecordBatch);
+            int writePosition = ProtobufOtlpLogSerializer.WriteLogsData(ref this.buffer, this.startWritePosition, this.sdkLimitOptions, this.experimentalOptions, this.Resource, logRecordBatch);
 
             if (this.startWritePosition == GrpcStartWritePosition)
             {

diff --git a/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/OtlpMetricExporter.cs b/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/OtlpMetricExporter.cs
@@ -65,7 +65,7 @@ public override ExportResult Export(in Batch<Metric> metrics)
 
         try
         {
-            int writePosition = ProtobufOtlpMetricSerializer.WriteMetricsData(this.buffer, this.startWritePosition, this.Resource, metrics);
+            int writePosition = ProtobufOtlpMetricSerializer.WriteMetricsData(ref this.buffer, this.startWritePosition, this.Resource, metrics);
 
             if (this.startWritePosition == GrpcStartWritePosition)
             {

diff --git a/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/OtlpTraceExporter.cs b/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/OtlpTraceExporter.cs
@@ -68,7 +68,7 @@ public override ExportResult Export(in Batch<Activity> activityBatch)
 
         try
         {
-            int writePosition = ProtobufOtlpTraceSerializer.WriteTraceData(this.buffer, this.startWritePosition, this.sdkLimitOptions, this.Resource, activityBatch);
+            int writePosition = ProtobufOtlpTraceSerializer.WriteTraceData(ref this.buffer, this.startWritePosition, this.sdkLimitOptions, this.Resource, activityBatch);
 
             if (this.startWritePosition == GrpcStartWritePosition)
             {

diff --git a/src/Shared/TagWriter/ArrayTagWriter.cs b/src/Shared/TagWriter/ArrayTagWriter.cs
@@ -19,4 +19,6 @@ internal abstract class ArrayTagWriter<TArrayState>
     public abstract void WriteStringValue(ref TArrayState state, ReadOnlySpan<char> value);
 
     public abstract void EndWriteArray(ref TArrayState state);
+
+    public virtual bool TryResize() => false;
 }
diff --git a/src/Shared/TagWriter/TagWriter.cs b/src/Shared/TagWriter/TagWriter.cs
@@ -71,6 +71,10 @@ public bool TryWriteTag(
                 {
                     this.WriteArrayTagInternal(ref state, key, array, tagValueMaxLength);
                 }
+                catch (Exception ex) when (ex is IndexOutOfRangeException || ex is ArgumentException)
+                {
+                    throw;
+                }
                 catch
                 {
                     // If an exception is thrown when calling ToString
@@ -152,27 +156,49 @@ private void WriteArrayTagInternal(ref TTagState state, string key, Array array,
     {
         var arrayState = this.arrayWriter.BeginWriteArray();
 
-        // This switch ensures the values of the resultant array-valued tag are of the same type.
-        switch (array)
+        try
         {
-            case char[] charArray: this.WriteStructToArray(ref arrayState, charArray); break;
-            case string?[] stringArray: this.WriteStringsToArray(ref arrayState, stringArray, tagValueMaxLength); break;
-            case bool[] boolArray: this.WriteStructToArray(ref arrayState, boolArray); break;
-            case byte[] byteArray: this.WriteToArrayCovariant(ref arrayState, byteArray); break;
-            case short[] shortArray: this.WriteToArrayCovariant(ref arrayState, shortArray); break;
+            // This switch ensures the values of the resultant array-valued tag are of the same type.
+            switch (array)
+            {
+                case char[] charArray: this.WriteStructToArray(ref arrayState, charArray); break;
+                case string?[] stringArray: this.WriteStringsToArray(ref arrayState, stringArray, tagValueMaxLength); break;
+                case bool[] boolArray: this.WriteStructToArray(ref arrayState, boolArray); break;
+                case byte[] byteArray: this.WriteToArrayCovariant(ref arrayState, byteArray); break;
+                case short[] shortArray: this.WriteToArrayCovariant(ref arrayState, shortArray); break;
 #if NETFRAMEWORK
-            case int[]: this.WriteArrayTagIntNetFramework(ref arrayState, array, tagValueMaxLength); break;
-            case long[]: this.WriteArrayTagLongNetFramework(ref arrayState, array, tagValueMaxLength); break;
+                case int[]: this.WriteArrayTagIntNetFramework(ref arrayState, array, tagValueMaxLength); break;
+                case long[]: this.WriteArrayTagLongNetFramework(ref arrayState, array, tagValueMaxLength); break;
 #else
-            case int[] intArray: this.WriteToArrayCovariant(ref arrayState, intArray); break;
-            case long[] longArray: this.WriteToArrayCovariant(ref arrayState, longArray); break;
+                case int[] intArray: this.WriteToArrayCovariant(ref arrayState, intArray); break;
+                case long[] longArray: this.WriteToArrayCovariant(ref arrayState, longArray); break;
 #endif
-            case float[] floatArray: this.WriteStructToArray(ref arrayState, floatArray); break;
-            case double[] doubleArray: this.WriteStructToArray(ref arrayState, doubleArray); break;
-            default: this.WriteToArrayTypeChecked(ref arrayState, array, tagValueMaxLength); break;
+                case float[] floatArray: this.WriteStructToArray(ref arrayState, floatArray); break;
+                case double[] doubleArray: this.WriteStructToArray(ref arrayState, doubleArray); break;
+                default: this.WriteToArrayTypeChecked(ref arrayState, array, tagValueMaxLength); break;
+            }
+
+            this.arrayWriter.EndWriteArray(ref arrayState);
         }
+        catch (Exception ex) when (ex is IndexOutOfRangeException || ex is ArgumentException)
+        {
+            // If the array writer cannot be resized, TryResize should log a message to the event source, return false.
+            if (this.arrayWriter.TryResize())
+            {
+                this.WriteArrayTagInternal(ref state, key, array, tagValueMaxLength);
+                return;
+            }
+
+            // Drop the array value and set "TRUNCATED" as value for easier isolation.
+            // This is a best effort to avoid dropping the entire tag.
+            this.WriteStringTag(
+                ref state,
+                key,
+                "TRUNCATED".AsSpan());
 
-        this.arrayWriter.EndWriteArray(ref arrayState);
+            this.LogUnsupportedTagTypeAndReturnFalse(key, array!.GetType().ToString());
+            return;
+        }
 
         this.WriteArrayTag(ref state, key, ref arrayState);
     }

diff --git a/...OpenTelemetryProtocol.Tests/Implementation/ExportClient/OtlpHttpTraceExportClientTests.cs b/...OpenTelemetryProtocol.Tests/Implementation/ExportClient/OtlpHttpTraceExportClientTests.cs
@@ -181,7 +181,7 @@ void RunTest(Batch<Activity> batch)
     private static (byte[] Buffer, int ContentLength) CreateTraceExportRequest(SdkLimitOptions sdkOptions, in Batch<Activity> batch, Resource resource)
     {
         var buffer = new byte[4096];
-        var writePosition = ProtobufOtlpTraceSerializer.WriteTraceData(buffer, 0, sdkOptions, resource, batch);
+        var writePosition = ProtobufOtlpTraceSerializer.WriteTraceData(ref buffer, 0, sdkOptions, resource, batch);
         return (buffer, writePosition);
     }
 }
-Original file line number
+Diff line change
@@ Expand Up @@
             try
             {
-                int writePosition = ProtobufOtlpLogSerializer.WriteLogsData(this.buffer, this.startWritePosition, this.sdkLimitOptions, this.experimentalOptions, this.Resource, logRecordBatch);
+                int writePosition = ProtobufOtlpLogSerializer.WriteLogsData(ref this.buffer, this.startWritePosition, this.sdkLimitOptions, this.experimentalOptions, this.Resource, logRecordBatch);
                 if (this.startWritePosition == GrpcStartWritePosition)
                 {
@@ Expand Down @@