Log attestation's journey published by local validators (#3534)

* Log attestation journey

* Fix lint and log

* New metric: missed attestations vs big inclusion distance

* Correct inclusionDistance in failed attestation log

* Track attestation publish delay second in metric and log

* Remove Math.floor() to calculate metric exactly

* Track seen time at the api side

* Add comments for sent peers bucket values

packages/lodestar/src/api/impl/beacon/pool/index.ts

@@ -8,7 +8,6 @@ import {validateGossipProposerSlashing} from "../../../../chain/validation/propo
 import {validateGossipVoluntaryExit} from "../../../../chain/validation/voluntaryExit";
 import {validateSyncCommitteeSigOnly} from "../../../../chain/validation/syncCommittee";
 import {ApiModules} from "../../types";
-import {OpSource} from "../../../../metrics/validatorMonitor";
 import {toHexString} from "@chainsafe/ssz";
 import {AttestationError, GossipAction, SyncCommitteeError} from "../../../../chain/errors";
 
@@ -51,12 +50,9 @@ export function getBeaconPoolApi({
           try {
             const {indexedAttestation, subnet} = await validateGossipAttestation(chain, attestation, null);
 
-            metrics?.registerUnaggregatedAttestation(OpSource.api, seenTimestampSec, indexedAttestation);
-
-            await Promise.all([
-              network.gossip.publishBeaconAttestation(attestation, subnet),
-              chain.attestationPool.add(attestation),
-            ]);
+            chain.attestationPool.add(attestation);
+            const sentPeers = await network.gossip.publishBeaconAttestation(attestation, subnet);
+            metrics?.submitUnaggregatedAttestation(seenTimestampSec, indexedAttestation, subnet, sentPeers);
           } catch (e) {
             errors.push(e as Error);
             logger.error(

packages/lodestar/src/api/impl/validator/index.ts

@@ -25,7 +25,6 @@ import {toGraffitiBuffer} from "../../../util/graffiti";
 import {ApiError, NodeIsSyncing} from "../errors";
 import {validateSyncCommitteeGossipContributionAndProof} from "../../../chain/validation/syncCommitteeContributionAndProof";
 import {CommitteeSubscription} from "../../../network/subnets";
-import {OpSource} from "../../../metrics/validatorMonitor";
 import {computeSubnetForCommitteesAtSlot, getPubkeysForIndices} from "./utils";
 import {ApiModules} from "../types";
 import {RegenCaller} from "../../../chain/regen";
@@ -436,21 +435,13 @@ export function getValidatorApi({chain, config, logger, metrics, network, sync}:
               signedAggregateAndProof
             );
 
-            metrics?.registerAggregatedAttestation(
-              OpSource.api,
-              seenTimestampSec,
-              signedAggregateAndProof,
-              indexedAttestation
+            chain.aggregatedAttestationPool.add(
+              signedAggregateAndProof.message.aggregate,
+              indexedAttestation.attestingIndices,
+              committeeIndices
             );
-
-            await Promise.all([
-              chain.aggregatedAttestationPool.add(
-                signedAggregateAndProof.message.aggregate,
-                indexedAttestation.attestingIndices,
-                committeeIndices
-              ),
-              network.gossip.publishBeaconAggregateAndProof(signedAggregateAndProof),
-            ]);
+            const sentPeers = await network.gossip.publishBeaconAggregateAndProof(signedAggregateAndProof);
+            metrics?.submitAggregatedAttestation(seenTimestampSec, indexedAttestation, sentPeers);
           } catch (e) {
             if (e instanceof AttestationError && e.type.code === AttestationErrorCode.AGGREGATOR_ALREADY_KNOWN) {
               logger.debug("Ignoring known signedAggregateAndProof");

packages/lodestar/src/metrics/metrics.ts

@@ -1,6 +1,7 @@
 /**
  * @module metrics
  */
+import {ILogger} from "@chainsafe/lodestar-utils";
 import {BeaconStateAllForks, getCurrentSlot} from "@chainsafe/lodestar-beacon-state-transition";
 import {IChainForkConfig} from "@chainsafe/lodestar-config";
 import {collectDefaultMetrics, Counter, Metric, Registry} from "prom-client";
@@ -18,14 +19,15 @@ export function createMetrics(
   opts: IMetricsOptions,
   config: IChainForkConfig,
   anchorState: BeaconStateAllForks,
+  logger: ILogger,
   externalRegistries: Registry[] = []
 ): IMetrics {
   const register = new RegistryMetricCreator();
   const beacon = createBeaconMetrics(register);
   const lodestar = createLodestarMetrics(register, opts.metadata, anchorState);
 
   const genesisTime = anchorState.genesisTime;
-  const validatorMonitor = createValidatorMonitor(lodestar, config, genesisTime);
+  const validatorMonitor = createValidatorMonitor(lodestar, config, genesisTime, logger);
   // Register a single collect() function to run all validatorMonitor metrics
   lodestar.validatorMonitor.validatorsTotal.addCollect(() => {
     const clockSlot = getCurrentSlot(config, genesisTime);

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

@@ -567,12 +567,23 @@ export function createLodestarMetrics(
       }),
       prevEpochOnChainAttesterHit: register.gauge<"index">({
         name: "validator_monitor_prev_epoch_on_chain_attester_hit_total",
-        help: "Incremented if the validator is flagged as a previous epoch attester during per epoch processing",
+        help: "Incremented if validator's submitted attestation is included in some blocks",
         labelNames: ["index"],
       }),
       prevEpochOnChainAttesterMiss: register.gauge<"index">({
         name: "validator_monitor_prev_epoch_on_chain_attester_miss_total",
-        help: "Incremented if the validator is not flagged as a previous epoch attester during per epoch processing",
+        help: "Incremented if validator's submitted attestation is not included in any blocks",
+        labelNames: ["index"],
+      }),
+      prevEpochOnChainSourceAttesterHit: register.gauge<"index">({
+        name: "validator_monitor_prev_epoch_on_chain_source_attester_hit_total",
+        help: "Incremented if the validator is flagged as a previous epoch source attester during per epoch processing",
+        labelNames: ["index"],
+      }),
+      prevEpochOnChainSourceAttesterMiss: register.gauge<"index">({
+        name: "validator_monitor_prev_epoch_on_chain_source_attester_miss_total",
+        help:
+          "Incremented if the validator is not flagged as a previous epoch source attester during per epoch processing",
         labelNames: ["index"],
       }),
       prevEpochOnChainHeadAttesterHit: register.gauge<"index">({
@@ -669,6 +680,15 @@ export function createLodestarMetrics(
         labelNames: ["index", "src"],
         buckets: [0.1, 1],
       }),
+      unaggregatedAttestationSubmittedSentPeers: register.histogram<"index">({
+        name: "validator_monitor_unaggregated_attestation_submited_sent_peers_count",
+        help: "Number of peers that an unaggregated attestation sent to",
+        labelNames: ["index"],
+        // as of Apr 2022, most of the time we sent to >30 peers per attestations
+        // these bucket values just base on that fact to get equal range
+        // refine if we want more reasonable values
+        buckets: [0, 10, 20, 30],
+      }),
       aggregatedAttestationTotal: register.gauge<"index" | "src">({
         name: "validator_monitor_aggregated_attestation_total",
         help: "Number of aggregated attestations seen",

packages/lodestar/src/metrics/validatorMonitor.ts

@@ -4,6 +4,7 @@ import {
   IAttesterStatus,
   parseAttesterFlags,
 } from "@chainsafe/lodestar-beacon-state-transition";
+import {ILogger} from "@chainsafe/lodestar-utils";
 import {allForks} from "@chainsafe/lodestar-beacon-state-transition";
 import {IChainForkConfig} from "@chainsafe/lodestar-config";
 import {MIN_ATTESTATION_INCLUSION_DELAY, SLOTS_PER_EPOCH} from "@chainsafe/lodestar-params";
@@ -26,13 +27,19 @@ export interface IValidatorMonitor {
   registerLocalValidator(index: number): void;
   registerValidatorStatuses(currentEpoch: Epoch, statuses: IAttesterStatus[], balances?: number[]): void;
   registerBeaconBlock(src: OpSource, seenTimestampSec: Seconds, block: allForks.BeaconBlock): void;
-  registerUnaggregatedAttestation(
-    src: OpSource,
-    seenTimestampSec: Seconds,
-    indexedAttestation: IndexedAttestation
+  submitUnaggregatedAttestation(
+    seenTimestampSec: number,
+    indexedAttestation: IndexedAttestation,
+    subnet: number,
+    sentPeers: number
+  ): void;
+  registerGossipUnaggregatedAttestation(seenTimestampSec: Seconds, indexedAttestation: IndexedAttestation): void;
+  submitAggregatedAttestation(
+    seenTimestampSec: number,
+    indexedAttestation: IndexedAttestation,
+    sentPeers: number
   ): void;
-  registerAggregatedAttestation(
-    src: OpSource,
+  registerGossipAggregatedAttestation(
     seenTimestampSec: Seconds,
     signedAggregateAndProof: SignedAggregateAndProof,
     indexedAttestation: IndexedAttestation
@@ -166,7 +173,8 @@ type MonitoredValidator = {
 export function createValidatorMonitor(
   metrics: ILodestarMetrics,
   config: IChainForkConfig,
-  genesisTime: number
+  genesisTime: number,
+  logger: ILogger
 ): IValidatorMonitor {
   /** The validators that require additional monitoring. */
   const validators = new Map<ValidatorIndex, MonitoredValidator>();
@@ -201,20 +209,24 @@ export function createValidatorMonitor(
         }
 
         const summary = statusToSummary(status);
+        let failedAttestation = false;
 
         if (summary.isPrevSourceAttester) {
-          metrics.validatorMonitor.prevEpochOnChainAttesterHit.inc({index});
+          metrics.validatorMonitor.prevEpochOnChainSourceAttesterHit.inc({index});
         } else {
-          metrics.validatorMonitor.prevEpochOnChainAttesterMiss.inc({index});
+          failedAttestation = true;
+          metrics.validatorMonitor.prevEpochOnChainSourceAttesterMiss.inc({index});
         }
         if (summary.isPrevHeadAttester) {
           metrics.validatorMonitor.prevEpochOnChainHeadAttesterHit.inc({index});
         } else {
+          failedAttestation = true;
           metrics.validatorMonitor.prevEpochOnChainHeadAttesterMiss.inc({index});
         }
         if (summary.isPrevTargetAttester) {
           metrics.validatorMonitor.prevEpochOnChainTargetAttesterHit.inc({index});
         } else {
+          failedAttestation = true;
           metrics.validatorMonitor.prevEpochOnChainTargetAttesterMiss.inc({index});
         }
 
@@ -236,12 +248,27 @@ export function createValidatorMonitor(
 
         if (inclusionDistance !== null) {
           metrics.validatorMonitor.prevEpochOnChainInclusionDistance.set({index}, inclusionDistance);
+          metrics.validatorMonitor.prevEpochOnChainAttesterHit.inc({index});
+        } else {
+          metrics.validatorMonitor.prevEpochOnChainAttesterMiss.inc({index});
         }
 
         const balance = balances && balances[index];
         if (balance !== undefined) {
           metrics.validatorMonitor.prevEpochOnChainBalance.set({index}, balance);
         }
+
+        if (failedAttestation) {
+          logger.debug("Failed attestation in previous epoch", {
+            validatorIndex: index,
+            prevEpoch: currentEpoch - 1,
+            isPrevSourceAttester: summary.isPrevSourceAttester,
+            isPrevHeadAttester: summary.isPrevHeadAttester,
+            isPrevTargetAttester: summary.isPrevTargetAttester,
+            // inclusionDistance is not available in summary since altair
+            inclusionDistance,
+          });
+        }
       }
     },
 
@@ -257,7 +284,29 @@ export function createValidatorMonitor(
       }
     },
 
-    registerUnaggregatedAttestation(src, seenTimestampSec, indexedAttestation) {
+    submitUnaggregatedAttestation(seenTimestampSec, indexedAttestation, subnet, sentPeers) {
+      const data = indexedAttestation.data;
+      // Returns the duration between when the attestation `data` could be produced (1/3rd through the slot) and `seenTimestamp`.
+      const delaySec = seenTimestampSec - (genesisTime + (data.slot + 1 / 3) * config.SECONDS_PER_SLOT);
+      for (const index of indexedAttestation.attestingIndices) {
+        const validator = validators.get(index);
+        if (validator) {
+          metrics.validatorMonitor.unaggregatedAttestationSubmittedSentPeers.observe({index}, sentPeers);
+          metrics.validatorMonitor.unaggregatedAttestationDelaySeconds.observe({src: OpSource.api, index}, delaySec);
+          logger.debug("Local validator published unaggregated attestation", {
+            validatorIndex: validator.index,
+            slot: data.slot,
+            committeeIndex: data.index,
+            subnet,
+            sentPeers,
+            delaySec,
+          });
+        }
+      }
+    },
+
+    registerGossipUnaggregatedAttestation(seenTimestampSec, indexedAttestation) {
+      const src = OpSource.gossip;
       const data = indexedAttestation.data;
       const epoch = computeEpochAtSlot(data.slot);
       // Returns the duration between when the attestation `data` could be produced (1/3rd through the slot) and `seenTimestamp`.
@@ -276,7 +325,28 @@ export function createValidatorMonitor(
       }
     },
 
-    registerAggregatedAttestation(src, seenTimestampSec, signedAggregateAndProof, indexedAttestation) {
+    submitAggregatedAttestation(seenTimestampSec, indexedAttestation, sentPeers) {
+      const data = indexedAttestation.data;
+      // Returns the duration between when a `AggregateAndproof` with `data` could be produced (2/3rd through the slot) and `seenTimestamp`.
+      const delaySec = seenTimestampSec - (genesisTime + (data.slot + 2 / 3) * config.SECONDS_PER_SLOT);
+
+      for (const index of indexedAttestation.attestingIndices) {
+        const validator = validators.get(index);
+        if (validator) {
+          metrics.validatorMonitor.aggregatedAttestationDelaySeconds.observe({src: OpSource.api, index}, delaySec);
+          logger.debug("Local validator published aggregated attestation", {
+            validatorIndex: validator.index,
+            slot: data.slot,
+            committeeIndex: data.index,
+            sentPeers,
+            delaySec,
+          });
+        }
+      }
+    },
+
+    registerGossipAggregatedAttestation(seenTimestampSec, signedAggregateAndProof, indexedAttestation) {
+      const src = OpSource.gossip;
       const data = indexedAttestation.data;
       const epoch = computeEpochAtSlot(data.slot);
       // Returns the duration between when a `AggregateAndproof` with `data` could be produced (2/3rd through the slot) and `seenTimestamp`.
@@ -302,6 +372,11 @@ export function createValidatorMonitor(
           withEpochSummary(validator, epoch, (summary) => {
             summary.attestationAggregateIncusions += 1;
           });
+          logger.debug("Local validator attestation is included in AggregatedAndProof", {
+            validatorIndex: validator.index,
+            slot: data.slot,
+            committeeIndex: data.index,
+          });
         }
       }
     },
@@ -336,6 +411,14 @@ export function createValidatorMonitor(
             }
             summary.attestationCorrectHead = correctHead;
           });
+
+          logger.debug("Local validator attestation is included in block", {
+            validatorIndex: validator.index,
+            slot: data.slot,
+            committeeIndex: data.index,
+            inclusionDistance,
+            correctHead,
+          });
         }
       }
     },

packages/lodestar/src/network/gossip/gossipsub.ts

@@ -148,12 +148,13 @@ export class Eth2Gossipsub extends Gossipsub {
   /**
    * Publish a `GossipObject` on a `GossipTopic`
    */
-  async publishObject<K extends GossipType>(topic: GossipTopicMap[K], object: GossipTypeMap[K]): Promise<void> {
+  async publishObject<K extends GossipType>(topic: GossipTopicMap[K], object: GossipTypeMap[K]): Promise<number> {
     const topicStr = this.getGossipTopicString(topic);
     const sszType = getGossipSSZType(topic);
     const messageData = (sszType.serialize as (object: GossipTypeMap[GossipType]) => Uint8Array)(object);
     const sentPeers = await this.publish(topicStr, messageData);
     this.logger.verbose("Publish to topic", {topic: topicStr, sentPeers});
+    return sentPeers;
   }
 
   /**
@@ -182,17 +183,17 @@ export class Eth2Gossipsub extends Gossipsub {
     await this.publishObject<GossipType.beacon_block>({type: GossipType.beacon_block, fork}, signedBlock);
   }
 
-  async publishBeaconAggregateAndProof(aggregateAndProof: phase0.SignedAggregateAndProof): Promise<void> {
+  async publishBeaconAggregateAndProof(aggregateAndProof: phase0.SignedAggregateAndProof): Promise<number> {
     const fork = this.config.getForkName(aggregateAndProof.message.aggregate.data.slot);
-    await this.publishObject<GossipType.beacon_aggregate_and_proof>(
+    return await this.publishObject<GossipType.beacon_aggregate_and_proof>(
       {type: GossipType.beacon_aggregate_and_proof, fork},
       aggregateAndProof
     );
   }
 
-  async publishBeaconAttestation(attestation: phase0.Attestation, subnet: number): Promise<void> {
+  async publishBeaconAttestation(attestation: phase0.Attestation, subnet: number): Promise<number> {
     const fork = this.config.getForkName(attestation.data.slot);
-    await this.publishObject<GossipType.beacon_attestation>(
+    return await this.publishObject<GossipType.beacon_attestation>(
       {type: GossipType.beacon_attestation, fork, subnet},
       attestation
     );

packages/lodestar/src/network/gossip/handlers/index.ts

@@ -161,12 +161,7 @@ export function getGossipHandlers(modules: ValidatorFnsModules, options: GossipH
 
       // Handler
       const {indexedAttestation, committeeIndices} = validationResult;
-      metrics?.registerAggregatedAttestation(
-        OpSource.gossip,
-        seenTimestampSec,
-        signedAggregateAndProof,
-        indexedAttestation
-      );
+      metrics?.registerGossipAggregatedAttestation(seenTimestampSec, signedAggregateAndProof, indexedAttestation);
       const aggregatedAttestation = signedAggregateAndProof.message.aggregate;
 
       chain.aggregatedAttestationPool.add(
@@ -206,7 +201,7 @@ export function getGossipHandlers(modules: ValidatorFnsModules, options: GossipH
 
       // Handler
       const {indexedAttestation} = validationResult;
-      metrics?.registerUnaggregatedAttestation(OpSource.gossip, seenTimestampSec, indexedAttestation);
+      metrics?.registerGossipUnaggregatedAttestation(seenTimestampSec, indexedAttestation);
 
       // Node may be subscribe to extra subnets (long-lived random subnets). For those, validate the messages
       // but don't import them, to save CPU and RAM

packages/lodestar/src/node/nodejs.ts

@@ -128,7 +128,9 @@ export class BeaconNode {
     // start db if not already started
     await db.start();
 
-    const metrics = opts.metrics.enabled ? createMetrics(opts.metrics, config, anchorState, metricsRegistries) : null;
+    const metrics = opts.metrics.enabled
+      ? createMetrics(opts.metrics, config, anchorState, logger.child({module: "VMON"}), metricsRegistries)
+      : null;
     if (metrics) {
       initBeaconMetrics(metrics, anchorState);
     }

packages/lodestar/test/unit/metrics/utils.ts

@@ -1,8 +1,10 @@
 import {config} from "@chainsafe/lodestar-config/default";
 import {ssz} from "@chainsafe/lodestar-types";
+import {WinstonLogger} from "@chainsafe/lodestar-utils";
 import {createMetrics, IMetrics} from "../../../src/metrics";
 
 export function createMetricsTest(): IMetrics {
   const state = ssz.phase0.BeaconState.defaultViewDU();
-  return createMetrics({enabled: true, timeout: 12000}, config, state);
+  const logger = new WinstonLogger();
+  return createMetrics({enabled: true, timeout: 12000}, config, state, logger);
 }