Refactor/Allow metrics before block processing

src/Nethermind/Nethermind.Consensus/Processing/BlockAddedEventArgs.cs

@@ -0,0 +1,12 @@
+// SPDX-FileCopyrightText: 2025 Demerzel Solutions Limited
+// SPDX-License-Identifier: LGPL-3.0-only
+
+using System;
+using Nethermind.Core.Crypto;
+
+namespace Nethermind.Consensus.Processing;
+
+public class BlockAddedEventArgs(Hash256 blockHash) : EventArgs
+{
+    public Hash256 BlockHash => blockHash;
+}

src/Nethermind/Nethermind.Consensus/Processing/BlockchainProcessor.cs

@@ -143,6 +143,8 @@ public async ValueTask Enqueue(Block block, ProcessingOptions processingOptions)
         if (!_recoveryComplete)
         {
             Interlocked.Increment(ref _queueCount);
+            BlockAdded?.Invoke(this, new BlockAddedEventArgs(blockHash));
+
             _lastProcessedBlock = DateTime.UtcNow;
             try
             {
@@ -424,6 +426,7 @@ private void FireProcessingQueueEmpty()
 
     public event EventHandler? ProcessingQueueEmpty;
     public event EventHandler<BlockRemovedEventArgs>? BlockRemoved;
+    public event EventHandler<BlockAddedEventArgs>? BlockAdded;
     public bool IsEmpty => Volatile.Read(ref _queueCount) == 0;
     public int Count => Volatile.Read(ref _queueCount);
 

src/Nethermind/Nethermind.Consensus/Processing/IBlockProcessingQueue.cs

@@ -26,6 +26,7 @@ public interface IBlockProcessingQueue
         /// </summary>
         event EventHandler ProcessingQueueEmpty;
 
+        event EventHandler<BlockAddedEventArgs> BlockAdded;
         event EventHandler<BlockRemovedEventArgs> BlockRemoved;
 
         /// <summary>

src/Nethermind/Nethermind.Core.Test/Db/TestMemDbProvider.cs

@@ -10,6 +10,7 @@
 using Nethermind.Blockchain.Synchronization;
 using Nethermind.Db;
 using Nethermind.Init.Modules;
+using Nethermind.Monitoring;
 
 namespace Nethermind.Core.Test.Db
 {
@@ -23,7 +24,12 @@ public static Task<IDbProvider> InitAsync()
         public static IDbProvider Init()
         {
             return new ContainerBuilder()
-                .AddModule(new DbModule(new InitConfig() { DiagnosticMode = DiagnosticMode.MemDb }, new ReceiptConfig(), new SyncConfig()))
+                .AddModule(new DbModule(
+                    new InitConfig() { DiagnosticMode = DiagnosticMode.MemDb },
+                    new ReceiptConfig(),
+                    new SyncConfig(),
+                    dontConfigureMetric: true
+                ))
                 .AddSingleton<IDbProvider, ContainerOwningDbProvider>()
                 .Build()
                 .Resolve<IDbProvider>();

src/Nethermind/Nethermind.Core/ContainerBuilderExtensions.cs

@@ -374,6 +374,15 @@ public static ContainerBuilder Intercept<T>(this ContainerBuilder builder, Actio
         });
     }
 
+    public static ContainerBuilder Intercept<T>(this ContainerBuilder builder, Action<T, IComponentContext> interceptor) where T : class
+    {
+        return builder.AddDecorator<T>((ctx, service) =>
+        {
+            interceptor(service, ctx);
+            return service;
+        });
+    }
+
     /// <summary>
     /// A convenient way of creating a service whose members can be configured independent of other instances of the same
     /// type (assuming the type is of lifetime scope). This is useful for same type with multiple configuration

src/Nethermind/Nethermind.Db.Test/DbTrackerTests.cs

@@ -1,9 +1,21 @@
 // SPDX-License-Identifier: LGPL-3.0-only
 
+using System;
 using System.Linq;
 using Autofac;
 using FluentAssertions;
+using Nethermind.Api;
+using Nethermind.Blockchain.Receipts;
+using Nethermind.Blockchain.Synchronization;
+using Nethermind.Consensus.Processing;
 using Nethermind.Core;
+using Nethermind.Core.Crypto;
+using Nethermind.Core.Test;
+using Nethermind.Init.Modules;
+using Nethermind.Logging;
+using Nethermind.Monitoring;
+using Nethermind.Monitoring.Config;
+using NSubstitute;
 using NUnit.Framework;
 
 namespace Nethermind.Db.Test;
@@ -15,6 +27,9 @@ public void TestTrackOnlyCreatedDb()
     {
         using IContainer container = new ContainerBuilder()
             .AddSingleton<DbTracker>()
+            .AddSingleton<IMetricsConfig>(new MetricsConfig())
+            .AddSingleton<ILogManager>(LimboLogs.Instance)
+            .AddSingleton<IMonitoringService>(NoopMonitoringService.Instance)
             .AddDecorator<IDbFactory, DbTracker.DbFactoryInterceptor>()
             .AddSingleton<IDbFactory, MemDbFactory>()
             .Build();
@@ -30,4 +45,106 @@ public void TestTrackOnlyCreatedDb()
         var firstEntry = tracker.GetAllDbMeta().First();
         firstEntry.Key.Should().Be("TestDb");
     }
+
+    [Parallelizable(ParallelScope.None)]
+    [TestCase(true)]
+    [TestCase(false)]
+    public void TestUpdateDbMetric(bool isProcessing)
+    {
+        IBlockProcessingQueue queue = Substitute.For<IBlockProcessingQueue>();
+        (IContainer container, Action updateAction, FakeDb fakeDb) = ConfigureMetricUpdater((builder) => builder.AddSingleton<IBlockProcessingQueue>(queue));
+        using var _ = container;
+
+        // Reset
+        Metrics.DbReads["TestDb"] = 0;
+
+        if (isProcessing)
+        {
+            container.Resolve<IBlockProcessingQueue>(); // Only setup is something requested the block processing queue.
+            queue.IsEmpty.Returns(false);
+            queue.BlockAdded += Raise.EventWith<BlockAddedEventArgs>(new BlockAddedEventArgs(Keccak.Zero));
+        }
+
+        updateAction!();
+
+        // Assert
+        Assert.That(Metrics.DbReads["TestDb"], isProcessing ? Is.EqualTo(0) : Is.EqualTo(10));
+    }
+
+    [Parallelizable(ParallelScope.None)]
+    [Test]
+    public void DoesNotUpdateIfIntervalHasNotPassed()
+    {
+        (IContainer container, Action updateAction, FakeDb fakeDb) = ConfigureMetricUpdater();
+        using var _ = container;
+
+        container.Resolve<IDbFactory>().CreateDb(new DbSettings("TestDb", "TestDb"));
+
+        // Reset
+        Metrics.DbReads["TestDb"] = 0;
+
+        updateAction!();
+        Assert.That(Metrics.DbReads["TestDb"], Is.EqualTo(10));
+
+        fakeDb.SetMetric(new IDbMeta.DbMetric()
+        {
+            TotalReads = 11
+        });
+
+        updateAction!();
+        Assert.That(Metrics.DbReads["TestDb"], Is.EqualTo(10));
+    }
+
+    private (IContainer, Action, FakeDb) ConfigureMetricUpdater(Action<ContainerBuilder>? configurer = null)
+    {
+        IDbFactory fakeDbFactory = Substitute.For<IDbFactory>();
+
+        IMonitoringService monitoringService = Substitute.For<IMonitoringService>();
+        ContainerBuilder builder = new ContainerBuilder()
+            .AddModule(new DbModule(new InitConfig(), new ReceiptConfig(), new SyncConfig()))
+            .AddSingleton<IMetricsConfig>(new MetricsConfig())
+            .AddSingleton<ILogManager>(LimboLogs.Instance)
+            .AddSingleton<IMonitoringService>(monitoringService)
+            .AddDecorator<IDbFactory, DbTracker.DbFactoryInterceptor>()
+            .AddSingleton<IDbFactory>(fakeDbFactory);
+
+        configurer?.Invoke(builder);
+
+        IContainer container = builder
+            .Build();
+
+        IDbMeta.DbMetric metric = new IDbMeta.DbMetric()
+        {
+            TotalReads = 10
+        };
+        FakeDb fakeDb = new FakeDb(metric);
+        fakeDbFactory.CreateDb(Arg.Any<DbSettings>()).Returns(fakeDb);
+
+        Action updateAction = null;
+        monitoringService
+            .When((m) => m.AddMetricsUpdateAction(Arg.Any<Action>()))
+            .Do((c) =>
+            {
+                updateAction = (Action)c[0];
+            });
+
+        container.Resolve<IDbFactory>().CreateDb(new DbSettings("TestDb", "TestDb"));
+
+        return (container, updateAction, fakeDb);
+    }
+
+    private class FakeDb(IDbMeta.DbMetric metric) : TestMemDb, IDbMeta
+    {
+        private IDbMeta.DbMetric _metric = metric;
+
+        public override IDbMeta.DbMetric GatherMetric(bool includeSharedCache = false)
+        {
+            return _metric;
+        }
+
+        internal void SetMetric(IDbMeta.DbMetric metric)
+        {
+            _metric = metric;
+        }
+    }
 }

src/Nethermind/Nethermind.Db.Test/StandardDbInitializerTests.cs

@@ -90,7 +90,7 @@ private Task<IDbProvider> InitializeStandardDb(bool useReceipts, bool useMemDb,
                 }, new SyncConfig()
                 {
                     DownloadReceiptsInFastSync = useReceipts
-                }))
+                }, dontConfigureMetric: true))
                 .AddModule(new WorldStateModule(initConfig)) // For the full pruning db
                 .AddSingleton<IPruningConfig>(new PruningConfig())
                 .AddSingleton<IDbConfig>(new DbConfig())

src/Nethermind/Nethermind.Db/DbTracker.cs

@@ -3,13 +3,31 @@
 
 using System;
 using System.Collections.Generic;
+using Nethermind.Logging;
+using Nethermind.Monitoring;
+using Nethermind.Monitoring.Config;
 using NonBlocking;
 
 namespace Nethermind.Db;
 
 public class DbTracker
 {
     private readonly ConcurrentDictionary<string, IDbMeta> _createdDbs = new ConcurrentDictionary<string, IDbMeta>();
+    private readonly int _intervalSec;
+    private long _lastDbMetricsUpdate = 0;
+
+    private ILogger _logger;
+
+    public DbTracker(IMonitoringService monitoringService, IMetricsConfig metricsConfig, ILogManager logManager)
+    {
+        _intervalSec = metricsConfig.DbMetricIntervalSeconds;
+        _logger = logManager.GetClassLogger<DbTracker>();
+
+        if (metricsConfig.EnableDbSizeMetrics)
+        {
+            monitoringService.AddMetricsUpdateAction(UpdateDbMetrics);
+        }
+    }
 
     public void AddDb(string name, IDbMeta dbMeta)
     {
@@ -21,6 +39,39 @@ public IEnumerable<KeyValuePair<string, IDbMeta>> GetAllDbMeta()
         return _createdDbs;
     }
 
+    public bool Paused { get; set; } = false;
+
+    private void UpdateDbMetrics()
+    {
+        try
+        {
+            if (Paused) return;
+
+            if (Environment.TickCount64 - _lastDbMetricsUpdate < _intervalSec * 1000)
+            {
+                // Update max every minute
+                return;
+            }
+
+            foreach (KeyValuePair<string, IDbMeta> kv in GetAllDbMeta())
+            {
+                // Note: At the moment, the metric for a columns db is combined across column.
+                IDbMeta.DbMetric dbMetric = kv.Value.GatherMetric(includeSharedCache: kv.Key == DbNames.State); // Only include shared cache if state db
+                Db.Metrics.DbSize[kv.Key] = dbMetric.Size;
+                Db.Metrics.DbBlockCacheSize[kv.Key] = dbMetric.CacheSize;
+                Db.Metrics.DbMemtableSize[kv.Key] = dbMetric.MemtableSize;
+                Db.Metrics.DbIndexFilterSize[kv.Key] = dbMetric.IndexSize;
+                Db.Metrics.DbReads[kv.Key] = dbMetric.TotalReads;
+                Db.Metrics.DbWrites[kv.Key] = dbMetric.TotalWrites;
+            }
+            _lastDbMetricsUpdate = Environment.TickCount64;
+        }
+        catch (Exception e)
+        {
+            if (_logger.IsError) _logger.Error("Error during updating db metrics", e);
+        }
+    }
+
     public class DbFactoryInterceptor(DbTracker tracker, IDbFactory baseFactory) : IDbFactory
     {
         public IDb CreateDb(DbSettings dbSettings)

src/Nethermind/Nethermind.Db/MemDb.cs

@@ -151,7 +151,7 @@ public virtual void Set(ReadOnlySpan<byte> key, byte[]? value, WriteFlags flags
             _spanDb[key] = value;
         }
 
-        public IDbMeta.DbMetric GatherMetric(bool includeSharedCache = false) => new() { Size = Count };
+        public virtual IDbMeta.DbMetric GatherMetric(bool includeSharedCache = false) => new() { Size = Count };
 
         private IEnumerable<KeyValuePair<byte[], byte[]?>> OrderedDb => _db.OrderBy(kvp => kvp.Key, Bytes.Comparer);
     }

src/Nethermind/Nethermind.Db/Nethermind.Db.csproj

@@ -6,6 +6,7 @@
 
   <ItemGroup>
     <ProjectReference Include="..\Nethermind.Config\Nethermind.Config.csproj" />
+    <ProjectReference Include="..\Nethermind.Monitoring\Nethermind.Monitoring.csproj" />
     <ProjectReference Include="..\Nethermind.Serialization.Rlp\Nethermind.Serialization.Rlp.csproj" />
   </ItemGroup>
 

src/Nethermind/Nethermind.Init/Modules/DbModule.cs

@@ -6,13 +6,15 @@
 using Nethermind.Api;
 using Nethermind.Blockchain.Receipts;
 using Nethermind.Blockchain.Synchronization;
+using Nethermind.Consensus.Processing;
 using Nethermind.Core;
 using Nethermind.Db;
 using Nethermind.Db.Rocks;
 using Nethermind.Db.Rocks.Config;
 using Nethermind.Db.Rpc;
 using Nethermind.JsonRpc.Client;
 using Nethermind.Logging;
+using Nethermind.Monitoring.Config;
 using Nethermind.Serialization.Json;
 
 namespace Nethermind.Init.Modules;
@@ -30,7 +32,8 @@ namespace Nethermind.Init.Modules;
 public class DbModule(
     IInitConfig initConfig,
     IReceiptConfig receiptConfig,
-    ISyncConfig syncConfig
+    ISyncConfig syncConfig,
+    bool dontConfigureMetric = false
 ) : Module
 {
     protected override void Load(ContainerBuilder builder)
@@ -57,11 +60,6 @@ protected override void Load(ContainerBuilder builder)
                 return sortedKeyValue;
             })
 
-            // Monitoring use these to track active db. We intercept db factory to keep them lazy. Does not
-            // track db that is not created by db factory though...
-            .AddSingleton<DbTracker>()
-            .AddDecorator<IDbFactory, DbTracker.DbFactoryInterceptor>()
-
             .AddDatabase(DbNames.State)
             .AddDatabase(DbNames.Code)
             .AddDatabase(DbNames.Metadata)
@@ -77,6 +75,30 @@ protected override void Load(ContainerBuilder builder)
             .AddColumnDatabase<BlobTxsColumns>(DbNames.BlobTransactions)
             ;
 
+        if (!dontConfigureMetric)
+        {
+            // Intercept created db to publish metric.
+            // Dont use constructor injection to get all db because that would resolve all db
+            // making them not lazy.
+            builder
+                .AddSingleton<DbTracker>()
+                .AddDecorator<IDbFactory, DbTracker.DbFactoryInterceptor>()
+
+                // Intercept block processing by checking the queue and pausing the metrics when that happen.
+                // Dont use constructor injection because this would prevent the metric from being updated before
+                // the block processing chain is constructed, eg: verifytrie or import jobs.
+                .Intercept<IBlockProcessingQueue>((processingQueue, ctx) =>
+                {
+                    if (!ctx.Resolve<IMetricsConfig>().PauseDbMetricDuringBlockProcessing) return;
+
+                    // Do not update db metrics while processing a block
+                    DbTracker updater = ctx.Resolve<DbTracker>();
+                    processingQueue.BlockAdded += (sender, args) => updater.Paused = !processingQueue.IsEmpty;
+                    processingQueue.BlockRemoved += (sender, args) => updater.Paused = !processingQueue.IsEmpty;
+                })
+                ;
+        }
+
         switch (initConfig.DiagnosticMode)
         {
             case DiagnosticMode.MemDb: