proposals: add processing metrics for major steps
## Motivation
add information about time spent on proposal processing
MODIFY
proposals/handler.go
proposals/handler.go
@@ -4,6 +4,7 @@ import (
"context"
"errors"
"fmt"
+ "time"
"github.com/prometheus/client_golang/prometheus"
@@ -13,7 +14,6 @@ import (
"github.com/spacemeshos/go-spacemesh/log"
"github.com/spacemeshos/go-spacemesh/p2p"
"github.com/spacemeshos/go-spacemesh/p2p/pubsub"
- "github.com/spacemeshos/go-spacemesh/proposals/metrics"
"github.com/spacemeshos/go-spacemesh/sql/ballots"
"github.com/spacemeshos/go-spacemesh/sql/blocks"
"github.com/spacemeshos/go-spacemesh/sql/identities"
@@ -110,8 +110,7 @@ func NewHandler(cdb *datastore.CachedDB, f system.Fetcher, bc system.BeaconColle
// HandleProposal is the gossip receiver for Proposal.
func (h *Handler) HandleProposal(ctx context.Context, peer p2p.Peer, msg []byte) pubsub.ValidationResult {
- newCtx := log.WithNewRequestID(ctx)
- err := h.handleProposalData(newCtx, msg, peer)
+ err := h.handleProposalData(ctx, msg, peer)
switch {
case err == nil:
return pubsub.ValidationAccept
@@ -120,7 +119,7 @@ func (h *Handler) HandleProposal(ctx context.Context, peer p2p.Peer, msg []byte)
case errors.Is(err, errKnownProposal) == true:
return pubsub.ValidationIgnore
default:
- h.logger.WithContext(newCtx).With().Error("failed to process proposal gossip", log.Err(err))
+ h.logger.WithContext(ctx).With().Error("failed to process proposal gossip", log.Err(err))
return pubsub.ValidationIgnore
}
}
@@ -131,6 +130,7 @@ func (h *Handler) HandleBallotData(ctx context.Context, data []byte) error {
logger := h.logger.WithContext(newCtx)
var b types.Ballot
+ t0 := time.Now()
if err := codec.Decode(data, &b); err != nil {
logger.With().Error("malformed ballot", log.Err(err))
return errMalformedData
@@ -141,9 +141,12 @@ func (h *Handler) HandleBallotData(ctx context.Context, data []byte) error {
logger.With().Error("failed to initialize ballot", log.Err(err))
return errInitialize
}
+ ballotDuration.WithLabelValues(decodeInit).Observe(float64(time.Since(t0)))
bidHash := b.ID().AsHash32()
+ t1 := time.Now()
h.fetcher.AddPeersFromHash(bidHash, collectHashes(&b))
+ ballotDuration.WithLabelValues(peerHashes).Observe(float64(time.Since(t1)))
logger = logger.WithFields(b.ID(), b.LayerIndex)
if err := h.processBallot(ctx, &b, logger); err != nil {
@@ -174,6 +177,7 @@ func (h *Handler) HandleProposalData(ctx context.Context, data []byte) error {
func (h *Handler) handleProposalData(ctx context.Context, data []byte, peer p2p.Peer) error {
logger := h.logger.WithContext(ctx)
+ t0 := time.Now()
var p types.Proposal
if err := codec.Decode(data, &p); err != nil {
logger.With().Error("malformed proposal", log.Err(err))
@@ -185,9 +189,10 @@ func (h *Handler) handleProposalData(ctx context.Context, data []byte, peer p2p.
logger.With().Warning("failed to initialize proposal", log.Err(err))
return errInitialize
}
+ proposalDuration.WithLabelValues(decodeInit).Observe(float64(time.Since(t0)))
logger = logger.WithFields(p.ID(), p.Ballot.ID(), p.LayerIndex)
-
+ t1 := time.Now()
if has, err := proposals.Has(h.cdb, p.ID()); err != nil {
logger.With().Error("failed to look up proposal", p.ID(), log.Err(err))
return fmt.Errorf("lookup proposal %v: %w", p.ID(), err)
@@ -195,39 +200,48 @@ func (h *Handler) handleProposalData(ctx context.Context, data []byte, peer p2p.
logger.Info("known proposal")
return fmt.Errorf("%w proposal %s", errKnownProposal, p.ID())
}
+ proposalDuration.WithLabelValues(dbLookup).Observe(float64(time.Since(t1)))
- logger.With().Info("new proposal", p.ID())
-
+ logger.With().Info("new proposal", p.ID(), log.Int("num_txs", len(p.TxIDs)))
+ t2 := time.Now()
h.fetcher.RegisterPeerHashes(peer, collectHashes(&p.Ballot))
+ proposalDuration.WithLabelValues(peerHashes).Observe(float64(time.Since(t2)))
+ t3 := time.Now()
if err := h.processBallot(ctx, &p.Ballot, logger); err != nil {
logger.With().Warning("failed to process ballot", log.Err(err))
return err
}
+ proposalDuration.WithLabelValues(ballot).Observe(float64(time.Since(t3)))
+ t4 := time.Now()
if err := h.checkTransactions(ctx, &p); err != nil {
logger.With().Warning("failed to fetch proposal TXs", log.Err(err))
return err
}
+ proposalDuration.WithLabelValues(fetchTXs).Observe(float64(time.Since(t4)))
logger.With().Debug("proposal is syntactically valid")
-
+ t5 := time.Now()
if err := proposals.Add(h.cdb, &p); err != nil {
logger.With().Error("failed to save proposal", log.Err(err))
return fmt.Errorf("save proposal: %w", err)
}
-
+ proposalDuration.WithLabelValues(dbSave).Observe(float64(time.Since(t5)))
logger.With().Info("added proposal to database", p.ID())
+ t6 := time.Now()
if err := h.mesh.AddTXsFromProposal(ctx, p.LayerIndex, p.ID(), p.TxIDs); err != nil {
return fmt.Errorf("proposal add TXs: %w", err)
}
+ proposalDuration.WithLabelValues(linkTxs).Observe(float64(time.Since(t6)))
reportProposalMetrics(&p)
return nil
}
func (h *Handler) processBallot(ctx context.Context, b *types.Ballot, logger log.Log) error {
+ t0 := time.Now()
if has, err := ballots.Has(h.cdb, b.ID()); err != nil {
h.logger.WithContext(ctx).With().Error("failed to look up ballot", b.ID(), log.Err(err))
return fmt.Errorf("lookup ballot %v: %w", b.ID(), err)
@@ -235,6 +249,7 @@ func (h *Handler) processBallot(ctx context.Context, b *types.Ballot, logger log
logger.Debug("known ballot", b.ID())
return nil
}
+ ballotDuration.WithLabelValues(dbLookup).Observe(float64(time.Since(t0)))
logger.With().Info("new ballot", log.Inline(b))
@@ -243,36 +258,46 @@ func (h *Handler) processBallot(ctx context.Context, b *types.Ballot, logger log
return fmt.Errorf("syntactic-check ballot: %w", err)
}
+ t1 := time.Now()
if err := ballots.Add(h.cdb, b); err != nil {
return fmt.Errorf("save ballot: %w", err)
}
+ ballotDuration.WithLabelValues(dbSave).Observe(float64(time.Since(t1)))
- reportBallotMetrics(b)
+ reportVotesMetrics(b)
return nil
}
func (h *Handler) checkBallotSyntacticValidity(ctx context.Context, b *types.Ballot) error {
h.logger.WithContext(ctx).With().Debug("checking proposal syntactic validity")
+ t0 := time.Now()
if err := h.checkBallotDataIntegrity(b); err != nil {
h.logger.WithContext(ctx).With().Warning("ballot integrity check failed", log.Err(err))
return err
}
+ ballotDuration.WithLabelValues(dataCheck).Observe(float64(time.Since(t0)))
+ t1 := time.Now()
if err := h.checkBallotDataAvailability(ctx, b); err != nil {
h.logger.WithContext(ctx).With().Warning("ballot data availability check failed", log.Err(err))
return err
}
+ ballotDuration.WithLabelValues(fetchRef).Observe(float64(time.Since(t1)))
+ t2 := time.Now()
if err := h.checkVotesConsistency(ctx, b); err != nil {
h.logger.WithContext(ctx).With().Warning("ballot votes consistency check failed", log.Err(err))
return err
}
+ ballotDuration.WithLabelValues(votes).Observe(float64(time.Since(t2)))
+ t3 := time.Now()
if eligible, err := h.validator.CheckEligibility(ctx, b); err != nil || !eligible {
h.logger.WithContext(ctx).With().Warning("ballot eligibility check failed", log.Err(err))
return errNotEligible
}
+ ballotDuration.WithLabelValues(eligible).Observe(float64(time.Since(t3)))
h.logger.WithContext(ctx).With().Debug("ballot is syntactically valid")
return nil
@@ -429,7 +454,6 @@ func (h *Handler) checkBallotDataAvailability(ctx context.Context, b *types.Ball
return fmt.Errorf("fetch blocks: %w", err)
}
}
-
return nil
}
@@ -463,12 +487,12 @@ func (h *Handler) checkTransactions(ctx context.Context, p *types.Proposal) erro
}
func reportProposalMetrics(p *types.Proposal) {
- metrics.ProposalSize.WithLabelValues().Observe(float64(len(p.Bytes())))
- metrics.NumTxsInProposal.WithLabelValues().Observe(float64(len(p.TxIDs)))
+ proposalSize.WithLabelValues().Observe(float64(len(p.Bytes())))
+ numTxsInProposal.WithLabelValues().Observe(float64(len(p.TxIDs)))
}
-func reportBallotMetrics(b *types.Ballot) {
- metrics.NumBlocksInException.With(prometheus.Labels{metrics.DiffTypeLabel: metrics.DiffTypeAgainst}).Observe(float64(len(b.Votes.Against)))
- metrics.NumBlocksInException.With(prometheus.Labels{metrics.DiffTypeLabel: metrics.DiffTypeFor}).Observe(float64(len(b.Votes.Support)))
- metrics.NumBlocksInException.With(prometheus.Labels{metrics.DiffTypeLabel: metrics.DiffTypeNeutral}).Observe(float64(len(b.Votes.Abstain)))
+func reportVotesMetrics(b *types.Ballot) {
+ numBlocksInException.With(prometheus.Labels{diffTypeLabel: diffTypeAgainst}).Observe(float64(len(b.Votes.Against)))
+ numBlocksInException.With(prometheus.Labels{diffTypeLabel: diffTypeFor}).Observe(float64(len(b.Votes.Support)))
+ numBlocksInException.With(prometheus.Labels{diffTypeLabel: diffTypeNeutral}).Observe(float64(len(b.Votes.Abstain)))
}

ADD
proposals/metrics.go
+package proposals
+
+import (
+ "github.com/prometheus/client_golang/prometheus"
+
+ "github.com/spacemeshos/go-spacemesh/metrics"
+)
+
+const (
+ subsystem = "proposals"
+
+ // diffTypeLabel is the label name for the type of different opinions with base block.
+ diffTypeLabel = "diff_type"
+
+ // diffTypeFor is the label value for different opinions with base ballot when voting for blocks.
+ diffTypeFor = "diff_for"
+ // diffTypeAgainst is the label value for different opinions with base ballot when voting against blocks.
+ diffTypeAgainst = "diff_against"
+ // diffTypeNeutral is the label value for different opinions with base ballot when voting neutral on blocks.
+ diffTypeNeutral = "diff_neutral"
+)
+
+// proposalSize records average size of proposals.
+var proposalSize = metrics.NewHistogramWithBuckets(
+ "proposal_size",
+ subsystem,
+ "proposal size in bytes",
+ []string{},
+ prometheus.ExponentialBuckets(100, 2, 8),
+)
+
+// numTxsInProposal records average number of transactions in a proposal.
+var numTxsInProposal = metrics.NewHistogramWithBuckets(
+ "num_txs_in_proposal",
+ subsystem,
+ "number of transactions in proposal",
+ []string{},
+ prometheus.ExponentialBuckets(1, 2, 8),
+)
+
+// numBlocksInException records the number of blocks encoded in a ballot for a given exception type (against, for, and neutral).
+var numBlocksInException = metrics.NewHistogramWithBuckets(
+ "num_blocks_in_exception",
+ subsystem,
+ "number of blocks in an exception list",
+ []string{
+ diffTypeLabel,
+ },
+ prometheus.ExponentialBuckets(1, 2, 8),
+)
+
+const (
+ // labels for each step of proposal/ballot processing.
+ decodeInit = "decode"
+ dbLookup = "db_r"
+ peerHashes = "hashes"
+ ballot = "ballot"
+ fetchTXs = "txs"
+ dbSave = "db_w"
+ linkTxs = "link"
+
+ dataCheck = "data" // check data integrity
+ fetchRef = "fetch" // check referenced ballots/atxs/blocks are available
+ votes = "votes" // check votes are consistent
+ eligible = "elig" // check ballot eligibility
+)
+
+var (
+ proposalDuration = metrics.NewHistogramWithBuckets(
+ "proposal_duration",
+ subsystem,
+ "Duration in ns for processing a proposal",
+ []string{"step"},
+ prometheus.ExponentialBuckets(1_000_000, 4, 10),
+ )
+ ballotDuration = metrics.NewHistogramWithBuckets(
+ "ballot_duration",
+ subsystem,
+ "Duration in ns for processing a ballot",
+ []string{"step"},
+ prometheus.ExponentialBuckets(1_000_000, 4, 10),
+ )
+)

DELETE
proposals/metrics/prometheus.go
-package metrics
-
-import (
- "github.com/prometheus/client_golang/prometheus"
-
- "github.com/spacemeshos/go-spacemesh/metrics"
-)
-
-const (
- subsystem = "proposals"
-
- // DiffTypeLabel is the label name for the type of different opinions with base block.
- DiffTypeLabel = "diff_type"
-
- // DiffTypeFor is the label value for different opinions with base ballot when voting for blocks.
- DiffTypeFor = "diff_for"
- // DiffTypeAgainst is the label value for different opinions with base ballot when voting against blocks.
- DiffTypeAgainst = "diff_against"
- // DiffTypeNeutral is the label value for different opinions with base ballot when voting neutral on blocks.
- DiffTypeNeutral = "diff_neutral"
-)
-
-// ProposalSize records average size of proposals.
-var ProposalSize = metrics.NewHistogramWithBuckets(
- "proposal_size",
- subsystem,
- "proposal size in bytes",
- []string{},
- prometheus.ExponentialBuckets(100, 2, 8),
-)
-
-// NumTxsInProposal records average number of transactions in a proposal.
-var NumTxsInProposal = metrics.NewHistogramWithBuckets(
- "num_txs_in_proposal",
- subsystem,
- "number of transactions in proposal",
- []string{},
- prometheus.ExponentialBuckets(1, 2, 8),
-)
-
-// NumBlocksInException records the number of blocks encoded in a ballot for a given exception type (against, for, and neutral).
-var NumBlocksInException = metrics.NewHistogramWithBuckets(
- "num_blocks_in_exception",
- subsystem,
- "number of blocks in an exception list",
- []string{
- DiffTypeLabel,
- },
- prometheus.ExponentialBuckets(1, 2, 8),
-)

MODIFY
systest/Makefile
systest/Makefile
@@ -32,7 +32,7 @@ run:
--labels="testid=$(test_id)" \
--image-pull-policy=IfNotPresent -- \
tests -test.v -test.timeout=0 -test.run=$(test_name) -namespace=$(namespace) -clusters=$(clusters) -storage=$(storage) -size=$(size) -image=$(smesher_image) -level=$(level) -node-selector=$(node_selector) -bootstrap=$(bootstrap) -keep=$(keep) -testid=$(test_id) -labels=$(label)
- -@kubectl wait --timeout=5s --for=condition=ready pod/$(test_pod_name)
+ -@kubectl wait --timeout=10s --for=condition=ready pod/$(test_pod_name)
@kubectl logs $(test_pod_name) -f
.PHONY: clean

proposals: add processing metrics for major steps (#3369)

## Motivation
add information about time spent on proposal processing