Add aggregation metrics (#12417)

beacon-chain/operations/attestations/metrics.go

@@ -30,6 +30,20 @@ var (
 		Name: "expired_block_atts_total",
 		Help: "The number of expired and deleted block attestations in the pool.",
 	})
+	batchForkChoiceAttsT1 = promauto.NewHistogram(
+		prometheus.HistogramOpts{
+			Name:    "aggregate_attestations_t1",
+			Help:    "Captures times of attestation aggregation in milliseconds during the first interval per slot",
+			Buckets: []float64{100, 200, 500, 1000, 1500, 2000, 2500, 3500},
+		},
+	)
+	batchForkChoiceAttsT2 = promauto.NewHistogram(
+		prometheus.HistogramOpts{
+			Name:    "aggregate_attestations_t2",
+			Help:    "Captures times of attestation aggregation in milliseconds during the second interval per slot",
+			Buckets: []float64{10, 40, 100, 200, 600},
+		},
+	)
 )
 
 func (s *Service) updateMetrics() {

beacon-chain/operations/attestations/prepare_forkchoice.go

@@ -31,15 +31,19 @@ func (s *Service) prepareForkChoiceAtts() {
 			break
 		}
 	}
-	ticker := slots.NewSlotTickerWithIntervals(time.Unix(int64(s.genesisTime), 0), intervals)
+	ticker := slots.NewSlotTickerWithIntervals(time.Unix(int64(s.genesisTime), 0), intervals[:])
 	for {
 		select {
 		case <-ticker.C():
 			t := time.Now()
 			if err := s.batchForkChoiceAtts(s.ctx); err != nil {
 				log.WithError(err).Error("Could not prepare attestations for fork choice")
 			}
-			log.WithField("latency", time.Since(t).Milliseconds()).Debug("batched forkchoice attestations")
+			if slots.TimeIntoSlot(s.genesisTime) < intervals[1] {
+				batchForkChoiceAttsT1.Observe(float64(time.Since(t).Milliseconds()))
+			} else if slots.TimeIntoSlot(s.genesisTime) < intervals[2] {
+				batchForkChoiceAttsT2.Observe(float64(time.Since(t).Milliseconds()))
+			}
 		case <-s.ctx.Done():
 			log.Debug("Context closed, exiting routine")
 			return

config/features/config.go

@@ -74,7 +74,7 @@ type Flags struct {
 	KeystoreImportDebounceInterval time.Duration
 
 	// AggregateIntervals specifies the time durations at which we aggregate attestations preparing for forkchoice.
-	AggregateIntervals []time.Duration
+	AggregateIntervals [3]time.Duration
 }
 
 var featureConfig *Flags
@@ -222,7 +222,7 @@ func ConfigureBeaconChain(ctx *cli.Context) error {
 		logEnabled(disableBuildBlockParallel)
 		cfg.BuildBlockParallel = false
 	}
-	cfg.AggregateIntervals = []time.Duration{aggregateFirstInterval.Value, aggregateSecondInterval.Value, aggregateThirdInterval.Value}
+	cfg.AggregateIntervals = [3]time.Duration{aggregateFirstInterval.Value, aggregateSecondInterval.Value, aggregateThirdInterval.Value}
 	Init(cfg)
 	return nil
 }

time/slots/slottime.go

@@ -252,3 +252,9 @@ func SecondsSinceSlotStart(s primitives.Slot, genesisTime, timeStamp uint64) (ui
 	}
 	return timeStamp - genesisTime - uint64(s)*params.BeaconConfig().SecondsPerSlot, nil
 }
+
+// TimeIntoSlot returns the time duration elapsed between the current time and
+// the start of the current slot
+func TimeIntoSlot(genesisTime uint64) time.Duration {
+	return time.Since(StartTime(genesisTime, CurrentSlot(genesisTime)))
+}

time/slots/slottime_test.go

@@ -563,3 +563,9 @@ func TestDuration(t *testing.T) {
 		})
 	}
 }
+
+func TestTimeIntoSlot(t *testing.T) {
+	genesisTime := uint64(time.Now().Add(-37 * time.Second).Unix())
+	require.Equal(t, true, TimeIntoSlot(genesisTime) > 900*time.Millisecond)
+	require.Equal(t, true, TimeIntoSlot(genesisTime) < 3000*time.Millisecond)
+}