Or 1834 metrics

AssociationRegistry.sln

@@ -20,8 +20,8 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = ".Solution Items", ".Solutio
 		definition_of_done.md = definition_of_done.md
 		readme.md = readme.md
 		CONTRIBUTING.md = CONTRIBUTING.md
-		otel-collector-config.yaml = otel-collector-config.yaml
 		regen-marten.sh = regen-marten.sh
+		otel-collector-config.yaml = otel-collector-config.yaml
 	EndProjectSection
 EndProject
 Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "test", "test", "{C812C887-6495-405A-8B99-4F686D243126}"

src/AssociationRegistry.Acm.Api/Infrastructure/Extensions/MartenExtensions.cs

@@ -14,9 +14,12 @@
 using Newtonsoft.Json;
 using Schema.VerenigingenPerInsz;
 
-public static class MartenExtentions
+public static class MartenExtensions
 {
-    public static IServiceCollection AddMarten(this IServiceCollection services, PostgreSqlOptionsSection postgreSqlOptions, IConfiguration configuration)
+    public static IServiceCollection AddMarten(
+        this IServiceCollection services,
+        PostgreSqlOptionsSection postgreSqlOptions,
+        IConfiguration configuration)
     {
         var martenConfiguration = services
                                  .AddSingleton(postgreSqlOptions)
@@ -28,7 +31,7 @@ public static IServiceCollection AddMarten(this IServiceCollection services, Pos
                                           opts.Events.StreamIdentity = StreamIdentity.AsString;
                                           opts.Serializer(CreateCustomMartenSerializer());
                                           opts.Events.MetadataConfig.EnableAll();
-                                          opts.AddPostgresProjections();
+                                          opts.AddPostgresProjections(serviceProvider);
 
                                           opts.RegisterDocumentType<VerenigingenPerInszDocument>();
                                           opts.RegisterDocumentType<VerenigingDocument>();

src/AssociationRegistry.Acm.Api/Infrastructure/Extensions/StoreOptionsExtensions.cs

@@ -2,12 +2,16 @@ namespace AssociationRegistry.Acm.Api.Infrastructure.Extensions;
 
 using Marten;
 using Marten.Events.Projections;
+using Metrics;
+using Microsoft.Extensions.DependencyInjection;
 using Projections;
+using System;
 
 public static class StoreOptionsExtensions
 {
-    public static void AddPostgresProjections(this StoreOptions source)
+    public static void AddPostgresProjections(this StoreOptions source, IServiceProvider serviceProvider)
     {
         source.Projections.Add(new VerenigingenPerInszProjection(), ProjectionLifecycle.Async);
+        source.Projections.AsyncListeners.Add(new ProjectionStateListener(serviceProvider.GetRequiredService<AcmInstrumentation>()));
     }
 }

src/AssociationRegistry.Acm.Api/Infrastructure/Metrics/AcmInstrumentation.cs

@@ -0,0 +1,41 @@
+namespace AssociationRegistry.Acm.Api.Infrastructure.Metrics;
+
+using OpenTelemetry;
+using System;
+using System.Diagnostics;
+using System.Diagnostics.Metrics;
+
+/// <summary>
+/// It is recommended to use a custom type to hold references for
+/// ActivitySource and Instruments. This avoids possible type collisions
+/// with other components in the DI container.
+/// </summary>
+public class AcmInstrumentation : IInstrumentation, IDisposable
+{
+    public string ActivitySourceName => "AssociationRegistry";
+    public string MeterName => "AcmProjections";
+    private readonly Meter meter;
+
+    public AcmInstrumentation()
+    {
+        var version = typeof(AcmInstrumentation).Assembly.GetName().Version?.ToString();
+        ActivitySource = new ActivitySource(ActivitySourceName, version);
+        meter = new Meter(MeterName, version);
+
+        _verenigingPerInszGauge =
+            meter.CreateObservableGauge(name: "ar.acm.p.verenigingPerInsz.g", unit: "events",
+                                              description: "vereniging per insz projection",
+                                              observeValue: () => VerenigingPerInszEventValue);
+    }
+
+    public ActivitySource ActivitySource { get; }
+    private ObservableGauge<long> _verenigingPerInszGauge;
+
+    public long VerenigingPerInszEventValue = 0;
+
+    public void Dispose()
+    {
+        ActivitySource.Dispose();
+        meter.Dispose();
+    }
+}

src/AssociationRegistry.Acm.Api/Infrastructure/Metrics/ProjectionStateListener.cs

@@ -0,0 +1,76 @@
+namespace AssociationRegistry.Acm.Api.Infrastructure.Metrics;
+
+using Marten;
+using Marten.Events.Daemon;
+using Marten.Internal.Operations;
+using Marten.Services;
+using Newtonsoft.Json;
+using Newtonsoft.Json.Linq;
+using Newtonsoft.Json.Serialization;
+using Projections;
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Reflection;
+using System.Threading;
+using System.Threading.Tasks;
+
+public class ProjectionStateListener : DocumentSessionListenerBase
+{
+    private readonly AcmInstrumentation _acmInstrumentation;
+
+    public ProjectionStateListener(AcmInstrumentation acmInstrumentation)
+    {
+        _acmInstrumentation = acmInstrumentation;
+    }
+
+    public override Task AfterCommitAsync(IDocumentSession session, IChangeSet commit, CancellationToken token)
+    {
+        if (commit is not IUnitOfWork uow) return Task.CompletedTask;
+        var operations = uow.OperationsFor<Marten.Events.IEvent>();
+
+        foreach (var operation in operations)
+        {
+            var range = GetEventRange(operation);
+
+            if (range is null) continue;
+
+            if (range.ShardName.ProjectionName ==
+                typeof(VerenigingenPerInszProjection).FullName)
+                _acmInstrumentation.VerenigingPerInszEventValue =range.SequenceCeiling;
+        }
+
+        return Task.CompletedTask;
+    }
+
+    private static EventRange? GetEventRange(IStorageOperation opperation)
+    {
+        return opperation.GetType().Name switch
+        {
+            "InsertProjectionProgress" => opperation.GetType().GetField("_progress", BindingFlags.NonPublic|BindingFlags.Instance).GetValue(opperation) as EventRange,
+            "UpdateProjectionProgress" => opperation.GetType().GetProperty("Range").GetValue(opperation) as EventRange,
+            _ => null
+        };
+    }
+}
+
+public class AllFieldsContractResolver : DefaultContractResolver
+{
+    protected override IList<JsonProperty> CreateProperties(Type type, MemberSerialization memberSerialization)
+    {
+        var props = type
+                   .GetProperties(BindingFlags.Public | BindingFlags.NonPublic | BindingFlags.Instance)
+                   .Select(p => base.CreateProperty(p, memberSerialization))
+                   .Union(type.GetFields(BindingFlags.Public | BindingFlags.NonPublic | BindingFlags.Instance)
+                              .Select(f => base.CreateProperty(f, memberSerialization)))
+                   .ToList();
+
+        props.ForEach(p =>
+        {
+            p.Writable = true;
+            p.Readable = true;
+        });
+
+        return props;
+    }
+}

src/AssociationRegistry.Acm.Api/Infrastructure/Monitoring/OpenTelemetryMetrics.cs

@@ -0,0 +1,83 @@
+namespace AssociationRegistry.Acm.Api.Infrastructure.Monitoring;
+
+using System;
+using System.Diagnostics.Metrics;
+
+public class OpenTelemetryMetrics
+{
+    public class ElasticSearchProjections
+    {
+        public static Func<string, string> MeterNameFunc = name => $"VR.ES.{name}";
+        private readonly Meter _meter;
+        public Histogram<int> NumberOfEnvelopesHandledHistogram { get; }
+        public int NumberOfEnvelopesHandledGauge { get; set; }
+        public int NumberOfEnvelopesHandledCounter { get; set; }
+        public Histogram<int> NumberOfEnvelopesBehindHistogram { get; }
+        public int NumberOfEnvelopesBehindGauge { get; set; }
+        public int NumberOfEnvelopesBehindCounter { get; set; }
+        public int LastProcessedEventNumberGauge { get; set; }
+        public int LastProcessedEventNumberCounter { get; set; }
+        public int MaxEventNumberToProcessGauge { get; set; }
+        public int MaxEventNumberToProcessCounter { get; set; }
+
+        private static class MeterNames
+        {
+            public static Func<string, string, string> EnvelopesHandled = (meterName, meterType)
+                => $"{meterName}.{nameof(EnvelopesHandled)}.{meterType}";
+
+            public static Func<string, string, string> EnvelopesBehind = (meterName, meterType)
+                => $"{meterName}.{nameof(EnvelopesBehind)}.{meterType}";
+
+            public static Func<string, string, string> LastProcessedEvent =
+                (meterName, meterType) => $"{meterName}.{nameof(LastProcessedEvent)}.{meterType}";
+
+            public static Func<string, string, string> MaxEventNumberToProcess =
+                (meterName, meterType) => $"{meterName}.{nameof(MaxEventNumberToProcess)}.{meterType}";
+
+            public static Func<string, string, string> OrganisationsToRebuild =
+                (meterName, meterType) => $"{meterName}.{nameof(OrganisationsToRebuild)}.{meterType}";
+        }
+
+        public ElasticSearchProjections(string runnerName)
+        {
+            const string histogram = "Histogram";
+            const string counter = "Counter";
+            const string gauge = "Gauge";
+
+            var meterName = MeterNameFunc(runnerName);
+            _meter = new Meter(meterName);
+
+            NumberOfEnvelopesHandledHistogram = _meter.CreateHistogram<int>(MeterNames.EnvelopesHandled(meterName, histogram),
+                                                                            unit: "envelopes", description: "number of envelopes handled");
+
+            _meter.CreateObservableGauge(MeterNames.EnvelopesHandled(meterName, gauge), observeValue: () => NumberOfEnvelopesHandledGauge,
+                                         unit: "envelopes", description: "number of envelopes handled");
+
+            _meter.CreateObservableUpDownCounter(MeterNames.EnvelopesHandled(meterName, counter),
+                                                 observeValue: () => NumberOfEnvelopesHandledCounter, unit: "envelopes",
+                                                 description: "number of envelopes handled");
+
+            _meter.CreateObservableUpDownCounter(MeterNames.LastProcessedEvent(meterName, counter),
+                                                 observeValue: () => LastProcessedEventNumberCounter);
+
+            _meter.CreateObservableGauge(MeterNames.LastProcessedEvent(meterName, gauge),
+                                         observeValue: () => LastProcessedEventNumberGauge);
+
+            _meter.CreateObservableUpDownCounter(MeterNames.MaxEventNumberToProcess(meterName, counter),
+                                                 observeValue: () => MaxEventNumberToProcessCounter);
+
+            _meter.CreateObservableGauge(MeterNames.MaxEventNumberToProcess(meterName, gauge),
+                                         observeValue: () => MaxEventNumberToProcessGauge);
+
+            NumberOfEnvelopesBehindHistogram = _meter.CreateHistogram<int>(MeterNames.EnvelopesBehind(meterName, histogram),
+                                                                           unit: "envelopes", description: "number of envelopes behind");
+
+            _meter.CreateObservableUpDownCounter(MeterNames.EnvelopesBehind(meterName, counter),
+                                                 observeValue: () => NumberOfEnvelopesBehindCounter, unit: "envelopes",
+                                                 description: "number of envelopes behind");
+
+            _meter.CreateObservableGauge(MeterNames.EnvelopesBehind(meterName, gauge), observeValue: () => NumberOfEnvelopesBehindGauge,
+                                         unit: "envelopes", description: "number of envelopes behind");
+        }
+    }
+}

src/AssociationRegistry.Acm.Api/Program.cs

@@ -16,6 +16,7 @@ namespace AssociationRegistry.Acm.Api;
 using Infrastructure.Configuration;
 using Infrastructure.ConfigurationBindings;
 using Infrastructure.Extensions;
+using Infrastructure.Metrics;
 using Microsoft.AspNetCore.Authentication.JwtBearer;
 using Microsoft.AspNetCore.Authorization;
 using Microsoft.AspNetCore.Builder;
@@ -214,7 +215,8 @@ private static void LoadConfiguration(WebApplicationBuilder builder, params stri
     {
         builder.Configuration
                .AddJsonFile("appsettings.json")
-               .AddJsonFile($"appsettings.{builder.Environment.EnvironmentName.ToLowerInvariant()}.json", optional: true, reloadOnChange: false)
+               .AddJsonFile($"appsettings.{builder.Environment.EnvironmentName.ToLowerInvariant()}.json", optional: true,
+                            reloadOnChange: false)
                .AddJsonFile($"appsettings.{Environment.MachineName.ToLowerInvariant()}.json", optional: true, reloadOnChange: false)
                .AddEnvironmentVariables()
                .AddCommandLine(args)
@@ -249,7 +251,7 @@ private static void ConfigureServices(WebApplicationBuilder builder)
         builder.Services
                .AddSingleton(appSettings)
                .AddMarten(postgreSqlOptionsSection, builder.Configuration)
-               .AddOpenTelemetry()
+               .AddOpenTelemetry(new AcmInstrumentation())
                .AddHttpContextAccessor()
                .AddControllers();
 
@@ -303,7 +305,8 @@ private static void ConfigureServices(WebApplicationBuilder builder)
                         cfg.AddPolicy(
                             StartupConstants.AllowSpecificOrigin,
                             configurePolicy: corsPolicy => corsPolicy
-                                                          .WithOrigins(builder.Configuration.GetValue<string[]>("Cors") ?? Array.Empty<string>())
+                                                          .WithOrigins(builder.Configuration.GetValue<string[]>("Cors") ??
+                                                                       Array.Empty<string>())
                                                           .WithMethods(StartupConstants.HttpMethodsAsString)
                                                           .WithHeaders(StartupConstants.Headers)
                                                           .WithExposedHeaders(StartupConstants.ExposedHeaders)
@@ -337,7 +340,9 @@ private static void ConfigureServices(WebApplicationBuilder builder)
                     JwtBearerDefaults.AuthenticationScheme,
                     configureOptions: options =>
                     {
-                        var configOptions = builder.Configuration.GetSection(nameof(OAuth2IntrospectionOptions)).Get<OAuth2IntrospectionOptions>();
+                        var configOptions = builder.Configuration.GetSection(nameof(OAuth2IntrospectionOptions))
+                                                   .Get<OAuth2IntrospectionOptions>();
+
                         options.ClientId = configOptions.ClientId;
                         options.ClientSecret = configOptions.ClientSecret;
                         options.Authority = configOptions.Authority;

src/AssociationRegistry.Admin.Api/Infrastructure/Extensions/MartenExtensions.cs

@@ -8,6 +8,7 @@
 using Marten;
 using Marten.Events;
 using Marten.Services;
+using Metrics;
 using Microsoft.Extensions.Configuration;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Hosting;
@@ -35,6 +36,8 @@ public static IServiceCollection AddMarten(
                 opts.Serializer(CreateCustomMartenSerializer());
                 opts.Events.MetadataConfig.EnableAll();
 
+                opts.Listeners.Add(new HighWatermarkListener(serviceProvider.GetRequiredService<Instrumentation>()));
+
                 opts.RegisterDocumentType<BeheerVerenigingDetailDocument>();
                 opts.RegisterDocumentType<BeheerVerenigingHistoriekDocument>();
 

src/AssociationRegistry.Admin.Api/Infrastructure/Metrics/HighWatermarkListener.cs

@@ -0,0 +1,28 @@
+namespace AssociationRegistry.Admin.Api.Infrastructure.Metrics;
+
+using Marten;
+using Marten.Services;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+
+public class HighWatermarkListener : DocumentSessionListenerBase
+{
+    private readonly Instrumentation _instrumentation;
+
+    public HighWatermarkListener(Instrumentation instrumentation)
+    {
+        _instrumentation = instrumentation;
+    }
+
+    public override async Task AfterCommitAsync(IDocumentSession session, IChangeSet commit, CancellationToken token)
+    {
+        var events = commit.GetEvents().ToArray();
+
+        if (events.Any())
+        {
+            var highWatermark = events.Max(x => x.Sequence);
+            _instrumentation.HighWatermarkEventValue = highWatermark;
+        }
+    }
+}

src/AssociationRegistry.Admin.Api/Infrastructure/Metrics/Instrumentation.cs

@@ -0,0 +1,38 @@
+namespace AssociationRegistry.Admin.Api.Infrastructure.Metrics;
+
+using OpenTelemetry;
+using System;
+using System.Diagnostics;
+using System.Diagnostics.Metrics;
+
+/// <summary>
+/// It is recommended to use a custom type to hold references for
+/// ActivitySource and Instruments. This avoids possible type collisions
+/// with other components in the DI container.
+/// </summary>
+public class Instrumentation : IInstrumentation, IDisposable
+{
+    public string ActivitySourceName => "AssociationRegistry";
+    public string MeterName => "AdminApi";
+    private readonly Meter meter;
+
+    public Instrumentation()
+    {
+        var version = typeof(Instrumentation).Assembly.GetName().Version?.ToString();
+        ActivitySource = new ActivitySource(ActivitySourceName, version);
+        meter = new Meter(MeterName, version);
+
+        _highWatermarkGauge = meter.CreateObservableGauge<long>(name: "ar.beheer.p.highWatermark.g", unit: "events",
+                                                               description: "high watermark", observeValue: () => HighWatermarkEventValue);
+    }
+
+    public ActivitySource ActivitySource { get; }
+    private ObservableGauge<long> _highWatermarkGauge;
+    public long HighWatermarkEventValue = 0;
+
+    public void Dispose()
+    {
+        ActivitySource.Dispose();
+        meter.Dispose();
+    }
+}