cardano-tracer: Add functionality to run cardano-tracer as a library, with shut-down functionality and internal/user messaging.

Author: Icelandjack

cardano-tracer/app/cardano-tracer.hs

@@ -1,14 +1,23 @@
-import           Cardano.Tracer.CLI (TracerParams, parseTracerParams)
+{-# LANGUAGE OverloadedRecordDot #-}
+
+import           Cardano.Tracer.CLI (TracerParams(..), parseTracerParams)
+import           Cardano.Tracer.MetaTrace
 import           Cardano.Tracer.Run (runCardanoTracer)
 
+import           Data.Functor (void)
 import           Data.Version (showVersion)
 import           Options.Applicative
 
 import           Paths_cardano_tracer (version)
 
 main :: IO ()
-main =
-  runCardanoTracer =<< customExecParser (prefs showHelpOnEmpty) tracerInfo
+main = void do
+  tracerParams :: TracerParams
+     <- customExecParser (prefs showHelpOnEmpty) tracerInfo
+  trace :: Trace IO TracerTrace <-
+    -- Default `Nothing' severity filter to Info.
+    mkTracerTracer $ SeverityF (tracerParams.logSeverity <|> Just Info)
+  runCardanoTracer trace tracerParams
 
 tracerInfo :: ParserInfo TracerParams
 tracerInfo = info
@@ -21,7 +30,9 @@ tracerInfo = info
 
 versionOption :: Parser (a -> a)
 versionOption = infoOption
-  (showVersion version)
-  (long "version" <>
-   short 'v' <>
-   help "Show version")
+  do showVersion version
+  do mconcat
+       [ long  "version"
+       , short 'v'
+       , help  "Show version"
+       ]

cardano-tracer/cardano-tracer.cabal

@@ -200,6 +200,7 @@ library
                       , trace-dispatcher ^>= 2.10.0
                       , trace-forward ^>= 2.3.0
                       , trace-resources ^>= 0.2.3
+                      , unagi-chan
                       , wai ^>= 3.2
                       , warp ^>= 3.4
                       , yaml
@@ -294,6 +295,7 @@ library demo-acceptor-lib
   exposed-modules:      Cardano.Tracer.Test.Acceptor
 
   build-depends:        bytestring
+                      , QuickCheck
                       , cardano-tracer
                       , containers
                       , extra
@@ -306,9 +308,9 @@ library demo-acceptor-lib
                       , text
                       , trace-dispatcher
                       , trace-forward
+                      , unagi-chan
                       , vector
                       , vector-algorithms
-                      , QuickCheck
 
 executable demo-acceptor
   import:               project-config

cardano-tracer/src/Cardano/Tracer/Acceptors/Run.hs

@@ -1,3 +1,4 @@
+{-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE PackageImports #-}
 
@@ -14,6 +15,7 @@ import           Cardano.Tracer.Utils
 import           Cardano.Logging.Types (TraceObject)
 import qualified Cardano.Logging.Types as Net
 
+import           Control.Concurrent.Chan.Unagi (dupChan)
 import           Control.Concurrent.Async (forConcurrently_)
 import           "contra-tracer" Control.Tracer (Tracer, contramap, nullTracer, stdoutTracer)
 import qualified Data.List.NonEmpty as NE
@@ -33,20 +35,28 @@ import qualified Trace.Forward.Protocol.TraceObject.Type as TOF
 --   1. Server mode, when the tracer accepts connections from any number of nodes.
 --   2. Client mode, when the tracer initiates connections to specified number of nodes.
 runAcceptors :: TracerEnv -> TracerEnvRTView -> IO ()
-runAcceptors tracerEnv@TracerEnv{teTracer} tracerEnvRTView = do
+runAcceptors tracerEnv@TracerEnv{teTracer, teInChan = inChan} tracerEnvRTView = do
   traceWith teTracer $ TracerStartedAcceptors network
   case network of
-    AcceptAt howToConnect ->
+    AcceptAt howToConnect -> let
       -- Run one server that accepts connections from the nodes.
-      runInLoop
-        (runAcceptorsServer tracerEnv tracerEnvRTView howToConnect $ acceptorsConfigs (Net.howToConnectString howToConnect))
-        verbosity howToConnect initialPauseInSec
-    ConnectTo localSocks ->
+
+      action :: IO ()
+      action = do
+        dieOnShutdown =<< dupChan inChan
+        runAcceptorsServer tracerEnv tracerEnvRTView howToConnect $ acceptorsConfigs (Net.howToConnectString howToConnect)
+
+      in runInLoop action verbosity howToConnect initialPauseInSec
+    ConnectTo localSocks -> do
       -- Run N clients that initiate connections to the nodes.
-      forConcurrently_ (NE.nub localSocks) \howToConnect ->
-        runInLoop
-          (runAcceptorsClient tracerEnv tracerEnvRTView howToConnect $ acceptorsConfigs (Net.howToConnectString howToConnect))
-          verbosity howToConnect initialPauseInSec
+      forConcurrently_ (NE.nub localSocks) \howToConnect -> let
+
+        action :: IO ()
+        action = runAcceptorsClient tracerEnv tracerEnvRTView howToConnect $ acceptorsConfigs (Net.howToConnectString howToConnect)
+
+        in do
+          dieOnShutdown =<< dupChan inChan
+          runInLoop action verbosity howToConnect initialPauseInSec
  where
   TracerConfig{network, ekgRequestFreq, verbosity, ekgRequestFull} = teConfig tracerEnv
   ekgUseFullRequests = fromMaybe False ekgRequestFull

cardano-tracer/src/Cardano/Tracer/Environment.hs

@@ -1,11 +1,23 @@
 {-# LANGUAGE CPP #-}
+{-# LANGUAGE GADTs #-}
+{-# LANGUAGE LambdaCase #-}
 
 module Cardano.Tracer.Environment
   ( TracerEnv (..)
   , TracerEnvRTView (..)
+  , ChanMessage (..)
+  , InternalMessage (..)
+  , Tag (..)
+  , CardanoTracerMessage
+  , onChanMessage
+  , onInternal
+  , onUser
+  , dieOnShutdown
+  , forever'tilShutdown
   ) where
 
 import           Cardano.Logging.Types
+import           Cardano.Logging.Resources.Types (ResourceStats)
 import           Cardano.Tracer.Configuration
 #if RTVIEW
 import           Cardano.Tracer.Handlers.Notifications.Types
@@ -16,10 +28,13 @@ import           Cardano.Tracer.Handlers.State.TraceObjects
 import           Cardano.Tracer.MetaTrace
 import           Cardano.Tracer.Types
 
+import           Control.Concurrent (myThreadId)
+import           Control.Exception (AsyncException(ThreadKilled), throwTo)
+import           Control.Concurrent.Chan.Unagi (InChan, OutChan, readChan)
 import           Control.Concurrent.Extra (Lock)
 import           Data.Text (Text)
 import           Data.Text.Lazy.Builder (Builder)
-
+import           Data.Kind (Type)
 
 -- | Environment for all functions.
 data TracerEnv = TracerEnv
@@ -36,6 +51,7 @@ data TracerEnv = TracerEnv
   , teRegistry              :: !HandleRegistry
   , teStateDir              :: !(Maybe FilePath)
   , teMetricsHelp           :: ![(Text, Builder)]
+  , teInChan                :: !(InChan (CardanoTracerMessage ()))
   }
 
 #if RTVIEW
@@ -51,3 +67,39 @@ data TracerEnvRTView = TracerEnvRTView
 #else
 data TracerEnvRTView = TracerEnvRTView
 #endif
+
+type CardanoTracerMessage userMsg = ChanMessage InternalMessage userMsg
+
+type ChanMessage :: Type -> Type -> Type
+data ChanMessage internal user
+  = Shutdown
+  | InternalMessage internal
+  | UserMessage     user
+
+onChanMessage :: (internal -> IO ()) -> (user -> IO ()) -> OutChan (ChanMessage internal user) -> IO ()
+onChanMessage internalAction userAction outChan = 
+  readChan outChan >>= \case
+    Shutdown                 -> myThreadId >>= (`throwTo` ThreadKilled)
+    InternalMessage internal -> internalAction internal
+    UserMessage user         -> userAction user
+
+onInternal :: (internal -> IO ()) -> OutChan (ChanMessage internal user) -> IO ()
+onInternal = (`onChanMessage` mempty)
+
+onUser :: (user -> IO ()) -> OutChan (ChanMessage internal user) -> IO ()
+onUser = (mempty `onChanMessage`)
+
+data InternalMessage where
+  ResourceMessage :: Tag ex -> (ex -> IO ()) -> InternalMessage
+
+data Tag a where
+  TagResource :: Tag (ResourceStats, Trace IO TracerTrace)
+
+dieOnShutdown :: OutChan (ChanMessage internal user) -> IO ()
+dieOnShutdown = onChanMessage mempty mempty
+
+forever'tilShutdown :: OutChan (ChanMessage internal user) -> IO () -> IO ()
+forever'tilShutdown outChan action = do
+  readChan outChan >>= \case
+    Shutdown -> pure ()
+    _        -> action *> forever'tilShutdown outChan action

cardano-tracer/src/Cardano/Tracer/Handlers/Logs/Rotator.hs

@@ -6,16 +6,17 @@ module Cardano.Tracer.Handlers.Logs.Rotator
   ) where
 
 import           Cardano.Tracer.Configuration
-import           Cardano.Tracer.Environment
+import           Cardano.Tracer.Environment (TracerEnv (..), forever'tilShutdown)
 import           Cardano.Tracer.Handlers.Logs.Utils (createOrUpdateEmptyLog, getTimeStampFromLog,
                    isItLog)
 import           Cardano.Tracer.MetaTrace
 import           Cardano.Tracer.Types (HandleRegistry, HandleRegistryKey, NodeName)
 import           Cardano.Tracer.Utils (showProblemIfAny, readRegistry)
 
 import           Control.Concurrent.Async (forConcurrently_)
+import           Control.Concurrent.Chan.Unagi (dupChan)
 import           Control.Concurrent.Extra (Lock)
-import           Control.Monad (forM_, forever, unless, when)
+import           Control.Monad (forM_, unless, when)
 import           Control.Monad.Extra (whenJust, whenM)
 import           Data.Foldable (for_)
 import           Data.List (nub, sort)
@@ -33,38 +34,40 @@ import           System.Time.Extra (sleep)
 
 -- | Runs rotation mechanism for the log files.
 runLogsRotator :: TracerEnv -> IO ()
-runLogsRotator TracerEnv
-  { teConfig = TracerConfig{rotation, verbosity, logging}
-  , teCurrentLogLock
-  , teTracer
-  , teRegistry
-  } = do
-  whenJust rotation \rotParams -> do
+runLogsRotator tracerEnv@TracerEnv { teConfig = TracerConfig{rotation}, teTracer } = do
+  whenJust rotation \rot -> do
     traceWith teTracer TracerStartedLogRotator
-    launchRotator loggingParamsForFiles rotParams verbosity teRegistry teCurrentLogLock
- where
+    launchRotator tracerEnv rot
+
+launchRotator
+  :: TracerEnv
+  -> RotationParams
+  -> IO ()
+launchRotator tracerEnv rot@RotationParams{rpFrequencySecs} = do
+  whenNonEmpty loggingParamsForFiles do
+    outChan <- dupChan teInChan
+    forever'tilShutdown outChan do
+      showProblemIfAny verbosity do
+        forM_ loggingParamsForFiles \loggingParam -> do
+          checkRootDir teCurrentLogLock teRegistry rot loggingParam
+      sleep (fromIntegral rpFrequencySecs)
+  where
+  whenNonEmpty :: Applicative f => [a] -> f () -> f ()
+  whenNonEmpty = unless . null
+
+  TracerEnv
+    { teConfig = TracerConfig{verbosity, logging}
+    , teCurrentLogLock
+    , teRegistry
+    , teInChan
+    } = tracerEnv
+
   loggingParamsForFiles :: [LoggingParams]
   loggingParamsForFiles = nub (NE.filter filesOnly logging)
 
   filesOnly :: LoggingParams -> Bool
   filesOnly LoggingParams{logMode} = logMode == FileMode
 
-launchRotator
-  :: [LoggingParams]
-  -> RotationParams
-  -> Maybe Verbosity
-  -> HandleRegistry
-  -> Lock
-  -> IO ()
-launchRotator [] _ _ _ _ = return ()
-launchRotator loggingParamsForFiles
-              rotParams@RotationParams{rpFrequencySecs} verb registry currentLogLock =
-  forever do
-    showProblemIfAny verb do
-      forM_ loggingParamsForFiles \loggingParam -> do
-        checkRootDir currentLogLock registry rotParams loggingParam
-    sleep $ fromIntegral rpFrequencySecs
-
 -- | All the logs with 'TraceObject's received from particular node
 --   will be stored in a separate subdirectory in the root directory.
 --

cardano-tracer/src/Cardano/Tracer/Handlers/Metrics/Monitoring.hs

@@ -1,3 +1,4 @@
+{-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE ScopedTypeVariables #-}
@@ -14,6 +15,9 @@ import           Cardano.Tracer.Types
 
 import           Prelude hiding (head)
 
+import           Control.Concurrent (myThreadId)
+import           Control.Concurrent.Chan.Unagi (OutChan, readChan, dupChan)
+import           Control.Exception (AsyncException(ThreadKilled), throwTo)
 import           Data.ByteString as ByteString (ByteString, isInfixOf)
 import           Data.ByteString.Builder (stringUtf8)
 import qualified Data.Text as T
@@ -39,19 +43,24 @@ runMonitoringServer
   -> Endpoint -- ^ (web page with list of connected nodes, EKG web page).
   -> IO RouteDictionary
   -> IO ()
-runMonitoringServer TracerEnv{teTracer} endpoint computeRoutes_autoUpdate = do
+runMonitoringServer TracerEnv{teTracer, teInChan = inChan} endpoint computeRoutes_autoUpdate = do
   -- Pause to prevent collision between "Listening"-notifications from servers.
   sleep 0.2
   traceWith teTracer TracerStartedMonitoring
     { ttMonitoringEndpoint = endpoint
     , ttMonitoringType     = "list"
     }
   dummyStore <- EKG.newStore
-  runSettings (setEndpoint endpoint defaultSettings) do
-    renderEkg dummyStore computeRoutes_autoUpdate
+  runSettings (setEndpoint endpoint defaultSettings) \req respond -> do
+    outChan <- dupChan inChan
+    renderEkg dummyStore outChan computeRoutes_autoUpdate req respond
+
+renderEkg :: EKG.Store -> OutChan (CardanoTracerMessage ()) -> IO RouteDictionary -> Application
+renderEkg dummyStore outChan computeRoutes_autoUpdate request send = do
+  readChan outChan >>= \case
+    Shutdown -> myThreadId >>= (`throwTo` ThreadKilled)
+    _ -> pure ()
 
-renderEkg :: EKG.Store -> IO RouteDictionary -> Application
-renderEkg dummyStore computeRoutes_autoUpdate request send = do
   routeDictionary :: RouteDictionary <-
     computeRoutes_autoUpdate
 

cardano-tracer/src/Cardano/Tracer/Handlers/Metrics/Prometheus.hs

@@ -1,3 +1,4 @@
+{-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE ScopedTypeVariables #-}
@@ -14,6 +15,9 @@ import           Cardano.Tracer.MetaTrace
 
 import           Prelude hiding (head)
 
+import           Control.Concurrent (myThreadId)
+import           Control.Concurrent.Chan.Unagi (OutChan, readChan, dupChan)
+import           Control.Exception (AsyncException(ThreadKilled), throwTo)
 import qualified Data.ByteString as ByteString
 import           Data.ByteString.Builder (stringUtf8)
 import           Data.Functor ((<&>))
@@ -67,26 +71,33 @@ runPrometheusServer tracerEnv endpoint computeRoutes_autoUpdate = do
   -- If everything is okay, the function 'simpleHttpServe' never returns.
   -- But if there is some problem, it never throws an exception, but just stops.
   -- So if it stopped - it will be re-started.
-  traceWith teTracer TracerStartedPrometheus
+  traceWith tracer TracerStartedPrometheus
     { ttPrometheusEndpoint = endpoint
     }
-  runSettings (setEndpoint endpoint defaultSettings) do
-    renderPrometheus computeRoutes_autoUpdate noSuffix teMetricsHelp
+  runSettings (setEndpoint endpoint defaultSettings) \req respond -> do
+    outChan <- dupChan inChan
+    renderPrometheus computeRoutes_autoUpdate outChan noSuffix metricsHelp req respond
   where
     TracerEnv
-      { teTracer
-      , teConfig = TracerConfig { metricsNoSuffix }
-      , teMetricsHelp
+      { teTracer      = tracer
+      , teConfig      = TracerConfig { metricsNoSuffix }
+      , teMetricsHelp = metricsHelp
+      , teInChan      = inChan
       } = tracerEnv
 
     noSuffix = or @Maybe metricsNoSuffix
 
 renderPrometheus
   :: IO RouteDictionary
+  -> OutChan (CardanoTracerMessage ())
   -> Bool
   -> [(Text, Builder)]
   -> Application
-renderPrometheus computeRoutes_autoUpdate noSuffix helpTextDict request send = do
+renderPrometheus computeRoutes_autoUpdate outChan noSuffix helpTextDict request send = do
+  readChan outChan >>= \case
+    Shutdown -> myThreadId >>= (`throwTo` ThreadKilled)
+    _ -> pure ()
+
   routeDictionary :: RouteDictionary <-
     computeRoutes_autoUpdate
 

cardano-tracer/src/Cardano/Tracer/Handlers/Notifications/Timer.hs

@@ -110,3 +110,5 @@ mkTimerOnFailure onFailure tracer io state callPeriod_sec = do
     , startTimer    = modifyIORef' isRunning (const True)
     , stopTimer     = modifyIORef' isRunning (const False)
     }
+
+    -- TODO: Store last timestamp