-
Notifications
You must be signed in to change notification settings - Fork 175
/
verification.go
341 lines (286 loc) · 15.5 KB
/
verification.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
package metrics
import (
"github.com/prometheus/client_golang/prometheus"
"github.com/onflow/flow-go/module"
)
type VerificationCollector struct {
tracer module.Tracer
// Job Consumers
lastProcessedBlockJobIndexBlockConsumer prometheus.Gauge
lastProcessedChunkJobIndexChunkConsumer prometheus.Gauge
// Assigner Engine
receivedFinalizedHeightAssigner prometheus.Gauge // the last finalized height received by assigner engine
assignedChunkTotalAssigner prometheus.Counter // total chunks assigned to this verification node
processedChunkTotalAssigner prometheus.Counter // total chunks sent by assigner engine to chunk consumer (i.e., fetcher input)
receivedResultsTotalAssigner prometheus.Counter // total execution results arrived at assigner
// Fetcher Engine
receivedAssignedChunkTotalFetcher prometheus.Counter // total assigned chunks received by fetcher engine from assigner engine.
sentVerifiableChunksTotalFetcher prometheus.Counter // total verifiable chunk sent by fetcher engine and sent to verifier engine.
receivedChunkDataPackTotalFetcher prometheus.Counter // total chunk data packs received by fetcher engine
requestedChunkDataPackTotalFetcher prometheus.Counter // total number of chunk data packs requested by fetcher engine
// Requester Engine
//
// total number of chunk data pack requests received by requester engine from fetcher engine.
receivedChunkDataPackRequestsTotalRequester prometheus.Counter
// total number of chunk data request messages dispatched in the network by requester engine.
sentChunkDataRequestMessagesTotalRequester prometheus.Counter
// total number of chunk data response messages received by requester from network.
receivedChunkDataResponseMessageTotalRequester prometheus.Counter
// total number of chunk data pack sent by requester to fetcher engine.
sentChunkDataPackByRequesterTotal prometheus.Counter
// maximum number of attempts made for requesting a chunk data pack belonging to the next unsealed height.
maxChunkDataPackRequestAttemptForNextUnsealedHeight prometheus.Gauge
// Verifier Engine
receivedVerifiableChunkTotalVerifier prometheus.Counter // total verifiable chunks received by verifier engine
sentResultApprovalTotalVerifier prometheus.Counter // total result approvals sent by verifier engine
}
func NewVerificationCollector(tracer module.Tracer, registerer prometheus.Registerer) *VerificationCollector {
// Job Consumers
lastProcessedBlockJobIndexBlockConsumer := prometheus.NewGauge(prometheus.GaugeOpts{
Name: "last_processed_block_job_index",
Namespace: namespaceVerification,
Subsystem: subsystemBlockConsumer,
Help: "the last block job index processed by block consumer",
})
lastProcessedChunkJobIndexChunkConsumer := prometheus.NewGauge(prometheus.GaugeOpts{
Name: "last_processed_chunk_job_index",
Namespace: namespaceVerification,
Subsystem: subsystemChunkConsumer,
Help: "the last chunk job index processed by chunk consumer",
})
// Assigner Engine
receivedFinalizedHeightAssigner := prometheus.NewGauge(prometheus.GaugeOpts{
Name: "finalized_height",
Namespace: namespaceVerification,
Subsystem: subsystemAssignerEngine,
Help: "the last finalized height received by assigner engine",
})
receivedResultsTotalAssigner := prometheus.NewGauge(prometheus.GaugeOpts{
Name: "received_result_total",
Namespace: namespaceVerification,
Subsystem: subsystemAssignerEngine,
Help: "total number of execution results received by assigner engine",
})
assignedChunksTotalAssigner := prometheus.NewCounter(prometheus.CounterOpts{
Name: "chunk_assigned_total",
Namespace: namespaceVerification,
Subsystem: subsystemAssignerEngine,
Help: "total number of chunks assigned to verification node",
})
sentChunksTotalAssigner := prometheus.NewCounter(prometheus.CounterOpts{
Name: "processed_chunk_sent_total",
Namespace: namespaceVerification,
Subsystem: subsystemAssignerEngine,
Help: "total number chunks sent by assigner engine to chunk consumer",
})
// Fetcher Engine
receivedAssignedChunksTotalFetcher := prometheus.NewCounter(prometheus.CounterOpts{
Name: "assigned_chunk_received_total",
Namespace: namespaceVerification,
Subsystem: subsystemFetcherEngine,
Help: "total number of chunks received by fetcher engine from assigner engine through chunk consumer",
})
// Requester Engine
receivedChunkDataPackRequestsTotalRequester := prometheus.NewCounter(prometheus.CounterOpts{
Name: "chunk_data_pack_request_received_total",
Namespace: namespaceVerification,
Subsystem: subsystemRequesterEngine,
Help: "total number of chunk data pack requests received by requester engine from fetcher engine",
})
sentChunkDataRequestMessagesTotalRequester := prometheus.NewCounter(prometheus.CounterOpts{
Name: "chunk_data_pack_request_message_sent_total",
Namespace: namespaceVerification,
Subsystem: subsystemRequesterEngine,
Help: "total number of chunk data pack request messages sent in the network by requester engine",
})
receivedChunkDataResponseMessagesTotalRequester := prometheus.NewCounter(prometheus.CounterOpts{
Name: "chunk_data_response_message_received_total",
Namespace: namespaceVerification,
Subsystem: subsystemRequesterEngine,
Help: "total number of chunk data response messages received from network by requester engine",
})
sentChunkDataPackByRequesterTotal := prometheus.NewCounter(prometheus.CounterOpts{
Name: "chunk_data_pack_sent_total",
Namespace: namespaceVerification,
Subsystem: subsystemRequesterEngine,
Help: "total number of chunk data packs sent by requester engine to fetcher engine",
})
sentVerifiableChunksTotalFetcher := prometheus.NewCounter(prometheus.CounterOpts{
Name: "verifiable_chunk_sent_total",
Namespace: namespaceVerification,
Subsystem: subsystemFetcherEngine,
Help: "total number of verifiable chunks sent by fetcher engine to verifier engine",
})
receivedChunkDataPackTotalFetcher := prometheus.NewCounter(prometheus.CounterOpts{
Name: "chunk_data_pack_received_total",
Namespace: namespaceVerification,
Subsystem: subsystemFetcherEngine,
Help: "total number of chunk data packs received by fetcher engine",
})
requestedChunkDataPackTotalFetcher := prometheus.NewCounter(prometheus.CounterOpts{
Name: "chunk_data_pack_requested_total",
Namespace: namespaceVerification,
Subsystem: subsystemFetcherEngine,
Help: "total number of chunk data packs requested by fetcher engine",
})
maxChunkDataPackRequestAttemptForNextUnsealedHeight := prometheus.NewGauge(prometheus.GaugeOpts{
Name: "next_unsealed_height_max_chunk_data_pack_request_attempt_times",
Namespace: namespaceVerification,
Subsystem: subsystemRequesterEngine,
// an indicator for when execution nodes is unresponsive to chunk data pack requests,
// in which case verification node will keep requesting the chunk data pack, and this
// metrics number will go up.
Help: "among all the pending chunk data packs requested by the requester engine for the next unsealed height the maximum number of times a" +
" certain chunk data pack was requested",
})
// Verifier Engine
receivedVerifiableChunksTotalVerifier := prometheus.NewCounter(prometheus.CounterOpts{
Name: "verifiable_chunk_received_total",
Namespace: namespaceVerification,
Subsystem: subsystemVerifierEngine,
Help: "total number verifiable chunks received by verifier engine from fetcher engine",
})
sentResultApprovalTotalVerifier := prometheus.NewCounter(prometheus.CounterOpts{
Name: "result_approvals_total",
Namespace: namespaceVerification,
Subsystem: subsystemVerifierEngine,
Help: "total number of emitted result approvals by verifier engine",
})
// registers all metrics and panics if any fails.
registerer.MustRegister(
// job consumers
lastProcessedBlockJobIndexBlockConsumer,
lastProcessedChunkJobIndexChunkConsumer,
// assigner
receivedFinalizedHeightAssigner,
assignedChunksTotalAssigner,
sentChunksTotalAssigner,
receivedResultsTotalAssigner,
// fetcher engine
receivedAssignedChunksTotalFetcher,
sentVerifiableChunksTotalFetcher,
receivedChunkDataPackTotalFetcher,
requestedChunkDataPackTotalFetcher,
// requester engine
receivedChunkDataPackRequestsTotalRequester,
sentChunkDataRequestMessagesTotalRequester,
receivedChunkDataResponseMessagesTotalRequester,
sentChunkDataPackByRequesterTotal,
maxChunkDataPackRequestAttemptForNextUnsealedHeight,
// verifier engine
receivedVerifiableChunksTotalVerifier,
sentResultApprovalTotalVerifier)
vc := &VerificationCollector{
tracer: tracer,
// job consumers
lastProcessedChunkJobIndexChunkConsumer: lastProcessedChunkJobIndexChunkConsumer,
lastProcessedBlockJobIndexBlockConsumer: lastProcessedBlockJobIndexBlockConsumer,
// assigner
receivedFinalizedHeightAssigner: receivedFinalizedHeightAssigner,
assignedChunkTotalAssigner: assignedChunksTotalAssigner,
processedChunkTotalAssigner: sentChunksTotalAssigner,
receivedResultsTotalAssigner: receivedResultsTotalAssigner,
// fetcher
receivedAssignedChunkTotalFetcher: receivedAssignedChunksTotalFetcher,
receivedChunkDataPackTotalFetcher: receivedChunkDataPackTotalFetcher,
requestedChunkDataPackTotalFetcher: requestedChunkDataPackTotalFetcher,
sentVerifiableChunksTotalFetcher: sentVerifiableChunksTotalFetcher,
// verifier
sentResultApprovalTotalVerifier: sentResultApprovalTotalVerifier,
receivedVerifiableChunkTotalVerifier: receivedVerifiableChunksTotalVerifier,
// requester
receivedChunkDataPackRequestsTotalRequester: receivedChunkDataPackRequestsTotalRequester,
sentChunkDataRequestMessagesTotalRequester: sentChunkDataRequestMessagesTotalRequester,
receivedChunkDataResponseMessageTotalRequester: receivedChunkDataResponseMessagesTotalRequester,
sentChunkDataPackByRequesterTotal: sentChunkDataPackByRequesterTotal,
maxChunkDataPackRequestAttemptForNextUnsealedHeight: maxChunkDataPackRequestAttemptForNextUnsealedHeight,
}
return vc
}
// OnExecutionResultReceivedAtAssignerEngine is called whenever a new execution result arrives
// at Assigner engine. It increments total number of received results.
func (vc *VerificationCollector) OnExecutionResultReceivedAtAssignerEngine() {
vc.receivedResultsTotalAssigner.Inc()
}
// OnVerifiableChunkReceivedAtVerifierEngine is called whenever a verifiable chunk is received by Verifier engine
// from Assigner engine.It increments the total number of sent verifiable chunks.
func (vc *VerificationCollector) OnVerifiableChunkReceivedAtVerifierEngine() {
vc.receivedVerifiableChunkTotalVerifier.Inc()
}
// OnResultApprovalDispatchedInNetwork is called whenever a result approval for is emitted to consensus nodes.
// It increases the total number of result approvals.
func (vc *VerificationCollector) OnResultApprovalDispatchedInNetworkByVerifier() {
// increases the counter of disseminated result approvals
// fo by one. Each result approval corresponds to a single chunk of the block
// the approvals disseminated by verifier engine
vc.sentResultApprovalTotalVerifier.Inc()
}
// OnFinalizedBlockArrivedAtAssigner sets a gauge that keeps track of number of the latest block height arrives
// at assigner engine. Note that it assumes blocks are coming to assigner engine in strictly increasing order of their height.
func (vc *VerificationCollector) OnFinalizedBlockArrivedAtAssigner(height uint64) {
vc.receivedFinalizedHeightAssigner.Set(float64(height))
}
// OnChunksAssignmentDoneAtAssigner increments a counter that keeps track of the total number of assigned chunks to
// the verification node.
func (vc *VerificationCollector) OnChunksAssignmentDoneAtAssigner(chunks int) {
vc.assignedChunkTotalAssigner.Add(float64(chunks))
}
// OnAssignedChunkProcessedAtAssigner increments a counter that keeps track of the total number of assigned chunks pushed by
// assigner engine to the fetcher engine.
func (vc *VerificationCollector) OnAssignedChunkProcessedAtAssigner() {
vc.processedChunkTotalAssigner.Inc()
}
// OnAssignedChunkReceivedAtFetcher increments a counter that keeps track of number of assigned chunks arrive at fetcher engine.
func (vc *VerificationCollector) OnAssignedChunkReceivedAtFetcher() {
vc.receivedAssignedChunkTotalFetcher.Inc()
}
// OnChunkDataPackRequestSentByFetcher increments a counter that keeps track of number of chunk data pack requests that fetcher engine
// sends to requester engine.
func (vc *VerificationCollector) OnChunkDataPackRequestSentByFetcher() {
vc.requestedChunkDataPackTotalFetcher.Inc()
}
// OnChunkDataPackRequestReceivedByRequester increments a counter that keeps track of number of chunk data pack requests
// arrive at the requester engine from the fetcher engine.
func (vc *VerificationCollector) OnChunkDataPackRequestReceivedByRequester() {
vc.receivedChunkDataPackRequestsTotalRequester.Inc()
}
// OnChunkDataPackRequestDispatchedInNetworkByRequester increments a counter that keeps track of number of chunk data pack requests that the
// requester engine dispatches in the network (to the execution nodes).
func (vc *VerificationCollector) OnChunkDataPackRequestDispatchedInNetworkByRequester() {
vc.sentChunkDataRequestMessagesTotalRequester.Inc()
}
// OnChunkDataPackResponseReceivedFromNetworkByRequester increments a counter that keeps track of number of chunk data pack responses that the
// requester engine receives from execution nodes (through network).
func (vc *VerificationCollector) OnChunkDataPackResponseReceivedFromNetworkByRequester() {
vc.receivedChunkDataResponseMessageTotalRequester.Inc()
}
// OnChunkDataPackSentToFetcher increases a counter that keeps track of number of chunk data packs sent to the fetcher engine from
// requester engine.
func (vc *VerificationCollector) OnChunkDataPackSentToFetcher() {
vc.sentChunkDataPackByRequesterTotal.Inc()
}
// OnChunkDataPackArrivedAtFetcher increments a counter that keeps track of number of chunk data packs arrived at fetcher engine from
// requester engine.
func (vc *VerificationCollector) OnChunkDataPackArrivedAtFetcher() {
vc.receivedChunkDataPackTotalFetcher.Inc()
}
// OnVerifiableChunkSentToVerifier increments a counter that keeps track of number of verifiable chunks fetcher engine sent to verifier engine.
func (vc *VerificationCollector) OnVerifiableChunkSentToVerifier() {
vc.sentVerifiableChunksTotalFetcher.Inc()
}
// OnChunkConsumerJobDone is invoked by chunk consumer whenever it is notified a job is done by a worker. It
// sets the last processed chunk job index.
func (vc *VerificationCollector) OnChunkConsumerJobDone(processedIndex uint64) {
vc.lastProcessedChunkJobIndexChunkConsumer.Set(float64(processedIndex))
}
// OnBlockConsumerJobDone is invoked by block consumer whenever it is notified a job is done by a worker. It
// sets the last processed block job index.
func (vc *VerificationCollector) OnBlockConsumerJobDone(processedIndex uint64) {
vc.lastProcessedBlockJobIndexBlockConsumer.Set(float64(processedIndex))
}
// SetMaxChunkDataPackAttemptsForNextUnsealedHeightAtRequester is invoked when a cycle of requesting chunk data packs is done by requester engine.
// It updates the maximum number of attempts made by requester engine for requesting the chunk data packs of the next unsealed height.
// The maximum is taken over the history of all chunk data packs requested during that cycle that belong to the next unsealed height.
func (vc *VerificationCollector) SetMaxChunkDataPackAttemptsForNextUnsealedHeightAtRequester(attempts uint64) {
vc.maxChunkDataPackRequestAttemptForNextUnsealedHeight.Set(float64(attempts))
}