observability: restart-resilient counters + proposal-parent labels

beacon/goclient/goclient.go

@@ -212,6 +212,12 @@ func New(ctx context.Context, logger *zap.Logger, opt Options) (*GoClient, error
 		}
 	}
 
+	// Pre-register baselines for the proposal_parent counters across the configured
+	// beacon clients so that PromQL increase()/rate() return correct values per
+	// ssv.beacon.client label after restart. The registered function runs when
+	// metrics.EmitBaselines is called from startup (after observability.Initialize).
+	registerProposalParentBaselines(client.clients)
+
 	client.log.Debug("connecting")
 
 	err := client.initMultiClient(ctx)

beacon/goclient/observability.go

@@ -70,68 +70,100 @@ var (
 			metric.WithUnit("{match}"),
 			metric.WithDescription("attestation data head matched cached HeadEvent")))
 
-	attestationDataHeadCacheMissCounter = metrics.New(
+	// Sparse: only fires when cache lookup fails for an attestation; registered for
+	// baseline emission so increase()/rate() work correctly after process restart.
+	attestationDataHeadCacheMissCounter = metrics.RegisterSparseCounter(metrics.New(
 		meter.Int64Counter(
 			observability.InstrumentName(observabilityNamespace, "attestation_data.head_cache_miss"),
 			metric.WithUnit("{miss}"),
-			metric.WithDescription("head root was not cached (no verification performed)")))
+			metric.WithDescription("head root was not cached (no verification performed)"))))
 
-	attestationDataHeadMismatchCounter = metrics.New(
+	// Sparse: only fires on a head-mismatch (rare condition).
+	attestationDataHeadMismatchCounter = metrics.RegisterSparseCounter(metrics.New(
 		meter.Int64Counter(
 			observability.InstrumentName(observabilityNamespace, "attestation_data.head_mismatch"),
 			metric.WithUnit("{mismatch}"),
-			metric.WithDescription("attestation data head did not match cached HeadEvent")))
+			metric.WithDescription("attestation data head did not match cached HeadEvent"))))
 
-	attestationDataRefetchSuccessCounter = metrics.New(
+	// Sparse: only fires when a re-fetch succeeds (which only happens after a mismatch).
+	attestationDataRefetchSuccessCounter = metrics.RegisterSparseCounter(metrics.New(
 		meter.Int64Counter(
 			observability.InstrumentName(observabilityNamespace, "attestation_data.refetch_success"),
 			metric.WithUnit("{refetch}"),
-			metric.WithDescription("re-fetch got correct head after mismatch")))
+			metric.WithDescription("re-fetch got correct head after mismatch"))))
 
-	attestationDataRefetchFailedCounter = metrics.New(
+	// Sparse: only fires on re-fetch failure (which only happens after a mismatch).
+	attestationDataRefetchFailedCounter = metrics.RegisterSparseCounter(metrics.New(
 		meter.Int64Counter(
 			observability.InstrumentName(observabilityNamespace, "attestation_data.refetch_failed"),
 			metric.WithUnit("{refetch}"),
-			metric.WithDescription("re-fetch failed or timed out")))
+			metric.WithDescription("re-fetch failed or timed out"))))
 
-	attestationDataRefetchStillMismatchCounter = metrics.New(
+	// Sparse: only fires when a re-fetch returns the same stale data (likely reorg).
+	attestationDataRefetchStillMismatchCounter = metrics.RegisterSparseCounter(metrics.New(
 		meter.Int64Counter(
 			observability.InstrumentName(observabilityNamespace, "attestation_data.refetch_still_mismatch"),
 			metric.WithUnit("{refetch}"),
-			metric.WithDescription("re-fetch still had wrong head (possible reorg)")))
+			metric.WithDescription("re-fetch still had wrong head (possible reorg)"))))
 
-	attestationDataRefetchSkippedCounter = metrics.New(
+	// Sparse: only fires when re-fetch is skipped due to insufficient deadline budget.
+	attestationDataRefetchSkippedCounter = metrics.RegisterSparseCounter(metrics.New(
 		meter.Int64Counter(
 			observability.InstrumentName(observabilityNamespace, "attestation_data.refetch_skipped"),
 			metric.WithUnit("{skip}"),
-			metric.WithDescription("retry skipped due to insufficient time before deadline")))
+			metric.WithDescription("retry skipped due to insufficient time before deadline"))))
 
-	// Proposal parent verification metrics (observability only, no re-fetch)
-	proposalParentVerifyCounter = metrics.New(
+	// Proposal parent verification metrics (observability only, no re-fetch). All four are
+	// sparse — proposals occur once per slot at most, and only when this operator is the
+	// proposer for the slot. Registered for baseline emission so increase()/rate() work.
+	proposalParentVerifyCounter = metrics.RegisterSparseCounter(metrics.New(
 		meter.Int64Counter(
 			observability.InstrumentName(observabilityNamespace, "proposal.parent_verify"),
 			metric.WithUnit("{verification}"),
-			metric.WithDescription("total proposals that attempted parent root verification")))
+			metric.WithDescription("total proposals that attempted parent root verification"))))
 
-	proposalParentMatchCounter = metrics.New(
+	proposalParentMatchCounter = metrics.RegisterSparseCounter(metrics.New(
 		meter.Int64Counter(
 			observability.InstrumentName(observabilityNamespace, "proposal.parent_match"),
 			metric.WithUnit("{match}"),
-			metric.WithDescription("proposal parent root matched cached HeadEvent")))
+			metric.WithDescription("proposal parent root matched cached HeadEvent"))))
 
-	proposalParentCacheMissCounter = metrics.New(
+	proposalParentCacheMissCounter = metrics.RegisterSparseCounter(metrics.New(
 		meter.Int64Counter(
 			observability.InstrumentName(observabilityNamespace, "proposal.parent_cache_miss"),
 			metric.WithUnit("{miss}"),
-			metric.WithDescription("parent slot head was not cached (no verification performed)")))
+			metric.WithDescription("parent slot head was not cached (no verification performed)"))))
 
-	proposalParentMismatchCounter = metrics.New(
+	proposalParentMismatchCounter = metrics.RegisterSparseCounter(metrics.New(
 		meter.Int64Counter(
 			observability.InstrumentName(observabilityNamespace, "proposal.parent_mismatch"),
 			metric.WithUnit("{mismatch}"),
-			metric.WithDescription("proposal parent root did not match cached HeadEvent")))
+			metric.WithDescription("proposal parent root did not match cached HeadEvent"))))
 )
 
+// registerProposalParentBaselines pre-emits Add(ctx, 0) for each (counter, beacon_client)
+// combination so that PromQL increase()/rate() return correct values for per-client
+// queries after process restart. Without this, each labeled time series starts at the
+// first real increment with no prior sample, and Prometheus has nothing to compute a
+// delta from. Configured beacon addresses are passed in because they're only known after
+// GoClient construction.
+//
+// Invariant: `clients` must be final at call time. The closure captures the slice header,
+// so any later append to the same backing array (or a re-slice via append-that-reallocates)
+// would not be visible here. Today this is satisfied because the caller in goclient.New
+// finishes all addSingleClient calls before invoking registerProposalParentBaselines.
+func registerProposalParentBaselines(clients []Client) {
+	metrics.RegisterLabeledBaseline(func(ctx context.Context) {
+		for _, c := range clients {
+			attr := metric.WithAttributes(observability.BeaconClientAttribute(c.Address()))
+			proposalParentVerifyCounter.Add(ctx, 0, attr)
+			proposalParentMatchCounter.Add(ctx, 0, attr)
+			proposalParentCacheMissCounter.Add(ctx, 0, attr)
+			proposalParentMismatchCounter.Add(ctx, 0, attr)
+		}
+	})
+}
+
 func recordRequest(
 	ctx context.Context,
 	logger *zap.Logger,

beacon/goclient/proposer.go

@@ -22,8 +22,10 @@ import (
 	"github.com/attestantio/go-eth2-client/spec/phase0"
 	ssz "github.com/ferranbt/fastssz"
 	spectypes "github.com/ssvlabs/ssv-spec/types"
+	"go.opentelemetry.io/otel/metric"
 	"go.uber.org/zap"
 
+	"github.com/ssvlabs/ssv/observability"
 	"github.com/ssvlabs/ssv/observability/log/fields"
 	"github.com/ssvlabs/ssv/observability/traces"
 )
@@ -97,24 +99,29 @@ func (gc *GoClient) GetBeaconBlock(
 	copy(graffiti[:], graffitiBytes[:])
 
 	var beaconBlock *api.VersionedProposal
+	// beaconClient is the address of the BN that produced the selected proposal — used
+	// for metric labeling so per-client staleness rates can be measured.
+	var beaconClient string
 	var err error
 
 	// For single client, use direct call to avoid multi-client overhead
 	if len(gc.clients) == 1 {
-		beaconBlock, err = gc.fetchProposal(ctx, gc.clients[0], slot, sig, graffiti)
+		client := gc.clients[0]
+		beaconBlock, err = gc.fetchProposal(ctx, client, slot, sig, graffiti)
 		if err != nil {
 			return nil, nil, err
 		}
+		beaconClient = client.Address()
 	} else {
 		// For multiple clients, race them in parallel for the fastest response
-		beaconBlock, err = gc.getProposalParallel(ctx, logger, slot, sig, graffiti)
+		beaconBlock, beaconClient, err = gc.getProposalParallel(ctx, logger, slot, sig, graffiti)
 		if err != nil {
 			return nil, nil, err
 		}
 	}
 
 	// Verify proposal parent root against cached HeadEvent (observability only).
-	gc.verifyProposalParent(ctx, logger, slot, beaconBlock)
+	gc.verifyProposalParent(ctx, logger, slot, beaconBlock, beaconClient)
 
 	// Check and log if fee recipient is missing (for both single and multi-client paths)
 	feeRecipient, err := beaconBlock.FeeRecipient()
@@ -173,13 +180,17 @@ func (gc *GoClient) GetBeaconBlock(
 // The parent context (from duty runner, bounded by slot timing) serves as the hard
 // deadline. We never give up early on getting a block proposal - missing a proposal
 // is catastrophic, so we wait as long as the slot allows.
+//
+// Returns the selected proposal along with the address of the beacon client that
+// produced it — used for metric labeling so we can attribute stale-parent proposals
+// (and other diagnostics) to specific beacon clients in operators' setups.
 func (gc *GoClient) getProposalParallel(
 	ctx context.Context,
 	logger *zap.Logger,
 	slot phase0.Slot,
 	sig phase0.BLSSignature,
 	graffiti [32]byte,
-) (*api.VersionedProposal, error) {
+) (*api.VersionedProposal, string, error) {
 	// Create a context for the collection period - during this time we gather
 	// proposals from multiple beacon nodes to select the best one.
 	// After this expires, we return the best seen so far or wait for the first valid one.
@@ -274,7 +285,7 @@ collect:
 			fields.Slot(slot),
 		)
 
-		return bestProposal, nil
+		return bestProposal, bestClient, nil
 	}
 
 	logger.Debug("did not receive any valid proposals during the collection period",
@@ -304,15 +315,15 @@ collect:
 				zap.Bool("blinded", res.proposal.Blinded),
 				fields.Slot(slot),
 			)
-			return res.proposal, nil
+			return res.proposal, res.client, nil
 
 		case <-ctx.Done():
 			// Parent context canceled (duty deadline reached)
-			return nil, ctx.Err()
+			return nil, "", ctx.Err()
 		}
 	}
 
-	return nil, fmt.Errorf("all %d clients failed to get proposal for slot %d, encountered errors: %w", len(gc.clients), slot, errs)
+	return nil, "", fmt.Errorf("all %d clients failed to get proposal for slot %d, encountered errors: %w", len(gc.clients), slot, errs)
 }
 
 // scoreProposal computes a score for a beacon proposal.
@@ -326,13 +337,26 @@ func (gc *GoClient) scoreProposal(
 
 // verifyProposalParent checks the proposal's parent root against cached HeadEvent.
 // This is observability only - no re-fetch, just metrics and logging.
+// beaconClient is the address of the beacon node that produced the proposal, attached as
+// a metric label so per-client staleness rates can be measured.
 func (gc *GoClient) verifyProposalParent(
 	ctx context.Context,
 	logger *zap.Logger,
 	slot phase0.Slot,
 	proposal *api.VersionedProposal,
+	beaconClient string,
 ) {
-	proposalParentVerifyCounter.Add(ctx, 1)
+	if slot == 0 {
+		// Guards against the slot-1 uint64 underflow below. In production this branch
+		// never fires (the slot ticker is well past 0 by the time GetBeaconBlock runs),
+		// but tests with synthetic zero slots can reach here. Genesis has no parent root
+		// to verify in any case.
+		return
+	}
+
+	clientAttr := metric.WithAttributes(observability.BeaconClientAttribute(beaconClient))
+
+	proposalParentVerifyCounter.Add(ctx, 1, clientAttr)
 
 	parentRoot, err := proposal.ParentRoot()
 	if err != nil {
@@ -344,22 +368,26 @@ func (gc *GoClient) verifyProposalParent(
 	parentSlot := slot - 1
 	item := gc.headCache.Get(parentSlot)
 	if item == nil {
-		proposalParentCacheMissCounter.Add(ctx, 1)
+		proposalParentCacheMissCounter.Add(ctx, 1, clientAttr)
 		return
 	}
 	expectedRoot := item.Value()
 
 	if parentRoot == expectedRoot {
-		proposalParentMatchCounter.Add(ctx, 1)
+		proposalParentMatchCounter.Add(ctx, 1, clientAttr)
 		return
 	}
 
-	proposalParentMismatchCounter.Add(ctx, 1)
-	logger.Warn("proposal parent root mismatch detected",
+	proposalParentMismatchCounter.Add(ctx, 1, clientAttr)
+	// Logged at Info: this is observability-only with no corrective action. The metric
+	// commonly fires during normal fork resolution (cache-vs-BN drift), not staleness —
+	// Warn would create alert fatigue. Revisit if the check becomes actionable.
+	logger.Info("proposal parent root mismatch detected",
 		fields.Slot(slot),
 		zap.Uint64("parent_slot", uint64(parentSlot)),
 		zap.Stringer("expected_root", expectedRoot),
 		zap.Stringer("got_root", parentRoot),
+		fields.BeaconClient(beaconClient),
 	)
 }
 

beacon/goclient/proposer_test.go

@@ -14,6 +14,7 @@ import (
 	"time"
 
 	eth2apiv1 "github.com/attestantio/go-eth2-client/api/v1"
+	apiv1electra "github.com/attestantio/go-eth2-client/api/v1/electra"
 	"github.com/attestantio/go-eth2-client/spec"
 	"github.com/attestantio/go-eth2-client/spec/bellatrix"
 	"github.com/attestantio/go-eth2-client/spec/phase0"
@@ -128,37 +129,37 @@ func createProposalBeaconServer(t *testing.T, options beaconProposalServerOption
 	return server, serverGotRequests
 }
 
-// Create a safe proposal response using ssv-spec utilities (called once during server setup)
+// createProposalResponseSafe builds a JSON-encoded beacon proposal response with the
+// requested slot + fee recipient. It deep-clones the shared spec-testing fixture before
+// mutating: TestingBeaconBlockV / TestingBlindedBeaconBlockV return wrappers whose inner
+// *electra.BlockContents / *electra.BlindedBeaconBlock fields point at package-level
+// singletons (TestingBlockContentsElectra, TestingBlindedBeaconBlockElectra). Mutating
+// those directly leaks state across every test in the binary; SSZ round-trip gives us a
+// per-call independent copy.
 func createProposalResponseSafe(slot phase0.Slot, feeRecipient bellatrix.ExecutionAddress, blinded bool) []byte {
 	if blinded {
-		// Get a blinded block from ssv-spec testing utilities
-		versionedBlinded := spectestingutils.TestingBlindedBeaconBlockV(spec.DataVersionElectra)
-		block := versionedBlinded.ElectraBlinded
+		shared := spectestingutils.TestingBlindedBeaconBlockV(spec.DataVersionElectra).ElectraBlinded
+		sszBytes, _ := shared.MarshalSSZ()
+		block := new(apiv1electra.BlindedBeaconBlock)
+		_ = block.UnmarshalSSZ(sszBytes)
 
-		// Modify the fields we need for our test
 		block.Slot = slot
 		block.Body.ExecutionPayloadHeader.FeeRecipient = feeRecipient
 
-		// Wrap in response structure
-		response := map[string]any{
-			"data": block,
-		}
+		response := map[string]any{"data": block}
 		data, _ := json.Marshal(response)
 		return data
 	}
 
-	// Get a regular block from ssv-spec testing utilities
-	versioned := spectestingutils.TestingBeaconBlockV(spec.DataVersionElectra)
-	blockContents := versioned.Electra
+	shared := spectestingutils.TestingBeaconBlockV(spec.DataVersionElectra).Electra
+	sszBytes, _ := shared.MarshalSSZ()
+	blockContents := new(apiv1electra.BlockContents)
+	_ = blockContents.UnmarshalSSZ(sszBytes)
 
-	// Modify the fields we need for our test
 	blockContents.Block.Slot = slot
 	blockContents.Block.Body.ExecutionPayload.FeeRecipient = feeRecipient
 
-	// Wrap in response structure
-	response := map[string]any{
-		"data": blockContents,
-	}
+	response := map[string]any{"data": blockContents}
 	data, _ := json.Marshal(response)
 	return data
 }