3128 export metrics for conflation blob and aggregation size (#571)

* feat: added gauge metrics for batch, blob, aggregation size

* feat: use histogram instead of gauge for aggregation, blob, batch size metrics

* fix: unit test

coordinator/core/src/main/kotlin/net/consensys/zkevm/ethereum/coordination/aggregation/GlobalAggregationCalculator.kt

@@ -7,7 +7,6 @@ import net.consensys.zkevm.domain.BlobCounters
 import net.consensys.zkevm.domain.BlobsToAggregate
 import org.apache.logging.log4j.LogManager
 import org.apache.logging.log4j.Logger
-import java.lang.IllegalStateException
 import java.util.PriorityQueue
 
 class GlobalAggregationCalculator(

coordinator/core/src/main/kotlin/net/consensys/zkevm/ethereum/coordination/aggregation/ProofAggregationCoordinatorService.kt

@@ -4,6 +4,7 @@ import build.linea.domain.BlockIntervals
 import build.linea.domain.toBlockIntervalsString
 import io.vertx.core.Vertx
 import kotlinx.datetime.Clock
+import net.consensys.linea.metrics.LineaMetricsCategory
 import net.consensys.linea.metrics.MetricsFacade
 import net.consensys.zkevm.LongRunningService
 import net.consensys.zkevm.PeriodicPollingService
@@ -27,6 +28,7 @@ import kotlin.time.Duration
 class ProofAggregationCoordinatorService(
   private val vertx: Vertx,
   private val config: Config,
+  private val metricsFacade: MetricsFacade,
   private var nextBlockNumberToPoll: Long,
   private val aggregationCalculator: AggregationCalculator,
   private val aggregationsRepository: AggregationsRepository,
@@ -46,6 +48,21 @@ class ProofAggregationCoordinatorService(
   )
 
   private val pendingBlobs = ConcurrentLinkedQueue<BlobAndBatchCounters>()
+  private val aggregationSizeInBlocksHistogram = metricsFacade.createHistogram(
+    category = LineaMetricsCategory.AGGREGATION,
+    name = "blocks.size",
+    description = "Number of blocks in each aggregation"
+  )
+  private val aggregationSizeInBatchesHistogram = metricsFacade.createHistogram(
+    category = LineaMetricsCategory.AGGREGATION,
+    name = "batches.size",
+    description = "Number of batches in each aggregation"
+  )
+  private val aggregationSizeInBlobsHistogram = metricsFacade.createHistogram(
+    category = LineaMetricsCategory.AGGREGATION,
+    name = "blobs.size",
+    description = "Number of blobs in each aggregation"
+  )
 
   init {
     aggregationCalculator.onAggregation(this)
@@ -132,6 +149,10 @@ class ProofAggregationCoordinatorService(
       blobCounters.blobCounters.numberOfBatches
     }
 
+    aggregationSizeInBlocksHistogram.record(blobsToAggregate.blocksRange.count().toDouble())
+    aggregationSizeInBatchesHistogram.record(batchCount.toDouble())
+    aggregationSizeInBlobsHistogram.record(compressionBlobs.size.toDouble())
+
     val compressionProofIndexes = compressionBlobs.map {
       ProofIndex(
         startBlockNumber = it.blobCounters.startBlockNumber,
@@ -250,11 +271,12 @@ class ProofAggregationCoordinatorService(
       )
 
       val proofAggregationService = ProofAggregationCoordinatorService(
-        vertx,
+        vertx = vertx,
         config = Config(
           pollingInterval = aggregationCoordinatorPollingInterval,
           proofsLimit = maxProofsPerAggregation
         ),
+        metricsFacade = metricsFacade,
         nextBlockNumberToPoll = startBlockNumberInclusive.toLong(),
         aggregationCalculator = globalAggregationCalculator,
         aggregationsRepository = aggregationsRepository,

coordinator/core/src/main/kotlin/net/consensys/zkevm/ethereum/coordination/blob/BlobCompressionProofCoordinator.kt

@@ -38,11 +38,22 @@ class BlobCompressionProofCoordinator(
   private val blobsToHandle = LinkedBlockingDeque<Blob>(defaultQueueCapacity)
   private var timerId: Long? = null
   private lateinit var blobPollingAction: Handler<Long>
+
   private val blobsCounter = metricsFacade.createCounter(
     category = LineaMetricsCategory.BLOB,
     name = "counter",
     description = "New blobs arriving to blob compression proof coordinator"
   )
+  private val blobSizeInBlocksHistogram = metricsFacade.createHistogram(
+    category = LineaMetricsCategory.BLOB,
+    name = "blocks.size",
+    description = "Number of blocks in each blob"
+  )
+  private val blobSizeInBatchesHistogram = metricsFacade.createHistogram(
+    category = LineaMetricsCategory.BLOB,
+    name = "batches.size",
+    description = "Number of batches in each blob"
+  )
 
   init {
     metricsFacade.createGauge(
@@ -167,6 +178,8 @@ class BlobCompressionProofCoordinator(
       blobsToHandle.size,
       blob.conflations.toBlockIntervalsString()
     )
+    blobSizeInBlocksHistogram.record(blob.blocksRange.count().toDouble())
+    blobSizeInBatchesHistogram.record(blob.conflations.size.toDouble())
     blobsToHandle.put(blob)
     log.trace("Blob was added to the handling queue {}", blob)
     return SafeFuture.completedFuture(Unit)

coordinator/core/src/main/kotlin/net/consensys/zkevm/ethereum/coordination/conflation/ConflationServiceImpl.kt

@@ -37,6 +37,11 @@ class ConflationServiceImpl(
     name = "blocks.imported",
     description = "New blocks arriving to conflation service counter"
   )
+  private val batchSizeInBlocksHistogram = metricsFacade.createHistogram(
+    category = LineaMetricsCategory.CONFLATION,
+    name = "blocks.size",
+    description = "Number of blocks in each conflated batch"
+  )
 
   init {
     metricsFacade.createGauge(
@@ -63,6 +68,8 @@ class ConflationServiceImpl(
       conflation.tracesCounters,
       conflation.blocksRange.joinToString(",", "[", "]") { it.toString() }
     )
+    batchSizeInBlocksHistogram.record(conflation.blocksRange.count().toDouble())
+
     val blocksToConflate =
       blocksInProgress
         .filter { it.number in conflation.blocksRange }

coordinator/core/src/test/kotlin/net/consensys/zkevm/ethereum/coordination/aggregation/ProofAggregationCoordinatorServiceTest.kt

@@ -1,9 +1,11 @@
 package net.consensys.zkevm.ethereum.coordination.aggregation
 
 import build.linea.domain.BlockIntervals
+import io.micrometer.core.instrument.simple.SimpleMeterRegistry
 import io.vertx.core.Vertx
 import kotlinx.datetime.Clock
 import kotlinx.datetime.Instant
+import net.consensys.linea.metrics.micrometer.MicrometerMetricsFacade
 import net.consensys.trimToSecondPrecision
 import net.consensys.zkevm.coordinator.clients.ProofAggregationProverClientV2
 import net.consensys.zkevm.domain.Aggregation
@@ -91,6 +93,7 @@ class ProofAggregationCoordinatorServiceTest {
       consecutiveProvenBlobsProvider = mockAggregationsRepository::findConsecutiveProvenBlobs,
       proofAggregationClient = mockProofAggregationClient,
       aggregationL2StateProvider = aggregationL2StateProvider,
+      metricsFacade = MicrometerMetricsFacade(registry = SimpleMeterRegistry()),
       provenAggregationEndBlockNumberConsumer = provenAggregationEndBlockNumberConsumer
     )
     verify(mockAggregationCalculator).onAggregation(proofAggregationCoordinatorService)