Add opPool metrics

packages/lodestar/src/chain/chain.ts

@@ -222,6 +222,8 @@ export class BeaconChain implements IBeaconChain {
     new PrecomputeNextEpochTransitionScheduler(this, this.config, metrics, this.logger, signal);
 
     handleChainEvents.bind(this)(this.abortController.signal);
+
+    metrics?.opPool.aggregatedAttestationPoolSize.addCollect(() => this.onScrapeMetrics());
   }
 
   close(): void {
@@ -342,6 +344,18 @@ export class BeaconChain implements IBeaconChain {
     this.logger.debug("Persisted invalid ssz object", {id: suffix, filepath});
   }
 
+  private onScrapeMetrics(): void {
+    const {attestationCount, attestationDataCount} = this.aggregatedAttestationPool.getAttestationCount();
+    this.metrics?.opPool.aggregatedAttestationPoolSize.set(attestationCount);
+    this.metrics?.opPool.aggregatedAttestationPoolUniqueData.set(attestationDataCount);
+    this.metrics?.opPool.attestationPoolSize.set(this.attestationPool.getAttestationCount());
+    this.metrics?.opPool.attesterSlashingPoolSize.set(this.opPool.attesterSlashingsSize);
+    this.metrics?.opPool.proposerSlashingPoolSize.set(this.opPool.proposerSlashingsSize);
+    this.metrics?.opPool.voluntaryExitPoolSize.set(this.opPool.voluntaryExitsSize);
+    this.metrics?.opPool.syncCommitteeMessagePoolSize.set(this.syncCommitteeMessagePool.size);
+    this.metrics?.opPool.syncContributionAndProofPoolSize.set(this.syncContributionAndProofPool.size);
+  }
+
   async updateBeaconProposerData(epoch: Epoch, proposers: ProposerPreparationData[]): Promise<void> {
     proposers.forEach((proposer) => {
       this.beaconProposerCache.add(epoch, proposer);

packages/lodestar/src/chain/opPools/aggregatedAttestationPool.ts

@@ -54,6 +54,19 @@ export class AggregatedAttestationPool {
   );
   private lowestPermissibleSlot = 0;
 
+  /** For metrics to track size of the pool */
+  getAttestationCount(): {attestationCount: number; attestationDataCount: number} {
+    let attestationCount = 0;
+    let attestationDataCount = 0;
+    for (const attestationGroupByData of this.attestationGroupByDataHashBySlot.values()) {
+      attestationDataCount += attestationGroupByData.size;
+      for (const attestationGroup of attestationGroupByData.values()) {
+        attestationCount += attestationGroup.getAttestationCount();
+      }
+    }
+    return {attestationCount, attestationDataCount};
+  }
+
   add(attestation: phase0.Attestation, attestingIndicesCount: number, committee: ValidatorIndex[]): InsertOutcome {
     const slot = attestation.data.slot;
     const lowestPermissibleSlot = this.lowestPermissibleSlot;
@@ -210,6 +223,10 @@ export class MatchingDataAttestationGroup {
 
   constructor(readonly committee: ValidatorIndex[], readonly data: phase0.AttestationData) {}
 
+  getAttestationCount(): number {
+    return this.attestations.length;
+  }
+
   /**
    * Add an attestation.
    * Try to preaggregate to existing attestations if possible.

packages/lodestar/src/chain/opPools/attestationPool.ts

@@ -60,7 +60,14 @@ export class AttestationPool {
   );
   private lowestPermissibleSlot = 0;
 
-  // TODO: Add metrics for total num of attestations in the pool
+  /** Returns current count of pre-aggregated attestations with unique data */
+  getAttestationCount(): number {
+    let attestationCount = 0;
+    for (const attestationByRoot of this.attestationByRootBySlot.values()) {
+      attestationCount += attestationByRoot.size;
+    }
+    return attestationCount;
+  }
 
   /**
    * Accepts an `VerifiedUnaggregatedAttestation` and attempts to apply it to the "naive

packages/lodestar/src/chain/opPools/opPool.ts

@@ -27,6 +27,18 @@ export class OpPool {
   /** Set of seen attester slashing indexes. No need to prune */
   private readonly attesterSlashingIndexes = new Set<ValidatorIndex>();
 
+  // Getters for metrics
+
+  get attesterSlashingsSize(): number {
+    return this.attesterSlashings.size;
+  }
+  get proposerSlashingsSize(): number {
+    return this.proposerSlashings.size;
+  }
+  get voluntaryExitsSize(): number {
+    return this.voluntaryExits.size;
+  }
+
   async fromPersisted(db: IBeaconDb): Promise<void> {
     const [attesterSlashings, proposerSlashings, voluntaryExits] = await Promise.all([
       db.attesterSlashing.entries(),

packages/lodestar/src/chain/opPools/syncCommitteeMessagePool.ts

@@ -44,6 +44,17 @@ export class SyncCommitteeMessagePool {
   >(() => new MapDef<Subnet, Map<BlockRootHex, ContributionFast>>(() => new Map<BlockRootHex, ContributionFast>()));
   private lowestPermissibleSlot = 0;
 
+  /** Returns current count of unique ContributionFast by block root and subnet */
+  get size(): number {
+    let count = 0;
+    for (const contributionsByRootBySubnet of this.contributionsByRootBySubnetBySlot.values()) {
+      for (const contributionsByRoot of contributionsByRootBySubnet.values()) {
+        count += contributionsByRoot.size;
+      }
+    }
+    return count;
+  }
+
   // TODO: indexInSubcommittee: number should be indicesInSyncCommittee
   add(subnet: Subnet, signature: altair.SyncCommitteeMessage, indexInSubcommittee: number): InsertOutcome {
     const {slot, beaconBlockRoot} = signature;

packages/lodestar/src/chain/opPools/syncContributionAndProofPool.ts

@@ -34,17 +34,18 @@ export type SyncContributionFast = {
 
 /** Hex string of `contribution.beaconBlockRoot` */
 type BlockRootHex = string;
+type Subnet = number;
 
 /**
  * Cache SyncCommitteeContribution and seen ContributionAndProof.
  * This is used for SignedContributionAndProof validation and block factory.
  * This stays in-memory and should be pruned per slot.
  */
 export class SyncContributionAndProofPool {
-  private readonly bestContributionBySubnetRootSlot = new MapDef<
+  private readonly bestContributionBySubnetRootBySlot = new MapDef<
     Slot,
-    MapDef<BlockRootHex, Map<number, SyncContributionFast>>
-  >(() => new MapDef<BlockRootHex, Map<number, SyncContributionFast>>(() => new Map<number, SyncContributionFast>()));
+    MapDef<BlockRootHex, Map<Subnet, SyncContributionFast>>
+  >(() => new MapDef<BlockRootHex, Map<Subnet, SyncContributionFast>>(() => new Map<number, SyncContributionFast>()));
 
   private lowestPermissibleSlot = 0;
 
@@ -55,6 +56,17 @@ export class SyncContributionAndProofPool {
     }
   }
 
+  /** Returns current count of unique SyncContributionFast by block root and subnet */
+  get size(): number {
+    let count = 0;
+    for (const bestContributionByRootBySubnet of this.bestContributionBySubnetRootBySlot.values()) {
+      for (const bestContributionByRoot of bestContributionByRootBySubnet.values()) {
+        count += bestContributionByRoot.size;
+      }
+    }
+    return count;
+  }
+
   /**
    * Only call this once we pass all validation.
    */
@@ -70,7 +82,7 @@ export class SyncContributionAndProofPool {
     }
 
     // Limit object per slot
-    const bestContributionBySubnetByRoot = this.bestContributionBySubnetRootSlot.getOrDefault(slot);
+    const bestContributionBySubnetByRoot = this.bestContributionBySubnetRootBySlot.getOrDefault(slot);
     if (bestContributionBySubnetByRoot.size >= MAX_ITEMS_PER_SLOT) {
       throw new OpPoolError({code: OpPoolErrorCode.REACHED_MAX_PER_SLOT});
     }
@@ -90,7 +102,7 @@ export class SyncContributionAndProofPool {
    * This is for the block factory, the same to process_sync_committee_contributions in the spec.
    */
   getAggregate(slot: Slot, prevBlockRoot: Root): altair.SyncAggregate {
-    const bestContributionBySubnet = this.bestContributionBySubnetRootSlot.get(slot)?.get(toHexString(prevBlockRoot));
+    const bestContributionBySubnet = this.bestContributionBySubnetRootBySlot.get(slot)?.get(toHexString(prevBlockRoot));
     if (!bestContributionBySubnet || bestContributionBySubnet.size === 0) {
       // TODO: Add metric for missing SyncAggregate
       // Must return signature as G2_POINT_AT_INFINITY when participating bits are empty
@@ -110,7 +122,7 @@ export class SyncContributionAndProofPool {
    * We don't want to prune by clock slot in case there's a long period of skipped slots.
    */
   prune(headSlot: Slot): void {
-    pruneBySlot(this.bestContributionBySubnetRootSlot, headSlot, SLOTS_RETAINED);
+    pruneBySlot(this.bestContributionBySubnetRootBySlot, headSlot, SLOTS_RETAINED);
     this.lowestPermissibleSlot = Math.max(headSlot - SLOTS_RETAINED, 0);
   }
 }

packages/lodestar/src/metrics/metrics/lodestar.ts

@@ -564,6 +564,44 @@ export function createLodestarMetrics(
       }),
     },
 
+    opPool: {
+      // Note: Current opPool metrics only track current size.
+      //       I don't believe tracking total add() count is relevant since that can be seen with gossip ACCEPTs
+      aggregatedAttestationPoolSize: register.gauge({
+        name: "lodestar_oppool_aggregated_attestation_pool_size",
+        help: "Current size of the AggregatedAttestationPool = total attestations",
+      }),
+      /** This metric helps view how many overlapping attestations we keep per data on average */
+      aggregatedAttestationPoolUniqueData: register.gauge({
+        name: "lodestar_oppool_aggregated_attestation_pool_unique_data_count",
+        help: "Current size of the AggregatedAttestationPool = total attestations unique by data",
+      }),
+      attestationPoolSize: register.gauge({
+        name: "lodestar_oppool_attestation_pool_size",
+        help: "Current size of the AttestationPool = total attestations unique by data and slot",
+      }),
+      attesterSlashingPoolSize: register.gauge({
+        name: "lodestar_oppool_attester_slashing_pool_size",
+        help: "Current size of the AttesterSlashingPool",
+      }),
+      proposerSlashingPoolSize: register.gauge({
+        name: "lodestar_oppool_proposer_slashing_pool_size",
+        help: "Current size of the ProposerSlashingPool",
+      }),
+      voluntaryExitPoolSize: register.gauge({
+        name: "lodestar_oppool_voluntary_exit_pool_size",
+        help: "Current size of the VoluntaryExitPool",
+      }),
+      syncCommitteeMessagePoolSize: register.gauge({
+        name: "lodestar_oppool_sync_committee_message_pool_size",
+        help: "Current size of the SyncCommitteeMessagePool unique by slot subnet and block root",
+      }),
+      syncContributionAndProofPoolSize: register.gauge({
+        name: "lodestar_oppool_sync_contribution_and_proof_pool_pool_size",
+        help: "Current size of the SyncContributionAndProofPool unique by slot subnet and block root",
+      }),
+    },
+
     // Validator monitoring
 
     validatorMonitor: {