Separate ingestion, query and combined benchmarks

Author: liammclennan

src/SeqCli/Cli/Commands/Bench/BenchCasesCollection.cs

@@ -22,5 +22,5 @@ namespace SeqCli.Cli.Commands.Bench;
 class BenchCasesCollection
 {
     // ReSharper disable once CollectionNeverUpdated.Global
-    public IList<BenchCase> Cases { get; } = new List<BenchCase>();
+    public IList<QueryBenchCase> Cases { get; } = new List<QueryBenchCase>();
 }

src/SeqCli/Cli/Commands/Bench/BenchCommand.cs

@@ -20,6 +20,7 @@
 using System.Threading.Tasks;
 using Newtonsoft.Json;
 using Seq.Api;
+using Seq.Api.Model.Data;
 using Seq.Api.Model.Signals;
 using SeqCli.Cli.Features;
 using SeqCli.Connection;
@@ -73,7 +74,7 @@ class BenchCommand : Command
     string _reportingServerApiKey = "";
     string _description = "";
     bool _withIngestion = false;
-    bool _withQueries = true;
+    bool _withQueries = false;
 
     public BenchCommand(SeqConnectionFactory connectionFactory)
     {
@@ -114,34 +115,74 @@ protected override async Task<int> Run()
     {
         try
         {
-            var (url, apiKey) = _connectionFactory.GetConnectionDetails(_connection);
+            var (_, apiKey) = _connectionFactory.GetConnectionDetails(_connection);
             var connection = _connectionFactory.Connect(_connection);
             var seqVersion = (await connection.Client.GetRootAsync()).Version;
             await using var reportingLogger = BuildReportingLogger();
 
             var runId = Guid.NewGuid().ToString("N")[..16];
             CancellationTokenSource cancellationTokenSource = new ();
             var cancellationToken = cancellationTokenSource.Token;
-
-            if (_withIngestion)
+            
+            using (LogContext.PushProperty("RunId", runId))
+            using (LogContext.PushProperty("WithIngestion", _withIngestion))
+            using (LogContext.PushProperty("WithQueries", _withQueries))
+            using (LogContext.PushProperty("Start", _range.Start))
+            using (LogContext.PushProperty("End", _range.End))
             {
-                var t = Task.WhenAll(IngestionBenchmark(reportingLogger, runId, connection, apiKey, seqVersion, cancellationToken))
-                    .ContinueWith(async t =>
+                if (_withIngestion)
+                {
+                    var t = IngestionBenchmark(reportingLogger, runId, connection, apiKey, seqVersion,
+                        isQueryBench: _withQueries, cancellationToken)
+                        .ContinueWith(async t =>
+                        {
+                            if (t.Exception is not null)
+                            {
+                                await Console.Error.WriteLineAsync(t.Exception.Message);
+                            }
+                        });
+
+                    if (!_withQueries)
                     {
-                        if (t.Exception is not null)
+                        int benchDurationMs = 120_000;
+                        await Task.Delay(benchDurationMs);
+                        cancellationTokenSource.Cancel();
+                                
+                        var response = await connection.Data.QueryAsync(
+                            "select count(*) from stream group by time(1s)",
+                            DateTime.Now.Add(-1 * TimeSpan.FromMilliseconds(benchDurationMs))
+                        );
+                                
+                        if (response.Slices == null)
                         {
-                            await Console.Error.WriteLineAsync(t.Exception.Message);
+                            throw new Exception("Failed to query ingestion benchmark results");
                         }
-                    }, TaskContinuationOptions.OnlyOnFaulted);
-                await Task.Delay(10000); // how long to ingest before beginning queries
-            }
+                                
+                        var counts = response.Slices.Select(s => Convert.ToDouble(s.Rows[0][0])).Where(c => c > 1000).ToArray();
+                        var countsMean = counts.Sum() / counts.Length;
+                        var countsRSD = QueryBenchCaseTimings.StandardDeviation(counts) / countsMean;
+                                
+                        using (LogContext.PushProperty("RunId", runId))
+                        using (LogContext.PushProperty("EventsPerSecond", counts))
+                        {
+                            reportingLogger.Information(
+                                "Ingestion benchmark {Description} ran for {RunDuration:N0}ms; ingested {TotalIngested:N0} at {EventsPerMinute:N0}events/min; with RSD {RelativeStandardDeviationPercentage,4:N1}%",
+                                _description,
+                                benchDurationMs,
+                                counts.Sum(),
+                                countsMean * 60,
+                                countsRSD * 100);
+                        }
+                    }
+                }
 
-            if (_withQueries)
-            {
-                var collectedTimings = await QueryBenchmark(reportingLogger, runId, connection, seqVersion);
-                collectedTimings.LogSummary();
+                if (_withQueries)
+                {
+                    var collectedTimings = await QueryBenchmark(reportingLogger, runId, connection, seqVersion);
+                    collectedTimings.LogSummary(_description);
+                    cancellationTokenSource.Cancel();
+                }
             }
-            cancellationTokenSource.Cancel();
 
             return 0;
         } 
@@ -152,69 +193,85 @@ protected override async Task<int> Run()
         }
     }
 
-    Task[] IngestionBenchmark(Logger reportingLogger, string runId, SeqConnection connection, string? apiKey, string seqVersion, CancellationToken cancellationToken = default)
+    async Task IngestionBenchmark(Logger reportingLogger, string runId, SeqConnection connection, string? apiKey, 
+        string seqVersion, bool isQueryBench, CancellationToken cancellationToken = default)
     {
-        return Enumerable.Range(1, 1000)
-            .Select(i => Simulation.RunAsync(connection, apiKey, 10000, echoToStdout: false, cancellationToken))
-            .ToArray();
+        using (!string.IsNullOrWhiteSpace(_description)
+                   ? LogContext.PushProperty("Description", _description)
+                   : null)
+        {
+            reportingLogger.Information(
+                "Ingestion bench run {RunId} against {ServerUrl} ({SeqVersion})",
+                runId, connection.Client.ServerUrl, seqVersion);
+        }
+
+        if (isQueryBench)
+        {
+            var simulationTasks = Enumerable.Range(1, 500)
+                .Select(i => Simulation.RunAsync(connection, apiKey, 10000, echoToStdout: false, cancellationToken))
+                .ToArray();
+            await Task.Delay(20_000); // how long to ingest before beginning queries
+        }
+        else
+        {
+            var simulationTasks = Enumerable.Range(1, 2000)
+                .Select(i => Simulation.RunAsync(connection, apiKey, 10000, echoToStdout: false, cancellationToken))
+                .ToArray();
+        }
     }
 
     async Task<CollectedTimings> QueryBenchmark(Logger reportingLogger, string runId, SeqConnection connection, string seqVersion)
     {
         var cases = ReadCases(_cases);
-        CollectedTimings collectedTimings = new();
+        CollectedTimings collectedTimings = new(reportingLogger);
         using (!string.IsNullOrWhiteSpace(_description)
                    ? LogContext.PushProperty("Description", _description)
                    : null)
         {
             reportingLogger.Information(
-                "Bench run {RunId} against {ServerUrl} ({SeqVersion}); {CaseCount} cases, {Runs} runs, from {Start} to {End}",
+                "Query bench run {RunId} against {ServerUrl} ({SeqVersion}); {CaseCount} cases, {Runs} runs, from {Start} to {End}",
                 runId, connection.Client.ServerUrl, seqVersion, cases.Cases.Count, _runs, _range.Start, _range.End);
         }
 
-        using (LogContext.PushProperty("RunId", runId))
-        using (LogContext.PushProperty("Start", _range.Start))
-        using (LogContext.PushProperty("End", _range.End))
+        foreach (var c in cases.Cases.OrderBy(c => c.Id)
+                     .Concat(new [] { CollectedTimings.FINAL_COUNT_CASE }))
         {
-            foreach (var c in cases.Cases.OrderBy(c => c.Id)
-                         .Concat(new [] { CollectedTimings.FINAL_COUNT_CASE }))
+            var timings = new QueryBenchCaseTimings(c);
+            collectedTimings.Add(timings);
+
+            foreach (var i in Enumerable.Range(1, _runs))
             {
-                var timings = new BenchCaseTimings(c);
-                collectedTimings.Add(timings);
-                object? lastResult = null;
+                var response = await connection.Data.QueryAsync(
+                    c.Query,
+                    _range.Start,
+                    _range.End,
+                    c.SignalExpression != null ? SignalExpressionPart.Signal(c.SignalExpression) : null,
+                    null,
+                    TimeSpan.FromMinutes(4)
+                );
+        
+                timings.PushElapsed(response.Statistics.ElapsedMilliseconds);
 
-                foreach (var i in Enumerable.Range(1, _runs))
+                if (response.Rows != null)
                 {
-                    var response = await connection.Data.QueryAsync(
-                        c.Query,
-                        _range.Start,
-                        _range.End,
-                        c.SignalExpression != null ? SignalExpressionPart.Signal(c.SignalExpression) : null
-                    );
-
-                    timings.PushElapsed(response.Statistics.ElapsedMilliseconds);
-
-                    if (response.Rows != null)
+                    var isScalarResult = response.Rows.Length == 1 && response.Rows[0].Length == 1;
+                    if (isScalarResult && i == _runs)
                     {
-                        var isScalarResult = response.Rows.Length == 1 && response.Rows[0].Length == 1;
-                        if (isScalarResult && i == _runs)
-                        {
-                            lastResult = response.Rows[0][0];
-                        }
+                        timings.LastResult = response.Rows[0][0];
                     }
                 }
+            }
 
-                using (lastResult != null ? LogContext.PushProperty("LastResult", lastResult) : null)
-                using (!string.IsNullOrWhiteSpace(c.SignalExpression)
-                           ? LogContext.PushProperty("SignalExpression", c.SignalExpression)
-                           : null)
-                using (LogContext.PushProperty("StandardDeviationElapsed", timings.StandardDeviationElapsed))
-                using (LogContext.PushProperty("Query", c.Query))
-                {
-                    reportingLogger.Information(
-                        "Case {Id,-40} ({LastResult}) mean {MeanElapsed,5:N0} ms (first {FirstElapsed,5:N0} ms, min {MinElapsed,5:N0} ms, max {MaxElapsed,5:N0} ms, RSD {RelativeStandardDeviationElapsed,4:N2})",
-                        c.Id, lastResult, timings.MeanElapsed, timings.FirstElapsed, timings.MinElapsed, timings.MaxElapsed, timings.RelativeStandardDeviationElapsed);
-                }
+            using (timings.LastResult != null ? LogContext.PushProperty("LastResult", timings.LastResult) : null)
+            using (!string.IsNullOrWhiteSpace(c.SignalExpression)
+                       ? LogContext.PushProperty("SignalExpression", c.SignalExpression)
+                       : null)
+            using (LogContext.PushProperty("StandardDeviationElapsed", timings.StandardDeviationElapsed))
+            using (LogContext.PushProperty("Query", c.Query))
+            {
+                reportingLogger.Information(
+                    "Case {Id,-40} ({LastResult}) mean {MeanElapsed,5:N0} ms (first {FirstElapsed,5:N0} ms, min {MinElapsed,5:N0} ms, max {MaxElapsed,5:N0} ms, RSD {RelativeStandardDeviationElapsed,4:N2})",
+                    c.Id, timings.LastResult, timings.MeanElapsed, timings.FirstElapsed, timings.MinElapsed, timings.MaxElapsed, timings.RelativeStandardDeviationElapsed);
             }
         }
 

…c/SeqCli/Cli/Commands/Bench/BenchCase.cs …Cli/Cli/Commands/Bench/QueryBenchCase.cssrc/SeqCli/Cli/Commands/Bench/BenchCase.cs renamed to src/SeqCli/Cli/Commands/Bench/QueryBenchCase.cs src/SeqCli/Cli/Commands/Bench/BenchCase.cs renamed to src/SeqCli/Cli/Commands/Bench/QueryBenchCase.cs

@@ -16,7 +16,7 @@ namespace SeqCli.Cli.Commands.Bench;
 
 // ReSharper disable ClassNeverInstantiated.Global AutoPropertyCanBeMadeGetOnly.Global UnusedAutoPropertyAccessor.Global
 
-class BenchCase
+class QueryBenchCase
 {
     public string Id { get; set; } = null!;
     public string Query { get; set; } = null!;

…i/Cli/Commands/Bench/BenchCaseTimings.cs …/Commands/Bench/QueryBenchCaseTimings.cssrc/SeqCli/Cli/Commands/Bench/BenchCaseTimings.cs renamed to src/SeqCli/Cli/Commands/Bench/QueryBenchCaseTimings.cs src/SeqCli/Cli/Commands/Bench/BenchCaseTimings.cs renamed to src/SeqCli/Cli/Commands/Bench/QueryBenchCaseTimings.cs

@@ -15,56 +15,92 @@
 using System;
 using System.Collections.Generic;
 using System.Linq;
+using Serilog.Core;
 
 namespace SeqCli.Cli.Commands.Bench;
 
 class CollectedTimings
 {
-    List<BenchCaseTimings> _collectedTimings = new();
+    readonly Logger _reportingLogger;
+    List<QueryBenchCaseTimings> _collectedTimings = new();
 
-    public static BenchCase FINAL_COUNT_CASE = new BenchCase()
+    public static QueryBenchCase FINAL_COUNT_CASE = new QueryBenchCase()
     {
         Id = "final-count-star",
         Query = "select count(*) from stream",
     };
 
-    public void Add(BenchCaseTimings caseTimings)
+    public CollectedTimings(Logger reportingLogger)
+    {
+        _reportingLogger = reportingLogger;
+    }
+
+    public void Add(QueryBenchCaseTimings caseTimings)
     {
         _collectedTimings.Add(caseTimings);
     }
 
-    public void LogSummary()
+    public void LogSummary(string description)
+    {
+        _reportingLogger.Information(
+            "Query benchmark {Description} complete in {TotalMeanElapsed:N0} ms with {MeanRelativeStandardDeviationPercentage:N1}% deviation, processed {FinalEventCount:N0} events at {EventsPerMs:N0} events/ms", 
+            description,
+            TotalMeanElapsed(),
+            MeanRelativeStandardDeviationPercentage(),
+            FinalEventCount(),
+            FinalEventCount() * _collectedTimings.Count / TotalMeanElapsed());
+    }
+
+    private double TotalMeanElapsed()
+    {
+        return _collectedTimings.Sum(c => c.MeanElapsed);
+    }
+
+    private double MeanRelativeStandardDeviationPercentage()
     {
-        throw new NotImplementedException();
+        return _collectedTimings.Average(c => c.RelativeStandardDeviationElapsed) * 100;
+    }
+
+    private int FinalEventCount()
+    {
+        var benchCase = _collectedTimings.Single(c => c.Id == FINAL_COUNT_CASE.Id);
+        return Convert.ToInt32(benchCase.LastResult);
     }
 }
 
 /*
  * Collects benchmarking elapsed time measurements and calculates statistics. 
  */
-class BenchCaseTimings
+class QueryBenchCaseTimings
 {
-    readonly BenchCase _benchCase;
+    readonly QueryBenchCase _queryBenchCase;
     readonly List<double> _timings = new();
-    
+    object? _lastResult;
+
     public double MeanElapsed => _timings.Sum() / _timings.Count;
     public double MinElapsed => _timings.Min();
     public double MaxElapsed => _timings.Max();
     public double FirstElapsed => _timings.First();
     public double StandardDeviationElapsed => StandardDeviation(_timings); 
     public double RelativeStandardDeviationElapsed => StandardDeviation(_timings) / MeanElapsed;
 
-    public BenchCaseTimings(BenchCase benchCase)
+    public object? LastResult
     {
-        _benchCase = benchCase;
+        get => _lastResult;
+        set => _lastResult = value;
+    }
+
+    public QueryBenchCaseTimings(QueryBenchCase queryBenchCase)
+    {
+        _queryBenchCase = queryBenchCase;
     }
 
     public void PushElapsed(double elapsed)
     {
         _timings.Add(elapsed);
     }
 
-    static double StandardDeviation(ICollection<double> population)
+    public static double StandardDeviation(ICollection<double> population)
     {
         if (population.Count < 2)
         {
@@ -74,4 +110,6 @@ static double StandardDeviation(ICollection<double> population)
         var mean = population.Sum() / population.Count;
         return Math.Sqrt(population.Select(e => Math.Pow(e - mean, 2)).Sum() / (population.Count - 1));
     }
+
+    public string Id => _queryBenchCase.Id;
 }