tracing: fix runInLoop; forwarding interruptions properly traced

Author: mgmeier

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
@@ -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,26 +110,30 @@ 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)
+          howToConnect = Net.LocalPipe tracerSocket
           initForwConf = InitForwardingWith
             { initNetworkMagic          = toNetworkMagic networkId
             , initEKGStore              = Nothing
-            , initHowToConnect          = Net.LocalPipe tracerSocket
+            , initHowToConnect          = howToConnect
             , initForwarderMode         = Initiator
-            , initOnForwardInterruption = Nothing
+            , initOnForwardInterruption = Just $ \(SomeException e) ->
+                traceWith (btTxSubmit_ benchTracer) (TraceBenchForwardingInterrupted howToConnect $ show e)
             , initOnQueueOverflow       = Nothing
             }
         (forwardSink, dpStore, kickoffForwarder) <-
@@ -214,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
@@ -285,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"]
@@ -304,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
 
@@ -327,6 +345,7 @@ instance MetaTrace (TraceBenchTxSubmit TxId) where
         , Namespace [] ["BenchTxSubDebug"]
         , Namespace [] ["BenchTxSubError"]
         , Namespace [] ["BenchPlutusBudgetSummary"]
+        , Namespace [] ["ForwardingInterrupted"]
         ]
 
 instance LogFormatting NodeToNodeSubmissionTrace where

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

@@ -3,6 +3,7 @@
 {-# LANGUAGE PackageImports #-}
 {-# LANGUAGE RankNTypes #-}
 {-# LANGUAGE RecordWildCards #-}
+{-# LANGUAGE RecursiveDo #-}
 {-# LANGUAGE ScopedTypeVariables #-}
 
 module Cardano.Node.Tracing.API
@@ -37,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)
@@ -112,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
@@ -127,15 +129,16 @@ 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 ->
+          (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 = Nothing         -- TODO:MKarg
+                      , initOnForwardInterruption = Just $ \(SomeException e) ->
+                          traceWith (nodeStateTracer tracers) (NodeTracingForwardingInterrupted initHowToConnect $ show e)
                       , initOnQueueOverflow       = Nothing
                       , ..
                       }
@@ -144,21 +147,21 @@ initTraceDispatcher nc p networkMagic nodeKernel noBlockForging = do
                 forwardingConf = fromMaybe defaultForwarder (tcForwarder trConfig)
             in initForwardingDelayed iomgr forwardingConf initForwConf
 
-          pure (forwardTracer (writeToSink forwardSink), dataPointTracer dpStore, kickoffForwarder)
+          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"]

trace-dispatcher/src/Cardano/Logging/Types.hs

@@ -3,6 +3,7 @@
 {-# LANGUAGE DerivingStrategies #-}
 {-# LANGUAGE GADTs #-}
 {-# LANGUAGE GeneralizedNewtypeDeriving #-}
+{-# LANGUAGE MultiWayIf #-}
 {-# LANGUAGE RankNTypes #-}
 {-# LANGUAGE RecordWildCards #-}
 {-# LANGUAGE ScopedTypeVariables #-}
@@ -56,13 +57,14 @@ module Cardano.Logging.Types (
   , TraceObject(..)
   , PreFormatted(..)
   , HowToConnect(..)
-  , howToConnectString
 ) where
 
-
 import           Codec.Serialise (Serialise (..))
+import           Control.Applicative ((<|>))
+import           Control.DeepSeq (NFData)
 import qualified Control.Tracer as T
 import qualified Data.Aeson as AE
+import qualified Data.Aeson.Types as AE (Parser)
 import           Data.Bool (bool)
 import           Data.ByteString (ByteString)
 import qualified Data.HashMap.Strict as HM
@@ -72,7 +74,8 @@ import           Data.Map.Strict (Map)
 import qualified Data.Map.Strict as Map
 import           Data.Set (Set)
 import qualified Data.Set as Set
-import           Data.Text as T (Text, intercalate, null, pack, singleton, unpack, words)
+import           Data.Text as T (Text, breakOnEnd, intercalate, null, pack, singleton, unpack,
+                   unsnoc, words)
 import           Data.Text.Read as T (decimal)
 import           Data.Time (UTCTime)
 import           Data.Word (Word16)
@@ -617,9 +620,42 @@ type HowToConnect :: Type
 data HowToConnect
   = LocalPipe    !FilePath    -- ^ Local pipe (UNIX or Windows).
   | RemoteSocket !Host !Port  -- ^ Remote socket (host and port).
-  deriving stock (Eq, Show, Generic)
-
-howToConnectString :: HowToConnect -> String
-howToConnectString = \case
-  LocalPipe pipe -> pipe
-  RemoteSocket host port -> T.unpack host ++ ":" ++ show port
+  deriving stock (Eq, Generic)
+  deriving anyclass (NFData)
+
+instance Show HowToConnect where
+  show = \case
+    LocalPipe pipe         -> pipe
+    RemoteSocket host port -> T.unpack host ++ ":" ++ show port
+
+instance AE.ToJSON HowToConnect where
+  toJSON     = AE.toJSON . show
+  toEncoding = AE.toEncoding . show
+
+-- first try to host:port, and if that fails revert to parsing any
+-- string literal and assume it is a localpipe.
+instance AE.FromJSON HowToConnect where
+  parseJSON = AE.withText "HowToConnect" $ \t ->
+        (uncurry RemoteSocket <$> parseHostPort t)
+    <|> (        LocalPipe    <$> parseLocalPipe t)
+
+parseLocalPipe :: Text -> AE.Parser FilePath
+parseLocalPipe t
+  | T.null t = fail "parseLocalPipe: empty Text"
+  | otherwise   = pure $ T.unpack t
+
+parseHostPort :: Text -> AE.Parser (Text, Word16)
+parseHostPort t
+  | T.null t
+  = fail "parseHostPort: empty Text"
+  | otherwise
+  = let
+    (host_, portText) = T.breakOnEnd ":" t
+    host              = maybe "" fst (T.unsnoc host_)
+  in if
+    | T.null host      -> fail "parseHostPort: Empty host or no colon found."
+    | T.null portText  -> fail "parseHostPort: Empty port."
+    | Right (port, remainder) <- T.decimal portText
+    , T.null remainder
+    , 0 <= port, port <= 65535 -> pure (host, port)
+    | otherwise -> fail "parseHostPort: Non-numeric port or value out of range."