New tracing: new features and some maintenance

bench/tx-generator/src/Cardano/Benchmarking/LogTypes.hs

@@ -123,6 +123,7 @@ data TraceBenchTxSubmit txid
   | TraceBenchTxSubError Text
   | TraceBenchPlutusBudgetSummary PlutusBudgetSummary
   -- ^ PlutusBudgetSummary.
+  | TraceBenchForwardingInterrupted HowToConnect String
   deriving stock (Show, Generic)
 
 data SubmissionSummary

bench/tx-generator/src/Cardano/Benchmarking/Tracer.hs

@@ -11,6 +11,7 @@
 {-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE PatternSynonyms #-}
 {-# LANGUAGE RankNTypes #-}
+{-# LANGUAGE RecursiveDo #-}
 {-# LANGUAGE ScopedTypeVariables #-}
 {-# LANGUAGE TypeApplications #-}
 {-# LANGUAGE TypeOperators #-}
@@ -37,6 +38,7 @@ import           Cardano.Node.Startup
 import           Cardano.Node.Tracing.NodeInfo ()
 import           Ouroboros.Network.IOManager (IOManager)
 
+import           Control.Exception (SomeException (..))
 import           Control.Monad (forM, guard)
 import           Data.Aeson as A
 import qualified Data.Aeson.KeyMap as KeyMap
@@ -49,7 +51,7 @@ import qualified Data.Text as Text
 import           Data.Time.Clock
 import           GHC.Generics
 
-import           Trace.Forward.Forwarding (initForwardingDelayed)
+import           Trace.Forward.Forwarding (InitForwardingConfig (..), initForwardingDelayed)
 import           Trace.Forward.Utils.TraceObject
 
 pattern TracerNameBench     :: Text
@@ -85,9 +87,9 @@ initNullTracers = BenchTracers
 -- if the first argument isJust, we assume we have a socket path
 -- and want to use trace-dispatcher, so we'll create a forwarding tracer
 initTxGenTracers :: Maybe (IOManager, NetworkId, FilePath) -> IO BenchTracers
-initTxGenTracers mbForwarding = do
+initTxGenTracers mbForwarding = mdo
   mbStdoutTracer <- fmap Just standardTracer
-  mbForwardingTracer <- prepareForwardingTracer
+  mbForwardingTracer <- prepareForwardingTracer tracers
   confState       <- emptyConfigReflection
 
   let
@@ -108,21 +110,34 @@ initTxGenTracers mbForwarding = do
   connectTracer   <- mkTracer TracerNameConnect   mbStdoutTracer mbForwardingTracer
   submitTracer    <- mkTracer TracerNameSubmit    mbStdoutTracer mbForwardingTracer
 
-  traceWith benchTracer (TraceTxGeneratorVersion Version.txGeneratorVersion)
-
-  return $ BenchTracers
-    { btTxSubmit_    = benchTracer
-    , btConnect_     = connectTracer
-    , btSubmission2_ = submitTracer
-    , btN2N_         = n2nSubmitTracer
-    }
+  let
+    tracers = BenchTracers
+      { btTxSubmit_    = benchTracer
+      , btConnect_     = connectTracer
+      , btSubmission2_ = submitTracer
+      , btN2N_         = n2nSubmitTracer
+      }
+
+  traceWith (btTxSubmit_ tracers) (TraceTxGeneratorVersion Version.txGeneratorVersion)
+  return tracers
  where
-  prepareForwardingTracer :: IO (Maybe (Trace IO FormattedMessage))
-  prepareForwardingTracer = forM mbForwarding $
+  prepareForwardingTracer :: BenchTracers -> IO (Maybe (Trace IO FormattedMessage))
+  prepareForwardingTracer benchTracer = forM mbForwarding $
     \(iomgr, networkId, tracerSocket) -> do
-        let forwardingConf = fromMaybe defaultForwarder (tcForwarder initialTraceConfig)
+        let
+          forwardingConf = fromMaybe defaultForwarder (tcForwarder initialTraceConfig)
+          howToConnect = Net.LocalPipe tracerSocket
+          initForwConf = InitForwardingWith
+            { initNetworkMagic          = toNetworkMagic networkId
+            , initEKGStore              = Nothing
+            , initHowToConnect          = howToConnect
+            , initForwarderMode         = Initiator
+            , initOnForwardInterruption = Just $ \(SomeException e) ->
+                traceWith (btTxSubmit_ benchTracer) (TraceBenchForwardingInterrupted howToConnect $ show e)
+            , initOnQueueOverflow       = Nothing
+            }
         (forwardSink, dpStore, kickoffForwarder) <-
-            initForwardingDelayed iomgr forwardingConf (toNetworkMagic networkId) Nothing $ Just (Net.LocalPipe tracerSocket, Initiator)
+            initForwardingDelayed iomgr forwardingConf initForwConf
 
         -- we need to provide NodeInfo DataPoint, to forward generator's name
         -- to the acceptor application (for example, 'cardano-tracer').
@@ -205,8 +220,14 @@ instance (ConstructorsOf f, ConstructorsOf g) => ConstructorsOf (f :+: g) where
 instance (Constructor ('MetaCons n f r)) => ConstructorsOf (C1 ('MetaCons n f r) x) where
   constructorsOf _ = [ conName @('MetaCons n f r) undefined ]
 
+
 instance LogFormatting (TraceBenchTxSubmit TxId) where
-  forHuman = Text.pack . show
+  forHuman = \case
+    TraceBenchForwardingInterrupted howToConnect errMsg ->
+      Text.pack $ "trace forwarding connection with " <> show howToConnect <> " failed: " <> errMsg
+    _
+      -> ""
+
   forMachine DMinimal _ = mempty
   forMachine DNormal t = mconcat [ "kind" .= A.String (genericName t) ]
   forMachine DDetailed t = forMachine DMaximum t
@@ -276,6 +297,11 @@ instance LogFormatting (TraceBenchTxSubmit TxId) where
       mconcat [ "kind"    .= A.String "TraceBenchPlutusBudgetSummary"
               , "summary" .= toJSON summary
               ]
+    TraceBenchForwardingInterrupted howToConnect msg ->
+      mconcat [ "kind"    .= A.String "TraceBenchForwardingInterrupted"
+              , "conn"    .= howToConnect
+              , "msg"     .= msg
+              ]
 
 instance MetaTrace (TraceBenchTxSubmit TxId) where
     namespaceFor TraceTxGeneratorVersion {} = Namespace [] ["TxGeneratorVersion"]
@@ -295,6 +321,7 @@ instance MetaTrace (TraceBenchTxSubmit TxId) where
     namespaceFor TraceBenchTxSubDebug {} = Namespace [] ["BenchTxSubDebug"]
     namespaceFor TraceBenchTxSubError {} = Namespace [] ["BenchTxSubError"]
     namespaceFor TraceBenchPlutusBudgetSummary {} = Namespace [] ["BenchPlutusBudgetSummary"]
+    namespaceFor TraceBenchForwardingInterrupted {} = Namespace [] ["ForwardingInterrupted"]
 
     severityFor _ _ = Just Info
 
@@ -318,6 +345,7 @@ instance MetaTrace (TraceBenchTxSubmit TxId) where
         , Namespace [] ["BenchTxSubDebug"]
         , Namespace [] ["BenchTxSubError"]
         , Namespace [] ["BenchPlutusBudgetSummary"]
+        , Namespace [] ["ForwardingInterrupted"]
         ]
 
 instance LogFormatting NodeToNodeSubmissionTrace where

cardano-node/cardano-node.cabal

@@ -211,8 +211,8 @@ library
                       , sop-extras
                       , text >= 2.0
                       , time
-                      , trace-dispatcher ^>= 2.10.0
-                      , trace-forward ^>= 2.3.0
+                      , trace-dispatcher ^>= 2.11.0
+                      , trace-forward ^>= 2.4.0
                       , trace-resources ^>= 0.2.4
                       , tracer-transformers
                       , transformers

cardano-node/src/Cardano/Node/Tracing/API.hs

@@ -2,6 +2,8 @@
 {-# LANGUAGE MonoLocalBinds #-}
 {-# LANGUAGE PackageImports #-}
 {-# LANGUAGE RankNTypes #-}
+{-# LANGUAGE RecordWildCards #-}
+{-# LANGUAGE RecursiveDo #-}
 {-# LANGUAGE ScopedTypeVariables #-}
 
 module Cardano.Node.Tracing.API
@@ -36,6 +38,7 @@ import           Ouroboros.Network.NodeToNode (RemoteAddress)
 import           Prelude
 
 import           Control.DeepSeq (deepseq)
+import           Control.Exception (SomeException (..))
 import           Control.Monad (forM_)
 import           "contra-tracer" Control.Tracer (traceWith)
 import           "trace-dispatcher" Control.Tracer (nullTracer)
@@ -46,7 +49,7 @@ import           Network.Mux.Trace (TraceLabelPeer (..))
 import           Network.Socket (HostName)
 import           System.Metrics as EKG
 
-import           Trace.Forward.Forwarding (initForwardingDelayed)
+import           Trace.Forward.Forwarding (InitForwardingConfig (..), initForwardingDelayed)
 import           Trace.Forward.Utils.TraceObject (writeToSink)
 
 
@@ -111,7 +114,7 @@ initTraceDispatcher nc p networkMagic nodeKernel noBlockForging = do
           , IO (Maybe String)
           , Tracers RemoteAddress LocalAddress blk IO
           )
-  mkTracers trConfig = do
+  mkTracers trConfig = mdo
     ekgStore <- EKG.newStore
     EKG.registerGcMetrics ekgStore
     ekgTrace <- ekgTracer trConfig ekgStore
@@ -126,28 +129,39 @@ initTraceDispatcher nc p networkMagic nodeKernel noBlockForging = do
       if forwarderBackendEnabled
         then do
           -- TODO: check if this is the correct way to use withIOManager
-          (forwardSink, dpStore, kickoffForwarder) <- withIOManager $ \iomgr -> do
-            let tracerSocketMode :: Maybe (HowToConnect, ForwarderMode)
-                tracerSocketMode = ncTraceForwardSocket nc
+          (forwardSink, dpStore, kickoffForwarder') <- withIOManager $ \iomgr ->
+            let initForwConf :: InitForwardingConfig
+                initForwConf = case ncTraceForwardSocket nc of
+                  Nothing -> InitForwardingNone
+                  Just (initHowToConnect, initForwarderMode) ->
+                    InitForwardingWith
+                      { initNetworkMagic          = networkMagic
+                      , initEKGStore              = Just ekgStore
+                      , initOnForwardInterruption = Just $ \(SomeException e) ->
+                          traceWith (nodeStateTracer tracers) (NodeTracingForwardingInterrupted initHowToConnect $ show e)
+                      , initOnQueueOverflow       = Nothing
+                      , ..
+                      }
 
                 forwardingConf :: TraceOptionForwarder
                 forwardingConf = fromMaybe defaultForwarder (tcForwarder trConfig)
-            initForwardingDelayed iomgr forwardingConf networkMagic (Just ekgStore) tracerSocketMode
-          pure (forwardTracer (writeToSink forwardSink), dataPointTracer dpStore, kickoffForwarder)
+            in initForwardingDelayed iomgr forwardingConf initForwConf
+
+          pure (forwardTracer (writeToSink forwardSink), dataPointTracer dpStore, kickoffForwarder')
         else
           -- Since 'Forwarder' backend isn't enabled, there is no forwarding.
           -- So we use nullTracers to ignore 'TraceObject's and 'DataPoint's.
           pure (Trace nullTracer, Trace nullTracer, pure ())
 
-    (,,) kickoffForwarder kickoffPrometheusSimple
-      <$> mkDispatchTracers
-        nodeKernel
-        stdoutTrace
-        fwdTracer
-        (Just ekgTrace)
-        dpTracer
-        trConfig
-        p
+    tracers <- mkDispatchTracers
+      nodeKernel
+      stdoutTrace
+      fwdTracer
+      (Just ekgTrace)
+      dpTracer
+      trConfig
+      p
+    pure (kickoffForwarder, kickoffPrometheusSimple, tracers)
 
    where
     -- This backend can only be used globally, i.e. will always apply to the namespace root.

cardano-node/src/Cardano/Node/Tracing/StateRep.hs

@@ -99,6 +99,7 @@ deriving instance (NFData StartupState)
 data NodeState
   = NodeTracingOnlineConfiguring
   | NodeTracingFailure String
+  | NodeTracingForwardingInterrupted HowToConnect String
   | NodeOpeningDbs OpeningDbs
   | NodeReplays Replays
   | NodeInitChainSelection InitChainSelection
@@ -130,16 +131,27 @@ instance LogFormatting NodeState where
       [ "kind" .= String "NodeShutdown",           "shutdown"  .= toJSON x]
     NodeTracingFailure x -> mconcat
       [ "kind" .= String "NodeTracingFailure",     "message"   .= toJSON x]
-
-  forHuman (NodeTracingFailure errMsg) = T.pack errMsg
-  forHuman _ = ""
-
+    NodeTracingForwardingInterrupted howToConnect x -> mconcat
+      [ "kind" .= String "NodeTracingForwardingInterrupted"
+      , "conn" .= howToConnect
+      , "message" .= toJSON x
+      ]
+
+  forHuman = \case
+    NodeTracingFailure errMsg ->
+      T.pack errMsg
+    NodeTracingForwardingInterrupted howToConnect errMsg ->
+      T.pack $ "trace forwarding connection with " <> show howToConnect <> " failed: " <> errMsg
+    _
+      -> ""
 
 instance MetaTrace NodeState where
   namespaceFor NodeTracingOnlineConfiguring {}  =
     Namespace [] ["NodeTracingOnlineConfiguring"]
-  namespaceFor NodeTracingFailure {}  =
+  namespaceFor NodeTracingFailure {} =
     Namespace [] ["NodeTracingFailure"]
+  namespaceFor NodeTracingForwardingInterrupted {} =
+    Namespace [] ["NodeTracingForwardingInterrupted"]
   namespaceFor NodeOpeningDbs {}  =
     Namespace [] ["OpeningDbs"]
   namespaceFor NodeReplays {}  =
@@ -159,6 +171,8 @@ instance MetaTrace NodeState where
     Just Info
   severityFor  (Namespace _ ["NodeTracingFailure"]) _ =
     Just Error
+  severityFor  (Namespace _ ["NodeTracingForwardingInterrupted"]) _ =
+    Just Warning
   severityFor  (Namespace _ ["OpeningDbs"]) _ =
     Just Info
   severityFor  (Namespace _ ["NodeReplays"]) _ =
@@ -180,6 +194,8 @@ instance MetaTrace NodeState where
     "Tracing system came online, system configuring now"
   documentFor  (Namespace _ ["NodeTracingFailure"]) = Just
     "Tracing system experienced a non-fatal failure during startup"
+  documentFor  (Namespace _ ["NodeTracingForwardingInterrupted"]) = Just
+    "Trace/metrics forwarding connection was interrupted"
   documentFor  (Namespace _ ["OpeningDbs"]) = Just
     "ChainDB components being opened"
   documentFor  (Namespace _ ["NodeReplays"]) = Just
@@ -199,6 +215,7 @@ instance MetaTrace NodeState where
   allNamespaces = [
           Namespace [] ["NodeTracingOnlineConfiguring"]
         , Namespace [] ["NodeTracingFailure"]
+        , Namespace [] ["NodeTracingForwardingInterrupted"]
         , Namespace [] ["OpeningDbs"]
         , Namespace [] ["NodeReplays"]
         , Namespace [] ["NodeInitChainSelection"]

cardano-node/src/Cardano/Node/Tracing/Tracers/ChainDB.hs

@@ -88,16 +88,16 @@ instance (  LogFormatting (Header blk)
           ) => LogFormatting (ChainDB.TraceEvent blk) where
   forHuman ChainDB.TraceLastShutdownUnclean        =
     "ChainDB is not clean. Validating all immutable chunks"
-  forHuman (ChainDB.TraceAddBlockEvent v)          = forHumanOrMachine v
-  forHuman (ChainDB.TraceFollowerEvent v)          = forHumanOrMachine v
-  forHuman (ChainDB.TraceCopyToImmutableDBEvent v) = forHumanOrMachine v
-  forHuman (ChainDB.TraceGCEvent v)                = forHumanOrMachine v
-  forHuman (ChainDB.TraceInitChainSelEvent v)      = forHumanOrMachine v
-  forHuman (ChainDB.TraceOpenEvent v)              = forHumanOrMachine v
-  forHuman (ChainDB.TraceIteratorEvent v)          = forHumanOrMachine v
-  forHuman (ChainDB.TraceLedgerDBEvent v)          = forHumanOrMachine v
-  forHuman (ChainDB.TraceImmutableDBEvent v)       = forHumanOrMachine v
-  forHuman (ChainDB.TraceVolatileDBEvent v)        = forHumanOrMachine v
+  forHuman (ChainDB.TraceAddBlockEvent v)          = forHuman v
+  forHuman (ChainDB.TraceFollowerEvent v)          = forHuman v
+  forHuman (ChainDB.TraceCopyToImmutableDBEvent v) = forHuman v
+  forHuman (ChainDB.TraceGCEvent v)                = forHuman v
+  forHuman (ChainDB.TraceInitChainSelEvent v)      = forHuman v
+  forHuman (ChainDB.TraceOpenEvent v)              = forHuman v
+  forHuman (ChainDB.TraceIteratorEvent v)          = forHuman v
+  forHuman (ChainDB.TraceLedgerDBEvent v)          = forHuman v
+  forHuman (ChainDB.TraceImmutableDBEvent v)       = forHuman v
+  forHuman (ChainDB.TraceVolatileDBEvent v)        = forHuman v
   forHuman (ChainDB.TraceChainSelStarvationEvent ev) = case ev of
         ChainDB.ChainSelStarvation RisingEdge ->
           "Chain Selection was starved."
@@ -429,12 +429,12 @@ instance ( LogFormatting (Header blk)
   forHuman (ChainDB.SwitchedToAFork es _ _ c) =
       "Switched to a fork, new tip: " <> renderPointAsPhrase (AF.headPoint c) <>
         Text.concat [ "\nEvent: " <> showT e | e <- es ]
-  forHuman (ChainDB.AddBlockValidation ev') = forHumanOrMachine ev'
+  forHuman (ChainDB.AddBlockValidation ev') = forHuman ev'
   forHuman (ChainDB.AddedBlockToVolatileDB pt _ _ enclosing) =
       case enclosing of
         RisingEdge  -> "Chain about to add block " <> renderRealPointAsPhrase pt
         FallingEdge -> "Chain added block " <> renderRealPointAsPhrase pt
-  forHuman (ChainDB.PipeliningEvent ev') = forHumanOrMachine ev'
+  forHuman (ChainDB.PipeliningEvent ev') = forHuman ev'
   forHuman (ChainDB.AddedReprocessLoEBlocksToQueue edgeSz) =
       case edgeSz of
         RisingEdge ->
@@ -1068,7 +1068,7 @@ instance MetaTrace (ChainDB.TraceGCEvent blk) where
 
 instance (ConvertRawHash blk, LedgerSupportsProtocol blk)
   => LogFormatting (ChainDB.TraceInitChainSelEvent blk) where
-    forHuman (ChainDB.InitChainSelValidation v) = forHumanOrMachine v
+    forHuman (ChainDB.InitChainSelValidation v) = forHuman v
     forHuman ChainDB.InitialChainSelected{} =
         "Initial chain selected"
     forHuman ChainDB.StartedInitChainSelection {} =
@@ -1336,7 +1336,7 @@ instance MetaTrace (ChainDB.TraceOpenEvent blk) where
 instance  ( StandardHash blk
           , ConvertRawHash blk
           ) => LogFormatting (ChainDB.TraceIteratorEvent blk) where
-  forHuman (ChainDB.UnknownRangeRequested ev') = forHumanOrMachine ev'
+  forHuman (ChainDB.UnknownRangeRequested ev') = forHuman ev'
   forHuman (ChainDB.BlockMissingFromVolatileDB realPt) = mconcat
     [ "This block is no longer in the VolatileDB because it has been garbage"
     , " collected. It might now be in the ImmDB if it was part of the"
@@ -2848,8 +2848,8 @@ instance (   LogFormatting (LedgerError blk)
     forMachine dtal (ExtValidationErrorLedger err) = forMachine dtal err
     forMachine dtal (ExtValidationErrorHeader err) = forMachine dtal err
 
-    forHuman (ExtValidationErrorLedger err) =  forHumanOrMachine err
-    forHuman (ExtValidationErrorHeader err) =  forHumanOrMachine err
+    forHuman (ExtValidationErrorLedger err) =  forHuman err
+    forHuman (ExtValidationErrorHeader err) =  forHuman err
 
     asMetrics (ExtValidationErrorLedger err) =  asMetrics err
     asMetrics (ExtValidationErrorHeader err) =  asMetrics err