Add monitoring support for the local-cluster

justfile

@@ -31,7 +31,9 @@ bench target:
 local-cluster:
   nix shell '.#local-cluster' '.#cardano-node' '.#cardano-wallet' \
     -c "local-cluster" \
-    --cluster-configs lib/local-cluster/test/data/cluster-configs
+    --cluster-configs lib/local-cluster/test/data/cluster-configs \
+    --faucet-funds ${FAUCET_FUNDS_FILE} \
+    --monitoring-port 12798
 
 # run unit tests on a match
 unit-tests-cabal-match match:

lib/application-extras/lib/Cardano/Wallet/Network/Ports.hs

@@ -24,6 +24,7 @@ module Cardano.Wallet.Network.Ports
     -- * Helpers
     , portFromURL
     , randomUnusedTCPPorts
+    , validPorts
     ) where
 
 import Prelude
@@ -127,6 +128,9 @@ portFromURL uri = fromMaybe fallback
   where
     fallback = if uriScheme uri == "https:" then 443 else 80
 
+validPorts :: [Int]
+validPorts = [1024..49151]
+
 -- | Get a list of random TCPv4 ports that currently do not have any servers
 -- listening on them. It may return less than the requested number of ports.
 --
@@ -135,7 +139,7 @@ portFromURL uri = fromMaybe fallback
 -- listening socket to the child process.
 randomUnusedTCPPorts :: Int -> IO [Int]
 randomUnusedTCPPorts count = do
-    usablePorts <- shuffleM [1024..49151]
+    usablePorts <- shuffleM validPorts
     sort <$> filterM unused (take count usablePorts)
   where
     unused = fmap not . isPortOpen . simpleSockAddr (127,0,0,1) . fromIntegral

lib/benchmarks/exe/latency-bench.hs

@@ -673,6 +673,7 @@ withShelleyServer tracers action = withFaucet $ \faucetClientEnv -> do
                         , cfgNodeOutputFile = Nothing
                         }
             withCluster
+                (const $ pure ())
                 clusterConfig
                 faucetFunds
                 (onClusterStart cfgTestnetMagic setupAction db)

lib/integration/framework/Test/Integration/Framework/Cluster/Launch.hs

@@ -119,12 +119,13 @@ localClusterProcess CommandLineOptions{..} era = do
 
 withLocalCluster
     :: HasCallStack
-    => Config
+    => Int  -- ^ Port for monitoring the local cluster.
+    -> Config
     -> FaucetFunds
     -> (RunningNode -> IO a)
     -- ^ Action to run once when all pools have started.
     -> IO a
-withLocalCluster Config{..} faucetFunds run = do
+withLocalCluster monitoringPort Config{..} faucetFunds run = do
     r <- withTempFile $ \faucetFundsPath -> do
         let faucetFundsFile = FileOf $ absFile faucetFundsPath
             clusterConfigsDir = cfgClusterConfigs

lib/integration/framework/Test/Integration/Framework/Setup.hs

@@ -351,7 +351,7 @@ withServer
                             , cfgTracer = tr'
                             , cfgNodeOutputFile = nodeOutputFile
                             }
-                withCluster clusterConfig faucetFunds
+                withCluster (const $ pure ()) clusterConfig faucetFunds
                     $ onClusterStart
                         ctx
                         (onReady (T.pack smashUrl))

lib/local-cluster/exe/local-cluster.hs

@@ -6,6 +6,7 @@
 {-# LANGUAGE NumericUnderscores #-}
 {-# LANGUAGE OverloadedLabels #-}
 {-# LANGUAGE ScopedTypeVariables #-}
+{-# LANGUAGE TypeApplications #-}
 
 import Prelude
 
@@ -29,26 +30,44 @@ import Cardano.Wallet.Launch.Cluster.CommandLine
     ( CommandLineOptions (..)
     , parseCommandLineOptions
     )
+import Cardano.Wallet.Launch.Cluster.Control.Server
+    ( server
+    )
+import Cardano.Wallet.Launch.Cluster.Control.State
+    ( changePhase
+    , withControlLayer
+    )
 import Cardano.Wallet.Launch.Cluster.FileOf
     ( DirOf (..)
     , FileOf (..)
     , mkRelDirOf
     )
+import Control.Concurrent
+    ( threadDelay
+    )
 import Control.Exception
     ( bracket
     )
 import Control.Lens
     ( over
     )
+import Control.Monad
+    ( void
+    , (<=<)
+    )
 import Control.Monad.Cont
     ( ContT (..)
     )
 import Control.Monad.Trans
-    ( lift
+    ( MonadIO (..)
+    , MonadTrans (..)
     )
 import Main.Utf8
     ( withUtf8
     )
+import Network.Wai.Handler.Warp
+    ( run
+    )
 import System.Environment.Extended
     ( isEnvSet
     )
@@ -66,8 +85,9 @@ import System.Path
 import System.Path.Directory
     ( createDirectoryIfMissing
     )
-import UnliftIO.Concurrent
-    ( threadDelay
+import UnliftIO
+    ( async
+    , link
     )
 
 import qualified Cardano.Node.Cli.Launcher as NC
@@ -143,16 +163,24 @@ main = withUtf8 $ do
             $ Just
             $ mkRelDirOf
             $ Cluster.clusterEraToString clusterEra
-    CommandLineOptions{clusterConfigsDir, faucetFundsFile, clusterDir} <-
+    CommandLineOptions
+        { clusterConfigsDir
+        , faucetFundsFile
+        , clusterDir
+        , monitoringPort
+        } <-
         parseCommandLineOptions
     funds <- retrieveFunds faucetFundsFile
     flip runContT pure $ do
+        monitoring <- withControlLayer
+        liftIO $ link <=< async $ server monitoring >>= run monitoringPort
         clusterPath <-
             case clusterDir of
                 Just path -> pure path
                 Nothing ->
-                    fmap (DirOf . absDir) $
-                    ContT $ withSystemTempDir tr "test-cluster" skipCleanup
+                    fmap (DirOf . absDir)
+                        $ ContT
+                        $ withSystemTempDir tr "test-cluster" skipCleanup
         let clusterCfg =
                 Cluster.Config
                     { cfgStakePools = Cluster.defaultPoolConfigs
@@ -168,7 +196,9 @@ main = withUtf8 $ do
         let clusterDirPath = absDirOf clusterPath
             walletDir = clusterDirPath </> relDir "wallet"
         lift $ createDirectoryIfMissing True walletDir
-        node <- ContT $ Cluster.withCluster clusterCfg funds
+        node <-
+            ContT
+                $ Cluster.withCluster (changePhase monitoring) clusterCfg funds
         nodeSocket <-
             case parse . nodeSocketFile $ Cluster.runningNodeSocketPath node of
                 Left e -> error e
@@ -187,9 +217,9 @@ main = withUtf8 $ do
                             $ clusterDirPath
                                 </> relFile "byron-genesis.json"
                     }
-        lift
+        void
+            $ ContT
             $ bracket
                 (WC.start walletProcessConfig)
                 (WC.stop . fst)
-            $ const
-            $ threadDelay maxBound -- wait for Ctrl+C
+        liftIO $ threadDelay maxBound -- wait for Ctrl+C

lib/local-cluster/lib/Cardano/Wallet/Launch/Cluster/Cluster.hs

@@ -35,6 +35,9 @@ import Cardano.Wallet.Launch.Cluster.ConfiguredPool
     ( ConfiguredPool (..)
     , configurePools
     )
+import Cardano.Wallet.Launch.Cluster.Control.State
+    ( Phase (..)
+    )
 import Cardano.Wallet.Launch.Cluster.Faucet
     ( readFaucetAddresses
     , resetGlobals
@@ -164,23 +167,26 @@ data FaucetFunds = FaucetFunds
 -- The onClusterStart actions are not guaranteed to use the same node.
 withCluster
     :: HasCallStack
-    => Config
+    => (Phase -> IO ())
+    -> Config
     -> FaucetFunds
     -> (RunningNode -> IO a)
     -- ^ Action to run once when all pools have started.
     -> IO a
-withCluster config@Config{..} faucetFunds onClusterStart = runClusterM config
+withCluster phaseChange config@Config{..} faucetFunds onClusterStart
+    = runClusterM config
     $ bracketTracer' "withCluster"
     $ do
         let clusterDir = absDirOf cfgClusterDir
         traceClusterLog $ MsgHardFork cfgLastHardFork
+        phase Metadata
         withPoolMetadataServer $ \metadataServer -> do
             liftIO $ createDirectoryIfMissing True clusterDir
             traceClusterLog $ MsgStartingCluster cfgClusterDir
             liftIO resetGlobals
 
             configuredPools <- configurePools metadataServer cfgStakePools
-
+            phase Genesis
             addGenesisPools <- do
                 genesisDeltas <-
                     liftIO
@@ -195,7 +201,7 @@ withCluster config@Config{..} faucetFunds onClusterStart = runClusterM config
                 generateGenesis
                     (pureAdaFunds <> faucetAddresses <> massiveWalletFunds)
                     (addGenesisPools : cfgShelleyGenesisMods)
-
+            phase Genesis
             extraPort : poolsTcpPorts <-
                 liftIO
                     $ randomUnusedTCPPorts (length cfgStakePools + 1)
@@ -210,11 +216,15 @@ withCluster config@Config{..} faucetFunds onClusterStart = runClusterM config
                         pool0port
                         cfgNodeLogging
                         cfgNodeOutputFile
+            phase Pool0
             liftIO $ operatePool pool0 pool0Cfg $ \runningPool0 ->
                 runClusterM config $ do
+                    phase Funding
                     extraClusterSetupUsingNode configuredPools runningPool0
                     case NE.nonEmpty otherPools of
-                        Nothing -> liftIO $ onClusterStart runningPool0
+                        Nothing -> do
+                            phase Cluster
+                            liftIO $ onClusterStart runningPool0
                         Just others -> do
                             let relayNodeParams =
                                     NodeParams
@@ -230,18 +240,22 @@ withCluster config@Config{..} faucetFunds onClusterStart = runClusterM config
                                         , nodeParamsOutputFile
                                             = cfgNodeOutputFile
                                         }
+                            phase Pools
                             launchPools
                                 others
                                 genesisFiles
                                 poolPorts
                                 runningPool0
-                                $ \_poolNode ->
+                                $ \_poolNode -> do
+                                    phase Relay
                                     withRelayNode
-                                        relayNodeParams
-                                        onClusterStart
+                                        relayNodeParams $ \c -> do
+                                            phaseChange Cluster
+                                            onClusterStart c
   where
     FaucetFunds pureAdaFunds maryAllegraFunds massiveWalletFunds
         = faucetFunds
+    phase = liftIO . phaseChange
     -- Important cluster setup to run without rollbacks
     extraClusterSetupUsingNode
         :: NonEmpty ConfiguredPool -> RunningNode -> ClusterM ()

lib/local-cluster/lib/Cardano/Wallet/Launch/Cluster/CommandLine.hs

@@ -1,3 +1,4 @@
+{-# LANGUAGE ApplicativeDo #-}
 {-# LANGUAGE DataKinds #-}
 {-# LANGUAGE DerivingStrategies #-}
 
@@ -15,14 +16,25 @@ import Cardano.Wallet.Launch.Cluster.FileOf
     , FileOf (..)
     , newAbsolutizer
     )
+import Cardano.Wallet.Network.Ports
+    ( validPorts
+    )
+import Control.Monad
+    ( unless
+    )
+import Network.Wai.Handler.Warp
+    ( Port
+    )
 import Options.Applicative
     ( Parser
+    , auto
     , execParser
     , help
     , helper
     , info
     , long
     , metavar
+    , option
     , optional
     , progDesc
     , strOption
@@ -36,6 +48,7 @@ data CommandLineOptions = CommandLineOptions
     { clusterConfigsDir :: DirOf "cluster-configs"
     , faucetFundsFile :: FileOf "faucet-funds"
     , clusterDir :: Maybe (DirOf "cluster")
+    , monitoringPort :: Port
     }
     deriving stock (Show)
 
@@ -48,6 +61,7 @@ parseCommandLineOptions = do
                 <$> clusterConfigsDirParser absolutizer
                 <*> faucetFundsParser absolutizer
                 <*> clusterDirParser absolutizer
+                <*> portParser
                 <**> helper
             )
             (progDesc "Local Cluster for testing")
@@ -79,3 +93,18 @@ clusterDirParser (Absolutizer absOf) =
                     <> metavar "LOCAL_CLUSTER"
                     <> help "Path to the local cluster directory"
                 )
+
+portParser :: Parser Port
+portParser = do
+    option
+        parse
+        ( long "monitor-port"
+            <> metavar "MONITOR_PORT"
+            <> help "Port for the monitoring server"
+        )
+    where parse = do
+            p <- auto
+            unless (p `elem` validPorts) $
+                fail $ "Invalid port number. Must be inside: " ++
+                    show (head validPorts) ++ ".." ++ show (last validPorts)
+            pure p

lib/local-cluster/lib/Cardano/Wallet/Launch/Cluster/Config.hs

@@ -64,5 +64,7 @@ data Config = Config
     , cfgShelleyGenesisMods :: [ShelleyGenesisModifier]
     -- ^ Shelley genesis modifications to apply.
     , cfgTracer :: Tracer IO ClusterLog
+    -- ^ Tracer for logging.
     , cfgNodeOutputFile :: Maybe (FileOf "node-output")
+    -- ^ File to write node output to.
     }

lib/local-cluster/lib/Cardano/Wallet/Launch/Cluster/Control/API.hs

@@ -0,0 +1,27 @@
+{-# LANGUAGE DataKinds #-}
+{-# LANGUAGE DerivingStrategies #-}
+{-# LANGUAGE TypeOperators #-}
+
+module Cardano.Wallet.Launch.Cluster.Control.API
+    ( API
+    , proxyAPI) where
+
+import Cardano.Wallet.Launch.Cluster.Control.State
+    ( Phase
+    )
+import Data.Proxy
+    ( Proxy (..)
+    )
+import Data.Set
+    ( Set
+    )
+import Servant.API
+    ( Get
+    , JSON
+    , (:>)
+    )
+
+type API = "phase" :> Get '[JSON] (Set Phase)
+
+proxyAPI :: Proxy API
+proxyAPI = Proxy