[OTLP] Add support for zstd compression

src/OpenTelemetry.Exporter.OpenTelemetryProtocol/.publicApi/Stable/PublicAPI.Unshipped.txt

@@ -0,0 +1 @@
+OpenTelemetry.Exporter.OtlpExportCompression.Zstd = 2 -> OpenTelemetry.Exporter.OtlpExportCompression

src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/ExportClient/ExportClientGrpcResponse.cs

@@ -12,14 +12,21 @@ public ExportClientGrpcResponse(
         DateTime deadlineUtc,
         Exception? exception,
         Status? status,
-        string? grpcStatusDetailsHeader)
+        string? grpcStatusDetailsHeader,
+        string? grpcAcceptEncodingHeader = null)
         : base(success, deadlineUtc, exception)
     {
         this.Status = status;
         this.GrpcStatusDetailsHeader = grpcStatusDetailsHeader;
+        this.GrpcAcceptEncodingHeader = grpcAcceptEncodingHeader;
     }
 
     public Status? Status { get; }
 
     public string? GrpcStatusDetailsHeader { get; }
+
+    /// <summary>
+    /// Gets the value of the <c>grpc-accept-encoding</c> header of the response, if it was captured.
+    /// </summary>
+    public string? GrpcAcceptEncodingHeader { get; }
 }

src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/ExportClient/Grpc/GrpcProtocolHelpers.cs

@@ -82,6 +82,38 @@ public static Status GetStatusCore(HttpHeaders headers)
         return new Status((StatusCode)statusValue, grpcMessage ?? string.Empty);
     }
 
+    /// <summary>
+    /// Gets the value of a header that is a comma-separated list, joining the values of
+    /// the header if it was sent as more than one field, which is equivalent to one field
+    /// with the values joined by commas (RFC 9110 section 5.3).
+    /// </summary>
+    /// <param name="headers">The headers to get the value from.</param>
+    /// <param name="name">The name of the header.</param>
+    /// <returns>The value of the header, or <see langword="null"/> if it was not sent.</returns>
+    public static string? GetListHeaderValue(HttpHeaders? headers, string name)
+    {
+        if (headers == null)
+        {
+            return null;
+        }
+
+#if NET
+        if (!headers.NonValidated.TryGetValues(name, out var values))
+        {
+            return null;
+        }
+#else
+        if (!headers.TryGetValues(name, out var values))
+        {
+            return null;
+        }
+#endif
+
+        var joined = string.Join(",", values);
+
+        return joined.Length == 0 ? null : joined;
+    }
+
     public static string? GetHeaderValue(HttpHeaders? headers, string name, bool first = false)
     {
         if (headers == null)

src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/ExportClient/OtlpExportClient.cs

@@ -17,6 +17,12 @@ internal abstract class OtlpExportClient : IExportClient
     /// </summary>
     internal const int MaxLoggedResponseSizeBytes = 4 * 1024;
 
+    /// <summary>
+    /// The number of consecutive requests that are rejected when sent with Zstandard,
+    /// and then fail again when sent with GZip, after which Zstandard is no longer used.
+    /// </summary>
+    private const int MaxConsecutiveZstdFallbackFailures = 3;
+
     private static readonly Version Http2RequestVersion = new(2, 0);
     private static readonly TimeSpan MaxDeadlineDelay = TimeSpan.FromMilliseconds(int.MaxValue);
 
@@ -33,6 +39,15 @@ internal abstract class OtlpExportClient : IExportClient
         !OperatingSystem.IsBrowser();
 #endif
 
+    private readonly OtlpExportCompression compression;
+
+#if NET11_0_OR_GREATER
+    private readonly ZstdCompressor? zstdCompressor;
+#endif
+
+    private int zstdRejected;
+    private int zstdFallbackFailures;
+
     protected OtlpExportClient(OtlpExporterOptions options, HttpClient httpClient, string signalPath)
     {
         Guard.ThrowIfNull(options);
@@ -56,7 +71,23 @@ protected OtlpExportClient(OtlpExporterOptions options, HttpClient httpClient, s
         this.Endpoint = new UriBuilder(exporterEndpoint).Uri;
         this.Headers = options.GetHeaders<Dictionary<string, string>>((d, k, v) => d.Add(k, v));
         this.HttpClient = httpClient;
-        this.CompressionEnabled = options.Compression == OtlpExportCompression.GZip;
+        this.compression = options.Compression;
+
+        if (this.compression == OtlpExportCompression.Zstd)
+        {
+#if NET11_0_OR_GREATER
+            if (ZstdCompressor.IsSupported)
+            {
+                this.zstdCompressor = new ZstdCompressor();
+            }
+            else
+#endif
+            {
+                OpenTelemetryProtocolExporterEventSource.Log.ZstdCompressionNotSupported();
+                this.compression = OtlpExportCompression.GZip;
+            }
+        }
+
         this.MaxResponseSizeBytes = options.MaxResponseSizeBytes;
     }
 
@@ -66,7 +97,19 @@ protected OtlpExportClient(OtlpExporterOptions options, HttpClient httpClient, s
 
     internal IReadOnlyDictionary<string, string> Headers { get; }
 
-    internal bool CompressionEnabled { get; }
+    /// <summary>
+    /// Gets the compression the next request will be sent with.
+    /// </summary>
+    /// <remarks>
+    /// This is <see cref="OtlpExportCompression.GZip"/> if Zstandard compression was configured,
+    /// but the receiver rejected a request using it and then accepted it using GZip, or if
+    /// <see cref="MaxConsecutiveZstdFallbackFailures"/> requests in a row were rejected using it
+    /// and then failed again using GZip.
+    /// </remarks>
+    internal OtlpExportCompression Compression
+        => this.compression == OtlpExportCompression.Zstd && Volatile.Read(ref this.zstdRejected) != 0
+            ? OtlpExportCompression.GZip
+            : this.compression;
 
     /// <summary>
     /// Gets the maximum size, in bytes, of a response the exporter will accept.
@@ -81,12 +124,53 @@ protected OtlpExportClient(OtlpExporterOptions options, HttpClient httpClient, s
 
     internal virtual HttpCompletionOption CompletionOption => HttpCompletionOption.ResponseHeadersRead;
 
-    public abstract ExportClientResponse SendExportRequest(byte[] buffer, int contentLength, DateTime deadlineUtc, CancellationToken cancellationToken = default);
+    /// <inheritdoc/>
+    public ExportClientResponse SendExportRequest(byte[] buffer, int contentLength, DateTime deadlineUtc, CancellationToken cancellationToken = default)
+    {
+        var compression = this.Compression;
+        var response = this.SendExportRequest(buffer, contentLength, compression, deadlineUtc, cancellationToken);
+
+        if (compression == OtlpExportCompression.Zstd && this.IsCompressionRejected(response))
+        {
+            // Receivers are not required to support Zstandard, so a request they reject is sent again
+            // once with GZip so that it is not dropped solely because of the compression used.
+            // See https://github.com/open-telemetry/opentelemetry-specification/pull/5321.
+            OpenTelemetryProtocolExporterEventSource.Log.ZstdCompressionRejected(this.Endpoint);
+
+            response = this.SendExportRequest(buffer, contentLength, OtlpExportCompression.GZip, deadlineUtc, cancellationToken);
+
+            if (response.Success)
+            {
+                // Only stop using Zstandard immediately if GZip succeeded, as otherwise the
+                // rejection was most likely caused by something other than the compression used.
+                Volatile.Write(ref this.zstdRejected, 1);
+            }
+            else if (Interlocked.Increment(ref this.zstdFallbackFailures) >= MaxConsecutiveZstdFallbackFailures)
+            {
+                // The receiver keeps rejecting requests whatever their compression, such as when
+                // it responds with 400 to a payload it does not accept. Stop sending each
+                // request twice, which doubles the bandwidth and CPU used, for no benefit.
+                Volatile.Write(ref this.zstdRejected, 1);
+            }
+        }
+        else if (compression == OtlpExportCompression.Zstd)
+        {
+            // The receiver did not reject the compression, so the failures were not consecutive.
+            Volatile.Write(ref this.zstdFallbackFailures, 0);
+        }
+
+        return response;
+    }
 
     /// <inheritdoc/>
     public bool Shutdown(int timeoutMilliseconds)
     {
         this.HttpClient.CancelPendingRequests();
+
+#if NET11_0_OR_GREATER
+        this.zstdCompressor?.Shutdown();
+#endif
+
         return true;
     }
 
@@ -207,7 +291,26 @@ protected bool IsResponseTooLarge(
         return false;
     }
 
-    protected HttpRequestMessage CreateHttpRequest(byte[] buffer, int contentLength)
+    /// <summary>
+    /// Sends an export request using the specified compression.
+    /// </summary>
+    /// <param name="buffer">The request body.</param>
+    /// <param name="contentLength">The number of bytes within <paramref name="buffer"/> that make up the message.</param>
+    /// <param name="compression">The compression to apply to the request body.</param>
+    /// <param name="deadlineUtc">The deadline time in UTC for the export request to finish.</param>
+    /// <param name="cancellationToken">An optional token for canceling the call.</param>
+    /// <returns><see cref="ExportClientResponse"/>.</returns>
+    protected abstract ExportClientResponse SendExportRequest(byte[] buffer, int contentLength, OtlpExportCompression compression, DateTime deadlineUtc, CancellationToken cancellationToken);
+
+    /// <summary>
+    /// Determines whether a failed response indicates that the receiver does not
+    /// support the compression the request was sent with.
+    /// </summary>
+    /// <param name="response">The response to inspect.</param>
+    /// <returns><see langword="true"/> if the receiver rejected the request's compression.</returns>
+    protected abstract bool IsCompressionRejected(ExportClientResponse response);
+
+    protected HttpRequestMessage CreateHttpRequest(byte[] buffer, int contentLength, OtlpExportCompression compression)
     {
         var request = new HttpRequestMessage(HttpMethod.Post, this.Endpoint);
 
@@ -232,7 +335,7 @@ protected HttpRequestMessage CreateHttpRequest(byte[] buffer, int contentLength)
             }
         }
 
-        request.Content = this.CreateHttpContent(buffer, contentLength);
+        request.Content = this.CreateHttpContent(buffer, contentLength, compression);
 
         return request;
     }
@@ -243,8 +346,9 @@ protected HttpRequestMessage CreateHttpRequest(byte[] buffer, int contentLength)
     /// </summary>
     /// <param name="buffer">The serialized protobuf payload buffer.</param>
     /// <param name="contentLength">The number of bytes within <paramref name="buffer"/> that make up the message.</param>
+    /// <param name="compression">The compression to apply to the payload.</param>
     /// <returns>An <see cref="HttpContent"/> representing the export payload.</returns>
-    protected virtual HttpContent CreateHttpContent(byte[] buffer, int contentLength)
+    protected virtual HttpContent CreateHttpContent(byte[] buffer, int contentLength, OtlpExportCompression compression)
     {
 #if NET
         var content = new ReadOnlyMemoryContent(buffer.AsMemory(0, contentLength));
@@ -255,6 +359,15 @@ protected virtual HttpContent CreateHttpContent(byte[] buffer, int contentLength
         return content;
     }
 
+#if NET11_0_OR_GREATER
+    /// <summary>
+    /// Gets the <see cref="ZstdCompressor"/> to use for requests compressed with Zstandard.
+    /// </summary>
+    /// <returns>The <see cref="ZstdCompressor"/> to use.</returns>
+    protected ZstdCompressor GetZstdCompressor()
+        => this.zstdCompressor ?? throw new InvalidOperationException("Zstandard compression is not enabled.");
+#endif
+
     protected HttpResponseMessage SendHttpRequest(HttpRequestMessage request, DateTime deadlineUtc, CancellationToken cancellationToken)
     {
         if (GetTimeUntilDeadline(deadlineUtc) is not { } remaining)