trace-forward: use config record type InitForwardingConfig

mgmeier · mgmeier · commit d9bda00a8eee · 2025-11-26T11:25:46.000+01:00
diff --git a/bench/tx-generator/src/Cardano/Benchmarking/Tracer.hs b/bench/tx-generator/src/Cardano/Benchmarking/Tracer.hs
@@ -49,7 +49,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
@@ -120,9 +120,18 @@ initTxGenTracers mbForwarding = do
   prepareForwardingTracer :: IO (Maybe (Trace IO FormattedMessage))
   prepareForwardingTracer = forM mbForwarding $
     \(iomgr, networkId, tracerSocket) -> do
-        let forwardingConf = fromMaybe defaultForwarder (tcForwarder initialTraceConfig)
+        let
+          forwardingConf = fromMaybe defaultForwarder (tcForwarder initialTraceConfig)
+          initForwConf = InitForwardingWith
+            { initNetworkMagic          = toNetworkMagic networkId
+            , initEKGStore              = Nothing
+            , initHowToConnect          = Net.LocalPipe tracerSocket
+            , initForwarderMode         = Initiator
+            , initOnForwardInterruption = Nothing
+            , 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').
diff --git a/cardano-node/src/Cardano/Node/Tracing/API.hs b/cardano-node/src/Cardano/Node/Tracing/API.hs
@@ -2,6 +2,7 @@
 {-# LANGUAGE MonoLocalBinds #-}
 {-# LANGUAGE PackageImports #-}
 {-# LANGUAGE RankNTypes #-}
+{-# LANGUAGE RecordWildCards #-}
 {-# LANGUAGE ScopedTypeVariables #-}
 
 module Cardano.Node.Tracing.API
@@ -46,7 +47,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)
 
 
@@ -126,13 +127,23 @@ 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 = Nothing         -- TODO:MKarg
+                      , initOnQueueOverflow       = Nothing
+                      , ..
+                      }
 
                 forwardingConf :: TraceOptionForwarder
                 forwardingConf = fromMaybe defaultForwarder (tcForwarder trConfig)
-            initForwardingDelayed iomgr forwardingConf networkMagic (Just ekgStore) tracerSocketMode
+            in initForwardingDelayed iomgr forwardingConf initForwConf
+
           pure (forwardTracer (writeToSink forwardSink), dataPointTracer dpStore, kickoffForwarder)
         else
           -- Since 'Forwarder' backend isn't enabled, there is no forwarding.
diff --git a/cardano-tracer/src/Cardano/Tracer/Handlers/ReForwarder.hs b/cardano-tracer/src/Cardano/Tracer/Handlers/ReForwarder.hs
@@ -1,5 +1,6 @@
 {-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE RecordWildCards #-}
 {-# LANGUAGE ScopedTypeVariables #-}
 {-# LANGUAGE TypeApplications #-}
 {-# LANGUAGE ViewPatterns #-}
@@ -19,7 +20,6 @@ module Cardano.Tracer.Handlers.ReForwarder
 import           Cardano.Logging.Trace
 import           Cardano.Logging.Tracer.DataPoint
 import qualified Cardano.Logging.Types as Log
-import qualified Cardano.Logging.Types as Net
 import           Cardano.Tracer.Configuration
 import           Cardano.Tracer.Handlers.Utils (normalizeNamespace)
 import           Cardano.Tracer.MetaTrace
@@ -47,34 +47,17 @@ initReForwarder TracerConfig{networkMagic, hasForwarding}
                 teTracer = do
   mForwarding <- case hasForwarding of
       Nothing -> pure Nothing
-      Just x  -> case x of
-        (ConnectTo{}, _, _) ->
+      Just (ConnectTo{}, _, _) ->
           error "initReForwarder:  unsupported mode of operation:  ConnectTo.  Use AcceptAt."
-        (AcceptAt (LocalPipe socket), flattenNS -> mFwdNames, forwConf) -> do
+      Just (AcceptAt howToConnect, flattenNS -> mFwdNames, forwConf) -> do
           (fwdsink, dpStore :: DataPointStore) <- withIOManager \iomgr -> do
             traceWith teTracer TracerStartedReforwarder
-            initForwarding iomgr forwConf
-                                 (NetworkMagic networkMagic)
-                                 Nothing
-                                 (Just (Net.LocalPipe socket, Log.Responder))
+            initForwarding iomgr forwConf $ initForwardingWith howToConnect
           pure $ Just ( filteredWriteToSink
                           (traceObjectHasPrefixIn mFwdNames)
                           fwdsink
                       , dataPointTracer @IO dpStore
                       )
-        (AcceptAt (RemoteSocket host port), flattenNS -> mFwdNames, forwConf) -> do
-          (fwdsink, dpStore :: DataPointStore) <- withIOManager \iomgr -> do
-            traceWith teTracer TracerStartedReforwarder
-            initForwarding iomgr forwConf
-                                 (NetworkMagic networkMagic)
-                                 Nothing
-                                 (Just (Net.RemoteSocket host port, Log.Responder))
-          pure $ Just ( filteredWriteToSink
-                          (traceObjectHasPrefixIn mFwdNames)
-                          fwdsink
-                      , dataPointTracer @IO dpStore
-                      )
-
   let traceDP = case mForwarding of
                   Just (_,tr) -> tr
                   Nothing     -> mempty
@@ -90,14 +73,22 @@ initReForwarder TracerConfig{networkMagic, hasForwarding}
   where
     flattenNS = fmap (map (Text.intercalate "."))
 
+    initForwardingWith initHowToConnect =
+      InitForwardingWith
+        { initNetworkMagic          = NetworkMagic networkMagic
+        , initEKGStore              = Nothing
+        , initForwarderMode         = Log.Responder
+        , initOnForwardInterruption = Nothing         -- TODO:MKarg
+        , initOnQueueOverflow       = Nothing
+        , ..
+        }
 
 traceObjectHasPrefixIn :: Maybe [Text.Text] -> Log.TraceObject -> Bool
 traceObjectHasPrefixIn mFwdNames (normalizeNamespace . Log.toNamespace -> ns) =
   case mFwdNames of
     Nothing       -> True -- forward everything in this case
     Just fwdNames -> any (`Text.isPrefixOf` ns) fwdNames
 
-
 filteredWriteToSink :: (Log.TraceObject -> Bool)
                     -> ForwardSink Log.TraceObject
                     -> Log.TraceObject -> IO ()
diff --git a/cardano-tracer/test/cardano-tracer-test-ext.hs b/cardano-tracer/test/cardano-tracer-test-ext.hs
@@ -30,7 +30,7 @@ import qualified System.Process as Sys
 import           Test.Tasty
 import           Test.Tasty.QuickCheck
 
-import           Trace.Forward.Forwarding (initForwarding)
+import           Trace.Forward.Forwarding (InitForwardingConfig (..), initForwarding)
 import           Trace.Forward.Utils.TraceObject (writeToSink)
 
 main :: IO ()
@@ -137,7 +137,14 @@ getExternalTracerState TestSetup{..} ref = do
      (forwardSink, _dpStore) <- withIOManager \iomgr -> do
        -- For simplicity, we are always 'Initiator',
        -- so 'cardano-tracer' is always a 'Responder'.
-       let tracerSocketMode = Just (Net.LocalPipe (unI tsSockExternal), Initiator)
-           forwardingConf = fromMaybe defaultForwarder (tcForwarder simpleTestConfig)
-       initForwarding iomgr forwardingConf (unI tsNetworkMagic) Nothing tracerSocketMode
+       let forwardingConf = fromMaybe defaultForwarder (tcForwarder simpleTestConfig)
+       initForwarding iomgr forwardingConf $
+         InitForwardingWith
+           { initNetworkMagic          = unI tsNetworkMagic
+           , initEKGStore              = Nothing
+           , initHowToConnect          = Net.LocalPipe (unI tsSockExternal)
+           , initForwarderMode         = Initiator
+           , initOnForwardInterruption = Nothing
+           , initOnQueueOverflow       = Nothing
+           }
      pure (externalTracerHdl, forwardTracer (writeToSink forwardSink))
diff --git a/trace-forward/src/Trace/Forward/Forwarding.hs b/trace-forward/src/Trace/Forward/Forwarding.hs
@@ -3,12 +3,16 @@
 {-# LANGUAGE DataKinds #-}
 {-# LANGUAGE FlexibleInstances #-}
 {-# LANGUAGE PackageImports #-}
+{-# LANGUAGE RecordWildCards #-}
 {-# LANGUAGE ScopedTypeVariables #-}
 {-# LANGUAGE TypeApplications #-}
 {-# LANGUAGE ViewPatterns #-}
 
+{-# OPTIONS_GHC -Wno-partial-fields #-}
+
 module Trace.Forward.Forwarding
-  ( initForwarding
+  ( InitForwardingConfig(..)
+  , initForwarding
   , initForwardingDelayed
   ) where
 
@@ -26,22 +30,21 @@ import           Ouroboros.Network.Protocol.Handshake.Codec (cborTermVersionData
 import           Ouroboros.Network.Protocol.Handshake.Type (Handshake)
 import           Ouroboros.Network.Protocol.Handshake.Version (acceptableVersion, queryVersion,
                    simpleSingletonVersions)
-import           Ouroboros.Network.Snocket (MakeBearer, Snocket, localAddressFromPath, localSnocket,
-                   makeLocalBearer, LocalAddress, socketSnocket, makeSocketBearer, LocalSocket)
-import           Ouroboros.Network.Socket (ConnectToArgs (..),
-                   HandshakeCallbacks (..), SomeResponderApplication (..),
-                   connectToNode, nullNetworkConnectTracers)
 import qualified Ouroboros.Network.Server.Simple as Server
+import           Ouroboros.Network.Snocket (LocalAddress, LocalSocket, MakeBearer, Snocket,
+                   localAddressFromPath, localSnocket, makeLocalBearer, makeSocketBearer,
+                   socketSnocket)
+import           Ouroboros.Network.Socket (ConnectToArgs (..), HandshakeCallbacks (..),
+                   SomeResponderApplication (..), connectToNode, nullNetworkConnectTracers)
 
 import           Codec.CBOR.Term (Term)
 import           Control.Concurrent.Async (async, wait)
-import           Control.Exception (throwIO)
+import           Control.Exception (SomeException, throwIO)
 import           Control.Monad.IO.Class
 import           "contra-tracer" Control.Tracer (Tracer, contramap, nullTracer, stdoutTracer)
 import qualified Data.ByteString.Lazy as LBS
 import           Data.Functor
 import           Data.List.NonEmpty (NonEmpty ((:|)))
-import           Data.Maybe (isNothing)
 import qualified Data.Text as Text
 import           Data.Void (Void, absurd)
 import           Data.Word (Word16)
@@ -61,15 +64,37 @@ import           Trace.Forward.Utils.ForwardSink (ForwardSink)
 import           Trace.Forward.Utils.TraceObject
 import           Trace.Forward.Utils.Version
 
+
+-- | Config record to initialise trace forwarding
+data InitForwardingConfig
+  = -- | Only construct relevant values, but do not actually run forwarding
+    InitForwardingNone
+  | -- | Run forwarding with the provided settings
+    InitForwardingWith
+    { initNetworkMagic          :: !NetworkMagic
+    -- ^ Forwarding is always tied to a singular networkId
+    , initEKGStore              :: !(Maybe EKG.Store)
+    -- ^ A metrics store to be forwarded (optional)
+    , initHowToConnect          :: !HowToConnect
+    -- ^ A LocalPipe or RemoteSocket
+    , initForwarderMode         :: !ForwarderMode
+    -- ^ Run as Initiator or Responder
+    , initOnForwardInterruption :: !(Maybe (SomeException -> IO ()))
+    -- ^ Optional handler when forwarding connection is interrupted (may be temporary or permanent)
+    --   default: no action
+    , initOnQueueOverflow       :: !(Maybe ([TraceObject] -> IO ()))
+    -- ^ Optional handler when forwarding queue overflows (argument are objects dropped from queue)
+    --   default: print one-liner to stderr, indicating object count and timestamps of first and last object
+    }
+
+
 initForwarding :: forall m. (MonadIO m)
   => IOManager
   -> TraceOptionForwarder
-  -> NetworkMagic
-  -> Maybe EKG.Store
-  -> Maybe (HowToConnect, ForwarderMode)
+  -> InitForwardingConfig
   -> m (ForwardSink TraceObject, DataPointStore)
-initForwarding iomgr config magic ekgStore tracerSocketMode = do
-  (a, b, kickoffForwarder) <- initForwardingDelayed iomgr config magic ekgStore tracerSocketMode
+initForwarding iomgr config forwarding = do
+  (a, b, kickoffForwarder) <- initForwardingDelayed iomgr config forwarding
   liftIO kickoffForwarder
   pure (a, b)
 
@@ -79,38 +104,34 @@ initForwardingDelayed :: forall m. ()
   => MonadIO m
   => IOManager
   -> TraceOptionForwarder
-  -> NetworkMagic
-  -> Maybe EKG.Store
-  -> Maybe (HowToConnect, ForwarderMode)
+  -> InitForwardingConfig
   -> m (ForwardSink TraceObject, DataPointStore, IO ())
-initForwardingDelayed iomgr config magic ekgStore tracerSocketMode = liftIO $ do
-  let ignoreOverflow, onOverflow :: [TraceObject] -> IO ()
-      ignoreOverflow _ =
-        pure ()
-      onOverflow | isNothing tracerSocketMode = ignoreOverflow
-                 | otherwise                  = handleOverflow
+initForwardingDelayed iomgr config forwarding = liftIO $ do
+  let onOverflow :: [TraceObject] -> IO ()
+      onOverflow = case forwarding of
+        InitForwardingNone                                      -> const $ pure ()
+        InitForwardingWith{initOnQueueOverflow = Just handler}  -> handler
+        InitForwardingWith{initOnQueueOverflow = Nothing}       -> handleOverflow
   forwardSink <- initForwardSink tfConfig onOverflow
   dpStore <- initDataPointStore
   let
     kickoffForwarder = launchForwarders
       iomgr
-      magic
+      forwarding
       ekgConfig
       tfConfig
       dpfConfig
-      ekgStore
       forwardSink
       dpStore
-      tracerSocketMode
       maxReconnectDelay
   pure (forwardSink, dpStore, kickoffForwarder)
  where
   endpoint :: EKGF.HowToConnect
   endpoint =
-    case tracerSocketMode of
-      Nothing -> EKGF.LocalPipe ""
-      Just (LocalPipe str, _mode) -> EKGF.LocalPipe str
-      Just (RemoteSocket host port, _mode) -> EKGF.RemoteSocket host port
+    case forwarding of
+      InitForwardingNone                                            -> EKGF.LocalPipe ""
+      InitForwardingWith{initHowToConnect = LocalPipe str}          -> EKGF.LocalPipe str
+      InitForwardingWith{initHowToConnect = RemoteSocket host port} -> EKGF.RemoteSocket host port
   queueSize         = tofQueueSize         config
   verbosity         = tofVerbosity         config
   maxReconnectDelay = tofMaxReconnectDelay config
@@ -157,39 +178,37 @@ handleOverflow (msg : msgs) =
 
 launchForwarders
   :: IOManager
-  -> NetworkMagic
+  -> InitForwardingConfig
   -> EKGF.ForwarderConfiguration
   -> TF.ForwarderConfiguration TraceObject
   -> DPF.ForwarderConfiguration
-  -> Maybe EKG.Store
   -> ForwardSink TraceObject
   -> DataPointStore
-  -> Maybe (HowToConnect, ForwarderMode)
   -> Word
   -> IO ()
-launchForwarders iomgr magic
+launchForwarders iomgr forwarding
                  ekgConfig tfConfig dpfConfig
-                 ekgStore sink dpStore tracerSocketMode maxReconnectDelay =
-  -- If 'tracerSocketMode' is not specified, it's impossible to establish
-  -- network connection with acceptor application (for example, 'cardano-tracer').
+                 sink dpStore maxReconnectDelay =
+  -- If InitForwardingNone is specified, it's impossible to establish
+  -- a connection with an acceptor application (for example, 'cardano-tracer').
   -- In this case, we should not launch forwarders.
-  case tracerSocketMode of
-    Nothing -> return ()
-    Just (socketPath, mode) ->
+  case forwarding of
+    InitForwardingNone     -> return ()
+    InitForwardingWith{..} ->
       void . async $
         runInLoop
           (launchForwardersViaLocalSocket
              iomgr
-             magic
-             socketPath
-             mode
+             initNetworkMagic
+             initHowToConnect
+             initForwarderMode
              ekgConfig
              tfConfig
              dpfConfig
              sink
-             ekgStore
+             initEKGStore
              dpStore)
-          socketPath
+          initHowToConnect
           1
           maxReconnectDelay
 
