Support for gzip compression in OTLP exporter

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

@@ -0,0 +1,5 @@
+OpenTelemetry.Exporter.OtlpExportCompression
+OpenTelemetry.Exporter.OtlpExportCompression.Gzip = 1 -> OpenTelemetry.Exporter.OtlpExportCompression
+OpenTelemetry.Exporter.OtlpExportCompression.None = 0 -> OpenTelemetry.Exporter.OtlpExportCompression
+OpenTelemetry.Exporter.OtlpExporterOptions.Compression.get -> OpenTelemetry.Exporter.OtlpExportCompression
+OpenTelemetry.Exporter.OtlpExporterOptions.Compression.set -> void

src/OpenTelemetry.Exporter.OpenTelemetryProtocol/CHANGELOG.md

@@ -7,6 +7,13 @@ Notes](../../RELEASENOTES.md).
 
 ## Unreleased
 
+* Added support for gzip compression in the OTLP exporter for HTTP and gRPC.
+  Compression can be configured using the environment variables `OTEL_EXPORTER_OTLP_COMPRESSION`,
+  `OTEL_EXPORTER_OTLP_TRACES_COMPRESSION`, `OTEL_EXPORTER_OTLP_METRICS_COMPRESSION`,
+  `OTEL_EXPORTER_OTLP_LOGS_COMPRESSION`. Setting the respective environment variable
+  to `gzip` activates compression.
+  ([#6494](https://github.com/open-telemetry/opentelemetry-dotnet/pull/6494))
+
 * Fixed an issue in .NET Framework where OTLP export of traces, logs, and
   metrics using `OtlpExportProtocol.Grpc` did not correctly set the initial
   write position, resulting in gRPC protocol errors.

src/OpenTelemetry.Exporter.OpenTelemetryProtocol/IOtlpExporterOptions.cs

@@ -69,6 +69,15 @@ internal interface IOtlpExporterOptions
     /// </summary>
     int TimeoutMilliseconds { get; set; }
 
+    /// <summary>
+    /// Gets or sets a value indicating how to compress the payload.
+    /// Currently gzip is the only supported compression method.
+    /// Note: Refer to the  <see
+    /// href="https://opentelemetry.io/docs/specs/otlp/#protocol-details">
+    /// OpenTelemetry Specification</see> for details />.
+    /// </summary>
+    OtlpExportCompression Compression { get; set; }
+
     /// <summary>
     /// Gets or sets the factory function called to create the <see
     /// cref="HttpClient"/> instance that will be used at runtime to

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

@@ -44,10 +44,13 @@ protected OtlpExportClient(OtlpExporterOptions options, HttpClient httpClient, s
         this.Endpoint = new UriBuilder(exporterEndpoint).Uri;
         this.Headers = options.GetHeaders<Dictionary<string, string>>((d, k, v) => d.Add(k, v));
         this.HttpClient = httpClient;
+        this.Compression = options.Compression;
     }
 
     internal HttpClient HttpClient { get; }
 
+    internal OtlpExportCompression Compression { get; }
+
     internal Uri Endpoint { get; }
 
     internal IReadOnlyDictionary<string, string> Headers { get; }
@@ -56,6 +59,8 @@ protected OtlpExportClient(OtlpExporterOptions options, HttpClient httpClient, s
 
     internal virtual bool RequireHttp2 => false;
 
+    protected abstract string? ContentEncodingHeader { get; }
+
     public abstract ExportClientResponse SendExportRequest(byte[] buffer, int contentLength, DateTime deadlineUtc, CancellationToken cancellationToken = default);
 
     /// <inheritdoc/>
@@ -83,14 +88,26 @@ protected HttpRequestMessage CreateHttpRequest(byte[] buffer, int contentLength)
             request.Headers.Add(header.Key, header.Value);
         }
 
-        // TODO: Support compression.
+        Stream data = new MemoryStream(buffer, 0, contentLength);
+
+        if (this.Compression == OtlpExportCompression.Gzip)
+        {
+            data = this.Compress(data);
+        }
 
-        request.Content = new ByteArrayContent(buffer, 0, contentLength);
+        request.Content = new StreamContent(data);
         request.Content.Headers.ContentType = this.MediaTypeHeader;
 
+        if (this.Compression == OtlpExportCompression.Gzip && this.ContentEncodingHeader != null)
+        {
+            request.Content.Headers.Add("Content-Encoding", this.ContentEncodingHeader);
+        }
+
         return request;
     }
 
+    protected abstract Stream Compress(Stream dataStream);
+
     protected HttpResponseMessage SendHttpRequest(HttpRequestMessage request, CancellationToken cancellationToken)
     {
 #if NET

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

@@ -4,6 +4,8 @@
 #if NETFRAMEWORK
 using System.Net.Http;
 #endif
+using System.Buffers.Binary;
+using System.IO.Compression;
 using System.Net.Http.Headers;
 using OpenTelemetry.Exporter.OpenTelemetryProtocol.Implementation.ExportClient.Grpc;
 
@@ -13,6 +15,7 @@ namespace OpenTelemetry.Exporter.OpenTelemetryProtocol.Implementation.ExportClie
 internal sealed class OtlpGrpcExportClient : OtlpExportClient
 {
     public const string GrpcStatusDetailsHeader = "grpc-status-details-bin";
+    private const int GrpcMessageHeaderSize = 5;
     private static readonly ExportClientHttpResponse SuccessExportResponse = new(success: true, deadlineUtc: default, response: null, exception: null);
     private static readonly MediaTypeHeaderValue MediaHeaderValue = new("application/grpc");
 
@@ -33,11 +36,22 @@ public OtlpGrpcExportClient(OtlpExporterOptions options, HttpClient httpClient,
 
     internal override bool RequireHttp2 => true;
 
+    protected override string? ContentEncodingHeader => null;
+
     /// <inheritdoc/>
     public override ExportClientResponse SendExportRequest(byte[] buffer, int contentLength, DateTime deadlineUtc, CancellationToken cancellationToken = default)
     {
         try
         {
+            // A gRPC message consists of 3 parts:
+            // byte 0     - Compression flag (0 = not compressed, 1 = compressed).
+            // bytes 1-4  - Message length in big-endian format (length of the serialized data only).
+            // bytes 5+   - Protobuf-encoded payload.
+            buffer[0] = this.Compression == OtlpExportCompression.Gzip ? (byte)1 : (byte)0;
+            Span<byte> data = new Span<byte>(buffer, 1, 4);
+            var dataLength = contentLength - GrpcMessageHeaderSize;
+            BinaryPrimitives.WriteUInt32BigEndian(data, (uint)dataLength);
+
             using var httpRequest = this.CreateHttpRequest(buffer, contentLength);
 
             // TE is required by some servers, e.g. C Core.
@@ -158,6 +172,27 @@ public override ExportClientResponse SendExportRequest(byte[] buffer, int conten
         }
     }
 
+    protected override Stream Compress(Stream dataStream)
+    {
+        var compressedStream = new MemoryStream();
+
+        compressedStream.WriteByte(1);
+        compressedStream.Write(new byte[4], 0, 4);
+
+        dataStream.Position = GrpcMessageHeaderSize;
+
+        using (var gzipStream = new GZipStream(compressedStream, CompressionLevel.Fastest, leaveOpen: true))
+        {
+            dataStream.CopyTo(gzipStream);
+        }
+
+        var compressedDataLength = (int)(compressedStream.Length - GrpcMessageHeaderSize);
+        BinaryPrimitives.WriteUInt32BigEndian(compressedStream.GetBuffer().AsSpan(1, 4), (uint)compressedDataLength);
+
+        compressedStream.Position = 0;
+        return compressedStream;
+    }
+
     private static bool IsTransientNetworkError(HttpRequestException ex)
     {
         return ex.InnerException is System.Net.Sockets.SocketException socketEx

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

@@ -4,6 +4,7 @@
 #if NETFRAMEWORK
 using System.Net.Http;
 #endif
+using System.IO.Compression;
 using System.Net.Http.Headers;
 
 namespace OpenTelemetry.Exporter.OpenTelemetryProtocol.Implementation.ExportClient;
@@ -21,6 +22,8 @@ internal OtlpHttpExportClient(OtlpExporterOptions options, HttpClient httpClient
 
     internal override MediaTypeHeaderValue MediaTypeHeader => MediaHeaderValue;
 
+    protected override string? ContentEncodingHeader => "gzip";
+
     /// <inheritdoc/>
     public override ExportClientResponse SendExportRequest(byte[] buffer, int contentLength, DateTime deadlineUtc, CancellationToken cancellationToken = default)
     {
@@ -47,4 +50,16 @@ public override ExportClientResponse SendExportRequest(byte[] buffer, int conten
             return new ExportClientHttpResponse(success: false, deadlineUtc: deadlineUtc, response: null, exception: ex);
         }
     }
+
+    protected override Stream Compress(Stream dataStream)
+    {
+        var compressedStream = new MemoryStream();
+        using (var gzipStream = new GZipStream(compressedStream, CompressionLevel.Fastest, leaveOpen: true))
+        {
+            dataStream.CopyTo(gzipStream);
+        }
+
+        compressedStream.Position = 0;
+        return compressedStream;
+    }
 }

src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/OtlpSpecConfigDefinitions.cs

@@ -15,20 +15,24 @@ internal static class OtlpSpecConfigDefinitions
     public const string DefaultHeadersEnvVarName = "OTEL_EXPORTER_OTLP_HEADERS";
     public const string DefaultTimeoutEnvVarName = "OTEL_EXPORTER_OTLP_TIMEOUT";
     public const string DefaultProtocolEnvVarName = "OTEL_EXPORTER_OTLP_PROTOCOL";
+    public const string DefaultCompressionEnvVarName = "OTEL_EXPORTER_OTLP_COMPRESSION";
 
     public const string LogsEndpointEnvVarName = "OTEL_EXPORTER_OTLP_LOGS_ENDPOINT";
     public const string LogsHeadersEnvVarName = "OTEL_EXPORTER_OTLP_LOGS_HEADERS";
     public const string LogsTimeoutEnvVarName = "OTEL_EXPORTER_OTLP_LOGS_TIMEOUT";
     public const string LogsProtocolEnvVarName = "OTEL_EXPORTER_OTLP_LOGS_PROTOCOL";
+    public const string LogsCompressionEnvVarName = "OTEL_EXPORTER_OTLP_LOGS_COMPRESSION";
 
     public const string MetricsEndpointEnvVarName = "OTEL_EXPORTER_OTLP_METRICS_ENDPOINT";
     public const string MetricsHeadersEnvVarName = "OTEL_EXPORTER_OTLP_METRICS_HEADERS";
     public const string MetricsTimeoutEnvVarName = "OTEL_EXPORTER_OTLP_METRICS_TIMEOUT";
     public const string MetricsProtocolEnvVarName = "OTEL_EXPORTER_OTLP_METRICS_PROTOCOL";
     public const string MetricsTemporalityPreferenceEnvVarName = "OTEL_EXPORTER_OTLP_METRICS_TEMPORALITY_PREFERENCE";
+    public const string MetricsCompressionEnvVarName = "OTEL_EXPORTER_OTLP_METRICS_COMPRESSION";
 
     public const string TracesEndpointEnvVarName = "OTEL_EXPORTER_OTLP_TRACES_ENDPOINT";
     public const string TracesHeadersEnvVarName = "OTEL_EXPORTER_OTLP_TRACES_HEADERS";
     public const string TracesTimeoutEnvVarName = "OTEL_EXPORTER_OTLP_TRACES_TIMEOUT";
     public const string TracesProtocolEnvVarName = "OTEL_EXPORTER_OTLP_TRACES_PROTOCOL";
+    public const string TracesCompressionEnvVarName = "OTEL_EXPORTER_OTLP_TRACES_COMPRESSION";
 }

src/OpenTelemetry.Exporter.OpenTelemetryProtocol/OtlpExportCompression.cs

@@ -0,0 +1,20 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+namespace OpenTelemetry.Exporter;
+
+/// <summary>
+/// Supported compression methods for OTLP exporter according to the specification https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/protocol/exporter.md.
+/// </summary>
+public enum OtlpExportCompression
+{
+    /// <summary>
+    /// Compression is disabled.
+    /// </summary>
+    None = 0,
+
+    /// <summary>
+    /// Gzip is the only specified compression method for now.
+    /// </summary>
+    Gzip = 1,
+}

src/OpenTelemetry.Exporter.OpenTelemetryProtocol/OtlpExporterOptions.cs

@@ -2,6 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 using System.Diagnostics;
+
 #if NETFRAMEWORK
 using System.Net.Http;
 #endif
@@ -136,6 +137,9 @@ public OtlpExportProtocol Protocol
     /// <remarks>Note: This only applies when exporting traces.</remarks>
     public BatchExportProcessorOptions<Activity> BatchExportProcessorOptions { get; set; }
 
+    /// <inheritdoc/>
+    public OtlpExportCompression Compression { get; set; } = OtlpExportCompression.None;
+
     /// <inheritdoc/>
     public Func<HttpClient> HttpClientFactory
     {
@@ -179,7 +183,8 @@ internal void ApplyConfigurationUsingSpecificationEnvVars(
         bool appendSignalPathToEndpoint,
         string protocolEnvVarKey,
         string headersEnvVarKey,
-        string timeoutEnvVarKey)
+        string timeoutEnvVarKey,
+        string compressPayloadEnvVarKey)
     {
         if (configuration.TryGetUriValue(OpenTelemetryProtocolExporterEventSource.Log, endpointEnvVarKey, out var endpoint))
         {
@@ -205,6 +210,18 @@ internal void ApplyConfigurationUsingSpecificationEnvVars(
         {
             this.TimeoutMilliseconds = timeout;
         }
+
+        if (configuration.TryGetStringValue(compressPayloadEnvVarKey, out var compressPayload))
+        {
+            if (Enum.TryParse<OtlpExportCompression>(compressPayload.Trim(), true, out var compression))
+            {
+                this.Compression = compression;
+            }
+            else
+            {
+                OpenTelemetryProtocolExporterEventSource.Log.InvalidConfigurationValue(compressPayloadEnvVarKey, compressPayload);
+            }
+        }
     }
 
     internal OtlpExporterOptions ApplyDefaults(OtlpExporterOptions defaultExporterOptions)
@@ -250,7 +267,8 @@ private void ApplyConfiguration(
                 appendSignalPathToEndpoint: true,
                 OtlpSpecConfigDefinitions.DefaultProtocolEnvVarName,
                 OtlpSpecConfigDefinitions.DefaultHeadersEnvVarName,
-                OtlpSpecConfigDefinitions.DefaultTimeoutEnvVarName);
+                OtlpSpecConfigDefinitions.DefaultTimeoutEnvVarName,
+                OtlpSpecConfigDefinitions.DefaultCompressionEnvVarName);
         }
         else if (configurationType == OtlpExporterOptionsConfigurationType.Logs)
         {
@@ -260,7 +278,8 @@ private void ApplyConfiguration(
                 appendSignalPathToEndpoint: false,
                 OtlpSpecConfigDefinitions.LogsProtocolEnvVarName,
                 OtlpSpecConfigDefinitions.LogsHeadersEnvVarName,
-                OtlpSpecConfigDefinitions.LogsTimeoutEnvVarName);
+                OtlpSpecConfigDefinitions.LogsTimeoutEnvVarName,
+                OtlpSpecConfigDefinitions.LogsCompressionEnvVarName);
         }
         else if (configurationType == OtlpExporterOptionsConfigurationType.Metrics)
         {
@@ -270,7 +289,8 @@ private void ApplyConfiguration(
                 appendSignalPathToEndpoint: false,
                 OtlpSpecConfigDefinitions.MetricsProtocolEnvVarName,
                 OtlpSpecConfigDefinitions.MetricsHeadersEnvVarName,
-                OtlpSpecConfigDefinitions.MetricsTimeoutEnvVarName);
+                OtlpSpecConfigDefinitions.MetricsTimeoutEnvVarName,
+                OtlpSpecConfigDefinitions.MetricsCompressionEnvVarName);
         }
         else if (configurationType == OtlpExporterOptionsConfigurationType.Traces)
         {
@@ -280,7 +300,8 @@ private void ApplyConfiguration(
                 appendSignalPathToEndpoint: false,
                 OtlpSpecConfigDefinitions.TracesProtocolEnvVarName,
                 OtlpSpecConfigDefinitions.TracesHeadersEnvVarName,
-                OtlpSpecConfigDefinitions.TracesTimeoutEnvVarName);
+                OtlpSpecConfigDefinitions.TracesTimeoutEnvVarName,
+                OtlpSpecConfigDefinitions.TracesCompressionEnvVarName);
         }
         else
         {

src/OpenTelemetry.Exporter.OpenTelemetryProtocol/OtlpLogExporter.cs

@@ -1,7 +1,6 @@
 // Copyright The OpenTelemetry Authors
 // SPDX-License-Identifier: Apache-2.0
 
-using System.Buffers.Binary;
 using System.Diagnostics;
 using OpenTelemetry.Exporter.OpenTelemetryProtocol.Implementation;
 using OpenTelemetry.Exporter.OpenTelemetryProtocol.Implementation.Serializer;
@@ -78,17 +77,6 @@ public override ExportResult Export(in Batch<LogRecord> logRecordBatch)
         {
             int writePosition = ProtobufOtlpLogSerializer.WriteLogsData(ref this.buffer, this.startWritePosition, this.sdkLimitOptions, this.experimentalOptions, this.Resource, logRecordBatch);
 
-            if (this.startWritePosition == GrpcStartWritePosition)
-            {
-                // Grpc payload consists of 3 parts
-                // byte 0 - Specifying if the payload is compressed.
-                // 1-4 byte - Specifies the length of payload in big endian format.
-                // 5 and above -  Protobuf serialized data.
-                Span<byte> data = new Span<byte>(this.buffer, 1, 4);
-                var dataLength = writePosition - GrpcStartWritePosition;
-                BinaryPrimitives.WriteUInt32BigEndian(data, (uint)dataLength);
-            }
-
             if (!this.transmissionHandler.TrySubmitRequest(this.buffer, writePosition))
             {
                 return ExportResult.Failure;

src/OpenTelemetry.Exporter.OpenTelemetryProtocol/OtlpMetricExporter.cs

@@ -1,7 +1,6 @@
 // Copyright The OpenTelemetry Authors
 // SPDX-License-Identifier: Apache-2.0
 
-using System.Buffers.Binary;
 using System.Diagnostics;
 using OpenTelemetry.Exporter.OpenTelemetryProtocol.Implementation;
 using OpenTelemetry.Exporter.OpenTelemetryProtocol.Implementation.Serializer;
@@ -71,17 +70,6 @@ public override ExportResult Export(in Batch<Metric> metrics)
         {
             int writePosition = ProtobufOtlpMetricSerializer.WriteMetricsData(ref this.buffer, this.startWritePosition, this.Resource, metrics);
 
-            if (this.startWritePosition == GrpcStartWritePosition)
-            {
-                // Grpc payload consists of 3 parts
-                // byte 0 - Specifying if the payload is compressed.
-                // 1-4 byte - Specifies the length of payload in big endian format.
-                // 5 and above -  Protobuf serialized data.
-                Span<byte> data = new Span<byte>(this.buffer, 1, 4);
-                var dataLength = writePosition - GrpcStartWritePosition;
-                BinaryPrimitives.WriteUInt32BigEndian(data, (uint)dataLength);
-            }
-
             if (!this.transmissionHandler.TrySubmitRequest(this.buffer, writePosition))
             {
                 return ExportResult.Failure;