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

@@ -50,22 +50,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);
@@ -232,20 +232,31 @@ 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.
+    /// 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 };
+
+    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 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 +270,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 +304,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);

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

@@ -2,6 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 using System.Diagnostics.Tracing;
+using System.Net;
 #if NETFRAMEWORK
 using System.Net.Http;
 #endif
@@ -23,14 +24,14 @@ internal OtlpHttpExportClient(OtlpExporterOptions options, HttpClient httpClient
     internal override MediaTypeHeaderValue MediaTypeHeader => MediaHeaderValue;
 
     /// <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)
     {
         try
         {
-            using var httpRequest = this.CreateHttpRequest(buffer, contentLength);
+            using var httpRequest = this.CreateHttpRequest(buffer, contentLength, compression);
 
 #if NETFRAMEWORK
-            if (this.CompressionEnabled)
+            if (compression != OtlpExportCompression.None)
             {
                 // The length of the compressed content is not known up front, so unless the request
                 // is sent chunked HttpClientHandler buffers the entire compressed body in memory first.
@@ -96,8 +97,20 @@ public override ExportClientResponse SendExportRequest(byte[] buffer, int conten
         }
     }
 
-    protected override HttpContent CreateHttpContent(byte[] buffer, int contentLength) =>
-        this.CompressionEnabled
-            ? new GZipHttpContent(buffer, contentLength, this.MediaTypeHeader)
-            : base.CreateHttpContent(buffer, contentLength);
+    /// <inheritdoc/>
+    /// <remarks>
+    /// Receivers that do not support a Content-Encoding should return 415, but may return 400.
+    /// </remarks>
+    protected override bool IsCompressionRejected(ExportClientResponse response) =>
+        response is ExportClientHttpResponse { StatusCode: HttpStatusCode.UnsupportedMediaType or HttpStatusCode.BadRequest };
+
+    protected override HttpContent CreateHttpContent(byte[] buffer, int contentLength, OtlpExportCompression compression) =>
+        compression switch
+        {
+            OtlpExportCompression.GZip => new GZipHttpContent(buffer, contentLength, this.MediaTypeHeader),
+#if NET11_0_OR_GREATER
+            OtlpExportCompression.Zstd => new ZstdHttpContent(buffer, contentLength, this.MediaTypeHeader, this.GetZstdCompressor()),
+#endif
+            _ => base.CreateHttpContent(buffer, contentLength, compression),
+        };
 }