serialize KeyValue lists

Author: juliuskoval

src/OpenTelemetry.Exporter.Console/Implementation/ConsoleTagWriter.cs

@@ -4,6 +4,7 @@
 using System.Diagnostics;
 using System.Globalization;
 using System.Text;
+using Microsoft.Extensions.Primitives;
 using OpenTelemetry.Internal;
 
 namespace OpenTelemetry.Exporter;
@@ -85,6 +86,18 @@ protected override bool TryWriteEmptyTag(ref ConsoleTag consoleTag, string key,
         return true;
     }
 
+    protected override void WriteKvListTag(ref ConsoleTag state, string key, IEnumerable<KeyValuePair<string, object?>> kvList, int? tagValueMaxLength)
+    {
+        var stringValue = Convert.ToString(kvList, CultureInfo.InvariantCulture);
+
+        if (stringValue is null)
+        {
+            return;
+        }
+
+        this.TryWriteTag(ref state, key, stringValue);
+    }
+
     internal struct ConsoleTag
     {
         public string? Key;

src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/Serializer/ProtobufOtlpCommonFieldNumberConstants.cs

@@ -29,4 +29,6 @@ internal static class ProtobufOtlpCommonFieldNumberConstants
     internal const int AnyValue_Bytes_Value = 7;
 
     internal const int ArrayValue_Value = 1;
+
+    internal const int KeyValueList_Values = 1;
 }

src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/Serializer/ProtobufOtlpTagWriter.cs

@@ -7,6 +7,8 @@ namespace OpenTelemetry.Exporter.OpenTelemetryProtocol.Implementation.Serializer
 
 internal sealed class ProtobufOtlpTagWriter : TagWriter<ProtobufOtlpTagWriter.OtlpTagWriterState, ProtobufOtlpTagWriter.OtlpTagWriterArrayState>
 {
+    private const int ReserveSizeForLength = 4;
+
     private ProtobufOtlpTagWriter()
         : base(new OtlpArrayTagWriter())
     {
@@ -101,6 +103,55 @@ protected override bool TryWriteByteArrayTag(ref OtlpTagWriterState state, strin
         return true;
     }
 
+    protected override void WriteKvListTag(ref OtlpTagWriterState state, string key, IEnumerable<KeyValuePair<string, object?>> kvList, int? tagValueMaxLength)
+    {
+        var startPosition = state.WritePosition;
+
+        state.WritePosition = ProtobufSerializer.WriteStringWithTag(state.Buffer, state.WritePosition, ProtobufOtlpCommonFieldNumberConstants.KeyValue_Key, key);
+
+        // Reserve space for KeyValue.value (AnyValue wrapper)
+        state.WritePosition = ProtobufSerializer.WriteTag(state.Buffer, state.WritePosition, ProtobufOtlpCommonFieldNumberConstants.KeyValue_Value, ProtobufWireType.LEN);
+        int anyValueLengthPos = state.WritePosition;
+        state.WritePosition += ReserveSizeForLength;
+
+        // Reserve space for AnyValue.kvlist_value (KeyValueList wrapper)
+        state.WritePosition = ProtobufSerializer.WriteTag(state.Buffer, state.WritePosition, ProtobufOtlpCommonFieldNumberConstants.AnyValue_Kvlist_Value, ProtobufWireType.LEN);
+        int kvlistLengthPos = state.WritePosition;
+        state.WritePosition += ReserveSizeForLength;
+
+        try
+        {
+            foreach (var kvp in kvList)
+            {
+                var startEntryPosition = state.WritePosition;
+
+                // Write KeyValueList.values tag (field 1, LEN) + reserve length per entry
+                state.WritePosition = ProtobufSerializer.WriteTag(state.Buffer, state.WritePosition, ProtobufOtlpCommonFieldNumberConstants.KeyValueList_Values, ProtobufWireType.LEN);
+                int entryLengthPos = state.WritePosition;
+                state.WritePosition += ReserveSizeForLength;
+
+                // Drop a tag if we fail to write it, and reset the position to the start of the entry to overwrite the reserved tag and length for the next entry.
+                if (!this.TryWriteTag(ref state, kvp.Key, kvp.Value, tagValueMaxLength))
+                {
+                    state.WritePosition = startEntryPosition;
+                    continue;
+                }
+
+                ProtobufSerializer.WriteReservedLength(state.Buffer, entryLengthPos, state.WritePosition - (entryLengthPos + ReserveSizeForLength));
+            }
+        }
+        catch (Exception)
+        {
+            state.WritePosition = startPosition;
+
+            throw;
+        }
+
+        // Fill in reserved lengths
+        ProtobufSerializer.WriteReservedLength(state.Buffer, kvlistLengthPos, state.WritePosition - (kvlistLengthPos + ReserveSizeForLength));
+        ProtobufSerializer.WriteReservedLength(state.Buffer, anyValueLengthPos, state.WritePosition - (anyValueLengthPos + ReserveSizeForLength));
+    }
+
     internal struct OtlpTagWriterState
     {
         public byte[] Buffer;

src/OpenTelemetry.Exporter.Zipkin/Implementation/ZipkinTagWriter.cs

@@ -2,6 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 using System.Buffers.Text;
+using System.Collections;
 using System.Globalization;
 using System.Text.Json;
 using OpenTelemetry.Internal;
@@ -66,4 +67,16 @@ protected override void OnUnsupportedTagDropped(
     }
 
     protected override bool TryWriteEmptyTag(ref Utf8JsonWriter state, string key, object? value) => false;
+
+    protected override void WriteKvListTag(ref Utf8JsonWriter writer, string key, IEnumerable<KeyValuePair<string, object?>> value, int? tagValueMaxLength)
+    {
+        var stringValue = Convert.ToString(value, CultureInfo.InvariantCulture);
+
+        if (stringValue is null)
+        {
+            return;
+        }
+
+        this.TryWriteTag(ref writer, key, stringValue);
+    }
 }

src/Shared/TagWriter/TagWriter.cs

@@ -64,6 +64,22 @@ public bool TryWriteTag(
             case double:
                 this.WriteFloatingPointTag(ref state, key, Convert.ToDouble(value, CultureInfo.InvariantCulture));
                 break;
+            case IEnumerable<KeyValuePair<string, object?>> kvList:
+                try
+                {
+                    this.WriteKvListTag(ref state, key, kvList, tagValueMaxLength);
+                }
+                catch (Exception ex) when (ex is IndexOutOfRangeException or ArgumentException)
+                {
+                    throw;
+                }
+                catch
+                {
+                    return this.LogUnsupportedTagTypeAndReturnFalse(key, value);
+                }
+
+                break;
+
             case Array array:
                 if (value.GetType() == typeof(byte[]) && this.TryWriteByteArrayTag(ref state, key, ((byte[])value).AsSpan()))
                 {
@@ -134,6 +150,8 @@ public bool TryWriteTag(
 
     protected abstract void WriteArrayTag(ref TTagState state, string key, ref TArrayState value);
 
+    protected abstract void WriteKvListTag(ref TTagState state, string key, IEnumerable<KeyValuePair<string, object?>> kvList, int? tagValueMaxLength);
+
     protected abstract void OnUnsupportedTagDropped(
         string tagKey,
         string tagValueTypeFullName);