Add V3 streaming event ingestion pipeline

.gitignore

@@ -87,3 +87,5 @@ docs/_cache/
 .devcontainer/devcontainer-lock.json
 
 *.lscache
+# BenchmarkDotNet output
+BenchmarkDotNet.Artifacts/

Exceptionless.slnx

@@ -23,6 +23,8 @@
     <File Path="tests/http/users.http" />
     <File Path="tests/http/webhooks.http" />
   </Folder>
+  <Project Path="benchmarks/Exceptionless.Benchmarks/Exceptionless.Benchmarks.csproj" />
+  <Project Path="benchmarks/Exceptionless.Ingestion.Load/Exceptionless.Ingestion.Load.csproj" />
   <Project Path="src/Exceptionless.AppHost/Exceptionless.AppHost.csproj" />
   <Project Path="src/Exceptionless.Core/Exceptionless.Core.csproj" />
   <Project Path="src/Exceptionless.Insulation/Exceptionless.Insulation.csproj" />

benchmarks/Directory.Build.props

@@ -0,0 +1,7 @@
+<Project>
+  <Import Project="..\src\Directory.Build.props" />
+
+  <PropertyGroup>
+    <IsPackable>false</IsPackable>
+  </PropertyGroup>
+</Project>

benchmarks/Exceptionless.Benchmarks/Exceptionless.Benchmarks.csproj

@@ -0,0 +1,19 @@
+<Project Sdk="Microsoft.NET.Sdk">
+  <PropertyGroup>
+    <OutputType>Exe</OutputType>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <PackageReference Include="BenchmarkDotNet" Version="0.15.8" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\..\src\Exceptionless.Core\Exceptionless.Core.csproj" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <!-- Compile the production reader source so the microbenchmark cannot drift to a substitute parser. -->
+    <Compile Include="..\..\src\Exceptionless.Web\Utility\EventIngestionV3StreamReader.cs"
+      Link="Production\EventIngestionV3StreamReader.cs" />
+  </ItemGroup>
+</Project>

benchmarks/Exceptionless.Benchmarks/Processing/EventIngestionProcessingBenchmarks.cs

@@ -0,0 +1,72 @@
+using BenchmarkDotNet.Attributes;
+using Exceptionless.Core.Models;
+using Exceptionless.Core.Models.Ingestion;
+using Exceptionless.Core.Services;
+
+namespace Exceptionless.Benchmarks.Processing;
+
+[MemoryDiagnoser]
+public class EventIngestionProcessingBenchmarks
+{
+    private const string StackTrace = """
+        at System.Threading.Tasks.Task.Run()
+        at Example.Services.OrderService.Save() in /src/OrderService.cs:line 42
+        at Example.Api.OrdersController.Post() in /src/OrdersController.cs:line 18
+        """;
+
+    private readonly StackTraceParser _parser = new();
+    private StackFingerprintService _fingerprintService = null!;
+    private EventMaterializer _materializer = null!;
+    private EventIngestionV3Event _source = null!;
+    private Organization _organization = null!;
+    private Project _project = null!;
+    private StackFingerprint _fingerprint = null!;
+
+    [GlobalSetup]
+    public void Setup()
+    {
+        _fingerprintService = new StackFingerprintService(_parser);
+        _materializer = new EventMaterializer(_parser, TimeProvider.System);
+        _organization = new Organization { Id = "507f191e810c19729de860ea", Name = "Benchmark" };
+        _project = new Project
+        {
+            Id = "507f191e810c19729de860eb",
+            OrganizationId = _organization.Id,
+            Name = "Benchmark"
+        };
+        _source = new EventIngestionV3Event
+        {
+            Id = "benchmark-event-1",
+            Type = Event.KnownTypes.Error,
+            Message = "Operation failed",
+            ExceptionType = "System.InvalidOperationException",
+            StackTrace = StackTrace,
+            Tags = ["benchmark", "v3"]
+        };
+        _fingerprint = _fingerprintService.Create(_source, _organization, _project);
+    }
+
+    [Benchmark]
+    public bool FingerprintRawStack()
+    {
+        return _parser.TryFindFrame(StackTrace, static frame => frame.DeclaringNamespace?.StartsWith("Example") is true, out _, out _, out _);
+    }
+
+    [Benchmark]
+    public StackFingerprint CreateFingerprint()
+    {
+        return _fingerprintService.Create(_source, _organization, _project);
+    }
+
+    [Benchmark]
+    public int ParseStructuredFrames()
+    {
+        return _parser.Parse(StackTrace).Count;
+    }
+
+    [Benchmark]
+    public PersistentEvent MaterializeSurvivor()
+    {
+        return _materializer.Materialize(_source, _fingerprint, _organization, _project);
+    }
+}

benchmarks/Exceptionless.Benchmarks/Program.cs

@@ -0,0 +1,11 @@
+using BenchmarkDotNet.Running;
+
+namespace Exceptionless.Benchmarks;
+
+public static class Program
+{
+    public static void Main(string[] args)
+    {
+        BenchmarkSwitcher.FromAssembly(typeof(Program).Assembly).Run(args);
+    }
+}

benchmarks/Exceptionless.Benchmarks/Serialization/EventIngestionDeserializationBenchmarks.cs

@@ -0,0 +1,222 @@
+using System.IO.Pipelines;
+using System.Text;
+using System.Text.Json;
+using System.Text.Json.Serialization;
+using BenchmarkDotNet.Attributes;
+using Exceptionless.Core.Extensions;
+using Exceptionless.Core.Models;
+using Exceptionless.Core.Models.Ingestion;
+using Exceptionless.Core.Serialization;
+using Exceptionless.Web.Utility;
+
+namespace Exceptionless.Benchmarks.Serialization;
+
+[MemoryDiagnoser]
+public class EventIngestionDeserializationBenchmarks
+{
+    private const int MaximumEventSize = 512 * 1024;
+    private byte[] _v2Payload = null!;
+    private byte[] _v3StreamPayload = null!;
+    private JsonSerializerOptions _v2JsonOptions = null!;
+
+    [Params(1, 100, 1000)]
+    public int EventCount { get; set; }
+
+    [GlobalSetup]
+    public void Setup()
+    {
+        _v2JsonOptions = new JsonSerializerOptions().ConfigureExceptionlessDefaults();
+        _v2JsonOptions.RespectNullableAnnotations = false;
+        var v2Events = new V2BenchmarkEvent[EventCount];
+        var v3Events = new EventIngestionV3Event[EventCount];
+        for (int index = 0; index < v3Events.Length; index++)
+        {
+            const string message = "Operation failed";
+            const string exceptionType = "System.InvalidOperationException";
+            const string stackTrace = "   at Example.Service.Run() in Service.cs:line 42";
+            v2Events[index] = new V2BenchmarkEvent(
+                Event.KnownTypes.Error,
+                DateTimeOffset.UnixEpoch.AddSeconds(index),
+                message,
+                new V2BenchmarkData(new V2BenchmarkError(exceptionType, message, stackTrace)));
+            v3Events[index] = new EventIngestionV3Event
+            {
+                Id = $"01J0000000000000000000{index:D4}",
+                Type = "error",
+                Date = DateTimeOffset.UnixEpoch.AddSeconds(index),
+                Message = message,
+                ExceptionType = exceptionType,
+                StackTrace = stackTrace
+            };
+        }
+
+        _v2Payload = EventCount == 1
+            ? JsonSerializer.SerializeToUtf8Bytes(v2Events[0])
+            : JsonSerializer.SerializeToUtf8Bytes(v2Events);
+
+        using var stream = new MemoryStream(_v2Payload.Length);
+        for (int index = 0; index < v3Events.Length; index++)
+        {
+            JsonSerializer.Serialize(
+                stream,
+                v3Events[index],
+                EventIngestionJsonContext.Default.EventIngestionV3Event);
+            stream.WriteByte((byte)'\n');
+        }
+
+        _v3StreamPayload = stream.ToArray();
+    }
+
+    [Benchmark(Baseline = true)]
+    public int DeserializeV2Payload()
+    {
+        string input = Encoding.UTF8.GetString(_v2Payload);
+        return input.GetJsonType() switch
+        {
+            JsonType.Object => JsonSerializer.Deserialize<PersistentEvent>(input, _v2JsonOptions) is null ? 0 : 1,
+            JsonType.Array => JsonSerializer.Deserialize<PersistentEvent[]>(input, _v2JsonOptions)?.Length ?? 0,
+            _ => 0
+        };
+    }
+
+    [Benchmark]
+    public Task<int> FrameAndRouteV3NdjsonAsync() => ReadV3NdjsonAsync(materializeSurvivors: false);
+
+    [Benchmark]
+    public Task<int> FrameRouteAndMaterializeV3SurvivorsAsync() => ReadV3NdjsonAsync(materializeSurvivors: true);
+
+    private async Task<int> ReadV3NdjsonAsync(bool materializeSurvivors)
+    {
+        using var stream = new MemoryStream(_v3StreamPayload, writable: false);
+        var reader = PipeReader.Create(stream);
+        int count = 0;
+        try
+        {
+            while (await EventIngestionV3StreamReader.ReadAsync(reader, MaximumEventSize, CancellationToken.None) is { } record)
+            {
+                try
+                {
+                    EventIngestionV3Event parsed = materializeSurvivors
+                        ? record.BufferedRecord.Materialize()
+                        : record.Event;
+                    if (!String.IsNullOrEmpty(parsed.Id))
+                        count++;
+                }
+                finally
+                {
+                    record.BufferedRecord.Dispose();
+                }
+            }
+        }
+        finally
+        {
+            await reader.CompleteAsync();
+        }
+
+        return count;
+    }
+
+    private sealed record V2BenchmarkEvent(
+        [property: JsonPropertyName("type")] string Type,
+        [property: JsonPropertyName("date")] DateTimeOffset Date,
+        [property: JsonPropertyName("message")] string Message,
+        [property: JsonPropertyName("data")] V2BenchmarkData Data);
+
+    private sealed record V2BenchmarkData(
+        [property: JsonPropertyName("@simple_error")] V2BenchmarkError SimpleError);
+
+    private sealed record V2BenchmarkError(
+        [property: JsonPropertyName("type")] string Type,
+        [property: JsonPropertyName("message")] string Message,
+        [property: JsonPropertyName("stack_trace")] string StackTrace);
+}
+
+/// <summary>
+/// Keeps large raw stacks visible in allocation results. One accepted error is enough to expose
+/// duplicate LOH strings without multiplying the benchmark process's retained payload by a batch.
+/// </summary>
+[MemoryDiagnoser]
+public class LargeStackEventIngestionDeserializationBenchmarks
+{
+    private const int MaximumEventSize = 512 * 1024;
+    private byte[] _v2Payload = null!;
+    private byte[] _v3Payload = null!;
+    private JsonSerializerOptions _v2JsonOptions = null!;
+
+    [Params(16 * 1024, 128 * 1024)]
+    public int StackTraceLength { get; set; }
+
+    [GlobalSetup]
+    public void Setup()
+    {
+        string stackTrace = new('x', StackTraceLength);
+        _v2JsonOptions = new JsonSerializerOptions().ConfigureExceptionlessDefaults();
+        _v2JsonOptions.RespectNullableAnnotations = false;
+        _v2Payload = JsonSerializer.SerializeToUtf8Bytes(new Dictionary<string, object?>
+        {
+            ["type"] = "error",
+            ["message"] = "Operation failed",
+            ["data"] = new Dictionary<string, object?>
+            {
+                ["@simple_error"] = new
+                {
+                    type = "Example.Exception",
+                    message = "Operation failed",
+                    stack_trace = stackTrace
+                }
+            }
+        });
+        _v3Payload = JsonSerializer.SerializeToUtf8Bytes(new
+        {
+            id = "large-stack-event",
+            type = "error",
+            message = "Operation failed",
+            exception_type = "Example.Exception",
+            stack_trace = stackTrace
+        });
+    }
+
+    [Benchmark(Baseline = true)]
+    public PersistentEvent? DeserializeV2Payload()
+    {
+        string input = Encoding.UTF8.GetString(_v2Payload);
+        return JsonSerializer.Deserialize<PersistentEvent>(input, _v2JsonOptions);
+    }
+
+    [Benchmark]
+    public Task<int> FrameAndRouteV3NdjsonAsync() => ReadV3NdjsonAsync(materializeSurvivor: false);
+
+    [Benchmark]
+    public Task<int> FrameRouteAndMaterializeV3SurvivorAsync() => ReadV3NdjsonAsync(materializeSurvivor: true);
+
+    private async Task<int> ReadV3NdjsonAsync(bool materializeSurvivor)
+    {
+        using var stream = new MemoryStream(_v3Payload, writable: false);
+        var reader = PipeReader.Create(stream);
+        try
+        {
+            EventIngestionV3StreamRecord? record = await EventIngestionV3StreamReader.ReadAsync(
+                reader,
+                MaximumEventSize,
+                CancellationToken.None);
+            if (record is null)
+                return 0;
+
+            try
+            {
+                EventIngestionV3Event parsed = materializeSurvivor
+                    ? record.Value.BufferedRecord.Materialize()
+                    : record.Value.Event;
+                return parsed.StackTrace?.Length ?? 0;
+            }
+            finally
+            {
+                record.Value.BufferedRecord.Dispose();
+            }
+        }
+        finally
+        {
+            await reader.CompleteAsync();
+        }
+    }
+}

benchmarks/Exceptionless.Ingestion.Load/Exceptionless.Ingestion.Load.csproj

@@ -0,0 +1,13 @@
+<Project Sdk="Microsoft.NET.Sdk">
+  <PropertyGroup>
+    <OutputType>Exe</OutputType>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <InternalsVisibleTo Include="Exceptionless.Tests" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\..\src\Exceptionless.Core\Exceptionless.Core.csproj" />
+  </ItemGroup>
+</Project>