Skip to content

Commit

Permalink
chore: merge
Browse files Browse the repository at this point in the history
  • Loading branch information
@ninabarbakadze
ninabarbakadze committed Mar 25, 2024
2 parents 74abc3b + 48abbee commit c085e04
Show file tree
Hide file tree
Showing 20 changed files with 621 additions and 100 deletions.
2 changes: 1 addition & 1 deletion .github/CODEOWNERS
View file
Validating CODEOWNERS rules …
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@
# global owners are only requested if there isn't a more specific
# codeowner specified below. For this reason, the global codeowners
# are often repeated in package-level definitions.
* @evan-forbes @cmwaters
* @evan-forbes @cmwaters @staheri14 @rach-id @ninabarbakadze @rootulp

# Overrides for tooling packages
docs/celestia-architecture @liamsi @adlerjohn
51 changes: 40 additions & 11 deletions consensus/metrics.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,8 @@ const (
type Metrics struct {
// Height of the chain.
Height metrics.Gauge
// The height when the metrics started from
StartHeight metrics.Gauge

// ValidatorLastSignedHeight of a validator.
ValidatorLastSignedHeight metrics.Gauge
Expand Down Expand Up @@ -88,6 +90,12 @@ type Metrics struct {
// timestamp and the timestamp of the latest prevote in a round where 100%
// of the voting power on the network issued prevotes.
FullPrevoteMessageDelay metrics.Gauge

// The amount of proposals that were rejected by the application.
ApplicationRejectedProposals metrics.Counter

// The amount of proposals that failed to be received in time
TimedOutProposals metrics.Counter
}

// PrometheusMetrics returns Metrics build using Prometheus client library.
Expand All @@ -105,6 +113,12 @@ func PrometheusMetrics(namespace string, labelsAndValues ...string) *Metrics {
Name: "height",
Help: "Height of the chain.",
}, labels).With(labelsAndValues...),
StartHeight: prometheus.NewGaugeFrom(stdprometheus.GaugeOpts{
Namespace: namespace,
Subsystem: MetricsSubsystem,
Name: "start_height",
Help: "Height that metrics began",
}, labels).With(labelsAndValues...),
Rounds: prometheus.NewGaugeFrom(stdprometheus.GaugeOpts{
Namespace: namespace,
Subsystem: MetricsSubsystem,
Expand Down Expand Up @@ -241,13 +255,26 @@ func PrometheusMetrics(namespace string, labelsAndValues ...string) *Metrics {
Help: "Difference in seconds between the proposal timestamp and the timestamp " +
"of the latest prevote that achieved 100% of the voting power in the prevote step.",
}, labels).With(labelsAndValues...),
ApplicationRejectedProposals: prometheus.NewCounterFrom(stdprometheus.CounterOpts{
Namespace: namespace,
Subsystem: MetricsSubsystem,
Name: "application_rejected_proposals",
Help: "Number of proposals rejected by the application",
}, labels).With(labelsAndValues...),
TimedOutProposals: prometheus.NewCounterFrom(stdprometheus.CounterOpts{
Namespace: namespace,
Subsystem: MetricsSubsystem,
Name: "timed_out_proposals",
Help: "Number of proposals that failed to be received in time",
}, labels).With(labelsAndValues...),
}
}

// NopMetrics returns no-op Metrics.
func NopMetrics() *Metrics {
return &Metrics{
Height: discard.NewGauge(),
Height: discard.NewGauge(),
StartHeight: discard.NewGauge(),

ValidatorLastSignedHeight: discard.NewGauge(),

Expand All @@ -265,16 +292,18 @@ func NopMetrics() *Metrics {

BlockIntervalSeconds: discard.NewHistogram(),

NumTxs: discard.NewGauge(),
BlockSizeBytes: discard.NewGauge(),
TotalTxs: discard.NewGauge(),
CommittedHeight: discard.NewGauge(),
FastSyncing: discard.NewGauge(),
StateSyncing: discard.NewGauge(),
BlockParts: discard.NewCounter(),
BlockGossipPartsReceived: discard.NewCounter(),
QuorumPrevoteMessageDelay: discard.NewGauge(),
FullPrevoteMessageDelay: discard.NewGauge(),
NumTxs: discard.NewGauge(),
BlockSizeBytes: discard.NewGauge(),
TotalTxs: discard.NewGauge(),
CommittedHeight: discard.NewGauge(),
FastSyncing: discard.NewGauge(),
StateSyncing: discard.NewGauge(),
BlockParts: discard.NewCounter(),
BlockGossipPartsReceived: discard.NewCounter(),
QuorumPrevoteMessageDelay: discard.NewGauge(),
FullPrevoteMessageDelay: discard.NewGauge(),
ApplicationRejectedProposals: discard.NewCounter(),
TimedOutProposals: discard.NewCounter(),
}
}

Expand Down
47 changes: 33 additions & 14 deletions consensus/reactor.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -353,8 +353,12 @@ func (conR *Reactor) ReceiveEnvelope(e p2p.Envelope) {
case *VoteMessage:
cs := conR.conS
cs.mtx.RLock()
height, valSize, lastCommitSize := cs.Height, cs.Validators.Size(), cs.LastCommit.Size()
height, round, valSize, lastCommitSize := cs.Height, cs.Round,
cs.Validators.Size(), cs.LastCommit.Size()
cs.mtx.RUnlock()

schema.WriteVote(conR.traceClient, height, round, msg.Vote, e.Src.ID(), schema.TransferTypeDownload)

ps.EnsureVoteBitArrays(height, valSize)
ps.EnsureVoteBitArrays(height-1, lastCommitSize)
ps.SetHasVote(msg.Vote)
Expand Down Expand Up @@ -763,7 +767,7 @@ OUTER_LOOP:
// Special catchup logic.
// If peer is lagging by height 1, send LastCommit.
if prs.Height != 0 && rs.Height == prs.Height+1 {
if ps.PickSendVote(rs.LastCommit) {
if conR.pickSendVoteAndTrace(rs.LastCommit, rs, ps) {
logger.Debug("Picked rs.LastCommit to send", "height", prs.Height)
continue OUTER_LOOP
}
Expand All @@ -776,8 +780,11 @@ OUTER_LOOP:
// Load the block commit for prs.Height,
// which contains precommit signatures for prs.Height.
if commit := conR.conS.blockStore.LoadBlockCommit(prs.Height); commit != nil {
if ps.PickSendVote(commit) {
vote := ps.PickSendVote(commit)
if vote != nil {
logger.Debug("Picked Catchup commit to send", "height", prs.Height)
schema.WriteVote(conR.traceClient, rs.Height, rs.Round, vote,
ps.peer.ID(), schema.TransferTypeUpload)
continue OUTER_LOOP
}
}
Expand All @@ -799,6 +806,18 @@ OUTER_LOOP:
}
}

// pickSendVoteAndTrace picks a vote to send and traces it.
// It returns true if a vote is sent.
// Note that it is a wrapper around PickSendVote with the addition of tracing the vote.
func (conR *Reactor) pickSendVoteAndTrace(votes types.VoteSetReader, rs *cstypes.RoundState, ps *PeerState) bool {
vote := ps.PickSendVote(votes)
if vote != nil { // if a vote is sent, trace it
schema.WriteVote(conR.traceClient, rs.Height, rs.Round, vote,
ps.peer.ID(), schema.TransferTypeUpload)
return true
}
return false
}
func (conR *Reactor) gossipVotesForHeight(
logger log.Logger,
rs *cstypes.RoundState,
Expand All @@ -808,15 +827,15 @@ func (conR *Reactor) gossipVotesForHeight(

// If there are lastCommits to send...
if prs.Step == cstypes.RoundStepNewHeight {
if ps.PickSendVote(rs.LastCommit) {
if conR.pickSendVoteAndTrace(rs.LastCommit, rs, ps) {
logger.Debug("Picked rs.LastCommit to send")
return true
}
}
// If there are POL prevotes to send...
if prs.Step <= cstypes.RoundStepPropose && prs.Round != -1 && prs.Round <= rs.Round && prs.ProposalPOLRound != -1 {
if polPrevotes := rs.Votes.Prevotes(prs.ProposalPOLRound); polPrevotes != nil {
if ps.PickSendVote(polPrevotes) {
if conR.pickSendVoteAndTrace(polPrevotes, rs, ps) {
logger.Debug("Picked rs.Prevotes(prs.ProposalPOLRound) to send",
"round", prs.ProposalPOLRound)
return true
Expand All @@ -825,29 +844,29 @@ func (conR *Reactor) gossipVotesForHeight(
}
// If there are prevotes to send...
if prs.Step <= cstypes.RoundStepPrevoteWait && prs.Round != -1 && prs.Round <= rs.Round {
if ps.PickSendVote(rs.Votes.Prevotes(prs.Round)) {
if conR.pickSendVoteAndTrace(rs.Votes.Prevotes(prs.Round), rs, ps) {
logger.Debug("Picked rs.Prevotes(prs.Round) to send", "round", prs.Round)
return true
}
}
// If there are precommits to send...
if prs.Step <= cstypes.RoundStepPrecommitWait && prs.Round != -1 && prs.Round <= rs.Round {
if ps.PickSendVote(rs.Votes.Precommits(prs.Round)) {
if conR.pickSendVoteAndTrace(rs.Votes.Precommits(prs.Round), rs, ps) {
logger.Debug("Picked rs.Precommits(prs.Round) to send", "round", prs.Round)
return true
}
}
// If there are prevotes to send...Needed because of validBlock mechanism
if prs.Round != -1 && prs.Round <= rs.Round {
if ps.PickSendVote(rs.Votes.Prevotes(prs.Round)) {
if conR.pickSendVoteAndTrace(rs.Votes.Prevotes(prs.Round), rs, ps) {
logger.Debug("Picked rs.Prevotes(prs.Round) to send", "round", prs.Round)
return true
}
}
// If there are POLPrevotes to send...
if prs.ProposalPOLRound != -1 {
if polPrevotes := rs.Votes.Prevotes(prs.ProposalPOLRound); polPrevotes != nil {
if ps.PickSendVote(polPrevotes) {
if conR.pickSendVoteAndTrace(polPrevotes, rs, ps) {
logger.Debug("Picked rs.Prevotes(prs.ProposalPOLRound) to send",
"round", prs.ProposalPOLRound)
return true
Expand Down Expand Up @@ -1163,8 +1182,8 @@ func (ps *PeerState) SetHasProposalBlockPart(height int64, round int32, index in
}

// PickSendVote picks a vote and sends it to the peer.
// Returns true if vote was sent.
func (ps *PeerState) PickSendVote(votes types.VoteSetReader) bool {
// Returns the vote if vote was sent. Otherwise, returns nil.
func (ps *PeerState) PickSendVote(votes types.VoteSetReader) *types.Vote {
if vote, ok := ps.PickVoteToSend(votes); ok {
ps.logger.Debug("Sending vote message", "ps", ps, "vote", vote)
if p2p.SendEnvelopeShim(ps.peer, p2p.Envelope{ //nolint: staticcheck
Expand All @@ -1174,11 +1193,11 @@ func (ps *PeerState) PickSendVote(votes types.VoteSetReader) bool {
},
}, ps.logger) {
ps.SetHasVote(vote)
return true
return vote
}
return false
return nil
}
return false
return nil
}

// PickVoteToSend picks a vote to send to the peer.
Expand Down
18 changes: 8 additions & 10 deletions consensus/replay.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -252,15 +252,10 @@ func (h *Handshaker) Handshake(proxyApp proxy.AppConns) (string, error) {
}
appHash := res.LastBlockAppHash

h.logger.Info("ABCI Handshake App Info",
"height", blockHeight,
"hash", appHash,
"software-version", res.Version,
"protocol-version", res.AppVersion,
)

// Only set the version if there is no existing state.
if h.initialState.LastBlockHeight == 0 {
appVersion := h.initialState.Version.Consensus.App
// set app version if it's not set via genesis
if h.initialState.LastBlockHeight == 0 && appVersion == 0 && res.AppVersion != 0 {
appVersion = res.AppVersion
h.initialState.Version.Consensus.App = res.AppVersion
}

Expand All @@ -271,7 +266,10 @@ func (h *Handshaker) Handshake(proxyApp proxy.AppConns) (string, error) {
}

h.logger.Info("Completed ABCI Handshake - CometBFT and App are synced",
"appHeight", blockHeight, "appHash", appHash)
"appHeight", blockHeight,
"appHash", appHash,
"appVersion", appVersion,
)

// TODO: (on restart) replay mempool

Expand Down
5 changes: 5 additions & 0 deletions consensus/state.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -315,6 +315,8 @@ func (cs *State) OnStart() error {
}
}

cs.metrics.StartHeight.Set(float64(cs.Height))

// we need the timeoutRoutine for replay so
// we don't block on the tick chan.
// NOTE: we will get a build up of garbage go routines
Expand Down Expand Up @@ -1281,6 +1283,7 @@ func (cs *State) defaultDoPrevote(height int64, round int32) {
// If ProposalBlock is nil, prevote nil.
if cs.ProposalBlock == nil {
logger.Debug("prevote step: ProposalBlock is nil")
cs.metrics.TimedOutProposals.Add(1)
cs.signAddVote(cmtproto.PrevoteType, nil, types.PartSetHeader{})
return
}
Expand All @@ -1297,12 +1300,14 @@ func (cs *State) defaultDoPrevote(height int64, round int32) {
stateMachineValidBlock, err := cs.blockExec.ProcessProposal(cs.ProposalBlock, cs.state)
if err != nil {
cs.Logger.Error("state machine returned an error when trying to process proposal block", "err", err)
return
}

// Vote nil if application invalidated the block
if !stateMachineValidBlock {
// The app says we must vote nil
logger.Error("prevote step: the application deems this block to be mustVoteNil", "err", err)
cs.metrics.ApplicationRejectedProposals.Add(1)
cs.signAddVote(cmtproto.PrevoteType, nil, types.PartSetHeader{})
return
}
Expand Down
53 changes: 19 additions & 34 deletions go.mod
View file
Original file line number Diff line number Diff line change
Expand Up @@ -5,70 +5,55 @@ go 1.21.5
require (
github.com/BurntSushi/toml v1.2.1
github.com/ChainSafe/go-schnorrkel v1.0.0
github.com/Masterminds/semver/v3 v3.2.0
github.com/Workiva/go-datastructures v1.0.53
github.com/adlio/schema v1.3.3
github.com/btcsuite/btcd/btcec/v2 v2.2.1
github.com/btcsuite/btcd/btcutil v1.1.2
github.com/bufbuild/buf v1.9.0
github.com/celestiaorg/nmt v0.20.0
github.com/cometbft/cometbft-db v0.7.0
github.com/creachadair/taskgroup v0.3.2
github.com/fortytw2/leaktest v1.3.0
github.com/go-git/go-git/v5 v5.11.0
github.com/go-kit/kit v0.12.0
github.com/go-kit/log v0.2.1
github.com/go-logfmt/logfmt v0.5.1
github.com/gofrs/uuid v4.3.0+incompatible
github.com/gogo/protobuf v1.3.2
github.com/golang/protobuf v1.5.3
github.com/golangci/golangci-lint v1.50.1
github.com/google/orderedcode v0.0.1
github.com/google/uuid v1.3.1
github.com/gorilla/websocket v1.5.0
github.com/grafana/pyroscope-go v1.0.3
github.com/gtank/merlin v0.1.1
github.com/influxdata/influxdb-client-go/v2 v2.12.3
github.com/informalsystems/tm-load-test v1.3.0
github.com/lib/pq v1.10.6
github.com/libp2p/go-buffer-pool v0.1.0
github.com/minio/highwayhash v1.0.2
github.com/ory/dockertest v3.3.5+incompatible
github.com/pkg/errors v0.9.1
github.com/prometheus/client_golang v1.14.0
github.com/pyroscope-io/otel-profiling-go v0.4.0
github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475
github.com/rs/cors v1.8.2
github.com/sasha-s/go-deadlock v0.3.1
github.com/snikch/goodman v0.0.0-20171125024755-10e37e294daa
github.com/spf13/cobra v1.6.1
github.com/spf13/viper v1.13.0
github.com/stretchr/testify v1.8.4
)

require (
github.com/google/uuid v1.3.1
golang.org/x/crypto v0.17.0
golang.org/x/net v0.19.0
google.golang.org/grpc v1.59.0
)

require (
github.com/gogo/protobuf v1.3.2
github.com/informalsystems/tm-load-test v1.3.0
)

require (
github.com/bufbuild/buf v1.9.0
github.com/creachadair/taskgroup v0.3.2
github.com/syndtr/goleveldb v1.0.1-0.20210819022825-2ae1ddf74ef7
)

require (
github.com/Masterminds/semver/v3 v3.2.0
github.com/btcsuite/btcd/btcec/v2 v2.2.1
github.com/btcsuite/btcd/btcutil v1.1.2
github.com/celestiaorg/nmt v0.20.0
github.com/cometbft/cometbft-db v0.7.0
github.com/go-git/go-git/v5 v5.11.0
github.com/vektra/mockery/v2 v2.14.0
gonum.org/v1/gonum v0.8.2
google.golang.org/protobuf v1.32.0
)

require (
github.com/grafana/pyroscope-go v1.0.3
github.com/influxdata/influxdb-client-go/v2 v2.12.3
github.com/pyroscope-io/otel-profiling-go v0.4.0
go.opentelemetry.io/otel v1.18.0
go.opentelemetry.io/otel/exporters/stdout/stdouttrace v1.18.0
go.opentelemetry.io/otel/sdk v1.18.0
golang.org/x/crypto v0.17.0
golang.org/x/net v0.19.0
gonum.org/v1/gonum v0.8.2
google.golang.org/grpc v1.59.0
google.golang.org/protobuf v1.31.0
)

require (
Expand Down
Loading

0 comments on commit c085e04

Please sign in to comment.