Merge pull request #440 from KodrAus/feat/sample-metrics

Initial metric support in `sample ingest`

Author: KodrAus

src/Roastery/Api/OrdersController.cs

@@ -2,6 +2,7 @@
 using System.Net;
 using System.Threading.Tasks;
 using Roastery.Data;
+using Roastery.Metrics;
 using Roastery.Model;
 using Roastery.Web;
 using Serilog;
@@ -15,8 +16,8 @@ class OrdersController : Controller
 {
     readonly Database _database;
 
-    public OrdersController(ILogger logger, Database database)
-        : base(logger)
+    public OrdersController(ILogger logger, RoasteryMetrics metrics, Database database)
+        : base(logger, metrics)
     {
         _database = database;
     }
@@ -41,7 +42,10 @@ public async Task<HttpResponse> Create(HttpRequest request)
         }
 
         await _database.InsertAsync(order);
+        
+        Metrics.RecordOrderCreated();
         Log.Information("Created new order {OrderId} for customer {CustomerName}", order.Id, order.CustomerName);
+        
         return Json(order, HttpStatusCode.Created);
     }
 
@@ -65,7 +69,11 @@ public async Task<HttpResponse> Update(HttpRequest request)
         if (order.Status == OrderStatus.PendingShipment)
             Log.Information("Order placed and ready for shipment");
         else if (order.Status == OrderStatus.Shipped)
-            Log.Information("Order shipped to {CustomerName} at {ShippingAddress}", order.CustomerName, order.ShippingAddress);
+        {
+            Metrics.RecordOrderShipped();
+            Log.Information("Order shipped to {CustomerName} at {ShippingAddress}", order.CustomerName,
+                order.ShippingAddress);
+        }
         else
             Log.Information("Order updated");
 

src/Roastery/Api/ProductsController.cs

@@ -1,5 +1,6 @@
 using System.Threading.Tasks;
 using Roastery.Data;
+using Roastery.Metrics;
 using Roastery.Model;
 using Roastery.Web;
 using Serilog;
@@ -12,8 +13,8 @@ class ProductsController : Controller
 {
     readonly Database _database;
 
-    public ProductsController(ILogger logger, Database database)
-        : base(logger)
+    public ProductsController(ILogger logger, RoasteryMetrics metrics, Database database)
+        : base(logger, metrics)
     {
         _database = database;
     }

src/Roastery/Metrics/ExponentialHistogram.cs

@@ -0,0 +1,65 @@
+using System;
+using System.Collections.Generic;
+
+namespace Roastery.Metrics;
+
+public class ExponentialHistogram
+{
+    public ExponentialHistogram(int initialScale = 20, int targetBuckets = 160)
+    {
+        _scale = initialScale;
+        _targetBuckets = targetBuckets;
+        _buckets = new Dictionary<double, ulong>();
+    }
+    
+    readonly int _targetBuckets;
+
+    int _scale;
+    Dictionary<double, ulong> _buckets;
+
+    double _min;
+    double _max;
+    ulong _total;
+
+    public void Record(double rawValue)
+    {
+        _min = Math.Min(_min, rawValue);
+        _max = Math.Max(_max, rawValue);
+        _total += 1;
+        
+        var midpoint = Midpoint(_scale, rawValue);
+        _buckets.TryAdd(midpoint, 0);
+        _buckets[midpoint] += 1;
+
+        if (_buckets.Count <= _targetBuckets) return;
+        
+        // Rescale
+        var newScale = _scale - 1;
+        var newBuckets = new Dictionary<double, ulong>();
+
+        foreach (var (oldMidpoint, count) in _buckets)
+        {
+            var newMidpoint = Midpoint(_scale, oldMidpoint);
+            newBuckets.TryAdd(newMidpoint, 0);
+            newBuckets[newMidpoint] += count;
+        }
+
+        _buckets = newBuckets;
+        _scale = newScale;
+    }
+
+    static double Midpoint(int scale, double rawValue)
+    {
+        var gamma = Math.Pow(2d, Math.Pow(2d, -scale));
+        var index = Math.Abs(Math.Log(rawValue, gamma));
+        
+        return (Math.Pow(gamma, index - 1) + Math.Pow(gamma, index)) / 2;
+    }
+
+    public IReadOnlyDictionary<double, ulong> Buckets => _buckets;
+    public int Scale => _scale;
+    
+    public double Min => _min;
+    public double Max => _max;
+    public ulong Total => _total;
+}

src/Roastery/Metrics/PropertyNameMapping.cs

@@ -0,0 +1,3 @@
+namespace Roastery.Metrics;
+
+public record struct PropertyNameMapping(string MetricDefinitions, string MetricSamples);

src/Roastery/Metrics/RoasteryMetrics.cs

@@ -0,0 +1,174 @@
+using System;
+using System.Collections.Generic;
+using System.Diagnostics;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using Serilog;
+using Serilog.Events;
+using Serilog.Parsing;
+
+namespace Roastery.Metrics;
+
+public class RoasteryMetrics
+{
+    public class Sample
+    {
+        /*
+        Adding new metrics:
+
+        1. Add a new key type, `TKey` for the metric's attributes using structural equality.
+        2. Add a `Dictionary<TKey, TMetric>` property for the metric where `TMetric` is its collection type.
+        3. Add a method to `RoasterMetrics` to add a sample to the metric for a given key.
+        4. Add support in `ToLogEvents` for the new metric.
+        */
+        
+        // `HttpRequestDuration`: histogram
+        public record struct HttpRequestDurationKey(string Path, int StatusCode);
+        public readonly Dictionary<HttpRequestDurationKey, ExponentialHistogram> HttpRequestDuration = new();
+        
+        // `OrdersCreated`: counter
+        public ulong OrdersCreated;
+        
+        // `OrdersShipped`: counter
+        public ulong OrdersShipped;
+        
+        static readonly MessageTemplate Template = new MessageTemplateParser().Parse("Metrics sampled");
+
+        public IEnumerable<LogEvent> ToLogEvents(ILogger logger, PropertyNameMapping propertyNameMapping, DateTimeOffset timestamp)
+        {
+            foreach (var (key, metric) in HttpRequestDuration)
+            {
+                yield return ToLogEvent(
+                    logger,
+                    propertyNameMapping,
+                    timestamp,
+                    new Dictionary<string, object>
+                    {
+                        { "HttpRequestDuration", new { kind = "Exponential", unit = "ms", description = "The time taken to fully process a request" } }
+                    },
+                    new
+                    {
+                        HttpRequestDuration = new {
+                            buckets = metric.Buckets
+                                .Select(bucket => new { midpoint = bucket.Key, count = bucket.Value }).ToArray(),
+                            scale = metric.Scale,
+                            min = metric.Min,
+                            max = metric.Max,
+                            count = metric.Total
+                        },
+                        key.Path,
+                        key.StatusCode
+                    }
+                );
+            }
+            
+            yield return ToLogEvent(
+                logger,
+                propertyNameMapping,
+                timestamp,
+                new Dictionary<string, object>
+                {
+                    { "OrdersCreated", new { kind = "Counter", unit = "orders", description = "The total number of orders created in the system" } },
+                    { "OrdersShipped", new { kind = "Counter", unit = "orders", description = "The total number of orders shipped in the system" } }
+                },
+                new
+                {
+                    OrdersCreated,
+                    OrdersShipped
+                }
+            );
+        }
+
+        static LogEvent ToLogEvent(ILogger logger, PropertyNameMapping propertyNameMapping, DateTimeOffset timestamp, Dictionary<string, object> definitions, object samples)
+        {
+            logger.BindProperty(propertyNameMapping.MetricDefinitions, definitions, true, out var definitionsProperty);
+            logger.BindProperty(propertyNameMapping.MetricSamples, samples, true, out var sampleProperty);
+
+            return new LogEvent(timestamp, LogEventLevel.Information, null, Template,
+                [definitionsProperty!, sampleProperty!]);
+        }
+    }
+    
+    // Access to the current sample is synchronized through a lock
+    // This is a simple way to implement deltas for arbitrary types
+    readonly Lock _lock = new();
+    Sample _current = new();
+
+    public void RecordHttpRequestDuration(Sample.HttpRequestDurationKey key, double rawValue)
+    {
+        lock (_lock)
+        {
+            if (!_current.HttpRequestDuration.TryGetValue(key, out var metric))
+            {
+                metric = new ExponentialHistogram();
+                _current.HttpRequestDuration.Add(key, metric);
+            }
+
+            metric.Record(rawValue);
+        }
+    }
+
+    public void RecordOrderCreated()
+    {
+        lock (_lock)
+        {
+            _current.OrdersCreated += 1;
+        }
+    }
+
+    public void RecordOrderShipped()
+    {
+        lock (_lock)
+        {
+            _current.OrdersShipped += 1;
+        }
+    }
+
+    public (DateTimeOffset, Sample) Take()
+    {
+        var timestamp = DateTimeOffset.UtcNow;
+        
+        var current = new Sample();
+
+        lock (_lock)
+        {
+            (current, _current) = (_current, current);
+        }
+
+        return (timestamp, current);
+    }
+
+    public static Task PeriodicSample(
+        RoasteryMetrics metrics,
+        TimeSpan samplingInterval,
+        Func<DateTimeOffset, Sample, CancellationToken, Task> sample,
+        CancellationToken cancellationToken)
+    {
+        return Task.Run(async () =>
+        {
+            var waitFor = samplingInterval;
+            while (!cancellationToken.IsCancellationRequested)
+            {
+                await Task.Delay(waitFor, cancellationToken);
+
+                var stopwatch = Stopwatch.StartNew();
+
+                try
+                {
+                    var (timestamp, current) = metrics.Take();
+                    await sample(timestamp, current, cancellationToken);
+                }
+                catch
+                {
+                    // Ignored
+                }
+
+                // Account for the time taken to produce the sample when computing
+                // the next interval to wait for
+                var elapsed = stopwatch.Elapsed;
+                waitFor = elapsed < samplingInterval ? samplingInterval - stopwatch.Elapsed : samplingInterval;
+            }
+        }, cancellationToken);
+    }
+}

src/Roastery/Program.cs

@@ -1,11 +1,13 @@
-using System.Collections.Generic;
+using System;
+using System.Collections.Generic;
 using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
 using Roastery.Agents;
 using Roastery.Api;
 using Roastery.Data;
 using Roastery.Fake;
+using Roastery.Metrics;
 using Roastery.Util;
 using Roastery.Web;
 using Serilog;
@@ -15,22 +17,35 @@ namespace Roastery;
 // Named this way to make stack traces a little more believable :-)
 public static class Program
 {
-    public static async Task Main(ILogger logger, CancellationToken cancellationToken = default)
+    public static async Task Main(ILogger logger, PropertyNameMapping propertyNameMapping, CancellationToken cancellationToken = default)
     {
+        var metrics = new RoasteryMetrics();
+        
         var webApplicationLogger = logger.ForContext("Application", "Roastery Web Frontend");
 
+        // Sample metrics
+        var periodicSample = RoasteryMetrics.PeriodicSample(metrics, TimeSpan.FromSeconds(5), (timestamp, sample, ct) =>
+        {
+            foreach (var evt in sample.ToLogEvents(webApplicationLogger, propertyNameMapping, timestamp))
+            {
+                webApplicationLogger.Write(evt);
+            }
+            
+            return Task.CompletedTask;
+        }, cancellationToken);
+
         var database = new Database(webApplicationLogger, "roastery");
         DatabaseMigrator.Populate(database);
 
         var client = new HttpClient(
             "https://roastery.datalust.co",
             new NetworkLatencyMiddleware(
-                new RequestLoggingMiddleware(webApplicationLogger,
+                new RequestLoggingMiddleware(webApplicationLogger, metrics,
                     new SchedulingLatencyMiddleware(
                         new FaultInjectionMiddleware(webApplicationLogger,
                             new Router([
-                                new OrdersController(logger, database),
-                                new ProductsController(logger, database)
+                                new OrdersController(logger, metrics, database),
+                                new ProductsController(logger, metrics, database)
                             ], webApplicationLogger))))));
 
         var agents = new List<Agent>();
@@ -46,5 +61,6 @@ public static async Task Main(ILogger logger, CancellationToken cancellationToke
         agents.Add(new ArchivingBatch(client, batchApplicationLogger));
 
         await Task.WhenAll(agents.Select(a => Agent.Run(a, cancellationToken)));
+        await periodicSample;
     }
 }

src/Roastery/Web/Controller.cs

@@ -1,15 +1,18 @@
 using System.Net;
+using Roastery.Metrics;
 using Serilog;
 
 namespace Roastery.Web;
 
 abstract class Controller
 {
     protected ILogger Log { get; }
-        
-    protected Controller(ILogger logger)
+    protected RoasteryMetrics Metrics { get; }
+    
+    protected Controller(ILogger logger, RoasteryMetrics metrics)
     {
         Log = logger.ForContext(GetType());
+        Metrics = metrics;
     }
 
     protected static HttpResponse Json(object? body, HttpStatusCode statusCode = HttpStatusCode.OK)