added more http compression (responses also)

This commit is contained in:
Oliver Kieselbach
2026-04-15 00:34:18 +02:00
parent 9b2f801530
commit a3d37c67ac
8 changed files with 97 additions and 59 deletions
@@ -95,6 +95,13 @@ namespace AutopilotMonitor.Agent.Core.Monitoring.Network
// Azure App Service extracts the cert from the TLS handshake and forwards it
// in the X-ARR-ClientCert header to the backend function.
var handler = new HttpClientHandler();
// AutomaticDecompression sends Accept-Encoding and transparently decompresses
// gzip/deflate responses. Bandwidth-sensitive links benefit a lot on JSON-heavy
// endpoints like GetAgentConfig.
if (handler.SupportsAutomaticDecompression)
{
handler.AutomaticDecompression = System.Net.DecompressionMethods.GZip | System.Net.DecompressionMethods.Deflate;
}
if (_clientCertificate != null)
{
handler.ClientCertificates.Add(_clientCertificate);
@@ -1 +1 @@
978
980
@@ -5,7 +5,9 @@ using Newtonsoft.Json;
namespace AutopilotMonitor.Functions.Tests;
/// <summary>
/// Tests for NdjsonParser — the NDJSON+gzip parsing contract.
/// Tests for NdjsonParser — the NDJSON parsing contract.
/// Gzip transport compression is handled by ASP.NET Core's UseRequestDecompression middleware,
/// so this parser only deals with already-decompressed NDJSON streams.
/// If the wire format between agent and backend ever drifts, these tests will catch it.
/// </summary>
public class NdjsonParserTests
@@ -137,19 +139,28 @@ public class NdjsonParserTests
Assert.Throws<InvalidOperationException>(() => NdjsonParser.ParseNdjson(""));
}
// --- ParseGzipAsync ---
// --- ParseNdjsonStreamAsync (decompressed stream — middleware handles gzip upstream) ---
private static byte[] BuildNdjsonBytes(string sessionId, string tenantId, IEnumerable<object> events)
{
var sb = new System.Text.StringBuilder();
sb.AppendLine(JsonConvert.SerializeObject(new { SessionId = sessionId, TenantId = tenantId }));
foreach (var evt in events)
sb.AppendLine(JsonConvert.SerializeObject(evt));
return System.Text.Encoding.UTF8.GetBytes(sb.ToString());
}
[Fact]
public async Task ParseGzipAsync_ValidPayload_ReturnsCorrectData()
public async Task ParseNdjsonStreamAsync_ValidPayload_ReturnsCorrectData()
{
var payload = NdjsonParser.BuildGzipPayload(ValidSessionId, ValidTenantId,
var payload = BuildNdjsonBytes(ValidSessionId, ValidTenantId,
[
new EnrollmentEvent { EventType = "phase_changed" },
new EnrollmentEvent { EventType = "app_install_started" },
]);
using var stream = new MemoryStream(payload);
var (sessionId, tenantId, events) = await NdjsonParser.ParseGzipAsync(stream, 5 * 1024 * 1024);
var (sessionId, tenantId, events) = await NdjsonParser.ParseNdjsonStreamAsync(stream, 5 * 1024 * 1024);
Assert.Equal(ValidSessionId, sessionId);
Assert.Equal(ValidTenantId, tenantId);
@@ -157,9 +168,9 @@ public class NdjsonParserTests
}
[Fact]
public async Task ParseGzipAsync_PayloadExceedsMaxSize_Throws()
public async Task ParseNdjsonStreamAsync_PayloadExceedsMaxSize_Throws()
{
var payload = NdjsonParser.BuildGzipPayload(ValidSessionId, ValidTenantId,
var payload = BuildNdjsonBytes(ValidSessionId, ValidTenantId,
Enumerable.Range(0, 100).Select(_ => new EnrollmentEvent
{
EventType = "test",
@@ -170,18 +181,16 @@ public class NdjsonParserTests
// Restrict to 1 byte — will throw immediately
await Assert.ThrowsAsync<InvalidOperationException>(
() => NdjsonParser.ParseGzipAsync(stream, maxSizeBytes: 1));
() => NdjsonParser.ParseNdjsonStreamAsync(stream, maxSizeBytes: 1));
}
// --- BuildGzipPayload (test helper verification) ---
[Fact]
public async Task BuildGzipPayload_RoundTrip_PreservesMetadata()
public async Task ParseNdjsonStreamAsync_EmptyEvents_ReturnsMetadataOnly()
{
var payload = NdjsonParser.BuildGzipPayload(ValidSessionId, ValidTenantId, []);
var payload = BuildNdjsonBytes(ValidSessionId, ValidTenantId, []);
using var stream = new MemoryStream(payload);
var (sessionId, tenantId, events) = await NdjsonParser.ParseGzipAsync(stream, 5 * 1024 * 1024);
var (sessionId, tenantId, events) = await NdjsonParser.ParseNdjsonStreamAsync(stream, 5 * 1024 * 1024);
Assert.Equal(ValidSessionId, sessionId);
Assert.Equal(ValidTenantId, tenantId);
@@ -15,7 +15,7 @@ namespace AutopilotMonitor.Functions.Functions.Ingest
/// </summary>
public partial class IngestEventsFunction
{
private async Task<IngestEventsRequest> ParseNdjsonGzipRequest(Stream body, string? tenantId = null)
private async Task<IngestEventsRequest> ParseNdjsonRequest(Stream body, string? tenantId = null)
{
var config = !string.IsNullOrEmpty(tenantId)
? await _configService.GetConfigurationAsync(tenantId)
@@ -23,7 +23,9 @@ namespace AutopilotMonitor.Functions.Functions.Ingest
var maxPayloadSizeBytes = (config?.MaxNdjsonPayloadSizeMB ?? 5) * 1024 * 1024;
var (sessionId, metaTenantId, events) = await NdjsonParser.ParseGzipAsync(body, maxPayloadSizeBytes);
// body is already gzip-decompressed by the UseRequestDecompression middleware if the
// client sent Content-Encoding: gzip — parser reads plain NDJSON.
var (sessionId, metaTenantId, events) = await NdjsonParser.ParseNdjsonStreamAsync(body, maxPayloadSizeBytes);
_logger.LogDebug("NDJSON payload parsed: {EventCount} events (limit: {LimitMB} MB)",
events.Count, config?.MaxNdjsonPayloadSizeMB ?? 5);
@@ -48,7 +48,7 @@ namespace AutopilotMonitor.Functions.Functions.Ingest
// Session-aware block: parse body to check if this is a different (new) session
if (!string.IsNullOrEmpty(blockedSessionIds))
{
earlyParsedRequest = await ParseNdjsonGzipRequest(req.Body, tenantId);
earlyParsedRequest = await ParseNdjsonRequest(req.Body, tenantId);
if (earlyParsedRequest != null && !string.IsNullOrEmpty(earlyParsedRequest.SessionId))
{
var (stillBlocked, _, _, _) = await _blockedDeviceService.IsBlockedAsync(
@@ -128,7 +128,7 @@ namespace AutopilotMonitor.Functions.Functions.Ingest
// --- Parse NDJSON+gzip request body (only after security is cleared) ---
// If session-aware block check already parsed the body, reuse that result (stream is consumed)
var request = earlyParsedRequest ?? await ParseNdjsonGzipRequest(req.Body, tenantId);
var request = earlyParsedRequest ?? await ParseNdjsonRequest(req.Body, tenantId);
if (request?.Events == null || request.Events.Count == 0)
{
@@ -1,4 +1,3 @@
using System.IO.Compression;
using System.Text;
using AutopilotMonitor.Shared.Models;
using Newtonsoft.Json;
@@ -7,38 +6,36 @@ using Newtonsoft.Json.Linq;
namespace AutopilotMonitor.Functions.Functions.Ingest
{
/// <summary>
/// Pure NDJSON+gzip parsing logic — no service dependencies.
/// Isolated here so it can be unit-tested without DI infrastructure.
/// Pure NDJSON parsing logic — no service dependencies.
/// Gzip transport compression is handled upstream by ASP.NET Core's UseRequestDecompression
/// middleware, so this parser sees an already-decompressed stream.
/// </summary>
internal static class NdjsonParser
{
/// <summary>
/// Decompresses a gzip stream and parses NDJSON content.
/// Reads an NDJSON stream with a size cap and parses it.
/// First line: metadata (sessionId, tenantId). Subsequent lines: EnrollmentEvent objects.
/// </summary>
internal static async Task<(string sessionId, string tenantId, List<EnrollmentEvent> events)>
ParseGzipAsync(Stream body, int maxSizeBytes)
ParseNdjsonStreamAsync(Stream body, int maxSizeBytes)
{
using var decompressed = new MemoryStream();
using (var gzip = new GZipStream(body, CompressionMode.Decompress, leaveOpen: true))
{
var buffer = new byte[8192];
int bytesRead;
long totalBytesRead = 0;
using var buffered = new MemoryStream();
var buffer = new byte[8192];
int bytesRead;
long totalBytesRead = 0;
while ((bytesRead = await gzip.ReadAsync(buffer, 0, buffer.Length)) > 0)
{
totalBytesRead += bytesRead;
if (totalBytesRead > maxSizeBytes)
throw new InvalidOperationException(
$"NDJSON payload size exceeds maximum allowed size. " +
$"Current size: {totalBytesRead / 1024.0 / 1024.0:F2} MB");
await decompressed.WriteAsync(buffer, 0, bytesRead);
}
while ((bytesRead = await body.ReadAsync(buffer, 0, buffer.Length)) > 0)
{
totalBytesRead += bytesRead;
if (totalBytesRead > maxSizeBytes)
throw new InvalidOperationException(
$"NDJSON payload size exceeds maximum allowed size. " +
$"Current size: {totalBytesRead / 1024.0 / 1024.0:F2} MB");
await buffered.WriteAsync(buffer, 0, bytesRead);
}
decompressed.Position = 0;
var ndjson = await new StreamReader(decompressed, Encoding.UTF8).ReadToEndAsync();
buffered.Position = 0;
var ndjson = await new StreamReader(buffered, Encoding.UTF8).ReadToEndAsync();
return ParseNdjson(ndjson);
}
@@ -80,25 +77,6 @@ namespace AutopilotMonitor.Functions.Functions.Ingest
return (metadata.SessionId, metadata.TenantId, events);
}
/// <summary>
/// Builds a gzip-compressed NDJSON payload for testing.
/// </summary>
internal static byte[] BuildGzipPayload(string sessionId, string tenantId, IEnumerable<object> events)
{
var sb = new StringBuilder();
sb.AppendLine(JsonConvert.SerializeObject(new { SessionId = sessionId, TenantId = tenantId }));
foreach (var evt in events)
sb.AppendLine(JsonConvert.SerializeObject(evt));
using var output = new MemoryStream();
using (var gzip = new GZipStream(output, CompressionMode.Compress, leaveOpen: true))
{
var bytes = Encoding.UTF8.GetBytes(sb.ToString());
gzip.Write(bytes, 0, bytes.Length);
}
return output.ToArray();
}
internal static void NormalizeEventData(EnrollmentEvent evt)
{
if (evt.Data == null || evt.Data.Count == 0) return;
@@ -0,0 +1,22 @@
using Microsoft.AspNetCore.Builder;
using Microsoft.AspNetCore.Hosting;
namespace AutopilotMonitor.Functions.Middleware
{
// Registers ASP.NET Core response-compression and request-decompression middleware into the
// Kestrel pipeline that the Azure Functions isolated worker hosts via
// Microsoft.Azure.Functions.Worker.Extensions.Http.AspNetCore. Using the standard middleware
// keeps gzip handling out of function code and NdjsonParser.
internal sealed class HttpCompressionStartupFilter : IStartupFilter
{
public Action<IApplicationBuilder> Configure(Action<IApplicationBuilder> next)
{
return builder =>
{
builder.UseResponseCompression();
builder.UseRequestDecompression();
next(builder);
};
}
}
}
@@ -1,3 +1,4 @@
using System.IO.Compression;
using System.Text.Json;
using System.Text.Json.Serialization;
using AutopilotMonitor.Functions.DataAccess;
@@ -8,6 +9,9 @@ using AutopilotMonitor.Functions.Helpers;
using AutopilotMonitor.Functions.Middleware;
using AutopilotMonitor.Functions.Services;
using Microsoft.AspNetCore.Authentication.JwtBearer;
using Microsoft.AspNetCore.Builder;
using Microsoft.AspNetCore.Hosting;
using Microsoft.AspNetCore.ResponseCompression;
using Microsoft.Azure.Functions.Worker;
using Microsoft.Azure.Functions.Worker.Builder;
using Microsoft.Azure.Functions.Worker.Middleware;
@@ -100,6 +104,22 @@ builder.Services.AddAuthorization();
// Enable ASP.NET Core integration for authentication
builder.Services.AddHttpContextAccessor();
// HTTP compression — bidirectional gzip for bandwidth-sensitive agent links.
// UseResponseCompression: backend → agent (config, ingest ack, ...). Triggered by Accept-Encoding.
// UseRequestDecompression: agent → backend (ingest NDJSON). Triggered by Content-Encoding.
// Registered via IStartupFilter because FunctionsApplication.CreateBuilder's Build() returns IHost,
// not WebApplication — we can't call app.UseXxx() directly.
builder.Services.AddResponseCompression(options =>
{
options.EnableForHttps = true;
options.Providers.Add<GzipCompressionProvider>();
options.MimeTypes = ResponseCompressionDefaults.MimeTypes
.Concat(new[] { "application/json", "application/x-ndjson" });
});
builder.Services.Configure<GzipCompressionProviderOptions>(o => o.Level = CompressionLevel.Fastest);
builder.Services.AddRequestDecompression();
builder.Services.AddTransient<IStartupFilter, HttpCompressionStartupFilter>();
// Register our services
builder.Services.AddMemoryCache();
builder.Services.AddHttpClient();