Add helper methods to IApiWebRequest for serializing JSON directly to a Stream (#8019)

## Summary of changes

Adds `PostAsJson<T>` method to `IApiWebRequest`

## Reason for change

Currently, if we want to send JSON, we serialize it to a string locally,
convert the string to utf-8 bytes, potentially compress those bytes, and
then copy that to the stream. Doing that as efficiently as we can is
somewhat tricky, and we haven't always got it right. By creating a
central method, and writing directly to the underlying stream where we
can, we can potentially see efficiency gains, and can also potentially
make it easier to move to a more modern serializer later.

## Implementation details

- Added `IApiRequest.PostAsJsonAsync<T>(T payload, MultipartCompression
compression)` (and an overload that accepts json settings)
- Implemented the method as a "push stream" approach in each of the
three implementations we currently have (`ApiRequest`, `HttpClient`,
`HttpStream`)
- Benchmarked the implementation to confirm no regressions (see below)

## Test coverage

Added unit tests by specifically serializing telemetry data, and
confirming we get the correct results when we deserialize the other end
(telemetry is one of the candidates for using this approach).

When running benchmarks, it became apparent that we had a serious
regression in our allocations when we added GZIP-ing of our telemetry 😅
I didn't investigate the root cause, because switching to the new
approach (in #8017) will resolve the issue anyway.

Overall conclusion:
- In general, the new approach allocates slightly less than before
- We have a big allocation and speed regression in GZip (specifically
for telemetry), which the new approach will resolve completely
- In general, the new approach allocates the same whether you use gzip
or not
- Throughput is roughly the same both before and after


**`ApiWebRequest` (.NET FX)**

| Method                    |     Mean | Allocated | Alloc Ratio |
| ------------------------- | -------: | --------: | ----------: |
| ApiWebRequest_Before_Gzip | 6.460 ms | 572.44 KB |        1.00 |
| ApiWebRequest_After_Gzip  | 2.037 ms |  20.75 KB |        0.04 |
|                           |          |           |             |
| ApiWebRequest_Before      | 1.949 ms |  22.34 KB |        1.00 |
| ApiWebRequest_After       | 1.908 ms |  20.75 KB |        0.93 |

**`HttpClientRequest` (.NET Core 3.1, .NET 6)** - had to re-enable
keep-alive to avoid connection exhaustion!

| Method | Runtime | Mean | Allocated | Alloc Ratio |
| ---------------------- | ------------- | ---------: | --------: |
----------: |
| HttpClient_Before_Gzip | .NET 6.0 | 4,980.1 us | 406.27 KB | 0.98 |
| HttpClient_After_Gzip | .NET 6.0 | 161.5 us | 12.04 KB | 0.03 |
| HttpClient_Before_Gzip | .NET Core 3.1 | 4,847.4 us | 414.43 KB | 1.00
|
| HttpClient_After_Gzip | .NET Core 3.1 | 166.3 us | 12.97 KB | 0.03 |
| | | | | |
| HttpClient_Before | .NET 6.0 | 129.2 us | 13.03 KB | 0.91 |
| HttpClient_After | .NET 6.0 | 154.9 us | 12.05 KB | 0.84 |
| HttpClient_Before | .NET Core 3.1 | 162.2 us | 14.27 KB | 1.00 |
| HttpClient_After | .NET Core 3.1 | 189.6 us | 13.2 KB | 0.92 |

**`HttpStreamRequest` over UDS (.NET Core 3.1, .NET 6)**

| Method | Runtime | Mean | Allocated | Alloc Ratio |
| ---------------------- | ------------- | ---------: | --------: |
----------: |
| HttpStream_Before_Gzip | .NET 6.0 | 5,362.8 us | 440.87 KB | 0.99 |
| HttpStream_After_Gzip | .NET 6.0 | 444.8 us | 42.63 KB | 0.10 |
| HttpStream_Before_Gzip | .NET Core 3.1 | 5,421.3 us | 446.78 KB | 1.00
|
| HttpStream_After_Gzip | .NET Core 3.1 | 462.6 us | 42.77 KB | 0.10 |
| | | | | |
| HttpStream_Before | .NET 6.0 | 428.7 us | 43.73 KB | 1.01 |
| HttpStream_After | .NET 6.0 | 433.3 us | 42.3 KB | 0.97 |
| HttpStream_Before | .NET Core 3.1 | 445.3 us | 43.5 KB | 1.00 |
| HttpStream_After | .NET Core 3.1 | 448.5 us | 42.62 KB | 0.98 |

**`SocketHandlerRequest` over UDS (..NET 6)**

| Method                    |        Mean | Allocated | Alloc Ratio |
| ------------------------- | ----------: | --------: | ----------: |
| SocketHandler_Before_Gzip | 5,070.65 us | 406.26 KB |        1.00 |
| SocketHandler_After_Gzip  |    97.66 us |  12.28 KB |        0.03 |
|                           |             |           |             |
| SocketHandler_Before      |    53.95 us |  13.01 KB |        1.00 |
| SocketHandler_After       |    87.64 us |  12.28 KB |        0.94 |



<details><summary>Benchmark used (approx)</summary>


```csharp
using System;
using System.IO;
using System.IO.Compression;
using System.Net;
using System.Text;
using System.Threading.Tasks;
using BenchmarkDotNet.Attributes;
using BenchmarkDotNet.Configs;
using Datadog.Trace;
using Datadog.Trace.Agent;
using Datadog.Trace.Agent.StreamFactories;
using Datadog.Trace.Agent.Transports;
using Datadog.Trace.Configuration;
using Datadog.Trace.DogStatsd;
using Datadog.Trace.HttpOverStreams;
using Datadog.Trace.Tagging;
using Datadog.Trace.Telemetry;
using Datadog.Trace.Telemetry.Transports;
using Datadog.Trace.Util;
using Datadog.Trace.Vendors.Newtonsoft.Json;
using Datadog.Trace.Vendors.Newtonsoft.Json.Serialization;

namespace Benchmarks.Trace;

[MemoryDiagnoser]
[GroupBenchmarksBy(BenchmarkLogicalGroupRule.ByCategory)]
[CategoriesColumn]
public class TelemetryHttpClientBenchmark
{
    private const string BaseUrl = "http://localhost:5035";
    private const string Socket = @"C:\repos\temp\temp74\bin\Release\net10.0\test.socket";

    private TelemetryData _telemetryData;
    private ApiWebRequestFactory _apiWebRequestFactory;
    private Uri _apiEndpointUri;

#if NETCOREAPP3_1_OR_GREATER
    private HttpClientRequestFactory _httpClientRequestFactory;
    private Uri _httpClientEndpointUri;

    private HttpStreamRequestFactory _httpStreamRequestFactory;
    private Uri _httpStreamEndpointUri;
#endif
#if NET5_0_OR_GREATER
    private SocketHandlerRequestFactory _socketHandlerRequestFactory;
    private Uri _socketHandlerEndpointUri;
#endif

    [GlobalSetup]
    public void GlobalSetup()
    {
        _telemetryData = GetData();

        var config = TracerHelper.DefaultConfig;
        config.Add(ConfigurationKeys.TraceEnabled, false);
        var settings = TracerSettings.Create(config);

        _apiWebRequestFactory = new ApiWebRequestFactory(new Uri(BaseUrl), AgentHttpHeaderNames.MinimalHeaders);
        _apiEndpointUri = _apiWebRequestFactory.GetEndpoint("/");

#if NETCOREAPP3_1_OR_GREATER
        _httpClientRequestFactory = new HttpClientRequestFactory(new Uri(BaseUrl), AgentHttpHeaderNames.MinimalHeaders);
        _httpClientEndpointUri = _httpClientRequestFactory.GetEndpoint("/");

        _httpStreamRequestFactory = new HttpStreamRequestFactory(
            new UnixDomainSocketStreamFactory(Socket),
            new DatadogHttpClient(new MinimalAgentHeaderHelper()),
            new Uri(BaseUrl));
        _httpStreamEndpointUri = _httpStreamRequestFactory.GetEndpoint("/");
#endif

#if NET5_0_OR_GREATER
        _socketHandlerRequestFactory = new SocketHandlerRequestFactory(
            new UnixDomainSocketStreamFactory(Socket),
            AgentHttpHeaderNames.MinimalHeaders,
            new Uri(BaseUrl));
        _socketHandlerEndpointUri = _socketHandlerRequestFactory.GetEndpoint("/");
#endif
    }

    [GlobalCleanup]
    public void GlobalCleanup()
    {
    }

    [BenchmarkCategory("ApiWebRequest", "Uncompressed"), Benchmark(Baseline = true)]
    public async Task<int> ApiWebRequest_Before()
    {
        var request = _apiWebRequestFactory.Create(_apiEndpointUri);
        var data = SerializeTelemetry(_telemetryData);
        using var response = await request.PostAsync(new ArraySegment<byte>(Encoding.UTF8.GetBytes(data)), "application/json", contentEncoding: null).ConfigureAwait(false);
        return response.StatusCode;
    }

    [BenchmarkCategory("ApiWebRequest", "Gzip"), Benchmark(Baseline = true)]
    public async Task<int> ApiWebRequest_Before_Gzip()
    {
        var request = _apiWebRequestFactory.Create(_apiEndpointUri);
        var data = SerializeTelemetryWithGzip(_telemetryData);
        using var response = await request.PostAsync(new ArraySegment<byte>(data), "application/json", contentEncoding: "gzip").ConfigureAwait(false);
        return response.StatusCode;
    }

    [BenchmarkCategory("ApiWebRequest", "Uncompressed"), Benchmark]
    public async Task<int> ApiWebRequest_After()
    {
        var request = _apiWebRequestFactory.Create(_apiEndpointUri);
        using var response = await request.PostAsJsonAsync(request, compression: MultipartCompression.None);
        return response.StatusCode;
    }

    [BenchmarkCategory("ApiWebRequest", "Gzip"), Benchmark]
    public async Task<int> ApiWebRequest_After_Gzip()
    {
        var request = _apiWebRequestFactory.Create(_apiEndpointUri);
        using var response = await request.PostAsJsonAsync(request, compression: MultipartCompression.None);
        return response.StatusCode;
    }

#if NETCOREAPP3_1_OR_GREATER

    [BenchmarkCategory("HttpClient", "Uncompressed"), Benchmark(Baseline = true)]
    public async Task<int> HttpClient_Before()
    {
        var request = _httpClientRequestFactory.Create(_httpClientEndpointUri);
        var data = SerializeTelemetry(_telemetryData);
        using var response = await request.PostAsync(new ArraySegment<byte>(Encoding.UTF8.GetBytes(data)), "application/json", contentEncoding: null).ConfigureAwait(false);
        return response.StatusCode;
    }

    [BenchmarkCategory("HttpClient", "Gzip"), Benchmark(Baseline = true)]
    public async Task<int> HttpClient_Before_Gzip()
    {
        var request = _httpClientRequestFactory.Create(_httpClientEndpointUri);
        var data = SerializeTelemetryWithGzip(_telemetryData);
        using var response = await request.PostAsync(new ArraySegment<byte>(data), "application/json", contentEncoding: "gzip").ConfigureAwait(false);
        return response.StatusCode;
    }

    [BenchmarkCategory("HttpClient", "Uncompressed"), Benchmark]
    public async Task<int> HttpClient_After()
    {
        var request = _httpClientRequestFactory.Create(_httpClientEndpointUri);
        using var response = await request.PostAsJsonAsync(request, compression: MultipartCompression.None);
        return response.StatusCode;
    }

    [BenchmarkCategory("HttpClient", "Gzip"), Benchmark]
    public async Task<int> HttpClient_After_Gzip()
    {
        var request = _httpClientRequestFactory.Create(_httpClientEndpointUri);
        using var response = await request.PostAsJsonAsync(request, compression: MultipartCompression.None);
        return response.StatusCode;
    }
#endif

#if NETCOREAPP3_1_OR_GREATER
    [BenchmarkCategory("HttpStream", "Uncompressed"), Benchmark(Baseline = true)]
    public async Task<int> HttpStream_Before()
    {
        var request = _httpStreamRequestFactory.Create(_httpStreamEndpointUri);
        var data = SerializeTelemetry(_telemetryData);
        using var response = await request.PostAsync(new ArraySegment<byte>(Encoding.UTF8.GetBytes(data)), "application/json", contentEncoding: null).ConfigureAwait(false);
        return response.StatusCode;
    }

    [BenchmarkCategory("HttpStream", "Gzip"), Benchmark(Baseline = true)]
    public async Task<int> HttpStream_Before_Gzip()
    {
        var request = _httpStreamRequestFactory.Create(_httpStreamEndpointUri);
        var data = SerializeTelemetryWithGzip(_telemetryData);
        using var response = await request.PostAsync(new ArraySegment<byte>(data), "application/json", contentEncoding: "gzip").ConfigureAwait(false);
        return response.StatusCode;
    }

    [BenchmarkCategory("HttpStream", "Uncompressed"), Benchmark]
    public async Task<int> HttpStream_After()
    {
        var request = _httpStreamRequestFactory.Create(_httpStreamEndpointUri);
        using var response = await request.PostAsJsonAsync(request, compression: MultipartCompression.None);
        return response.StatusCode;
    }

    [BenchmarkCategory("HttpStream", "Gzip"), Benchmark]
    public async Task<int> HttpStream_After_Gzip()
    {
        var request = _httpStreamRequestFactory.Create(_httpStreamEndpointUri);
        using var response = await request.PostAsJsonAsync(request, compression: MultipartCompression.None);
        return response.StatusCode;
    }
#endif

#if NET5_0_OR_GREATER
    [BenchmarkCategory("SocketHandler", "Uncompressed"), Benchmark(Baseline = true)]
    public async Task<int> SocketHandler_Before()
    {
        var request = _socketHandlerRequestFactory.Create(_socketHandlerEndpointUri);
        var data = SerializeTelemetry(_telemetryData);
        using var response = await request.PostAsync(new ArraySegment<byte>(Encoding.UTF8.GetBytes(data)), "application/json", contentEncoding: null).ConfigureAwait(false);
        return response.StatusCode;
    }

    [BenchmarkCategory("SocketHandler", "Gzip"), Benchmark(Baseline = true)]
    public async Task<int> SocketHandler_Before_Gzip()
    {
        var request = _socketHandlerRequestFactory.Create(_socketHandlerEndpointUri);
        var data = SerializeTelemetryWithGzip(_telemetryData);
        using var response = await request.PostAsync(new ArraySegment<byte>(data), "application/json", contentEncoding: "gzip").ConfigureAwait(false);
        return response.StatusCode;
    }

    [BenchmarkCategory("SocketHandler", "Uncompressed"), Benchmark]
    public async Task<int> SocketHandler_After()
    {
        var request = _socketHandlerRequestFactory.Create(_socketHandlerEndpointUri);
        using var response = await request.PostAsJsonAsync(request, compression: MultipartCompression.None);
        return response.StatusCode;
    }

    [BenchmarkCategory("SocketHandler", "Gzip"), Benchmark]
    public async Task<int> SocketHandler_After_Gzip()
    {
        var request = _socketHandlerRequestFactory.Create(_socketHandlerEndpointUri);
        using var response = await request.PostAsJsonAsync(request, compression: MultipartCompression.None);
        return response.StatusCode;
    }
#endif

    internal static string SerializeTelemetry<T>(T data) => JsonConvert.SerializeObject(data, Formatting.None, JsonTelemetryTransport.SerializerSettings);

    internal static byte[] SerializeTelemetryWithGzip<T>(T data)
    {
        using var memStream = new MemoryStream();
        using (var zipStream = new GZipStream(memStream, CompressionMode.Compress, true))
        {
            using var streamWriter = new StreamWriter(zipStream);
            using var jsonWriter = new JsonTextWriter(streamWriter);
            var serializer = new JsonSerializer { NullValueHandling = NullValueHandling.Ignore, ContractResolver = new DefaultContractResolver { NamingStrategy = new SnakeCaseNamingStrategy(), }, Formatting = Formatting.None };

            serializer.Serialize(jsonWriter, data);
        }

        return memStream.ToArray();
    }

    private TelemetryData GetData() =>
        new TelemetryData(
            requestType: TelemetryRequestTypes.GenerateMetrics,
            runtimeId: "20338dfd-f700-4e5c-b3f6-0d470f054ae8",
            seqId: 5672,
            tracerTime: 1628099086,
            application: new ApplicationTelemetryData(
                serviceName: "myapp",
                env: "prod",
                serviceVersion: "1.2.3",
                tracerVersion: "0.33.1",
                languageName: "node.js",
                languageVersion: "14.16.1",
                runtimeName: "dotnet",
                runtimeVersion: "7.0.3",
                commitSha: "testCommitSha",
                repositoryUrl: "testRepositoryUrl",
                processTags: "entrypoint.basedir:Users,entrypoint.workdir:Downloads"),
            host: new HostTelemetryData(
                hostname: "i-09ecf74c319c49be8",
                os: "GNU/Linux",
                architecture: "x86_64")
            {
                OsVersion = "ubuntu 18.04.5 LTS (Bionic Beaver)",
                KernelName = "Linux",
                KernelRelease = "5.4.0-1037-gcp",
                KernelVersion = "#40~18.04.1-Ubuntu SMP Fri Feb 5 15:41:35 UTC 2021"
            },
            payload: new GenerateMetricsPayload(
                new MetricData[]
                {
                    new(
                        "tracer_init_time",
                        new MetricSeries()
                        {
                            new(1575317847, 2241),
                            new(1575317947, 2352),
                        },
                        common: true,
                        type: MetricTypeConstants.Count)
                    {
                        Tags = new[]
                        {
                            "org_id: 2",
                            "environment:test"
                        }
                    },
                    new(
                        "app_sec_initialization_time",
                        new MetricSeries()
                        {
                            new(1575317447, 254),
                            new(1575317547, 643),
                        },
                        common: false,
                        type: MetricTypeConstants.Gauge)
                    {
                        Namespace = MetricNamespaceConstants.ASM,
                        Interval = 60,
                    },
                }));
}
```

</details> 


## Other details

Part of a small stack

- #8019 👈 
- #8017

Author: andrewlock

tracer/src/Datadog.Trace/Agent/IApiRequest.cs

@@ -7,6 +7,7 @@
 using System.IO;
 using System.Threading.Tasks;
 using Datadog.Trace.Agent.Transports;
+using Datadog.Trace.Vendors.Newtonsoft.Json;
 
 namespace Datadog.Trace.Agent
 {
@@ -20,6 +21,10 @@ internal interface IApiRequest
 
         Task<IApiResponse> PostAsync(ArraySegment<byte> bytes, string contentType, string contentEncoding);
 
+        Task<IApiResponse> PostAsJsonAsync<T>(T payload, MultipartCompression compression);
+
+        Task<IApiResponse> PostAsJsonAsync<T>(T payload, MultipartCompression compression, JsonSerializerSettings settings);
+
         Task<IApiResponse> PostAsync(Func<Stream, Task> writeToRequestStream, string contentType, string contentEncoding, string multipartBoundary);
 
         Task<IApiResponse> PostAsync(MultipartFormItem[] items, MultipartCompression multipartCompression = MultipartCompression.None);

tracer/src/Datadog.Trace/Agent/Transports/ApiWebRequest.cs

@@ -11,6 +11,8 @@
 using System.Threading.Tasks;
 using Datadog.Trace.Logging;
 using Datadog.Trace.Util;
+using Datadog.Trace.Vendors.Newtonsoft.Json;
+using Datadog.Trace.Vendors.Serilog.Events;
 using static Datadog.Trace.HttpOverStreams.DatadogHttpValues;
 
 namespace Datadog.Trace.Agent.Transports
@@ -58,6 +60,42 @@ public async Task<IApiResponse> PostAsync(ArraySegment<byte> bytes, string conte
             return await FinishAndGetResponse().ConfigureAwait(false);
         }
 
+        public Task<IApiResponse> PostAsJsonAsync<T>(T payload, MultipartCompression compression)
+            => PostAsJsonAsync(payload, compression, SerializationHelpers.DefaultJsonSettings);
+
+        public async Task<IApiResponse> PostAsJsonAsync<T>(T payload, MultipartCompression compression, JsonSerializerSettings settings)
+        {
+            var contentEncoding = compression == MultipartCompression.GZip ? "gzip" : null;
+            if (Log.IsEnabled(LogEventLevel.Debug))
+            {
+                Log.Debug("Sending {Type} data as JSON with compression '{Compression}'", typeof(T).FullName, contentEncoding ?? "none");
+            }
+
+            ResetRequest(method: "POST", contentType: MimeTypes.Json, contentEncoding: contentEncoding);
+
+            using (var reqStream = await _request.GetRequestStreamAsync().ConfigureAwait(false))
+            {
+                // wrap in gzip if requested
+                using Stream gzip = (compression == MultipartCompression.GZip
+                                      ? new GZipStream(reqStream, CompressionMode.Compress, leaveOpen: true)
+                                      : null);
+                var streamToWriteTo = gzip ?? reqStream;
+
+                using var streamWriter = new StreamWriter(streamToWriteTo, EncodingHelpers.Utf8NoBom, bufferSize: 1024, leaveOpen: true);
+                using var jsonWriter = new JsonTextWriter(streamWriter)
+                {
+                    CloseOutput = false
+                };
+                var serializer = JsonSerializer.Create(settings);
+                serializer.Serialize(jsonWriter, payload);
+                await streamWriter.FlushAsync().ConfigureAwait(false);
+                await streamToWriteTo.FlushAsync().ConfigureAwait(false);
+                await reqStream.FlushAsync().ConfigureAwait(false);
+            }
+
+            return await FinishAndGetResponse().ConfigureAwait(false);
+        }
+
         public async Task<IApiResponse> PostAsync(Func<Stream, Task> writeToRequestStream, string contentType, string contentEncoding, string multipartBoundary)
         {
             ResetRequest(method: "POST", ContentTypeHelper.GetContentType(contentType, multipartBoundary), contentEncoding);

tracer/src/Datadog.Trace/Agent/Transports/HttpClientRequest.cs

@@ -6,12 +6,16 @@
 #if NETCOREAPP
 using System;
 using System.IO;
+using System.IO.Compression;
 using System.Net.Http;
 using System.Net.Http.Headers;
 using System.Threading.Tasks;
 using Datadog.Trace.AppSec;
 using Datadog.Trace.HttpOverStreams;
 using Datadog.Trace.Logging;
+using Datadog.Trace.Util;
+using Datadog.Trace.Vendors.Newtonsoft.Json;
+using Datadog.Trace.Vendors.Serilog.Events;
 
 namespace Datadog.Trace.Agent.Transports
 {
@@ -67,6 +71,50 @@ public async Task<IApiResponse> PostAsync(ArraySegment<byte> bytes, string conte
             }
         }
 
+        public Task<IApiResponse> PostAsJsonAsync<T>(T payload, MultipartCompression compression)
+            => PostAsJsonAsync(payload, compression, SerializationHelpers.DefaultJsonSettings);
+
+        public async Task<IApiResponse> PostAsJsonAsync<T>(T payload, MultipartCompression compression, JsonSerializerSettings settings)
+        {
+            if (Log.IsEnabled(LogEventLevel.Debug))
+            {
+                Log.Debug("Sending {Type} data as JSON with compression '{Compression}'", typeof(T).FullName, compression == MultipartCompression.GZip ? "gzip" : "none");
+            }
+
+            using var content = new PushStreamContent(stream => WriteAsJson(stream, payload, settings, compression));
+            content.Headers.ContentType = new MediaTypeHeaderValue(MimeTypes.Json);
+
+            if (compression == MultipartCompression.GZip)
+            {
+                content.Headers.ContentEncoding.Add("gzip");
+            }
+
+            _postRequest.Content = content;
+
+            var response = await _client.SendAsync(_postRequest).ConfigureAwait(false);
+            return new HttpClientResponse(response);
+
+            static async Task WriteAsJson(Stream requestStream, T payload, JsonSerializerSettings serializationSettings, MultipartCompression compression)
+            {
+                // wrap in gzip if requested
+                using Stream gzip = compression == MultipartCompression.GZip
+                                        ? new GZipStream(requestStream, CompressionMode.Compress, leaveOpen: true)
+                                        : null;
+                var streamToWriteTo = gzip ?? requestStream;
+
+                using var streamWriter = new StreamWriter(streamToWriteTo, EncodingHelpers.Utf8NoBom, bufferSize: 1024, leaveOpen: true);
+                using var jsonWriter = new JsonTextWriter(streamWriter)
+                {
+                    CloseOutput = false
+                };
+                var serializer = JsonSerializer.Create(serializationSettings);
+                serializer.Serialize(jsonWriter, payload);
+                await streamWriter.FlushAsync().ConfigureAwait(false);
+                await streamToWriteTo.FlushAsync().ConfigureAwait(false);
+                await requestStream.FlushAsync().ConfigureAwait(false);
+            }
+        }
+
         public async Task<IApiResponse> PostAsync(Func<Stream, Task> writeToRequestStream, string contentType, string contentEncoding, string multipartBoundary)
         {
             // re-create HttpContent on every retry because some versions of HttpClient always dispose of it, so we can't reuse.

tracer/src/Datadog.Trace/Agent/Transports/HttpStreamRequest.cs

@@ -5,16 +5,22 @@
 
 using System;
 using System.IO;
+using System.IO.Compression;
 using System.Net;
 using System.Threading.Tasks;
 using Datadog.Trace.HttpOverStreams;
 using Datadog.Trace.HttpOverStreams.HttpContent;
+using Datadog.Trace.Logging;
 using Datadog.Trace.Util;
+using Datadog.Trace.Vendors.Newtonsoft.Json;
+using Datadog.Trace.Vendors.Serilog.Events;
 
 namespace Datadog.Trace.Agent.Transports
 {
     internal sealed class HttpStreamRequest : IApiRequest
     {
+        private static readonly IDatadogLogger Log = DatadogLogging.GetLoggerFor<HttpStreamRequest>();
+
         private readonly Uri _uri;
         private readonly DatadogHttpClient _client;
         private readonly IStreamFactory _streamFactory;
@@ -41,6 +47,48 @@ public Task<IApiResponse> PostAsync(ArraySegment<byte> bytes, string contentType
         public async Task<IApiResponse> PostAsync(ArraySegment<byte> bytes, string contentType, string contentEncoding)
             => (await SendAsync(WebRequestMethods.Http.Post, contentType, new BufferContent(bytes), contentEncoding, chunkedEncoding: false).ConfigureAwait(false)).Item1;
 
+        public Task<IApiResponse> PostAsJsonAsync<T>(T payload, MultipartCompression compression)
+            => PostAsJsonAsync(payload, compression, SerializationHelpers.DefaultJsonSettings);
+
+        public async Task<IApiResponse> PostAsJsonAsync<T>(T payload, MultipartCompression compression, JsonSerializerSettings settings)
+        {
+            var contentEncoding = compression == MultipartCompression.GZip ? "gzip" : null;
+            if (Log.IsEnabled(LogEventLevel.Debug))
+            {
+                Log.Debug("Sending {Type} data as JSON with compression '{Compression}'", typeof(T).FullName, contentEncoding ?? "none");
+            }
+
+            var result = await SendAsync(
+                                 WebRequestMethods.Http.Post,
+                                 contentType: MimeTypes.Json,
+                                 content: new HttpOverStreams.HttpContent.PushStreamContent(stream => WriteAsJson(stream, payload, settings, compression)),
+                                 contentEncoding: contentEncoding,
+                                 chunkedEncoding: true) // must use chunked encoding because push-stream content
+                            .ConfigureAwait(false);
+
+            return result.Item1;
+
+            static async Task WriteAsJson(Stream requestStream, T payload, JsonSerializerSettings serializationSettings, MultipartCompression compression)
+            {
+                // wrap in gzip if requested
+                using Stream gzip = compression == MultipartCompression.GZip
+                                        ? new GZipStream(requestStream, CompressionMode.Compress, leaveOpen: true)
+                                        : null;
+                var streamToWriteTo = gzip ?? requestStream;
+
+                using var streamWriter = new StreamWriter(streamToWriteTo, EncodingHelpers.Utf8NoBom, bufferSize: 1024, leaveOpen: true);
+                using var jsonWriter = new JsonTextWriter(streamWriter)
+                {
+                    CloseOutput = false
+                };
+                var serializer = JsonSerializer.Create(serializationSettings);
+                serializer.Serialize(jsonWriter, payload);
+                await streamWriter.FlushAsync().ConfigureAwait(false);
+                await streamToWriteTo.FlushAsync().ConfigureAwait(false);
+                await requestStream.FlushAsync().ConfigureAwait(false);
+            }
+        }
+
         public async Task<IApiResponse> PostAsync(Func<Stream, Task> writeToRequestStream, string contentType, string contentEncoding, string multipartBoundary)
             => (await SendAsync(WebRequestMethods.Http.Post, contentType, new HttpOverStreams.HttpContent.PushStreamContent(writeToRequestStream), contentEncoding, chunkedEncoding: true, multipartBoundary).ConfigureAwait(false)).Item1;
 

tracer/src/Datadog.Trace/Agent/Transports/SerializationHelpers.cs

@@ -0,0 +1,23 @@
+// <copyright file="SerializationHelpers.cs" company="Datadog">
+// Unless explicitly stated otherwise all files in this repository are licensed under the Apache 2 License.
+// This product includes software developed at Datadog (https://www.datadoghq.com/). Copyright 2017 Datadog, Inc.
+// </copyright>
+
+#nullable enable
+
+using Datadog.Trace.Vendors.Newtonsoft.Json;
+using Datadog.Trace.Vendors.Newtonsoft.Json.Serialization;
+
+namespace Datadog.Trace.Agent.Transports;
+
+internal static class SerializationHelpers
+{
+    public static readonly JsonSerializerSettings DefaultJsonSettings = new()
+    {
+        NullValueHandling = NullValueHandling.Ignore,
+        ContractResolver = new DefaultContractResolver
+        {
+            NamingStrategy = new SnakeCaseNamingStrategy(),
+        }
+    };
+}

tracer/test/Datadog.Trace.TestHelpers/TransportHelpers/TestApiRequest.cs

@@ -10,6 +10,7 @@
 using System.Threading.Tasks;
 using Datadog.Trace.Agent;
 using Datadog.Trace.Agent.Transports;
+using Datadog.Trace.Vendors.Newtonsoft.Json;
 
 namespace Datadog.Trace.TestHelpers.TransportHelpers;
 
@@ -64,6 +65,18 @@ public virtual Task<IApiResponse> PostAsync(ArraySegment<byte> bytes, string con
         return Task.FromResult((IApiResponse)response);
     }
 
+    public virtual Task<IApiResponse> PostAsJsonAsync<T>(T payload, MultipartCompression compression)
+        => PostAsJsonAsync(payload, compression, SerializationHelpers.DefaultJsonSettings);
+
+    public virtual Task<IApiResponse> PostAsJsonAsync<T>(T payload, MultipartCompression compression, JsonSerializerSettings settings)
+    {
+        var response = new TestApiResponse(_statusCode, _responseContent, _responseContentType);
+        Responses.Add(response);
+        ContentType = MimeTypes.Json;
+
+        return Task.FromResult((IApiResponse)response);
+    }
+
     public async Task<IApiResponse> PostAsync(Func<Stream, Task> writeToRequestStream, string contentType, string contentEncoding, string multipartBoundary)
     {
         using (var ms = new MemoryStream())