Merge remote-tracking branch 'origin' into jasmina/4457-blocksync-verification_part1

internal/blocksync/reactor.go

@@ -502,11 +502,21 @@ func (r *Reactor) poolRoutine(ctx context.Context, stateSynced bool, blockSyncCh
 				newBlockPartSetHeader = newBlockParts.Header()
 				newBlockID            = types.BlockID{Hash: newBlock.Hash(), PartSetHeader: newBlockPartSetHeader}
 			)
-
+			// Finally, verify the first block using the second's commit.
+			//
+			// NOTE: We can probably make this more efficient, but note that calling
+			// first.Hash() doesn't verify the tx contents, so MakePartSet() is
+			// currently necessary.
 			if r.lastTrustedBlock != nil {
 				err := VerifyNextBlock(newBlock, newBlockID, verifyBlock, r.lastTrustedBlock.block, r.lastTrustedBlock.commit, state.NextValidators)
 
 				if err != nil {
+					r.logger.Error(
+						err.Error(),
+						"last_commit", verifyBlock.LastCommit,
+						"block_id", newBlock,
+						"height", newBlock.Height,
+					)
 					switch err.(type) {
 					case ErrBlockIDDiff:
 					case ErrValidationFailed:
@@ -517,7 +527,7 @@ func (r *Reactor) poolRoutine(ctx context.Context, stateSynced bool, blockSyncCh
 						}); serr != nil {
 							return
 						}
-						continue
+
 					case ErrInvalidVerifyBlock:
 						r.logger.Error(
 							err.Error(),
@@ -532,11 +542,20 @@ func (r *Reactor) poolRoutine(ctx context.Context, stateSynced bool, blockSyncCh
 						}); serr != nil {
 							return
 						}
-						continue
-					}
 
+					}
+					continue // was return previously
 				}
 
+				// // Finally, verify the first block using the second's commit.
+				// //
+				// // NOTE: We can probably make this more efficient, but note that calling
+				// // first.Hash() doesn't verify the tx contents, so MakePartSet() is
+				// // currently necessary.
+				// err = state.Validators.VerifyCommitLight(chainID, firstID, first.Height, second.LastCommit)
+
+				r.lastTrustedBlock = &BlockResponse{block: newBlock, commit: verifyBlock.LastCommit}
+
 			} else {
 				// we need to load the last block we trusted
 				if r.initialState.LastBlockHeight != 0 {
@@ -559,22 +578,42 @@ func (r *Reactor) poolRoutine(ctx context.Context, stateSynced bool, blockSyncCh
 					}); serr != nil {
 						return
 					}
-					continue
+					continue // was return previously
 				}
-				r.lastTrustedBlock = &BlockResponse{block: newBlock, commit: verifyBlock.LastCommit}
+
+				r.lastTrustedBlock.block = newBlock
+				// r.lastTrustedBlock.commit
+			}
+			var err error
+			// validate the block before we persist it
+			err = r.blockExec.ValidateBlock(ctx, state, newBlock)
+			if err != nil {
+				r.logger.Error("The validator set provided by the new block does not match the expected validator set",
+					"initial hash ", r.initialState.Validators.Hash(),
+					"new hash ", newBlock.ValidatorsHash,
+				)
+
+				peerID := r.pool.RedoRequest(r.lastTrustedBlock.block.Height + 1)
+				if serr := blockSyncCh.SendError(ctx, p2p.PeerError{
+					NodeID: peerID,
+					Err:    ErrValidationFailed{},
+				}); serr != nil {
+					return
+				}
+				continue // was return previously
 			}
-			r.lastTrustedBlock.block = newBlock
-			r.lastTrustedBlock.commit = verifyBlock.LastCommit
 			r.pool.PopRequest()
 
 			// TODO: batch saves so we do not persist to disk every block
 			r.store.SaveBlock(newBlock, newBlockParts, verifyBlock.LastCommit)
 
-			var err error
-
 			// TODO: Same thing for app - but we would need a way to get the hash
 			// without persisting the state.
 			state, err = r.blockExec.ApplyBlock(ctx, state, newBlockID, newBlock)
+			if err != nil {
+				panic(fmt.Sprintf("failed to process committed block (%d:%X): %v", newBlock.Height, newBlock.Hash(), err))
+			}
+
 			if err != nil {
 				// TODO: This is bad, are we zombie?
 				panic(fmt.Sprintf("failed to process committed block (%d:%X): %v", newBlock.Height, newBlock.Hash(), err))

internal/consensus/metrics.go

@@ -8,6 +8,7 @@ import (
 	"github.com/go-kit/kit/metrics/discard"
 
 	cstypes "github.com/tendermint/tendermint/internal/consensus/types"
+	tmproto "github.com/tendermint/tendermint/proto/tendermint/types"
 	"github.com/tendermint/tendermint/types"
 
 	prometheus "github.com/go-kit/kit/metrics/prometheus"
@@ -103,6 +104,33 @@ type Metrics struct {
 	// the proposal message and the local time of the validator at the time
 	// that the validator received the message.
 	ProposalTimestampDifference metrics.Histogram
+
+	// VoteExtensionReceiveCount is the number of vote extensions received by this
+	// node. The metric is annotated by the status of the vote extension from the
+	// application, either 'accepted' or 'rejected'.
+	VoteExtensionReceiveCount metrics.Counter
+
+	// ProposalReceiveCount is the total number of proposals received by this node
+	// since process start.
+	// The metric is annotated by the status of the proposal from the application,
+	// either 'accepted' or 'rejected'.
+	ProposalReceiveCount metrics.Counter
+
+	// ProposalCreationCount is the total number of proposals created by this node
+	// since process start.
+	// The metric is annotated by the status of the proposal from the application,
+	// either 'accepted' or 'rejected'.
+	ProposalCreateCount metrics.Counter
+
+	// RoundVotingPowerPercent is the percentage of the total voting power received
+	// with a round. The value begins at 0 for each round and approaches 1.0 as
+	// additional voting power is observed. The metric is labeled by vote type.
+	RoundVotingPowerPercent metrics.Gauge
+
+	// LateVotes stores the number of votes that were received by this node that
+	// correspond to earlier heights and rounds than this node is currently
+	// in.
+	LateVotes metrics.Counter
 }
 
 // PrometheusMetrics returns Metrics build using Prometheus client library.
@@ -280,6 +308,43 @@ func PrometheusMetrics(namespace string, labelsAndValues ...string) *Metrics {
 				"Only calculated when a new block is proposed.",
 			Buckets: []float64{-10, -.5, -.025, 0, .1, .5, 1, 1.5, 2, 10},
 		}, append(labels, "is_timely")).With(labelsAndValues...),
+		VoteExtensionReceiveCount: prometheus.NewCounterFrom(stdprometheus.CounterOpts{
+			Namespace: namespace,
+			Subsystem: MetricsSubsystem,
+			Name:      "vote_extension_receive_count",
+			Help: "Number of vote extensions received by the node since process start, labeled by " +
+				"the application's response to VerifyVoteExtension, either accept or reject.",
+		}, append(labels, "status")).With(labelsAndValues...),
+
+		ProposalReceiveCount: prometheus.NewCounterFrom(stdprometheus.CounterOpts{
+			Namespace: namespace,
+			Subsystem: MetricsSubsystem,
+			Name:      "proposal_receive_count",
+			Help: "Number of vote proposals received by the node since process start, labeled by " +
+				"the application's response to ProcessProposal, either accept or reject.",
+		}, append(labels, "status")).With(labelsAndValues...),
+
+		ProposalCreateCount: prometheus.NewCounterFrom(stdprometheus.CounterOpts{
+			Namespace: namespace,
+			Subsystem: MetricsSubsystem,
+			Name:      "proposal_create_count",
+			Help:      "Number of proposals created by the node since process start.",
+		}, labels).With(labelsAndValues...),
+
+		RoundVotingPowerPercent: prometheus.NewGaugeFrom(stdprometheus.GaugeOpts{
+			Namespace: namespace,
+			Subsystem: MetricsSubsystem,
+			Name:      "round_voting_power_percent",
+			Help: "Percentage of the total voting power received with a round. " +
+				"The value begins at 0 for each round and approaches 1.0 as additional " +
+				"voting power is observed.",
+		}, append(labels, "vote_type")).With(labelsAndValues...),
+		LateVotes: prometheus.NewCounterFrom(stdprometheus.CounterOpts{
+			Namespace: namespace,
+			Subsystem: MetricsSubsystem,
+			Name:      "late_votes",
+			Help:      "Number of votes received by the node since process start that correspond to earlier heights and rounds than this node is currently in.",
+		}, append(labels, "vote_type")).With(labelsAndValues...),
 	}
 }
 
@@ -317,6 +382,11 @@ func NopMetrics() *Metrics {
 		QuorumPrevoteDelay:          discard.NewGauge(),
 		FullPrevoteDelay:            discard.NewGauge(),
 		ProposalTimestampDifference: discard.NewHistogram(),
+		VoteExtensionReceiveCount:   discard.NewCounter(),
+		ProposalReceiveCount:        discard.NewCounter(),
+		ProposalCreateCount:         discard.NewCounter(),
+		RoundVotingPowerPercent:     discard.NewGauge(),
+		LateVotes:                   discard.NewCounter(),
 	}
 }
 
@@ -336,10 +406,45 @@ func (m *Metrics) MarkBlockGossipComplete() {
 	m.BlockGossipReceiveLatency.Observe(time.Since(m.blockGossipStart).Seconds())
 }
 
+func (m *Metrics) MarkProposalProcessed(accepted bool) {
+	status := "accepted"
+	if !accepted {
+		status = "rejected"
+	}
+	m.ProposalReceiveCount.With("status", status).Add(1)
+}
+
+func (m *Metrics) MarkVoteExtensionReceived(accepted bool) {
+	status := "accepted"
+	if !accepted {
+		status = "rejected"
+	}
+	m.VoteExtensionReceiveCount.With("status", status).Add(1)
+}
+
+func (m *Metrics) MarkVoteReceived(vt tmproto.SignedMsgType, power, totalPower int64) {
+	p := float64(power) / float64(totalPower)
+	n := strings.ToLower(strings.TrimPrefix(vt.String(), "SIGNED_MSG_TYPE_"))
+	m.RoundVotingPowerPercent.With("vote_type", n).Add(p)
+}
+
 func (m *Metrics) MarkRound(r int32, st time.Time) {
 	m.Rounds.Set(float64(r))
 	roundTime := time.Since(st).Seconds()
 	m.RoundDuration.Observe(roundTime)
+
+	pvt := tmproto.PrevoteType
+	pvn := strings.ToLower(strings.TrimPrefix(pvt.String(), "SIGNED_MSG_TYPE_"))
+	m.RoundVotingPowerPercent.With("vote_type", pvn).Set(0)
+
+	pct := tmproto.PrecommitType
+	pcn := strings.ToLower(strings.TrimPrefix(pct.String(), "SIGNED_MSG_TYPE_"))
+	m.RoundVotingPowerPercent.With("vote_type", pcn).Set(0)
+}
+
+func (m *Metrics) MarkLateVote(vt tmproto.SignedMsgType) {
+	n := strings.ToLower(strings.TrimPrefix(vt.String(), "SIGNED_MSG_TYPE_"))
+	m.LateVotes.With("vote_type", n).Add(1)
 }
 
 func (m *Metrics) MarkStep(s cstypes.RoundStepType) {

internal/consensus/state.go

@@ -1334,6 +1334,7 @@ func (cs *State) defaultDecideProposal(ctx context.Context, height int64, round
 		} else if block == nil {
 			return
 		}
+		cs.metrics.ProposalCreateCount.Add(1)
 		blockParts, err = block.MakePartSet(types.BlockPartSizeBytes)
 		if err != nil {
 			cs.logger.Error("unable to create proposal block part set", "error", err)
@@ -1531,6 +1532,7 @@ func (cs *State) defaultDoPrevote(ctx context.Context, height int64, round int32
 	if err != nil {
 		panic(fmt.Sprintf("ProcessProposal: %v", err))
 	}
+	cs.metrics.MarkProposalProcessed(isAppValid)
 
 	// Vote nil if the Application rejected the block
 	if !isAppValid {
@@ -2297,6 +2299,10 @@ func (cs *State) addVote(
 		"cs_height", cs.Height,
 	)
 
+	if vote.Height < cs.Height || (vote.Height == cs.Height && vote.Round < cs.Round) {
+		cs.metrics.MarkLateVote(vote.Type)
+	}
+
 	// A precommit for the previous height?
 	// These come in while we wait timeoutCommit
 	if vote.Height+1 == cs.Height && vote.Type == tmproto.PrecommitType {
@@ -2337,7 +2343,9 @@ func (cs *State) addVote(
 
 	// Verify VoteExtension if precommit
 	if vote.Type == tmproto.PrecommitType {
-		if err = cs.blockExec.VerifyVoteExtension(ctx, vote); err != nil {
+		err := cs.blockExec.VerifyVoteExtension(ctx, vote)
+		cs.metrics.MarkVoteExtensionReceived(err == nil)
+		if err != nil {
 			return false, err
 		}
 	}
@@ -2348,6 +2356,11 @@ func (cs *State) addVote(
 		// Either duplicate, or error upon cs.Votes.AddByIndex()
 		return
 	}
+	if vote.Round == cs.Round {
+		vals := cs.state.Validators
+		_, val := vals.GetByIndex(vote.ValidatorIndex)
+		cs.metrics.MarkVoteReceived(vote.Type, val.VotingPower, vals.TotalVotingPower())
+	}
 
 	if err := cs.eventBus.PublishEventVote(types.EventDataVote{Vote: vote}); err != nil {
 		return added, err

internal/state/execution.go

@@ -247,6 +247,10 @@ func (blockExec *BlockExecutor) ApplyBlock(
 	}
 	if len(validatorUpdates) > 0 {
 		blockExec.logger.Debug("updates to validators", "updates", types.ValidatorListString(validatorUpdates))
+		blockExec.metrics.ValidatorSetUpdates.Add(1)
+	}
+	if finalizeBlockResponse.ConsensusParamUpdates != nil {
+		blockExec.metrics.ConsensusParamUpdates.Add(1)
 	}
 
 	// Update the state with the block and responses.

internal/state/indexer/metrics.gen.go

@@ -0,0 +1,51 @@
+// Code generated by metricsgen. DO NOT EDIT.
+
+package indexer
+
+import (
+	"github.com/go-kit/kit/metrics/discard"
+	prometheus "github.com/go-kit/kit/metrics/prometheus"
+	stdprometheus "github.com/prometheus/client_golang/prometheus"
+)
+
+func PrometheusMetrics(namespace string, labelsAndValues ...string) *Metrics {
+	labels := []string{}
+	for i := 0; i < len(labelsAndValues); i += 2 {
+		labels = append(labels, labelsAndValues[i])
+	}
+	return &Metrics{
+		BlockEventsSeconds: prometheus.NewHistogramFrom(stdprometheus.HistogramOpts{
+			Namespace: namespace,
+			Subsystem: MetricsSubsystem,
+			Name:      "block_events_seconds",
+			Help:      "Latency for indexing block events.",
+		}, labels).With(labelsAndValues...),
+		TxEventsSeconds: prometheus.NewHistogramFrom(stdprometheus.HistogramOpts{
+			Namespace: namespace,
+			Subsystem: MetricsSubsystem,
+			Name:      "tx_events_seconds",
+			Help:      "Latency for indexing transaction events.",
+		}, labels).With(labelsAndValues...),
+		BlocksIndexed: prometheus.NewCounterFrom(stdprometheus.CounterOpts{
+			Namespace: namespace,
+			Subsystem: MetricsSubsystem,
+			Name:      "blocks_indexed",
+			Help:      "Number of complete blocks indexed.",
+		}, labels).With(labelsAndValues...),
+		TransactionsIndexed: prometheus.NewCounterFrom(stdprometheus.CounterOpts{
+			Namespace: namespace,
+			Subsystem: MetricsSubsystem,
+			Name:      "transactions_indexed",
+			Help:      "Number of transactions indexed.",
+		}, labels).With(labelsAndValues...),
+	}
+}
+
+func NopMetrics() *Metrics {
+	return &Metrics{
+		BlockEventsSeconds:  discard.NewHistogram(),
+		TxEventsSeconds:     discard.NewHistogram(),
+		BlocksIndexed:       discard.NewCounter(),
+		TransactionsIndexed: discard.NewCounter(),
+	}
+}

internal/state/indexer/metrics.go

@@ -2,12 +2,10 @@ package indexer
 
 import (
 	"github.com/go-kit/kit/metrics"
-	"github.com/go-kit/kit/metrics/discard"
-
-	prometheus "github.com/go-kit/kit/metrics/prometheus"
-	stdprometheus "github.com/prometheus/client_golang/prometheus"
 )
 
+//go:generate go run github.com/tendermint/tendermint/scripts/metricsgen -struct=Metrics
+
 // MetricsSubsystem is a the subsystem label for the indexer package.
 const MetricsSubsystem = "indexer"
 
@@ -25,49 +23,3 @@ type Metrics struct {
 	// Number of transactions indexed.
 	TransactionsIndexed metrics.Counter
 }
-
-// PrometheusMetrics returns Metrics build using Prometheus client library.
-// Optionally, labels can be provided along with their values ("foo",
-// "fooValue").
-func PrometheusMetrics(namespace string, labelsAndValues ...string) *Metrics {
-	labels := []string{}
-	for i := 0; i < len(labelsAndValues); i += 2 {
-		labels = append(labels, labelsAndValues[i])
-	}
-	return &Metrics{
-		BlockEventsSeconds: prometheus.NewHistogramFrom(stdprometheus.HistogramOpts{
-			Namespace: namespace,
-			Subsystem: MetricsSubsystem,
-			Name:      "block_events_seconds",
-			Help:      "Latency for indexing block events.",
-		}, labels).With(labelsAndValues...),
-		TxEventsSeconds: prometheus.NewHistogramFrom(stdprometheus.HistogramOpts{
-			Namespace: namespace,
-			Subsystem: MetricsSubsystem,
-			Name:      "tx_events_seconds",
-			Help:      "Latency for indexing transaction events.",
-		}, labels).With(labelsAndValues...),
-		BlocksIndexed: prometheus.NewCounterFrom(stdprometheus.CounterOpts{
-			Namespace: namespace,
-			Subsystem: MetricsSubsystem,
-			Name:      "blocks_indexed",
-			Help:      "Number of complete blocks indexed.",
-		}, labels).With(labelsAndValues...),
-		TransactionsIndexed: prometheus.NewCounterFrom(stdprometheus.CounterOpts{
-			Namespace: namespace,
-			Subsystem: MetricsSubsystem,
-			Name:      "transactions_indexed",
-			Help:      "Number of transactions indexed.",
-		}, labels).With(labelsAndValues...),
-	}
-}
-
-// NopMetrics returns an indexer metrics stub that discards all samples.
-func NopMetrics() *Metrics {
-	return &Metrics{
-		BlockEventsSeconds:  discard.NewHistogram(),
-		TxEventsSeconds:     discard.NewHistogram(),
-		BlocksIndexed:       discard.NewCounter(),
-		TransactionsIndexed: discard.NewCounter(),
-	}
-}

internal/state/metrics.go

@@ -17,6 +17,14 @@ const (
 type Metrics struct {
 	// Time between BeginBlock and EndBlock.
 	BlockProcessingTime metrics.Histogram
+
+	// ConsensusParamUpdates is the total number of times the application has
+	// udated the consensus params since process start.
+	ConsensusParamUpdates metrics.Counter
+
+	// ValidatorSetUpdates is the total number of times the application has
+	// udated the validator set since process start.
+	ValidatorSetUpdates metrics.Counter
 }
 
 // PrometheusMetrics returns Metrics build using Prometheus client library.
@@ -35,12 +43,29 @@ func PrometheusMetrics(namespace string, labelsAndValues ...string) *Metrics {
 			Help:      "Time between BeginBlock and EndBlock in ms.",
 			Buckets:   stdprometheus.LinearBuckets(1, 10, 10),
 		}, labels).With(labelsAndValues...),
+		ConsensusParamUpdates: prometheus.NewCounterFrom(stdprometheus.CounterOpts{
+			Namespace: namespace,
+			Subsystem: MetricsSubsystem,
+			Name:      "consensus_param_updates",
+			Help: "The total number of times the application as updated the consensus " +
+				"parameters since process start.",
+		}, labels).With(labelsAndValues...),
+
+		ValidatorSetUpdates: prometheus.NewCounterFrom(stdprometheus.CounterOpts{
+			Namespace: namespace,
+			Subsystem: MetricsSubsystem,
+			Name:      "validator_set_updates",
+			Help: "The total number of times the application as updated the validator " +
+				"set since process start.",
+		}, labels).With(labelsAndValues...),
 	}
 }
 
 // NopMetrics returns no-op Metrics.
 func NopMetrics() *Metrics {
 	return &Metrics{
-		BlockProcessingTime: discard.NewHistogram(),
+		BlockProcessingTime:   discard.NewHistogram(),
+		ConsensusParamUpdates: discard.NewCounter(),
+		ValidatorSetUpdates:   discard.NewCounter(),
 	}
 }