[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/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)

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

@@ -17,6 +17,7 @@ namespace OpenTelemetry.Exporter.OpenTelemetryProtocol.Implementation.ExportClie
 internal sealed class OtlpGrpcExportClient : OtlpExportClient
 {
     public const string GrpcStatusDetailsHeader = "grpc-status-details-bin";
+    public const string GrpcAcceptEncodingHeader = "grpc-accept-encoding";
 
     // A gRPC message frame header is 5 bytes:
     //   byte 0     - Compression flag (0 = not compressed, 1 = compressed).
@@ -50,22 +51,22 @@ public OtlpGrpcExportClient(OtlpExporterOptions options, HttpClient httpClient,
 #endif
 
     /// <inheritdoc/>
-    public override ExportClientResponse SendExportRequest(byte[] buffer, int contentLength, DateTime deadlineUtc, CancellationToken cancellationToken = default)
+    protected override ExportClientResponse SendExportRequest(byte[] buffer, int contentLength, OtlpExportCompression compression, DateTime deadlineUtc, CancellationToken cancellationToken)
     {
         HttpResponseMessage? httpResponse = null;
 
         try
         {
-            using var httpRequest = this.CreateHttpRequest(buffer, contentLength);
+            using var httpRequest = this.CreateHttpRequest(buffer, contentLength, compression);
 
             // TE is required by some servers, e.g. C Core.
             // A missing TE header results in servers aborting the gRPC call.
             httpRequest.Headers.TryAddWithoutValidation("TE", "trailers");
 
-            if (this.CompressionEnabled)
+            if (compression != OtlpExportCompression.None)
             {
                 httpRequest.Headers.Remove("grpc-encoding");
-                httpRequest.Headers.TryAddWithoutValidation("grpc-encoding", "gzip");
+                httpRequest.Headers.TryAddWithoutValidation("grpc-encoding", GetEncodingName(compression));
             }
 
             httpResponse = this.SendHttpRequest(httpRequest, deadlineUtc, cancellationToken);
@@ -152,14 +153,26 @@ public override ExportClientResponse SendExportRequest(byte[] buffer, int conten
                 grpcStatusDetailsHeader = GrpcProtocolHelpers.GetHeaderValue(trailingHeaders, GrpcStatusDetailsHeader);
             }
 
+            // The encodings the server supports are only needed to decide whether a request that
+            // was rejected as UNIMPLEMENTED should be sent again with a different compression, and
+            // the response is disposed of before that decision is made.
+            string? grpcAcceptEncodingHeader = null;
+            if (status.StatusCode == StatusCode.Unimplemented)
+            {
+                grpcAcceptEncodingHeader =
+                    GrpcProtocolHelpers.GetHeaderValue(httpResponse.Headers, GrpcAcceptEncodingHeader) ??
+                    GrpcProtocolHelpers.GetHeaderValue(trailingHeaders, GrpcAcceptEncodingHeader);
+            }
+
             OpenTelemetryProtocolExporterEventSource.Log.ExportFailure(this.Endpoint, "Export failed due to unexpected status code.", status);
 
             return new ExportClientGrpcResponse(
                 success: false,
                 deadlineUtc: deadlineUtc,
                 exception: null,
                 status: status,
-                grpcStatusDetailsHeader: grpcStatusDetailsHeader);
+                grpcStatusDetailsHeader: grpcStatusDetailsHeader,
+                grpcAcceptEncodingHeader: grpcAcceptEncodingHeader);
         }
         catch (HttpRequestException ex) when (ex.InnerException is TimeoutException || IsTransientNetworkError(ex))
         {
@@ -232,20 +245,54 @@ public override ExportClientResponse SendExportRequest(byte[] buffer, int conten
         }
     }
 
-    protected override HttpContent CreateHttpContent(byte[] buffer, int contentLength) =>
-        this.CompressionEnabled
-            ? this.CreateGZipHttpContent(buffer, contentLength)
-            : base.CreateHttpContent(buffer, contentLength);
+    /// <inheritdoc/>
+    /// <remarks>
+    /// Servers that do not support a grpc-encoding return UNIMPLEMENTED and may list the encodings
+    /// they do support in grpc-accept-encoding. If that lists zstd, the error is unrelated to the
+    /// compression used. Many servers (such as grpc-go and so the OpenTelemetry Collector) do not
+    /// send the header, so its absence is treated as the encoding not being supported.
+    /// See https://github.com/grpc/grpc/blob/master/doc/compression.md.
+    /// </remarks>
+    protected override bool IsCompressionRejected(ExportClientResponse response) =>
+        response is ExportClientGrpcResponse { Status.StatusCode: StatusCode.Unimplemented } grpcResponse &&
+        !AcceptsEncoding(grpcResponse.GrpcAcceptEncodingHeader, "zstd");
+
+    protected override HttpContent CreateHttpContent(byte[] buffer, int contentLength, OtlpExportCompression compression) =>
+        compression != OtlpExportCompression.None
+            ? this.CreateCompressedHttpContent(buffer, contentLength, compression)
+            : base.CreateHttpContent(buffer, contentLength, compression);
+
+    private static string GetEncodingName(OtlpExportCompression compression) =>
+        compression == OtlpExportCompression.Zstd ? "zstd" : "gzip";
+
+    private static bool AcceptsEncoding(string? acceptEncodingHeader, string encoding)
+    {
+        if (acceptEncodingHeader is not { Length: > 0 })
+        {
+            return false;
+        }
+
+        // The value is a comma-separated list of encodings, such as "identity,deflate,gzip".
+        foreach (var candidate in acceptEncodingHeader.Split(','))
+        {
+            if (string.Equals(candidate.Trim(), encoding, StringComparison.OrdinalIgnoreCase))
+            {
+                return true;
+            }
+        }
+
+        return false;
+    }
 
     private static bool IsTransientNetworkError(HttpRequestException ex) =>
         ex.InnerException is SocketException { SocketErrorCode: SocketError.TimedOut or SocketError.ConnectionReset or SocketError.HostUnreachable or SocketError.ConnectionRefused };
 
-    private StreamContent CreateGZipHttpContent(byte[] buffer, int contentLength)
+    private StreamContent CreateCompressedHttpContent(byte[] buffer, int contentLength, OtlpExportCompression compression)
     {
-        // Build a gzip-compressed gRPC message frame:
-        //   byte 0     - Compression flag = 1 (gzip).
+        // Build a compressed gRPC message frame:
+        //   byte 0     - Compression flag = 1 (compressed with the grpc-encoding).
         //   bytes 1-4  - Compressed payload length in big-endian format.
-        //   bytes 5+   - Gzip-compressed protobuf payload.
+        //   bytes 5+   - Compressed protobuf payload.
 #if NET
         var compressedStream = new PooledBufferStream();
 #else
@@ -259,8 +306,15 @@ private StreamContent CreateGZipHttpContent(byte[] buffer, int contentLength)
         compressedStream.Write(GrpcFrameHeader, 0, GrpcFrameHeader.Length);
 #endif
 
-        using (var gzipStream = new GZipStream(compressedStream, CompressionLevel.Fastest, leaveOpen: true))
+#if NET11_0_OR_GREATER
+        if (compression == OtlpExportCompression.Zstd)
+        {
+            this.GetZstdCompressor().Compress(buffer.AsSpan(GrpcMessageHeaderSize, contentLength - GrpcMessageHeaderSize), compressedStream);
+        }
+        else
+#endif
         {
+            using var gzipStream = new GZipStream(compressedStream, CompressionLevel.Fastest, leaveOpen: true);
             gzipStream.Write(buffer, GrpcMessageHeaderSize, contentLength - GrpcMessageHeaderSize);
         }
 
@@ -286,7 +340,7 @@ private StreamContent CreateGZipHttpContent(byte[] buffer, int contentLength)
 
         if (OpenTelemetryProtocolExporterEventSource.Log.IsEnabled(EventLevel.Verbose, EventKeywords.All))
         {
-            OpenTelemetryProtocolExporterEventSource.Log.CompressedGrpcPayload("gzip", contentLength, compressedStream.Length);
+            OpenTelemetryProtocolExporterEventSource.Log.CompressedGrpcPayload(GetEncodingName(compression), contentLength, compressedStream.Length);
         }
 
         var content = new StreamContent(compressedStream);