From f117ef6702fab3f12cd3d7be45626fa8d2fba318 Mon Sep 17 00:00:00 2001 From: oleg-ssvlabs Date: Mon, 9 Dec 2024 12:23:52 +0100 Subject: [PATCH] Refactor attributes --- message/validation/observability.go | 10 +++------- observability/attributes.go | 7 +++++-- operator/duties/scheduler.go | 9 ++++++--- operator/validator/controller.go | 7 ++++--- protocol/v2/ssv/runner/aggregator.go | 13 +++++++------ protocol/v2/ssv/runner/committee.go | 15 ++++++++------- protocol/v2/ssv/runner/observability.go | 5 ----- protocol/v2/ssv/runner/proposer.go | 15 ++++++++------- protocol/v2/ssv/runner/runner.go | 5 +++-- .../v2/ssv/runner/sync_committee_aggregator.go | 13 +++++++------ protocol/v2/ssv/validator/committee.go | 9 +++++---- protocol/v2/ssv/validator/duty_executer.go | 3 ++- protocol/v2/ssv/validator/validator.go | 3 ++- 13 files changed, 60 insertions(+), 54 deletions(-) diff --git a/message/validation/observability.go b/message/validation/observability.go index 238e2b5f46..d88a56c8ea 100644 --- 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 index 393aeeda6c..8d2cf0499e 100644 --- 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 index 6af8979f34..1011306375 100644 --- 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 index 6819bb5203..dc4d90d6fa 100644 --- 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 index e7dde78f5f..b1b1142a19 100644 --- 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 index 0fafe43660..a5824d2835 100644 --- 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 index 2908b1724c..d9cd83b59a 100644 --- 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 index 4b8c8e71cd..136fca94a3 100644 --- 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 index fe6545d79f..5d0d245834 100644 --- 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() diff --git a/protocol/v2/ssv/runner/sync_committee_aggregator.go b/protocol/v2/ssv/runner/sync_committee_aggregator.go index 50c162b7ee..50e246359f 100644 --- a/protocol/v2/ssv/runner/sync_committee_aggregator.go +++ b/protocol/v2/ssv/runner/sync_committee_aggregator.go @@ -17,6 +17,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" @@ -89,7 +90,7 @@ func (r *SyncCommitteeAggregatorRunner) ProcessPreConsensus(ctx context.Context, } r.measurements.EndPreConsensus() - preConsensusDurationHistogram.Record(ctx, r.measurements.PreConsensusTime().Seconds(), metric.WithAttributes(roleAttribute(spectypes.RoleSyncCommitteeContribution))) + preConsensusDurationHistogram.Record(ctx, r.measurements.PreConsensusTime().Seconds(), metric.WithAttributes(observability.RunnerRoleAttribute(spectypes.RoleSyncCommitteeContribution))) // collect selection proofs and subnets var ( @@ -174,7 +175,7 @@ func (r *SyncCommitteeAggregatorRunner) ProcessConsensus(ctx context.Context, lo } r.measurements.EndConsensus() - consensusDurationHistogram.Record(ctx, r.measurements.ConsensusTime().Seconds(), metric.WithAttributes(roleAttribute(spectypes.RoleSyncCommitteeContribution))) + consensusDurationHistogram.Record(ctx, r.measurements.ConsensusTime().Seconds(), metric.WithAttributes(observability.RunnerRoleAttribute(spectypes.RoleSyncCommitteeContribution))) r.measurements.StartPostConsensus() @@ -246,7 +247,7 @@ func (r *SyncCommitteeAggregatorRunner) ProcessPostConsensus(ctx context.Context } r.measurements.EndPostConsensus() - postConsensusDurationHistogram.Record(ctx, r.measurements.PostConsensusTime().Seconds(), metric.WithAttributes(roleAttribute(spectypes.RoleSyncCommitteeContribution))) + postConsensusDurationHistogram.Record(ctx, r.measurements.PostConsensusTime().Seconds(), metric.WithAttributes(observability.RunnerRoleAttribute(spectypes.RoleSyncCommitteeContribution))) // get contributions validatorConsensusData := &spectypes.ValidatorConsensusData{} @@ -294,7 +295,7 @@ func (r *SyncCommitteeAggregatorRunner) ProcessPostConsensus(ctx context.Context } if err := r.GetBeaconNode().SubmitSignedContributionAndProof(signedContribAndProof); err != nil { - failedSubmissionCounter.Add(ctx, 1, metric.WithAttributes(roleAttribute(spectypes.RoleSyncCommitteeContribution))) + failedSubmissionCounter.Add(ctx, 1, metric.WithAttributes(observability.RunnerRoleAttribute(spectypes.RoleSyncCommitteeContribution))) logger.Error("❌ could not submit to Beacon chain reconstructed contribution and proof", fields.SubmissionTime(time.Since(start)), zap.Error(err)) @@ -303,8 +304,8 @@ func (r *SyncCommitteeAggregatorRunner) ProcessPostConsensus(ctx context.Context r.measurements.EndDutyFlow() dutyDurationHistogram.Record(ctx, r.measurements.DutyDurationTime().Seconds(), - metric.WithAttributes(roleAttribute(spectypes.RoleSyncCommitteeContribution), roundAttribute(r.GetState().RunningInstance.State.Round))) - submissionCounter.Add(ctx, 1, metric.WithAttributes(roleAttribute(spectypes.RoleSyncCommitteeContribution))) + metric.WithAttributes(observability.RunnerRoleAttribute(spectypes.RoleSyncCommitteeContribution), roundAttribute(r.GetState().RunningInstance.State.Round))) + submissionCounter.Add(ctx, 1, metric.WithAttributes(observability.RunnerRoleAttribute(spectypes.RoleSyncCommitteeContribution))) logger.Debug("✅ successfully submitted sync committee aggregator", fields.SubmissionTime(time.Since(start)), diff --git a/protocol/v2/ssv/validator/committee.go b/protocol/v2/ssv/validator/committee.go index 1bbf1a3e60..6db1616d85 100644 --- a/protocol/v2/ssv/validator/committee.go +++ b/protocol/v2/ssv/validator/committee.go @@ -18,6 +18,7 @@ import ( "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/message" "github.com/ssvlabs/ssv/protocol/v2/ssv/queue" "github.com/ssvlabs/ssv/protocol/v2/ssv/runner" @@ -99,7 +100,7 @@ func (c *Committee) StartConsumeQueue(ctx context.Context, logger *zap.Logger, d ctx, span := tracer.Start(ctx, fmt.Sprintf("%s.start_consume_queue", observabilityNamespace), trace.WithAttributes( - attribute.String("ssv.runner.role", duty.RunnerRole().String()), + observability.RunnerRoleAttribute(duty.RunnerRole()), attribute.Int("ssv.validator.duty_count", len(duty.ValidatorDuties)), attribute.Int64("ssv.validator.duty.slot", int64(duty.Slot)), )) @@ -142,7 +143,7 @@ func (c *Committee) StartDuty(ctx context.Context, logger *zap.Logger, duty *spe ctx, span := tracer.Start(ctx, fmt.Sprintf("%s.start_duty", observabilityNamespace), trace.WithAttributes( - attribute.String("ssv.runner.role", duty.RunnerRole().String()), + observability.RunnerRoleAttribute(duty.RunnerRole()), attribute.Int("ssv.validator.duty_count", len(duty.ValidatorDuties)), attribute.Int64("ssv.validator.duty.slot", int64(duty.Slot)), )) @@ -171,7 +172,7 @@ func (c *Committee) StartDuty(ctx context.Context, logger *zap.Logger, duty *spe if !exists { span.AddEvent("no share for validator duty", trace.WithAttributes( attribute.Int64("ssv.validator.index", int64(beaconDuty.ValidatorIndex)), - attribute.String("ssv.beacon.role", beaconDuty.Type.String()), + observability.BeaconRoleAttribute(beaconDuty.Type), attribute.String("ssv.validator.pubkey", beaconDuty.PubKey.String()), )) continue @@ -249,7 +250,7 @@ func (c *Committee) ProcessMessage(ctx context.Context, logger *zap.Logger, msg trace.WithAttributes( attribute.String("ssv.validator.msg_id", msg.GetID().String()), attribute.Int64("ssv.validator.msg_type", int64(msg.GetType())), - attribute.String("ssv.runner.role", msg.GetID().GetRoleType().String()), + observability.RunnerRoleAttribute(msg.GetID().GetRoleType()), )) defer span.End() diff --git a/protocol/v2/ssv/validator/duty_executer.go b/protocol/v2/ssv/validator/duty_executer.go index 9eeebfb5b4..0efe5eae6a 100644 --- a/protocol/v2/ssv/validator/duty_executer.go +++ b/protocol/v2/ssv/validator/duty_executer.go @@ -10,6 +10,7 @@ import ( "go.uber.org/zap" "github.com/ssvlabs/ssv/logging/fields" + "github.com/ssvlabs/ssv/observability" "github.com/ssvlabs/ssv/protocol/v2/types" ) @@ -49,7 +50,7 @@ func (c *Committee) OnExecuteDuty(ctx context.Context, logger *zap.Logger, msg * span.SetAttributes( attribute.Int64("ssv.validator.duty.slot", int64(executeDutyData.Duty.Slot)), - attribute.String("ssv.runner.role", executeDutyData.Duty.RunnerRole().String()), + observability.RunnerRoleAttribute(executeDutyData.Duty.RunnerRole()), attribute.Int("ssv.validator.duty_count", len(executeDutyData.Duty.ValidatorDuties)), ) span.AddEvent("start duty") diff --git a/protocol/v2/ssv/validator/validator.go b/protocol/v2/ssv/validator/validator.go index c589b00734..70aac6e805 100644 --- a/protocol/v2/ssv/validator/validator.go +++ b/protocol/v2/ssv/validator/validator.go @@ -17,6 +17,7 @@ import ( "github.com/ssvlabs/ssv/logging/fields" "github.com/ssvlabs/ssv/message/validation" "github.com/ssvlabs/ssv/networkconfig" + "github.com/ssvlabs/ssv/observability" "github.com/ssvlabs/ssv/protocol/v2/message" "github.com/ssvlabs/ssv/protocol/v2/ssv/queue" "github.com/ssvlabs/ssv/protocol/v2/ssv/runner" @@ -126,7 +127,7 @@ func (v *Validator) ProcessMessage(ctx context.Context, logger *zap.Logger, msg trace.WithAttributes( attribute.String("ssv.validator.msg_id", msg.GetID().String()), attribute.Int64("ssv.validator.msg_type", int64(msg.GetType())), - attribute.String("ssv.runner.role", msg.GetID().GetRoleType().String()), + observability.RunnerRoleAttribute(msg.GetID().GetRoleType()), )) defer span.End()