-
Notifications
You must be signed in to change notification settings - Fork 721
/
Tracers.hs
1642 lines (1520 loc) · 71.9 KB
/
Tracers.hs
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
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
{-# LANGUAGE BangPatterns #-}
{-# LANGUAGE ConstraintKinds #-}
{-# LANGUAGE DataKinds #-}
{-# LANGUAGE FlexibleContexts #-}
{-# LANGUAGE FlexibleInstances #-}
{-# LANGUAGE GADTs #-}
{-# LANGUAGE KindSignatures #-}
{-# LANGUAGE LambdaCase #-}
{-# LANGUAGE MultiParamTypeClasses #-}
{-# LANGUAGE NamedFieldPuns #-}
{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE PackageImports #-}
{-# LANGUAGE QuantifiedConstraints #-}
{-# LANGUAGE ScopedTypeVariables #-}
{-# LANGUAGE TupleSections #-}
{-# LANGUAGE TypeApplications #-}
{-# LANGUAGE UndecidableInstances #-}
{-# OPTIONS_GHC -Wno-orphans #-}
{-# OPTIONS_GHC -fno-warn-redundant-constraints #-}
-- needs different instances on ghc8 and on ghc9
module Cardano.Tracing.Tracers
( Tracers (..)
, TraceOptions
, mkTracers
, nullTracersP2P
, nullTracersNonP2P
, traceCounter
) where
import Cardano.BM.Data.Aggregated (Measurable (..))
import Cardano.BM.Data.Tracer (WithSeverity (..), annotateSeverity)
import Cardano.BM.Data.Transformers
import Cardano.BM.Internal.ElidingTracer
import Cardano.BM.Trace (traceNamedObject)
import Cardano.BM.Tracing
import Cardano.Node.Configuration.Logging
import Cardano.Node.Protocol.Byron ()
import Cardano.Node.Protocol.Shelley ()
import Cardano.Node.Queries
import qualified Cardano.Node.STM as STM
import Cardano.Node.Startup
import Cardano.Node.TraceConstraints
import Cardano.Node.Tracing
import Cardano.Protocol.TPraos.OCert (KESPeriod (..))
import Cardano.Slotting.Slot (EpochNo (..), SlotNo (..), WithOrigin (..))
import Cardano.Tracing.Config
import Cardano.Tracing.HasIssuer (BlockIssuerVerificationKeyHash (..), HasIssuer (..))
import Cardano.Tracing.Metrics
import Cardano.Tracing.Render (renderChainHash, renderHeaderHash)
import Cardano.Tracing.Shutdown ()
import Cardano.Tracing.Startup ()
import Ouroboros.Consensus.Block (BlockConfig, BlockProtocol, CannotForge,
ConvertRawHash (..), ForgeStateInfo, ForgeStateUpdateError, Header,
realPointHash, realPointSlot)
import Ouroboros.Consensus.BlockchainTime (SystemStart (..),
TraceBlockchainTimeEvent (..))
import Ouroboros.Consensus.HeaderValidation (OtherHeaderEnvelopeError)
import Ouroboros.Consensus.Ledger.Abstract (LedgerErr, LedgerState)
import Ouroboros.Consensus.Ledger.Extended (ledgerState)
import Ouroboros.Consensus.Ledger.Inspect (InspectLedger, LedgerEvent)
import Ouroboros.Consensus.Ledger.Query (BlockQuery)
import Ouroboros.Consensus.Ledger.SupportsMempool (ApplyTxErr, GenTx, GenTxId, HasTxs,
LedgerSupportsMempool)
import Ouroboros.Consensus.Ledger.SupportsProtocol (LedgerSupportsProtocol)
import Ouroboros.Consensus.Mempool (MempoolSize (..), TraceEventMempool (..))
import Ouroboros.Consensus.MiniProtocol.BlockFetch.Server
import Ouroboros.Consensus.MiniProtocol.ChainSync.Server
import qualified Ouroboros.Consensus.Network.NodeToClient as NodeToClient
import qualified Ouroboros.Consensus.Network.NodeToNode as NodeToNode
import Ouroboros.Consensus.Node (NetworkP2PMode (..))
import qualified Ouroboros.Consensus.Node.Run as Consensus (RunNode)
import qualified Ouroboros.Consensus.Node.Tracers as Consensus
import Ouroboros.Consensus.Protocol.Abstract (SelectView, ValidationErr)
import qualified Ouroboros.Consensus.Protocol.Ledger.HotKey as HotKey
import qualified Ouroboros.Consensus.Storage.ChainDB as ChainDB
import qualified Ouroboros.Consensus.Storage.LedgerDB as LedgerDB
import Ouroboros.Consensus.Util.Enclose
import qualified Ouroboros.Network.AnchoredFragment as AF
import Ouroboros.Network.Block (BlockNo (..), ChainUpdate (..), HasHeader (..), Point,
StandardHash, blockNo, pointSlot, unBlockNo)
import Ouroboros.Network.BlockFetch.ClientState (TraceFetchClientState (..),
TraceLabelPeer (..))
import Ouroboros.Network.BlockFetch.Decision (FetchDecision, FetchDecline (..))
import Ouroboros.Network.ConnectionId (ConnectionId)
import Ouroboros.Network.ConnectionManager.Types (ConnectionManagerCounters (..),
ConnectionManagerTrace (..))
import qualified Ouroboros.Network.Diffusion as Diffusion
import qualified Ouroboros.Network.Diffusion.NonP2P as NonP2P
import qualified Ouroboros.Network.Diffusion.P2P as P2P
import Ouroboros.Network.InboundGovernor (InboundGovernorTrace (..))
import Ouroboros.Network.InboundGovernor.State (InboundGovernorCounters (..))
import Ouroboros.Network.NodeToClient (LocalAddress)
import Ouroboros.Network.NodeToNode (RemoteAddress)
import Ouroboros.Network.PeerSelection.Governor (PeerSelectionCounters (..))
import Ouroboros.Network.Point (fromWithOrigin)
import Ouroboros.Network.Protocol.LocalStateQuery.Type (ShowQuery)
import Ouroboros.Network.TxSubmission.Inbound
import Codec.CBOR.Read (DeserialiseFailure)
import Control.Concurrent (MVar, modifyMVar_)
import Control.Concurrent.STM (STM, atomically)
import qualified Control.Concurrent.STM as STM
import Control.Monad (forM_, when)
import "contra-tracer" Control.Tracer
import Control.Tracer.Transformers
import Data.Aeson (ToJSON (..), Value (..))
import qualified Data.Aeson.KeyMap as KeyMap
import qualified Data.ByteString.Base16 as B16
import Data.Functor ((<&>))
import Data.Int (Int64)
import Data.IntPSQ (IntPSQ)
import qualified Data.IntPSQ as Pq
import qualified Data.Map.Strict as Map
import Data.Proxy (Proxy (..))
import Data.Text (Text)
import qualified Data.Text as Text
import qualified Data.Text.Encoding as Text
import Data.Time (NominalDiffTime, UTCTime)
import Data.Word (Word64)
import GHC.Clock (getMonotonicTimeNSec)
import GHC.TypeLits (KnownNat, Nat, natVal)
import qualified System.Metrics.Counter as Counter
import qualified System.Metrics.Gauge as Gauge
import qualified System.Metrics.Label as Label
import qualified System.Remote.Monitoring as EKG
{-# OPTIONS_GHC -Wno-redundant-constraints #-}
-- needs different instances on ghc8 and on ghc9
{- HLINT ignore "Redundant bracket" -}
{- HLINT ignore "Use record patterns" -}
data ForgeTracers = ForgeTracers
{ ftForged :: Trace IO Text
, ftForgeAboutToLead :: Trace IO Text
, ftCouldNotForge :: Trace IO Text
, ftAdopted :: Trace IO Text
, ftDidntAdoptBlock :: Trace IO Text
, ftForgedInvalid :: Trace IO Text
, ftTraceNodeNotLeader :: Trace IO Text
, ftTraceNodeCannotForge :: Trace IO Text
, ftTraceForgeStateUpdateError :: Trace IO Text
, ftTraceBlockFromFuture :: Trace IO Text
, ftTraceSlotIsImmutable :: Trace IO Text
, ftTraceNodeIsLeader :: Trace IO Text
, ftTraceAdoptionThreadDied :: Trace IO Text
}
nullTracersP2P :: Tracers peer localPeer blk 'Diffusion.P2P
nullTracersP2P = Tracers
{ chainDBTracer = nullTracer
, consensusTracers = Consensus.nullTracers
, nodeToClientTracers = NodeToClient.nullTracers
, nodeToNodeTracers = NodeToNode.nullTracers
, diffusionTracers = Diffusion.nullTracers
, diffusionTracersExtra = Diffusion.P2PTracers P2P.nullTracers
, startupTracer = nullTracer
, shutdownTracer = nullTracer
, nodeInfoTracer = nullTracer
, nodeStartupInfoTracer = nullTracer
, nodeStateTracer = nullTracer
, resourcesTracer = nullTracer
, peersTracer = nullTracer
}
nullTracersNonP2P :: Tracers peer localPeer blk 'Diffusion.NonP2P
nullTracersNonP2P = Tracers
{ chainDBTracer = nullTracer
, consensusTracers = Consensus.nullTracers
, nodeToClientTracers = NodeToClient.nullTracers
, nodeToNodeTracers = NodeToNode.nullTracers
, diffusionTracers = Diffusion.nullTracers
, diffusionTracersExtra = Diffusion.NonP2PTracers NonP2P.nullTracers
, startupTracer = nullTracer
, shutdownTracer = nullTracer
, nodeInfoTracer = nullTracer
, nodeStartupInfoTracer = nullTracer
, nodeStateTracer = nullTracer
, resourcesTracer = nullTracer
, peersTracer = nullTracer
}
indexGCType :: ChainDB.TraceGCEvent a -> Int
indexGCType ChainDB.ScheduledGC{} = 1
indexGCType ChainDB.PerformedGC{} = 2
indexReplType :: ChainDB.TraceReplayEvent a -> Int
indexReplType LedgerDB.ReplayFromGenesis{} = 1
indexReplType LedgerDB.ReplayFromSnapshot{} = 2
indexReplType LedgerDB.ReplayedBlock{} = 3
instance ElidingTracer (WithSeverity (ChainDB.TraceEvent blk)) where
-- equivalent by type and severity
isEquivalent (WithSeverity s1 (ChainDB.TraceLedgerReplayEvent ev1))
(WithSeverity s2 (ChainDB.TraceLedgerReplayEvent ev2)) =
s1 == s2 && indexReplType ev1 == indexReplType ev2
isEquivalent (WithSeverity s1 (ChainDB.TraceGCEvent ev1))
(WithSeverity s2 (ChainDB.TraceGCEvent ev2)) =
s1 == s2 && indexGCType ev1 == indexGCType ev2
isEquivalent (WithSeverity _s1 (ChainDB.TraceAddBlockEvent _))
(WithSeverity _s2 (ChainDB.TraceAddBlockEvent _)) = True
isEquivalent (WithSeverity _s1 (ChainDB.TraceGCEvent _ev1))
(WithSeverity _s2 (ChainDB.TraceAddBlockEvent _)) = True
isEquivalent (WithSeverity _s1 (ChainDB.TraceAddBlockEvent _))
(WithSeverity _s2 (ChainDB.TraceGCEvent _ev2)) = True
isEquivalent (WithSeverity _s1 (ChainDB.TraceGCEvent _ev1))
(WithSeverity _s2 (ChainDB.TraceCopyToImmutableDBEvent _)) = True
isEquivalent (WithSeverity _s1 (ChainDB.TraceCopyToImmutableDBEvent _))
(WithSeverity _s2 (ChainDB.TraceGCEvent _ev2)) = True
isEquivalent (WithSeverity _s1 (ChainDB.TraceCopyToImmutableDBEvent _))
(WithSeverity _s2 (ChainDB.TraceAddBlockEvent _)) = True
isEquivalent (WithSeverity _s1 (ChainDB.TraceAddBlockEvent _))
(WithSeverity _s2 (ChainDB.TraceCopyToImmutableDBEvent _)) = True
isEquivalent (WithSeverity _s1 (ChainDB.TraceCopyToImmutableDBEvent _))
(WithSeverity _s2 (ChainDB.TraceCopyToImmutableDBEvent _)) = True
isEquivalent (WithSeverity _s1 (ChainDB.TraceInitChainSelEvent ev1))
(WithSeverity _s2 (ChainDB.TraceInitChainSelEvent ev2)) =
case (ev1, ev2) of
(ChainDB.InitChainSelValidation (
ChainDB.UpdateLedgerDbTraceEvent (
LedgerDB.StartedPushingBlockToTheLedgerDb _ _ _)),
ChainDB.InitChainSelValidation (
ChainDB.UpdateLedgerDbTraceEvent (
LedgerDB.StartedPushingBlockToTheLedgerDb _ _ _))) -> True
_ -> False
isEquivalent _ _ = False
-- the types to be elided
doelide (WithSeverity _ (ChainDB.TraceLedgerReplayEvent _)) = True
doelide (WithSeverity _ (ChainDB.TraceGCEvent _)) = True
doelide (WithSeverity _ (ChainDB.TraceAddBlockEvent (ChainDB.IgnoreBlockOlderThanK _))) = False
doelide (WithSeverity _ (ChainDB.TraceAddBlockEvent (ChainDB.IgnoreInvalidBlock _ _))) = False
doelide (WithSeverity _ (ChainDB.TraceAddBlockEvent (ChainDB.BlockInTheFuture _ _))) = False
doelide (WithSeverity _ (ChainDB.TraceAddBlockEvent (ChainDB.StoreButDontChange _))) = False
doelide (WithSeverity _ (ChainDB.TraceAddBlockEvent (ChainDB.TrySwitchToAFork _ _))) = False
doelide (WithSeverity _ (ChainDB.TraceAddBlockEvent (ChainDB.SwitchedToAFork{}))) = False
doelide (WithSeverity _ (ChainDB.TraceAddBlockEvent (ChainDB.AddBlockValidation (ChainDB.InvalidBlock _ _)))) = False
doelide (WithSeverity _ (ChainDB.TraceAddBlockEvent (ChainDB.AddBlockValidation ChainDB.CandidateContainsFutureBlocksExceedingClockSkew{}))) = False
doelide (WithSeverity _ (ChainDB.TraceAddBlockEvent (ChainDB.AddBlockValidation _))) = True
doelide (WithSeverity _ (ChainDB.TraceAddBlockEvent (ChainDB.AddedToCurrentChain events _ _ _))) = null events
doelide (WithSeverity _ (ChainDB.TraceAddBlockEvent (ChainDB.PipeliningEvent{}))) = True
doelide (WithSeverity _ (ChainDB.TraceAddBlockEvent _)) = True
doelide (WithSeverity _ (ChainDB.TraceCopyToImmutableDBEvent _)) = True
doelide (WithSeverity _ (ChainDB.TraceInitChainSelEvent (ChainDB.InitChainSelValidation (ChainDB.UpdateLedgerDbTraceEvent{})))) = True
doelide _ = False
conteliding _tverb _tr _ (Nothing, _count) = return (Nothing, 0)
conteliding tverb tr ev@(WithSeverity _ (ChainDB.TraceAddBlockEvent ChainDB.AddedToCurrentChain{})) (_old, oldt) = do
tnow <- getMonotonicTimeNSec
let tnow' = fromIntegral tnow
deltat = tnow' - oldt
if deltat > 1250000000 -- report at most every 1250 ms
then do
traceWith (toLogObject' tverb tr) ev
return (Just ev, tnow')
else return (Just ev, oldt)
conteliding _tverb _tr ev@(WithSeverity _ (ChainDB.TraceAddBlockEvent _)) (_old, count) =
return (Just ev, count)
conteliding _tverb _tr ev@(WithSeverity _ (ChainDB.TraceCopyToImmutableDBEvent _)) (_old, count) =
return (Just ev, count)
conteliding _tverb _tr ev@(WithSeverity _ (ChainDB.TraceGCEvent _)) (_old, count) =
return (Just ev, count)
conteliding _tverb _tr ev@(WithSeverity _ (ChainDB.TraceLedgerReplayEvent (LedgerDB.ReplayedBlock {}))) (_old, count) = do
return (Just ev, count)
conteliding _tverb _tr ev@(WithSeverity _ (ChainDB.TraceInitChainSelEvent
(ChainDB.InitChainSelValidation
(ChainDB.UpdateLedgerDbTraceEvent
(LedgerDB.StartedPushingBlockToTheLedgerDb
_ _ (LedgerDB.Pushing curr)))))) (_old, count) = return $
let currSlot = fromIntegral $ unSlotNo $ realPointSlot curr in
if count == 0
then (Just ev, currSlot)
else if count + 10000 < currSlot
then (Nothing, 0)
else (Just ev, count)
conteliding _ _ _ _ = return (Nothing, 0)
reportelided _tverb _tr (WithSeverity _ (ChainDB.TraceLedgerReplayEvent (LedgerDB.ReplayedBlock{}))) _count = pure ()
reportelided t tr ev count = defaultelidedreporting t tr ev count
instance (StandardHash header, Eq peer) => ElidingTracer
(WithSeverity [TraceLabelPeer peer (FetchDecision [Point header])]) where
-- equivalent by type and severity
isEquivalent (WithSeverity s1 _peers1)
(WithSeverity s2 _peers2) = s1 == s2
-- the types to be elided
doelide (WithSeverity _ peers) =
let checkDecision :: TraceLabelPeer peer (Either FetchDecline result) -> Bool
checkDecision (TraceLabelPeer _peer (Left FetchDeclineChainNotPlausible)) = True
checkDecision (TraceLabelPeer _peer (Left (FetchDeclineConcurrencyLimit _ _))) = True
checkDecision (TraceLabelPeer _peer (Left (FetchDeclinePeerBusy _ _ _))) = True
checkDecision _ = False
in any checkDecision peers
conteliding _tverb _tr _ (Nothing, _count) = return (Nothing, 0)
conteliding tverb tr ev (_old, count) = do
when (count > 0 && count `mod` 1000 == 0) $ -- report every 1000th message
traceWith (toLogObject' tverb tr) ev
return (Just ev, count + 1)
-- | Tracers for all system components.
--
mkTracers
:: forall blk p2p.
( Consensus.RunNode blk
, TraceConstraints blk
)
=> BlockConfig blk
-> TraceOptions
-> Trace IO Text
-> NodeKernelData blk
-> Maybe EKGDirect
-> NetworkP2PMode p2p
-> IO (Tracers (ConnectionId RemoteAddress) (ConnectionId LocalAddress) blk p2p)
mkTracers blockConfig tOpts@(TracingOnLegacy trSel) tr nodeKern ekgDirect enableP2P = do
fStats <- mkForgingStats
consensusTracers <- mkConsensusTracers ekgDirect trSel verb tr nodeKern fStats
elidedChainDB <- newstate -- for eliding messages in ChainDB tracer
tForks <- STM.newTVarIO 0
pure Tracers
{ chainDBTracer = tracerOnOff' (traceChainDB trSel) $
annotateSeverity $ teeTraceChainTip
blockConfig
fStats
tOpts elidedChainDB
ekgDirect
tForks
(appendName "ChainDB" tr)
(appendName "metrics" tr)
, consensusTracers = consensusTracers
, nodeToClientTracers = nodeToClientTracers' trSel verb tr
, nodeToNodeTracers = nodeToNodeTracers' trSel verb tr
, diffusionTracers
, diffusionTracersExtra = diffusionTracersExtra' enableP2P
-- TODO: startupTracer should ignore severity level (i.e. it should always
-- be printed)!
, startupTracer = toLogObject' verb (appendName "startup" tr)
<> Tracer (\(ev :: StartupTrace blk) -> traceForgeEnabledMetric ekgDirect ev)
, shutdownTracer = toLogObject' verb $ appendName "shutdown" tr
-- The remaining tracers are completely unused by the legacy tracing:
, nodeInfoTracer = nullTracer
, nodeStartupInfoTracer = nullTracer
, nodeStateTracer = nullTracer
, resourcesTracer = nullTracer
, peersTracer = nullTracer
}
where
traceForgeEnabledMetric :: Maybe EKGDirect -> StartupTrace blk -> IO ()
traceForgeEnabledMetric mbEKGDirect ev =
case mbEKGDirect of
Just ekgDirect' ->
case ev of
BlockForgingUpdate b -> sendEKGDirectInt ekgDirect' "can_forge_blocks"
(case b of
EnabledBlockForging -> 1 :: Int
DisabledBlockForging -> 0 :: Int
NotEffective -> 0 :: Int)
_ -> pure ()
Nothing -> pure ()
diffusionTracers = Diffusion.Tracers
{ Diffusion.dtMuxTracer = muxTracer
, Diffusion.dtHandshakeTracer = handshakeTracer
, Diffusion.dtLocalMuxTracer = localMuxTracer
, Diffusion.dtLocalHandshakeTracer = localHandshakeTracer
, Diffusion.dtDiffusionTracer = initializationTracer
}
diffusionTracersExtra' enP2P =
case enP2P of
EnabledP2PMode ->
Diffusion.P2PTracers P2P.TracersExtra
{ P2P.dtTraceLocalRootPeersTracer =
tracerOnOff (traceLocalRootPeers trSel)
verb "LocalRootPeers" tr
, P2P.dtTracePublicRootPeersTracer =
tracerOnOff (tracePublicRootPeers trSel)
verb "PublicRootPeers" tr
, P2P.dtTracePeerSelectionTracer =
tracerOnOff (tracePeerSelection trSel)
verb "PeerSelection" tr
, P2P.dtDebugPeerSelectionInitiatorTracer =
tracerOnOff (traceDebugPeerSelectionInitiatorTracer trSel)
verb "DebugPeerSelection" tr
, P2P.dtDebugPeerSelectionInitiatorResponderTracer =
tracerOnOff (traceDebugPeerSelectionInitiatorResponderTracer trSel)
verb "DebugPeerSelection" tr
, P2P.dtTracePeerSelectionCounters =
tracePeerSelectionCountersMetrics
(tracePeerSelectionCounters trSel)
ekgDirect
<> tracerOnOff (tracePeerSelectionCounters trSel)
verb "PeerSelectionCounters" tr
, P2P.dtPeerSelectionActionsTracer =
tracerOnOff (tracePeerSelectionActions trSel)
verb "PeerSelectionActions" tr
, P2P.dtConnectionManagerTracer =
traceConnectionManagerTraceMetrics
(traceConnectionManagerCounters trSel)
ekgDirect
<> tracerOnOff (traceConnectionManager trSel)
verb "ConnectionManager" tr
, P2P.dtConnectionManagerTransitionTracer =
tracerOnOff (traceConnectionManagerTransitions trSel)
verb "ConnectionManagerTransition" tr
, P2P.dtServerTracer =
tracerOnOff (traceServer trSel) verb "Server" tr
, P2P.dtInboundGovernorTracer =
traceInboundGovernorCountersMetrics
(traceInboundGovernorCounters trSel)
ekgDirect
<> tracerOnOff (traceInboundGovernor trSel)
verb "InboundGovernor" tr
, P2P.dtInboundGovernorTransitionTracer =
tracerOnOff (traceInboundGovernorTransitions trSel)
verb "InboundGovernorTransition" tr
, P2P.dtLocalConnectionManagerTracer =
tracerOnOff (traceLocalConnectionManager trSel)
verb "LocalConnectionManager" tr
, P2P.dtLocalServerTracer =
tracerOnOff (traceLocalServer trSel)
verb "LocalServer" tr
, P2P.dtLocalInboundGovernorTracer =
tracerOnOff (traceLocalInboundGovernor trSel)
verb "LocalInboundGovernor" tr
, P2P.dtTraceLedgerPeersTracer =
tracerOnOff (traceLedgerPeers trSel)
verb "LedgerPeers" tr
}
DisabledP2PMode ->
Diffusion.NonP2PTracers NonP2P.TracersExtra
{ NonP2P.dtIpSubscriptionTracer =
tracerOnOff (traceIpSubscription trSel) verb "IpSubscription" tr
, NonP2P.dtDnsSubscriptionTracer =
tracerOnOff (traceDnsSubscription trSel) verb "DnsSubscription" tr
, NonP2P.dtDnsResolverTracer =
tracerOnOff (traceDnsResolver trSel) verb "DnsResolver" tr
, NonP2P.dtErrorPolicyTracer =
tracerOnOff (traceErrorPolicy trSel) verb "ErrorPolicy" tr
, NonP2P.dtLocalErrorPolicyTracer =
tracerOnOff (traceLocalErrorPolicy trSel) verb "LocalErrorPolicy" tr
, NonP2P.dtAcceptPolicyTracer =
tracerOnOff (traceAcceptPolicy trSel) verb "AcceptPolicy" tr
}
verb :: TracingVerbosity
verb = traceVerbosity trSel
muxTracer =
tracerOnOff (traceMux trSel) verb "Mux" tr
localMuxTracer =
tracerOnOff (traceLocalMux trSel) verb "MuxLocal" tr
localHandshakeTracer =
tracerOnOff (traceLocalHandshake trSel) verb "LocalHandshake" tr
handshakeTracer =
tracerOnOff (traceHandshake trSel) verb "Handshake" tr
initializationTracer =
tracerOnOff (traceDiffusionInitialization trSel) verb
"DiffusionInitializationTracer" tr
mkTracers _ _ _ _ _ enableP2P =
pure Tracers
{ chainDBTracer = nullTracer
, consensusTracers = Consensus.Tracers
{ Consensus.chainSyncClientTracer = nullTracer
, Consensus.chainSyncServerHeaderTracer = nullTracer
, Consensus.chainSyncServerBlockTracer = nullTracer
, Consensus.blockFetchDecisionTracer = nullTracer
, Consensus.blockFetchClientTracer = nullTracer
, Consensus.blockFetchServerTracer = nullTracer
, Consensus.keepAliveClientTracer = nullTracer
, Consensus.forgeStateInfoTracer = nullTracer
, Consensus.txInboundTracer = nullTracer
, Consensus.txOutboundTracer = nullTracer
, Consensus.localTxSubmissionServerTracer = nullTracer
, Consensus.mempoolTracer = nullTracer
, Consensus.forgeTracer = nullTracer
, Consensus.blockchainTimeTracer = nullTracer
, Consensus.consensusErrorTracer = nullTracer
, Consensus.gsmTracer = nullTracer
}
, nodeToClientTracers = NodeToClient.Tracers
{ NodeToClient.tChainSyncTracer = nullTracer
, NodeToClient.tTxMonitorTracer = nullTracer
, NodeToClient.tTxSubmissionTracer = nullTracer
, NodeToClient.tStateQueryTracer = nullTracer
}
, nodeToNodeTracers = NodeToNode.Tracers
{ NodeToNode.tChainSyncTracer = nullTracer
, NodeToNode.tChainSyncSerialisedTracer = nullTracer
, NodeToNode.tBlockFetchTracer = nullTracer
, NodeToNode.tBlockFetchSerialisedTracer = nullTracer
, NodeToNode.tTxSubmission2Tracer = nullTracer
}
, diffusionTracers = Diffusion.nullTracers
, diffusionTracersExtra =
case enableP2P of
EnabledP2PMode -> Diffusion.P2PTracers P2P.nullTracers
DisabledP2PMode -> Diffusion.NonP2PTracers NonP2P.nullTracers
, startupTracer = nullTracer
, shutdownTracer = nullTracer
, nodeInfoTracer = nullTracer
, nodeStartupInfoTracer = nullTracer
, nodeStateTracer = nullTracer
, resourcesTracer = nullTracer
, peersTracer = nullTracer
}
--------------------------------------------------------------------------------
-- Chain DB Tracers
--------------------------------------------------------------------------------
teeTraceChainTip
:: ( ConvertRawHash blk
, HasIssuer blk
, LedgerSupportsProtocol blk
, InspectLedger blk
, ToObject (Header blk)
, ToObject (LedgerEvent blk)
, ToObject (SelectView (BlockProtocol blk))
)
=> BlockConfig blk
-> ForgingStats
-> TraceOptions
-> MVar (Maybe (WithSeverity (ChainDB.TraceEvent blk)), Integer)
-> Maybe EKGDirect
-> STM.TVar Word64
-> Trace IO Text
-> Trace IO Text
-> Tracer IO (WithSeverity (ChainDB.TraceEvent blk))
teeTraceChainTip _ _ TracingOff _ _ _ _ _ = nullTracer
teeTraceChainTip _ _ TraceDispatcher{} _ _ _ _ _ = nullTracer
teeTraceChainTip blockConfig fStats (TracingOnLegacy trSel) elided ekgDirect tFork trTrc trMet =
Tracer $ \ev -> do
traceWith (teeTraceChainTipElide (traceVerbosity trSel) elided trTrc) ev
traceWith (ignoringSeverity (traceChainMetrics ekgDirect tFork blockConfig fStats trMet)) ev
teeTraceChainTipElide
:: ( ConvertRawHash blk
, LedgerSupportsProtocol blk
, InspectLedger blk
, ToObject (Header blk)
, ToObject (LedgerEvent blk)
, ToObject (SelectView (BlockProtocol blk))
)
=> TracingVerbosity
-> MVar (Maybe (WithSeverity (ChainDB.TraceEvent blk)), Integer)
-> Trace IO Text
-> Tracer IO (WithSeverity (ChainDB.TraceEvent blk))
teeTraceChainTipElide = elideToLogObject
{-# INLINE teeTraceChainTipElide #-}
ignoringSeverity :: Tracer IO a -> Tracer IO (WithSeverity a)
ignoringSeverity tr = Tracer $ \(WithSeverity _ ev) -> traceWith tr ev
{-# INLINE ignoringSeverity #-}
traceChainMetrics
:: forall blk. ()
=> HasHeader (Header blk)
=> ConvertRawHash blk
=> HasIssuer blk
=> Maybe EKGDirect
-> STM.TVar Word64
-> BlockConfig blk
-> ForgingStats
-> Trace IO Text
-> Tracer IO (ChainDB.TraceEvent blk)
traceChainMetrics Nothing _ _ _ _ = nullTracer
traceChainMetrics (Just _ekgDirect) tForks _blockConfig _fStats tr = do
Tracer $ \ev ->
maybe (pure ()) doTrace (chainTipInformation ev)
where
chainTipInformation :: ChainDB.TraceEvent blk -> Maybe ChainInformation
chainTipInformation = \case
ChainDB.TraceAddBlockEvent ev -> case ev of
ChainDB.SwitchedToAFork _warnings selChangedInfo oldChain newChain ->
let fork = not $ AF.withinFragmentBounds (AF.headPoint oldChain)
newChain in
Just $ chainInformation selChangedInfo fork oldChain newChain 0
ChainDB.AddedToCurrentChain _warnings selChangedInfo oldChain newChain ->
Just $ chainInformation selChangedInfo False oldChain newChain 0
_ -> Nothing
_ -> Nothing
doTrace :: ChainInformation -> IO ()
doTrace
ChainInformation { slots, blocks, density, epoch, slotInEpoch, fork, tipBlockHash, tipBlockParentHash, tipBlockIssuerVerificationKeyHash } = do
-- TODO this is executed each time the newChain changes. How cheap is it?
meta <- mkLOMeta Critical Public
traceD tr meta "density" (fromRational density)
traceI tr meta "slotNum" slots
traceI tr meta "blockNum" blocks
traceI tr meta "slotInEpoch" slotInEpoch
traceI tr meta "epoch" (unEpochNo epoch)
when fork $
traceI tr meta "forks" =<< STM.modifyReadTVarIO tForks succ
let tipBlockIssuerVkHashText :: Text
tipBlockIssuerVkHashText =
case tipBlockIssuerVerificationKeyHash of
NoBlockIssuer -> "NoBlockIssuer"
BlockIssuerVerificationKeyHash bs ->
Text.decodeLatin1 (B16.encode bs)
traceNamedObject
(appendName "tipBlockHash" tr)
(meta, LogMessage tipBlockHash)
traceNamedObject
(appendName "tipBlockParentHash" tr)
(meta, LogMessage tipBlockParentHash)
traceNamedObject
(appendName "tipBlockIssuerVerificationKeyHash" tr)
(meta, LogMessage tipBlockIssuerVkHashText)
traceD :: Trace IO a -> LOMeta -> Text -> Double -> IO ()
traceD tr meta msg d = traceNamedObject tr (meta, LogValue msg (PureD d))
traceI :: Integral i => Trace IO a -> LOMeta -> Text -> i -> IO ()
traceI tr meta msg i = traceNamedObject tr (meta, LogValue msg (PureI (fromIntegral i)))
sendEKGDirectCounter :: EKGDirect -> Text -> IO ()
sendEKGDirectCounter ekgDirect name = do
modifyMVar_ (ekgCounters ekgDirect) $ \registeredMap -> do
case Map.lookup name registeredMap of
Just counter -> do
Counter.inc counter
pure registeredMap
Nothing -> do
counter <- EKG.getCounter name (ekgServer ekgDirect)
Counter.inc counter
pure $ Map.insert name counter registeredMap
sendEKGDirectInt :: Integral a => EKGDirect -> Text -> a -> IO ()
sendEKGDirectInt ekgDirect name val = do
modifyMVar_ (ekgGauges ekgDirect) $ \registeredMap -> do
case Map.lookup name registeredMap of
Just gauge -> do
Gauge.set gauge (fromIntegral val)
pure registeredMap
Nothing -> do
gauge <- EKG.getGauge name (ekgServer ekgDirect)
Gauge.set gauge (fromIntegral val)
pure $ Map.insert name gauge registeredMap
sendEKGDirectDouble :: EKGDirect -> Text -> Double -> IO ()
sendEKGDirectDouble ekgDirect name val = do
modifyMVar_ (ekgLabels ekgDirect) $ \registeredMap -> do
case Map.lookup name registeredMap of
Just label -> do
Label.set label (Text.pack (show val))
pure registeredMap
Nothing -> do
label <- EKG.getLabel name (ekgServer ekgDirect)
Label.set label (Text.pack (show val))
pure $ Map.insert name label registeredMap
--------------------------------------------------------------------------------
-- Consensus Tracers
--------------------------------------------------------------------------------
isRollForward :: TraceChainSyncServerEvent blk -> Bool
isRollForward (TraceChainSyncServerUpdate _tip (AddBlock _pt) _blocking FallingEdge) = True
isRollForward _ = False
mkConsensusTracers
:: forall blk peer localPeer.
( Show peer
, Eq peer
, LedgerQueries blk
, ToJSON (GenTxId blk)
, ToObject (ApplyTxErr blk)
, ToObject (CannotForge blk)
, ToObject (GenTx blk)
, ToObject (LedgerErr (LedgerState blk))
, ToObject (OtherHeaderEnvelopeError blk)
, ToObject (ValidationErr (BlockProtocol blk))
, ToObject (ForgeStateUpdateError blk)
, ToObject peer
, Consensus.RunNode blk
, HasKESMetricsData blk
, HasKESInfo blk
)
=> Maybe EKGDirect
-> TraceSelection
-> TracingVerbosity
-> Trace IO Text
-> NodeKernelData blk
-> ForgingStats
-> IO (Consensus.Tracers' peer localPeer blk (Tracer IO))
mkConsensusTracers mbEKGDirect trSel verb tr nodeKern fStats = do
let trmet = appendName "metrics" tr
elidedFetchDecision <- newstate -- for eliding messages in FetchDecision tr
forgeTracers <- mkForgeTracers
meta <- mkLOMeta Critical Public
tBlocksServed <- STM.newTVarIO 0
tLocalUp <- STM.newTVarIO 0
tMaxSlotNo <- STM.newTVarIO $ SlotNo 0
tSubmissionsCollected <- STM.newTVarIO 0
tSubmissionsAccepted <- STM.newTVarIO 0
tSubmissionsRejected <- STM.newTVarIO 0
tBlockDelayM <- STM.newTVarIO Pq.empty
tBlockDelayCDF1s <- STM.newTVarIO $ CdfCounter 0
tBlockDelayCDF3s <- STM.newTVarIO $ CdfCounter 0
tBlockDelayCDF5s <- STM.newTVarIO $ CdfCounter 0
pure Consensus.Tracers
{ Consensus.chainSyncClientTracer = tracerOnOff (traceChainSyncClient trSel) verb "ChainSyncClient" tr
, Consensus.chainSyncServerHeaderTracer =
tracerOnOff' (traceChainSyncHeaderServer trSel)
(annotateSeverity . toLogObject' verb $ appendName "ChainSyncHeaderServer" tr)
<> (\(TraceLabelPeer _ ev) -> ev) `contramap` Tracer (traceServedCount mbEKGDirect)
, Consensus.chainSyncServerBlockTracer = tracerOnOff (traceChainSyncBlockServer trSel) verb "ChainSyncBlockServer" tr
, Consensus.blockFetchDecisionTracer = tracerOnOff' (traceBlockFetchDecisions trSel) $
annotateSeverity $ teeTraceBlockFetchDecision verb elidedFetchDecision tr
, Consensus.blockFetchClientTracer = traceBlockFetchClientMetrics mbEKGDirect tBlockDelayM
tBlockDelayCDF1s tBlockDelayCDF3s tBlockDelayCDF5s $
tracerOnOff (traceBlockFetchClient trSel) verb "BlockFetchClient" tr
, Consensus.blockFetchServerTracer = traceBlockFetchServerMetrics trmet meta tBlocksServed
tLocalUp tMaxSlotNo $ tracerOnOff (traceBlockFetchServer trSel) verb "BlockFetchServer" tr
, Consensus.keepAliveClientTracer = tracerOnOff (traceKeepAliveClient trSel) verb "KeepAliveClient" tr
, Consensus.forgeStateInfoTracer = tracerOnOff' (traceForgeStateInfo trSel) $
forgeStateInfoTracer (Proxy @blk) trSel tr
, Consensus.txInboundTracer = tracerOnOff' (traceTxInbound trSel) $
Tracer $ \ev -> do
traceWith (annotateSeverity . toLogObject' verb $ appendName "TxInbound" tr) ev
case ev of
TraceLabelPeer _ (TraceTxSubmissionCollected collected) ->
traceI trmet meta "submissions.submitted.count" =<<
STM.modifyReadTVarIO tSubmissionsCollected (+ collected)
TraceLabelPeer _ (TraceTxSubmissionProcessed processed) -> do
traceI trmet meta "submissions.accepted.count" =<<
STM.modifyReadTVarIO tSubmissionsAccepted (+ ptxcAccepted processed)
traceI trmet meta "submissions.rejected.count" =<<
STM.modifyReadTVarIO tSubmissionsRejected (+ ptxcRejected processed)
TraceLabelPeer _ TraceTxInboundTerminated -> return ()
TraceLabelPeer _ (TraceTxInboundCanRequestMoreTxs _) -> return ()
TraceLabelPeer _ (TraceTxInboundCannotRequestMoreTxs _) -> return ()
, Consensus.txOutboundTracer = tracerOnOff (traceTxOutbound trSel) verb "TxOutbound" tr
, Consensus.localTxSubmissionServerTracer = tracerOnOff (traceLocalTxSubmissionServer trSel) verb "LocalTxSubmissionServer" tr
, Consensus.mempoolTracer = tracerOnOff' (traceMempool trSel) $ mempoolTracer trSel tr fStats
, Consensus.forgeTracer = tracerOnOff' (traceForge trSel) $
Tracer $ \tlcev@Consensus.TraceLabelCreds{} -> do
traceWith (annotateSeverity
$ traceLeadershipChecks forgeTracers nodeKern verb tr) tlcev
traceWith (forgeTracer verb tr forgeTracers fStats) tlcev
, Consensus.blockchainTimeTracer = tracerOnOff' (traceBlockchainTime trSel) $
Tracer $ \ev ->
traceWith (toLogObject tr) (readableTraceBlockchainTimeEvent ev)
, Consensus.consensusErrorTracer =
Tracer $ \err -> traceWith (toLogObject tr) (ConsensusStartupException err)
, Consensus.gsmTracer = tracerOnOff (traceGsm trSel) verb "GSM" tr
}
where
mkForgeTracers :: IO ForgeTracers
mkForgeTracers = do
-- We probably don't want to pay the extra IO cost per-counter-increment. -- sk
staticMeta <- mkLOMeta Critical Confidential
let name :: LoggerName = "metrics.Forge"
ForgeTracers
<$> counting (liftCounting staticMeta name "forged" tr)
<*> counting (liftCounting staticMeta name "forge-about-to-lead" tr)
<*> counting (liftCounting staticMeta name "could-not-forge" tr)
<*> counting (liftCounting staticMeta name "adopted" tr)
<*> counting (liftCounting staticMeta name "didnt-adopt" tr)
<*> counting (liftCounting staticMeta name "forged-invalid" tr)
<*> counting (liftCounting staticMeta name "node-not-leader" tr)
<*> counting (liftCounting staticMeta name "cannot-forge" tr)
<*> counting (liftCounting staticMeta name "forge-state-update-error" tr)
<*> counting (liftCounting staticMeta name "block-from-future" tr)
<*> counting (liftCounting staticMeta name "slot-is-immutable" tr)
<*> counting (liftCounting staticMeta name "node-is-leader" tr)
<*> counting (liftCounting staticMeta name "adoption-thread-died" tr)
traceServedCount :: Maybe EKGDirect -> TraceChainSyncServerEvent blk -> IO ()
traceServedCount Nothing _ = pure ()
traceServedCount (Just ekgDirect) ev =
when (isRollForward ev) $
sendEKGDirectCounter ekgDirect
"cardano.node.metrics.served.header.counter.int"
traceBlockFetchServerMetrics
:: forall blk peer. ()
=> Tracer IO (LoggerName, LogObject Text)
-> LOMeta
-> STM.TVar Int64
-> STM.TVar Int64
-> STM.TVar SlotNo
-> Tracer IO (TraceLabelPeer peer (TraceBlockFetchServerEvent blk))
-> Tracer IO (TraceLabelPeer peer (TraceBlockFetchServerEvent blk))
traceBlockFetchServerMetrics trMeta meta tBlocksServed tLocalUp tMaxSlotNo tracer = Tracer bsTracer
where
bsTracer :: TraceLabelPeer peer (TraceBlockFetchServerEvent blk) -> IO ()
bsTracer e@(TraceLabelPeer _p (TraceBlockFetchServerSendBlock p)) = do
traceWith tracer e
(served, mbLocalUpstreamyness) <- atomically $ do
served <- STM.modifyReadTVar' tBlocksServed (+1)
maxSlotNo <- STM.readTVar tMaxSlotNo
case pointSlot p of
Origin -> return (served, Nothing)
At slotNo ->
case compare maxSlotNo slotNo of
LT -> do
STM.writeTVar tMaxSlotNo slotNo
lu <- STM.modifyReadTVar' tLocalUp (+1)
return (served, Just lu)
GT -> do
return (served, Nothing)
EQ -> do
lu <- STM.modifyReadTVar' tLocalUp (+1)
return (served, Just lu)
traceI trMeta meta "served.block.count" served
forM_ mbLocalUpstreamyness $ \localUpstreamyness ->
traceI trMeta meta "served.block.latest.count" localUpstreamyness
-- | CdfCounter tracks the number of time a value below 'limit' has been seen.
newtype CdfCounter (limit :: Nat) = CdfCounter Int64
-- | Estimates the CDF for a specific limit 'l' by counting the number of times
-- a value 'v' is below the limit.
cdfCounter :: forall a l.
( Num a, Ord a
, KnownNat l)
=> a -> Int -> Int64 -> STM.TVar (CdfCounter l) -> STM Double
cdfCounter v !size !step tCdf= do
when (v < lim) $
STM.modifyTVar' tCdf (\(CdfCounter c) -> CdfCounter $ c + step)
(CdfCounter cdf) <- STM.readTVar tCdf
return $! (fromIntegral cdf / fromIntegral size)
where
lim :: a
lim = fromInteger $ natVal (Proxy :: Proxy l)
-- Add an observation to the CdfCounter.
incCdfCounter :: Ord a => Num a => KnownNat l => a -> Int -> STM.TVar (CdfCounter l) -> STM Double
incCdfCounter v size = cdfCounter v size 1
-- Remove an observation from the CdfCounter.
decCdfCounter :: Ord a => Num a => KnownNat l => a -> Int -> STM.TVar (CdfCounter l) -> STM Double
decCdfCounter v size = cdfCounter v size (-1)
-- Track the fraction of times forgeDelay was above 1s, 3s, and 5s.
-- Only the first sample per slot number is counted.
cdf135Counters
:: Integral a
=> STM.TVar (IntPSQ a NominalDiffTime)
-> STM.TVar (CdfCounter 1)
-> STM.TVar (CdfCounter 3)
-> STM.TVar (CdfCounter 5)
-> a
-> NominalDiffTime
-> STM (Bool, Double, Double, Double)
cdf135Counters slotMapVar cdf1sVar cdf3sVar cdf5sVar slotNo forgeDelay = do
slotMap <- STM.readTVar slotMapVar
if Pq.null slotMap && forgeDelay > 20
then return (False, 0, 0, 0) -- During startup wait until we are in sync
else case Pq.lookup (fromIntegral slotNo) slotMap of
Nothing -> do
let slotMap' = Pq.insert (fromIntegral slotNo) slotNo forgeDelay slotMap
if Pq.size slotMap' > 1080 -- TODO k/2, should come from config file
then
case Pq.minView slotMap' of
Nothing -> return (False, 0, 0, 0) -- Err. We just inserted an element!
Just (_, minSlotNo, minDelay, slotMap'') ->
if minSlotNo == slotNo
then return (False, 0, 0, 0) -- Nothing to do
else do
decCdfs minDelay (Pq.size slotMap'')
(cdf1s, cdf3s, cdf5s) <- incCdfs forgeDelay (Pq.size slotMap'')
STM.writeTVar slotMapVar slotMap''
return (True, cdf1s, cdf3s, cdf5s)
else do
(cdf1s, cdf3s, cdf5s) <- incCdfs forgeDelay (Pq.size slotMap')
STM.writeTVar slotMapVar slotMap'
-- Wait until we have at least 45 samples before we start providing
-- cdf estimates.
if Pq.size slotMap >= 45
then return (True, cdf1s, cdf3s, cdf5s)
else return (True, -1, -1, -1)
Just _ -> return (False, 0, 0, 0) -- dupe, we only track the first
where
incCdfs :: NominalDiffTime -> Int -> STM (Double, Double, Double)
incCdfs delay size = do
cdf1s <- incCdfCounter delay size cdf1sVar
cdf3s <- incCdfCounter delay size cdf3sVar
cdf5s <- incCdfCounter delay size cdf5sVar
return (cdf1s, cdf3s, cdf5s)
decCdfs :: NominalDiffTime -> Int -> STM ()
decCdfs delay size =
decCdfCounter delay size cdf1sVar
>> decCdfCounter delay size cdf3sVar
>> decCdfCounter delay size cdf5sVar
>> return ()
traceBlockFetchClientMetrics
:: forall blk remotePeer.
( )
=> Maybe EKGDirect
-> STM.TVar (IntPSQ Word64 NominalDiffTime)
-> STM.TVar (CdfCounter 1)
-> STM.TVar (CdfCounter 3)
-> STM.TVar (CdfCounter 5)
-> Tracer IO (TraceLabelPeer remotePeer (TraceFetchClientState (Header blk)))
-> Tracer IO (TraceLabelPeer remotePeer (TraceFetchClientState (Header blk)))
traceBlockFetchClientMetrics Nothing _ _ _ _ tracer = tracer
traceBlockFetchClientMetrics (Just ekgDirect) slotMapVar cdf1sVar cdf3sVar cdf5sVar tracer = Tracer bfTracer
where
bfTracer :: TraceLabelPeer remotePeer (TraceFetchClientState (Header blk)) -> IO ()
bfTracer e@(TraceLabelPeer _ (CompletedBlockFetch p _ _ _ delay blockSize)) = do
traceWith tracer e
case pointSlot p of
Origin -> return () -- Nothing to do.
At slotNo -> do
(fresh, cdf1s, cdf3s, cdf5s) <- atomically $
cdf135Counters slotMapVar cdf1sVar cdf3sVar cdf5sVar (slotMapKey slotNo) delay
when fresh $ do
-- TODO: Revisit ekg counter access once there is a faster way.
sendEKGDirectDouble ekgDirect "cardano.node.metrics.blockfetchclient.blockdelay.s"
$ realToFrac delay
sendEKGDirectInt ekgDirect "cardano.node.metrics.blockfetchclient.blocksize"
blockSize
when (cdf1s >= 0) $
sendEKGDirectDouble ekgDirect
"cardano.node.metrics.blockfetchclient.blockdelay.cdfOne"
cdf1s
when (cdf3s >= 0) $
sendEKGDirectDouble ekgDirect
"cardano.node.metrics.blockfetchclient.blockdelay.cdfThree"
cdf3s
when (cdf5s >= 0) $
sendEKGDirectDouble ekgDirect
"cardano.node.metrics.blockfetchclient.blockdelay.cdfFive"
cdf5s
when (delay > 5) $
sendEKGDirectCounter ekgDirect "cardano.node.metrics.blockfetchclient.lateblocks"
bfTracer e =
traceWith tracer e
slotMapKey :: SlotNo -> Word64
slotMapKey (SlotNo s) = s
traceLeadershipChecks ::
forall blk
. ( Consensus.RunNode blk
, LedgerQueries blk
)
=> ForgeTracers
-> NodeKernelData blk
-> TracingVerbosity
-> Trace IO Text
-> Tracer IO (WithSeverity (Consensus.TraceLabelCreds (Consensus.TraceForgeEvent blk)))
traceLeadershipChecks _ft nodeKern _tverb tr = Tracer $
\(WithSeverity sev (Consensus.TraceLabelCreds creds event)) ->
case event of
Consensus.TraceStartLeadershipCheck slot -> do
!query <- mapNodeKernelDataIO
(\nk ->
(,,)
<$> nkQueryLedger (ledgerUtxoSize . ledgerState) nk
<*> nkQueryLedger (ledgerDelegMapSize . ledgerState) nk
<*> nkQueryChain fragmentChainDensity nk)
nodeKern
meta <- mkLOMeta sev Public
fromSMaybe (pure ()) $
query <&>
\(utxoSize, delegMapSize, _) -> do
traceCounter "utxoSize" tr utxoSize
traceCounter "delegMapSize" tr delegMapSize
traceNamedObject (appendName "LeadershipCheck" tr)
( meta
, LogStructured $ KeyMap.fromList $
[("kind", String "TraceStartLeadershipCheck")
,("credentials", String creds)
,("slot", toJSON $ unSlotNo slot)]