From 332fcec1cc322e17697887a45ae9665fded63521 Mon Sep 17 00:00:00 2001 From: martincostello Date: Wed, 28 Oct 2026 14:19:07 +0000 Subject: [PATCH 1/7] [OTLP] Add support for zstd compression Add support for using zstd compression for OTLP when targeting `net11.0`. Based on https://github.com/open-telemetry/opentelemetry-specification/pull/5321. --- .../.publicApi/Stable/PublicAPI.Unshipped.txt | 1 + .../ExportClient/OtlpExportClient.cs | 104 +++++++- .../ExportClient/OtlpGrpcExportClient.cs | 46 +++- .../ExportClient/OtlpHttpExportClient.cs | 27 +- .../ExportClient/ZstdCompressor.cs | 115 ++++++++ .../ExportClient/ZstdHttpContent.cs | 40 +++ ...penTelemetryProtocolExporterEventSource.cs | 23 ++ .../OtlpExportCompression.cs | 16 ++ .../OtlpExporterOptions.cs | 4 + .../README.md | 19 +- .../OtlpExporterCompressionBenchmarks.cs | 4 + .../OtlpExporterOptionsTests.cs | 3 + .../OtlpGrpcExportClientTests.cs | 124 +++++++++ .../OtlpHttpExportClientTests.cs | 249 ++++++++++++++++++ 14 files changed, 743 insertions(+), 32 deletions(-) create mode 100644 src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/ExportClient/ZstdCompressor.cs create mode 100644 src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/ExportClient/ZstdHttpContent.cs diff --git a/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/.publicApi/Stable/PublicAPI.Unshipped.txt b/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/.publicApi/Stable/PublicAPI.Unshipped.txt index e69de29bb2d..ecc968e4fc8 100644 --- a/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/.publicApi/Stable/PublicAPI.Unshipped.txt +++ b/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/.publicApi/Stable/PublicAPI.Unshipped.txt @@ -0,0 +1 @@ +OpenTelemetry.Exporter.OtlpExportCompression.Zstd = 2 -> OpenTelemetry.Exporter.OtlpExportCompression diff --git a/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/ExportClient/OtlpExportClient.cs b/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/ExportClient/OtlpExportClient.cs index d3c59e9c4dc..83c67d6d479 100644 --- a/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/ExportClient/OtlpExportClient.cs +++ b/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/ExportClient/OtlpExportClient.cs @@ -33,6 +33,14 @@ 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; + protected OtlpExportClient(OtlpExporterOptions options, HttpClient httpClient, string signalPath) { Guard.ThrowIfNull(options); @@ -56,7 +64,23 @@ protected OtlpExportClient(OtlpExporterOptions options, HttpClient httpClient, s this.Endpoint = new UriBuilder(exporterEndpoint).Uri; this.Headers = options.GetHeaders>((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 +90,17 @@ protected OtlpExportClient(OtlpExporterOptions options, HttpClient httpClient, s internal IReadOnlyDictionary Headers { get; } - internal bool CompressionEnabled { get; } + /// + /// Gets the compression the next request will be sent with. + /// + /// + /// This is if Zstandard compression was configured, + /// but the receiver rejected a request using it and then accepted it using GZip. + /// + internal OtlpExportCompression Compression + => this.compression == OtlpExportCompression.Zstd && Volatile.Read(ref this.zstdRejected) != 0 + ? OtlpExportCompression.GZip + : this.compression; /// /// Gets the maximum size, in bytes, of a response the exporter will accept. @@ -81,12 +115,41 @@ 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); + /// + 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 if GZip succeeded, as otherwise the rejection + // was most likely caused by something other than the compression used. + Volatile.Write(ref this.zstdRejected, 1); + } + } + + return response; + } /// public bool Shutdown(int timeoutMilliseconds) { this.HttpClient.CancelPendingRequests(); + +#if NET11_0_OR_GREATER + this.zstdCompressor?.Shutdown(); +#endif + return true; } @@ -207,7 +270,26 @@ protected bool IsResponseTooLarge( return false; } - protected HttpRequestMessage CreateHttpRequest(byte[] buffer, int contentLength) + /// + /// Sends an export request using the specified compression. + /// + /// The request body. + /// The number of bytes within that make up the message. + /// The compression to apply to the request body. + /// The deadline time in UTC for the export request to finish. + /// An optional token for canceling the call. + /// . + protected abstract ExportClientResponse SendExportRequest(byte[] buffer, int contentLength, OtlpExportCompression compression, DateTime deadlineUtc, CancellationToken cancellationToken); + + /// + /// Determines whether a failed response indicates that the receiver does not + /// support the compression the request was sent with. + /// + /// The response to inspect. + /// if the receiver rejected the request's compression. + 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 +314,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 +325,9 @@ protected HttpRequestMessage CreateHttpRequest(byte[] buffer, int contentLength) /// /// The serialized protobuf payload buffer. /// The number of bytes within that make up the message. + /// The compression to apply to the payload. /// An representing the export payload. - 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 +338,15 @@ protected virtual HttpContent CreateHttpContent(byte[] buffer, int contentLength return content; } +#if NET11_0_OR_GREATER + /// + /// Gets the to use for requests compressed with Zstandard. + /// + /// The to use. + 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) diff --git a/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/ExportClient/OtlpGrpcExportClient.cs b/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/ExportClient/OtlpGrpcExportClient.cs index 9951ed6a8a7..0b431db3c43 100644 --- a/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/ExportClient/OtlpGrpcExportClient.cs +++ b/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/ExportClient/OtlpGrpcExportClient.cs @@ -50,22 +50,22 @@ public OtlpGrpcExportClient(OtlpExporterOptions options, HttpClient httpClient, #endif /// - 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); + /// + /// + /// Servers that do not support a grpc-encoding return UNIMPLEMENTED. + /// See https://github.com/grpc/grpc/blob/master/doc/compression.md. + /// + 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); diff --git a/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/ExportClient/OtlpHttpExportClient.cs b/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/ExportClient/OtlpHttpExportClient.cs index 89a8146562f..540333f7b24 100644 --- a/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/ExportClient/OtlpHttpExportClient.cs +++ b/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; /// - 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); + /// + /// + /// Receivers that do not support a Content-Encoding should return 415, but may return 400. + /// + 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), + }; } diff --git a/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/ExportClient/ZstdCompressor.cs b/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/ExportClient/ZstdCompressor.cs new file mode 100644 index 00000000000..af47c0ef774 --- /dev/null +++ b/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/ExportClient/ZstdCompressor.cs @@ -0,0 +1,115 @@ +// Copyright The OpenTelemetry Authors +// SPDX-License-Identifier: Apache-2.0 + +#if NET11_0_OR_GREATER + +using System.Diagnostics; +using System.IO.Compression; +using System.Runtime.Versioning; + +namespace OpenTelemetry.Exporter.OpenTelemetryProtocol.Implementation.ExportClient; + +/// +/// Compresses OTLP payloads with Zstandard, reusing a single native encoder between exports. +/// +/// +/// See https://github.com/open-telemetry/opentelemetry-specification/pull/5321. +/// +internal sealed class ZstdCompressor +{ + /// + /// The largest window an encoder should use (8 MiB), as recommended for HTTP + /// interoperability by RFC 9659 section 3. + /// + internal const int MaxWindowLog2 = 23; + + private ZstandardEncoder? cachedEncoder; + + [UnsupportedOSPlatformGuard("browser")] + [UnsupportedOSPlatformGuard("wasi")] + internal static bool IsSupported => !OperatingSystem.IsBrowser() && !OperatingSystem.IsWasi(); + + public void Compress(ReadOnlySpan source, Stream destination) + { + Debug.Assert(IsSupported, "Zstandard compression is not supported on this platform."); + + var encoder = this.RentEncoder(); + + try + { + using var zstd = CreateStream(destination, encoder, source.Length); + zstd.Write(source); + } + catch + { + // The encoder's state is unknown, so do not reuse it. + encoder.Dispose(); + throw; + } + + this.ReturnEncoder(encoder); + } + + public async Task CompressAsync(ReadOnlyMemory source, Stream destination, CancellationToken cancellationToken) + { + Debug.Assert(IsSupported, "Zstandard compression is not supported on this platform."); + + var encoder = this.RentEncoder(); + + try + { + var zstd = CreateStream(destination, encoder, source.Length); + await using (zstd.ConfigureAwait(false)) + { + await zstd.WriteAsync(source, cancellationToken).ConfigureAwait(false); + } + } + catch + { + encoder.Dispose(); + throw; + } + + this.ReturnEncoder(encoder); + } + + public void Shutdown() + { + Debug.Assert(IsSupported, "Zstandard compression is not supported on this platform."); + + Interlocked.Exchange(ref this.cachedEncoder, null)?.Dispose(); + } + + [UnsupportedOSPlatform("browser")] + [UnsupportedOSPlatform("wasi")] + private static ZstandardStream CreateStream(Stream destination, ZstandardEncoder encoder, int sourceLength) + { + var zstd = new ZstandardStream(destination, encoder, leaveOpen: true); + + // Pledging the size up front writes a single frame that records the content size, and lets + // the encoder size its window (and so its memory use) to the payload rather than MaxWindowLog2. + zstd.SetSourceLength(sourceLength); + + return zstd; + } + + [UnsupportedOSPlatform("browser")] + [UnsupportedOSPlatform("wasi")] + private ZstandardEncoder RentEncoder() + => Interlocked.Exchange(ref this.cachedEncoder, null) ?? new ZstandardEncoder(ZstandardCompressionOptions.DefaultQuality, MaxWindowLog2); + + [UnsupportedOSPlatform("browser")] + [UnsupportedOSPlatform("wasi")] + private void ReturnEncoder(ZstandardEncoder encoder) + { + encoder.Reset(); + + if (Interlocked.CompareExchange(ref this.cachedEncoder, encoder, null) != null) + { + // Another export returned an encoder first. + encoder.Dispose(); + } + } +} + +#endif diff --git a/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/ExportClient/ZstdHttpContent.cs b/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/ExportClient/ZstdHttpContent.cs new file mode 100644 index 00000000000..1984f07e5a1 --- /dev/null +++ b/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/ExportClient/ZstdHttpContent.cs @@ -0,0 +1,40 @@ +// Copyright The OpenTelemetry Authors +// SPDX-License-Identifier: Apache-2.0 + +#if NET11_0_OR_GREATER + +using System.Net; +using System.Net.Http.Headers; + +namespace OpenTelemetry.Exporter.OpenTelemetryProtocol.Implementation.ExportClient; + +internal sealed class ZstdHttpContent : HttpContent +{ + private readonly ReadOnlyMemory buffer; + private readonly ZstdCompressor compressor; + + public ZstdHttpContent(byte[] buffer, int contentLength, MediaTypeHeaderValue mediaType, ZstdCompressor compressor) + { + this.buffer = buffer.AsMemory(0, contentLength); + this.compressor = compressor; + this.Headers.ContentType = mediaType; + this.Headers.ContentEncoding.Add("zstd"); + } + + protected override bool TryComputeLength(out long length) + { + length = -1; + return false; + } + + protected override Task SerializeToStreamAsync(Stream stream, TransportContext? context) + => this.SerializeToStreamAsync(stream, context, CancellationToken.None); + + protected override Task SerializeToStreamAsync(Stream stream, TransportContext? context, CancellationToken cancellationToken) + => this.compressor.CompressAsync(this.buffer, stream, cancellationToken); + + protected override void SerializeToStream(Stream stream, TransportContext? context, CancellationToken cancellationToken) + => this.compressor.Compress(this.buffer.Span, stream); +} + +#endif diff --git a/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/OpenTelemetryProtocolExporterEventSource.cs b/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/OpenTelemetryProtocolExporterEventSource.cs index eb5dcb61e57..fb3aa3fa941 100644 --- a/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/OpenTelemetryProtocolExporterEventSource.cs +++ b/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/OpenTelemetryProtocolExporterEventSource.cs @@ -386,6 +386,29 @@ internal void HttpClientFactoryUnavailable(string httpClientName, string excepti internal void RetryDeadlineExceeded() => this.WriteEvent(43); + [Event( + 44, + Message = "Zstandard compression is not supported on this platform. GZip compression will be used instead.", + Level = EventLevel.Warning)] + internal void ZstdCompressionNotSupported() + => this.WriteEvent(44); + + [NonEvent] + internal void ZstdCompressionRejected(Uri endpoint) + { + if (Log.IsEnabled(EventLevel.Warning, EventKeywords.All)) + { + this.ZstdCompressionRejected(RedactEndpointUri(endpoint)); + } + } + + [Event( + 45, + Message = "The request to endpoint '{0}' compressed with Zstandard was rejected. The request will be sent again compressed with GZip.", + Level = EventLevel.Warning)] + internal void ZstdCompressionRejected(string endpoint) + => this.WriteEvent(45, endpoint); + private static string RedactEndpointUri(Uri endpoint) { var authority = endpoint.GetComponents(UriComponents.SchemeAndServer, UriFormat.UriEscaped); diff --git a/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/OtlpExportCompression.cs b/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/OtlpExportCompression.cs index 04881a05b1d..2bb2f717b56 100644 --- a/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/OtlpExportCompression.cs +++ b/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/OtlpExportCompression.cs @@ -20,4 +20,20 @@ public enum OtlpExportCompression /// Compress with GZip. /// GZip = 1, + + /// + /// Compress with Zstandard (zstd). + /// + /// + /// Notes: + /// + /// Zstandard compression is only available when running on .NET 11 + /// or later, and is not available in the browser or on WASI. Where it is + /// not available, is used instead. + /// If the receiver rejects a Zstandard-compressed request, the + /// request is sent again once using . If that succeeds, + /// is used for all subsequent requests. + /// + /// + Zstd = 2, } diff --git a/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/OtlpExporterOptions.cs b/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/OtlpExporterOptions.cs index cdea758a8c3..36c9a9fc430 100644 --- a/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/OtlpExporterOptions.cs +++ b/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/OtlpExporterOptions.cs @@ -335,6 +335,10 @@ internal static bool TryParseCompression(string value, out OtlpExportCompression result = OtlpExportCompression.GZip; return true; + case "ZSTD": + result = OtlpExportCompression.Zstd; + return true; + default: result = default; return false; diff --git a/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/README.md b/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/README.md index 27d66549d32..d78fb66a01f 100644 --- a/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/README.md +++ b/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/README.md @@ -343,7 +343,16 @@ appBuilder.Services.AddOptions() > traces, the full URL will look like `http://your-custom-endpoint/v1/traces`. * `Compression`: Optionally compress the export payload. The default is no - compression. The only supported compression option is `OtlpExportCompression.GZip`. + compression. The supported compression options are `OtlpExportCompression.GZip` + and `OtlpExportCompression.Zstd`. + + > [!NOTE] + > `OtlpExportCompression.Zstd` requires .NET 11 or later and is not available + > in when using Blazor. Where Zstandard is not available, GZip is used + > instead. Receivers are not required to support Zstandard, so if a receiver + > rejects a Zstandard-compressed request (with `415`/`400` for + > `http/protobuf`, or `UNIMPLEMENTED` for `grpc`), the request is sent again + > using GZip, and GZip is then used for all further requests if that succeeds. * `Headers`: Optional headers for the connection. @@ -453,7 +462,7 @@ or reader | Environment variable | `OtlpExporterOptions` property | | ---------------------------------| ---------------------------------------| - | `OTEL_EXPORTER_OTLP_COMPRESSION` | `Compression` (`none` or `gzip`) | + | `OTEL_EXPORTER_OTLP_COMPRESSION` | `Compression` (`none`/`gzip`/`zstd`) | | `OTEL_EXPORTER_OTLP_ENDPOINT` | `Endpoint` | | `OTEL_EXPORTER_OTLP_HEADERS` | `Headers` | | `OTEL_EXPORTER_OTLP_PROTOCOL` | `Protocol` (`grpc` or `http/protobuf`) | @@ -491,7 +500,7 @@ or reader | Environment variable | `OtlpExporterOptions` property | UseOtlpExporter | AddOtlpExporter | | --------------------------------------| ---------------------------------------|-----------------|-----------------| - | `OTEL_EXPORTER_OTLP_LOGS_COMPRESSION` | `Compression` (`none` or `gzip`) | Supported | Not supported | + | `OTEL_EXPORTER_OTLP_LOGS_COMPRESSION` | `Compression` (`none`/`gzip`/`zstd`) | Supported | Not supported | | `OTEL_EXPORTER_OTLP_LOGS_ENDPOINT` | `Endpoint` | Supported | Not supported | | `OTEL_EXPORTER_OTLP_LOGS_HEADERS` | `Headers` | Supported | Not supported | | `OTEL_EXPORTER_OTLP_LOGS_PROTOCOL` | `Protocol` (`grpc` or `http/protobuf`) | Supported | Not supported | @@ -529,7 +538,7 @@ or reader | Environment variable | `OtlpExporterOptions` property | UseOtlpExporter | AddOtlpExporter | | -----------------------------------------| ---------------------------------------|-----------------|-----------------| - | `OTEL_EXPORTER_OTLP_METRICS_COMPRESSION` | `Compression` (`none` or `gzip`) | Supported | Not supported | + | `OTEL_EXPORTER_OTLP_METRICS_COMPRESSION` | `Compression` (`none`/`gzip`/`zstd`) | Supported | Not supported | | `OTEL_EXPORTER_OTLP_METRICS_ENDPOINT` | `Endpoint` | Supported | Not supported | | `OTEL_EXPORTER_OTLP_METRICS_HEADERS` | `Headers` | Supported | Not supported | | `OTEL_EXPORTER_OTLP_METRICS_PROTOCOL` | `Protocol` (`grpc` or `http/protobuf`) | Supported | Not supported | @@ -553,7 +562,7 @@ or reader | Environment variable | `OtlpExporterOptions` property | UseOtlpExporter | AddOtlpExporter | | ----------------------------------------| ---------------------------------------|-----------------|-----------------| - | `OTEL_EXPORTER_OTLP_TRACES_COMPRESSION` | `Compression` (`none` or `gzip`) | Supported | Not supported | + | `OTEL_EXPORTER_OTLP_TRACES_COMPRESSION` | `Compression` (`none`/`gzip`/`zstd`) | Supported | Not supported | | `OTEL_EXPORTER_OTLP_TRACES_ENDPOINT` | `Endpoint` | Supported | Not supported | | `OTEL_EXPORTER_OTLP_TRACES_HEADERS` | `Headers` | Supported | Not supported | | `OTEL_EXPORTER_OTLP_TRACES_PROTOCOL` | `Protocol` (`grpc` or `http/protobuf`) | Supported | Not supported | diff --git a/test/Benchmarks/Exporter/OtlpExporterCompressionBenchmarks.cs b/test/Benchmarks/Exporter/OtlpExporterCompressionBenchmarks.cs index 9a2cea4304c..aec6927952e 100644 --- a/test/Benchmarks/Exporter/OtlpExporterCompressionBenchmarks.cs +++ b/test/Benchmarks/Exporter/OtlpExporterCompressionBenchmarks.cs @@ -31,7 +31,11 @@ public class OtlpExporterCompressionBenchmarks private Activity? activity; private CircularBuffer? activityBatch; +#if NET11_0_OR_GREATER + [Params(OtlpExportCompression.None, OtlpExportCompression.GZip, OtlpExportCompression.Zstd)] +#else [Params(OtlpExportCompression.None, OtlpExportCompression.GZip)] +#endif public OtlpExportCompression Compression { get; set; } [GlobalSetup] diff --git a/test/OpenTelemetry.Exporter.OpenTelemetryProtocol.Tests/OtlpExporterOptionsTests.cs b/test/OpenTelemetry.Exporter.OpenTelemetryProtocol.Tests/OtlpExporterOptionsTests.cs index 401f536de27..8d45b649760 100644 --- a/test/OpenTelemetry.Exporter.OpenTelemetryProtocol.Tests/OtlpExporterOptionsTests.cs +++ b/test/OpenTelemetry.Exporter.OpenTelemetryProtocol.Tests/OtlpExporterOptionsTests.cs @@ -204,6 +204,9 @@ public void OtlpExporterOptions_SetterOverridesEnvironmentVariable() [InlineData("none", OtlpExportCompression.None)] [InlineData("None", OtlpExportCompression.None)] [InlineData("NONE", OtlpExportCompression.None)] + [InlineData("zstd", OtlpExportCompression.Zstd)] + [InlineData("Zstd", OtlpExportCompression.Zstd)] + [InlineData("ZSTD", OtlpExportCompression.Zstd)] public void OtlpExporterOptions_AppliesCompressionFromEnvironment(string value, OtlpExportCompression expected) { var values = new Dictionary diff --git a/test/OpenTelemetry.Exporter.OpenTelemetryProtocol.Tests/OtlpGrpcExportClientTests.cs b/test/OpenTelemetry.Exporter.OpenTelemetryProtocol.Tests/OtlpGrpcExportClientTests.cs index 1454c6e866a..4fc85b7dde3 100644 --- a/test/OpenTelemetry.Exporter.OpenTelemetryProtocol.Tests/OtlpGrpcExportClientTests.cs +++ b/test/OpenTelemetry.Exporter.OpenTelemetryProtocol.Tests/OtlpGrpcExportClientTests.cs @@ -228,6 +228,94 @@ public void SendExportRequest_GrpcExport_GzipCompression_PayloadDecompressesToOr h => h.Key == "grpc-encoding" && h.Value.Contains("gzip")); } +#if NET11_0_OR_GREATER + [Fact] + public void SendExportRequest_GrpcExport_ZstdCompression_PayloadDecompressesToOriginalProtobuf() + { + var protobufPayload = "grpc test payload for compression"u8.ToArray(); + var buffer = BuildGrpcFrame(protobufPayload); + + using var testHandler = new TestGrpcMessageHandler(); + using var httpClient = new HttpClient(testHandler, disposeHandler: false); + + var exportClient = CreateZstdExportClient(httpClient); + + var response = exportClient.SendExportRequest( + buffer, + buffer.Length, + DateTime.UtcNow.AddSeconds(10), + TestContext.Current.CancellationToken); + + Assert.True(response.Success); + + Assert.NotNull(testHandler.CapturedRequestBytes); + var content = testHandler.CapturedRequestBytes; + + Assert.Equal(1, content[0]); + + var compressedLength = (int)BinaryPrimitives.ReadUInt32BigEndian(content.AsSpan(1, 4)); + Assert.Equal(content.Length - GrpcHeaderSize, compressedLength); + + using (var input = new MemoryStream(content, GrpcHeaderSize, compressedLength)) + using (var zstd = new ZstandardStream(input, CompressionMode.Decompress)) + using (var output = new MemoryStream()) + { + zstd.CopyTo(output); + Assert.Equal(protobufPayload, output.ToArray()); + } + + Assert.NotNull(testHandler.CapturedRequestHeaders); + Assert.Contains( + testHandler.CapturedRequestHeaders, + h => h.Key == "grpc-encoding" && h.Value.Contains("zstd")); + } + + [Fact] + public void SendExportRequest_GrpcExport_ZstdCompression_Unimplemented_RetriesWithGzipAndThenUsesGzip() + { + const int ZstdCompressionRejectedEventId = 45; + + var buffer = BuildGrpcFrame("grpc test payload for compression"u8.ToArray()); + + using var listener = new TestEventListener(OpenTelemetryProtocolExporterEventSource.Log, EventLevel.Warning); + using var testHandler = new GrpcEncodingMessageHandler(encoding => encoding == "zstd" ? StatusCode.Unimplemented : StatusCode.OK); + using var httpClient = new HttpClient(testHandler, disposeHandler: false); + + var exportClient = CreateZstdExportClient(httpClient); + + var response = exportClient.SendExportRequest(buffer, buffer.Length, DateTime.UtcNow.AddSeconds(10), TestContext.Current.CancellationToken); + + Assert.True(response.Success); + Assert.Equal(["zstd", "gzip"], testHandler.GrpcEncodings); + Assert.Contains(listener.Messages, e => e.EventId == ZstdCompressionRejectedEventId); + Assert.Equal(OtlpExportCompression.GZip, exportClient.Compression); + + response = exportClient.SendExportRequest(buffer, buffer.Length, DateTime.UtcNow.AddSeconds(10), TestContext.Current.CancellationToken); + + Assert.True(response.Success); + Assert.Equal(["zstd", "gzip", "gzip"], testHandler.GrpcEncodings); + } + + [Theory] + [InlineData((int)StatusCode.InvalidArgument)] + [InlineData((int)StatusCode.Unavailable)] + public void SendExportRequest_GrpcExport_ZstdCompression_OtherFailure_DoesNotRetryWithGzip(int statusCode) + { + var buffer = BuildGrpcFrame("grpc test payload for compression"u8.ToArray()); + + using var testHandler = new GrpcEncodingMessageHandler(_ => (StatusCode)statusCode); + using var httpClient = new HttpClient(testHandler, disposeHandler: false); + + var exportClient = CreateZstdExportClient(httpClient); + + var response = exportClient.SendExportRequest(buffer, buffer.Length, DateTime.UtcNow.AddSeconds(10), TestContext.Current.CancellationToken); + + Assert.False(response.Success); + Assert.Equal(["zstd"], testHandler.GrpcEncodings); + Assert.Equal(OtlpExportCompression.Zstd, exportClient.Compression); + } +#endif + [Fact] public void SendExportRequest_Timeout_ReturnsRetryableDeadlineExceededResponse() { @@ -400,6 +488,18 @@ private static byte[] BuildGrpcFrame(byte[] protobufPayload) return frame; } +#if NET11_0_OR_GREATER + private static OtlpGrpcExportClient CreateZstdExportClient(HttpClient httpClient) => + new( + new OtlpExporterOptions + { + Endpoint = new Uri("http://localhost:4317"), + Compression = OtlpExportCompression.Zstd, + }, + httpClient, + string.Empty); +#endif + private static byte[] Decompress(byte[] compressed) { using var input = new MemoryStream(compressed); @@ -530,6 +630,30 @@ public override void Flush() } } +#if NET11_0_OR_GREATER + private sealed class GrpcEncodingMessageHandler(Func getStatusCode) : HttpMessageHandler + { + public List GrpcEncodings { get; } = []; + + protected override Task SendAsync(HttpRequestMessage request, CancellationToken cancellationToken) + { + var encoding = request.Headers.TryGetValues("grpc-encoding", out var values) ? values.SingleOrDefault() : null; + + this.GrpcEncodings.Add(encoding); + + var response = new HttpResponseMessage(System.Net.HttpStatusCode.OK) + { + RequestMessage = request, + Content = new ByteArrayContent([]), + }; + + response.Headers.Add("grpc-status", ((int)getStatusCode(encoding)).ToString(System.Globalization.CultureInfo.InvariantCulture)); + + return Task.FromResult(response); + } + } +#endif + private sealed class TestGrpcMessageHandler : HttpMessageHandler { public byte[]? CapturedRequestBytes { get; private set; } diff --git a/test/OpenTelemetry.Exporter.OpenTelemetryProtocol.Tests/OtlpHttpExportClientTests.cs b/test/OpenTelemetry.Exporter.OpenTelemetryProtocol.Tests/OtlpHttpExportClientTests.cs index dd33809a527..8922e7be4af 100644 --- a/test/OpenTelemetry.Exporter.OpenTelemetryProtocol.Tests/OtlpHttpExportClientTests.cs +++ b/test/OpenTelemetry.Exporter.OpenTelemetryProtocol.Tests/OtlpHttpExportClientTests.cs @@ -4,6 +4,9 @@ #if NETFRAMEWORK using System.Net.Http; #endif +#if NET11_0_OR_GREATER +using System.Buffers.Binary; +#endif using System.Diagnostics; using System.Diagnostics.Tracing; using System.IO.Compression; @@ -197,6 +200,170 @@ public void SendExportRequest_WithGzipCompression_IsCompressed() #endif } +#if NET11_0_OR_GREATER + [Fact] + public void SendExportRequest_WithZstdCompression_IsCompressed() + { + var payload = "00000000000000000000000000000000"u8.ToArray(); + + using var testHandler = new TestHttpMessageHandler(); + using var httpClient = new HttpClient(testHandler, disposeHandler: false); + + var exportClient = CreateZstdExportClient(httpClient); + + var response = exportClient.SendExportRequest( + payload, + payload.Length, + DateTime.UtcNow.AddSeconds(10), + TestContext.Current.CancellationToken); + + Assert.True(response.Success); + + var content = testHandler.HttpRequestContent; + + Assert.NotNull(content); + Assert.True(content.Length < payload.Length, "The payload was not compressed."); + Assert.Equal(payload, DecompressZstd(content)); + AssertZstdFrameIsInteroperable(content); + + var request = testHandler.HttpRequestMessage; + + Assert.NotNull(request); + Assert.NotNull(request.Content); + Assert.Contains(request.Content.Headers, h => h.Key == "Content-Encoding" && h.Value.Contains("zstd")); + } + + [Fact] + public void SendExportRequest_WithZstdCompression_LargePayloadWindowDoesNotExceed8MiB() + { + // Larger than the 8 MiB window, so the frame cannot be single-segment. + var payload = new byte[(9 * 1024 * 1024) + 1]; + System.Security.Cryptography.RandomNumberGenerator.Fill(payload); + + using var testHandler = new TestHttpMessageHandler(); + using var httpClient = new HttpClient(testHandler, disposeHandler: false); + + var exportClient = CreateZstdExportClient(httpClient); + + exportClient.SendExportRequest( + payload, + payload.Length, + DateTime.UtcNow.AddSeconds(30), + TestContext.Current.CancellationToken); + + var content = testHandler.HttpRequestContent; + + Assert.NotNull(content); + Assert.Equal(payload, DecompressZstd(content)); + + var windowSize = AssertZstdFrameIsInteroperable(content); + Assert.True(windowSize <= 8 * 1024 * 1024, $"The window size of {windowSize} bytes exceeds 8 MiB."); + } + + [Theory] + [InlineData(HttpStatusCode.UnsupportedMediaType)] + [InlineData(HttpStatusCode.BadRequest)] + public void SendExportRequest_WithZstdCompression_RejectedByReceiver_RetriesWithGzipAndThenUsesGzip(HttpStatusCode rejectionStatusCode) + { + const int ZstdCompressionRejectedEventId = 45; + + var payload = "00000000000000000000000000000000"u8.ToArray(); + + using var listener = new TestEventListener(OpenTelemetryProtocolExporterEventSource.Log, EventLevel.Warning); + using var testHandler = new ContentEncodingHttpMessageHandler(encoding => encoding == "zstd" ? rejectionStatusCode : HttpStatusCode.OK); + using var httpClient = new HttpClient(testHandler, disposeHandler: false); + + var exportClient = CreateZstdExportClient(httpClient); + + var response = exportClient.SendExportRequest(payload, payload.Length, DateTime.UtcNow.AddSeconds(10), TestContext.Current.CancellationToken); + + Assert.True(response.Success); + Assert.Equal(["zstd", "gzip"], testHandler.ContentEncodings); + Assert.Contains(listener.Messages, e => e.EventId == ZstdCompressionRejectedEventId); + Assert.Equal(OtlpExportCompression.GZip, exportClient.Compression); + + response = exportClient.SendExportRequest(payload, payload.Length, DateTime.UtcNow.AddSeconds(10), TestContext.Current.CancellationToken); + + Assert.True(response.Success); + Assert.Equal(["zstd", "gzip", "gzip"], testHandler.ContentEncodings); + } + + [Fact] + public void SendExportRequest_WithZstdCompression_RejectedForOtherReason_KeepsUsingZstd() + { + var payload = "00000000000000000000000000000000"u8.ToArray(); + + using var testHandler = new ContentEncodingHttpMessageHandler(_ => HttpStatusCode.BadRequest); + using var httpClient = new HttpClient(testHandler, disposeHandler: false); + + var exportClient = CreateZstdExportClient(httpClient); + + var response = exportClient.SendExportRequest(payload, payload.Length, DateTime.UtcNow.AddSeconds(10), TestContext.Current.CancellationToken); + + var httpResponse = Assert.IsType(response); + Assert.False(httpResponse.Success); + Assert.Equal(HttpStatusCode.BadRequest, httpResponse.StatusCode); + Assert.Equal(["zstd", "gzip"], testHandler.ContentEncodings); + Assert.Equal(OtlpExportCompression.Zstd, exportClient.Compression); + } + + [Theory] + [InlineData(HttpStatusCode.InternalServerError)] + [InlineData(HttpStatusCode.ServiceUnavailable)] + [InlineData(HttpStatusCode.Unauthorized)] + public void SendExportRequest_WithZstdCompression_OtherFailure_DoesNotRetryWithGzip(HttpStatusCode statusCode) + { + var payload = "00000000000000000000000000000000"u8.ToArray(); + + using var testHandler = new ContentEncodingHttpMessageHandler(_ => statusCode); + using var httpClient = new HttpClient(testHandler, disposeHandler: false); + + var exportClient = CreateZstdExportClient(httpClient); + + var response = exportClient.SendExportRequest(payload, payload.Length, DateTime.UtcNow.AddSeconds(10), TestContext.Current.CancellationToken); + + Assert.False(response.Success); + Assert.Equal(["zstd"], testHandler.ContentEncodings); + } +#else + [Fact] + public void SendExportRequest_WithZstdCompression_UsesGzipWhenZstdIsNotSupported() + { + const int ZstdCompressionNotSupportedEventId = 44; + + var payload = "00000000000000000000000000000000"u8.ToArray(); + + using var listener = new TestEventListener(OpenTelemetryProtocolExporterEventSource.Log, EventLevel.Warning); + using var testHandler = new TestHttpMessageHandler(); + using var httpClient = new HttpClient(testHandler, disposeHandler: false); + + var exportClient = new OtlpHttpExportClient( + new OtlpExporterOptions + { + Endpoint = new Uri("http://localhost:4318"), + Protocol = OtlpExportProtocol.HttpProtobuf, + Compression = OtlpExportCompression.Zstd, + }, + httpClient, + string.Empty); + + exportClient.SendExportRequest( + payload, + payload.Length, + DateTime.UtcNow.AddSeconds(10), + TestContext.Current.CancellationToken); + + Assert.Equal(OtlpExportCompression.GZip, exportClient.Compression); + Assert.Contains(listener.Messages, e => e.EventId == ZstdCompressionNotSupportedEventId); + + var request = testHandler.HttpRequestMessage; + + Assert.NotNull(request); + Assert.NotNull(request.Content); + Assert.Contains(request.Content.Headers, h => h.Key == "Content-Encoding" && h.Value.Contains("gzip")); + } +#endif + [Fact] public void SendExportRequest_UnexpectedException_ReturnsRetryableFailureResponse() { @@ -480,6 +647,65 @@ private static ExportClientResponse SendExportRequestThatThrows(Exception except return exportClient.SendExportRequest(payload, payload.Length, DateTime.UtcNow.AddSeconds(10)); } +#if NET11_0_OR_GREATER + private static OtlpHttpExportClient CreateZstdExportClient(HttpClient httpClient) => + new( + new OtlpExporterOptions + { + Endpoint = new Uri("http://localhost:4318"), + Protocol = OtlpExportProtocol.HttpProtobuf, + Compression = OtlpExportCompression.Zstd, + }, + httpClient, + string.Empty); + + private static byte[] DecompressZstd(byte[] compressed) + { + using var input = new MemoryStream(compressed); + using var zstd = new ZstandardStream(input, CompressionMode.Decompress); + using var output = new MemoryStream(); + + zstd.CopyTo(output); + + return output.ToArray(); + } + + /// + /// Asserts the content is a standard Zstandard frame that does not use a dictionary. + /// + /// The Zstandard frame. + /// The window size the frame requires a decoder to use. + private static long AssertZstdFrameIsInteroperable(byte[] frame) + { + // See https://www.rfc-editor.org/rfc/rfc8878#section-3.1.1. + Assert.Equal(0xFD2FB528u, BinaryPrimitives.ReadUInt32LittleEndian(frame)); + + var descriptor = frame[4]; + var dictionaryIdFlag = descriptor & 0b11; + var singleSegment = (descriptor & 0b0010_0000) != 0; + var contentSizeFlag = descriptor >> 6; + + Assert.Equal(0, dictionaryIdFlag); + + if (singleSegment) + { + // The window size is the content size. + return contentSizeFlag switch + { + 0 => frame[5], + 1 => BinaryPrimitives.ReadUInt16LittleEndian(frame.AsSpan(5)) + 256, + 2 => BinaryPrimitives.ReadUInt32LittleEndian(frame.AsSpan(5)), + _ => (long)BinaryPrimitives.ReadUInt64LittleEndian(frame.AsSpan(5)), + }; + } + + var windowDescriptor = frame[5]; + var windowBase = 1L << (10 + (windowDescriptor >> 3)); + + return windowBase + ((windowBase / 8) * (windowDescriptor & 0b111)); + } +#endif + private sealed class ThrowingHttpMessageHandler : HttpMessageHandler { private readonly Exception exception; @@ -498,6 +724,29 @@ protected override HttpResponseMessage Send(HttpRequestMessage request, Cancella #endif } +#if NET11_0_OR_GREATER + private sealed class ContentEncodingHttpMessageHandler(Func getStatusCode) : HttpMessageHandler + { + public List ContentEncodings { get; } = []; + + protected override Task SendAsync(HttpRequestMessage request, CancellationToken cancellationToken) + => Task.FromResult(this.Send(request, cancellationToken)); + + protected override HttpResponseMessage Send(HttpRequestMessage request, CancellationToken cancellationToken) + { + var encoding = request.Content?.Headers.ContentEncoding.SingleOrDefault(); + + this.ContentEncodings.Add(encoding); + + return new(getStatusCode(encoding)) + { + RequestMessage = request, + Content = new ByteArrayContent([]), + }; + } + } +#endif + private sealed class StatusCodeHttpMessageHandler : HttpMessageHandler { private readonly HttpStatusCode statusCode; From 964e2e14c5d36ddfdaae6daf6893bb3a05699469 Mon Sep 17 00:00:00 2001 From: martincostello Date: Wed, 28 Oct 2026 15:36:52 +0000 Subject: [PATCH 2/7] [OTLP] Extend test coverage Add more test coverage for zstd support. --- .../OtlpGrpcExportClientTests.cs | 10 ++ .../OtlpHttpExportClientTests.cs | 132 ++++++++++++++++++ 2 files changed, 142 insertions(+) diff --git a/test/OpenTelemetry.Exporter.OpenTelemetryProtocol.Tests/OtlpGrpcExportClientTests.cs b/test/OpenTelemetry.Exporter.OpenTelemetryProtocol.Tests/OtlpGrpcExportClientTests.cs index 4fc85b7dde3..9ff70142e27 100644 --- a/test/OpenTelemetry.Exporter.OpenTelemetryProtocol.Tests/OtlpGrpcExportClientTests.cs +++ b/test/OpenTelemetry.Exporter.OpenTelemetryProtocol.Tests/OtlpGrpcExportClientTests.cs @@ -232,9 +232,12 @@ public void SendExportRequest_GrpcExport_GzipCompression_PayloadDecompressesToOr [Fact] public void SendExportRequest_GrpcExport_ZstdCompression_PayloadDecompressesToOriginalProtobuf() { + const int CompressedGrpcPayloadEventId = 37; + var protobufPayload = "grpc test payload for compression"u8.ToArray(); var buffer = BuildGrpcFrame(protobufPayload); + using var listener = new TestEventListener(OpenTelemetryProtocolExporterEventSource.Log, EventLevel.Verbose); using var testHandler = new TestGrpcMessageHandler(); using var httpClient = new HttpClient(testHandler, disposeHandler: false); @@ -268,6 +271,13 @@ public void SendExportRequest_GrpcExport_ZstdCompression_PayloadDecompressesToOr Assert.Contains( testHandler.CapturedRequestHeaders, h => h.Key == "grpc-encoding" && h.Value.Contains("zstd")); + + var compressedEvent = Assert.Single(listener.Messages, e => e.EventId == CompressedGrpcPayloadEventId); + + Assert.NotNull(compressedEvent.Payload); + Assert.Equal("zstd", compressedEvent.Payload[0]); + Assert.Equal((long)buffer.Length, compressedEvent.Payload[1]); + Assert.Equal((long)content.Length, compressedEvent.Payload[2]); } [Fact] diff --git a/test/OpenTelemetry.Exporter.OpenTelemetryProtocol.Tests/OtlpHttpExportClientTests.cs b/test/OpenTelemetry.Exporter.OpenTelemetryProtocol.Tests/OtlpHttpExportClientTests.cs index 8922e7be4af..7a0f88e5df2 100644 --- a/test/OpenTelemetry.Exporter.OpenTelemetryProtocol.Tests/OtlpHttpExportClientTests.cs +++ b/test/OpenTelemetry.Exporter.OpenTelemetryProtocol.Tests/OtlpHttpExportClientTests.cs @@ -325,6 +325,104 @@ public void SendExportRequest_WithZstdCompression_OtherFailure_DoesNotRetryWithG Assert.False(response.Success); Assert.Equal(["zstd"], testHandler.ContentEncodings); } + + [Fact] + public void Shutdown_WithZstdCompression_LaterRequestsAreStillCompressed() + { + var payload = "00000000000000000000000000000000"u8.ToArray(); + + using var testHandler = new TestHttpMessageHandler(); + using var httpClient = new HttpClient(testHandler, disposeHandler: false); + + var exportClient = CreateZstdExportClient(httpClient); + + Assert.True(exportClient.SendExportRequest(payload, payload.Length, DateTime.UtcNow.AddSeconds(10), TestContext.Current.CancellationToken).Success); + + // The first shutdown releases the cached encoder, the second finds none to release. + Assert.True(exportClient.Shutdown(0)); + Assert.True(exportClient.Shutdown(0)); + + Assert.True(exportClient.SendExportRequest(payload, payload.Length, DateTime.UtcNow.AddSeconds(10), TestContext.Current.CancellationToken).Success); + + Assert.NotNull(testHandler.HttpRequestContent); + Assert.Equal(payload, DecompressZstd(testHandler.HttpRequestContent)); + } + + [Fact] + public async Task ZstdHttpContent_CopyToAsync_IsCompressed() + { + var payload = "00000000000000000000000000000000"u8.ToArray(); + var compressor = new ZstdCompressor(); + + // The second copy reuses the encoder cached by the first. + for (var i = 0; i < 2; i++) + { + using var content = new ZstdHttpContent(payload, payload.Length, OtlpHttpExportClient.MediaHeaderValue, compressor); + using var destination = new MemoryStream(); + + await content.CopyToAsync(destination, TestContext.Current.CancellationToken); + + Assert.Equal(payload, DecompressZstd(destination.ToArray())); + Assert.Equal(["zstd"], content.Headers.ContentEncoding); + Assert.Equal(OtlpHttpExportClient.MediaHeaderValue, content.Headers.ContentType); + } + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task ZstdCompressor_DestinationThrows_EncoderIsNotReused(bool async) + { + var payload = "00000000000000000000000000000000"u8.ToArray(); + var compressor = new ZstdCompressor(); + + using (var failing = new ThrowingStream()) + { + var exception = async + ? await Assert.ThrowsAsync(() => compressor.CompressAsync(payload, failing, TestContext.Current.CancellationToken)) + : Assert.Throws(() => compressor.Compress(payload, failing)); + + Assert.Equal(ThrowingStream.Message, exception.Message); + } + + // Had the encoder that failed been disposed and then cached, this would throw ObjectDisposedException. + using var destination = new MemoryStream(); + + if (async) + { + await compressor.CompressAsync(payload, destination, TestContext.Current.CancellationToken); + } + else + { + compressor.Compress(payload, destination); + } + + Assert.Equal(payload, DecompressZstd(destination.ToArray())); + } + + [Fact] + public void ZstdCompressor_OverlappingCompressions_EachUseTheirOwnEncoder() + { + var outerPayload = "outer outer outer outer outer outer"u8.ToArray(); + var innerPayload = "inner inner inner inner inner inner"u8.ToArray(); + var compressor = new ZstdCompressor(); + + using var inner = new MemoryStream(); + + // Starts a second compression while the first holds the cached encoder, so the second + // creates its own and caches it, and the first then finds the cache occupied when it finishes. + using var outer = new CallbackOnFirstWriteStream(() => compressor.Compress(innerPayload, inner)); + + compressor.Compress(outerPayload, outer); + + Assert.Equal(outerPayload, DecompressZstd(outer.ToArray())); + Assert.Equal(innerPayload, DecompressZstd(inner.ToArray())); + + using var next = new MemoryStream(); + compressor.Compress(outerPayload, next); + + Assert.Equal(outerPayload, DecompressZstd(next.ToArray())); + } #else [Fact] public void SendExportRequest_WithZstdCompression_UsesGzipWhenZstdIsNotSupported() @@ -745,6 +843,40 @@ protected override HttpResponseMessage Send(HttpRequestMessage request, Cancella }; } } + + private sealed class ThrowingStream : MemoryStream + { + public const string Message = "The destination stream failed."; + + public override void Write(byte[] buffer, int offset, int count) + => throw new IOException(Message); + + public override void Write(ReadOnlySpan buffer) + => throw new IOException(Message); + + public override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) + => throw new IOException(Message); + + public override ValueTask WriteAsync(ReadOnlyMemory buffer, CancellationToken cancellationToken = default) + => throw new IOException(Message); + } + + private sealed class CallbackOnFirstWriteStream(Action onFirstWrite) : MemoryStream + { + private Action? onFirstWrite = onFirstWrite; + + public override void Write(byte[] buffer, int offset, int count) + { + Interlocked.Exchange(ref this.onFirstWrite, null)?.Invoke(); + base.Write(buffer, offset, count); + } + + public override void Write(ReadOnlySpan buffer) + { + Interlocked.Exchange(ref this.onFirstWrite, null)?.Invoke(); + base.Write(buffer); + } + } #endif private sealed class StatusCodeHttpMessageHandler : HttpMessageHandler From 3bcc4f47b83f74056b3e972997928aceedba37b1 Mon Sep 17 00:00:00 2001 From: martincostello Date: Thu, 29 Oct 2026 13:38:58 +0000 Subject: [PATCH 3/7] [OTLP] Address feedback Place a limit on how often a HTTP 400 will cause a change in the compression codec. --- .../ExportClient/OtlpExportClient.cs | 27 ++++++++-- .../OtlpExportCompression.cs | 1 + .../README.md | 3 ++ .../OtlpHttpExportClientTests.cs | 53 +++++++++++++++++++ 4 files changed, 81 insertions(+), 3 deletions(-) diff --git a/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/ExportClient/OtlpExportClient.cs b/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/ExportClient/OtlpExportClient.cs index 83c67d6d479..270bf4b7f4f 100644 --- a/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/ExportClient/OtlpExportClient.cs +++ b/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/ExportClient/OtlpExportClient.cs @@ -17,6 +17,12 @@ internal abstract class OtlpExportClient : IExportClient /// internal const int MaxLoggedResponseSizeBytes = 4 * 1024; + /// + /// 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. + /// + private const int MaxConsecutiveZstdFallbackFailures = 3; + private static readonly Version Http2RequestVersion = new(2, 0); private static readonly TimeSpan MaxDeadlineDelay = TimeSpan.FromMilliseconds(int.MaxValue); @@ -40,6 +46,7 @@ internal abstract class OtlpExportClient : IExportClient #endif private int zstdRejected; + private int zstdFallbackFailures; protected OtlpExportClient(OtlpExporterOptions options, HttpClient httpClient, string signalPath) { @@ -95,7 +102,9 @@ protected OtlpExportClient(OtlpExporterOptions options, HttpClient httpClient, s /// /// /// This is if Zstandard compression was configured, - /// but the receiver rejected a request using it and then accepted it using GZip. + /// but the receiver rejected a request using it and then accepted it using GZip, or if + /// requests in a row were rejected using it + /// and then failed again using GZip. /// internal OtlpExportCompression Compression => this.compression == OtlpExportCompression.Zstd && Volatile.Read(ref this.zstdRejected) != 0 @@ -132,11 +141,23 @@ public ExportClientResponse SendExportRequest(byte[] buffer, int contentLength, if (response.Success) { - // Only stop using Zstandard if GZip succeeded, as otherwise the rejection - // was most likely caused by something other than the compression used. + // 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; } diff --git a/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/OtlpExportCompression.cs b/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/OtlpExportCompression.cs index 2bb2f717b56..876be14db4e 100644 --- a/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/OtlpExportCompression.cs +++ b/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/OtlpExportCompression.cs @@ -32,6 +32,7 @@ public enum OtlpExportCompression /// not available, is used instead. /// If the receiver rejects a Zstandard-compressed request, the /// request is sent again once using . If that succeeds, + /// or if three requests in a row are rejected and then fail again, /// is used for all subsequent requests. /// /// diff --git a/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/README.md b/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/README.md index d78fb66a01f..fe8d83926f6 100644 --- a/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/README.md +++ b/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/README.md @@ -353,6 +353,9 @@ appBuilder.Services.AddOptions() > rejects a Zstandard-compressed request (with `415`/`400` for > `http/protobuf`, or `UNIMPLEMENTED` for `grpc`), the request is sent again > using GZip, and GZip is then used for all further requests if that succeeds. + > If three requests in a row are rejected and then fail again using GZip, + > GZip is also used for all further requests, so that each request is not + > sent twice indefinitely. * `Headers`: Optional headers for the connection. diff --git a/test/OpenTelemetry.Exporter.OpenTelemetryProtocol.Tests/OtlpHttpExportClientTests.cs b/test/OpenTelemetry.Exporter.OpenTelemetryProtocol.Tests/OtlpHttpExportClientTests.cs index 7a0f88e5df2..c9afe272072 100644 --- a/test/OpenTelemetry.Exporter.OpenTelemetryProtocol.Tests/OtlpHttpExportClientTests.cs +++ b/test/OpenTelemetry.Exporter.OpenTelemetryProtocol.Tests/OtlpHttpExportClientTests.cs @@ -307,6 +307,59 @@ public void SendExportRequest_WithZstdCompression_RejectedForOtherReason_KeepsUs Assert.Equal(OtlpExportCompression.Zstd, exportClient.Compression); } + [Fact] + public void SendExportRequest_WithZstdCompression_RepeatedlyRejectedForOtherReason_StopsSendingRequestsTwice() + { + var payload = "00000000000000000000000000000000"u8.ToArray(); + + using var testHandler = new ContentEncodingHttpMessageHandler(_ => HttpStatusCode.BadRequest); + using var httpClient = new HttpClient(testHandler, disposeHandler: false); + + var exportClient = CreateZstdExportClient(httpClient); + + for (var i = 0; i < 3; i++) + { + var response = exportClient.SendExportRequest(payload, payload.Length, DateTime.UtcNow.AddSeconds(10), TestContext.Current.CancellationToken); + + Assert.False(response.Success); + } + + Assert.Equal(["zstd", "gzip", "zstd", "gzip", "zstd", "gzip"], testHandler.ContentEncodings); + Assert.Equal(OtlpExportCompression.GZip, exportClient.Compression); + + var finalResponse = exportClient.SendExportRequest(payload, payload.Length, DateTime.UtcNow.AddSeconds(10), TestContext.Current.CancellationToken); + + Assert.False(finalResponse.Success); + Assert.Equal(["zstd", "gzip", "zstd", "gzip", "zstd", "gzip", "gzip"], testHandler.ContentEncodings); + } + + [Fact] + public void SendExportRequest_WithZstdCompression_RejectedForOtherReasonNonConsecutively_KeepsUsingZstd() + { + var payload = "00000000000000000000000000000000"u8.ToArray(); + var zstdStatusCode = HttpStatusCode.BadRequest; + + using var testHandler = new ContentEncodingHttpMessageHandler( + encoding => encoding == "zstd" ? zstdStatusCode : HttpStatusCode.BadRequest); + using var httpClient = new HttpClient(testHandler, disposeHandler: false); + + var exportClient = CreateZstdExportClient(httpClient); + + for (var round = 0; round < 3; round++) + { + exportClient.SendExportRequest(payload, payload.Length, DateTime.UtcNow.AddSeconds(10), TestContext.Current.CancellationToken); + exportClient.SendExportRequest(payload, payload.Length, DateTime.UtcNow.AddSeconds(10), TestContext.Current.CancellationToken); + + // A request that is not rejected because of its compression resets the count + zstdStatusCode = HttpStatusCode.InternalServerError; + exportClient.SendExportRequest(payload, payload.Length, DateTime.UtcNow.AddSeconds(10), TestContext.Current.CancellationToken); + zstdStatusCode = HttpStatusCode.BadRequest; + } + + Assert.Equal(OtlpExportCompression.Zstd, exportClient.Compression); + Assert.Equal(15, testHandler.ContentEncodings.Count); + } + [Theory] [InlineData(HttpStatusCode.InternalServerError)] [InlineData(HttpStatusCode.ServiceUnavailable)] From de8e84c4f7045223b8172b9317295b9f0800229d Mon Sep 17 00:00:00 2001 From: martincostello Date: Sun, 1 Nov 2026 19:26:49 +0000 Subject: [PATCH 4/7] [OTLP] Address zstd review feedback - Only fall back to gzip for gRPC UNIMPLEMENTED when grpc-accept-encoding does not list zstd. - Only fall back to gzip for HTTP 415; 400 is no longer treated as a compression rejection. - Dispose of a Zstandard encoder returned after shutdown instead of caching it. Assisted-by: Claude Code --- .../ExportClient/ExportClientGrpcResponse.cs | 9 +++- .../ExportClient/OtlpGrpcExportClient.cs | 42 ++++++++++++++-- .../ExportClient/OtlpHttpExportClient.cs | 6 ++- .../ExportClient/ZstdCompressor.cs | 18 +++++++ .../README.md | 7 +-- .../OtlpGrpcExportClientTests.cs | 38 ++++++++++++-- .../OtlpHttpExportClientTests.cs | 49 ++++++++++++++----- 7 files changed, 145 insertions(+), 24 deletions(-) diff --git a/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/ExportClient/ExportClientGrpcResponse.cs b/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/ExportClient/ExportClientGrpcResponse.cs index 339e0ab78ad..4766cf7ce67 100644 --- a/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/ExportClient/ExportClientGrpcResponse.cs +++ b/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; } + + /// + /// Gets the value of the grpc-accept-encoding header of the response, if it was captured. + /// + public string? GrpcAcceptEncodingHeader { get; } } diff --git a/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/ExportClient/OtlpGrpcExportClient.cs b/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/ExportClient/OtlpGrpcExportClient.cs index 0b431db3c43..1ad621418dd 100644 --- a/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/ExportClient/OtlpGrpcExportClient.cs +++ b/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). @@ -152,6 +153,17 @@ protected override ExportClientResponse SendExportRequest(byte[] buffer, int con 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( @@ -159,7 +171,8 @@ protected override ExportClientResponse SendExportRequest(byte[] buffer, int con deadlineUtc: deadlineUtc, exception: null, status: status, - grpcStatusDetailsHeader: grpcStatusDetailsHeader); + grpcStatusDetailsHeader: grpcStatusDetailsHeader, + grpcAcceptEncodingHeader: grpcAcceptEncodingHeader); } catch (HttpRequestException ex) when (ex.InnerException is TimeoutException || IsTransientNetworkError(ex)) { @@ -234,11 +247,15 @@ protected override ExportClientResponse SendExportRequest(byte[] buffer, int con /// /// - /// Servers that do not support a grpc-encoding return UNIMPLEMENTED. + /// 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. /// protected override bool IsCompressionRejected(ExportClientResponse response) => - response is ExportClientGrpcResponse { Status.StatusCode: StatusCode.Unimplemented }; + 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 @@ -248,6 +265,25 @@ protected override HttpContent CreateHttpContent(byte[] buffer, int contentLengt 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 }; diff --git a/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/ExportClient/OtlpHttpExportClient.cs b/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/ExportClient/OtlpHttpExportClient.cs index 540333f7b24..2879e4e8f51 100644 --- a/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/ExportClient/OtlpHttpExportClient.cs +++ b/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/ExportClient/OtlpHttpExportClient.cs @@ -99,10 +99,12 @@ protected override ExportClientResponse SendExportRequest(byte[] buffer, int con /// /// - /// Receivers that do not support a Content-Encoding should return 415, but may return 400. + /// Receivers that do not support a Content-Encoding must return 415. A 400 is not treated as a + /// rejection of the compression as OTLP/HTTP uses it for payloads that are permanently invalid, + /// which must not be sent again. /// protected override bool IsCompressionRejected(ExportClientResponse response) => - response is ExportClientHttpResponse { StatusCode: HttpStatusCode.UnsupportedMediaType or HttpStatusCode.BadRequest }; + response is ExportClientHttpResponse { StatusCode: HttpStatusCode.UnsupportedMediaType }; protected override HttpContent CreateHttpContent(byte[] buffer, int contentLength, OtlpExportCompression compression) => compression switch diff --git a/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/ExportClient/ZstdCompressor.cs b/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/ExportClient/ZstdCompressor.cs index af47c0ef774..b981760e565 100644 --- a/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/ExportClient/ZstdCompressor.cs +++ b/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/ExportClient/ZstdCompressor.cs @@ -24,6 +24,7 @@ internal sealed class ZstdCompressor internal const int MaxWindowLog2 = 23; private ZstandardEncoder? cachedEncoder; + private int isShutdown; [UnsupportedOSPlatformGuard("browser")] [UnsupportedOSPlatformGuard("wasi")] @@ -77,6 +78,9 @@ public void Shutdown() { Debug.Assert(IsSupported, "Zstandard compression is not supported on this platform."); + // Set before releasing the cached encoder, so that an encoder returned by an export that is + // still compressing either sees the flag and is disposed, or is cached in time to be released here. + Interlocked.Exchange(ref this.isShutdown, 1); Interlocked.Exchange(ref this.cachedEncoder, null)?.Dispose(); } @@ -102,6 +106,14 @@ private ZstandardEncoder RentEncoder() [UnsupportedOSPlatform("wasi")] private void ReturnEncoder(ZstandardEncoder encoder) { + if (Volatile.Read(ref this.isShutdown) != 0) + { + // Shutdown ran while the encoder was in use, so do not retain its native memory. + // Later exports can still compress, but each uses (and then disposes of) its own encoder. + encoder.Dispose(); + return; + } + encoder.Reset(); if (Interlocked.CompareExchange(ref this.cachedEncoder, encoder, null) != null) @@ -109,6 +121,12 @@ private void ReturnEncoder(ZstandardEncoder encoder) // Another export returned an encoder first. encoder.Dispose(); } + else if (Volatile.Read(ref this.isShutdown) != 0) + { + // Shutdown ran after the check above but before the encoder was cached, and so may have + // missed it. Whichever of this and Shutdown removes it from the cache is the one that disposes of it. + Interlocked.Exchange(ref this.cachedEncoder, null)?.Dispose(); + } } } diff --git a/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/README.md b/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/README.md index fe8d83926f6..f57e1bc4dff 100644 --- a/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/README.md +++ b/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/README.md @@ -350,9 +350,10 @@ appBuilder.Services.AddOptions() > `OtlpExportCompression.Zstd` requires .NET 11 or later and is not available > in when using Blazor. Where Zstandard is not available, GZip is used > instead. Receivers are not required to support Zstandard, so if a receiver - > rejects a Zstandard-compressed request (with `415`/`400` for - > `http/protobuf`, or `UNIMPLEMENTED` for `grpc`), the request is sent again - > using GZip, and GZip is then used for all further requests if that succeeds. + > rejects a Zstandard-compressed request (with `415` for `http/protobuf`, or + > `UNIMPLEMENTED` for `grpc` unless the response's `grpc-accept-encoding` + > header lists `zstd`), the request is sent again using GZip, and GZip is then + > used for all further requests if that succeeds. > If three requests in a row are rejected and then fail again using GZip, > GZip is also used for all further requests, so that each request is not > sent twice indefinitely. diff --git a/test/OpenTelemetry.Exporter.OpenTelemetryProtocol.Tests/OtlpGrpcExportClientTests.cs b/test/OpenTelemetry.Exporter.OpenTelemetryProtocol.Tests/OtlpGrpcExportClientTests.cs index 9ff70142e27..7c246b30a7c 100644 --- a/test/OpenTelemetry.Exporter.OpenTelemetryProtocol.Tests/OtlpGrpcExportClientTests.cs +++ b/test/OpenTelemetry.Exporter.OpenTelemetryProtocol.Tests/OtlpGrpcExportClientTests.cs @@ -280,15 +280,19 @@ public void SendExportRequest_GrpcExport_ZstdCompression_PayloadDecompressesToOr Assert.Equal((long)content.Length, compressedEvent.Payload[2]); } - [Fact] - public void SendExportRequest_GrpcExport_ZstdCompression_Unimplemented_RetriesWithGzipAndThenUsesGzip() + [Theory] + [InlineData(null)] + [InlineData("")] + [InlineData("identity,gzip")] + [InlineData("identity, gzip, zstdx")] + public void SendExportRequest_GrpcExport_ZstdCompression_Unimplemented_RetriesWithGzipAndThenUsesGzip(string? acceptEncoding) { const int ZstdCompressionRejectedEventId = 45; var buffer = BuildGrpcFrame("grpc test payload for compression"u8.ToArray()); using var listener = new TestEventListener(OpenTelemetryProtocolExporterEventSource.Log, EventLevel.Warning); - using var testHandler = new GrpcEncodingMessageHandler(encoding => encoding == "zstd" ? StatusCode.Unimplemented : StatusCode.OK); + using var testHandler = new GrpcEncodingMessageHandler(encoding => encoding == "zstd" ? StatusCode.Unimplemented : StatusCode.OK, acceptEncoding); using var httpClient = new HttpClient(testHandler, disposeHandler: false); var exportClient = CreateZstdExportClient(httpClient); @@ -306,6 +310,27 @@ public void SendExportRequest_GrpcExport_ZstdCompression_Unimplemented_RetriesWi Assert.Equal(["zstd", "gzip", "gzip"], testHandler.GrpcEncodings); } + [Theory] + [InlineData("zstd")] + [InlineData("identity,gzip,zstd")] + [InlineData("identity, ZSTD ,gzip")] + public void SendExportRequest_GrpcExport_ZstdCompression_UnimplementedWhileZstdIsAdvertised_DoesNotRetryWithGzip(string acceptEncoding) + { + var buffer = BuildGrpcFrame("grpc test payload for compression"u8.ToArray()); + + // The receiver supports zstd, so the error is unrelated to the compression used. + using var testHandler = new GrpcEncodingMessageHandler(_ => StatusCode.Unimplemented, acceptEncoding); + using var httpClient = new HttpClient(testHandler, disposeHandler: false); + + var exportClient = CreateZstdExportClient(httpClient); + + var response = exportClient.SendExportRequest(buffer, buffer.Length, DateTime.UtcNow.AddSeconds(10), TestContext.Current.CancellationToken); + + Assert.False(response.Success); + Assert.Equal(["zstd"], testHandler.GrpcEncodings); + Assert.Equal(OtlpExportCompression.Zstd, exportClient.Compression); + } + [Theory] [InlineData((int)StatusCode.InvalidArgument)] [InlineData((int)StatusCode.Unavailable)] @@ -641,7 +666,7 @@ public override void Flush() } #if NET11_0_OR_GREATER - private sealed class GrpcEncodingMessageHandler(Func getStatusCode) : HttpMessageHandler + private sealed class GrpcEncodingMessageHandler(Func getStatusCode, string? acceptEncoding = null) : HttpMessageHandler { public List GrpcEncodings { get; } = []; @@ -659,6 +684,11 @@ protected override Task SendAsync(HttpRequestMessage reques response.Headers.Add("grpc-status", ((int)getStatusCode(encoding)).ToString(System.Globalization.CultureInfo.InvariantCulture)); + if (acceptEncoding != null) + { + response.Headers.Add("grpc-accept-encoding", acceptEncoding); + } + return Task.FromResult(response); } } diff --git a/test/OpenTelemetry.Exporter.OpenTelemetryProtocol.Tests/OtlpHttpExportClientTests.cs b/test/OpenTelemetry.Exporter.OpenTelemetryProtocol.Tests/OtlpHttpExportClientTests.cs index c9afe272072..8d898a1c0ca 100644 --- a/test/OpenTelemetry.Exporter.OpenTelemetryProtocol.Tests/OtlpHttpExportClientTests.cs +++ b/test/OpenTelemetry.Exporter.OpenTelemetryProtocol.Tests/OtlpHttpExportClientTests.cs @@ -260,17 +260,15 @@ public void SendExportRequest_WithZstdCompression_LargePayloadWindowDoesNotExcee Assert.True(windowSize <= 8 * 1024 * 1024, $"The window size of {windowSize} bytes exceeds 8 MiB."); } - [Theory] - [InlineData(HttpStatusCode.UnsupportedMediaType)] - [InlineData(HttpStatusCode.BadRequest)] - public void SendExportRequest_WithZstdCompression_RejectedByReceiver_RetriesWithGzipAndThenUsesGzip(HttpStatusCode rejectionStatusCode) + [Fact] + public void SendExportRequest_WithZstdCompression_RejectedByReceiver_RetriesWithGzipAndThenUsesGzip() { const int ZstdCompressionRejectedEventId = 45; var payload = "00000000000000000000000000000000"u8.ToArray(); using var listener = new TestEventListener(OpenTelemetryProtocolExporterEventSource.Log, EventLevel.Warning); - using var testHandler = new ContentEncodingHttpMessageHandler(encoding => encoding == "zstd" ? rejectionStatusCode : HttpStatusCode.OK); + using var testHandler = new ContentEncodingHttpMessageHandler(encoding => encoding == "zstd" ? HttpStatusCode.UnsupportedMediaType : HttpStatusCode.OK); using var httpClient = new HttpClient(testHandler, disposeHandler: false); var exportClient = CreateZstdExportClient(httpClient); @@ -293,7 +291,9 @@ public void SendExportRequest_WithZstdCompression_RejectedForOtherReason_KeepsUs { var payload = "00000000000000000000000000000000"u8.ToArray(); - using var testHandler = new ContentEncodingHttpMessageHandler(_ => HttpStatusCode.BadRequest); + // A 415 for a reason other than the Content-Encoding, such as an unsupported Content-Type, + // also fails when the request is sent again with GZip. + using var testHandler = new ContentEncodingHttpMessageHandler(_ => HttpStatusCode.UnsupportedMediaType); using var httpClient = new HttpClient(testHandler, disposeHandler: false); var exportClient = CreateZstdExportClient(httpClient); @@ -302,7 +302,7 @@ public void SendExportRequest_WithZstdCompression_RejectedForOtherReason_KeepsUs var httpResponse = Assert.IsType(response); Assert.False(httpResponse.Success); - Assert.Equal(HttpStatusCode.BadRequest, httpResponse.StatusCode); + Assert.Equal(HttpStatusCode.UnsupportedMediaType, httpResponse.StatusCode); Assert.Equal(["zstd", "gzip"], testHandler.ContentEncodings); Assert.Equal(OtlpExportCompression.Zstd, exportClient.Compression); } @@ -312,7 +312,7 @@ public void SendExportRequest_WithZstdCompression_RepeatedlyRejectedForOtherReas { var payload = "00000000000000000000000000000000"u8.ToArray(); - using var testHandler = new ContentEncodingHttpMessageHandler(_ => HttpStatusCode.BadRequest); + using var testHandler = new ContentEncodingHttpMessageHandler(_ => HttpStatusCode.UnsupportedMediaType); using var httpClient = new HttpClient(testHandler, disposeHandler: false); var exportClient = CreateZstdExportClient(httpClient); @@ -337,10 +337,10 @@ public void SendExportRequest_WithZstdCompression_RepeatedlyRejectedForOtherReas public void SendExportRequest_WithZstdCompression_RejectedForOtherReasonNonConsecutively_KeepsUsingZstd() { var payload = "00000000000000000000000000000000"u8.ToArray(); - var zstdStatusCode = HttpStatusCode.BadRequest; + var zstdStatusCode = HttpStatusCode.UnsupportedMediaType; using var testHandler = new ContentEncodingHttpMessageHandler( - encoding => encoding == "zstd" ? zstdStatusCode : HttpStatusCode.BadRequest); + encoding => encoding == "zstd" ? zstdStatusCode : HttpStatusCode.UnsupportedMediaType); using var httpClient = new HttpClient(testHandler, disposeHandler: false); var exportClient = CreateZstdExportClient(httpClient); @@ -353,7 +353,7 @@ public void SendExportRequest_WithZstdCompression_RejectedForOtherReasonNonConse // A request that is not rejected because of its compression resets the count zstdStatusCode = HttpStatusCode.InternalServerError; exportClient.SendExportRequest(payload, payload.Length, DateTime.UtcNow.AddSeconds(10), TestContext.Current.CancellationToken); - zstdStatusCode = HttpStatusCode.BadRequest; + zstdStatusCode = HttpStatusCode.UnsupportedMediaType; } Assert.Equal(OtlpExportCompression.Zstd, exportClient.Compression); @@ -361,6 +361,7 @@ public void SendExportRequest_WithZstdCompression_RejectedForOtherReasonNonConse } [Theory] + [InlineData(HttpStatusCode.BadRequest)] [InlineData(HttpStatusCode.InternalServerError)] [InlineData(HttpStatusCode.ServiceUnavailable)] [InlineData(HttpStatusCode.Unauthorized)] @@ -377,6 +378,7 @@ public void SendExportRequest_WithZstdCompression_OtherFailure_DoesNotRetryWithG Assert.False(response.Success); Assert.Equal(["zstd"], testHandler.ContentEncodings); + Assert.Equal(OtlpExportCompression.Zstd, exportClient.Compression); } [Fact] @@ -476,6 +478,31 @@ public void ZstdCompressor_OverlappingCompressions_EachUseTheirOwnEncoder() Assert.Equal(outerPayload, DecompressZstd(next.ToArray())); } + + [Fact] + public void ZstdCompressor_ShutdownDuringCompression_CompressionCompletesAndLaterCompressionsStillWork() + { + var payload = "shutdown shutdown shutdown shutdown shutdown"u8.ToArray(); + var compressor = new ZstdCompressor(); + + // Shuts down while the encoder is in use, so it is not cached when it is returned + // as nothing would then release it, which would retain its native memory. + using var destination = new CallbackOnFirstWriteStream(compressor.Shutdown); + + compressor.Compress(payload, destination); + + Assert.Equal(payload, DecompressZstd(destination.ToArray())); + + for (var i = 0; i < 2; i++) + { + using var next = new MemoryStream(); + compressor.Compress(payload, next); + + Assert.Equal(payload, DecompressZstd(next.ToArray())); + } + + compressor.Shutdown(); + } #else [Fact] public void SendExportRequest_WithZstdCompression_UsesGzipWhenZstdIsNotSupported() From 1d4c4cd5df2c16a12bd92e9252387ace19f03038 Mon Sep 17 00:00:00 2001 From: martincostello Date: Sun, 1 Nov 2026 20:12:50 +0000 Subject: [PATCH 5/7] [OTLP] Join repeated grpc-accept-encoding fields grpc-accept-encoding may be sent as more than one header field, which GetHeaderValue rejects by throwing. That turned an UNIMPLEMENTED status into a retryable Unavailable one and skipped the gzip fallback. Join the values of the header instead. Assisted-by: Claude Code --- .../ExportClient/Grpc/GrpcProtocolHelpers.cs | 32 +++++++++++ .../ExportClient/OtlpGrpcExportClient.cs | 7 ++- .../OtlpGrpcExportClientTests.cs | 53 +++++++++++++++++-- 3 files changed, 85 insertions(+), 7 deletions(-) diff --git a/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/ExportClient/Grpc/GrpcProtocolHelpers.cs b/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/ExportClient/Grpc/GrpcProtocolHelpers.cs index d0201e64ad8..f42e8392946 100644 --- a/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/ExportClient/Grpc/GrpcProtocolHelpers.cs +++ b/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); } + /// + /// 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). + /// + /// The headers to get the value from. + /// The name of the header. + /// The value of the header, or if it was not sent. + 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) diff --git a/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/ExportClient/OtlpGrpcExportClient.cs b/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/ExportClient/OtlpGrpcExportClient.cs index 1ad621418dd..5bb09c08c3a 100644 --- a/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/ExportClient/OtlpGrpcExportClient.cs +++ b/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/ExportClient/OtlpGrpcExportClient.cs @@ -159,9 +159,12 @@ protected override ExportClientResponse SendExportRequest(byte[] buffer, int con string? grpcAcceptEncodingHeader = null; if (status.StatusCode == StatusCode.Unimplemented) { + // The header may be sent as more than one field, so the values are joined rather + // than using GetHeaderValue(), which throws for repeated fields and would cause + // the UNIMPLEMENTED status to be replaced by a retryable Unavailable one. grpcAcceptEncodingHeader = - GrpcProtocolHelpers.GetHeaderValue(httpResponse.Headers, GrpcAcceptEncodingHeader) ?? - GrpcProtocolHelpers.GetHeaderValue(trailingHeaders, GrpcAcceptEncodingHeader); + GrpcProtocolHelpers.GetListHeaderValue(httpResponse.Headers, GrpcAcceptEncodingHeader) ?? + GrpcProtocolHelpers.GetListHeaderValue(trailingHeaders, GrpcAcceptEncodingHeader); } OpenTelemetryProtocolExporterEventSource.Log.ExportFailure(this.Endpoint, "Export failed due to unexpected status code.", status); diff --git a/test/OpenTelemetry.Exporter.OpenTelemetryProtocol.Tests/OtlpGrpcExportClientTests.cs b/test/OpenTelemetry.Exporter.OpenTelemetryProtocol.Tests/OtlpGrpcExportClientTests.cs index 7c246b30a7c..0fea7308f0d 100644 --- a/test/OpenTelemetry.Exporter.OpenTelemetryProtocol.Tests/OtlpGrpcExportClientTests.cs +++ b/test/OpenTelemetry.Exporter.OpenTelemetryProtocol.Tests/OtlpGrpcExportClientTests.cs @@ -292,7 +292,7 @@ public void SendExportRequest_GrpcExport_ZstdCompression_Unimplemented_RetriesWi var buffer = BuildGrpcFrame("grpc test payload for compression"u8.ToArray()); using var listener = new TestEventListener(OpenTelemetryProtocolExporterEventSource.Log, EventLevel.Warning); - using var testHandler = new GrpcEncodingMessageHandler(encoding => encoding == "zstd" ? StatusCode.Unimplemented : StatusCode.OK, acceptEncoding); + using var testHandler = new GrpcEncodingMessageHandler(encoding => encoding == "zstd" ? StatusCode.Unimplemented : StatusCode.OK, acceptEncoding is null ? null : [acceptEncoding]); using var httpClient = new HttpClient(testHandler, disposeHandler: false); var exportClient = CreateZstdExportClient(httpClient); @@ -319,7 +319,7 @@ public void SendExportRequest_GrpcExport_ZstdCompression_UnimplementedWhileZstdI var buffer = BuildGrpcFrame("grpc test payload for compression"u8.ToArray()); // The receiver supports zstd, so the error is unrelated to the compression used. - using var testHandler = new GrpcEncodingMessageHandler(_ => StatusCode.Unimplemented, acceptEncoding); + using var testHandler = new GrpcEncodingMessageHandler(_ => StatusCode.Unimplemented, [acceptEncoding]); using var httpClient = new HttpClient(testHandler, disposeHandler: false); var exportClient = CreateZstdExportClient(httpClient); @@ -331,6 +331,45 @@ public void SendExportRequest_GrpcExport_ZstdCompression_UnimplementedWhileZstdI Assert.Equal(OtlpExportCompression.Zstd, exportClient.Compression); } + [Theory] + [InlineData("identity", "gzip", true)] + [InlineData("identity", "zstd", false)] + [InlineData("zstd", "gzip", false)] + public void SendExportRequest_GrpcExport_ZstdCompression_UnimplementedWithRepeatedAcceptEncodingFields_RetriesWithGzipOnlyIfZstdIsNotAdvertised( + string firstEncoding, + string secondEncoding, + bool retriesWithGzip) + { + var buffer = BuildGrpcFrame("grpc test payload for compression"u8.ToArray()); + + // The header is sent as two fields, which must be treated as one list rather than as an error. + using var testHandler = new GrpcEncodingMessageHandler( + encoding => encoding == "zstd" ? StatusCode.Unimplemented : StatusCode.OK, + [firstEncoding, secondEncoding]); + using var httpClient = new HttpClient(testHandler, disposeHandler: false); + + var exportClient = CreateZstdExportClient(httpClient); + + var response = exportClient.SendExportRequest(buffer, buffer.Length, DateTime.UtcNow.AddSeconds(10), TestContext.Current.CancellationToken); + + if (retriesWithGzip) + { + Assert.True(response.Success); + Assert.Equal(["zstd", "gzip"], testHandler.GrpcEncodings); + Assert.Equal(OtlpExportCompression.GZip, exportClient.Compression); + } + else + { + var grpcResponse = Assert.IsType(response); + Assert.False(grpcResponse.Success); + + // Not the retryable Unavailable status that an exception reading the header would produce + Assert.Equal(StatusCode.Unimplemented, grpcResponse.Status?.StatusCode); + Assert.Equal(["zstd"], testHandler.GrpcEncodings); + Assert.Equal(OtlpExportCompression.Zstd, exportClient.Compression); + } + } + [Theory] [InlineData((int)StatusCode.InvalidArgument)] [InlineData((int)StatusCode.Unavailable)] @@ -666,7 +705,7 @@ public override void Flush() } #if NET11_0_OR_GREATER - private sealed class GrpcEncodingMessageHandler(Func getStatusCode, string? acceptEncoding = null) : HttpMessageHandler + private sealed class GrpcEncodingMessageHandler(Func getStatusCode, string[]? acceptEncodings = null) : HttpMessageHandler { public List GrpcEncodings { get; } = []; @@ -684,9 +723,13 @@ protected override Task SendAsync(HttpRequestMessage reques response.Headers.Add("grpc-status", ((int)getStatusCode(encoding)).ToString(System.Globalization.CultureInfo.InvariantCulture)); - if (acceptEncoding != null) + if (acceptEncodings != null) { - response.Headers.Add("grpc-accept-encoding", acceptEncoding); + // Each value is sent as a separate field + foreach (var acceptEncoding in acceptEncodings) + { + response.Headers.Add("grpc-accept-encoding", acceptEncoding); + } } return Task.FromResult(response); From ce5dd7c0397177f9fffdbbf277b3f26266e5d1f8 Mon Sep 17 00:00:00 2001 From: martincostello Date: Sun, 1 Nov 2026 21:10:17 +0000 Subject: [PATCH 6/7] [OTLP] Release the zstd encoder on permanent downgrade Once the exporter permanently switches to GZip, nothing can reuse the cached Zstandard encoder, so shut the compressor down instead of retaining its native memory until the client shuts down. Assisted-by: Claude Code --- .../ExportClient/OtlpExportClient.cs | 19 +++++++++++++++++-- 1 file changed, 17 insertions(+), 2 deletions(-) diff --git a/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/ExportClient/OtlpExportClient.cs b/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/ExportClient/OtlpExportClient.cs index 270bf4b7f4f..333b019f38c 100644 --- a/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/ExportClient/OtlpExportClient.cs +++ b/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/ExportClient/OtlpExportClient.cs @@ -143,14 +143,14 @@ public ExportClientResponse SendExportRequest(byte[] buffer, int contentLength, { // 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); + this.StopUsingZstd(); } 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); + this.StopUsingZstd(); } } else if (compression == OtlpExportCompression.Zstd) @@ -415,4 +415,19 @@ private HttpResponseMessage SendHttpRequest(HttpRequestMessage request, Cancella #else this.HttpClient.SendAsync(request, this.CompletionOption, cancellationToken).GetAwaiter().GetResult(); #endif + + /// + /// Permanently switches to using GZip, and releases the Zstandard encoder as nothing can use it again. + /// + /// + /// Any export that is still using the encoder disposes of it once it has finished with it. + /// + private void StopUsingZstd() + { + Volatile.Write(ref this.zstdRejected, 1); + +#if NET11_0_OR_GREATER + this.zstdCompressor?.Shutdown(); +#endif + } } From 589b4b31b1a243b8489e22f0872c6df9860e7857 Mon Sep 17 00:00:00 2001 From: martincostello Date: Mon, 2 Nov 2026 12:47:17 +0000 Subject: [PATCH 7/7] [OTLP] Update tests Update expected event IDs. --- .../OtlpGrpcExportClientTests.cs | 2 +- .../OtlpHttpExportClientTests.cs | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/test/OpenTelemetry.Exporter.OpenTelemetryProtocol.Tests/OtlpGrpcExportClientTests.cs b/test/OpenTelemetry.Exporter.OpenTelemetryProtocol.Tests/OtlpGrpcExportClientTests.cs index 0fea7308f0d..3dc2b7b88b6 100644 --- a/test/OpenTelemetry.Exporter.OpenTelemetryProtocol.Tests/OtlpGrpcExportClientTests.cs +++ b/test/OpenTelemetry.Exporter.OpenTelemetryProtocol.Tests/OtlpGrpcExportClientTests.cs @@ -287,7 +287,7 @@ public void SendExportRequest_GrpcExport_ZstdCompression_PayloadDecompressesToOr [InlineData("identity, gzip, zstdx")] public void SendExportRequest_GrpcExport_ZstdCompression_Unimplemented_RetriesWithGzipAndThenUsesGzip(string? acceptEncoding) { - const int ZstdCompressionRejectedEventId = 45; + const int ZstdCompressionRejectedEventId = 46; var buffer = BuildGrpcFrame("grpc test payload for compression"u8.ToArray()); diff --git a/test/OpenTelemetry.Exporter.OpenTelemetryProtocol.Tests/OtlpHttpExportClientTests.cs b/test/OpenTelemetry.Exporter.OpenTelemetryProtocol.Tests/OtlpHttpExportClientTests.cs index 8d898a1c0ca..bc627069267 100644 --- a/test/OpenTelemetry.Exporter.OpenTelemetryProtocol.Tests/OtlpHttpExportClientTests.cs +++ b/test/OpenTelemetry.Exporter.OpenTelemetryProtocol.Tests/OtlpHttpExportClientTests.cs @@ -263,7 +263,7 @@ public void SendExportRequest_WithZstdCompression_LargePayloadWindowDoesNotExcee [Fact] public void SendExportRequest_WithZstdCompression_RejectedByReceiver_RetriesWithGzipAndThenUsesGzip() { - const int ZstdCompressionRejectedEventId = 45; + const int ZstdCompressionRejectedEventId = 46; var payload = "00000000000000000000000000000000"u8.ToArray(); @@ -507,7 +507,7 @@ public void ZstdCompressor_ShutdownDuringCompression_CompressionCompletesAndLate [Fact] public void SendExportRequest_WithZstdCompression_UsesGzipWhenZstdIsNotSupported() { - const int ZstdCompressionNotSupportedEventId = 44; + const int ZstdCompressionNotSupportedEventId = 45; var payload = "00000000000000000000000000000000"u8.ToArray();