Refactor attributes

ssvlabs · Dec 9, 2024 · f117ef6 · f117ef6
1 parent 5335f0f
commit f117ef6
Show file tree

Hide file tree

Showing 13 changed files with 60 additions and 54 deletions.
diff --git a/message/validation/observability.go b/message/validation/observability.go
@@ -60,22 +60,18 @@ func reasonAttribute(reason string) attribute.KeyValue {
 	return attribute.String("ssv.p2p.message.validation.discard_reason", reason)
 }
 
-func roleAttribute(role types.RunnerRole) attribute.KeyValue {
-	return attribute.String("ssv.runner.role", role.String())
-}
-
 func recordMessage(ctx context.Context) {
 	messageValidationsCounter.Add(ctx, 1)
 }
 
 func recordAcceptedMessage(ctx context.Context, role types.RunnerRole) {
-	messageValidationsAcceptedCounter.Add(ctx, 1, metric.WithAttributes(roleAttribute(role)))
+	messageValidationsAcceptedCounter.Add(ctx, 1, metric.WithAttributes(observability.RunnerRoleAttribute(role)))
 }
 
 func recordRejectedMessage(ctx context.Context, role types.RunnerRole, reason string) {
-	messageValidationsRejectedCounter.Add(ctx, 1, metric.WithAttributes(reasonAttribute(reason), roleAttribute(role)))
+	messageValidationsRejectedCounter.Add(ctx, 1, metric.WithAttributes(reasonAttribute(reason), observability.RunnerRoleAttribute(role)))
 }
 
 func recordIgnoredMessage(ctx context.Context, role types.RunnerRole, reason string) {
-	messageValidationsIgnoredCounter.Add(ctx, 1, metric.WithAttributes(reasonAttribute(reason), roleAttribute(role)))
+	messageValidationsIgnoredCounter.Add(ctx, 1, metric.WithAttributes(reasonAttribute(reason), observability.RunnerRoleAttribute(role)))
 }
diff --git a/observability/attributes.go b/observability/attributes.go
@@ -7,6 +7,9 @@ import (
 )
 
 func BeaconRoleAttribute(role types.BeaconRole) attribute.KeyValue {
-	const eventNameAttrName = "ssv.beacon.role"
-	return attribute.String(eventNameAttrName, role.String())
+	return attribute.String("ssv.beacon.role", role.String())
+}
+
+func RunnerRoleAttribute(role types.RunnerRole) attribute.KeyValue {
+	return attribute.String("ssv.runner.role", role.String())
 }
diff --git a/operator/duties/scheduler.go b/operator/duties/scheduler.go
@@ -26,6 +26,7 @@ import (
 	"github.com/ssvlabs/ssv/logging/fields"
 	"github.com/ssvlabs/ssv/network"
 	"github.com/ssvlabs/ssv/networkconfig"
+	"github.com/ssvlabs/ssv/observability"
 	"github.com/ssvlabs/ssv/operator/duties/dutystore"
 	"github.com/ssvlabs/ssv/operator/slotticker"
 	"github.com/ssvlabs/ssv/protocol/v2/types"
@@ -377,8 +378,8 @@ func (s *Scheduler) ExecuteDuties(ctx context.Context, logger *zap.Logger, dutie
 			span.AddEvent("late duty execution",
 				trace.WithAttributes(
 					attribute.Int64("slot_delay_ms", slotDelay.Milliseconds()),
-					attribute.String("ssv.beacon.role", duty.Type.String()),
-					attribute.String("ssv.runner.role", duty.RunnerRole().String())))
+					observability.BeaconRoleAttribute(duty.Type),
+					observability.RunnerRoleAttribute(duty.RunnerRole())))
 		}
 
 		slotDelayHistogram.Record(ctx, slotDelay.Seconds())
@@ -417,7 +418,9 @@ func (s *Scheduler) ExecuteCommitteeDuties(ctx context.Context, logger *zap.Logg
 		slotDelayHistogram.Record(ctx, slotDelay.Seconds())
 		go func() {
 			s.waitOneThirdOrValidBlock(duty.Slot)
-			committeeDutiesExecutedCounter.Add(ctx, 1, metric.WithAttributes(attribute.String("ssv.runner.role", committee.duty.RunnerRole().String())))
+			committeeDutiesExecutedCounter.Add(ctx, 1,
+				metric.WithAttributes(
+					observability.RunnerRoleAttribute(committee.duty.RunnerRole())))
 			s.dutyExecutor.ExecuteCommitteeDuty(ctx, logger, committee.id, duty)
 		}()
 	}

diff --git a/operator/validator/controller.go b/operator/validator/controller.go
@@ -29,6 +29,7 @@ import (
 	"github.com/ssvlabs/ssv/message/validation"
 	"github.com/ssvlabs/ssv/network"
 	"github.com/ssvlabs/ssv/networkconfig"
+	"github.com/ssvlabs/ssv/observability"
 	operatordatastore "github.com/ssvlabs/ssv/operator/datastore"
 	"github.com/ssvlabs/ssv/operator/duties"
 	nodestorage "github.com/ssvlabs/ssv/operator/storage"
@@ -638,8 +639,8 @@ func (c *controller) ExecuteDuty(ctx context.Context, logger *zap.Logger, duty *
 		fmt.Sprintf("%s.execute_duty", observabilityNamespace),
 		trace.WithAttributes(
 			attribute.Int("ssv.validator.duty.slot", int(duty.Slot)),
-			attribute.String("ssv.beacon.role", duty.Type.String()),
-			attribute.String("ssv.runner.role", duty.RunnerRole().String()),
+			observability.BeaconRoleAttribute(duty.Type),
+			observability.RunnerRoleAttribute(duty.RunnerRole()),
 			attribute.String("ssv.validator.pubkey", duty.PubKey.String()),
 		))
 	defer span.End()
@@ -676,7 +677,7 @@ func (c *controller) ExecuteCommitteeDuty(ctx context.Context, logger *zap.Logge
 		trace.WithAttributes(
 			attribute.Int("ssv.validator.duty.slot", int(duty.Slot)),
 			attribute.String("ssv.comittee.id", hex.EncodeToString(committeeID[:])),
-			attribute.String("ssv.runner.role", duty.RunnerRole().String()),
+			observability.RunnerRoleAttribute(duty.RunnerRole()),
 		))
 	defer span.End()
 

diff --git a/protocol/v2/ssv/runner/aggregator.go b/protocol/v2/ssv/runner/aggregator.go
@@ -15,6 +15,7 @@ import (
 	specqbft "github.com/ssvlabs/ssv-spec/qbft"
 	spectypes "github.com/ssvlabs/ssv-spec/types"
 	"github.com/ssvlabs/ssv/logging/fields"
+	"github.com/ssvlabs/ssv/observability"
 	"github.com/ssvlabs/ssv/protocol/v2/blockchain/beacon"
 	"github.com/ssvlabs/ssv/protocol/v2/qbft/controller"
 	ssvtypes "github.com/ssvlabs/ssv/protocol/v2/types"
@@ -88,7 +89,7 @@ func (r *AggregatorRunner) ProcessPreConsensus(ctx context.Context, logger *zap.
 	}
 
 	r.measurements.EndPreConsensus()
-	preConsensusDurationHistogram.Record(ctx, r.measurements.PreConsensusTime().Seconds(), metric.WithAttributes(roleAttribute(spectypes.RoleAggregator)))
+	preConsensusDurationHistogram.Record(ctx, r.measurements.PreConsensusTime().Seconds(), metric.WithAttributes(observability.RunnerRoleAttribute(spectypes.RoleAggregator)))
 
 	// only 1 root, verified by basePreConsensusMsgProcessing
 	root := roots[0]
@@ -145,7 +146,7 @@ func (r *AggregatorRunner) ProcessConsensus(ctx context.Context, logger *zap.Log
 	}
 
 	r.measurements.EndConsensus()
-	consensusDurationHistogram.Record(ctx, r.measurements.ConsensusTime().Seconds(), metric.WithAttributes(roleAttribute(spectypes.RoleAggregator)))
+	consensusDurationHistogram.Record(ctx, r.measurements.ConsensusTime().Seconds(), metric.WithAttributes(observability.RunnerRoleAttribute(spectypes.RoleAggregator)))
 
 	r.measurements.StartPostConsensus()
 
@@ -209,7 +210,7 @@ func (r *AggregatorRunner) ProcessPostConsensus(ctx context.Context, logger *zap
 	}
 
 	r.measurements.EndPostConsensus()
-	postConsensusDurationHistogram.Record(ctx, r.measurements.PostConsensusTime().Seconds(), metric.WithAttributes(roleAttribute(spectypes.RoleAggregator)))
+	postConsensusDurationHistogram.Record(ctx, r.measurements.PostConsensusTime().Seconds(), metric.WithAttributes(observability.RunnerRoleAttribute(spectypes.RoleAggregator)))
 
 	for _, root := range roots {
 		sig, err := r.GetState().ReconstructBeaconSig(r.GetState().PostConsensusContainer, root, r.GetShare().ValidatorPubKey[:], r.GetShare().ValidatorIndex)
@@ -241,7 +242,7 @@ func (r *AggregatorRunner) ProcessPostConsensus(ctx context.Context, logger *zap
 		start := time.Now()
 
 		if err := r.GetBeaconNode().SubmitSignedAggregateSelectionProof(msg); err != nil {
-			failedSubmissionCounter.Add(ctx, 1, metric.WithAttributes(roleAttribute(spectypes.RoleAggregator)))
+			failedSubmissionCounter.Add(ctx, 1, metric.WithAttributes(observability.RunnerRoleAttribute(spectypes.RoleAggregator)))
 
 			logger.Error("❌ could not submit to Beacon chain reconstructed contribution and proof",
 				fields.SubmissionTime(time.Since(start)),
@@ -250,8 +251,8 @@ func (r *AggregatorRunner) ProcessPostConsensus(ctx context.Context, logger *zap
 		}
 		r.measurements.EndDutyFlow()
 		dutyDurationHistogram.Record(ctx, r.measurements.DutyDurationTime().Seconds(),
-			metric.WithAttributes(roleAttribute(spectypes.RoleAggregator), roundAttribute(r.GetState().RunningInstance.State.Round)))
-		submissionCounter.Add(ctx, 1, metric.WithAttributes(roleAttribute(spectypes.RoleAggregator)))
+			metric.WithAttributes(observability.RunnerRoleAttribute(spectypes.RoleAggregator), roundAttribute(r.GetState().RunningInstance.State.Round)))
+		submissionCounter.Add(ctx, 1, metric.WithAttributes(observability.RunnerRoleAttribute(spectypes.RoleAggregator)))
 
 		logger.Debug("✅ successful submitted aggregate",
 			fields.SubmissionTime(time.Since(start)),

diff --git a/protocol/v2/ssv/runner/committee.go b/protocol/v2/ssv/runner/committee.go
@@ -23,6 +23,7 @@ import (
 	spectypes "github.com/ssvlabs/ssv-spec/types"
 	"github.com/ssvlabs/ssv/logging/fields"
 	"github.com/ssvlabs/ssv/networkconfig"
+	"github.com/ssvlabs/ssv/observability"
 	"github.com/ssvlabs/ssv/protocol/v2/blockchain/beacon"
 	"github.com/ssvlabs/ssv/protocol/v2/qbft/controller"
 	ssvtypes "github.com/ssvlabs/ssv/protocol/v2/types"
@@ -88,7 +89,7 @@ func (cr *CommitteeRunner) StartNewDuty(ctx context.Context, logger *zap.Logger,
 	ctx, span := tracer.Start(ctx,
 		fmt.Sprintf("%s.runner.start_new_duty", observabilityNamespace),
 		trace.WithAttributes(
-			roleAttribute(duty.RunnerRole()),
+			observability.RunnerRoleAttribute(duty.RunnerRole()),
 			attribute.Int64("ssv.validator.quorum", int64(quorum)),
 			attribute.Int64("ssv.validator.duty.slot", int64(duty.DutySlot()))))
 	defer span.End()
@@ -224,7 +225,7 @@ func (cr *CommitteeRunner) ProcessConsensus(ctx context.Context, logger *zap.Log
 		trace.WithAttributes(
 			attribute.String("ssv.validator.msg_id", msg.SSVMessage.MsgID.String()),
 			attribute.Int64("ssv.validator.msg_type", int64(msg.SSVMessage.MsgType)),
-			roleAttribute(msg.SSVMessage.GetID().GetRoleType()),
+			observability.RunnerRoleAttribute(msg.SSVMessage.GetID().GetRoleType()),
 		))
 	defer span.End()
 
@@ -242,7 +243,7 @@ func (cr *CommitteeRunner) ProcessConsensus(ctx context.Context, logger *zap.Log
 	}
 
 	cr.measurements.EndConsensus()
-	consensusDurationHistogram.Record(ctx, cr.measurements.ConsensusTime().Seconds(), metric.WithAttributes(roleAttribute(spectypes.RoleCommittee)))
+	consensusDurationHistogram.Record(ctx, cr.measurements.ConsensusTime().Seconds(), metric.WithAttributes(observability.RunnerRoleAttribute(spectypes.RoleCommittee)))
 
 	cr.measurements.StartPostConsensus()
 	// decided means consensus is done
@@ -260,7 +261,7 @@ func (cr *CommitteeRunner) ProcessConsensus(ctx context.Context, logger *zap.Log
 		span.SetAttributes(
 			attribute.Int64("ssv.validator.index", int64(duty.ValidatorIndex)),
 			attribute.String("ssv.validator.pubkey", duty.PubKey.String()),
-			attribute.String("ssv.beacon.role", duty.Type.String()),
+			observability.BeaconRoleAttribute(duty.Type),
 		)
 		if err := cr.DutyGuard.ValidDuty(duty.Type, spectypes.ValidatorPK(duty.PubKey), duty.DutySlot()); err != nil {
 			eventMsg := "duty is no longer valid"
@@ -435,7 +436,7 @@ func (cr *CommitteeRunner) ProcessPostConsensus(ctx context.Context, logger *zap
 		}
 		eventMsg := "found validators for root"
 		span.AddEvent(eventMsg, trace.WithAttributes(
-			attribute.String("ssv.beacon.role", role.String()),
+			observability.BeaconRoleAttribute(role),
 			attribute.String("ssv.validator.duty.root", hex.EncodeToString(root[:])),
 		))
 		logger.Debug(eventMsg,
@@ -516,7 +517,7 @@ func (cr *CommitteeRunner) ProcessPostConsensus(ctx context.Context, logger *zap
 	}
 
 	cr.measurements.EndPostConsensus()
-	postConsensusDurationHistogram.Record(ctx, cr.measurements.PostConsensusTime().Seconds(), metric.WithAttributes(roleAttribute(spectypes.RoleCommittee)))
+	postConsensusDurationHistogram.Record(ctx, cr.measurements.PostConsensusTime().Seconds(), metric.WithAttributes(observability.RunnerRoleAttribute(spectypes.RoleCommittee)))
 
 	logger = logger.With(fields.PostConsensusTime(cr.measurements.PostConsensusTime()))
 
@@ -777,7 +778,7 @@ func (cr *CommitteeRunner) executeDuty(ctx context.Context, logger *zap.Logger,
 	ctx, span := tracer.Start(ctx,
 		fmt.Sprintf("%s.runner.execute_duty", observabilityNamespace),
 		trace.WithAttributes(
-			roleAttribute(duty.RunnerRole()),
+			observability.RunnerRoleAttribute(duty.RunnerRole()),
 			attribute.Int64("ssv.validator.duty.slot", int64(duty.DutySlot()))))
 	defer span.End()
 

diff --git a/protocol/v2/ssv/runner/observability.go b/protocol/v2/ssv/runner/observability.go
@@ -9,7 +9,6 @@ import (
 	"go.opentelemetry.io/otel/metric"
 
 	"github.com/ssvlabs/ssv-spec/qbft"
-	"github.com/ssvlabs/ssv-spec/types"
 	"github.com/ssvlabs/ssv/observability"
 )
 
@@ -67,10 +66,6 @@ func metricName(name string) string {
 	return fmt.Sprintf("%s.%s", observabilityNamespace, name)
 }
 
-func roleAttribute(role types.RunnerRole) attribute.KeyValue {
-	return attribute.String("ssv.runner.role", role.String())
-}
-
 func roundAttribute(qbftRound qbft.Round) attribute.KeyValue {
 	var round int64
 	r := uint64(qbftRound)

diff --git a/protocol/v2/ssv/runner/proposer.go b/protocol/v2/ssv/runner/proposer.go
@@ -22,6 +22,7 @@ import (
 	specqbft "github.com/ssvlabs/ssv-spec/qbft"
 	spectypes "github.com/ssvlabs/ssv-spec/types"
 	"github.com/ssvlabs/ssv/logging/fields"
+	"github.com/ssvlabs/ssv/observability"
 	"github.com/ssvlabs/ssv/protocol/v2/blockchain/beacon"
 	"github.com/ssvlabs/ssv/protocol/v2/qbft/controller"
 	ssvtypes "github.com/ssvlabs/ssv/protocol/v2/types"
@@ -103,7 +104,7 @@ func (r *ProposerRunner) ProcessPreConsensus(ctx context.Context, logger *zap.Lo
 	}
 
 	r.measurements.EndPreConsensus()
-	preConsensusDurationHistogram.Record(ctx, r.measurements.PreConsensusTime().Seconds(), metric.WithAttributes(roleAttribute(spectypes.RoleProposer)))
+	preConsensusDurationHistogram.Record(ctx, r.measurements.PreConsensusTime().Seconds(), metric.WithAttributes(observability.RunnerRoleAttribute(spectypes.RoleProposer)))
 
 	// only 1 root, verified in basePreConsensusMsgProcessing
 	root := roots[0]
@@ -167,7 +168,7 @@ func (r *ProposerRunner) ProcessConsensus(ctx context.Context, logger *zap.Logge
 	}
 
 	r.measurements.EndConsensus()
-	consensusDurationHistogram.Record(ctx, r.measurements.ConsensusTime().Seconds(), metric.WithAttributes(roleAttribute(spectypes.RoleProposer)))
+	consensusDurationHistogram.Record(ctx, r.measurements.ConsensusTime().Seconds(), metric.WithAttributes(observability.RunnerRoleAttribute(spectypes.RoleProposer)))
 
 	r.measurements.StartPostConsensus()
 
@@ -255,7 +256,7 @@ func (r *ProposerRunner) ProcessPostConsensus(ctx context.Context, logger *zap.L
 		copy(specSig[:], sig)
 
 		r.measurements.EndPostConsensus()
-		postConsensusDurationHistogram.Record(ctx, r.measurements.PostConsensusTime().Seconds(), metric.WithAttributes(roleAttribute(spectypes.RoleProposer)))
+		postConsensusDurationHistogram.Record(ctx, r.measurements.PostConsensusTime().Seconds(), metric.WithAttributes(observability.RunnerRoleAttribute(spectypes.RoleProposer)))
 
 		logger.Debug("🧩 reconstructed partial post consensus signatures proposer",
 			zap.Uint64s("signers", getPostConsensusProposerSigners(r.GetState(), root)),
@@ -294,7 +295,7 @@ func (r *ProposerRunner) ProcessPostConsensus(ctx context.Context, logger *zap.L
 			)
 
 			if err := r.GetBeaconNode().SubmitBlindedBeaconBlock(vBlindedBlk, specSig); err != nil {
-				failedSubmissionCounter.Add(ctx, 1, metric.WithAttributes(roleAttribute(spectypes.RoleProposer)))
+				failedSubmissionCounter.Add(ctx, 1, metric.WithAttributes(observability.RunnerRoleAttribute(spectypes.RoleProposer)))
 				logger.Error("❌ could not submit blinded Beacon block",
 					fields.SubmissionTime(time.Since(start)),
 					zap.Error(err))
@@ -312,7 +313,7 @@ func (r *ProposerRunner) ProcessPostConsensus(ctx context.Context, logger *zap.L
 			)
 
 			if err := r.GetBeaconNode().SubmitBeaconBlock(vBlk, specSig); err != nil {
-				failedSubmissionCounter.Add(ctx, 1, metric.WithAttributes(roleAttribute(spectypes.RoleProposer)))
+				failedSubmissionCounter.Add(ctx, 1, metric.WithAttributes(observability.RunnerRoleAttribute(spectypes.RoleProposer)))
 				logger.Error("❌ could not submit Beacon block",
 					fields.SubmissionTime(time.Since(start)),
 					zap.Error(err))
@@ -322,8 +323,8 @@ func (r *ProposerRunner) ProcessPostConsensus(ctx context.Context, logger *zap.L
 
 		r.measurements.EndDutyFlow()
 		dutyDurationHistogram.Record(ctx, r.measurements.DutyDurationTime().Seconds(),
-			metric.WithAttributes(roleAttribute(spectypes.RoleProposer), roundAttribute(r.GetState().RunningInstance.State.Round)))
-		submissionCounter.Add(ctx, 1, metric.WithAttributes(roleAttribute(spectypes.RoleProposer)))
+			metric.WithAttributes(observability.RunnerRoleAttribute(spectypes.RoleProposer), roundAttribute(r.GetState().RunningInstance.State.Round)))
+		submissionCounter.Add(ctx, 1, metric.WithAttributes(observability.RunnerRoleAttribute(spectypes.RoleProposer)))
 
 		logger.Info("✅ successfully submitted block proposal",
 			fields.Slot(validatorConsensusData.Duty.Slot),

diff --git a/protocol/v2/ssv/runner/runner.go b/protocol/v2/ssv/runner/runner.go
@@ -16,6 +16,7 @@ import (
 
 	specqbft "github.com/ssvlabs/ssv-spec/qbft"
 	spectypes "github.com/ssvlabs/ssv-spec/types"
+	"github.com/ssvlabs/ssv/observability"
 	"github.com/ssvlabs/ssv/protocol/v2/blockchain/beacon"
 	"github.com/ssvlabs/ssv/protocol/v2/qbft/controller"
 	"github.com/ssvlabs/ssv/protocol/v2/ssv"
@@ -150,7 +151,7 @@ func (b *BaseRunner) baseStartNewDuty(ctx context.Context, logger *zap.Logger, r
 	ctx, span := tracer.Start(ctx,
 		fmt.Sprintf("%s.base_runner.start_new_duty", observabilityNamespace),
 		trace.WithAttributes(
-			roleAttribute(duty.RunnerRole()),
+			observability.RunnerRoleAttribute(duty.RunnerRole()),
 			attribute.Int64("ssv.validator.quorum", int64(quorum)),
 			attribute.Int64("ssv.validator.duty.slot", int64(duty.DutySlot()))))
 	defer span.End()
@@ -314,7 +315,7 @@ func (b *BaseRunner) decide(ctx context.Context, logger *zap.Logger, runner Runn
 	ctx, span := tracer.Start(ctx,
 		fmt.Sprintf("%s.base_runner.decide", observabilityNamespace),
 		trace.WithAttributes(
-			roleAttribute(runner.GetBaseRunner().RunnerRoleType),
+			observability.RunnerRoleAttribute(runner.GetBaseRunner().RunnerRoleType),
 			attribute.Int64("ssv.validator.duty.slot", int64(slot))))
 	defer span.End()